fix(xtest): post-review fixes for the km3 KAO-URI tests

.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 }}
 

otdf-local/AGENTS.md

@@ -18,7 +18,13 @@ 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 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
 

otdf-local/README.md

@@ -183,6 +183,17 @@ 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 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.
 
 ## Configuration
 

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():

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,21 @@ 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 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"

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 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. 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
+            # 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,

otdf-local/tests/test_kas_config.py

@@ -0,0 +1,199 @@
+"""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 must not pick up km3's settings, or the negative test proves nothing.
+
+    `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)
+    assert get_nested(config, "services.kas.key_cache_expiration") is None

xtest/README.md

@@ -128,6 +128,32 @@ 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 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"
+```
+
+The override only opens the test gate; the KAS must separately be started with
+`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
 
 ```shell