"""Local Booking admission proof: Cloud store + staff ports BFF. Run from Core: python scripts/probe_booking_admission_11f2.py [--generated] """ from __future__ import annotations import asyncio import base64 from datetime import datetime, timedelta, timezone import json from pathlib import Path import sys import time import tomllib 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.integrations.cloud_control_client import _rsa_private_jwk, _rsa_public_jwk 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": "11f2"}).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()))}" def weekday_slot(): now = datetime.now(timezone.utc).replace(minute=0, second=0, microsecond=0) start = now + timedelta(days=1) while start.weekday() >= 5: start += timedelta(days=1) start = start.replace(hour=10) later = start + timedelta(minutes=30) return start.isoformat().replace("+00:00", "Z"), later.isoformat().replace("+00:00", "Z") async def main(): config = tomllib.loads((ROOT.parent / "P2HSCloud/apps/pc-gateway/wrangler.toml").read_text()) token = config["vars"]["INTERNAL_CONTROL_TOKEN"] workspace, project, other = 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() observations = [] def check(name, actual, expected): observations.append({"check": name, "actual": actual, "expected": expected, "pass": actual == expected}) first_slot, second_slot = weekday_slot() 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": "11F2 booking probe", "status": "active", "cloud_enabled": True, "suspended": False, }) for item in (project, other): await control(f"projects/{item}", { "project_id": str(item), "workspace_id": str(workspace), "name": "11F2 probe", "status": "active", "cloud_enabled": True, "suspended": False, "allowed_request_types": ["booking.reserve", "booking.cancel"], "origins": [], }) await control(f"projects/{item}/auth-keys", { "project_id": str(item), "key_id": "11f2", "public_key_jwk": _rsa_public_jwk(public_pem, key_id="11f2"), "private_key_jwk": _rsa_private_jwk(private_pem, key_id="11f2"), }) resource_key = "consult" created = await req("POST", f"/v1/projects/{project}/bookings/resources", { "resource_key": resource_key, "name": "Consult", "timezone": "UTC", "duration_minutes": 30, "buffer_minutes": 0, }) check("upsert resource", created.status_code, 201) created.raise_for_status() availability = await req( "GET", f"/v1/projects/{project}/bookings/resources/{resource_key}/availability", ) check("availability executable", availability.status_code, 200) slots = availability.json().get("slots") or [] if slots: first_slot = slots[0]["starts_at"] second_slot = slots[1]["starts_at"] if len(slots) > 1 else first_slot check("availability has slots", bool(slots), True) reserved = await req("POST", f"/v1/projects/{project}/bookings/reserve", { "resource_key": resource_key, "starts_at": first_slot, "customer_name": "Pat", "customer_email": "pat@example.invalid", "idempotency_key": "idem-1", }) check("reserve", reserved.status_code, 201) reserved.raise_for_status() reservation_id = reserved.json()["reservation"]["id"] replay = await req("POST", f"/v1/projects/{project}/bookings/reserve", { "resource_key": resource_key, "starts_at": first_slot, "customer_name": "Pat", "customer_email": "pat@example.invalid", "idempotency_key": "idem-1", }) check("idempotent replay", replay.status_code, 201) check("idempotent same id", replay.json()["reservation"]["id"], reservation_id) overlap = await req("POST", f"/v1/projects/{project}/bookings/reserve", { "resource_key": resource_key, "starts_at": first_slot, "customer_name": "Other", "customer_email": "other@example.invalid", "idempotency_key": "idem-2", }) check("overlap conflict", overlap.status_code, 409) held = await req("POST", f"/v1/projects/{project}/bookings/reserve", { "resource_key": resource_key, "starts_at": second_slot, "confirm": False, "hold_minutes": 0, "customer_name": "Hold", "customer_email": "hold@example.invalid", }) check("hold created", held.status_code, 201) expired = await req( "GET", f"/v1/projects/{project}/bookings/resources/{resource_key}/availability", ) check("hold expiry request accepted", expired.status_code, 200) cancelled = await req("POST", f"/v1/projects/{project}/bookings/{reservation_id}/cancel", {}) check("cancel", cancelled.status_code, 200) if held.status_code == 201: hold_id = held.json()["reservation"]["id"] released = await req("POST", f"/v1/projects/{project}/bookings/{hold_id}/cancel", {}) check("release hold", released.status_code, 200) app_jwt = jwt(key, project, workspace, "admin") denied = await req("POST", f"/v1/projects/{project}/bookings/reserve", { "resource_key": resource_key, "starts_at": second_slot, }, app_jwt) check("app JWT cannot call booking store directly", denied.status_code, 401) 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-booking-generated-11f3.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), "service_token": token, "resource_key": resource_key, "starts_at": second_slot, "member_starts_at": second_slot, "admin": jwt(key, project, workspace, "admin", users["admin"]), "member": jwt(key, project, workspace, "member", users["member"]), }).encode()) print(output.decode()) if child.returncode: print(error.decode()) check("generated host to booking BFF to gateway proof", child.returncode, 0) failed = sum(not item["pass"] for item in observations) print(json.dumps({ "phase": "11F2-booking", "gateway": BASE, "failed": failed, "observations": observations, "note": "Synthetic local fixtures only; Cloud D1 booking store is canonical.", }, indent=2)) return 1 if failed else 0 if __name__ == "__main__": raise SystemExit(asyncio.run(main()))