"""Local-only product-catalog admission proof. Installs stamped product-catalog onto a disposable Cloud project, then exercises staff/member records HTTP. Optional --generated runs HTML hosts through the existing records BFF. No checkout pack is installed. Run from Core: python scripts/probe_catalog_admission_11g2.py [--generated] """ from __future__ import annotations import asyncio import base64 import json from pathlib import Path import sys import time import tomllib from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock, patch from uuid import uuid4 import httpx from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric import padding, rsa ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT)) from copy import deepcopy from src.domain.feature_packs.crm_authorization import classify_policy, staff_policy from src.domain.feature_packs.service import PackInstallService from src.integrations.cloud_control_client import ( CloudControlClient, _rsa_private_jwk, _rsa_public_jwk, ) from tests.test_phase11_pack_migration_materialization import _captured_composition BASE = "http://127.0.0.1:8801" def b64(value: bytes) -> str: return base64.urlsafe_b64encode(value).rstrip(b"=").decode() def jwt(key, project, workspace, role, sub=None): now = int(time.time()) header = b64(json.dumps({"alg": "RS256", "typ": "JWT", "kid": "11g2"}).encode()) body = b64(json.dumps({ "sub": sub or f"probe-{role}", "project_id": str(project), "workspace_id": str(workspace), "role": role, "email": f"{role}@example.invalid", "iat": now, "exp": now + 600, "app_metadata": {}, "user_metadata": {}, }).encode()) message = f"{header}.{body}" return f"{message}.{b64(key.sign(message.encode(), padding.PKCS1v15(), hashes.SHA256()))}" async def install(cloud, project, composition): pack = SimpleNamespace( id=uuid4(), slug="product-catalog", status="active", version="1.0.0", composition=composition, input_schema=None, ) service = PackInstallService(AsyncMock(), MagicMock(), cloud) service._get_pack_by_slug = AsyncMock(return_value=pack) service._installed_config = AsyncMock(return_value={}) service._upsert_install_state = AsyncMock() with patch( "src.domain.projects.cloud_schema_catalog.ProjectCloudSchemaSnapshotRepository.upsert", new_callable=AsyncMock, ): return await service.install_pack(project_id=project, pack_slug=pack.slug, config={}) async def main(): config = tomllib.loads((ROOT.parent / "P2HSCloud/apps/pc-gateway/wrangler.toml").read_text()) token = config["vars"]["INTERNAL_CONTROL_TOKEN"] cloud = CloudControlClient(api_url=BASE, token=token) workspace, project, other, absent_project, conflict_project = ( uuid4(), uuid4(), uuid4(), uuid4(), uuid4(), ) key = rsa.generate_private_key(public_exponent=65537, key_size=2048) private_pem = key.private_bytes( serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, serialization.NoEncryption() ).decode() public_pem = key.public_key().public_bytes( serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo ).decode() composition = _captured_composition( str(ROOT / "alembic/versions/145_product_catalog_pack.py"), "product-catalog" ) observations = [] def check(name, actual, expected): observations.append({ "check": name, "actual": actual, "expected": expected, "pass": actual == expected, }) async with httpx.AsyncClient(base_url=BASE, timeout=15) as http: async def req(method, path, body=None, bearer=token): kwargs = {"headers": {"Authorization": f"Bearer {bearer}"}} if body is not None: kwargs["json"] = body return await http.request(method, path, **kwargs) async def control(path, body): (await req("PUT", "/v1/cloud/control/" + path, body)).raise_for_status() await control(f"workspaces/{workspace}", { "workspace_id": str(workspace), "name": "11G2 disposable catalog probe", "status": "active", "cloud_enabled": True, "suspended": False, }) for target in (project, other, absent_project, conflict_project): await control(f"projects/{target}", { "project_id": str(target), "workspace_id": str(workspace), "name": "11G2 probe", "status": "active", "cloud_enabled": True, "suspended": False, "allowed_request_types": [], "origins": [], }) await control(f"projects/{target}/auth-keys", { "project_id": str(target), "key_id": "11g2", "public_key_jwk": _rsa_public_jwk(public_pem, key_id="11g2"), "private_key_jwk": _rsa_private_jwk(private_pem, key_id="11g2"), }) result = await install(cloud, project, composition) check("catalog installs", result["status"], "installed") live = await cloud.get_project_entity(str(project), "catalog_item") schema = json.loads(live["entity"]["schema_json"]) keys = {field["key"] for field in schema["fields"]} check("entity_key catalog_item", live["entity"]["entity_key"], "catalog_item") check("no stripe product id on catalog_item", "stripe_product_id" in keys, False) missing = await req("GET", f"/v1/projects/{project}/entities/product") check("checkout product entity absent", missing.status_code, 404) path = f"/v1/projects/{project}/entities/catalog_item/records" admin = jwt(key, project, workspace, "admin") member = jwt(key, project, workspace, "member") response = await req("POST", path, { "data": {"name": "Fixture item", "sku": "SKU-1", "status": "active", "price_amount": 19}, }, admin) check("admin create", response.status_code, 201) response.raise_for_status() record_id = response.json()["record"]["id"] response = await req("GET", path, bearer=admin) check("admin list", response.status_code, 200) response = await req("PATCH", f"{path}/{record_id}", {"data": {"status": "inactive"}}, admin) check("admin update", response.status_code, 200) for method, route, body in ( ("POST", path, {"data": {"name": "Denied", "status": "active"}}), ("GET", path, None), ("GET", f"{path}/{record_id}", None), ("PATCH", f"{path}/{record_id}", {"data": {"status": "active"}}), ("DELETE", f"{path}/{record_id}", None), ): response = await req(method, route, body, member) check(f"member {method} denied", response.status_code, 403) response = await req( "GET", path, bearer=jwt(key, other, workspace, "admin") ) check("cross-project JWT denied", response.status_code, 401) live = await cloud.get_project_entity(str(project), "catalog_item") check( "fresh install live staff policy", classify_policy(json.loads(live["entity"]["schema_json"])), "canonical_policy", ) reinstall = await install(cloud, project, composition) check("canonical already_exists reinstall", reinstall["status"], "installed") after = await cloud.get_project_entity(str(project), "catalog_item") check( "canonical reinstall left policy", json.loads(after["entity"]["schema_json"])["access_policy"], staff_policy(), ) bare = deepcopy(composition["entities"][0]) bare["schema"] = dict(bare["schema"]) bare["schema"].pop("access_policy", None) await cloud.create_project_entity(str(absent_project), bare) absent_result = await install(cloud, absent_project, composition) check("already_exists without policy blocked", absent_result["status"], "blocked") absent_live = await cloud.get_project_entity(str(absent_project), "catalog_item") check( "absent policy not auto-upgraded", classify_policy(json.loads(absent_live["entity"]["schema_json"])), "historical_no_policy", ) conflicting = deepcopy(composition["entities"][0]) conflicting["schema"] = dict(conflicting["schema"]) conflicting["schema"]["access_policy"] = { "version": 1, "roles": {"member": {"read": {"effect": "allow", "scope": "all"}}}, } await cloud.create_project_entity(str(conflict_project), conflicting) before_conflict = (await cloud.get_project_entity(str(conflict_project), "catalog_item"))["entity"] conflict_result = await install(cloud, conflict_project, composition) check("conflicting policy blocked", conflict_result["status"], "blocked") after_conflict = (await cloud.get_project_entity(str(conflict_project), "catalog_item"))["entity"] check("conflicting policy left untouched", after_conflict, before_conflict) if "--generated" in sys.argv: await control(f"projects/{project}/auth-config", { "project_id": str(project), "workspace_id": str(workspace), "providers": ["email"], "jwt_expiry_seconds": 600, "refresh_token_expiry_seconds": 3600, "allow_signup": False, "require_email_verify": False, "google_oauth_mode": "platform", "roles": [{"name": role, "is_default": role == "member", "permissions": []} for role in ("admin", "super_admin", "member")], }) users = {} for role in ("admin", "member"): response = await req("POST", f"/v1/projects/{project}/auth/users", { "email": f"{role}@example.invalid", "role": role, "email_verified": True, }) response.raise_for_status() users[role] = response.json()["id"] child = await asyncio.create_subprocess_exec( "node", str(ROOT / "scripts/prove-catalog-generated-11g3.mjs"), stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) output, error = await child.communicate(json.dumps({ "gateway": BASE, "project": str(project), "admin": jwt(key, project, workspace, "admin", users["admin"]), "member": jwt(key, project, workspace, "member", users["member"]), "other_project": jwt(key, other, workspace, "admin"), }).encode()) print(output.decode()) if child.returncode: print(error.decode()) check("generated host to BFF to gateway proof", child.returncode, 0) failed = sum(not item["pass"] for item in observations) print(json.dumps({ "phase": "11G2-admission", "gateway": BASE, "fixture_projects": [str(project), str(other)], "core_database": "mocked lookup and install/snapshot persistence", "cloud_runtime": "real local gateway, D1, JWT, install and records HTTP", "observations": observations, "failed": failed, "note": "Synthetic local fixtures only; checkout pack was not installed.", }, indent=2)) return 1 if failed else 0 if __name__ == "__main__": raise SystemExit(asyncio.run(main()))