'
+ + html.escape(m["body"]) + "
" for m in messages)
+ return response(start, body, "text/html")
+ if path == "/state":
+ return response(start, {"messages": len(messages), "provisions": len(provisions),
+ "events": len(events),
+ "users": service.operability_snapshot().metrics["users"]})
+ if env["REQUEST_METHOD"] != "POST":
+ return response(start, {}, status="404 Not Found")
+ raw = env["wsgi.input"].read(int(env.get("CONTENT_LENGTH") or 0))
+ data = json.loads(raw or b"{}")
+ if path == "/events":
+ assert env.get("HTTP_AUTHORIZATION") == "Bearer synthetic-event-token"
+ assert "primary_email" not in data["data"]
+ events.append(data)
+ elif path == "/deliver":
+ service.deliver_outbox(operator, portal.outbox_delivery, worker_id="fixture-worker")
+ elif path == "/expire":
+ with ledger.lock, ledger.db:
+ ledger.db.execute("UPDATE verifications SET expires_at=? WHERE email=?",
+ ("2000-01-01T00:00:00+00:00", data["email"]))
+ elif path == "/suppress":
+ ledger.suppress(data["email"])
+ else:
+ return response(start, {}, status="404 Not Found")
+ return response(start, {"ok": True})
+
+ with server(fixtures, ports) as fixture_url, server(portal, ports) as portal_url:
+ policies = [CallerPolicy(
+ caller_id=name,
+ token=TOKEN if name == "user-engine" else "synthetic-other-caller-token",
+ source="user-engine", bindings=frozenset({(TENANT, CLIENT)}),
+ operations=frozenset({"send", "status", "delivery-status",
+ "verify", "consume", "cancel"}),
+ templates=frozenset({"family_member.invited", "family_invitation.resent"}),
+ rate_per_minute=120, burst=100, max_payload_bytes=32768,
+ ) for name in ("user-engine", "other-caller")]
+ receiver = ScopedTransactionalApplication(ledger, MailSink(), portal_url, policies)
+ with server(receiver, ports) as mail_url:
+ portal.registration_verification = HTTPRegistrationVerificationAdapter(
+ base_url=mail_url, bearer_token=TOKEN)
+ portal.outbox_delivery = HTTPOutboxDeliveryAdapter(
+ event_url=fixture_url + "/events", event_bearer_token="synthetic-event-token",
+ mail_url=mail_url + "/v1/send", mail_bearer_token=TOKEN,
+ mail_client_bindings=json.dumps([[TENANT, CLIENT]]))
+ result = subprocess.run(
+ [os.environ["OWNER_NODE"], os.environ["OWNER_BROWSER_SCRIPT"],
+ portal_url, fixture_url, mail_url, os.environ["OWNER_PLAYWRIGHT_MODULE"]],
+ capture_output=True, text=True, timeout=150)
+ assert result.returncode == 0, result.stdout + result.stderr
+ checks = json.loads(result.stdout)
+ assert len(provisions) == 1
+ assert provisions[0].preferred_username == "verified.person"
+ assert all(p.primary_email.endswith("@example.test") for p in provisions)
+ with ledger.lock:
+ assert ledger.db.execute("SELECT count(*) FROM verifications WHERE consumed_at IS NOT NULL"
+ ).fetchone()[0] == 1
+ assert ledger.db.execute("SELECT count(*) FROM verifications WHERE canceled_at IS NOT NULL"
+ ).fetchone()[0] == 1
+ ledger.db.close()
+ assert checks["passed"] and checks["browser_requests_outside_fixture"] == 0
+ Path(os.environ["OWNER_CHECKS_RESULT"]).write_text(json.dumps(checks))
diff --git a/tests/test_owner_communication_gate.py b/tests/test_owner_communication_gate.py
new file mode 100644
index 0000000..cff6151
--- /dev/null
+++ b/tests/test_owner_communication_gate.py
@@ -0,0 +1,79 @@
+"""The explicit owner-browser runner must isolate configuration and refuse weak reports."""
+
+import importlib.util
+import json
+import sys
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def runner():
+ spec = importlib.util.spec_from_file_location("owner_browser_gate", ROOT / "scripts"
+ / "check-owner-communication.py")
+ module = importlib.util.module_from_spec(spec)
+ spec.loader.exec_module(module)
+ return module
+
+
+@pytest.mark.parametrize("report_kind", ["passed", "skipped", "wrong_node"])
+def test_snapshot_strips_credentials_runtime_data_and_requires_exact_unskipped_node(
+ tmp_path, monkeypatch, report_kind):
+ module = runner()
+ owners = []
+ for package in ("user_engine", "email_connect"):
+ owner = tmp_path / package
+ source = owner / "src" / package
+ source.mkdir(parents=True)
+ (source / "__init__.py").write_text("# synthetic owner source\n")
+ (owner / ".env").write_text("SMTP_PASSWORD=synthetic-not-to-inherit")
+ (source / "runtime.db").write_bytes(b"synthetic-private-database")
+ owners.append(owner)
+ output = tmp_path / "receipt.json"
+ monkeypatch.setattr(sys, "argv", ["gate", "--user-engine", str(owners[0]),
+ "--email-connect", str(owners[1]),
+ "--receipt", str(output)])
+ monkeypatch.setenv("SMTP_PASSWORD", "synthetic-not-to-inherit")
+ monkeypatch.setenv("USER_ENGINE_FLEX_AUTH_TOKEN_FILE", "/synthetic/production-config")
+
+ def execute(argv, *, cwd, env, **kwargs):
+ assert "SMTP_PASSWORD" not in env and "USER_ENGINE_FLEX_AUTH_TOKEN_FILE" not in env
+ assert not list(cwd.rglob(".env")) and not list(cwd.rglob("*.db"))
+ assert argv[argv.index("-c") + 1] == str(cwd / "pytest.ini")
+ assert module.NODE in argv
+ node = module.NODE.split("::")[-1] if report_kind != "wrong_node" else "another_test"
+ skipped = "