SeedVault v2 backup decryptor (GrapheneOS, Android)

#!/usr/bin/env python3
"""
Decrypt SeedVault v2 (chunked) backups.

Format:
  - BIP39 mnemonic -> PBKDF2-SHA512(salt="mnemonic", rounds=2048) -> 64-byte seed
  - seed[:32] = legacy_key (unused for v2)
  - seed[32:] = main_key
  - stream_key = HKDF-expand(prk=main_key, info=b"app backup stream key", len=32)
  - Each blob: [0x02 version] [Tink header: len(1)+salt(32)+nonce_prefix(7)] [1MB segments...]
  - Per-file key = HKDF(ikm=stream_key, salt=header_salt, info=aad, len=32)
  - Segment nonce = nonce_prefix(7) + segment_nr(4,big) + is_last(1)
  - Decrypted content is zstd-compressed protobuf (snapshot) or raw data (chunks)
"""

import sys
import os
import hashlib
import struct
import getpass
from pathlib import Path

try:
    from cryptography.hazmat.primitives.kdf.hkdf import HKDF, HKDFExpand
    from cryptography.hazmat.primitives import hashes
    from cryptography.hazmat.primitives.ciphers.aead import AESGCM
    from cryptography.exceptions import InvalidTag
except ImportError:
    sys.exit("Missing: nix shell nixpkgs#python3Packages.cryptography")

try:
    import zstandard as zstd
except ImportError:
    sys.exit("Missing: nix shell nixpkgs#python3Packages.zstandard")

SEGMENT_SIZE = 1 << 20  # 1 MB ciphertext segment
KEY_SIZE = 32
NONCE_PREFIX_SIZE = 7
TAG_SIZE = 16
TINK_HEADER_SIZE = 1 + KEY_SIZE + NONCE_PREFIX_SIZE  # 40


def derive_keys(mnemonic: str):
    seed = hashlib.pbkdf2_hmac("sha512", mnemonic.encode(), b"mnemonic", 2048)
    return seed[:32], seed[32:]  # legacy_key, main_key


def derive_stream_key(main_key: bytes) -> bytes:
    return HKDFExpand(
        algorithm=hashes.SHA256(), length=KEY_SIZE, info=b"app backup stream key"
    ).derive(main_key)


def tink_decrypt(stream_key: bytes, raw: bytes, aad: bytes = b"") -> bytes:
    if raw[0] != 0x02:
        raise ValueError(f"Unexpected version byte: {raw[0]:#x}")
    raw = raw[1:]  # strip SeedVault version byte

    header_len = raw[0]
    if header_len != TINK_HEADER_SIZE:
        raise ValueError(f"Unexpected Tink header length: {header_len}")
    salt = raw[1:33]
    nonce_prefix = raw[33:40]
    segments_data = raw[40:]

    file_key = HKDF(
        algorithm=hashes.SHA256(), length=KEY_SIZE, salt=salt, info=aad
    ).derive(stream_key)

    aesgcm = AESGCM(file_key)
    plaintext = b""
    i = 0

    while segments_data:
        chunk = segments_data[:SEGMENT_SIZE]
        segments_data = segments_data[SEGMENT_SIZE:]
        is_last = len(segments_data) == 0
        nonce = nonce_prefix + struct.pack(">I", i) + (b"\x01" if is_last else b"\x00")
        try:
            plaintext += aesgcm.decrypt(nonce, chunk, None)
        except InvalidTag:
            if i == 0:
                raise ValueError("Decryption failed on first segment — wrong mnemonic or wrong AAD?")
            raise ValueError(f"Decryption failed on segment {i}")
        i += 1

    return plaintext


def decompress(data: bytes) -> bytes:
    dctx = zstd.ZstdDecompressor()
    return dctx.stream_reader(data).read()


def decrypt_blob(stream_key: bytes, path: Path, aad: bytes = b"") -> bytes:
    raw = path.read_bytes()
    decrypted = tink_decrypt(stream_key, raw, aad)
    # First 4 bytes are big-endian int32 compressed size; skip them
    return decompress(decrypted[4:])


def list_snapshot_apps(snapshot_bytes: bytes):
    """
    Parse enough of the protobuf snapshot to print app names.
    We scan for length-prefixed UTF-8 strings that look like package names.
    (Full proto parsing would need the .proto schema + protobuf library.)
    """
    import re
    # Look for package name patterns (field tag 0x0a = field 1, type 2 = length-delimited)
    # In protobuf, string fields are: tag varint + length varint + bytes
    # We do a simple regex scan for android package names in the raw bytes
    text = snapshot_bytes.decode("latin-1")
    packages = re.findall(r'[a-z][a-z0-9_]*(?:\.[a-z][a-z0-9_]*){2,}', text)
    seen = set()
    unique = []
    for p in packages:
        if p not in seen and len(p) > 8:
            seen.add(p)
            unique.append(p)
    return unique


def main():
    backup_dir = Path(sys.argv[1]) if len(sys.argv) > 1 else \
        Path("/home/bogdb/seedvault_backup_apr20/.SeedVaultAndroidBackup")

    # Find the token directory (long hex name)
    token_dirs = [d for d in backup_dir.iterdir()
                  if d.is_dir() and len(d.name) == 64]
    if not token_dirs:
        sys.exit(f"No token directory found in {backup_dir}")
    token_dir = token_dirs[0]
    print(f"Token dir: {token_dir.name[:16]}...")

    # Find snapshot
    snapshots = list(token_dir.glob("*.snapshot"))
    if not snapshots:
        sys.exit("No .snapshot file found")
    snapshot_path = snapshots[0]
    print(f"Snapshot:  {snapshot_path.name[:16]}... ({snapshot_path.stat().st_size} bytes)")

    # Get mnemonic — from env var, file, or interactive prompt
    mnemonic = os.environ.get("SV_MNEMONIC", "").strip()
    if mnemonic:
        print("Using mnemonic from $SV_MNEMONIC")
    else:
        mnemonic_file = Path("~/.sv_mnemonic").expanduser()
        if mnemonic_file.exists():
            mnemonic = mnemonic_file.read_text().strip()
            print(f"Using mnemonic from {mnemonic_file}")
        else:
            vis = input("\nShow mnemonic while typing? [y/N]: ").strip().lower()
            if vis == "y":
                mnemonic = input("Mnemonic (12 words, lowercase, space-separated): ").strip()
            else:
                mnemonic = getpass.getpass("Mnemonic: ").strip()

    print("\nDeriving keys...")
    legacy_key, main_key = derive_keys(mnemonic)
    stream_key = derive_stream_key(main_key)
    print(f"  stream_key (first 8 bytes): {stream_key[:8].hex()}")

    # AAD = single version byte (0x02) per getAdForVersion() in SeedVault Crypto.kt
    aad = bytes([0x02])

    print("Decrypting snapshot...")
    try:
        snapshot_bytes = decrypt_blob(stream_key, snapshot_path, aad)
    except ValueError as e:
        sys.exit(f"Failed: {e}")
    except Exception as e:
        sys.exit(f"Error: {e}")

    print(f"Snapshot decrypted: {len(snapshot_bytes)} bytes")

    # Save raw snapshot for inspection
    out_dir = Path("/home/bogdb/seedvault_backup_apr20/decrypted")
    out_dir.mkdir(exist_ok=True)
    raw_out = out_dir / "snapshot.bin"
    raw_out.write_bytes(snapshot_bytes)
    print(f"Saved to: {raw_out}")


if __name__ == "__main__":
    main()