scripts/message_history_reader.py — Message History Reader for iMessage and SMS

#!/usr/bin/env python3
# Message History Reader for iMessage and SMS, version 0.1.1.
# By Zero Width Lab (zerowidthlab.com)
"""Read a private snapshot of the local macOS Messages database."""

from __future__ import annotations

import argparse
import hashlib
import json
import os
import sqlite3
import stat
import sys
import tempfile
import unicodedata
from contextlib import contextmanager
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Iterator

__author__ = "Zero Width Lab"
SIGNATURE = "By Zero Width Lab (zerowidthlab.com)"
APPLE_EPOCH = 978307200
MAX_SOURCE_BYTES = 128 * 1024 * 1024
DEFAULT_DB = Path.home() / "Library" / "Messages" / "chat.db"
REQUIRED = {
    "chat": {"ROWID", "chat_identifier"},
    "handle": {"ROWID", "id"},
    "message": {"ROWID", "text", "attributedBody", "date", "is_from_me", "handle_id"},
    "chat_message_join": {"chat_id", "message_id"},
    "chat_handle_join": {"chat_id", "handle_id"},
}


class ReaderError(Exception):
    pass


def display_text(value: Any) -> str:
    output = []
    for character in str(value):
        category = unicodedata.category(character)
        if category.startswith("C") or category in {"Zl", "Zp"}:
            code = ord(character)
            output.append(f"\\u{code:04x}" if code <= 0xFFFF else f"\\U{code:08x}")
        else:
            output.append(character)
    return "".join(output)


def positive(value: str) -> int:
    try:
        number = int(value)
    except ValueError as exc:
        raise argparse.ArgumentTypeError("must be a positive integer") from exc
    if number < 1:
        raise argparse.ArgumentTypeError("must be a positive integer")
    return number


def iso_date(value: Any) -> str | None:
    if value is None:
        return None
    try:
        number = int(value)
        seconds = number / 1_000_000_000 if abs(number) >= 100_000_000_000 else number
        return datetime.fromtimestamp(APPLE_EPOCH + seconds, timezone.utc).isoformat(timespec="seconds")
    except (ValueError, TypeError, OverflowError, OSError):
        return None


def date_filter(value: str) -> str:
    try:
        datetime.fromisoformat(value.replace("Z", "+00:00"))
    except ValueError as exc:
        raise argparse.ArgumentTypeError("use an ISO 8601 date or timestamp") from exc
    return value


def filter_seconds(value: str) -> float:
    parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    if parsed.tzinfo is None:
        parsed = parsed.replace(tzinfo=timezone.utc)
    return parsed.timestamp()


def extract_attributed_body(blob: bytes | None) -> str | None:
    """Decode a bounded NSString UTF-8 payload from a typedstream subset.

    Unsupported object graphs are rejected; guessed text would be misleading.
    """
    if not isinstance(blob, bytes) or len(blob) > 8 * 1024 * 1024:
        return None
    if b"streamtyped" not in blob[:32]:
        return None
    marker = blob.find(b"NSString")
    if marker < 0:
        return None
    region = blob[marker + len(b"NSString"):marker + len(b"NSString") + 64]
    for offset in range(len(region)):
        absolute = marker + len(b"NSString") + offset
        if blob[absolute] == 0x84 and absolute + 5 <= len(blob):
            length = int.from_bytes(blob[absolute + 1:absolute + 5], "little")
            start = absolute + 5
        elif blob[absolute] == 0x2B and absolute + 2 <= len(blob):
            length = blob[absolute + 1]
            start = absolute + 2
        else:
            continue
        if length > 0 and length <= 1024 * 1024 and start + length <= len(blob):
            try:
                candidate = blob[start:start + length].decode("utf-8")
            except UnicodeDecodeError:
                continue
            if candidate and all(char.isprintable() or char in "\r\n\t" for char in candidate):
                return candidate
    return None


def require_platform() -> None:
    if sys.platform != "darwin" or not all(hasattr(os, name) for name in ("O_DIRECTORY", "O_NOFOLLOW")) or os.open not in os.supports_dir_fd:
        raise ReaderError("macOS with POSIX directory-descriptor and no-follow APIs is required")


def selected_path(path: Path) -> Path:
    raw = str(path)
    if "\\" in raw or "%" in raw or ".." in path.parts:
        raise ReaderError("selected paths reject traversal, encoded and backslash forms")
    return path.expanduser().absolute()


@contextmanager
def directory_fd(path: Path) -> Iterator[int]:
    descriptor = os.open(path.anchor, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
    try:
        for part in path.parts[1:]:
            following = os.open(part, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=descriptor)
            os.close(descriptor)
            descriptor = following
        yield descriptor
    finally:
        os.close(descriptor)


def identity(descriptor: int) -> list[int]:
    info = os.fstat(descriptor)
    return [info.st_dev, info.st_ino]


@contextmanager
def snapshot(path: Path, skipped: list[str] | None = None) -> Iterator[sqlite3.Connection]:
    path = selected_path(path)
    with directory_fd(path.parent) as parent:
        with tempfile.TemporaryDirectory(prefix="message-history-reader-") as directory:
            private = Path(directory)
            private.chmod(0o700)
            copied = private / "chat.db"
            for suffix in ("", "-wal", "-shm"):
                name = path.name + suffix
                try:
                    descriptor = os.open(name, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=parent)
                except FileNotFoundError as exc:
                    if suffix:
                        continue
                    raise ReaderError("Messages database is missing. Full Disk Access may be required.") from exc
                except OSError as exc:
                    if suffix:
                        if skipped is not None:
                            skipped.append(f"{name}: refused symlink or irregular file")
                        continue
                    raise ReaderError("database path must be a regular file, not a symlink") from exc
                info = os.fstat(descriptor)
                if not stat.S_ISREG(info.st_mode):
                    os.close(descriptor)
                    if suffix:
                        if skipped is not None:
                            skipped.append(f"{name}: refused symlink or irregular file")
                        continue
                    raise ReaderError("database path must be a regular file, not a symlink")
                with os.fdopen(descriptor, "rb") as source:
                    if info.st_size > MAX_SOURCE_BYTES:
                        raise ReaderError("database or sidecar exceeds the 128 MiB input limit")
                    with Path(str(copied) + suffix).open("xb") as target:
                        remaining = MAX_SOURCE_BYTES
                        while True:
                            chunk = source.read(min(1024 * 1024, remaining + 1))
                            if not chunk:
                                break
                            remaining -= len(chunk)
                            if remaining < 0:
                                raise ReaderError("database or sidecar exceeds the 128 MiB input limit")
                            target.write(chunk)
            connection = sqlite3.connect(f"{copied.as_uri()}?mode=ro", uri=True)
            connection.row_factory = sqlite3.Row
            try:
                for table, columns in REQUIRED.items():
                    found = {row[1] for row in connection.execute(f"PRAGMA table_info({table})")}
                    if not columns <= found:
                        raise ReaderError(f"unsupported Messages database schema: {table}")
                yield connection
            finally:
                connection.close()


def participants(connection: sqlite3.Connection, chat_id: int) -> list[str]:
    values = [row[0] for row in connection.execute(
        "SELECT DISTINCT h.id FROM chat_handle_join j JOIN handle h ON h.ROWID=j.handle_id "
        "WHERE j.chat_id=? ORDER BY h.id", (chat_id,)
    )]
    if any(not isinstance(value, str) for value in values):
        raise ReaderError("malformed participant handle: expected a string")
    return values


def chats(connection: sqlite3.Connection, limit: int) -> list[dict[str, Any]]:
    rows = connection.execute(
        "SELECT c.ROWID AS chat_id,c.chat_identifier,COUNT(m.ROWID) AS message_count,"
        "MAX(m.date) AS last_date FROM chat c LEFT JOIN chat_message_join j ON j.chat_id=c.ROWID "
        "LEFT JOIN message m ON m.ROWID=j.message_id GROUP BY c.ROWID "
        "ORDER BY last_date DESC,c.ROWID LIMIT ?", (limit,)
    )
    rows = list(rows)
    if any(not isinstance(row["chat_identifier"], str) for row in rows):
        raise ReaderError("malformed chat identifier: expected a string")
    if any(row["last_date"] is not None and iso_date(row["last_date"]) is None for row in rows):
        raise ReaderError("malformed chat last date")
    return [{"chat_id": row["chat_id"], "identifier": row["chat_identifier"],
             "participants": participants(connection, row["chat_id"]),
             "message_count": row["message_count"], "last_date": iso_date(row["last_date"])} for row in rows]


def messages(connection: sqlite3.Connection, *, chat_id: int | None = None,
             term: str | None = None, handle: str | None = None, sender: str | None = None,
             since: str | None = None, until: str | None = None,
             limit: int | None = None) -> tuple[list[dict[str, Any]], list[str]]:
    where = []
    params: list[Any] = []
    if chat_id is not None:
        where.append("c.ROWID=?")
        params.append(chat_id)
    if handle is not None:
        where.append("EXISTS (SELECT 1 FROM chat_handle_join hj JOIN handle hh ON hh.ROWID=hj.handle_id WHERE hj.chat_id=c.ROWID AND hh.id=?)")
        params.append(handle)
    if sender is not None:
        where.append("m.is_from_me=0 AND h.id=?")
        params.append(sender)
    statement = ("SELECT c.ROWID AS chat_id,m.ROWID AS message_id,m.text,m.attributedBody,"
                 "m.date,m.is_from_me,h.id AS sender FROM chat_message_join j "
                 "JOIN chat c ON c.ROWID=j.chat_id JOIN message m ON m.ROWID=j.message_id "
                 "LEFT JOIN handle h ON h.ROWID=m.handle_id")
    if where:
        statement += " WHERE " + " AND ".join(where)
    statement += " ORDER BY m.date,m.ROWID"
    result: list[dict[str, Any]] = []
    skipped: list[str] = []
    for row in connection.execute(statement, params):
        timestamp = iso_date(row["date"])
        if timestamp is None:
            skipped.append(f"message {row['message_id']}: invalid date")
            continue
        unix_time = filter_seconds(timestamp)
        if since and unix_time < filter_seconds(since):
            continue
        if until and unix_time > filter_seconds(until):
            continue
        body = row["text"]
        if body is not None and not isinstance(body, str):
            raise ReaderError(f"message {row['message_id']}: text must be a string or null")
        if row["is_from_me"] not in (0, 1) or (row["sender"] is not None and not isinstance(row["sender"], str)):
            raise ReaderError(f"message {row['message_id']}: malformed sender or direction")
        if body is None and row["attributedBody"] is not None:
            body = extract_attributed_body(row["attributedBody"])
            if body is None:
                skipped.append(f"message {row['message_id']}: undecodable attributedBody")
                continue
        if body is None:
            body = ""
        if term is not None and term.casefold() not in body.casefold():
            continue
        result.append({"chat_id": row["chat_id"], "message_id": row["message_id"],
                       "timestamp": timestamp, "sender": "me" if row["is_from_me"] else row["sender"] or "unknown",
                       "text": body})
        if limit is not None and len(result) >= limit:
            break
    return result, skipped


def transcript(rows: list[dict[str, Any]]) -> str:
    lines: list[str] = []
    for row in rows:
        content = str(row["text"]).splitlines() or [""]
        lines.append(f"{row['timestamp'][:19]}  {display_text(row['sender'])}: {display_text(content[0])}")
        lines.extend("    | " + display_text(part) for part in content[1:])
    return "\n".join(lines) + ("\n" if lines else "")


def output_files(rows: list[dict[str, Any]]) -> dict[str, bytes]:
    jsonl = "".join(json.dumps(row, ensure_ascii=True, sort_keys=True) + "\n" for row in rows)
    return {"messages.jsonl": jsonl.encode("utf-8"), "transcript.txt": transcript(rows).encode("utf-8")}


def output_plan(out: Path) -> dict[str, Any]:
    out = selected_path(out)
    try:
        with directory_fd(out.parent) as parent:
            plan: dict[str, Any] = {"destination": str(out.resolve()), "parent_identity": identity(parent), "directory_identity": None, "create_directory": True}
            try:
                descriptor = os.open(out.name, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=parent)
            except FileNotFoundError:
                return plan
            except OSError as exc:
                raise ReaderError("--out must be a real directory, not a symlink") from exc
            try:
                if os.listdir(descriptor):
                    raise ReaderError("--out must be empty")
                plan.update(directory_identity=identity(descriptor), create_directory=False)
            finally:
                os.close(descriptor)
            return plan
    except OSError as exc:
        raise ReaderError("--out parent must already exist without symlinks") from exc


def export_digest(chat_id: int, out: Path, files: dict[str, bytes], destination_plan: dict[str, Any] | None = None) -> str:
    plan = {"chat_id": chat_id, **(destination_plan if destination_plan is not None else output_plan(out)), "files": [
        {"name": name, "bytes": len(body), "sha256": hashlib.sha256(body).hexdigest()}
        for name, body in sorted(files.items())]}
    return hashlib.sha256(json.dumps(plan, sort_keys=True, separators=(",", ":")).encode()).hexdigest()


def ensure_output(out: Path) -> None:
    output_plan(out)


def create_export(out: Path, files: dict[str, bytes], approved_plan: dict[str, Any] | None = None) -> None:
    out = selected_path(out)
    plan = approved_plan if approved_plan is not None else output_plan(out)
    with directory_fd(out.parent) as parent:
        if str(out.resolve()) != plan["destination"] or identity(parent) != plan["parent_identity"]:
            raise ReaderError("output parent changed after approval")
        if plan["create_directory"]:
            os.mkdir(out.name, mode=0o700, dir_fd=parent)
        descriptor = os.open(out.name, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=parent)
        try:
            if plan["directory_identity"] is not None and identity(descriptor) != plan["directory_identity"]:
                raise ReaderError("output directory changed after approval")
            if os.listdir(descriptor):
                raise ReaderError("--out must be empty")
            for name, body in files.items():
                if name not in {"messages.jsonl", "transcript.txt"}:
                    raise ReaderError("invalid export filename")
                target = os.open(name, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600, dir_fd=descriptor)
                with os.fdopen(target, "wb") as stream:
                    stream.write(body)
        finally:
            os.close(descriptor)


def parser() -> argparse.ArgumentParser:
    app = argparse.ArgumentParser(description="Read local Messages history on macOS.", epilog=SIGNATURE)
    app.add_argument("--db", type=Path, default=DEFAULT_DB, help="Messages database path (default: user's chat.db)")
    app.add_argument("--json", action="store_true", help="machine-readable output")
    app.add_argument("--allow-partial", action="store_true", help="accept skipped messages")
    commands = app.add_subparsers(dest="command", required=True)
    listing = commands.add_parser("chats", help="list conversations")
    listing.add_argument("--limit", type=positive, default=50)
    search = commands.add_parser("search", help="search message text")
    search.add_argument("--text", required=True)
    search.add_argument("--handle", help="chats that include this handle")
    search.add_argument("--from", dest="sender", help="messages sent by this handle (excludes messages from me)")
    search.add_argument("--since", type=date_filter)
    search.add_argument("--until", type=date_filter)
    search.add_argument("--limit", type=positive, default=100)
    export = commands.add_parser("export", help="export one conversation")
    export.add_argument("--chat", type=positive, required=True)
    export.add_argument("--out", type=Path, required=True)
    export.add_argument("--dry-run", action="store_true")
    export.add_argument("--approve", help="digest printed by dry run")
    return app


def main(argv: list[str] | None = None) -> int:
    args = parser().parse_args(argv)
    require_platform()
    path = selected_path(args.db)
    if args.command == "export":
        out = selected_path(args.out)
        plan = output_plan(out)
    snapshot_skipped: list[str] = []
    with snapshot(path, snapshot_skipped) as connection:
        if args.command == "chats":
            data = chats(connection, args.limit)
            skipped: list[str] = snapshot_skipped
        elif args.command == "search":
            if args.since and args.until and filter_seconds(args.since) > filter_seconds(args.until):
                raise ReaderError("--since must not follow --until")
            data, skipped = messages(connection, term=args.text, handle=args.handle, sender=args.sender,
                                     since=args.since, until=args.until, limit=args.limit)
            skipped = snapshot_skipped + skipped
        else:
            found = connection.execute("SELECT 1 FROM chat WHERE ROWID=?", (args.chat,)).fetchone()
            if found is None:
                raise ReaderError("chat ID not found")
            data, skipped = messages(connection, chat_id=args.chat)
            skipped = snapshot_skipped + skipped
    incomplete = bool(skipped)
    if args.command == "export":
        files = output_files(data)
        digest = export_digest(args.chat, out, files, plan)
        if not args.dry_run:
            if args.approve != digest:
                raise ReaderError("export requires the matching --approve digest from --dry-run")
            if incomplete and not args.allow_partial:
                result = {"chat_id": args.chat, "count": len(data), "out": str(out),
                          "files": sorted(files), "approval_digest": digest,
                          "dry_run": False, "incomplete": True, "skipped": skipped}
                if args.json:
                    print(json.dumps(result, ensure_ascii=True))
                print("incomplete: " + "; ".join(display_text(item) for item in skipped), file=sys.stderr)
                return 2
            create_export(out, files, plan)
        result: dict[str, Any] = {"chat_id": args.chat, "count": len(data), "out": str(out),
                                  "files": sorted(files), "approval_digest": digest,
                                  "dry_run": args.dry_run, "incomplete": incomplete, "skipped": skipped, "destination_plan": plan}
    else:
        result = {"results": data, "incomplete": incomplete, "skipped": skipped}
    if args.json:
        print(json.dumps(result, ensure_ascii=True))
    elif args.command == "export":
        print(f"{'Dry run' if args.dry_run else 'Exported'}: {len(data)} messages; destination {display_text(result['out'])}")
        print(f"Directory action: {'create' if plan['create_directory'] else 'use existing'} {display_text(plan['destination'])}")
        print(f"Parent identity: {plan['parent_identity']}; directory identity: {plan['directory_identity']}")
        for name in sorted(files):
            print(f"Planned file: {display_text(out / name)}")
        print(f"Approval digest: {result['approval_digest']}")
    else:
        for item in data:
            print(display_text(json.dumps(item, ensure_ascii=False, sort_keys=True)))
    if incomplete:
        print("incomplete: " + "; ".join(display_text(item) for item in skipped), file=sys.stderr)
    return 2 if incomplete and not args.allow_partial else 0


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except PermissionError:
        print("error: Messages access denied. Grant Full Disk Access to the terminal or agent harness in macOS System Settings, then retry.", file=sys.stderr)
        raise SystemExit(1)
    except (ReaderError, sqlite3.Error, OSError) as exc:
        print(f"error: {display_text(exc)}", file=sys.stderr)
        raise SystemExit(1)

License: MIT (read it).

Back to the listing