"""Regression coverage for delegation, OAuth replay, and token attenuation.""" from datetime import timedelta from types import SimpleNamespace from unittest.mock import AsyncMock from uuid import uuid4 import pytest from fastapi import HTTPException from pydantic import ValidationError as SchemaError from src.core.exceptions import ForbiddenError, NotFoundError, UnauthorizedError, ValidationError from src.domain.agent_access.dependencies import require_agent_access from src.domain.agent_access.forms import read_oauth_form from src.domain.agent_access.schemas import ( ClientCreate, GrantCreate, GrantUpdate, PurchasePolicy, WorkspaceAccess, ) from src.domain.agent_access.service import AgentAccessService, fingerprint, pkce_challenge, utcnow from src.domain.agent_access.tools import AgentToolService from starlette.requests import Request @pytest.fixture def service(): repo, workspaces, projects, policy = (AsyncMock() for _ in range(4)) repo.add.side_effect = lambda row: row workspaces.check_permission.return_value = True return AgentAccessService(repo, workspaces, projects, policy) def grant(**overrides): return SimpleNamespace( **{ "id": uuid4(), "workspace_id": uuid4(), "created_by": uuid4(), "name": "Agent", "kind": "user", "client_id": "client", "scopes": ["projects.read", "builds.run"], "project_ids": [], "workspace_wide": False, "purchase_policy": {}, "expires_at": utcnow() + timedelta(days=1), "revoked_at": None, "created_at": utcnow(), **overrides, } ) def access_row(workspace_id, scopes=("projects.read", "builds.run"), **overrides): """One agent_grant_workspaces row: what a grant may do in a single workspace.""" return SimpleNamespace( **{ "workspace_id": workspace_id, "scopes": list(scopes), "project_ids": [], "workspace_wide": True, "purchase_policy": {}, "created_at": utcnow(), **overrides, } ) def grant_access(service, *rows): service.repo.grant_workspaces.return_value = list(rows) def workspace_body(*workspace_ids, scopes=("projects.read",)): return GrantCreate( name="Agent", workspaces=[WorkspaceAccess(workspace_id=w, scopes=list(scopes)) for w in workspace_ids], expires_at=utcnow() + timedelta(days=1), ) def credential(service, kind="refresh", **overrides): return SimpleNamespace( **{ "grant_id": uuid4(), "kind": kind, "client_id": "client", "consumed_at": None, "expires_at": utcnow() + timedelta(hours=1), "details": {"resource": service.settings.agent_mcp_url, "scopes": ["projects.read"]}, **overrides, } ) def token_form(service, **overrides): return { "grant_type": "refresh_token", "refresh_token": "secret", "client_id": "client", "resource": service.settings.agent_mcp_url, **overrides, } async def test_refresh_preserves_narrowed_scope_by_default(service): row = credential(service) original = grant() service.repo.credential.return_value = row service.repo.grant.return_value = original result = await service.exchange(token_form(service)) assert result["scope"] == "projects.read" assert original.scopes == ["projects.read", "builds.run"] assert row.consumed_at assert "refresh_token" in result assert all( call.args[0].details["scopes"] == ["projects.read"] for call in service.repo.add.await_args_list ) assert all( call.args[0].fingerprint != result["access_token"] for call in service.repo.add.await_args_list ) async def test_refresh_cannot_restore_removed_or_ungranted_scope(service): service.repo.credential.return_value = credential(service) service.repo.grant.return_value = grant() with pytest.raises(ValidationError): await service.exchange(token_form(service, scope="builds.run")) service.repo.add.assert_not_awaited() @pytest.mark.parametrize( "kind,grant_type,field", [ ("refresh", "refresh_token", "refresh_token"), ("code", "authorization_code", "code"), ], ) async def test_replay_revokes_and_commits_before_error(service, kind, grant_type, field): original = grant() service.repo.credential.return_value = credential(service, kind, consumed_at=utcnow()) service.repo.grant.return_value = original with pytest.raises(UnauthorizedError): await service.exchange(token_form(service, grant_type=grant_type, **{field: "secret"})) assert original.revoked_at service.repo.db.commit.assert_awaited_once() service.repo.add.assert_not_awaited() @pytest.mark.parametrize( "verifier,redirect", [("x" * 43, "https://client.test/cb"), ("v" * 43, "https://evil.test/cb")] ) async def test_code_requires_pkce_and_exact_redirect(service, verifier, redirect): row = credential( service, "code", details={ "code_challenge": pkce_challenge("v" * 43), "redirect_uri": "https://client.test/cb", }, ) service.repo.credential.return_value = row service.repo.grant.return_value = grant() with pytest.raises(UnauthorizedError): await service.exchange( token_form( service, grant_type="authorization_code", code="secret", code_verifier=verifier, redirect_uri=redirect, ) ) assert row.consumed_at is None service.repo.add.assert_not_awaited() async def test_service_secret_is_reusable_without_refresh_token(service): row = credential(service, "service", client_id="service-id") service.repo.credential.return_value = row service.repo.grant.return_value = grant(kind="service") result = await service.exchange( token_form( service, grant_type="client_credentials", client_id="service-id", client_secret="secret" ) ) assert row.consumed_at is None assert "refresh_token" not in result async def test_authentication_never_mutates_stored_grant(service): original = grant() service.repo.credential.return_value = credential(service, "access") service.repo.grant.return_value = original principal = await service.authenticate("secret") assert principal.scopes == ["projects.read"] assert original.scopes == ["projects.read", "builds.run"] service.repo.credential.assert_awaited_once_with(fingerprint("secret")) async def test_wrong_resource_fails_before_credential_lookup(service): with pytest.raises(ValidationError): await service.exchange(token_form(service, resource="https://other.test/mcp")) service.repo.credential.assert_not_awaited() async def test_a2a_resource_is_also_a_valid_audience(service): row = credential(service, details={"resource": service.settings.agent_a2a_url, "scopes": ["projects.read"]}) service.repo.credential.return_value = row service.repo.grant.return_value = grant() result = await service.exchange(token_form(service, resource=service.settings.agent_a2a_url)) assert result["access_token"] async def test_refresh_token_cannot_mint_access_for_a_different_resource(service): """A refresh token issued for MCP must not silently mint an A2A-audience access token.""" row = credential(service, details={"resource": service.settings.agent_mcp_url, "scopes": ["projects.read"]}) service.repo.credential.return_value = row service.repo.grant.return_value = grant() with pytest.raises(ValidationError): await service.exchange(token_form(service, resource=service.settings.agent_a2a_url)) async def test_access_token_audience_is_checked_on_authenticate(service): row = credential(service, "access", details={"resource": service.settings.agent_a2a_url, "scopes": ["projects.read"]}) service.repo.credential.return_value = row service.repo.grant.return_value = grant() principal = await service.authenticate("secret") assert principal.scopes == ["projects.read"] @pytest.mark.parametrize("foreign", [True, False]) async def test_project_reads_reject_foreign_or_unselected_projects(service, foreign): principal = grant() project = uuid4() grant_access(service, access_row(principal.workspace_id, workspace_wide=False)) service.projects.get_by_id.return_value = SimpleNamespace( workspace_id=uuid4() if foreign else principal.workspace_id, deleted_at=None ) with pytest.raises(NotFoundError): await service.require_scope(principal, "projects.read", project) async def test_demotion_prevents_action(service): service.workspaces.check_permission.return_value = False principal = grant() grant_access(service, access_row(principal.workspace_id)) with pytest.raises(ForbiddenError): await service.require_scope(principal, "builds.run") service.productive_policy.require_for_workspace.assert_not_awaited() async def test_workspace_billing_and_mfa_policy_applies(service): principal = grant() grant_access(service, access_row(principal.workspace_id)) service.productive_policy.require_for_workspace.side_effect = ForbiddenError() with pytest.raises(ForbiddenError): await service.require_scope(principal, "builds.run") service.productive_policy.require_for_workspace.assert_awaited_once_with( principal.workspace_id, principal.created_by ) async def test_grant_cannot_delegate_unowned_permissions(service): # manager check, projects.read and codegen.run pass; delegating the scope fails service.workspaces.check_permission.side_effect = [True, True, True, False] body = workspace_body(uuid4()) with pytest.raises(ForbiddenError): await service.create_service_account(body, uuid4()) service.repo.add.assert_not_awaited() async def test_unattended_purchase_consent_requires_workspace_owner_and_policy(service): user = uuid4() service.workspaces.get_workspace_summary.return_value = SimpleNamespace(owner_id=uuid4()) body = workspace_body(uuid4(), scopes=("credits.purchase",)) with pytest.raises(ForbiddenError): await service.create_service_account(body, user) service.productive_policy.require_for_workspace.assert_awaited_once_with( body.workspaces[0].workspace_id, user ) service.repo.add.assert_not_awaited() @pytest.mark.parametrize( "url", ["http://client.test/cb", "https://user:pass@client.test/cb", "https://client.test/#cb"] ) def test_registration_rejects_unsafe_redirects(url): with pytest.raises(SchemaError): ClientCreate(client_name="Client", redirect_uris=[url]) def test_registration_ignores_dcr_metadata_it_does_not_understand(): """RFC 7591 servers MAY ignore unrecognized client metadata; they must not 422 the whole registration over it. Real clients (Claude among them) send fields like these that this server has no use for.""" client = ClientCreate( client_name="Claude", redirect_uris=["https://claude.ai/api/mcp/auth_callback"], scope="projects.read builds.run", client_uri="https://claude.ai", logo_uri="https://claude.ai/logo.png", contacts=["support@anthropic.com"], software_id="claude-mcp-connector", software_version="1.0", jwks_uri="https://claude.ai/.well-known/jwks.json", ) assert client.client_name == "Claude" assert not hasattr(client, "scope") assert "scope" not in client.model_dump() def test_registration_still_rejects_client_secret_post_because_nothing_verifies_one(): """"client_secret_post" is deliberately NOT in the AS metadata's token_endpoint_auth_methods_supported (see discovery.py) for exactly this reason: no client secret is ever issued or checked at /token for an authorization_code/refresh_token client. If a caller requests it anyway (ignoring the metadata), it must still fail here, not silently succeed.""" with pytest.raises(SchemaError): ClientCreate( client_name="Claude", redirect_uris=["https://claude.ai/api/mcp/auth_callback"], token_endpoint_auth_method="client_secret_post", ) def test_purchase_caps_require_explicit_consent(): with pytest.raises(SchemaError): PurchasePolicy(per_purchase_minor=500, monthly_minor=1000) with pytest.raises(SchemaError): PurchasePolicy(per_purchase_minor=2000, monthly_minor=1000, saved_payment_consent=True) @pytest.mark.parametrize( "chunks", [[b"client_id=a&client_id=b"], [b"x=\xff"], [b"x=" + b"a" * 16384]] ) async def test_oauth_form_rejects_duplicate_invalid_or_large_input(chunks): async def receive(): return {"type": "http.request", "body": chunks.pop(0), "more_body": bool(chunks)} request = Request( {"type": "http", "headers": [(b"content-type", b"application/x-www-form-urlencoded")]}, receive, ) with pytest.raises(ValidationError): await read_oauth_form(request) @pytest.mark.parametrize("supplied", [b"", b"wrong", b"\xff"]) def test_direct_core_access_requires_dashboard_proxy_secret(monkeypatch, supplied): monkeypatch.setattr( "src.domain.agent_access.dependencies.get_settings", lambda: SimpleNamespace( agent_access_enabled=True, agent_proxy_secret="private-proxy-secret" ), ) request = Request({"type": "http", "headers": [(b"x-p2hs-agent-proxy", supplied)]}) with pytest.raises(HTTPException) as error: require_agent_access(request) assert error.value.status_code == 404 def test_dashboard_proxy_secret_allows_enabled_access(monkeypatch): monkeypatch.setattr( "src.domain.agent_access.dependencies.get_settings", lambda: SimpleNamespace( agent_access_enabled=True, agent_proxy_secret="private-proxy-secret" ), ) require_agent_access( Request({"type": "http", "headers": [(b"x-p2hs-agent-proxy", b"private-proxy-secret")]}) ) # ---- one grant, several workspaces ---------------------------------------- async def test_a_project_resolves_to_its_own_workspace_and_is_permission_checked_there(service): a, b = uuid4(), uuid4() principal = grant() grant_access(service, access_row(a), access_row(b)) project = uuid4() service.projects.get_by_id.return_value = SimpleNamespace(workspace_id=b, deleted_at=None) assert await service.require_scope(principal, "builds.run", project) == b service.workspaces.check_permission.assert_awaited_with(b, principal.created_by, "codegen.run") service.productive_policy.require_for_workspace.assert_awaited_once_with( b, principal.created_by ) async def test_several_workspaces_require_an_explicit_choice(service): grant_access(service, access_row(uuid4()), access_row(uuid4())) with pytest.raises(ValidationError, match="list_workspaces"): await service.require_scope(grant(), "projects.read") async def test_a_single_workspace_grant_needs_no_workspace_argument(service): only = uuid4() grant_access(service, access_row(only)) assert await service.require_scope(grant(), "projects.read") == only async def test_a_workspace_outside_the_grant_is_not_found(service): grant_access(service, access_row(uuid4()), access_row(uuid4())) with pytest.raises(NotFoundError): await service.require_scope(grant(), "projects.read", workspace_id=uuid4()) service.workspaces.check_permission.assert_not_awaited() async def test_a_project_in_a_workspace_outside_the_grant_is_not_found(service): grant_access(service, access_row(uuid4())) service.projects.get_by_id.return_value = SimpleNamespace(workspace_id=uuid4(), deleted_at=None) with pytest.raises(NotFoundError): await service.require_scope(grant(), "projects.read", uuid4()) async def test_scopes_are_per_workspace_not_just_per_grant(service): """builds.run is in the grant's union, but only workspace A was given it.""" a, b = uuid4(), uuid4() principal = grant(scopes=["projects.read", "builds.run"]) grant_access( service, access_row(a, ["projects.read", "builds.run"]), access_row(b, ["projects.read"]) ) assert await service.require_scope(principal, "builds.run", workspace_id=a) == a with pytest.raises(ForbiddenError): await service.require_scope(principal, "builds.run", workspace_id=b) async def test_a_demoted_workspace_drops_out_without_breaking_the_others(service): a, b = uuid4(), uuid4() principal = grant() grant_access(service, access_row(a), access_row(b)) async def permission(workspace_id, user_id, permission): return workspace_id == b service.workspaces.check_permission.side_effect = permission assert await service.scoped_workspaces(principal, "builds.run") == [b] async def test_grant_is_refused_in_a_workspace_without_developer_access(service): """A viewer (projects.read only) may not hand an agent a workspace they cannot build in.""" ok, viewer_only = uuid4(), uuid4() async def permission(workspace_id, user_id, permission): return workspace_id == ok or permission == "projects.read" service.workspaces.check_permission.side_effect = permission with pytest.raises(ForbiddenError, match="Developer access"): await service.create_grant(workspace_body(ok, viewer_only), uuid4(), client_id="client") service.repo.add.assert_not_awaited() async def test_consent_stores_a_workspace_row_with_its_own_policy_for_each_workspace(service): a, b = uuid4(), uuid4() body = GrantCreate( name="Agent", workspaces=[ WorkspaceAccess(workspace_id=a, scopes=["projects.read"]), WorkspaceAccess( workspace_id=b, scopes=["projects.read", "credits.purchase"], purchase_policy=PurchasePolicy( per_purchase_minor=500, monthly_minor=1000, saved_payment_consent=True ), ), ], expires_at=utcnow() + timedelta(days=1), ) user = uuid4() service.workspaces.get_workspace_summary.return_value = SimpleNamespace(owner_id=user) created = await service.create_grant(body, user, client_id="client") rows = { c.args[0].workspace_id: c.args[0] for c in service.repo.add.await_args_list if hasattr(c.args[0], "workspace_wide") and hasattr(c.args[0], "grant_id") } assert set(rows) == {a, b} assert rows[a].purchase_policy["monthly_minor"] == 0 assert rows[b].purchase_policy["monthly_minor"] == 1000 assert rows[a].workspace_wide and rows[a].project_ids == [] assert created.scopes == ["credits.purchase", "projects.read"] # union, for OAuth negotiation def test_a_workspace_may_only_be_listed_once(): workspace = uuid4() with pytest.raises(SchemaError): workspace_body(workspace, workspace) def test_spending_limits_need_the_purchase_scope_in_that_same_workspace(): with pytest.raises(SchemaError): WorkspaceAccess( workspace_id=uuid4(), scopes=["projects.read"], purchase_policy=PurchasePolicy( per_purchase_minor=500, monthly_minor=1000, saved_payment_consent=True ), ) async def test_editing_a_connection_cannot_add_permissions_it_never_had(service): user = uuid4() original = grant(created_by=user, scopes=["projects.read"]) service.repo.grant.return_value = original body = GrantUpdate(workspaces=[WorkspaceAccess(workspace_id=uuid4(), scopes=["builds.run"])]) with pytest.raises(ValidationError, match="not originally approved"): await service.update_grant(original.id, body, user) async def test_only_the_creator_may_edit_a_connection(service): original = grant(created_by=uuid4()) service.repo.grant.return_value = original with pytest.raises(NotFoundError): await service.update_grant(original.id, GrantUpdate(name="Renamed"), uuid4()) assert original.name == "Agent" async def test_editing_replaces_the_workspace_set(service): keep, drop, new = uuid4(), uuid4(), uuid4() user = uuid4() original = grant(created_by=user, scopes=["projects.read", "builds.run"], workspace_id=drop) kept_row = access_row(keep, ["projects.read", "builds.run"]) service.repo.grant.return_value = original grant_access(service, kept_row, access_row(drop)) body = GrantUpdate( workspaces=[ WorkspaceAccess(workspace_id=keep, scopes=["projects.read"]), WorkspaceAccess(workspace_id=new, scopes=["projects.read", "builds.run"]), ] ) await service.update_grant(original.id, body, user) service.repo.remove_grant_workspace.assert_awaited_once_with(original.id, drop) assert kept_row.scopes == ["projects.read"] added = [c.args[0] for c in service.repo.add.await_args_list] assert [row.workspace_id for row in added] == [new] assert original.workspace_id == keep async def test_a_non_creator_manager_only_detaches_their_own_workspace(service): mine, other = uuid4(), uuid4() original = grant(created_by=uuid4(), workspace_id=mine) service.repo.grant.return_value = original grant_access(service, access_row(mine), access_row(other)) await service.revoke(mine, original.id, uuid4()) service.repo.remove_grant_workspace.assert_awaited_once_with(original.id, mine) assert original.revoked_at is None assert original.workspace_id == other async def test_a_creator_revoking_cuts_off_every_workspace(service): a, b = uuid4(), uuid4() user = uuid4() original = grant(created_by=user, workspace_id=a) service.repo.grant.return_value = original grant_access(service, access_row(a), access_row(b)) await service.revoke(a, original.id, user) assert original.revoked_at service.repo.remove_grant_workspace.assert_not_awaited() async def test_workspace_options_freeze_what_the_person_cannot_delegate(service): dev_ws, billing_ws, viewer_ws = uuid4(), uuid4(), uuid4() user = uuid4() developer = {"projects.read", "projects.create", "codegen.run", "invoices.read"} held = { dev_ws: developer, billing_ws: developer | {"billing.manage"}, viewer_ws: {"projects.read"}, } service.workspaces.list_workspace_summaries_for_user.return_value = [ SimpleNamespace(id=dev_ws, name="Dev", owner_id=uuid4()), SimpleNamespace(id=billing_ws, name="Bill", owner_id=user), SimpleNamespace(id=viewer_ws, name="View", owner_id=uuid4()), ] async def permission(workspace_id, user_id, permission): return permission in held[workspace_id] service.workspaces.check_permission.side_effect = permission options = {o["name"]: o for o in await service.workspace_options(user)} assert options["Dev"]["eligible"] and not options["Dev"]["can_manage_billing"] assert "credits.checkout" not in options["Dev"]["scopes"] assert "credits.purchase" not in options["Dev"]["scopes"] assert options["Bill"]["eligible"] and options["Bill"]["is_owner"] assert "credits.purchase" in options["Bill"]["scopes"] assert not options["View"]["eligible"] async def test_a_billing_manager_who_is_not_the_owner_cannot_delegate_purchases(service): service.workspaces.list_workspace_summaries_for_user.return_value = [ SimpleNamespace(id=uuid4(), name="Shared", owner_id=uuid4()) ] [option] = await service.workspace_options(uuid4()) # check_permission is True for all assert option["can_manage_billing"] and not option["is_owner"] assert "credits.checkout" in option["scopes"] assert "credits.purchase" not in option["scopes"] async def test_the_authenticated_principal_never_carries_a_default_workspace(service): """Every tool must resolve its workspace through require_scope(); an ambient principal.workspace_id would silently pick one of several.""" service.repo.credential.return_value = credential(service, "access") service.repo.grant.return_value = grant() principal = await service.authenticate("secret") assert not hasattr(principal, "workspace_id") async def test_capabilities_list_only_workspaces_where_a_scope_still_works(service): a, b = uuid4(), uuid4() principal = grant(scopes=["projects.read"]) grant_access(service, access_row(a, ["projects.read"]), access_row(b, ["projects.read"])) async def permission(workspace_id, user_id, permission): return workspace_id == a service.workspaces.check_permission.side_effect = permission service.workspaces.get_workspace_summary.return_value = SimpleNamespace(name="Alpha") result = await AgentToolService(service, service.repo).capabilities(principal) assert [w["id"] for w in result["workspaces"]] == [str(a)] assert result["workspace_id"] == str(a) assert "list_workspaces" not in result["tools"] # only one usable workspace async def test_list_workspaces_is_offered_only_when_there_is_a_choice(service): a, b = uuid4(), uuid4() principal = grant(scopes=["projects.read"]) grant_access(service, access_row(a, ["projects.read"]), access_row(b, ["projects.read"])) service.workspaces.get_workspace_summary.return_value = SimpleNamespace(name="W") result = await AgentToolService(service, service.repo).capabilities(principal) assert result["tools"][0] == "list_workspaces" assert {w["id"] for w in result["workspaces"]} == {str(a), str(b)} async def test_rate_limit_accounting_ignores_a_workspace_that_is_not_in_the_grant(service): """Otherwise any connection could burn another tenant's workspace budget just by naming its id in an argument.""" mine = uuid4() service.redis = object() # limiter enabled grant_access(service, access_row(mine), access_row(uuid4())) tools = AgentToolService(service, service.repo) principal = grant() assert await tools._rate_limit_workspace(principal, {"workspace_id": str(uuid4())}) is None assert await tools._rate_limit_workspace(principal, {"workspace_id": str(mine)}) == mine assert await tools._rate_limit_workspace(principal, {"workspace_id": "not-a-uuid"}) is None async def test_only_the_creator_of_a_live_connection_can_edit_it(service): user = uuid4() mine = grant(created_by=user) theirs = grant(created_by=uuid4()) revoked = grant(created_by=user, revoked_at=utcnow()) service.repo.grant_workspaces_by_grant.return_value = { g.id: [access_row(g.workspace_id)] for g in (mine, theirs, revoked) } views = await service.grant_views([mine, theirs, revoked], user) assert [v["can_edit"] for v in views] == [True, False, False] assert all(len(v["workspaces"]) == 1 for v in views)