railiance-platform/tests/task_execution_source_servers.py

"""Real Approval/Access servers; every admission/key/identity is synthetic local.

The positive fixture policy is separately named and exact-descriptor scoped. It
never edits either existing unadmitted production task policy or their false gate.
"""
import base64
from contextlib import contextmanager
from dataclasses import replace
from datetime import datetime,timedelta,timezone
import hashlib
from http.client import HTTPConnection
from http.server import BaseHTTPRequestHandler,ThreadingHTTPServer
import json
import os
from pathlib import Path
import shutil
import socket
import subprocess
import threading
import time
from types import SimpleNamespace
from wsgiref.simple_server import make_server,WSGIRequestHandler

from cryptography import x509
from cryptography.hazmat.primitives import serialization,hashes
from cryptography.hazmat.primitives.asymmetric import ed25519,rsa
from cryptography.x509.oid import NameOID

ACCESS=Path(os.environ.get('ACCESS_ENGINE_SOURCE_ROOT','/home/worsch/access-engine'))
APPROVAL=Path(os.environ.get('APPROVAL_ENGINE_SOURCE_ROOT','/home/worsch/approval-engine'))
SECRETS=Path(os.environ.get('SECRETS_ENGINE_SOURCE_ROOT','/home/worsch/secrets-engine'))
FIXTURE_PACKAGE='platform.disposable-task-seam.fixture'
FIXTURE_VERSION='v1'
PRINCIPAL='system:serviceaccount:secrets-engine:secrets-engine'


def private(path,text):path.write_text(text);path.chmod(0o600);return path

def free_port():
    with socket.socket() as sock:sock.bind(('127.0.0.1',0));return sock.getsockname()[1]


def http(origin,method,path,body=None,token=None):
    from urllib.parse import urlsplit
    parsed=urlsplit(origin);conn=HTTPConnection(parsed.hostname,parsed.port,timeout=4)
    try:
        headers={'Content-Type':'application/json'}
        if token is not None:headers['Authorization']='Bearer '+token
        conn.request(method,path,body=json.dumps(body) if body is not None else None,headers=headers)
        response=conn.getresponse();raw=response.read(1024*1024)
        return response.status,json.loads(raw) if raw else {}
    finally:conn.close()


class QuietWSGI(WSGIRequestHandler):
    def log_message(self,*args):pass


@contextmanager
def approval_server(directory):
    from approval_engine.api import App
    from approval_engine.auth import Identity,StaticTokenAuthenticator
    from approval_engine.store import Engine
    service='synthetic-source-service';human='synthetic-source-human';no_read='synthetic-source-no-read'
    def identity(subject,kind,scopes):
        return Identity(subject,'https://synthetic.invalid',('approval-engine',),kind,'tenant:platform',
                        frozenset(),frozenset(scopes),{'level':'aal2','methods':['synthetic']},
                        'fixture:synthetic-identity',tenant_source='directory' if kind=='human' else 'registration')
    identities={service:identity('secrets-engine','service',['approval:create','approval:read','approval:consume']),
                human:identity('synthetic-founder','human',['approval:approve','approval:read']),
                no_read:identity('platform-fixture-confirmation','service',['approval:consume'])}
    engine=Engine(str(directory/'approval.sqlite'));app=App(engine,StaticTokenAuthenticator(identities))
    calls=[]
    def logged(environ,start_response):
        calls.append((environ['REQUEST_METHOD'],environ['PATH_INFO']))
        return app(environ,start_response)
    server=make_server('127.0.0.1',0,logged,handler_class=QuietWSGI)
    thread=threading.Thread(target=server.serve_forever,daemon=True);thread.start()
    try:
        yield SimpleNamespace(origin='http://127.0.0.1:'+str(server.server_port),engine=engine,
                              service_token=service,human_token=human,no_read_token=no_read,calls=calls)
    finally:server.shutdown();server.server_close();thread.join(5);engine.close()


@contextmanager
def reviewer_server(directory):
    # Actual Access TokenReview client consumes this independently admitted
    # synthetic reviewer. This is not a Kubernetes authentication proof.
    exp=int(time.time())+900
    payload=base64.urlsafe_b64encode(json.dumps({'exp':exp}).encode()).decode().rstrip('=')
    token='synthetic.'+payload+'.synthetic';reviewer='synthetic-reviewer';calls=[]
    class Handler(BaseHTTPRequestHandler):
        def log_message(self,*args):pass
        def do_POST(self):
            size=int(self.headers.get('Content-Length','0'));data=json.loads(self.rfile.read(size))
            valid=(self.path=='/apis/authentication.k8s.io/v1/tokenreviews'
                   and self.headers.get('Authorization')=='Bearer '+reviewer
                   and data.get('spec',{}).get('token')==token
                   and 'flex-auth' in data.get('spec',{}).get('audiences',[]))
            calls.append(valid)
            result={'apiVersion':'authentication.k8s.io/v1','kind':'TokenReview',
                    'status':{'authenticated':valid,'audiences':['flex-auth'],
                              'user':{'username':PRINCIPAL}}}
            raw=json.dumps(result).encode();self.send_response(201)
            self.send_header('Content-Type','application/json');self.send_header('Content-Length',str(len(raw)));self.end_headers();self.wfile.write(raw)
    server=ThreadingHTTPServer(('127.0.0.1',0),Handler);thread=threading.Thread(target=server.serve_forever,daemon=True);thread.start()
    # The actual reviewer constructor requires a valid CA even for its explicitly
    # synthetic HTTP endpoint; this CA never trusts an estate endpoint.
    key=rsa.generate_private_key(public_exponent=65537,key_size=2048)
    name=x509.Name([x509.NameAttribute(NameOID.COMMON_NAME,'synthetic-local-reviewer')])
    now=datetime.now(timezone.utc)
    cert=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(days=1)).sign(key,hashes.SHA256())
    ca=directory/'fixture-ca.pem';ca.write_bytes(cert.public_bytes(serialization.Encoding.PEM))
    reviewer_file=private(directory/'reviewer.token',reviewer)
    try:yield SimpleNamespace(origin='http://127.0.0.1:'+str(server.server_port),caller_token=token,reviewer_file=reviewer_file,ca=ca,calls=calls)
    finally:server.shutdown();server.server_close();thread.join(5)


def fixture_policy(directory,descriptor,request):
    policy=directory/'fixture-policy.md'
    fixture=directory/'fixture-cases.json'
    fixture.write_text(json.dumps([{'id':'synthetic-claim-free-deny','request':request,
                                   'expect':{'effect':'deny','reason':'fixture_unapproved'}}]))
    document='''---
id: platform.disposable-task-seam.fixture
name: Disposable exact task seam fixture
namespace: secrets-engine:secret-catalog-lane
version: v1
status: proposed
package: fixture.platform.task_seam
allow_ttl: 5m
actions: [execute-ccr]
owner: team:synthetic-fixture
fixtures: [fixture-cases.json]
caring: {profile: caring-0.4.0-rc2, enforce: false}
activation: {mode: local}
---
Synthetic local policy only. No production task policy is edited or enabled.

```rego
default decision := {"effect":"deny","reason":"fixture_unapproved"}
decision := {"effect":"allow","reason":"synthetic_exact_task_approved"} {
  input.action == "execute-ccr"
  input.tenant == "tenant:platform"
  input.subject.id == "secrets-engine"
  input.subject.type == "service"
  input.resource.system == "secrets-engine"
  input.context.human_control == true
  input.context.approval.valid_now == true
  input.context.approval.consumed == false
  input.context.approval.binding.human_control == true
  input.context.execution_contract == DESCRIPTOR
}
```
''' + '''
```rego test
import data.fixture.platform.task_seam
test_missing_approval_denies { task_seam.decision.effect == "deny" with input as {} }
```
'''
    document=document.replace('DESCRIPTOR',json.dumps(descriptor,separators=(',',':')))
    policy.write_text(document)
    return policy


@contextmanager
def access_server(directory,descriptor,request,reviewer,*,actual_policy=False):
    binary=Path(os.environ.get('PLATFORM_ACCESS_FIXTURE_BINARY','/tmp/netkingdom-platform-wave4-access'))
    verifier=Path(os.environ.get('PLATFORM_PDP_FIXTURE_VERIFIER','/tmp/netkingdom-platform-wave4-verify'))
    if not binary.is_file() or not verifier.is_file():raise RuntimeError('build source fixture binaries first')
    registry=directory/'registry.json';registry.write_bytes((ACCESS/'examples/secrets-engine/registry_snapshot.json').read_bytes())
    policy=ACCESS/'examples/secrets-engine-service-task-execution/policy_package.md' if actual_policy else fixture_policy(directory,descriptor,request)
    requestfile=directory/'claim-free-request.json';requestfile.write_text(json.dumps(request))
    # Trusted owner-local source CLI derives the real package pin independently
    # from the network responder. Its DENY is used only as digest provenance.
    offline=subprocess.run([str(binary),'check','--registry',str(registry),'--policy',str(policy),'--request',str(requestfile)],capture_output=True,text=True,timeout=15)
    if offline.returncode!=0:raise RuntimeError('source policy fixture invalid: '+offline.stderr[:300])
    produced=json.loads(offline.stdout);provenance=produced['provenance']
    key=ed25519.Ed25519PrivateKey.generate()
    seed=key.private_bytes(serialization.Encoding.Raw,serialization.PrivateFormat.Raw,serialization.NoEncryption())
    public=key.public_key().public_bytes(serialization.Encoding.Raw,serialization.PublicFormat.Raw)
    b64=lambda value:base64.urlsafe_b64encode(value).decode().rstrip('=')
    seedfile=private(directory/'signing.seed',b64(seed))
    trust={name:provenance[name] for name in ['evaluator','mode','policy_package','policy_version','policy_package_digest']}
    trust.update(schema='secrets-engine.pdp-trust.v1',keys={'synthetic-owner-key':b64(public)})
    trustfile=directory/'pdp-trust.json';trustfile.write_text(json.dumps(trust))
    port=free_port();origin='http://127.0.0.1:'+str(port)
    args=[str(binary),'serve','--addr','127.0.0.1:'+str(port),'--registry',str(registry),'--policy',str(policy),
          '--envelope-signing-key-file',str(seedfile),'--envelope-signing-kid','synthetic-owner-key',
          '--caller-auth-mode','enforce','--caller-audience','flex-auth','--caller-kubernetes-url',reviewer.origin,
          '--caller-reviewer-token-file',str(reviewer.reviewer_file),'--caller-ca-file',str(reviewer.ca),
          '--caller-binding','secrets-engine='+PRINCIPAL]
    process=subprocess.Popen(args,stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL)
    for _ in range(100):
        if process.poll() is not None:raise RuntimeError('source Access server exited')
        try:
            with socket.create_connection(('127.0.0.1',port),timeout=.1):break
        except OSError:time.sleep(.05)
    else:process.kill();process.wait();raise RuntimeError('source Access server failed to start')
    try:yield SimpleNamespace(origin=origin,trustfile=trustfile,verifier=verifier,
                              package=provenance['policy_package'],version=provenance['policy_version'],
                              process=process,offline_envelope=produced)
    finally:
        process.terminate()
        try:process.wait(timeout=5)
        except subprocess.TimeoutExpired:process.kill();process.wait()