ce1d870d54
Mod for TikTok 47.1.3 (arm64) in Material 3 style: bottom bar, profile menu, TikTok settings and comments; downloads without the watermark; switches for ads, live streams and the Shop tab; badges of other mods next to names and the niche mod badge (members service); app icons; mod settings screen. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
269 lines
13 KiB
Python
269 lines
13 KiB
Python
"""Self-reported LihonT presence registry. Standard library only; bind to localhost.
|
|
|
|
This is a community marker, not proof of TikTok account ownership or official
|
|
verification. A modified client can claim another UID. No TikTok credentials
|
|
are accepted. Installation secrets permit withdrawing only that installation's
|
|
contribution. Public lookup responses expose only requested membership bits.
|
|
"""
|
|
from __future__ import annotations
|
|
import argparse
|
|
import base64
|
|
import hmac
|
|
import secrets
|
|
import urllib.request
|
|
import hashlib
|
|
import json
|
|
import re
|
|
import sqlite3
|
|
import threading
|
|
import time
|
|
from collections import defaultdict, deque
|
|
from contextlib import contextmanager
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from pathlib import Path
|
|
|
|
UID = re.compile(r"[1-9][0-9]{4,24}\Z")
|
|
INSTALL = re.compile(r"[a-f0-9]{64}\Z")
|
|
LIFETIME = 30 * 86400
|
|
UNIQUE = re.compile(r"[A-Za-z0-9_.]{2,24}\Z")
|
|
# The niche-mod badge can be claimed until this moment (Moscow time).
|
|
CLAIM_DEADLINE = 1794735240 # 2026-11-15 12:34:00 +03:00
|
|
|
|
|
|
def fetch_profile(unique_id):
|
|
"""Public TikTok profile page of @unique_id as (account id, bio), or None. Anonymous GET only."""
|
|
request = urllib.request.Request(
|
|
"https://www.tiktok.com/@" + unique_id,
|
|
headers={"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
|
|
"(KHTML, like Gecko) Chrome/124.0 Safari/537.36"})
|
|
with urllib.request.urlopen(request, timeout=20) as response:
|
|
page = response.read(3_000_000).decode("utf-8", "replace")
|
|
match = re.search(r'<script id="__UNIVERSAL_DATA_FOR_REHYDRATION__"[^>]*>(.*?)</script>', page, re.S)
|
|
if not match:
|
|
return None
|
|
user = (json.loads(match.group(1)).get("__DEFAULT_SCOPE__", {})
|
|
.get("webapp.user-detail", {}).get("userInfo", {}).get("user", {}))
|
|
if not user.get("id"):
|
|
return None
|
|
return str(user["id"]), str(user.get("signature") or "")
|
|
|
|
|
|
def valid_uid(value):
|
|
return isinstance(value, str) and UID.fullmatch(value) is not None
|
|
|
|
|
|
class Registry:
|
|
def __init__(self, path, clock=time.time, fetcher=fetch_profile):
|
|
self.path, self.clock, self.fetcher = str(path), clock, fetcher
|
|
with self.connect() as db:
|
|
db.execute("PRAGMA journal_mode=WAL")
|
|
db.execute("CREATE TABLE IF NOT EXISTS presence (installation TEXT NOT NULL, uid TEXT NOT NULL, last_seen INTEGER NOT NULL, frame_color TEXT NOT NULL DEFAULT 'green', PRIMARY KEY(installation, uid))")
|
|
columns = {row[1] for row in db.execute("PRAGMA table_info(presence)")}
|
|
if "frame_color" not in columns:
|
|
db.execute("ALTER TABLE presence ADD COLUMN frame_color TEXT NOT NULL DEFAULT 'green'")
|
|
db.execute("CREATE INDEX IF NOT EXISTS presence_uid ON presence(uid, last_seen)")
|
|
db.execute("CREATE TABLE IF NOT EXISTS blocked (uid TEXT PRIMARY KEY)")
|
|
db.execute("CREATE TABLE IF NOT EXISTS supporters (uid TEXT PRIMARY KEY, granted_at INTEGER NOT NULL)")
|
|
db.execute("CREATE TABLE IF NOT EXISTS niche (uid TEXT PRIMARY KEY, unique_id TEXT NOT NULL, granted_at INTEGER NOT NULL)")
|
|
db.execute("CREATE TABLE IF NOT EXISTS meta (key TEXT PRIMARY KEY, value TEXT NOT NULL)")
|
|
db.execute("INSERT OR IGNORE INTO meta(key,value) VALUES('code_secret',?)", (secrets.token_hex(32),))
|
|
|
|
@contextmanager
|
|
def connect(self):
|
|
db = sqlite3.connect(self.path, timeout=10)
|
|
try:
|
|
with db:
|
|
yield db
|
|
finally:
|
|
db.close()
|
|
|
|
def presence(self, uid, installation, withdraw=False, frame_color="green"):
|
|
if not valid_uid(uid) or not isinstance(installation, str) or not INSTALL.fullmatch(installation):
|
|
raise ValueError("Invalid account ID or installation code")
|
|
digest = hashlib.sha256(installation.encode("ascii")).hexdigest()
|
|
if frame_color not in ("none", "green", "red", "yellow"):
|
|
frame_color = "green"
|
|
now = int(self.clock())
|
|
with self.connect() as db:
|
|
db.execute("DELETE FROM presence WHERE last_seen < ?", (now - LIFETIME,))
|
|
if withdraw:
|
|
db.execute("DELETE FROM presence WHERE installation=? AND uid=?", (digest, uid))
|
|
else:
|
|
if db.execute("SELECT 1 FROM blocked WHERE uid=?", (uid,)).fetchone():
|
|
raise ValueError("Account excluded from the registry")
|
|
existing = db.execute("SELECT 1 FROM presence WHERE installation=? AND uid=?", (digest, uid)).fetchone()
|
|
if not existing and db.execute("SELECT COUNT(*) FROM presence WHERE installation=?", (digest,)).fetchone()[0] >= 20:
|
|
raise ValueError("Too many accounts for one installation")
|
|
db.execute("INSERT INTO presence(installation,uid,last_seen,frame_color) VALUES(?,?,?,?) ON CONFLICT(installation,uid) DO UPDATE SET last_seen=excluded.last_seen,frame_color=excluded.frame_color", (digest, uid, now, frame_color))
|
|
return {"ok": True}
|
|
|
|
def code(self, uid):
|
|
"""The proof string for an account: it only ever goes into that account's own bio."""
|
|
if not valid_uid(uid):
|
|
raise ValueError("Invalid account ID")
|
|
with self.connect() as db:
|
|
secret = db.execute("SELECT value FROM meta WHERE key='code_secret'").fetchone()[0]
|
|
digest = hmac.new(bytes.fromhex(secret), uid.encode("ascii"), "sha256").digest()
|
|
return "LT-" + base64.b32encode(digest)[:8].decode("ascii")
|
|
|
|
def claim(self, uid, unique_id):
|
|
"""Gives the niche-mod badge to `uid` when its public bio holds the account's code."""
|
|
if not valid_uid(uid) or not isinstance(unique_id, str) or not UNIQUE.fullmatch(unique_id):
|
|
raise ValueError("Invalid account")
|
|
with self.connect() as db:
|
|
if db.execute("SELECT 1 FROM niche WHERE uid=?", (uid,)).fetchone():
|
|
return {"ok": True, "granted": True}
|
|
if db.execute("SELECT 1 FROM blocked WHERE uid=?", (uid,)).fetchone():
|
|
raise ValueError("Account excluded")
|
|
if self.clock() > CLAIM_DEADLINE:
|
|
raise ValueError("The badge can no longer be claimed")
|
|
try:
|
|
profile = self.fetcher(unique_id)
|
|
except Exception:
|
|
raise ValueError("Could not read the TikTok profile, try again later")
|
|
if profile is None:
|
|
raise ValueError("Profile not found")
|
|
found_uid, bio = profile
|
|
if found_uid != uid:
|
|
raise ValueError("That profile belongs to another account")
|
|
if self.code(uid) not in bio:
|
|
return {"ok": True, "granted": False, "reason": "code_missing"}
|
|
with self.connect() as db:
|
|
db.execute("INSERT OR IGNORE INTO niche(uid,unique_id,granted_at) VALUES(?,?,?)",
|
|
(uid, unique_id, int(self.clock())))
|
|
return {"ok": True, "granted": True}
|
|
|
|
def lookup(self, uids):
|
|
if not isinstance(uids, list) or len(uids) > 100 or any(not valid_uid(uid) for uid in uids):
|
|
raise ValueError("Request at most 100 account IDs")
|
|
uids = list(dict.fromkeys(uids))
|
|
if not uids:
|
|
return {"members": {}}
|
|
placeholders = ",".join("?" for _ in uids)
|
|
with self.connect() as db:
|
|
rows = db.execute("SELECT uid,frame_color FROM presence WHERE uid IN (" + placeholders + ") AND last_seen >= ? AND uid NOT IN (SELECT uid FROM blocked)", uids + [int(self.clock()) - LIFETIME]).fetchall()
|
|
niche = {row[0] for row in db.execute(
|
|
"SELECT uid FROM niche WHERE uid IN (" + placeholders + ") AND uid NOT IN (SELECT uid FROM blocked)", uids
|
|
).fetchall()}
|
|
supporters = {row[0] for row in db.execute(
|
|
"SELECT uid FROM supporters WHERE uid IN (" + placeholders + ")", uids
|
|
).fetchall()}
|
|
colors = {}
|
|
for uid, color in rows:
|
|
colors.setdefault(uid, color if color in ("none", "green", "red", "yellow") else "green")
|
|
return {"members": {uid: uid in colors for uid in uids}, "frames": colors,
|
|
"supporters": {uid: uid in supporters for uid in uids},
|
|
"niche": {uid: uid in niche for uid in uids}, "ttl": 300}
|
|
|
|
def set_supporter(self, uid, enabled=True):
|
|
"""Admin-only when invoked from the server CLI; never exposed over HTTP."""
|
|
if not valid_uid(uid):
|
|
raise ValueError("Invalid account ID")
|
|
with self.connect() as db:
|
|
if enabled:
|
|
db.execute("INSERT OR IGNORE INTO supporters(uid,granted_at) VALUES(?,?)",
|
|
(uid, int(self.clock())))
|
|
else:
|
|
db.execute("DELETE FROM supporters WHERE uid=?", (uid,))
|
|
return {"uid": uid, "supporter": bool(enabled)}
|
|
|
|
|
|
class Limits:
|
|
def __init__(self):
|
|
self.events = defaultdict(deque)
|
|
self.lock = threading.Lock()
|
|
|
|
def allow(self, ip, operation):
|
|
key = ip, operation
|
|
window, count = (60, 120) if operation == "lookup" else (3600, 10) if operation == "claim" else (3600, 40)
|
|
now = time.monotonic()
|
|
with self.lock:
|
|
if len(self.events) > 10000:
|
|
for old in list(self.events):
|
|
if not self.events[old] or self.events[old][-1] < now - 3600:
|
|
del self.events[old]
|
|
if len(self.events) > 10000 and key not in self.events:
|
|
return False
|
|
queue = self.events[key]
|
|
while queue and queue[0] < now - window:
|
|
queue.popleft()
|
|
if len(queue) >= count:
|
|
return False
|
|
queue.append(now)
|
|
return True
|
|
|
|
|
|
class Handler(BaseHTTPRequestHandler):
|
|
server_version = "LihonT-Members/1"
|
|
|
|
def log_message(self, fmt, *args):
|
|
# Do not log payloads, installation codes or account IDs.
|
|
pass
|
|
|
|
def send_json(self, status, payload):
|
|
raw = json.dumps(payload, separators=(",", ":")).encode("utf-8")
|
|
self.send_response(status)
|
|
self.send_header("Content-Type", "application/json; charset=utf-8")
|
|
self.send_header("Cache-Control", "no-store")
|
|
self.send_header("Content-Length", str(len(raw)))
|
|
self.end_headers()
|
|
self.wfile.write(raw)
|
|
|
|
def do_GET(self):
|
|
self.send_json(200 if self.path == "/members/v1/health" else 404,
|
|
{"ok": True} if self.path == "/members/v1/health" else {"error": "Not found"})
|
|
|
|
def do_POST(self):
|
|
operation = self.path.removeprefix("/members/v1/")
|
|
if self.path != "/members/v1/" + operation or operation not in ("presence", "lookup", "withdraw", "code", "claim"):
|
|
return self.send_json(404, {"error": "Not found"})
|
|
# This server binds to localhost; nginx overwrites this header.
|
|
ip = self.headers.get("X-Real-IP", self.client_address[0])
|
|
if not self.server.limits.allow(ip, operation if operation in ("lookup", "claim") else "presence"):
|
|
return self.send_json(429, {"error": "Try again later"})
|
|
try:
|
|
length = int(self.headers.get("Content-Length", "0"))
|
|
if not 0 < length <= 8192:
|
|
return self.send_json(413, {"error": "Invalid payload size"})
|
|
self.connection.settimeout(10)
|
|
payload = json.loads(self.rfile.read(length))
|
|
if not isinstance(payload, dict):
|
|
raise ValueError("Expected an object")
|
|
if operation == "lookup":
|
|
result = self.server.registry.lookup(payload.get("uids"))
|
|
elif operation == "code":
|
|
result = {"code": self.server.registry.code(payload.get("uid")),
|
|
"deadline": CLAIM_DEADLINE, "open": time.time() <= CLAIM_DEADLINE}
|
|
elif operation == "claim":
|
|
result = self.server.registry.claim(payload.get("uid"), payload.get("unique_id"))
|
|
else:
|
|
result = self.server.registry.presence(payload.get("uid"), payload.get("installation"), operation == "withdraw", payload.get("frame_color", "green"))
|
|
self.send_json(200, result)
|
|
except (ValueError, TypeError, UnicodeError) as exc:
|
|
self.send_json(400, {"error": str(exc)})
|
|
except Exception:
|
|
self.send_json(503, {"error": "Service temporarily unavailable"})
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--database", required=True)
|
|
parser.add_argument("--port", type=int, default=18746)
|
|
parser.add_argument("--grant-supporter", metavar="UID", help="Grant the LihonT supporter badge locally")
|
|
parser.add_argument("--revoke-supporter", metavar="UID", help="Revoke the LihonT supporter badge locally")
|
|
args = parser.parse_args()
|
|
Path(args.database).resolve().parent.mkdir(parents=True, exist_ok=True)
|
|
if args.grant_supporter or args.revoke_supporter:
|
|
registry = Registry(args.database)
|
|
print(json.dumps(registry.set_supporter(args.grant_supporter or args.revoke_supporter,
|
|
enabled=bool(args.grant_supporter))))
|
|
return
|
|
server = ThreadingHTTPServer(("127.0.0.1", args.port), Handler)
|
|
server.registry = Registry(args.database)
|
|
server.limits = Limits()
|
|
server.serve_forever()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|