134 lines
5.2 KiB
Python
134 lines
5.2 KiB
Python
#!/usr/bin/env python3
|
|
"""Resolve a host-specific route from the Guanghu Lighthouse registries."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import subprocess
|
|
from pathlib import Path
|
|
|
|
SCRIPT_PATH = Path(__file__).resolve()
|
|
REPO_ROOT = SCRIPT_PATH.parents[4]
|
|
SHARED_ROOT = SCRIPT_PATH.parents[3]
|
|
if (REPO_ROOT / "routing" / "lighthouse-path-registry.json").is_file():
|
|
DEFAULT_LIGHTHOUSE = REPO_ROOT / "routing" / "lighthouse-path-registry.json"
|
|
DEFAULT_HOST_MAP = REPO_ROOT / "routing" / "host-skill-navigation-map.json"
|
|
else:
|
|
DEFAULT_LIGHTHOUSE = SHARED_ROOT / "registries" / "lighthouse-path-registry.json"
|
|
DEFAULT_HOST_MAP = SHARED_ROOT / "registries" / "host-skill-navigation-map.json"
|
|
CURRENT_STATES = {"CURRENT", "ACTIVE", "ACTIVE_CANDIDATE", "ACTIVE_LOCAL_PRIVATE"}
|
|
|
|
|
|
def load_json(path: Path) -> dict:
|
|
with path.open(encoding="utf-8") as handle:
|
|
return json.load(handle)
|
|
|
|
|
|
def normalise(value: str) -> str:
|
|
return "".join(value.lower().split())
|
|
|
|
|
|
def resolve_host(host_map: dict, requested: str) -> dict | None:
|
|
key = normalise(requested)
|
|
for host in host_map["hosts"]:
|
|
if key in {normalise(host["id"]), *(normalise(alias) for alias in host.get("aliases", []))}:
|
|
return host
|
|
return None
|
|
|
|
|
|
def resolve_intent(host_map: dict, text: str) -> dict | None:
|
|
value = normalise(text)
|
|
scored = []
|
|
for intent in host_map["intents"]:
|
|
score = sum(len(normalise(phrase)) for phrase in intent["phrases"] if normalise(phrase) in value)
|
|
if score:
|
|
scored.append((score, intent["id"], intent))
|
|
return max(scored, default=(0, "", None))[2]
|
|
|
|
|
|
def resolve_path(registry: dict, route_id: str) -> dict:
|
|
for path in registry["paths"]:
|
|
if path["id"].upper() == route_id.upper():
|
|
return {
|
|
"status": "VALID_REGISTERED" if path["state"] in CURRENT_STATES else "INVALID_STATE",
|
|
"path": path,
|
|
}
|
|
for redirect in registry.get("redirects", []):
|
|
if redirect["id"].upper() == route_id.upper():
|
|
return {"status": "REDIRECT_ONLY_NOT_CURRENT", "redirect": redirect}
|
|
return {"status": "INVALID_NO_GUESS", "route_id": route_id}
|
|
|
|
|
|
def keychain_helper_status() -> dict:
|
|
try:
|
|
result = subprocess.run(
|
|
["git", "config", "--global", "--get-all", "credential.helper"],
|
|
check=False,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=3,
|
|
)
|
|
except (OSError, subprocess.TimeoutExpired):
|
|
return {"status": "UNAVAILABLE", "secret_read": False}
|
|
helpers = [line.strip() for line in result.stdout.splitlines() if line.strip()]
|
|
return {
|
|
"status": "AVAILABLE" if "osxkeychain" in helpers else "UNAVAILABLE",
|
|
"helper": "osxkeychain" if "osxkeychain" in helpers else None,
|
|
"secret_read": False,
|
|
}
|
|
|
|
|
|
def compile_route(registry: dict, host_map: dict, host_name: str, intent_text: str) -> dict:
|
|
host = resolve_host(host_map, host_name)
|
|
if not host:
|
|
return {"decision": "BLOCK", "error": "HOST_UNKNOWN_NO_GUESS", "requested_host": host_name}
|
|
intent = resolve_intent(host_map, intent_text)
|
|
if not intent:
|
|
return {"decision": "BLOCK", "error": "INTENT_UNKNOWN_NO_GUESS", "requested_intent": intent_text}
|
|
target = resolve_path(registry, intent["target_id"])
|
|
if target["status"] != "VALID_REGISTERED":
|
|
return {"decision": "BLOCK", "error": "LIGHTHOUSE_TARGET_INVALID", "target": target}
|
|
result = {
|
|
"schema": "guanghu.host-navigation-receipt/v1",
|
|
"decision": "ALLOW_NAVIGATION",
|
|
"lighthouse_id": registry["lighthouse_id"],
|
|
"registry_id": registry["registry_id"],
|
|
"registry_version": registry["version"],
|
|
"host": host,
|
|
"intent": intent,
|
|
"target": target["path"],
|
|
"authority_granted": False,
|
|
}
|
|
if intent.get("transport") == "LOCAL_GIT_OSXKEYCHAIN":
|
|
result["transport_probe"] = keychain_helper_status()
|
|
if result["transport_probe"]["status"] != "AVAILABLE":
|
|
result["decision"] = "BLOCK"
|
|
result["error"] = "KEYCHAIN_HELPER_UNAVAILABLE"
|
|
offline = target["path"].get("offline")
|
|
if offline and offline.startswith("/"):
|
|
result["offline_reality"] = {
|
|
"path": offline,
|
|
"exists": Path(os.path.expanduser(offline)).exists(),
|
|
}
|
|
if target["path"]["state"] == "ACTIVE_LOCAL_PRIVATE" and not result["offline_reality"]["exists"]:
|
|
result["decision"] = "BLOCK"
|
|
result["error"] = "PRIVATE_VOLUME_MISSING"
|
|
return result
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--host", required=True)
|
|
parser.add_argument("--intent", required=True)
|
|
parser.add_argument("--lighthouse", type=Path, default=DEFAULT_LIGHTHOUSE)
|
|
parser.add_argument("--host-map", type=Path, default=DEFAULT_HOST_MAP)
|
|
args = parser.parse_args()
|
|
result = compile_route(load_json(args.lighthouse), load_json(args.host_map), args.host, args.intent)
|
|
print(json.dumps(result, ensure_ascii=False, indent=2))
|
|
return 0 if result["decision"] == "ALLOW_NAVIGATION" else 2
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|