Exporting media from Signal Desktop

# List your group conversations
uv run signal-export-media.py --list-groups

# Export all media from a group
uv run signal-export-media.py --group "OK C'est Cool" --output ./photos

#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.10"
# dependencies = [
#     "cryptography",
#     "secretstorage",
# ]
# ///
"""Export media from a Signal Desktop group conversation to a folder."""

import argparse
import base64
import binascii
import hashlib
import hmac
import json
import mimetypes
import subprocess
import sys
from pathlib import Path

import secretstorage
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes

SIGNAL_DIR = Path.home() / ".config" / "Signal"
DB_PATH = SIGNAL_DIR / "sql" / "db.sqlite"
ATTACHMENTS_DIR = SIGNAL_DIR / "attachments.noindex"
CONFIG_PATH = SIGNAL_DIR / "config.json"


def get_db_key() -> str:
    """Get the Signal DB Key from the Keyring"""
    encrypted_raw = binascii.unhexlify(
        json.loads(CONFIG_PATH.read_text())["encryptedKey"]
    )

    conn = secretstorage.dbus_init()
    collection = secretstorage.get_default_collection(conn)
    password = None
    for item in collection.get_all_items():
        attrs = item.get_attributes()
        if (
            attrs.get("xdg:schema") == "chrome_libsecret_os_crypt_password_v2"
            and attrs.get("application") == "Signal"
        ):
            password = item.get_secret()
            break

    if password is None:
        sys.exit("Could not find Signal keyring entry")

    derived = hashlib.pbkdf2_hmac("sha1", password, b"saltysalt", 1, dklen=16)
    ciphertext = encrypted_raw[3:]
    dec = Cipher(algorithms.AES(derived), modes.CBC(b" " * 16)).decryptor()
    plaintext = dec.update(ciphertext) + dec.finalize()
    return plaintext[: -plaintext[-1]].decode()


def query_db(db_key: str, sql: str) -> list[dict]:
    """Run a SQL query via sqlcipher CLI and return parsed JSON rows."""
    full_sql = f"PRAGMA key = \"x'{db_key}'\";\nPRAGMA cipher_compatibility = 4;\n.mode json\n{sql}\n"
    result = subprocess.run(
        ["sqlcipher", str(DB_PATH)], input=full_sql, capture_output=True, text=True
    )
    lines = [l for l in result.stdout.strip().splitlines() if l != "ok"]
    return json.loads("\n".join(lines)) if lines else []


def list_groups(db_key: str) -> list[dict]:
    return query_db(
        db_key,
        "SELECT id, name FROM conversations WHERE type = 'group' AND active_at IS NOT NULL ORDER BY active_at DESC;",
    )


def get_attachments(db_key: str, conversation_id: str) -> list[dict]:
    return query_db(
        db_key,
        f"""SELECT path, contentType, fileName, size, localKey, sentAt
            FROM message_attachments
            WHERE conversationId = '{conversation_id}'
              AND path IS NOT NULL
              AND localKey IS NOT NULL
              AND attachmentType = 'attachment'
            ORDER BY sentAt ASC;""",
    )


def decrypt_attachment(path: str, local_key: str) -> bytes:
    data = (ATTACHMENTS_DIR / path).read_bytes()
    key_bytes = base64.b64decode(local_key)
    iv, ciphertext, hmac_sig = data[:16], data[16:-32], data[-32:]

    expected = hmac.new(key_bytes[32:], iv + ciphertext, hashlib.sha256).digest()
    if not hmac.compare_digest(expected, hmac_sig):
        raise ValueError(f"HMAC verification failed for {path}")

    dec = Cipher(algorithms.AES(key_bytes[:32]), modes.CBC(iv)).decryptor()
    plaintext = dec.update(ciphertext) + dec.finalize()
    return plaintext[: -plaintext[-1]]


def extension_for(content_type, filename):
    if filename and (ext := Path(filename).suffix):
        return ext
    ext = mimetypes.guess_extension(content_type or "")
    return ".jpg" if ext == ".jpe" else (ext or "")


def unique_path(dest: Path) -> Path:
    if not dest.exists():
        return dest
    stem, suffix = dest.stem, dest.suffix
    counter = 1
    while (candidate := dest.with_name(f"{stem}_{counter}{suffix}")).exists():
        counter += 1
    return candidate


def main():
    parser = argparse.ArgumentParser(
        description="Export media from a Signal Desktop group conversation."
    )
    parser.add_argument("--list-groups", action="store_true")
    parser.add_argument("--group", help="Group name (or substring)")
    parser.add_argument("--output", type=Path, help="Output directory")
    args = parser.parse_args()

    if not args.list_groups and not args.group:
        parser.error("provide --list-groups or --group")

    db_key = get_db_key()
    groups = list_groups(db_key)
    if not groups:
        sys.exit("No groups found.")

    if args.list_groups:
        for g in groups:
            print(g["name"])
        return

    matches = [g for g in groups if args.group.lower() in g["name"].lower()]
    if not matches:
        sys.exit(f"No group matching '{args.group}'")
    if len(matches) > 1:
        sys.exit(f"Ambiguous: {', '.join(m['name'] for m in matches)}")
    group = matches[0]

    output_dir = (args.output or Path(f"./{group['name']}")).resolve()
    output_dir.mkdir(parents=True, exist_ok=True)

    print(f"Exporting '{group['name']}' to {output_dir}")
    attachments = get_attachments(db_key, group["id"])
    if not attachments:
        print("No media found.")
        return

    exported = skipped = errors = 0
    for att in attachments:
        ext = extension_for(att["contentType"], att["fileName"])
        name = att["fileName"] or (Path(att["path"]).name + ext)
        dest = unique_path(output_dir / name)

        try:
            dest.write_bytes(decrypt_attachment(att["path"], att["localKey"]))
            exported += 1
            print(f"  [{exported}/{len(attachments)}] {dest.name}")
        except FileNotFoundError:
            skipped += 1
        except Exception as e:
            errors += 1
            print(f"  ERROR: {name}: {e}", file=sys.stderr)

    print(f"\nDone: {exported} exported, {skipped} missing, {errors} errors.")


if __name__ == "__main__":
    main()