#!/usr/bin/env python3 """Reproducible local Vault dev-mode experiment. No production credentials used.""" from __future__ import annotations import argparse from datetime import datetime, timezone import hashlib import json import os from pathlib import Path import platform import secrets import socket import subprocess import sys import time import urllib.error import urllib.request def run(binary: Path, output: Path): version = subprocess.check_output([str(binary), 'version'], text=True).strip() with socket.socket() as sock: sock.bind(('127.0.0.1', 0)) port = sock.getsockname()[1] base = f'http://127.0.0.1:{port}' root_token = secrets.token_urlsafe(32) env = {key: value for key, value in os.environ.items() if not key.startswith('VAULT_')} env.update(VAULT_DEV_ROOT_TOKEN_ID=root_token, VAULT_DEV_LISTEN_ADDRESS=f'127.0.0.1:{port}') # Never emit the server's generated root token or persist it in a token helper. process = subprocess.Popen([str(binary), 'server', '-dev', '-dev-no-store-token'], env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) checks = [] expiry_observation = {} opener = urllib.request.build_opener(urllib.request.ProxyHandler({})) def api(method, path, token=None, body=None): headers = {'Content-Type': 'application/json'} if token: headers['X-Vault-Token'] = token request = urllib.request.Request(base+'/v1/'+path, method=method, headers=headers, data=json.dumps(body).encode() if body is not None else None) try: with opener.open(request, timeout=3) as response: raw = response.read() return response.status, json.loads(raw) if raw else {} except urllib.error.HTTPError as error: # Do not serialize arbitrary response bodies that may contain tokens. return error.code, {} def expect(name, status, expected): checks.append({'check':name, 'observedHttpStatus':status, 'expectedHttpStatus':expected, 'passed':status==expected}) if status != expected: raise RuntimeError(f'{name}: expected HTTP {expected}, observed {status}') error = None try: for _ in range(50): if process.poll() is not None: raise RuntimeError('Local dev server exited before readiness') try: status, _ = api('GET', 'auth/token/lookup-self', root_token) if status == 200: break except (urllib.error.URLError, TimeoutError): pass time.sleep(0.1) else: raise RuntimeError('Local dev server did not become ready') for name in ['allowed', 'other']: status, _ = api('POST', f'secret/data/{name}', root_token, {'data':{'demo':'synthetic-lab-value'}}) expect('seed_'+name, status, 200) policy = 'path "secret/data/allowed" { capabilities = ["read"] }' status, _ = api('PUT', 'sys/policies/acl/lab-reader', root_token, {'policy':policy}) expect('install_scoped_policy', status, 204) status, issued = api('POST', 'auth/token/create', root_token, {'policies':['lab-reader'], 'no_default_policy':True, 'ttl':'60s', 'renewable':False}) expect('issue_reader', status, 200) token = issued['auth']['client_token'] status, body = api('GET', 'secret/data/allowed', token) expect('allowed_read', status, 200) if body.get('data', {}).get('data', {}).get('demo') != 'synthetic-lab-value': raise RuntimeError('Authorized read returned unexpected synthetic content') expect('other_path_denied', api('GET','secret/data/other',token)[0],403) expect('write_denied', api('POST','secret/data/allowed',token,{'data':{'demo':'changed'}})[0],403) expect('revoke_reader', api('POST','auth/token/revoke',root_token,{'token':token})[0],204) expect('read_after_revocation_denied', api('GET','secret/data/allowed',token)[0],403) status, issued = api('POST', 'auth/token/create', root_token, {'policies':['lab-reader'],'no_default_policy':True, 'ttl':'2s','explicit_max_ttl':'2s','renewable':False}) expect('issue_short_lived_reader',status,200) issued_at = time.monotonic() expiry_observation.update(requestedTtlSeconds=2, issuedTtlSeconds=issued['auth'].get('lease_duration')) short_token = issued['auth']['client_token'] expect('read_before_expiry',api('GET','secret/data/allowed',short_token)[0],200) time.sleep(3) expect('read_after_expiry_denied',api('GET','secret/data/allowed',short_token)[0],403) expiry_observation['secondsFromIssueResponseToDeniedRead'] = round(time.monotonic()-issued_at, 6) except Exception as exc: error = str(exc) finally: process.terminate() try: process.wait(timeout=5) except subprocess.TimeoutExpired: process.kill() process.wait() report = {'schemaVersion':1, 'experiment':'vault-dev-kv-access-revocation-expiry', 'observedAt':datetime.now(timezone.utc).isoformat(), 'version':version, 'binarySha256':hashlib.sha256(binary.read_bytes()).hexdigest(), 'scriptSha256':hashlib.sha256(Path(__file__).read_bytes()).hexdigest(), 'invocation':[sys.executable, *sys.argv], 'workingDirectory':str(Path.cwd()), 'expiryObservation':expiry_observation, 'platform':platform.platform(), 'architecture':platform.machine(), 'mode':'isolated dev server bound to loopback, in-memory, synthetic data', 'checks':checks, 'passed':error is None and all(c['passed'] for c in checks), 'error':error, 'serverStopped':process.poll() is not None, 'limits':['Single local dev-mode configuration; not a production deployment or security certification.', 'Revoking a token does not withdraw secret values already read. Timing brackets rejection; it does not establish the exact expiry instant.', 'Tests access-token expiry/revocation to stored KV secrets; not dynamic database credential generation, AppRole bootstrap or plugin WIF.', 'Does not measure availability, performance, audits, cloud integration or comparative superiority.']} output.parent.mkdir(parents=True,exist_ok=True) output.write_text(json.dumps(report,ensure_ascii=False,indent=2)+'\n') print(f'{"PASS" if report["passed"] else "FAIL"}: {len(checks)} observations; report {output}; server stopped.') return 0 if report['passed'] else 1 if __name__ == '__main__': parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--vault-binary',type=Path,required=True) parser.add_argument('--output',type=Path,required=True) args = parser.parse_args() raise SystemExit(run(args.vault_binary.resolve(),args.output.resolve()))