81 lines
3.5 KiB
Python
81 lines
3.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
|
|
from typing import Any, Callable, Mapping
|
|
|
|
from aios_target_branch_policy import validate_target_branch
|
|
|
|
|
|
@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"
|
|
|
|
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}
|
|
|
|
|
|
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) -> dict[str, Any]:
|
|
if repository != "aios/aios" or not validate_target_branch(target_branch) or generation < 1 or not 1 <= lease_seconds <= 86400:
|
|
raise ValueError("CREDENTIAL_LEASE_BINDING_INVALID")
|
|
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__)
|
|
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")
|
|
return self.provider.environment(lease)
|
|
|
|
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()
|