From 83f529841b9b3bd04bdcaafa5a6de20cd9155efa Mon Sep 17 00:00:00 2001
From: Dave Mihalcik <dmihalcik@virtru.com>
Date: Thu, 17 Sep 2026 10:15:16 -0400
Subject: [PATCH 1/6] fix(xtest): make the KAO URI tests runnable outside CI
 dispatch

Rebased onto main after #606, #612 and #615 landed; findings those PRs already
addressed are dropped, including the km3 action pin, which #612 synchronized
with its siblings.

- Skip the km3 tests when no km3 is listening, instead of failing with
  connection-refused inside an SDK CLI. The feature gate answers "is the
  override set?", not "does a km3 exist?". (A later commit in this stack
  turns that skip into a failure.)
- Add kas-km3 (8787) to otdf-local, with the kas_uri_from_kao setting, the
  5-minute key cache and debug logging its CI step uses, so all three km3
  tests are runnable locally.
- Document that kas_uri_from_kao is forced via XT_FORCE_PLATFORM_SUPPORTS
  (not XT_FORCE_SUPPORTS) and that km3 starts on the PR gate and nightlies
  while every test behind the gate skips.
---
 otdf-local/AGENTS.md                      |  6 +++++-
 otdf-local/README.md                      |  9 +++++++++
 otdf-local/src/otdf_local/cli.py          |  3 ++-
 otdf-local/src/otdf_local/config/ports.py | 13 +++++++++++-
 otdf-local/src/otdf_local/services/kas.py | 21 +++++++++++++++++++-
 xtest/README.md                           | 22 +++++++++++++++++++++
 xtest/fixtures/kas.py                     | 24 +++++++++++++++++++++++
 xtest/tdfs.py                             | 16 +++++++++++----
 xtest/test_abac.py                        | 20 +++++++++++++++++++
 9 files changed, 126 insertions(+), 8 deletions(-)

diff --git a/otdf-local/AGENTS.md b/otdf-local/AGENTS.md
index 0c116ef4b..ef466862f 100644
--- a/otdf-local/AGENTS.md
+++ b/otdf-local/AGENTS.md
@@ -18,7 +18,11 @@ uv run pytest --sdks go -v
 
 Auto-configured by otdf-local:
 - Keycloak: 8888, Postgres: 5432, Platform: 8080
-- KAS: alpha=8181, beta=8282, gamma=8383, delta=8484, km1=8585, km2=8686
+- KAS: alpha=8181, beta=8282, gamma=8383, delta=8484, km1=8585, km2=8686, km3=8787
+- km3 alone sets `services.kas.kas_uri_from_kao: true`; km1/km2 leave it off so
+  they serve as the negative control for that feature. km3 also runs at `debug`
+  with a 5-minute `key_cache_expiration`, matching its CI step — the cache test
+  asserts on a debug-only log line.
 
 ## Restart Procedures
 
diff --git a/otdf-local/README.md b/otdf-local/README.md
index 1a897c06a..407cf6183 100644
--- a/otdf-local/README.md
+++ b/otdf-local/README.md
@@ -183,6 +183,15 @@ otdf-local clean --keep-logs
 | kas-delta | 8484 | Subprocess | Standard KAS |
 | kas-km1 | 8585 | Subprocess | Key management KAS |
 | kas-km2 | 8686 | Subprocess | Key management KAS |
+| kas-km3 | 8787 | Subprocess | Key management KAS, `kas_uri_from_kao` enabled |
+
+`kas-km3` is the only instance started with `services.kas.kas_uri_from_kao: true`,
+so it resolves managed keys by the KAS URI in the KAO rather than by its own
+`registered_kas_uri`. km1 and km2 leave the setting off, which lets xtest use
+them as the negative control. km3 also mirrors the CI step's
+`key_cache_expiration` (5 minutes) and runs at `debug`, because
+`test_decrypt_same_kid_in_different_registries_with_cache` asserts on a
+debug-level cache-hit line.
 
 ## Configuration
 
diff --git a/otdf-local/src/otdf_local/cli.py b/otdf-local/src/otdf_local/cli.py
index 1b6c1d753..660a70b47 100644
--- a/otdf-local/src/otdf_local/cli.py
+++ b/otdf-local/src/otdf_local/cli.py
@@ -546,7 +546,7 @@ def restart(
 
     print_error(f"Unknown service: {service}")
     print_info(
-        "Valid services: docker, platform, kas-alpha, kas-beta, kas-gamma, kas-delta, kas-km1, kas-km2"
+        "Valid services: docker, platform, kas-alpha, kas-beta, kas-gamma, kas-delta, kas-km1, kas-km2, kas-km3"
     )
     raise typer.Exit(1)
 
@@ -601,6 +601,7 @@ def env(
         "delta": "KAS_DELTA_LOG_FILE",
         "km1": "KAS_KM1_LOG_FILE",
         "km2": "KAS_KM2_LOG_FILE",
+        "km3": "KAS_KM3_LOG_FILE",
     }
 
     for kas_name, env_var in kas_env_mapping.items():
diff --git a/otdf-local/src/otdf_local/config/ports.py b/otdf-local/src/otdf_local/config/ports.py
index 21d193358..ba1eee7d5 100644
--- a/otdf-local/src/otdf_local/config/ports.py
+++ b/otdf-local/src/otdf_local/config/ports.py
@@ -22,6 +22,7 @@ class Ports:
     KAS_DELTA: int = 8484
     KAS_KM1: int = 8585
     KAS_KM2: int = 8686
+    KAS_KM3: int = 8787
 
     # Mapping from KAS name to class attribute name
     _KAS_NAMES: ClassVar[dict[str, str]] = {
@@ -31,6 +32,7 @@ class Ports:
         "delta": "KAS_DELTA",
         "km1": "KAS_KM1",
         "km2": "KAS_KM2",
+        "km3": "KAS_KM3",
     }
 
     @classmethod
@@ -54,9 +56,18 @@ def standard_kas_names(cls) -> list[str]:
     @classmethod
     def km_kas_names(cls) -> list[str]:
         """Return key management KAS instance names."""
-        return ["km1", "km2"]
+        return ["km1", "km2", "km3"]
 
     @classmethod
     def is_km_kas(cls, name: str) -> bool:
         """Check if a KAS instance is a key management instance."""
         return name in cls.km_kas_names()
+
+    @classmethod
+    def is_kao_uri_kas(cls, name: str) -> bool:
+        """Whether this instance resolves managed keys by the KAO's KAS URI.
+
+        Only km3. km1 and km2 deliberately leave the setting off so xtest can use them as
+        the negative control for the kas_uri_from_kao feature.
+        """
+        return name == "km3"
diff --git a/otdf-local/src/otdf_local/services/kas.py b/otdf-local/src/otdf_local/services/kas.py
index 7c8afeb6f..5d6086a72 100644
--- a/otdf-local/src/otdf_local/services/kas.py
+++ b/otdf-local/src/otdf_local/services/kas.py
@@ -14,6 +14,9 @@
 from otdf_local.services.base import Service, ServiceInfo, ServiceType
 from otdf_local.utils.yaml import copy_yaml_with_updates, get_nested, load_yaml
 
+# 5 minutes, in nanoseconds -- the unit services.kas.key_cache_expiration takes.
+KM3_KEY_CACHE_EXPIRATION_NS = 300_000_000_000
+
 
 class KASService(Service):
     """Manages a single KAS instance."""
@@ -50,6 +53,11 @@ def is_key_management(self) -> bool:
         """Check if this is a key management KAS instance."""
         return Ports.is_km_kas(self._kas_name)
 
+    @property
+    def is_kao_uri(self) -> bool:
+        """Check if this instance resolves managed keys by the KAO's KAS URI."""
+        return Ports.is_kao_uri_kas(self._kas_name)
+
     def _generate_config(self) -> Path:
         """Generate the KAS config file from template."""
         config_path = self.settings.get_kas_config_path(self._kas_name)
@@ -81,6 +89,17 @@ def _generate_config(self) -> Path:
             # registered_kas_uri should NOT have /kas suffix
             updates["services.kas.registered_kas_uri"] = f"http://localhost:{self.port}"
 
+        # Off by default, and left off for km1/km2 so they stay usable as the negative
+        # control: with this unset, a KAO naming a URI other than registered_kas_uri
+        # above should fail to resolve.
+        if self.is_kao_uri:
+            updates["services.kas.kas_uri_from_kao"] = True
+            # Matches the km3 step in .github/workflows/xtest.yml. The cache test asserts
+            # a "found private key in cache" line per registry key, which is only emitted
+            # at debug and only if the entry is still live -- hence the 5-minute window.
+            updates["services.kas.key_cache_expiration"] = KM3_KEY_CACHE_EXPIRATION_NS
+            updates["logger.level"] = "debug"
+
         copy_yaml_with_updates(template_path, config_path, updates)
         return config_path
 
@@ -111,7 +130,7 @@ def start(self) -> bool:
         self.start_error = None
         # See PlatformService.start: OPENTDF_LOG_LEVEL resolved to the config key
         # "log.level", not "logger.level", so it was never read. Level belongs in
-        # the generated config.
+        # the generated config -- km3's debug level is set in _generate_config.
         self._process = self._process_manager.start(
             name=self.name,
             cmd=cmd,
diff --git a/xtest/README.md b/xtest/README.md
index e3272355c..bb9d8377e 100644
--- a/xtest/README.md
+++ b/xtest/README.md
@@ -128,6 +128,28 @@ Both reject unknown feature names. A few features are platform-only
 rejected too, because it would force the feature on for every SDK while leaving
 the platform gate the tests read untouched — a green run that tested nothing.
 
+`kas_uri_from_kao` is force-only: no release detects it, so the gate never opens
+on its own. The `force-platform-supports` input exists only on
+`workflow_dispatch` and `workflow_call`, so on the PR gate and the nightlies the
+km3 KAS still starts — its workflow step is gated on `multikas`, not on the
+feature — but **every test behind this gate skips**:
+`test_decrypt_uses_kao_kas_registration`,
+`test_decrypt_rejects_kao_kas_registration_when_disabled`, and
+`test_decrypt_same_kid_in_different_registries_with_cache`. They are
+dispatch-only until the platform release ships and the feature gets a semver
+gate in `tdfs.py`. To run them:
+
+```shell
+# CI: dispatch X-Test with force-platform-supports: kas_uri_from_kao
+# Local: otdf-local starts km3 (port 8787) with the setting enabled
+XT_FORCE_PLATFORM_SUPPORTS=kas_uri_from_kao pytest test_abac.py \
+  -k "kao_kas_registration or same_kid_in_different_registries"
+```
+
+The override only opens the test gate; the KAS must separately be started with
+`services.kas.kas_uri_from_kao: true`. If no km3 is listening the tests skip
+rather than failing against a dead port.
+
 #### Run TDF Tests
 
 ```shell
diff --git a/xtest/fixtures/kas.py b/xtest/fixtures/kas.py
index feafce43a..e75260553 100644
--- a/xtest/fixtures/kas.py
+++ b/xtest/fixtures/kas.py
@@ -6,16 +6,40 @@
 - Key management KAS instances (km1, km2, and dedicated KAO-enabled km3)
 """
 
+import logging
 import os
+import urllib.parse
+import urllib.request
 
 import pytest
 
 import abac
 from otdfctl import OpentdfCommandLineTool
 
+logger = logging.getLogger("xtest")
+
 PLATFORM_DIR = os.getenv("PLATFORM_DIR", "../../platform")
 
 
+def kas_reachable(kas_url: str) -> bool:
+    """Whether a KAS is actually listening at ``kas_url``.
+
+    The registry fixtures never touch the KAS itself -- ``kas_registry_create_if_not_present``
+    and ``_get_or_create_key`` talk only to the policy service on PLATFORMURL. So a KAS that
+    was never started stays invisible right up until a rewrap, where it surfaces as a
+    connection refused buried in an SDK CLI's stderr. Callers use this to turn that into a
+    skip that names the port.
+    """
+    parsed = urllib.parse.urlparse(kas_url)
+    url = f"{parsed.scheme}://{parsed.netloc}/healthz"
+    try:
+        with urllib.request.urlopen(url, timeout=5) as resp:
+            return resp.status == 200
+    except Exception as e:
+        logger.debug("KAS health probe at %s failed: %s", url, e)
+        return False
+
+
 def load_cached_kas_keys() -> abac.PublicKey:
     """Load RSA and EC public keys from platform directory."""
     keyset: list[abac.KasPublicKey] = []
diff --git a/xtest/tdfs.py b/xtest/tdfs.py
index 60bf26e9c..e433e2dfd 100644
--- a/xtest/tdfs.py
+++ b/xtest/tdfs.py
@@ -169,10 +169,18 @@ def is_sdk_type(val: str) -> TypeIs[sdk_type]:
     "hexless",
     "hexaflexible",
     "kasallowlist",
-    # Platform: resolve managed keys using the KAS URI from the KAO. Force-only
-    # until the first supported release is known -- via XT_FORCE_PLATFORM_SUPPORTS,
-    # since it is listed in PLATFORM_ONLY_FEATURES below. KAS also needs the
-    # setting on.
+    # Platform: resolve managed keys by the KAS URI recorded in the KAO rather than
+    # by the KAS's own ``services.kas.registered_kas_uri``.
+    #
+    # Force-only via ``XT_FORCE_PLATFORM_SUPPORTS``; it is in
+    # ``PLATFORM_ONLY_FEATURES`` below, so ``XT_FORCE_SUPPORTS`` rejects it. No
+    # release detects it yet, so nothing in ``PlatformFeatureSet`` adds it -- when
+    # one ships, drop this note and add a semver gate beside the others there.
+    #
+    # Opening the gate is necessary but not sufficient: the KAS must also run with
+    # ``services.kas.kas_uri_from_kao: true``, which only km3 does. See "Testing
+    # unreleased platform features" in xtest/README.md for why that makes these
+    # tests dispatch-only in CI.
     "kas_uri_from_kao",
     # Allow and respect assigning specific keys (kas url + key id) to attributes,
     # including splitting with multiple keys on the same kas (sdk feature),
diff --git a/xtest/test_abac.py b/xtest/test_abac.py
index a86c588db..2d4d64497 100644
--- a/xtest/test_abac.py
+++ b/xtest/test_abac.py
@@ -17,6 +17,7 @@
 )
 from audit_logs import AuditLogAsserter
 from fixtures.encryption import EncryptFactory
+from fixtures.kas import kas_reachable
 from fixtures.keys import _create_keyed_attribute, _get_or_create_key
 from otdfctl import OpentdfCommandLineTool
 from test_policytypes import skip_rts_as_needed
@@ -909,6 +910,25 @@ def test_obligations_client_scoped(
 """
 
 
+@pytest.fixture(scope="module")
+def kas_entry_km3(otdfctl: OpentdfCommandLineTool, kas_url_km3: str) -> KasEntry:
+    """KAS registry entry for the dedicated KAO-enabled key management KAS km3.
+
+    Skips unless the platform reports key_management and kas_uri_from_kao, and unless a
+    km3 is actually listening. The feature gate answers "is the override set?", not "does
+    a km3 exist?" -- see kas_reachable.
+    """
+    tdfs.get_platform_features().skip_if_unsupported(
+        "key_management", "kas_uri_from_kao"
+    )
+    if not kas_reachable(kas_url_km3):
+        pytest.skip(
+            f"no km3 KAS at {kas_url_km3}; start one (otdf-local up) "
+            "or unset XT_FORCE_PLATFORM_SUPPORTS=kas_uri_from_kao"
+        )
+    return otdfctl.kas_registry_create_if_not_present(kas_url_km3)
+
+
 @pytest.fixture(scope="module")
 def attribute_with_alternate_kas_registration(
     otdfctl: OpentdfCommandLineTool,

From 05d3d680e4d7c3b7dbc244d2046646cdff0339c5 Mon Sep 17 00:00:00 2001
From: Dave Mihalcik <dmihalcik@virtru.com>
Date: Mon, 21 Sep 2026 11:05:40 -0400
Subject: [PATCH 2/6] fix(xtest): report why a KAS probe failed, not just that
 it did

`kas_reachable` returned a bool, so a caller could only say "no km3 at
:8787". Connection refused, a proxy in the way, and a typo in KASURL7 all
collapsed to the same message, and the one that matters least -- the typo --
looks exactly like the one that matters most.

Return the reason instead, mirroring `ManagedProcess.startup_error`. Three
related fixes fall out:

- A URL that is not absolute http(s) now raises rather than being probed.
  `urlparse("localhost:8787")` yields no netloc, so the old code built
  "://healthz", failed to connect, and blamed the KAS -- sending you to
  debug a service that is running fine.
- Only `URLError`/`TimeoutError` are caught. The bare `except Exception`
  also swallowed bugs in the probe itself and reported them as a missing KAS.
- The probe ignores `http_proxy`. These are loopback services; an inherited
  proxy had it reporting on the proxy's health instead.

`/healthz` is still taken from the server root, dropping any `/kas` path on
the URL, and there is now a test pinning that.

New test_kas_units.py covers all of it offline against a throwaway HTTP
server -- no platform, no SDK.
---
 xtest/fixtures/kas.py   |  42 ++++++++++++---
 xtest/test_abac.py      |   8 +--
 xtest/test_kas_units.py | 112 ++++++++++++++++++++++++++++++++++++++++
 3 files changed, 150 insertions(+), 12 deletions(-)
 create mode 100644 xtest/test_kas_units.py

diff --git a/xtest/fixtures/kas.py b/xtest/fixtures/kas.py
index e75260553..2a425c673 100644
--- a/xtest/fixtures/kas.py
+++ b/xtest/fixtures/kas.py
@@ -8,6 +8,7 @@
 
 import logging
 import os
+import urllib.error
 import urllib.parse
 import urllib.request
 
@@ -20,24 +21,49 @@
 
 PLATFORM_DIR = os.getenv("PLATFORM_DIR", "../../platform")
 
+HEALTH_TIMEOUT_S = 5
 
-def kas_reachable(kas_url: str) -> bool:
-    """Whether a KAS is actually listening at ``kas_url``.
+
+def kas_health_error(kas_url: str) -> str | None:
+    """Describe why no KAS is answering at ``kas_url``, or None if one is.
 
     The registry fixtures never touch the KAS itself -- ``kas_registry_create_if_not_present``
     and ``_get_or_create_key`` talk only to the policy service on PLATFORMURL. So a KAS that
     was never started stays invisible right up until a rewrap, where it surfaces as a
-    connection refused buried in an SDK CLI's stderr. Callers use this to turn that into a
-    skip that names the port.
+    connection refused buried in an SDK CLI's stderr. Probing ``/healthz`` first turns that
+    into a message naming the port and the reason, which is why this returns the reason
+    rather than a bool.
+
+    ``/healthz`` is served from the root, so any path on ``kas_url`` -- ``/kas`` for most of
+    these -- is deliberately dropped.
+
+    Raises:
+        ValueError: if ``kas_url`` is not an absolute http(s) URL.
     """
     parsed = urllib.parse.urlparse(kas_url)
+    if parsed.scheme not in ("http", "https") or not parsed.netloc:
+        # Left alone, a typo'd KASURL7 becomes "://healthz", fails to connect, and
+        # reports itself as an unreachable KAS -- sending you to debug a service that
+        # is running fine. The URL is the bug, so say so.
+        raise ValueError(
+            f"not a usable KAS URL: {kas_url!r} "
+            "(expected an absolute URL, e.g. http://localhost:8787)"
+        )
+
     url = f"{parsed.scheme}://{parsed.netloc}/healthz"
+    # No proxy: these are loopback services, and an http_proxy inherited from the
+    # environment would have the probe report on the proxy's health instead.
+    opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
     try:
-        with urllib.request.urlopen(url, timeout=5) as resp:
-            return resp.status == 200
-    except Exception as e:
+        with opener.open(url, timeout=HEALTH_TIMEOUT_S) as resp:
+            if resp.status != 200:
+                return f"{url} returned HTTP {resp.status}"
+            return None
+    except (urllib.error.URLError, TimeoutError) as e:
+        # Only connection-shaped failures mean "no KAS here". Anything else is a bug
+        # in this probe and should reach the caller as one.
         logger.debug("KAS health probe at %s failed: %s", url, e)
-        return False
+        return f"{url}: {e}"
 
 
 def load_cached_kas_keys() -> abac.PublicKey:
diff --git a/xtest/test_abac.py b/xtest/test_abac.py
index 2d4d64497..67664983e 100644
--- a/xtest/test_abac.py
+++ b/xtest/test_abac.py
@@ -17,7 +17,7 @@
 )
 from audit_logs import AuditLogAsserter
 from fixtures.encryption import EncryptFactory
-from fixtures.kas import kas_reachable
+from fixtures.kas import kas_health_error
 from fixtures.keys import _create_keyed_attribute, _get_or_create_key
 from otdfctl import OpentdfCommandLineTool
 from test_policytypes import skip_rts_as_needed
@@ -916,14 +916,14 @@ def kas_entry_km3(otdfctl: OpentdfCommandLineTool, kas_url_km3: str) -> KasEntry
 
     Skips unless the platform reports key_management and kas_uri_from_kao, and unless a
     km3 is actually listening. The feature gate answers "is the override set?", not "does
-    a km3 exist?" -- see kas_reachable.
+    a km3 exist?" -- see kas_health_error.
     """
     tdfs.get_platform_features().skip_if_unsupported(
         "key_management", "kas_uri_from_kao"
     )
-    if not kas_reachable(kas_url_km3):
+    if reason := kas_health_error(kas_url_km3):
         pytest.skip(
-            f"no km3 KAS at {kas_url_km3}; start one (otdf-local up) "
+            f"no km3 KAS at {kas_url_km3} ({reason}); start one (otdf-local up) "
             "or unset XT_FORCE_PLATFORM_SUPPORTS=kas_uri_from_kao"
         )
     return otdfctl.kas_registry_create_if_not_present(kas_url_km3)
diff --git a/xtest/test_kas_units.py b/xtest/test_kas_units.py
new file mode 100644
index 000000000..cc815e262
--- /dev/null
+++ b/xtest/test_kas_units.py
@@ -0,0 +1,112 @@
+"""Offline tests for the KAS health probe in ``fixtures/kas.py``.
+
+No platform, no SDK. ``kas_health_error`` is what stands between "km3 was never
+started" and a connection-refused buried in an SDK CLI's stderr, so what matters
+is that it reports the right *reason* -- and in particular that a typo'd
+``KASURL7`` is reported as a bad URL rather than as a dead service, which would
+send you to debug a KAS that is running fine.
+"""
+
+import http.server
+import socket
+import threading
+
+import pytest
+
+from fixtures.kas import kas_health_error
+
+
+def _closed_port() -> int:
+    """A port that was bound and then released, so nothing is listening on it."""
+    with socket.socket() as s:
+        s.bind(("127.0.0.1", 0))
+        return s.getsockname()[1]
+
+
+@pytest.fixture
+def kas_stub():
+    """Start throwaway HTTP servers; returns ``(url, requested_paths)`` per call."""
+    servers: list[http.server.ThreadingHTTPServer] = []
+
+    def _start(status: int = 200) -> tuple[str, list[str]]:
+        paths: list[str] = []
+
+        class Handler(http.server.BaseHTTPRequestHandler):
+            def do_GET(self) -> None:
+                paths.append(self.path)
+                self.send_response(status)
+                self.end_headers()
+
+            def log_message(self, format: str, *args: object) -> None:
+                """Silence the default stderr access log."""
+
+        server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+        servers.append(server)
+        threading.Thread(target=server.serve_forever, daemon=True).start()
+        return f"http://127.0.0.1:{server.server_port}", paths
+
+    yield _start
+
+    for server in servers:
+        server.shutdown()
+        server.server_close()
+
+
+def test_healthy_kas_reports_no_error(kas_stub):
+    url, _ = kas_stub()
+    assert kas_health_error(url) is None
+
+
+def test_probe_hits_healthz_at_the_root_not_under_the_url_path(kas_stub):
+    """/healthz is served from the root, so a /kas suffix must not be carried over."""
+    url, paths = kas_stub()
+
+    assert kas_health_error(f"{url}/kas") is None
+    assert paths == ["/healthz"]
+
+
+def test_unreachable_kas_reports_the_url_and_the_cause():
+    reason = kas_health_error(f"http://127.0.0.1:{_closed_port()}")
+
+    assert reason is not None
+    assert "/healthz" in reason
+    # Without the cause the message is just "it didn't work", which is where the
+    # old bool-returning probe left you.
+    assert "refused" in reason.lower() or "connection" in reason.lower()
+
+
+def test_non_200_is_an_error_not_a_pass(kas_stub):
+    url, _ = kas_stub(status=204)
+
+    reason = kas_health_error(url)
+
+    assert reason is not None
+    assert "204" in reason
+
+
+def test_server_error_reports_the_status(kas_stub):
+    url, _ = kas_stub(status=503)
+
+    reason = kas_health_error(url)
+
+    assert reason is not None
+    assert "503" in reason
+
+
+@pytest.mark.parametrize(
+    "bad", ["", "   ", "localhost:8787", "/kas", "ftp://localhost:8787", "8787"]
+)
+def test_a_url_that_is_not_an_absolute_http_url_raises(bad: str):
+    """A bad KASURL7 is a bug in the environment, not evidence about the KAS."""
+    with pytest.raises(ValueError, match="not a usable KAS URL"):
+        kas_health_error(bad)
+
+
+def test_proxy_environment_is_ignored(kas_stub, monkeypatch: pytest.MonkeyPatch):
+    """These are loopback services; a proxy would have the probe report on itself."""
+    url, _ = kas_stub()
+    # Points at a port nothing is on, so if the proxy were honoured this fails.
+    monkeypatch.setenv("http_proxy", f"http://127.0.0.1:{_closed_port()}")
+    monkeypatch.setenv("HTTP_PROXY", f"http://127.0.0.1:{_closed_port()}")
+
+    assert kas_health_error(url) is None

From 8dc8041e88dad58f08f62808342f8456e6a87166 Mon Sep 17 00:00:00 2001
From: Dave Mihalcik <dmihalcik@virtru.com>
Date: Mon, 21 Sep 2026 11:07:46 -0400
Subject: [PATCH 3/6] fix(xtest): put the km3 gate in one place, and fail when
 km3 is missing

Two changes that only make sense together.

test_abac.py defined its own module-scoped `kas_entry_km3` carrying the
feature gate, shadowing the ungated one in fixtures/kas.py that
`pytest_plugins` registers. pytest resolves the module-local fixture with no
warning, so the plugin copy was dead code that still looked authoritative --
and any other module using `kas_entry_km3` would silently get the ungated
one, talking to a KAS that may not exist. There is now a single gated fixture
in fixtures/kas.py.

Past the feature gate, an unreachable km3 now fails instead of skipping. The
gate answers "was XT_FORCE_PLATFORM_SUPPORTS set?", so once it is open the
caller has asked for these tests; a km3 that isn't listening is a broken
environment, not an unsupported build. Reporting it as a second skip made a
mistyped KASURL7 read exactly like "this platform can't do it", and pytest
prints captured logs for errors but not for skips, so the skip was also the
harder of the two to diagnose afterwards. The failure message names the three
ways out.

The gate lives in `require_km3` rather than inline in the fixture so it can be
tested directly -- pytest 8.4+ will not let you call a fixture function, and
exercising this through the fixture needs installed SDKs and a live platform,
which is exactly the environment the gate exists to cope with the absence of.
---
 xtest/README.md         |  6 +++--
 xtest/fixtures/kas.py   | 25 ++++++++++++++++++
 xtest/test_abac.py      | 20 --------------
 xtest/test_kas_units.py | 58 ++++++++++++++++++++++++++++++++++++++++-
 4 files changed, 86 insertions(+), 23 deletions(-)

diff --git a/xtest/README.md b/xtest/README.md
index bb9d8377e..937b2278a 100644
--- a/xtest/README.md
+++ b/xtest/README.md
@@ -147,8 +147,10 @@ XT_FORCE_PLATFORM_SUPPORTS=kas_uri_from_kao pytest test_abac.py \
 ```
 
 The override only opens the test gate; the KAS must separately be started with
-`services.kas.kas_uri_from_kao: true`. If no km3 is listening the tests skip
-rather than failing against a dead port.
+`services.kas.kas_uri_from_kao: true`. Once the gate is open, a km3 that isn't
+listening is a **failure**, not a skip — you asked for these tests, so a missing
+km3 is a broken environment rather than an unsupported build, and skipping there
+would read identically to the feature gate being shut.
 
 #### Run TDF Tests
 
diff --git a/xtest/fixtures/kas.py b/xtest/fixtures/kas.py
index 2a425c673..8a87f5067 100644
--- a/xtest/fixtures/kas.py
+++ b/xtest/fixtures/kas.py
@@ -15,6 +15,7 @@
 import pytest
 
 import abac
+import tdfs
 from otdfctl import OpentdfCommandLineTool
 
 logger = logging.getLogger("xtest")
@@ -251,7 +252,31 @@ def kas_url_km3():
     return os.getenv("KASURL7", "http://localhost:8787")
 
 
+def require_km3(kas_url: str) -> None:
+    """Skip if the platform can't do KAO-URI lookup; fail if it can but km3 is absent.
+
+    Two gates, deliberately with different outcomes. The feature gate answers "was
+    the override set?" -- a build that doesn't do this is not a failure, so it
+    skips. Past that the caller has said to run these tests, so a km3 that isn't
+    listening is a broken environment, and reporting it as a second skip would make
+    a mistyped KASURL7 read exactly like a correct "this build can't do it". pytest
+    prints captured logs for errors but not for skips, so the skip is also the
+    harder of the two to diagnose after the fact.
+    """
+    tdfs.get_platform_features().skip_if_unsupported(
+        "key_management", "kas_uri_from_kao"
+    )
+    if reason := kas_health_error(kas_url):
+        pytest.fail(
+            f"km3 KAS is not answering ({reason}). Start one with `otdf-local up`, "
+            "point KASURL7 at it, or drop kas_uri_from_kao from "
+            "XT_FORCE_PLATFORM_SUPPORTS to skip these tests instead.",
+            pytrace=False,
+        )
+
+
 @pytest.fixture(scope="module")
 def kas_entry_km3(otdfctl: OpentdfCommandLineTool, kas_url_km3: str) -> abac.KasEntry:
     """KAS registry entry for the dedicated KAO-enabled key management KAS km3."""
+    require_km3(kas_url_km3)
     return otdfctl.kas_registry_create_if_not_present(kas_url_km3)
diff --git a/xtest/test_abac.py b/xtest/test_abac.py
index 67664983e..a86c588db 100644
--- a/xtest/test_abac.py
+++ b/xtest/test_abac.py
@@ -17,7 +17,6 @@
 )
 from audit_logs import AuditLogAsserter
 from fixtures.encryption import EncryptFactory
-from fixtures.kas import kas_health_error
 from fixtures.keys import _create_keyed_attribute, _get_or_create_key
 from otdfctl import OpentdfCommandLineTool
 from test_policytypes import skip_rts_as_needed
@@ -910,25 +909,6 @@ def test_obligations_client_scoped(
 """
 
 
-@pytest.fixture(scope="module")
-def kas_entry_km3(otdfctl: OpentdfCommandLineTool, kas_url_km3: str) -> KasEntry:
-    """KAS registry entry for the dedicated KAO-enabled key management KAS km3.
-
-    Skips unless the platform reports key_management and kas_uri_from_kao, and unless a
-    km3 is actually listening. The feature gate answers "is the override set?", not "does
-    a km3 exist?" -- see kas_health_error.
-    """
-    tdfs.get_platform_features().skip_if_unsupported(
-        "key_management", "kas_uri_from_kao"
-    )
-    if reason := kas_health_error(kas_url_km3):
-        pytest.skip(
-            f"no km3 KAS at {kas_url_km3} ({reason}); start one (otdf-local up) "
-            "or unset XT_FORCE_PLATFORM_SUPPORTS=kas_uri_from_kao"
-        )
-    return otdfctl.kas_registry_create_if_not_present(kas_url_km3)
-
-
 @pytest.fixture(scope="module")
 def attribute_with_alternate_kas_registration(
     otdfctl: OpentdfCommandLineTool,
diff --git a/xtest/test_kas_units.py b/xtest/test_kas_units.py
index cc815e262..cce444636 100644
--- a/xtest/test_kas_units.py
+++ b/xtest/test_kas_units.py
@@ -10,10 +10,12 @@
 import http.server
 import socket
 import threading
+from types import SimpleNamespace
 
 import pytest
 
-from fixtures.kas import kas_health_error
+import tdfs
+from fixtures.kas import kas_health_error, require_km3
 
 
 def _closed_port() -> int:
@@ -102,6 +104,60 @@ def test_a_url_that_is_not_an_absolute_http_url_raises(bad: str):
         kas_health_error(bad)
 
 
+def _stub_platform_features(
+    monkeypatch: pytest.MonkeyPatch, *, supported: bool
+) -> None:
+    def skip_if_unsupported(*features: str) -> None:
+        if not supported:
+            pytest.skip(f"platform does not support {features}")
+
+    monkeypatch.setattr(
+        tdfs,
+        "get_platform_features",
+        lambda: SimpleNamespace(skip_if_unsupported=skip_if_unsupported),
+    )
+
+
+def test_require_km3_skips_when_the_platform_gate_is_shut(
+    monkeypatch: pytest.MonkeyPatch,
+):
+    """A build that can't do KAO-URI lookup is not a failure."""
+    _stub_platform_features(monkeypatch, supported=False)
+
+    with pytest.raises(pytest.skip.Exception):
+        require_km3(f"http://127.0.0.1:{_closed_port()}")
+
+
+def test_require_km3_fails_when_the_gate_is_open_but_km3_is_absent(
+    monkeypatch: pytest.MonkeyPatch,
+):
+    """Past the gate the caller asked for these tests, so a missing km3 is an error.
+
+    This is the case the old skip hid: pytest prints captured logs for errors but
+    not for skips, so a mistyped KASURL7 read exactly like an unsupported build.
+    """
+    _stub_platform_features(monkeypatch, supported=True)
+
+    with pytest.raises(pytest.fail.Exception) as excinfo:
+        require_km3(f"http://127.0.0.1:{_closed_port()}")
+
+    message = str(excinfo.value)
+    assert "km3 KAS is not answering" in message
+    # The three ways out, so the reader does not have to go find them.
+    assert "otdf-local up" in message
+    assert "KASURL7" in message
+    assert "XT_FORCE_PLATFORM_SUPPORTS" in message
+
+
+def test_require_km3_passes_when_the_gate_is_open_and_km3_answers(
+    kas_stub, monkeypatch: pytest.MonkeyPatch
+):
+    _stub_platform_features(monkeypatch, supported=True)
+    url, _ = kas_stub()
+
+    require_km3(url)
+
+
 def test_proxy_environment_is_ignored(kas_stub, monkeypatch: pytest.MonkeyPatch):
     """These are loopback services; a proxy would have the probe report on itself."""
     url, _ = kas_stub()

From d705294eea0b733e287001424c3393e9b7f19bcf Mon Sep 17 00:00:00 2001
From: Dave Mihalcik <dmihalcik@virtru.com>
Date: Mon, 21 Sep 2026 11:11:39 -0400
Subject: [PATCH 4/6] fix(ci): give km3 the pqc input, and pin its config
 against otdf-local

km3 was the only KAS step in the X-Test workflow without pqc-enabled, so CI
started it without hybrid_tdf_enabled while otdf-local sets that unconditionally
for every key-management KAS. A hybrid-TDF test against km3 would therefore pass
locally and fail in CI, with nothing in either failure pointing at the config.

The new test reads the km3 step out of the workflow rather than restating its
values, and requires every input on that step to be classified as either mapped
to a config key or deliberately local-only -- so the next input added to one side
fails here instead of surfacing later as an unexplained CI-only failure.
---
 .github/workflows/xtest.yml         |   1 +
 otdf-local/tests/test_kas_config.py | 197 ++++++++++++++++++++++++++++
 2 files changed, 198 insertions(+)
 create mode 100644 otdf-local/tests/test_kas_config.py

diff --git a/.github/workflows/xtest.yml b/.github/workflows/xtest.yml
index c14af20c3..68c0ee000 100644
--- a/.github/workflows/xtest.yml
+++ b/.github/workflows/xtest.yml
@@ -774,6 +774,7 @@ jobs:
           kas-port: 8787
           log-level: debug  # The cache test verifies a hit for each registry key.
           log-type: json
+          pqc-enabled: ${{ steps.pqc-check.outputs.supported == 'true' }}
           root-key: ${{ steps.km-check.outputs.root_key }}
           dpop-challenge-enabled: ${{ inputs.dpop-challenge || false }}
 
diff --git a/otdf-local/tests/test_kas_config.py b/otdf-local/tests/test_kas_config.py
new file mode 100644
index 000000000..780e8e12c
--- /dev/null
+++ b/otdf-local/tests/test_kas_config.py
@@ -0,0 +1,197 @@
+"""km3's generated KAS config, pinned against the km3 step in CI.
+
+The KAO-URI tests run against a km3 started two different ways: by the
+`start-additional-kas` action in `.github/workflows/xtest.yml`, and by
+`otdf-local up` on a developer machine. Nothing but these tests connects the
+two, so a setting added to one side stays absent from the other until some
+test fails in CI and passes locally (or the reverse) for reasons that look
+nothing like a config drift. Reading the workflow here is deliberate: a
+hand-copied table of expected values would drift in exactly the same way.
+"""
+
+from pathlib import Path
+from typing import Any
+
+import pytest
+from otdf_local.config.features import PlatformFeatures
+from otdf_local.config.settings import Settings
+from otdf_local.services.kas import KASService
+from otdf_local.utils.yaml import get_nested, load_yaml, save_yaml
+
+WORKFLOW = Path(__file__).resolve().parents[2] / ".github/workflows/xtest.yml"
+
+ROOT_KEY = "0123456789abcdef0123456789abcdef"
+
+#: A build new enough that no feature gate in `_generate_config` trims anything.
+MODERN_PLATFORM = PlatformFeatures(
+    version="99.0.0", semver=(99, 0, 0), features={"logger_stderr"}
+)
+
+#: Inputs on the CI km3 step, and the config key otdf-local must set for each.
+#: Anything here is a setting both sides control; the tests below check that they
+#: agree on it.
+CI_INPUT_TO_CONFIG_KEY = {
+    "ec-tdf-enabled": "services.kas.preview.ec_tdf_enabled",
+    "key-management": "services.kas.preview.key_management",
+    "pqc-enabled": "services.kas.preview.hybrid_tdf_enabled",
+    "kas-uri-from-kao": "services.kas.kas_uri_from_kao",
+    "key-cache-expiration": "services.kas.key_cache_expiration",
+    "kas-port": "server.port",
+    "log-level": "logger.level",
+    "log-type": "logger.type",
+    "root-key": "services.kas.root_key",
+}
+
+#: Inputs with nothing for otdf-local to match: ``kas-name`` names the instance
+#: rather than configuring it, and the DPoP nonce tests are CI-only.
+CI_INPUTS_WITHOUT_CONFIG_KEY = {"kas-name", "dpop-challenge-enabled"}
+
+#: Inputs CI supplies as a version-gated expression rather than a literal. There is
+#: no value to compare against, so these are checked for being enabled instead --
+#: otdf-local targets a current platform and has no gate to mirror.
+CI_EXPRESSION_INPUTS = {"key-management", "pqc-enabled"}
+
+
+def _km3_step_inputs() -> dict[str, Any]:
+    """The `with:` block of the km3 step in the X-Test workflow."""
+    workflow = load_yaml(WORKFLOW)
+    for job in workflow["jobs"].values():
+        for step in job.get("steps", []):
+            if step.get("id") == "kas-km3":
+                return dict(step["with"])
+    raise AssertionError(f"no step with `id: kas-km3` in {WORKFLOW}")
+
+
+def _ci_value(ci_input: str) -> Any:
+    """One input's value, with a readable failure if CI stopped passing it.
+
+    Bare subscripting would raise a KeyError here, which reads as a broken test rather
+    than as the drift these tests exist to report.
+    """
+    inputs = _km3_step_inputs()
+    assert ci_input in inputs, f"the km3 CI step no longer passes {ci_input}"
+    return inputs[ci_input]
+
+
+def _is_expression(value: Any) -> bool:
+    return isinstance(value, str) and "${{" in value
+
+
+def _same_scalar(ci: Any, local: Any) -> bool:
+    """Compare a workflow input to a config value across YAML's scalar types.
+
+    The action takes every input as a string, so CI writes `true` where the config
+    wants a bool and quotes `'300000000000'` where it wants an int. Comparing the
+    rendered text is what the action itself effectively does.
+    """
+    return str(ci).strip().lower() == str(local).strip().lower()
+
+
+@pytest.fixture
+def settings(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Settings:
+    """A Settings pointing at a throwaway platform dir with the files KAS reads."""
+    platform_dir = tmp_path / "platform"
+    platform_dir.mkdir()
+    save_yaml(
+        platform_dir / "opentdf-dev.yaml", {"services": {"kas": {"root_key": ROOT_KEY}}}
+    )
+    save_yaml(
+        platform_dir / "opentdf-kas-mode.yaml",
+        {
+            "logger": {"level": "info"},
+            "server": {"port": 8080},
+            "services": {"kas": {}},
+        },
+    )
+
+    # PlatformFeatures.detect shells out to `go run ./service version`, which needs a
+    # real platform checkout and a Go toolchain. Feature detection is not what these
+    # tests are about, so pin it to a build that has everything.
+    monkeypatch.setattr(
+        PlatformFeatures, "detect", classmethod(lambda *_args: MODERN_PLATFORM)
+    )
+
+    settings = Settings(xtest_root=tmp_path, platform_dir=platform_dir)
+    settings.ensure_directories()
+    return settings
+
+
+def _generated(settings: Settings, kas_name: str) -> dict[str, Any]:
+    return load_yaml(KASService(settings, kas_name)._generate_config())  # noqa: SLF001
+
+
+def test_every_ci_input_is_accounted_for() -> None:
+    """A new input on the km3 step has to be classified before the rest can pass.
+
+    Without this the parity tests only cover the inputs someone remembered to list,
+    so adding a setting to CI and forgetting otdf-local would stay green.
+    """
+    assert set(_km3_step_inputs()) == set(CI_INPUT_TO_CONFIG_KEY) | (
+        CI_INPUTS_WITHOUT_CONFIG_KEY
+    )
+
+
+def test_km3_config_sets_everything_the_ci_step_sets(settings: Settings) -> None:
+    config = _generated(settings, "km3")
+    missing = [
+        key
+        for key in CI_INPUT_TO_CONFIG_KEY.values()
+        if get_nested(config, key) is None
+    ]
+    assert not missing, (
+        f"the km3 CI step sets these; the generated config does not: {missing}"
+    )
+
+
+@pytest.mark.parametrize(
+    "ci_input",
+    sorted(set(CI_INPUT_TO_CONFIG_KEY) - CI_EXPRESSION_INPUTS - {"root-key"}),
+)
+def test_km3_config_matches_the_literal_ci_values(
+    settings: Settings, ci_input: str
+) -> None:
+    ci_value = _ci_value(ci_input)
+    assert not _is_expression(ci_value), (
+        f"{ci_input} became an expression in CI; move it to CI_EXPRESSION_INPUTS"
+    )
+    local = get_nested(_generated(settings, "km3"), CI_INPUT_TO_CONFIG_KEY[ci_input])
+    assert _same_scalar(ci_value, local), (
+        f"{ci_input}: CI sets {ci_value!r}, otdf-local generates {local!r}"
+    )
+
+
+@pytest.mark.parametrize("ci_input", sorted(CI_EXPRESSION_INPUTS))
+def test_version_gated_ci_inputs_are_enabled_locally(
+    settings: Settings, ci_input: str
+) -> None:
+    """CI gates these on a platform-version check; otdf-local just turns them on."""
+    assert _is_expression(_ci_value(ci_input)), (
+        f"{ci_input} is now a literal in CI; drop it from CI_EXPRESSION_INPUTS so its "
+        "value gets compared"
+    )
+    key = CI_INPUT_TO_CONFIG_KEY[ci_input]
+    assert get_nested(_generated(settings, "km3"), key) is True
+
+
+def test_km3_root_key_comes_from_the_platform_config(settings: Settings) -> None:
+    """CI passes the platform's root key through; locally it is read off disk.
+
+    Same requirement either way -- a km3 with its own root key cannot unwrap anything
+    the platform wrapped -- but only this side can be checked against a known value.
+    """
+    assert get_nested(_generated(settings, "km3"), "services.kas.root_key") == ROOT_KEY
+
+
+@pytest.mark.parametrize("kas_name", ["km1", "km2"])
+def test_the_negative_control_kas_leave_kao_lookup_off(
+    settings: Settings, kas_name: str
+) -> None:
+    """km1/km2 must not pick up km3's settings, or the negative test proves nothing.
+
+    `test_decrypt_rejects_kao_kas_registration_when_disabled` asserts a KAO naming a
+    URI other than the KAS's own `registered_kas_uri` fails to resolve. Enable
+    kas_uri_from_kao on these and it passes for the wrong reason.
+    """
+    config = _generated(settings, kas_name)
+    assert not get_nested(config, "services.kas.kas_uri_from_kao", False)
+    assert get_nested(config, "services.kas.key_cache_expiration") is None

From 079b4660cbfa7cb23d15ec6e4862e69397e5152f Mon Sep 17 00:00:00 2001
From: Dave Mihalcik <dmihalcik@virtru.com>
Date: Mon, 21 Sep 2026 11:18:54 -0400
Subject: [PATCH 5/6] docs: name km1, not km1/km2, as the negative control

Four places said km1 and km2 together serve as the negative control for
kas_uri_from_kao. Only km1 does: test_decrypt_rejects_kao_kas_registration_when_disabled
registers a key under km1's /kas URI and requires the rewrap to fail, which
holds only while the setting stays off there. km2 is off as well, but that is
just the default and nothing asserts on it.

The distinction matters when someone changes the KAS config: turning the
setting on for km2 costs nothing, turning it on for km1 silently converts a
negative test into one that passes for the wrong reason.

Also fixes the xtest/README recipe, which told you to start km3 for a -k filter
that also selects the km1 test.
---
 otdf-local/AGENTS.md                      | 10 ++++++----
 otdf-local/README.md                      |  6 ++++--
 otdf-local/src/otdf_local/config/ports.py |  7 +++++--
 otdf-local/src/otdf_local/services/kas.py |  4 ++--
 otdf-local/tests/test_kas_config.py       | 10 ++++++----
 xtest/README.md                           |  4 +++-
 6 files changed, 26 insertions(+), 15 deletions(-)

diff --git a/otdf-local/AGENTS.md b/otdf-local/AGENTS.md
index ef466862f..c04db179a 100644
--- a/otdf-local/AGENTS.md
+++ b/otdf-local/AGENTS.md
@@ -19,10 +19,12 @@ uv run pytest --sdks go -v
 Auto-configured by otdf-local:
 - Keycloak: 8888, Postgres: 5432, Platform: 8080
 - KAS: alpha=8181, beta=8282, gamma=8383, delta=8484, km1=8585, km2=8686, km3=8787
-- km3 alone sets `services.kas.kas_uri_from_kao: true`; km1/km2 leave it off so
-  they serve as the negative control for that feature. km3 also runs at `debug`
-  with a 5-minute `key_cache_expiration`, matching its CI step — the cache test
-  asserts on a debug-only log line.
+- km3 alone sets `services.kas.kas_uri_from_kao: true`. km1 is the negative
+  control — `test_decrypt_rejects_kao_kas_registration_when_disabled` requires the
+  rewrap to fail there, which only happens while the setting stays off. km2 is
+  off too, but nothing asserts on that. km3 also runs at `debug` with a 5-minute
+  `key_cache_expiration`, matching its CI step — the cache test asserts on a
+  debug-only log line.
 
 ## Restart Procedures
 
diff --git a/otdf-local/README.md b/otdf-local/README.md
index 407cf6183..a1b770f43 100644
--- a/otdf-local/README.md
+++ b/otdf-local/README.md
@@ -187,8 +187,10 @@ otdf-local clean --keep-logs
 
 `kas-km3` is the only instance started with `services.kas.kas_uri_from_kao: true`,
 so it resolves managed keys by the KAS URI in the KAO rather than by its own
-`registered_kas_uri`. km1 and km2 leave the setting off, which lets xtest use
-them as the negative control. km3 also mirrors the CI step's
+`registered_kas_uri`. km1 leaves the setting off and is the negative control:
+`test_decrypt_rejects_kao_kas_registration_when_disabled` registers a key under
+km1's `/kas` URI and requires the rewrap to fail. km2 is off as well, but only
+because that is the default — no test depends on it. km3 also mirrors the CI step's
 `key_cache_expiration` (5 minutes) and runs at `debug`, because
 `test_decrypt_same_kid_in_different_registries_with_cache` asserts on a
 debug-level cache-hit line.
diff --git a/otdf-local/src/otdf_local/config/ports.py b/otdf-local/src/otdf_local/config/ports.py
index ba1eee7d5..749136ed7 100644
--- a/otdf-local/src/otdf_local/config/ports.py
+++ b/otdf-local/src/otdf_local/config/ports.py
@@ -67,7 +67,10 @@ def is_km_kas(cls, name: str) -> bool:
     def is_kao_uri_kas(cls, name: str) -> bool:
         """Whether this instance resolves managed keys by the KAO's KAS URI.
 
-        Only km3. km1 and km2 deliberately leave the setting off so xtest can use them as
-        the negative control for the kas_uri_from_kao feature.
+        Only km3. km1 is the negative control: xtest's
+        test_decrypt_rejects_kao_kas_registration_when_disabled registers a key under
+        km1's /kas URI and requires the rewrap to fail, which it only does while the
+        setting stays off there. km2 leaves it off as well, but that is just the default
+        -- no test asserts on it.
         """
         return name == "km3"
diff --git a/otdf-local/src/otdf_local/services/kas.py b/otdf-local/src/otdf_local/services/kas.py
index 5d6086a72..513827609 100644
--- a/otdf-local/src/otdf_local/services/kas.py
+++ b/otdf-local/src/otdf_local/services/kas.py
@@ -89,9 +89,9 @@ def _generate_config(self) -> Path:
             # registered_kas_uri should NOT have /kas suffix
             updates["services.kas.registered_kas_uri"] = f"http://localhost:{self.port}"
 
-        # Off by default, and left off for km1/km2 so they stay usable as the negative
+        # Off by default, and left off for km1 so it stays usable as the negative
         # control: with this unset, a KAO naming a URI other than registered_kas_uri
-        # above should fail to resolve.
+        # above should fail to resolve. km2 is off too, but only by default.
         if self.is_kao_uri:
             updates["services.kas.kas_uri_from_kao"] = True
             # Matches the km3 step in .github/workflows/xtest.yml. The cache test asserts
diff --git a/otdf-local/tests/test_kas_config.py b/otdf-local/tests/test_kas_config.py
index 780e8e12c..1f7b281d5 100644
--- a/otdf-local/tests/test_kas_config.py
+++ b/otdf-local/tests/test_kas_config.py
@@ -186,11 +186,13 @@ def test_km3_root_key_comes_from_the_platform_config(settings: Settings) -> None
 def test_the_negative_control_kas_leave_kao_lookup_off(
     settings: Settings, kas_name: str
 ) -> None:
-    """km1/km2 must not pick up km3's settings, or the negative test proves nothing.
+    """km1 must not pick up km3's settings, or the negative test proves nothing.
 
-    `test_decrypt_rejects_kao_kas_registration_when_disabled` asserts a KAO naming a
-    URI other than the KAS's own `registered_kas_uri` fails to resolve. Enable
-    kas_uri_from_kao on these and it passes for the wrong reason.
+    `test_decrypt_rejects_kao_kas_registration_when_disabled` asserts that a KAO naming
+    a URI other than km1's own `registered_kas_uri` fails to resolve. Enable
+    kas_uri_from_kao there and it passes for the wrong reason. km2 is covered here as
+    well: nothing asserts on it today, but it is configured from the same branch, so a
+    change that leaked the setting to one would leak it to both.
     """
     config = _generated(settings, kas_name)
     assert not get_nested(config, "services.kas.kas_uri_from_kao", False)
diff --git a/xtest/README.md b/xtest/README.md
index 937b2278a..dc5570a46 100644
--- a/xtest/README.md
+++ b/xtest/README.md
@@ -141,7 +141,9 @@ gate in `tdfs.py`. To run them:
 
 ```shell
 # CI: dispatch X-Test with force-platform-supports: kas_uri_from_kao
-# Local: otdf-local starts km3 (port 8787) with the setting enabled
+# Local: `otdf-local up` starts km3 (port 8787) with the setting enabled and km1
+# (port 8585) with it off. Both are needed -- the "when_disabled" test is the
+# negative control and runs against km1.
 XT_FORCE_PLATFORM_SUPPORTS=kas_uri_from_kao pytest test_abac.py \
   -k "kao_kas_registration or same_kid_in_different_registries"
 ```

From e13671e9324bf0b84ff42a85cc6925d619aaaa9b Mon Sep 17 00:00:00 2001
From: Dave Mihalcik <dmihalcik@virtru.com>
Date: Mon, 21 Sep 2026 16:56:53 -0400
Subject: [PATCH 6/6] test(xtest): stop the require_km3 tests racing on a
 released port

_closed_port binds an ephemeral port and releases it, so another process can
bind it before the probe connects. Neither require_km3 test cares which failure
the probe hits, so both now stub it:

- the gate-shut test asserts the probe is never called at all, which is the
  actual claim -- the feature gate has to raise before anything touches the
  network -- and is stronger than the released port it used to pass;
- the km3-absent test pins a known reason string, so it can also assert the
  reason survives into the failure message. Dropping it from the f-string in
  require_km3 previously went unnoticed.

test_unreachable_kas_reports_the_url_and_the_cause keeps the real socket: it
exists to prove a live connection-refused reaches the except clause in
kas_health_error, which a stubbed transport would reduce to asserting that an
f-string interpolates. The race there costs a flaky failure, never a false
pass. Holding the socket bound-but-unlistening would reserve the port, but on
macOS that drops the SYN rather than refusing it, so the probe times out and
reports the wrong reason.

The proxy test's URLs were never connected to -- kas_health_error builds its
opener with ProxyHandler({}) -- so those use a literal now too.
---
 xtest/test_kas_units.py | 50 +++++++++++++++++++++++++++++++++++------
 1 file changed, 43 insertions(+), 7 deletions(-)

diff --git a/xtest/test_kas_units.py b/xtest/test_kas_units.py
index cce444636..4c1166a38 100644
--- a/xtest/test_kas_units.py
+++ b/xtest/test_kas_units.py
@@ -14,12 +14,29 @@
 
 import pytest
 
+import fixtures.kas
 import tdfs
 from fixtures.kas import kas_health_error, require_km3
 
+#: For the two tests that never reach the network. Keeping them off `_closed_port`
+#: leaves it with the single caller that genuinely needs a live socket.
+UNCONNECTED_URL = "http://127.0.0.1:1"
+
 
 def _closed_port() -> int:
-    """A port that was bound and then released, so nothing is listening on it."""
+    """A port that was bound and then released, so nothing is listening on it.
+
+    Racy in principle -- another process can bind the port between the release and
+    the connect -- and deliberately kept anyway, for the one test that has to see a
+    real connection refused reach the ``except`` clause in ``kas_health_error``.
+    Stubbing the transport there would reduce it to asserting that an f-string
+    interpolates. The race costs a flaky failure, never a false pass: anything that
+    did answer on that port would fail every assertion in the test.
+
+    Holding the socket bound-but-unlistening would reserve the port, but on macOS
+    that drops the SYN instead of refusing it, so the probe times out after
+    HEALTH_TIMEOUT_S and reports the wrong reason.
+    """
     with socket.socket() as s:
         s.bind(("127.0.0.1", 0))
         return s.getsockname()[1]
@@ -121,11 +138,20 @@ def skip_if_unsupported(*features: str) -> None:
 def test_require_km3_skips_when_the_platform_gate_is_shut(
     monkeypatch: pytest.MonkeyPatch,
 ):
-    """A build that can't do KAO-URI lookup is not a failure."""
+    """A build that can't do KAO-URI lookup is not a failure.
+
+    The URL is never probed -- the gate raises first -- which is the point: reaching
+    the network here at all would mean the gate ran in the wrong order.
+    """
     _stub_platform_features(monkeypatch, supported=False)
 
+    def unreachable_probe(kas_url: str) -> str | None:
+        pytest.fail(f"probed {kas_url} despite the platform gate being shut")
+
+    monkeypatch.setattr(fixtures.kas, "kas_health_error", unreachable_probe)
+
     with pytest.raises(pytest.skip.Exception):
-        require_km3(f"http://127.0.0.1:{_closed_port()}")
+        require_km3(UNCONNECTED_URL)
 
 
 def test_require_km3_fails_when_the_gate_is_open_but_km3_is_absent(
@@ -135,14 +161,24 @@ def test_require_km3_fails_when_the_gate_is_open_but_km3_is_absent(
 
     This is the case the old skip hid: pytest prints captured logs for errors but
     not for skips, so a mistyped KASURL7 read exactly like an unsupported build.
+
+    What the probe failed *on* is covered by the kas_health_error tests above; here
+    it is stubbed, so the branch under test cannot flake on host port activity and
+    the reason can be asserted against a known string.
     """
     _stub_platform_features(monkeypatch, supported=True)
+    monkeypatch.setattr(
+        fixtures.kas, "kas_health_error", lambda _url: "nothing listening on 8787"
+    )
 
     with pytest.raises(pytest.fail.Exception) as excinfo:
-        require_km3(f"http://127.0.0.1:{_closed_port()}")
+        require_km3(UNCONNECTED_URL)
 
     message = str(excinfo.value)
     assert "km3 KAS is not answering" in message
+    # The probe's reason has to survive into the failure, or the message says only
+    # that something is wrong and not what.
+    assert "nothing listening on 8787" in message
     # The three ways out, so the reader does not have to go find them.
     assert "otdf-local up" in message
     assert "KASURL7" in message
@@ -161,8 +197,8 @@ def test_require_km3_passes_when_the_gate_is_open_and_km3_answers(
 def test_proxy_environment_is_ignored(kas_stub, monkeypatch: pytest.MonkeyPatch):
     """These are loopback services; a proxy would have the probe report on itself."""
     url, _ = kas_stub()
-    # Points at a port nothing is on, so if the proxy were honoured this fails.
-    monkeypatch.setenv("http_proxy", f"http://127.0.0.1:{_closed_port()}")
-    monkeypatch.setenv("HTTP_PROXY", f"http://127.0.0.1:{_closed_port()}")
+    # Points somewhere nothing can answer, so if the proxy were honoured this fails.
+    monkeypatch.setenv("http_proxy", UNCONNECTED_URL)
+    monkeypatch.setenv("HTTP_PROXY", UNCONNECTED_URL)
 
     assert kas_health_error(url) is None