#!/usr/bin/env python3 """photosafe_cli — sign up, back up, list, export and restore, from a terminal. Everything is encrypted and decrypted here, on your machine. The server only ever receives ciphertext. `restore --export` works entirely offline against a downloaded export file: you never need PhotoSafe, or a network, to get your photos back. pip install requests cryptography photosafe_cli.py signup --email you@example.com photosafe_cli.py backup --email you@example.com ~/Pictures photosafe_cli.py list --email you@example.com photosafe_cli.py export --email you@example.com --out export.tar.gz photosafe_cli.py restore --email you@example.com --out ./restored photosafe_cli.py restore --export export.tar.gz --out ./restored # offline The passphrase is read from PHOTOSAFE_PASSPHRASE or prompted for. Losing it means losing the photos: nobody, including us, can recover them. """ import argparse import base64 import getpass import hashlib import json import mimetypes import os import sys import tarfile import requests sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import crypto_core as cc # noqa: E402 DEFAULT_BASE = os.environ.get("PHOTOSAFE_URL", "https://photosafe.eliteaiempire.com") b64e = lambda b: base64.b64encode(b).decode() # noqa: E731 def passphrase(): return os.environ.get("PHOTOSAFE_PASSPHRASE") or getpass.getpass("PhotoSafe passphrase: ") def login(base, email, pw): s = requests.Session() p = s.post(f"{base}/api/login/params", json={"email": email}, timeout=30).json() salt = base64.b64decode(p["kdf_salt"]) r = s.post(f"{base}/api/login", json={"email": email, "auth_hash": b64e(cc.derive_auth_hash(pw, salt))}, timeout=60) if r.status_code != 200: sys.exit(f"login failed: {r.status_code} {r.text[:200]}") d = r.json() mk = cc.unwrap_master_key(cc.derive_kek(pw, salt), base64.b64decode(d["wrapped_master_key"])) return s, mk def cmd_signup(a): pw = passphrase() salt = cc.new_salt() mk = cc.new_master_key() r = requests.post(f"{a.base}/api/signup", timeout=60, json={ "email": a.email, "kdf_salt": b64e(salt), "wrapped_master_key": b64e(cc.wrap_master_key(cc.derive_kek(pw, salt), mk)), "auth_hash": b64e(cc.derive_auth_hash(pw, salt))}) print(r.status_code, r.text[:300]) return 0 if r.status_code == 200 else 1 def cmd_backup(a): s, mk = login(a.base, a.email, passphrase()) files = [] for p in a.paths: if os.path.isdir(p): for root, _d, names in os.walk(p): files += [os.path.join(root, n) for n in sorted(names)] else: files.append(p) up = dedup = 0 for path in sorted(files): data = open(path, "rb").read() meta = {"name": os.path.basename(path), "path": os.path.relpath(path), "mime": mimetypes.guess_type(path)[0] or "application/octet-stream", "mtime": int(os.path.getmtime(path)), "size": len(data)} ct, wfk, mb, sha = cc.encrypt_object(mk, data, meta) r = s.post(f"{a.base}/api/objects", timeout=300, files={"ciphertext": ("blob", ct, "application/octet-stream")}, data={"wrapped_file_key": b64e(wfk), "meta_blob": b64e(mb), "plain_sha256": sha, "plain_bytes": str(len(data))}) if r.status_code != 200: print(f"FAIL {path}: {r.status_code} {r.text[:160]}") continue if r.json().get("deduped"): dedup += 1 else: up += 1 print(f"{'dedup' if r.json().get('deduped') else 'up '} {sha[:12]} {path}") print(f"\nuploaded={up} deduped={dedup} total={len(files)}") return 0 def cmd_list(a): s, mk = login(a.base, a.email, passphrase()) for o in s.get(f"{a.base}/api/objects", timeout=60).json()["objects"]: m = cc.decrypt_meta(mk, base64.b64decode(o["meta_blob"])) print(f"{o['id'][:12]} {o['plain_bytes']:>10} {o['plain_sha256'][:12]} {m['name']}") return 0 def cmd_export(a): s, _mk = login(a.base, a.email, passphrase()) r = s.get(f"{a.base}/api/export", timeout=1800, stream=True) r.raise_for_status() with open(a.out, "wb") as fh: for chunk in r.iter_content(1 << 20): fh.write(chunk) print(f"wrote {a.out} ({os.path.getsize(a.out)} bytes)") return 0 def _write(out, name, data, sha, report): dest = os.path.join(out, name) os.makedirs(os.path.dirname(dest) or out, exist_ok=True) if os.path.exists(dest): stem, ext = os.path.splitext(dest) dest = f"{stem}.{sha[:8]}{ext}" open(dest, "wb").write(data) got = hashlib.sha256(data).hexdigest() report["ok" if got == sha else "mismatch"] += 1 if got != sha: report["bad_files"].append(dest) return dest def cmd_restore(a): os.makedirs(a.out, exist_ok=True) report = {"ok": 0, "mismatch": 0, "bad_files": []} pw = passphrase() if a.export: # fully offline path with tarfile.open(a.export, "r:gz") as tar: man = json.load(tar.extractfile("manifest.json")) mk = cc.unwrap_master_key(cc.derive_kek(pw, base64.b64decode(man["kdf_salt"])), base64.b64decode(man["wrapped_master_key"])) for o in man["objects"]: ct = tar.extractfile(o["blob"]).read() pt = cc.decrypt_object(mk, ct, base64.b64decode(o["wrapped_file_key"])) meta = cc.decrypt_meta(mk, base64.b64decode(o["meta_blob"])) _write(a.out, meta["name"], pt, o["plain_sha256"], report) else: # live account s, mk = login(a.base, a.email, pw) for o in s.get(f"{a.base}/api/objects", timeout=120).json()["objects"]: ct = s.get(f"{a.base}/api/objects/{o['id']}", timeout=600).content pt = cc.decrypt_object(mk, ct, base64.b64decode(o["wrapped_file_key"])) meta = cc.decrypt_meta(mk, base64.b64decode(o["meta_blob"])) _write(a.out, meta["name"], pt, o["plain_sha256"], report) print(json.dumps(report, indent=1)) return 1 if report["mismatch"] else 0 def main(): p = argparse.ArgumentParser(description="PhotoSafe client") p.add_argument("--base", default=DEFAULT_BASE) sub = p.add_subparsers(dest="cmd", required=True) for name, fn, extra in ( ("signup", cmd_signup, []), ("backup", cmd_backup, [("paths", {"nargs": "+"})]), ("list", cmd_list, []), ("export", cmd_export, [("--out", {"required": True})]), ("restore", cmd_restore, [("--out", {"required": True}), ("--export", {})])): sp = sub.add_parser(name) sp.add_argument("--email", default=os.environ.get("PHOTOSAFE_EMAIL")) for an, kw in extra: sp.add_argument(an, **kw) sp.set_defaults(fn=fn) a = p.parse_args() sys.exit(a.fn(a)) if __name__ == "__main__": main()