"""Core-owned consent, least-privilege grants, and opaque OAuth credentials.""" import base64 import hashlib import secrets from datetime import UTC, datetime, timedelta from types import SimpleNamespace from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit from uuid import UUID, uuid4 from src.config import get_settings from src.core.exceptions import ForbiddenError, NotFoundError, UnauthorizedError, ValidationError from src.domain.agent_access.models import ( AgentClient, AgentCredential, AgentGrant, AgentGrantWorkspace, ) from src.domain.agent_access.repository import AgentRepository from src.domain.agent_access.schemas import ( DELEGABLE_BILLING_SCOPES, MIN_DELEGATION_PERMISSIONS, SCOPE_PERMISSIONS, AuthorizationRequest, ClientCreate, ConsentCreate, GrantCreate, GrantUpdate, PurchasePolicy, WorkspaceAccess, ) from src.domain.projects.repository import ProjectRepository from src.domain.workspaces.facades import WorkspaceFacadeImpl # Scopes that read or spend a workspace's resources; these additionally require the # workspace to be in good standing (billing/MFA policy). PRODUCTIVE_SCOPES = {"projects.read", "projects.create", "builds.run", "credits.purchase"} def utcnow(): return datetime.now(UTC) def fingerprint(secret: str) -> str: return hashlib.sha256(secret.encode()).hexdigest() def pkce_challenge(verifier: str) -> str: return ( base64.urlsafe_b64encode(hashlib.sha256(verifier.encode()).digest()).rstrip(b"=").decode() ) def valid_resources(settings) -> set[str]: """OAuth `resource` (RFC 8707) values this authorization server issues tokens for. Each protected resource (MCP, A2A) gets its own audience-bound token even though both draw from the same grants/scopes. """ return { settings.agent_mcp_url, settings.agent_openai_mcp_url, settings.agent_a2a_url, } def prompt_source_for(settings, principal) -> str: """Label which interaction surface a principal's token was issued for. Distinct from ``initiating_grant_id`` (which grant acted): this answers through which admitted surface the action arrived (MCP vs A2A), for downstream provenance/audit only — never an authorization input. """ resource = getattr(principal, "resource", None) if resource in {settings.agent_mcp_url, settings.agent_openai_mcp_url}: return "mcp_agent" if resource == settings.agent_a2a_url: return "a2a_agent" return "api" def grant_response(grant: AgentGrant) -> dict: return { key: getattr(grant, key) for key in ( "id", "workspace_id", "name", "kind", "client_id", "scopes", "project_ids", "workspace_wide", "purchase_policy", "expires_at", "revoked_at", "created_at", ) } def workspace_response(row: AgentGrantWorkspace) -> dict: return { "workspace_id": row.workspace_id, "scopes": row.scopes, "purchase_policy": row.purchase_policy, } class AgentAccessService: def __init__( self, repo: AgentRepository, workspaces: WorkspaceFacadeImpl, projects: ProjectRepository, productive_policy, redis=None, ): self.repo = repo self.workspaces = workspaces self.projects = projects self.productive_policy = productive_policy self.settings = get_settings() # Only the HTTP tool-execution path wires a real client (see dependencies.py); # durable-worker callers of make_agent_service() never enforce tool rate limits. self.redis = redis async def register_client(self, body: ClientCreate): client = await self.repo.add( AgentClient( id=str(uuid4()), name=body.client_name, redirect_uris=body.redirect_uris, created_at=utcnow(), ) ) return { "client_id": client.id, **body.model_dump(), "client_id_issued_at": int(utcnow().timestamp()), } async def authorization_context(self, request: AuthorizationRequest): client = await self.repo.client(request.client_id) if not client or request.redirect_uri not in client.redirect_uris: raise ValidationError("Invalid client or redirect URI") if request.resource not in valid_resources(self.settings): raise ValidationError("Invalid resource") scopes = set(request.scope.split()) if not scopes or scopes - SCOPE_PERMISSIONS.keys(): raise ValidationError("Invalid scope") return { "client_name": client.name, "client_id": client.id, "redirect_uri": request.redirect_uri, "scopes": sorted(scopes), } async def require_manager(self, workspace_id: UUID, user_id: UUID): if not await self.workspaces.check_permission(workspace_id, user_id, "members.manage"): raise ForbiddenError() def _delegation_stands(self, row, workspace) -> bool: """An approved delegation only stands while the owner who gave it still owns the workspace, so a transfer of ownership never silently keeps someone else's grant.""" return bool( getattr(row, "status", None) == "approved" and getattr(row, "resolved_by", None) and workspace and workspace.owner_id == row.resolved_by ) async def approved_billing_delegation(self, workspace_id: UUID, user_id: UUID) -> bool: row = await self.repo.billing_delegation(workspace_id, user_id) if getattr(row, "status", None) != "approved": return False return self._delegation_stands(row, await self.workspaces.get_workspace_summary(workspace_id)) async def can_delegate_scope(self, workspace_id: UUID, user_id: UUID, scope: str) -> bool: """The person's own permission, or -- for the delegable billing scopes only -- an owner-approved billing delegation (models.WorkspaceBillingDelegation).""" if await self.workspaces.check_permission(workspace_id, user_id, SCOPE_PERMISSIONS[scope]): return True return scope in DELEGABLE_BILLING_SCOPES and await self.approved_billing_delegation( workspace_id, user_id ) async def validate_workspace_access( self, access: WorkspaceAccess, user_id: UUID, *, require_manager: bool ): """Everything the approving person must hold in one workspace before an agent may be given access to it. Raises ForbiddenError; never partially applies.""" workspace_id = access.workspace_id if require_manager: await self.require_manager(workspace_id, user_id) for permission in MIN_DELEGATION_PERMISSIONS: if not await self.workspaces.check_permission(workspace_id, user_id, permission): raise ForbiddenError("Developer access is required in every connected workspace") for scope in access.scopes: if not await self.can_delegate_scope(workspace_id, user_id, scope): raise ForbiddenError("Cannot delegate the requested permissions") if set(access.scopes) & PRODUCTIVE_SCOPES: await self.productive_policy.require_for_workspace(workspace_id, user_id) if "credits.purchase" in access.scopes: workspace = await self.workspaces.get_workspace_summary(workspace_id) if not workspace or workspace.owner_id != user_id: raise ForbiddenError("Only the workspace owner can approve unattended spending") def new_workspace_row(self, grant_id: UUID, access: WorkspaceAccess) -> AgentGrantWorkspace: return AgentGrantWorkspace( grant_id=grant_id, workspace_id=access.workspace_id, scopes=access.scopes, project_ids=[], workspace_wide=True, purchase_policy=access.purchase_policy.model_dump(), created_at=utcnow(), ) async def create_grant(self, body: GrantCreate, user_id: UUID, *, client_id=None): for access in body.workspaces: await self.validate_workspace_access(access, user_id, require_manager=client_id is None) grant = await self.repo.add( AgentGrant( id=uuid4(), # First workspace only, for legacy readers. Authority lives in the # per-workspace rows below; scopes here is the union (OAuth ceiling). workspace_id=body.workspaces[0].workspace_id, created_by=user_id, client_id=client_id, name=body.name, kind="user" if client_id else "service", scopes=body.all_scopes, project_ids=[], workspace_wide=True, purchase_policy=PurchasePolicy().model_dump(), expires_at=body.expires_at, revoked_at=None, created_at=utcnow(), ) ) for access in body.workspaces: await self.repo.add(self.new_workspace_row(grant.id, access)) return grant async def credential(self, grant, kind, client_id, expires_at, details=None): raw = "p2hsa_" + secrets.token_urlsafe(48) await self.repo.add( AgentCredential( fingerprint=fingerprint(raw), grant_id=grant.id, kind=kind, client_id=client_id, expires_at=min(expires_at, grant.expires_at), details=details or {}, consumed_at=None, ) ) return raw async def create_service_account(self, body: GrantCreate, user_id: UUID): grant = await self.create_grant(body, user_id) secret = await self.credential(grant, "service", str(grant.id), grant.expires_at) return {**await self.grant_view(grant), "client_secret": secret} async def grant_view(self, grant: AgentGrant) -> dict: rows = await self.repo.grant_workspaces(grant.id) return {**grant_response(grant), "workspaces": [workspace_response(r) for r in rows]} async def approve(self, body: ConsentCreate, user_id: UUID): await self.authorization_context(body.authorization) if set(body.grant.all_scopes) - set(body.authorization.scope.split()): raise ValidationError("Consent cannot add scopes the client did not request") grant = await self.create_grant(body.grant, user_id, client_id=body.authorization.client_id) code = await self.credential( grant, "code", body.authorization.client_id, utcnow() + timedelta(minutes=5), { "redirect_uri": body.authorization.redirect_uri, "code_challenge": body.authorization.code_challenge, "resource": body.authorization.resource, }, ) url = urlsplit(body.authorization.redirect_uri) query = [ (k, v) for k, v in parse_qsl(url.query, keep_blank_values=True) if k not in ("code", "state") ] query.extend([("code", code), ("state", body.authorization.state)]) return {"redirect_uri": urlunsplit(url._replace(query=urlencode(query)))} def require_active(self, grant): if not grant or grant.revoked_at or grant.expires_at <= utcnow(): raise UnauthorizedError("Invalid or expired agent grant") async def authenticate(self, token: str): row = await self.repo.credential(fingerprint(token)) if not row or row.kind != "access" or row.consumed_at or row.expires_at <= utcnow(): raise UnauthorizedError("Invalid or expired agent token") if row.details.get("resource") not in valid_resources(self.settings): raise UnauthorizedError("Invalid token audience") grant = await self.repo.grant(row.grant_id) self.require_active(grant) # Token-specific scope attenuation must not mutate the durable grant. # No `workspace_id`: a grant can span several workspaces, so every tool must # resolve its target explicitly through require_scope() rather than read one here. principal = {k: v for k, v in grant_response(grant).items() if k != "workspace_id"} return SimpleNamespace( **{ **principal, "created_by": grant.created_by, "scopes": sorted(set(grant.scopes) & set(row.details.get("scopes", []))), # The audience the access token was issued for (MCP vs A2A) — used only # to label build/operation provenance, never as an authorization input. "resource": row.details.get("resource"), } ) async def _authorize_workspace(self, grant, scope: str, access_row): """Recheck, right now, that the authorizing person may still delegate `scope` here.""" if scope not in access_row.scopes: raise ForbiddenError("Agent scope not granted") workspace_id = access_row.workspace_id # Also recheck the authorizing person's current role (or owner-approved billing # delegation). Removal, demotion or a revoked delegation fails closed. if not await self.can_delegate_scope(workspace_id, grant.created_by, scope): raise ForbiddenError("Authorizing user's permissions no longer permit this action") if scope == "credits.purchase": workspace = await self.workspaces.get_workspace_summary(workspace_id) if not workspace or workspace.owner_id != grant.created_by: raise ForbiddenError("The approving owner no longer owns this workspace") if scope in PRODUCTIVE_SCOPES: await self.productive_policy.require_for_workspace(workspace_id, grant.created_by) async def resolve_access(self, grant, *, project_id: UUID | None, workspace_id: UUID | None): """Pick the one granted workspace a call targets. A project implies its own workspace; otherwise the caller names it, or it is the grant's only one. Anything outside the grant is reported as not found.""" rows = {row.workspace_id: row for row in await self.repo.grant_workspaces(grant.id)} if project_id is not None: project = await self.projects.get_by_id(project_id) if not project or project.deleted_at or project.workspace_id not in rows: raise NotFoundError("Project not found") if workspace_id is not None and workspace_id != project.workspace_id: raise NotFoundError("Project not found") row = rows[project.workspace_id] if not row.workspace_wide and str(project_id) not in row.project_ids: raise NotFoundError("Project not found") return row if workspace_id is not None: if workspace_id not in rows: raise NotFoundError("Workspace not found") return rows[workspace_id] if len(rows) == 1: return next(iter(rows.values())) raise ValidationError( "workspace_id is required: this connection has access to several workspaces. " "Call list_workspaces to see them." ) async def require_scope( self, grant, scope: str, project_id: UUID | None = None, workspace_id: UUID | None = None, ) -> UUID: """Authorize `scope` for one workspace and return that workspace's id.""" self.require_active(grant) if scope not in grant.scopes: raise ForbiddenError("Agent scope not granted") row = await self.resolve_access(grant, project_id=project_id, workspace_id=workspace_id) await self._authorize_workspace(grant, scope, row) return row.workspace_id async def scoped_workspaces(self, grant, scope: str) -> list[UUID]: """Granted workspaces where `scope` is currently usable. A workspace that has lost the authorizing person's permission is skipped, not fatal to the others.""" self.require_active(grant) if scope not in grant.scopes: return [] usable = [] for row in await self.repo.grant_workspaces(grant.id): try: await self._authorize_workspace(grant, scope, row) except ForbiddenError: continue usable.append(row.workspace_id) return usable async def access_row(self, grant, workspace_id: UUID) -> AgentGrantWorkspace: rows = await self.repo.grant_workspaces(grant.id) row = next((r for r in rows if r.workspace_id == workspace_id), None) if not row: raise NotFoundError("Workspace not found") return row async def exchange(self, form: dict): kind = { "authorization_code": "code", "refresh_token": "refresh", "client_credentials": "service", }.get(form.get("grant_type")) if not kind: raise ValidationError("Unsupported grant type") resource = form.get("resource") if resource not in valid_resources(self.settings): raise ValidationError("Invalid resource") raw = form.get( {"code": "code", "refresh": "refresh_token", "service": "client_secret"}[kind], "" ) row = await self.repo.credential(fingerprint(raw), lock=True) if ( not row or row.kind != kind or row.client_id != form.get("client_id") or row.expires_at <= utcnow() ): raise UnauthorizedError("Invalid grant") if kind == "refresh" and row.details.get("resource") != resource: # A refresh token is bound to the resource it was issued for; it must # not mint an access token for a different protected resource (MCP vs A2A). raise ValidationError("Invalid resource") grant = await self.repo.grant(row.grant_id, lock=True) self.require_active(grant) if row.consumed_at: if kind in ("refresh", "code"): grant.revoked_at = utcnow() grant.revocation_reason = "credential_replay" # Persist replay revocation even though this request returns an error. await self.repo.db.commit() raise UnauthorizedError("Credential already used") if kind == "code": verifier = form.get("code_verifier", "") import re if not re.fullmatch(r"[A-Za-z0-9._~-]{43,128}", verifier) or ( not secrets.compare_digest(pkce_challenge(verifier), row.details["code_challenge"]) or form.get("redirect_uri") != row.details["redirect_uri"] ): raise UnauthorizedError("Invalid code verifier or redirect URI") allowed = set(row.details.get("scopes", grant.scopes)) & set(grant.scopes) requested = set(form.get("scope", " ".join(sorted(allowed))).split()) if not requested or requested - allowed: raise ValidationError("Scope escalation is not allowed") if kind != "service": row.consumed_at = utcnow() details = {"resource": resource, "scopes": sorted(requested)} expires_at = min( utcnow() + timedelta(seconds=self.settings.agent_access_ttl_seconds), grant.expires_at ) access = await self.credential(grant, "access", row.client_id, expires_at, details) response = { "access_token": access, "token_type": "Bearer", "scope": " ".join(sorted(requested)), "expires_in": max(0, int((expires_at - utcnow()).total_seconds())), } if kind != "service": response["refresh_token"] = await self.credential( grant, "refresh", row.client_id, utcnow() + timedelta(days=self.settings.agent_refresh_ttl_days), details, ) return response async def revoke(self, workspace_id: UUID, grant_id: UUID, user_id: UUID): grant = await self.repo.grant(grant_id, lock=True) rows = await self.repo.grant_workspaces(grant_id) if grant else [] if not grant or workspace_id not in {row.workspace_id for row in rows}: raise NotFoundError("Agent account not found") if grant.kind == "service" or grant.created_by != user_id: await self.require_manager(workspace_id, user_id) if grant.revoked_at: return if grant.created_by != user_id and len(rows) > 1: # A manager who did not create this connection can cut it off from their # workspace, but must not silently break it for the creator's other ones. await self.repo.remove_grant_workspace(grant_id, workspace_id) if grant.workspace_id == workspace_id: grant.workspace_id = next( r.workspace_id for r in rows if r.workspace_id != workspace_id ) return grant.revoked_at = utcnow() grant.revoked_by = user_id grant.revocation_reason = "user_revoked" async def list_grants(self, workspace_id: UUID, user_id: UUID): manager = await self.workspaces.check_permission(workspace_id, user_id, "members.manage") rows = await self.repo.list_grants(workspace_id, None if manager else user_id) return await self.grant_views(rows, user_id) async def list_my_grants(self, user_id: UUID): """Every connection this person created, across all their workspaces.""" return await self.grant_views(await self.repo.list_grants_created_by(user_id), user_id) async def grant_views(self, grants: list[AgentGrant], viewer_id: UUID) -> list[dict]: by_grant = await self.repo.grant_workspaces_by_grant([g.id for g in grants]) return [ { **grant_response(grant), "workspaces": [workspace_response(r) for r in by_grant[grant.id]], # Only the person who approved a connection may reshape it (see update_grant). "can_edit": grant.created_by == viewer_id and not grant.revoked_at, } for grant in grants ] async def update_grant(self, grant_id: UUID, body: GrantUpdate, user_id: UUID): grant = await self.repo.grant(grant_id, lock=True) if not grant or grant.created_by != user_id: raise NotFoundError("Agent account not found") self.require_active(grant) if body.name is not None: grant.name = body.name if body.workspaces is not None: ceiling = set(grant.scopes) for access in body.workspaces: if set(access.scopes) - ceiling: raise ValidationError( "A connection cannot gain permissions it was not originally approved for" ) await self.validate_workspace_access( access, user_id, require_manager=grant.kind == "service" ) existing = {row.workspace_id: row for row in await self.repo.grant_workspaces(grant_id)} wanted = {access.workspace_id for access in body.workspaces} for workspace_id in existing.keys() - wanted: await self.repo.remove_grant_workspace(grant_id, workspace_id) for access in body.workspaces: row = existing.get(access.workspace_id) if row is None: await self.repo.add(self.new_workspace_row(grant_id, access)) else: row.scopes = access.scopes row.purchase_policy = access.purchase_policy.model_dump() # An edit always yields whole-workspace access (legacy per-project # restrictions are not editable, only replaced). row.workspace_wide = True row.project_ids = [] grant.workspace_id = body.workspaces[0].workspace_id return await self.grant_view(grant) async def workspace_options(self, user_id: UUID) -> list[dict]: """What this person could delegate in each of their workspaces, so the consent screen can disable what they cannot instead of failing at submit.""" options = [] for workspace in await self.workspaces.list_workspace_summaries_for_user(user_id): held = { permission: await self.workspaces.check_permission( workspace.id, user_id, permission ) for permission in {*MIN_DELEGATION_PERMISSIONS, *SCOPE_PERMISSIONS.values()} } is_owner = workspace.owner_id == user_id delegation = None if held["billing.manage"] else await self.repo.billing_delegation( workspace.id, user_id ) delegated = self._delegation_stands(delegation, workspace) scopes = [ scope for scope, permission in SCOPE_PERMISSIONS.items() if (held[permission] or (delegated and scope in DELEGABLE_BILLING_SCOPES)) and (scope != "credits.purchase" or is_owner) ] # pending | declined | revoked are the person's own request state; "approved" # only while the approving owner still owns the workspace (see _delegation_stands). status = getattr(delegation, "status", None) billing_delegation = ( "approved" if delegated else (status if status in {"pending", "declined", "revoked"} else "none") ) options.append( { "id": str(workspace.id), "name": workspace.name, "eligible": all(held[p] for p in MIN_DELEGATION_PERMISSIONS), "scopes": scopes, "can_manage_billing": held["billing.manage"], # Billing scopes are usable here: by the person's own permission or by # an owner-approved delegation. credits.purchase stays owner-only. "billing_unlocked": held["billing.manage"] or delegated, "billing_delegation": billing_delegation, "is_owner": is_owner, } ) return options async def revoke_token(self, form: dict): row = await self.repo.credential(fingerprint(form.get("token", "")), lock=True) if row and row.client_id == form.get("client_id") and row.kind in ("access", "refresh"): grant = await self.repo.grant(row.grant_id, lock=True) if grant: if not grant.revoked_at: grant.revoked_at = utcnow() grant.revocation_reason = "token_revoked"