fifth-domain/zero-point/core-channel/HRC-INTERCEPTOR.py

#!/usr/bin/env python3
# ════════════════════════════════════════════════════════
# HRC-INTERCEPTOR.py · 人类登录接口统一拦截
# ════════════════════════════════════════════════════════
# 用途:任何人类登录路径(SSH / 控制台 / Web shell)都强制 HRC 验证
# 路径:部署到 /usr/local/bin/hrc-interceptor.py(可由 PAM 触发)
# 设计原则:
#   - 不依赖网络(本地校验)
#   - 验证通过才放行,失败立即拒绝
#   - 时间窗口 ±2 分钟(防 replay)
#   - 失败的尝试写日志(供事后审计)
# ════════════════════════════════════════════════════════

import hmac
import hashlib
import time
import sys
import os
import subprocess

HRC_SECRET_PATHS = [
    "/etc/hrc/recovery-secret",
    "/etc/light-lake-driver/hrc-secret",
    "/root/.hrc/secret",
]
HRC_VALIDITY_MINUTES = 5
HRC_WINDOW = 2  # ±2 分钟


def get_hrc_secret():
    """获取 HRC 共享密钥(优先级:文件 → 环境变量 → 报错)"""
    for path in HRC_SECRET_PATHS:
        if os.path.exists(path):
            try:
                with open(path) as f:
                    return f.read().strip()
            except Exception:
                pass
    env = os.environ.get("HRC_SECRET")
    if env:
        return env
    return None


def compute_hrc(secret, minute_ts):
    """计算 HRC 期望值"""
    msg = f"HRC-{minute_ts}"
    return hmac.new(
        secret.encode(),
        msg.encode(),
        hashlib.sha256
    ).hexdigest()[:16]


def verify_hrc(provided_hrc, secret):
    """验证 HRC(±2 分钟窗口)"""
    if not provided_hrc or len(provided_hrc) != 16:
        return False, "format_invalid"

    current_minute = int(time.time() // 60)
    for offset in range(-HRC_WINDOW, HRC_WINDOW + 1):
        ts = current_minute + offset
        expected = compute_hrc(secret, ts)
        if hmac.compare_digest(provided_hrc, expected):
            return True, f"ok_minute_{ts}_offset_{offset}"

    return False, "expired_or_invalid"


def log_attempt(result, user="unknown", source="unknown"):
    """记录验证尝试"""
    log_file = "/var/log/hrc-interceptor.log"
    try:
        ts = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
        line = f"[{ts}] user={user} source={source} result={result}\n"
        with open(log_file, "a") as f:
            f.write(line)
    except Exception:
        pass


def prompt_for_hrc():
    """提示用户输入 HRC"""
    print("╔════════════════════════════════════════════════════════╗")
    print("║  🔐 光湖驱动引擎 · 主机恢复码 (HRC) 验证           ║")
    print("╠════════════════════════════════════════════════════════╣")
    print("║  任何人类登录都必须通过 HRC 验证                    ║")
    print("║  请联系铸渊 ICE-GL-ZY001 获取 HRC                  ║")
    print("║  HRC = 16 位 hex 字符 · 5 分钟内有效               ║")
    print("╚════════════════════════════════════════════════════════╝")
    print()
    print("请输入 HRC:")
    hrc = input().strip()
    return hrc


def main():
    """主流程"""
    # 1. 检查环境
    if os.geteuid() != 0:
        print("❌ HRC-INTERCEPTOR 必须以 root 运行")
        sys.exit(1)

    # 2. 获取 secret
    secret = get_hrc_secret()
    if not secret:
        print("❌ 拿不到 HRC secret(检查 /etc/hrc/recovery-secret)")
        sys.exit(1)

    # 3. 检查是否已通过(从环境变量)
    hrc_validated = os.environ.get("HRC_VALIDATED")
    if hrc_validated == "1":
        # 已经在前置 PAM 模块通过,直接放行
        sys.exit(0)

    # 4. 检查命令行参数(便于非交互)
    if len(sys.argv) > 1:
        provided_hrc = sys.argv[1].strip()
    else:
        provided_hrc = prompt_for_hrc()

    # 5. 验证
    user = os.environ.get("USER", "unknown")
    source = os.environ.get("HRC_SOURCE", "unknown")
    is_valid, reason = verify_hrc(provided_hrc, secret)

    if is_valid:
        log_attempt(f"OK ({reason})", user=user, source=source)
        # 标记已验证(PAM 链下个模块看到 HRC_VALIDATED=1 放行)
        os.environ["HRC_VALIDATED"] = "1"
        # 启动真实 shell
        user_shell = os.environ.get("SHELL", "/bin/bash")
        os.execvpe(user_shell, [user_shell, "-l"], os.environ)
    else:
        log_attempt(f"FAIL ({reason})", user=user, source=source)
        print()
        print("❌ HRC 验证失败")
        print(f"   原因:{reason}")
        print(f"   时间窗口:±{HRC_WINDOW} 分钟")
        print()
        print("   请联系铸渊 ICE-GL-ZY001 获取最新 HRC")
        print("   或者从腾讯云 web 控制台 → VPS → 终端(绕过此拦截)")
        print()
        sys.exit(1)


if __name__ == "__main__":
    main()