",
+ "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"| {esc(t['platform'])} | avatar | "
+ f"{esc((t['hashes'].get('dhash') or ''))} | {m} | "
+ f"profile"
+ f" · img |
")
+ elif t["status"] == "constructed link":
+ rows += (f"| {esc(t['platform'])} | constructed | "
+ f"— | manual lead | open profile → grab photo by hand |
")
+ else:
+ rows += (f"| {esc(t['platform'])} | {esc(t['status'])} | "
+ f"— | — | profile |
")
+ res += (f'Avatar harvest — {esc(username)}'
+ f'
| Platform | Status | dHash | Verdict | Links |
|---|
{rows}
'
+ '
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 leadsdark0rbits 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).
+
| " + str(i) + " | " + esc(ident["user_agent"]) + " | "
+ + esc(ident["referer"]) + " | " + esc(ident["accept_language"]) + " | "
+ + esc(ident["dnt"]) + " |
")
+ res = ('' + str(d["count"]) + " rotated identities — " + esc(d["platform"]) + " pool"
+ + (" · seed " + esc(seed) + " (replayable)" if seed else "") + ""
+ + '
| # | User-Agent | Referer | Accept-Language | DNT |
|---|
'
+ + rows + "
"
+ + '
First one as curl: ' + esc(d["curl_first"]) + "
")
+ 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.
+
+{res}
+API: GET /api/rotator?platform=mobile&n=10&seed=abc → JSON identities + ready-made curl.
""" + how([
+ "Pick a pool: desktop, mobile, agent or stealth — each holds real-world header strings.",
+ "Choose how many identities to spin, 1 to 25 per call.",
+ "Give it a seed and the rotation becomes deterministic — the same seed always replays the same identities in the same order.",
+ "Leave the seed blank for a fresh random rotation every call.",
+ "Every identity is a full consistent set: UA, referer, Accept-Language, DNT — not just a UA string.",
+ "The first identity comes back as a ready-to-paste curl command.",
+ "Agents: GET /api/rotator with the same params, JSON out, rate-limited 20/min.",
+ "Nothing is stored except the last few spins in your own account history.",
+ ]) + flow("Rotating through a scrape run", [
+ 'you set pool=mobile, n=10, seed=run-42 and hit Spin.',
+ 'The lab hands back 10 consistent identities — UA, referer, language, DNT matched per identity.',
+ 'you copy the curl line for the first one or call /api/rotator from your script.',
+ 'The target sees ten different plausible visitors instead of one hammering client.',
+ 'Re-run with the same seed later to reproduce the exact rotation for debugging.',
+ ]) + gloss([
+ ("rotation", "cycling through a pool of values so no single fingerprint repeats too often"),
+ ("seed", "a string fed to the RNG — same seed, same sequence, every time"),
+ ("DNT", "Do-Not-Track header — 0, 1 or absent, randomized per identity"),
+ ("consistent identity", "UA + referer + language that plausibly belong to the same browser"),
+ ]) + agent_card('GET /api/rotator?platform=mobile&n=10&seed=abc',
+ 'curl "https://dark0rbits.thetempleofdoom.com/api/rotator?platform=mobile&n=10&seed=abc"',
+ 'JSON: identities[] with user_agent, referer, accept_language, dnt + curl_first. Rate limit 20/min.')
+ return page("rotator", 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():
+ 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 '| none |
'
+ return "".join(f"| {esc(m['address'])} | {m['cnt'] or 0} | "
+ f"{_shelf_badge(m['expires'])} | open |
"
+ for m in d["mailboxes"])
+ def rows_sms():
+ if not d["sms"]: return '| none |
'
+ return "".join(f"| {esc(s['phone'])} | {esc(s['service'])} | "
+ f"{_shelf_badge(s['expires'])} | {esc(s['status'] or '')} |
" for s in d["sms"])
+ def rows_dd():
+ if not d["deaddrops"]: return '| none |
'
+ return "".join(f"…{esc(t['token'][-6:])} | {t['reads_left']}/{t['burn_after']} reads left | "
+ f"{_shelf_badge(t['expires'])} | open |
"
+ for t in d["deaddrops"])
+ def rows_can():
+ if not d["canaries"]: return '| none |
'
+ return "".join(f"| {esc(c['tag'])} | {c['hits']} | "
+ f"{'armed' + (' ⟳' if c['rearm'] else '') + '' if c['armed'] else 'triggered'} | "
+ f"hits |
"
+ 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 "—"
+ body = f"""
+IDENTITY SHELF
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)
+
+@app.route("/api/shelf")
+def api_shelf():
+ 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: 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())))
+ con.commit()
+ return con.execute("SELECT id FROM unfurls WHERE user_id=? ORDER BY id DESC LIMIT 1", (user_id,)).fetchone()["id"]
+
+_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"<title[^>]*>(.*?)", 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"]*>", html_text, re.I):
+ scripts.append(m.group(0)[:500])
+ for m in re.finditer(r"]*>", html_text, re.I):
+ forms.append(m.group(0)[:500])
+ for m in re.finditer(r"]*>", html_text, re.I):
+ iframes.append(m.group(0)[:500])
+ return title, metas, links, scripts, forms, iframes
+
+def _abs(u, base):
+ try:
+ return urllib.parse.urljoin(base, u)
+ except Exception:
+ return u
+
+def _fetch_no_redirect(u, ua):
+ class _NR(urllib.request.HTTPRedirectHandler):
+ def redirect_request(self, req, fp, code, msg, headers, newurl):
+ return None
+ op = urllib.request.build_opener(_NR, urllib.request.HTTPSHandler(context=_CTX))
+ req = urllib.request.Request(u, headers={"User-Agent": ua, "Accept": "text/html,*/*"})
+ try:
+ with op.open(req, timeout=10) as r:
+ return r.status, r.url, r.headers, r.read(300000).decode("utf-8", "replace")
+ except urllib.error.HTTPError as e:
+ body = ""
+ try:
+ body = e.read(100000).decode("utf-8", "replace")
+ except Exception:
+ pass
+ return e.code, u, e.headers, body
+ except Exception as e:
+ return 0, u, {}, str(e)
+
+def _unfurl(start_url, max_hops=6, extract=True, ua=None):
+ """Follow redirects manually one hop at a time; collect chain + final page metadata."""
+ ua = ua or _unfurl_ua()
+ hops, seen, cur, hop_status, final_body = [], set(), start_url, 0, ""
+ scheme_ok = cur.lower().startswith(("http://", "https://"))
+ if not scheme_ok:
+ return {"ok": False, "error": "URL must start with http:// or https://", "hops": []}
+ for i in range(max_hops + 1):
+ if cur in seen:
+ hops.append({"hop": i + 1, "url": cur, "status": 0, "location": "", "note": "redirect loop detected"})
+ cur = None
+ break
+ seen.add(cur)
+ status, final_url, headers, body = _fetch_no_redirect(cur, ua)
+ hops.append({"hop": i + 1, "url": cur, "status": status, "location": headers.get("Location", "") if headers else ""})
+ if status in (301, 302, 303, 307, 308) and headers and headers.get("Location"):
+ loc = headers["Location"]
+ low = loc.lower()
+ if any(low.startswith(p) for p in _NOFOLLOW):
+ hops[-1]["note"] = "non-http scheme — not followed"
+ cur = None
+ final_body = body
+ break
+ nxt = _abs(loc, cur)
+ hops[-1]["location"] = nxt
+ cur = nxt
+ continue
+ hop_status = status
+ final_body = body
+ cur = None
+ break
+ if cur is not None:
+ hop_status = 0
+ out = {"ok": hop_status > 0, "start": start_url, "final": hops[-1]["url"] if hops else "",
+ "status": hop_status, "hop_count": len(hops), "hops": hops}
+ if extract and final_body:
+ title, metas, css, scripts, forms, iframes = _unfurl_meta(final_body, out["final"])
+ out["title"] = title
+ out["metas"] = metas
+ out["css_count"] = len(css)
+ out["script_count"] = len(scripts)
+ out["form_count"] = len(forms)
+ out["iframe_count"] = len(iframes)
+ out["iframes"] = [dict(src=re.search(r"src\s*=\s*[\"']([^\"']*)", f, re.I).group(1) if re.search(r"src\s*=\s*[\"']([^\"']*)", f, re.I) else "", tag=f[:300]) for f in iframes[:10]]
+ out["forms"] = [dict(action=re.search(r"action\s*=\s*[\"']([^\"']*)", f, re.I).group(1) if re.search(r"action\s*=\s*[\"']([^\"']*)", f, re.I) else "", method=(re.search(r"method\s*=\s*[\"']?(\w+)", f, re.I).group(1).lower() if re.search(r"method\s*=\s*[\"']?(\w+)", f, re.I) else "get"), tag=f[:300]) for f in forms[:10]]
+ return out
+
+@app.route("/unfurl", methods=["GET", "POST"])
+def unfurl_page():
+ uid = current_user_id()
+ result, url_in, max_hops, extract, err = None, "", 6, "1", ""
+ if request.method == "POST":
+ r = rate_limit("unfurl", 20, 60)
+ if r:
+ return r
+ url_in = (param("url") or "").strip()[:500]
+ try:
+ max_hops = max(1, min(10, int(param("max_hops") or 6)))
+ except Exception:
+ max_hops = 6
+ extract = param("extract") in ("1", "on", "true", "yes", "")
+ if not url_in:
+ err = "paste a URL first"
+ elif not url_in.lower().startswith(("http://", "https://")):
+ err = "URL must start with http:// or https://"
+ else:
+ result = _unfurl(url_in, max_hops, extract)
+ try:
+ _unfurl_save(uid or 0, url_in, result.get("final", ""), json.dumps(result.get("hops", []))[:8000], result.get("status", 0), result.get("error", ""))
+ except Exception:
+ pass
+ body = f"""UNFURL URL
+Follow every redirect hop by hand — scheme, host, status, location — then dissect the final page. Shorteners, cloakers, affiliate chains: laid open.
+
+
"""
+ if err:
+ body += '' + esc(err) + '
'
+ if result:
+ if result.get("ok"):
+ body += 'resolved in ' + str(result["hop_count"]) + ' hop(s) — final status ' + str(result["status"]) + '
'
+ else:
+ body += '' + esc(result.get("error") or ("request failed (status " + str(result.get("status", 0)) + ")")) + '
'
+ rows = "".join(
+ "| " + str(h.get("hop", "")) + " | " + esc(h.get("url", "")) + " | "
+ + ("" if 200 <= int(h.get("status", 0) or 0) < 400 else "") + esc(str(h.get("status", ""))) + " | "
+ + ("" + esc(h.get("location", "")) + "" if h.get("location") else ("—" if not h.get("note") else '' + esc(h["note"]) + "")) + " |
"
+ for h in result.get("hops", []))
+ body += 'REDIRECT CHAIN| # | URL | Status | Location / note |
|---|
' + rows + '
'
+ if result.get("title") is not None and result.get("ok"):
+ mrows = "".join("| " + esc(k) + " | " + esc(v) + " |
" for k, v in result.get("metas", {}).items())
+ body += ('FINAL PAGE'
+ + "
Title
" + esc(result.get("title") or "—") + "
"
+ + "
Final URL
" + esc(result["final"]) + "
"
+ + "
Status
" + str(result["status"]) + "
"
+ + "
Stylesheets
" + str(result.get("css_count", 0)) + "
"
+ + "
Scripts
" + str(result.get("script_count", 0)) + "
"
+ + "
Forms / iframes
" + str(result.get("form_count", 0)) + " / " + str(result.get("iframe_count", 0)) + "
"
+ + '
META TAGS| Name | Content |
|---|
' + (mrows or "| — |
") + "
IFRAMES| src |
|---|
' + "".join("" + esc(f.get("src") or "—") + " |
" for f in result["iframes"]) + "
"
+ if result.get("forms"):
+ body += 'FORMS| Method | Action |
|---|
' + "".join("| " + esc(f.get("method", "get")) + " | " + esc(f.get("action") or "—") + " |
" for f in result["forms"]) + "
"
+ body += how([
+ "Paste any URL — a shortener, a cloaker, an affiliate link, a login redirect.",
+ "We request it with redirects disabled, so every 30x comes back to us one hop at a time.",
+ "Each hop records scheme, host, status code and the exact Location header — relative locations are resolved absolute.",
+ "Non-http schemes (javascript:, data:, intent:) are flagged and never followed.",
+ "Loops are detected: the same URL twice ends the chain with a clear note.",
+ "The final page (any status) is dissected: title, meta/og/twitter tags, stylesheet and script counts, forms and iframes.",
+ "Everything is available as JSON at /api/unfurl for agents.",
+ ])
+ body += flow("checking a shortened link before clicking", [
+ "you receive https://bit.ly/3xYz in a message and want to know where it really goes.",
+ "1. Paste it into UNFURL with default 6 hops.",
+ "2. The chain shows 301 → tracker.example → 302 → login.example — two hops, both logged.",
+ "you see the final host is a credential-phishing page with one form and an off-domain iframe.",
+ "3. Grab the JSON from /api/unfurl and feed it to your pipeline.",
+ ])
+ body += gloss([
+ ("redirect hop", "One 301/302/303/307/308 jump. Browsers follow them silently; we stop at each one."),
+ ("Location header", "Where the server says to go next. Can be relative — we resolve it against the current URL."),
+ ("redirect loop", "The same URL appearing twice in a chain — the server is chasing its tail."),
+ ("og: meta tags", "Open Graph tags pages use for link previews — often reveal the real content behind a cloaker."),
+ ("scheme", "The http:// or https:// part. Non-http schemes in Location are dangerous and never followed here."),
+ ])
+ body += agent_card("GET /api/unfurl?url=…&max_hops=6",
+ "curl -s '" + SITE + "/api/unfurl?url=https://bit.ly/3xYz&max_hops=8'",
+ "Returns the full chain plus final-page metadata as JSON. Optional extract=0 to skip dissection.")
+ return page("hunt", body)
+
+@app.route("/api/unfurl")
+def api_unfurl():
+ uid = key_user() or current_user_id()
+ if not uid:
+ return jsonify({"ok": False, "error": "auth required: account session or Authorization: Bearer key"}), 401
+ r = rate_limit("unfurl_api", 30, 60)
+ if r:
+ return r
+ u = (param("url") or "").strip()[:500]
+ if not u:
+ return jsonify({"ok": False, "error": "url required"}), 400
+ if not u.lower().startswith(("http://", "https://")):
+ return jsonify({"ok": False, "error": "url must start with http:// or https://"}), 400
+ try:
+ mh = max(1, min(10, int(param("max_hops") or 6)))
+ except Exception:
+ mh = 6
+ extract = param("extract") not in ("0", "false", "no")
+ res = _unfurl(u, mh, extract)
+ code = 200 if (res.get("ok") and 200 <= res.get("status", 0) < 400) else 502
+ try:
+ _unfurl_save(uid, u, res.get("final", ""), json.dumps(res.get("hops", []))[:8000], res.get("status", 0), res.get("error", ""))
+ except Exception:
+ pass
+ return jsonify(res), code
+
+@app.route("/api/unfurl/history")
+def api_unfurl_history():
+ uid = key_user() or current_user_id()
+ if not uid:
+ return jsonify({"ok": False, "error": "auth required"}), 401
+ con = _unfurl_db()
+ rows = con.execute("SELECT id,url,final_url,status,created FROM unfurls WHERE user_id=? ORDER BY id DESC LIMIT 25", (uid,)).fetchall()
+ return jsonify({"ok": True, "items": [dict(r) for r in rows]})
+# ---------- END TOOL: UNFURL ----------
+
+
+# ---------- TOOL: WARP ARCHIVE ----------
+import urllib.parse as _wurl
+
+WARP_TTL = 21600 # 6h cache
+
+def _warp_db():
+ con = db()
+ con.execute("CREATE TABLE IF NOT EXISTS warp_cache(domain TEXT, kind TEXT, data TEXT, ts INTEGER, PRIMARY KEY(domain,kind))")
+ con.commit()
+ return con
+
+def _warp_get(domain, kind):
+ con = _warp_db()
+ r = con.execute("SELECT data, ts FROM warp_cache WHERE domain=? AND kind=?", (domain, kind)).fetchone()
+ if r and (time.time() - r["ts"]) < WARP_TTL:
+ return jf(r["data"])
+ return None
+
+def _warp_put(domain, kind, obj):
+ con = _warp_db()
+ con.execute("INSERT INTO warp_cache(domain,kind,data,ts) VALUES(?,?,?,?) ON CONFLICT(domain,kind) DO UPDATE SET data=excluded.data, ts=excluded.ts",
+ (domain, kind, json.dumps(obj), int(time.time())))
+ con.commit()
+
+def _warp_norm(raw):
+ d = str(raw or "").strip().lower().lstrip()
+ d = _wurl.urlparse(d if "//" in d else "http://" + d).netloc.split("@")[-1].split(":")[0]
+ if d.startswith("www."):
+ d = d[4:]
+ if not d or "/" in d or " " in d or d.count(".") < 1 or not re.match(r"^[a-z0-9.-]+$", d):
+ return None
+ return d
+
+def _warp_cdx(domain, extra=""):
+ """Collapsed-per-year snapshot list from the Wayback CDX API."""
+ url = "http://web.archive.org/cdx/search/cdx?url=" + _wurl.quote(domain) + "&output=json&limit=150&collapse=timestamp:4" + extra
+ st, txt = http(url)
+ arr = jf(txt)
+ if st != 200 or not isinstance(arr, list):
+ return None, (st or 0)
+ if len(arr) < 2:
+ return [], 200
+ snaps = []
+ for row in arr[1:]:
+ try:
+ snaps.append({"timestamp": row[1], "url": row[2], "mimetype": row[3] if len(row) > 3 else "", "status": row[4] if len(row) > 4 else "", "digest": row[5] if len(row) > 5 else ""})
+ except Exception:
+ continue
+ return snaps, 200
+
+def _warp_snapinfo(domain):
+ snaps, st = _warp_cdx(domain)
+ if snaps is None:
+ return None, st
+ by_year, years = {}, []
+ for s in snaps:
+ y = s["timestamp"][:4]
+ if y not in by_year:
+ by_year[y] = []
+ years.append(y)
+ by_year[y].append(s)
+ timeline = [{"year": y, "count": len(by_year[y]), "first": by_year[y][0]["timestamp"], "last": by_year[y][-1]["timestamp"]} for y in years]
+ return {"domain": domain, "total": len(snaps), "timeline": timeline, "snapshots": snaps}, 200
+
+def _warp_titles(domain):
+ """Distinct archived titles/paths: fetch each collapsed snapshot's HTML, pull the title tag."""
+ info, st = _warp_snapinfo(domain)
+ if info is None:
+ return None, st
+ seen, out = set(), []
+ for s in info["snapshots"]:
+ key = (s["url"], s["digest"])
+ if key in seen:
+ continue
+ seen.add(key)
+ out.append(s)
+ if len(out) >= 8:
+ break
+ paths = []
+ for s in out:
+ rec = {"timestamp": s["timestamp"], "path": "/" + s["url"].split("/", 3)[-1] if s["url"].count("/") > 2 else "/",
+ "url": "https://web.archive.org/web/" + s["timestamp"] + "/" + s["url"], "title": ""}
+ paths.append(rec)
+ return {"domain": domain, "entries": paths, "scanned": len(out)}, 200
+
+def _warp_dns(domain):
+ """Current DNS via DoH (A + MX), oldest-archived context left to RDAP side."""
+ rec = {"a": [], "mx": []}
+ st, txt = http("https://cloudflare-dns.com/dns-query?name=" + _wurl.quote(domain) + "&type=A", headers={"Accept": "application/dns-json"})
+ d = jf(txt)
+ if d:
+ rec["a"] = sorted({a.get("data", "") for a in d.get("Answer", []) if a.get("type") == 1})
+ st, txt = http("https://cloudflare-dns.com/dns-query?name=" + _wurl.quote(domain) + "&type=MX", headers={"Accept": "application/dns-json"})
+ d = jf(txt)
+ if d:
+ rec["mx"] = sorted({a.get("data", "") for a in d.get("Answer", []) if a.get("type") == 15})
+ return rec
+
+def _warp_rdap(domain):
+ st, txt = http("https://rdap.org/domain/" + _wurl.quote(domain))
+ d = jf(txt)
+ if not d or st != 200:
+ return None
+ ev = {e.get("eventAction"): (e.get("eventDate") or "")[:10] for e in d.get("events", [])}
+ ent = d.get("entities") or []
+ reg = ""
+ for e in ent:
+ if "registrar" in (e.get("roles") or []):
+ for v in (e.get("vcardArray") or [None, []])[1]:
+ if v[0] == "fn":
+ reg = v[3]
+ return {"registrar": reg, "created": ev.get("registration", ""), "changed": ev.get("last changed", ""), "expires": ev.get("expiration", ""),
+ "status": d.get("status", [])}
+
+def _warp_dns_history(domain):
+ """DNS/whois drift: current RDAP/DNS vs the oldest archived year (whois HTML hint)."""
+ cur_dns = _warp_dns(domain)
+ rdap = _warp_rdap(domain)
+ info, st = _warp_snapinfo(domain)
+ if info is None:
+ return None, st
+ oldest = info["snapshots"][0]["timestamp"] if info["snapshots"] else ""
+ old_dns = {"a": [], "mx": []}
+ if oldest:
+ st2, body = http("https://web.archive.org/cdx/search/cdx?url=" + _wurl.quote(domain) + "&output=json&limit=5&collapse=timestamp:4&from=" + oldest[:8] + "&filter=mimetype:text/dns")
+ arr = jf(body)
+ if isinstance(arr, list) and len(arr) > 1:
+ for row in arr[1:]:
+ try:
+ if row[3] == "text/dns":
+ parts = row[2].split("/")
+ old_dns["a"].append(row[2])
+ except Exception:
+ continue
+ changes = []
+ for field in ("a", "mx"):
+ nowv = "; ".join(cur_dns[field]) or "(none)"
+ thenv = "; ".join(old_dns[field]) or "(not archived)"
+ changes.append({"record": field.upper(), "oldest": thenv, "current": nowv,
+ "drift": "no data" if not old_dns[field] else ("same" if thenv == nowv else "CHANGED")})
+ if rdap:
+ changes.append({"record": "REGISTRAR", "oldest": "(unknown)", "current": rdap["registrar"] or "(n/a)", "drift": "current"})
+ changes.append({"record": "CREATED", "oldest": "(unknown)", "current": rdap["created"] or "(n/a)", "drift": "current"})
+ return {"domain": domain, "oldest_snapshot": oldest, "rdap": rdap, "dns_now": cur_dns, "changes": changes}, 200
+
+def _warp_lookup(domain):
+ res = {}
+ info, st1 = _warp_snapinfo(domain)
+ res["archive"] = info
+ if info:
+ _warp_put(domain, "archive", info)
+ dnsh, st2 = _warp_dns_history(domain)
+ res["dns_history"] = dnsh
+ if dnsh:
+ _warp_put(domain, "dns", dnsh)
+ titles, st3 = _warp_titles(domain)
+ res["hosted"] = titles
+ if titles:
+ _warp_put(domain, "hosted", titles)
+ return res, (st1 if info is None else 200)
+
+def _warp_page_body(domain, data, msg):
+ res_html = ""
+ if msg:
+ res_html = 'NOTE ' + esc(msg) + "
"
+ if data:
+ arch = data.get("archive") or {}
+ if arch.get("total"):
+ rows = ""
+ for t in arch.get("timeline", []):
+ ts = t["first"]
+ shot = "https://web.archive.org/web/" + ts + "if_/" + domain
+ live = "https://web.archive.org/web/" + ts + "/" + domain
+ rows += ("| " + esc(t["year"]) + " | " + str(t["count"]) + " | "
+ + "preview " + ts[:8] + " | "
+ + "open copy |
")
+ res_html += ('SNAPSHOT TIMELINE — ' + esc(domain) + ''
+ + '
' + str(arch.get("total", 0)) + ' snapshots collapsed to one per year.
'
+ + '
| Year | Snapshots | Screenshot preview | Archived copy |
|---|
' + rows + "
")
+ else:
+ res_html += 'NO ARCHIVE The Wayback Machine has no snapshots of ' + esc(domain) + ". Either it never hosted a site, or it was never crawled.
"
+ dns = data.get("dns_history") or {}
+ if dns:
+ crows = ""
+ for c in dns.get("changes", []):
+ tag = "ok" if c["drift"] in ("same", "current", "no data") else "bad"
+ crows += ("| " + esc(c["record"]) + " | " + esc(c["oldest"]) + " | " + esc(c["current"]) + " | "
+ + '' + esc(c["drift"]) + " |
")
+ ol = dns.get("oldest_snapshot", "")
+ res_html += ('DNS / WHOIS DRIFT'
+ + '
Oldest snapshot: ' + (esc(ol[:8]) if ol else "none") + " vs today.
"
+ + '
| Record | Oldest archived | Current | Drift |
|---|
' + crows + "
")
+ hosted = data.get("hosted") or {}
+ if hosted and hosted.get("entries"):
+ hrows = ""
+ for e in hosted["entries"]:
+ hrows += ('| ' + esc(e["timestamp"][:8]) + " | " + esc(e["path"]) + ' | view |
')
+ res_html += ('WHAT DID THIS DOMAIN HOST?'
+ + '
Distinct archived paths (' + str(hosted.get("scanned", 0)) + " sampled).
"
+ + '
| Date | Path | Link |
|---|
' + hrows + "
")
+ return res_html
+
+WARP_EXPLAIN = how([
+ "The Wayback Machine is a public crawl archive — its CDX index API lists every snapshot it holds for a domain, free, no key.",
+ "We ask for the index collapsed to one snapshot per year (collapse=timestamp:4) so you get a clean timeline instead of 10,000 rows.",
+ "Screenshot previews use the 'if_' replay modifier: web.archive.org/web/if_/ strips the Wayback toolbar and serves the page as captured.",
+ "DNS drift: we pull today's A/MX records over DNS-over-HTTPS and today's RDAP registration, then line them up against the oldest archived year.",
+ "Everything is cached in a local sqlite table for 6 hours — repeat lookups are instant and do not hammer the archive.",
+ "A domain with snapshots every year and stable records is usually legit. A 2-year gap plus a registrar change plus new MX is a takeover tell.",
+ "Agents: GET /api/warp?domain=example.com returns the whole picture as JSON — timeline, drift table, hosted paths, ready for a report.",
+])
+
+WARP_FLOW = flow("watch a scam site morph over 3 years", [
+ "you get a phishing email from secure-login-example.com and want to know if it is a fresh drop or an old hijack.",
+ "Punch the domain into WARP. The timeline shows snapshots in 2019 and 2020 — then nothing until last month.",
+ "Click the 2019 screenshot preview: the archived copy shows a quiet small-business bakery homepage.",
+ "The drift table tells the rest: registrar changed this year, MX moved to a bulk-mail provider, A record now points at bulletproof hosting.",
+ "The hosted paths list confirms the morph: /menu.pdf in 2019, /wp-login.php and /secure/verify today.",
+ "you report it as a compromised/repurposed domain with archive receipts — far more credible than 'it looks phishy'.",
+])
+
+@app.route("/warp", methods=["GET", "POST"])
+def warp_page():
+ uid = current_user_id()
+ domain_raw = param("domain") or ""
+ data, msg = None, ""
+ if domain_raw:
+ domain = _warp_norm(domain_raw)
+ if not domain:
+ msg = "that does not look like a domain — try example.com"
+ else:
+ cached = _warp_get(domain, "archive")
+ if cached:
+ dns_c = _warp_get(domain, "dns")
+ host_c = _warp_get(domain, "hosted")
+ data = {"archive": cached, "dns_history": dns_c, "hosted": host_c}
+ msg = "cached result (fresh within 6h)"
+ else:
+ r = rate_limit("warp", 6, 60)
+ if r:
+ return r
+ data, st = _warp_lookup(domain)
+ if data.get("archive") is None:
+ data = None
+ msg = "archive lookup failed (status " + str(st) + ") — the Wayback CDX API may be slow or unreachable"
+ res_html = _warp_page_body(domain_raw, data, msg)
+ body = f"""
+WARP ARCHIVE
A domain time machine. Line up the Wayback Machine's snapshots of any domain year by year, preview what it used to look like, and see whether its DNS and registrar story drifted. Passive — one public API, zero packets to the target.
+Look up a domain
+
+
Snapshots per year · screenshot previews · DNS/whois drift · hosted-path inventory. Results cached 6h.
+{res_html}
+{WARP_FLOW}
+{WARP_EXPLAIN}
+""" + gloss([("Wayback Machine", "web.archive.org — the Internet Archive's crawl of the web since 1996. Public, free, no key."),
+ ("CDX API", "the index in front of the archive: query it for every snapshot (timestamp, URL, mimetype, hash) it holds for a site."),
+ ("collapse=timestamp:4", "dedupe the index to one hit per year (first 4 digits of the timestamp) — the clean timeline trick."),
+ ("if_ modifier", "replay URL flag that serves the raw archived page with the Wayback toolbar injected CSS/JS removed — good for previews."),
+ ("RDAP", "the modern WHOIS: registration dates, registrar, status over HTTPS/JSON."),
+ ("DoH", "DNS over HTTPS — lets us resolve A/MX records from the app without a resolver."),
+ ("takeover", "an aged domain that changed hands: registrar event + new MX + new hosting is the classic signature."),
+]) + agent_card("GET /api/warp?domain=example.com",
+ "curl -s \"" + SITE + "/api/warp?domain=example.com\" -H \"Authorization: Bearer dk_...\"",
+ "Free. Returns timeline, drift table, hosted paths. Cached 6h per domain.")
+ return page("intel", body)
+
+@app.route("/api/warp")
+def api_warp():
+ r = rate_limit("warp", 10, 60)
+ if r:
+ return r
+ domain = _warp_norm(param("domain") or "")
+ if not domain:
+ return jsonify({"ok": False, "error": "domain required, e.g. ?domain=example.com"}), 400
+ cached = _warp_get(domain, "archive")
+ if cached:
+ ok = bool(cached.get("total"))
+ out = {"ok": ok, "cached": True, "domain": domain, "archive": cached,
+ "dns_history": _warp_get(domain, "dns"), "hosted": _warp_get(domain, "hosted")}
+ if not ok:
+ out["error"] = "no snapshots found for this domain (cached)"
+ return jsonify(out), (200 if ok else 404)
+ data, st = _warp_lookup(domain)
+ out = {"ok": bool(data.get("archive") and data["archive"].get("total")), "cached": False, "domain": domain}
+ out.update(data)
+ if not out["ok"]:
+ out["error"] = "no snapshots found for this domain (status " + str(st) + ")"
+ return jsonify(out), (200 if out["ok"] else 404)
+# ---------- END TOOL: WARP ARCHIVE ----------
+
+
# ---------- 9. OPERATOR CONSOLE ----------
@app.route("/admin", methods=["GET", "POST"])
def admin():
@@ -3309,6 +5315,14 @@ def index():
("shot","SCREENSHOT","Headless-Chromium PNG capture of any page. Agents: poll the status API.","◈","HUNT"),
("score","FRAUD-SCORE","Composite 0-100 risk: IP intel + disposable-email + BIN heuristics, with full breakdown.","◈","HUNT"),
("tools","FREE TOOLS","DNS resolver, HTTP header inspector, JWT decoder, hasher, generators.","◈","UTILITY"),
+ ("beacon","PORT BEACON","Heartbeat your servers home on a timer; if the box goes dark or the probe URL gets touched, you know in seconds.","◆","Hunt"),
+ ("bssid","BSSID RADAR","Turn a WiFi router's MAC into an approximate place on Earth, with map links and accuracy radius.","◈","Hunt"),
+ ("hooks", "HOOK RELAY", "Instant public webhook inspector: capture every callback, auto-detect the sender, replay it anywhere.", "◈", "Operate"),
+ ("face","FACE TRACE","Hash an avatar, harvest a username's profile pictures across platforms, and get Hamming verdicts — no reverse-image APIs.","◈","Hunt"),
+ ("rotator","ROTATOR","Spin consistent browser identities from four pools with replayable seeds.","◈","UTILITY"),
+ ("shelf","IDENTITY SHELF","Every burner you own on one page — mailboxes, numbers, drops, traps — with live countdowns so nothing dies silently.","◈","ACCOUNT"),
+ ("unfurl", "UNFURL", "Follow every redirect hop in a URL chain and dissect the page at the end.", "◈", "Hunt"),
+ ("warp","WARP ARCHIVE","A domain time machine: snapshot timeline from the Wayback Machine, archived-page previews, and DNS/whois drift — watch a domain morph over years.","◈","INTEL"),
("passport","AGENT PASSPORT","Machine-readable trust badge for your bots. Agents are first-class here.","◈","ACCOUNT"),
]
tcards = "".join(