#!/usr/bin/env python3 """Admin-managed Codex hooks for the bounded Zhulan Remote SSH lane.""" from __future__ import annotations import datetime as dt import hashlib import json import os import re import subprocess import sys from pathlib import Path from typing import Any WORKSPACE_ROOT = Path("/srv/guanghu/zhulan-codex/workspaces") LANE_PATH = Path("/etc/guanghu/zhulan-codex/LANE.hdlp") PROTECTED_MARKERS = ( "/etc/guanghu", "/var/lib/guanghu", "/opt/guanghu", "/srv/guanghu/zhulan-cell", "/etc/codex", "/root", ) FORBIDDEN_COMMANDS = re.compile( r"(^|[;&|()\s])(?:sudo|su|doas|systemctl|service|loginctl|mount|umount|" r"useradd|usermod|userdel|groupadd|visudo|sshd|iptables|nft|reboot|shutdown|" r"ssh|scp|sftp)(?=$|\s)", re.IGNORECASE, ) GIT_PUSH = re.compile(r"\bgit\b[^\n;&|]*\bpush\b", re.IGNORECASE) CANDIDATE_PUSH = re.compile(r"\bgit\b[^\n;&|]*\bpush\b[^\n;&|]*\bcandidate\b", re.IGNORECASE) PATCH_TARGET = re.compile(r"^\*\*\* (?:Add|Update|Delete) File: (.+)$", re.MULTILINE) def emit(value: dict[str, Any]) -> None: print(json.dumps(value, ensure_ascii=False, sort_keys=True)) def deny(reason: str) -> None: emit( { "hookSpecificOutput": { "hookEventName": "PreToolUse", "permissionDecision": "deny", "permissionDecisionReason": reason, }, "systemMessage": reason, } ) def under_workspace(path: Path) -> bool: try: path.resolve().relative_to(WORKSPACE_ROOT) return True except (OSError, ValueError): return False def git(workspace: Path, *args: str) -> str: try: result = subprocess.run( ["/usr/bin/git", "-C", str(workspace), *args], capture_output=True, text=True, timeout=8, check=False, ) except (OSError, subprocess.TimeoutExpired): return "unavailable" return result.stdout.strip()[:12000] if result.returncode == 0 else "unavailable" def checkpoint(payload: dict[str, Any], reason: str) -> Path | None: cwd = Path(str(payload.get("cwd") or "")) if not under_workspace(cwd): return None repo = Path(git(cwd, "rev-parse", "--show-toplevel")) if not under_workspace(repo): return None store = repo / ".zhulan" / "continuity" store.mkdir(parents=True, exist_ok=True) now = dt.datetime.now(dt.timezone.utc) record = { "schema": "guanghu.zhulan-server-checkpoint/v1", "persona_id": "ICE-GL-ZL-001", "reason": reason, "created_at": now.isoformat(), "session_id": payload.get("session_id"), "turn_id": payload.get("turn_id"), "repository": repo.name, "branch": git(repo, "branch", "--show-current"), "head": git(repo, "rev-parse", "HEAD"), "status": git(repo, "status", "--short"), "next_action": "restore LANE.hdlp and CURRENT.json, then continue the same candidate branch", } canonical = json.dumps(record, ensure_ascii=False, sort_keys=True).encode() record["sha256"] = hashlib.sha256(canonical).hexdigest() target = store / f"checkpoint-{now.strftime('%Y%m%dT%H%M%SZ')}-{record['sha256'][:10]}.json" target.write_text(json.dumps(record, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8") current = store / "CURRENT.json" temporary = store / ".CURRENT.json.tmp" temporary.write_text(target.read_text(encoding="utf-8"), encoding="utf-8") os.replace(temporary, current) return target def context(payload: dict[str, Any], event: str) -> str: lane = LANE_PATH.read_text(encoding="utf-8")[:12000] if LANE_PATH.is_file() else "LANE_MISSING_FAIL_CLOSED" cwd = Path(str(payload.get("cwd") or "")) current = "NO_SERVER_CHECKPOINT_YET" if under_workspace(cwd): repo = Path(git(cwd, "rev-parse", "--show-toplevel")) candidate = repo / ".zhulan" / "continuity" / "CURRENT.json" if under_workspace(repo) and candidate.is_file(): current = candidate.read_text(encoding="utf-8")[:8000] return ( f"ZHULAN_REMOTE_GUARD event={event}. Restore this server-owned lane before acting.\n" f"{lane}\nLATEST_SERVER_CHECKPOINT:\n{current}" ) def main() -> int: try: payload = json.load(sys.stdin) except (ValueError, TypeError): deny("ZHULAN_GUARD_INPUT_INVALID") return 0 event = str(payload.get("hook_event_name") or "") cwd = Path(str(payload.get("cwd") or "")) if event == "PreToolUse": if not under_workspace(cwd): deny("ZHULAN_WORKSPACE_BOUNDARY: tools may run only under the registered workspace root") return 0 tool_input = payload.get("tool_input") raw = json.dumps(tool_input, ensure_ascii=False) if not isinstance(tool_input, str) else tool_input lowered = raw.lower() if any(marker in lowered for marker in PROTECTED_MARKERS): deny("ZHULAN_PROTECTED_PATH_BOUNDARY: production, secrets, policy and root paths are read/write denied") return 0 command = str(tool_input.get("command") if isinstance(tool_input, dict) else raw) tool_name = str(payload.get("tool_name") or "") if tool_name == "apply_patch": for target in PATCH_TARGET.findall(command): candidate = Path(target.strip()) resolved = candidate if candidate.is_absolute() else cwd / candidate if not under_workspace(resolved): deny("ZHULAN_PATCH_BOUNDARY: patch targets must remain under the registered workspace root") return 0 if FORBIDDEN_COMMANDS.search(command): deny("ZHULAN_PRIVILEGE_BOUNDARY: privilege, service-manager and SSH-pivot commands are denied") return 0 if GIT_PUSH.search(command) and not CANDIDATE_PUSH.search(command): deny("ZHULAN_PUBLICATION_BOUNDARY: only the server-local candidate remote may receive a push") return 0 emit( { "hookSpecificOutput": { "hookEventName": "PreToolUse", "additionalContext": "Zhulan OS boundary checked. Candidate work only; approval and central publication remain separate.", } } ) return 0 if event in {"PreCompact", "Stop", "SessionEnd"}: saved = checkpoint(payload, event) message = f"Zhulan server checkpoint saved: {saved.name}" if saved else "Zhulan checkpoint skipped outside registered workspace" emit({"continue": True, "systemMessage": message}) return 0 if event in {"SessionStart", "PostCompact", "UserPromptSubmit"}: if event == "PostCompact": checkpoint(payload, event) emit( { "hookSpecificOutput": { "hookEventName": event, "additionalContext": context(payload, event), } } ) return 0 emit({"continue": True}) return 0 if __name__ == "__main__": raise SystemExit(main())