mrcharles
47343d53a2
Route every compile/vet/test path through a digest-pinned golang:1.26
builder container (dev.sh + Makefile) so the host never runs a Go
toolchain. Add an end-to-end smoke that serves keyproxy in a container
with a throwaway config and drives 401/200/404/400/501/405 paths with
python urllib, asserting the server log is redacted. Ship a commented
keyproxy.toml.example (the real config stays gitignored along with any
*.env tripwire) and rewrite the README as the full quickstart, config,
HTTP, CLI, and redaction reference.
💘 Generated with Crush
Assisted-by: Crush:glm-5.2
109 lines
3.4 KiB
Python
109 lines
3.4 KiB
Python
#!/usr/bin/env python3
|
|
"""keyproxy resolve smoke probe.
|
|
|
|
Drives `keyproxy serve` on BASE (default http://127.0.0.1:8082) with
|
|
plain stdlib urllib: the host has no curl by policy. Checks: healthz,
|
|
401 without/with wrong token, 200 file + env resolves, 404 unknown ref,
|
|
400 invalid ref format (without echo), 501 bitwarden/vault stubs, 502
|
|
empty-value failure. Prints one PASS/FAIL line per check; exits non-zero
|
|
on any failure.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
import time
|
|
import urllib.error
|
|
import urllib.request
|
|
|
|
BASE = os.environ.get("BASE_URL", "http://127.0.0.1:8082")
|
|
TOKEN = os.environ.get("KEYPROXY_TOKEN", "smoke-bearer-token-1111")
|
|
|
|
failures = 0
|
|
|
|
|
|
def req(method, path, body=None, headers=None):
|
|
r = urllib.request.Request(BASE + path, data=body, method=method,
|
|
headers=headers or {})
|
|
try:
|
|
with urllib.request.urlopen(r, timeout=5) as resp:
|
|
return resp.status, resp.read().decode()
|
|
except urllib.error.HTTPError as e:
|
|
return e.code, e.read().decode()
|
|
except Exception as e: # transport errors surface as FAIL lines, not tracebacks
|
|
return 0, f"transport error: {e}"
|
|
|
|
|
|
def check(name, cond, detail=""):
|
|
global failures
|
|
print(("PASS" if cond else "FAIL"), name, ("- " + detail) if detail else "")
|
|
if not cond:
|
|
failures += 1
|
|
|
|
|
|
def wait_healthz(tries=50):
|
|
for _ in range(tries):
|
|
try:
|
|
code, _ = req("GET", "/healthz")
|
|
if code == 200:
|
|
return True
|
|
except Exception:
|
|
pass
|
|
time.sleep(0.2)
|
|
return False
|
|
|
|
|
|
def resolve(token, ref):
|
|
body = json.dumps({"ref": ref}).encode()
|
|
headers = {"Authorization": "Bearer " + token} if token else {}
|
|
return req("POST", "/v1/resolve", body, headers)
|
|
|
|
|
|
def main():
|
|
check("healthz answers (unauthenticated)", wait_healthz())
|
|
|
|
code, text = resolve(None, "mpk-smoke")
|
|
check("no bearer token rejected 401", code == 401, text.strip())
|
|
|
|
code, text = resolve("wrong-token", "mpk-smoke")
|
|
check("wrong bearer token rejected 401", code == 401, text.strip())
|
|
|
|
code, text = resolve(TOKEN, "mpk-smoke")
|
|
obj = json.loads(text) if text else {}
|
|
check("file ref resolves 200 with value",
|
|
code == 200 and obj.get("value") == "smoke-secret-material-2222",
|
|
text.strip())
|
|
|
|
code, text = resolve(TOKEN, "mpk-smoke-env")
|
|
obj = json.loads(text) if text else {}
|
|
check("env ref resolves 200 with value",
|
|
code == 200 and obj.get("value") == "smoke-env-material-4444",
|
|
text.strip())
|
|
|
|
code, text = resolve(TOKEN, "mpk-ghost")
|
|
check("unknown ref 404", code == 404 and "mpk-ghost" in text, text.strip())
|
|
|
|
code, text = resolve(TOKEN, "sk-live-pasted-secret")
|
|
check("invalid ref format 400 without echo",
|
|
code == 400 and "pasted" not in text, text.strip())
|
|
|
|
code, text = resolve(TOKEN, "mpk-smoke-bw")
|
|
check("bitwarden stub 501 not_implemented",
|
|
code == 501 and "not_implemented" in text and "bitwarden" in text,
|
|
text.strip())
|
|
|
|
code, text = resolve(TOKEN, "mpk-smoke-vault")
|
|
check("vault stub 501 not_implemented",
|
|
code == 501 and "not_implemented" in text and "vault" in text,
|
|
text.strip())
|
|
|
|
code, text = req("GET", "/v1/resolve")
|
|
check("GET resolve rejected 405", code == 405, text.strip())
|
|
|
|
print("probe:", "OK" if failures == 0 else f"{failures} FAILURE(S)")
|
|
return 0 if failures == 0 else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|