CA/backend/auth/service.py

"""JWT token creation and password hashing utilities."""
import os
import logging
from datetime import datetime, timedelta, timezone

from jose import JWTError, jwt
from passlib.context import CryptContext

logger = logging.getLogger("cbpoa.auth")

_DEFAULT_SECRET = "cbpoa-dev-secret-change-in-production"
SECRET_KEY = os.getenv("AUTH_SECRET_KEY", _DEFAULT_SECRET)
if SECRET_KEY == _DEFAULT_SECRET:
    logger.warning(
        "AUTH_SECRET_KEY is not set — using the built-in development secret. "
        "Set AUTH_SECRET_KEY in the environment before deploying; the default is public and allows token forgery."
    )
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = int(os.getenv("AUTH_TOKEN_EXPIRE_MINUTES", "480"))

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

# In-memory user store (replace with DB table when auth matures)
_users: dict[str, str] = {}


def seed_default_admin() -> None:
    """Create default admin user if no users exist."""
    if not _users:
        default_user = os.getenv("AUTH_DEFAULT_USER", "admin")
        default_pass = os.getenv("AUTH_DEFAULT_PASSWORD", "admin123")
        _users[default_user] = pwd_context.hash(default_pass)
        logger.info("Seeded default user '%s'", default_user)


def verify_password(plain: str, hashed: str) -> bool:
    return pwd_context.verify(plain, hashed)


def hash_password(password: str) -> str:
    return pwd_context.hash(password)


def authenticate_user(username: str, password: str) -> bool:
    hashed = _users.get(username)
    if not hashed:
        return False
    return verify_password(password, hashed)


def create_user(username: str, password: str) -> bool:
    """Register a new user. Returns False if username already exists."""
    if username in _users:
        return False
    _users[username] = hash_password(password)
    logger.info("Registered new user '%s'", username)
    return True


def create_access_token(data: dict) -> str:
    to_encode = data.copy()
    expire = datetime.now(timezone.utc) + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)


def decode_access_token(token: str) -> dict | None:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        # python-jose ignores PyJWT's options={"require": [...]}, so enforce exp manually:
        # a token with no exp claim would otherwise never expire.
        if "exp" not in payload:
            return None
        return payload
    except JWTError:
        return None