#!/usr/bin/env python3 import argparse import os import secrets import shutil import string import sys import tempfile import time from pathlib import Path from pykeepass import PyKeePass ALLOWED = { "infra/nas/automation": {"username": "xtao-deploy", "password": "secret"}, "infra/xtao-app/automation": {"username": "xtao-deploy", "password": "secret"}, "infra/xtao-ai/automation": {"username": "xtao-deploy", "password": "secret"}, "infra/traefik/monitor": {"username": "traefik-monitor", "token": "token"}, "infra/forgejo/admin": {"username": "forgejo-admin", "password": "secret"}, "infra/forgejo/postgres": {"username": "forgejo", "database": "forgejo", "password": "secret"}, "infra/nextcloud/admin": {"username": "xtao-admin", "password": "secret"}, "infra/nextcloud/postgres": {"username": "nextcloud", "database": "nextcloud", "password": "secret"}, "infra/nextcloud/redis": {"password": "secret"}, "backup/google-drive/restic-repository": {"repository": "gdrive:xtao-backup/restic/nas", "password": "secret"}, } READ_ALLOWED = { **{path: set(profile) for path, profile in ALLOWED.items()}, "external/google-drive/offsite-backup": { "oauth_client_id", "oauth_client_secret", "rclone_token_json", "root_folder_id", }, } ALPHABET = string.ascii_letters + string.digits + "-_" def master_from_stdin(): value = sys.stdin.read().rstrip("\r\n") if not value: raise SystemExit("FAIL missing_master_stdin") return value def ensure_group(kp, parts): group = kp.root_group for name in parts: found = kp.find_groups(name=name, group=group, recursive=False, first=True) group = found or kp.add_group(group, name) return group def find_entry(kp, path): group = kp.root_group parts = path.split("/") for name in parts[:-1]: group = kp.find_groups(name=name, group=group, recursive=False, first=True) if group is None: return None return kp.find_entries(title=parts[-1], group=group, recursive=False, first=True) def gen(kind): if kind == "token": return secrets.token_urlsafe(32) return "".join(secrets.choice(ALPHABET) for _ in range(40)) def upsert(kp, path): if path not in ALLOWED: raise SystemExit(f"FAIL disallowed_entry path={path}") profile = ALLOWED[path] group = ensure_group(kp, path.split("/")[:-1]) entry = kp.find_entries(title=path.split("/")[-1], group=group, recursive=False, first=True) if entry is None: entry = kp.add_entry(group, path.split("/")[-1], "", "", force_creation=True) entry.username = profile.get("username", "") entry.url = f"credential://{path}" entry.notes = "generated_by=nas-secret-agent" for key, value in profile.items(): if value in ("secret", "token"): current = entry.get_custom_property(key) if current in (None, ""): entry.set_custom_property(key, gen(value), protect=True) elif not entry.is_custom_property_protected(key): raise SystemExit(f"FAIL unprotected_existing_field path={path} field={key}") else: entry.set_custom_property(key, value, protect=False) return sorted(profile) def protected_fields(kp, path): entry = find_entry(kp, path) if entry is None: raise SystemExit(f"FAIL entry_missing path={path}") fields = [] for key in sorted((ALLOWED.get(path) or {}).keys()): value = entry.get_custom_property(key) if value in (None, ""): raise SystemExit(f"FAIL empty_field path={path} field={key}") if key in ("password", "token"): if not entry.is_custom_property_protected(key): raise SystemExit(f"FAIL unprotected_field path={path} field={key}") fields.append(key) return fields def backup_file(db, backup_dir): backup_dir.mkdir(parents=True, exist_ok=True) os.chmod(backup_dir, 0o700) ts = time.strftime("%Y%m%dT%H%M%SZ", time.gmtime()) target = backup_dir / f"{Path(db).name}.{ts}.bak.kdbx" shutil.copy2(db, target) os.chmod(target, 0o600) return target def atomic_upsert(db, master, backup_dir, entry_path): db = Path(db) backup = backup_file(db, Path(backup_dir)) fd, tmp_name = tempfile.mkstemp(prefix=db.name + ".", suffix=".candidate", dir=str(db.parent)) os.close(fd) candidate = Path(tmp_name) try: shutil.copy2(db, candidate) os.chmod(candidate, 0o600) kp = PyKeePass(str(candidate), password=master) fields = upsert(kp, entry_path) kp.save() reopened = PyKeePass(str(candidate), password=master) protected_fields(reopened, entry_path) os.replace(candidate, db) PyKeePass(str(db), password=master) return backup, fields except Exception: if candidate.exists(): candidate.unlink() raise def cmd_probe(args, master): kp = PyKeePass(args.db, password=master) count = len(kp.entries) print(f"PASS secret_agent_probe entries={count} values=redacted") def cmd_check(args, master): kp = PyKeePass(args.db, password=master) fields = protected_fields(kp, args.entry) print(f"PASS field_check entry={args.entry} protected={','.join(fields) or 'none'} values=redacted") def cmd_upsert(args, master): backup, fields = atomic_upsert(args.db, master, args.backup_dir, args.entry) print(f"PASS upsert entry={args.entry} fields={','.join(fields)} backup_ref={backup.name} values=redacted") def cmd_rollback_test(args, master): before = Path(args.db).read_bytes() backup, fields = atomic_upsert(args.db, master, args.backup_dir, "infra/traefik/monitor") shutil.copy2(backup, args.db) os.chmod(args.db, 0o600) PyKeePass(args.db, password=master) after = Path(args.db).read_bytes() if before != after: raise SystemExit("FAIL rollback_mismatch") print(f"PASS atomic_write_readback_rollback entry=infra/traefik/monitor fields={','.join(fields)} backup_ref={backup.name} values=redacted") def cmd_raw_field(args, master): allowed_fields = READ_ALLOWED.get(args.entry) if not allowed_fields or args.field not in allowed_fields: raise SystemExit(f"FAIL read_disallowed entry={args.entry} field={args.field}") kp = PyKeePass(args.db, password=master) entry = find_entry(kp, args.entry) if entry is None: raise SystemExit(f"FAIL entry_missing path={args.entry}") value = entry.get_custom_property(args.field) if value in (None, ""): raise SystemExit(f"FAIL empty_field entry={args.entry} field={args.field}") sys.stdout.write(value) def main(): parser = argparse.ArgumentParser() parser.add_argument("--db", required=True) parser.add_argument("--backup-dir", required=True) sub = parser.add_subparsers(dest="cmd", required=True) sub.add_parser("probe") check = sub.add_parser("check") check.add_argument("--entry", required=True) upsert_p = sub.add_parser("upsert") upsert_p.add_argument("--entry", required=True) sub.add_parser("rollback-test") raw = sub.add_parser("raw-field") raw.add_argument("--entry", required=True) raw.add_argument("--field", required=True) args = parser.parse_args() master = master_from_stdin() if args.cmd == "probe": cmd_probe(args, master) elif args.cmd == "check": cmd_check(args, master) elif args.cmd == "upsert": cmd_upsert(args, master) elif args.cmd == "rollback-test": cmd_rollback_test(args, master) elif args.cmd == "raw-field": cmd_raw_field(args, master) if __name__ == "__main__": main()