call-center/services/shared/security.py

from __future__ import annotations

from base64 import urlsafe_b64decode, urlsafe_b64encode
from datetime import datetime, timedelta, timezone
import hashlib
import hmac
import json
import os
from typing import Callable

from fastapi import Depends, Header, HTTPException

from services.shared.core import Role


def _bool_env(name: str, default: bool) -> bool:
    raw = os.getenv(name)
    if raw is None:
        return default
    return raw.strip().lower() in {"1", "true", "yes", "on"}


def legacy_header_auth_allowed() -> bool:
    return _bool_env("ALLOW_LEGACY_HEADER_AUTH", True)


def _app_token_secret() -> str:
    return os.getenv("APP_TOKEN_SECRET", "dev-secret-change-me")


def _app_token_ttl_seconds() -> int:
    raw = os.getenv("APP_TOKEN_TTL_SECONDS", "3600").strip()
    try:
        value = int(raw)
    except ValueError:
        value = 3600
    return max(value, 60)


def _b64url_encode(raw: bytes) -> str:
    return urlsafe_b64encode(raw).decode("utf-8").rstrip("=")


def _b64url_decode(raw: str) -> bytes:
    padding = "=" * (-len(raw) % 4)
    return urlsafe_b64decode((raw + padding).encode("utf-8"))


def _sign(message: str) -> str:
    digest = hmac.new(
        _app_token_secret().encode("utf-8"),
        message.encode("utf-8"),
        hashlib.sha256,
    ).digest()
    return _b64url_encode(digest)


def issue_app_token(
    *,
    subject: str,
    username: str,
    role: str,
    auth_source: str,
    provider: str | None = None,
    full_name: str | None = None,
    email: str | None = None,
    tenant_id: str | None = None,
    ttl_seconds: int | None = None,
) -> str:
    now = datetime.now(timezone.utc)
    ttl = ttl_seconds if ttl_seconds is not None else _app_token_ttl_seconds()
    header = {"alg": "HS256", "typ": "JWT"}
    payload = {
        "sub": subject,
        "username": username,
        "role": role,
        "auth_source": auth_source,
        "iat": int(now.timestamp()),
        "exp": int((now + timedelta(seconds=ttl)).timestamp()),
    }
    if provider:
        payload["provider"] = provider
    if full_name:
        payload["full_name"] = full_name
    if email:
        payload["email"] = email
    if tenant_id:
        payload["tenant_id"] = tenant_id

    encoded_header = _b64url_encode(json.dumps(header, separators=(",", ":")).encode("utf-8"))
    encoded_payload = _b64url_encode(json.dumps(payload, separators=(",", ":")).encode("utf-8"))
    message = f"{encoded_header}.{encoded_payload}"
    signature = _sign(message)
    return f"{message}.{signature}"


def decode_app_token(token: str) -> dict:
    try:
        encoded_header, encoded_payload, encoded_signature = token.split(".")
    except ValueError as exc:
        raise HTTPException(status_code=401, detail="Invalid token format") from exc

    message = f"{encoded_header}.{encoded_payload}"
    expected_signature = _sign(message)
    if not hmac.compare_digest(encoded_signature, expected_signature):
        raise HTTPException(status_code=401, detail="Invalid token signature")

    try:
        payload = json.loads(_b64url_decode(encoded_payload).decode("utf-8"))
    except Exception as exc:  # noqa: BLE001
        raise HTTPException(status_code=401, detail="Invalid token payload") from exc

    exp = int(payload.get("exp", 0) or 0)
    now_ts = int(datetime.now(timezone.utc).timestamp())
    if exp and exp < now_ts:
        raise HTTPException(status_code=401, detail="Token expired")

    return payload


def get_actor(
    authorization: str | None = Header(default=None, alias="Authorization"),
    x_user: str | None = Header(default=None, alias="X-User"),
    x_role: str | None = Header(default=None, alias="X-Role"),
    x_tenant_id: str | None = Header(default=None, alias="X-Tenant-ID"),
) -> dict:
    if authorization:
        scheme, _, value = authorization.partition(" ")
        if scheme.lower() != "bearer" or not value.strip():
            raise HTTPException(status_code=401, detail="Invalid authorization header")
        payload = decode_app_token(value.strip())
        tenant_id = str(payload.get("tenant_id") or "").strip()
        tenant_source = "token" if tenant_id else None
        if not tenant_id:
            tenant_id = str(x_tenant_id or "").strip()
            tenant_source = "header" if tenant_id else None
        return {
            "sub": payload.get("sub", ""),
            "user": payload.get("username", "anonymous"),
            "role": str(payload.get("role", "anonymous")).strip().lower() or "anonymous",
            "auth_source": payload.get("auth_source", "token"),
            "provider": payload.get("provider"),
            "full_name": payload.get("full_name"),
            "email": payload.get("email"),
            "tenant_id": tenant_id or None,
            "tenant_source": tenant_source,
        }

    if legacy_header_auth_allowed():
        role = (x_role or "").strip().lower()
        tenant_id = str(x_tenant_id or "").strip()
        return {
            "user": (x_user or "anonymous").strip(),
            "role": role or "anonymous",
            "auth_source": "legacy",
            "tenant_id": tenant_id or None,
            "tenant_source": "header" if tenant_id else None,
        }

    return {"user": "anonymous", "role": "anonymous", "auth_source": "none", "tenant_id": None, "tenant_source": None}


def require_roles(*allowed: Role) -> Callable:
    allowed_values = {a.value for a in allowed}

    def dependency(actor: dict = Depends(get_actor)) -> dict:
        if actor["role"] not in allowed_values:
            raise HTTPException(status_code=403, detail="Insufficient role")
        return actor

    return dependency