guanghu-ice-heart/server-tools/fifth-domain-lighthouse-mirror/public_mirror_agent.py

293 lines
14 KiB
Python

#!/usr/bin/env python3
"""Deterministic Fifth-Domain -> Lighthouse public projection and skill contribution gate."""
from __future__ import annotations
import argparse
import hashlib
import json
import os
from pathlib import Path, PurePosixPath
import re
import subprocess
import time
from typing import Any
DEFAULT_REPO = Path("/Volumes/JZAO/HoloLake/persona-runtime/repo-012-main")
DEFAULT_STATE = Path("/Volumes/JZAO/HoloLake/persona-runtime/shared/lighthouse-persona-skill-harbor")
MAP_PATH = "routing/fifth-domain-lighthouse-sanitized-mirror-map.json"
HARBOR_MAP_PATH = "routing/lighthouse-persona-skill-harbor-map.json"
DENIED_KEYS = re.compile(r"(?:password|passwd|token|secret|credential|private[_-]?key|private[_-]?memory|continuity[_-]?memory|raw[_-]?dialogue|relationship[_-]?memory|local[_-]?path|offline|home)", re.I)
PRIVATE_VALUE_PATTERNS = [
("PRIVATE_PATH", re.compile(r"(?:/Users/|/Volumes/|/home/|~[/\\])[^\s\"']*")),
("EMAIL", re.compile(r"(?<![\w.+-])[\w.+-]+@[\w.-]+\.[A-Za-z]{2,}")),
("PEM_PRIVATE_KEY", re.compile(r"-----BEGIN [A-Z ]*PRIVATE KEY-----")),
("CREDENTIAL_ASSIGNMENT", re.compile(r"(?i)\b(?:token|secret|password|passwd|api[_-]?key)\s*[:=]\s*[^\s,;]+")),
("URL_CREDENTIAL", re.compile(r"(?i)(?:https?|ssh)://[^\s/@:]+:[^\s/@]+@")),
]
PUBLIC_PAYLOAD_KEYS = {"instructions", "input_schema", "output_schema", "examples", "limitations", "dependencies", "license", "compatibility"}
class MirrorError(RuntimeError):
pass
def stable(value: Any) -> bytes:
return (json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")) + "\n").encode()
def sha(value: bytes) -> str:
return hashlib.sha256(value).hexdigest()
def atomic_json(path: Path, value: Any) -> str:
path.parent.mkdir(parents=True, exist_ok=True)
body = json.dumps(value, ensure_ascii=False, indent=2, sort_keys=True).encode() + b"\n"
temp = path.with_name(f".{path.name}.{os.getpid()}.tmp")
with temp.open("wb") as handle:
os.chmod(temp, 0o600)
handle.write(body)
handle.flush()
os.fsync(handle.fileno())
os.replace(temp, path)
return sha(body)
def git(repo: Path, *args: str) -> str:
result = subprocess.run(["git", "-C", str(repo), *args], text=True, capture_output=True, timeout=20)
if result.returncode:
raise MirrorError(f"GIT_READ_FAILED:{(result.stderr or result.stdout).strip()[-200:]}")
return result.stdout.strip()
def committed_json(repo: Path, commit: str, relative: str) -> dict[str, Any]:
path = PurePosixPath(relative)
if path.is_absolute() or ".." in path.parts:
raise MirrorError("INVALID_SOURCE_PATH")
result = subprocess.run(["git", "-C", str(repo), "show", f"{commit}:{relative}"], capture_output=True, timeout=20)
if result.returncode:
raise MirrorError(f"COMMITTED_SOURCE_MISSING:{relative}")
try:
value = json.loads(result.stdout)
except json.JSONDecodeError as error:
raise MirrorError(f"COMMITTED_JSON_INVALID:{relative}") from error
if not isinstance(value, dict):
raise MirrorError("SOURCE_OBJECT_REQUIRED")
return value
def sanitize_text(value: str) -> str:
output = value
for label, pattern in PRIVATE_VALUE_PATTERNS:
output = pattern.sub(f"<REDACTED:{label}>", output)
return output
def sanitize(value: Any) -> Any:
if isinstance(value, dict):
return {key: sanitize(item) for key, item in value.items() if not DENIED_KEYS.search(str(key))}
if isinstance(value, list):
return [sanitize(item) for item in value]
if isinstance(value, str):
return sanitize_text(value)
return value
def contains_private(value: Any) -> bool:
if isinstance(value, dict):
return any(DENIED_KEYS.search(str(key)) or contains_private(item) for key, item in value.items())
if isinstance(value, list):
return any(contains_private(item) for item in value)
if isinstance(value, str):
return any(pattern.search(value) for _, pattern in PRIVATE_VALUE_PATTERNS)
return False
def sync_architecture(repo: Path, state: Path, commit: str | None = None) -> dict[str, Any]:
commit = commit or git(repo, "rev-parse", "HEAD")
config = committed_json(repo, commit, MAP_PATH)
projections = []
for source in config["public_projection_sources"]:
raw = committed_json(repo, commit, source["path"])
projected = {key: raw[key] for key in source["allow"] if key in raw}
projected = sanitize(projected)
if contains_private(projected):
raise MirrorError(f"PRIVATE_DATA_REMAINS:{source['path']}")
projections.append({"source_id": raw.get("map_id") or raw.get("registry_id"), "source_path": source["path"], "source_sha256": sha(stable(raw)), "public": projected})
snapshot = {
"schema": "guanghu.zero-sense-lighthouse-public-language-mirror/v1",
"state": "CURRENT_LOCAL_SANITIZED_MIRROR_NOT_PUBLISHED",
"source_domain": "DOM-FIFTH-0001",
"target_domain": "DOMAIN-ZS",
"lighthouse_id": "SYS-GLW-LTH-0001",
"source_commit": commit,
"projections": projections,
"contains_private_source_body": False,
"authority_granted": False,
}
snapshot["freshness_token"] = sha(stable(snapshot))
current = state / "architecture" / "CURRENT.json"
previous = json.loads(current.read_text()) if current.is_file() else None
if previous and previous.get("freshness_token") == snapshot["freshness_token"]:
return {"outcome":"PASS", "state":"MIRROR_IDEMPOTENT", "freshness_token":snapshot["freshness_token"]}
current_sha = atomic_json(current, snapshot)
atomic_json(state / "architecture" / "receipts" / f"{commit}.json", {"outcome":"PASS","state":"SANITIZED_MIRROR_UPDATED_NOT_PUBLISHED","source_commit":commit,"current_sha256":current_sha,"freshness_token":snapshot["freshness_token"],"authority_granted":False})
return {"outcome":"PASS", "state":"SANITIZED_MIRROR_UPDATED_NOT_PUBLISHED", "current_sha256":current_sha, "freshness_token":snapshot["freshness_token"]}
def load_input(path: Path) -> dict[str, Any]:
if path.is_symlink():
raise MirrorError("SYMLINK_INPUT_REJECTED")
value = json.loads(path.read_text())
if not isinstance(value, dict):
raise MirrorError("INPUT_OBJECT_REQUIRED")
return value
def required_string(value: dict[str, Any], key: str) -> str:
item = value.get(key)
if not isinstance(item, str) or not item.strip():
raise MirrorError(f"REQUIRED_FIELD:{key}")
return item.strip()
def prepare_contribution(event: dict[str, Any], state: Path) -> dict[str, Any]:
required = ["contribution_id", "module_id", "name", "version", "contributor_human_id", "contributor_persona_id", "summary", "usage"]
values = {key: required_string(event, key) for key in required}
if event.get("persona_share_decision") != "SHARE":
raise MirrorError("PERSONA_EXPLICIT_SHARE_DECISION_REQUIRED")
if event.get("contains_private_human_data", True) and event.get("human_data_consent") != "GRANTED":
raise MirrorError("HUMAN_DATA_CONSENT_REQUIRED")
if not isinstance(event.get("capabilities"), list) or not event["capabilities"]:
raise MirrorError("CAPABILITIES_REQUIRED")
if not isinstance(event.get("provenance"), dict):
raise MirrorError("PROVENANCE_REQUIRED")
public_payload = event.get("public_payload", {})
if not isinstance(public_payload, dict):
raise MirrorError("PUBLIC_PAYLOAD_OBJECT_REQUIRED")
public_payload = {key: public_payload[key] for key in PUBLIC_PAYLOAD_KEYS if key in public_payload}
candidate = {
"schema": "guanghu.lighthouse-persona-skill-candidate/v1",
"state": "SANITIZED_CANDIDATE_MOTHER_REVIEW_PENDING",
**values,
"capabilities": sanitize(event["capabilities"]),
"provenance": sanitize(event["provenance"]),
"public_payload": sanitize(public_payload),
"persona_share_decision": "SHARE",
"human_data_consent": event.get("human_data_consent", "NOT_APPLICABLE"),
"source_body_included": False,
"external_effect": "NONE",
}
if contains_private(candidate):
raise MirrorError("PRIVATE_DATA_REMAINS_AFTER_SANITIZATION")
candidate["candidate_sha256"] = sha(stable(candidate))
destination = state / "quarantine" / values["module_id"] / f"{values['contribution_id']}.json"
if destination.exists():
old = json.loads(destination.read_text())
if old.get("candidate_sha256") != candidate["candidate_sha256"]:
raise MirrorError("CONTRIBUTION_ID_REPLAY_WITH_DIFFERENT_CONTENT")
return old
atomic_json(destination, candidate)
return candidate
def candidate_path(state: Path, module_id: str, contribution_id: str) -> Path:
path = state / "quarantine" / module_id / f"{contribution_id}.json"
if not path.is_file():
raise MirrorError("CANDIDATE_NOT_FOUND")
return path
def review(event: dict[str, Any], state: Path, kind: str) -> dict[str, Any]:
module_id = required_string(event, "module_id")
contribution_id = required_string(event, "contribution_id")
reviewer_id = required_string(event, "reviewer_id")
decision = required_string(event, "decision")
if decision not in {"ACCEPT", "HOLD", "REJECT"}:
raise MirrorError("INVALID_REVIEW_DECISION")
candidate = json.loads(candidate_path(state, module_id, contribution_id).read_text())
if event.get("candidate_sha256") != candidate["candidate_sha256"]:
raise MirrorError("CANDIDATE_HASH_MISMATCH")
if kind == "team":
mother_path = state / "reviews" / "mother" / f"{contribution_id}.json"
if not mother_path.is_file() or json.loads(mother_path.read_text()).get("decision") != "ACCEPT":
raise MirrorError("MOTHER_ACCEPT_REQUIRED_BEFORE_TEAM_REVIEW")
receipt = {
"schema": f"guanghu.lighthouse-persona-skill-{kind}-review/v1",
"reviewer_kind": "TCS_MOTHER" if kind == "mother" else "GUANGHU_HUMAN_TEAM",
"reviewer_id": reviewer_id,
"module_id": module_id,
"contribution_id": contribution_id,
"candidate_sha256": candidate["candidate_sha256"],
"decision": decision,
"reason_code": event.get("reason_code", "NONE"),
"external_effect": "NONE",
}
atomic_json(state / "reviews" / kind / f"{contribution_id}.json", receipt)
return receipt
def register(module_id: str, contribution_id: str, state: Path) -> dict[str, Any]:
candidate = json.loads(candidate_path(state, module_id, contribution_id).read_text())
reviews = []
for kind in ("mother", "team"):
path = state / "reviews" / kind / f"{contribution_id}.json"
if not path.is_file():
raise MirrorError(f"{kind.upper()}_REVIEW_REQUIRED")
item = json.loads(path.read_text())
if item.get("decision") != "ACCEPT" or item.get("candidate_sha256") != candidate["candidate_sha256"]:
raise MirrorError(f"{kind.upper()}_ACCEPT_HASH_BOUND_REVIEW_REQUIRED")
reviews.append(item)
if reviews[0]["reviewer_id"] == reviews[1]["reviewer_id"]:
raise MirrorError("INDEPENDENT_REVIEWERS_REQUIRED")
record = {key: candidate[key] for key in ["module_id","name","version","contributor_human_id","contributor_persona_id","summary","usage","capabilities","provenance","public_payload","candidate_sha256"]}
record.update({"schema":"guanghu.light-arrival-skill-registration/v1","state":"ARRIVAL_REGISTERED_LOCAL_NOT_PUBLISHED","contribution_id":contribution_id,"channel_id":"CH-LIGHT-ARRIVAL-SKILL-0001","mother_review_sha256":sha(stable(reviews[0])),"team_review_sha256":sha(stable(reviews[1])),"runtime_enabled":False,"authority_granted":False})
atomic_json(state / "registered" / f"{module_id}.json", record)
registered = []
for path in sorted((state / "registered").glob("*.json")):
registered.append(json.loads(path.read_text()))
catalog = {"schema":"guanghu.lighthouse-persona-skill-catalog/v1","catalog_id":"SYS-GLW-LTH-SKILL-0001","state":"LOCAL_REGISTERED_NOT_PUBLISHED","modules":registered,"external_publication":False}
atomic_json(state / "catalog" / "CURRENT.json", catalog)
return record
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("command", choices=("sync-architecture", "prepare", "mother-review", "team-review", "register", "status"))
parser.add_argument("--repo", default=str(DEFAULT_REPO))
parser.add_argument("--state-root", default=str(DEFAULT_STATE))
parser.add_argument("--input")
parser.add_argument("--module-id")
parser.add_argument("--contribution-id")
args = parser.parse_args()
repo, state = Path(args.repo), Path(args.state_root)
try:
if args.command == "sync-architecture":
result = sync_architecture(repo, state)
elif args.command in {"prepare", "mother-review", "team-review"}:
if not args.input:
raise MirrorError("INPUT_REQUIRED")
event = load_input(Path(args.input))
result = prepare_contribution(event, state) if args.command == "prepare" else review(event, state, "mother" if args.command == "mother-review" else "team")
elif args.command == "register":
if not args.module_id or not args.contribution_id:
raise MirrorError("MODULE_AND_CONTRIBUTION_REQUIRED")
result = register(args.module_id, args.contribution_id, state)
else:
current = state / "architecture" / "CURRENT.json"
catalog = state / "catalog" / "CURRENT.json"
result = {"outcome":"PASS","architecture_mirror":current.is_file(),"catalog":catalog.is_file(),"state_root":str(state),"external_publication":False}
print(json.dumps(result, ensure_ascii=False, indent=2, sort_keys=True))
return 0
except Exception as error:
rejection = {"outcome":"FAIL","error":str(error)[:500],"private_content_echoed":False,"last_known_good_preserved":True,"at_unix":int(time.time())}
try:
atomic_json(state / "rejected" / f"{time.time_ns()}.json", rejection)
except Exception:
pass
print(json.dumps(rejection, ensure_ascii=False, indent=2), file=os.sys.stderr)
return 2
if __name__ == "__main__":
raise SystemExit(main())