875b4c2381
- pre-op-guard.py: 前置操作守门人，人格体登录时锁定服务器，必须完成导航地图验证 - module-registry.hdlp: 模块编码体系 + 注册协议 - module-register.sh: 一键注册模块到服务器 - OPERATION-MANUAL.hdlp v1.1: 新增第八节 服务器操作前导航挑战 - INDEX.hdlp: 注册新文件 设计: 人格体登录 → 服务器自我审计 → 锁定 → 挑战:找出所有模块编码 人格体探索 → 收集编码 → 提交 → 服务器比对 一致 → 解锁 · 人格体知道全局地图 · 不会再改A坏B [SEC-CLEAN] · pre-op-guard 已就绪
422 lines
16 KiB
Python
422 lines
16 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
光湖语言系统 · 前置操作守门人 v1.0
|
||
Guanghu Language System · Pre-Operation Guard
|
||
|
||
每次人格体连接服务器 → 触发自我审计 → 锁定 → 人格体必须探索+提交导航地图
|
||
地图编码与服务器自审一致 → 解锁操作
|
||
|
||
设计哲学:
|
||
⊢ 铸渊上服务器就动手 → 不知道还跑了别的模块 → 改A坏B → 冰朔每次都要说
|
||
⊢ 服务器全天运行 → 人格体接入时触发自我审计 → "我现在跑了这些模块"
|
||
⊢ 人格体必须探索 → 画出完整导航地图 → 提交编码列表
|
||
⊢ 服务器比对: 人格体提交的编码 == 服务器自审编码?
|
||
⊢ 一致 → 解锁 · 人格体知道完整地图了 · 不会再改A坏B
|
||
⊢ 不一致 → "你漏了 MOD-X · 继续探索"
|
||
|
||
用法:
|
||
pre-op-guard.py --audit # 服务器自我审计
|
||
pre-op-guard.py --challenge [--for PERSONA_NAME] # 触达挑战
|
||
pre-op-guard.py --submit '<json_codes>' # 人格体提交验证
|
||
pre-op-guard.py --status # 查看锁定状态
|
||
pre-op-guard.py --unlock --force # 紧急解锁（冰朔用）
|
||
"""
|
||
|
||
import os, sys, json, time, subprocess, hashlib
|
||
from datetime import datetime, timedelta
|
||
|
||
# ═══════════════════════════════════════════════════════
|
||
# 配置
|
||
# ═══════════════════════════════════════════════════════
|
||
REGISTRY_DIR = os.environ.get("REGISTRY_DIR", "/opt/zhuyuan/registry/modules.d")
|
||
MAPS_DIR = os.environ.get("MAPS_DIR", "/opt/zhuyuan/registry/maps")
|
||
LOCK_DIR = os.environ.get("LOCK_DIR", "/opt/zhuyuan/registry/lock")
|
||
LOCK_FILE = os.path.join(LOCK_DIR, "pre-op-lock.json")
|
||
|
||
# 锁定超时（秒）：超过此时间无人提交 → 自动解锁（防止死锁）
|
||
LOCK_TIMEOUT = int(os.environ.get("PRE_OP_LOCK_TIMEOUT", "1800")) # 30 分钟
|
||
|
||
# 是否强制启用（可以通过环境变量禁用）
|
||
ENABLED = os.environ.get("PRE_OP_GUARD_ENABLED", "1") == "1"
|
||
|
||
|
||
def ensure_dirs():
|
||
"""确保所有目录存在"""
|
||
for d in [REGISTRY_DIR, MAPS_DIR, LOCK_DIR]:
|
||
os.makedirs(d, exist_ok=True)
|
||
|
||
|
||
def get_server_identity():
|
||
"""获取服务器身份"""
|
||
hostname = os.uname().nodename if hasattr(os, 'uname') else os.environ.get("HOSTNAME", "unknown")
|
||
return hostname
|
||
|
||
|
||
def audit_server():
|
||
"""服务器自我审计: 读取所有注册模块 → 返回编码集"""
|
||
ensure_dirs()
|
||
modules = []
|
||
if os.path.isdir(REGISTRY_DIR):
|
||
for filename in sorted(os.listdir(REGISTRY_DIR)):
|
||
if not filename.endswith('.json'):
|
||
continue
|
||
filepath = os.path.join(REGISTRY_DIR, filename)
|
||
try:
|
||
with open(filepath, 'r') as f:
|
||
data = json.load(f)
|
||
if data.get('status') == 'running':
|
||
modules.append({
|
||
'code': data['code'],
|
||
'name': data.get('name', ''),
|
||
'port': data.get('port', 0),
|
||
'function': data.get('function', ''),
|
||
})
|
||
except (json.JSONDecodeError, IOError):
|
||
pass
|
||
|
||
codes = sorted([m['code'] for m in modules])
|
||
return {
|
||
'server': get_server_identity(),
|
||
'audited_at': datetime.utcnow().isoformat() + 'Z',
|
||
'module_count': len(codes),
|
||
'codes': codes,
|
||
'modules': modules,
|
||
}
|
||
|
||
|
||
def is_locked():
|
||
"""检查当前是否锁定"""
|
||
if not os.path.exists(LOCK_FILE):
|
||
return False
|
||
try:
|
||
with open(LOCK_FILE, 'r') as f:
|
||
lock = json.load(f)
|
||
# 检查超时
|
||
locked_at = datetime.fromisoformat(lock.get('locked_at', '').replace('Z', '+00:00'))
|
||
if datetime.utcnow() - locked_at.replace(tzinfo=None) > timedelta(seconds=LOCK_TIMEOUT):
|
||
# 超时 → 自动解锁
|
||
os.remove(LOCK_FILE)
|
||
return False
|
||
return True
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
def create_lock(challenger="unknown"):
|
||
"""锁定服务器"""
|
||
ensure_dirs()
|
||
audit = audit_server()
|
||
lock = {
|
||
'locked': True,
|
||
'locked_at': datetime.utcnow().isoformat() + 'Z',
|
||
'challenger': challenger,
|
||
'module_count': audit['module_count'],
|
||
'server_codes_hash': hashlib.sha256(
|
||
','.join(audit['codes']).encode()
|
||
).hexdigest()[:16],
|
||
'timeout_seconds': LOCK_TIMEOUT,
|
||
}
|
||
with open(LOCK_FILE, 'w') as f:
|
||
json.dump(lock, f, indent=2)
|
||
return lock, audit
|
||
|
||
|
||
def release_lock():
|
||
"""解锁服务器"""
|
||
if os.path.exists(LOCK_FILE):
|
||
os.remove(LOCK_FILE)
|
||
return True
|
||
return False
|
||
|
||
|
||
def present_challenge(audit, persona_name="铸渊"):
|
||
"""向人格体呈现挑战"""
|
||
module_count = audit['module_count']
|
||
codes = audit['codes']
|
||
modules = audit['modules']
|
||
|
||
lines = []
|
||
lines.append("")
|
||
lines.append("╔══════════════════════════════════════════════╗")
|
||
lines.append("║ 🗺️ 光湖前置操作守门人 · 导航地图挑战 ║")
|
||
lines.append("╚══════════════════════════════════════════════╝")
|
||
lines.append("")
|
||
lines.append(f" 服务器: {audit['server']}")
|
||
lines.append(f" 挑战者: {persona_name}")
|
||
lines.append(f" 状态: 🔒 已锁定 · 操作被拦截")
|
||
lines.append(f" 当前运行模块数: {module_count}")
|
||
lines.append("")
|
||
lines.append(" ── 挑战说明 ──")
|
||
lines.append(f" 此服务器上正在运行 {module_count} 个模块。")
|
||
lines.append(" 每个模块有唯一编码（格式: SERVER-FUNC-SEQ）。")
|
||
lines.append("")
|
||
lines.append(" 你需要:")
|
||
lines.append(f" ① 探索服务器，发现全部 {module_count} 个模块")
|
||
lines.append(" ② 用 HLDP 格式画出导航地图")
|
||
lines.append(" ③ 通过 pre-push-clean 过滤敏感信息")
|
||
lines.append(" ④ 提交编码列表到验证端点")
|
||
lines.append("")
|
||
lines.append(" 探索指引:")
|
||
lines.append(f" ls {REGISTRY_DIR}/")
|
||
lines.append(" systemctl list-units --type=service --state=running")
|
||
lines.append(" ss -tlnp # 查看监听端口")
|
||
lines.append("")
|
||
lines.append(" ── 已注册模块（供参考 · 仅显示名称 · 编码需自行发现）──")
|
||
for i, m in enumerate(modules, 1):
|
||
lines.append(f" [{i}] {m['name']}")
|
||
|
||
lines.append("")
|
||
lines.append(" ── 提交方式 ──")
|
||
lines.append(" 在服务器上执行:")
|
||
lines.append(" pre-op-guard.py --submit '[")
|
||
lines.append(' "GZ006-GTW-01",')
|
||
lines.append(' "GZ006-LAMP-01",')
|
||
lines.append(' "GZ006-FRG-01",')
|
||
lines.append(' ... # 你收集到的所有编码')
|
||
lines.append(" ]'")
|
||
lines.append("")
|
||
lines.append(f" ⏰ 超时: {LOCK_TIMEOUT // 60} 分钟后自动解锁")
|
||
lines.append("")
|
||
|
||
return '\n'.join(lines)
|
||
|
||
|
||
def validate_submission(submitted_codes):
|
||
"""验证人格体提交的编码列表"""
|
||
audit = audit_server()
|
||
server_codes = set(audit['codes'])
|
||
submitted_set = set(submitted_codes)
|
||
|
||
missing = server_codes - submitted_set
|
||
extra = submitted_set - server_codes
|
||
|
||
result = {
|
||
'valid': len(missing) == 0 and len(extra) == 0,
|
||
'server_codes': sorted(server_codes),
|
||
'submitted_codes': sorted(submitted_set),
|
||
'missing': sorted(missing),
|
||
'extra': sorted(extra),
|
||
'module_count': len(server_codes),
|
||
'submitted_count': len(submitted_set),
|
||
}
|
||
|
||
if result['valid']:
|
||
# 解锁
|
||
release_lock()
|
||
# 保存导航地图
|
||
save_map(submitted_codes, audit)
|
||
else:
|
||
# 更新锁定（重置超时，给更多时间）
|
||
if is_locked():
|
||
with open(LOCK_FILE, 'r') as f:
|
||
lock = json.load(f)
|
||
lock['locked_at'] = datetime.utcnow().isoformat() + 'Z'
|
||
lock['last_submission'] = {
|
||
'submitted_codes': sorted(submitted_set),
|
||
'missing': sorted(missing),
|
||
'extra': sorted(extra),
|
||
}
|
||
with open(LOCK_FILE, 'w') as f:
|
||
json.dump(lock, f, indent=2)
|
||
|
||
return result
|
||
|
||
|
||
def save_map(codes, audit):
|
||
"""保存人格体提交的导航地图为持久化 artifact"""
|
||
ensure_dirs()
|
||
map_data = {
|
||
'server': audit['server'],
|
||
'created_at': datetime.utcnow().isoformat() + 'Z',
|
||
'validated_at': datetime.utcnow().isoformat() + 'Z',
|
||
'codes': sorted(codes),
|
||
'module_count': len(codes),
|
||
'full_modules': audit['modules'],
|
||
}
|
||
timestamp = datetime.utcnow().strftime('%Y%m%d-%H%M%S')
|
||
map_file = os.path.join(MAPS_DIR, f"map-{audit['server']}-{timestamp}.json")
|
||
with open(map_file, 'w') as f:
|
||
json.dump(map_data, f, indent=2, ensure_ascii=False)
|
||
|
||
# 同时保存一个 HLDP 格式的副本
|
||
hldp_file = os.path.join(MAPS_DIR, f"map-{audit['server']}-{timestamp}.hdlp")
|
||
with open(hldp_file, 'w') as f:
|
||
f.write(f"# 光湖导航地图 · {audit['server']}\n")
|
||
f.write(f"## 验证通过 · {datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S')}\n\n")
|
||
f.write(f"服务器: {audit['server']}\n")
|
||
f.write(f"模块数: {len(codes)}\n\n")
|
||
f.write("| 编码 | 名称 | 端口 |\n")
|
||
f.write("|------|------|------|\n")
|
||
for m in audit['modules']:
|
||
f.write(f"| `{m['code']}` | {m['name']} | {m['port']} |\n")
|
||
f.write(f"\n> ⊢ 本地图由 pre-op-guard 自动生成 · 验证通过后存档\n")
|
||
f.write(f"> ⊢ 后续人格体可引用本地图快速验证\n")
|
||
f.write(f"> ⊢ 冰朔 ICE-GL∞ · 国作登字-2026-A-00037559\n")
|
||
|
||
return map_file
|
||
|
||
|
||
def get_status():
|
||
"""获取当前状态"""
|
||
status = {
|
||
'enabled': ENABLED,
|
||
'locked': is_locked(),
|
||
'server': get_server_identity(),
|
||
}
|
||
if status['locked']:
|
||
try:
|
||
with open(LOCK_FILE, 'r') as f:
|
||
lock = json.load(f)
|
||
status['lock_details'] = lock
|
||
except Exception:
|
||
pass
|
||
if not status['locked']:
|
||
audit = audit_server()
|
||
status['modules_running'] = audit['module_count']
|
||
status['codes'] = audit['codes']
|
||
return status
|
||
|
||
|
||
def main():
|
||
if not ENABLED:
|
||
print("⊢ pre-op-guard 已禁用 (PRE_OP_GUARD_ENABLED=0)")
|
||
sys.exit(0)
|
||
|
||
if '--audit' in sys.argv:
|
||
audit = audit_server()
|
||
print(json.dumps(audit, indent=2, ensure_ascii=False))
|
||
|
||
elif '--challenge' in sys.argv:
|
||
persona = "铸渊"
|
||
for i, arg in enumerate(sys.argv):
|
||
if arg == '--for' and i + 1 < len(sys.argv):
|
||
persona = sys.argv[i + 1]
|
||
|
||
if is_locked():
|
||
print("⚠️ 服务器已锁定 · 上一个挑战尚未完成")
|
||
status = get_status()
|
||
print(json.dumps(status, indent=2, ensure_ascii=False))
|
||
sys.exit(1)
|
||
|
||
lock, audit = create_lock(persona)
|
||
challenge = present_challenge(audit, persona)
|
||
print(challenge)
|
||
|
||
elif '--submit' in sys.argv:
|
||
if not is_locked():
|
||
audit = audit_server()
|
||
print("✅ 服务器未锁定 · 但你可以直接提交导航地图验证")
|
||
print(f" 当前运行 {audit['module_count']} 个模块")
|
||
print(f" 编码: {', '.join(audit['codes'])}")
|
||
sys.exit(0)
|
||
|
||
# 查找 JSON 参数
|
||
submit_idx = None
|
||
for i, arg in enumerate(sys.argv):
|
||
if arg == '--submit' and i + 1 < len(sys.argv):
|
||
submit_idx = i + 1
|
||
break
|
||
|
||
if submit_idx is None:
|
||
print("❌ 用法: pre-op-guard.py --submit '[\"CODE1\",\"CODE2\",...]'")
|
||
sys.exit(1)
|
||
|
||
try:
|
||
submitted_codes = json.loads(sys.argv[submit_idx])
|
||
if not isinstance(submitted_codes, list):
|
||
raise ValueError("必须是 JSON 数组")
|
||
except (json.JSONDecodeError, ValueError) as e:
|
||
print(f"❌ JSON 解析失败: {e}")
|
||
sys.exit(1)
|
||
|
||
result = validate_submission(submitted_codes)
|
||
|
||
print("")
|
||
if result['valid']:
|
||
print("╔══════════════════════════════════════════════╗")
|
||
print("║ ✅ 导航地图验证通过 · 服务器已解锁 ║")
|
||
print("╚══════════════════════════════════════════════╝")
|
||
print("")
|
||
print(f" 提交 {result['submitted_count']} 个编码")
|
||
print(f" 服务器运行 {result['module_count']} 个模块")
|
||
print(f" 编码: {', '.join(result['submitted_codes'])}")
|
||
print("")
|
||
print(" 🗺️ 导航地图已存档 · 后续可直接引用")
|
||
print(" 🔓 服务器已解锁 · 可以操作了")
|
||
print("")
|
||
else:
|
||
print("╔══════════════════════════════════════════════╗")
|
||
print("║ ❌ 导航地图不完整 · 服务器保持锁定 ║")
|
||
print("╚══════════════════════════════════════════════╝")
|
||
print("")
|
||
if result['missing']:
|
||
print(f" ⚠️ 缺少 {len(result['missing'])} 个模块:")
|
||
for code in result['missing']:
|
||
print(f" - {code}")
|
||
if result['extra']:
|
||
print(f" ⚠️ 多出 {len(result['extra'])} 个不存在的编码:")
|
||
for code in result['extra']:
|
||
print(f" - {code}")
|
||
print("")
|
||
print(f" ⊢ 请继续探索，重新提交。")
|
||
print(f" ⊢ 已注册模块有 {result['module_count']} 个。")
|
||
print("")
|
||
|
||
elif '--status' in sys.argv:
|
||
status = get_status()
|
||
if status['locked']:
|
||
print("🔒 服务器已锁定")
|
||
print(json.dumps(status, indent=2, ensure_ascii=False))
|
||
else:
|
||
print(f"🔓 服务器未锁定 · {status.get('modules_running', '?')} 个模块运行中")
|
||
if 'codes' in status:
|
||
print(f" 编码: {', '.join(status['codes'])}")
|
||
|
||
elif '--unlock' in sys.argv and '--force' in sys.argv:
|
||
release_lock()
|
||
print("🔓 服务器已强制解锁 · 冰朔特权操作")
|
||
|
||
elif '--install-to' in sys.argv:
|
||
# 安装 pre-op-guard 到服务器的 SSH 登录钩子
|
||
install_idx = sys.argv.index('--install-to') + 1
|
||
target = sys.argv[install_idx] if install_idx < len(sys.argv) else '/etc/profile.d/pre-op-guard.sh'
|
||
install_ssh_hook(target)
|
||
|
||
else:
|
||
print("用法:")
|
||
print(" pre-op-guard.py --audit 服务器自我审计")
|
||
print(" pre-op-guard.py --challenge 触达挑战（锁定并呈现）")
|
||
print(" pre-op-guard.py --submit '[...]' 提交编码验证")
|
||
print(" pre-op-guard.py --status 查看锁定状态")
|
||
print(" pre-op-guard.py --unlock --force 紧急解锁")
|
||
print(" pre-op-guard.py --install-to PATH 安装SSH登录钩子")
|
||
|
||
|
||
def install_ssh_hook(target_path):
|
||
"""安装 SSH 登录钩子：每次 SSH 登录时触发 pre-op-guard"""
|
||
script_dir = os.path.dirname(os.path.abspath(__file__))
|
||
hook_content = f'''#!/bin/bash
|
||
# 光湖 pre-op-guard SSH 登录钩子
|
||
# 每次 SSH 登录自动触发 · 锁定操作 · 需要完成导航地图验证
|
||
|
||
GUARD="{script_dir}/pre-op-guard.py"
|
||
|
||
# 只在交互式 shell 中触发（非 scp/sftp）
|
||
if [[ $- == *i* ]] && [ -z "$PRE_OP_BYPASS" ]; then
|
||
python3 "$GUARD" --challenge --for "$USER"
|
||
fi
|
||
'''
|
||
try:
|
||
with open(target_path, 'w') as f:
|
||
f.write(hook_content)
|
||
os.chmod(target_path, 0o755)
|
||
print(f"✅ SSH 登录钩子已安装: {target_path}")
|
||
print(f" 此后每次 SSH 登录都会触发 pre-op-guard 导航挑战")
|
||
print(f" 若要临时绕过: PRE_OP_BYPASS=1 ssh user@host")
|
||
except IOError as e:
|
||
print(f"❌ 安装失败: {e}")
|
||
|
||
|
||
if __name__ == '__main__':
|
||
main()
|