Per-lane ca_bundle and insecure for http lanes (#78), v1.2.0

BACKLOG.md

@@ -80,9 +80,10 @@ The UI lives in its own private repo,
 - [x] Env scrub has no test on the `acp` harness — `child_env()` is shared by
       `cli` and `acp`, but only `cli` asserts a key stays out; drift reopens #22
       ([#77](https://github.com/CryptoJones/FlatlineRoundtable/issues/77)) — fixed on fix/open-issues-batch
-- [ ] No per-lane CA bundle — a self-hosted/private-CA endpoint forces a global
+- [x] No per-lane CA bundle — a self-hosted/private-CA endpoint forces a global
       TLS bypass that also weakens the OpenRouter lanes
-      ([#78](https://github.com/CryptoJones/FlatlineRoundtable/issues/78))
+      ([#78](https://github.com/CryptoJones/FlatlineRoundtable/issues/78)) — fixed on feat/per-lane-ca-bundle (v1.2.0):
+      `ca_bundle` and a per-lane-only `insecure`
 - [ ] Report per-lane `cost` and run-level spend in `--json` (and under `--diff`) —
       the number cron/CI callers need is computed but not surfaced machine-readably
       ([#79](https://github.com/CryptoJones/FlatlineRoundtable/issues/79))

README.md

@@ -308,6 +308,19 @@ for every option, is kept as
 [`tests/fixtures/legacy-example.yaml`](tests/fixtures/legacy-example.yaml); each
 option it documents is a `--set` key now.
 
+An `http` lane behind a private or corporate CA names that CA's bundle:
+
+```sh
+roundtable lanes edit Local --set ca_bundle=~/certs/internal-ca.pem
+```
+
+The bundle replaces the system roots for that lane only, so the private CA cannot
+vouch for any other host and no other lane changes. For a dev endpoint with a
+self-signed cert, `--set insecure=true` turns verification off for that one lane
+and prints a warning on every call. `insecure` cannot be a default, and a lane
+cannot set both. Do not use `PYTHONHTTPSVERIFY=0`, which turns verification off
+for every lane, OpenRouter included.
+
 Every run writes a transcript to
 `~/.local/share/flatline-roundtable/transcripts/`, because answers that exist
 only in a terminal scrollback are answers waiting to be lost.

roundtable

@@ -83,6 +83,7 @@ import shutil
 import signal
 import socket
 import sqlite3
+import ssl
 import subprocess
 import tempfile
 import sys
@@ -96,7 +97,7 @@ from pathlib import Path
 
 # Semantic version. Bump it in the PR that changes behaviour: patch for a fix,
 # minor for a new flag or mode, major for a change that breaks a caller.
-__version__ = "1.1.1"
+__version__ = "1.2.0"
 
 TRANSCRIPT_DIR = Path.home() / ".local" / "share" / "flatline-roundtable" / "transcripts"
 RUN_ID_RE = re.compile(r"[A-Za-z0-9_-][A-Za-z0-9._-]{0,63}")
@@ -452,6 +453,11 @@ def validate_config(cfg: dict, where: str) -> dict:
         die(f"{where} defines no lanes")
 
     defaults = cfg.get("defaults") or {}
+    # Turning verification off has to be a per-lane decision someone typed. As a
+    # default it would quietly cover every lane, OpenRouter included -- the
+    # global bypass #78 exists to make unnecessary.
+    if "insecure" in defaults:
+        die("insecure may not be a default — set it on the one lane that needs it")
     seen: set[str] = set()
     lanes = []
     for i, raw in enumerate(cfg["lanes"]):
@@ -515,6 +521,21 @@ def validate_config(cfg: dict, where: str) -> dict:
         bad = EXTRA_BODY_DENY & set((lane.get("extra_body") or {}).keys())
         if bad:
             die(f"{name}: extra_body may not set {', '.join(sorted(bad))}")
+
+        # TLS (#78). A default ca_bundle is fine -- a cli/acp lane simply has no
+        # TLS to apply it to -- but naming one on the lane itself is a no-op the
+        # operator meant to do something, so it fails.
+        if harness != "http":
+            for k in ("ca_bundle", "insecure"):
+                if k in raw:
+                    die(f"{name}: {k} applies to http lanes only")
+        if lane.get("ca_bundle") is not None and (
+                not isinstance(lane["ca_bundle"], str) or not lane["ca_bundle"]):
+            die(f"{name}: ca_bundle must be a path to a PEM file")
+        if lane.get("insecure") is not None and not isinstance(lane["insecure"], bool):
+            die(f"{name}: insecure must be true or false, got {lane['insecure']!r}")
+        if lane.get("insecure") and lane.get("ca_bundle"):
+            die(f"{name}: set ca_bundle or insecure, not both")
         lanes.append(lane)
     cfg["lanes"] = lanes
     return cfg
@@ -2301,6 +2322,31 @@ def parse_retry_after(value: str | None) -> float | None:
     return max(0.0, (when - dt.datetime.now(dt.timezone.utc)).total_seconds())
 
 
+def tls_context(lane: dict) -> ssl.SSLContext | None:
+    """The SSL context for one http lane, or None for urllib's default (#78).
+
+    `ca_bundle` trusts that PEM file *instead of* the system roots, for this lane
+    only: a private CA should not also be able to vouch for OpenRouter.
+    `insecure: true` turns verification off for this lane only, and says so on
+    every call. Neither touches any other lane, which a global
+    PYTHONHTTPSVERIFY=0 would.
+    """
+    if lane.get("insecure"):
+        print(f"warning: {lane.get('name', '?')}: TLS verification is OFF (insecure: true)",
+              file=sys.stderr)
+        ctx = ssl.create_default_context()
+        ctx.check_hostname = False
+        ctx.verify_mode = ssl.CERT_NONE
+        return ctx
+    if lane.get("ca_bundle"):
+        path = os.path.expanduser(lane["ca_bundle"])
+        try:
+            return ssl.create_default_context(cafile=path)
+        except (OSError, ssl.SSLError) as e:
+            raise RuntimeError(f"ca_bundle {path}: {e}") from None
+    return None
+
+
 def http_lane(lane: dict, prompt: str, key: str | None, retries: int) -> dict:
     body = {
         "model": lane["model"],
@@ -2319,12 +2365,13 @@ def http_lane(lane: dict, prompt: str, key: str | None, retries: int) -> dict:
     if key:
         # In-process only. Never argv -- argv is world-readable via ps.
         headers["Authorization"] = f"Bearer {key}"
+    ctx = tls_context(lane)
 
     attempt = 0
     while True:
         req = urllib.request.Request(url, data=json.dumps(body).encode(), headers=headers)
         try:
-            with urllib.request.urlopen(req, timeout=lane.get("timeout", 300)) as r:
+            with urllib.request.urlopen(req, timeout=lane.get("timeout", 300), context=ctx) as r:
                 data = json.load(r)
             # A 200 carrying an error body is how an upstream failure arrives when
             # the gateway itself succeeded -- "Upstream error from Nvidia" on a

skill/toolpanel.py

@@ -91,11 +91,12 @@ def post(lane, key, body):
     headers = {"Content-Type": "application/json"}
     if key:
         headers["Authorization"] = f"Bearer {key}"
+    ctx = rt.tls_context(lane)   # the lane's ca_bundle / insecure, as a run uses (#78)
     attempts = 6
     for attempt in range(attempts):
         try:
             req = urllib.request.Request(url, data=json.dumps(body).encode(), headers=headers)
-            with urllib.request.urlopen(req, timeout=lane.get("timeout", 300)) as r:
+            with urllib.request.urlopen(req, timeout=lane.get("timeout", 300), context=ctx) as r:
                 data = json.load(r)
             if "error" in data:
                 raise RuntimeError(str(data["error"])[:300])

tests/test_roundtable.py

@@ -206,6 +206,75 @@ def __exit__(self, *a):
         self.srv.server_close()   # else the socket leaks a ResourceWarning per test
 
 
+def make_ca_and_cert(d: Path) -> tuple[Path, Path, Path]:
+    """A throwaway CA and a 127.0.0.1 server cert it signed: ca.pem, cert.pem,
+    key.pem under d. Stands in for a private-CA endpoint (#78)."""
+    from cryptography import x509
+    from cryptography.hazmat.primitives import hashes, serialization
+    from cryptography.hazmat.primitives.asymmetric import ec
+    from cryptography.x509.oid import NameOID
+    import datetime as dt
+    import ipaddress
+
+    now = dt.datetime.now(dt.timezone.utc)
+
+    def name(cn):
+        return x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, cn)])
+
+    ca_key = ec.generate_private_key(ec.SECP256R1())
+    ca = (x509.CertificateBuilder().subject_name(name("roundtable test CA"))
+          .issuer_name(name("roundtable test CA")).public_key(ca_key.public_key())
+          .serial_number(x509.random_serial_number())
+          .not_valid_before(now - dt.timedelta(minutes=5))
+          .not_valid_after(now + dt.timedelta(days=1))
+          .add_extension(x509.BasicConstraints(ca=True, path_length=None), critical=True)
+          .add_extension(x509.KeyUsage(digital_signature=True, key_cert_sign=True,
+                                       crl_sign=True, content_commitment=False,
+                                       key_encipherment=False, data_encipherment=False,
+                                       key_agreement=False, encipher_only=False,
+                                       decipher_only=False), critical=True)
+          .add_extension(x509.SubjectKeyIdentifier.from_public_key(ca_key.public_key()),
+                         critical=False)
+          .sign(ca_key, hashes.SHA256()))
+    key = ec.generate_private_key(ec.SECP256R1())
+    cert = (x509.CertificateBuilder().subject_name(name("127.0.0.1"))
+            .issuer_name(ca.subject).public_key(key.public_key())
+            .serial_number(x509.random_serial_number())
+            .not_valid_before(now - dt.timedelta(minutes=5))
+            .not_valid_after(now + dt.timedelta(days=1))
+            .add_extension(x509.SubjectAlternativeName(
+                [x509.IPAddress(ipaddress.ip_address("127.0.0.1"))]), critical=False)
+            .add_extension(x509.BasicConstraints(ca=False, path_length=None), critical=True)
+            .add_extension(x509.ExtendedKeyUsage([x509.oid.ExtendedKeyUsageOID.SERVER_AUTH]),
+                           critical=False)
+            .add_extension(x509.AuthorityKeyIdentifier.from_issuer_public_key(
+                ca_key.public_key()), critical=False)
+            .sign(ca_key, hashes.SHA256()))
+    pem = serialization.Encoding.PEM
+    paths = d / "ca.pem", d / "cert.pem", d / "key.pem"
+    paths[0].write_bytes(ca.public_bytes(pem))
+    paths[1].write_bytes(cert.public_bytes(pem))
+    paths[2].write_bytes(key.private_bytes(pem, serialization.PrivateFormat.PKCS8,
+                                           serialization.NoEncryption()))
+    return paths
+
+
+class TLSStubServer(StubServer):
+    """StubServer behind TLS, with a cert from make_ca_and_cert()."""
+
+    def __init__(self, cert: Path, key: Path):
+        self.cert, self.key = cert, key
+
+    def __enter__(self):
+        import ssl
+        super().__enter__()
+        ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
+        ctx.load_cert_chain(self.cert, self.key)
+        self.srv.socket = ctx.wrap_socket(self.srv.socket, server_side=True)
+        self.url = self.url.replace("http://", "https://")
+        return self
+
+
 PRINTF_ANSI = "printf '\\033[32mgreen\\033[0m answer\\n'\n"
 PRINTF_EMOJI = "printf '\\xf0\\x9f\\x95\\x90 the answer\\n'\n"
 PRINTF_BANNER = "printf '>> updating\\nreal answer\\n'\n"
@@ -218,6 +287,57 @@ def write_exe(path: Path, body: str) -> Path:
 
 
 # --------------------------------------------------------------------------- #
+class TestPerLaneTLS(unittest.TestCase):
+    """#78: a private-CA endpoint is reachable without weakening any other lane."""
+
+    @classmethod
+    def setUpClass(cls):
+        cls.dir = Path(tempfile.mkdtemp())
+        cls.ca, cls.cert, cls.key = make_ca_and_cert(cls.dir)
+
+    @classmethod
+    def tearDownClass(cls):
+        shutil.rmtree(cls.dir, True)
+
+    def lane(self, url, **kw):
+        return {"name": "Private", "model": "ok", "base_url": url, "timeout": 10, **kw}
+
+    def test_without_a_bundle_the_private_ca_is_refused(self):
+        with TLSStubServer(self.cert, self.key) as s:
+            with self.assertRaises(RuntimeError) as e:
+                rt.http_lane(self.lane(s.url), "hi", None, 0)
+        self.assertIn("CERTIFICATE_VERIFY_FAILED", str(e.exception))
+
+    def test_ca_bundle_reaches_the_private_endpoint(self):
+        with TLSStubServer(self.cert, self.key) as s:
+            out = rt.http_lane(self.lane(s.url, ca_bundle=str(self.ca)), "hi", None, 0)
+        self.assertTrue(out["answer"])
+
+    def test_ca_bundle_replaces_the_system_roots_for_that_lane(self):
+        """A private CA must not also vouch for public hosts, so the bundle is
+        the lane's whole trust store, not an addition to it."""
+        ctx = rt.tls_context({"ca_bundle": str(self.ca)})
+        self.assertEqual(len(ctx.get_ca_certs()), 1)
+
+    def test_a_missing_bundle_names_the_path(self):
+        with self.assertRaises(RuntimeError) as e:
+            rt.http_lane(self.lane("https://127.0.0.1:9/v1",
+                                   ca_bundle=str(self.dir / "nope.pem")), "hi", None, 0)
+        self.assertIn("nope.pem", str(e.exception))
+
+    def test_insecure_connects_and_says_so(self):
+        import contextlib
+        import io
+        err = io.StringIO()
+        with TLSStubServer(self.cert, self.key) as s, contextlib.redirect_stderr(err):
+            out = rt.http_lane(self.lane(s.url, insecure=True), "hi", None, 0)
+        self.assertTrue(out["answer"])
+        self.assertIn("Private: TLS verification is OFF", err.getvalue())
+
+    def test_no_tls_key_means_urllib_default(self):
+        self.assertIsNone(rt.tls_context({"name": "A"}))
+
+
 class TestConfigValidation(unittest.TestCase):
     """Bad config must be rejected up front, not discovered mid-run."""
 
@@ -314,6 +434,41 @@ def test_extra_body_cannot_override_the_request(self):
             "extra_body may not set",
         )
 
+    def test_tls_keys_are_checked(self):
+        """#78: ca_bundle and insecure are validated before any lane runs."""
+        http = {"harness": "http", "model": "m", "base_url": "https://x/v1"}
+        cli = {"harness": "cli", "command": "x", "args": ["{prompt}"]}
+        for lane, needle in (
+            ({**http, "ca_bundle": 3}, "ca_bundle must be a path"),
+            ({**http, "ca_bundle": ""}, "ca_bundle must be a path"),
+            ({**http, "insecure": "yes"}, "insecure must be true or false"),
+            ({**http, "insecure": True, "ca_bundle": "/ca.pem"}, "not both"),
+            ({**cli, "ca_bundle": "/ca.pem"}, "applies to http lanes only"),
+            ({**cli, "insecure": False}, "applies to http lanes only"),
+        ):
+            with self.subTest(lane=lane):
+                with self.assertRaises(SystemExit) as e:
+                    rt.validate_config({"lanes": [{"name": "A", **lane}]}, "test")
+                self.assertIn(needle, str(e.exception))
+
+    def test_insecure_can_never_be_a_default(self):
+        """It would quietly turn verification off for every lane (#78)."""
+        with self.assertRaises(SystemExit) as e:
+            rt.validate_config(
+                {"defaults": {"insecure": True},
+                 "lanes": [{"name": "A", "harness": "http", "model": "m",
+                            "base_url": "https://x/v1"}]}, "test")
+        self.assertIn("insecure may not be a default", str(e.exception))
+
+    def test_a_default_ca_bundle_skips_cli_lanes(self):
+        cfg = rt.validate_config(
+            {"defaults": {"ca_bundle": "/ca.pem"},
+             "lanes": [{"name": "A", "harness": "cli", "command": "x",
+                        "args": ["{prompt}"]},
+                       {"name": "B", "harness": "http", "model": "m",
+                        "base_url": "https://x/v1"}]}, "test")
+        self.assertEqual(cfg["lanes"][1]["ca_bundle"], "/ca.pem")
+
     def test_cli_lane_must_actually_deliver_the_prompt(self):
         """#20: args non-empty was the only check, so a lane could send nothing.