docker: pin tenant networks to 172.16.0.0/16 via explicit --subnet

Tenant networks were created bare (docker network create <name>),
letting Docker's default pool hand out subnets outside the iptables
8889 allowlist (172.16.0.0/12). The smoke tenant landed on
192.168.32.0/20, so containers never reached the egress proxy and
every model turn timed out at 600s twice.

Fix (B): derive a deterministic /24 inside 172.16.0.0/16 from
sha256(network name) with linear probing on IPAM overlap, max 32
probes. 172.16.0.0/16 is the one /16 in the allowlist Docker never
auto-assigns (default pool starts at 172.17.0.0/16). No persistent
state; capacity 256 tenant networks per host, explicit failure when
exhausted. Pre-existing out-of-range networks are reused as-is and
need manual removal (see README).

Tests: 7 new (candidate properties incl. wrap-around, pinned argv,
overlap probe, exhaustion, empty name).

cloud/tests/test_tenant_isolation.py

@@ -6,6 +6,7 @@
 执行器 per-call 挂载/网络覆盖、网络创建失败不降级。
 """
 
+import ipaddress
 import os
 import sys
 import tempfile
@@ -15,12 +16,12 @@ from types import SimpleNamespace
 from unittest import mock
 
 try:
-    from cloud.sandbox import DockerExecutor
+    from cloud.sandbox import DockerExecutor, tenant_subnet_candidates
     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.sandbox import DockerExecutor, tenant_subnet_candidates
     from cloud.server import init_db
     from cloud.worker import Worker
 
@@ -118,6 +119,34 @@ class TenantDirsTestCase(unittest.TestCase):
             worker._tenant_dirs("projA")
 
 
+class TenantSubnetCandidatesTestCase(unittest.TestCase):
+    def test_candidates_inside_allowlist_distinct_and_24(self):
+        allow = ipaddress.IPv4Network("172.16.0.0/12")
+        seen = list(tenant_subnet_candidates("codex-p"))
+        self.assertEqual(len(seen), 32)
+        self.assertEqual(len(set(seen)), 32)
+        for item in seen:
+            net = ipaddress.IPv4Network(item)
+            self.assertEqual(net.prefixlen, 24)
+            self.assertTrue(net.subnet_of(allow))
+
+    def test_candidates_deterministic_and_name_dependent(self):
+        first = list(tenant_subnet_candidates("codex-a"))
+        self.assertEqual(first, list(tenant_subnet_candidates("codex-a")))
+        # sha256("codex-a") 首字节 59；顺延一格是 60。
+        self.assertEqual(first[0], "172.16.59.0/24")
+        self.assertEqual(first[1], "172.16.60.0/24")
+        self.assertNotEqual(
+            first[0], list(tenant_subnet_candidates("codex-b"))[0])
+
+    def test_candidates_wrap_around_last_slot(self):
+        # sha256("codex-zz") 首字节 250：槽位到 255 后绕回 0。
+        seen = list(tenant_subnet_candidates("codex-zz"))
+        self.assertEqual(seen[0], "172.16.250.0/24")
+        self.assertEqual(seen[5], "172.16.255.0/24")
+        self.assertEqual(seen[6], "172.16.0.0/24")
+
+
 class DockerTenantCommandTestCase(unittest.TestCase):
     def test_per_call_mounts_network_and_drop_caps(self):
         seen = []
@@ -165,6 +194,68 @@ class DockerTenantCommandTestCase(unittest.TestCase):
                                 env={"XCM_API_KEY": "test-placeholder"})
             self.assertFalse(ex.ensure_network("codex-p"))
 
+    def test_ensure_network_creates_with_pinned_subnet(self):
+        seen = []
+        with mock.patch("cloud.sandbox.shutil.which",
+                        return_value="/usr/bin/docker"):
+            ex = DockerExecutor(
+                "/fake/host.py",
+                runner=_ok_runner_factory(seen),
+                env={"XCM_API_KEY": "test-placeholder"})
+            self.assertTrue(ex.ensure_network("codex-p"))
+        self.assertEqual(len(seen), 1)
+        # sha256("codex-p") 首字节 138：首候选 172.16.138.0/24。
+        self.assertEqual(seen[0], ["docker", "network", "create",
+                                   "--subnet", "172.16.138.0/24",
+                                   "codex-p"])
+
+    def test_ensure_network_overlap_probes_next_subnet(self):
+        seen = []
+
+        def overlap_then_ok(cmd, timeout=None):
+            seen.append(list(cmd))
+            if len(seen) == 1:
+                return SimpleNamespace(
+                    returncode=1, stdout="",
+                    stderr="Error response from daemon: Pool overlaps "
+                           "with other one on this address space")
+            return SimpleNamespace(returncode=0, stdout="", stderr="")
+        with mock.patch("cloud.sandbox.shutil.which",
+                        return_value="/usr/bin/docker"):
+            ex = DockerExecutor("/fake/host.py", runner=overlap_then_ok,
+                                env={"XCM_API_KEY": "test-placeholder"})
+            self.assertTrue(ex.ensure_network("codex-p"))
+        self.assertEqual(len(seen), 2)
+        self.assertEqual(seen[0][4], "172.16.138.0/24")
+        self.assertEqual(seen[1][4], "172.16.139.0/24")
+        self.assertEqual(seen[1][5], "codex-p")
+
+    def test_ensure_network_overlap_exhausted_is_false(self):
+        seen = []
+
+        def always_overlap(cmd, timeout=None):
+            seen.append(list(cmd))
+            return SimpleNamespace(
+                returncode=1, stdout="",
+                stderr="Error response from daemon: Pool overlaps "
+                       "with other one on this address space")
+        with mock.patch("cloud.sandbox.shutil.which",
+                        return_value="/usr/bin/docker"):
+            ex = DockerExecutor("/fake/host.py", runner=always_overlap,
+                                env={"XCM_API_KEY": "test-placeholder"})
+            self.assertFalse(ex.ensure_network("codex-p"))
+        self.assertEqual(len(seen), 32)
+
+    def test_ensure_network_empty_name_is_false(self):
+        def must_not_run(cmd, timeout=None):
+            raise AssertionError("empty name must not call docker")
+        with mock.patch("cloud.sandbox.shutil.which",
+                        return_value="/usr/bin/docker"):
+            ex = DockerExecutor("/fake/host.py", runner=must_not_run,
+                                env={"XCM_API_KEY": "test-placeholder"})
+            self.assertFalse(ex.ensure_network(""))
+            self.assertFalse(ex.ensure_network(None))
+
 
 if __name__ == "__main__":
     unittest.main()