62 lines
2.4 KiB
Python
62 lines
2.4 KiB
Python
"""Tenant and Cell scoped, value-free secret reference model."""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from typing import Any, Literal
|
|
|
|
from .tenant_context import ScopeError, ValidationError, require_ident
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class SecretRef:
|
|
secret_id: str
|
|
tenant_id: str
|
|
cell_id: str
|
|
kind: Literal["channel", "model", "tool", "runtime"]
|
|
version: str
|
|
broker_path: str
|
|
|
|
def __post_init__(self) -> None:
|
|
for name in ("secret_id", "cell_id", "version"):
|
|
value = getattr(self, name)
|
|
if require_ident(name, value) != value:
|
|
raise ValidationError(f"{name} must be canonical")
|
|
if require_ident("tenant_id", self.tenant_id, tenant=True) != self.tenant_id:
|
|
raise ValidationError("tenant_id must be canonical")
|
|
if self.kind not in {"channel", "model", "tool", "runtime"}:
|
|
raise ValidationError("invalid secret kind")
|
|
self._validate_broker_path()
|
|
|
|
def _validate_broker_path(self) -> None:
|
|
if not isinstance(self.broker_path, str) or not self.broker_path.startswith("secret://"):
|
|
raise ValidationError("secret broker path required")
|
|
suffix = self.broker_path[len("secret://"):]
|
|
if (
|
|
not suffix or "?" in suffix or "#" in suffix or chr(92) in suffix
|
|
or any(part in {"", ".", ".."} for part in suffix.split("/"))
|
|
):
|
|
raise ValidationError("invalid secret broker path")
|
|
for part in suffix.split("/"):
|
|
if require_ident("secret broker path segment", part) != part:
|
|
raise ValidationError("invalid secret broker path")
|
|
|
|
def can_be_used_by(self, *, tenant_id: str, cell_id: str) -> bool:
|
|
return self.tenant_id == tenant_id and self.cell_id == cell_id
|
|
|
|
def resolve(self, broker: Any) -> str:
|
|
"""Resolve through one explicit broker boundary without leaking provider errors."""
|
|
failure: ScopeError | None = None
|
|
value: object = None
|
|
try:
|
|
reader = getattr(broker, "read_secret")
|
|
if not callable(reader):
|
|
raise TypeError("invalid broker")
|
|
value = reader(self)
|
|
if not isinstance(value, str) or not value:
|
|
raise ValueError("invalid broker result")
|
|
except Exception:
|
|
failure = ScopeError("secret broker unavailable")
|
|
if failure is not None:
|
|
raise failure
|
|
return value
|