大道至简：用 700 行纯 Python 从零手搓 NVIDIA OpenShell 零信任 Agent 沙箱

+-----------------------------------------------------------------------------------+
| 隔离边界内 (Sandbox Boundary: CLONE_NEWUSER | CLONE_NEWNET | CLONE_NEWNS)         |
|                                                                                   |
|   +------------------------+        127.0.0.1 Proxy       +-------------------+   |
|   | Agent / 子进程工作负载  | ---------------------------> |   sandbox.py      |   |
|   | (/usr/bin/python3,     |                              | (边界内可信观察者) |   |
|   |  /usr/bin/curl)        | <--- SIGSTOP (断连瞬间冻结) -- | 1. 查 /proc 溯源  |   |
|   +------------------------+                              | 2. 绝不做策略决策  |   |
|                                                           +---------+---------+   |
+---------------------------------------------------------------------|-------------+
                                                                      |
                                  Mediated Channel (UDS + X-Nano-Channel-Token)
                                  [启动握手锁 Generation ID + 透传可信 BinaryIdentity]
                                                                      |
+---------------------------------------------------------------------v-------------+
| 隔离边界外 (Trusted Host Boundary)                                                 |
|                                                                                   |
|   +-------------------------------------------------------+                       |
|   | supervisor.py (唯一策略执行点 PEP & 进程外凭证注入器)   |                       |
|   | 1. 校验 Mediated Token 防绕过                         |                       |
|   | 2. 调用 policy.py 执行 L4 -> Binary -> L7 三层评估     |                       |
|   | 3. 放行时从宿主机内存动态注入 Authorization: Bearer   | ---> 上游 API 服务     |
|   | 4. 拒绝时通过 DenialAggregator 生成候选策略提案        |                       |
|   +---------------------------+---------------------------+                       |
|                               | 输出 denials.json & policy-proposal.yaml          |
|                               v                                                   |
|   +-------------------------------------------------------+                       |
|   | prover.py (静态策略边界形式化验证器)                    |                       |
|   | 校验 5 大安全不变量：云元数据防护 / 凭证外发新域名 /   |                       |
|   | 未授权 L4 逃逸 / 二进制扩权 / L7 Method-Path 放宽      |                       |
|   +-------------------------------------------------------+                       |
+-----------------------------------------------------------------------------------+

nano-openshell/
├── policy.yaml          # [配置] 基准零信任策略声明
├── policy.py            # [Milestone 1] BinaryIdentity 解析器与 L4/Binary/L7 评估引擎 (~150 行)
├── supervisor.py        # [Milestone 2] 边界外 Supervisor、凭证代填与 DenialAggregator (~215 行)
├── sandbox.py           # [Milestone 3] 边界内 Namespace 隔离、/proc 溯源与 SIGSTOP 看门狗 (~215 行)
├── prover.py            # [Milestone 4] 静态策略包含性证明器与五大阻断规则 (~155 行)
└── test_regression.sh   # [验收] 覆盖 4 大阶段、19 项断言的端到端自动化回归测试脚本

version: 1
default_action: deny

network_policies:
  - name: openai_agent_inference
    hosts:
      - "api.openai.com"
      - "api.openai.local"
      - "127.0.0.1"
    ports: [443, 80, 18080]
    # Per-Binary Network Policy: 仅允许 Python Agent 运行时访问，阻止派生的 curl/wget/sh
    allowed_binaries:
      - "/usr/bin/python3*"
      - "*/python3*"
    l7_rules:
      - methods: ["GET"]
        paths: ["/v1/models"]
      - methods: ["POST"]
        paths: ["/v1/chat/completions"]
    credential:
      env_var: "REAL_OPENAI_KEY"
      header: "Authorization"
      value_template: "Bearer ${REAL_OPENAI_KEY}"

  - name: github_public_readonly
    hosts:
      - "api.github.com"
    ports: [443]
    allowed_binaries:
      - "/usr/bin/python3*"
      - "/usr/bin/git"
    l7_rules:
      - methods: ["GET"]
        paths: ["/repos/*"]

#!/usr/bin/env python3
"""
Nano-OpenShell Milestone 1: Per-Binary Policy Schema & L4/Binary/L7 Policy Engine.
"""
from __future__ import annotations

import dataclasses
import fnmatch
import hashlib
import json
import os
import posixpath
import sys
import urllib.parse
from typing import Any, Optional
import yaml

UNTRUSTED_EXEC_PREFIXES = ("/tmp/", "/var/tmp/", "/dev/shm/")


@dataclasses.dataclass
class BinaryIdentity:
    """沙箱内发起网络请求的进程真实二进制身份 (对齐 openshell-binary-identity)"""
    path: str
    realpath: str
    sha256: str
    pid: Optional[int] = None

    @classmethod
    def from_pid(cls, pid: int) -> "BinaryIdentity":
        exe_link = f"/proc/{pid}/exe"
        realpath = os.path.realpath(exe_link)
        return cls.from_path(realpath, pid=pid)

    @classmethod
    def from_path(cls, path: str, pid: Optional[int] = None) -> "BinaryIdentity":
        realpath = os.path.realpath(path) if os.path.exists(path) else path
        sha256 = ""
        if os.path.isfile(realpath):
            h = hashlib.sha256()
            with open(realpath, "rb") as f:
                for chunk in iter(lambda: f.read(65536), b""):
                    h.update(chunk)
            sha256 = h.hexdigest()
        return cls(path=path, realpath=realpath, sha256=sha256, pid=pid)


@dataclasses.dataclass
class Decision:
    """策略评估结果与结构化拒绝情报 (对齐 DenialEvent)"""
    allowed: bool
    stage: str                      # ALLOWED | L4_HOST_PORT | BINARY_IDENTITY | L7_HTTP
    reason: str
    rule_name: Optional[str] = None
    inject_headers: dict[str, str] = dataclasses.field(default_factory=dict)
    credential_env: Optional[str] = None
    denial_record: Optional[dict[str, Any]] = None


class PolicyEngine:
    def __init__(self, policy_data: dict[str, Any]) -> None:
        self.raw = policy_data
        self.version = policy_data.get("version", 1)
        self.default_action = policy_data.get("default_action", "deny")
        self.rules: list[dict[str, Any]] = policy_data.get("network_policies", [])
        canonical = json.dumps(self.raw, sort_keys=True).encode("utf-8")
        self.generation_id = f"gen-{hashlib.sha256(canonical).hexdigest()[:12]}"

    @classmethod
    def from_yaml_file(cls, path: str) -> "PolicyEngine":
        with open(path, "r", encoding="utf-8") as f:
            data = yaml.safe_load(f) or {}
        return cls(data)

    @staticmethod
    def _match_binary(binary: BinaryIdentity, patterns: list[Any]) -> bool:
        # 拦截位于临时可写目录 (/tmp, /var/tmp, /dev/shm) 的伪造二进制
        for candidate_path in (binary.path, binary.realpath):
            norm_cand = posixpath.normpath(candidate_path)
            if any(norm_cand.startswith(prefix) for prefix in UNTRUSTED_EXEC_PREFIXES):
                return False

        if not patterns:
            return True
        candidates = {binary.path, binary.realpath}
        for pat in patterns:
            if isinstance(pat, str):
                if pat.startswith("sha256:"):
                    prefix = pat.removeprefix("sha256:").lower()
                    if binary.sha256 and binary.sha256.lower().startswith(prefix):
                        return True
                else:
                    pat_real = os.path.realpath(pat) if os.path.exists(pat) else pat
                    if any(
                        fnmatch.fnmatch(c, pat) or fnmatch.fnmatch(c, pat_real)
                        for c in candidates
                    ):
                        return True
            elif isinstance(pat, dict):
                p_glob = pat.get("path", "*")
                sha_prefix = pat.get("sha256", "").removeprefix("sha256:").lower()
                path_ok = any(fnmatch.fnmatch(c, p_glob) for c in candidates)
                sha_ok = (not sha_prefix) or (
                    bool(binary.sha256) and binary.sha256.lower().startswith(sha_prefix)
                )
                if path_ok and sha_ok:
                    return True
        return False

    def evaluate(
        self,
        method: str,
        url_or_path: str,
        host: Optional[str] = None,
        port: Optional[int] = None,
        binary: Optional[BinaryIdentity] = None,
        env: Optional[dict[str, str]] = None,
    ) -> Decision:
        env_map = env if env is not None else os.environ
        parsed = urllib.parse.urlsplit(url_or_path)
        req_host = (host or parsed.hostname or "").lower()
        req_port = port or parsed.port or (443 if parsed.scheme == "https" else 80)
        raw_path = urllib.parse.unquote(parsed.path or "/")
        req_path = posixpath.normpath(raw_path)
        if not req_path.startswith("/"):
            req_path = "/" + req_path
        req_method = method.upper()
        bin_id = binary or BinaryIdentity(path="<unknown>", realpath="<unknown>", sha256="")

        # 1. L4 (Host + Port) 过滤
        l4_rules = []
        for rule in self.rules:
            host_ok = any(fnmatch.fnmatch(req_host, h.lower()) for h in rule.get("hosts", []))
            ports = rule.get("ports", [])
            port_ok = (not ports) or (req_port in ports)
            if host_ok and port_ok:
                l4_rules.append(rule)

        if not l4_rules:
            return self._deny(
                "L4_HOST_PORT",
                f"Host '{req_host}:{req_port}' not in egress allowlist",
                bin_id, req_host, req_port, req_method, req_path,
            )

        # 2. Binary Identity (进程真实二进制路径 / SHA-256) 过滤
        bin_rules = [
            r for r in l4_rules if self._match_binary(bin_id, r.get("allowed_binaries", []))
        ]
        if not bin_rules:
            return self._deny(
                "BINARY_IDENTITY",
                f"Binary '{bin_id.path}' (realpath='{bin_id.realpath}', "
                f"sha256={bin_id.sha256[:12]}...) not authorized for '{req_host}:{req_port}'",
                bin_id, req_host, req_port, req_method, req_path,
            )

        # 3. L7 (HTTP Method + Path) 过滤与凭证注入
        for rule in bin_rules:
            l7_rules = rule.get("l7_rules", [])
            l7_ok = not l7_rules
            for l7 in l7_rules:
                methods = [m.upper() for m in l7.get("methods", ["*"])]
                paths = l7.get("paths", ["*"])
                if ("*" in methods or req_method in methods) and any(
                    fnmatch.fnmatch(req_path, p) for p in paths
                ):
                    l7_ok = True
                    break

            if l7_ok:
                inject_headers: dict[str, str] = {}
                cred_env = None
                cred = rule.get("credential")
                if cred:
                    cred_env = cred.get("env_var")
                    secret_val = env_map.get(cred_env, f"<UNSET:{cred_env}>")
                    tpl = cred.get("value_template", f"${{{cred_env}}}")
                    inject_headers[cred["header"]] = tpl.replace(f"${{{cred_env}}}", secret_val)

                return Decision(
                    allowed=True,
                    stage="ALLOWED",
                    rule_name=rule.get("name"),
                    reason=f"Matched rule '{rule.get('name')}'",
                    inject_headers=inject_headers,
                    credential_env=cred_env,
                )

        return self._deny(
            "L7_HTTP",
            f"L7 request '{req_method} {req_path}' denied on '{req_host}:{req_port}'",
            bin_id, req_host, req_port, req_method, req_path,
        )

    @staticmethod
    def _deny(
        stage: str,
        reason: str,
        bin_id: BinaryIdentity,
        host: str,
        port: int,
        method: str,
        path: str,
    ) -> Decision:
        return Decision(
            allowed=False,
            stage=stage,
            reason=reason,
            denial_record={
                "binary": bin_id.realpath if bin_id.realpath != "<unknown>" else bin_id.path,
                "binary_invoked_as": bin_id.path,
                "binary_sha256": bin_id.sha256,
                "host": host,
                "port": port,
                "method": method,
                "path": path,
                "stage": stage,
                "reason": reason,
            },
        )


def main() -> None:
    if len(sys.argv) < 4 or sys.argv[1] != "check":
        print("Usage: python3 policy.py check <METHOD> <URL> [--binary <PATH>] [--policy <YAML>]")
        sys.exit(2)

    method, url = sys.argv[2], sys.argv[3]
    binary_path = "/usr/bin/python3"
    policy_path = os.path.join(os.path.dirname(__file__) or ".", "policy.yaml")

    args = sys.argv[4:]
    while args:
        flag = args.pop(0)
        if flag == "--binary" and args:
            binary_path = args.pop(0)
        elif flag == "--policy" and args:
            policy_path = args.pop(0)

    engine = PolicyEngine.from_yaml_file(policy_path)
    bin_id = BinaryIdentity.from_path(binary_path)
    decision = engine.evaluate(method=method, url_or_path=url, binary=bin_id)

    if decision.allowed:
        inj = ", ".join(f"{k}={v}" for k, v in decision.inject_headers.items()) or "none"
        print(
            f"ALLOW [rule={decision.rule_name}, gen={engine.generation_id}] "
            f"(binary={bin_id.realpath}, sha256={bin_id.sha256[:12]}..., inject: {inj})"
        )
        sys.exit(0)
    else:
        print(f"DENY [{decision.stage}]: {decision.reason}")
        sys.exit(1)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Nano-OpenShell Milestone 2: Out-of-Process Supervisor, Mediated Handshake,
Credential Injection & Structured Denial Aggregator.
"""
from __future__ import annotations

import copy
import fnmatch
import http.client
import http.server
import json
import os
import socket
import socketserver
import struct
import threading
import time
import urllib.parse
from typing import Any, Optional
import yaml

from policy import BinaryIdentity, PolicyEngine

DEFAULT_SOCK_PATH = "/tmp/nano-supervisor.sock"
DEFAULT_UPSTREAM_PORT = 18080
DEFAULT_CHANNEL_TOKEN = "nano-mediated-token-v1"
SO_PEERCRED = getattr(socket, "SO_PEERCRED", 17)


def get_peer_pid(sock: socket.socket) -> Optional[int]:
    """通过 Linux UDS SO_PEERCRED 获取直连客户端的内核可信 PID"""
    try:
        creds = sock.getsockopt(socket.SOL_SOCKET, SO_PEERCRED, struct.calcsize("3i"))
        pid, _uid, _gid = struct.unpack("3i", creds)
        return pid if pid > 0 else None
    except OSError:
        return None


class DenialAggregator:
    """结构化拒绝感知聚合器与候选策略生成器 (对齐 openshell-sandbox/src/denial_aggregator.rs)"""

    def __init__(self, base_dir: str, baseline_policy: dict[str, Any]) -> None:
        self._lock = threading.Lock()
        self.denials_path = os.path.join(base_dir, "denials.json")
        self.proposal_path = os.path.join(base_dir, "policy-proposal.yaml")
        self.baseline = baseline_policy
        self.records: dict[tuple[str, str, int, str, str], dict[str, Any]] = {}

    def record(self, denial: dict[str, Any]) -> None:
        key = (
            denial["binary"],
            denial["host"],
            int(denial["port"]),
            denial["method"],
            denial["path"],
        )
        now_ts = int(time.time())
        with self._lock:
            if key in self.records:
                entry = self.records[key]
                entry["count"] += 1
                entry["last_seen"] = now_ts
            else:
                entry = {
                    **denial,
                    "count": 1,
                    "first_seen": now_ts,
                    "last_seen": now_ts,
                }
                self.records[key] = entry
            self._flush_locked()

    def _flush_locked(self) -> None:
        items = list(self.records.values())
        with open(self.denials_path, "w", encoding="utf-8") as f:
            json.dump({"denials": items}, f, indent=2, ensure_ascii=False)

        proposal = copy.deepcopy(self.baseline)
        rules: list[dict[str, Any]] = proposal.setdefault("network_policies", [])

        for d in items:
            host, port, binary, method, path = (
                d["host"],
                int(d["port"]),
                d["binary"],
                d["method"],
                d["path"],
            )
            matched_rule = None
            for r in rules:
                host_ok = any(fnmatch.fnmatch(host, h.lower()) for h in r.get("hosts", []))
                ports = r.get("ports", [])
                if host_ok and (not ports or port in ports):
                    matched_rule = r
                    break

            if matched_rule is not None:
                bins = matched_rule.setdefault("allowed_binaries", [])
                if binary not in bins:
                    bins.append(binary)
                l7s = matched_rule.setdefault("l7_rules", [])
                if not any(
                    method in l7.get("methods", []) and path in l7.get("paths", [])
                    for l7 in l7s
                ):
                    l7s.append({"methods": [method], "paths": [path]})
            else:
                safe_name = host.replace(".", "_").replace(":", "_")
                rules.append(
                    {
                        "name": f"proposed_{safe_name}_{port}",
                        "hosts": [host],
                        "ports": [port],
                        "allowed_binaries": [binary],
                        "l7_rules": [{"methods": [method], "paths": [path]}],
                    }
                )

        with open(self.proposal_path, "w", encoding="utf-8") as f:
            yaml.safe_dump(proposal, f, sort_keys=False, allow_unicode=True)


class ThreadingUnixHTTPServer(socketserver.ThreadingMixIn, http.server.HTTPServer):
    address_family = socket.AF_UNIX
    daemon_threads = True


class SupervisorHandler(http.server.BaseHTTPRequestHandler):
    engine: PolicyEngine
    aggregator: DenialAggregator
    channel_token: str
    upstream_port: int

    def log_message(self, fmt: str, *args: Any) -> None:
        print(f"[supervisor] {fmt % args}", flush=True)

    def _handle_all(self) -> None:
        # 1. Mediated Channel 鉴权：防止沙箱内进程绕过 sandbox.py 直连 UDS 伪造身份
        token = self.headers.get("X-Nano-Channel-Token", "")
        if token != self.channel_token:
            self._send_json(
                401,
                {
                    "error": "Unauthorized",
                    "reason": "Direct connection bypassing sandbox mediated channel is forbidden",
                },
            )
            return

        # 2. 控制面端点：启动握手锁与断连冻结监控流
        if self.path == "/__nano_control__/handshake":
            self._send_json(
                200,
                {
                    "status": "ok",
                    "generation_id": self.engine.generation_id,
                },
            )
            return

        if self.path == "/__nano_control__/watch":
            self.send_response(200)
            self.send_header("Content-Type", "text/plain")
            self.end_headers()
            try:
                while True:
                    self.wfile.write(f"alive {self.engine.generation_id}\n".encode("utf-8"))
                    self.wfile.flush()
                    time.sleep(0.5)
            except (BrokenPipeError, ConnectionResetError, OSError):
                return

        # 3. 提取可信进程二进制身份（优先使用 sandbox.py 观察到的身份，回退使用 SO_PEERCRED）
        bin_path = self.headers.get("X-Nano-Binary-Path")
        bin_realpath = self.headers.get("X-Nano-Binary-Realpath")
        bin_sha256 = self.headers.get("X-Nano-Binary-Sha256")
        if bin_path:
            binary = BinaryIdentity(
                path=bin_path,
                realpath=bin_realpath or bin_path,
                sha256=bin_sha256 or "",
            )
        else:
            peer_pid = get_peer_pid(self.connection)
            binary = (
                BinaryIdentity.from_pid(peer_pid)
                if peer_pid
                else BinaryIdentity("<unknown>", "<unknown>", "")
            )

        # 4. 解析目标 Host / Port / Path 并执行三层策略评估
        parsed = urllib.parse.urlsplit(self.path)
        host_hdr = self.headers.get("Host", "127.0.0.1")
        if ":" in host_hdr:
            h_name, h_port_str = host_hdr.rsplit(":", 1)
            default_port = int(h_port_str)
        else:
            h_name, default_port = host_hdr, (443 if parsed.scheme == "https" else 80)

        target_host = (parsed.hostname or h_name).lower()
        target_port = parsed.port or default_port
        target_path = parsed.path or "/"
        if parsed.query:
            target_path += f"?{parsed.query}"

        decision = self.engine.evaluate(
            method=self.command,
            url_or_path=target_path,
            host=target_host,
            port=target_port,
            binary=binary,
        )

        if not decision.allowed:
            if decision.denial_record:
                self.aggregator.record(decision.denial_record)
            self._send_json(
                403,
                {
                    "error": "Blocked by Nano-OpenShell Supervisor",
                    "stage": decision.stage,
                    "reason": decision.reason,
                    "binary": binary.realpath,
                },
            )
            return

        # 5. 策略放行：剥离内部 X-Nano-* 头与客户端伪造的 Authorization 头，注入真实凭证
        content_len = int(self.headers.get("Content-Length", "0"))
        body = self.rfile.read(content_len) if content_len > 0 else None

        forward_headers = {
            k: v
            for k, v in self.headers.items()
            if not k.lower().startswith("x-nano-")
            and k.lower() not in ("host", "authorization")
        }
        forward_headers["Host"] = f"{target_host}:{target_port}"
        forward_headers.update(decision.inject_headers)

        try:
            conn = http.client.HTTPConnection("127.0.0.1", self.upstream_port, timeout=5)
            conn.request(self.command, target_path, body=body, headers=forward_headers)
            resp = conn.getresponse()
            resp_body = resp.read()
            self.send_response(resp.status)
            self.send_header("Content-Type", resp.getheader("Content-Type", "application/json"))
            self.send_header("Content-Length", str(len(resp_body)))
            self.end_headers()
            self.wfile.write(resp_body)
            conn.close()
        except Exception as exc:
            self._send_json(502, {"error": "Upstream connection failed", "detail": str(exc)})

    def _send_json(self, status: int, payload: dict[str, Any]) -> None:
        raw = (json.dumps(payload, ensure_ascii=False) + "\n").encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(raw)))
        self.end_headers()
        self.wfile.write(raw)

    do_GET = do_POST = do_PUT = do_DELETE = do_HEAD = _handle_all


def start_mock_upstream(port: int) -> socketserver.ThreadingTCPServer:
    """启动本地回显测试上游，用于零外部依赖验证 Supervisor 是否成功注入真实凭证"""

    class EchoUpstreamHandler(http.server.BaseHTTPRequestHandler):
        def log_message(self, fmt: str, *args: Any) -> None:
            pass

        def _echo(self) -> None:
            payload = {
                "upstream": "mock-openai-server",
                "method": self.command,
                "path": self.path,
                "host": self.headers.get("Host"),
                "received_authorization": self.headers.get("Authorization", "<NONE>"),
            }
            raw = (json.dumps(payload, ensure_ascii=False) + "\n").encode("utf-8")
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(raw)))
            self.end_headers()
            self.wfile.write(raw)

        do_GET = do_POST = do_PUT = do_DELETE = _echo

    class ReusableTCPServer(socketserver.ThreadingTCPServer):
        allow_reuse_address = True
        daemon_threads = True

    srv = ReusableTCPServer(("127.0.0.1", port), EchoUpstreamHandler)
    threading.Thread(target=srv.serve_forever, daemon=True).start()
    return srv


def main() -> None:
    base_dir = os.path.dirname(os.path.abspath(__file__))
    policy_path = os.environ.get("NANO_POLICY_PATH", os.path.join(base_dir, "policy.yaml"))
    sock_path = os.environ.get("NANO_SOCK_PATH", DEFAULT_SOCK_PATH)
    channel_token = os.environ.get("NANO_CHANNEL_TOKEN", DEFAULT_CHANNEL_TOKEN)
    upstream_port = int(os.environ.get("NANO_UPSTREAM_PORT", str(DEFAULT_UPSTREAM_PORT)))

    engine = PolicyEngine.from_yaml_file(policy_path)
    aggregator = DenialAggregator(base_dir=base_dir, baseline_policy=engine.raw)

    start_mock_upstream(upstream_port)
    if os.path.exists(sock_path):
        os.unlink(sock_path)

    SupervisorHandler.engine = engine
    SupervisorHandler.aggregator = aggregator
    SupervisorHandler.channel_token = channel_token
    SupervisorHandler.upstream_port = upstream_port

    server = ThreadingUnixHTTPServer(sock_path, SupervisorHandler)
    print(
        f"[supervisor] Ready on UDS {sock_path} | generation={engine.generation_id} "
        f"| mock_upstream=127.0.0.1:{upstream_port}",
        flush=True,
    )
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        pass
    finally:
        server.server_close()
        if os.path.exists(sock_path):
            os.unlink(sock_path)


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Nano-OpenShell Milestone 3: Inside the Sandbox Boundary.
Linux User+Net Namespace Isolation, /proc Socket-to-Binary Observer,
Fail-Closed Handshake Lock & SIGSTOP Disconnect Freeze.
"""
from __future__ import annotations

import fcntl
import glob
import http.client
import http.server
import json
import os
import signal
import socket
import socketserver
import struct
import subprocess
import sys
import threading
from typing import Any, Optional

from policy import BinaryIdentity

DEFAULT_SOCK_PATH = "/tmp/nano-supervisor.sock"
DEFAULT_CHANNEL_TOKEN = "nano-mediated-token-v1"
CLONE_NEWNS = getattr(os, "CLONE_NEWNS", 0x00020000)
CLONE_NEWUSER = getattr(os, "CLONE_NEWUSER", 0x10000000)
CLONE_NEWNET = getattr(os, "CLONE_NEWNET", 0x40000000)
SIOCSIFFLAGS = 0x8914
IFF_UP_RUNNING = 0x1 | 0x40  # IFF_UP (0x1) | IFF_RUNNING (0x40)


def enter_isolated_namespaces() -> None:
    """创建隔离的 User + Network + Mount Namespace 并原生激活 lo 回环网卡"""
    orig_uid, orig_gid = os.getuid(), os.getgid()
    os.unshare(CLONE_NEWUSER | CLONE_NEWNET | CLONE_NEWNS)

    # 1. 严格按内核要求顺序写入 setgroups -> uid_map -> gid_map
    with open("/proc/self/setgroups", "w", encoding="utf-8") as f:
        f.write("deny\n")
    with open("/proc/self/uid_map", "w", encoding="utf-8") as f:
        f.write(f"0 {orig_uid} 1\n")
    with open("/proc/self/gid_map", "w", encoding="utf-8") as f:
        f.write(f"0 {orig_gid} 1\n")

    # 2. 使用 ioctl(SIOCSIFFLAGS) 原生拉起沙箱内的 lo (127.0.0.1) 接口
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    try:
        ifr = struct.pack("16sH", b"lo", IFF_UP_RUNNING)
        fcntl.ioctl(sock.fileno(), SIOCSIFFLAGS, ifr)
    finally:
        sock.close()


def resolve_client_binary(client_port: int) -> BinaryIdentity:
    """通过沙箱内 /proc/self/net/tcp 与 /proc/<pid>/fd 反查发起连接的真实进程二进制"""
    port_suffix = f":{client_port:04X}"
    target_inode: Optional[str] = None

    for tcp_table in ("/proc/self/net/tcp", "/proc/self/net/tcp6"):
        if not os.path.exists(tcp_table):
            continue
        try:
            with open(tcp_table, "r", encoding="utf-8") as f:
                for line in f.readlines()[1:]:
                    cols = line.split()
                    if len(cols) >= 10 and cols[1].upper().endswith(port_suffix):
                        target_inode = cols[9]
                        break
        except OSError:
            continue
        if target_inode and target_inode != "0":
            break

    if not target_inode or target_inode == "0":
        return BinaryIdentity("<unknown>", "<unknown>", "")

    socket_tag = f"socket:[{target_inode}]"
    self_pid = os.getpid()
    for fd_path in glob.glob("/proc/[0-9]*/fd/*"):
        try:
            pid_str = fd_path.split("/")[2]
            pid = int(pid_str)
            if pid == self_pid:
                continue
            if os.readlink(fd_path) == socket_tag:
                return BinaryIdentity.from_pid(pid)
        except (OSError, ValueError):
            continue

    return BinaryIdentity("<unknown>", "<unknown>", "")


class UnixHTTPConnection(http.client.HTTPConnection):
    """通过 Unix Domain Socket 连接宿主机 Supervisor 的 HTTP 客户端"""

    def __init__(self, sock_path: str, timeout: float = 10.0) -> None:
        super().__init__("localhost", timeout=timeout)
        self.sock_path = sock_path

    def connect(self) -> None:
        self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        self.sock.settimeout(self.timeout)
        self.sock.connect(self.sock_path)


class SandboxBrokerHandler(http.server.BaseHTTPRequestHandler):
    sock_path: str
    channel_token: str

    def log_message(self, fmt: str, *args: Any) -> None:
        pass

    def _bridge_to_supervisor(self) -> None:
        # 1. 在隔离边界内通过 /proc 采样真实调用进程身份（绝不在沙箱内做策略决策）
        client_port = self.client_address[1]
        bin_id = resolve_client_binary(client_port)

        content_len = int(self.headers.get("Content-Length", "0"))
        body = self.rfile.read(content_len) if content_len > 0 else None

        headers = {
            k: v
            for k, v in self.headers.items()
            if not k.lower().startswith("x-nano-")
        }
        # 2. 注入 Mediated Channel 令牌与可信进程身份元数据
        headers["X-Nano-Channel-Token"] = self.channel_token
        headers["X-Nano-Binary-Path"] = bin_id.path
        headers["X-Nano-Binary-Realpath"] = bin_id.realpath
        headers["X-Nano-Binary-Sha256"] = bin_id.sha256

        try:
            conn = UnixHTTPConnection(self.sock_path, timeout=5.0)
            conn.request(self.command, self.path, body=body, headers=headers)
            resp = conn.getresponse()
            resp_body = resp.read()
            self.send_response(resp.status)
            for k, v in resp.getheaders():
                if k.lower() not in ("transfer-encoding", "connection"):
                    self.send_header(k, v)
            self.end_headers()
            self.wfile.write(resp_body)
            conn.close()
        except Exception as exc:
            err = (
                json.dumps({"error": "Mediated channel error", "detail": str(exc)}) + "\n"
            ).encode("utf-8")
            self.send_response(502)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(err)))
            self.end_headers()
            self.wfile.write(err)

    do_GET = do_POST = do_PUT = do_DELETE = do_HEAD = _bridge_to_supervisor


def perform_startup_handshake(sock_path: str, channel_token: str) -> str:
    """Fail-Closed 启动锁：与 Supervisor 完成 generation_id 握手，失败则拒绝启动 Agent"""
    try:
        conn = UnixHTTPConnection(sock_path, timeout=3.0)
        conn.request(
            "GET",
            "/__nano_control__/handshake",
            headers={"X-Nano-Channel-Token": channel_token},
        )
        resp = conn.getresponse()
        data = json.loads(resp.read().decode("utf-8"))
        conn.close()
        if resp.status == 200 and data.get("status") == "ok":
            return data["generation_id"]
        raise RuntimeError(f"Unexpected handshake response: {resp.status} {data}")
    except Exception as exc:
        print(
            f"[sandbox][FAIL-CLOSED] Supervisor mediated handshake failed ({exc}), "
            "refusing to launch workload!",
            file=sys.stderr,
            flush=True,
        )
        sys.exit(125)


def watch_supervisor_or_freeze(
    sock_path: str, channel_token: str, workload_proc: subprocess.Popen[bytes]
) -> None:
    """后台监控 Mediated 控制流：一旦与 Supervisor 断连，立即发送 SIGSTOP 冻结整个 Agent 进程组"""
    try:
        sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        sock.connect(sock_path)
        req = (
            "GET /__nano_control__/watch HTTP/1.1\r\n"
            "Host: localhost\r\n"
            f"X-Nano-Channel-Token: {channel_token}\r\n\r\n"
        ).encode("utf-8")
        sock.sendall(req)
        while workload_proc.poll() is None:
            chunk = sock.recv(256)
            if not chunk:
                break
    except OSError:
        pass

    if workload_proc.poll() is None:
        print(
            f"\n[sandbox][FAIL-CLOSED] Mediated channel lost! "
            f"Sending SIGSTOP to workload pgid={workload_proc.pid}",
            file=sys.stderr,
            flush=True,
        )
        try:
            os.killpg(workload_proc.pid, signal.SIGSTOP)
        except ProcessLookupError:
            pass


def build_sanitized_env(proxy_port: int) -> dict[str, str]:
    """清洗宿主机敏感环境变量，防止任何真实密钥或 Channel Token 泄露进沙箱"""
    clean_env: dict[str, str] = {}
    for k, v in os.environ.items():
        upper_k = k.upper()
        if (
            upper_k.startswith(("REAL_", "NANO_"))
            or upper_k.endswith(("_KEY", "_SECRET", "_TOKEN"))
            or upper_k in ("NO_PROXY", "HTTP_PROXY", "HTTPS_PROXY")
        ):
            continue
        clean_env[k] = v

    proxy_url = f"http://127.0.0.1:{proxy_port}"
    clean_env["HTTP_PROXY"] = proxy_url
    clean_env["http_proxy"] = proxy_url
    clean_env["HTTPS_PROXY"] = proxy_url
    clean_env["https_proxy"] = proxy_url
    return clean_env


def main() -> None:
    cmd = sys.argv[1:]
    if cmd and cmd[0] == "--":
        cmd = cmd[1:]
    if not cmd:
        print("Usage: python3 sandbox.py -- <COMMAND> [ARGS...]", file=sys.stderr)
        sys.exit(2)

    sock_path = os.environ.get("NANO_SOCK_PATH", DEFAULT_SOCK_PATH)
    channel_token = os.environ.get("NANO_CHANNEL_TOKEN", DEFAULT_CHANNEL_TOKEN)

    # 1. 进入 Linux User + Network + Mount 隔离命名空间并激活 lo
    enter_isolated_namespaces()

    # 2. Fail-Closed 启动锁：未确认 Supervisor 在线与 generation_id 前绝不启动子进程
    gen_id = perform_startup_handshake(sock_path, channel_token)

    # 3. 启动沙箱内本地代理观察器 (127.0.0.1:<随机空闲端口>)
    SandboxBrokerHandler.sock_path = sock_path
    SandboxBrokerHandler.channel_token = channel_token

    class ThreadedLocalBroker(socketserver.ThreadingMixIn, http.server.HTTPServer):
        daemon_threads = True

    broker = ThreadedLocalBroker(("127.0.0.1", 0), SandboxBrokerHandler)
    proxy_port = broker.server_address[1]
    threading.Thread(target=broker.serve_forever, daemon=True).start()

    # 4. 擦除敏感环境变量并在独立进程组 (start_new_session=True) 中启动 Agent 工作负载
    child_env = build_sanitized_env(proxy_port)
    print(
        f"[sandbox] Boundary active (gen={gen_id}, local_broker=127.0.0.1:{proxy_port})",
        file=sys.stderr,
        flush=True,
    )
    proc = subprocess.Popen(cmd, env=child_env, start_new_session=True)

    # 5. 启动断连 SIGSTOP 看门狗线程
    threading.Thread(
        target=watch_supervisor_or_freeze,
        args=(sock_path, channel_token, proc),
        daemon=True,
    ).start()

    sys.exit(proc.wait())


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Nano-OpenShell Milestone 4: Static Policy Prover & Containment Verifier.
Zero-dependency static verifier blocking unsafe auto-approval of policy proposals.
"""
from __future__ import annotations

import dataclasses
import fnmatch
import posixpath
import sys
from typing import Any
import yaml

CLOUD_METADATA_ENDPOINTS = (
    "169.254.169.254",
    "fd00:ec2::254",
    "metadata.google.internal",
    "100.100.100.200",
)


@dataclasses.dataclass(frozen=True)
class Finding:
    """静态策略边界审计发现项 (对齐 openshell-prover::finding::Finding)"""
    severity: str       # CRITICAL | HIGH | MEDIUM
    code: str           # CLOUD_METADATA_GUARD | NEW_CREDENTIALED_REACH | UNAPPROVED_EGRESS_ENDPOINT | BINARY_PRIVILEGE_EXPANSION | L7_METHOD_PATH_WIDENING
    rule_name: str
    detail: str


def is_glob_subset(cand_pat: str, base_pat: str) -> bool:
    """判定候选通配符模式 cand_pat 是否为基准模式 base_pat 的子集 (cand_pat ⊆ base_pat)"""
    if cand_pat == base_pat:
        return True
    starred = cand_pat.replace("*", "__WILD_STAR__")
    probe_a = starred.replace("?", "a")
    probe_b = starred.replace("?", "b")
    return fnmatch.fnmatchcase(probe_a, base_pat) and fnmatch.fnmatchcase(probe_b, base_pat)


def _extract_binary_pattern(entry: Any) -> str:
    if isinstance(entry, str):
        return entry
    if isinstance(entry, dict):
        return str(entry.get("path", "*"))
    return str(entry)


def _normalize_ports(rule: dict[str, Any]) -> list[int]:
    ports = rule.get("ports", [])
    return [int(p) for p in ports] if ports else [0]


def _port_covered(cand_port: int, base_ports: list[int]) -> bool:
    if 0 in base_ports:
        return True
    if cand_port == 0:
        return False
    return cand_port in base_ports


def _expand_l7_pairs(rule: dict[str, Any]) -> list[tuple[str, str]]:
    l7_rules = rule.get("l7_rules", [])
    if not l7_rules:
        return [("*", "/*")]
    pairs: list[tuple[str, str]] = []
    for item in l7_rules:
        methods = [str(m).upper() for m in item.get("methods", ["*"])]
        paths = [str(p) for p in item.get("paths", ["/*"])]
        for m in methods:
            for p in paths:
                norm_p = posixpath.normpath(p) if not p.endswith("*") else p
                if not norm_p.startswith("/"):
                    norm_p = "/" + norm_p
                pairs.append((m, norm_p))
    return pairs


class PolicyProver:
    """策略包含性与安全不变量静态验证器 (对齐 openshell-prover)"""

    def __init__(self, baseline: dict[str, Any], candidate: dict[str, Any]) -> None:
        self.baseline_rules: list[dict[str, Any]] = baseline.get("network_policies", [])
        self.candidate_rules: list[dict[str, Any]] = candidate.get("network_policies", [])

    def verify(self) -> list[Finding]:
        findings: list[Finding] = []
        seen: set[tuple[str, str, str]] = set()

        def emit(severity: str, code: str, rule_name: str, detail: str) -> None:
            key = (code, rule_name, detail)
            if key not in seen:
                seen.add(key)
                findings.append(
                    Finding(severity=severity, code=code, rule_name=rule_name, detail=detail)
                )

        for cand_rule in self.candidate_rules:
            r_name = cand_rule.get("name", "<unnamed>")
            cand_hosts = [str(h).lower() for h in cand_rule.get("hosts", [])]
            cand_ports = _normalize_ports(cand_rule)

            # 1. [CRITICAL] CLOUD_METADATA_GUARD: 拦截云厂商元数据与链路本地地址 (含通配符双向检测)
            for host_pat in cand_hosts:
                for meta_ep in CLOUD_METADATA_ENDPOINTS:
                    if fnmatch.fnmatchcase(meta_ep, host_pat) or fnmatch.fnmatchcase(
                        host_pat, meta_ep
                    ):
                        emit(
                            "CRITICAL",
                            "CLOUD_METADATA_GUARD",
                            r_name,
                            f"Host pattern '{host_pat}' exposes cloud metadata endpoint '{meta_ep}'",
                        )

            # 2. [CRITICAL] NEW_CREDENTIALED_REACH: 拦截凭证向未授权新 (host, port) 外发
            cred = cand_rule.get("credential") or {}
            cred_env = cred.get("env_var")
            if cred_env:
                same_cred_base_rules = [
                    br
                    for br in self.baseline_rules
                    if (br.get("credential") or {}).get("env_var") == cred_env
                ]
                for c_host in cand_hosts:
                    for c_port in cand_ports:
                        covered = any(
                            any(
                                is_glob_subset(c_host, str(b_host).lower())
                                for b_host in br.get("hosts", [])
                            )
                            and _port_covered(c_port, _normalize_ports(br))
                            for br in same_cred_base_rules
                        )
                        if not covered:
                            port_disp = "*" if c_port == 0 else str(c_port)
                            emit(
                                "CRITICAL",
                                "NEW_CREDENTIALED_REACH",
                                r_name,
                                f"Credential '{cred_env}' extended to unapproved endpoint '{c_host}:{port_disp}'",
                            )

            # 3. 校验 L4 端点覆盖关系、Binary 扩权与 L7 放宽
            for c_host in cand_hosts:
                for c_port in cand_ports:
                    covering_base = [
                        br
                        for br in self.baseline_rules
                        if any(
                            is_glob_subset(c_host, str(b_host).lower())
                            for b_host in br.get("hosts", [])
                        )
                        and _port_covered(c_port, _normalize_ports(br))
                    ]
                    port_disp = "*" if c_port == 0 else str(c_port)
                    if not covering_base:
                        emit(
                            "HIGH",
                            "UNAPPROVED_EGRESS_ENDPOINT",
                            r_name,
                            f"Endpoint '{c_host}:{port_disp}' is not covered by any baseline L4 rule",
                        )
                        continue

                    # 4. [HIGH] BINARY_PRIVILEGE_EXPANSION: 同一 (host, port) 下允许的二进制集合扩权
                    cand_bins = [
                        _extract_binary_pattern(b)
                        for b in cand_rule.get("allowed_binaries", [])
                    ]
                    if not cand_bins:
                        if not any(
                            len(br.get("allowed_binaries", [])) == 0 for br in covering_base
                        ):
                            emit(
                                "HIGH",
                                "BINARY_PRIVILEGE_EXPANSION",
                                r_name,
                                f"Rule allows unrestricted binaries ('*') on '{c_host}:{port_disp}' where baseline restricts binaries",
                            )
                    else:
                        for c_bin in cand_bins:
                            bin_ok = any(
                                len(br.get("allowed_binaries", [])) == 0
                                or any(
                                    is_glob_subset(c_bin, _extract_binary_pattern(b_bin))
                                    for b_bin in br.get("allowed_binaries", [])
                                )
                                for br in covering_base
                            )
                            if not bin_ok:
                                emit(
                                    "HIGH",
                                    "BINARY_PRIVILEGE_EXPANSION",
                                    r_name,
                                    f"Binary '{c_bin}' is not contained in baseline allowed_binaries for '{c_host}:{port_disp}'",
                                )

                    # 5. [MEDIUM] L7_METHOD_PATH_WIDENING: 同一 (host, port) 下 HTTP Method/Path 放宽
                    cand_l7_pairs = _expand_l7_pairs(cand_rule)
                    for c_method, c_path in cand_l7_pairs:
                        l7_ok = False
                        for br in covering_base:
                            for b_method, b_path in _expand_l7_pairs(br):
                                if is_glob_subset(c_method, b_method) and is_glob_subset(
                                    c_path, b_path
                                ):
                                    l7_ok = True
                                    break
                            if l7_ok:
                                break
                        if not l7_ok:
                            emit(
                                "MEDIUM",
                                "L7_METHOD_PATH_WIDENING",
                                r_name,
                                f"L7 capability '{c_method} {c_path}' widens baseline L7 rules on '{c_host}:{port_disp}'",
                            )

        return findings


def load_yaml(path: str) -> dict[str, Any]:
    with open(path, "r", encoding="utf-8") as f:
        return yaml.safe_load(f) or {}


def main() -> None:
    if len(sys.argv) != 3:
        print(
            "Usage: python3 prover.py <BASELINE_POLICY.yaml> <CANDIDATE_POLICY.yaml>",
            file=sys.stderr,
        )
        sys.exit(2)

    base_path, cand_path = sys.argv[1], sys.argv[2]
    baseline = load_yaml(base_path)
    candidate = load_yaml(cand_path)

    prover = PolicyProver(baseline=baseline, candidate=candidate)
    findings = prover.verify()

    if not findings:
        print(
            f"PASS: Candidate policy '{cand_path}' is strictly contained within baseline '{base_path}' (0 findings). Auto-approval PERMITTED."
        )
        sys.exit(0)

    print(
        f"FAIL: Candidate policy '{cand_path}' violates safety/containment boundaries ({len(findings)} finding(s)). Auto-approval BLOCKED:"
    )
    for idx, f in enumerate(findings, 1):
        print(f"  {idx}. FAIL [{f.severity}] {f.code} (rule={f.rule_name}): {f.detail}")
    sys.exit(1)


if __name__ == "__main__":
    main()

==================================================================
Stage 1: Milestone 1 (policy.py) L4 / Binary / L7 Unit Assertions
==================================================================
  [PASS] Authorized python3 POST /v1/chat/completions allowed
  [PASS] Credential template rendered for authorized request
  [PASS] Path traversal '/repos/../admin' blocked with exit code 1 (got '1')
  [PASS] Path normalization prevented glob bypass
  [PASS] Untrusted /tmp binary blocked with exit code 1 (got '1')
  [PASS] Untrusted execution prefix (/tmp/) rejected at BINARY_IDENTITY stage

==================================================================
Stage 2: Milestone 2 & 3 (supervisor.py + sandbox.py) Runtime E2E
==================================================================
  [PASS] Sandbox refuses to start when Supervisor is offline (exit 125) (got '125')
  [PASS] Fail-closed startup handshake message verified
  [PASS] Supervisor started (pid=14150) on /tmp/nano-supervisor-regression.sock
  [PASS] Host secret REAL_OPENAI_KEY scrubbed from sandbox environment
  [PASS] CLONE_NEWNET blocked direct raw socket bypass to 1.1.1.1:80
  [PASS] Authorized python3 request received injected Bearer sk-live-999
  [PASS] Spawned /usr/bin/curl inside sandbox received HTTP 403
  [PASS] /proc socket-to-inode resolver attributed request to /usr/bin/curl
  [PASS] Unapproved domain evil.example.com:18080 received HTTP 403
  [PASS] Supervisor blocked evil.example.com at L4_HOST_PORT stage

==================================================================
Stage 3: Milestone 4 (prover.py) Closed-Loop Proposal Verification
==================================================================
  [PASS] Tightened policy verified as strict subset (exit 0)
FAIL: Candidate policy '/home/ryan/OpenShell/nano-openshell/policy-proposal.yaml' violates safety/containment boundaries (10 finding(s)). Auto-approval BLOCKED:
  1. FAIL [HIGH] BINARY_PRIVILEGE_EXPANSION (rule=openai_agent_inference): Binary '/usr/bin/curl' is not contained in baseline allowed_binaries for 'api.openai.com:443'
  ...
  6. FAIL [HIGH] BINARY_PRIVILEGE_EXPANSION (rule=openai_agent_inference): Binary '/usr/bin/curl' is not contained in baseline allowed_binaries for 'api.openai.local:18080'
  ...
  10. FAIL [HIGH] UNAPPROVED_EGRESS_ENDPOINT (rule=proposed_evil_example_com_18080): Endpoint 'evil.example.com:18080' is not covered by any baseline L4 rule
  [PASS] Auto-generated policy-proposal.yaml blocked from auto-approval (exit 1) (got '1')
  [PASS] Prover caught /usr/bin/curl binary privilege expansion
  [PASS] Prover caught unapproved L4 endpoint evil.example.com:18080

==================================================================
Stage 4: Supervisor Disconnect -> SIGSTOP Workload Freeze & Cleanup
==================================================================
    PID    PGID STAT COMMAND
  14200   14200 Ts   python3
  [PASS] Workload process group frozen with SIGSTOP (state=Ts)
  [PASS] Frozen process group (-14200) and sandbox (14200) cleanly terminated

==================================================================
ALL NANO-OPENSHELL REGRESSION TESTS PASSED SUCCESSFULLY!
==================================================================