"""Phase 5A X API v2 OAuth and account-discovery coverage. All provider traffic is mocked. Normal CI never needs X credentials, API credits, or an interactive browser authorization flow. """ from __future__ import annotations import base64 from datetime import datetime, timedelta, timezone from pathlib import Path from urllib.parse import parse_qs, urlparse import httpx import pytest from pydantic import ValidationError from sqlalchemy import select from app.container import build_container from app.core.config import Settings from app.social.domain.errors import ( SocialAccountNotFoundError, SocialOAuthStateError, SocialPermissionDeniedError, SocialProviderUnavailableError, SocialReauthRequiredError, ) from app.social.models import OAuthState, SocialAccountToken from app.social.providers.x import XProvider from app.social.schemas.accounts import SocialAccountConnectRequest _REDIRECT_URI = "https://api.example.com/v1/social/accounts/x/callback" def x_settings(tmp_path: Path) -> Settings: return Settings( _env_file=None, auth_enabled=False, database_url=f"sqlite+aiosqlite:///{tmp_path / 'security.db'}", social_database_url=f"sqlite+aiosqlite:///{tmp_path / 'social.db'}", social_auto_migrate=True, social_worker_enabled=False, social_oauth_encryption_key="phase-5a-test-encryption-material", social_oauth_redirect_base_url="https://api.example.com", x_client_id="x-client-id", x_client_secret="x-client-secret", x_redirect_uri=_REDIRECT_URI, x_publishing_enabled=True, temp_dir=tmp_path / "temp", output_dir=tmp_path / "outputs", cleanup_interval_seconds=3600, whisper_model="tiny", ) def assert_confidential_client(request: httpx.Request) -> None: scheme, encoded = request.headers["authorization"].split(" ", 1) assert scheme == "Basic" assert base64.b64decode(encoded).decode() == "x-client-id:x-client-secret" async def test_x_authorization_uses_official_url_minimum_scopes_and_s256_pkce( tmp_path: Path, ) -> None: provider = XProvider(x_settings(tmp_path)) try: url = await provider.get_authorization_url( state="s" * 43, redirect_uri=_REDIRECT_URI, code_challenge="s256-code-challenge", ) with pytest.raises(SocialPermissionDeniedError): await provider.get_authorization_url( state="s" * 43, redirect_uri=_REDIRECT_URI, code_challenge=None, ) finally: await provider.close() parsed = urlparse(url) query = parse_qs(parsed.query) assert f"{parsed.scheme}://{parsed.netloc}{parsed.path}" == ( "https://x.com/i/oauth2/authorize" ) assert query["client_id"] == ["x-client-id"] assert query["redirect_uri"] == [_REDIRECT_URI] assert query["response_type"] == ["code"] assert query["scope"] == ["tweet.read users.read offline.access"] assert query["state"] == ["s" * 43] assert query["code_challenge"] == ["s256-code-challenge"] assert query["code_challenge_method"] == ["S256"] assert "tweet.write" not in query["scope"][0] assert "media.write" not in query["scope"][0] async def test_x_exchange_refresh_discovery_and_revoke_use_official_v2_endpoints( tmp_path: Path, ) -> None: calls: list[str] = [] async def handler(request: httpx.Request) -> httpx.Response: calls.append(request.url.path) assert request.url.host == "api.x.com" if request.url.path == "/2/oauth2/token": assert_confidential_client(request) form = parse_qs(request.content.decode()) assert "client_secret" not in form assert "client_id" not in form if form["grant_type"] == ["authorization_code"]: assert form == { "code": ["authorization-code"], "grant_type": ["authorization_code"], "redirect_uri": [_REDIRECT_URI], "code_verifier": ["pkce-verifier"], } else: assert form == { "refresh_token": ["refresh-token"], "grant_type": ["refresh_token"], } return httpx.Response( 200, json={ "access_token": "x-access-token", "refresh_token": "x-rotated-refresh-token", "expires_in": 7200, "scope": "tweet.read users.read offline.access", "token_type": "bearer", }, ) if request.url.path == "/2/users/me": assert request.headers["authorization"] == "Bearer x-access-token" assert parse_qs(request.url.query.decode()) == { "user.fields": [ "created_at,description,profile_image_url,protected,verified" ] } return httpx.Response( 200, json={ "data": { "id": "2244994945", "username": "XDevelopers", "name": "X Developers", "profile_image_url": "https://pbs.twimg.com/profile.jpg", "created_at": "2013-12-14T04:35:55.000Z", "description": "Official developer account", "protected": False, "verified": True, } }, ) assert request.url.path == "/2/oauth2/revoke" assert_confidential_client(request) assert parse_qs(request.content.decode()) == { "token": ["x-rotated-refresh-token"] } return httpx.Response(200) client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) provider = XProvider(x_settings(tmp_path), http_client=client) try: token = await provider.exchange_code( code="authorization-code", redirect_uri=_REDIRECT_URI, code_verifier="pkce-verifier", ) account = await provider.get_account(token) refreshed = await provider.refresh_token( {"access_token": "old-token", "refresh_token": "refresh-token"} ) await provider.revoke_token(refreshed) finally: await client.aclose() assert account == { "external_account_id": "2244994945", "account_type": "user", "username": "XDevelopers", "display_name": "X Developers", "avatar_url": "https://pbs.twimg.com/profile.jpg", "metadata": { "x_user_id": "2244994945", "created_at": "2013-12-14T04:35:55.000Z", "verified": True, "protected": False, "description": "Official developer account", }, } assert refreshed["refresh_token"] == "x-rotated-refresh-token" assert calls == [ "/2/oauth2/token", "/2/users/me", "/2/oauth2/token", "/2/oauth2/revoke", ] async def test_x_invalid_code_and_pkce_failure_are_normalized_without_secrets( tmp_path: Path, ) -> None: secret_code = "x-code-that-must-not-leak" async def handler(_: httpx.Request) -> httpx.Response: return httpx.Response( 400, json={ "error": "invalid_grant", "error_description": f"invalid code {secret_code}", }, ) client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) provider = XProvider(x_settings(tmp_path), http_client=client) try: with pytest.raises(SocialPermissionDeniedError): await provider.exchange_code( code=secret_code, redirect_uri=_REDIRECT_URI, code_verifier=None, ) with pytest.raises(SocialReauthRequiredError) as raised: await provider.exchange_code( code=secret_code, redirect_uri=_REDIRECT_URI, code_verifier="incorrect-verifier", ) finally: await client.aclose() assert secret_code not in str(raised.value) async def test_x_invalid_client_is_configuration_failure_not_consent_loop( tmp_path: Path, ) -> None: async def handler(_: httpx.Request) -> httpx.Response: return httpx.Response( 401, json={ "error": "invalid_client", "error_description": "client secret is not accepted", }, ) client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) provider = XProvider(x_settings(tmp_path), http_client=client) try: with pytest.raises(SocialProviderUnavailableError) as raised: await provider.exchange_code( code="authorization-code", redirect_uri=_REDIRECT_URI, code_verifier="pkce-verifier", ) finally: await client.aclose() assert "client secret is not accepted" not in str(raised.value) async def test_x_account_discovery_rejects_non_ascii_or_oversized_user_ids( tmp_path: Path, ) -> None: invalid_ids = ["٢٢٤٤٩٩٤٩٤٥", "12345678901234567890"] for user_id in invalid_ids: async def handler(_: httpx.Request, value: str = user_id) -> httpx.Response: return httpx.Response( 200, json={"data": {"id": value, "username": "invalid"}}, ) client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) provider = XProvider(x_settings(tmp_path), http_client=client) try: with pytest.raises(SocialProviderUnavailableError): await provider.get_account({"access_token": "x-access-token"}) finally: await client.aclose() async def test_x_callback_is_single_use_duplicate_safe_and_workspace_bound( tmp_path: Path, ) -> None: container = build_container(x_settings(tmp_path)) await container.social.initialize() adapter = container.social.accounts.providers.get("x") assert isinstance(adapter, XProvider) await adapter._client.aclose() async def handler(request: httpx.Request) -> httpx.Response: if request.url.path == "/2/oauth2/token": form = parse_qs(request.content.decode()) assert form.get("code_verifier", [""])[0] return httpx.Response( 200, json={ "access_token": "x-token-that-must-stay-encrypted", "refresh_token": "x-refresh-that-must-stay-encrypted", "expires_in": 7200, "scope": "tweet.read users.read offline.access", "token_type": "bearer", }, ) return httpx.Response( 200, json={ "data": { "id": "2244994945", "username": "workspace_user", "name": "Workspace User", } }, ) adapter._client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) adapter._owns_client = True try: first_connect = await container.social.oauth.connect( provider="x", workspace_id="workspace-a", user_id="user-a", payload=SocialAccountConnectRequest(), ) first_query = parse_qs(urlparse(first_connect.authorization_url or "").query) first_state = first_query["state"][0] assert first_query["code_challenge_method"] == ["S256"] assert first_query["code_challenge"][0] first = await container.social.oauth.callback( provider="x", state=first_state, code="first-code", ) with pytest.raises(SocialOAuthStateError): await container.social.oauth.callback( provider="x", state=first_state, code="replayed-code", ) second_connect = await container.social.oauth.connect( provider="x", workspace_id="workspace-a", user_id="user-a", payload=SocialAccountConnectRequest(), ) second_state = parse_qs( urlparse(second_connect.authorization_url or "").query )["state"][0] second = await container.social.oauth.callback( provider="x", state=second_state, code="second-code", ) assert first.id == second.id accounts = await container.social.accounts.list("workspace-a") assert [account.id for account in accounts if account.provider.value == "x"] == [ first.id ] assert "x-token-that-must-stay-encrypted" not in first.model_dump_json() with pytest.raises(SocialAccountNotFoundError): await container.social.accounts.get("workspace-b", first.id) async with container.social.database.session("workspace-a") as session: stored = await session.scalar( select(SocialAccountToken).where( SocialAccountToken.social_account_id == first.id ) ) assert stored is not None assert stored.expires_at is not None assert stored.encrypted_payload assert "x-token-that-must-stay-encrypted" not in stored.encrypted_payload finally: await container.social.close() await container.security_database.close() async def test_x_state_redirect_provider_and_expiry_validation(tmp_path: Path) -> None: container = build_container(x_settings(tmp_path)) await container.social.initialize() try: assert container.social.oauth._redirect_uri("x", None) == _REDIRECT_URI with pytest.raises(SocialPermissionDeniedError): container.social.oauth._redirect_uri( "x", "https://attacker.example/v1/social/accounts/x/callback", ) state = await container.social.oauth.states.create( provider="x", workspace_id="workspace-a", user_id="user-a", redirect_uri=_REDIRECT_URI, ) with pytest.raises(SocialOAuthStateError): await container.social.oauth.states.consume( state=state.state, provider="linkedin", ) consumed = await container.social.oauth.states.consume( state=state.state, provider="x", ) assert consumed.workspace_id == "workspace-a" assert consumed.user_id == "user-a" expired = OAuthState( state="expired-x-state-value-that-is-long-enough", provider="x", workspace_id="workspace-a", user_id="user-a", redirect_uri=_REDIRECT_URI, expires_at=datetime.now(timezone.utc) - timedelta(seconds=1), ) async with container.social.database.session("workspace-a") as session: session.add(expired) await session.commit() with pytest.raises(SocialOAuthStateError): await container.social.oauth.states.consume( state=expired.state, provider="x", ) finally: await container.social.close() await container.security_database.close() def test_x_redirect_configuration_is_fail_closed() -> None: invalid_redirects = [ "https://attacker.example/not-the-x-callback", "ftp://localhost/v1/social/accounts/x/callback", "http://api.example.com/v1/social/accounts/x/callback", "https://api.example.com/v1/social/accounts/x/callback?next=attacker", ] for redirect in invalid_redirects: with pytest.raises(ValidationError): Settings(_env_file=None, x_redirect_uri=redirect) settings = Settings( _env_file=None, x_redirect_uri="http://localhost/v1/social/accounts/x/callback", ) assert settings.x_redirect_uri.startswith("http://localhost/") async def test_x_capability_discovery_advertises_implemented_publishing( tmp_path: Path, ) -> None: container = build_container(x_settings(tmp_path)) try: provider = container.social.accounts.get_provider("x") assert provider.available assert provider.configured assert provider.capabilities.implementation_status == "implemented" assert provider.capabilities.account_types == ["user"] assert provider.capabilities.required_scopes == [ "tweet.read", "users.read", "offline.access", ] assert provider.capabilities.video assert provider.capabilities.video_upload assert provider.capabilities.video_status assert provider.capabilities.image assert provider.capabilities.direct_publish assert not provider.capabilities.draft_upload assert provider.capabilities.scheduled_publish assert not provider.capabilities.native_scheduling assert provider.capabilities.delete_post assert provider.capabilities.publishing_required_scopes == [ "tweet.write", "media.write", ] assert provider.capabilities.analytics assert provider.capabilities.analytics_required_scopes == ["tweet.read"] finally: await container.social.close() await container.security_database.close() async def test_x_publishing_capabilities_are_fail_closed_without_operator_gate( tmp_path: Path, ) -> None: settings = x_settings(tmp_path).model_copy( update={"x_publishing_enabled": False} ) provider = XProvider(settings) try: assert provider.configuration_ready assert not provider.publishing_ready assert not provider.capabilities.direct_publish assert not provider.capabilities.video_upload assert not provider.capabilities.delete_post assert provider.capabilities.publishing_required_scopes == [] finally: await provider.close()