Adopt canonical flex-auth credential checks
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-5.6-sol
Assistant-Session: 01a02e56-e4ad-71a2-b3e2-b6193e0d8093
This commit is contained in:
codex 2026-08-23 14:03:40 +02:00
parent 8ae77ca006
commit c9d02147d3
8 changed files with 338 additions and 50 deletions

View file

@ -9,6 +9,7 @@ import tempfile
import unittest
from argparse import Namespace
from pathlib import Path
from unittest import mock
REPO_DIR = Path(__file__).resolve().parents[1]
SPEC = importlib.util.spec_from_file_location(
@ -50,6 +51,139 @@ def sample_grant() -> dict:
class CredentialHelperTests(unittest.TestCase):
def test_flex_auth_request_uses_canonical_contract_and_numeric_ttl(self) -> None:
payload = credential.flex_auth_check_request(
grant=sample_grant(),
ttl="15m",
purpose="flex-auth-openbao-smoke",
delivery="exec-env",
actor="codex:operator",
actor_type="approved-agent",
subject="agent:codex/railiance-platform",
)
self.assertEqual(payload["tenant"], "tenant:platform")
self.assertEqual(
payload["subject"],
{"id": "agent:codex/railiance-platform", "type": "Agent"},
)
self.assertEqual(payload["action"], "issue")
self.assertEqual(
payload["resource"],
{
"id": "credential-grant:ops-warden/warden-sign",
"type": "credential-grant",
"system": "railiance-platform",
},
)
self.assertEqual(payload["context"]["requested_ttl_seconds"], 900)
self.assertEqual(
payload["context"]["bound_subject"],
"agent:codex/railiance-platform",
)
def test_authorize_request_reads_decision_envelope(self) -> None:
args = Namespace(
decision_id=None,
dry_run=False,
flex_auth_url="http://127.0.0.1:19099",
flex_auth_path="/v1/check",
require_flex_auth=True,
actor="codex:operator",
actor_type="approved-agent",
subject="agent:codex/railiance-platform",
http_timeout=1.0,
)
response = {
"id": "decision:credential-grant-allow",
"effect": "allow",
"reason": "credential_grant_allowed",
"subject": {
"id": "agent:codex/railiance-platform",
"type": "Agent",
},
"resource": {
"id": "credential-grant:ops-warden/warden-sign",
"type": "credential-grant",
"system": "railiance-platform",
},
"provenance": {"evaluator": "standalone", "mode": "embedded"},
}
with mock.patch.object(credential, "post_json", return_value=response) as post:
result = credential.authorize_request(
args=args,
grant=sample_grant(),
ttl="15m",
purpose="flex-auth-openbao-smoke",
delivery="exec-env",
)
self.assertEqual(result.mode, "flex-auth")
self.assertEqual(result.decision_id, "decision:credential-grant-allow")
endpoint, payload = post.call_args.args
self.assertEqual(endpoint, "http://127.0.0.1:19099/v1/check")
self.assertEqual(payload["context"]["requested_ttl_seconds"], 900)
def test_authorize_request_fails_closed_on_non_allow_effect(self) -> None:
args = Namespace(
decision_id=None,
dry_run=False,
flex_auth_url="http://127.0.0.1:19099",
flex_auth_path="/v1/check",
require_flex_auth=True,
actor="codex:operator",
actor_type="approved-agent",
subject="agent:codex/railiance-platform",
http_timeout=1.0,
)
with mock.patch.object(
credential,
"post_json",
return_value={
"id": "decision:deny",
"effect": "deny",
"reason": "ttl_out_of_bounds",
},
):
with self.assertRaisesRegex(SystemExit, "ttl_out_of_bounds"):
credential.authorize_request(
args=args,
grant=sample_grant(),
ttl="15m",
purpose="flex-auth-openbao-smoke",
delivery="exec-env",
)
def test_authorize_request_rejects_unbound_allow_response(self) -> None:
args = Namespace(
decision_id=None,
dry_run=False,
flex_auth_url="http://127.0.0.1:19099",
flex_auth_path="/v1/check",
require_flex_auth=True,
actor="codex:operator",
actor_type="approved-agent",
subject="agent:codex/railiance-platform",
http_timeout=1.0,
)
with mock.patch.object(
credential,
"post_json",
return_value={"id": "decision:unbound", "effect": "allow"},
):
with self.assertRaisesRegex(SystemExit, "subject does not match"):
credential.authorize_request(
args=args,
grant=sample_grant(),
ttl="15m",
purpose="flex-auth-openbao-smoke",
delivery="exec-env",
)
def test_flex_auth_default_path_is_canonical_check(self) -> None:
args = credential.build_parser().parse_args(
["request", "--purpose", "flex-auth-openbao-smoke"]
)
self.assertEqual(args.flex_auth_path, "/v1/check")
def test_database_credential_response_is_parsed_without_emission(self) -> None:
lease_id, username, password = credential.parse_database_credential(
'{"lease_id":"database/creds/runtime/lease-1","data":{"username":"leased-user","password":"leased-secret"}}'