group_fqcd_jr/app/core/security.py

from dataclasses import dataclass
from datetime import UTC, datetime, timedelta
from pathlib import Path
from typing import Protocol
from uuid import uuid4

import jwt

from app.core.actor import anonymous_context
from app.core.config import Settings
from app.core.contracts import RequestContext
from app.core.errors import UnauthorizedAgentError


class RevocationStore(Protocol):
    def is_revoked(self, jti: str) -> bool: ...


@dataclass(frozen=True)
class EmptyRevocationStore:
    def is_revoked(self, jti: str) -> bool:
        return False


class VisitorTokenIssuer:
    """Issues short-lived anonymous tokens for public customer-service access."""

    def __init__(self, settings: Settings) -> None:
        self._settings = settings
        self._private_key = self._load_private_key()

    def _load_private_key(self) -> str:
        path = Path(self._settings.jwt_private_key_path)
        if not path.is_absolute():
            path = Path.cwd() / path
        try:
            return path.read_text(encoding="utf-8")
        except OSError as exc:
            raise RuntimeError(f"JWT private key cannot be read: {path}") from exc

    def issue(self) -> tuple[str, datetime]:
        now = datetime.now(UTC)
        expires_at = now + timedelta(seconds=self._settings.visitor_token_ttl_seconds)
        subject = str(uuid4().int % 9_000_000_000_000_000_000 + 1)
        token = jwt.encode(
            {"sub": subject, "iss": self._settings.jwt_issuer,
             "aud": self._settings.jwt_audience, "iat": now, "nbf": now,
             "exp": expires_at, "jti": str(uuid4()), "visitor": True},
            self._private_key, algorithm=self._settings.jwt_algorithm,
        )
        return token, expires_at


class JwtAuthenticator:
    def __init__(self, settings: Settings, revocation_store: RevocationStore | None = None) -> None:
        self._settings = settings
        self._revocation_store = revocation_store or EmptyRevocationStore()
        self._public_key = self._load_public_key()

    def _load_public_key(self) -> str:
        path = Path(self._settings.jwt_public_key_path)
        if not path.is_absolute():
            path = Path.cwd() / path
        try:
            return path.read_text(encoding="utf-8")
        except OSError as exc:
            raise RuntimeError(f"JWT public key cannot be read: {path}") from exc

    def authenticate(self, token: str) -> RequestContext:
        if not token:
            raise UnauthorizedAgentError("missing bearer token")
        try:
            claims = jwt.decode(
                token,
                self._public_key,
                algorithms=[self._settings.jwt_algorithm],
                issuer=self._settings.jwt_issuer,
                audience=self._settings.jwt_audience,
                leeway=self._settings.jwt_clock_skew_seconds,
                options={"require": ["sub", "iss", "aud", "exp", "nbf", "jti"]},
            )
        except jwt.PyJWTError as exc:
            raise UnauthorizedAgentError("invalid bearer token") from exc

        jti = str(claims["jti"])
        if self._revocation_store.is_revoked(jti):
            raise UnauthorizedAgentError("revoked bearer token")
        subject = claims["sub"]
        if (not isinstance(subject, str) or not subject.isascii()
                or not subject.isdecimal() or len(subject) > 20
                or not 0 < int(subject) <= 18446744073709551615):
            raise UnauthorizedAgentError("invalid subject")
        if claims.get("visitor") is True:
            # `G-01`：访客三元组的**唯一构造点**在 `app.core.actor`。
            # 这里只负责「令牌是否声明自己是访客」这一个判断，不再拼装身份。
            return anonymous_context(user_id=str(subject), trace_id=str(uuid4()))
        return RequestContext(user_id=str(claims["sub"]), trace_id=str(uuid4()))