Align authorization request digests with Go canonical wire profile
All checks were successful
CI Smoke / host-smoke (push) Successful in 0s
CI Smoke / container-smoke (push) Successful in 1s

Assistant: codex
Assistant-Model: gpt-6.1-sol
Assistant-Session: 01a126e1-76f8-7643-8cf3-e93336539ade
This commit is contained in:
tegwick 2026-10-11 19:47:45 +02:00
parent d5ff4c7421
commit 3ffb6a138e
7 changed files with 399 additions and 9 deletions

View file

@ -152,7 +152,8 @@ def test_each_lane_is_gated_separately_before_backend(bound, monkeypatch, tmp_pa
table = {entry.id: entry, lane.id: lane}
monkeypatch.setattr(cli, "get_entry", lambda _dir, cid: table[cid])
gated = []
def gate(cfg, e, action, evidence, *, fields=()):
def gate(cfg, e, action, evidence, *, fields=(), auth_args=None):
assert auth_args is args
gated.append((e.id, action, fields))
if e.id == "test-worker":
raise DeliveryError("companion lane denied")
@ -161,8 +162,9 @@ def test_each_lane_is_gated_separately_before_backend(bound, monkeypatch, tmp_pa
monkeypatch.setattr(cli, "require_delivery_state", lambda *a: None)
monkeypatch.setattr(cli, "_open_backend", lambda *a, **k: pytest.fail("no backend after a lane refusal"))
command = data["delivery_config"]["exec_owner"]["command"]
args = SimpleNamespace(field=None, catalog=entry.id, command=command, mode="exec-env")
with pytest.raises(DeliveryError, match="companion lane denied"):
cli.cmd_exec(_cfg(tmp_path), SimpleNamespace(field=None, catalog=entry.id, command=command, mode="exec-env"))
cli.cmd_exec(_cfg(tmp_path), args)
assert gated == [("test-lane", "exec", ("api_token",)), ("test-worker", "exec", ("worker_token",))]

View file

@ -0,0 +1,109 @@
"""Real Access Engine oracle, source-only; no synthetic reimplementation."""
import copy
import json
import os
from pathlib import Path
import subprocess
import pytest
from secrets_engine.authorization import request_digest, approval_binding_digest
from secrets_engine.errors import DecisionError
from secrets_engine.execution_contract import catalog_target_digest
from tests.test_action_authorization import _request
@pytest.fixture(scope="session")
def go_oracle(tmp_path_factory):
access = Path(os.environ.get("ACCESS_ENGINE_SOURCE", "/home/worsch/access-engine"))
directory = tmp_path_factory.mktemp("go-canonical")
source = directory / "main.go"
source.write_text('''package main
import ("encoding/json"; "os"; "github.com/netkingdom/access-engine/pkg/api")
func main() {
var request api.CheckRequest
if err := json.NewDecoder(os.Stdin).Decode(&request); err != nil { panic(err) }
json.NewEncoder(os.Stdout).Encode([]string{api.RequestDigest(request), api.ApprovalBindingDigest(request)})
}
''')
binary = directory / "oracle"
subprocess.run(["go", "build", "-o", str(binary), str(source)], cwd=access, check=True, timeout=120)
return binary
def _oracle(binary, request):
result = subprocess.run([str(binary)], input=json.dumps(request, ensure_ascii=False, allow_nan=False),
text=True, capture_output=True, check=True, timeout=5)
return json.loads(result.stdout)
@pytest.mark.parametrize("purpose", ["café<&>", "漢字😀", "separator\u2028paragraph\u2029", 'quoted"\\\n\t', "\\u003c", "ascii"])
def test_real_go_unicode_and_approval_join(go_oracle, purpose):
request = _request()
request["context"]["purpose"] = purpose
request["context"]["nested"] = {"ключ<&>": [purpose, None, True, False]}
expected = _oracle(go_oracle, request)
assert [request_digest(request), approval_binding_digest(request)] == expected
before = request_digest(request)
request["context"]["approval"] = {"id": "approval:é<&>", "binding": {"pdp_digest": before}}
expected = _oracle(go_oracle, request)
assert [request_digest(request), approval_binding_digest(request)] == expected
assert expected[1] == before and expected[0] != before
@pytest.mark.parametrize("number", [0, 1, -1, 1.0, 0.0, -0.0, 1.5, 1e-7, 1e-6, 1.2e-6, 1e20, 1e21, 1e23, 5e-324, 1.7976931348623157e308, 1000000000000000, 2**53-1, -(2**53-1)])
def test_numeric_admitted_profile(go_oracle, number):
request = _request()
request["context"]["number"] = {"nested": [number]}
assert request_digest(request) == _oracle(go_oracle, request)[0]
@pytest.mark.parametrize("number", [float("nan"), float("inf"), -float("inf"), 2**53, 2**53+1, 10**100])
def test_numeric_outside_profile_fails_closed(number):
request = _request()
request["context"]["number"] = number
with pytest.raises(DecisionError):
request_digest(request)
with pytest.raises(DecisionError):
approval_binding_digest(request)
def test_empty_optional_maps_and_approval_only(go_oracle):
request = _request()
request["subject"]["attributes"] = {}
request["resource"]["attributes"] = {}
request["context"] = {"approval": {"id": "test"}}
assert [request_digest(request), approval_binding_digest(request)] == _oracle(go_oracle, request)
request["purpose"] = "readback"
assert [request_digest(request), approval_binding_digest(request)] == _oracle(go_oracle, request)
def test_unicode_fingerprint_is_distinct_and_unchanged(go_oracle):
request = _request()
target = request["context"]["catalog_target"]
target["org"] = "café<&>"
import hashlib
expected = "sha256:" + hashlib.sha256(json.dumps(target, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode()).hexdigest()
assert catalog_target_digest(target) == expected
assert request_digest(request) == _oracle(go_oracle, request)[0]
def test_invalid_unicode_fails_closed():
request = _request()
request["context"]["purpose"] = "bad\ud800"
with pytest.raises(DecisionError):
request_digest(request)
def test_binary64_samples_against_real_go(go_oracle):
import random
import struct
import math
randomizer = random.Random(20261011)
for _ in range(256):
number = struct.unpack(">d", randomizer.getrandbits(64).to_bytes(8, "big"))[0]
if not math.isfinite(number):
continue
request = _request()
request["context"]["number"] = number
assert request_digest(request) == _oracle(go_oracle, request)[0], repr(number)