From aed80ee14347848de1f53da0987d2b57f1d6f60f Mon Sep 17 00:00:00 2001 From: kisfenyo Date: Tue, 6 Oct 2026 11:26:23 +0200 Subject: [PATCH] R-624: the bench (only) seeds vaultwarden through its admin invite; run secrets redacted and shredded MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `09` §3 decision 146. vaultwarden's fixture tries the household's own /identity/accounts/register first (400 while sign-up is closed, R-512); on the BENCH ONLY it then signs in to /admin with the ADMIN_TOKEN the bench generated for this run, invites the drill address and registers it — the route measured on 9202 2026-09-15 (E1-vaultwarden-spike). The token goes to curl on stdin, the admin cookie in a 0600 header file that is shredded. The dead /api/accounts/register (404 on 1.36) is gone. bench_admin_seed_allowed(): the venue is the bench's (upgrade_boxport.Venue VENUE="bench"), FELHOM_BENCH_ADMIN_SEED=1, and /opt/docker/stacks does not exist (every Felhom box has it). Any one missing refuses; the edge stays inconclusive with what was tried. upgrade-test.py: the run's .env is written 0600 and shredded after the teardown; every printed line and every evidence file is redacted of the generated deploy secrets and the fixture's own password/key. zipline needs no held secret: its first-run /api/setup already makes the SUPERADMIN with a per-run password (measured 2026-09-30), now redacted too. Tests: BenchAdminSeedGuard, SecretHygiene (red-proved). Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_0159rPz1ZhFKsS53msqPYxtS --- scripts/test_upgrade_bench.py | 167 ++++++++++++++++++++++++++++++++ scripts/upgrade-test.py | 94 +++++++++++++++++- scripts/upgrade_boxport.py | 4 + scripts/upgrade_fixtures_box.py | 131 ++++++++++++++++++++++--- 4 files changed, 382 insertions(+), 14 deletions(-) diff --git a/scripts/test_upgrade_bench.py b/scripts/test_upgrade_bench.py index 926cc28..5fb970e 100644 --- a/scripts/test_upgrade_bench.py +++ b/scripts/test_upgrade_bench.py @@ -146,5 +146,172 @@ class MarkerIgnore(unittest.TestCase): self.assertTrue(suffix and path and max_size > 0 and len(reason) > 40, (app, path)) +# --- R-624 (`09` §3 decision 146): the bench-only admin seed, and the run's secrets ----------------------------------- +import sys # noqa: E402 +from unittest import mock # noqa: E402 +sys.path.insert(0, HERE) +import upgrade_fixtures_box as fxbox # noqa: E402 + +ADMIN = "a" * 8 + "0123456789abcdef0123456789abcdef0123456789abcdef01234567" # stands in for a generated hex:32 + + +class FakeVenue: + """walk.py's interface without a network: records every call; answers like vaultwarden 1.36 did on 9202 + (audits/evidence-p1fixes-2026-09-15/E1-vaultwarden-spike.txt): a stranger's register 400, POST /admin with the + right token sets VW_ADMIN, /admin/invite 200 with that cookie, an invited address registers 200.""" + + def __init__(self, bench): + if bench: + self.VENUE = "bench" + self.GENERATED = {"vaultwarden": {"ADMIN_TOKEN": ADMIN, "DOMAIN": "gate.invalid"}} + self.calls, self.invited = [], set() + + def wait_app(self, *a, **k): + return True + + def app_curl(self, sub, path, *extra, method=None, data=None, timeout=45): + headers = [] + for i, x in enumerate(extra): + if x == "-H" and extra[i + 1].startswith("@"): + headers.append(open(extra[i + 1][1:]).read().strip()) + self.header_file = extra[i + 1][1:] + self.calls.append({"path": path, "argv": list(extra), "data": data, "file_headers": headers}) + if path == "/identity/accounts/register": + email = json.loads(data)["email"] + return 0, ("200" if email in self.invited else "400"), '{"message":"Registration not allowed"}' + if path == "/admin": + if data == "token=" + ADMIN: + return 0, "200", "HTTP/1.1 200 OK\r\nset-cookie: VW_ADMIN=jwt.admin.session; Path=/admin\r\n\r\n" + return 0, "401", "HTTP/1.1 401\r\n\r\nInvalid admin token" + if path == "/admin/invite": + if "Cookie: VW_ADMIN=jwt.admin.session" in headers: + self.invited.add(json.loads(data)["email"]) + return 0, "200", "{}" + return 0, "401", "" + return 0, "404", "" + + +import json # noqa: E402 + + +class BenchAdminSeedGuard(unittest.TestCase): + """The admin seed runs on the bench ONLY. RED-PROOF (REPORT): make bench_admin_seed_allowed return (True, "") — + test_the_box_walk_never_signs_in_as_admin and each single-condition refusal fail.""" + + def setUp(self): + self.tmp = tempfile.mkdtemp(prefix="bench-guard-") + self.nobox = os.path.join(self.tmp, "no-such-stacks") + self.said = [] + + def tearDown(self): + shutil.rmtree(self.tmp, ignore_errors=True) + + def allowed(self, venue, env, box_marker): + with mock.patch.dict(os.environ, env, clear=False), mock.patch.object(fxbox, "BOX_MARKER", box_marker): + if "FELHOM_BENCH_ADMIN_SEED" not in env: + os.environ.pop("FELHOM_BENCH_ADMIN_SEED", None) + return fxbox.bench_admin_seed_allowed(venue) + + def test_all_three_conditions_allow(self): + self.assertEqual(self.allowed(FakeVenue(True), {"FELHOM_BENCH_ADMIN_SEED": "1"}, self.nobox), (True, "")) + + def test_each_condition_alone_refuses(self): + ok, why = self.allowed(FakeVenue(False), {"FELHOM_BENCH_ADMIN_SEED": "1"}, self.nobox) + self.assertFalse(ok) + self.assertIn("not the bench venue", why) + ok, why = self.allowed(FakeVenue(True), {}, self.nobox) + self.assertFalse(ok) + self.assertIn("FELHOM_BENCH_ADMIN_SEED", why) + ok, why = self.allowed(FakeVenue(True), {"FELHOM_BENCH_ADMIN_SEED": "yes"}, self.nobox) + self.assertFalse(ok) + ok, why = self.allowed(FakeVenue(True), {"FELHOM_BENCH_ADMIN_SEED": "1"}, self.tmp) # a box: stacks exists + self.assertFalse(ok) + self.assertIn("Felhom box", why) + + def test_the_bench_venue_names_itself(self): + import upgrade_boxport + self.assertEqual(upgrade_boxport.Venue.VENUE, "bench") + + def seed(self, venue, env, box_marker): + with mock.patch.dict(os.environ, env, clear=False), mock.patch.object(fxbox, "BOX_MARKER", box_marker): + if "FELHOM_BENCH_ADMIN_SEED" not in env: + os.environ.pop("FELHOM_BENCH_ADMIN_SEED", None) + fx_ = fxbox.Vaultwarden() + return fx_, fx_.seed(venue, "vault", self.said.append) + + def test_the_box_walk_never_signs_in_as_admin(self): + v = FakeVenue(False) + fx_, got = self.seed(v, {"FELHOM_BENCH_ADMIN_SEED": "1"}, self.nobox) + self.assertIsNone(got) + self.assertEqual([c["path"] for c in v.calls], ["/identity/accounts/register"]) + self.assertIn("NOT tried", fx_.tried) + + def test_the_bench_seeds_through_the_admin_invite_and_never_shows_the_token(self): + v = FakeVenue(True) + _, got = self.seed(v, {"FELHOM_BENCH_ADMIN_SEED": "1"}, self.nobox) + self.assertIsNotNone(got) + self.assertEqual([c["path"] for c in v.calls], ["/identity/accounts/register", "/admin", "/admin/invite", + "/identity/accounts/register"]) + for c in v.calls: + self.assertFalse(any(ADMIN in a or "VW_ADMIN" in a for a in c["argv"]), "a secret on the command line") + self.assertEqual(v.calls[2]["file_headers"], ["Cookie: VW_ADMIN=jwt.admin.session"]) + self.assertFalse(os.path.exists(v.header_file), "the cookie header file was not shredded") + self.assertFalse(any(ADMIN in s or "jwt.admin.session" in s for s in self.said), "a secret was printed") + + +class SecretHygiene(unittest.TestCase): + """The run's secrets: .env 0600 and shredded, every evidence file redacted. RED-PROOF (REPORT): make redact_tree + return [] without rewriting — test_evidence_files_are_redacted fails.""" + + FELHOM = ("deploy_fields:\n - env_var: DOMAIN\n type: domain\n - env_var: ADMIN_TOKEN\n type: secret\n" + " generate: \"hex:32\"\n - env_var: SIGNUPS_ALLOWED\n type: text\n default: \"false\"\n") + + def setUp(self): + self.tmp = tempfile.mkdtemp(prefix="bench-secrets-") + + def tearDown(self): + shutil.rmtree(self.tmp, ignore_errors=True) + + def test_secret_values_are_the_generated_fields_and_the_seed_password(self): + if ut.cvp is None: + import importlib.util as iu + sp = iu.spec_from_file_location("cvp", os.path.join(HERE, "check-volume-persistence.py")) + m = iu.module_from_spec(sp) + sp.loader.exec_module(m) + ut.cvp = m + env = {"DOMAIN": "gate.invalid", "ADMIN_TOKEN": ADMIN, "SIGNUPS_ALLOWED": "false"} + got = ut.secret_values(self.FELHOM, env, {"email": "drill-1@gate.invalid", "pw": "Drill-0011223344"}) + self.assertEqual(set(got), {ADMIN, "Drill-0011223344"}) + + def test_env_is_0600_and_shredded(self): + p = ut.Path(self.tmp) / ".env" + ut.write_secret_file(p, "ADMIN_TOKEN=%s\n" % ADMIN) + self.assertEqual(os.stat(p).st_mode & 0o777, 0o600) + ut.write_secret_file(p, "ADMIN_TOKEN=%s\n" % ADMIN) # a re-render replaces it, still 0600 + self.assertEqual(os.stat(p).st_mode & 0o777, 0o600) + ut.shred_file(p) + self.assertFalse(p.exists()) + ut.shred_file(p) # a missing file is fine + + def test_evidence_files_are_redacted(self): + ev = ut.Path(self.tmp) / "evidence" / "MV-vaultwarden" + (ev / "sub").mkdir(parents=True) + (ev / "run.log").write_text("token=%s ok\n" % ADMIN) + (ev / "sub" / "to-full.log").write_text("clean line\n") + (ev / "verdict.json").write_text(json.dumps({"abort_detail": "pw Drill-0011223344"})) + held = ut.redact_tree(ev, [ADMIN, "Drill-0011223344"]) + self.assertEqual(len(held), 2) + for p in ev.rglob("*"): + if p.is_file(): + t = p.read_text() + self.assertNotIn(ADMIN, t) + self.assertNotIn("Drill-0011223344", t) + self.assertIn(ut.REDACTED, (ev / "run.log").read_text()) + self.assertEqual((ev / "sub" / "to-full.log").read_text(), "clean line\n") + + def test_redact_longest_first(self): + self.assertEqual(ut.redact("x abcdefgh123 y", ["abcdefgh123", "abcdefgh"]), "x y") + + if __name__ == "__main__": unittest.main(verbosity=2) diff --git a/scripts/upgrade-test.py b/scripts/upgrade-test.py index 19f914f..09211c2 100755 --- a/scripts/upgrade-test.py +++ b/scripts/upgrade-test.py @@ -205,10 +205,93 @@ def render(app: str, images: dict, workdir: Path, env: dict, template: str = Non out.append(line) workdir.mkdir(parents=True, exist_ok=True) (workdir / "docker-compose.yml").write_text(apply_bench_overrides(app, pg_mounts_for("\n".join(out) + "\n"))) - (workdir / ".env").write_text("".join(f"{k}={v}\n" for k, v in env.items())) + write_secret_file(workdir / ".env", "".join(f"{k}={v}\n" for k, v in env.items())) return workdir / "docker-compose.yml" +# --- R-624 (`09` §3 decision 146): the run's secrets never reach a file that outlives the run, or any output -------- +# +# The bench GENERATES each app's deploy secrets per run (build_env) — vaultwarden's ADMIN_TOKEN among them, which the +# bench-only admin seed uses. They live in memory and in ONE file compose must read: the run's `.env`, written 0600 and +# shredded after the teardown. Every line the run prints and every evidence file it leaves is passed through redact() +# (pinned by scripts/test_upgrade_bench.py: SecretHygiene). +REDACTED = "" +SECRET_FIELD_TYPES = ("password", "secret", "secret_input") +SEED_SECRET_KEYS = ("pw", "key", "password", "token", "admin_token") + + +def write_secret_file(path: Path, text: str): + """Write `text` to `path` readable by its owner only (0600 from the first byte, not chmod after).""" + path = Path(path) + try: + path.unlink() + except FileNotFoundError: + pass + fd = os.open(str(path), os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) + with os.fdopen(fd, "w") as fh: + fh.write(text) + + +def shred_file(path: Path): + """Overwrite with zeros, fsync, remove. A missing file is fine.""" + try: + n = os.path.getsize(path) + with open(path, "r+b") as fh: + fh.write(b"\0" * n) + fh.flush() + os.fsync(fh.fileno()) + os.unlink(path) + except OSError: + pass + + +def secret_values(felhom_text: str, env: dict, seeded=None) -> list: + """Every value this run must never print: each deploy field the template generates or types as a secret, and the + fixture's own secret fields (its password / key). Longest first, so a value inside another is not half-redacted. + Values shorter than 8 characters are not secrets the bench made (and would redact ordinary words).""" + out = set() + for f in cvp.parse_deploy_fields(felhom_text) if cvp else []: + if f.get("generate") or f.get("type") in SECRET_FIELD_TYPES: + v = env.get(f["env_var"]) + if v: + out.add(str(v)) + if isinstance(seeded, dict): + for k, v in seeded.items(): + if k in SEED_SECRET_KEYS and isinstance(v, str): + out.add(v) + return sorted((v for v in out if len(v) >= 8), key=len, reverse=True) + + +def redact(text, secrets_: list): + if not text or not secrets_: + return text + for v in secrets_: + text = text.replace(v, REDACTED) + return text + + +def redact_tree(root: Path, secrets_: list) -> list: + """Redact every file under the run's evidence directory in place; returns the files that held a secret.""" + hit = [] + if not secrets_: + return hit + enc = [(v.encode(), REDACTED.encode()) for v in secrets_] + for p in sorted(Path(root).rglob("*")): + if not p.is_file() or p.is_symlink(): + continue + try: + b = p.read_bytes() + except OSError: + continue + nb = b + for v, r in enc: + nb = nb.replace(v, r) + if nb != b: + p.write_bytes(nb) + hit.append(str(p)) + return hit + + def compose(workdir: Path, project: str, *args, timeout=1800): return cvp._sh(["docker", "compose", "-p", project, "-f", str(workdir / "docker-compose.yml"), "--env-file", str(workdir / ".env")] + list(args), timeout=timeout) @@ -877,9 +960,10 @@ def run_edge(edge_id: str) -> dict: "duration_s": 0, "measured_at": None, "evidence": f"evidence/{edge_id}"} t0 = time.time() log = [] + secrets_ = secret_values(felhom, env) # R-624: grows by the fixture's own secrets after the seed def say(msg): - line = f"[{datetime.now(timezone.utc).strftime('%H:%M:%S')}] {msg}" + line = redact(f"[{datetime.now(timezone.utc).strftime('%H:%M:%S')}] {msg}", secrets_) print(line, flush=True) log.append(line) @@ -911,6 +995,7 @@ def run_edge(edge_id: str) -> dict: say("no fixture for this app — inconclusive") return rec seeded = fixture.seed(container_ip, say) + secrets_[:] = sorted(set(secrets_) | set(secret_values(felhom, env, seeded)), key=len, reverse=True) if seeded is None: rec["verdict"] = "inconclusive" rec["abort_detail"] = "no non-browser seed route" + ( @@ -1038,6 +1123,11 @@ def run_edge(edge_id: str) -> dict: (ev / "compose-final.log").write_text((lg.stdout + lg.stderr)[-400000:]) # post-abort state only — see to-full.log (ev / "verdict.json").write_text(json.dumps(rec, indent=2)) compose(workdir, project, "down", "-v", "--remove-orphans", timeout=900) + # R-624: no evidence file keeps a secret this run made, and the run's .env does not outlive it. + held = redact_tree(ev, secrets_) + if held: + print(f"[redact] a run secret was removed from {len(held)} evidence file(s): {held}", flush=True) + shred_file(workdir / ".env") SOAK_SECONDS = 600 diff --git a/scripts/upgrade_boxport.py b/scripts/upgrade_boxport.py index 93c9f58..a2a04cb 100644 --- a/scripts/upgrade_boxport.py +++ b/scripts/upgrade_boxport.py @@ -72,6 +72,10 @@ def routes(compose_text): class Venue: """walk.py's interface, on the bench.""" + # R-624: the ONE place the bench names itself. upgrade_fixtures_box.bench_admin_seed_allowed reads it (with the + # run's opt-in and the not-a-box check); walk.py's object on a box has no VENUE, so an admin seed refuses there. + VENUE = "bench" + def __init__(self, compose_text, env, ipfn): self.routes = routes(compose_text) self.env = env diff --git a/scripts/upgrade_fixtures_box.py b/scripts/upgrade_fixtures_box.py index 027afc9..b58dd0f 100644 --- a/scripts/upgrade_fixtures_box.py +++ b/scripts/upgrade_fixtures_box.py @@ -283,26 +283,131 @@ class Navidrome: # ============================================================================================= +# --- R-624 (`09` §3 decision 146): the BENCH may hold an app's admin secret to seed it, the bench ONLY ---------------- +BENCH_ADMIN_SEED_ENV = "FELHOM_BENCH_ADMIN_SEED" # the run's explicit opt-in; the bench command sets it to 1 +BOX_MARKER = "/opt/docker/stacks" # every Felhom box has it: the controller syncs templates there + + +def bench_admin_seed_allowed(w): + """(True, "") only on the test bench; (False, why) everywhere else. THE BENCH IS RECOGNISED BY ALL THREE: + 1. the venue is the bench's (`upgrade_boxport.Venue` sets VENUE = "bench"; the box walk's object has none); + 2. the run opted in: FELHOM_BENCH_ADMIN_SEED=1 in the environment (the bench command sets it, nothing else); + 3. the machine is not a Felhom box: /opt/docker/stacks does not exist (a box's controller syncs the catalog + there — this is what tells guest 9202, which also ran /opt/upg on 2026-09-23, from the bench LXC 9401). + Pinned by scripts/test_upgrade_bench.py (BenchAdminSeedGuard: each condition alone refuses).""" + if getattr(w, "VENUE", None) != "bench": + return False, "not the bench venue (the box walk never holds an admin secret)" + if os.environ.get(BENCH_ADMIN_SEED_ENV) != "1": + return False, "%s=1 is not set for this run" % BENCH_ADMIN_SEED_ENV + if os.path.exists(BOX_MARKER): + return False, "%s exists — this machine is a Felhom box, not the bench" % BOX_MARKER + return True, "" + + +def _curl_with_header_file(w, sub, path, header, *extra, **kw): + """w.app_curl with ONE secret header read by curl from a 0600 temp file (`-H @file`), so the value is never + on a command line; the file is overwritten and removed afterwards, whatever happens.""" + import tempfile + fd, hp = tempfile.mkstemp(prefix=".felhom-h-") + try: + os.fchmod(fd, 0o600) + os.write(fd, (header + "\n").encode()) + os.close(fd) + fd = None + return w.app_curl(sub, path, "-H", "@" + hp, *extra, **kw) + finally: + if fd is not None: + os.close(fd) + _shred(hp) + + +def _shred(path): + """Overwrite a small secret file with zeros, then remove it. A missing file is fine.""" + try: + n = os.path.getsize(path) + with open(path, "r+b") as fh: + fh.write(b"\0" * n) + fh.flush() + os.fsync(fh.fileno()) + os.unlink(path) + except OSError: + pass + + +def _encstring(n=32): + """A Bitwarden EncString-shaped opaque value ("2.||"). The server stores it and never opens it.""" + b = lambda k: base64.b64encode(secrets.token_bytes(k)).decode() + return "2.%s|%s|%s" % (b(16), b(n), b(32)) + + class Vaultwarden: - """Vaultwarden's own account API: register an account, then prove it survives by asking the app - to issue a token for it (its own login endpoint, the household's own route).""" + """Vaultwarden's own account API. The catalog CLOSES self-registration on purpose (SIGNUPS_ALLOWED=false, R-512), so + a stranger's `POST /identity/accounts/register` answers 400 „Registration not allowed" — measured on 9202 + 2026-09-15 (`audits/evidence-p1fixes-2026-09-15/E1-vaultwarden-spike.txt`), where the route that DOES make an + account was measured too: the admin page signs in with ADMIN_TOKEN (`POST /admin`, form `token=`), invites an + address (`POST /admin/invite`, JSON `{"email"}`; with mail off the invitation is stored, nothing is sent), and that + address may then register (`POST /identity/accounts/register` → 200). `/api/accounts/register` is 404 on 1.36. + + ON THE BENCH ONLY (R-624, `09` §3 decision 146; bench_admin_seed_allowed). ADMIN_TOKEN is the deploy secret the + bench itself GENERATED for this run (build_env, `generate: hex:32`) — held in memory, sent to curl on stdin, never + printed; the admin session cookie travels in a 0600 header file that is shredded. Everywhere else the seed tries + the household's own route only and the edge stays `inconclusive`, with what was tried — the honest answer while + sign-up is closed. The readback asks the app to issue a token for the account (its own login endpoint).""" sub = "vault" + def _register(self, w, sub, email, key): + body = json.dumps({"email": email, "name": "drill", "masterPasswordHash": key, "masterPasswordHint": None, + "key": _encstring(), + "keys": {"encryptedPrivateKey": _encstring(64), + "publicKey": base64.b64encode(secrets.token_bytes(64)).decode()}, + "kdf": 0, "kdfIterations": 600000}) + return w.app_curl(sub, "/identity/accounts/register", "-H", "Content-Type: application/json", + data=body, method="POST") + def seed(self, w, sub, say): if not w.wait_app(sub, "/alive", want=("200",)): return None email = f"drill-{secrets.token_hex(4)}@gate.invalid" # Vaultwarden stores an already-hashed master key; the value is opaque to the server. key = base64.b64encode(secrets.token_bytes(32)).decode() - body = json.dumps({"email": email, "name": "drill", "masterPasswordHash": key, - "key": "0." + base64.b64encode(secrets.token_bytes(48)).decode(), - "kdf": 0, "kdfIterations": 600000}) - rc, code, out = w.app_curl(sub, "/api/accounts/register", - "-H", "Content-Type: application/json", - data=body, method="POST") - say(f" vaultwarden: register http={code}") + rc, code, out = self._register(w, sub, email, key) + say(f" vaultwarden: self-registration http={code} (closed by design, R-512 — 400 expected)") + if code in ("200", "204"): + return {"email": email, "key": key} + tried = f"POST /identity/accounts/register -> {code} (sign-up closed by design)" + ok, why = bench_admin_seed_allowed(w) + if not ok: + self.tried = tried + f"; the admin invite was NOT tried: {why}" + say(f" vaultwarden: {self.tried}") + return None + token = (getattr(w, "GENERATED", {}).get("vaultwarden") or {}).get("ADMIN_TOKEN") or "" + if not token: + self.tried = tried + "; the bench generated no ADMIN_TOKEN for this run" + say(f" vaultwarden: {self.tried}") + return None + import urllib.parse + rc, code, page = w.app_curl(sub, "/admin", "-i", "-H", "Content-Type: application/x-www-form-urlencoded", + data="token=" + urllib.parse.quote(token, safe=""), method="POST") + cookie = "; ".join(p for p in _set_cookies(page).split("; ") if p.startswith("VW_ADMIN=")) + say(f" vaultwarden: bench admin sign-in http={code} session={'yes' if cookie else 'no'}") + if not cookie: + self.tried = tried + f"; POST /admin -> {code}, no admin session" + return None + try: + rc, code, out = _curl_with_header_file(w, sub, "/admin/invite", "Cookie: " + cookie, + "-H", "Content-Type: application/json", + data=json.dumps({"email": email}), method="POST") + finally: + cookie = None + say(f" vaultwarden: bench admin invite http={code}") + if code != "200": + self.tried = tried + f"; POST /admin/invite -> {code}" + return None + rc, code, out = self._register(w, sub, email, key) + say(f" vaultwarden: invited registration http={code}") if code not in ("200", "204"): - say(f" vaultwarden: refused {out[:250]}") + self.tried = tried + f"; invited POST /identity/accounts/register -> {code}" + say(f" vaultwarden: refused {out[:200]}") return None return {"email": email, "key": key} @@ -310,17 +415,19 @@ class Vaultwarden: if not w.wait_app(sub, "/alive", want=("200",), tries=36): return False def login(pwhash): + import urllib.parse return w.app_curl(sub, "/identity/connect/token", "-H", "Content-Type: application/x-www-form-urlencoded", data=("grant_type=password&scope=api%20offline_access" f"&client_id=web&deviceType=9&deviceIdentifier=drill" - f"&deviceName=drill&username={t['email']}&password={pwhash}"), + f"&deviceName=drill&username={urllib.parse.quote(t['email'], safe='')}" + f"&password={urllib.parse.quote(pwhash, safe='')}"), method="POST") rc, code, _ = login(base64.b64encode(secrets.token_bytes(32)).decode()) if code == "200": say(" vaultwarden: READBACK UNUSABLE — a wrong master key authenticated") return False - rc, code, out = login(t["key"].replace("+", "%2B").replace("=", "%3D").replace("/", "%2F")) + rc, code, out = login(t["key"]) ok = code == "200" and "access_token" in out say(f" vaultwarden: token for the seeded account http={code} ok={ok}") if not ok: