ghostwire/whatsapp_suite/analyzer.py

"""Read-only WhatsApp storage analyzer; optional reads through WhatsApp's own decryption path."""
from __future__ import annotations

import argparse
import datetime as dt
import hashlib
import json
import os
from pathlib import Path
import queue
import subprocess
import threading
import tomllib
import urllib.parse
import urllib.request
from .common import PROFILE
from .exports import inventory


def endpoint(value: str) -> str:
    parsed = urllib.parse.urlsplit(value)
    if parsed.scheme not in ("http", "https") or parsed.hostname not in ("127.0.0.1", "::1") or parsed.username or parsed.password or parsed.query or parsed.fragment:
        raise argparse.ArgumentTypeError("CDP must be an HTTP(S) literal loopback endpoint")
    return value.rstrip("/")


PROBE = r"""(async () => {
const limit = __LIMIT__, includeText = __TEXT__, readDb = __READ_DB__;
const hash = async s => Array.from(new Uint8Array(await crypto.subtle.digest('SHA-256', new TextEncoder().encode(s)))).map(x=>x.toString(16).padStart(2,'0')).join('');
const len = v => v == null ? null : typeof v==='string' ? v.length : v.byteLength ?? v.length ?? null;
const encrypted = v => v == null ? {present:false} : {present:true, envelope: typeof v==='object' && '_data' in v && 'iv' in v, ciphertext_bytes:len(v._data), iv_bytes:len(v.iv), scheme:v._scheme??null, key_id_present:v._keyId!=null};
let models=[], modelError=null;
try { models=require('WAWebMsgCollection').MsgCollection.getModelsArray(); }
catch(e) { modelError=String(e); }
const selected=models.slice(0,limit), ids=selected.map(m=>m.id?.toString()).filter(Boolean);
const summaries=await Promise.all(selected.map(async m=>{
 const s={id_sha256:m.id ? await hash(m.id.toString()):null, type:m.type??null, body_length:len(m.body), caption_length:len(m.caption), plaintext_present:typeof m.body==='string'||typeof m.caption==='string', model_encrypted_data_present:m.encryptedData!=null};
 if(includeText){s.body=typeof m.body==='string'?m.body:null;s.caption=typeof m.caption==='string'?m.caption:null;}
 return s;
}));
let appReads=[];
if(readDb){
 for(const m of selected){
  if(!m.id)continue;
  try {
   const row=await require('WAWebModelStorageUtils').getStorage().table('message').get(m.id.toString());
   const restored=row?require('WAWebDBMessageSerialization').messageFromDbRow(row):null;
   const result={id_sha256:await hash(m.id.toString()),row_found:!!row,decoded_opaque_bytes:len(row?.msgRowOpaqueData),restored_body_length:len(restored?.body),restored_caption_length:len(restored?.caption),body_matches_loaded_model:restored?restored.body===m.body:null,caption_matches_loaded_model:restored?restored.caption===m.caption:null};
   if(includeText){result.body=restored?.body??null;result.caption=restored?.caption??null;}
   appReads.push(result);
  }catch(e){appReads.push({id_sha256:await hash(m.id.toString()),error:String(e)});}
 }
}
// Refuse to create a database: enumerate names first, then abort any upgrade.
const dbs=typeof indexedDB.databases==='function'?await indexedDB.databases():null;
if(!dbs) return {model_count:models.length,models:summaries,model_error:modelError,database_error:'indexedDB.databases unavailable; refusing open to avoid creating a DB'};
if(!dbs.some(d=>d.name==='model-storage')) return {model_count:models.length,models:summaries,model_error:modelError,database_names:dbs.map(d=>d.name),database_error:'model-storage absent; no database opened'};
const db=await new Promise((resolve,reject)=>{ const r=indexedDB.open('model-storage');r.onupgradeneeded=()=>{r.transaction.abort();reject(new Error('Upgrade aborted'));};r.onsuccess=()=>resolve(r.result);r.onerror=()=>reject(new Error(r.error?.name??'Open failed'));r.onblocked=()=>reject(new Error('Database open blocked'));});
let rows=[],matching=[];
try {
 if(!db.objectStoreNames.contains('message'))return {model_count:models.length,models:summaries,model_error:modelError,database_error:'message store absent',stores:Array.from(db.objectStoreNames)};
 const request = r => new Promise((resolve,reject)=>{r.onsuccess=()=>resolve(r.result);r.onerror=()=>reject(new Error(r.error?.name??'Read failed'));});
 const tx=db.transaction(['message'],'readonly'), store=tx.objectStore('message');
 const reads=[request(store.getAll(undefined,limit)),...ids.map(id=>request(store.get(id)))];
 const values=await Promise.all(reads);rows=values[0];matching=values.slice(1).filter(Boolean);
 const summarize=async r=>({id_sha256:await hash(String(r.id)),type:r.type??null,field_names:Object.keys(r).sort(),payload:encrypted(r.msgRowOpaqueData),body_field_present:Object.prototype.hasOwnProperty.call(r,'body'),caption_field_present:Object.prototype.hasOwnProperty.call(r,'caption')});
 const rowSummary=await Promise.all(rows.map(summarize)), matchingSummary=await Promise.all(matching.map(summarize));
 const byId=new Map(matchingSummary.map(r=>[r.id_sha256,r]));
 return {model_count:models.length,models:summaries,model_error:modelError,application_read_path:appReads,database_names:dbs.map(d=>d.name),stores:Array.from(db.objectStoreNames),bounded_row_sample:rowSummary,matching_rows:matchingSummary,comparisons:summaries.filter(m=>byId.has(m.id_sha256)).map(m=>({id_sha256:m.id_sha256,ram_plaintext_present:m.plaintext_present,ram_body_length:m.body_length,ram_caption_length:m.caption_length,disk_payload:byId.get(m.id_sha256).payload}))};
} finally { db.close(); }
})()"""


def cdp_probe(url: str, limit: int, include_text: bool, read_db: bool = False) -> tuple[dict, str]:
    from websockets.sync.client import connect
    with urllib.request.urlopen(url + "/json/list", timeout=10) as response:
        targets = json.load(response)
    target = next((t for t in targets if t.get("type") == "page" and urllib.parse.urlsplit(t.get("url", "")).hostname == "web.whatsapp.com"), None)
    if not target:
        raise RuntimeError("No WhatsApp page target is exposed; start WhatsApp with CDP enabled.")
    websocket_url = target["webSocketDebuggerUrl"]
    if urllib.parse.urlsplit(websocket_url).hostname not in ("127.0.0.1", "::1"):
        raise RuntimeError("Refusing a non-loopback WebSocket target")
    expression = PROBE.replace("__LIMIT__", str(limit)).replace("__TEXT__", "true" if include_text else "false").replace("__READ_DB__", "true" if read_db else "false")
    with connect(websocket_url, max_size=20_000_000, open_timeout=10) as ws:
        ws.send(json.dumps({"id": 1, "method": "Runtime.evaluate", "params": {"expression": expression, "returnByValue": True, "awaitPromise": True}}))
        while True:
            result = json.loads(ws.recv(timeout=30))
            if result.get("id") == 1:
                break
    if "error" in result or result.get("result", {}).get("exceptionDetails"):
        raise RuntimeError("CDP evaluation failed: " + json.dumps(result.get("error") or result["result"]["exceptionDetails"]))
    return result["result"]["result"]["value"], target["id"]


class MCP:
    def __init__(self, config: Path):
        with config.open("rb") as stream:
            entry = tomllib.load(stream)["mcp_servers"]["rea"]
        env = dict(os.environ, **entry.get("env", {}))
        self.proc = subprocess.Popen([entry["command"], *entry.get("args", [])], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, text=True, encoding="utf-8", env=env)
        self.messages = queue.Queue()
        self.counter = 0
        threading.Thread(target=self._read, daemon=True).start()
        self.call("initialize", {"protocolVersion": "2024-11-05", "capabilities": {}, "clientInfo": {"name": "whatsapp-readonly-analyzer", "version": "1.0"}})
        self.send({"jsonrpc": "2.0", "method": "notifications/initialized"})

    def _read(self):
        for line in self.proc.stdout:
            try:
                self.messages.put(json.loads(line))
            except json.JSONDecodeError:
                continue
        self.messages.put({"error": {"message": "REA server exited"}})

    def send(self, payload):
        self.proc.stdin.write(json.dumps(payload) + "\n")
        self.proc.stdin.flush()

    def call(self, method, params):
        self.counter += 1
        ident = self.counter
        self.send({"jsonrpc": "2.0", "id": ident, "method": method, "params": params})
        while True:
            message = self.messages.get(timeout=60)
            if message.get("id") == ident or ("error" in message and "id" not in message):
                if "error" in message:
                    raise RuntimeError(str(message["error"]))
                return message["result"]

    def close(self):
        self.proc.terminate()
        try:
            self.proc.wait(timeout=5)
        except subprocess.TimeoutExpired:
            self.proc.kill()


def rea_inspect(url: str, target_id: str, config: Path) -> dict:
    client = MCP(config)
    try:
        results = []
        for name, arguments in (("list_browser_targets", {"cdp_endpoint": url, "allowed_origins": ["https://web.whatsapp.com"]}), ("inspect_web_page", {"cdp_endpoint": url, "target_id": target_id, "allowed_origins": ["https://web.whatsapp.com"], "observation_ms": 100, "include_script_sources": False, "include_accessibility_text": False, "include_console_text": False, "include_json_body_shapes": False, "include_storage_fingerprints": False, "include_storage_keys": False, "include_websocket_shapes": False})):
            result = client.call("tools/call", {"name": name, "arguments": arguments})
            # Keep provenance only: never save raw DOM, console, source, or storage values.
            evidence = result.get("structuredContent")
            if evidence is None:
                for block in result.get("content", []):
                    if block.get("type") == "text":
                        try:
                            evidence = json.loads(block["text"])
                            break
                        except json.JSONDecodeError:
                            pass
            evidence = evidence if isinstance(evidence, dict) else {}
            results.append({"tool": name, "is_error": result.get("isError", False), "evidence_id": evidence.get("evidence_id"), "operation": evidence.get("operation"), "provider": evidence.get("provider"), "confidence": evidence.get("confidence"), "limitations": evidence.get("limitations", []), "response_sha256": hashlib.sha256(json.dumps(result, sort_keys=True).encode()).hexdigest()})
        return {"source": "REA stdio MCP", "evidence": results, "retention": "Provenance and response digests only; full payload deliberately omitted."}
    finally:
        client.close()


def main(argv=None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--profile", type=Path, default=PROFILE)
    parser.add_argument("--cdp", type=endpoint, help="Existing loopback CDP endpoint, e.g. http://127.0.0.1:9223")
    parser.add_argument("--limit", type=int, default=20, help="Maximum models and DB sample rows (1..100)")
    parser.add_argument("--include-message-text", action="store_true", help="Explicitly save loaded own-account message body/caption in output")
    parser.add_argument("--read-db", action="store_true", help="Use WhatsApp's own message-table getter to decrypt bounded matching rows and compare text in memory")
    parser.add_argument("--rea", action="store_true", help="Also request REA passive page evidence over configured stdio MCP")
    parser.add_argument("--config", type=Path, default=Path.home() / ".codex/config.toml")
    parser.add_argument("--output", type=Path, default=Path("whatsapp-analyzer-report.json"))
    args = parser.parse_args(argv)
    if not 1 <= args.limit <= 100:
        parser.error("--limit must be between 1 and 100")
    if (args.rea or args.include_message_text or args.read_db) and not args.cdp:
        parser.error("--rea, --read-db and --include-message-text require --cdp")
    report = {"timestamp_utc": dt.datetime.now(dt.timezone.utc).isoformat(), "offline": inventory(args.profile), "message_text_included": args.include_message_text,
              "application_db_read_enabled": args.read_db,
              "limitations": ["Static REA findings guide the probe; Runtime.evaluate results are independent CDP observations, not REA attestations.", "Reads only existing loaded models and bounded read-only IndexedDB rows. No send, navigation, record writes or key extraction. --read-db calls WhatsApp's own getter, which decrypts payloads in memory.", "RAM strings are already decrypted by WhatsApp. Raw-envelope comparison alone proves storage shape; --read-db additionally checks restored content equality against loaded models without exporting text by default.", "Not all history is loaded; database getAll sample is bounded and has no recency guarantee. Metadata such as field names and message type remains visible.", "WebView2 and WhatsApp caches are not a forensic snapshot. OS swap, media caches, notifications and native SQLite are outside this probe."]}
    failed = False
    if args.cdp:
        try:
            result, target = cdp_probe(args.cdp, args.limit, args.include_message_text, args.read_db)
            report["runtime"] = {"source": "Independent read-only CDP Runtime.evaluate", "target_id": target, "result": result}
            if args.rea:
                report["rea"] = rea_inspect(args.cdp, target, args.config)
        except Exception as error:
            report["error"] = f"{type(error).__name__}: {error}"
            failed = True
    args.output.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8")
    print(f"Report saved: {args.output.resolve()}")
    if failed:
        print(report["error"])
    return 1 if failed else 0


if __name__ == "__main__":
    raise SystemExit(main())