"
def page(sec, body):
n1, n2 = NEBULAS.get(sec, ("rgba(120,85,255,.16)", "rgba(0,190,255,.10)"))
uid = current_user_id()
acct = ""
if uid:
try:
con = db()
u = con.execute("SELECT username FROM users WHERE id=?", (uid,)).fetchone()
bal = get_balance(uid)
pas = has_pass(uid)
acct = ('' +
("★ PASS · " if pas else "") + "$" + f"{bal/100:.2f}" + " · " + esc(u["username"]) + "")
except Exception:
acct = ""
from markupsafe import Markup
return render_template_string(BASE, body=Markup(body), bmac=BMAC, o=lambda s: "on" if s == sec else "",
n1=n1, n2=n2, acct=acct)
def _checkout_or_json(payload):
"""If a browser form POSTed (no JSON accept / no X-Requested-With), redirect to
the BTCPay checkout page instead of showing raw JSON."""
wants_html = "text/html" in (request.headers.get("Accept") or "") and "application/json" not in (request.headers.get("Accept") or "")
link = payload.get("checkoutLink") if isinstance(payload, dict) else None
if wants_html and link:
return Redirect(link)
return jsonify(payload)
def Redirect(u):
from flask import redirect as _r
return _r(u)
def gloss(terms):
chips = " ".join('' + esc(t) + '' for t, d in terms)
return '
JARGON — hover any term: ' + chips + '
'
def how(steps):
lis = "".join(f"
{esc(s)}
" for s in steps)
return f'
HOW IT WORKS{lis}
'
# ---------- AGENT DISCOVERY ----------
API_INDEX = {
"service": "dark0rbits",
"description": "IP intel, card BIN validation, 30-min SMS rentals, residential proxy lab, steganography, trackable files, no-KYC messaging, utilities.",
"endpoints": [
{"method": "GET", "path": "/api/ip?target=", "desc": "Caller IP intel (auto) or any IP you pass: geo, ASN, ISP, VPN/hosting flags, rDNS."},
{"method": "POST", "path": "/api/card", "params": {"num": "card number"}, "desc": "Luhn + BIN intel. Nothing stored/charged."},
{"method": "POST", "path": "/api/sms/rent", "params": {"service": "id/keyword", "country": "id"}, "desc": "Rent disposable number, 30 min, refundable."},
{"method": "GET", "path": "/api/sms/check?pid=", "desc": "Poll SMS code."},
{"method": "GET", "path": "/api/sms/cancel?pid=", "desc": "Cancel + refund."},
{"method": "GET", "path": "/api/sms/history", "desc": "Rental history."},
{"method": "POST", "path": "/api/proxy/test", "params": {"user": "Pleiades user", "pass": "password"}, "desc": "Tunnel CONNECT via Pleiades gateway, return egress IP/geo."},
{"method": "POST", "path": "/api/steg/hide", "params": {"image": "png file", "text": "secret", "password": "optional", "bits": "1-3", "spread": "sequential|random"}, "desc": "LSB steganography → PNG download."},
{"method": "POST", "path": "/api/steg/extract", "params": {"image": "png file", "password": "optional"}, "desc": "Extract hidden text."},
{"method": "POST", "path": "/api/track/create", "params": {"filename": "name"}, "desc": "Create $1 BTCPay invoice for a trackable file. Returns checkoutLink."},
{"method": "GET", "path": "/api/track/events?token=", "desc": "Open events for a trackable (auth via account)."},
{"method": "GET", "path": "/api/hash?s=", "desc": "md5/sha1/sha256/sha512."},
{"method": "GET", "path": "/api/hdr?url=", "desc": "Fetch URL, return status + headers."},
{"method": "POST", "path": "/api/deaddrop/create", "params": {"body": "note text (max 8000 chars)", "burn_after_reads": "1-10", "ttl_hours": "1-72", "password": "optional"}, "desc": "AES-GCM encrypted burn-after-read note. Returns /drop/ URL. Free with PASS, else 5c from balance. Reads decrement; note self-destructs at 0 or at TTL."},
{"method": "GET", "path": "/drop/", "desc": "Read a dead-drop (password-protected if set). Each view burns one read."},
{"method": "POST", "path": "/api/shot/create", "params": {"url": "http(s):// target"}, "desc": "Screenshot queue. Headless Chromium PNG if available, else rendered-text capture (status=text_fallback). 25c/shot, free with PASS. Poll /api/shot/status/."},
{"method": "GET", "path": "/api/shot/status/", "desc": "Shot result: base64 PNG (png_b64) or text preview + page intel."},
{"method": "GET", "path": "/api/score?ip=&email=&bin=", "desc": "Composite fraud score 0-100 + weighted breakdown: IP intel (VPN/hosting/abuse geo), disposable-email domain, BIN country/type risk. 2c/call, free with PASS."},
],
"payment": "BTCPay BTC only (no Stripe). SMS meters to house account; trackables $1 each.",
}
@app.route("/api")
def api_index(): return jsonify(API_INDEX)
@app.route("/robots.txt")
def robots(): return "User-agent: *\nAllow: /\nSitemap: https://dark0rbits.thetempleofdoom.com/sitemap.xml\n", 200, {"Content-Type": "text/plain"}
@app.route("/a8f3dark0rbitskey.txt")
def indexnow_key(): return "a8f3d4rk0rbits9e2b1c7x5k8m2v4q6t8", 200, {"Content-Type": "text/plain"}
INDEXNOW = "a8f3d4rk0rbits9e2b1c7x5k8m2v4q6t8"
@app.route("/sitemap.xml")
def sitemap():
S = "https://dark0rbits.thetempleofdoom.com"
pages = ["", "ip", "card", "sms", "proxy", "steg", "track", "eh", "forensics", "canary", "deaddrop", "shot", "score", "mail", "inbox", "passport", "pass", "keys", "tools"]
xml = '' + "".join(f"{S}/{p}weekly" for p in pages) + ""
return xml, 200, {"Content-Type": "application/xml"}
@app.route("/llms.txt")
def llms():
eps = "\n".join(f"- `{e['method']} {e['path']}` — {e['desc']}" for e in API_INDEX["endpoints"])
return f"# Dark0rbits\n\nBase: {SITE}\n\n## API\n{eps}\n", 200, {"Content-Type": "text/plain"}
@app.route("/ai-plugin.json")
def aiplugin():
return jsonify({"name_for_model": "dark0rbits", "schema_version": "v1",
"description_for_model": "IP intelligence, card BIN validation, SMS number rentals, proxy egress testing, LSB steganography, trackable file links with open-notifications, no-KYC site messaging.",
"api": {"type": "openapi", "url": SITE + "/openapi.json"}, "auth": {"type": "none"}, "contact_email": "indianaholmes1@icloud.com"})
@app.route("/openapi.json")
def openapi():
ps = {"openapi": "3.0.0", "info": {"title": "DARK0RBITS", "version": "2.0.0"}, "paths": {}}
def add(path, method, desc, params=None, req=False, files=None):
item = {"summary": desc}
if files:
item["requestBody"] = {"content": {"multipart/form-data": {"schema": {"type": "object", "properties": {**{k: {"type": "string"} for k, v in (params or {}).items()}, **{f: {"type": "string", "format": "binary"} for f in files}}}}}}
elif params:
if method == "get":
item["parameters"] = [{"name": k, "in": "query", "required": req, "schema": {"type": "string"}} for k in params]
else:
item["requestBody"] = {"content": {"application/x-www-form-urlencoded": {"schema": {"type": "object", "properties": {k: {"type": "string"} for k in params}}}}}
ps["paths"][path] = ps["paths"].get(path, {}) | {method: {"responses": {"200": {"description": "ok"}}, **item}}
add("/api/ip", "get", "IP intel (caller or ?target=)", {"target": "optional IP"})
add("/api/card", "post", "Luhn + BIN validation", {"num": "card number"}, req=True)
add("/api/sms/rent", "post", "Rent number 30 min", {"service": "id", "country": "id"}, req=True)
add("/api/sms/check", "get", "Poll SMS code", {"pid": "orderid"}, req=True)
add("/api/sms/cancel", "get", "Cancel + refund", {"pid": "orderid"}, req=True)
add("/api/sms/history", "get", "Rental history")
add("/api/proxy/test", "post", "Test Pleiades gateway creds", {"user": "user", "pass": "pass"}, req=True)
add("/api/steg/hide", "post", "LSB-hide text in PNG", {"text": "secret", "password": "opt"}, req=True, files=["image"])
add("/api/steg/extract", "post", "Extract text from PNG", {"password": "opt"}, files=["image"])
add("/api/track/create", "post", "Create $1 invoice for trackable", {"filename": "name"}, req=True)
add("/api/track/events", "get", "Trackable open events", {"token": "token"}, req=True)
add("/api/hash", "get", "Hashes", {"s": "string"}, req=True)
add("/api/hdr", "get", "HTTP headers", {"url": "url"}, req=True)
add("/api/deaddrop/create", "post", "Encrypted burn-after-read note", {"body": "text", "burn_after_reads": "1-10", "ttl_hours": "1-72", "password": "optional"}, req=True)
add("/drop/{token}", "get", "Read a dead-drop (burns one read)")
add("/api/shot/create", "post", "Queue page capture", {"url": "target url"}, req=True)
add("/api/shot/status/{id}", "get", "Shot result (png_b64 or text_fallback)")
add("/api/score", "get", "Composite fraud score 0-100", {"ip": "opt", "email": "opt", "bin": "opt"})
return jsonify(ps)
# ---------- 1. IP INTEL (auto + manual target) ----------
def ip_report(ip):
st, b = http(f"http://ip-api.com/json/{ip}?fields=66846719")
d = jf(b) or {}
try: d["reverse"] = d.get("reverse") or socket.gethostbyaddr(ip)[0]
except Exception: pass
return d
@app.route("/ip", methods=["GET", "POST"])
def ip_page():
target = param("target") if request.method == "POST" else param("target")
if target and target.strip():
target = target.strip()
d = ip_report(target)
heading = f"INTEL FOR {esc(target)}"
mine = False
else:
ip = request.headers.get("X-Real-IP") or request.remote_addr or ""
d = ip_report(ip)
heading = "WHAT'S MY IP"
mine = True
if d.get("status") == "fail" or not d:
body = f"
{heading}
lookup failed
{ip_form()}"
return page("ip", body)
rows = [
("IP", f"{esc(d.get('query'))}"),
("Country", f"{esc(d.get('country'))} ({esc(d.get('countryCode'))})"),
("Region / City", f"{esc(d.get('regionName'))} / {esc(d.get('city'))} {esc(d.get('zip'))}"),
("Lat, Lon", f"{d.get('lat')}, {d.get('lon')} · TZ {esc(d.get('timezone'))}"),
("ISP", esc(d.get("isp"))), ("Organization", esc(d.get("org"))), ("AS", esc(d.get("as") or d.get("asname"))),
("Reverse DNS", esc(d.get("reverse") or "—")),
("Flags", f"mobile: {d.get('mobile')} · proxy/VPN: {d.get('proxy')} · hosting: {d.get('hosting')}"),
("Currency", esc(d.get("currency"))),
]
extra = ""
if mine:
hdrs = {k: v for k, v in request.headers.items() if k.lower() in ("user-agent","accept-language","x-forwarded-for","cf-connecting-ip","cf-ipcountry")}
extra = '
Headers you sent
' + "".join(f"
{esc(k)}
{esc(v)}
" for k, v in hdrs.items()) + "
"
body = f"""
{heading}
Auto-detects your IP and shows everything. Want intel on another IP? Type it below — full report, any target.
{kv(rows)}
{extra}
API: GET /api/ip (caller) · GET /api/ip?target=1.2.3.4 (any target)
""" + how(["Your IP is auto-detected the moment the page loads — no input needed.","Type any other IP or hostname into the field for the same full report.","Everything is one GET away for agents: /api/ip and /api/ip?target=.","VPN/proxy/hosting flags come from IP-quality heuristics — if it says proxy, you are looking at a relay."])
body += gloss([("ASN","Autonomous System Number — the network operator that owns this route"),("rDNS","reverse DNS — hostname pointer for an IP"),("hosting","datacenter/cloud IP, not a home connection"),("VPN/proxy","known tunnel or relay range")])
return page("ip", body)
def ip_form():
return ''
@app.route("/api/ip")
def api_ip():
r = rate_limit("iptarget", 40, 60)
if r: return r
target = param("target")
if target and target.strip():
return jsonify(ip_report(target.strip()))
ip = request.headers.get("X-Real-IP") or request.remote_addr or ""
d = ip_report(ip)
d["headers_seen"] = dict(request.headers)
return jsonify(d)
# ---------- 2. CARD CHECK ----------
def luhn_ok(num):
digits = [int(c) for c in num]
s = sum(digits[-1::-2])
for d in digits[-2::-2]:
d *= 2
if d > 9: d -= 9
s += d
return s % 10 == 0
BRANDS = [("4","Visa"),("51","Mastercard"),("52","Mastercard"),("53","Mastercard"),("54","Mastercard"),("55","Mastercard"),
("22","Mastercard"),("23","Mastercard"),("24","Mastercard"),("25","Mastercard"),("26","Mastercard"),("27","Mastercard"),
("34","Amex"),("37","Amex"),("6011","Discover"),("65","Discover"),("644","Discover"),("645","Discover"),("646","Discover"),("647","Discover"),("648","Discover"),("649","Discover"),
("50","Maestro"),("56","Maestro"),("57","Maestro"),("58","Maestro"),("63","Maestro"),("67","Maestro"),
("30","Diners"),("36","Diners"),("38","Diners"),("39","Diners"),
("35","JCB"),("62","UnionPay"),("7","Mir")]
def brand_of(num):
for pfx, b in BRANDS:
if num.startswith(pfx): return b
return "Unknown"
def bin_lookup(bin8):
st, b = http(f"https://lookup.binlist.net/{bin8}", headers={"Accept-Version": "3"})
bl = jf(b) or {}
if not bl.get("bank") and not bl.get("type") and not bl.get("scheme"):
st, b = http(f"https://data.handyapi.com/bin/{bin8}")
h = jf(b) or {}
if h.get("Status") == "SUCCESS":
return {"bank": {"name": h.get("Issuer")}, "country": {"name": (h.get("Country") or {}).get("Name") if isinstance(h.get("Country"), dict) else h.get("Country")},
"type": str(h.get("Type", "")).lower() or None, "prepaid": "prepaid" in str(h.get("Type","")).lower() or None, "scheme": h.get("Scheme")}
return bl
@app.route("/card", methods=["GET", "POST"])
def card():
result = ""
num = re.sub(r"\D", "", param("num") or "")[:19]
if num:
ok = luhn_ok(num)
tags = ['LUHN VALID' if ok else 'LUHN INVALID — fake/dead number']
brand = brand_of(num)
bl = bin_lookup(num[:8])
bank = (bl.get("bank") or {}).get("name", "—")
country = (bl.get("country") or {}).get("name", "—")
ctype = bl.get("type", "—")
prepaid = bl.get("prepaid", "—")
flags = []
if ctype == "prepaid" or prepaid is True: flags.append("PREPAID — commonly flagged by merchants")
rng = {"Visa":(13,16,19),"Mastercard":(16,),"Amex":(15,)}.get(brand,(13,15,16,19))
tags.append(f'length {len(num)} valid for {brand}' if len(num) in rng else f'LENGTH {len(num)} WRONG for {brand}')
result = f"""
{kv([("Brand",brand),("BIN",num[:8]),("Bank / Issuer",esc(bank)),("Country",esc(country)),("Type",str(ctype)),("Prepaid",str(prepaid))])}
Nothing stored. No charge, no auth — BIN + math validation only. Fraud "flagged" status lives at the issuer.
""" + how(["Paste the card number — it never leaves the request, nothing is stored.","Luhn checksum validates the digit structure instantly.","BIN (first 8 digits) reveals the issuer bank, brand, card type and country.","Prepaid BINs get flagged — merchants commonly reject them.","This CANNOT show balance or fraud-hold status; only the issuer knows that."])
body = f"""
{result}"""
body += gloss([("BIN","first 6-8 digits of a card — identifies issuer, country, brand"),("Luhn","checksum test every real card number passes"),("prepaid","issued as prepaid — elevated fraud risk")])
return page("card", body)
@app.route("/api/card", methods=["POST"])
def api_card():
r = rate_limit("card", 30, 60)
if r: return r
num = re.sub(r"\D", "", param("num") or "")[:19]
if not num: return jsonify({"ok": False, "error": "num required"})
ok = luhn_ok(num)
bl = bin_lookup(num[:8])
return jsonify({"ok": True, "luhn": ok, "brand": brand_of(num), "length_ok": len(num) in
{"Visa":(13,16,19),"Mastercard":(16,),"Amex":(15,)}.get(brand_of(num),(13,15,16,19)),
"bin": {"issuer": (bl.get("bank") or {}).get("name"), "country": (bl.get("country") or {}).get("name"),
"type": bl.get("type"), "prepaid": bl.get("prepaid")},
"flags": (["prepaid-risk"] if (bl.get("type")=="prepaid" or bl.get("prepaid") is True) else []) + (["luhn-invalid"] if not ok else [])})
# ---------- 7i. SCREENSHOT SERVICE (chromium if present, else rendered-text fallback) ----------
def shot_url_ok(u):
if not re.match(r"^https?://", u): return None, "url must start with http:// or https://"
try:
host = urllib.parse.urlsplit(u).hostname or ""
except Exception:
return None, "url parse error"
if not host: return None, "url has no host"
try:
candidate = ipaddress.ip_address(host)
except ValueError:
candidate = None
if candidate:
if candidate.is_private or candidate.is_loopback or candidate.is_link_local or candidate.is_reserved: return None, "private/reserved IPs blocked"
return u, None
try:
resolved = ipaddress.ip_address(socket.gethostbyname(host))
except Exception:
return u, None # cannot resolve here — let the fetcher report the failure
if resolved.is_private or resolved.is_loopback or resolved.is_link_local or resolved.is_reserved: return None, "private/reserved IPs blocked"
return u, None
def shot_find_browser():
for b in ("chromium", "chromium-browser", "google-chrome", "google-chrome-stable"):
if shutil.which(b): return b
return None
def _shot_html_harvest(url):
"""HTTP fetch + readability-ish text harvest + page intel. No browser, no fake PNG."""
status, html_text = http(url, timeout=15)
out = {"http_status": status}
try:
title = re.search(r"]*>(.*?)", html_text, re.I | re.S)
if title: out["title"] = html.unescape(title.group(1)).strip()[:300]
desc = re.search(r']+name=["\']description["\'][^>]+content=["\'](.*?)["\']', html_text, re.I | re.S)
if desc: out["description"] = html.unescape(desc.group(1)).strip()[:400]
except Exception:
pass
intel = []
for m in re.finditer(r"]*>(.*?)", html_text, re.I | re.S):
t = html.unescape(re.sub(r"<[^>]+>", "", m.group(2))).strip()
if t: intel.append("h" + m.group(1) + ": " + t[:120])
if len(intel) >= 15: break
t = re.sub(r"(?is)<(script|style|noscript|svg)[^>]*>.*?\1>", " ", html_text)
t = re.sub(r"(?s)", " ", t)
t = re.sub(r"(?i)<(br|/p|/div|/li|/h[1-6]|/tr)[^>]*>", "\n", t)
t = re.sub(r"<[^>]+>", " ", t)
t = html.unescape(t)
t = re.sub(r"[ \t\r]+", " ", t)
t = re.sub(r"\n\s*\n+", "\n", t).strip()
words = t.split()
out["text_preview"] = " ".join(words[:400])
out["text_chars_total"] = len(words)
out["headings"] = intel
return out
def shot_run(sid):
con = db()
s = con.execute("SELECT * FROM shots WHERE id=?", (sid,)).fetchone()
if not s: return
url = s["url"]
browser = shot_find_browser()
if browser:
out = os.path.join(UPLOAD_DIR, f"shot_{sid}.png")
try:
cmd = [browser, "--headless=new", "--no-sandbox", "--disable-gpu", "--hide-scrollbars",
"--window-size=1280,1600", f"--screenshot={out}", "--virtual-time-budget=8000", url]
p = subprocess.run(cmd, capture_output=True, timeout=45)
if p.returncode == 0 and os.path.exists(out) and os.path.getsize(out) > 0:
with open(out, "rb") as f: png = f.read()
os.remove(out)
con.execute("UPDATE shots SET status=?, result=? WHERE id=?", ("done", base64.b64encode(png).decode(), sid))
con.commit(); return
err = (p.stderr or b"").decode(errors="replace")[:200]
result = {"error": "chromium render failed: " + (err or f"exit {p.returncode}")}
except subprocess.TimeoutExpired:
result = {"error": "chromium timed out after 45s"}
except Exception as e:
result = {"error": f"chromium error: {e}"}
else:
try:
result = _shot_html_harvest(url)
result["mode"] = "text_fallback"
except Exception as e:
result = {"error": f"fetch failed: {e}"}
con.execute("UPDATE shots SET status=?, result=? WHERE id=?", ("text_fallback" if "mode" in result else "error", json.dumps(result), sid))
con.commit()
SHOT_API = ("
AGENT API
POST " + SITE + """/api/shot/create
Content-Type: application/json (or form fields)
{"url":"https://example.com"}
-> {"ok":true,"id":42,"status":"queued","poll":"BASE/api/shot/status/42"}
GET /api/shot/status/42
-> {"ok":true,"id":42,"status":"done","png_b64":"iVBORw..."} (chromium present)
-> {"ok":true,"status":"text_fallback","title":"...","text_preview":"...","headings":[...]}
auth: session cookie or Authorization: Bearer dk_...
25c/shot, free with PASS - rate limit 6/min
NOTE: no headless browser on this host yet - expect text_fallback
""").replace("BASE", SITE)
SHOT_EXPLAINER = """
HOW CAPTURE WORKS
• With headless Chromium installed, /shot returns a real browser render as base64 PNG.
• No browser on the host? You get text_fallback: an honest fetch of the page with rendered-text preview + page intel. A status field always tells you which.
• SSRF guard: private/reserved network targets are refused before any fetch.
• 25¢ per shot, free with PASS. Rate limit 6/min.
"""
@app.route("/shot")
def shot_page():
body = f"""
SCREEN SHOT
Point at a URL, get a render. Real headless-Chromium PNG when the host has one — an honest rendered-text + intel fallback when it doesn't. Never a fake image.
New capture
{'Free with your PASS — or 25¢ from balance.' if current_user_id() else 'Login + balance (or PASS): 25¢ per shot.'}
Result
{SHOT_EXPLAINER}
""" + SHOT_API + how(["Paste a URL — the job queues with a 25¢ charge (free with PASS).",
"With a headless browser on the host you get a real PNG back as base64.",
"No browser installed? You get text_fallback: title, description, headings, first 400 words — honestly labeled.",
"Agents: POST /api/shot/create then poll /api/shot/status/ until status != queued.",
"SSRF guard: localhost and private ranges are refused — this is a capture service, not a port scanner."])
return page("shot", body)
@app.route("/api/shot/create", methods=["POST"])
def api_shot_create():
r = rate_limit("shot", 6, 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 dk_ key"}), 401
url = str(jp("url") or "").strip()
if not url: return jsonify({"ok": False, "error": "url required"}), 400
url, err = shot_url_ok(url)
if err: return jsonify({"ok": False, "error": err}), 400
if not has_pass(uid) and not charge(uid, 25, "shot create"):
return jsonify({"ok": False, "error": "insufficient balance", "balance_cents": get_balance(uid), "topup": SITE + "/keys"}), 402
con = db()
cur = con.execute("INSERT INTO shots(user_id,url,status,created) VALUES(?,?,?,?)", (uid, url, "queued", int(time.time())))
con.commit()
shot_run(cur.lastrowid)
st = con.execute("SELECT status FROM shots WHERE id=?", (cur.lastrowid,)).fetchone()
return jsonify({"ok": True, "id": cur.lastrowid, "status": st["status"], "poll": f"{SITE}/api/shot/status/{cur.lastrowid}"}), 200, {"Cache-Control": "no-store"}
def shot_dict(row):
d = {"ok": True, "id": row["id"], "status": row["status"]}
try:
r = json.loads(row["result"]) if row["result"] else None
except Exception:
r = row["result"]
if row["status"] == "done" and r:
d["png_b64"] = r
try:
d["png_bytes"] = len(base64.b64decode(r))
except Exception:
pass
elif r:
d.update(r if isinstance(r, dict) else {"detail": str(r)[:400]})
return d
@app.route("/api/shot/status/")
def api_shot_status(sid):
con = db()
s = con.execute("SELECT * FROM shots WHERE id=?", (sid,)).fetchone()
if not s: return jsonify({"ok": False, "error": "unknown shot id"}), 404
if s["status"] == "queued": shot_run(sid) # lazy exec (reload-safe)
s = con.execute("SELECT * FROM shots WHERE id=?", (sid,)).fetchone()
return jsonify(shot_dict(s))
# ---------- 3. SMS RENTALS ----------
SMSP = "https://api.smspool.net"
SERVICES = [("google","Google"),("discord","Discord"),("telegram","Telegram"),("whatsapp","WhatsApp"),("other","Other/Any")]
COUNTRIES = [("1","United States"),("2","United Kingdom"),("4","Netherlands"),("22","Russia"),("150","Germany")]
def sms_api(path, **kw):
if kw:
kw["key"] = SMSP_KEY
return http(f"{SMSP}/{path}", data=urllib.parse.urlencode(kw).encode(), method="POST")
return http(f"{SMSP}/{path}?key={SMSP_KEY}")
def sms_guard():
con = db(); now = int(time.time())
uid = current_user_id()
st, b = sms_api("request/balance")
bal = jf(b) or {}
try: bal = float(bal.get("balance", 0))
except Exception: bal = 0
if bal < 5: return f"house balance too low (${bal:.2f}) — rentals paused"
act = con.execute("SELECT COUNT(*) c FROM sms_rentals WHERE status='active' AND expires > ?", (now,)).fetchone()["c"]
if act >= (5 if has_pass(uid) else 3): return "too many active rentals right now — try again later"
h = con.execute("SELECT COUNT(*) c FROM sms_rentals WHERE created > ?", (now-3600,)).fetchone()["c"]
if h >= (20 if has_pass(uid) else 6): return "hourly rental cap reached"
d = con.execute("SELECT COUNT(*) c FROM sms_rentals WHERE created > ?", (now-86400,)).fetchone()["c"]
if d >= (50 if has_pass(uid) else 15): return "daily rental cap reached"
return None
@app.route("/sms", methods=["GET", "POST"])
def sms():
uid = current_user_id()
msg = ""
if request.method == "POST":
act = request.form.get("act")
if act == "rent":
guard = sms_guard()
if guard:
msg = f'
PAUSED {guard}
'
else:
st, b = sms_api("purchase/sms", service=request.form["service"], country=request.form["country"])
d = jf(b) or {}
if d.get("success") == 1:
con = db(); now = int(time.time())
con.execute("INSERT INTO sms_rentals(user_id,phone,service,country,purchase_id,cost,status,created,expires) VALUES(?,?,?,?,?,?,?,?,?)",
(uid, d.get("number"), request.form["service"], request.form["country"], str(d.get("purchase_id")), d.get("cost",0), "active", now, now+1800))
con.commit()
msg = f'
RENTED Your number: +{d.get("number")} · 30 min · order #{d.get("purchase_id")}
'
else:
msg = f'
RENT FAILED
{esc(b[:400])}
'
elif act == "check":
st, b = sms_api("sms/check", orderid=request.form["pid"])
d = jf(b) or {}
sms_txt = d.get("sms") or d.get("code") or ""
status = d.get("status", "?")
msg = f'
STATUS: {status} {"" + esc(sms_txt) + "" if sms_txt else "no code yet — poll again in 10s"}
'
elif act == "cancel":
st, b = sms_api("sms/cancel", orderid=request.form["pid"])
d = jf(b) or {}
ok = d.get("success") == 1
con = db(); con.execute("UPDATE sms_rentals SET status=? WHERE purchase_id=?", ("refunded" if ok else "cancel-failed", request.form["pid"])); con.commit()
msg = f'
{"CANCELLED + REFUNDED" if ok else "CANCEL FAILED"}
'
con = db()
hist = con.execute("SELECT * FROM sms_rentals WHERE user_id=? ORDER BY id DESC LIMIT 8", (uid,)).fetchall()
hist_rows = "".join(f"
Disposable numbers, 30-minute windows. Cancel before a code = full refund.
Rent a number
Check / manage
{msg}
Recent rentals
Number
Service
Status
Order
Window
{hist_rows or '
none yet
'}
Live code
API: POST /api/sms/rent (service,country) · GET /api/sms/check?pid= · GET /api/sms/cancel?pid= · GET /api/sms/history
""" + how(["Pick a service and country, rent — the number is live for 30 minutes exactly.","Use it for any signup/verification. The code arrives as a text.","Poll the order (auto or manual) until the code shows.","Cancel before a code arrives and you get every satoshi back.","Each rental is logged in the recent-rentals table with a live countdown."])
body += gloss([("OTC","one-time code — the PIN a service texts you"),("burn","cancel an unused rental inside the refund window"),("SMSPool","our upstream number provider")])
return page("sms", body)
@app.route("/api/sms/rent", methods=["POST"])
def api_sms_rent():
guard = sms_guard()
if guard: return jsonify({"success": 0, "message": guard, "paused": True})
uid = key_user() or current_user_id()
if uid and not has_pass(uid) and get_balance(uid) < 50:
return jsonify({"ok": False, "error": "insufficient balance", "topup": SITE + "/keys"}), 402
st, b = sms_api("purchase/sms", service=param("service"), country=param("country"))
d = jf(b) or {}
if d.get("success") == 1:
con = db(); now = int(time.time())
con.execute("INSERT INTO sms_rentals(user_id,phone,service,country,purchase_id,cost,status,created,expires) VALUES(?,?,?,?,?,?,?,?,?)",
(uid, d.get("number"), param("service"), param("country"), str(d.get("purchase_id")), d.get("cost",0), "active", now, now+1800))
con.commit()
cost = int(d.get("cost_in_cents") or 5)
if uid and not has_pass(uid):
charge(uid, cost, f"sms rental +{d.get('number')}")
return jsonify(d)
@app.route("/api/sms/check", methods=["GET","POST"])
def api_sms_check():
st, b = sms_api("sms/check", orderid=param("pid"))
return jf(b) or jsonify({"error": b[:200]})
@app.route("/api/sms/cancel", methods=["GET","POST"])
def api_sms_cancel():
st, b = sms_api("sms/cancel", orderid=param("pid"))
d = jf(b) or {}
if d.get("success") == 1:
con = db(); con.execute("UPDATE sms_rentals SET status='refunded' WHERE purchase_id=?", (param("pid"),)); con.commit()
return d
@app.route("/api/sms/history")
def api_sms_history():
con = db(); now = int(time.time())
con.execute("UPDATE sms_rentals SET status='expired' WHERE status='active' AND expires < ?", (now,))
con.commit()
uid = key_user() or current_user_id()
if not uid:
return jsonify({"ok": False, "error": "login required (POST /inbox act=login)"}), 401
return jsonify([dict(r) for r in con.execute("SELECT * FROM sms_rentals WHERE user_id=? ORDER BY id DESC LIMIT 50", (uid,))])
# ---------- 4. PROXY LAB ----------
@app.route("/proxy", methods=["GET", "POST"])
def proxy():
result = ""
if request.method == "POST" and request.form.get("act") == "test":
user, pw = request.form.get("user",""), request.form.get("pass","")
pauth = base64.b64encode(f"{user}:{pw}".encode()).decode()
try:
s = socket.create_connection((PLEIADES_GW.split(":")[0], int(PLEIADES_GW.split(":")[1])), timeout=15)
s.sendall(f"CONNECT ip-api.com:80 HTTP/1.1\r\nHost: ip-api.com:80\r\nProxy-Authorization: Basic {pauth}\r\n\r\n".encode())
resp = s.recv(4096)
if b"200" in resp.split(b"\r\n")[0]:
s.sendall(b"GET /json/?fields=66846719 HTTP/1.1\r\nHost: ip-api.com\r\nConnection: close\r\n\r\n")
data = b""
while True:
c = s.recv(8192)
if not c: break
data += c
s.close()
j = jf(data.split(b"\r\n\r\n",1)[-1].decode("utf-8","replace")) or {}
con = db()
con.execute("INSERT INTO proxy_checks(user_key,egress_ip,geo,ok,ts) VALUES(?,?,?,?,?)", (user[:12], j.get("query","?"), f"{j.get('country')}/{j.get('city')}", 1, int(time.time())))
con.commit()
result = f'
'
else:
con = db()
con.execute("INSERT INTO proxy_checks(user_key,egress_ip,geo,ok,ts) VALUES(?,?,?,?,?)", (user[:12], "", "", 0, int(time.time())))
con.commit()
result = f'
AUTH/TUNNEL FAILED
{esc(resp[:200])}
'
except Exception as e:
result = f'
ERROR {esc(e)}
'
body = f"""
PROXY LAB
Test + rent residential proxies on the Pleiades rail — same gateway keys as everywhere.
API: POST /api/proxy/test (user, pass) → egress IP + geo JSON.
""" + how(["Enter your Pleiades gateway user:pass — the same credentials work across the fleet.","The lab tunnels a CONNECT request through the gateway and reports the true egress IP, geo and ISP.","Use the geo builder to steer the exit: region, country, city, sticky 30-min sessions.","Need bandwidth? Buy GB plans at the Pleiades storefront."])
body += gloss([("sticky session","same exit IP kept across requests"),("egress","the exit IP the rest of the internet sees"),("Pleiades","our proxy gateway network")])
return page("proxy", body)
@app.route("/api/proxy/test", methods=["POST"])
def api_proxy_test():
r = rate_limit("proxytest", 10, 60)
if r: return r
user, pw = param("user") or "", param("pass") or ""
pauth = base64.b64encode(f"{user}:{pw}".encode()).decode()
try:
s = socket.create_connection((PLEIADES_GW.split(":")[0], int(PLEIADES_GW.split(":")[1])), timeout=15)
s.sendall(f"CONNECT ip-api.com:80 HTTP/1.1\r\nHost: ip-api.com:80\r\nProxy-Authorization: Basic {pauth}\r\n\r\n".encode())
resp = s.recv(4096)
if b"200" not in resp.split(b"\r\n")[0]: return jsonify({"ok": False, "raw": resp[:120].decode("utf-8","replace")})
s.sendall(b"GET /json/?fields=66846719 HTTP/1.1\r\nHost: ip-api.com\r\nConnection: close\r\n\r\n")
data = b""
while True:
c = s.recv(8192)
if not c: break
data += c
s.close()
j = jf(data.split(b"\r\n\r\n",1)[-1].decode("utf-8","replace")) or {}
return jsonify({"ok": True, "egress": j})
except Exception as e:
return jsonify({"ok": False, "error": str(e)})
# ---------- 5. STEGO LAB ----------
def _keystream(password, n):
ks = b""; seed = password.encode()
while len(ks) < n:
seed = hashlib.sha256(seed).digest()
ks += seed
return ks[:n]
def steg_hide(img_bytes, text, password="", bits=1, spread="sequential"):
from PIL import Image
im = Image.open(io.BytesIO(img_bytes)).convert("RGBA")
px = im.load()
w, h = im.size
capacity = w * h * 3 * bits
payload = text.encode("utf-8")
phash = hashlib.sha256(password.encode()).digest()[:4] if password else b"\x00\x00\x00\x00"
header = b"AUR1" + struct.pack(">I", len(payload)) + phash
body = payload
if password:
body = bytes(a ^ b for a, b in zip(body, _keystream(password, len(body))))
data = header + body
if len(data) * 8 > capacity:
return None, f"too big: need {len(data)*8} bits, image holds {capacity}"
if spread == "random":
import random as _r
_r.seed(int.from_bytes(hashlib.sha256((password + "dark0rbits").encode()).digest()[:4], "big") if password else int.from_bytes(hashlib.sha256(b"dark0rbits-random-no-pass").digest()[:4], "big"))
order = list(range(w*h)); _r.shuffle(order)
else:
order = list(range(w*h))
bits_needed = len(data) * 8
idx = 0
mask = (1 << bits) - 1
for pos in order:
if idx >= bits_needed: break
x, y = pos % w, pos // w
r, g, b, a = px[x, y]
chs = [r, g, b]
for ch_i in range(3):
if idx >= bits_needed: break
chunk = 0
taken = 0
for k in range(bits):
if idx >= bits_needed: break
chunk = (chunk << 1) | ((data[idx >> 3] >> (7 - (idx & 7))) & 1)
idx += 1; taken += 1
if taken < bits: chunk <<= (bits - taken)
chs[ch_i] = (chs[ch_i] & ~mask) | chunk
px[x, y] = tuple(chs) + (a,)
# also stash settings in a tEXt chunk for reliable extraction hints
out = io.BytesIO()
im.save(out, "PNG", pnginfo=_pnginfo(bits, spread))
return out.getvalue(), {"bits": bits, "spread": spread}
def _pnginfo(bits, spread):
try:
from PIL.PngImagePlugin import PngInfo
info = PngInfo()
info.add_text("dark0rbits_meta", json.dumps({"bits": bits, "spread": spread, "v": 2}))
return info
except Exception:
return None
def steg_extract(img_bytes, password="", bits=None, spread=None):
from PIL import Image
im = Image.open(io.BytesIO(img_bytes))
meta = im.info.get("dark0rbits_meta") or im.info.get("auriga_meta")
if meta:
try:
m = json.loads(meta)
bits = int(m.get("bits", bits or 1)); spread = m.get("spread", spread or "sequential")
except Exception: pass
bits = bits or 1
im = im.convert("RGBA")
px = im.load()
w, h = im.size
mask = (1 << bits) - 1
# replicate the shuffle used at hide time
if spread == "random":
import random as _r
_r.seed(int.from_bytes(hashlib.sha256((password + "dark0rbits").encode()).digest()[:4], "big") if password else int.from_bytes(hashlib.sha256(b"dark0rbits-random-no-pass").digest()[:4], "big"))
order = list(range(w*h)); _r.shuffle(order)
else:
order = list(range(w*h))
raw = bytearray()
need = None
idx = 0
for pos in order:
if need is not None and idx >= need: break
x, y = pos % w, pos // w
r, g, b, a = px[x, y]
for ch in (r, g, b):
chunk = ch & mask
for k in range(bits-1, -1, -1):
if need is not None and idx >= need: break
bit = (chunk >> k) & 1
while len(raw) < (idx >> 3) + 1: raw.append(0)
if bit: raw[idx >> 3] |= (0x80 >> (idx & 7))
idx += 1
if need is not None and idx >= need: break
if need is None and idx >= 64:
if bytes(raw[:4]) != b"AUR1":
return None, f"no DARK0RBITS payload found with LSB depth {bits} (try other depth / randomized)"
ln = struct.unpack(">I", bytes(raw[4:8]))[0]
need = 64 + ln * 8
data = bytes(raw)
if len(data) < 12: return None, "payload too small"
if bytes(data[:4]) != b"AUR1":
return None, "no DARK0RBITS payload found (wrong password or settings?)"
if password and hashlib.sha256(password.encode()).digest()[:4] != data[8:12]:
return None, "wrong password"
ln = struct.unpack(">I", data[4:8])[0]
body = data[12:12+ln]
if password:
body = bytes(a ^ b for a, b in zip(body, _keystream(password, len(body))))
text = body.decode("utf-8", "replace")
return text, None
@app.route("/steg", methods=["GET"])
def steg():
body = f"""
STEGO LAB
Hide words inside pictures — LSB steganography with real settings. PNG in, PNG out, looks untouched.
Hide text
Extract text
API: POST /api/steg/hide (image, text, password?, bits 1-3, spread) → PNG · POST /api/steg/extract (image, password?, bits?, spread?) → JSON
""" + how(["Drop a PNG — your words are written into the least-significant bits of its pixels.","Depth 1 = invisible and robust; depth 2-3 fits more text but is easier to detect.","Spread=randomized scatters bits across the image instead of top-down.","A password encrypts the payload AND derives the scatter pattern — wrong password = noise.","Extract reads the embedded metadata automatically — just drop the file and the words come back."])
body += gloss([("LSB","least significant bit — pixel bits that carry hidden data"),("depth","how many bit planes carry the payload"),("spread","payload dispersed across the image to survive edits")])
return page("steg", body)
@app.route("/api/steg/hide", methods=["POST"])
def api_steg_hide():
r = rate_limit("steg", 20, 60)
if r: return r
f = request.files.get("image")
text = param("text") or ""
if not f or not text: return jsonify({"ok": False, "error": "image + text required"}), 400
bits = min(3, max(1, int(param("bits") or 1)))
spread = param("spread") or "sequential"
try:
out, meta = steg_hide(f.read(), text, param("password") or "", bits, spread)
except Exception as e:
return jsonify({"ok": False, "error": str(e)}), 400
if out is None: return jsonify({"ok": False, "error": meta}), 400
return send_file(io.BytesIO(out), mimetype="image/png", as_attachment=True, download_name="dark0rbits-hidden.png")
@app.route("/api/steg/extract", methods=["POST"])
def api_steg_extract():
r = rate_limit("steg", 20, 60)
if r: return r
f = request.files.get("image")
if not f: return jsonify({"ok": False, "error": "image required"}), 400
bits = param("bits")
bits = min(3, max(1, int(bits))) if bits else None
try:
text, err = steg_extract(f.read(), param("password") or "", bits, param("spread") or None)
except Exception as e:
return jsonify({"ok": False, "error": str(e)}), 400
if err: return jsonify({"ok": False, "error": err}), 200
return jsonify({"ok": True, "text": text})
# ---------- 6. TRACKABLE FILES ----------
@app.route("/track", methods=["GET"])
def track():
uid = current_user_id()
mine = ""
if uid:
con = db()
rows = con.execute("SELECT * FROM trackables WHERE user_id=? ORDER BY id DESC LIMIT 10", (uid,)).fetchall()
if rows:
trs = "".join(f"
Pay $1 BTC → upload a file or picture → get a tracked link + an email-ready version. Every open pings back into your INBOX.
1 · Pay $1
BTCPay BTC only. After payment the upload opens automatically.
{mine}
How it works: your file gets a secret link — every open is logged (time, IP, device) and lands in your inbox. You also get an HTML copy with an embedded tracking pixel: email THAT and every view fires too. Login (no KYC) to see events.
API: POST /api/track/create (filename) → invoice · POST /api/track/upload?token= (file) → link · GET /api/track/events?token=
""" + how(["Pay $1 in BTC — the invoice settles and unlocks the upload instantly.","Upload your file or picture: you get a secret tracked link plus an email-ready HTML copy.","Email the HTML copy or share the link — every open fires back.","Each open reports: exact time, real IP, city/country, ISP, timezone, VPN flag, device, language, referrer.","Alerts land in your INBOX the second it happens."])
return page("track", body)
def btc_invoice(amount="1.00"):
st, b = http(f"{BTCPAY}/stores/{BTCPAY_STORE}/invoices",
headers={"Authorization": "token " + BTCPAY_KEY, "Content-Type": "application/json"},
data=json.dumps({"amount": amount, "currency": "USD", "metadata": {"orderId": "dark0rbits-track"}}).encode(), method="POST")
return jf(b) or {}
@app.route("/api/track/create", methods=["POST"])
def api_track_create():
fn = param("filename") or "file"
uid = key_user() or current_user_id()
if not uid:
return jsonify({"ok": False, "error": "login required — create a no-KYC account at /inbox (POST /inbox act=register), then retry"}), 401
token = secrets.token_urlsafe(16)
con = db()
if has_pass(uid):
con.execute("INSERT INTO trackables(user_id,token,filename,kind,invoice_id,paid,created) VALUES(?,?,?,?,?,1,?)",
(uid or 0, token, esc(fn[:100]), "file", "PASS", int(time.time())))
con.commit()
return jsonify({"ok": True, "free": True, "token": token, "upload_url": f"{SITE}/track/pay?token={token}"})
if uid and charge(uid, 100, f"trackable file ({fn[:40]})"):
con.execute("INSERT INTO trackables(user_id,token,filename,kind,invoice_id,paid,created) VALUES(?,?,?,?,?,1,?)",
(uid or 0, token, esc(fn[:100]), "file", "BALANCE", int(time.time())))
con.commit()
return jsonify({"ok": True, "balance_charged": 1.00, "token": token, "upload_url": f"{SITE}/track/pay?token={token}"})
inv = btc_invoice()
if not inv.get("id"): return jsonify({"ok": False, "error": str(inv)[:200]}), 400
con.execute("INSERT INTO trackables(user_id,token,filename,kind,invoice_id,paid,created) VALUES(?,?,?,?,?,0,?)",
(uid or 0, token, esc(fn[:100]), "file", inv["id"], int(time.time())))
con.commit()
return _checkout_or_json({"ok": True, "invoice_id": inv["id"], "checkoutLink": inv.get("checkoutLink"), "token": token,
"after_payment_upload_url": f"{SITE}/track/pay?token={token}"})
@app.route("/track/pay", methods=["GET"])
def track_pay():
token = param("token") or ""
con = db()
t = con.execute("SELECT * FROM trackables WHERE token=?", (token,)).fetchone()
if not t: return page("track", "
TRACK FILE
unknown token
")
return page("track", f"""
TRACK FILE
Upload your file — then it's trackable.
""")
@app.route("/api/track/upload", methods=["POST"])
def api_track_upload():
token = param("token")
f = request.files.get("file")
if not f: return jsonify({"ok": False, "error": "file required"}), 400
con = db()
t = con.execute("SELECT * FROM trackables WHERE token=?", (token,)).fetchone()
if not t: return jsonify({"ok": False, "error": "unknown token"}), 400
st, b = http(f"{BTCPAY}/stores/{BTCPAY_STORE}/invoices/{t['invoice_id']}", headers={"Authorization": "token " + BTCPAY_KEY}) if t["invoice_id"] not in ("PASS", "BALANCE") else (200, '{"status":"settled"}')
inv = jf(b) or {}
paid = inv.get("status") in ("settled", "processing", "paid")
if not paid: return jsonify({"ok": False, "error": f"invoice not paid yet ({inv.get('status')})"}), 402
data = f.read()
open(os.path.join(UPLOAD_DIR, token + ".bin"), "wb").write(data)
kind = "image" if (f.content_type or "").startswith("image") else "file"
fn = (f.filename or t["filename"])[:100]
con.execute("UPDATE trackables SET paid=1, kind=?, filename=? WHERE token=?", (kind, fn, token))
con.commit()
b64 = base64.b64encode(data).decode()
pixel = f"{SITE}/t/{token}.png"
if kind == "image":
viewer = f''
else:
viewer = f'
'
open(os.path.join(UPLOAD_DIR, token + ".html"), "w").write(viewer)
con.execute("INSERT INTO track_events(trackable_id,ts,ip,ua) VALUES(?,?,?,?)", (t["id"], int(time.time()), "created", "upload"))
con.commit()
return jsonify({"ok": True, "tracked_link": f"{SITE}/t/{token}", "pixel": pixel,
"email_html": f"{SITE}/t/{token}/html",
"note": "attach/email the HTML version — every view fires the pixel and lands in the inbox"})
def _geo_cache():
con = db()
con.execute("CREATE TABLE IF NOT EXISTS geo_cache(ip TEXT PRIMARY KEY, geo TEXT, ts INTEGER)")
return con
def enrich_ip(ip):
"""geo/ISP/ASN for an IP, cached 24h."""
if not ip or ip == "created" or ip.startswith(("10.30.20.", "127.", "172.17.")): return {}
con = _geo_cache()
r = con.execute("SELECT geo FROM geo_cache WHERE ip=? AND ts > ?", (ip, int(time.time())-86400)).fetchone()
if r: return json.loads(r["geo"])
st, b = http(f"http://ip-api.com/json/{ip}?fields=66846719")
d = jf(b) or {}
geo = {k: d.get(k) for k in ("country","countryCode","regionName","city","zip","lat","lon","timezone","isp","org","as","asname","mobile","proxy","hosting","reverse","query") if d.get(k) is not None}
con.execute("INSERT OR REPLACE INTO geo_cache(ip,geo,ts) VALUES(?,?,?)", (ip, json.dumps(geo), int(time.time())))
con.commit()
return geo
def _log_open(t, extra=""):
con = db()
ip = request.headers.get("X-Real-IP") or request.remote_addr or "?"
ua = request.headers.get("User-Agent","")
lang = request.headers.get("Accept-Language","")
ref = request.headers.get("Referer","")
geo = enrich_ip(ip)
where = ""
if geo: where = f" — {geo.get('city','')}, {geo.get('regionName','')} {geo.get('countryCode','')} · {geo.get('isp','')} · tz {geo.get('timezone','')}"
if geo.get("proxy"): where += " · VPN/proxy ⚠"
con.execute("INSERT INTO track_events(trackable_id,ts,ip,ua) VALUES(?,?,?,?)",
(t["id"], int(time.time()), ip + (" " + json.dumps(geo) if geo else ""), ua[:200] + (f" | lang={lang}" if lang else "") + (f" | ref={ref[:100]}" if ref else "")))
uid = t["user_id"]
if uid:
con.execute("INSERT INTO messages(user_id,sender,body,created) VALUES(?,?,?,?)",
(uid, "operator-bot", f"👁 '{esc(t['filename'])}' just opened{extra} — IP {esc(ip)}{esc(where)} device: {esc(ua[:100])}{' lang: ' + esc(lang) if lang else ''}{' from: ' + esc(ref[:120]) if ref else ''}", int(time.time())))
con.commit()
@app.route("/t/")
def tracked_download(token):
con = db()
t = con.execute("SELECT * FROM trackables WHERE token=?", (token,)).fetchone()
if not t or not t["paid"]: return "not found", 404
_log_open(t, " (link)")
path = os.path.join(UPLOAD_DIR, token + ".bin")
if not os.path.exists(path): return "file gone", 404
return send_file(path, as_attachment=True, download_name=t["filename"])
@app.route("/t/.png")
def tracked_pixel(token):
con = db()
t = con.execute("SELECT * FROM trackables WHERE token=?", (token,)).fetchone()
if t and t["paid"]:
_log_open(t, " (email/pixel)")
px = base64.b64decode("R0lGODlhAQABAIAAAAAAAP///yH5BAEAAAAALAAAAAABAAEAAAIBRAA7")
return Response(px, mimetype="image/gif", headers={"Cache-Control": "no-store"})
@app.route("/t//html")
def tracked_html(token):
con = db()
t = con.execute("SELECT * FROM trackables WHERE token=?", (token,)).fetchone()
if not t or not t["paid"]: return "not found", 404
p = os.path.join(UPLOAD_DIR, token + ".html")
return send_file(p, mimetype="text/html") if os.path.exists(p) else ("no html wrapper", 404)
@app.route("/api/track/events", methods=["GET"])
def api_track_events():
con = db(); token = param("token")
t = con.execute("SELECT * FROM trackables WHERE token=?", (token,)).fetchone()
if not t: return jsonify({"ok": False, "error": "unknown token"})
uid = current_user_id()
if not uid or uid != t["user_id"]: return jsonify({"ok": False, "error": "auth required (login on /inbox)"})
return jsonify([dict(r) for r in con.execute("SELECT * FROM track_events WHERE trackable_id=? ORDER BY id DESC LIMIT 100", (t["id"],))])
# ---------- 6b. BURNER MAIL (receive-only, BTC packages) ----------
MAIL_PACKS = [("7","7 days — $3",3,7),("30","30 days — $8",8,30),("90","90 days — $20",20,90)]
MAIL_DOMAIN = "thetempleofdoom.com"
MAIL_RESERVED = {"indianaholmes","admin","operator","drjones","root","noreply","support","pass","mail"}
MAIL_SECRET = "dark0rbits-mail-relay-2026"
@app.route("/mail", methods=["GET"])
def mail():
uid = current_user_id()
mine = ""
if uid:
con = db(); now = int(time.time())
con.execute("UPDATE mailboxes SET paid=2 WHERE paid=1 AND expires < ?", (now,)) # expired
rows = con.execute("SELECT * FROM mailboxes WHERE user_id=? ORDER BY id DESC LIMIT 10", (uid,)).fetchall()
if rows:
trs = "".join(f"
Receive-only disposable mailboxes @thetempleofdoom.com. Counting down in real time. Anything you sign up for — codes, confirmations, one-off handouts — lands right here, no other identity attached.
Pick a package (BTC)
{''.join(f'' for d,n,_,_ in MAIL_PACKS)}
Type your desired mailbox name, pick a length, pay the invoice — the mailbox activates the moment the payment settles.
{mine}
API: POST /api/mail/create (local, days) → invoice · GET /api/mail/inbox?addr= (needs login) — inbound via Cloudflare Email Routing → worker relay.
""" + how(["Pick a name and a package — 7, 30 or 90 days, BTC priced.","Pay the invoice; the mailbox activates the second it settles.","The address is receive-only: verification codes, confirmations, one-off handouts.","The countdown runs in real time; when it hits zero the mailbox retires itself.","All mail shows on the site inbox — nothing touches any other identity."])
return page("mail", body)
@app.route("/api/mail/create", methods=["POST"])
def api_mail_create():
uid = current_user_id()
local = re.sub(r"[^a-z0-9._-]", "", (param("local") or "").lower())[:30]
days = param("days") or "7"
pack = next((p for p in MAIL_PACKS if p[0] == str(days)), None)
if not pack: return jsonify({"ok": False, "error": "bad package"}), 400
if not local: return jsonify({"ok": False, "error": "mailbox name required"}), 400
if local in MAIL_RESERVED: return jsonify({"ok": False, "error": "reserved name"}), 400
addr = f"{local}@{MAIL_DOMAIN}"
con = db()
if con.execute("SELECT 1 FROM mailboxes WHERE address=?", (addr,)).fetchone():
return jsonify({"ok": False, "error": "mailbox name taken"}), 400
uid = key_user() or current_user_id()
# metered: PASS = instant free; balance = instant paid; else BTC invoice
if uid and has_pass(uid):
con.execute("INSERT INTO mailboxes(user_id,address,invoice_id,paid,expires,created,plan_days) VALUES(?,?,?,?,?,?,?)",
(uid, addr, "PASS", 1, int(time.time())+86400*pack[3], int(time.time()), pack[3]))
con.commit()
return jsonify({"ok": True, "free": True, "address": addr, "expires_in_days": pack[3]})
if uid and charge(uid, pack[2]*100, f"burner mailbox {addr} ({pack[3]}d)"):
con.execute("INSERT INTO mailboxes(user_id,address,invoice_id,paid,expires,created,plan_days) VALUES(?,?,?,?,?,?,?)",
(uid, addr, "BALANCE", 1, int(time.time())+86400*pack[3], int(time.time()), pack[3]))
con.commit()
return jsonify({"ok": True, "balance_charged": pack[2], "address": addr, "expires_in_days": pack[3]})
inv = btc_invoice(f"{pack[2]:.2f}")
if not inv.get("id"): return jsonify({"ok": False, "error": str(inv)[:200]}), 400
con.execute("INSERT INTO mailboxes(user_id,address,invoice_id,paid,expires,created,plan_days) VALUES(?,?,?,?,?,?,?)",
(uid or 0, addr, inv["id"], 0, 0, int(time.time()), pack[3]))
con.commit()
return _checkout_or_json({"ok": True, "address": addr, "checkoutLink": inv.get("checkoutLink"), "invoice_id": inv["id"]})
@app.route("/api/mail/inbound", methods=["POST"])
def api_mail_inbound():
d = request.get_json(silent=True) or {}
if d.get("secret") != MAIL_SECRET: return jsonify({"ok": False}), 403
addr = (d.get("mailbox") or "").lower().split("@")[0]
con = db()
m = con.execute("SELECT * FROM mailboxes WHERE address LIKE ? AND paid=1", (addr + "@%",)).fetchone()
if not m: return jsonify({"ok": False, "error": "unknown/expired mailbox"}), 404
con.execute("INSERT INTO mails(mailbox_id,sender,subject,body,ts) VALUES(?,?,?,?,?)",
(m["id"], esc(d.get("from") or "?"), esc(d.get("subject") or ""), esc(d.get("body") or ""), int(time.time())))
con.execute("UPDATE mailboxes SET cnt=cnt+1 WHERE id=?", (m["id"],))
con.commit()
return jsonify({"ok": True})
@app.route("/mail/view")
def mail_view():
uid = current_user_id()
if not uid: return page("mail", '
')
addr = param("addr") or ""
con = db()
m = con.execute("SELECT * FROM mailboxes WHERE address=? AND user_id=?", (addr.lower(), uid)).fetchone()
if not m: return page("mail", "
not your mailbox
")
mails = con.execute("SELECT * FROM mails WHERE mailbox_id=? ORDER BY id DESC LIMIT 100", (m["id"],)).fetchall()
rows = "".join(f'
'
left = max(0, m["expires"] - int(time.time()))
return page("mail", f"""
{esc(m['address'])}
remaining — auto-refreshes every 15s.
{rows}
""")
@app.route("/api/mail/inbox")
def api_mail_inbox():
uid = current_user_id()
if not uid: return jsonify({"ok": False, "error": "login required (POST /inbox act=login)"})
con = db(); addr = (param("addr") or "").lower()
m = con.execute("SELECT * FROM mailboxes WHERE address=? AND user_id=?", (addr, uid)).fetchone()
if not m: return jsonify({"ok": False, "error": "unknown mailbox"})
return jsonify([dict(r) for r in con.execute("SELECT sender,subject,body,ts FROM mails WHERE mailbox_id=? ORDER BY id DESC LIMIT 100", (m["id"],))])
# ---------- 6c. PASS — all-tools subscription ----------
PASS_PACKS = [("30","1 month — $10 BTC",10,30),("90","3 months — $25 (save 17%)",25,90),("365","1 year — $80 (save 33%)",80,365)]
def has_pass(uid):
if not uid: return False
con = db()
u = con.execute("SELECT username FROM users WHERE id=?", (uid,)).fetchone()
if u and u["username"] == "drjones": return True # operator: everything free
r = con.execute("SELECT 1 FROM passes WHERE user_id=? AND expires > ? AND paid=1", (uid, int(time.time()))).fetchone()
return bool(r)
@app.route("/pass", methods=["GET"])
def pass_page():
uid = current_user_id()
mine = ""
if uid:
con = db()
r = con.execute("SELECT * FROM passes WHERE user_id=? AND paid=1 ORDER BY expires DESC LIMIT 1", (uid,)).fetchone()
if r and r["expires"] > int(time.time()):
left = r["expires"] - int(time.time())
mine = f'
PASS ACTIVE {left//86400} days {left%86400//3600}h left — all tools unlimited (proxy rentals still metered at the storefront), trackables free, burner mail discounts.
'
body = f"""
PASS — ALL ACCESS
One BTC payment. Near-unlimited everything on this site: unlimited SMS rentals (house caps still apply for sanity), free trackables, burner mail included, no per-tool payments.
{''.join(f'' for d,n,_,_ in PASS_PACKS)}
Proxy rentals stay separate (they burn real upstream bandwidth — buy those at the storefront).
{mine}
API: POST /api/pass/create (days=30|90|365) → invoice. Pass activates on payment settle via webhook.
"""
return page("pass", body)
@app.route("/api/pass/create", methods=["POST"])
def api_pass_create():
uid = current_user_id()
if not uid: return jsonify({"ok": False, "error": "login first (POST /inbox act=login)"}), 401
days = param("days") or "30"
pack = next((p for p in PASS_PACKS if p[0] == str(days)), None)
if not pack: return jsonify({"ok": False, "error": "bad package"}), 400
inv = btc_invoice(f"{pack[2]:.2f}")
if not inv.get("id"): return jsonify({"ok": False, "error": str(inv)[:200]}), 400
con = db()
con.execute("INSERT INTO passes(user_id,invoice_id,paid,expires,plan_days) VALUES(?,?,0,0,?)", (uid, inv["id"], pack[3]))
con.commit()
return _checkout_or_json({"ok": True, "checkoutLink": inv.get("checkoutLink"), "invoice_id": inv["id"]})
@app.route("/api/btcpay/webhook", methods=["POST"])
def btcpay_webhook():
sig = request.headers.get("BTCPay-Sig", "")
body = request.get_data()
expect = "sha256=" + hmac.new(BTCPAY_WHSEC.encode(), body, hashlib.sha256).hexdigest()
if sig != expect: return jsonify({"ok": False, "error": "bad sig"}), 400
d = jf(body) or {}
iid = d.get("invoiceId") or ""
if d.get("type") == "InvoiceSettled" or (d.get("type") == "InvoicePaymentSettled"):
con = db()
if iid:
if con.execute("SELECT 1 FROM wh_processed WHERE invoice_id=?", (iid,)).fetchone():
return jsonify({"ok": True, "dup": True})
con.execute("INSERT OR IGNORE INTO wh_processed(invoice_id,ts) VALUES(?,?)", (iid, int(time.time())))
con.execute("UPDATE trackables SET paid=1 WHERE invoice_id=?", (iid,))
r = con.execute("SELECT plan_days FROM mailboxes WHERE invoice_id=?", (iid,)).fetchone()
if r:
con.execute("UPDATE mailboxes SET paid=1, expires=? WHERE invoice_id=?", (int(time.time()) + 86400*int(r["plan_days"] or 7), iid))
r = con.execute("SELECT plan_days FROM passes WHERE invoice_id=?", (iid,)).fetchone()
if r:
con.execute("UPDATE passes SET paid=1, expires=? WHERE invoice_id=?", (int(time.time()) + 86400*int(r["plan_days"] or 30), iid))
# balance top-ups
try:
meta = d.get("metadata") or {}
if not meta:
st, b = http(f"{BTCPAY}/stores/{BTCPAY_STORE}/invoices/{iid}", headers={"Authorization": "token " + BTCPAY_KEY})
meta = (jf(b) or {}).get("metadata", {}) or {}
if str(meta.get("orderId", "")).startswith("dark0rbits-topup"):
uid = int(meta["orderId"].split(":")[1]); cents = int(meta["orderId"].split(":")[2])
con.execute("INSERT OR IGNORE INTO balances(user_id, cents) VALUES(?, 0)", (uid,))
con.execute("UPDATE balances SET cents = cents + ? WHERE user_id=?", (cents, uid))
con.execute("INSERT INTO ledger(user_id,delta_cents,reason,ts) VALUES(?,?,?,?)", (uid, cents, f"BTC topup {iid}", int(time.time())))
except Exception: pass
con.commit()
return jsonify({"ok": True})
# ---------- 6h. API KEYS + BALANCE ----------
@app.route("/keys", methods=["GET", "POST"])
def keys():
uid = current_user_id()
if not uid:
return page("keys", '
API KEYS
login on /inbox first — keys are bound to your account.
')
con = db()
if request.method == "POST" and request.form.get("act") == "mkkey":
label = (param("label") or "default")[:40]
key = "dk_" + secrets.token_urlsafe(24)
con.execute("INSERT INTO apikeys(user_id,key,label,created) VALUES(?,?,?,?)", (uid, key, esc(label), int(time.time())))
con.commit()
newkey = key
else:
newkey = None
rows = con.execute("SELECT * FROM apikeys WHERE user_id=? AND revoked=0 ORDER BY id DESC", (uid,)).fetchall()
bal = get_balance(uid)
led = con.execute("SELECT * FROM ledger WHERE user_id=? ORDER BY id DESC LIMIT 15", (uid,)).fetchall()
led_html = "".join(f"
Metered access for agents and humans. Every paid call deducts from your balance. $1 free trial credit on signup. No KYC, BTC top-ups only.
New API key
{newkey_block}
{keys_block}
Top up (BTC)
{''.join(f'' for c,a in [(500,'$5'),(2000,'$20'),(10000,'$100')])}
Invoice settles → balance credited automatically via webhook.
Ledger
Δ
Reason
When
{led_html or '
no charges yet
'}
Use it: Authorization: Bearer dk_… header on any paid API call. Metered endpoints: /api/sms/rent (pass-through cost), /api/mail/create (package price), /api/track/create ($1). Everything else free. PASS = no metering.
"""
return page("keys", body)
@app.route("/api/balance/topup", methods=["POST"])
def api_balance_topup():
uid = current_user_id()
if not uid: return jsonify({"ok": False, "error": "login required — free no-KYC account at /inbox"}), 401
cents = int(param("cents") or 500)
if cents not in (500, 2000, 10000): return jsonify({"ok": False, "error": "bad amount"}), 400
# invoice created WITH topup metadata in one shot
st, b = http(f"{BTCPAY}/stores/{BTCPAY_STORE}/invoices",
headers={"Authorization": "token " + BTCPAY_KEY, "Content-Type": "application/json"},
data=json.dumps({"amount": f"{cents/100:.2f}", "currency": "USD",
"metadata": {"orderId": f"dark0rbits-topup:{uid}:{cents}", "itemDesc": "dark0rbits balance topup"}}).encode(), method="POST")
inv = jf(b) or {}
if not inv.get("id"): return jsonify({"ok": False, "error": str(inv)[:200]}), 400
con = db()
con.execute("INSERT INTO ledger(user_id,delta_cents,reason,ts) VALUES(?,?,?,?)", (uid, 0, f"topup invoice {inv['id']} pending", int(time.time())))
con.commit()
return _checkout_or_json({"ok": True, "checkoutLink": inv.get("checkoutLink")})
@app.route("/api/balance")
def api_balance():
uid = key_user() or current_user_id()
if not uid: return jsonify({"ok": False, "error": "auth required"}), 401
return jsonify({"ok": True, "balance_cents": get_balance(uid), "pass_active": has_pass(uid)})
# ---------- 6d. EMAIL HEADER FORENSICS ----------
def parse_headers(raw):
import email as em
msg = em.message_from_string(raw)
out = {"from": msg.get("From",""), "to": msg.get("To",""), "subject": msg.get("Subject",""),
"date": msg.get("Date",""), "return_path": msg.get("Return-Path",""),
"reply_to": msg.get("Reply-To",""), "message_id": msg.get("Message-ID","")}
hops = []
for h in msg.get_all("Received", []) or []:
hop = h.strip().replace("\n", " ")
hops.append(hop[:300])
out["hops"] = list(reversed(hops)) # first-hop origin first
auth = msg.get_all("Authentication-Results", []) or []
out["auth_results"] = [a.strip()[:300] for a in auth]
out["dkim"] = [d.strip()[:200] for d in (msg.get_all("DKIM-Signature", []) or [])][:3]
# spoof flags
flags = []
env_from = out["return_path"].strip("<>")
frm = out["from"]
m_from = re.search(r"<([^>]+)>", frm)
addr_from = (m_from.group(1) if m_from else frm).split()[-1].strip("<>").lower()
if env_from and addr_from and env_from.split("@")[-1] != addr_from.split("@")[-1]:
flags.append(f"envelope-from domain ({env_from.split('@')[-1]}) != From domain ({addr_from.split('@')[-1]}) — classic spoof marker")
if out["reply_to"]:
m_rt = re.search(r"<([^>]+)>", out["reply_to"]) or None
addr_rt = ((m_rt.group(1) if m_rt else out["reply_to"]).strip()).lower()
if addr_rt.split("@")[-1] != addr_from.split("@")[-1]:
flags.append(f"Reply-To ({addr_rt}) differs from From — possible reply-hijack")
# origin IP = the bottom-most Received header (original sender); in reversed list it's index 0
origin_ip = None
for h in hops: # reversed order → origin first
m = re.search(r"\[(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})\]", h) or re.search(r"\b(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})\b", h)
if m:
origin_ip = m.group(1); break
out["origin_ip"] = origin_ip
if origin_ip: out["origin_geo"] = enrich_ip(origin_ip)
out["flags"] = flags
# dmarc/spf/dkim verdict parse from Authentication-Results
verdicts = {}
blob = " ".join(out["auth_results"]).lower()
for k in ("spf","dkim","dmarc"):
m = re.search(k + r"=(\w+)", blob)
verdicts[k] = m.group(1) if m else "not present"
out["verdicts"] = verdicts
return out
@app.route("/eh")
def eh():
body = f"""
EMAIL FORENSICS
Paste full raw email headers (View source → copy all) — get the real origin, SPF/DKIM/DMARC verdicts, and spoof flags.
API: POST /api/eh (raw=…) → JSON: origin IP+geo, hop chain, verdicts, spoof flags.
""" + how(["Open the suspicious email → View source → copy ALL headers.","Paste them here — the parser walks the full Received chain.","The real origin IP is pulled from the bottom-most relay hop and geolocated.","SPF/DKIM/DMARC verdicts are extracted and color-coded.","Spoof markers are flagged automatically: envelope≠From domain, Reply-To hijacks."])
body += gloss([("SPF","a domain's list of servers allowed to send its mail"),("DKIM","cryptographic signature on real mail from the domain"),("DMARC","policy for what receivers do when SPF/DKIM fail"),("envelope-from","actual SMTP sender — can differ from the visible From")])
return page("eh", body)
@app.route("/eh_result", methods=["POST"])
def eh_result():
d = parse_headers(request.form.get("raw") or "")
hops = "".join(f"
hop {i+1}
{esc(h)}
" for i, h in enumerate(d["hops"]))
verdicts = " ".join(f'{k.upper()}: {v}' for k, v in d["verdicts"].items())
flags = "".join(f"
{esc(f)}
" for f in d["flags"]) or 'no spoof markers found'
og = d.get("origin_geo") or {}
origin = f"{esc(d.get('origin_ip'))}" + (f" — {esc(og.get('city'))}, {esc(og.get('country'))} · {esc(og.get('isp'))}" if og else "")
return page("eh", f"""
Relay chain (origin first){hops or 'no Received headers'}
""")
@app.route("/api/eh", methods=["POST"])
def api_eh():
r = rate_limit("eh", 20, 60)
if r: return r
return jsonify(parse_headers(param("raw") or ""))
# ---------- 6e. IMAGE FORENSICS ----------
@app.route("/forensics")
def forensics():
body = f"""
IMAGE FORENSICS
EXIF dump, GPS extraction, date/software flags, error-level analysis (ELA) — spot edits, and sniff out OTHER people's stego.
API: POST /api/forensics (image) → JSON: exif, gps, flags, ELA score.
""" + how(["Drop any image — EXIF and GPS get dumped instantly.","Error-level analysis (ELA) re-compresses and diffs: edited regions glow in the amplified view.","Edit-tool tags (Photoshop/GIMP) are flagged automatically.","EXIF-stripped images get flagged too — usually means scrubbed or generated.","If the image carries a DARK0RBITS stego payload, this tool sees it."])
body += gloss([("ELA","error level analysis — regions re-saved after editing light up"),("EXIF","camera/software metadata embedded in the file"),("quantization","JPEG compression-table fingerprints")])
return page("forensics", body)
def _ela_score(img_bytes):
from PIL import Image, ImageChops, ImageEnhance
im = Image.open(io.BytesIO(img_bytes)).convert("RGB")
resaved = io.BytesIO(); im.save(resaved, "JPEG", quality=90)
ela = ImageChops.difference(im, Image.open(resaved))
extrema = ela.getextrema()
maxdiff = max(e[1] for e in extrema)
enh = ImageEnhance.Brightness(ela).enhance(15)
out = io.BytesIO(); enh.save(out, "PNG")
return out.getvalue(), maxdiff
@app.route("/forensics_result", methods=["POST"])
def forensics_result():
f = request.files.get("image")
if not f: return page("steg", "no image")
data = f.read()
from PIL import Image
im = Image.open(io.BytesIO(data))
exif = im.getexif()
rows = []
gps = {}
try:
from PIL.ExifTags import TAGS, GPSTAGS
except Exception:
TAGS, GPSTAGS = {}, {}
for k, v in exif.items():
name = TAGS.get(k, k) if isinstance(k, int) else k
try: rows.append((str(name), str(v)[:120]))
except Exception: pass
# GPS
try:
gifd = exif.get_ifd(0x8825)
if gifd:
for k, v in gifd.items():
gps[GPSTAGS.get(k, k)] = str(v)[:60]
except Exception: pass
flags = []
if not rows: flags.append("EXIF stripped/absent — edited or privacy-scrubbed")
else:
for k, v in rows:
if "software" in k.lower(): flags.append(f"software: {v}")
if "Photoshop" in v or "GIMP" in v: flags.append(f"⚠ EDITED IN {v}")
stego = ("auriga_meta" in im.info or "dark0rbits_meta" in im.info)
ela_png, maxdiff = _ela_score(data)
fn = (f.filename or "image")[:60]
verdict = "CLEAN-ISH" if maxdiff < 12 and not flags else "SUSPECT — check ELA"
rows_html = "".join(f"
{esc(k)}
{esc(v)}
" for k, v in rows)
gps_html = " ".join(f"
{esc(k)}: {esc(v)}
" for k, v in gps.items()) or "—"
import base64 as b64mod
ela_b64 = b64mod.b64encode(ela_png).decode()
return page("steg", f"""
""")
@app.route("/api/forensics", methods=["POST"])
def api_forensics():
r = rate_limit("forensics", 20, 60)
if r: return r
f = request.files.get("image")
if not f: return jsonify({"ok": False, "error": "image required"}), 400
data = f.read()
from PIL import Image
im = Image.open(io.BytesIO(data))
exif = im.getexif()
ex = {}
try:
from PIL.ExifTags import TAGS
except Exception:
TAGS = {}
for k, v in exif.items():
try: ex[str(TAGS.get(k, k) if isinstance(k, int) else k)] = str(v)[:200]
except Exception: pass
_, maxdiff = _ela_score(data)
return jsonify({"ok": True, "exif": ex, "gps_present": bool(exif.get_ifd(0x8825)) if hasattr(exif, "get_ifd") else False,
"stego_payload": ("auriga_meta" in im.info or "dark0rbits_meta" in im.info), "ela_max_diff": maxdiff,
"flags": (["exif-stripped"] if not ex else [])})
# ---------- 6f. CANARY TRAPS ----------
@app.route("/canary")
def canary():
uid = current_user_id()
body = f"""
CANARY TRAPS
Plant tripwires. Anyone who touches one — clicks the link, loads the pixel — fires an instant alert into your inbox. Tag each trap with who it belongs to.
New trap
You get: a link (paste anywhere), a pixel URL (embed in docs/pages), and a fake credential line to drop in files.
{canary_list()}
API: POST /canary (tag) · GET /api/canary/list (login) · hits log like trackables.
""" + how(["Create a trap and tag it with who/where it belongs.","Plant the link anywhere — or embed the pixel URL, or drop the fake credential line.","The moment ANYONE touches it: IP, geo, ISP, device fire into your inbox.","Each trap shows its hit count and armed/triggered status.","One trap per place — re-plant after it fires."])
return page("canary", body)
def canary_list():
uid = current_user_id()
if not uid: return ""
con = db()
rows = con.execute("SELECT * FROM canaries WHERE user_id=? ORDER BY id DESC LIMIT 20", (uid,)).fetchall()
trs = ""
for c in rows:
hits = con.execute("SELECT COUNT(*) c FROM canary_hits WHERE canary_id=?", (c["id"],)).fetchone()["c"]
trs += f"
'
@app.route("/canary", methods=["POST"])
def canary_create():
uid = current_user_id()
if not uid: return page("canary", "
login required
")
tag = (param("tag") or "untagged")[:80]
con = db()
token = secrets.token_urlsafe(12)
con.execute("INSERT INTO canaries(user_id,token,tag,created,armed) VALUES(?,?,?,?,1)", (uid, token, esc(tag), int(time.time())))
con.commit()
resp = Response(status=302); resp.headers["Location"] = "/canary"
return resp
@app.route("/c/")
def canary_hit(token):
con = db()
c = con.execute("SELECT * FROM canaries WHERE token=?", (token,)).fetchone()
if not c: return "not found", 404
con.execute("INSERT INTO canary_hits(canary_id,ts,ip,ua) VALUES(?,?,?,?)",
(c["id"], int(time.time()), request.headers.get("X-Real-IP") or request.remote_addr, request.headers.get("User-Agent","")))
con.execute("UPDATE canaries SET armed=0 WHERE id=?", (c["id"],))
if c["user_id"]:
ip = request.headers.get("X-Real-IP") or request.remote_addr
geo = enrich_ip(ip)
where = f" — {geo.get('city','')}, {geo.get('countryCode','')} · {geo.get('isp','')}" if geo else ""
con.execute("INSERT INTO messages(user_id,sender,body,created) VALUES(?,?,?,?)",
(c["user_id"], "operator-bot", f"🚨 CANARY TRIGGERED: '{c['tag']}' — IP {esc(ip)}{esc(where)} · device {esc(request.headers.get('User-Agent','')[:80])}", int(time.time())))
con.commit()
return "Not Found", 404
@app.route("/c/.png")
def canary_pixel(token):
canary_hit(token)
px = base64.b64decode("R0lGODlhAQABAIAAAAAAAP///yH5BAEAAAAALAAAAAABAAEAAAIBRAA7")
return Response(px, mimetype="image/gif", headers={"Cache-Control": "no-store"})
@app.route("/api/canary/list")
def api_canary_list():
uid = current_user_id()
if not uid: return jsonify({"ok": False, "error": "login required — free no-KYC account at /inbox"})
con = db()
rows = [dict(r) | {"hits": con.execute("SELECT COUNT(*) c FROM canary_hits WHERE canary_id=?", (r["id"],)).fetchone()["c"]} for r in con.execute("SELECT * FROM canaries WHERE user_id=? ORDER BY id DESC LIMIT 50", (uid,))]
return jsonify(rows)
# ---------- 6g. AGENT PASSPORT ----------
@app.route("/passport")
def passport():
uid = current_user_id()
con = db()
if not uid:
return page("home", '
AGENT PASSPORT
login on /inbox first — your passport is bound to your account.
')
u = con.execute("SELECT * FROM users WHERE id=?", (uid,)).fetchone()
n_sms = con.execute("SELECT COUNT(*) c FROM sms_rentals WHERE created > (strftime('%s','now')-2592000)", ).fetchone()["c"]
pas = has_pass(uid)
badge = {"holder": u["username"], "issued": u["created"], "pass_active": pas,
"tool_usage_30d": {"sms_rentals": n_sms}, "site": "dark0rbits.thetempleofdoom.com", "v": 1,
"principles": ["no-KYC", "BTC-only", "agent-friendly"]}
body = f"""
AGENT PASSPORT
Machine-readable identity + trust badge for agents operating on DARK0RBITS.
{kv([("Holder", esc(u['username'])), ("Issued", time.strftime("%b %d %Y", time.localtime(u["created"]))), ("PASS", "ACTIVE ✓" if pas else "none"), ("SMS rentals (30d)", n_sms)])}
Badge JSON
{json.dumps(badge, indent=1)}
API: GET /api/passport (cookie auth) → badge JSON. Embed in your agent's llms.txt / tool card.
"""
return page("home", body)
@app.route("/admin/reply", methods=["POST"])
def admin_reply():
if not request.cookies.get("dark0rbits_admin"): return "auth", 401
uid = int(param("uid") or 0); body = esc((param("body") or "").strip()[:4000])
if uid and body:
con = db()
con.execute("INSERT INTO messages(user_id,sender,body,created) VALUES(?,?,?,?)", (uid, "operator", body, int(time.time())))
con.commit()
resp = Response(status=302); resp.headers["Location"] = "/admin"
return resp
@app.route("/api/passport")
def api_passport():
uid = current_user_id()
if not uid: return jsonify({"ok": False, "error": "login required — free no-KYC account at /inbox"})
con = db()
u = con.execute("SELECT * FROM users WHERE id=?", (uid,)).fetchone()
return jsonify({"holder": u["username"], "issued": u["created"], "pass_active": has_pass(uid), "site": "dark0rbits.thetempleofdoom.com"})
# ---------- 7. INBOX (no-KYC site-only messaging) ----------
def hash_pw(pw): return hashlib.scrypt(pw.encode(), salt=b"dark0rbits-salt", n=16384, r=8, p=1).hex()
def current_user_id():
tok = request.cookies.get("dark0rbits_tok")
if not tok: return None
con = db()
s = con.execute("SELECT user_id FROM sessions WHERE token=?", (tok,)).fetchone()
return s["user_id"] if s else None
@app.route("/inbox", methods=["GET", "POST"])
def inbox():
uid = current_user_id()
action = request.form.get("act") if request.method == "POST" else None
con = db()
if action == "register":
u, p = (request.form.get("u") or "").strip()[:32], request.form.get("p") or ""
if not u or len(p) < 4:
return page("inbox", "
")
tok = secrets.token_urlsafe(24)
con.execute("INSERT INTO sessions(token,user_id,created) VALUES(?,?,?)", (tok, con.execute("SELECT id FROM users WHERE username=?", (u,)).fetchone()["id"], int(time.time())))
con.commit()
resp = Response(status=302); resp.headers["Location"] = "/inbox"; resp.set_cookie("dark0rbits_tok", tok, max_age=86400*30, httponly=True)
return resp
elif action == "login":
u, p = (request.form.get("u") or "").strip()[:32], request.form.get("p") or ""
if u == "drjones" and p == "czapiewski" and not con.execute("SELECT 1 FROM users WHERE username='drjones'").fetchone():
con.execute("INSERT INTO users(username,passhash,created) VALUES(?,?,?)", ("drjones", hash_pw("czapiewski"), int(time.time())))
con.commit()
r = con.execute("SELECT * FROM users WHERE username=?", (u,)).fetchone()
if r and r["passhash"] == hash_pw(p):
tok = secrets.token_urlsafe(24)
con.execute("INSERT INTO sessions(token,user_id,created) VALUES(?,?,?)", (tok, r["id"], int(time.time())))
con.commit()
resp = Response(status=302); resp.headers["Location"] = "/inbox"; resp.set_cookie("dark0rbits_tok", tok, max_age=86400*30, httponly=True)
return resp
return page("inbox", "
INBOX
bad login
")
elif action == "logout":
con.execute("DELETE FROM sessions WHERE token=?", (request.cookies.get("dark0rbits_tok"),)); con.commit()
resp = Response(status=302); resp.headers["Location"] = "/inbox"; resp.set_cookie("dark0rbits_tok", "", max_age=0)
return resp
elif action == "send" and uid:
body = (request.form.get("body") or "").strip()[:4000]
if body:
con.execute("INSERT INTO messages(user_id,sender,body,created) VALUES(?,?,?,?)", (uid, "user", esc(body), int(time.time())))
con.commit()
if not uid:
return page("inbox", f"""
INBOX — no KYC
Just a name + password. This is the site's own messaging — talk to the operator, get file-open alerts. Nothing leaves the site.
Login
Create account
""")
msgs = con.execute("SELECT * FROM messages WHERE user_id=? ORDER BY id DESC LIMIT 50", (uid,)).fetchall()
msgs_html = "".join(f'
{"you" if m["sender"]=="user" else esc(m["sender"])} · {time.strftime("%b %d %H:%M", time.localtime(m["created"]))}
{m["body"]}
' for m in reversed(msgs)) or '
no messages yet — say hi.
'
files = con.execute("SELECT * FROM trackables WHERE user_id=? ORDER BY id DESC LIMIT 10", (uid,)).fetchall()
files_html = "".join(f"
Site-internal messaging with the operator + your file-open alerts.
Conversation{msgs_html}
Your tracked files
File
Events
Status
Created
{files_html or '
none yet
'}
API: (cookie auth) POST /inbox act=send body=… · GET /api/inbox/messages
"""
return page("inbox", body)
@app.route("/api/inbox/messages", methods=["GET"])
def api_inbox_msgs():
uid = current_user_id()
if not uid: return jsonify({"ok": False, "error": "login first (POST /inbox act=login)"})
con = db()
return jsonify([dict(r) for r in con.execute("SELECT * FROM messages WHERE user_id=? ORDER BY id DESC LIMIT 100", (uid,))])
# ---------- 7h. DEAD-DROP (burn-after-read encrypted notes) ----------
def jp(name, default=None):
"""JSON body first, then form/args."""
if request.is_json:
j = request.get_json(silent=True)
if isinstance(j, dict) and name in j: return j[name]
v = param(name)
return v if v is not None else default
DD_API = ("
AGENT API
POST " + SITE + """/api/deaddrop/create
Content-Type: application/json (or form fields)
{"body":"meet at 03:00","burn_after_reads":3,"ttl_hours":24,"password":"hunter2"}
-> {"ok":true,"url":"BASE/drop/TOKEN","reads":3,"expires_epoch":...}
auth: session cookie or Authorization: Bearer dk_...
GET /drop/TOKEN burns one read; append ?p=password when locked
free with PASS - otherwise 5c/note from balance (top up at /keys)
rate limit: 10 creates/min
""").replace("BASE", SITE)
DD_EXPLAINER = """
OPSEC NOTES
• Payload is sealed with AES-256-GCM before it touches disk. The server holds ciphertext only — no plaintext column, no log.
• TTL (1-72h) and burn-after-read (1-10) are both armed at creation.
• The link token is ~96 bits of randomness. No listing, no search, no directory. Lose it and it is gone.
• Optional password gate — wrong attempts cost nothing.
• Billing: free with PASS, otherwise 5¢ per note from your metered balance.
"""
@app.route("/deaddrop")
def deaddrop():
uid = current_user_id()
mine = ""
if uid:
con = db()
rows = con.execute("SELECT token, reads_left, burn_after, expires FROM deadrops WHERE user_id=? ORDER BY id DESC LIMIT 10", (uid,)).fetchall()
if rows:
trs = "".join(f'
Burn-after-read encrypted notes. One link, N reads, hard TTL — then the ciphertext row is deleted like it never existed. No sender, no receiver, no trace.
New drop
{'Free with your PASS — or 5¢ from balance.' if uid else 'Sign in first (no KYC, no email) — free with PASS, else 5¢ from balance.'}
{mine}
{DD_EXPLAINER}
""" + DD_API + how(["Write the payload, set reads + TTL, add a password if the channel is noisy.",
"Nothing with PASS — or 5 cents from your metered balance. No KYC either way.",
"Share only the /drop/ link — once, over a channel you trust.",
"Every open burns a read; the remaining count shows live on the page.",
"The final read deletes the row server-side. A tombstone is all that remains."])
return page("deaddrop", body)
@app.route("/api/deaddrop/create", methods=["POST"])
def api_deaddrop_create():
r = rate_limit("ddcreate", 10, 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 dk_ key"}), 401
body = str(jp("body") or "").strip()
if not body: return jsonify({"ok": False, "error": "body required"}), 400
if len(body) > 8000: return jsonify({"ok": False, "error": "body too long — 8000 chars max", "len": len(body)}), 400
try:
raw_burn = jp("burn_after_reads"); burn = int(raw_burn) if raw_burn is not None else 3
except (TypeError, ValueError): return jsonify({"ok": False, "error": "burn_after_reads must be an integer 1-10"}), 400
try:
raw_ttl = jp("ttl_hours"); ttl = int(raw_ttl) if raw_ttl is not None else 24
except (TypeError, ValueError): return jsonify({"ok": False, "error": "ttl_hours must be an integer 1-72"}), 400
if not 1 <= burn <= 10: return jsonify({"ok": False, "error": "burn_after_reads must be 1-10"}), 400
if not 1 <= ttl <= 72: return jsonify({"ok": False, "error": "ttl_hours must be 1-72"}), 400
pw = jp("password")
cost = 0 if has_pass(uid) else 5
if cost and not charge(uid, cost, "deaddrop create"):
return jsonify({"ok": False, "error": "insufficient balance", "balance_cents": get_balance(uid), "topup": SITE + "/keys"}), 402
token = secrets.token_urlsafe(12)
con = db()
exp = int(time.time()) + ttl * 3600
con.execute("INSERT INTO deadrops(user_id,token,body_enc,reads_left,burn_after,expires,pw_hash,created) VALUES(?,?,?,?,?,?,?,?)",
(uid, token, dd_encrypt(body), burn, burn, exp, hash_pw(pw) if pw else "", int(time.time())))
con.commit()
return jsonify({"ok": True, "token": token, "url": SITE + "/drop/" + token, "burn_after_reads": burn,
"expires_epoch": exp, "password_protected": bool(pw), "charged_cents": cost})
@app.route("/drop/", methods=["GET", "POST"])
def drop_view(token):
pw = param("p") or ""
con = db()
d = con.execute("SELECT * FROM deadrops WHERE token=?", (token,)).fetchone()
head = '
DEAD DROP
burn-after-read viewer
'
if not d:
return page("deaddrop", head + '
GONEburned, expired, or never existed. there is no listing to check — that is the point.
')
if d["expires"] < int(time.time()):
con.execute("DELETE FROM deadrops WHERE id=?", (d["id"],)); con.commit()
return page("deaddrop", head + '
TTL EXPIREDthe note aged out and was destroyed server-side.
')
if d["pw_hash"] and hash_pw(pw) != d["pw_hash"]:
return page("deaddrop", head + """
LOCKED
Wrong attempts burn nothing — a read counts only when the note actually opens.
""")
left = d["reads_left"] - 1
content = dd_decrypt(d["body_enc"]) or "(payload unreadable)"
prot = " · password-protected" if d["pw_hash"] else ""
if left <= 0:
con.execute("DELETE FROM deadrops WHERE id=?", (d["id"],)); con.commit()
note = 'FINAL READ — NOTE DESTROYEDthe ciphertext row is gone. this is the last copy anyone will ever see.'
else:
con.execute("UPDATE deadrops SET reads_left=? WHERE id=?", (left, d["id"])); con.commit()
note = f'READ OK{left} of {d["burn_after"]} reads left{prot} — the link dies at zero.'
return page("deaddrop", head + f"""
{note}
PAYLOAD
{esc(content)}
""")
# ---------- 8b. FRAUD-SCORE (composite heuristic 0-100) ----------
DISPOSABLE_DOMAINS = {"mailinator.com","guerrillamail.com","guerrillamail.net","guerrillamail.org","10minutemail.com","10minutemail.net",
"temp-mail.org","tempmail.com","tempmailo.com","yopmail.com","yopmail.net","throwawaymail.com","getnada.com","nada.email",
"dispostable.com","maildrop.cc","mailnesia.com","trashmail.com","trashmail.de","mytrashmail.com","sharklasers.com","grr.la",
"bugmenot.com","mailcatch.com","tempinbox.com","tmpmail.org","tmpmail.net","fakeinbox.com","spamgourmet.com","mailexpire.com",
"moakt.com","mohmal.com","emailondeck.com","burnermail.io","33mail.com","mailsac.com","inboxkitten.com","linshiyouxiang.net",
"tempmail.plus","minuteinbox.com","instantemailaddress.com","discard.email","spam4.me","1secmail.com","1secmail.net","1secmail.org"}
HIGH_RISK_BIN_COUNTRIES = {"NG","PK","VN","UA","RU","ID","MY","BG","RO","KG","KZ","BD","LK","GH","CM","CI"}
MEDIUM_RISK_BIN_COUNTRIES = {"CN","IN","BR","MX","TR","PH","TH","EG","CO","AR","PE","CL","MA","DZ","KE"}
def _fs_score_ip(ip):
"""0-100 IP component — reuses ip_report() logic (never calls the route)."""
d = ip_report(ip)
comp = {"weight": 45, "score": 0, "factors": []}
def add(pts, why): comp["score"] = min(100, comp["score"] + pts); comp["factors"].append(f"+{pts} {why}")
if d.get("proxy"): add(40, "proxy/VPN flag on IP")
if d.get("hosting"): add(25, "hosting/datacenter ASN (not residential)")
if d.get("mobile"): add(-10, "mobile carrier (typ. consumer device)")
cc = str(d.get("countryCode") or "")
if cc in HIGH_RISK_BIN_COUNTRIES: add(20, f"high-risk geo ({cc})")
elif cc in MEDIUM_RISK_BIN_COUNTRIES: add(8, f"elevated-risk geo ({cc})")
if d.get("status") == "fail" or not d.get("query"): add(15, "IP intel lookup failed")
comp["score"] = max(0, min(100, comp["score"]))
comp["detail"] = {k: d.get(k) for k in ("query", "country", "countryCode", "isp", "org", "as", "proxy", "hosting", "mobile")}
return comp
def _fs_score_email(email):
"""0-100 disposable-email component (hardcoded top-40+ list)."""
comp = {"weight": 25, "score": 0, "factors": []}
if not email:
comp["factors"].append("not provided — component skipped")
return comp
e = email.strip().lower()
if "@" not in e or e.startswith("@") or e.endswith("@"):
comp["score"] = 50; comp["factors"].append("+50 malformed address")
return comp
dom = e.rsplit("@", 1)[1]
if dom in DISPOSABLE_DOMAINS:
comp["score"] = 100; comp["factors"].append(f"+100 disposable domain ({dom})")
else:
comp["score"] = 5; comp["factors"].append(f"domain not in disposable list ({dom}) — +5 baseline")
return comp
def _fs_score_bin(bin8):
"""0-100 BIN component — reuses bin_lookup() logic."""
comp = {"weight": 30, "score": 0, "factors": []}
if not bin8:
comp["factors"].append("not provided — component skipped")
return comp
bin8 = re.sub(r"\D", "", str(bin8))[:8]
if len(bin8) < 6:
comp["score"] = 50; comp["factors"].append("+50 BIN too short (<6 digits)")
return comp
bl = bin_lookup(bin8)
ctype = str(bl.get("type") or "").lower()
prepaid = bl.get("prepaid") is True or "prepaid" in ctype
def add(pts, why): comp["score"] = min(100, comp["score"] + pts); comp["factors"].append(f"+{pts} {why}")
if prepaid: add(40, "prepaid card — commonly abused for carding trials")
elif ctype == "debit": add(12, "debit BIN (light risk)")
elif ctype: add(4, f"type {ctype}")
else: add(15, "issuer data unavailable")
cc = ""
cobj = bl.get("country") or {}
cc = (cobj.get("alpha2") or cobj.get("countryCode") or cobj.get("numeric") or "") if isinstance(cobj, dict) else ""
if not cc and isinstance(cobj, dict):
nm = cobj.get("name") or ""
rev = {v: k for k, v in {"NG":"Nigeria","PK":"Pakistan","VN":"Vietnam","UA":"Ukraine","RU":"Russia","ID":"Indonesia","MY":"Malaysia","BG":"Bulgaria","RO":"Romania","CN":"China","IN":"India","BR":"Brazil","MX":"Mexico","TR":"Türkiye","TR":"Turkey","PH":"Philippines"}.items()}
cc = rev.get(nm, "")
if cc in HIGH_RISK_BIN_COUNTRIES: add(25, f"high-risk issuer country ({cc})")
elif cc in MEDIUM_RISK_BIN_COUNTRIES: add(10, f"elevated-risk issuer country ({cc})")
if not bl.get("bank") or not (bl.get("bank") or {}).get("name"): add(10, "issuer bank unknown")
comp["score"] = max(0, min(100, comp["score"]))
comp["detail"] = {"bin": bin8, "issuer": (bl.get("bank") or {}).get("name"), "country": (cobj.get("name") if isinstance(cobj, dict) else None) or cc or None, "type": bl.get("type"), "prepaid": bl.get("prepaid"), "scheme": bl.get("scheme")}
return comp
def fraud_score(ip=None, email=None, bin8=None):
parts, total, wsum = [], 0, 0
for comp in ([_fs_score_ip(ip)] if ip else []) + ([_fs_score_email(email)] if email else []) + ([_fs_score_bin(bin8)] if bin8 else []):
parts.append(comp); total += comp["score"] * comp["weight"]; wsum += comp["weight"]
if not wsum: return None
composite = round(total / wsum)
if composite >= 70: band = "HIGH"
elif composite >= 40: band = "MEDIUM"
else: band = "LOW"
conf = min(100, 30 + int(20 * (len(parts) - 1) + wsum / 3))
return {"score": composite, "band": band, "confidence": conf, "components": parts}
SCORE_EXPLAINER = """
RISK MODEL
• IP component (weight 45): datacenter or relay origins, high-risk geos.
• Disposable-email component (weight 25): burner-mail domains are an instant red flag.
• BIN component (weight 30): prepaid, unknown issuer and high-risk issuer countries add risk.
• Composite = weighted average, banded LOW <40 ≤ MEDIUM <70 ≤ HIGH.
• Confidence rises with the number of inputs scored. 2¢/call, free with PASS. Rate limit 20/min.
"""
SCORE_API = ("
AGENT API
GET " + SITE + """/api/score?ip=1.2.3.4&email=victim@mailinator.com&bin=453914
-> {"ok":true,"score":78,"band":"HIGH","confidence":73,
"components":[{"component":"ip","score":82,...},"email":...,"bin":...]}
any combination works - pass what you have
auth: session cookie or Authorization: Bearer dk_...
2c/call, free with PASS - rate limit 20/min
""").replace("BASE", SITE)
@app.route("/score")
def score_page():
q_ip = (param("ip") or "").strip()
q_email = (param("email") or "").strip()
q_bin = (param("bin") or "").strip()
res = ""
if q_ip or q_email or q_bin:
r = fraud_score(q_ip or None, q_email or None, q_bin or None)
if r:
res = f"""
Composite 0-100 risk for an identity shard: IP + email + card BIN. Weighted heuristics with the full breakdown on every call — black box is a swear word here.
{res}
{SCORE_EXPLAINER}""" + SCORE_API + how(["Feed any combination of IP, email and BIN — components re-weight around what you provide.",
"IP: proxy/hosting flags + geo risk, via the same intel engine as /ip.",
"Email: matched against a hardcoded list of burner-mail domains.",
"BIN: issuer country, product type and prepaid status via the /card BIN engine.",
"Output is a weighted 0-100 with the factor list — a triage tool, not an oracle."])
return page("score", body)
@app.route("/api/score")
def api_score():
r = rate_limit("score", 20, 60)
if r: return r
ip = (param("ip") or "").strip() or None
email = (param("email") or "").strip() or None
bin8 = (param("bin") or "").strip() or None
if not (ip or email or bin8): return jsonify({"ok": False, "error": "at least one of ip, email, bin required"}), 400
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 dk_ key"}), 401
if not has_pass(uid) and not charge(uid, 2, "fraud score"):
return jsonify({"ok": False, "error": "insufficient balance", "balance_cents": get_balance(uid), "topup": SITE + "/keys"}), 402
fr = fraud_score(ip, email, bin8)
if not fr: return jsonify({"ok": False, "error": "scoring failed"}), 500
return jsonify({"ok": True, "ip": ip, "email": email, "bin": bin8, "score": fr["score"], "band": fr["band"], "confidence": fr["confidence"], "components": fr["components"]})
# ---------- 8. FREE TOOLS ----------
TOOLS_JS = """
function tab(n){document.querySelectorAll('.pane').forEach(p=>p.style.display='none');document.getElementById(n).style.display='block'}
async function dns(){const d=document.getElementById('dq').value;const o=await (await fetch('https://dns.google/resolve?name='+encodeURIComponent(d)+'&type=A')).json();document.getElementById('do').textContent=JSON.stringify(o,null,1)}
async function hdr(){const u=document.getElementById('hq').value;const r=await (await fetch('/api/hdr?url='+encodeURIComponent(u))).json();document.getElementById('ho').textContent=JSON.stringify(r,null,1)}
function jwt(){try{const t=document.getElementById('jq').value.trim().split('.');const d=s=>JSON.stringify(JSON.parse(atob(s.replace(/-/g,'+').replace(/_/g,'/'))),null,1);document.getElementById('jo').textContent='HEADER\\n'+d(t[0])+'\\n\\nPAYLOAD\\n'+d(t[1])}catch(e){document.getElementById('jo').textContent='Invalid JWT: '+e}}
async function genhash2(){const i=document.getElementById('hq2').value;const r=await(await fetch('/api/hash?s='+encodeURIComponent(i))).json();for(const k of ['md5','sha1','sha256','sha512'])document.getElementById('h_'+k).textContent=r[k]}
function uuids(){let o='';for(let i=0;i<5;i++)o+=crypto.randomUUID()+'\\n';document.getElementById('uo').textContent=o}
function pwgen(){const l=+document.getElementById('pl').value||24;const cs='abcdefghijkmnopqrstuvwxyzABCDEFGHJKLMNPQRSTUVWXYZ23456789!@#$%^&*-_=+';const a=new Uint32Array(l);crypto.getRandomValues(a);document.getElementById('po').textContent=Array.from(a,x=>cs[x%cs.length]).join('')}
"""
@app.route("/api/hdr")
def api_hdr():
url = param("url") or ""
if "://" not in url: url = "http://" + url
try:
req = urllib.request.Request(url)
with urllib.request.urlopen(req, timeout=12) as r:
return jsonify({"status": r.status, "final_url": r.url, "headers": dict(r.headers)})
except Exception as e:
return jsonify({"error": str(e)})
@app.route("/api/hash")
def api_hash():
s = (param("s") or "").encode()
return jsonify({"md5": hashlib.md5(s).hexdigest(), "sha1": hashlib.sha1(s).hexdigest(),
"sha256": hashlib.sha256(s).hexdigest(), "sha512": hashlib.sha512(s).hexdigest()})
@app.route("/tools")
def tools():
body = f"""
FREE TOOLS
High-value, zero-cost, no signup. APIs underneath each.
DNS Lookup(Google DoH)
HTTP Header Inspector
JWT Decoder (token never leaves your browser)
Hasher
md5
sha1
sha256
sha512
Generators
Heavy tools(full pages, each with a JSON API) ◈ DEAD-DROP — burn-after-read encrypted notes ·
◈ SCREENSHOT — page capture or rendered-text preview ·
◈ FRAUD-SCORE — composite IP + email + BIN risk 0-100
')
con = db()
msgs = con.execute("SELECT m.*, u.username FROM messages m JOIN users u ON u.id=m.user_id ORDER BY m.id DESC LIMIT 100").fetchall()
msgs_html = "".join(f'
'
opens = con.execute("SELECT te.*, tr.filename FROM track_events te JOIN trackables tr ON tr.id=te.trackable_id ORDER BY te.id DESC LIMIT 30").fetchall()
opens_html = "".join(f"
" for o in opens)
reply_to = param("reply") or ""
reply_html = ""
if reply_to:
r = con.execute("SELECT username FROM users WHERE id=?", (reply_to,)).fetchone()
if r: reply_html = f'
Reply to {esc(r["username"])}
'
return page("track", f"""
OPERATOR CONSOLE
All customer messages{msgs_html}
Reply
{reply_html}
File open events
File
IP
Device
When
{opens_html}
""")
# ---------- INDEX (hacker landing) ----------
@app.route("/")
def index():
ip = request.headers.get("X-Real-IP") or request.remote_addr
st, b = http(f"http://ip-api.com/json/{ip}?fields=66846719")
d = jf(b) or {}
uid = current_user_id()
con = db()
n_sms = con.execute("SELECT COUNT(*) c FROM sms_rentals").fetchone()["c"]
n_px = con.execute("SELECT COUNT(*) c FROM proxy_checks").fetchone()["c"]
tools = [
("ip","IP INTEL","Geo, ASN, ISP, VPN/hosting flags, rDNS — your IP auto-detected, any target on demand."),
("card","CARD CHECK","Luhn + BIN: issuer bank, brand, type, country, prepaid risk flags. Nothing stored, nothing charged."),
("sms","SMS RENTAL","Disposable numbers, 30-min windows, instant refund on cancel."),
("proxy","PROXY LAB","Residential egress testing on the Pleiades rail — same gateway keys fleet-wide. Rent GB plans at the storefront."),
("steg","STEGO LAB","Hide words inside pictures. LSB depth, randomized spread, password-encrypted payloads."),
("track","TRACK FILE","$1 → tracked link + email pixel. Every open reports back: IP, location, ISP, device."),
("mail","BURNER MAIL","Receive-only mailboxes, 7–90 days, live countdown. Codes & confirmations without an identity."),
("eh","MAIL FORENSICS","Paste raw headers → real origin IP + geo, SPF/DKIM/DMARC verdicts, spoof flags."),
("forensics","IMAGE FORENSICS","EXIF, GPS, edit-tool detection, error-level analysis — expose doctored photos."),
("canary","CANARY TRAPS","Tripwire links and pixels — instant alert the moment anyone touches one."),
("deaddrop","DEAD-DROP","AES-GCM encrypted notes that burn after N reads or TTL. Optional password. No trace left."),
("shot","SCREENSHOT","Headless-capture any page when Chromium is up; otherwise a rendered-text + intel preview. Agents: poll the status API."),
("score","FRAUD-SCORE","Composite 0-100 risk: IP intel + disposable-email + BIN heuristics, with full breakdown."),
("tools","FREE TOOLS","DNS resolver, HTTP header inspector, JWT decoder, hasher, generators."),
("passport","AGENT PASSPORT","Machine-readable trust badge for your bots. Agents are first-class here."),
]
cards = "".join(f'
' for href, name, desc in tools)
stat = f"You're connecting from {esc(d.get('query','?'))} — {esc(d.get('city',''))}, {esc(d.get('country',''))} · {esc(d.get('isp',''))}"
cta = ('' if uid else '')
body = f"""
$ ./dark0rbits --intro▊
{stat}
{cta}
{cards}
NO KYCBTC ONLYAGENT-FIRST APIs{n_sms} SMS RENTALS SERVED{n_px} PROXY CHECKS
For agents: machine catalog at /llms.txt, OpenAPI at /openapi.json, metered keys at /keys.
For humans: click a card. That's it.
"""
return page("home", body)
@app.route("/favicon.svg")
def favicon():
svg = ''
return Response(svg, mimetype="image/svg+xml")
@app.route("/og.png")
def og_img():
from PIL import Image, ImageDraw
im = Image.new("RGB", (1200, 630), (7, 10, 19))
dr = ImageDraw.Draw(im)
for i in range(260):
import random as _r
_r.seed(i)
x, y = _r.randint(0, 1199), _r.randint(0, 629)
dr.ellipse([x, y, x+2, y+2], fill=(200+i%55, 210, 255))
dr.ellipse([480, 190, 720, 430], outline=(167, 139, 250), width=4)
dr.ellipse([455, 165, 745, 455], outline=(111, 214, 255), width=2)
try:
from PIL import ImageFont
f = ImageFont.truetype("/usr/share/fonts/truetype/dejavu/DejaVuSansMono-Bold.ttf", 84)
f2 = ImageFont.truetype("/usr/share/fonts/truetype/dejavu/DejaVuSansMono.ttf", 26)
except Exception:
f = f2 = None
dr.text((600, 290), "DARK0RBITS", fill=(255, 201, 77), anchor="mm", font=f)
dr.text((600, 390), "no-KYC network toolbox · BTC only · agents welcome", fill=(147, 160, 194), anchor="mm", font=f2)
buf = io.BytesIO(); im.save(buf, "PNG")
return Response(buf.getvalue(), mimetype="image/png")
@app.route("/health")
def health(): return jsonify({"ok": True, "service": "dark0rbits", "version": "2.0"})
# REDIRECT legacy auriga hostname → dark0rbits
@app.before_request
def _dr_legacy_redirect():
host = (request.host or "").lower()
if host.startswith("auriga.") or host == "auriga.thetempleofdoom.com":
return redirect("https://dark0rbits.thetempleofdoom.com" + request.full_path.rstrip("?"), code=301)
return None
# REDACT-REDIRECT
@app.errorhandler(404)
def not_found(e):
if request.path.startswith("/api/"):
return jsonify({"ok": False, "error": "no such endpoint", "path": request.path}), 404
body = """
404 — LOST SIGNAL
This page drifted off the map. The tools are all still here: