125 lines
6.5 KiB
Python
125 lines
6.5 KiB
Python
"""Metadata-only, server-side credential lease broker."""
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timedelta, timezone
|
|
import os
|
|
from pathlib import Path
|
|
import secrets
|
|
import re
|
|
from typing import Any, Callable, Mapping
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class CredentialLease:
|
|
lease_id: str
|
|
task_id: str
|
|
repository: str
|
|
target_branch: str
|
|
generation: int
|
|
issued_at: str
|
|
expires_at: str
|
|
scopes: tuple[str, ...]
|
|
provider: str
|
|
state: str = "ACTIVE"
|
|
scope_type: str = "system"
|
|
tenant_id: str | None = None
|
|
user_id: str | None = None
|
|
project_id: str | None = None
|
|
run_id: str | None = None
|
|
|
|
def as_dict(self) -> dict[str, Any]:
|
|
return {"lease_id": self.lease_id, "task_id": self.task_id, "repository": self.repository, "target_branch": self.target_branch, "generation": self.generation, "issued_at": self.issued_at, "expires_at": self.expires_at, "scopes": list(self.scopes), "provider": self.provider, "state": self.state, "scope_type": self.scope_type, "tenant_id": self.tenant_id, "user_id": self.user_id, "project_id": self.project_id, "run_id": self.run_id}
|
|
|
|
|
|
class FileCredentialProvider:
|
|
def __init__(self, token_path: str | os.PathLike[str], *, actor: str = "aios-autonomy-agent") -> None:
|
|
self.token_path = Path(token_path)
|
|
self.actor = actor
|
|
|
|
def environment(self, lease: CredentialLease) -> dict[str, str]:
|
|
if not self.token_path.is_file():
|
|
raise RuntimeError("CREDENTIAL_PROVIDER_UNAVAILABLE")
|
|
return {"CREDENTIAL_FILE": str(self.token_path), "GITEA_MACHINE_ACTOR": self.actor}
|
|
|
|
|
|
class CredentialBroker:
|
|
def __init__(self, provider: Any, *, now: Callable[[], datetime] | None = None) -> None:
|
|
self.provider = provider
|
|
self.now = now or (lambda: datetime.now(timezone.utc))
|
|
self._leases: dict[str, CredentialLease] = {}
|
|
|
|
def issue(self, *, task_id: str, repository: str, target_branch: str, generation: int, scopes: tuple[str, ...] = ("git:push", "gitea:read"), lease_seconds: int = 900, scope_type: str = "system", tenant_id: str | None = None, user_id: str | None = None, project_id: str | None = None, run_id: str | None = None) -> dict[str, Any]:
|
|
if repository != "aios/aios" or not target_branch.startswith("agent/") or generation < 1 or not 1 <= lease_seconds <= 86400:
|
|
raise ValueError("CREDENTIAL_LEASE_BINDING_INVALID")
|
|
if scope_type == "tenant":
|
|
if not tenant_id:
|
|
raise ValueError("CREDENTIAL_SCOPE_DENY")
|
|
elif scope_type in {"system", "global"}:
|
|
if any((tenant_id, user_id, project_id, run_id)):
|
|
raise ValueError("CREDENTIAL_SCOPE_DENY")
|
|
else:
|
|
raise ValueError("CREDENTIAL_SCOPE_DENY")
|
|
now = self.now().astimezone(timezone.utc)
|
|
lease = CredentialLease("lease-" + secrets.token_hex(16), task_id, repository, target_branch, generation, now.isoformat(), (now + timedelta(seconds=lease_seconds)).isoformat(), tuple(scopes), type(self.provider).__name__, "ACTIVE", scope_type, tenant_id, user_id, project_id, run_id)
|
|
self._leases[lease.lease_id] = lease
|
|
return lease.as_dict()
|
|
|
|
def _get(self, lease_id: str) -> CredentialLease:
|
|
lease = self._leases.get(lease_id)
|
|
if lease is None:
|
|
raise KeyError("CREDENTIAL_LEASE_NOT_FOUND")
|
|
if lease.state == "ACTIVE" and self.now().astimezone(timezone.utc) >= datetime.fromisoformat(lease.expires_at):
|
|
lease = CredentialLease(**{**lease.as_dict(), "scopes": tuple(lease.scopes), "state": "EXPIRED"})
|
|
self._leases[lease_id] = lease
|
|
return lease
|
|
|
|
def environment(self, lease_id: str) -> Mapping[str, str]:
|
|
lease = self._get(lease_id)
|
|
if lease.state != "ACTIVE":
|
|
raise RuntimeError("CREDENTIAL_LEASE_NOT_ACTIVE")
|
|
environment = self.provider.environment(lease)
|
|
for key, value in environment.items():
|
|
upper = str(key).upper()
|
|
if not isinstance(value, str):
|
|
raise RuntimeError("CREDENTIAL_SECRET_EXPOSURE_DENY")
|
|
sensitive = any(marker in upper for marker in ("CREDENTIAL", "TOKEN", "SECRET", "PASSWORD", "PRIVATE_KEY", "API_KEY"))
|
|
if re.search(r"(?i)bearer\s+\S+|(?:token|secret|password|api[_-]?key)\s*[:=]", value):
|
|
raise RuntimeError("CREDENTIAL_SECRET_EXPOSURE_DENY")
|
|
if sensitive and upper.endswith("_FILE"):
|
|
candidate = Path(value)
|
|
credential_root = Path("/run/credentials")
|
|
try:
|
|
resolved = candidate.resolve(strict=False)
|
|
except OSError as exc:
|
|
raise RuntimeError("CREDENTIAL_SECRET_EXPOSURE_DENY") from exc
|
|
if not candidate.is_absolute() or not (resolved == credential_root or credential_root in resolved.parents):
|
|
raise RuntimeError("CREDENTIAL_SECRET_EXPOSURE_DENY")
|
|
elif sensitive and upper.endswith("_REFERENCE"):
|
|
if re.fullmatch(r"(?:systemd-credential|vault|secret-manager):[A-Za-z0-9][A-Za-z0-9._:/#-]{0,255}", value) is None:
|
|
raise RuntimeError("CREDENTIAL_SECRET_EXPOSURE_DENY")
|
|
elif sensitive:
|
|
raise RuntimeError("CREDENTIAL_SECRET_EXPOSURE_DENY")
|
|
return environment
|
|
|
|
def authorize(self, lease_id: str, *, task_id: str, operation: str, scope_type: str = "system", tenant_id: str | None = None, user_id: str | None = None, project_id: str | None = None, run_id: str | None = None) -> Mapping[str, str]:
|
|
lease = self._get(lease_id)
|
|
if lease.state != "ACTIVE" or lease.task_id != task_id or operation not in lease.scopes:
|
|
raise RuntimeError("CREDENTIAL_SCOPE_DENY")
|
|
requested = (scope_type, tenant_id, user_id, project_id, run_id)
|
|
actual = (lease.scope_type, lease.tenant_id, lease.user_id, lease.project_id, lease.run_id)
|
|
if requested != actual:
|
|
raise RuntimeError("CREDENTIAL_SCOPE_DENY")
|
|
return self.environment(lease_id)
|
|
|
|
def status(self, lease_id: str) -> dict[str, Any]:
|
|
return self._get(lease_id).as_dict()
|
|
|
|
def revoke(self, lease_id: str) -> dict[str, Any]:
|
|
lease = self._get(lease_id)
|
|
if lease.state == "REVOKED":
|
|
return lease.as_dict()
|
|
updated = CredentialLease(**{**lease.as_dict(), "scopes": tuple(lease.scopes), "state": "REVOKED"})
|
|
self._leases[lease_id] = updated
|
|
return updated.as_dict()
|