From 760d652f5f2d5681d7e1b416dd91bc61bb01979a Mon Sep 17 00:00:00 2001 From: Majjo <90888719+NoxRare@users.noreply.github.com> Date: Sat, 18 Apr 2026 21:01:07 +0200 Subject: [PATCH] feat: story capture API + Python downloader script --- .../features/impl/global/CompanionServer.kt | 67 ++++ tools/snap_story_dl.py | 286 ++++++++++++++++++ 2 files changed, 353 insertions(+) create mode 100644 tools/snap_story_dl.py diff --git a/core/src/main/kotlin/me/eternal/purrfectsnap/core/features/impl/global/CompanionServer.kt b/core/src/main/kotlin/me/eternal/purrfectsnap/core/features/impl/global/CompanionServer.kt index fbc0b9c6..7c0e76b5 100644 --- a/core/src/main/kotlin/me/eternal/purrfectsnap/core/features/impl/global/CompanionServer.kt +++ b/core/src/main/kotlin/me/eternal/purrfectsnap/core/features/impl/global/CompanionServer.kt @@ -2,6 +2,7 @@ package me.eternal.purrfectsnap.core.features.impl.global import android.util.Log import com.google.gson.Gson +import me.eternal.purrfectsnap.common.util.protobuf.ProtoReader import me.eternal.purrfectsnap.core.event.events.impl.NativeUnaryCallEvent import me.eternal.purrfectsnap.core.event.events.impl.NetworkApiRequestEvent import me.eternal.purrfectsnap.core.features.Feature @@ -28,6 +29,10 @@ class CompanionServer : Feature("CompanionServer") { private val networkById = java.util.concurrent.ConcurrentHashMap>() private val bufferLock = Any() + // userId → list of {url, key, iv, postedAt, createdAt, capturedAt} + private val capturedStories = java.util.concurrent.ConcurrentHashMap>>() + private val capturedStoryUrls = java.util.concurrent.ConcurrentHashMap>() + // Prevent re-entrant log capture while the server itself logs private val isCapturing = ThreadLocal() @@ -73,6 +78,7 @@ class CompanionServer : Feature("CompanionServer") { context.log.warn("Companion Server: log hooking unavailable (${it.message})", "CompanionServer") } subscribeToNetworkEvents() + subscribeToStoryEvents() startServer(port, token) } @@ -177,6 +183,56 @@ class CompanionServer : Feature("CompanionServer") { } } + // ---- Story capture ---- + + private fun subscribeToStoryEvents() { + context.event.subscribe(NetworkApiRequestEvent::class) { event -> + if (!event.url.contains("df-mixer-prod") || + (!event.url.endsWith("stories") && !event.url.endsWith("batch_stories"))) return@subscribe + event.onSuccess { buffer -> + buffer ?: return@onSuccess + runCatching { + ProtoReader(buffer).eachBuffer(3) { + eachBuffer(3) { + eachBuffer(3) { + followPath(36) { + eachBuffer(1) { + val userId = getString(8, 1) ?: return@eachBuffer + val url = getString(2, 2)?.substringBefore("?") ?: return@eachBuffer + val postedAt = getVarInt(3) ?: return@eachBuffer + val createdAt = getVarInt(27) ?: -1L + val keyB64 = getString(2, 5) ?: return@eachBuffer + val ivB64 = getString(2, 4) ?: return@eachBuffer + + // Deduplicate by URL + val seen = capturedStoryUrls.getOrPut(userId) { + java.util.concurrent.ConcurrentHashMap() + } + if (seen.putIfAbsent(url, true) == null) { + capturedStories.getOrPut(userId) { + java.util.concurrent.CopyOnWriteArrayList() + }.add(mapOf( + "userId" to userId, + "url" to url, + "postedAt" to postedAt, + "createdAt" to createdAt, + "key" to keyB64, + "iv" to ivB64, + "capturedAt" to System.currentTimeMillis() + )) + } + } + } + } + } + } + }.onFailure { + context.log.warn("CompanionServer: story proto parse failed: ${it.message}", "CompanionServer") + } + } + } + } + // ---- Server ---- private fun startServer(port: Int, token: String) { @@ -269,6 +325,17 @@ class CompanionServer : Feature("CompanionServer") { val entry = networkById[id] if (entry != null) serveJson(socket, entry) else respond404(socket) } + rawPath == "/stories" -> { + val userId = query["user_id"] + val result: Any = if (userId != null) { + capturedStories[userId]?.toList() ?: emptyList>() + } else { + capturedStories.map { (uid, stories) -> + mapOf("userId" to uid, "count" to stories.size, "stories" to stories.toList()) + } + } + serveJson(socket, result) + } else -> respond404(socket) } } diff --git a/tools/snap_story_dl.py b/tools/snap_story_dl.py new file mode 100644 index 00000000..c2643beb --- /dev/null +++ b/tools/snap_story_dl.py @@ -0,0 +1,286 @@ +#!/usr/bin/env python3 +""" +PurrSnap Story Downloader +========================= +Downloads and decrypts Snapchat friend stories using metadata captured by the +PurrfectSnap Companion Server running on your rooted Android device. + +How it works +------------ +1. The PurrfectSnap Xposed mod intercepts the df-mixer-prod/stories HTTP response + while you browse the Snapchat story feed normally. +2. It extracts each story's CDN URL, AES-128 key, and IV from the response proto. +3. This script authenticates to the Companion Server (using your own token), + fetches that metadata, downloads each encrypted file directly from Snapchat's + CDN, and decrypts it locally. + +No Snapchat credentials are used or transmitted by this script. +The Snapchat session on the device is not affected. + +Requirements +------------ + pip install pycryptodome requests + +Usage +----- + python snap_story_dl.py --server http://192.168.1.5:8484 --token YOUR_TOKEN + python snap_story_dl.py --server http://192.168.1.5:8484 --token YOUR_TOKEN --user USER_ID_HERE + python snap_story_dl.py --server http://192.168.1.5:8484 --token YOUR_TOKEN --list-users + +Note: Browse to the Snapchat story feed on the device BEFORE running this script +so that the mod has a chance to intercept and cache the story metadata. +""" + +import argparse +import base64 +import hashlib +import json +import sys +from pathlib import Path +from typing import Optional + +import requests +from Crypto.Cipher import AES + + +# ── Companion Server auth ───────────────────────────────────────────────────── + +def login(server: str, token: str) -> requests.Session: + """POST /login with the token; returns a Session with the session cookie set.""" + session = requests.Session() + r = session.post( + f"{server}/login", + data={"token": token}, + allow_redirects=False, + timeout=10, + ) + if "session=" not in r.headers.get("Set-Cookie", ""): + print("ERROR: Login failed — wrong token or server not reachable.", file=sys.stderr) + sys.exit(1) + print(f" Logged in to {server}") + return session + + +# ── Story metadata from Companion Server ────────────────────────────────────── + +def fetch_stories(session: requests.Session, server: str, user_id: Optional[str]) -> list[dict]: + """ + GET /stories or GET /stories?user_id=xxx + + Returns a flat list of story dicts: + { userId, url, key, iv, postedAt, createdAt, capturedAt } + """ + url = f"{server}/stories" + if user_id: + url += f"?user_id={user_id}" + + r = session.get(url, timeout=15) + r.raise_for_status() + data = r.json() + + if user_id: + # API returns list of story objects directly + return [{"userId": user_id, **s} for s in (data if isinstance(data, list) else [])] + + # API returns list of { userId, count, stories: [...] } + flat: list[dict] = [] + for group in data if isinstance(data, list) else []: + for story in group.get("stories", []): + flat.append({"userId": group["userId"], **story}) + return flat + + +def list_users(session: requests.Session, server: str) -> None: + r = session.get(f"{server}/stories", timeout=15) + r.raise_for_status() + groups = r.json() + if not groups: + print("No stories captured yet. Open Snapchat and browse the story feed first.") + return + print(f"{'User ID':<40} Stories") + print("-" * 55) + for g in groups: + print(f"{g['userId']:<40} {g['count']}") + + +# ── Decryption ──────────────────────────────────────────────────────────────── + +def _b64decode_padded(s: str) -> bytes: + """Base64-decode with automatic padding.""" + s = s.strip() + pad = (4 - len(s) % 4) % 4 + return base64.b64decode(s + "=" * pad) + + +def decrypt_story(encrypted: bytes, key_b64: str, iv_b64: str) -> bytes: + """ + AES-128-CBC decrypt. + Snapchat uses PKCS5/7 padding for most media; strip it if present. + Falls back gracefully if padding is absent (NoPadding mode). + """ + key = _b64decode_padded(key_b64) + iv = _b64decode_padded(iv_b64) + + if len(key) not in (16, 24, 32): + raise ValueError(f"Unexpected key length {len(key)} bytes") + if len(iv) != 16: + raise ValueError(f"Unexpected IV length {len(iv)} bytes") + + cipher = AES.new(key, AES.MODE_CBC, iv) + data = cipher.decrypt(encrypted) + + # Strip PKCS7 padding if present and plausible + if data and 1 <= data[-1] <= 16: + pad_len = data[-1] + if data[-pad_len:] == bytes([pad_len] * pad_len): + data = data[:-pad_len] + + return data + + +# ── File type detection ─────────────────────────────────────────────────────── + +def detect_ext(data: bytes) -> str: + if data[:3] == b"\xff\xd8\xff": + return ".jpg" + if data[:8] == b"\x89PNG\r\n\x1a\n": + return ".png" + if data[:4] == b"RIFF" and data[8:12] == b"WEBP": + return ".webp" + if data[4:8] == b"ftyp": + return ".mp4" + if data[:4] in (b"GIF8", b"GIF9"): + return ".gif" + return ".bin" + + +# ── Download + decrypt ──────────────────────────────────────────────────────── + +_SNAP_CDN_HEADERS = { + "User-Agent": ( + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " + "AppleWebKit/537.36 (KHTML, like Gecko) " + "Chrome/125.0.0.0 Safari/537.36" + ), + "Accept": "*/*", +} + + +def download_and_decrypt(story: dict, out_dir: Path) -> Optional[Path]: + url = story.get("url", "") + key_b64 = story.get("key", "") + iv_b64 = story.get("iv", "") + user_id = story.get("userId", "unknown") + posted = story.get("postedAt", 0) + + if not url or not key_b64 or not iv_b64: + return None + + # Deterministic filename: userId + postedAt + URL hash + url_hash = hashlib.md5(url.encode()).hexdigest()[:10] + stem = f"{user_id}_{posted}_{url_hash}" + + # Skip if any extension of this stem already exists + for ext in (".jpg", ".mp4", ".png", ".webp", ".gif", ".bin"): + if (out_dir / (stem + ext)).exists(): + return out_dir / (stem + ext) + + r = requests.get(url, headers=_SNAP_CDN_HEADERS, timeout=60) + r.raise_for_status() + + plaintext = decrypt_story(r.content, key_b64, iv_b64) + ext = detect_ext(plaintext) + dest = out_dir / (stem + ext) + dest.write_bytes(plaintext) + return dest + + +# ── CLI ─────────────────────────────────────────────────────────────────────── + +def main() -> None: + parser = argparse.ArgumentParser( + description="Download + decrypt Snapchat stories via PurrfectSnap Companion Server", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=__doc__, + ) + parser.add_argument( + "--server", + default="http://192.168.1.x:8484", + metavar="URL", + help="Companion Server address, e.g. http://192.168.1.5:8484", + ) + parser.add_argument( + "--token", + required=True, + metavar="TOKEN", + help="Companion Server access token (set in PurrfectSnap settings)", + ) + parser.add_argument( + "--user", + default=None, + metavar="USER_ID", + help="Snapchat internal user ID to filter (omit for all captured users)", + ) + parser.add_argument( + "--out", + default="stories_out", + metavar="DIR", + help="Output directory (created if absent). Default: stories_out/", + ) + parser.add_argument( + "--list-users", + action="store_true", + help="List captured user IDs and story counts, then exit", + ) + args = parser.parse_args() + + if args.server.endswith("x:8484"): + print("ERROR: Set --server to your device's LAN IP, e.g. http://192.168.1.5:8484", file=sys.stderr) + sys.exit(1) + + print(f"Connecting to {args.server} …") + session = login(args.server, args.token) + + if args.list_users: + list_users(session, args.server) + return + + out_dir = Path(args.out) + out_dir.mkdir(parents=True, exist_ok=True) + + print("Fetching story metadata …") + stories = fetch_stories(session, args.server, args.user) + + if not stories: + print( + "No stories found.\n" + "Open Snapchat on the device, browse the Friends story feed (scroll down to see all),\n" + "then re-run this script." + ) + return + + print(f"Found {len(stories)} stories. Downloading …\n") + ok = fail = skip = 0 + + for story in stories: + uid = story.get("userId", "?") + url = story.get("url", "?")[:60] + try: + path = download_and_decrypt(story, out_dir) + if path is None: + skip += 1 + elif path.stat().st_size > 0: + print(f" [{uid[:20]}] {path.name}") + ok += 1 + else: + skip += 1 + except Exception as e: + print(f" FAIL [{uid[:20]}] {url}… — {e}", file=sys.stderr) + fail += 1 + + print(f"\nDone — {ok} saved, {skip} skipped (already existed), {fail} failed") + print(f"Output: {out_dir.resolve()}") + + +if __name__ == "__main__": + main()