"""Broker security/lifecycle tests with a provider and persistence boundary fake.""" from datetime import UTC, datetime, timedelta from types import SimpleNamespace from unittest.mock import AsyncMock from urllib.parse import parse_qs, urlsplit from uuid import uuid4 import pytest from src.core.exceptions import ValidationError from src.core.security.encryption import decrypt_value from src.domain.provider_auth.identity import bind_identity, consume_identity, start_identity from src.domain.provider_auth.manifests import ( FAMILY_SCOPES, HUBSPOT_FAMILY_SCOPES, SALESFORCE_FAMILY_SCOPES, authorization_manifest, ) from src.domain.provider_auth.models import AuthorizationSession, ProviderConnection, ProviderGrant from src.domain.provider_auth.schemas import CompleteAuthorization, StartAuthorization from src.domain.provider_auth.service import ProviderAuthService, safe_return_to from src.integrations.google_oauth import GoogleOAuthError class MemoryRepo: def __init__(self): self.rows = [] self.db = AsyncMock() async def add(self, row): self.rows.append(row) return row async def session_by_state(self, hashed): return next((r for r in self.rows if isinstance(r, AuthorizationSession) and r.state_hash == hashed), None) async def session_by_completion(self, hashed, project_id): return next((r for r in self.rows if isinstance(r, AuthorizationSession) and r.completion_hash == hashed and r.project_id == project_id), None) async def connection(self, workspace_id, project_id, connection_id): return next((r for r in self.rows if isinstance(r, ProviderConnection) and r.id == connection_id and r.project_id == project_id and r.workspace_id == workspace_id), None) async def by_subject(self, session, subject): return next((r for r in self.rows if isinstance(r, ProviderConnection) and r.subject == subject and r.project_id == session.project_id and r.actor_id == session.actor_id), None) async def grants(self, connection_id): return [r for r in self.rows if isinstance(r, ProviderGrant) and r.connection_id == connection_id] @pytest.fixture def broker(): google = SimpleNamespace(token=AsyncMock(return_value={"access_token": "access", "refresh_token": "refresh", "expires_in": 3600, "scope": " ".join(FAMILY_SCOPES["identity"] + FAMILY_SCOPES["docs"])}), userinfo=AsyncMock(return_value={"sub": "subject", "email": "test@example.com", "email_verified": True})) cfg = SimpleNamespace(google_platform_client_id="test-client", google_platform_client_secret="test-secret", google_platform_redirect_uri="https://dashboard.p2hs.com/api/oauth/google/callback", google_platform_enabled_families="identity,docs,sheets", google_platform_session_ttl_seconds=600, secret_key="test-signing-key") return ProviderAuthService(MemoryRepo(), google, cfg) @pytest.fixture def hubspot_broker(): hubspot = SimpleNamespace( token=AsyncMock(return_value={"access_token": "hs-access", "refresh_token": "hs-refresh", "expires_in": 3600, "scope": " ".join(HUBSPOT_FAMILY_SCOPES["crm"])}), userinfo=AsyncMock(return_value={"sub": "hubspot-user", "portal_id": "12345"}), ) cfg = SimpleNamespace( hubspot_platform_client_id="hs-client", hubspot_platform_client_secret="hs-secret", hubspot_platform_redirect_uri="https://dashboard.p2hs.com/api/oauth/hubspot/callback", hubspot_platform_enabled_families="crm", hubspot_platform_session_ttl_seconds=600, secret_key="test-signing-key", ) return ProviderAuthService(MemoryRepo(), hubspot, cfg, provider="hubspot") async def begin(broker, family="docs", project=None, connection_id=None): project = project or SimpleNamespace(id=uuid4(), workspace_id=uuid4()) result = await broker.start(workspace_id=project.workspace_id, project_id=project.id, initiated_by=uuid4(), body=StartAuthorization(grant_family=family, browser_binding="b" * 32, connection_id=connection_id)) state = parse_qs(urlsplit(result.authorize_url).query)["state"][0] return project, result, CompleteAuthorization(state=state, code="code", browser_binding="b" * 32) @pytest.mark.asyncio async def test_callback_single_use_encrypted_and_exact_redirect(broker): _, result, callback = await begin(broker) params = parse_qs(urlsplit(result.authorize_url).query) assert params["include_granted_scopes"] == ["true"] assert params["code_challenge_method"] == ["S256"] assert (await broker.complete(callback)).status == "completed" broker.repo.db.commit.assert_awaited_once() assert broker.google.token.call_args.kwargs["redirect_uri"] == params["redirect_uri"][0] connection = next(r for r in broker.repo.rows if isinstance(r, ProviderConnection)) assert connection.secrets_enc["access_token"] != "access" assert decrypt_value(connection.secrets_enc["refresh_token"]) == "refresh" with pytest.raises(ValidationError): await broker.complete(callback) assert broker.google.token.await_count == 1 @pytest.mark.asyncio async def test_wrong_browser_and_tampered_state_never_exchange(broker): _, _, callback = await begin(broker) with pytest.raises(ValidationError): await broker.complete(callback.model_copy(update={"browser_binding": "x" * 32})) with pytest.raises(ValidationError): await broker.complete(callback.model_copy(update={"state": callback.state + "x"})) broker.google.token.assert_not_awaited() @pytest.mark.asyncio async def test_partial_consent_never_invents_scopes(broker): _, _, callback = await begin(broker) broker.google.token.return_value["scope"] = "openid email profile" assert (await broker.complete(callback)).status == "scope_escalation_pending" connection = next(r for r in broker.repo.rows if isinstance(r, ProviderConnection)) assert FAMILY_SCOPES["docs"][0] not in connection.effective_scopes @pytest.mark.asyncio async def test_scope_elevation_preserves_refresh_and_rejects_account_swap(broker): project, _, callback = await begin(broker) await broker.complete(callback) connection = next(r for r in broker.repo.rows if isinstance(r, ProviderConnection)) _, _, callback2 = await begin(broker, "sheets", project, connection.id) broker.google.token.return_value = {"access_token": "next", "expires_in": 3600, "scope": " ".join(FAMILY_SCOPES["docs"] + FAMILY_SCOPES["sheets"])} assert (await broker.complete(callback2)).status == "completed" assert decrypt_value(connection.secrets_enc["refresh_token"]) == "refresh" assert set(connection.effective_scopes) == set(FAMILY_SCOPES["docs"] + FAMILY_SCOPES["sheets"]) _, _, callback3 = await begin(broker, "docs", project, connection.id) broker.google.userinfo.return_value["sub"] = "someone-else" assert (await broker.complete(callback3)).status == "account_mismatch" @pytest.mark.asyncio async def test_expired_session_and_denial(broker): _, _, callback = await begin(broker) broker.repo.rows[0].expires_at = datetime.now(UTC) - timedelta(seconds=1) with pytest.raises(ValidationError): await broker.complete(callback) _, _, callback = await begin(broker) assert (await broker.complete(callback.model_copy(update={"error": "access_denied"}))).status == "denied" broker.google.token.assert_not_awaited() @pytest.mark.asyncio async def test_invalid_grant_has_distinct_reason(broker): _, _, callback = await begin(broker) broker.google.token.side_effect = GoogleOAuthError("invalid_grant") assert (await broker.complete(callback)).status == "reauthentication_required" @pytest.mark.parametrize("target", ["https://evil.test", "//evil.test", "/\\evil.test", "/connectors\n", "/api/anything"]) def test_return_destination_rejected(target): with pytest.raises(ValidationError): safe_return_to(target) @pytest.mark.asyncio async def test_identity_plane_never_creates_connector_credentials(broker): project = SimpleNamespace(id=uuid4(), workspace_id=uuid4()) result = await start_identity(broker, project=project, return_to="https://app.test/login", cloud_binding="c" * 32, completion_url="https://cloud.test/complete", allowed_origins={"https://app.test"}) state = parse_qs(urlsplit(result.authorize_url).query)["state"][0] await bind_identity(broker, state, "b" * 32) with pytest.raises(ValidationError): await bind_identity(broker, state, "b" * 32) done = await broker.complete(CompleteAuthorization(state=state, code="code", browser_binding="b" * 32)) assert done.purpose == "identity" assert not any(isinstance(r, (ProviderConnection, ProviderGrant)) for r in broker.repo.rows) ticket = parse_qs(urlsplit(done.return_to).query)["ticket"][0] with pytest.raises(ValidationError): await consume_identity(broker, project.id, ticket, "wrong" * 8) consumed = await consume_identity(broker, project.id, ticket, "c" * 32) assert consumed["identity"]["sub"] == "subject" assert "access_token" not in str(consumed) with pytest.raises(ValidationError): await consume_identity(broker, project.id, ticket, "c" * 32) @pytest.mark.asyncio async def test_unshipped_family_does_not_request_consent(broker): broker.settings.google_platform_enabled_families = "identity" with pytest.raises(ValidationError): await begin(broker) assert not broker.repo.rows @pytest.mark.asyncio async def test_hubspot_platform_oauth_uses_generic_broker(hubspot_broker): _, result, callback = await begin(hubspot_broker, "crm") params = parse_qs(urlsplit(result.authorize_url).query) assert urlsplit(result.authorize_url).netloc == "app.hubspot.com" assert params["client_id"] == ["hs-client"] assert (await hubspot_broker.complete(callback)).status == "completed" connection = next(r for r in hubspot_broker.repo.rows if isinstance(r, ProviderConnection)) assert connection.provider == "hubspot" assert decrypt_value(connection.secrets_enc["refresh_token"]) == "hs-refresh" def test_connector_authorization_manifests_identify_provider(): assert authorization_manifest("google_docs")["provider"] == "google" assert authorization_manifest("hubspot")["provider"] == "hubspot" assert authorization_manifest("salesforce")["scopes"] == list(SALESFORCE_FAMILY_SCOPES["crm"]) assert authorization_manifest("x_ads")["provider"] == "x_ads" assert authorization_manifest("microsoft_ads")["provider"] == "microsoft_ads"