#!/usr/bin/env python3 """Reproducible local Vault AppRole 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) -> int: 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) demo_value = secrets.token_urlsafe(24) 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}", ) process = subprocess.Popen( [str(binary), "server", "-dev", "-dev-no-store-token"], env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) checks: list[dict[str, object]] = [] opener = urllib.request.build_opener(urllib.request.ProxyHandler({})) def api(method: str, path: str, token: str | None = None, body: dict | None = 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: # Error bodies are deliberately discarded: they are unnecessary and # could contain values that must not enter the report. return error.code, {} def expect(name: str, status: int, expected: int) -> None: 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": demo_value}}, ) expect(f"seed_{name}", status, 200) policy = 'path "secret/data/allowed" { capabilities = ["read"] }' expect( "install_scoped_policy", api("PUT", "sys/policies/acl/lab-reader", root_token, {"policy": policy})[0], 204, ) expect( "enable_approle", api("POST", "sys/auth/approle", root_token, {"type": "approle"})[0], 204, ) expect( "configure_approle", api( "POST", "auth/approle/role/lab-machine", root_token, { "token_policies": ["lab-reader"], "token_no_default_policy": True, "token_ttl": "60s", "token_max_ttl": "60s", "secret_id_num_uses": 0, }, )[0], 204, ) status, role_response = api( "GET", "auth/approle/role/lab-machine/role-id", root_token ) expect("read_role_id", status, 200) role_id = role_response["data"]["role_id"] status, secret_response = api( "POST", "auth/approle/role/lab-machine/secret-id", root_token, {} ) expect("issue_secret_id", status, 200) secret_id = secret_response["data"]["secret_id"] secret_id_accessor = secret_response["data"]["secret_id_accessor"] status, login_response = api( "POST", "auth/approle/login", body={"role_id": role_id, "secret_id": secret_id}, ) expect("login_with_approle", status, 200) issued_token = login_response["auth"]["client_token"] if "default" in login_response["auth"].get("policies", []): raise RuntimeError("AppRole login unexpectedly received the default policy") status, allowed_body = api("GET", "secret/data/allowed", issued_token) expect("allowed_read", status, 200) if allowed_body.get("data", {}).get("data", {}).get("demo") != demo_value: raise RuntimeError("Authorized read returned unexpected synthetic content") expect( "other_path_denied", api("GET", "secret/data/other", issued_token)[0], 403, ) expect( "destroy_secret_id", api( "POST", "auth/approle/role/lab-machine/secret-id-accessor/destroy", root_token, {"secret_id_accessor": secret_id_accessor}, )[0], 204, ) expect( "login_after_secret_id_destroyed_denied", api( "POST", "auth/approle/login", body={"role_id": role_id, "secret_id": secret_id}, )[0], 400, ) expect( "issued_token_still_reads", api("GET", "secret/data/allowed", issued_token)[0], 200, ) expect( "revoke_issued_token", api("POST", "auth/token/revoke", root_token, {"token": issued_token})[0], 204, ) expect( "read_after_token_revocation_denied", api("GET", "secret/data/allowed", issued_token)[0], 403, ) except Exception as exc: # The report records the failing step, never response bodies. error = str(exc) finally: process.terminate() try: process.wait(timeout=5) except subprocess.TimeoutExpired: process.kill() process.wait() report = { "schemaVersion": 1, "experiment": "vault-dev-approle-bootstrap-secret-id-and-token-revocation", "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()), "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(check["passed"] for check in checks), "error": error, "serverStopped": process.poll() is not None, "limits": [ "Single local dev-mode configuration; not a production deployment or security certification.", "Destroying a SecretID prevents its later use but does not revoke a token already issued from it.", "Revoking the issued token prevents later Vault API use; it does not withdraw values already read.", "Does not test response wrapping, SecretID delivery, limited-use SecretIDs, TLS, HA, namespaces, audit devices, performance, availability 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; ' f"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) arguments = parser.parse_args() raise SystemExit(run(arguments.vault_binary.resolve(), arguments.output.resolve()))