from datetime import datetime, timedelta, timezone from hashlib import sha256 import secrets from typing import Annotated, Any, Literal import jwt from fastapi import Depends, Header, HTTPException, Request, Security, status from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer from pydantic import BaseModel, Field from app.config import Settings, get_settings Role = Literal["VIEWER", "SALES", "SUPERVISOR", "OWNER"] class Principal(BaseModel): user_code: str = Field(max_length=50) user_name: str = Field(max_length=50) role: Role kode_sales: str | None = None @property def is_salesman(self) -> bool: return self.role == "SALES" or bool(self.kode_sales) @property def is_owner(self) -> bool: return self.role in ("OWNER", "SUPERVISOR") bearer = HTTPBearer(auto_error=False) def create_access_token( data: dict[str, Any], settings: Settings, expires_delta: timedelta | None = None, ) -> tuple[str, int]: to_encode = data.copy() if expires_delta: expire_seconds = int(expires_delta.total_seconds()) else: expire_seconds = settings.jwt_expires_minutes * 60 now = datetime.now(timezone.utc) expire = now + timedelta(seconds=expire_seconds) to_encode.update({"iat": now, "exp": expire}) encoded_jwt = jwt.encode( to_encode, settings.jwt_secret_key.get_secret_value(), algorithm=settings.jwt_algorithm, ) return encoded_jwt, expire_seconds def decode_jwt_token(token: str, settings: Settings) -> dict[str, Any] | None: try: return jwt.decode( token, settings.jwt_secret_key.get_secret_value(), algorithms=[settings.jwt_algorithm], ) except jwt.ExpiredSignatureError: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Token telah kedaluwarsa", headers={"WWW-Authenticate": "Bearer"}, ) except jwt.InvalidTokenError: return None def _configured_principal( credentials: HTTPAuthorizationCredentials | None, settings: Settings, ) -> Principal: if credentials is None or credentials.scheme.lower() != "bearer": raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Bearer token diperlukan", headers={"WWW-Authenticate": "Bearer"}, ) raw_token = credentials.credentials # 1. Try decoding as JWT token payload = decode_jwt_token(raw_token, settings) if payload is not None: user_code = payload.get("sub") or payload.get("user_code") user_name = payload.get("name") or payload.get("user_name", "") role = payload.get("role") kode_sales = payload.get("kode_sales") if not kode_sales and (role == "SALES" or payload.get("source") == "SALES"): kode_sales = user_code if user_code and role: try: return Principal( user_code=user_code, user_name=user_name, role=role, kode_sales=kode_sales, ) except ValueError: pass # 2. Try static configured API keys digest = sha256(raw_token.encode("utf-8")).hexdigest() matched = next( ( value for key, value in settings.auth_api_keys.items() if secrets.compare_digest(key.lower(), digest) ), None, ) if matched is None: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Bearer token tidak valid", headers={"WWW-Authenticate": "Bearer"}, ) try: data = dict(matched) if data.get("role") == "SALES" and "kode_sales" not in data: data["kode_sales"] = data.get("user_code") return Principal.model_validate(data) except ValueError as exc: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Konfigurasi principal API tidak valid", ) from exc def get_current_principal( request: Request, credentials: Annotated[ HTTPAuthorizationCredentials | None, Security(bearer) ], settings: Annotated[Settings, Depends(get_settings)], ) -> Principal: if credentials is not None: principal = _configured_principal(credentials, settings) elif not settings.auth_enabled: role = (request.headers.get("X-User-Role") or "OWNER").upper() if role not in ("VIEWER", "SALES", "SUPERVISOR", "OWNER"): role = "OWNER" user_code = request.headers.get("X-User-Code") or "DEV" user_name = request.headers.get("X-User-Name") or "DEVELOPMENT" kode_sales = request.headers.get("X-Kode-Sales") if not kode_sales and role == "SALES": kode_sales = user_code principal = Principal( user_code=user_code, user_name=user_name, role=role, kode_sales=kode_sales, ) else: principal = _configured_principal(credentials, settings) request.state.principal = principal return principal CurrentPrincipal = Annotated[Principal, Depends(get_current_principal)] def get_audit_user_code( principal: CurrentPrincipal, settings: Annotated[Settings, Depends(get_settings)], legacy_user_code: Annotated[ str | None, Header(alias="X-User-Code", max_length=6) ] = None, ) -> str: if settings.auth_enabled or (principal and principal.user_code != "DEV"): if principal.role not in {"SALES", "SUPERVISOR", "OWNER"}: raise HTTPException(status_code=403, detail="Role hanya dapat membaca") return principal.user_code if not legacy_user_code: raise HTTPException(status_code=422, detail="X-User-Code diperlukan") return legacy_user_code def get_audit_user_name( principal: CurrentPrincipal, settings: Annotated[Settings, Depends(get_settings)], legacy_user_name: Annotated[ str | None, Header(alias="X-User-Name", max_length=50) ] = None, ) -> str: if settings.auth_enabled or (principal and principal.user_code != "DEV"): return principal.user_name if not legacy_user_name: raise HTTPException(status_code=422, detail="X-User-Name diperlukan") return legacy_user_name def get_audit_user_role( principal: CurrentPrincipal, settings: Annotated[Settings, Depends(get_settings)], legacy_user_role: Annotated[ str | None, Header(alias="X-User-Role", max_length=30) ] = None, ) -> str: if settings.auth_enabled: return principal.role if not legacy_user_role: raise HTTPException(status_code=422, detail="X-User-Role diperlukan") return legacy_user_role.upper() AuditUserCode = Annotated[str, Depends(get_audit_user_code)] AuditUserName = Annotated[str, Depends(get_audit_user_name)] AuditUserRole = Annotated[str, Depends(get_audit_user_role)] def get_salesman_filter(principal: CurrentPrincipal) -> str | None: if principal.role == "SALES": return principal.kode_sales or principal.user_code return None SalesmanFilter = Annotated[str | None, Depends(get_salesman_filter)]