feat: integrate durable authorization audit and runtime composition

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a0e747-8f27-7242-8df8-8bc44f88c929

hub_core/security/config.py

@@ -0,0 +1,62 @@
+"""Explicit runtime composition: owner facts remain an injected trust adapter."""
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass
+from pathlib import Path
+
+import httpx
+
+from hub_core.security.audit import AuditCoreSink
+from hub_core.security.boundary import AccessController, FactSource
+from hub_core.security.identity import OIDCVerifier, require_https
+from hub_core.security.policy import FlexPolicy
+
+
+@dataclass(frozen=True)
+class SecuritySettings:
+    issuer: str
+    audience: str
+    root_subject: str
+    policy_url: str
+    policy_caller: str
+    policy_token_file: Path
+    policy_keys_file: Path
+    audit_url: str
+    audit_token_file: Path
+
+    def __post_init__(self):
+        for url in (self.issuer, self.policy_url, self.audit_url):
+            require_https(url)
+        if not self.audience or not self.root_subject:
+            raise ValueError("explicit audience and immutable root subject required")
+        if not self.policy_caller.startswith("system:serviceaccount:"):
+            raise ValueError("explicit policy workload principal required")
+        for path in (self.policy_token_file, self.policy_keys_file, self.audit_token_file):
+            if not isinstance(path, Path) or not path.is_absolute():
+                raise ValueError("absolute credential/trust paths required")
+        if self.policy_token_file == self.audit_token_file:
+            raise ValueError("policy and audit require separate credentials")
+
+    @classmethod
+    def from_env(cls) -> SecuritySettings | None:
+        fields = tuple(cls.__dataclass_fields__)
+        values = {field: os.getenv("HUB_CORE_SECURITY_" + field.upper(), "") for field in fields}
+        if not any(values.values()):
+            return None
+        missing = [field for field, value in values.items() if not value]
+        if missing:
+            raise ValueError("incomplete Hub security configuration: " + ", ".join(missing))
+        return cls(**{field: Path(value) if field.endswith("_file") else value
+                      for field, value in values.items()})
+
+    def compose(self, *, facts: FactSource, client: httpx.AsyncClient) -> AccessController:
+        return AccessController(
+            identity=OIDCVerifier(issuer=self.issuer, audience=self.audience, client=client),
+            facts=facts,
+            policy=FlexPolicy(base_url=self.policy_url, client=client,
+                              caller_token_file=self.policy_token_file,
+                              trusted_keys_file=self.policy_keys_file, caller=self.policy_caller),
+            audit=AuditCoreSink(base_url=self.audit_url, token_file=self.audit_token_file, client=client),
+            root_issuer=self.issuer, root_subject=self.root_subject,
+        )