feat: story capture API + Python downloader script
This commit is contained in:
Majjo
@@ -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<String, MutableMap<String, Any?>>()
|
||||
private val bufferLock = Any()
|
||||
|
||||
// userId → list of {url, key, iv, postedAt, createdAt, capturedAt}
|
||||
private val capturedStories = java.util.concurrent.ConcurrentHashMap<String, java.util.concurrent.CopyOnWriteArrayList<Map<String, Any>>>()
|
||||
private val capturedStoryUrls = java.util.concurrent.ConcurrentHashMap<String, java.util.concurrent.ConcurrentHashMap<String, Boolean>>()
|
||||
|
||||
// Prevent re-entrant log capture while the server itself logs
|
||||
private val isCapturing = ThreadLocal<Boolean>()
|
||||
|
||||
@@ -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<Map<String, Any>>()
|
||||
} else {
|
||||
capturedStories.map { (uid, stories) ->
|
||||
mapOf("userId" to uid, "count" to stories.size, "stories" to stories.toList())
|
||||
}
|
||||
}
|
||||
serveJson(socket, result)
|
||||
}
|
||||
else -> respond404(socket)
|
||||
}
|
||||
}
|
||||
|
||||
286
tools/snap_story_dl.py
Normal file
286
tools/snap_story_dl.py
Normal file
@@ -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()
|
||||
Reference in New Issue
Block a user