Initial import with GitLab CI/CD and registry deploy flow

services/shared/security.py

@@ -0,0 +1,153 @@
+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,
+    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
+
+    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"),
+) -> 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())
+        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"),
+        }
+
+    if legacy_header_auth_allowed():
+        role = (x_role or "").strip().lower()
+        return {"user": (x_user or "anonymous").strip(), "role": role or "anonymous", "auth_source": "legacy"}
+
+    return {"user": "anonymous", "role": "anonymous", "auth_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