Bind native OpenRouter approval to custody and delivery inputs

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a09cbb-87c6-7900-a145-4ce53ba9f1a6

tests/test_decision_replay.py

@@ -3,7 +3,8 @@ import copy
 import json
 from datetime import datetime, timezone
 from pathlib import Path
-from types import SimpleNamespace
+from secrets_engine.catalog import validate_entry
+from tests.test_catalog import VALID
 
 import pytest
 
@@ -122,7 +123,7 @@ def test_real_expired_allow_refused():
 
 
 def test_request_builder_tenant_boundary():
-    entry = SimpleNamespace(id="glas-primary", stage="prod")
+    entry = validate_entry({**VALID, "id": "glas-primary", "stage": "prod"})
     args = dict(subject_id="secrets-engine", subject_type="service", purpose="rotation")
     request = build_action_request(entry, "rotate", **args)
     assert request["tenant"] == REQUEST_TENANT

tests/test_openrouter_binding.py

@@ -0,0 +1,61 @@
+"""A named native lane must not hide a changed custody or delivery target."""
+import copy
+from pathlib import Path
+
+import pytest
+
+from secrets_engine.authorization import build_action_request, approval_binding_digest, validate_decision_envelope
+from secrets_engine.catalog import load_entry, validate_entry
+from secrets_engine.errors import DecisionError
+from secrets_engine.plan import build_plan
+from tests.test_action_authorization import _envelope
+
+CATALOG = Path(__file__).resolve().parents[1] / "catalog/openrouter-llm-connect.yaml"
+
+
+def request(entry):
+    return build_action_request(entry, "apply", subject_id="secrets-engine",
+        subject_type="service", request_id="check:openrouter-apply", purpose="IR-WP-0004 native OpenRouter access",
+        policy_targets=[entry.policy_name], auth_targets=[entry.role_name])
+
+
+@pytest.mark.parametrize("change", ["path", "mount", "repo", "consumer", "ttl", "uses", "field", "delivery"])
+def test_old_approval_cannot_follow_changed_native_target(change):
+    entry = load_entry(CATALOG)
+    original = request(entry)
+    data = copy.deepcopy(entry.raw)
+    if change in {"path", "mount", "repo"}: data[change] += "-other"
+    elif change == "consumer": data["consumers"][1]["name"] = "unrelated-recipient"
+    elif change == "ttl": data["delivery_auth"]["token_max_ttl"] = "24h"
+    elif change == "uses": data["delivery_auth"]["secret_id_num_uses"] = 0
+    elif change == "field": data["fields"].append("OTHER_KEY")
+    elif change == "delivery": data["delivery_modes"].append("exec-file")
+    changed = request(validate_entry(data))
+    assert approval_binding_digest(original) != approval_binding_digest(changed)
+    with pytest.raises(DecisionError, match="submitted request digest"):
+        validate_decision_envelope(_envelope(original), changed,
+            accepted_policy_packages={"secrets-engine.catalog-lane.lifecycle"},
+            accepted_policy_versions={"v2"})
+
+
+def test_issuing_approval_does_not_change_its_own_target():
+    entry = load_entry(CATALOG)
+    original = request(entry)
+    data = copy.deepcopy(entry.raw)
+    data["approval"]["authorization_id"] = "new-approval-object"
+    assert approval_binding_digest(original) == approval_binding_digest(request(validate_entry(data)))
+    entry.delivery_auth["token_max_ttl"] = "24h"
+    assert original["context"]["catalog_target"]["delivery_auth"]["token_max_ttl"] == "30m"
+
+
+def test_openrouter_plan_exposes_all_limits_and_preserves_existing_custody():
+    entry = load_entry(CATALOG)
+    plan = build_plan(entry, "prod")
+    assert [a.kind for a in plan.actions] == ["kv-mount-check", "policy", "approle"]
+    assert plan.policy_name == plan.role_name == "se-prod-openrouter-llm-connect"
+    assert entry.kv_data_path in plan.policy_hcl
+    assert plan.actions[-1].detail == {
+        "token_policies": plan.policy_name, "auth": "approle",
+        "token_ttl": "15m", "token_max_ttl": "30m", "token_num_uses": 8,
+        "secret_id_ttl": "15m", "secret_id_num_uses": 1,
+    }

tests/test_openrouter_key_check.py

@@ -0,0 +1,54 @@
+import importlib.util
+import json
+from pathlib import Path
+
+import pytest
+
+SPEC = importlib.util.spec_from_file_location("openrouter_key_check", Path(__file__).resolve().parents[1] / "tools/openrouter_key_check.py")
+probe = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(probe)
+
+
+@pytest.mark.parametrize("status,body,result", [
+    (200, b'{"data":{"label":"SYNTHETIC-KEY","usage":999}}', "authenticated"),
+    (401, b'SYNTHETIC-KEY', "refused"),
+    (302, b'SYNTHETIC-KEY', "refused"),
+    (500, b'SYNTHETIC-KEY', "refused"),
+    (200, b'bad SYNTHETIC-KEY', "check_failed"),
+    (200, b'{"data":null}', "invalid_response"),
+    (200, b'x' * (probe.MAX_BODY + 1), "invalid_response"),
+])
+def test_probe_is_one_fixed_read_and_never_returns_provider_content(status, body, result):
+    calls = []
+    class Connection:
+        def __init__(self, host, timeout):
+            assert (host, timeout) == ("openrouter.ai", 10)
+        def request(self, method, path, headers):
+            calls.append((method, path))
+            assert headers["Authorization"] == "Bearer SYNTHETIC-KEY"
+        def getresponse(self): return self
+        def read(self, size):
+            assert size == probe.MAX_BODY + 1
+            return body[:size]
+        def close(self): calls.append("closed")
+    Connection.status = status
+    output = probe.check("SYNTHETIC-KEY", connection_factory=Connection)
+    assert output["result"] == result
+    assert "SYNTHETIC-KEY" not in json.dumps(output)
+    assert calls == [("GET", "/api/v1/key"), "closed"]
+
+
+def test_transport_error_is_sanitized_and_connection_closed():
+    closed = []
+    class Broken:
+        def __init__(self, *a, **kw): pass
+        def request(self, *a, **kw): raise OSError("SYNTHETIC-KEY")
+        def close(self): closed.append(True)
+    assert probe.check("SYNTHETIC-KEY", connection_factory=Broken) == {"result": "check_failed"}
+    assert closed == [True]
+
+
+@pytest.mark.parametrize("key", ["", "key\r\nInjected: value", "key with spaces", "non-ascii-ä"])
+def test_bad_input_never_opens_connection(key):
+    def forbidden(*a, **kw): pytest.fail("must not connect")
+    assert probe.check(key, connection_factory=forbidden) == {"result": "invalid_input"}