"""Local-only passing CRM authorization proof, extended from the 11E2 preflight. Runs real PackInstallService + CloudControlClient + local gateway records routes. Only Core pack lookup/install-state persistence and schema-snapshot persistence are mocked. Seed 132 and current 134/135 stamping supply the installed payload. Candidate policy is confined to disposable projects, never the product catalog. Run from Core: python scripts/probe_crm_admission_11e2.py """ from __future__ import annotations import asyncio import base64 import copy 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 src.domain.feature_packs.crm_authorization import staff_policy from src.domain.feature_packs.normalize import stamp_composition 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, _load_migration BASE = "http://127.0.0.1:8801" # Intentionally no remote override. POLICY = staff_policy() 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": "11e2"}).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, slug="lead-lifecycle"): pack = SimpleNamespace(id=uuid4(), slug=slug, status="active", version="1.1.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, legacy_project, fresh_project = 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() raw = _captured_composition(str(ROOT / "alembic/versions/132_lead_lifecycle_pack.py"), "lead-lifecycle") composition = stamp_composition("lead-lifecycle", raw, access_policy={"minimum_plan": "free"}) candidate = _load_migration(str(ROOT / "alembic/versions/143_lead_staff_authorization.py")).upgrade_composition(composition) 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): response = await req("PUT", "/v1/cloud/control/" + path, body) response.raise_for_status() await control(f"workspaces/{workspace}", { "workspace_id": str(workspace), "name": "11E2 disposable authorization probe", "status": "active", "cloud_enabled": True, "suspended": False, }) for project in (legacy_project, fresh_project): await control(f"projects/{project}", { "project_id": str(project), "workspace_id": str(workspace), "name": "11E2 probe", "status": "active", "cloud_enabled": True, "suspended": False, "allowed_request_types": [], "origins": [], }) await control(f"projects/{project}/auth-keys", { "project_id": str(project), "key_id": "11e2", "public_key_jwk": _rsa_public_jwk(public_pem, key_id="11e2"), "private_key_jwk": _rsa_private_jwk(private_pem, key_id="11e2"), }) async def exercise(project, prefix): path = f"/v1/projects/{project}/entities/lead/records" member = jwt(key, project, workspace, "member") admin = jwt(key, project, workspace, "admin") response = await req("POST", path, {"data": { "name": "Synthetic lead", "email": "probe@example.invalid", "status": "new", }}, admin) check(prefix + ": admin create", response.status_code, 201) response.raise_for_status() record = response.json()["record"]["id"] response = await req("POST", path, {"data": { "name": "Member lead", "email": "member@example.invalid", "status": "new", }}, member) check(prefix + ": member create denied", response.status_code, 403) response = await req("GET", path, bearer=member) check(prefix + ": member list denied", response.status_code, 403) response = await req("GET", f"{path}/{record}", bearer=member) check(prefix + ": member reads admin lead denied", response.status_code, 403) response = await req("PATCH", f"{path}/{record}", {"data": {"status": "contacted"}}, member) check(prefix + ": member pipeline transition denied", response.status_code, 403) response = await req("DELETE", f"{path}/{record}", bearer=member) check(prefix + ": member deletes admin lead denied", response.status_code, 403) if prefix == "candidate fresh": response = await req("GET", path, bearer=admin) check(prefix + ": admin list", response.status_code, 200) response = await req("GET", f"{path}/{record}", bearer=admin) check(prefix + ": admin read", response.status_code, 200) response = await req("PATCH", f"{path}/{record}", {"data": {"status": "contacted"}}, admin) check(prefix + ": admin transition", response.status_code, 200) response = await req("DELETE", f"{path}/{record}", bearer=admin) check(prefix + ": admin delete", response.status_code, 200) super_admin = jwt(key, project, workspace, "super_admin") response = await req("POST", path, {"data": { "name": "Super-admin lead", "email": "super@example.invalid", "status": "new", }}, super_admin) check(prefix + ": super-admin create", response.status_code, 201) response.raise_for_status() super_record = response.json()["record"]["id"] for method, route, body in ( ("GET", path, None), ("GET", f"{path}/{super_record}", None), ("PATCH", f"{path}/{super_record}", {"data": {"status": "contacted"}}), ("DELETE", f"{path}/{super_record}", None), ): response = await req(method, route, body, super_admin) check(prefix + f": super-admin {method} {route.rsplit('/', 1)[-1] == 'records'}", response.status_code, 200) historical = copy.deepcopy(composition["entities"][0]) historical["schema"]["fields"].append({"key": "custom", "type": "text", "indexed": True}) historical["schema"]["custom_metadata"] = {"retain": [1, 2]} await cloud.create_project_entity(str(legacy_project), historical) first = await install(cloud, legacy_project, candidate) check("historical upgrade installs", first["status"], "installed") live = await cloud.get_project_entity(str(legacy_project), "lead") after = json.loads(live["entity"]["schema_json"]) after.pop("access_policy") check("historical custom fields and metadata preserved", after, historical["schema"]) initial_version = live["entity"]["version"] await exercise(legacy_project, "historical upgraded") second = await install(cloud, legacy_project, candidate) check("candidate reinstall reports installed", second["status"], "installed") live = await cloud.get_project_entity(str(legacy_project), "lead") check("canonical reinstall is no-op", live["entity"]["version"], initial_version) await exercise(legacy_project, "candidate reinstall") third = await install(cloud, fresh_project, candidate) check("candidate fresh install", third["status"], "installed") await exercise(fresh_project, "candidate fresh") response = await req("GET", f"/v1/projects/{fresh_project}/entities/lead/records", bearer=jwt(key, legacy_project, workspace, "admin")) check("cross-project JWT denied", response.status_code, 401) # Test conflicting and malformed policies on the disposable entity only. for custom in ({"version": 1, "roles": {"member": {"read": {"effect": "allow", "scope": "all"}}}}, None): changed = copy.deepcopy(candidate["entities"][0]["schema"]) changed["access_policy"] = custom await cloud.patch_project_entity(str(fresh_project), "lead", {"schema": changed}) before = (await cloud.get_project_entity(str(fresh_project), "lead"))["entity"] for retry in range(2): result = await install(cloud, fresh_project, candidate) check(f"custom={custom is not None} retry={retry} blocked", result["status"], "blocked") after = (await cloud.get_project_entity(str(fresh_project), "lead"))["entity"] check(f"custom={custom is not None} left untouched", after, before) # A stale conditional write must not overwrite a policy installed after GET. before = await cloud.get_project_entity(str(fresh_project), "lead") await cloud.patch_project_entity(str(fresh_project), "lead", {"schema": candidate["entities"][0]["schema"]}) response = await http.patch(f"/v1/projects/{fresh_project}/entities/lead", headers={"Authorization": f"Bearer {token}", "If-Match": f'"{before["entity"]["version"]}"'}, json={"schema": changed}) check("stale entity version rejects concurrent overwrite", response.status_code, 409) live = (await cloud.get_project_entity(str(fresh_project), "lead"))["entity"] check("concurrent policy retained", json.loads(live["schema_json"])["access_policy"], POLICY) customer = _captured_composition(str(ROOT / "alembic/versions/144_customer_records_pack.py"), "customer-records") result = await install(cloud, fresh_project, customer, "customer-records") check("customer installs with verified staff policy", result["status"], "installed") customer_path = f"/v1/projects/{fresh_project}/entities/customer/records" admin = jwt(key, fresh_project, workspace, "admin") member = jwt(key, fresh_project, workspace, "member") response = await req("POST", customer_path, {"data": {"name": "Customer", "status": "active"}}, admin) check("customer admin create", response.status_code, 201) response.raise_for_status() customer_id = response.json()["record"]["id"] for method, path, body in ( ("POST", customer_path, {"data": {"name": "Denied", "status": "active"}}), ("GET", customer_path, None), ("GET", f"{customer_path}/{customer_id}", None), ("PATCH", f"{customer_path}/{customer_id}", {"data": {"status": "inactive"}}), ("DELETE", f"{customer_path}/{customer_id}", None), ): response = await req(method, path, body, member) check(f"customer member {method} denied", response.status_code, 403) if "--generated" in sys.argv: # Real auth/me lookup in the generated BFF needs real project users. await control(f"projects/{fresh_project}/auth-config", { "project_id": str(fresh_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/{fresh_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-crm-generated-11e3.mjs"), stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) output, error = await child.communicate(json.dumps({ "gateway": BASE, "project": str(fresh_project), "admin": jwt(key, fresh_project, workspace, "admin", users["admin"]), "member": jwt(key, fresh_project, workspace, "member", users["member"]), "other_project": jwt(key, legacy_project, 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": "11E2-authorization", "gateway": BASE, "fixture_projects": [str(legacy_project), str(fresh_project)], "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; no production catalog or existing project changed.", }, indent=2)) return 1 if failed else 0 if __name__ == "__main__": raise SystemExit(asyncio.run(main()))