admin/felhom.eu
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

dooplex-offsite: Vaultwarden (the password manager) rides the nightly encrypted copy (R-923); R-923 filed (operator finding); R-922 ruling A recorded
gates / gates (push) Successful in 5m32s
Details

Browse Source 
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0159rPz1ZhFKsS53msqPYxtS
This commit is contained in:
authentik Default Admin
2026-10-09 11:53:35 +02:00
parent 81c835827e
commit 4cee21acf7
6 changed files with 193 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/dooplex-offsite/test_dooplex_offsite.py
+106 -2
View File
@@ -25,6 +25,27 @@ import os, subprocess, sys
a = sys.argv[1:]
pod = os.environ["FAKE_POD_DATA"]
cmd = a[a.index("--") + 1:]
if "vaultwarden-system" in a:
vw = os.environ["FAKE_VW_DATA"]
if cmd == ["/vaultwarden", "backup"]:
if os.environ.get("FAKE_VW_BACKUP_FAIL"):
print("[ERROR] backup failed: database is locked"); sys.exit(1)
name = "db_20261009_094401.sqlite3"
import shutil as sh; sh.copy(os.path.join(vw, "db.sqlite3"), os.path.join(vw, name))
print("[NOTICE] You are using a plain text `ADMIN_TOKEN` which is insecure.")
print("Backup to 'data/%s' was successful" % name); sys.exit(0)
if cmd[0] == "cat":
sys.exit(subprocess.call(["cat", vw + cmd[1][len("/data"):]]))
if cmd[:2] == ["rm", "-f"]:
p = vw + cmd[2][len("/data"):]
open(os.path.join(os.environ["FAKE_STATE"], "vw-removed"), "a").write(cmd[2] + "\n")
if os.path.exists(p): os.remove(p)
sys.exit(0)
if cmd[:2] == ["sh", "-c"]:
sys.exit(subprocess.call(["sh", "-c", cmd[2].replace("/data", vw)]))
if cmd[0] == "tar":
sys.exit(subprocess.call([vw if x == "/data" else x for x in cmd]))
sys.exit(96)
if cmd[0] == "tar":
cnt = os.path.join(os.environ["FAKE_STATE"], "tar-calls")
n = int(open(cnt).read()) if os.path.exists(cnt) else 0
@@ -73,6 +94,31 @@ head -c 5 "$2" | grep -q '^PGDMP' || { echo "pg_restore: error: input file does
'''
# The CI runner has no sqlite3 CLI: a stand-in on Python's own sqlite3 module (the same SQLite library) answers the calls
# the scripts make — `sqlite3 -readonly <db> '<sql>'`, one row per line, exit 1 on an error.
FAKE_SQLITE3 = r'''#!/usr/bin/env python3
import sqlite3, sys
a = [x for x in sys.argv[1:] if x != "-readonly"]
try:
db = sqlite3.connect("file:" + a[0] + "?mode=ro", uri=True)
for row in db.execute(a[1]):
print("|".join("" if v is None else str(v) for v in row))
except Exception as e:
print("Error: " + str(e), file=sys.stderr); sys.exit(1)
'''
REAL_SQLITE3 = None if os.environ.get("FORCE_FAKE_SQLITE3") else shutil.which("sqlite3")
def make_vw_db(path, users=1, items=3):
import sqlite3
db = sqlite3.connect(path)
db.execute("CREATE TABLE users (uuid TEXT PRIMARY KEY, email TEXT)")
db.execute("CREATE TABLE ciphers (uuid TEXT PRIMARY KEY, data TEXT)")
for i in range(users): db.execute("INSERT INTO users VALUES (?, ?)", ("u%d" % i, "u%d@example.invalid" % i))
for i in range(items): db.execute("INSERT INTO ciphers VALUES (?, ?)", ("c%d" % i, "2.encrypted"))
db.commit(); db.close()
def git(*a, cwd=None):
env = dict(os.environ, GIT_AUTHOR_NAME="t", GIT_AUTHOR_EMAIL="t@t", GIT_COMMITTER_NAME="t", GIT_COMMITTER_EMAIL="t@t")
subprocess.run(["git", *a], cwd=cwd, check=True, capture_output=True, env=env)
@@ -109,7 +155,14 @@ class Base(unittest.TestCase):
open(os.path.join(self.conf, "enc.key"), "w").write("{}")
open(os.path.join(self.conf, "env"), "w").write(
"PBS_REPOSITORY_PUSH='u!push@h:1:s'\nPBS_REPOSITORY_RESTORE='u!restore@h:1:s'\nPBS_FINGERPRINT='aa'\n")
for name, body in (("kubectl", FAKE_KUBECTL), ("proxmox-backup-client", FAKE_PBS), ("pg_restore", FAKE_PG_RESTORE)):
self.vw = j("vw")
os.makedirs(self.vw)
make_vw_db(os.path.join(self.vw, "db.sqlite3"))
open(os.path.join(self.vw, "rsa_key.pem"), "w").write("-----TEST KEY-----\n")
fakes = [("kubectl", FAKE_KUBECTL), ("proxmox-backup-client", FAKE_PBS), ("pg_restore", FAKE_PG_RESTORE)]
if not REAL_SQLITE3:
fakes.append(("sqlite3", FAKE_SQLITE3))
for name, body in fakes:
p = os.path.join(self.bin, name)
open(p, "w").write(body); os.chmod(p, 0o755)
@@ -130,7 +183,7 @@ class Base(unittest.TestCase):
def env(self, **kw):
e = dict(os.environ, PATH=self.bin + ":" + os.environ["PATH"], FAKE_POD_DATA=self.pod, FAKE_PBS_DIR=self.pbs,
FAKE_STATE=self.t, FELHOM_DXOFF_CONF=self.conf, FELHOM_DXOFF_TOKENS=self.tokens,
FAKE_STATE=self.t, FAKE_VW_DATA=self.vw, FELHOM_DXOFF_CONF=self.conf, FELHOM_DXOFF_TOKENS=self.tokens,
FELHOM_DXOFF_STATE=self.state, FELHOM_DXOFF_TEXTFILE_DIR=self.text, FELHOM_DXOFF_DUMPS=self.dumps,
FELHOM_DXOFF_SECRETS=self.secrets)
e.update({k: str(v) for k, v in kw.items()})
@@ -228,6 +281,39 @@ class Push(Base):
self.assertIn("link(s)", r.stderr)
self.assertEqual(self.pushed(), []); self.assertFalse(self.signal())
def test_vaultwarden_is_copied_and_its_backup_file_removed(self):
r = self.push()
self.assertEqual(r.returncode, 0, r.stderr)
snap = os.path.join(self.pbs, "snaps", self.pushed()[0])
self.assertTrue(os.path.isfile(os.path.join(snap, "vaultwarden/db.sqlite3")))
self.assertTrue(os.path.isfile(os.path.join(snap, "vaultwarden/rsa_key.pem")))
self.assertEqual(open(os.path.join(snap, "vaultwarden/USERS")).read().strip(), "1")
self.assertFalse(os.path.exists(os.path.join(self.vw, "db_20261009_094401.sqlite3")), "the backup file must be removed")
self.assertEqual(open(os.path.join(self.t, "vw-removed")).read().strip(), "/data/db_20261009_094401.sqlite3")
self.assertIn("vaultwarden_users 1", open(os.path.join(self.text, "felhom_dooplex_offsite.prom")).read())
def test_vaultwarden_backup_failing_refuses(self):
r = self.push(FAKE_VW_BACKUP_FAIL=1)
self.assertNotEqual(r.returncode, 0)
self.assertIn("vaultwarden backup", r.stderr)
self.assertEqual(self.pushed(), []); self.assertFalse(self.signal())
def test_vaultwarden_with_no_user_refuses(self):
os.remove(os.path.join(self.vw, "db.sqlite3"))
make_vw_db(os.path.join(self.vw, "db.sqlite3"), users=0)
r = self.push()
self.assertNotEqual(r.returncode, 0)
self.assertIn("no user", r.stderr)
self.assertEqual(self.pushed(), []); self.assertFalse(self.signal())
self.assertFalse(os.path.exists(os.path.join(self.vw, "db_20261009_094401.sqlite3")), "removed even on refusal")
def test_a_corrupt_vaultwarden_copy_refuses(self):
open(os.path.join(self.vw, "db.sqlite3"), "wb").write(b"SQLite format 3\x00" + b"\xff" * 4096)
r = self.push()
self.assertNotEqual(r.returncode, 0)
self.assertIn("Vaultwarden integrity_check", r.stderr)
self.assertEqual(self.pushed(), []); self.assertFalse(self.signal())
def test_failed_push_writes_no_signal(self):
r = self.push(FAKE_PBS_FAIL=1)
self.assertNotEqual(r.returncode, 0)
@@ -285,6 +371,24 @@ class RestoreTest(Base):
r = self.restore()
self.assertEqual(r.returncode, 0, r.stderr)
def test_restore_reports_vaultwarden_rows_never_contents(self):
self.pushed_copy()
r = self.restore()
self.assertEqual(r.returncode, 0, r.stderr)
self.assertIn("Vaultwarden 1 user(s) / 3 item(s)", r.stdout)
self.assertNotIn("encrypted", r.stdout + r.stderr)
def test_restore_without_vaultwarden_fails(self):
snap = self.pushed_copy()
os.remove(os.path.join(snap, "vaultwarden/db.sqlite3"))
man = os.path.join(snap, "MANIFEST.sha256")
kept = [l for l in open(man).readlines() if "vaultwarden/db.sqlite3" not in l]
open(man, "w").writelines(kept)
r = self.restore()
self.assertNotEqual(r.returncode, 0)
self.assertIn("no Vaultwarden database", r.stderr)
self.assertFalse(self.signal("felhom_dooplex_offsite_restore.prom"))
def test_an_unreadable_dump_fails(self):
shutil.rmtree(self.dumps); os.makedirs(self.dumps)
self.add_dump("20261008-220001", self.now - 1200, magic=b"XXXXX")