Your mailboxes
'
body = f"""
| Address | Expires | Status |
|---|
BURNER MAIL
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)}
{mine}
Type your desired mailbox name, pick a length, pay the invoice — the mailbox activates the moment the payment settles.
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("steg", 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 jsonify({"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("inbox", "login required
")
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("inbox", "not your mailbox
")
mails = con.execute("SELECT * FROM mails WHERE mailbox_id=? ORDER BY id DESC LIMIT 100", (m["id"],)).fetchall()
rows = "".join(f'{esc(x["sender"])} · {time.strftime("%b %d %H:%M", time.localtime(x["ts"]))}
{esc(x["subject"])}{esc(x["body"])}
empty — waiting for mail…
'
left = max(0, m["expires"] - int(time.time()))
return page("steg", 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()
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)}
{mine}
Proxy rentals stay separate (they burn real upstream bandwidth — buy those at the storefront).
API: POST /api/pass/create (days=30|90|365) → invoice. Pass activates on payment settle via webhook.
"""
return page("sms", 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 jsonify({"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("inbox", '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""+esc(k['key'][:14])+"… copyNEW KEY (shown once)
') if newkey else ''
keys_block = (''+esc(newkey)+' Keys
') if rows else ''
body = f"""
| Key | Label | Created |
|---|
API KEYS — balance: ${bal/100:.2f}
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 |
|---|---|---|
| no charges yet | ||
Use it:
"""
return page("inbox", 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"}), 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 jsonify({"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"""
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.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."])
return page("tools", 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)}{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("tools", f"""
VERDICT {esc(d.get('subject') or '(no subject)')}
{kv([("From", esc(d.get('from'))), ("Envelope-from", esc(d.get('return_path'))), ("Reply-To", esc(d.get('reply_to') or '—')), ("Origin IP", origin)])}Authentication
{verdicts}
Spoof flags
{flags}
{verdicts}
Spoof flags
{flags}
Relay chain (origin first){hops or 'no Received headers'}
""")
@app.route("/api/eh", methods=["POST"])
def api_eh():
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."])
return page("steg", 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 gps.items()) or "—"
import base64 as b64mod
ela_b64 = b64mod.b64encode(ela_png).decode()
return page("steg", f"""
FORENSICS {esc(fn)}
{kv([("Verdict", f'{verdict}'), ("ELA max diff", f"{maxdiff} (low=uniform=re-saved clean)"), ("EXIF", f"{len(rows)} tags"), ("Stego", "DARK0RBITS payload present ✓" if stego else "none detected")])}Flags
{'
'.join(esc(x) for x in flags) or 'none'}
{'
'.join(esc(x) for x in flags) or 'none'}
EXIF table{rows_html or '
| no EXIF |
GPS{gps_html}
""")
@app.route("/api/forensics", methods=["POST"])
def api_forensics():
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_auriga": ("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
{canary_list()}
You get: a link (paste anywhere), a pixel URL (embed in docs/pages), and a fake credential line to drop in files.
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("track", 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"{SITE}/c/{c['token']} copyYour traps
'
@app.route("/canary", methods=["POST"])
def canary_create():
uid = current_user_id()
if not uid: return page("track", "| Tag | Link | Pixel | Hits | Status |
|---|---|---|---|---|
| none yet | ||||
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/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("/api/passport")
def api_passport():
uid = current_user_id()
if not uid: return jsonify({"ok": False, "error": "login required"})
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", "INBOX
username + password (4+ chars) required
")
try:
con.execute("INSERT INTO users(username,passhash,created) VALUES(?,?,?)", (u, hash_pw(p), int(time.time())))
con.commit()
except sqlite3.IntegrityError:
return page("inbox", "INBOX
name taken
")
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 ""
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
{"you" if m["sender"]=="user" else esc(m["sender"])} · {time.strftime("%b %d %H:%M", time.localtime(m["created"]))}
{m["body"]}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"INBOX
Site-internal messaging with the operator + your file-open alerts.
Conversation{msgs_html}
Your tracked files
| File | Events | Status | Created |
|---|---|---|---|
| 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,))])
# ---------- 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)
"""
return page("tools", body)
# ---------- 9. OPERATOR CONSOLE ----------
@app.route("/admin", methods=["GET", "POST"])
def admin():
if request.method == "POST" and request.form.get("pw") == ADMIN_PW:
resp = Response(status=302); resp.headers["Location"] = "/admin"
resp.set_cookie("dark0rbits_admin", secrets.token_urlsafe(16), max_age=86400, httponly=True)
return resp
if not request.cookies.get("dark0rbits_admin"):
return page("ip", 'OPERATOR
') 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'{esc(m["username"])} · {time.strftime("%b %d %H:%M", time.localtime(m["created"]))}
{m["body"]}empty
'
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"OPERATOR CONSOLE
All customer messages{msgs_html}
File open events
""")
# ---------- 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."),
("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"""
| File | IP | Device | When |
|---|
$ ./dark0rbits --intro▊
DARK0RBITS — the toolbox that treats you like an operator, not a product.
No KYC. No email required. No Stripe. BTC only. Agents welcome.
{stat}
{cta}
{cards}
NO KYC BTC ONLY AGENT-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("/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
if __name__ == "__main__":
app.run(host="0.0.0.0", port=5000, threaded=True)