206 lines
7.1 KiB
Python
206 lines
7.1 KiB
Python
#!/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"},
|
|
"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"):
|
|
entry.set_custom_property(key, gen(value), protect=True)
|
|
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()
|