"""Phase 6A LinkedIn OIDC and organization-discovery coverage. All LinkedIn traffic is mocked. Normal CI needs no developer application, member credential, organization role, or interactive authorization flow. """ from __future__ import annotations 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 func, select from app.container import build_container from app.core.config import Settings from app.social.domain.errors import ( SocialAccountNotFoundError, SocialCapabilityUnsupportedError, SocialOAuthStateError, SocialPermissionDeniedError, SocialReauthRequiredError, ) from app.social.models import OAuthState, SocialAccountToken from app.social.providers.linkedin import LINKEDIN_API_VERSION, LinkedInProvider from app.social.schemas.accounts import SocialAccountConnectRequest _REDIRECT_URI = "https://api.example.com/v1/social/accounts/linkedin/callback" def linkedin_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-6a-linkedin-test-encryption-material", social_oauth_redirect_base_url="https://api.example.com", linkedin_client_id="linkedin-client-id", linkedin_client_secret="linkedin-client-secret", linkedin_redirect_uri=_REDIRECT_URI, temp_dir=tmp_path / "temp", output_dir=tmp_path / "outputs", cleanup_interval_seconds=3600, whisper_model="tiny", ) async def test_linkedin_member_authorization_uses_official_oidc_without_pkce( tmp_path: Path, ) -> None: provider = LinkedInProvider(linkedin_settings(tmp_path)) try: url = await provider.get_authorization_url( state="s" * 43, redirect_uri=_REDIRECT_URI, ) with pytest.raises(SocialPermissionDeniedError): await provider.get_authorization_url( state="s" * 43, redirect_uri=_REDIRECT_URI, code_challenge="undocumented-pkce-challenge", ) with pytest.raises(SocialCapabilityUnsupportedError): await provider.get_authorization_url( state="s" * 43, redirect_uri=_REDIRECT_URI, additional_scopes=["w_member_social"], ) finally: await provider.close() parsed = urlparse(url) query = parse_qs(parsed.query) assert f"{parsed.scheme}://{parsed.netloc}{parsed.path}" == ( "https://www.linkedin.com/oauth/v2/authorization" ) assert query == { "client_id": ["linkedin-client-id"], "redirect_uri": [_REDIRECT_URI], "response_type": ["code"], "state": ["s" * 43], "scope": ["openid profile"], } assert "code_challenge" not in query async def test_linkedin_exchange_member_and_organization_discovery_use_official_apis( tmp_path: Path, ) -> None: calls: list[str] = [] async def handler(request: httpx.Request) -> httpx.Response: calls.append(request.url.path) if request.url.path == "/oauth/v2/accessToken": assert request.url.host == "www.linkedin.com" form = parse_qs(request.content.decode()) assert form == { "grant_type": ["authorization_code"], "code": ["authorization-code"], "redirect_uri": [_REDIRECT_URI], "client_id": ["linkedin-client-id"], "client_secret": ["linkedin-client-secret"], } return httpx.Response( 200, json={ "access_token": "linkedin-access-token", "expires_in": 5184000, "token_type": "Bearer", }, ) assert request.headers["authorization"] == "Bearer linkedin-access-token" if request.url.path == "/v2/userinfo": return httpx.Response( 200, json={ "sub": "oidc-member-subject_123", "name": "Ada Lovelace", "given_name": "Ada", "family_name": "Lovelace", "picture": "https://media.licdn.com/member.jpg", "locale": {"country": "US", "language": "en"}, "email": "not-persisted@example.com", "email_verified": True, }, ) assert request.headers["linkedin-version"] == LINKEDIN_API_VERSION assert request.headers["x-restli-protocol-version"] == "2.0.0" if request.url.path == "/rest/organizationAcls": assert parse_qs(request.url.query.decode()) == { "q": ["roleAssignee"], "role": ["ADMINISTRATOR"], "state": ["APPROVED"], "count": ["100"], "start": ["0"], } return httpx.Response( 200, json={ "elements": [ {"organization": "urn:li:organization:123456"}, { "organizationTarget": "urn:li:organization:789012" }, ], "paging": {"start": 0, "count": 2, "total": 2, "links": []}, }, ) organization_id = request.url.path.rsplit("/", 1)[-1] return httpx.Response( 200, json={ "id": int(organization_id), "localizedName": f"Organization {organization_id}", "vanityName": f"organization-{organization_id}", "logoV2": { "digitalmediaAsset": "urn:li:digitalmediaAsset:logo_asset", "original~": { "elements": [ { "identifiers": [ { "identifier": f"https://media.licdn.com/{organization_id}.png" } ] } ] }, }, "primaryOrganizationType": "NONE", "defaultLocale": {"country": "US", "language": "en"}, "localizedWebsite": "https://example.com", }, ) client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) provider = LinkedInProvider(linkedin_settings(tmp_path), http_client=client) try: token = await provider.exchange_code( code="authorization-code", redirect_uri=_REDIRECT_URI, ) accounts = await provider.discover_accounts( token, account_type="linkedin_organization" ) finally: await client.aclose() member, first, second = accounts assert member["external_account_id"] == "oidc-member-subject_123" assert member["account_type"] == "linkedin_member" assert member["connection_status"] == "connected" assert member["metadata"]["email_verified"] is True assert "email" not in member["metadata"] assert [first["external_account_id"], second["external_account_id"]] == [ "123456", "789012", ] assert first["account_type"] == "linkedin_organization" assert first["connection_status"] == "pending" assert first["avatar_url"] == "https://media.licdn.com/123456.png" assert first["metadata"]["parent_member_id"] == "oidc-member-subject_123" assert first["metadata"]["logo_asset"] == ( "urn:li:digitalmediaAsset:logo_asset" ) assert calls == [ "/oauth/v2/accessToken", "/v2/userinfo", "/rest/organizationAcls", "/rest/organizations/123456", "/rest/organizations/789012", ] async def test_linkedin_organization_scope_is_explicit_and_bound_to_state( tmp_path: Path, ) -> None: container = build_container(linkedin_settings(tmp_path)) await container.social.initialize() try: member = await container.social.oauth.connect( provider="linkedin", workspace_id="workspace-a", user_id="user-a", payload=SocialAccountConnectRequest(), ) member_query = parse_qs(urlparse(member.authorization_url or "").query) assert member_query["scope"] == ["openid profile"] assert "code_challenge" not in member_query organization = await container.social.oauth.connect( provider="linkedin", workspace_id="workspace-a", user_id="user-a", payload=SocialAccountConnectRequest( account_type="linkedin_organization" ), ) organization_query = parse_qs( urlparse(organization.authorization_url or "").query ) assert organization_query["scope"] == [ "openid profile rw_organization_admin" ] assert "w_organization_social" not in organization_query["scope"][0] state = await container.social.oauth.states.consume( state=organization_query["state"][0], provider="linkedin" ) assert state.workspace_id == "workspace-a" assert state.user_id == "user-a" assert state.requested_account_type == "linkedin_organization" assert state.requested_scopes == [ "openid", "profile", "rw_organization_admin", ] with pytest.raises(SocialCapabilityUnsupportedError): await container.social.oauth.connect( provider="linkedin", workspace_id="workspace-a", user_id="user-a", payload=SocialAccountConnectRequest(account_type="organization"), ) finally: await container.social.close() await container.security_database.close() async def test_linkedin_callback_is_duplicate_safe_selectable_and_workspace_bound( tmp_path: Path, ) -> None: container = build_container(linkedin_settings(tmp_path)) await container.social.initialize() adapter = container.social.accounts.providers.get("linkedin") assert isinstance(adapter, LinkedInProvider) await adapter._client.aclose() async def handler(request: httpx.Request) -> httpx.Response: if request.url.path == "/oauth/v2/accessToken": return httpx.Response( 200, json={ "access_token": "linkedin-token-that-must-remain-encrypted", "expires_in": 3600, "token_type": "Bearer", }, ) if request.url.path == "/v2/userinfo": return httpx.Response( 200, json={"sub": "stable-member-sub", "name": "Workspace Member"}, ) if request.url.path == "/rest/organizationAcls": return httpx.Response( 200, json={ "elements": [ {"organization": "urn:li:organization:123456"} ], "paging": {"start": 0, "count": 1, "total": 1}, }, ) return httpx.Response( 200, json={ "id": 123456, "localizedName": "Workspace Organization", "vanityName": "workspace-organization", }, ) adapter._client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) adapter._owns_client = True try: connect = await container.social.oauth.connect( provider="linkedin", workspace_id="workspace-a", user_id="user-a", payload=SocialAccountConnectRequest( account_type="linkedin_organization" ), ) state = parse_qs(urlparse(connect.authorization_url or "").query)["state"][0] member = await container.social.oauth.callback( provider="linkedin", state=state, code="first-code" ) assert member.account_type == "linkedin_member" assert member.status.value == "connected" with pytest.raises(SocialOAuthStateError): await container.social.oauth.callback( provider="linkedin", state=state, code="replayed-code" ) accounts = await container.social.accounts.list("workspace-a") assert len(accounts) == 2 organization = next( item for item in accounts if item.account_type == "linkedin_organization" ) assert organization.status.value == "pending" assert "linkedin-token-that-must-remain-encrypted" not in ( organization.model_dump_json() ) with pytest.raises(SocialAccountNotFoundError): await container.social.accounts.select_discovered( "workspace-b", [organization.id] ) selected = await container.social.accounts.select_discovered( "workspace-a", [organization.id, organization.id] ) assert len(selected) == 1 assert selected[0].status.value == "connected" second_connect = await container.social.oauth.connect( provider="linkedin", workspace_id="workspace-a", user_id="user-a", payload=SocialAccountConnectRequest( account_type="linkedin_organization" ), ) second_state = parse_qs( urlparse(second_connect.authorization_url or "").query )["state"][0] await container.social.oauth.callback( provider="linkedin", state=second_state, code="second-code" ) duplicate_safe = await container.social.accounts.list("workspace-a") assert len(duplicate_safe) == 2 assert next( item for item in duplicate_safe if item.account_type == "linkedin_organization" ).status.value == "connected" async with container.social.database.session("workspace-a") as session: token_count = await session.scalar(select(func.count(SocialAccountToken.id))) encrypted_payloads = list( ( await session.scalars( select(SocialAccountToken.encrypted_payload) ) ).all() ) assert token_count == 2 assert all(encrypted_payloads) assert all( "linkedin-token-that-must-remain-encrypted" not in str(payload) for payload in encrypted_payloads ) finally: await container.social.close() await container.security_database.close() async def test_linkedin_state_redirect_expiry_and_provider_binding( tmp_path: Path, ) -> None: container = build_container(linkedin_settings(tmp_path)) await container.social.initialize() try: assert container.social.oauth._redirect_uri("linkedin", None) == _REDIRECT_URI with pytest.raises(SocialPermissionDeniedError): container.social.oauth._redirect_uri( "linkedin", "https://attacker.example/v1/social/accounts/linkedin/callback", ) state = await container.social.oauth.states.create( provider="linkedin", workspace_id="workspace-a", user_id="user-a", redirect_uri=_REDIRECT_URI, requested_account_type="linkedin_member", requested_scopes=["openid", "profile"], ) with pytest.raises(SocialOAuthStateError): await container.social.oauth.states.consume( state=state.state, provider="x" ) consumed = await container.social.oauth.states.consume( state=state.state, provider="linkedin" ) assert consumed.workspace_id == "workspace-a" expired = OAuthState( state="expired-linkedin-state-value-that-is-long-enough", provider="linkedin", workspace_id="workspace-a", user_id="user-a", redirect_uri=_REDIRECT_URI, requested_account_type="linkedin_member", requested_scopes=["openid", "profile"], 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="linkedin" ) finally: await container.social.close() await container.security_database.close() async def test_linkedin_invalid_code_and_refresh_rules_do_not_leak_secrets( tmp_path: Path, ) -> None: secret_code = "linkedin-code-that-must-not-leak" async def handler(_: httpx.Request) -> httpx.Response: return httpx.Response( 400, json={ "error": "invalid_grant", "error_description": f"bad code {secret_code}", }, ) client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) provider = LinkedInProvider(linkedin_settings(tmp_path), http_client=client) try: with pytest.raises(SocialPermissionDeniedError) as raised: await provider.exchange_code( code=secret_code, redirect_uri=_REDIRECT_URI ) with pytest.raises(SocialReauthRequiredError): await provider.refresh_token({"access_token": "expired"}) finally: await client.aclose() assert secret_code not in str(raised.value) assert "linkedin-client-secret" not in str(raised.value) async def test_linkedin_refresh_is_used_only_when_provider_issued_it( tmp_path: Path, ) -> None: async def handler(request: httpx.Request) -> httpx.Response: form = parse_qs(request.content.decode()) assert form == { "grant_type": ["refresh_token"], "refresh_token": ["partner-refresh-token"], "client_id": ["linkedin-client-id"], "client_secret": ["linkedin-client-secret"], } return httpx.Response( 200, json={ "access_token": "refreshed-access-token", "expires_in": 5184000, "token_type": "Bearer", }, ) client = httpx.AsyncClient(transport=httpx.MockTransport(handler)) provider = LinkedInProvider(linkedin_settings(tmp_path), http_client=client) try: refreshed = await provider.refresh_token( { "access_token": "expired-access-token", "refresh_token": "partner-refresh-token", } ) finally: await client.aclose() assert refreshed["access_token"] == "refreshed-access-token" assert refreshed["refresh_token"] == "partner-refresh-token" async def test_linkedin_capabilities_are_discovery_only(tmp_path: Path) -> None: container = build_container(linkedin_settings(tmp_path)) try: linkedin = container.social.accounts.get_provider("linkedin") assert linkedin.available assert linkedin.configured assert linkedin.capabilities.implementation_status == "implemented" assert linkedin.capabilities.account_types == [ "linkedin_member", "linkedin_organization", ] assert linkedin.capabilities.required_scopes == ["openid", "profile"] assert linkedin.capabilities.optional_scopes == ["rw_organization_admin"] assert not linkedin.capabilities.video assert not linkedin.capabilities.image assert not linkedin.capabilities.direct_publish assert not linkedin.capabilities.draft_upload assert not linkedin.capabilities.scheduled_publish assert not linkedin.capabilities.analytics assert not linkedin.capabilities.delete_post assert not linkedin.capabilities.personal_publishing assert not linkedin.capabilities.organization_publishing assert linkedin.capabilities.publish_metadata_schema == {} finally: await container.social.close() await container.security_database.close() def test_linkedin_redirect_configuration_is_fail_closed() -> None: invalid = [ "https://attacker.example/not-the-linkedin-callback", "http://api.example.com/v1/social/accounts/linkedin/callback", "https://api.example.com/v1/social/accounts/linkedin/callback?next=bad", "ftp://localhost/v1/social/accounts/linkedin/callback", ] for redirect in invalid: with pytest.raises(ValidationError): Settings(_env_file=None, linkedin_redirect_uri=redirect) local = Settings( _env_file=None, linkedin_redirect_uri=( "http://localhost/v1/social/accounts/linkedin/callback" ), ) assert local.linkedin_redirect_uri.startswith("http://localhost/")