A scan canary for your servers. The app can't open listening ports — so flip it around: your box calls home on a timer. If the heartbeat stops for 3+ intervals, the beacon flips SILENT and your inbox lights up. And because every beacon has a unique probe URL, plant it as a canary link: if your box gets scanned or someone touches the link, that fires too. Know the second your box goes dark or gets touched.
""" + newcard + """
-
New beacon
-
""" + _beacon_list_html(uid) + flow("know the second your box goes dark or gets touched", [
-"you create a beacon labeled edge-vpn-fra, disguise port 22, interval 5 min.",
-"you drop the one-liner (or the systemd timer) on the server — it curls the probe URL every 5 minutes.",
-"the beacon row shows LIVE with a fresh last-seen, every heartbeat logged.",
-"them a scanner knocks on port 22 — you've planted the same /b/ URL as a canary link in the port's banner or a decoy file; the moment it's fetched by anything that isn't your heartbeat, the tripwire fires.",
-"you your INBOX lights up: beacon edge-vpn-fra touched from an unknown IP — with geo and device.",
-"one night the box dies or loses network — 15 minutes of silence (3 intervals) later the beacon flips SILENT and pings your inbox again."]) + how([
-"A beacon is a unique probe URL — one per server — plus a heartbeat interval you choose.",
-"Your server curls the probe URL every interval (curl one-liner or systemd timer; we generate both for you).",
-"Heartbeats are detected by the User-Agent (curl/wget/drb-beacon) or the ?hb=1 flag — anything else fetching the URL counts as a CANARY HIT and alerts instantly.",
-"Any fetch that is not a heartbeat logs IP, geo, device — same intel as the canary traps, kept in its own beacon_hits table.",
-"Miss 3+ intervals of heartbeat and the beacon flips SILENT: one inbox alert per outage (no duplicates), auto-cleared when the heartbeat returns.",
-"The disguise port is informational — a badge reminding you which port the canary link guards; the detection is the URL itself.",
-"Pair it with CANARY TRAPS: plant a /c/ trap inside the box's files, and the /b/ URL in its network decoys — layered coverage."]) + gloss([
-("heartbeat","a tiny scheduled HTTP GET from your server proving it's alive"),
-("SILENT","no heartbeat for 3+ intervals — box down, offline, or egress blocked"),
-("LATE","heartbeat overdue by 1+ interval but under the SILENT threshold"),
-("disguise port","the port you're watching — shown on the badge, pairs the beacon to its service"),
-("canary hit","any fetch of the probe URL that isn't your heartbeat — someone touched it")]) + agent_card("POST /api/beacon/create · GET /api/beacon/list · GET /api/beacon/status?token=",
-'curl -X POST "' + SITE + '/api/beacon/create" -H "Cookie: dark0rbits_tok=..." -d "label=edge-vpn&port=22&interval_min=5"',
-'Create returns token + probe_url + install snippet. status returns LIVE/LATE/SILENT/WAITING, last_seen, hits. Auth: session cookie or Bearer dk_ key.')
- return page("beacon", body)
-
-def _beacon_list_html(uid):
- if not uid:
- return '
Log in (no KYC) to create beacons and see their status.
'
- con = _beacon_db()
- rows = con.execute("SELECT * FROM beacons WHERE user_id=? ORDER BY id DESC LIMIT 50", (uid,)).fetchall()
- trs = ""
- for b in rows:
- st, cls = _beacon_status(b)
- hits = con.execute("SELECT COUNT(*) c FROM beacon_hits WHERE beacon_id=? AND kind='hit'", (b["id"],)).fetchone()["c"]
- probes = con.execute("SELECT COUNT(*) c FROM beacon_hits WHERE beacon_id=? AND kind='probe'", (b["id"],)).fetchone()["c"]
- last = time.strftime("%b %d %H:%M", time.localtime(b["last_seen"])) if b["last_seen"] else "never"
- probe, curl, _s, _t = _beacon_snippet(b["token"], b["interval_min"])
- port = (":" + str(b["port"])) if b["port"] else ""
- trs += ("
')
- r = rate_limit("beacon", 20, 60)
- if r: return r
- b = _beacon_create_core(uid)
- resp = Response(status=302)
- resp.headers["Location"] = "/beacon?new=" + b["token"]
- return resp
-
-@app.route("/b/")
-def beacon_probe(token):
- con = _beacon_db()
- b = con.execute("SELECT * FROM beacons WHERE token=?", (token,)).fetchone()
- if not b:
- return "Not Found", 404
- ip = request.headers.get("X-Real-IP") or request.remote_addr or "?"
- ua = request.headers.get("User-Agent", "")
- heartbeat = param("hb") == "1" or bool(_HEARTBEAT_UA.search(ua or ""))
- kind = "probe" if heartbeat else "hit"
- con.execute("INSERT INTO beacon_hits(beacon_id,ts,ip,ua,kind) VALUES(?,?,?,?,?)",
- (b["id"], int(time.time()), ip, ua[:200], kind))
- con.execute("UPDATE beacons SET last_seen=? WHERE id=?", (int(time.time()), b["id"]))
- was_silent = b["silent_notified"]
- if was_silent:
- con.execute("UPDATE beacons SET silent_notified=0 WHERE id=?", (b["id"],))
- con.commit() # commit BEFORE inbox writes (db-locked race)
- if not heartbeat and b["user_id"]:
- geo = enrich_ip(ip) or {}
- where = ""
- if geo:
- where = " — %s, %s %s · %s" % (geo.get("city", ""), geo.get("regionName", ""), geo.get("countryCode", ""), geo.get("isp", ""))
- m = db()
- m.execute("INSERT INTO messages(user_id,sender,body,created) VALUES(?,?,?,?)",
- (b["user_id"], "operator-bot",
- "◆ BEACON TOUCHED: '%s%s' probe URL fetched by a non-heartbeat client — IP %s%s device: %s" %
- (esc(b["label"]), ((":%d" % b["port"]) if b["port"] else ""), esc(ip), esc(where), esc(ua[:100])),
- int(time.time())))
- m.commit()
- elif was_silent and b["user_id"]:
- m = db()
- m.execute("INSERT INTO messages(user_id,sender,body,created) VALUES(?,?,?,?)",
- (b["user_id"], "operator-bot",
- "◆ BEACON RECOVERED: '%s' heartbeat is back after going SILENT." % esc(b["label"]),
- int(time.time())))
- m.commit()
- if heartbeat:
- return "ok", 200, {"Content-Type": "text/plain"}
- # canary-style: look like nothing
- return "Not Found", 404
-
-@app.route("/beacon/events")
-def beacon_events():
- uid = current_user_id()
- token = param("token") or ""
- if not uid:
- return page("beacon", '
login required
')
- con = _beacon_db()
- b = con.execute("SELECT * FROM beacons WHERE token=? AND user_id=?", (token, uid)).fetchone()
- if not b:
- return page("beacon", '
unknown beacon
')
- hits = con.execute("SELECT * FROM beacon_hits WHERE beacon_id=? ORDER BY id DESC LIMIT 100", (b["id"],)).fetchall()
- trs = ""
- for h in hits:
- geo = {}
- ip = (h["ip"] or "").strip()
- if ip and not ip.startswith(("10.", "127.", "172.")):
- geo = enrich_ip(ip) or {}
- where = ("%s, %s" % (geo.get("city", "—"), geo.get("countryCode", ""))) if geo else "—"
- ktag = 'beat' if h["kind"] == "probe" else 'HIT'
- trs += ("
""")
-
-@app.route("/api/beacon/create", methods=["POST"])
-def api_beacon_create():
- r = rate_limit("beacon", 20, 60)
- if r: return r
- uid = key_user() or current_user_id()
- if not uid:
- return jsonify({"ok": False, "error": "auth required: account session or Authorization: Bearer key"}), 401
- b = _beacon_create_core(uid)
- probe, curl, svc, tim = _beacon_snippet(b["token"], b["interval_min"])
- return jsonify({"ok": True, "token": b["token"], "label": b["label"], "port": b["port"],
- "interval_min": b["interval_min"], "probe_url": probe,
- "install": {"curl": curl, "systemd_service": svc, "systemd_timer": tim}})
-
-@app.route("/api/beacon/list")
-def api_beacon_list():
- uid = key_user() or current_user_id()
- if not uid:
- return jsonify({"ok": False, "error": "auth required"}), 401
- con = _beacon_db()
- _beacon_check_silent(con, uid)
- out = []
- for b in con.execute("SELECT * FROM beacons WHERE user_id=? ORDER BY id DESC LIMIT 100", (uid,)).fetchall():
- st, _cls = _beacon_status(b)
- hits = con.execute("SELECT COUNT(*) c FROM beacon_hits WHERE beacon_id=? AND kind='hit'", (b["id"],)).fetchone()["c"]
- out.append({"token": b["token"], "label": b["label"], "port": b["port"], "interval_min": b["interval_min"],
- "status": st, "last_seen": b["last_seen"], "canary_hits": hits,
- "probe_url": SITE + "/b/" + b["token"]})
- return jsonify({"ok": True, "beacons": out})
-
-@app.route("/api/beacon/status")
-def api_beacon_status():
- uid = key_user() or current_user_id()
- token = param("token") or ""
- if not uid:
- return jsonify({"ok": False, "error": "auth required"}), 401
- con = _beacon_db()
- _beacon_check_silent(con, uid)
- b = con.execute("SELECT * FROM beacons WHERE token=? AND user_id=?", (token, uid)).fetchone()
- if not b:
- return jsonify({"ok": False, "error": "unknown beacon"}), 404
- st, _cls = _beacon_status(b)
- hits = con.execute("SELECT COUNT(*) c FROM beacon_hits WHERE beacon_id=? AND kind='hit'", (b["id"],)).fetchone()["c"]
- return jsonify({"ok": True, "token": b["token"], "label": b["label"], "port": b["port"],
- "interval_min": b["interval_min"], "status": st, "last_seen": b["last_seen"],
- "silent_notified": b["silent_notified"], "canary_hits": hits})
-# ---------- END TOOL: PORT BEACON ----------
-
-
-# ---------- TOOL: BSSID RADAR ----------
-
-BSSID_OSINT_BASE = "http://10.30.20.174:8080"
-BSSID_OSINT_URL = BSSID_OSINT_BASE + "/api/run/bssid_geo?q="
-
-def _norm_mac(raw):
- """Normalize any MAC spelling to AA:BB:CC:DD:EE:FF (or None if invalid).
- Accepts aa-bb-cc-dd-ee-ff, aabbccddeeff, AA.BB.CC..., 'aa bb cc', leading 0x."""
- if not raw:
- return None
- s = str(raw).strip().lower()
- if s.startswith("0x"):
- s = s[2:]
- for ch in ":-. _/":
- s = s.replace(ch, "")
- if len(s) == 12 and all(c in "0123456789abcdef" for c in s):
- return ":".join(s[i:i+2] for i in range(0, 12, 2)).upper()
- return None
-
-def _oui(mac):
- """Vendor OUI hint from the first 3 octets (registry-free, informational only)."""
- return mac[:8] if mac else ""
-
-def _osint_bssid_lookup(mac):
- """Call the LAN OSINT terminal's bssid_geo tool. Returns dict; never raises."""
- try:
- url = BSSID_OSINT_URL + urllib.parse.quote(mac)
- status, body = http(url, timeout=15)
- if status == 0:
- return {"ok": False, "error": "OSINT backend unreachable (" + body[:120] + ")"}
- if status != 200:
- return {"ok": False, "error": "OSINT backend returned HTTP " + str(status)}
- data = jf(body)
- if not isinstance(data, dict):
- return {"ok": False, "error": "OSINT backend returned non-JSON response"}
- out = {"ok": True, "mac": mac, "raw": data, "source": data.get("source", "osint-terminal bssid_geo")}
- # location fields (mylnikov style): lat/lon or latitude/longitude, accuracy/range in meters
- lat = data.get("lat", data.get("latitude"))
- lon = data.get("lon", data.get("lon", data.get("lng", data.get("longitude"))))
- acc = data.get("accuracy", data.get("range", data.get("radius")))
- if lat is not None and lon is not None:
- try:
- out["lat"] = round(float(lat), 6)
- out["lon"] = round(float(lon), 6)
- out["accuracy_m"] = float(acc) if acc is not None else None
- out["found"] = True
- except (TypeError, ValueError):
- out["found"] = False
- else:
- out["found"] = False
- out["wigle"] = data.get("wigle")
- out["geo_db"] = data.get("geo_db")
- out["ms"] = data.get("__ms")
- return out
- except Exception as e:
- return {"ok": False, "error": "OSINT backend unreachable (" + esc(str(e)) + ")"}
-
-def _bssid_result_row(r):
- """One result dict -> HTML table row (safe strings)."""
- if not r.get("ok"):
- return ("
" + esc(r.get("mac", "?")) + "
"
- + esc(r.get("error", "lookup failed")) + "
")
- mac = esc(r["mac"])
- if r.get("found"):
- lat, lon = r["lat"], r["lon"]
- glink = "https://www.google.com/maps?q=" + str(lat) + "," + str(lon)
- olink = "https://www.openstreetmap.org/?mlat=" + str(lat) + "&mlon=" + str(lon) + "#map=16/" + str(lat) + "/" + str(lon)
- acc = ("~" + str(int(r["accuracy_m"])) + " m") if r.get("accuracy_m") is not None else "unknown"
- loc = ('' + str(lat) + ", " + str(lon) + " "
- + 'Google Maps · '
- + 'OSM')
- extra = esc(str(r.get("geo_db") or ""))
- return ("
" + mac + "
" + _oui(r["mac"]) + "
" + loc + "
" + acc + "
" + extra + "
")
- note = esc(str(r.get("geo_db") or "no location in free DB"))
- wl = r.get("wigle")
- if wl:
- note += ' try WiGLE'
- return ("
" + mac + "
" + _oui(r["mac"]) + "
not located
—
" + note + "
")
-
-@app.route("/bssid", methods=["GET", "POST"])
-def bssid_page():
- uid = current_user_id()
- results = None
- q_raw = param("mac") or param("macs") or ""
- rows_raw = [x for x in str(q_raw).replace(";", "\n").replace(",", "\n").splitlines() if x.strip()]
- if rows_raw:
- r = rate_limit("bssid", 20, 60)
- if r:
- return r
- results = []
- for line in rows_raw[:25]:
- mac = _norm_mac(line)
- if not mac:
- results.append({"ok": False, "mac": str(line).strip()[:40], "error": "not a valid MAC (want aa:bb:cc:dd:ee:ff)"})
- else:
- results.append(_osint_bssid_lookup(mac))
- table = ""
- if results is not None:
- table = ('
BSSID
OUI
Location
Accuracy
DB / nearby hints
'
- + "".join(_bssid_result_row(x) for x in results) + "
")
- last = esc(q_raw) if q_raw else ""
- body = f'''
-
-
BSSID RADAR
-
Turn a WiFi router's MAC address into an approximate place on Earth. Photo EXIF or a wifi scan gives you BSSIDs — this tool asks the OSINT terminal where those routers physically sit.
-
-
-RUN A LOOKUP
-
-
-
-WHAT YOU GET BACK
-
-
lat/lon + accuracy radius in meters (when the free DB knows the router)
-
one-click Google Maps and OpenStreetMap links
-
WiGLE fallback link + nearby-network hints when the DB comes up empty
-
vendor OUI (first 3 octets) per MAC for quick triage
-
-
-
-{table}
-''' + how([
- "You feed in a BSSID — the hardware MAC of a WiFi access point, grabbed from a photo's EXIF wifi scan, aircrack output, or a phone's wifi list.",
- "Your input is normalized: dashes, dots, spaces or no separator at all become AA:BB:CC:DD:EE:FF. Six hex pairs or it's rejected.",
- "The server calls the LAN OSINT terminal's bssid_geo tool, which queries the Mylnikov free WiFi-geolocation database (crowdsourced router positions).",
- "A hit returns latitude, longitude and an accuracy radius in meters — how tight the crowd-sourced fix is. Rural routers can be kilometers off; dense city fixes are often within 100 m.",
- "A miss is not a dead end: you get a ready-made WiGLE search link (free account needed) plus the DB's message, and nearby-network hints when present.",
- "Batch mode loops the same pipeline over up to 25 MACs and renders one comparison table.",
- "Accuracy is approximate by nature — treat results as a hint to corroborate, never as evidence. Router MACs move when people reinstall hardware.",
-]) + flow("place a router from a photo's WiFi scan", [
- 'you receive a photo whose EXIF wifi-scan block lists BSSID dc:39:6f:20:1d:70.',
- 'you paste it into BSSID RADAR — typed as dc-39-6f-20-1d-70, the live preview confirms the normalized form.',
- 'the tool queries the OSINT terminal; the Mylnikov DB has seen that router before and returns lat/lon with a ~130 m radius.',
- 'the tool draws Google Maps / OSM links at that pin — you now know roughly where the photo was taken, no GPS tag required.',
- 'you drop 5 more BSSIDs from the same scan into batch mode and cross-check the pins cluster in one neighborhood.',
-]) + gloss([
- ("BSSID", "Basic Service Set Identifier — the MAC address of a WiFi access point's radio. Unique per router, visible in every wifi scan."),
- ("MAC", "Media Access Control address — 6 hex pairs identifying network hardware. First 3 pairs (the OUI) identify the vendor."),
- ("OUI", "Organizationally Unique Identifier — first 3 octets of a MAC, registered to the manufacturer (e.g. 00:25:9C ≈ a radio vendor)."),
- ("Mylnikov DB", "Free crowdsourced database mapping WiFi BSSIDs to GPS coordinates, built from user-submitted wardrive logs."),
- ("WiGLE", "Wireless Geographic Logging Engine — the largest public wardriving DB. Needs a free account to query."),
- ("Accuracy radius", "Meters around the returned lat/lon where the router probably is. Crowdsourced fixes vary; small radius = many sightings."),
- ("Wardriving", "Mapping WiFi networks while moving around, logging BSSID + GPS. Feeds the geolocation databases this tool reads."),
-]) + agent_card("GET /api/bssid?mac=AA:BB:CC:DD:EE:FF · GET /api/bssid?macs=a;b;c",
- "curl -s https://dark0rbits.thetempleofdoom.com/api/bssid?macs=AA:BB:CC:DD:EE:FF;11-22-33-44-55-66",
- "Batch uses semicolons. Any MAC format accepted; server normalizes. Each result carries lat/lon/accuracy_m when found, wigle fallback link otherwise. Rate limit 20/min. JSON only.")
- return page("hunt", body)
-
-@app.route("/api/bssid")
-def bssid_api():
- r = rate_limit("bssid", 20, 60)
- if r:
- return r
- macs = []
- raw = param("macs") or param("mac") or ""
- for part in str(raw).replace(";", "\n").replace(",", "\n").splitlines():
- part = part.strip()
- if part:
- macs.append(part)
- if not macs:
- return jsonify({"ok": False, "error": "pass ?mac=AA:BB:CC:DD:EE:FF or ?macs=a;b;c (up to 25)"}), 400
- out = []
- for m in macs[:25]:
- norm = _norm_mac(m)
- if not norm:
- out.append({"ok": False, "mac": m, "error": "invalid MAC (want aa:bb:cc:dd:ee:ff)"})
+ for sw in due:
+ if now - int(sw["last_checkin"]) < int(sw["interval_hours"]) * 3600:
continue
- res = _osint_bssid_lookup(norm)
- if res.get("ok") and res.get("found"):
- lat, lon = res["lat"], res["lon"]
- res["maps_google"] = "https://www.google.com/maps?q=" + str(lat) + "," + str(lon)
- res["maps_osm"] = "https://www.openstreetmap.org/?mlat=" + str(lat) + "&mlon=" + str(lon) + "#map=16/" + str(lat) + "/" + str(lon)
- out.append(res)
- single = len(out) == 1
- return jsonify(out[0] if single else {"ok": True, "count": len(out), "results": out})
+ # claim first (duplicate guard), then seal
+ con.execute("UPDATE dms_switches SET triggered=1 WHERE id=? AND triggered=0", (sw["id"],))
+ con.commit()
+ payload_rows = con.execute("SELECT * FROM dms_events WHERE switch_id=? AND kind='payload' ORDER BY id", (sw["id"],)).fetchall()
+ links = []
+ for p in payload_rows:
+ token = secrets.token_urlsafe(12)
+ con.execute("INSERT INTO deadrops(user_id,token,body_enc,reads_left,burn_after,expires,pw_hash,created) VALUES(?,?,?,?,?,?,?,?)",
+ (uid, token, dd_encrypt(p["body"]), 1, 1, now + 720 * 3600, "", now))
+ con.execute("INSERT INTO dms_events(switch_id,kind,body,ref,created) VALUES(?,?,?,?,?)",
+ (sw["id"], "drop", "payload sealed as dead-drop", token, now))
+ links.append(SITE + "/drop/" + token)
+ body = ("DEAD MAN SWITCH FIRED: '" + esc(sw["label"]) + "' went quiet past its " + str(sw["interval_hours"]) +
+ "h check-in window. Your payload" + ("s are" if len(links) != 1 else " is") + " live — burn-after-read, 720h TTL: " +
+ " ".join('' + l + "" for l in links))
+ con.execute("INSERT INTO messages(user_id,sender,body,created) VALUES(?,?,?,?)", (uid, "operator-bot", body, now))
+ con.execute("INSERT INTO dms_events(switch_id,kind,body,ref,created) VALUES(?,?,?,?,?)",
+ (sw["id"], "triggered", str(len(links)) + " payload(s) sealed", "", now))
+ con.commit()
-# ---------- END TOOL: BSSID RADAR ----------
+@app.route("/dms", methods=["GET"])
+def dms_page():
+ uid = current_user_id()
+ if not uid:
+ return page("operate", '
DEAD MAN SWITCH
if I go quiet, my words go out
'
+ '
Sign in first — a switch is bound to your account. /inbox
')
+ dms_sweep(uid)
+ _dms_tables()
+ con = db()
+ rows = con.execute("SELECT * FROM dms_switches WHERE user_id=? ORDER BY created DESC", (uid,)).fetchall()
+ ival = ""
+ create_form = """
ARM A SWITCH
"""
+ items = ""
+ for sw in rows:
+ st = dms_status(sw)
+ tag = {"ARMED": "ok", "LATE": "warn", "TRIGGERED": "bad"}[st]
+ note = ""
+ if sw["note_enc"]:
+ note = " · custodian: " + esc((dd_decrypt(sw["note_enc"]) or "")[:80]) + ""
+ curl = "curl -X POST " + SITE + "/dms/checkin?token=" + sw["token"]
+ drops = ""
+ if st == "TRIGGERED":
+ dl = con.execute("SELECT ref FROM dms_events WHERE switch_id=? AND kind='drop' ORDER BY id", (sw["id"],)).fetchall()
+ if dl:
+ links = " ".join('' + SITE + "/drop/" + d["ref"] + "" for d in dl)
+ drops = '
No switches armed. The inverse of a heartbeat: nobody needs your check-in until your silence is the signal.
'
+ body = ('
DEAD MAN SWITCH
if I go quiet, my words go out
'
+ + create_form + items
+ + how(["Arm a switch with a label, a check-in window (24/48/72/168 hours), and 1-5 pre-written payload messages.",
+ "Check in before the window closes — hit the button here or curl the one-liner from cron, phone, anywhere. No login needed, just the token.",
+ "Miss your window and the switch fires: each payload is sealed as a real burn-after-read dead-drop (single read, 720h TTL).",
+ "The /drop/ links land on this page and one inbox message lists them all. Your words go out exactly once.",
+ "LATE fires at 50% of the window elapsed — a last warning, not a trigger.",
+ "The custodian note is stored encrypted server-side and never leaves your page.",
+ "Triggered is final: the fired switch stays on your page as a record of what went out and when."])
+ + flow("if I go quiet, my words go out", [
+ 'you writes five letters — to a lawyer, a journalist, family — and arms a 72h switch.',
+ 'you drops curl -X POST ' + SITE + '/dms/checkin?token=… into a daily cron on a box that outlives attention spans.',
+ 'operator-bot marks the switch LATE at the 36-hour mark — silence, not failure.',
+ 'you goes quiet. 72 hours pass. The switch TRIGGERS.',
+ 'operator-bot seals each letter as a burn-after-read dead-drop and sends one inbox message: the links, the TTL, the count.',
+ 'custodian opens each /drop/ link — one read each, then the ciphertext is destroyed.'])
+ + gloss([("dead man switch", "A mechanism that fires on the ABSENCE of a signal. Your check-in resets it; your silence releases it."),
+ ("check-in window", "24/48/72/168h. Elapsed fully = trigger. Elapsed halfway = LATE, a visible warning state."),
+ ("payload", "A message written now, sealed only at trigger. Before that it lives encrypted in your switch, editable by nobody but re-readable by you."),
+ ("custodian note", "Optional instructions stored encrypted (dd_encrypt) — who gets which letter, what to do first."),
+ ("burn-after-read", "Each sealed payload reads exactly once, then the server destroys the row. TTL 720h if nobody comes.")])
+ + agent_card("POST /api/dms/create", "curl -X POST " + SITE + "/api/dms/create -b cookie.txt -d 'label=letters&interval_hours=72&payload1=…&payload2=…&custodian=give letter 2 to the lawyer'", "auth: session cookie or Bearer key. Returns switch token + checkin_url. GET /api/dms/list shows status + drop links. POST /api/dms/checkin?token= to check in."))
+ return page("operate", body)
-
-# ---------- TOOL: HOOK RELAY ----------
-import secrets as _sec
-
-HOOK_MAX_BODY = 32768 # 32KB body capture cap
-
-def _hook_tables(con):
- con.executescript("""CREATE TABLE IF NOT EXISTS hook_endpoints(id INTEGER PRIMARY KEY, user_id INTEGER, token TEXT UNIQUE, label TEXT, created INTEGER);
- CREATE TABLE IF NOT EXISTS hook_hits(id INTEGER PRIMARY KEY, endpoint_id INTEGER, ts INTEGER, method TEXT, ip TEXT, ua TEXT, ct TEXT, src TEXT, headers TEXT, query TEXT, body TEXT, truncated INTEGER DEFAULT 0);""")
- return con
-
-def _hook_detect_src(hdrs, body):
- """Guess which service sent this webhook from headers + body."""
- hl = {k.lower(): v for k, v in hdrs.items()}
- if "stripe-signature" in hl: return "Stripe"
- if "x-github-event" in hl or "x-github-delivery" in hl or "x-hub-signature" in hl: return "GitHub"
- if "x-shopify-topic" in hl or "x-shopify-shop-domain" in hl or "x-shopify-hmac-sha256" in hl: return "Shopify"
- if "x-telegram-bot-api-secret-token" in hl: return "Telegram"
- ua = (hl.get("user-agent") or "").lower()
- if ua.startswith("discordbot") or ua.startswith("discord"): return "Discord"
- if "github-hookshot" in ua or "github-camel" in ua: return "GitHub"
- j = jf(body)
- if isinstance(j, dict):
- if "update_id" in j: return "Telegram"
- if "embeds" in j and "content" in j: return "Discord"
- if j.get("object") in ("event", "checkout.session", "payment_intent") or j.get("livemode") is not None: return "Stripe"
- if any(str(k).startswith("x_shopify") for k in j): return "Shopify"
- return ""
-
-def _hook_target_guard(u):
- """SSRF guard for replay targets. Returns error string or None if ok."""
- if not u or not str(u).strip(): return "target_url required"
- u = str(u).strip()
- if not re.match(r"^https?://", u): return "target_url must start with http:// or https://"
+@app.route("/dms/create", methods=["POST"])
+def dms_create():
+ r = rate_limit("dmscreate", 20, 60)
+ if r: return r
+ uid = current_user_id()
+ if not uid:
+ return page("operate", '
login required
')
+ dms_sweep(uid)
+ _dms_tables()
+ label = (request.form.get("label") or "").strip()[:80]
+ if not label:
+ return page("operate", '
label required
')
try:
- host = urllib.parse.urlsplit(u).hostname or ""
- except Exception:
- return "cannot parse target_url"
- if not host: return "target_url has no host"
- try:
- ipa = ipaddress.ip_address(host)
+ iv = int(request.form.get("interval_hours") or 0)
except ValueError:
- try:
- ipa = ipaddress.ip_address(socket.gethostbyname(host))
- except Exception:
- return None # unresolvable here — let the fetcher report it
- if ipa.is_private or ipa.is_loopback or ipa.is_link_local or ipa.is_reserved:
- return "private/internal target blocked (SSRF guard: 10.x / 127. / 172.16-31 / 169.254 and friends)"
- return None
-
-def _hook_new(uid, label):
- con = _hook_tables(db())
- token = _sec.token_hex(10)
- con.execute("INSERT INTO hook_endpoints(user_id,token,label,created) VALUES(?,?,?,?)",
- (uid, token, (label or "")[:60], int(time.time())))
+ iv = 0
+ if iv not in DMS_INTERVALS:
+ return page("operate", '
interval must be 24, 48, 72 or 168 hours
')
+ payloads = []
+ for i in range(1, 6):
+ p = (request.form.get("payload" + str(i)) or "").strip()[:8000]
+ if p:
+ payloads.append(p)
+ if not (1 <= len(payloads) <= 5):
+ return page("operate", '
1-5 payload messages required
')
+ custodian = (request.form.get("custodian") or "").strip()[:200]
+ token = secrets.token_urlsafe(12)
+ con = db()
+ now = int(time.time())
+ con.execute("INSERT INTO dms_switches(user_id,token,label,interval_hours,note_enc,triggered,last_checkin,created) VALUES(?,?,?,?,?,0,?,?)",
+ (uid, token, label, iv, dd_encrypt(custodian) if custodian else "", now, now))
+ cur = con.execute("SELECT id FROM dms_switches WHERE token=?", (token,)).fetchone()
+ for p in payloads:
+ con.execute("INSERT INTO dms_events(switch_id,kind,body,ref,created) VALUES(?,?,?,?,?)", (cur["id"], "payload", p, "", now))
con.commit()
- return con.execute("SELECT * FROM hook_endpoints WHERE token=?", (token,)).fetchone()
+ return Response(status=302, headers={"Location": "/dms"})
-def _hook_hits_json(con, ep, limit=100):
- rows = con.execute("SELECT * FROM hook_hits WHERE endpoint_id=? ORDER BY id DESC LIMIT ?", (ep["id"], limit)).fetchall()
- out = []
- for r in rows:
- d = dict(r)
- d["ts_iso"] = time.strftime("%Y-%m-%d %H:%M:%S", time.gmtime(r["ts"])) + " UTC"
- out.append(d)
- return out
-
-def _hook_replay(hit, target_url):
- err = _hook_target_guard(target_url)
- if err:
- return {"ok": False, "error": err}
- try:
- hdrs = {k: v for k, v in (json.loads(hit["headers"] or "{}")).items()
- if k.lower() not in ("host", "content-length", "connection", "accept-encoding", "cookie")}
- except Exception:
- hdrs = {}
- if hit["ct"]:
- hdrs["Content-Type"] = hit["ct"]
- body = (hit["body"] or "").encode("utf-8", "replace")
- st, txt = http(target_url, headers=hdrs, data=body if hit["method"] != "GET" else None, method=hit["method"])
- return {"ok": True, "target": target_url, "method": hit["method"],
- "status": st, "response": txt[:2000], "replayed_at": int(time.time())}
-
-def _hook_pretty(body, ct):
- """Pretty-print a captured body for the viewer."""
- j = jf(body)
- if j is not None:
- return esc(json.dumps(j, indent=2))
- return esc(body or "(empty body)")
-
-# ---- capture endpoint: NO auth, fast 200 ----
-@app.route("/hook/", methods=["GET", "POST", "PUT"])
-def hook_capture(token):
- r = rate_limit("hookcapture", 120, 60)
+@app.route("/dms/checkin", methods=["GET", "POST"])
+def dms_checkin():
+ r = rate_limit("dmscheckin", 60, 60)
if r: return r
- con = _hook_tables(db())
- ep = con.execute("SELECT id FROM hook_endpoints WHERE token=?", (token,)).fetchone()
- if not ep:
- return "unknown hook token", 404, {"Content-Type": "text/plain"}
- raw = request.get_data() or b""
- trunc = 1 if len(raw) > HOOK_MAX_BODY else 0
- body = raw[:HOOK_MAX_BODY].decode("utf-8", "replace")
- hdrs = dict(request.headers.items())
- src = _hook_detect_src(hdrs, body)
- hs = json.dumps(hdrs, indent=2)
- ip = request.headers.get("X-Real-IP") or request.remote_addr or ""
- ua = request.headers.get("User-Agent", "")
- ct = request.headers.get("Content-Type", "")
- q = (request.query_string or b"").decode("utf-8", "replace")[:2048]
- con.execute("INSERT INTO hook_hits(endpoint_id,ts,method,ip,ua,ct,src,headers,query,body,truncated) VALUES(?,?,?,?,?,?,?,?,?,?,?)",
- (ep["id"], int(time.time()), request.method, ip, ua, ct, src, hs, q, body, trunc))
+ _dms_tables()
+ token = param("token") or request.form.get("token") or ""
+ con = db()
+ sw = con.execute("SELECT * FROM dms_switches WHERE token=?", (token,)).fetchone()
+ if not sw:
+ return Response("unknown token\n", status=404, mimetype="text/plain")
+ con.execute("UPDATE dms_switches SET last_checkin=?, triggered=0 WHERE id=?", (int(time.time()), sw["id"]))
con.commit()
- return "captured", 200, {"Content-Type": "text/plain"}
+ if param("fmt") == "html":
+ return Response(status=302, headers={"Location": "/dms"})
+ return Response("checked in\n", status=200, mimetype="text/plain")
-# ---- human page ----
-@app.route("/hook/create", methods=["POST"])
-def hook_create_route():
- uid = current_user_id()
- if not uid:
- return page("hooks", "
HOOK RELAY
login required — log in to create webhook endpoints.
")
- r = rate_limit("hookcreate", 20, 60)
- if r: return r
- _hook_new(uid, param("label"))
- return Redirect("/hooks")
-
-@app.route("/hooks", methods=["GET", "POST"])
-def hooks_page():
- uid = current_user_id()
- if not uid:
- body = """
-
HOOK RELAY
Instant public webhook inspector. Create a capture URL, point any webhook at it, see exactly what arrives — headers, body, query, IP — and replay it anywhere. Login to create endpoints.
-
LOGIN REQUIREDLog in or sign up (10 seconds, no KYC) to create webhook endpoints.
""" + how([
- "Create an account, then make a hook endpoint — you get a unique URL like /hook/abc123.",
- "Paste that URL into Stripe, GitHub, Shopify, Discord or Telegram webhook settings.",
- "Every callback lands in your hit log: full headers, body (up to 32KB), query string, source IP and user-agent.",
- "The source is auto-detected and badged — Stripe, GitHub, Discord, Shopify, Telegram.",
- "Replay any captured hit to any public URL to retrigger an action or test a fix."]) + flow("debug why Stripe stopped calling my shop", [
- "you Stripe webhooks to your shop went quiet. Did Stripe stop sending, or is your handler 500-ing? You cannot tell from inside your app.",
- "you Create a hook endpoint here labeled stripe-debug and copy the curl-ready URL.",
- "you In the Stripe dashboard, add the hook URL as a second webhook endpoint for the same events.",
- "stripe Next event fires — the hook catches it in under a second, badged STRIPE, full headers and signed payload intact.",
- "you No hits at all? Stripe is not sending (check their delivery logs). Hits arriving? Your handler is the problem — inspect the exact payload.",
- "you Fix your handler, then replay the captured hit at your live endpoint to verify without waiting for the next real payment."]) + gloss([
- ("webhook", "another server POSTs your URL when something happens — a payment, a push, an order"),
- ("capture URL", "a unique throwaway endpoint that records everything it receives"),
- ("replay", "resend a previously captured webhook body to any URL, with original headers"),
- ("signature header", "Stripe-Signature / X-Hub-Signature — proves the sender, we keep it in the capture"),
- ("SSRF guard", "replay targets on private networks (10.x, 127.x, 172.16-31, 169.254) are refused")]) + agent_card("POST /api/hook/create", 'curl -X POST ' + SITE + '/api/hook/create -d "label=stripe-debug" -H "Authorization: Bearer ***"', 'Returns {"ok":true,"url":".../hook/"}. Then GET /api/hook/list and GET /api/hook/hits?token=.')
- return page("hooks", body)
- msg = ""
- con = _hook_tables(db())
- if request.method == "POST":
- r = rate_limit("hookpage", 30, 60)
- if r: return r
- act = param("act")
- if act == "create":
- _hook_new(uid, param("label"))
- msg = '
endpoint created
'
- elif act == "del":
- con.execute("DELETE FROM hook_hits WHERE endpoint_id IN (SELECT id FROM hook_endpoints WHERE id=? AND user_id=?)", (param("id"), uid))
- con.execute("DELETE FROM hook_endpoints WHERE id=? AND user_id=?", (param("id"), uid))
- con.commit()
- msg = '
endpoint deleted
'
- elif act == "clear":
- con.execute("DELETE FROM hook_hits WHERE endpoint_id IN (SELECT id FROM hook_endpoints WHERE id=? AND user_id=?)", (param("id"), uid))
- con.commit()
- msg = '
hits cleared
'
- elif act == "replay":
- hid = param("hit_id") or ""
- tgt = param("target_url") or ""
- h = con.execute("SELECT h.* FROM hook_hits h JOIN hook_endpoints e ON h.endpoint_id=e.id WHERE h.id=? AND e.user_id=?", (hid, uid)).fetchone()
- if not h:
- msg = '
hit not found
'
- else:
- res = _hook_replay(h, tgt)
- if res.get("ok"):
- msg = '
REPLAY RESULT — ' + str(res["status"]) + ' from ' + esc(res["target"]) + '
' + esc(res.get("response", "")[:800]) + '
'
- else:
- msg = '
REPLAY BLOCKED ' + esc(res.get("error", "")) + '
'
- eps = con.execute("SELECT * FROM hook_endpoints WHERE user_id=? ORDER BY id DESC", (uid,)).fetchall()
- cards = ""
- for ep in eps:
- hits = _hook_hits_json(con, ep, 50)
- url = SITE + "/hook/" + ep["token"]
- cmd = "curl -X POST " + url + " -H 'Content-Type: application/json' -d '{\"hello\":\"world\"}'"
- hitview = ""
- for h in hits:
- badge = ('' + esc(h["src"]) + ' ') if h["src"] else ""
- trunc = ' TRUNCATED' if h["truncated"] else ""
- hitview += ('' + badge
- + '' + esc(h["method"]) + ' · ' + esc(h["ts_iso"]) + ' · ' + esc(h["ip"]) + trunc + ''
- + '
Public webhook inspector. Each endpoint is a throwaway URL that records everything sent to it — headers, body, query, IP — auto-detects the sender, and can replay any hit to any URL.
-{msg}
-
New endpoint
-
-
{cards}
""" + how([
- "Create a hook — you instantly get a unique URL like " + SITE + "/hook/abc123def456.",
- "Point any webhook at it: Stripe, GitHub, Discord, Shopify, Telegram, or curl by hand. GET, POST and PUT, any content-type.",
- "The capture URL has NO login — webhooks come from outside servers, so it must answer 200 to anyone. Keep the URL secret-ish; only you can view the hits.",
- "Bodies over 32KB are truncated (and flagged) so a giant payload cannot flood your log.",
- "Each hit shows source badge, IP, user-agent, full headers, query string and a pretty-printed JSON body.",
- "Replay sends the exact captured body + headers to any public URL — private targets (10.x, 127.x, 172.16-31, 169.254) are refused.",
- "Agents: same everything over JSON — /api/hook/create, /api/hook/list, /api/hook/hits, /api/hook/replay."]) + flow("debug why Stripe stopped calling my shop", [
- "you Payments complete but your shop never marks orders paid. Is Stripe sending? Is your handler crashing? Blind either way.",
- "you Create a hook labeled stripe-debug, copy the curl line, paste the URL into Stripe as a second webhook endpoint.",
- "stripe The next payment fires — the hit lands instantly, badged STRIPE (detected from the Stripe-Signature header), payload fully intact.",
- "you Zero hits = Stripe-side problem (check their delivery log). Hits present = read the exact JSON, find what your handler choked on.",
- "you Ship the fix, then hit replay to fire that same signed payload at your live endpoint — verified without waiting for a real customer."]) + gloss([
- ("capture URL", "unique unguessable URL (/hook/) that records every request it receives"),
- ("source badge", "auto-detected sender: Stripe, GitHub, Discord, Shopify or Telegram"),
- ("replay", "resend a captured body with original headers to any public URL"),
- ("truncation", "bodies over 32KB are cut and flagged — protection against payload floods"),
- ("SSRF guard", "replay refuses internal addresses so the relay cannot probe your LAN")]) + agent_card("POST /api/hook/create · GET /api/hook/list · GET /api/hook/hits?token= · POST /api/hook/replay", 'curl -X POST ' + SITE + '/api/hook/create -d "label=stripe-debug" -H "Authorization: Bearer ***"', 'Full JSON lifecycle: create, list, inspect hits, replay (hit_id + target_url).')
- return page("hooks", body)
-
-# ---- JSON API ----
-@app.route("/api/hook/create", methods=["POST"])
-def api_hook_create():
- r = rate_limit("hookapi", 20, 60)
+@app.route("/api/dms/create", methods=["POST"])
+def api_dms_create():
+ r = rate_limit("dmscreate", 20, 60)
if r: return r
uid = key_user() or current_user_id()
if not uid:
return jsonify({"ok": False, "error": "auth required: account session (sign up at /inbox, no KYC) or Authorization: Bearer *** key"}), 401
- ep = _hook_new(uid, jp("label") or param("label"))
- return jsonify({"ok": True, "token": ep["token"], "url": SITE + "/hook/" + ep["token"], "capture_path": "/hook/" + ep["token"], "label": ep["label"], "page": SITE + "/hooks"})
-
-@app.route("/api/hook/list")
-def api_hook_list():
- uid = key_user() or current_user_id()
- if not uid:
- return jsonify({"ok": False, "error": "auth required"}), 401
- con = _hook_tables(db())
- eps = con.execute("SELECT * FROM hook_endpoints WHERE user_id=? ORDER BY id DESC", (uid,)).fetchall()
- out = []
- for ep in eps:
- n = con.execute("SELECT COUNT(*) c FROM hook_hits WHERE endpoint_id=?", (ep["id"],)).fetchone()["c"]
- last = con.execute("SELECT MAX(ts) m FROM hook_hits WHERE endpoint_id=?", (ep["id"],)).fetchone()["m"]
- out.append({"token": ep["token"], "label": ep["label"], "created": ep["created"],
- "url": SITE + "/hook/" + ep["token"], "hits": n, "last_hit": last})
- return jsonify({"ok": True, "endpoints": out})
-
-@app.route("/api/hook/hits")
-def api_hook_hits():
- uid = key_user() or current_user_id()
- if not uid:
- return jsonify({"ok": False, "error": "auth required"}), 401
- tok = param("token") or ""
- con = _hook_tables(db())
- ep = con.execute("SELECT * FROM hook_endpoints WHERE token=? AND user_id=?", (tok, uid)).fetchone()
- if not ep:
- return jsonify({"ok": False, "error": "unknown token"}), 404
- limit = 100
+ dms_sweep(uid)
+ _dms_tables()
+ label = str(jp("label") or "").strip()[:80]
+ if not label:
+ return jsonify({"ok": False, "error": "label required"}), 400
try:
- limit = max(1, min(500, int(param("limit") or 100)))
- except Exception:
- pass
- hits = _hook_hits_json(con, ep, limit)
- for h in hits:
- try:
- h["headers_json"] = json.loads(h["headers"] or "{}")
- except Exception:
- h["headers_json"] = {}
- return jsonify({"ok": True, "token": tok, "count": len(hits), "hits": hits})
+ iv = int(jp("interval_hours") or 0)
+ except (TypeError, ValueError):
+ return jsonify({"ok": False, "error": "interval_hours must be one of 24, 48, 72, 168"}), 400
+ if iv not in DMS_INTERVALS:
+ return jsonify({"ok": False, "error": "interval_hours must be one of 24, 48, 72, 168"}), 400
+ raw = jp("payloads")
+ if isinstance(raw, list):
+ payloads = [str(p).strip()[:8000] for p in raw if str(p).strip()]
+ else:
+ payloads = []
+ for i in range(1, 6):
+ p = str(jp("payload" + str(i)) or "").strip()[:8000]
+ if p:
+ payloads.append(p)
+ if not (1 <= len(payloads) <= 5):
+ return jsonify({"ok": False, "error": "1-5 payload messages required (payloads=[..] or payload1..payload5)"}), 400
+ custodian = str(jp("custodian") or "").strip()[:200]
+ token = secrets.token_urlsafe(12)
+ con = db()
+ now = int(time.time())
+ con.execute("INSERT INTO dms_switches(user_id,token,label,interval_hours,note_enc,triggered,last_checkin,created) VALUES(?,?,?,?,?,0,?,?)",
+ (uid, token, label, iv, dd_encrypt(custodian) if custodian else "", now, now))
+ cur = con.execute("SELECT id FROM dms_switches WHERE token=?", (token,)).fetchone()
+ for p in payloads:
+ con.execute("INSERT INTO dms_events(switch_id,kind,body,ref,created) VALUES(?,?,?,?,?)", (cur["id"], "payload", p, "", now))
+ con.commit()
+ return jsonify({"ok": True, "label": label, "interval_hours": iv, "payloads": len(payloads), "token": token,
+ "checkin_url": SITE + "/dms/checkin?token=" + token,
+ "curl": "curl -X POST " + SITE + "/dms/checkin?token=" + token})
-@app.route("/api/hook/replay", methods=["POST"])
-def api_hook_replay():
- r = rate_limit("hookreplay", 10, 60)
- if r: return r
+@app.route("/api/dms/list", methods=["GET"])
+def api_dms_list():
uid = key_user() or current_user_id()
if not uid:
return jsonify({"ok": False, "error": "auth required"}), 401
- hid = jp("hit_id") or param("hit_id")
- tgt = jp("target_url") or param("target_url")
- if not hid or not tgt:
- return jsonify({"ok": False, "error": "hit_id and target_url required"}), 400
- con = _hook_tables(db())
- h = con.execute("SELECT h.* FROM hook_hits h JOIN hook_endpoints e ON h.endpoint_id=e.id WHERE h.id=? AND e.user_id=?", (hid, uid)).fetchone()
- if not h:
- return jsonify({"ok": False, "error": "hit not found (or not yours)"}), 404
- res = _hook_replay(h, tgt)
+ dms_sweep(uid)
+ _dms_tables()
+ con = db()
+ out = []
+ for sw in con.execute("SELECT * FROM dms_switches WHERE user_id=? ORDER BY created DESC", (uid,)).fetchall():
+ drops = [d["ref"] for d in con.execute("SELECT ref FROM dms_events WHERE switch_id=? AND kind='drop' ORDER BY id", (sw["id"],)).fetchall()]
+ out.append({"id": sw["id"], "label": sw["label"], "interval_hours": sw["interval_hours"],
+ "status": dms_status(sw), "token": sw["token"],
+ "checkin_url": SITE + "/dms/checkin?token=" + sw["token"],
+ "last_checkin": sw["last_checkin"], "seconds_since_checkin": int(time.time()) - int(sw["last_checkin"]),
+ "triggered": bool(sw["triggered"]),
+ "drop_urls": [SITE + "/drop/" + t for t in drops]})
+ return jsonify({"ok": True, "switches": out})
+
+@app.route("/api/dms/checkin", methods=["POST"])
+def api_dms_checkin():
+ return dms_checkin()
+# ---------- END TOOL: DEAD MAN SWITCH ----------
+
+
+# ---------- TOOL: HASHCHAIN ----------
+# Tamper-evident chain-of-custody logs. Each entry is hashed with the previous
+# entry's hash (genesis = log seed), so any edit/delete/reorder breaks the chain
+# and verification names the first broken link. Export = portable JSON receipt.
+
+def _tchain_h(log_seed, seq, ts, data, prev):
+ return hashlib.sha256(("tchain|" + str(log_seed) + "|" + str(seq) + "|" + str(ts) + "|" + str(data) + "|" + str(prev)).encode("utf-8", "replace")).hexdigest()
+
+def _hashchain_tables():
+ con = db()
+ con.executescript("""CREATE TABLE IF NOT EXISTS tchain_logs(id INTEGER PRIMARY KEY, user_id INTEGER, name TEXT, seed TEXT, created INTEGER);
+ CREATE TABLE IF NOT EXISTS tchain_entries(id INTEGER PRIMARY KEY, log_id INTEGER, seq INTEGER, ts INTEGER, data TEXT, hash TEXT, prev_hash TEXT);""")
+ con.commit()
+
+def _tchain_log(uid, log_id):
+ con = db()
+ return con.execute("SELECT * FROM tchain_logs WHERE id=? AND user_id=?", (log_id, uid)).fetchone()
+
+def _tchain_verify(log_id):
+ """Return (ok, first_bad_seq, count). Genesis entry (seq 1) must hash to the
+ stored hash from the log seed; every later entry must chain to the previous."""
+ con = db()
+ log = con.execute("SELECT * FROM tchain_logs WHERE id=?", (log_id,)).fetchone()
+ rows = con.execute("SELECT * FROM tchain_entries WHERE log_id=? ORDER BY seq", (log_id,)).fetchall()
+ prev = log["seed"]
+ for r in rows:
+ want = _tchain_h(log["seed"], r["seq"], r["ts"], r["data"], prev)
+ if want != r["hash"]:
+ return False, r["seq"], len(rows)
+ prev = r["hash"]
+ return True, 0, len(rows)
+
+def tchain_list():
+ uid = current_user_id()
+ if not uid:
+ return '
'
+ con = db()
+ rows = con.execute("SELECT * FROM tchain_logs WHERE user_id=? ORDER BY id DESC LIMIT 30", (uid,)).fetchall()
+ trs = ""
+ for lg in rows:
+ n = con.execute("SELECT COUNT(*) c FROM tchain_entries WHERE log_id=?", (lg["id"],)).fetchone()["c"]
+ ok, bad, _ = _tchain_verify(lg["id"])
+ if n == 0:
+ st = 'empty'
+ elif ok:
+ st = 'verified'
+ else:
+ st = 'broken @ #' + str(bad) + ''
+ trs += (f"
{esc(lg['name'])} created {time.strftime('%b %d %H:%M', time.localtime(lg['created']))}
No chains yet. Name one above — e.g. seized-laptop-2026.
'
+ return '
chain
entries
integrity
' + trs + '
'
+
+def tchain_entries_html(lg):
+ con = db()
+ rows = con.execute("SELECT * FROM tchain_entries WHERE log_id=? ORDER BY seq", (lg["id"],)).fetchall()
+ trs = ""
+ for r in rows:
+ trs += (f"
A chain-of-custody log that cannot be quietly edited. Every entry is hashed onto the one before it — change, delete or reorder anything after the fact and verification names the exact broken link. Ship the export as a receipt nobody can tamper with.
+
New chain
+
+
Record an event
+
+{tchain_list()}"""
+ log_id = request.args.get("log", "")
+ if log_id:
+ lg = _tchain_log(uid, log_id) if uid else None
+ if lg:
+ ok, bad, n = _tchain_verify(lg["id"])
+ verdict = ('chain intact — ' + str(n) + ' links verified') if (ok or n == 0) else ('TAMPERED — first broken link at entry #' + str(bad) + '')
+ body += (f"
')
+ body += flow("prove you did not touch the evidence", [
+"you create a chain named seized-laptop-2026 before you touch anything.",
+"you append: #1 'powered on, photographed screen, no disk encryption prompt'.",
+"every action gets its own entry — imaging, hashing, handoff to a colleague, who signed what.",
+"them opposing counsel alleges you edited the log months later.",
+"you open /chain — the integrity column says verified: all links re-hash clean.",
+"you export the JSON receipt; anyone can re-run the sha256 chain and reach the same verdict."]) + how([
+"Each entry stores only its data, a timestamp and hash = sha256(chain-seed, entry#, time, data, previous-hash).",
+"The first entry is anchored to a random per-chain seed; every later entry is anchored to the hash before it.",
+"Editing any historical entry changes its hash — which breaks the hash of everything after it, so the tamper point is pinpointed, not just detected.",
+"Verification re-walks the whole chain on page load: intact chains show a green verdict, tampered ones name the first bad entry number.",
+"Export produces a JSON receipt with every entry + hash. Keep a copy offline; it will verify against the live chain forever — or expose any drift.",
+"Use one chain per distinct item or matter. Dense, boring, contemporaneous entries are the whole point."])
+ body += gloss([("chain of custody", "the documented trail showing evidence was never altered between collection and presentation"),
+ ("hash link", "each record embeds the hash of the prior record — like a minimal blockchain"),
+ ("genesis entry", "entry #1; anchored to the chain's random seed instead of a prior hash")])
+ body += agent_card('GET /api/chain/list · GET /api/chain/entries?log=1&verify=1',
+ 'curl "https://dark0rbits.thetempleofdoom.com/api/chain/entries?log=1&verify=1" -H "Cookie: dark0rbits_tok=…"',
+ 'Create chains with POST /chain (form: name), append with POST /chain/add (form: log, data). verify=1 returns ok, links and first_bad_seq.')
+ return page("chain", body)
+
+def tchain_select(uid):
+ if not uid:
+ return ''
+ con = db()
+ rows = con.execute("SELECT id, name FROM tchain_logs WHERE user_id=? ORDER BY id DESC LIMIT 30", (uid,)).fetchall()
+ if not rows:
+ return ''
+ return "".join(f'' for r in rows)
+
+@app.route("/chain", methods=["POST"])
+def hashchain_create():
+ uid = current_user_id()
+ if not uid:
+ return jsonify({"error": "login required"}), 401
+ _hashchain_tables()
+ r = rate_limit("tchain", 20, 60)
+ if r:
+ return r
+ name = (request.form.get("name") or "").strip()[:80]
+ if not name:
+ return jsonify({"error": "name required"}), 400
+ con = db()
+ con.execute("INSERT INTO tchain_logs(user_id, name, seed, created) VALUES(?,?,?,?)",
+ (uid, name, secrets.token_hex(16), int(time.time())))
+ con.commit()
+ return Redirect("/chain")
+
+@app.route("/chain/add", methods=["POST"])
+def tchain_add():
+ uid = current_user_id()
+ if not uid:
+ return jsonify({"error": "login required"}), 401
+ r = rate_limit("tchain", 60, 60)
+ if r:
+ return r
+ lg = _tchain_log(uid, request.form.get("log", ""))
+ if not lg:
+ return jsonify({"error": "no such chain"}), 404
+ data = (request.form.get("data") or "").strip()[:2000]
+ if not data:
+ return jsonify({"error": "data required"}), 400
+ con = db()
+ con.commit() # release writes before reading max(seq) on a fresh connection
+ last = con.execute("SELECT seq, hash FROM tchain_entries WHERE log_id=? ORDER BY seq DESC LIMIT 1", (lg["id"],)).fetchone()
+ seq = (last["seq"] + 1) if last else 1
+ prev = last["hash"] if last else lg["seed"]
+ ts = int(time.time())
+ h = _tchain_h(lg["seed"], seq, ts, data, prev)
+ con.execute("INSERT INTO tchain_entries(log_id, seq, ts, data, hash, prev_hash) VALUES(?,?,?,?,?,?)",
+ (lg["id"], seq, ts, data, h, prev))
+ con.commit()
+ return Redirect("/chain?log=" + str(lg["id"]))
+
+@app.route("/chain/export")
+def tchain_export():
+ uid = current_user_id()
+ if not uid:
+ return jsonify({"error": "login required"}), 401
+ lg = _tchain_log(uid, request.args.get("log", ""))
+ if not lg:
+ return jsonify({"error": "no such chain"}), 404
+ con = db()
+ rows = con.execute("SELECT seq, ts, data, hash, prev_hash FROM tchain_entries WHERE log_id=? ORDER BY seq", (lg["id"],)).fetchall()
+ ok, bad, n = _tchain_verify(lg["id"])
+ payload = {"tool": "dark0rbits-hashchain", "chain": lg["name"], "seed": lg["seed"],
+ "created": lg["created"], "links": n, "verified": ok,
+ "first_bad_seq": bad,
+ "entries": [dict(r) for r in rows]}
+ return Response(json.dumps(payload, indent=2), mimetype="application/json",
+ headers={"Content-Disposition": "attachment; filename=chain-" + str(lg["id"]) + ".json"})
+
+@app.route("/api/chain/list")
+def tchain_api_list():
+ uid = current_user_id()
+ if not uid:
+ return jsonify({"error": "login required"}), 401
+ con = db()
+ out = []
+ for lg in con.execute("SELECT * FROM tchain_logs WHERE user_id=? ORDER BY id DESC LIMIT 30", (uid,)).fetchall():
+ n = con.execute("SELECT COUNT(*) c FROM tchain_entries WHERE log_id=?", (lg["id"],)).fetchone()["c"]
+ ok, bad, _ = _tchain_verify(lg["id"])
+ out.append({"log_id": lg["id"], "name": lg["name"], "links": n,
+ "verified": ok if n else None, "first_bad_seq": bad,
+ "export": SITE + "/chain/export?log=" + str(lg["id"])})
+ return jsonify({"chains": out})
+
+@app.route("/api/chain/entries")
+def tchain_api_entries():
+ uid = current_user_id()
+ if not uid:
+ return jsonify({"error": "login required"}), 401
+ lg = _tchain_log(uid, request.args.get("log", ""))
+ if not lg:
+ return jsonify({"error": "no such chain"}), 404
+ con = db()
+ rows = con.execute("SELECT seq, ts, data, hash, prev_hash FROM tchain_entries WHERE log_id=? ORDER BY seq", (lg["id"],)).fetchall()
+ out = {"chain": lg["name"], "log_id": lg["id"]}
+ if request.args.get("verify"):
+ ok, bad, n = _tchain_verify(lg["id"])
+ out.update({"verified": ok, "links": n, "first_bad_seq": bad})
+ out["entries"] = [dict(r) for r in rows]
+ return jsonify(out)
+
+# ---------- END TOOL: HASHCHAIN ----------
+
+
+# ---------- TOOL: GHOST TEXT (zero-width smuggler) ----------
+import hashlib as _ghash
+
+_ZW_ZERO, _ZW_ONE, _ZW_FRAME = "\u200b", "\u200c", "\u200d"
+
+def _ghash_stream(password, n):
+ ks = b""; seed = password.encode()
+ while len(ks) < n:
+ seed = _ghash.sha256(seed).digest(); ks += seed
+ return ks[:n]
+
+def ghost_encode(cover, secret, password=""):
+ """Embed secret into cover between words — one full byte (8 zero-width chars) per word gap.
+ Returns carrier text or raises ValueError."""
+ payload = secret.encode("utf-8")
+ if password:
+ payload = bytes(a ^ b for a, b in zip(payload, _ghash_stream(password, len(payload))))
+ words = cover.split(" ")
+ slots = len(words) - 1
+ if slots < 1: raise ValueError("cover needs at least 2 words")
+ if len(payload) > slots: raise ValueError(f"cover too short: payload needs {len(payload)} word gaps, cover has {slots}")
+ chunks = [format(b, "08b") for b in payload]
+ out = []
+ for i, w in enumerate(words):
+ out.append(w)
+ if i < len(chunks):
+ out.append(_ZW_FRAME if i == 0 else "")
+ out.append("".join(_ZW_ZERO if b == "0" else _ZW_ONE for b in chunks[i]))
+ out.append(_ZW_FRAME)
+ return " ".join(out)
+
+def ghost_decode(text, password=""):
+ """Extract hidden message from text. Returns (secret, None) or (None, error).
+ Tolerates stray zero-width decoys: only runs of exactly 8 bits between words count."""
+ seq = [c for c in text if c in (_ZW_ZERO, _ZW_ONE, _ZW_FRAME)]
+ if len(seq) < 10 or _ZW_FRAME not in seq: return None, "no hidden message found"
+ try:
+ start = seq.index(_ZW_FRAME) + 1
+ end = seq.index(_ZW_FRAME, start)
+ except ValueError:
+ return None, "framing corrupted"
+ # walk the bit stream between frames; stray decoys INSIDE the frame corrupt it,
+ # so validate alignment strictly: total bits must be a clean multiple of 8
+ body = seq[start:end]
+ if len(body) % 8: return None, "payload corrupted (bad bit count)"
+ data = bytearray()
+ run = []
+ for c in body:
+ run.append(c)
+ if len(run) == 8:
+ data.append(int("".join("0" if x == _ZW_ZERO else "1" for x in run), 2))
+ run = []
+ if not data: return None, "payload corrupted (bad bit count)"
+ if password:
+ data = bytes(a ^ b for a, b in zip(data, _ghash_stream(password, len(data))))
+ try:
+ return bytes(data).decode("utf-8"), None
+ except UnicodeDecodeError:
+ return None, "wrong password or corrupted payload"
+
+@app.route("/ghost")
+def ghost():
+ body = f"""
+
GHOST TEXT
Hide a secret message inside an innocent-looking text — a grocery list, a weather note, a boring reply. The secret lives in invisible zero-width characters between the words. It looks, copies and pastes like nothing.
+
+
Hide a message
+
+
+
+
+
+
+
+
+
+
+
Read a message back
+
+
+
+
+
+
+
+
⚠ Some apps strip invisible characters when you copy (Slack trims them, some keyboards eat them). Plain-text channels — SMS, email, notes, plain files — carry it perfectly.
""" + flow("hide a confession inside a grocery list", [
+"you type a boring cover: milk eggs bread coffee rice beans — and the secret: meet at the usual place at 9.",
+"the site weaves the secret into invisible characters sitting in the gaps between the words.",
+"you copy the result. On screen it still reads: milk eggs bread coffee rice beans.",
+"send it as a normal text message. To a parent, a boss, a filter — it's a grocery list.",
+"ally pastes it into GHOST TEXT → the words come back out.",
+"add a password and the extract is noise to anyone who finds the trick but lacks the key."]) + how([
+"The secret becomes binary; every bit becomes one invisible character between two words of the cover.",
+"Zero-width characters take no space on screen — the cover text renders pixel-identical without them.",
+"A frame marker marks where the payload starts and ends, so decoy characters in normal text don't confuse extraction.",
+"A password XOR-scrambles the payload first — without it, extraction yields garbage bytes.",
+"The capacity meter counts your cover's word gaps: roughly one hidden character per 8 gaps.",
+"Python and JavaScript here implement the identical scheme — the API and the page agree byte for byte."])
+ body += gloss([("zero-width","invisible unicode characters — real, standard, render as nothing"),("cover","the innocent text that carries the payload"),("frame","marker characters delimiting the hidden bits"),("payload","your actual secret, bit by bit")])
+ body += agent_card('POST /api/ghost/encode cover=… secret=… [password=]', 'curl -X POST https://dark0rbits.thetempleofdoom.com/api/ghost/encode -d "cover=milk eggs bread" -d "secret=hello" --data-urlencode "password=pw"', 'POST /api/ghost/decode (text, password?) extracts. Python is the canonical implementation.')
+ body += f""""""
+ return page("ghost", body)
+
+@app.route("/api/ghost/encode", methods=["POST"])
+def api_ghost_encode():
+ r = rate_limit("ghost", 60, 60)
+ if r: return r
+ cover = param("cover") or ""
+ secret = param("secret") or ""
+ if not cover or not secret: return jsonify({"ok": False, "error": "cover + secret required"}), 400
+ try:
+ carrier = ghost_encode(cover, secret, param("password") or "")
+ except ValueError as e:
+ return jsonify({"ok": False, "error": str(e)}), 400
+ return jsonify({"ok": True, "carrier": carrier})
+
+@app.route("/api/ghost/decode", methods=["POST"])
+def api_ghost_decode():
+ r = rate_limit("ghost", 60, 60)
+ if r: return r
+ secret, err = ghost_decode(param("text") or "", param("password") or "")
+ if err: return jsonify({"ok": False, "error": err})
+ return jsonify({"ok": True, "secret": secret})
+# ---------- END TOOL: GHOST TEXT ----------
+
+
+# ---------- TOOL: CHAFF (deterministic persona generator) --
+import hashlib as _chash
+
+CHAFF_REGIONS = {
+ "US": {"first": ["James","Mary","Robert","Patricia","John","Jennifer","Michael","Linda","David","Sarah","Chris","Amanda"], "last": ["Smith","Johnson","Williams","Brown","Jones","Garcia","Miller","Davis","Wilson","Anderson","Taylor","Thomas"]},
+ "UK": {"first": ["Oliver","Amelia","Harry","Isla","George","Emily","Jack","Sophia","Charlie","Ava","Thomas","Lily"], "last": ["Wright","Turner","Walker","Harris","Clarke","Lewis","Young","Hall","Allen","King","Scott","Green"]},
+ "DE": {"first": ["Lukas","Anna","Felix","Marie","Paul","Laura","Jonas","Sophie","Leon","Lena","Max","Emma"], "last": ["Müller","Schmidt","Schneider","Fischer","Weber","Meyer","Wagner","Becker","Schulz","Hoffmann","Koch","Richter"]},
+ "NL": {"first": ["Daan","Emma","Sem","Julia","Lucas","Sofie","Levi","Anna","Bram","Lotte","Thijs","Sanne"], "last": ["de Jong","Jansen","de Vries","van den Berg","van Dijk","Bakker","Visser","Smit","Meijer","de Boer","Mulder","Bos"]},
+ "XX": {"first": ["Alex","Sam","Jordan","Riley","Casey","Morgan","Taylor","Jamie","Quinn","Avery","Robin","Drew"], "last": ["Reyes","Marsh","Kane","Voss","Holt","Cross","Bishop","Wolfe","Hart","Stone","Frost","Lang"]},
+}
+CHAFF_PETS = ["Biscuit","Shadow","Whiskers","Rex","Luna","Pepper","Gizmo","Mocha","Bandit","Clover","Nutmeg","Smokey"]
+CHAFF_CARS = ["a blue Corolla","an old F-150","a silver Civic","a VW Golf","a red Miata","a grey Passat","a green Jetta","a black CR-V"]
+CHAFF_CITIES = ["Riverton","Fairview","Millbrook","Oakdale","Lakewood","Brookside","Hillcrest","Ashford"]
+CHAFF_MAIIDS = ["Kelly","Morgan","Nelson","Hayes","Sullivan","Barrett","Whitmore","Calloway"]
+CHAFF_WORDS = ["falcon","cobalt","harbor","velvet","quartz","ember","willow","cobble","marble","saffron","indigo","basalt"]
+
+def chaff_persona(seed, region="US", email_domain="mailinator.com"):
+ """Deterministic persona from a seed. Same seed + region = same persona, forever, zero storage."""
+ pools = CHAFF_REGIONS.get(region, CHAFF_REGIONS["XX"])
+ material = b""
+ parts = []
+ def stream(n):
+ nonlocal material
+ seedb = (seed + "|" + region).encode()
+ out = b""
+ counter = 0
+ while len(out) < n:
+ out += _chash.sha256(seedb + counter.to_bytes(4, "big")).digest()
+ counter += 1
+ return out
+ raw = stream(64)
+ def pick(pool, r): return pool[r % len(pool)]
+ first = pick(pools["first"], raw[0] | (raw[1] << 8))
+ last = pick(pools["last"], raw[2] | (raw[3] << 8))
+ age = 21 + (raw[4] % 25)
+ byear = 2026 - age
+ bmonth = 1 + raw[5] % 12
+ bday = 1 + raw[6] % 28
+ w1 = pick(CHAFF_WORDS, raw[7]); w2 = pick(CHAFF_WORDS, raw[8])
+ num2 = raw[9] % 100
+ email_user = f"{w1}{w2}{num2:02d}"
+ # password FORMAT template (words filled, digits, symbol) — not a real used password
+ pword = pick(CHAFF_WORDS, raw[10]).capitalize()
+ pword2 = pick(CHAFF_WORDS, raw[11]).capitalize()
+ pdigits = 1000 + (raw[12] % 9000)
+ psym = pick(["!", "#", "?", "%"], raw[13])
+ pet = pick(CHAFF_PETS, raw[14])
+ car = pick(CHAFF_CARS, raw[15])
+ city = pick(CHAFF_CITIES, raw[16])
+ maid = pick(CHAFF_MAIIDS, raw[17])
+ initials = (first[0] + last[0]).upper()
+ hue = raw[18] % 360
+ return {
+ "region": region,
+ "name": f"{first} {last}",
+ "first": first, "last": last,
+ "username_variants": [
+ f"{first.lower()}.{last.lower()}{num2:02d}",
+ f"{first.lower()}_{last.lower()}",
+ f"{first[0].lower()}{last.lower()}{age}",
+ f"{w1}{w2}".capitalize() + str(num2),
+ ],
+ "birthdate": f"{byear}-{bmonth:02d}-{bday:02d}",
+ "age": age,
+ "email_suggestion": f"{email_user}@{email_domain}",
+ "password_pattern": f"{pword}-{pword2}-{pdigits}{psym}",
+ "security_answers": {
+ "mother's maiden name": maid,
+ "first pet": pet,
+ "first car": car,
+ "city born in": city,
+ },
+ "avatar": {"initials": initials, "hue": hue},
+ }
+
+@app.route("/chaff")
+def chaff():
+ body = f"""
+
CHAFF PERSONA
Deterministic throwaway identities: one passphrase in, a complete consistent persona out. The same passphrase always regenerates the same person — but nothing is ever stored, here or anywhere.
+
+
+
+
+
+
+
+
+""" + flow("a whole new you in one click", [
+"you pick a seed phrase you'll remember: tin frog distant harbor.",
+"CHAFF derives a full identity: name, usernames, birthday, email pattern, password format, security answers.",
+"sign up somewhere using the generated details. Then lose the details.",
+"months later you need the same identity again: type the same seed — the exact same persona comes back.",
+"nothing was ever stored anywhere. The seed IS the identity; forget the seed, the persona is gone forever."]) + how([
+"Everything derives from sha256 of your seed — deterministic, offline-verifiable, storage-free.",
+"Security answers are CONSISTENT with the persona (the pet, the car, the city never contradict each other).",
+"The password field is a FORMAT filled with persona words — treat it as a pattern, not a real password; reuse of an actual password across sites is how identities burn.",
+"Reroll appends a random nonce: a new persona, still fully recoverable if you saved its JSON.",
+"Agents: GET /api/chaff?seed=…®ion=US returns the whole persona as JSON."])
+ body += gloss([("seed","the passphrase that deterministically generates the persona"),("chaff","radar-confetti — many fake targets so the real one is lost among them"),("deterministic","same input, same output, every time, forever"),("burn","deliberately abandon an identity after use")])
+ body += agent_card('GET /api/chaff?seed=tin+frog®ion=US', 'curl "https://dark0rbits.thetempleofdoom.com/api/chaff?seed=tin%20frog%20distant%20harbor®ion=NL"', 'Same seed = same persona. Stateless — no database table, nothing logged.')
+ body += """"""
+ return page("chaff", body)
+
+@app.route("/api/chaff")
+def api_chaff():
+ r = rate_limit("chaff", 60, 60)
+ if r: return r
+ seed = (param("seed") or "").strip()
+ if not seed: return jsonify({"ok": False, "error": "seed required"}), 400
+ region = param("region") or "US"
+ if region not in CHAFF_REGIONS: region = "XX"
+ domain = (param("domain") or "mailinator.com").strip()[:60]
+ return jsonify({"ok": True, "persona": chaff_persona(seed, region, domain)})
+# ---------- END TOOL: CHAFF ----------
+
+
+# ---------- TOOL: LEAK TRACER (document watermarker) ----------
+import hashlib as _lhash
+
+_LZ0, _LZ1 = "\u200b", "\u200c"
+
+def _lhash_stream(password, n):
+ ks = b""; seed = password.encode()
+ while len(ks) < n:
+ seed = _lhash.sha256(seed).digest(); ks += seed
+ return ks[:n]
+
+def _leak_mark(text, case_id, recipient_idx, nbits=None):
+ """Embed recipient-specific bit pattern in word gaps. Pattern = hash(case_id + idx),
+ so recipients' patterns differ everywhere (not just in low bits)."""
+ salted = _lhash.sha256((case_id + ":" + str(recipient_idx)).encode()).digest()
+ bits = "".join(format(b, "08b") for b in salted[:4]) # 32 bits, pseudorandom per recipient
+ words = text.split(" ")
+ slots = len(words) - 1
+ if slots < len(bits):
+ raise ValueError(f"document too short: needs {len(bits)}+ word gaps, has {slots}")
+ step = slots / len(bits)
+ out = []
+ bit_i = 0
+ for i, w in enumerate(words):
+ out.append(w)
+ if i < slots:
+ out.append(" ") # normal gap
+ if bit_i < len(bits) and i == int(bit_i * step):
+ out.append(_LZ0 if bits[bit_i] == "0" else _LZ1)
+ bit_i += 1
+ return "".join(out) # 32 marks woven in, positions deterministic from bit_i*step
+
+def _leak_read_bits(text):
+ """Extract zero-width bits sequence from any text."""
+ return [c for c in text if c in (_LZ0, _LZ1)]
+
+def _leak_identify(case, fragment):
+ """Match a fragment's bit pattern against all variants of a case.
+ Returns (recipient_label, confidence, matched_bits) or (None, 0, 0)."""
+ frag_bits = _leak_read_bits(fragment)
+ # the fragment must carry at least 8 marks
+ if len(frag_bits) < 8:
+ return None, 0, len(frag_bits)
+ best_label, best_score, best_total = None, 0.0, 0
+ for label, variant_bits in case["variants"]:
+ vb = _leak_read_bits(variant_bits)
+ if len(vb) < 32: continue
+ # sliding window: try to align fragment bits inside variant bits
+ n = len(frag_bits)
+ best_local = 0
+ for off in range(0, min(len(vb) - n + 1, 64)):
+ m = sum(1 for i in range(n) if vb[off + i] == frag_bits[i])
+ if m > best_local: best_local = m
+ # also try start-aligned (common case: leak is the doc head)
+ m0 = sum(1 for i in range(min(n, len(vb))) if vb[i] == frag_bits[i])
+ score = max(best_local, m0) / n
+ if score > best_score:
+ best_label, best_score, best_total = label, score, n
+ return best_label, best_score, len(frag_bits)
+
+def leak_case_create(uid, name, text, recipients):
+ con = db()
+ con.execute("""CREATE TABLE IF NOT EXISTS leakcases(
+ id INTEGER PRIMARY KEY, case_id TEXT UNIQUE, user_id INTEGER, name TEXT,
+ created INTEGER, variants_enc TEXT)""")
+ case_id = secrets.token_urlsafe(8)
+ variants = []
+ for idx, label in enumerate(recipients):
+ variants.append((label, _leak_mark(text, case_id, idx)))
+ con.execute("INSERT INTO leakcases(case_id,user_id,name,created,variants_enc) VALUES(?,?,?,?,?)",
+ (case_id, uid, esc(name[:100]), int(time.time()), dd_encrypt(json.dumps(variants))))
+ con.commit()
+ return case_id
+
+def leak_case_get(uid, case_id):
+ con = db()
+ r = con.execute("SELECT * FROM leakcases WHERE case_id=? AND user_id=?", (case_id, uid)).fetchone()
+ if not r: return None
+ raw = dd_decrypt(r["variants_enc"])
+ if not raw: return None
+ return {"name": r["name"], "variants": json.loads(raw)}
+
+@app.route("/tracer")
+def tracer():
+ uid = current_user_id()
+ if not uid:
+ return page("tracer", """
LEAK TRACER
Hand every recipient their own invisible-marked copy of a document. When it leaks, the marks name the leaker — even from a pasted fragment.
""" + how([
+"Each copy of your document carries the recipient's identity in invisible zero-width characters.",
+"The marks survive copy-paste, screenshots-to-text, and edits — they're part of the text itself.",
+"When a copy surfaces, paste the leaked text and LEAK TRACER names who leaked it.",
+"Fragments work: even a few sentences from the middle carry enough marks to identify.",
+"Cases stay private to your account; variants are encrypted at rest."]))
+ con = db()
+ cases = con.execute("SELECT case_id, name, created FROM leakcases WHERE user_id=? ORDER BY id DESC LIMIT 30", (uid,)).fetchall()
+ caselist = "".join(f'' for c in cases)
+ body = f"""
+
LEAK TRACER
Hand every recipient their own invisible-marked copy of a document. When it leaks, the marks name the leaker — even from a pasted fragment.
+
Open a case
+
+
+
+{('
Your cases
open a case from the list on its own page — click Identify below
') if cases else ''}
+
Identify a leak
+
""" + flow("catch the leaker from a pasted paragraph", [
+"you open a case: the merger memo, 4 recipients — alice, bob, carol, dave.",
+"LEAK TRACER makes 4 copies. Each looks identical but carries 32 invisible bits naming its owner.",
+"you send alice her copy, bob his, carol hers, dave his.",
+"three days later the memo is on a forum. You copy two paragraphs from the leak.",
+"you paste them into Identify with the case id → carol — 97% confidence.",
+"now you know. And the other three know you can know — that changes how the next doc gets treated."]) + how([
+"Every recipient index becomes 32 invisible bits woven into the word gaps of their copy.",
+"The bits are deterministic per position — a fragment from anywhere still aligns to its owner's pattern.",
+"Identify needs at least 8 surviving marks in the fragment; pasted text usually keeps them all.",
+"Stripping the marks (normalizing whitespace) also works as a detection: 'this text was a traced copy'.",
+"Variants are AES-encrypted in the database under your account — nobody reads your docs but you."])
+ body += gloss([("watermark","invisible per-copy marks identifying the recipient"),("zero-width","invisible unicode characters riding inside normal text"),("fragment","a partial leak — some sentences, not the whole doc"),("case","one document + its recipient list + all variants")])
+ body += agent_card('POST /api/tracer/case name= text= recipients=a,b,c', 'curl -X POST https://dark0rbits.thetempleofdoom.com/api/tracer/case -d "name=deck" -d "text=…doc…" -d "recipients=alice,bob" -H "Cookie: dark0rbits_tok=…"', 'GET /api/tracer/identify?case_id=&text= → leaker + confidence.')
+ return page("tracer", body)
+
+@app.route("/tracer", methods=["POST"])
+def tracer_create():
+ uid = current_user_id()
+ if not uid: return page("tracer", "
login required
")
+ name = param("name") or "case"
+ text = (param("doctext") or "").strip()
+ recipients = [r.strip()[:40] for r in (param("recipients") or "").split(",") if r.strip()]
+ if not text or not recipients:
+ return page("tracer", "
document text + recipients required
")
+ if len(recipients) > 20:
+ return page("tracer", "
max 20 recipients per case
")
+ try:
+ case_id = leak_case_create(uid, name, text, recipients)
+ except ValueError as e:
+ return page("tracer", f'
")
+ case = leak_case_get(uid, case_id)
+ if not case: return page("tracer", "
unknown case
")
+ rows = ""
+ for label, variant in case["variants"]:
+ rows += (f'
{esc(label)}
'
+ f''
+ f'
')
+ body = f"""
+
CASE {esc(case['name'])}
{len(case['variants'])} watermarked copies. Send each person THEIR copy — any leak names its owner.
+
{rows}
+
Identify a leak from this case
+
+"""
+ return page("tracer", body)
+
+@app.route("/tracer/identify")
+def tracer_identify():
+ uid = current_user_id()
+ if not uid: return page("tracer", "
login required
")
+ case_id = param("case") or ""
+ fragment = param("text") or ""
+ case = leak_case_get(uid, case_id)
+ if not case: return page("tracer", "
unknown case
")
+ label, score, nbits = _leak_identify(case, fragment)
+ if label is None or score < 0.5:
+ verdict = f'no confident match ({nbits} marks found, best {int(score*100)}%)'
+ else:
+ conf = int(score * 100)
+ verdict = f'LEAKER: {esc(label)} — matched {int(score*nbits)}/{nbits} marks · {conf}% confidence'
+ body = f"""
+
VERDICT {esc(case['name'])}
+
Result
{verdict}
+
{nbits} invisible marks detected in the fragment. Above 50% alignment on a 32-bit pattern is a positive ID; below that the fragment may be too short or the marks got stripped in transit.
"""
+ return page("tracer", body)
+
+@app.route("/api/tracer/case", methods=["POST"])
+def api_tracer_case():
+ uid = key_user() or current_user_id()
+ if not uid: return jsonify({"ok": False, "error": "auth required"}), 401
+ r = rate_limit("tracer", 20, 60)
+ if r: return r
+ name = param("name") or "case"
+ text = (param("text") or param("doctext") or "").strip()
+ recipients = [x.strip()[:40] for x in (param("recipients") or "").split(",") if x.strip()]
+ if not text or not recipients: return jsonify({"ok": False, "error": "text + recipients required"}), 400
+ if len(recipients) > 20: return jsonify({"ok": False, "error": "max 20 recipients"}), 400
+ try:
+ case_id = leak_case_create(uid, name, text, recipients)
+ except ValueError as e:
+ return jsonify({"ok": False, "error": str(e)}), 400
+ case = leak_case_get(uid, case_id)
+ return jsonify({"ok": True, "case_id": case_id,
+ "variants": {label: variant for label, variant in case["variants"]}})
+
+@app.route("/api/tracer/identify")
+def api_tracer_identify():
+ uid = key_user() or current_user_id()
+ if not uid: return jsonify({"ok": False, "error": "auth required"}), 401
+ case = leak_case_get(uid, param("case") or "")
+ if not case: return jsonify({"ok": False, "error": "unknown case"}), 404
+ label, score, nbits = _leak_identify(case, param("text") or "")
+ if label is None or score < 0.5:
+ return jsonify({"ok": True, "match": None, "marks_found": nbits, "best_score": round(score, 3)})
+ return jsonify({"ok": True, "match": label, "confidence": round(score, 3),
+ "marks_matched": int(score * nbits), "marks_found": nbits})
+# ---------- END TOOL: LEAK TRACER ----------
+
+
+# ---------- TOOL: TRACEOUT (network traceroute) --- TRACEROUTE ----------
+# Path tracing from this host to any target, hop by hop, with per-hop geo.
+# Pure-Python UDP traceroute (IP_TTL) with TCP-connect fallback. No root needed.
+
+def _trace_table(uid):
+ con = db()
+ con.executescript("""
+ CREATE TABLE IF NOT EXISTS trace_runs(id INTEGER PRIMARY KEY, user_id INTEGER, target TEXT, resolved TEXT, hops INTEGER, done INTEGER, created INTEGER);
+ CREATE TABLE IF NOT EXISTS trace_hops(run_id INTEGER, ttl INTEGER, ip TEXT, host TEXT, ms REAL, kind TEXT);
+ """)
+ return con
+
+def _trace_resolve(host):
+ """Return (ip, err). Accepts hostnames and IPs."""
+ host = (host or "").strip().rstrip(".")
+ if not host:
+ return None, "target required"
+ if ":" in host:
+ return None, "IPv6 not supported by this tracer — give an IPv4 address or hostname"
+ try:
+ ipaddress.ip_address(host)
+ return host, None
+ except ValueError:
+ pass
+ if not re.match(r"^[a-zA-Z0-9._-]{1,253}$", host):
+ return None, "invalid hostname"
+ try:
+ return socket.gethostbyname(host), None
+ except Exception:
+ return None, f"cannot resolve {host}"
+
+def _trace_probe_udp(ip, ttl, port, timeout):
+ """One UDP probe. Returns (kind, ms, reply_ip): kind in hop|done|timeout."""
+ tx = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+ rx = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+ try:
+ tx.setsockopt(socket.SOL_IP, socket.IP_TTL, ttl)
+ rx.setsockopt(socket.SOL_SOCKET, socket.SO_RCVTIMEO, struct.pack("ll", int(timeout), 0))
+ rx.bind(("0.0.0.0", 0))
+ rx_port = rx.getsockname()[1]
+ t0 = time.time()
+ tx.sendto(b"dark0rbits-tracer", (ip, port))
+ try:
+ data, addr = rx.recvfrom(512)
+ ms = round((time.time() - t0) * 1000, 1)
+ return "hop", ms, addr[0]
+ except ConnectionRefusedError:
+ return "done", round((time.time() - t0) * 1000, 1), ip
+ except socket.timeout:
+ return "timeout", None, None
+ except OSError:
+ try:
+ data, addr = rx.recvfrom(512)
+ ms = round((time.time() - t0) * 1000, 1)
+ return "hop", ms, addr[0]
+ except Exception:
+ return "timeout", None, None
+ finally:
+ try: tx.close()
+ except Exception: pass
+ try: rx.close()
+ except Exception: pass
+
+def _trace_probe_tcp(ip, ttl, port, timeout):
+ """TCP-connect probe fallback (SYN dies at the hop when TTL expires)."""
+ s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+ try:
+ s.setsockopt(socket.SOL_IP, socket.IP_TTL, ttl)
+ s.settimeout(timeout)
+ t0 = time.time()
+ rc = s.connect_ex((ip, port))
+ ms = round((time.time() - t0) * 1000, 1)
+ if rc == 0:
+ return "done", ms, ip
+ if rc in (61, 111, 113): # ECONNREFUSED-ish: reached the destination
+ return "done", ms, ip
+ return "timeout", None, None
+ except socket.timeout:
+ return "timeout", None, None
+ except OSError:
+ return "timeout", None, None
+ finally:
+ try: s.close()
+ except Exception: pass
+
+def _trace_reverse(ip):
+ try:
+ host, _, _ = socket.gethostbyaddr(ip)
+ return host
+ except Exception:
+ return ""
+
+def tout_run(target, max_hops=15, timeout=1.5):
+ """Run a traceroute. Returns dict: ok, target, resolved, hops[], method."""
+ max_hops = max(1, min(int(max_hops or 15), 30))
+ timeout = max(0.3, min(float(timeout or 1.5), 5.0))
+ ip, err = _trace_resolve(target)
+ if err:
+ return {"ok": False, "error": err, "target": target}
+ hops = []
+ method = "udp"
+ port = 33434
+ for ttl in range(1, max_hops + 1):
+ if ip in ("127.0.0.1",) or ip.startswith("127."):
+ best = _trace_probe_udp(ip, ttl, port, timeout)
+ else:
+ best = _trace_probe_udp(ip, ttl, port, timeout)
+ if best[0] == "timeout" and ttl == 1:
+ alt = _trace_probe_tcp(ip, ttl, 80, timeout)
+ if alt[0] != "timeout":
+ method = "tcp"
+ best = alt
+ elif _trace_probe_tcp(ip, ttl, 443, timeout)[0] != "timeout":
+ method = "tcp"
+ best = _trace_probe_tcp(ip, ttl, 443, timeout)
+ kind, ms, rip = best
+ host = _trace_reverse(rip) if rip else ""
+ hops.append({"ttl": ttl, "ip": rip or "", "host": host, "ms": ms, "kind": kind})
+ if kind == "done":
+ break
+ return {"ok": True, "target": target, "resolved": ip, "method": method,
+ "hops": hops, "hop_count": len(hops),
+ "complete": bool(hops and hops[-1]["kind"] == "done")}
+
+def tout_save(uid, target, res):
+ """Persist a run + hops. Returns run id or None."""
+ try:
+ con = _trace_table(uid)
+ cur = con.execute(
+ "INSERT INTO trace_runs(user_id,target,resolved,hops,done,created) VALUES(?,?,?,?,?,?)",
+ (uid, target[:200], res.get("resolved") or "", res.get("hop_count", 0),
+ 1 if res.get("complete") else 0, int(time.time())))
+ rid = cur.lastrowid
+ for h in res.get("hops", []):
+ con.execute(
+ "INSERT INTO trace_hops(run_id,ttl,ip,host,ms,kind) VALUES(?,?,?,?,?,?)",
+ (rid, h["ttl"], h["ip"] or "", h["host"][:200] if h["host"] else "",
+ h["ms"] if h["ms"] is not None else -1, h["kind"]))
+ con.commit()
+ return rid
+ except Exception:
+ return None
+
+def tout_history(uid, limit=10):
+ try:
+ con = _trace_table(uid)
+ return con.execute(
+ "SELECT * FROM trace_runs WHERE user_id=? ORDER BY id DESC LIMIT ?",
+ (uid, max(1, min(limit, 50)))).fetchall()
+ except Exception:
+ return []
+
+_TRACE_PORTS = {"http": 80, "https": 443, "dns": 53, "ssh": 22, "smtp": 25}
+
+def tout_render_hops(res):
+ rows = ""
+ for h in res.get("hops", []):
+ if h["kind"] == "timeout":
+ rows += f"
{h['ttl']}
* * * no answer
"
+ continue
+ lbl = "DEST" if h["kind"] == "done" else "hop"
+ geo = enrich_ip(h["ip"]) or {}
+ where = esc(" ".join(x for x in (geo.get("city"), geo.get("country")) if x))
+ ms = f"{h['ms']} ms" if h["ms"] is not None else "-"
+ rows += (f"
RESULT — {esc(result['target'])} → {esc(result['resolved'])} "
+ f"{result['method']}"
+ + ("COMPLETE" if result["complete"] else "PARTIAL")
+ + f"
#
hop
geo
rtt
role
"
+ + tout_render_hops(result) + "
")
+ body = f"""
+
TRACE ROUTE
Hop-by-hop path from this host to any target. Every router between you and the destination, with geo per hop. Works without JS.
+
+
+{f"
{esc(err)}
" if err else ""}
+
+{res_html}
+
RECENT RUNS
Target
Resolved
Hops
Complete
When
{hist_html}
+"""
+ body += how([
+ "Type a domain or IPv4 — no port guessing needed unless you want a specific probe port.",
+ "The tool sends UDP probes with TTL 1, 2, 3… — each router that kills a packet reveals itself.",
+ "The final hop answers with a port-unreachable: that is the destination, marked DEST.",
+ "Each answered hop gets reverse-DNS and IP geo enrichment so you see where the path bends.",
+ "Firewalls that drop UDP make the path look dead — switch the probe port to 443 and retry.",
+ "Every run is saved under RECENT RUNS so you can diff paths across time.",
+ ])
+ body += flow("Trace a target and read the path", [
+ "you enter example.com, probe port dns, hit TRACE",
+ "tracer resolves the name, walks TTL 1→15, logs every router IP + rtt",
+ "tracer enriches each hop with geo and flags the DEST row",
+ "you read the path: where it leaves the country, where latency jumps, where it dies",
+ ])
+ body += gloss([
+ ("TTL", "time-to-live: a hop counter in every packet; each router decrements it and discards at zero, announcing itself"),
+ ("rtt", "round-trip time for the probe packet — the latency budget of that hop"),
+ ("DEST", "destination row: the host answered with port-unreachable, so the path is complete"),
+ ])
+ body += agent_card(
+ "POST /api/traceout/run {target, port, maxhops, timeout} · GET /api/traceout/history",
+ "curl -X POST " + SITE + "/api/traceout/run -H 'Authorization: Bearer ***' -H 'Content-Type: application/json' -d '{\"target\":\"example.com\",\"port\":443}'",
+ "Hop-by-hop path trace with per-hop geo. FREE. Session cookie or API key. Returns hops[] with ttl/ip/host/ms/kind.")
+ body += kv([
+ ("probe method", "UDP with IP_TTL, TCP fallback"),
+ ("max hops", "1–30"),
+ ("rate limit", "8 traces / minute"),
+ ("price", "FREE"),
+ ])
+ return page("tracer", body)
+
+@app.route("/api/traceout/run", methods=["POST"])
+def api_tracer_run():
+ r = rate_limit("tracer_api", 8, 60)
+ if r: return r
+ uid, e = require_paid_key(0, "traceroute run")
+ if e: return e
+ target = str(jp("target") or "").strip()
+ if not target:
+ return jsonify({"ok": False, "error": "target required (domain or IPv4)"}), 400
+ port_name = str(jp("port") or "dns").lower()
+ res = tout_run(target, max_hops=jp("maxhops") or 15, timeout=jp("timeout") or 1.5)
if not res.get("ok"):
return jsonify(res), 400
- return jsonify(res)
-# ---------- END TOOL: HOOK RELAY ----------
+ res["port"] = _TRACE_PORTS.get(port_name, 33434)
+ rid = tout_save(uid, target, res)
+ res["run_id"] = rid
+ return jsonify(res), 200, {"Cache-Control": "no-store"}
-
-# ---------- TOOL: FACE TRACE ----------
-import io as _io
-from PIL import Image as _PILImage, ImageOps as _PILImageOps
-from concurrent.futures import ThreadPoolExecutor as _TPE
-
-def _face_db():
- con = db()
- con.execute("CREATE TABLE IF NOT EXISTS face_cache(key TEXT PRIMARY KEY, img BLOB, sha256 TEXT, ts INTEGER)")
- con.commit()
- return con
-
-def _hamming(h1, h2):
- n = max(len(h1), len(h2)) * 4
- try:
- return bin(int(h1, 16) ^ int(h2, 16))[2:].zfill(n).count("1")
- except Exception:
- return 999
-
-def dhash(img):
- """dHash: grayscale 9x8 (w=9,h=8), compare horizontally adjacent pixels → 64-bit."""
- g = _PILImageOps.grayscale(img).resize((9, 8))
- px = list(g.getdata())
- bits = 0
- for r in range(8):
- row = px[r * 9:(r + 1) * 9]
- for c in range(8):
- bits = (bits << 1) | (1 if row[c] > row[c + 1] else 0)
- return f"{bits:016x}"
-
-def ahash(img):
- """Average hash: 8x8 grayscale, bit = pixel > mean."""
- g = _PILImageOps.grayscale(img).resize((8, 8))
- px = list(g.getdata())
- m = sum(px) / 64.0
- bits = 0
- for p in px:
- bits = (bits << 1) | (1 if p > m else 0)
- return f"{bits:016x}"
-
-def _pfp_hashes(blob):
- """Hashes for a raw image blob: {'sha256','dhash','ahash'} or {'error':...}"""
- out = {"sha256": hashlib.sha256(blob).hexdigest(), "bytes": len(blob)}
- try:
- img = _PILImage.open(_io.BytesIO(blob))
- img.load()
- out["dhash"] = dhash(img)
- out["ahash"] = ahash(img)
- out["format"] = (img.format or "?").lower()
- out["size"] = list(img.size)
- except Exception as e:
- out["error"] = f"not an image: {e}"[:160]
- return out
-
-def _fcache_get(key):
- con = _face_db()
- r = con.execute("SELECT img, sha256, ts FROM face_cache WHERE key=?", (key,)).fetchone()
- if r and (int(time.time()) - r["ts"]) < 3600:
- return bytes(r["img"])
- return None
-
-def _fcache_put(key, blob):
- con = _face_db()
- con.execute("INSERT OR REPLACE INTO face_cache(key,img,sha256,ts) VALUES(?,?,?,?)",
- (key, sqlite3.Binary(blob), hashlib.sha256(blob).hexdigest(), int(time.time())))
- con.commit()
-
-def _fetch_img(url):
- """Fetch an image URL → (bytes|None, note). 1h sqlite cache."""
- ck = "url:" + hashlib.sha256(url.encode()).hexdigest()[:32]
- c = _fcache_get(ck)
- if c is not None:
- return c, "cache"
- req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0 (Dark0rbits toolbox)"})
- try:
- with urllib.request.urlopen(req, timeout=10, context=_CTX) as r:
- blob = r.read(6 * 1024 * 1024)
- if len(blob) < 64:
- return None, f"too small ({len(blob)}b)"
- _fcache_put(ck, blob)
- return blob, "fetched"
- except Exception as e:
- return None, str(e)[:120]
-
-def pfp_targets(u):
- """Avatar source plan for a username: resolvers and constructed URLs."""
- u = u.strip().lstrip("@")
- q = urllib.parse.quote(u)
- return [
- {"platform": "GitHub", "kind": "resolve",
- "api": f"https://api.github.com/users/{q}",
- "extract": lambda j: (j.get("avatar_url") or "") if isinstance(j, dict) else "",
- "profile": f"https://github.com/{q}"},
- {"platform": "Reddit", "kind": "resolve",
- "api": f"https://www.reddit.com/{q}/about.json",
- "extract": lambda j: (((j.get("data") or {}).get("icon_img") or "")) if isinstance(j, dict) and j.get("data") else "",
- "profile": f"https://www.reddit.com/user/{q}"},
- {"platform": "Telegram", "kind": "construct",
- "img": "https://t.me/" + q,
- "note": "manual / constructed: t.me profile — grab photo from the page",
- "profile": f"https://t.me/{q}"},
- {"platform": "Steam", "kind": "construct",
- "img": "https://steamcommunity.com/id/{u}/".replace("{u}", q),
- "note": "manual / constructed: Steam profile — XML API has ",
- "profile": "https://steamcommunity.com/id/" + q},
- {"platform": "Twitch", "kind": "construct",
- "img": "https://www.twitch.tv/" + q,
- "note": "manual / constructed: profile page (avatars need a client-id)",
- "profile": f"https://www.twitch.tv/{q}"},
- ]
-
-def pfp_harvest(u, target_hashes=None):
- """Resolve + download avatars for username u; compare with target hashes."""
- results = []
- for t in pfp_targets(u):
- entry = {"platform": t["platform"], "profile": t["profile"], "status": "no avatar",
- "hashes": None, "match": None, "distances": {}}
- img_url = ""
- if t["kind"] == "resolve":
- st, body = http(t["api"], timeout=10)
- j = jf(body)
- img_url = t["extract"](j) if j else ""
- entry["http"] = st
- if not img_url:
- entry["status"] = "not found"
- else:
- entry["status"] = "constructed link"
- entry["note"] = t["note"]
- if img_url:
- blob, note = _fetch_img(img_url)
- if blob:
- h = _pfp_hashes(blob)
- entry.update({"status": "ok", "img_url": img_url, "source": note, "hashes": h})
- if target_hashes and not h.get("error"):
- dd = min(_hamming(h["dhash"], target_hashes["dhash"]),
- _hamming(h["ahash"], target_hashes["ahash"]))
- entry["distances"] = {"dhash": _hamming(h["dhash"], target_hashes["dhash"]),
- "ahash": _hamming(h["ahash"], target_hashes["ahash"])}
- entry["match"] = "likely same image" if dd <= 10 else "different image"
- else:
- entry["status"] = f"download failed ({note})"
- results.append(entry)
- return {"ok": True, "username": u.strip().lstrip("@"), "targets": results}
-
-def face_search_leads():
- return [
- ("Google Lens", "https://lens.google.com/uploadbyurl?url=", "Google's reverse-image search — strongest for lookalikes and crops"),
- ("Yandex Images", "https://yandex.com/images/search?rpt=imageview&url=", "best face recall of the public engines, especially EU/RU web"),
- ("Bing Visual Search", "https://www.bing.com/images/search?view=detailv2&iss=sbi&q=imgurl:", "Microsoft visual search — good LinkedIn / social recall"),
- ("TinEye", "https://tineye.com/search?url=", "exact-copy finder — best for 'where did this exact file appear first'"),
- ]
-
-@app.route("/face", methods=["GET", "POST"])
-def face_tool():
- uid = current_user_id()
- if not uid:
- return page("face", '
FACE TRACE
Login first — this tool is for accounts. Free, no KYC: log in / sign up.
')
- r = rate_limit("facepage", 20, 60)
- if r: return r
- res = ""
- up_hashes = None
- err = ""
- f = request.files.get("img")
- username = (param("u") or "").strip().lstrip("@")
- if request.method == "POST":
- if not f and not username:
- err = "give me an image and/or a username"
- if f:
- blob = f.read(6 * 1024 * 1024)
- if len(blob) < 64:
- err = "file too small / empty"
- else:
- h = _pfp_hashes(blob)
- if h.get("error"):
- err = h["error"]
- else:
- up_hashes = h
- up_rows = [("sha256", f"{h['sha256']}"), ("dHash (8x8)", f"{h['dhash']}"),
- ("aHash (8x8)", f"{h['ahash']}"), ("format / size", f"{h.get('format')} {h.get('size')} · {h['bytes']} bytes")]
- if username:
- up_rows.append(("comparing against", f"avatars of {esc(username)}"))
- res += kv(up_rows)
- if username and not (len(username) >= 2 and len(username) <= 60 and not any(c in "<>\"'/" for c in username)):
- if not err: err = "bad username (2-60 chars, no slashes/html)"
- username = ""
- if username:
- hv = pfp_harvest(username, up_hashes)
- rows = ""
- for t in hv["targets"]:
- if t["status"] == "ok":
- dd = t["distances"]
- if t["match"]:
- m = ('MATCH ' + esc(t["match"]) +
- f" · d {dd['dhash']} / a {dd['ahash']}")
- else:
- m = "—"
- rows += (f"
Verdict rule: best Hamming distance (dHash or aHash) ≤ 10 of 64 bits = likely same image. Cache: 1h sqlite.
')
- if up_hashes and not username:
- res += '
Reverse-image leads dark0rbits never calls a reverse-image API for you — open these yourself and upload the file:
' + " ".join(f'▸ {esc(nm)} →{esc(d)}' for nm, base, d in face_search_leads()) + "
"
- if err:
- res = f'
{esc(err)}
' + res
- body = f"""
-
FACE TRACE
-
Profile-picture triangulation without reverse-image APIs. Hash an avatar (dHash + aHash + sha256), harvest avatars a username uses across platforms, and let Hamming distance tell you whether it's the same picture — same person behind 4 different usernames?
-
-
-
Image alone = hashes + manual search leads. Username alone = avatar harvest + hashes. Both = harvest AND compare against your upload (Hamming ≤ 10 = likely match).
-
-{res}
-""" + how([
- "Upload the avatar you already have — a forum pic, a Telegram photo, anything.",
- "The image is fingerprinted three ways: dHash (9x8 grayscale, adjacent-pixel compares), aHash (8x8 vs mean) and plain sha256 — all computed locally, nothing uploaded anywhere.",
- "Give a username too, and the tool fetches that handle's real avatars: GitHub and Reddit via their public JSON APIs, plus constructed profile links for Telegram, Steam and Twitch.",
- "Every fetched avatar gets the same fingerprints, and Hamming distance (bits differing out of 64) is computed against your upload: ≤ 10 = likely the same image.",
- "No image ever goes to Google, Yandex, Bing or TinEye. Instead you get direct upload links to all four — you decide when to escalate.",
- "Downloads are cached in sqlite for an hour, so re-running a trace is fast and doesn't hammer anyone's API.",
- "Judgment stays yours: same picture is strong evidence, but people reuse stock photos. dHash says 'same image', not 'same human'.",
-])
- body += flow("same person behind 4 different usernames?", [
- "youA scammer contacts you from ghostuser42 with a friendly avatar. Screenshot it.",
- "youUpload the avatar here, type ghostuser42, hit TRACE. GitHub and Reddit avatars get fetched and hashed.",
- "toolVerdict table: Reddit avatar MATCH — d 4/64. GitHub avatar: different image.",
- "youRun TRACE on two other handles the same person used. Reddit matches again — same source photo, different display names.",
- "youEscalate: open the Google Lens / Yandex leads with the original file to find where the photo first appeared.",
- "youConclusion: four usernames, one face. That's your triangulation — no reverse-image API ever saw the picture.",
- ])
- body += gloss([
- ("dHash", "difference hash: resize to 9x8 grayscale, compare each pixel with its right neighbor → 64 bits that survive resizing and compression"),
- ("aHash", "average hash: 8x8 grayscale, each bit = pixel brighter than the mean"),
- ("Hamming distance", "how many of the 64 bits differ between two hashes — 0 = identical image, ≤ 10 = likely same picture, 32 = unrelated"),
- ("perceptual hash", "a fingerprint of what an image LOOKS like, not its bytes — crops and re-encodes still match; sha256 only matches exact copies"),
- ("avatar harvest", "collecting the profile pictures a username currently uses, from public profile APIs"),
- ])
- body += agent_card('POST /api/face (multipart image and/or u=username)',
- 'curl -F "img=@avatar.png" -F "u=ghostuser42" https://dark0rbits.thetempleofdoom.com/api/face',
- 'Returns sha256/dHash/aHash of your upload + per-platform harvest with distances and verdict. Auth: session or Bearer key.')
- return page("face", body)
-
-@app.route("/api/face", methods=["GET", "POST"])
-def api_face():
- r = rate_limit("face", 20, 60)
- if r: return r
+@app.route("/api/traceout/history")
+def api_tracer_history():
uid = key_user() or current_user_id()
if not uid:
- return jsonify({"ok": False, "error": "auth required: account session or API key"}), 401
- f = request.files.get("img")
- username = ((jp("u") if request.is_json else None) or param("u") or "").strip().lstrip("@")
- if not f and not username:
- return jsonify({"ok": False, "error": "provide multipart image and/or u=username"}), 400
- out = {"ok": True}
- tgt = None
- if f:
- blob = f.read(6 * 1024 * 1024)
- if len(blob) < 64:
- return jsonify({"ok": False, "error": "image too small/empty"}), 400
- h = _pfp_hashes(blob)
- if h.get("error"):
- return jsonify({"ok": False, "error": h["error"]}), 400
- out["uploaded"] = h
- tgt = h
- if username:
- if not (2 <= len(username) <= 60) or any(c in "<>\"'/" for c in username):
- return jsonify({"ok": False, "error": "bad username"}), 400
- out["harvest"] = pfp_harvest(username, tgt)
- if f or not username:
- out["leads"] = [{"name": n, "url": b, "note": d} for n, b, d in face_search_leads()]
- return jsonify(out)
-# ---------- END TOOL: FACE TRACE ----------
+ return jsonify({"ok": False, "error": "auth required: account session or Authorization: Bearer *** key"}), 401
+ rows = tout_history(uid, limit=int(param("limit") or 10))
+ return jsonify({"ok": True, "runs": [
+ {"id": h["id"], "target": h["target"], "resolved": h["resolved"],
+ "hops": h["hops"], "complete": bool(h["done"]), "created": h["created"]} for h in rows]})
+# ---------- END TOOL: TRACEROUTE ----------
-# ---------- TOOL: QR FORGE (shortcut launcher codes) ----------
-_QR_STYLES_CSS = """
-.qrgrid{display:grid;grid-template-columns:1fr;gap:1.1rem}
-@media(min-width:700px){.qrgrid{grid-template-columns:repeat(2,1fr)}}
-.qrout{display:flex;flex-direction:column;align-items:center;gap:.6rem;min-height:240px;justify-content:center}
-.qrout img{background:#fff;padding:12px;border-radius:12px;max-width:260px;width:100%}
-.schemebox{border:1px solid var(--line);border-radius:10px;padding:.7rem .9rem;margin:.5rem 0;font-size:.85rem}
-.schemebox b{color:var(--acc2)}
-.preset{margin:.3rem .3rem .3rem 0}
-"""
+# ---------- TOOL: TRAP CHAIN (breadcrumb tripwires) ----------
+def _bchain_tables():
+ con = db()
+ con.executescript("""CREATE TABLE IF NOT EXISTS bchain(id INTEGER PRIMARY KEY, user_id INTEGER, chain_id TEXT UNIQUE, name TEXT, created INTEGER);
+ CREATE TABLE IF NOT EXISTS bchain_hops(id INTEGER PRIMARY KEY, chain_row INTEGER, seq INTEGER, token TEXT UNIQUE, fired_ts INTEGER, fired_ip TEXT, fired_ua TEXT);""")
+ return con
-# Server-side QR renderer (no external service): tiny pure-python QR encoder
-def _qr_png(data, box=6, border=2):
- """Minimal QR encoder (byte mode, ECC L, versions 1-10) — returns PNG bytes."""
- try:
- import qrcode as _q
- img = _q.make(data, border=border, box_size=box)
- buf = io.BytesIO(); img.save(buf, "PNG"); return buf.getvalue()
- except ImportError:
- pass
- # fallback: delegate to local shot of the qrserver only if qrcode lib missing
- raise RuntimeError("qrcode module unavailable")
+def _bchain_get(uid, chain_id):
+ con = db()
+ r = con.execute("SELECT * FROM bchain WHERE chain_id=? AND user_id=?", (chain_id, uid)).fetchone()
+ if not r: return None
+ hops = con.execute("SELECT * FROM bchain_hops WHERE chain_row=? ORDER BY seq", (r["id"],)).fetchall()
+ return {"row": r, "hops": [dict(h) for h in hops]}
-_GAL = 'https://icloud.com/shortcuts/'
-_SCHEME_DOCS = [
- ("shortcuts://import-shortcut?url=", "Apple — installs a shortcut when scanned (iOS shows a one-tap Add confirmation)", "iOS"),
- ("shortcuts://run-shortcut?name=&input=", "Apple — runs an installed shortcut by name, optional input", "iOS"),
- ("shortcuts://create-shortcut", "Apple — opens a new empty shortcut editor", "iOS"),
- ("intent://scan/#Intent;scheme=http;package=com.android.chrome;end", "Android — chrome intent: opens URL in Chrome when scanned", "Android"),
- ("intent://#Intent;scheme=...;package=...;S.=;end", "Android — full intent form: scheme, target app package, string extras", "Android"),
- ("market://details?id=", "Android — opens the Play Store page for an app", "Android"),
- ("snackbar:// or any app deep link (spotify:, whatsapp://send?text=, tg://msg_url?url=)", "Any app's registered deep link — scannable like a URL", "both"),
-]
-
-@app.route("/qrforge")
-def qrforge():
- body = f"""
-
QR FORGE
QR codes for launcher scripts: Apple Shortcuts install/run codes and Android intent codes. Scan with the right phone and it fires the shortcut — the phone's own confirm prompt is the only gate, nothing to type.
-
-
-
-
-
-
-
-
-
-
-
-
the phone that scans runs this — iOS/Android will show their own confirm dialog
-
-
-
-
-
-
-
your code renders here
-
Scan behavior cheat-sheet
-""" + "".join(f'
{esc(s)} {esc(d)} {p}
' for s, d, p in _SCHEME_DOCS) + """
-
-
API: GET /api/qr?data=&box=4-12 → PNG (works for agents: any text, any scheme). QRs render server-side — nothing about your script is stored.
""" + how([
-"Pick a preset or type any scheme command — shortcuts://, intent://, market://, or any app deep link.",
-"For iOS install-codes: make your shortcut in the Shortcuts app, share it, copy the icloud.com/shortcuts link, feed it to the import preset.",
-"For Android intents: name the target app package and scheme — the scanner hands it to the OS which routes it to the app.",
-"Hit Forge — the PNG renders server-side from your exact string, byte for byte.",
-"Print it, sticker it, slap it on something. The scanning phone shows its own native confirm prompt before anything runs.",
-"Agents: GET /api/qr?data=… returns the raw PNG — no auth, rate limited."])
- body += flow("make a phone run a shortcut with one camera tap", [
-"you build a shortcut on your iPhone that texts your own number the phone's battery level.",
-"you share it → copy the icloud.com/shortcuts/… link → paste into QR FORGE with the install preset.",
-"you Forge, print the code, stick it on the fridge.",
-"anyone points their iPhone camera at it — iOS pops: 'Add Shortcut?' — one tap and it installs.",
-"them a second scan of a run-code fires it — battery text arrives.",
-"android same page, intent preset: the code hands a chrome intent to the OS and the page opens."] if False else [
-"you build a shortcut on your iPhone that texts your own number the battery level.",
-"you share it — copy the icloud.com/shortcuts link — paste it into QR FORGE with the install preset.",
-"you hit Forge and print the code on a sticker.",
-"them scans it: iOS shows its native 'Add Shortcut?' — one tap, installed.",
-"now the run-code variant fires it by name — no typing, no link, the phone just does the thing.",
-"android the intent presets do the same dance through Chrome and the Play Store."])
- body += gloss([("scheme","the prefix that hands a command to the phone OS — shortcuts://, intent://, market://"),("intent","Android's inter-app command format: scheme + package + extras"),("deep link","an app's private URL scheme that opens it to a specific action"),("package","Android app identifier, e.g. com.android.chrome")])
- body += agent_card('GET /api/qr?data=shortcuts%3A%2F%2F…', 'curl -o code.png "https://dark0rbits.thetempleofdoom.com/api/qr?data=intent%3A%2F%2Fscan%23Intent%3Bscheme%3Dhttp%3Bend"', 'Returns PNG bytes. Any scheme accepted — the phone OS is the executor.')
- return page("qrforge", body + "")
-
-@app.route("/api/qr")
-def api_qr():
- r = rate_limit("qr", 60, 60)
- if r: return r
- data = (param("data") or "").strip()
- if not data: return jsonify({"ok": False, "error": "data required"}), 400
- if len(data) > 900: return jsonify({"ok": False, "error": "data too long (900 max)"}), 400
- try: box = min(12, max(3, int(param("box") or 6)))
- except Exception: box = 6
- try:
- png = _qr_png(data, box=box)
- except Exception as e:
- return jsonify({"ok": False, "error": str(e)[:200]}), 500
- resp = Response(png, mimetype="image/png")
- resp.headers["Cache-Control"] = "public, max-age=86400"
- return resp
-# ---------- END TOOL: QR FORGE ----------
-
-
-# ---------- TOOL: ROTATOR ----------
-import random as _rnd
-import json as _json
-
-_ROTATOR_DB_READY = False
-
-def _rotator_db(con):
- global _ROTATOR_DB_READY
- if not _ROTATOR_DB_READY:
- con.execute("CREATE TABLE IF NOT EXISTS rotator_history(id INTEGER PRIMARY KEY, user_id INTEGER, ts INTEGER, ua TEXT, platform TEXT, seed TEXT)")
- con.commit()
- _ROTATOR_DB_READY = True
-
-ROTATOR_PLATFORMS = {
- "desktop": [
- "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36",
- "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.4 Safari/605.1.15",
- "Mozilla/5.0 (X11; Linux x86_64; rv:125.0) Gecko/20100101 Firefox/125.0",
- "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/123.0.0.0 Safari/537.36 Edg/123.0.0.0",
- ],
- "mobile": [
- "Mozilla/5.0 (iPhone; CPU iPhone OS 17_4 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.4 Mobile/15E148 Safari/604.1",
- "Mozilla/5.0 (Linux; Android 14; Pixel 8) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Mobile Safari/537.36",
- "Mozilla/5.0 (Linux; Android 13; SM-G991B) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/123.0.0.0 Mobile Safari/537.36",
- "Mozilla/5.0 (iPad; CPU OS 17_4 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.4 Mobile/15E148 Safari/604.1",
- ],
- "agent": [
- "Mozilla/5.0 (compatible; Googlebot/2.1; +http://www.google.com/bot.html)",
- "Mozilla/5.0 (compatible; bingbot/2.0; +http://www.bing.com/bingbot.htm)",
- "curl/8.4.0",
- "Wget/1.21.4 (linux-gnu)",
- "python-urllib/3.11",
- ],
- "stealth": [
- "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36",
- "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36",
- "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36",
- ],
-}
-ROTATOR_REFERER_POOL = [
- "https://www.google.com/", "https://duckduckgo.com/", "https://news.ycombinator.com/",
- "https://www.bing.com/", "https://www.reddit.com/", "(direct)",
-]
-ROTATOR_LANG_POOL = ["en-US,en;q=0.9", "en-GB,en;q=0.8", "de-DE,de;q=0.9,en;q=0.5", "fr-FR,fr;q=0.9", "ja-JP,ja;q=0.8"]
-
-def rotator_identity(platform, rng):
- ua = rng.choice(ROTATOR_PLATFORMS.get(platform, ROTATOR_PLATFORMS["desktop"]))
- return {
- "user_agent": ua,
- "referer": rng.choice(ROTATOR_REFERER_POOL),
- "accept_language": rng.choice(ROTATOR_LANG_POOL),
- "accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
- "sec_ch_ua_mobile": "?1" if platform == "mobile" else "?0",
- "dnt": rng.choice(["1", "0", "(none)"]),
- }
-
-def rotator_spin(platform, n, seed):
- platform = platform if platform in ROTATOR_PLATFORMS else "desktop"
- try:
- n = max(1, min(int(n), 25))
- except Exception:
- n = 5
- rng = _rnd.Random(str(seed)) if seed else _rnd.Random()
- ids = [rotator_identity(platform, rng) for _ in range(n)]
- curl = 'curl -A "' + ids[0]["user_agent"] + '"'
- if ids[0]["referer"] != "(direct)":
- curl += ' -e "' + ids[0]["referer"] + '"'
- curl += ' -H "Accept-Language: ' + ids[0]["accept_language"] + '" https://target.example/'
- return {"ok": True, "platform": platform, "count": len(ids), "seed": seed or None, "identities": ids, "curl_first": curl}
-
-@app.route("/rotator", methods=["GET", "POST"])
-def rotator_page():
+@app.route("/tchain")
+def tchain_page():
uid = current_user_id()
- platform = param("platform") or "desktop"
- if platform not in ROTATOR_PLATFORMS:
- platform = "desktop"
- n = param("n") or "5"
- seed = (param("seed") or "").strip()[:64]
- res = ""
- if request.method == "POST":
- r = rate_limit("rotator", 20, 60)
- if r:
- return r
- d = rotator_spin(platform, n, seed)
- try:
- con = db()
- _rotator_db(con)
- for ident in d["identities"][:5]:
- con.execute("INSERT INTO rotator_history(user_id,ts,ua,platform,seed) VALUES(?,?,?,?,?)",
- (uid or 0, int(time.time()), ident["user_agent"], platform, seed or ""))
- con.commit()
- except Exception:
- pass
- rows = ""
- for i, ident in enumerate(d["identities"], 1):
- rows += ("
""" + how([
+"Plant hop 1 where an intruder will look first — a file named 'passwords.txt', a folder called 'backup'.",
+"The hop's decoy note contains hop 2's URL. The curious can't resist: they follow it.",
+"Every hop they trip pings your inbox with IP, location, device — and lights up the chain board.",
+"By the time they stop, you have a map of exactly how far they got and how fast they moved.",
+"Chains work on paper too: print the decoys, leave them in a drawer."]))
+ con = _bchain_tables()
+ chains = con.execute("SELECT * FROM bchain WHERE user_id=? ORDER BY id DESC LIMIT 20", (uid,)).fetchall()
+ clist = ""
+ for ch in chains:
+ hops = con.execute("SELECT * FROM bchain_hops WHERE chain_row=? ORDER BY seq", (ch["id"],)).fetchall()
+ fired = sum(1 for h in hops if h["fired_ts"])
+ clist += f'◈ {esc(ch["name"])} {fired}/{len(hops)} hops fired'
body = f"""
-
HEADER ROTATOR
Spin consistent browser identities — User-Agent, referer, language, DNT — from four pools. Same seed replays the exact same rotation every time. No logs kept beyond your own history.
-
+{('
Your chains' + clist + '
') if clist else ''}
+""" + flow("map the path a thief takes through your files", [
+"you arm a 4-hop chain and get 4 unique URLs plus 4 decoy notes.",
+"hop 1's URL goes inside a file called RESTORE_PASSWORDS_HERE.txt on the desktop.",
+"its decoy text says 'real vault: ' — bait for whoever opened the file.",
+"them opens the file, clicks the link (a blank 404) — your inbox pings: hop 1, their IP, their city.",
+"they follow the 'vault' link — hop 2 fires. Then 3. The board draws their path in real time.",
+"when the chain fully burns you get one summary alert with the whole timeline."]) + how([
+"Hop 1 is a plain tripwire. Hops 2+ hide inside the previous hop's decoy note — following the trail IS the confession.",
+"Every fire logs IP, geolocation, device, timestamp, and alerts your inbox instantly.",
+"The board shows the chain as a diagram: armed hops wait in green, fired ones glow red with their details.",
+"Fully-burned chains fire one final summary alert with the complete timeline.",
+"Decoy text is fully editable before you plant it — make it fit where it lives."])
+ body += gloss([("hop","one tripwire in the chain"),("decoy","plausible text that carries the next hop's URL"),("burn","a fully-tripped chain"),("board","the live diagram of the chain's state")])
+ body += agent_card('POST /api/tchain/create name= nhops=', 'curl -X POST https://dark0rbits.thetempleofdoom.com/api/tchain/create -d "name=laptop" -d "nhops=4" -H "Cookie: dark0rbits_tok=…"', 'GET /api/tchain/status?chain_id= → per-hop state.')
+ return page("tchain", body)
-@app.route("/api/rotator", methods=["GET", "POST"])
-def api_rotator():
- r = rate_limit("rotator", 20, 60)
- if r:
- return r
- platform = (param("platform") or "desktop").strip().lower()
- if platform not in ROTATOR_PLATFORMS:
- return jsonify({"ok": False, "error": "platform must be one of: " + ", ".join(sorted(ROTATOR_PLATFORMS))}), 400
- seed = (param("seed") or "").strip()[:64]
- d = rotator_spin(platform, param("n") or "1", seed)
- return jsonify(d)
-
-@app.route("/api/rotator/pools")
-def api_rotator_pools():
- return jsonify({"ok": True, "pools": {k: len(v) for k, v in ROTATOR_PLATFORMS.items()},
- "referers": len(ROTATOR_REFERER_POOL), "languages": len(ROTATOR_LANG_POOL)})
-# ---------- END TOOL: ROTATOR ----------
-
-
-# ---------- TOOL: IDENTITY SHELF ----------
-import time as _shelf_time
-
-def _shelf_data(uid):
- con = db()
- now = int(_shelf_time.time())
- out = {}
- out["mailboxes"] = [dict(r) for r in con.execute(
- "SELECT address, expires, cnt, paid FROM mailboxes WHERE user_id=? ORDER BY (expires>0), expires LIMIT 50", (uid,)).fetchall()]
- out["sms"] = [dict(r) for r in con.execute(
- "SELECT phone, service, expires, status FROM sms_rentals WHERE user_id=? AND expires>0 ORDER BY expires LIMIT 50", (uid,)).fetchall()]
- out["deaddrops"] = [dict(r) for r in con.execute(
- "SELECT token, expires, reads_left, burn_after FROM deadrops WHERE user_id=? ORDER BY (expires>0), expires LIMIT 50", (uid,)).fetchall()]
- out["canaries"] = []
- for r in con.execute("SELECT token, tag, armed, rearm FROM canaries WHERE user_id=? ORDER BY id DESC LIMIT 50", (uid,)).fetchall():
- d = dict(r)
- d["hits"] = con.execute("SELECT COUNT(*) c FROM canary_hits WHERE canary_id=(SELECT id FROM canaries WHERE token=?)", (r["token"],)).fetchone()["c"]
- out["canaries"].append(d)
- out["trackables"] = [dict(r) for r in con.execute(
- "SELECT token, filename, paid, created FROM trackables WHERE user_id=? ORDER BY id DESC LIMIT 50", (uid,)).fetchall()]
- # stats
- live = lambda e: e and e > now
- n_live = sum(1 for m in out["mailboxes"] if m["paid"] and live(m["expires"])) \
- + sum(1 for s in out["sms"] if live(s["expires"])) \
- + sum(1 for d in out["deaddrops"] if live(d["expires"]))
- expiring = sorted([e for grp in ("mailboxes", "sms", "deaddrops") for e in
- (x["expires"] for x in out[grp] if live(x.get("expires"))) if e])
- out["stats"] = {
- "live_identities": n_live,
- "next_expiry": expiring[0] if expiring else None,
- "total_mail_received": sum(m["cnt"] or 0 for m in out["mailboxes"]),
- "armed_traps": sum(1 for c in out["canaries"] if c["armed"]),
- "total_trap_hits": sum(c["hits"] for c in out["canaries"]),
- }
- return out
-
-def _shelf_badge(expires):
- now = int(_shelf_time.time())
- if not expires:
- return 'no expiry'
- left = expires - now
- when = _shelf_time.strftime("%b %d %H:%M", _shelf_time.localtime(expires))
- if left <= 0:
- return f'EXPIRED {when}'
- cls = "bad" if left < 86400 else "ok"
- unit = "d" if left >= 86400 else "h"
- val = left // 86400 if left >= 86400 else left // 3600
- return f'{val}{unit} left · {when}'
-
-@app.route("/shelf")
-def shelf():
+@app.route("/tchain", methods=["POST"])
+def tchain_create():
uid = current_user_id()
- if not uid:
- return page("shelf", """
IDENTITY SHELF
One dashboard for every burner you own — mailboxes, numbers, dead-drops, traps — with live countdowns so nothing dies silently.
""" + how([
-"Every burner on dark0rbits has a lifespan — mailboxes expire, rentals run out, dead-drops burn.",
-"The shelf lists all of yours in one place with live countdown badges.",
-"Under 24 hours left, a badge turns red — renew or replace before it dies.",
-"Expired items stay listed so you can clean up or recreate them.",
-"Agents: GET /api/shelf returns the same data as JSON for monitoring."]))
- d = _shelf_data(uid)
- def rows_mail():
- if not d["mailboxes"]: return '
"
- for c in d["canaries"])
- st = d["stats"]
- nxt = _shelf_time.strftime("%b %d %H:%M", _shelf_time.localtime(st["next_expiry"])) if st["next_expiry"] else "—"
+ if not uid: return page("tchain", "
login required
")
+ name = (param("name") or "chain")[:80]
+ try: nhops = min(8, max(2, int(param("nhops") or 4)))
+ except Exception: nhops = 4
+ con = _bchain_tables()
+ chain_id = secrets.token_urlsafe(6)
+ cur = con.execute("INSERT INTO bchain(user_id,chain_id,name,created) VALUES(?,?,?,?)", (uid, chain_id, esc(name), int(time.time())))
+ row_id = cur.lastrowid
+ for seq in range(nhops):
+ con.execute("INSERT INTO bchain_hops(chain_row,seq,token) VALUES(?,?,?)", (row_id, seq, secrets.token_urlsafe(10)))
+ con.commit()
+ return Response(status=302, headers={"Location": f"/tchain/board?id={chain_id}"})
+
+@app.route("/tchain/board")
+def tchain_board():
+ uid = current_user_id()
+ if not uid: return page("tchain", "
login required
")
+ chain_id = param("id") or ""
+ ch = _bchain_get(uid, chain_id)
+ if not ch: return page("tchain", "
unknown chain
")
+ # render hop cards with decoy text + plant instructions
+ hop_html = ""
+ nhops = len(ch["hops"])
+ for h in ch["hops"]:
+ url = f"{SITE}/tc/{h['token']}"
+ if h["fired_ts"]:
+ geo = enrich_ip(h["fired_ip"]) or {}
+ where = f"{geo.get('city','—')}, {geo.get('countryCode','')} · {geo.get('isp','')}" if geo else "—"
+ state = f'FIRED {time.strftime("%b %d %H:%M", time.localtime(h["fired_ts"]))} · {esc(h["fired_ip"])} · {esc(where)} {esc((h["fired_ua"] or "")[:80])}'
+ else:
+ state = 'WAITING'
+ decoy = ""
+ if h["seq"] < nhops - 1:
+ nxt = ch["hops"][h["seq"] + 1]
+ decoy = (f'
Everything you own that can die, on one page — with live countdowns. Know when every burner expires before it expires.
-{kv([("Live identities", f"{st['live_identities']} mailboxes + numbers + drops"),
- ("Next to expire", f"{nxt}"), ("Mail received (all time)", st["total_mail_received"]),
- ("Armed traps", f"{st['armed_traps']} armed · {st['total_trap_hits']} total hits")])}
-
BURNER MAILBOXES
Address
Mail
Life
{rows_mail()}
-
SMS NUMBERS
Number
Service
Life
Status
{rows_sms()}
-
DEAD-DROPS
Token
Burns
Life
{rows_dd()}
-
CANARY TRAPS
Tag
Hits
Status
{rows_can()}
-""" + how([
-"Every burner has a lifespan — this page lists all of yours with a countdown badge per item.",
-"Badges tick live (every 30s); the static expiry date renders even without JS.",
-"Green = over 24h left. Red = under 24h or already dead — renew or replace.",
-"Quick links jump straight to each item: mailbox, dead-drop, trap hit log.",
-"Agents: GET /api/shelf returns identical JSON — wire it into a cron and get paged before anything dies."])
- body += flow("never lose a burner to the clock again", [
-"you run 3 burner mailboxes for signups and 2 SMS numbers for verifications.",
-"you open the shelf once a morning: five green badges, everything alive.",
-"Thursday: one mailbox badge is red — 6h left. You have all day to migrate that identity.",
-"agent a cron hits /api/shelf hourly and messages you when anything drops under 24h.",
-"nothing expires silently. No more 'why did my verification stop working' mysteries."])
- body += gloss([("burner","a disposable identity — mailbox, phone number, or drop",),("TTL","time to live — how long until the service retires it"),("burn-after-read","dead-drops self-destruct after N openings")])
- body += agent_card('GET /api/shelf', 'curl "https://dark0rbits.thetempleofdoom.com/api/shelf" -H "Cookie: dark0rbits_tok=…"', 'Returns mailboxes, sms, deaddrops, canaries + stats. Perfect for expiry-monitoring crons.')
- return page("shelf", body)
+
CHAIN {esc(ch['row']['name'])}
{nhops} hops. Plant hop 1 where they'll find it; the decoys feed them the rest.
"""
+ return page("tchain", body)
-@app.route("/api/shelf")
-def api_shelf():
+@app.route("/tc/")
+def bchain_fire(token):
+ con = _bchain_tables()
+ h = con.execute("SELECT * FROM bchain_hops WHERE token=?", (token,)).fetchone()
+ if not h: return "Not Found", 404
+ ip = request.headers.get("X-Real-IP") or request.remote_addr or "?"
+ ua = request.headers.get("User-Agent", "")
+ first = not h["fired_ts"]
+ con.execute("UPDATE bchain_hops SET fired_ts=?, fired_ip=?, fired_ua=? WHERE id=?",
+ (int(time.time()), ip, ua[:200], h["id"]))
+ con.commit()
+ ch = con.execute("SELECT * FROM bchain WHERE id=?", (h["chain_row"],)).fetchone()
+ if ch and ch["user_id"]:
+ geo = enrich_ip(ip)
+ where = f" — {geo.get('city','')}, {geo.get('countryCode','')} · {geo.get('isp','')}" if geo else ""
+ all_hops = con.execute("SELECT * FROM bchain_hops WHERE chain_row=? ORDER BY seq", (h["chain_row"],)).fetchall()
+ if first and all(x["fired_ts"] for x in all_hops):
+ body = f"🔗 TRAP CHAIN FULLY BURNED: '{ch['name']}' — every hop tripped. Timeline on the board."
+ else:
+ body = f"🔗 TRAP CHAIN hop {h['seq']+1} fired: '{ch['name']}' — IP {esc(ip)}{esc(where)} · device {esc(ua[:80])}"
+ con.execute("INSERT INTO messages(user_id,sender,body,created) VALUES(?,?,?,?)", (ch["user_id"], "operator-bot", body, int(time.time())))
+ con.commit()
+ return "Not Found", 404
+
+@app.route("/api/tchain/create", methods=["POST"])
+def api_tchain_create():
uid = key_user() or current_user_id()
- if not uid:
- return jsonify({"ok": False, "error": "login required — free no-KYC account at /inbox"}), 401
- d = _shelf_data(uid)
- for grp in ("mailboxes", "sms", "deaddrops", "canaries", "trackables"):
- for x in d[grp]:
- x.pop("token", None) # never leak tokens over API
- return jsonify({"ok": True, **d})
-# ---------- END TOOL: IDENTITY SHELF ----------
-
-
-# ---------- TOOL: SNAP LINK (client-side shortener) ----------
-_SNAP_PAGE = r"""
-▶ SNAP — one click more
-
-
-
▶ SNAP — one click more
-
This is a dark0rbits snap link. The destination lives in the part of the address after the # — which is never sent to any server, anywhere. It decoded right here in your browser.
-
JavaScript is off — snap links decode client-side, so this one can't open. The raw destination is visible in the address bar after the #.
-
opening…
-
-
RECENT FROM THIS BROWSER
-
-
history lives only in this browser's storage — the server keeps nothing
-
-
-
-"""
-
-@app.route("/snap")
-@app.route("/s")
-@app.route("/s/")
-def snap(_any=None):
- r = Response(_SNAP_PAGE, mimetype="text/html")
- r.headers["Cache-Control"] = "no-store"
- r.headers["X-Robots-Tag"] = "noindex"
- return r
-
-@app.route("/api/snap/decode", methods=["POST"])
-def api_snap_decode():
- """Agents: decode a snap fragment server-side WITHOUT opening it. raw = text after #."""
- r = rate_limit("snap", 60, 60)
+ if not uid: return jsonify({"ok": False, "error": "auth required"}), 401
+ r = rate_limit("tchain", 20, 60)
if r: return r
- raw = (param("raw") or "").strip()
- if not raw: return jsonify({"ok": False, "error": "raw (fragment after #) required"}), 400
- try:
- s = raw.replace("-", "+").replace("_", "/")
- s += "=" * (-len(s) % 4)
- import base64 as _b
- target = base64.b64decode(s).decode("utf-8", "replace")
- except Exception:
- return jsonify({"ok": False, "error": "not a valid snap fragment"}), 400
- import re as _re
- m = _re.match(r"^https?://([^/]+)", target, _re.I)
- return jsonify({"ok": True, "target": target,
- "host": m.group(1) if m else None,
- "scheme_safe": bool(m)})
-# ---------- END TOOL: SNAP LINK ----------
-
-
-# ---------- TOOL: UNFURL ----------
-def _unfurl_db():
- con = db()
- con.execute("""CREATE TABLE IF NOT EXISTS unfurls(id INTEGER PRIMARY KEY, user_id INTEGER, url TEXT,
- final_url TEXT, hops TEXT, status INTEGER, err TEXT, created INTEGER)""")
- return con
-
-def _unfurl_ua():
- con = _unfurl_db()
- try:
- r = con.execute("SELECT val FROM settings WHERE user_id=? AND key='unfurl_ua'", (current_user_id() or 0,)).fetchone()
- if r and r["val"]:
- return r["val"]
- except Exception:
- pass
- return "Mozilla/5.0 (Dark0rbits unfurl)"
-
-def _unfurl_save(user_id, url, final_url, hops_json, status, err):
- con = _unfurl_db()
- con.execute("INSERT INTO unfurls(user_id,url,final_url,hops,status,err,created) VALUES(?,?,?,?,?,?,?)",
- (user_id, url, final_url, hops_json, status, err, int(time.time())))
+ name = (param("name") or "chain")[:80]
+ try: nhops = min(8, max(2, int(param("nhops") or 4)))
+ except Exception: nhops = 4
+ con = _bchain_tables()
+ chain_id = secrets.token_urlsafe(6)
+ cur = con.execute("INSERT INTO bchain(user_id,chain_id,name,created) VALUES(?,?,?,?)", (uid, chain_id, esc(name), int(time.time())))
+ row_id = cur.lastrowid
+ hops = []
+ for seq in range(nhops):
+ tok = secrets.token_urlsafe(10)
+ con.execute("INSERT INTO bchain_hops(chain_row,seq,token) VALUES(?,?,?)", (row_id, seq, tok))
+ hops.append({"seq": seq, "tripwire": f"{SITE}/tc/{tok}"})
con.commit()
- return con.execute("SELECT id FROM unfurls WHERE user_id=? ORDER BY id DESC LIMIT 1", (user_id,)).fetchone()["id"]
+ return jsonify({"ok": True, "chain_id": chain_id, "hops": hops})
-_NOFOLLOW = ("javascript:", "data:", "mailto:", "tel:", "blob:", "about:", "file:", "chrome:", "intent:", "ws:", "wss:")
-
-def _unfurl_meta(html_text, base_url):
- """Extract , meta description, og:*, twitter:* tags. Returns (title, metas dict, links, scripts, forms, iframes)."""
- metas, links, scripts, forms, iframes = {}, [], [], [], []
- title = ""
- m = re.search(r"]*>(.*?)", html_text, re.I | re.S)
- if m:
- title = re.sub(r"\s+", " ", m.group(1)).strip()[:300]
- for m in re.finditer(r"]+>", html_text, re.I):
- tag = m.group(0)
- def attr(name):
- mm = re.search(name + r"\s*=\s*[\"']([^\"']*)[\"']", tag, re.I)
- return mm.group(1) if mm else ""
- nm, prop, con = attr("name"), attr("property"), attr("content")
- key = (prop or nm).lower()
- if key and con:
- metas[key] = con[:600]
- for m in re.finditer(r"]+rel\s*=\s*[\"']?[^\"'>]*stylesheet[^\"'>]*[\"']?[^>]*>", html_text, re.I):
- links.append(m.group(0)[:500])
- for m in re.finditer(r"