import jwt import os import time from fastapi import HTTPException, Security from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials from passlib.context import CryptContext from pydantic import BaseModel # Secret key for JWT signing. In production, this must be a secure random string loaded from .env # For this deployment, we use a fallback to ensure it works out of the box if missing. SECRET_KEY = os.getenv("JWT_SECRET", "kia_shqiperia_2026_super_secret_key_12345") ALGORITHM = "HS256" ACCESS_TOKEN_EXPIRE_MINUTES = 60 * 24 * 7 # 7 days security = HTTPBearer() pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") # Hardcoded Access Codes mapping to roles (for the C4ISR prototype) # In a real deployed system, these would be in a DB mapped to user emails. ROLE_CREDENTIALS = { "visitor": "", # no password required for unclassified "officer": "oficer2025", "commander": "komanda2025", "general": "kia_gjeneral" } class LoginRequest(BaseModel): role: str access_code: str = "" def verify_password(plain_password, target_password): # Standard string comparison for demo. In production, use pwd_context.verify return plain_password == target_password def create_access_token(data: dict, expires_delta: int = ACCESS_TOKEN_EXPIRE_MINUTES): to_encode = data.copy() expire = time.time() + (expires_delta * 60) to_encode.update({"exp": expire}) encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) return encoded_jwt def authenticate_role(role: str, access_code: str): if role not in ROLE_CREDENTIALS: return False expected_code = ROLE_CREDENTIALS[role] if expected_code == "" or verify_password(access_code, expected_code): return True return False def verify_token(credentials: HTTPAuthorizationCredentials = Security(security)): """FastAPI Depends function that extracts and validates the JWT, returning the role.""" token = credentials.credentials try: payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) role: str = payload.get("role") if role is None: raise HTTPException(status_code=401, detail="Token i pavlefshëm (Rol nuk gjendet)") # Check expiration manually since pyjwt does it too, but just to be safe exp = payload.get("exp") if exp and time.time() > exp: raise HTTPException(status_code=401, detail="Sesioni ka skaduar") return role except jwt.ExpiredSignatureError: raise HTTPException(status_code=401, detail="Sesioni ka skaduar. Ju lutem mbyllni dhe hyni përsëri.") except jwt.PyJWTError: raise HTTPException(status_code=401, detail="Lidhje e paautorizuar. Kërkohet verifikim kredencialesh.")