冰朔
db729a26bd
Human-Authorizer: ICE-GL∞ Persona-Executor: ICE-P-ZY001 Technical-Controller: Awen Authorization-Evidence-SHA256: 8b08ee5af5ce98e2d30871d527c6a5941a04617e46f5ea59dc6db785fd4a0f1f Authorization-Form: HUMAN_MESSAGE_SCREENSHOT Scope: CURRENT_ENTERPRISE_SERVER_BOOTSTRAP Persona-Registration: PENDING_HANDOFF
225 lines
9.1 KiB
Python
225 lines
9.1 KiB
Python
#!/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")
|