import base64 import copy import json from datetime import datetime, timedelta, timezone from pathlib import Path import pytest from cryptography.exceptions import InvalidSignature from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat from hub_core.security.policy import go_json, parse_json, submitted_digest, verify_decision, verify_signature FIXTURES = Path(__file__).parent / 'fixtures/flex-auth' def test_real_go_signer_fixture_and_tampered_pair(): keys = parse_json((FIXTURES / 'keys.json').read_bytes()) verify_signature(parse_json((FIXTURES / 'decision_rotate_signed.json').read_bytes()), keys) with pytest.raises(InvalidSignature): verify_signature(parse_json((FIXTURES / 'decision_rotate_signed_tampered.json').read_bytes()), keys) def test_go_submitted_digest_known_answer(): request = parse_json((FIXTURES / 'check_request_allow_rotate.json').read_bytes()) envelope = parse_json((FIXTURES / 'decision_rotate_signed.json').read_bytes()) assert submitted_digest(request) == envelope['binding']['submitted_request_digest'] def case(): now = datetime.now(timezone.utc) request = {'id': 'request:1', 'tenant': 'tenant:platform', 'subject': {'id': 'root-sub', 'type': 'human', 'tenant': 'tenant:platform'}, 'action': 'hub.read', 'resource': {'id': '/docs', 'type': 'hub-route', 'system': 'hub-core', 'tenant': 'tenant:platform'}, 'context': {'http_request_digest': 'request-hash', 'root_entitled': True}} envelope = {'id': 'decision:1', 'contract_version': 'flex-auth.decision-record.v1', 'request_id': request['id'], 'effect': 'allow', 'resource': copy.deepcopy(request['resource']), 'subject': copy.deepcopy(request['subject']), 'binding': {**copy.deepcopy(request), 'submitted_request_digest': submitted_digest(request)}, 'lifetime': {'kind': 'ttl', 'not_before': now.isoformat(), 'expires_at': (now+timedelta(seconds=300)).isoformat()}, 'provenance': {'policy_version': 'v1', 'policy_package_digest': 'sha256:'+'a'*64, 'decision_time': now.isoformat(), 'caller': { 'mode': 'enforce', 'principal': 'workload:hub', 'audience': 'flex-auth', 'not_after': (now+timedelta(seconds=300)).isoformat()}}} return request, envelope, now def sign(envelope): key = Ed25519PrivateKey.generate() encode = lambda value: base64.urlsafe_b64encode(value).decode().rstrip('=') envelope.pop('signature', None) signature = key.sign(go_json(envelope)) envelope['signature'] = {'mode': 'signed', 'alg': 'ed25519', 'kid': 'test', 'value': encode(signature)} return {'keys': [{'kid': 'test', 'alg': 'ed25519', 'public_key': encode(key.public_key().public_bytes(Encoding.Raw, PublicFormat.Raw))}]} def test_signed_bound_allow(): request, envelope, now = case() decision = verify_decision(envelope, request=request, keys=sign(envelope), caller='workload:hub', now=now) assert decision.allowed @pytest.mark.parametrize('mutation', [ lambda d: d.update(effect='redact'), lambda d: d.update(request_id='other'), lambda d: d['binding'].update(submitted_request_digest='sha256:wrong'), lambda d: d['binding'].update(action='destroy'), lambda d: d['binding']['subject'].update(id='other'), lambda d: d['binding']['resource'].update(tenant='tenant:other'), lambda d: d['binding']['resource'].update(id='/secrets'), lambda d: d['binding']['context'].update(root_entitled=False), lambda d: d['lifetime'].update(expires_at='2000-01-01T00:00:00Z'), lambda d: d['lifetime'].update(not_before='2099-01-01T00:00:00Z'), lambda d: d['provenance'].update(decision_time='2000-01-01T00:00:00Z'), lambda d: d['provenance']['caller'].update(mode='warn'), lambda d: d['provenance']['caller'].update(principal='other'), lambda d: d.update(obligations=[{'type': 'approval'}]), ]) def test_even_authentically_signed_wrong_decisions_are_rejected(mutation): request, envelope, now = case() mutation(envelope) with pytest.raises(ValueError): verify_decision(envelope, request=request, keys=sign(envelope), caller='workload:hub', now=now) def test_unsigned_untrusted_and_ambiguous_inputs_fail(): request, envelope, now = case() keys = sign(envelope) for wrong in ({'keys': []}, {'keys': keys['keys']*2}): with pytest.raises(ValueError): verify_decision(envelope, request=request, keys=wrong, caller='workload:hub', now=now) envelope['signature'] = {'mode': 'unsigned'} with pytest.raises(ValueError): verify_signature(envelope, keys) for raw in ['{"effect":"deny","effect":"allow"}', '{"x":NaN}', '{"x":1.1}']: with pytest.raises(ValueError): parse_json(raw) def test_policy_client_rotates_workload_credentials_and_retains_no_user_token(tmp_path): import asyncio import time import httpx from hub_core.security.boundary import Actor, Authorization, LiveFacts from hub_core.security.policy import FlexPolicy from hub_core.security.identity import AccessFailure token_file, keys_file = tmp_path/'caller', tmp_path/'keys' token_file.write_text('first-workload-token') seen = [] unavailable = False caller = 'system:serviceaccount:hub-core:hub-core' actor = Actor('https://issuer.example', 'root-sub', 'tenant:platform', 'human', 'aal2', int(time.time()), int(time.time())+300) facts = LiveFacts(actor.issuer, actor.subject, actor.tenant, 'tenant:platform', True, True, True, True, time.time(), 'owner:receipt') authorization = Authorization(actor, 'hub.read', '/docs', facts, 'request:1', 'body-hash') def handle(request): seen.append(request.headers['authorization']) if unavailable: return httpx.Response(503, text='sensitive backend details') check = json.loads(request.content) _, envelope, _ = case() envelope['subject'], envelope['resource'] = check['subject'], check['resource'] envelope['binding'] = {k: v for k, v in check.items() if k != 'id'} envelope['binding']['submitted_request_digest'] = submitted_digest(check) envelope['provenance']['caller']['principal'] = caller # Key publication precedes this call in reality; write the fixture before # evaluation and sign with the corresponding ephemeral test key below. envelope.pop('signature', None) encode = lambda v: base64.urlsafe_b64encode(v).decode().rstrip('=') envelope['signature'] = {'mode': 'signed', 'alg': 'ed25519', 'kid': 'test', 'value': encode(signing_key.sign(go_json(envelope)))} assert 'token' not in request.content.decode() return httpx.Response(200, content=go_json(envelope)) signing_key = Ed25519PrivateKey.generate() keys_file.write_text(json.dumps({'keys': [{'kid': 'test', 'alg': 'ed25519', 'public_key': base64.urlsafe_b64encode(signing_key.public_key().public_bytes( Encoding.Raw, PublicFormat.Raw)).decode().rstrip('=')}]})) async def run(): nonlocal unavailable async with httpx.AsyncClient(transport=httpx.MockTransport(handle)) as client: policy = FlexPolicy(base_url='https://policy.example', client=client, caller=caller, caller_token_file=token_file, trusted_keys_file=keys_file) assert (await policy.evaluate(authorization)).allowed token_file.write_text('second-workload-token') assert (await policy.evaluate(authorization)).allowed unavailable = True with pytest.raises(AccessFailure) as error: await policy.evaluate(authorization) assert error.value.status == 503 assert 'sensitive' not in str(error.value) asyncio.run(run()) assert seen == ['Bearer first-workload-token', 'Bearer second-workload-token', 'Bearer second-workload-token'] def test_signed_decision_is_retained_exactly_and_sensitive_fields_are_refused(): request,envelope,now = case() keys = sign(envelope) result = verify_decision(envelope,request=request,keys=keys,caller='workload:hub',now=now) assert result.signed_envelope == go_json(envelope).decode() verify_signature(parse_json(result.signed_envelope),keys) envelope['diagnostics'] = {'access_token':'must-not-be-archived'} with pytest.raises(ValueError,match='sensitive decision field'): verify_decision(envelope,request=request,keys=sign(envelope),caller='workload:hub',now=now)