import io import json import ssl import time import jwt import pytest from informed_decision.disposition import ActorKind from informed_decision.intake_service import IntakeError, AUDIENCE from informed_decision.workload import KubernetesTokenReview, LifecycleWorkloadIdentity, LIFECYCLE_AUDIENCE, PrivateWorkloadApp from informed_decision.runtime import Runtime from test_review_runtime import config SA = 'system:serviceaccount:state-hub:state-hub' def verified(body): return {'status': {'authenticated': True, 'audiences': body['spec']['audiences'], 'user': {'username': SA, 'uid': 'actual-sa-uid'}}} def token(exp=None): return jwt.encode({'exp': int(time.time())+300 if exp is None else exp}, 'synthetic-test-key-with-at-least-thirty-two-bytes', algorithm='HS256') def test_workload_identity_uses_verified_subject_and_separate_audience(tmp_path): path = tmp_path/'callers.json' path.write_text(json.dumps({SA: {'tenant': 'tenant:platform'}})) requests = [] def review(body): requests.append(body) return verified(body) identity = LifecycleWorkloadIdentity(review, path) caller = identity(token()) assert requests[0]['spec']['audiences'] == [LIFECYCLE_AUDIENCE] assert caller.attribution.actor.sub == SA and caller.attribution.actor.kind is ActorKind.SERVICE assert caller.expires_at <= time.time()+30 assert json.loads(caller.attributes_json)['workload_uid'] == 'actual-sa-uid' assert 'token' not in caller.attributes_json path.write_text('{}') with pytest.raises(ValueError): identity(token()) @pytest.mark.parametrize('fault', ['wrong_audience', 'unauthenticated', 'human', 'missing_uid', 'expired', 'no_expiry', 'tenant']) def test_workload_identity_fails_closed(tmp_path, fault): path = tmp_path/'callers.json' path.write_text(json.dumps({SA: {'tenant': 'tenant:other' if fault == 'tenant' else 'tenant:platform'}})) def review(body): result = verified(body) if fault == 'wrong_audience': result['status']['audiences'] = [AUDIENCE] if fault == 'unauthenticated': result['status']['authenticated'] = False if fault == 'human': result['status']['user']['username'] = 'human' if fault == 'missing_uid': result['status']['user']['uid'] = '' return result value = token(int(time.time())-1) if fault == 'expired' else token() if fault == 'no_expiry': value = jwt.encode({}, 'synthetic-test-key-with-at-least-thirty-two-bytes', algorithm='HS256') with pytest.raises(ValueError): LifecycleWorkloadIdentity(review, path)(value) def test_tokenreview_uses_owned_ca_rotating_caller_and_bounded_transport(tmp_path, monkeypatch): seen = [] sentinel = ssl.create_default_context() def context(*, cafile): assert cafile == str(tmp_path/'ca.crt') return sentinel monkeypatch.setattr(ssl, 'create_default_context', context) current = ['synthetic-caller-one'] class Transport: def __init__(self, **kwargs): assert kwargs['ssl_context'] is sentinel def request(self, method, url, **kwargs): seen.append((method, url, kwargs)) return 201, verified(json.loads(kwargs['body'])) review = KubernetesTokenReview('https://kubernetes.default.svc:443', tmp_path/'ca.crt', lambda: current[0], transport_factory=Transport) body = {'spec': {'token': 'synthetic-submitter', 'audiences': [AUDIENCE]}} assert review(body)['status']['authenticated'] current[0] = 'synthetic-caller-two' review(body) assert seen[0][2]['headers']['Authorization'] != seen[1][2]['headers']['Authorization'] assert seen[0][1] == 'https://kubernetes.default.svc:443/apis/authentication.k8s.io/v1/tokenreviews' def fail(**kwargs): raise OSError('private-path-or-response') monkeypatch.setattr(ssl, 'create_default_context', fail) with pytest.raises(IntakeError, match='workload_identity_unavailable') as error: review(body) assert 'private-path' not in str(error.value) def test_explicit_runtime_configuration_does_not_read_custody_or_start_listener(tmp_path): path, data = config(tmp_path) data['lifecycle_policy'] = {**data['policy'], 'package': 'informed-decision.lifecycle'} data['workload_api'] = {'token_review': {'origin': 'https://kubernetes.default.svc:443', 'ca_file': str(tmp_path/'absent-ca'), 'token_file': str(tmp_path/'absent-token')}, 'lifecycle_callers_file': str(tmp_path/'absent-callers'), 'intake': {'secret_file': str(tmp_path/'absent-secret'), 'submitters_file': str(tmp_path/'absent-submitters'), 'groups_file': str(tmp_path/'absent-groups')}} data['intake_attributes'] = True path.write_text(json.dumps(data)) runtime = Runtime.from_file(path) assert isinstance(runtime.workload, PrivateWorkloadApp) assert runtime.workload.intake.store is runtime.controller.store assert runtime.workload.lifecycle.service is runtime.lifecycle assert runtime.controller.intake_attributes del data['lifecycle_policy'] path.write_text(json.dumps(data)) with pytest.raises(ValueError): Runtime.from_file(path) def test_private_router_does_not_expose_browser_or_unknown_routes(): calls = [] def app(environ, start): calls.append(environ['PATH_INFO']); start('200 OK', []); return [b'ok'] router = PrivateWorkloadApp(intake=app, lifecycle=app) for path in ('/v1/memos', '/v1/decisions', '/v1/decisions/example/events'): assert b''.join(router({'PATH_INFO': path}, lambda s, h: None)) == b'ok' for path in ('/', '/auth/start', '/review', '/v1/decisionsevil'): statuses = [] router({'PATH_INFO': path}, lambda s, h: statuses.append(s)) assert statuses == ['404 Not Found'] assert len(calls) == 3 def test_two_listener_startup_failure_closes_first_socket_and_dispatcher(monkeypatch): import waitress from informed_decision.web import serve_with_workload calls = [] class Server: def close(self): calls.append('close') def shutdown(self): calls.append('shutdown') @property def task_dispatcher(self): return self def create(app, **kwargs): calls.append(kwargs) if kwargs['port'] == 8081: raise OSError('occupied') return Server() monkeypatch.setattr(waitress, 'create_server', create) with pytest.raises(OSError): serve_with_workload(object(), object(), host='127.0.0.1') assert calls[0]['map'] is calls[1]['map'] assert calls[-2:] == ['close', 'shutdown'] def test_real_tls_tokenreview_requires_ca_and_hostname(tmp_path): from datetime import datetime, timedelta, timezone from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import ipaddress import threading from cryptography import x509 from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric import rsa from cryptography.x509.oid import NameOID key = rsa.generate_private_key(public_exponent=65537, key_size=2048) now = datetime.now(timezone.utc) name = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, 'synthetic-api')]) certificate = (x509.CertificateBuilder().subject_name(name).issuer_name(name).public_key(key.public_key()) .serial_number(x509.random_serial_number()).not_valid_before(now-timedelta(minutes=1)) .not_valid_after(now+timedelta(hours=1)) .add_extension(x509.SubjectAlternativeName([x509.IPAddress(ipaddress.ip_address('127.0.0.1'))]), critical=False) .sign(key, hashes.SHA256())) ca, private_key = tmp_path/'ca.pem', tmp_path/'key.pem' ca.write_bytes(certificate.public_bytes(serialization.Encoding.PEM)) private_key.write_bytes(key.private_bytes(serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, serialization.NoEncryption())) received = [] class Handler(BaseHTTPRequestHandler): def log_message(self, *args): pass def do_POST(self): received.append(self.path) assert self.headers['Authorization'] == 'Bearer synthetic-reviewer' body = json.loads(self.rfile.read(int(self.headers['Content-Length']))) raw = json.dumps(verified(body)).encode() self.send_response(201); self.end_headers(); self.wfile.write(raw) server = ThreadingHTTPServer(('127.0.0.1', 0), Handler) tls = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) tls.load_cert_chain(ca, private_key) server.socket = tls.wrap_socket(server.socket, server_side=True) thread = threading.Thread(target=server.serve_forever, daemon=True); thread.start() body = {'spec': {'token': 'synthetic-caller', 'audiences': [LIFECYCLE_AUDIENCE]}} try: origin = f'https://127.0.0.1:{server.server_port}' review = KubernetesTokenReview(origin, ca, lambda: 'synthetic-reviewer') assert review(body)['status']['authenticated'] wrong_host = KubernetesTokenReview(origin.replace('127.0.0.1', 'localhost'), ca, lambda: 'synthetic-reviewer') with pytest.raises(IntakeError): wrong_host(body) untrusted = KubernetesTokenReview(origin, '/etc/ssl/certs/ca-certificates.crt', lambda: 'synthetic-reviewer') with pytest.raises(IntakeError): untrusted(body) assert len(received) == 1 finally: server.shutdown(); server.server_close(); thread.join(timeout=5)