"""WordPress connect codes: opaque, single-use, credential-free, return-URL bound (EDGE-11, C-3).""" from __future__ import annotations import asyncio import base64 import json from unittest.mock import AsyncMock, MagicMock from uuid import uuid4 import pytest from src.core.exceptions import ForbiddenError, ValidationError from src.domain.external_admin.models import ( INSTALLATION_STATUS_CONNECTED, INSTALLATION_STATUS_PENDING, ExternalAdminInstallation, ) from src.domain.external_admin.schemas import ( WordPressConnectCompleteRequest, WordPressConnectStartRequest, ) from src.domain.external_admin.service import ExternalAdminService from src.domain.external_admin.wordpress_connect import ( CODE_PREFIX, issue_wordpress_connect_code, redeem_wordpress_connect_code, validate_wordpress_return_url, ) SITE = "https://shop.example.com" INSTALLATION_ID = str(uuid4()) class FakeRedis: """Just enough of redis.asyncio: SET NX EX, and MULTI/EXEC pipelines.""" def __init__(self): self.data: dict[str, str] = {} self.ttls: dict[str, int] = {} async def set(self, key, value, ex=None, nx=False): if nx and key in self.data: return None self.data[key] = value if ex is not None: self.ttls[key] = ex return True def pipeline(self, transaction=True): return _Pipeline(self) class _Pipeline: def __init__(self, redis): self._redis = redis self._ops = [] async def __aenter__(self): return self async def __aexit__(self, *exc): return False def get(self, key): self._ops.append(("get", key)) def delete(self, key): self._ops.append(("delete", key)) async def execute(self): # MULTI/EXEC: no other coroutine runs between the queued commands. out = [] for op, key in self._ops: if op == "get": out.append(self._redis.data.get(key)) else: out.append(1 if self._redis.data.pop(key, None) is not None else 0) return out # ── the code itself ────────────────────────────────────────────────────────── async def test_code_is_opaque_and_carries_nothing(): redis = FakeRedis() code = await issue_wordpress_connect_code(redis, installation_id=INSTALLATION_ID, site_url=SITE) assert code.startswith(CODE_PREFIX) assert INSTALLATION_ID not in code assert "shop.example.com" not in code # Nothing in it is decodable JSON (the old format was base64(JSON).signature). body = code[len(CODE_PREFIX) :] padded = body + "=" * (-len(body) % 4) with pytest.raises(Exception): # noqa: B017 - any decode/parse failure is the point json.loads(base64.urlsafe_b64decode(padded)) async def test_the_server_stores_a_hash_and_no_credential(): redis = FakeRedis() code = await issue_wordpress_connect_code(redis, installation_id=INSTALLATION_ID, site_url=SITE) (key,) = redis.data assert code not in key # only a hash of the code is used as the key stored = json.loads(redis.data[key]) assert set(stored) == {"installation_id", "site_url"} # no token, no URLs to follow assert redis.ttls[key] == 600 async def test_codes_are_unique(): redis = FakeRedis() codes = { await issue_wordpress_connect_code(redis, installation_id=INSTALLATION_ID, site_url=SITE) for _ in range(200) } assert len(codes) == 200 # ── redemption ─────────────────────────────────────────────────────────────── async def test_redeem_returns_the_installation_once(): redis = FakeRedis() code = await issue_wordpress_connect_code(redis, installation_id=INSTALLATION_ID, site_url=SITE) assert await redeem_wordpress_connect_code(redis, code, SITE) == { "installation_id": INSTALLATION_ID } with pytest.raises(ForbiddenError): # replaying a leaked URL gets nothing await redeem_wordpress_connect_code(redis, code, SITE) async def test_concurrent_redemptions_cannot_both_succeed(): redis = FakeRedis() code = await issue_wordpress_connect_code(redis, installation_id=INSTALLATION_ID, site_url=SITE) results = await asyncio.gather( *(redeem_wordpress_connect_code(redis, code, SITE) for _ in range(8)), return_exceptions=True, ) assert sum(1 for r in results if isinstance(r, dict)) == 1 assert sum(1 for r in results if isinstance(r, ForbiddenError)) == 7 async def test_a_wrong_site_is_refused_and_burns_the_code(): redis = FakeRedis() code = await issue_wordpress_connect_code(redis, installation_id=INSTALLATION_ID, site_url=SITE) with pytest.raises(ForbiddenError, match="connect_code_site_mismatch"): await redeem_wordpress_connect_code(redis, code, "https://attacker.example") with pytest.raises(ForbiddenError): # the legitimate site cannot reuse it either await redeem_wordpress_connect_code(redis, code, SITE) @pytest.mark.parametrize( "presented", [ "https://SHOP.example.com", "https://shop.example.com/", "https://shop.example.com/blog/wp-admin", ], ) async def test_site_matching_ignores_case_trailing_slash_and_path(presented): redis = FakeRedis() code = await issue_wordpress_connect_code(redis, installation_id=INSTALLATION_ID, site_url=SITE) assert (await redeem_wordpress_connect_code(redis, code, presented))[ "installation_id" ] == INSTALLATION_ID async def test_unknown_used_and_expired_codes_look_identical(): redis = FakeRedis() code = await issue_wordpress_connect_code(redis, installation_id=INSTALLATION_ID, site_url=SITE) await redeem_wordpress_connect_code(redis, code, SITE) # used expired = await issue_wordpress_connect_code( redis, installation_id=INSTALLATION_ID, site_url=SITE ) redis.data.clear() # what Redis does when the TTL elapses messages = set() for candidate in (code, expired, CODE_PREFIX + "never-issued-0123456789abcdef"): with pytest.raises(ForbiddenError) as exc: await redeem_wordpress_connect_code(redis, candidate, SITE) messages.add(exc.value.message) assert messages == {"invalid_connect_code"} @pytest.mark.parametrize( "bad", [ "", " ", "eyJhIjoxfQ.deadbeef", # the legacy signed format is no longer accepted "not-a-code", "p2c0." + "a" * 40, CODE_PREFIX + "a" * 500, ], ) async def test_malformed_and_legacy_codes_are_rejected(bad): with pytest.raises(ValidationError): await redeem_wordpress_connect_code(FakeRedis(), bad, SITE) async def test_a_corrupted_stored_value_is_refused(): redis = FakeRedis() code = await issue_wordpress_connect_code(redis, installation_id=INSTALLATION_ID, site_url=SITE) (key,) = redis.data redis.data[key] = "{not json" with pytest.raises(ForbiddenError): await redeem_wordpress_connect_code(redis, code, SITE) # ── return URL ─────────────────────────────────────────────────────────────── @pytest.mark.parametrize( "return_url", [ "https://shop.example.com/wp-admin/admin.php?page=p2hs-ai&p2hs_state=abc", "https://SHOP.example.com/wp-admin/admin.php?page=p2hs-ai", "https://shop.example.com/blog/wp-admin/admin.php?page=p2hs-ai", ], ) def test_return_urls_on_the_site_wp_admin_are_accepted(return_url): assert validate_wordpress_return_url(SITE, return_url) == return_url @pytest.mark.parametrize( "return_url", [ "", "https://attacker.example/wp-admin/admin.php", "https://shop.example.com.attacker.example/wp-admin/admin.php", "https://attacker.example/?u=https://shop.example.com/wp-admin/", "https://shop.example.com@attacker.example/wp-admin/admin.php", "https://user:pw@shop.example.com/wp-admin/admin.php", "http://shop.example.com/wp-admin/admin.php", # scheme downgrade "https://shop.example.com:8443/wp-admin/admin.php", # different port = different origin "https://shop.example.com/not-admin/page", "https://shop.example.com/wp-admin-evil/page", "https://shop.example.com/wp-admin/admin.php#frag", "//shop.example.com/wp-admin/admin.php", "javascript:alert(1)", "wp-admin/admin.php", ], ) def test_return_urls_anywhere_else_are_rejected(return_url): with pytest.raises(ValidationError, match="invalid_return_url"): validate_wordpress_return_url(SITE, return_url) # ── service ────────────────────────────────────────────────────────────────── def _installation(**over) -> ExternalAdminInstallation: inst = ExternalAdminInstallation( id=uuid4(), project_id=uuid4(), workspace_id=uuid4(), installation_type="wordpress", status=INSTALLATION_STATUS_PENDING, metadata_json={}, ) for key, value in over.items(): setattr(inst, key, value) return inst def _service(installation=None, project=None): installs = MagicMock() installs.get_by_id = AsyncMock(return_value=installation) installs.get_by_project = AsyncMock(return_value=installation) installs.create = AsyncMock() installs.save = AsyncMock() projects = MagicMock() projects.get_by_id = AsyncMock(return_value=project) others = [MagicMock() for _ in range(9)] svc = ExternalAdminService( installs, others[0], others[1], others[2], others[3], projects, *others[4:] ) svc._encrypt = lambda value: f"enc({value})" svc._wordpress_embed_page_url = lambda *a, **k: "https://dash.example/embed" return svc, installs async def test_complete_mints_the_credential_at_redemption_and_returns_it_in_the_body_only(): inst = _installation(id=uuid4()) svc, installs = _service(inst) redis = FakeRedis() code = await issue_wordpress_connect_code(redis, installation_id=str(inst.id), site_url=SITE) out = await svc.handle_wordpress_connect_complete( WordPressConnectCompleteRequest(connect_code=code, site_url=SITE), redis ) assert out.ok and out.installation_id == str(inst.id) assert len(out.credential_token) >= 40 assert out.credential_token not in code # the code never contained it assert inst.installation_token_enc == f"enc({out.credential_token})" assert inst.status == INSTALLATION_STATUS_CONNECTED installs.save.assert_awaited() async def test_complete_cannot_be_repeated_with_the_same_code(): inst = _installation() svc, installs = _service(inst) redis = FakeRedis() code = await issue_wordpress_connect_code(redis, installation_id=str(inst.id), site_url=SITE) body = WordPressConnectCompleteRequest(connect_code=code, site_url=SITE) await svc.handle_wordpress_connect_complete(body, redis) installs.save.reset_mock() with pytest.raises(ForbiddenError): await svc.handle_wordpress_connect_complete(body, redis) installs.save.assert_not_awaited() # the second attempt changed nothing async def test_each_redemption_gets_a_different_credential(): inst = _installation() svc, _ = _service(inst) redis = FakeRedis() tokens = set() for _ in range(3): code = await issue_wordpress_connect_code( redis, installation_id=str(inst.id), site_url=SITE ) out = await svc.handle_wordpress_connect_complete( WordPressConnectCompleteRequest(connect_code=code, site_url=SITE), redis ) tokens.add(out.credential_token) assert len(tokens) == 3 async def test_a_code_for_a_shopify_installation_is_refused(): inst = _installation(installation_type="shopify") svc, _ = _service(inst) redis = FakeRedis() code = await issue_wordpress_connect_code(redis, installation_id=str(inst.id), site_url=SITE) with pytest.raises(ForbiddenError, match="installation_type_mismatch"): await svc.handle_wordpress_connect_complete( WordPressConnectCompleteRequest(connect_code=code, site_url=SITE), redis ) async def test_issuing_returns_a_credential_free_response(): inst = _installation() project = MagicMock(workspace_id=uuid4(), type="ai_assist_wordpress") svc, _ = _service(inst, project) redis = FakeRedis() out = await svc.issue_wordpress_connect_code(uuid4(), project.workspace_id, SITE, redis) assert out.connect_code.startswith(CODE_PREFIX) assert out.core_direct_api_url == "" # no direct-to-Core address is handed out any more assert "token" not in out.model_dump_json().replace("connect_code", "") assert len(redis.data) == 1 async def test_issuing_refuses_a_foreign_return_url_before_writing_anything(): inst = _installation() project = MagicMock(workspace_id=uuid4(), type="ai_assist_wordpress") svc, _ = _service(inst, project) redis = FakeRedis() with pytest.raises(ValidationError, match="invalid_return_url"): await svc.issue_wordpress_connect_code( uuid4(), project.workspace_id, SITE, redis, return_url="https://attacker.example/wp-admin/x", ) assert redis.data == {} async def test_connect_start_refuses_to_build_a_link_to_a_foreign_return_address(): svc, _ = _service() ok = await svc.build_wordpress_connect_start_url( WordPressConnectStartRequest( site_url=SITE, return_url=f"{SITE}/wp-admin/admin.php?page=p2hs-ai" ) ) assert "return_url=" in ok.url with pytest.raises(ValidationError, match="invalid_return_url"): await svc.build_wordpress_connect_start_url( WordPressConnectStartRequest( site_url=SITE, return_url="https://attacker.example/wp-admin/x" ) )