#!/usr/bin/env python3 import json import os import sqlite3 import subprocess import sys import tempfile import time import urllib.error import urllib.request from pathlib import Path ROOT = Path(__file__).parent NODE_MAP = ROOT.parents[1] / "deployment" / "navigation-maps" / "AW-GZ-001.json" BASE = "http://127.0.0.1:48031" HEADERS = { "X-Lighthouse-Admin-Token": "test-token", "X-Guanghu-Principal-Id": "ICE-GL-ZY001:TEST-INSTANCE", "X-Guanghu-Subject-Kind": "persona", "X-Guanghu-Human-Anchor": "ICE-GL-INF", } def post(path, payload, headers=HEADERS): request = urllib.request.Request( BASE + path, data=json.dumps(payload).encode(), method="POST", headers=headers, ) return json.load(urllib.request.urlopen(request)) def expect_error(code, path, payload, headers=HEADERS): try: post(path, payload, headers) except urllib.error.HTTPError as error: assert error.code == code, (error.code, json.load(error)) return json.load(error) raise AssertionError(f"{path} unexpectedly succeeded") with tempfile.TemporaryDirectory() as temp: env = { **os.environ, "LIGHTHOUSE_DB": f"{temp}/lighthouse.db", "LIGHTHOUSE_ADMIN_TOKEN": "test-token", "LIGHTHOUSE_PORT": "48031", "LIGHTHOUSE_NODE_MAP": str(NODE_MAP), } process = subprocess.Popen( [sys.executable, "lighthouse.py"], cwd=ROOT, env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) try: for _ in range(30): try: health = json.load(urllib.request.urlopen(BASE + "/health", timeout=1)) assert health["execution"] == "disabled" assert health["mode"] == "map-intent-agent-gate" break except OSError: time.sleep(.1) else: raise AssertionError("server did not start") status = json.load(urllib.request.urlopen(BASE + "/v1/status")) assert status["node_id"] == "AW-GZ-001" assert status["host_state"] == "ONLINE" assert status["technical_control"] == { "enterprise_human_controller": "Awen", "enterprise_human_controller_id": None, "enterprise_persona_executor_id": None, "state": "HUMAN_DECLARED_PERSONA_PENDING_REGISTRATION", "final_deployment_signature_state": "NOT_PRESENT", "authorization_receipt": None, "production_deployment": "BLOCKED", } assert {domain["state"] for domain in status["domains"] if domain["id"] != "DOMAIN-FIFTH"} == { "HOSTED_READ_ONLY" } assert { domain["steward_state"] for domain in status["domains"] if domain["id"] != "DOMAIN-FIFTH" } == {"UNBOUND"} assert { domain["mutation_state"] for domain in status["domains"] if domain["id"] != "DOMAIN-FIFTH" } == {"BLOCKED_UNTIL_PERSONA_STEWARD_BOUND"} assert { domain["technical_controller_human_name"] for domain in status["domains"] if domain["id"] != "DOMAIN-FIFTH" } == {"Awen"} assert { domain["deployment_state"] for domain in status["domains"] if domain["id"] != "DOMAIN-FIFTH" } == {"BLOCKED"} assert all( domain["human_steward_id"] is None and domain["persona_steward_id"] is None for domain in status["domains"] if domain["id"] != "DOMAIN-FIFTH" ) fifth = next(domain for domain in status["domains"] if domain["id"] == "DOMAIN-FIFTH") assert fifth["human_steward_id"] == "ICE-GL∞" assert fifth["persona_steward_id"] == "ICE-P-ZY001" assert fifth["responsibility_state"] == "EXTERNAL_SOVEREIGN_BOUND" assert fifth["technical_controller_persona_id"] == "ICE-P-ZY001" assert fifth["final_deployment_signature_state"] == "REQUIRED_PER_CHANGE" assert status["node_counts"]["ACTIVE"] == 4 with sqlite3.connect(env["LIGHTHOUSE_DB"]) as db: db.execute( """INSERT INTO technical_controllers (scope_id, human_display_name, human_controller_id, persona_executor_id, persona_public_key, runtime_node_id, authorization_receipt, final_signature_receipt, state, registered_at) VALUES (?, ?, NULL, NULL, NULL, NULL, ?, ?, 'AUTHORIZED_BOOTSTRAP_PERSONA_PENDING_REGISTRATION', ?)""", ( "ENTERPRISE-FOUR-DOMAINS", "Awen", "sha256:test-awen-message", "sha256:test-awen-message", int(time.time()), ), ) authorized = json.load(urllib.request.urlopen(BASE + "/v1/status")) assert authorized["technical_control"]["production_deployment"] == "AUTHORIZED_FOR_CURRENT_BOOTSTRAP" assert authorized["technical_control"]["authorization_receipt"] == "sha256:test-awen-message" assert { domain["deployment_state"] for domain in authorized["domains"] if domain["id"] != "DOMAIN-FIFTH" } == {"AUTHORIZED_FOR_CURRENT_BOOTSTRAP"} nodes = json.load(urllib.request.urlopen(BASE + "/v1/nodes"))["nodes"] hosted = {node["domain_id"]: node for node in nodes} assert set(hosted) == {"DOMAIN-MAIN", "DOMAIN-SUB", "DOMAIN-ZERO", "DOMAIN-ZS"} assert all(node["state"] == "ACTIVE" for node in hosted.values()) assert all(node["display_name"].startswith("AW-GZ-001") for node in hosted.values()) intake = { "human_name": "Test", "email": "test@example.invalid", "server_ip": "203.0.113.8", "domain_id": "DOMAIN-ZS", } expect_error(423, "/v1/intakes", intake) nav = json.load(urllib.request.urlopen(BASE + "/v2/global-navigation-map")) assert nav["navigation_map"]["schema"] == "guanghu.enterprise-global-navigation-map/v2" assert nav["navigation_map"]["node_map"]["node_id"] == "AW-GZ-001" assert post("/v1/navigation-map/ack", {"map_hash": nav["map_hash"]})["ok"] is True expect_error(423, "/v1/intakes", intake) capsule = { "schema": "guanghu.intent-state-capsule/v1", "human_anchor": "ICE-GL-INF", "persona_id": "ICE-GL-ZY001", "task_intent": "Test the enterprise navigation and Agent gate.", "identity_boundary": "The public operations system is not a persona.", "established_facts": ["AW map is current."], "decisions": ["Use only a registered Agent."], "rejected_routes": ["raw shell"], "authorization_state": "test-only", "current_checkpoint": "map acknowledged", "next_action": "unlock inspector", "completion_definition": ["preflight is gated"], "evidence": ["deployment/navigation-maps/AW-GZ-001.json"], } restored = post("/v1/intent-state/restore", capsule) assert restored["ok"] is True assert json.load(urllib.request.urlopen(urllib.request.Request( BASE + "/v1/intakes", data=json.dumps(intake).encode(), method="POST", headers=HEADERS, )))["state"] == "PENDING_REVIEW" bootstrap_error = expect_error(423, "/v1/nodes/bootstrap", { "id": "AW-GZ-001", "domain_id": "DOMAIN-ZS", "display_name": "Enterprise root node", "server_ip": "203.0.113.8", }) assert bootstrap_error["required"] == "ACTIVE_HUMAN_BACKED_PERSONA_STEWARD" preflight = {"action": "health_check", "target_node_id": "AW-GZ-001", "agent_id": "AW-INSPECTOR"} expect_error(423, "/v1/preflight", preflight) unlocked = post("/v1/agents/unlock", { "agent_id": "AW-INSPECTOR", "action": "health_check", "target_node_id": "AW-GZ-001", }) preflight["unlock_id"] = unlocked["unlock_id"] expect_error(404, "/v1/preflight", preflight) assert post("/v1/agents/consume", { **preflight, "unlock_id": unlocked["unlock_id"], })["consumed"] is True expect_error(423, "/v1/preflight", preflight) expect_error(403, "/v1/nodes/bootstrap", { "id": "NODE-FIFTH-001", "domain_id": "DOMAIN-FIFTH", "display_name": "Not allowed", "server_ip": "203.0.113.8", }) expect_error(400, "/v1/intakes", {"cmd": "unsafe"}) other_headers = { **HEADERS, "X-Guanghu-Principal-Id": "ANOTHER-PERSONA:TEST", } expect_error(423, "/v1/intent-state/restore", capsule, other_headers) assert post("/v1/navigation-map/ack", {"map_hash": nav["map_hash"]}, other_headers)["ok"] is True expect_error(400, "/v1/intent-state/restore", {**capsule, "token": "forbidden"}, other_headers) finally: process.terminate() process.wait(timeout=5) print("enterprise lighthouse tests passed")