#!/usr/bin/env python3 """Dark0rbits v2 — toolbox: IP intel, card validator, SMS rentals, proxy lab, stego lab, trackable files (BTCPay), no-KYC site-only messaging inbox. Single-file Flask + SQLite.""" import base64, binascii, hashlib, hmac, html, io, ipaddress, json, os, re, secrets, shutil, socket, sqlite3, struct, subprocess, time, uuid import urllib.request, urllib.parse from flask import Flask, request, jsonify, render_template_string, Response, send_file from flask import redirect import importlib.util as _ilu _HAVE_AESGCM = _ilu.find_spec("cryptography") is not None app = Flask(__name__) DB_PATH = os.environ.get("DARK0RBITS_DB", "/opt/dark0rbits/dark0rbits.db") UPLOAD_DIR = os.environ.get("DARK0RBITS_UPLOADS", "/opt/dark0rbits/uploads") os.makedirs(UPLOAD_DIR, exist_ok=True) SMSP_KEY = os.environ.get("SMSP_KEY", "") PLEIADES_GW = os.environ.get("PLEIADES_GW", "10.30.20.178:8080") PLEIADES_APP = os.environ.get("PLEIADES_APP", "https://pleiades.thetempleofdoom.com") BTCPAY = "https://10.30.20.140/api/v1" BTCPAY_KEY = os.environ.get("BTCPAY_KEY", "6026288e2e315984661c748baafd509e81a75f22") BTCPAY_STORE = os.environ.get("BTCPAY_STORE", "7h79ndYyZX2yF6CPa12xt2uVGQ5Fd6nrSDG4Koy86x6u") WEBCHECK = os.environ.get("WEBCHECK", "http://10.30.20.13:3000") ADMIN_PW = os.environ.get("DARK0RBITS_ADMIN", "Czapiewski1!") BTCPAY_WHSEC = os.environ.get("BTCPAY_WHSEC", "QnrkV2XPFD3P6ULjMxspHQ") BMAC = "https://buymeacoffee.com/r26xrthzttg" SITE = "https://dark0rbits.thetempleofdoom.com" def _migrate(con): cols = [r[1] for r in con.execute("PRAGMA table_info(settings)")] if not cols: con.execute("CREATE TABLE IF NOT EXISTS settings(user_id INTEGER, key TEXT, val TEXT, PRIMARY KEY(user_id,key))") cols = [r[1] for r in con.execute("PRAGMA table_info(sms_rentals)")] if "user_id" not in cols: con.execute("ALTER TABLE sms_rentals ADD COLUMN user_id INTEGER DEFAULT 0") def db(): con = sqlite3.connect(DB_PATH); con.row_factory = sqlite3.Row con.executescript("""CREATE TABLE IF NOT EXISTS sms_rentals(id INTEGER PRIMARY KEY, phone TEXT, service TEXT, country TEXT, purchase_id TEXT, cost REAL, status TEXT, created INTEGER, expires INTEGER); CREATE TABLE IF NOT EXISTS proxy_checks(id INTEGER PRIMARY KEY, user_key TEXT, egress_ip TEXT, geo TEXT, ok INTEGER, ts INTEGER); CREATE TABLE IF NOT EXISTS users(id INTEGER PRIMARY KEY, username TEXT UNIQUE, passhash TEXT, created INTEGER); CREATE TABLE IF NOT EXISTS sessions(id INTEGER PRIMARY KEY, token TEXT UNIQUE, user_id INTEGER, created INTEGER); CREATE TABLE IF NOT EXISTS trackables(id INTEGER PRIMARY KEY, user_id INTEGER, token TEXT UNIQUE, filename TEXT, kind TEXT, invoice_id TEXT, paid INTEGER DEFAULT 0, created INTEGER); CREATE TABLE IF NOT EXISTS track_events(id INTEGER PRIMARY KEY, trackable_id INTEGER, ts INTEGER, ip TEXT, ua TEXT); CREATE TABLE IF NOT EXISTS messages(id INTEGER PRIMARY KEY, user_id INTEGER, sender TEXT, body TEXT, created INTEGER); CREATE TABLE IF NOT EXISTS mailboxes(id INTEGER PRIMARY KEY, user_id INTEGER, address TEXT UNIQUE, invoice_id TEXT, paid INTEGER DEFAULT 0, expires INTEGER DEFAULT 0, created INTEGER, plan_days INTEGER DEFAULT 7, cnt INTEGER DEFAULT 0); CREATE TABLE IF NOT EXISTS mails(id INTEGER PRIMARY KEY, mailbox_id INTEGER, sender TEXT, subject TEXT, body TEXT, ts INTEGER); CREATE TABLE IF NOT EXISTS passes(id INTEGER PRIMARY KEY, user_id INTEGER, invoice_id TEXT, paid INTEGER DEFAULT 0, expires INTEGER DEFAULT 0, plan_days INTEGER DEFAULT 30); CREATE TABLE IF NOT EXISTS canaries(id INTEGER PRIMARY KEY, user_id INTEGER, token TEXT UNIQUE, tag TEXT, created INTEGER, armed INTEGER DEFAULT 1); CREATE TABLE IF NOT EXISTS canary_hits(id INTEGER PRIMARY KEY, canary_id INTEGER, ts INTEGER, ip TEXT, ua TEXT); CREATE TABLE IF NOT EXISTS balances(user_id INTEGER PRIMARY KEY, cents INTEGER DEFAULT 0); CREATE TABLE IF NOT EXISTS apikeys(id INTEGER PRIMARY KEY, user_id INTEGER, key TEXT UNIQUE, label TEXT, created INTEGER, revoked INTEGER DEFAULT 0); CREATE TABLE IF NOT EXISTS ledger(id INTEGER PRIMARY KEY, user_id INTEGER, delta_cents INTEGER, reason TEXT, ts INTEGER); CREATE TABLE IF NOT EXISTS wh_processed(invoice_id TEXT PRIMARY KEY, ts INTEGER); CREATE TABLE IF NOT EXISTS rate_hits(bucket TEXT, ip TEXT, ts INTEGER); CREATE TABLE IF NOT EXISTS deadrops(id INTEGER PRIMARY KEY, user_id INTEGER, token TEXT UNIQUE, body_enc TEXT, reads_left INTEGER, burn_after INTEGER, expires INTEGER, pw_hash TEXT, created INTEGER); CREATE TABLE IF NOT EXISTS shots(id INTEGER PRIMARY KEY, user_id INTEGER, url TEXT, status TEXT, result TEXT, created INTEGER);""") _migrate(con) return con # ---------- USER SETTINGS (visible tunables, agent-settable) ---------- DEFAULT_SETTINGS = { "bg": "1", "warp": "1", "parallax": "1", "density": "1.0", "speed": "1.0", "twinkle": "1.0", "hue": "0", "grid": "1", "scan": "1", "toast": "1", "type": "1", } BOOL_SETTINGS = {"bg", "warp", "parallax", "grid", "scan", "toast", "type"} RANGE_SETTINGS = {"density": (0, 2.5), "speed": (0, 3), "twinkle": (0, 3), "hue": (-180, 180)} def get_settings(uid): out = dict(DEFAULT_SETTINGS) if not uid: return out con = db() try: for r in con.execute("SELECT key,val FROM settings WHERE user_id=?", (uid,)): if r["key"] in out: out[r["key"]] = r["val"] except Exception: pass return out def set_setting(uid, key, val): if key not in DEFAULT_SETTINGS: return False if key in BOOL_SETTINGS: val = "1" if str(val) in ("1", "true", "on", "yes") else "0" elif key in RANGE_SETTINGS: try: lo, hi = RANGE_SETTINGS[key] val = str(max(lo, min(hi, float(val)))) except Exception: return False con = db() con.execute("INSERT INTO settings(user_id,key,val) VALUES(?,?,?) ON CONFLICT(user_id,key) DO UPDATE SET val=excluded.val", (uid, key, str(val))) con.commit() return True # ---------- BILLING CORE (per-call metering for outside users) ---------- def get_balance(uid): con = db() con.execute("INSERT OR IGNORE INTO balances(user_id, cents) VALUES(?, 100)", (uid,)) # $1 free trial credit con.commit() return con.execute("SELECT cents FROM balances WHERE user_id=?", (uid,)).fetchone()["cents"] def charge(uid, cents, reason): """Deduct from balance; return False if insufficient.""" if cents <= 0: return True if get_balance(uid) < cents: return False con = db() 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, reason, int(time.time()))) con.commit() return True def key_user(): """API-key auth: Authorization: Bearer dk_... → user_id or None.""" auth = request.headers.get("Authorization", "") if not auth.startswith("Bearer dk_"): return None con = db() r = con.execute("SELECT user_id FROM apikeys WHERE key=? AND revoked=0", (auth[7:],)).fetchone() return r["user_id"] if r else None def require_paid_key(cents, reason): """For API calls: key or session auth; metered charge. Returns (uid, error_json).""" uid = key_user() or current_user_id() if not uid: return None, (jsonify({"ok": False, "error": "auth required: account session or Authorization: Bearer dk_… key"}), 401) if has_pass(uid): return uid, None # PASS = unlimited tools (proxy excluded) if not charge(uid, cents, reason): return None, (jsonify({"ok": False, "error": "insufficient balance", "balance_cents": get_balance(uid), "topup": SITE + "/keys"}), 402) return uid, None # ---------- RATE LIMITING ---------- _RL = {} def rate_limit(bucket, limit, window): """Sliding-window per-IP limiter. Returns None if ok, else a 429 response.""" key = request.headers.get("X-Real-IP") or request.remote_addr or "?" now = time.time() con = db() con.execute("DELETE FROM rate_hits WHERE bucket=? AND ts < ?", (bucket, now-window)) n = con.execute("SELECT COUNT(*) c FROM rate_hits WHERE bucket=? AND ip=?", (bucket, key)).fetchone()["c"] if n >= limit: return jsonify({"ok": False, "error": "rate limited — slow down"}), 429 con.execute("INSERT INTO rate_hits(bucket,ip,ts) VALUES(?,?,?)", (bucket, key, now)) con.commit() return None import ssl as _ssl _CTX = _ssl.create_default_context() _CTX.check_hostname = False _CTX.verify_mode = _ssl.CERT_NONE def http(url, headers=None, data=None, method="GET", timeout=12): h = {"User-Agent": "Mozilla/5.0 (Dark0rbits toolbox)"} h.update(headers or {}) req = urllib.request.Request(url, headers=h, data=data, method=method) try: with urllib.request.urlopen(req, timeout=timeout, context=_CTX) as r: return r.status, r.read().decode("utf-8", "replace") except urllib.error.HTTPError as e: return e.code, e.read().decode("utf-8", "replace") except Exception as e: return 0, str(e) def jf(b): try: return json.loads(b) except Exception: return None def param(name): return request.form.get(name) or request.args.get(name) def esc(s): return html.escape(str(s)) # ---------- DEAD-DROP CRYPTO (AES-GCM on CT768, XOR-HMAC stream fallback) ---------- def _dd_master_key(): return hashlib.sha256(("dark0rbits-deaddrop-v1:" + (os.environ.get("DARK0RBITS_SECRET", "ct768-fallback-secret"))).encode()).digest() def dd_encrypt(plaintext): """AES-256-GCM when cryptography is present, else HMAC-verified XOR stream. Returns 'mode:vault' string.""" if _HAVE_AESGCM: from cryptography.hazmat.primitives.ciphers.aead import AESGCM nonce = secrets.token_bytes(12) vault = AESGCM(_dd_master_key()).encrypt(nonce, plaintext.encode(), None) return "aesgcm:" + base64.urlsafe_b64encode(nonce + vault).decode() key = secrets.token_bytes(32) stream = bytes(a ^ b for a, b in zip(plaintext.encode(), hashlib.shake_256(key + str(len(plaintext)).encode()).digest(len(plaintext) + 64))) mac = hmac.new(_dd_master_key(), stream, hashlib.sha256).hexdigest() return "xor:" + base64.urlsafe_b64encode(key + stream).decode() + ":" + mac def dd_decrypt(vault): try: if vault.startswith("aesgcm:"): from cryptography.hazmat.primitives.ciphers.aead import AESGCM raw = base64.urlsafe_b64decode(vault[7:].encode()) return AESGCM(_dd_master_key()).decrypt(raw[:12], raw[12:], None).decode() if vault.startswith("xor:"): k64, mac = vault[4:].rsplit(":", 1) raw = base64.urlsafe_b64decode(k64.encode()) if not hmac.compare_digest(hmac.new(_dd_master_key(), raw[32:], hashlib.sha256).hexdigest(), mac): return None n = len(raw) - 32 - 64 stream = bytes(a ^ b for a, b in zip(raw[32:], hashlib.shake_256(raw[:32] + str(n).encode()).digest(n + 64))) return stream.decode() except Exception: pass return None BASE = """ DARK0RBITS — No-KYC Network Toolbox: IP Intel, Stego, Burner Mail, SMS Rentals, Proxy Lab
{{acct}}

Intel

◈ IP INTEL geo, ASN, ISP, VPN flags — any target ◈ CARD CHECK luhn + BIN issuer intelligence ◈ MAG-LAB browser magstripe studio — ISO 7811 encode, read, batch issue (closed-loop only) ◈ MAIL FORENSICS origin + SPF/DKIM/DMARC + spoof flags ◈ IMAGE FORENSICS EXIF, GPS, ELA, edit detection

Operate

◈ SMS RENTAL 30-min numbers, refundable ◈ PROXY LAB residential egress, geo builder ◈ STEGO LAB hide words in pictures ◈ BURNER MAIL receive-only mailboxes, countdown

Hunt

◈ TRACK FILE opens report back: IP, city, ISP ◈ CANARY TRAPS tripwires with instant alerts ◈ DEAD-DROP burn-after-read encrypted notes ◈ SCREENSHOT page capture or rendered-text fallback ◈ FRAUD-SCORE composite IP + email + BIN risk 0-100 ◈ FREE TOOLS DNS, headers, JWT, hasher

Account

◈ INBOX no-KYC messaging ◈ SIGN UP username + password, 10 seconds, no KYC ◈ API KEYS account, balance, metered keys ◈ PASS $10/mo all-access ◈ AGENT PASSPORT machine-readable badge
{{body}}
✦ REACH THE DEV
⚙
""" NEBULAS = { "home": ("rgba(120,85,255,.17)", "rgba(0,190,255,.10)"), "ip": ("rgba(255,170,60,.13)", "rgba(120,85,255,.10)"), "card": ("rgba(66,232,164,.10)", "rgba(0,190,255,.09)"), "sms": ("rgba(0,190,255,.13)", "rgba(167,139,250,.10)"), "proxy": ("rgba(167,139,250,.14)", "rgba(255,170,60,.08)"), "steg": ("rgba(255,110,180,.11)", "rgba(120,85,255,.11)"), "track": ("rgba(255,90,90,.11)", "rgba(255,170,60,.08)"), "mail": ("rgba(66,232,164,.10)", "rgba(0,190,255,.08)"), "forensics": ("rgba(0,210,255,.12)", "rgba(255,110,180,.07)"), "canary": ("rgba(255,201,77,.12)", "rgba(255,90,90,.08)"), "deaddrop": ("rgba(45,226,200,.13)", "rgba(160,225,255,.09)"), "shot": ("rgba(111,214,255,.12)", "rgba(120,85,255,.10)"), "score": ("rgba(255,110,180,.10)", "rgba(66,232,164,.10)"), } def kv(pairs): rows = "".join(f"
{k}
{v}
" for k, v in pairs) return '
' + rows + "
" 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"]) + "" + ' EXIT') except Exception: acct = "" if not uid or not acct: nxt = ("?next=" + urllib.parse.quote(request.path)) if request.path not in ("/", "/inbox", "/signup", "/logout") else "" acct = ('LOG IN' ' SIGN UP') from markupsafe import Markup st = get_settings(uid) return render_template_string(BASE, body=Markup(body), bmac=BMAC, o=lambda s2: "on" if s2 == sec else "", n1=n1, n2=n2, acct=Markup(acct), CFG_JS="window.DRB=" + json.dumps(st) + ";") 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 agent_card(endpoint, example, notes): """Interactive-for-LLMs card: exact curl + auth + link to openapi.""" return ('
FOR AGENTS ' '?
' '' + esc(endpoint) + '
' '' + esc(example) + '
' + esc(notes) + ' · spec: /openapi.json · catalog: /llms.txt
') 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 ---------- LLMS_SETTINGS = """ ## ACCOUNT TUNABLES (machine-settable) GET/POST /api/settings — keys: bg, warp, parallax, density (0-2.5), speed (0-3), twinkle (0-3), hue (-180-180), grid, scan, toast, type (1|0). Agents driving browsers (or building clients) can persist a theme per API key: POST form-encoded key=value. Values validated server-side. ## KEYBOARD Ctrl+K / Cmd+K — command palette on any page. Type tool name, Enter navigates. """ 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": "/signup", "desc": "No-KYC signup page (humans): username + password, 4+ chars, ~10 seconds. Agents: POST /inbox form act=register&u=NAME&p=PASS -> session cookie dark0rbits_tok (30 days) + $1 free trial credit."}, {"method": "GET/POST", "path": "/api/settings", "desc": "Per-account UI tunables (bg, warp, parallax, density, speed, twinkle, hue, grid, scan, toast, type). Agents can theme their own client. GET returns current; POST form key=val applies (validated + clamped)."}, {"method": "GET", "path": "/deaddrop", "desc": "Burn-after-read encrypted notes. POST /api/deaddrop/create (body, burn_after 1-10, ttl_hours 1-72, password optional) -> token. 5c, free with PASS."}, {"method": "GET", "path": "/shot", "desc": "Page capture: POST /api/shot/create {url} then GET /api/shot/status/. SSRF-guarded. 25c, free with PASS."}, {"method": "GET", "path": "/score", "desc": "Composite fraud score: IP 45% + disposable-email 25% + BIN 30%. 2c, free with PASS."}, {"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/settings", methods=["GET", "POST"]) def api_settings(): uid = key_user() or current_user_id() if not uid: return jsonify({"ok": False, "error": "auth required: account session or API key"}), 401 if request.method == "GET": return jsonify({"ok": True, "settings": get_settings(uid)}) out = {} for k in DEFAULT_SETTINGS: if k in request.form: out[k] = set_setting(uid, k, request.form[k]) return jsonify({"ok": True, "applied": out, "settings": get_settings(uid)}) @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", "maglab", "tools", "signup"] 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{LLMS_SETTINGS}", 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": "makemoneys8@proton.me"}) @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"}) add("/signup", "get", "No-KYC signup page (username + password only)") 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"" for k, v in hdrs.items()) + "
    {esc(k)}{esc(v)}
    " 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")]) body += agent_card('GET /api/ip?target=1.2.3.4', 'curl "https://dark0rbits.thetempleofdoom.com/api/ip?target=1.2.3.4" -H "Authorization: Bearer drb_..."', 'Auto-detects caller IP if target omitted.') 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))])}
    Fraud & structure flags
    {' '.join(tags)}{'
    ⚠ ' + ' · '.join(flags) if flags else ''}
    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"""

    CARD CHECK

    Luhn + BIN intelligence: issuer, brand, type, country, prepaid risk flags.

    Paste anything — auto-formats. Nothing stored.
    {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")]) body += agent_card('POST /api/card num=4539148803436467', 'curl -X POST https://dark0rbits.thetempleofdoom.com/api/card -d num=4539148803436467', 'Luhn + BIN intel. 2c/metered with API key, free with PASS.') 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)[^>]*>.*?", " ", 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
    PNG mode: status "done" + png_b64. If chromium vanishes, 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 real 1280×1600 headless-Chromium PNG render (base64 in the API). Honest text+intel fallback only if the renderer is down. 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.'}
    {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"+{h['phone']} copy{h['service']}{h['status']}#{h['purchase_id']}…" for h in hist) body = f"""

    SMS RENTAL

    Disposable numbers, 30-minute windows. Cancel before a code = full refund.

    Rent a number
    Check / manage
    {msg}
    Recent rentals{hist_rows or ''}
    NumberServiceStatusOrderWindow
    none yet
    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'
    PROXY LIVE Egress: {esc(j.get("query"))} — {esc(j.get("country"))} / {esc(j.get("city"))} · ISP {esc(j.get("isp"))} · tz {esc(j.get("timezone"))}
    ' 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.





    {result}
    Geo session builder:
    yourpassword
    Rent more — storefront: {PLEIADES_APP}
    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 = 96 + ln * 8 # header is 12 bytes (AUR1+len+phash) = 96 bits; old 64 truncated 4 bytes off every payload 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
    📤 drop a PNG here or click
    Extract text
    📥 drop the carrier PNG
    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")]) body += agent_card('POST /api/steg/hide image= text=hi [password= bits= spread=]', 'curl -X POST https://dark0rbits.thetempleofdoom.com/api/steg/hide -F image=@x.png -F text=hi -F password=hunter2', 'Extract: POST /api/steg/extract. Free with PASS.') 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"{esc(t['filename'])}{'events' if t['paid'] else '—'}{'paid ✓' if t['paid'] else 'unpaid'}" for t in rows) mine = f'
    Your trackables{trs}
    FileEventsStatus
    ' body = f"""

    TRACK FILE

    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) BTCPAY_PUBLIC = "https://btcpay.thetempleofdoom.com" def public_checkout(link): """LAN invoices must be payable from the open internet — swap host on checkout links.""" if not link: return link return link.replace("https://10.30.20.140", BTCPAY_PUBLIC) 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": public_checkout(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.

    📤 drop file / picture here
    """) @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'

    📎 {esc(fn)} ({len(data)} bytes)

    Open / download the file

    ' 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"{esc(m['address'])} copyview mail…{'live' if m['paid']==1 else 'expired'}{m['cnt']}" for m in rows) mine = f'
    Your mailboxes{trs}
    AddressExpiresStatusMail
    ' body = f"""

    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)}
    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": public_checkout(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", '
    login required — sign in / create account
    ') 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'
    {esc(x["sender"])} · {time.strftime("%b %d %H:%M", time.localtime(x["ts"]))}
    {esc(x["subject"])}
    {esc(x["body"])}
    ' for x in mails) or '
    empty — waiting for mail…
    ' 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": public_checkout(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"{'$%.2f' % (l['delta_cents']/100)}{esc(l['reason'])}{time.strftime('%b %d %H:%M', time.localtime(l['ts']))}" for l in led) keys_html = "".join(""+esc(k['key'][:14])+"… copy"+esc(k['label'])+""+time.strftime('%b %d', time.localtime(k['created']))+"" for k in rows) newkey_block = ('
    NEW KEY (shown once)
    '+esc(newkey)+'
    ') if newkey else '' keys_block = ('
    Keys'+keys_html+'
    KeyLabelCreated
    ') if rows else '' body = f"""

    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{led_html or ''}
    ΔReasonWhen
    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": public_checkout(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() or [""])[-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")]) body += agent_card('POST /api/eh raw=', 'curl -X POST https://dark0rbits.thetempleofdoom.com/api/eh --data-urlencode raw@headers.txt', 'Returns origin IP, hop chain, SPF/DKIM/DMARC verdicts, spoof flags.') 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"""

    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}
    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.

    🖼 drop an image
    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")]) body += agent_card('POST /api/forensics image=', 'curl -X POST https://dark0rbits.thetempleofdoom.com/api/forensics -F image=@img.jpg', 'EXIF dump, GPS, ELA score, editor flags.') 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"""

    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'}
    ELA (amplified 15×)
    full PNG
    EXIF table{rows_html or ''}
    no EXIF
    GPS{gps_html}
    """) @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"{esc(c['tag'])}{SITE}/c/{c['token']} copy{SITE}/c/{c['token']}.png{hits}{'armed' if c['armed'] else 'triggered ⚠'}" return f'
    Your traps{trs or ""}
    TagLinkPixelHitsStatus
    none yet
    ' @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", "

    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() nxt = request.form.get("next") or "/inbox" if not nxt.startswith("/") or nxt.startswith("//"): nxt = "/inbox" resp = Response(status=302); resp.headers["Location"] = nxt; 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() nxt = request.form.get("next") or "/inbox" if not nxt.startswith("/") or nxt.startswith("//"): nxt = "/inbox" resp = Response(status=302); resp.headers["Location"] = nxt; 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"{esc(f['filename'])}{'events' if f['paid'] else '—'}{'paid ✓' if f['paid'] else 'unpaid'}{time.strftime('%b %d', time.localtime(f['created']))}" for f in files) body = f"""

    INBOX

    Site-internal messaging with the operator + your file-open alerts.

    Conversation{msgs_html}
    Your tracked files{files_html or ''}
    FileEventsStatusCreated
    none yet
    API: (cookie auth) POST /inbox act=send body=… · GET /api/inbox/messages
    """ return page("inbox", body) @app.route("/signup", methods=["GET"]) def signup(): if current_user_id(): return Redirect("/keys") nxt = esc(request.args.get("next") or "") body = f"""

    SIGN UP

    Name + password. That's the whole form — no email, no phone, no KYC, nothing to verify. $1 free credit lands in your balance the moment you're in.

    Create account — 10 seconds
    What you get immediately
    • ◈ $1 free trial credit — metered API calls work instantly
    • ▣ API keys — machine access on the same balance
    • ✉ Inbox — no-KYC messaging with the operator + file-open alerts
    • ⚙ Tunables — your theme follows your account
    Already have an account? Log in →
    """ return page("signup", body) @app.route("/logout", methods=["GET"]) def logout(): con = db() con.execute("DELETE FROM sessions WHERE token=?", (request.cookies.get("dark0rbits_tok"),)); con.commit() resp = Response(status=302); resp.headers["Location"] = "/" resp.set_cookie("dark0rbits_tok", "", max_age=0) return resp @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("/dead-drop") def deaddrop_alias(): from flask import redirect return redirect("/deaddrop", 301) @app.route("/burner-mail") def burnermail_alias(): from flask import redirect return redirect("/mail", 301) @app.route("/fraud-score") def fraudscore_alias(): from flask import redirect return redirect("/score", 301) @app.route("/mag-lab") def maglab_alias(): from flask import redirect return redirect("/maglab", 301) @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'/drop/{r["token"][:10]}…{r["reads_left"]}/{r["burn_after"]}' for r in rows) mine = f'
    Your drops{trs}
    LinkReadsSelf-destructs
    ' body = f"""

    DEAD DROP

    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."]) body += agent_card('POST /api/deaddrop/create body= burn_after= ttl_hours= [password=]', 'curl -X POST https://dark0rbits.thetempleofdoom.com/api/deaddrop/create -d body=secret -d burn_after=1 -d ttl_hours=24', 'Returns /drop/. Reader destroys the note at the last read.') 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 + '
    GONE burned, 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 EXPIRED the 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 DESTROYED the 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"""
    SCORE: {r['score']}/100 — {r['band']} RISK confidence {r['confidence']}% {''.join(f"" for c in r['components'])}
    ComponentScoreFactors
    {esc(c['weight'])}{esc(c['score'])}{esc('; '.join(c['factors']))}
    """ body = f"""

    FRAUD SCORE

    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."]) body += agent_card('GET /api/score?ip=&email=&bin=', 'curl "https://dark0rbits.thetempleofdoom.com/api/score?ip=1.2.3.4&email=a@mailinator.com&bin=453914" -H "Authorization: Bearer drb_..."', 'Weighted composite; re-normalizes on partial input.') 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)
    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
    """ 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("track", '

    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"]}
    ' for m in msgs) or '
    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"{esc(o['filename'])}{esc(o['ip'])}{esc(o['ua'][:50])}{time.strftime('%b %d %H:%M', time.localtime(o['ts']))}" 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{opens_html}
    FileIPDeviceWhen
    """) # ---------- 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.","◈","INTEL"), ("card","CARD CHECK","Luhn + BIN: issuer bank, brand, type, country, prepaid risk flags. Nothing stored, nothing charged.","◈","INTEL"), ("eh","MAIL FORENSICS","Paste raw headers → real origin IP + geo, SPF/DKIM/DMARC verdicts, spoof flags.","◈","INTEL"), ("forensics","IMAGE FORENSICS","EXIF, GPS, edit-tool detection, error-level analysis — expose doctored photos.","◈","INTEL"), ("sms","SMS RENTAL","Disposable numbers, 30-min windows, instant refund on cancel.","◈","ACQUIRE"), ("mail","BURNER MAIL","Receive-only mailboxes, 7–90 days, live countdown. Codes & confirmations without an identity.","◈","ACQUIRE"), ("proxy","PROXY LAB","Residential egress testing on the Pleiades rail — same gateway keys fleet-wide.","◈","ACQUIRE"), ("deaddrop","DEAD-DROP","AES-GCM encrypted notes that burn after N reads or TTL. Optional password. No trace left.","◈","ACQUIRE"), ("steg","STEGO LAB","Hide words inside pictures. LSB depth, randomized spread, password-encrypted payloads.","◈","OPERATE"), ("track","TRACK FILE","$1 → tracked link + email pixel. Every open reports back: IP, location, ISP, device.","◈","HUNT"), ("canary","CANARY TRAPS","Tripwire links and pixels — instant alert the moment anyone touches one.","◈","HUNT"), ("shot","SCREENSHOT","Headless-Chromium PNG capture of any page. Agents: poll the status API.","◈","HUNT"), ("score","FRAUD-SCORE","Composite 0-100 risk: IP intel + disposable-email + BIN heuristics, with full breakdown.","◈","HUNT"), ("tools","FREE TOOLS","DNS resolver, HTTP header inspector, JWT decoder, hasher, generators.","◈","UTILITY"), ("passport","AGENT PASSPORT","Machine-readable trust badge for your bots. Agents are first-class here.","◈","ACCOUNT"), ] tcards = "".join( f'
    {ico}

    {name}

    {cat}
    ' f'

    {desc}

    ' for href, name, desc, ico, cat in tools) stat = f"you're connecting from {esc(d.get('query','?'))} · {esc(d.get('country',''))}" cta = (' ' if uid else ' ') from markupsafe import escape as _e stat_line = '▸ ' + stat + '' if stat else "" body = f"""
    $ ./dark0rbits --intro ▊

    The toolbox that treats you like an operator, not a product.

    No KYC. No email. No Stripe — BTC only. Every tool has a JSON API, metered per call, so scripts and agents are first-class customers.

    {stat_line}
    {cta}
    {tcards}
    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("/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:
    """ return page("home", body), 404 # ---------- MAG-LAB (browser magstripe studio, closed-loop only) ---------- _MAG_POLICY = ( "MAG-LAB encodes CLOSED-LOOP cards only: your own gift, loyalty, membership, " "event or staff cards. Payment-network tracks (bank/debit/credit layouts, " "13-19 digit Luhn-valid PANs, bank service codes 101/121/201-220, EMV/JCOP " "dumps) are refused at encode AND decode. This is a hard policy, not a " "toggle you can switch off.") def _mag_sentinel(track): """True if a track looks like a payment-network card rather than closed-loop.""" import re as _re t = (track or "").strip() if not t: return False body = t[1:] if t[0] in "%;" else t pan = _re.sub(r"[^0-9]", "", body.split("^")[0] if "^" in body else (body.split("=")[0] if "=" in body else body)) if pan and 13 <= len(pan) <= 19: s, alt = 0, False for ch in reversed(pan): d = ord(ch) - 48 if alt: d *= 2 if d > 9: d -= 9 s += d alt = not alt if s % 10 == 0: return True m = _re.search(r"\^([0-9]{4})([0-9]{3})", t) if m and m.group(2)[0] in ("1", "2"): return True m = _re.search(r"=([0-9]{4})([0-9]{3})", t) if m and m.group(2)[0] in ("1", "2"): return True return False _MAGLAB_HTML = r"""

    MAG LAB

    Browser magstripe studio — Chrome + Web Serial talks straight to your MSR605/606-class writer. No drivers, no desktop app, any OS. Encode, read, decode, batch-issue closed-loop cards: gift, loyalty, membership, event tickets, staff badges.

    POLICY
    __POLICY__
    CONNECT
    not connected
    Chrome/Edge/Opera on Windows, macOS, Linux or ChromeOS. Firefox/Safari do not ship Web Serial. Writer must be an MSR605/606 or compatible serial MagStripe encoder.
    ISSUE A CARD
    BATCH ISSUE (20c/card — paste CSV lines: label,value)
    TRACK VISUALIZER
    paste any track to see its structure decoded (read-only field map; payment shapes are masked)
    
    
    API (agent-first — 30c/encode, 20c/card batch, via key) __AGENT__
    """ @app.route("/maglab") def maglab(): body = _MAGLAB_HTML.replace("__POLICY__", _MAG_POLICY) body = body.replace("__AGENT__", agent_card( "POST /api/maglab/encode {t1,t2} · POST /api/maglab/batch {rows:[{label,value}]}", "curl -X POST " + SITE + "/api/maglab/encode -H 'Authorization: Bearer KEY' -d '{\"t2\":\";GIFT000123=4321?\"}'", "Closed-loop magstripe studio. Encode validates + LRC-checks (30c), batch issues up to 100 cards with PINs + QR twins (20c/card). Refuses payment-card shapes.")) body += gloss([("ISO 7811", "the magstripe track format standard - track1 79-bit alnum, track2/3 5-bit numeric"), ("LRC", "longitudinal redundancy check - the trailing ? sentinel; wrong LRC = unreadable card"), ("closed-loop", "a card scheme you own end-to-end: your shop issues it, your shop redeems it")]) return page("maglab", body) @app.route("/api/maglab/encode", methods=["POST"]) def api_maglab_encode(): r = rate_limit("maglab", 20, 60) if r: return r uid = key_user() or current_user_id() if not uid: return jsonify({"ok": False, "error": "auth required: account session (sign up at /inbox, no KYC) or Authorization: Bearer *** key"}), 401 if not has_pass(uid) and not charge(uid, 30, "maglab encode"): return jsonify({"ok": False, "error": "insufficient balance", "balance_cents": get_balance(uid), "topup": SITE + "/keys"}), 402 t1 = str(jp("t1") or "").strip() t2 = str(jp("t2") or "").strip() if not t1 and not t2: return jsonify({"ok": False, "error": "at least one track required"}), 400 for t in (t1, t2): if t and _mag_sentinel(t): return jsonify({"ok": False, "error": "refused: payment-network card shape detected - " + _MAG_POLICY[:120]}), 403 for name, t in (("t1", t1), ("t2", t2)): if t and not t.endswith("?"): return jsonify({"ok": False, "error": name + " missing terminator '?' (LRC sentinel)"}), 400 return jsonify({"ok": True, "ready": True, "t1": t1, "t2": t2}) @app.route("/api/maglab/batch", methods=["POST"]) def api_maglab_batch(): r = rate_limit("maglab_batch", 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 *** key"}), 401 rows = jp("rows") if not isinstance(rows, list) or not rows: return jsonify({"ok": False, "error": "rows: array of {label,value} required"}), 400 if len(rows) > 100: return jsonify({"ok": False, "error": "max 100 cards per batch"}), 400 pas = has_pass(uid) if not pas and not charge(uid, 20 * len(rows), "maglab batch x%d" % len(rows)): return jsonify({"ok": False, "error": "insufficient balance", "balance_cents": get_balance(uid), "topup": SITE + "/keys"}), 402 made = [] for i, row in enumerate(rows): label = str(row.get("label") or ("CARD%03d" % (i + 1)))[:40] value = str(row.get("value") or "").strip() if not value: return jsonify({"ok": False, "error": "row %d: value required" % (i + 1)}), 400 pin = "".join(str(random.randrange(10)) for _ in range(6)) t2 = ";" + re.sub(r"[^A-Z0-9]", "", value.upper()) + "=" + pin + "?" if _mag_sentinel(t2): return jsonify({"ok": False, "error": "row %d: refused, payment-card shape" % (i + 1)}), 403 made.append({"label": label, "t2": t2, "pin": pin, "qr": SITE + "/card?card=" + label + ":" + pin}) return jsonify({"ok": True, "count": len(made), "cards": made, "cost_cents": 0 if pas else 20 * len(rows)}) @app.route("/api/maglab/decode", methods=["POST"]) def api_maglab_decode(): r = rate_limit("maglab_dec", 40, 60) if r: return r t = str(jp("track") or "").strip() if not t: return jsonify({"ok": False, "error": "track required"}), 400 if _mag_sentinel(t): return jsonify({"ok": False, "masked": True, "error": "payment-card shape - fields masked by policy"}), 200 t1 = t.startswith("%") sep = "^" if t1 else "=" body = t[1:] if t[0] in "%;" else t fields = body.rstrip("?").split(sep) return jsonify({"ok": True, "track": 1 if t1 else 2, "format": "ISO7811-A" if t1 else "ISO7811-B", "fields": [f[:40] for f in fields], "lrc_sentinel": body.endswith("?"), "note": "closed-loop decode"}) if __name__ == "__main__": app.run(host="0.0.0.0", port=5000, threaded=True)