79 lines
2.2 KiB
Python
79 lines
2.2 KiB
Python
"""
|
|
Authentication Service
|
|
JWT token management and password hashing
|
|
"""
|
|
import uuid
|
|
import bcrypt
|
|
from datetime import datetime, timedelta
|
|
from jose import jwt
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app import config
|
|
from database.models import User
|
|
|
|
|
|
def hash_password(password: str) -> str:
|
|
"""Hash a password using bcrypt"""
|
|
return bcrypt.hashpw(password.encode(), bcrypt.gensalt()).decode()
|
|
|
|
|
|
def verify_password(password: str, hashed: str) -> bool:
|
|
"""Verify a password against its hash"""
|
|
return bcrypt.checkpw(password.encode(), hashed.encode())
|
|
|
|
|
|
def create_access_token(user_id: str, email: str) -> str:
|
|
"""Create a JWT access token"""
|
|
expire = datetime.utcnow() + timedelta(minutes=config.ACCESS_TOKEN_EXPIRE_MINUTES)
|
|
payload = {
|
|
"sub": user_id,
|
|
"email": email,
|
|
"exp": expire
|
|
}
|
|
return jwt.encode(payload, config.SECRET_KEY, algorithm=config.ALGORITHM)
|
|
|
|
|
|
def decode_token(token: str) -> dict | None:
|
|
"""Decode and validate a JWT token"""
|
|
try:
|
|
return jwt.decode(token, config.SECRET_KEY, algorithms=[config.ALGORITHM])
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def get_user_by_email(db: Session, email: str) -> User | None:
|
|
"""Get user by email address"""
|
|
return db.query(User).filter(User.email == email).first()
|
|
|
|
|
|
def get_user_by_id(db: Session, user_id: str) -> User | None:
|
|
"""Get user by ID"""
|
|
return db.query(User).filter(User.id == user_id).first()
|
|
|
|
|
|
def authenticate_user(db: Session, email: str, password: str) -> User | None:
|
|
"""Authenticate user with email and password"""
|
|
user = get_user_by_email(db, email)
|
|
if not user or not user.password_hash:
|
|
return None
|
|
if not verify_password(password, user.password_hash):
|
|
return None
|
|
return user
|
|
|
|
|
|
def create_user(db: Session, email: str, password: str, name: str, role: str = "employee") -> User:
|
|
"""Create a new user"""
|
|
user = User(
|
|
id=str(uuid.uuid4()),
|
|
email=email,
|
|
password_hash=hash_password(password),
|
|
name=name,
|
|
role=role,
|
|
created_at=datetime.utcnow(),
|
|
updated_at=datetime.utcnow()
|
|
)
|
|
db.add(user)
|
|
db.commit()
|
|
db.refresh(user)
|
|
return user
|