#!/usr/bin/env python3 """ Two local experiments for "revocation latency" claims. Python 3 stdlib only. Everything runs on 127.0.0.1; nothing leaves the machine. experiment 1 stream-outlives-token A resource server validates an HS256 JWT (signature + exp) at request admission, then serves an SSE stream. The token expires 5 seconds in. The demonstration: events keep arriving after exp, because expiry is checked when the request is admitted, not while the response is open. experiment 2 pdp-in-path Measures the per-call cost of asking a policy decision point before every tool call, AuthZEN-shaped: POST /access/v1/evaluation over HTTP/1.1 keep-alive to a local stub returning {"decision": true}. Reports p50/p95/p99 for: in-process check, HTTP keep-alive PDP, HTTP no-keep-alive PDP. Localhost numbers are a floor, not a claim about your network. Usage: python3 part2-experiments.py """ import base64, hmac, hashlib, json, socket, socketserver, threading, time SECRET = b"demo-signing-key" def b64u(b): return base64.urlsafe_b64encode(b).rstrip(b"=") def mint(exp_in): h = b64u(json.dumps({"alg":"HS256","typ":"JWT"}).encode()) p = b64u(json.dumps({"sub":"agent-7","iat":int(time.time()), "exp":int(time.time())+exp_in}).encode()) sig = b64u(hmac.new(SECRET, h+b"."+p, hashlib.sha256).digest()) return (h+b"."+p+b"."+sig).decode() def validate(tok, leeway=0): try: h,p,s = tok.encode().split(b".") if not hmac.compare_digest(s, b64u(hmac.new(SECRET,h+b"."+p,hashlib.sha256).digest())): return False, "bad signature" claims = json.loads(base64.urlsafe_b64decode(p+b"==")) if time.time() >= claims["exp"] + leeway: return False, "expired" return True, claims except Exception as e: return False, str(e) # ---------- experiment 1 ---------- class SSEHandler(socketserver.StreamRequestHandler): def handle(self): req = self.rfile.readline().decode() tok = None for line in iter(self.rfile.readline, b"\r\n"): if line.lower().startswith(b"authorization: bearer "): tok = line.split(b" ",2)[2].strip().decode() ok, why = validate(tok) # <-- the only auth check if not ok: self.wfile.write(b"HTTP/1.1 401 Unauthorized\r\n\r\n"); return self.wfile.write(b"HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\n\r\n") for i in range(15): # 1 event/second for 15 s self.wfile.write(f"data: event {i} at t={i}s\n\n".encode()) self.wfile.flush(); time.sleep(1) def experiment_1(): print("== experiment 1: the stream outlives the token ==") srv = socketserver.ThreadingTCPServer(("127.0.0.1",0), SSEHandler) port = srv.server_address[1] threading.Thread(target=srv.serve_forever, daemon=True).start() tok = mint(exp_in=5) print(f" token minted, exp = now+5s; validator checks signature and exp at admission") s = socket.create_connection(("127.0.0.1",port)) s.sendall(f"GET /events HTTP/1.1\r\nHost: x\r\nAuthorization: Bearer {tok}\r\n\r\n".encode()) t0, expired_at, last_after = time.time(), None, None buf = b"" while True: chunk = s.recv(4096) if not chunk: break buf += chunk while b"\n\n" in buf: line, buf = buf.split(b"\n\n",1) if b"data:" in line: dt = time.time()-t0 ok,_ = validate(tok) if not ok and expired_at is None: expired_at = dt print(f" t={dt:4.1f}s TOKEN NOW EXPIRED, stream still open") if not ok: last_after = dt print(f" stream closed at t={time.time()-t0:.1f}s") print(f" -> events kept arriving for {last_after-expired_at+1:.0f}s after expiry;") print(f" nothing in the request path ever looked at the token again\n") srv.shutdown() # ---------- experiment 2 ---------- class PDPHandler(socketserver.StreamRequestHandler): def handle(self): while True: line = self.rfile.readline() if not line: return clen = 0 while True: h = self.rfile.readline() if h in (b"\r\n", b""): break if h.lower().startswith(b"content-length:"): clen = int(h.split(b":")[1]) if clen: self.rfile.read(clen) body = b'{"decision": true}' self.wfile.write(b"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\n" b"Content-Length: "+str(len(body)).encode()+b"\r\n\r\n"+body) self.wfile.flush() EVAL = json.dumps({"subject":{"type":"identity","id":"alice@example.com"}, "context":{"agent":"mcp-client-7"}, "action":{"name":"tools/call"}, "resource":{"type":"tool","id":"send_payment"}}).encode() def pct(xs, p): xs=sorted(xs); return xs[min(len(xs)-1, int(p/100*len(xs)))] def report(name, us): print(f" {name:<28} p50 {pct(us,50):7.0f} us p95 {pct(us,95):7.0f} us p99 {pct(us,99):7.0f} us") def experiment_2(n=2000): print("== experiment 2: cost of a PDP call per tool call (localhost floor) ==") srv = socketserver.ThreadingTCPServer(("127.0.0.1",0), PDPHandler) port = srv.server_address[1] threading.Thread(target=srv.serve_forever, daemon=True).start() grants = {("alice@example.com","tools/call","send_payment"): True} us=[] for _ in range(n): t=time.perf_counter_ns() grants.get(("alice@example.com","tools/call","send_payment"), False) us.append((time.perf_counter_ns()-t)/1000) report("in-process dict check", us) req=(b"POST /access/v1/evaluation HTTP/1.1\r\nHost: x\r\nContent-Type: application/json\r\n" b"Content-Length: "+str(len(EVAL)).encode()+b"\r\n\r\n"+EVAL) s=socket.create_connection(("127.0.0.1",port)); us=[] for _ in range(n): t=time.perf_counter_ns(); s.sendall(req) r=b"" while b'"decision"' not in r: r+=s.recv(4096) us.append((time.perf_counter_ns()-t)/1000) report("HTTP PDP, keep-alive", us); s.close() us=[] for _ in range(min(n,400)): t=time.perf_counter_ns() s=socket.create_connection(("127.0.0.1",port)); s.sendall(req) r=b"" while b'"decision"' not in r: r+=s.recv(4096) s.close(); us.append((time.perf_counter_ns()-t)/1000) report("HTTP PDP, new conn each", us) srv.shutdown() print(" -> add your real network RTT to the keep-alive line; the floor is not the story,") print(" the story is that it is a number you choose, unlike the token TTL\n") if __name__ == "__main__": experiment_1(); experiment_2()