Outer hardening: AppArmor profile, pids limit, egress allowlist proxy

Inner codex sandbox stays danger-full-access by choice (bwrap
path is a separate decision); this hardens everything around it:
- DockerExecutor: --security-opt plumbing (repeatable, dedups
  no-new-privileges), --pids-limit (CLI default 512, fork-bomb
  guard), per-call extra_env (-e NAME=VALUE, non-secrets only,
  e.g. proxy URLs; the model key still goes by -e NAME only).
- Worker: --egress-proxy-port (0 = off). When on, Docker task
  containers get HTTP(S)_PROXY pointing at the host filtering
  proxy via their network gateway IP (tenant network gateway
  or default bridge gateway). Gateway unresolvable => the
  attempt fails instead of running unproxied (fail closed,
  same rule as tenant network creation failure).
- AppArmor profile deploy/apparmor/docker-codex: docker-default
  base + deny network raw/packet, deny ptrace, deny mount/
  umount/pivot_root, deny dangerous capabilities, deny writes
  to /proc/sys, /sys, /proc/keys. Deliberately NO userns rule
  (that would enable the bwrap path, not this one). File rules
  mirror docker-default (broad + sensitive denies); a strict
  file allowlist would need profiling codex's file access set
  and is not claimed here.
- Egress proxy deploy/egress/: Tinyproxy with
  FilterDefaultDeny + anchored domain allowlist (relay,
  github, pypi, npm). Anchoring matters: an unanchored
  'github.com' regex would also match 'evilgithub.com'.
- Seccomp honesty (README): Docker default seccomp already
  enforces on all task containers; no new stricter allowlist
  is claimed in this change (building one from memory risks
  silently breaking the toolchain; it needs syscall profiling
  first).
Tests: 3 outer-hardening tests (security-opt dedup, pids
limit, proxy env without leaking the key value, gateway
parse/fail, proxy fail-closed). Full suites pass.

cloud/tests/test_outer_hardening.py

@@ -0,0 +1,122 @@
+"""外层加固单测：security-opt / pids-limit / extra env / 网关解析。
+
+只用假 runner，不跑真实 Docker、不联网。覆盖：命令里
+出现 --security-opt（不重复 no-new-privileges）、--pids-limit、
+-e NAME=VALUE（代理地址，非密钥）；网关解析成功/失败；
+worker 代理启用时网关查不到则任务 failed（fail closed）。
+"""
+
+import sys
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+from unittest import mock
+
+try:
+    from cloud.sandbox import DockerExecutor
+    from cloud.server import init_db
+    from cloud.worker import Worker
+except ImportError:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+    from cloud.sandbox import DockerExecutor
+    from cloud.server import init_db
+    from cloud.worker import Worker
+
+import os
+import tempfile
+
+
+def _executor(seen, **kwargs):
+    def fake(cmd, timeout=None):
+        seen.append(list(cmd))
+        return SimpleNamespace(returncode=0, stdout="", stderr="")
+    kwargs.setdefault("env", {"XCM_API_KEY": "test-placeholder"})
+    kwargs.setdefault("runner", fake)
+    with mock.patch("cloud.sandbox.shutil.which",
+                    return_value="/usr/bin/docker"):
+        return DockerExecutor("/fake/host.py", **kwargs)
+
+
+class HardeningCommandTestCase(unittest.TestCase):
+    def test_security_opts_pids_limit_and_extra_env(self):
+        seen = []
+        ex = _executor(
+            seen, security_opts=["apparmor=docker-codex",
+                                 "no-new-privileges"],
+            pids_limit=512,
+            extra_env={"HTTP_PROXY": "http://172.18.0.1:8888"})
+        with mock.patch("cloud.sandbox.shutil.which",
+                        return_value="/usr/bin/docker"):
+            cmd = ex.build_run_command(
+                {"id": "t1", "worktree": "/tmp/wt"}, ["echo", "hi"],
+                extra_env={"HTTPS_PROXY": "http://172.18.0.1:8888"})
+        # no-new-privileges 只出现一次（硬编码 + 传入去重；
+        # 注意它是 --security-opt=no-new-privileges 单个元素）。
+        self.assertEqual(
+            sum(1 for a in cmd if "no-new-privileges" in a), 1)
+        self.assertEqual(cmd[cmd.index("apparmor=docker-codex") - 1],
+                         "--security-opt")
+        self.assertEqual(cmd[cmd.index("--pids-limit") + 1], "512")
+        self.assertIn("-e", cmd)
+        self.assertIn("HTTP_PROXY=http://172.18.0.1:8888", cmd)
+        self.assertIn("HTTPS_PROXY=http://172.18.0.1:8888", cmd)
+        # 密钥仍只传名，不带值。
+        self.assertIn("XCM_API_KEY", cmd)
+        self.assertNotIn("XCM_API_KEY=test-placeholder", cmd)
+
+    def test_network_gateway_parses_and_fails_closed(self):
+        def gw_runner(cmd, timeout=None):
+            if "network" in cmd and "inspect" in cmd:
+                return SimpleNamespace(returncode=0,
+                                       stdout="172.19.0.1\n", stderr="")
+            return SimpleNamespace(returncode=0, stdout="", stderr="")
+        with mock.patch("cloud.sandbox.shutil.which",
+                        return_value="/usr/bin/docker"):
+            ex = DockerExecutor("/fake/host.py", runner=gw_runner,
+                                env={"XCM_API_KEY": "test-placeholder"})
+            self.assertEqual(ex.network_gateway("codex-projA"),
+                             "172.19.0.1")
+            # 默认 bridge：传 None 时查 bridge 网络。
+            self.assertEqual(ex.network_gateway(None), "172.19.0.1")
+
+        def bad_runner(cmd, timeout=None):
+            return SimpleNamespace(returncode=1, stdout="", stderr="nope")
+        with mock.patch("cloud.sandbox.shutil.which",
+                        return_value="/usr/bin/docker"):
+            ex = DockerExecutor("/fake/host.py", runner=bad_runner,
+                                env={"XCM_API_KEY": "test-placeholder"})
+            self.assertIsNone(ex.network_gateway("codex-missing"))
+
+
+class WorkerProxyFailClosedTestCase(unittest.TestCase):
+    def test_proxy_gateway_failure_fails_attempt(self):
+        tmp = tempfile.TemporaryDirectory()
+        db_path = os.path.join(tmp.name, "control.db")
+        init_db(db_path)
+        seen = []
+
+        def fake_runner(cmd, timeout=None):
+            seen.append(list(cmd))
+            # 网关解析失败（inspect 非零）。
+            return SimpleNamespace(returncode=1, stdout="", stderr="nope")
+        ex = DockerExecutor("/fake/host.py", runner=fake_runner,
+                            env={"XCM_API_KEY": "test-placeholder"})
+        worker = Worker(db_path, "/fake/host.py", executor=ex,
+                        env={"XCM_API_KEY": "test-placeholder"},
+                        egress_proxy_port=8888)
+        task = {"id": "px1", "project_id": "p1", "prompt": "hi",
+                "status": "queued"}
+        # _run_attempt 的 Docker 分支在网关失败时直接返回 failed，
+        # 不调 dispatch（seen 里不应出现 docker run）。
+        with mock.patch("cloud.sandbox.shutil.which",
+                        return_value="/usr/bin/docker"):
+            result, error = worker._run_attempt(task)
+        self.assertEqual(result["status"], "failed")
+        self.assertIn("egress", error)
+        self.assertFalse(any(c and c[0] == "docker" and "run" in c
+                             for c in seen))
+        tmp.cleanup()
+
+
+if __name__ == "__main__":
+    unittest.main()