"""Authentifizierte Autorisierung: HMAC-SHA256-Clearance. Abgrenzung (bewusst getrennt benannt): - INTEGRITÄT (context.py): "Ist der Kontext unverändert?" -> SHA-256-Digest. - AUTORISIERUNG (hier): "Hat die Policy-Instanz genau DIESEN Auftrag für DIESES Backend freigegeben?" -> HMAC über job_id, Backend, Datenschutz- klasse, Kontext-Digest, Absender, Chat, Ausstellungs-/Ablaufzeit, Nonce. Nur wer den Schlüssel besitzt, kann eine gültige Clearance ausstellen. - Die Absenderauthentifizierung (Transport-Identität + Allowlist) liegt im Gateway (gateway.Gateway._authenticate); sie ist Voraussetzung dafür, dass überhaupt eine Clearance ausgestellt wird. Schlüssel: Datei, Pfad aus der Konfiguration. Fehlt sie, ist sie zu kurz, nicht regulär oder für Gruppe/Andere lesbar -> KEY_UNAVAILABLE (fail-closed). Je Backend wird ein eigener Schlüssel abgeleitet; so kann z. B. das Telegram-Plugin später nur Claude-Clearances prüfen, aber keine Codex-Clearances erzeugen. """ from __future__ import annotations import hashlib import hmac import os import re import secrets import stat import time from dataclasses import asdict, dataclass, replace from pathlib import Path from typing import Any, Callable, Mapping from .context import canonical_json from .core import PrivacyMatrix, PrivacyTier from .state import NonceStore CLEARANCE_VERSION = 1 CLEARANCE_DOMAIN = b"aria-v3/clearance/v1\n" KEY_LABEL = b"aria-v3/clearance-key/v1/" MIN_KEY_BYTES = 32 MAX_KEY_BYTES = 4096 MAX_TTL_SECONDS = 600 _HEX64 = re.compile(r"^[0-9a-f]{64}$") _HEX32 = re.compile(r"^[0-9a-f]{32}$") class AuthorizationError(ValueError): def __init__(self, code: str): super().__init__(code) self.code = code class KeyUnavailableError(AuthorizationError): pass class ClearanceKeyring: """Liest den Hauptschlüssel bei jeder Verwendung neu (Entzug wirkt sofort).""" def __init__(self, key_file: Path): self.key_file = Path(key_file) def _load_master(self) -> bytes: try: fd = os.open(self.key_file, os.O_RDONLY | os.O_NOFOLLOW | os.O_CLOEXEC) except OSError as exc: raise KeyUnavailableError("KEY_UNAVAILABLE") from exc try: info = os.fstat(fd) if not stat.S_ISREG(info.st_mode): raise KeyUnavailableError("KEY_NOT_REGULAR_FILE") if info.st_uid != os.geteuid(): raise KeyUnavailableError("KEY_WRONG_OWNER") if info.st_mode & 0o077: raise KeyUnavailableError("KEY_PERMISSIONS_TOO_OPEN") data = os.read(fd, MAX_KEY_BYTES + 1) finally: os.close(fd) if len(data) < MIN_KEY_BYTES: raise KeyUnavailableError("KEY_TOO_SHORT") if len(data) > MAX_KEY_BYTES: raise KeyUnavailableError("KEY_TOO_LONG") return data def available(self) -> bool: try: self._load_master() except KeyUnavailableError: return False return True def key_for(self, backend: str) -> bytes: if not isinstance(backend, str) or not backend: raise AuthorizationError("INVALID_BACKEND") master = self._load_master() return hmac.new(master, KEY_LABEL + backend.encode("utf-8"), hashlib.sha256).digest() @dataclass(frozen=True) class Clearance: v: int job_id: str backend: str privacy: str context_digest: str sender_id: str chat_id: str issued_at: int expires_at: int nonce: str mac: str = "" def signing_payload(self) -> bytes: fields = asdict(self) fields.pop("mac") return CLEARANCE_DOMAIN + canonical_json(fields) def to_dict(self) -> dict: return asdict(self) @classmethod def from_dict(cls, data: Mapping[str, Any]) -> "Clearance": expected = {"v": int, "job_id": str, "backend": str, "privacy": str, "context_digest": str, "sender_id": str, "chat_id": str, "issued_at": int, "expires_at": int, "nonce": str, "mac": str} if not isinstance(data, Mapping) or set(data) != set(expected): raise AuthorizationError("MALFORMED_CLEARANCE") for key, kind in expected.items(): if type(data[key]) is not kind: raise AuthorizationError("MALFORMED_CLEARANCE") return cls(**{k: data[k] for k in expected}) def _sign(key: bytes, clearance: Clearance) -> str: return hmac.new(key, clearance.signing_payload(), hashlib.sha256).hexdigest() class ClearanceIssuer: """Stellt Clearances nur für Kombinationen aus, die die Matrix erlaubt.""" def __init__(self, keyring: ClearanceKeyring, matrix: PrivacyMatrix, *, ttl_seconds: int = 120, clock: Callable[[], float] = time.time): if type(ttl_seconds) is not int or not 1 <= ttl_seconds <= MAX_TTL_SECONDS: raise ValueError("INVALID_TTL") self._keyring = keyring self._matrix = matrix self._ttl = ttl_seconds self._clock = clock def issue(self, *, job_id: str, backend: str, privacy: PrivacyTier, context_digest: str, sender_id: str, chat_id: str) -> Clearance: if not isinstance(privacy, PrivacyTier): raise AuthorizationError("INVALID_PRIVACY") if not self._matrix.permits(backend, privacy): raise AuthorizationError("PRIVACY_NOT_PERMITTED") if not isinstance(context_digest, str) or not _HEX64.match(context_digest): raise AuthorizationError("INVALID_CONTEXT_DIGEST") for value in (job_id, sender_id, chat_id): if not isinstance(value, str) or not value: raise AuthorizationError("INVALID_SUBJECT") now = int(self._clock()) unsigned = Clearance( v=CLEARANCE_VERSION, job_id=job_id, backend=backend, privacy=privacy.value, context_digest=context_digest, sender_id=sender_id, chat_id=chat_id, issued_at=now, expires_at=now + self._ttl, nonce=secrets.token_hex(16), ) return replace(unsigned, mac=_sign(self._keyring.key_for(backend), unsigned)) class ClearanceVerifier: """Prüft vor JEDER Adapter-Ausführung (siehe adapter_contract.BackendAdapter).""" def __init__(self, keyring: ClearanceKeyring, matrix: PrivacyMatrix, nonce_store: NonceStore, classifier, *, max_ttl_seconds: int = MAX_TTL_SECONDS, max_clock_skew_seconds: int = 5, clock: Callable[[], float] = time.time): self._keyring = keyring self._matrix = matrix self._nonces = nonce_store self.classifier = classifier self._max_ttl = max_ttl_seconds self._skew = max_clock_skew_seconds self._clock = clock def verify(self, clearance, *, backend: str, job_id: str, context_digest: str, privacy: PrivacyTier, sender_id: str, chat_id: str) -> Clearance: if clearance is None: raise AuthorizationError("MISSING_CLEARANCE") if isinstance(clearance, Mapping): clearance = Clearance.from_dict(clearance) if not isinstance(clearance, Clearance): raise AuthorizationError("INVALID_CLEARANCE") if clearance.v != CLEARANCE_VERSION: raise AuthorizationError("UNSUPPORTED_CLEARANCE_VERSION") if clearance.backend != backend: raise AuthorizationError("BACKEND_MISMATCH") expected_mac = _sign(self._keyring.key_for(backend), clearance) if not hmac.compare_digest(expected_mac, clearance.mac): raise AuthorizationError("INVALID_SIGNATURE") # Ab hier sind alle Felder authentisch (von der Policy-Instanz signiert). now = self._clock() lifetime = clearance.expires_at - clearance.issued_at if lifetime <= 0 or lifetime > self._max_ttl: raise AuthorizationError("INVALID_LIFETIME") if clearance.issued_at > now + self._skew: raise AuthorizationError("NOT_YET_VALID") if now >= clearance.expires_at: raise AuthorizationError("EXPIRED") if clearance.job_id != job_id: raise AuthorizationError("JOB_MISMATCH") if clearance.sender_id != sender_id or clearance.chat_id != chat_id: raise AuthorizationError("SUBJECT_MISMATCH") if not hmac.compare_digest(clearance.context_digest, context_digest): raise AuthorizationError("CONTEXT_DIGEST_MISMATCH") if clearance.privacy != privacy.value: raise AuthorizationError("PRIVACY_MISMATCH") if not self._matrix.permits(backend, privacy): raise AuthorizationError("PRIVACY_NOT_PERMITTED") if not _HEX32.match(clearance.nonce): raise AuthorizationError("MALFORMED_CLEARANCE") if not self._nonces.consume(clearance.nonce, clearance.expires_at, now): raise AuthorizationError("REPLAY") return clearance