Files
s3-gateway/tests/vault-update/watch.py
T
57_WolveandClaude Opus 5.5 a52f92c100 patches: add v1.11.0 atomic Vault IAM update patch and its leil-rig proof
0004-vault-atomic-update-user.patch makes the Vault IAM backend's
UpdateUserAccount -- what admin update-user, and so every key rotation,
runs -- one KV v2 write with cas set to the version it read, retried
from a fresh read on a check-and-set conflict (up to 5 attempts).
Upstream deleted the entry and created it again: in between, another
instance on the same KV path whose IAM cache missed answered
InvalidAccessKeyId, a failure between the two calls lost the account,
and two updates at once lost one of them. Touches only
auth/iam_vault.go and a new auth/iam_vault_update_test.go.

tests/vault-update/proof.sh proves it on leil-rig against an OpenBao
dev server, with two gateways on one KV path and gateway A's OpenBao
calls held 3 s by a proxy so upstream's gap can be observed. On
leil-test-1: patched 20 passed, 0 failed; 0001-0003 control 13 passed,
7 failed (entry missing 3.00 s, 28 InvalidAccessKeyId on the second
gateway, concurrent rotation lost, account gone after a refused
write). go test on the node: 35 of 35 packages ok.

docs/vault-atomic-update.md has the problem, the change, the proof and
its result; README and the multi-version pipeline doc point at it.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-11-11 02:04:31 -06:00

176 lines
6.6 KiB
Python

#!/usr/bin/env python3
"""Samplers and record checks for proof.sh.
watch.py kv BAO_URL PATH UNTIL OUT read a KV v2 entry until epoch UNTIL,
logging "<epoch> <status> <version>"
watch.py s3 ENDPOINT UNTIL OUT sign ListBuckets with S3_KEY/S3_SECRET until
epoch UNTIL, logging "<epoch> <status> <code>"
watch.py never-missing OUT the entry was there at every kv sample
watch.py never-invalid OUT no s3 request was answered InvalidAccessKeyId
watch.py cache-expiry OUT EXPIRY the key worked until epoch EXPIRY, not after
watch.py expect BAO_URL PATH FIELD=VALUE... the account in the entry has these fields
(secret=NAME compares with $NAME, unprinted)
watch.py meta BAO_URL PATH "<current_version> <created_time>", or "missing"
The KV commands authenticate with $BAO_TOKEN. The samplers run as fast as a round trip
allows, about 5 ms apart for the entry and 50 ms for the key. Every check prints one
line and exits 0 when it holds, 1 when it does not. Python standard library only;
s3req.py comes from ../owner-scope.
"""
import http.client
import json
import os
import re
import sys
import time
import urllib.parse
sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "owner-scope"))
from s3req import sign # noqa: E402
def bao(url):
u = urllib.parse.urlsplit(url)
return http.client.HTTPConnection(u.hostname, u.port, timeout=10)
def bao_get(conn, path):
conn.request("GET", f"/v1/{path}", headers={"X-Vault-Token": os.environ["BAO_TOKEN"]})
resp = conn.getresponse()
body = resp.read()
return resp.status, json.loads(body) if resp.status == 200 else None
def sample(connect, request, until, out, pause):
conn = None
with open(out, "w") as log:
while time.time() < until:
try:
conn = conn or connect()
status, detail = request(conn)
except (OSError, http.client.HTTPException) as e:
conn = None
status, detail = "err", type(e).__name__
log.write(f"{time.time():.3f} {status} {detail}\n")
log.flush()
time.sleep(pause)
def kv(url, path, until, out):
def request(conn):
status, body = bao_get(conn, path)
return status, body["data"]["metadata"]["version"] if body else "-"
sample(lambda: bao(url), request, until, out, 0.005)
def s3(endpoint, until, out):
u = urllib.parse.urlsplit(endpoint)
key, secret = os.environ["S3_KEY"], os.environ["S3_SECRET"]
def request(conn):
headers = {}
sign("GET", u.netloc, "/", "", headers, b"", key, secret, os.environ.get("S3_REGION", "us-east-1"))
conn.request("GET", "/", headers=headers)
resp = conn.getresponse()
code = re.search(rb"<Code>(.*?)</Code>", resp.read())
return resp.status, code.group(1).decode() if code else "-"
sample(lambda: http.client.HTTPConnection(u.hostname, u.port, timeout=30), request, until, out, 0.05)
def read(out):
with open(out) as f:
return [(float(t), s, d) for t, s, d in (line.split() for line in f)]
def never_missing(out):
rows = read(out)
if not rows:
print("no samples")
return False
span = f"{len(rows)} samples over {rows[-1][0] - rows[0][0]:.0f} s"
missing = [t for t, s, _ in rows if s != "200"]
if missing:
statuses = "/".join(sorted({s for _, s, _ in rows if s != "200"}))
print(f"missing ({statuses}) in {len(missing)} of {span}, "
f"for {missing[-1] - missing[0]:.2f} s from {missing[0] - rows[0][0]:.2f} s in")
return False
versions = sorted({d for _, _, d in rows}, key=int)
print(f"present in all {span}, at version {' then '.join(versions)}")
return True
def never_invalid(out):
rows = read(out)
counts = {}
for _, s, d in rows:
counts[f"{s} {d}"] = counts.get(f"{s} {d}", 0) + 1
print(f"{len(rows)} requests: " + ", ".join(f"{n} x {k}" for k, n in sorted(counts.items())))
return bool(rows) and set(counts) <= {"200 -", "403 SignatureDoesNotMatch"}
def cache_expiry(out, expiry):
rows = [(t - expiry, s, d) for t, s, d in read(out)]
ok = [t for t, s, _ in rows if s == "200"]
refused = [t for t, s, d in rows if s == "403" and d == "SignatureDoesNotMatch"]
if not ok or not refused:
print(f"{len(ok)} accepted, {len(refused)} refused as SignatureDoesNotMatch")
return False
print(f"accepted until {ok[-1]:+.2f} s from the cache expiry, "
f"refused as SignatureDoesNotMatch from {refused[0]:+.2f} s")
return -1 <= ok[-1] <= 1 and refused[0] > ok[-1]
def expect(url, path, wants):
status, body = bao_get(bao(url), path)
if status != 200:
print(f"no entry (HTTP {status})")
return False
acc = next(iter(body["data"]["data"].values()))
got, held = [f"version {body['data']['metadata']['version']}:"], True
for want in wants:
field, value = want.split("=", 1)
if field == "secret":
same = acc["secret"] == os.environ[value]
got.append(f"secret {'is' if same else 'is not'} {value}")
elif field == "version":
same = str(body["data"]["metadata"]["version"]) == value
got[0] = f"version {body['data']['metadata']['version']}{'' if same else f' (want {value})'}:"
else:
same = str(acc.get(field)) == value
got.append(f"{field}={acc.get(field)}" + ("" if same else f" (want {value})"))
held = held and same
print(" ".join(got))
return held
def meta(url, path):
status, body = bao_get(bao(url), path)
print(f"{body['data']['current_version']} {body['data']['created_time']}" if status == 200 else "missing")
def main(argv):
cmd, args = (argv[0], argv[1:]) if argv else ("", [])
if cmd == "kv":
kv(args[0], args[1], float(args[2]), args[3])
elif cmd == "s3":
s3(args[0], float(args[1]), args[2])
elif cmd == "never-missing":
sys.exit(0 if never_missing(args[0]) else 1)
elif cmd == "never-invalid":
sys.exit(0 if never_invalid(args[0]) else 1)
elif cmd == "cache-expiry":
sys.exit(0 if cache_expiry(args[0], float(args[1])) else 1)
elif cmd == "expect":
sys.exit(0 if expect(args[0], args[1], args[2:]) else 1)
elif cmd == "meta":
meta(args[0], args[1])
else:
sys.exit(__doc__)
if __name__ == "__main__":
main(sys.argv[1:])