#!/usr/bin/env python3 """ LYNX — the sharp-eyed inspector. No-KYC, Bitcoin-paid API for file inspection, steganography, forensics, and recon. Powered by local tooling (file/steg/forensics) + Kali (isolated recon egress). """ import os, io, re, json, time, math, uuid, hashlib, secrets, shutil, subprocess, tempfile, base64 import sqlite3, threading, collections from functools import wraps from flask import Flask, request, jsonify, send_from_directory, Response, g import urllib.request, urllib.error, ssl # ---------------------------------------------------------------------------- # Config # ---------------------------------------------------------------------------- PORT = int(os.environ.get("LYNX_PORT", 5059)) DB_PATH = os.environ.get("LYNX_DB", "/opt/lynx/lynx.db") WORK_DIR = "/opt/lynx/work" PUBLIC_DOMAIN = os.environ.get("LYNX_DOMAIN", "lynx.thetempleofdoom.com") # Kali recon engine (isolated red-team segment) KALI_API = os.environ.get("KALI_API", "http://10.30.30.177:5000") # BTCPay BTCPAY_URL = os.environ.get("BTCPAY_URL", "https://10.30.20.140") BTCPAY_KEY = os.environ.get("BTCPAY_KEY", "b1d5016e3b2197882c0671cefe080ccf7c2ebb2e") BTCPAY_STORE = os.environ.get("BTCPAY_STORE", "2riBfb6pMUnSKsmB2RwokExz37f2spyghE9WmWsn2iWo") BTCPAY_WEBHOOK_SECRET = os.environ.get("BTCPAY_WSEC", "BHhQC4ZfBnbQemYqSoksEA") BTCPAY_PUBLIC = os.environ.get("BTCPAY_PUBLIC", "https://btcpay.thetempleofdoom.com") ADMIN_KEY = os.environ.get("LYNX_ADMIN_KEY", "sk-lynx-admin-" + secrets.token_hex(12)) # Pricing (USD -> credits) TIERS = {2: 20, 5: 60, 10: 150, 20: 400} # Tool cost (credits) COSTS = { "inspect/file": 1, "steg/extract": 2, "steg/crack": 4, "forensics/disk": 3, "recon/nmap": 3, "recon/subfinder": 2, "recon/nuclei": 4, "recon/theharvester": 2, "recon/dnsrecon": 2, } MAX_UPLOAD = 20 * 1024 * 1024 # 20MB # ---------------------------------------------------------------------------- # DB # ---------------------------------------------------------------------------- def db(): c = getattr(g, "_db", None) if c is None: c = g._db = sqlite3.connect(DB_PATH, timeout=15) c.row_factory = sqlite3.Row return c def init_db(): os.makedirs(os.path.dirname(DB_PATH), exist_ok=True) c = sqlite3.connect(DB_PATH, timeout=15) c.executescript(""" CREATE TABLE IF NOT EXISTS users ( id INTEGER PRIMARY KEY AUTOINCREMENT, email TEXT UNIQUE NOT NULL, api_key TEXT UNIQUE NOT NULL, credits INTEGER DEFAULT 0, total_calls INTEGER DEFAULT 0, is_admin INTEGER DEFAULT 0, created_at INTEGER, last_used_at INTEGER ); CREATE TABLE IF NOT EXISTS orders ( id INTEGER PRIMARY KEY AUTOINCREMENT, order_id TEXT UNIQUE NOT NULL, api_key TEXT, invoice_id TEXT, credits INTEGER, status TEXT DEFAULT 'pending', created_at INTEGER, settled_at INTEGER ); CREATE TABLE IF NOT EXISTS usage_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, api_key TEXT, tool TEXT, target TEXT, credits INTEGER, created_at INTEGER ); """) c.commit() c.close() # ---------------------------------------------------------------------------- # Helpers # ---------------------------------------------------------------------------- TOOL_PATHS = {} def resolve_tools(): """Resolve tool binary paths (some pip/gem tools land in /usr/local/bin).""" for name, candidates in { "steghide": ["steghide"], "binwalk": ["binwalk"], "foremost": ["foremost"], "exiftool": ["exiftool"], "ssdeep": ["ssdeep"], "strings": ["strings"], "file": ["file"], "mmls": ["mmls"], "fls": ["fls"], "fsstat": ["fsstat"], "tsk_recover": ["tsk_recover"], "stegcracker": ["stegcracker"], "zsteg": ["zsteg"], "olevba": ["olevba"], "pdftotext": ["pdftotext"], }.items(): for cand in candidates: p = shutil.which(cand) or (("/usr/local/bin/" + cand) if os.path.exists("/usr/local/bin/" + cand) else None) if p: TOOL_PATHS[name] = p break def run(cmd, timeout=120): try: r = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout) return r.returncode, r.stdout, r.stderr except subprocess.TimeoutExpired: return -1, "", "timeout" except Exception as e: return -1, "", str(e) def sha256_file(p): h = hashlib.sha256() with open(p, "rb") as f: for chunk in iter(lambda: f.read(65536), b""): h.update(chunk) return h.hexdigest() def file_hashes(p): md5 = hashlib.md5(); sha1 = hashlib.sha1(); sha256 = hashlib.sha256() with open(p, "rb") as f: for chunk in iter(lambda: f.read(65536), b""): md5.update(chunk); sha1.update(chunk); sha256.update(chunk) return md5.hexdigest(), sha1.hexdigest(), sha256.hexdigest() def entropy(p): freq = [0] * 256 total = 0 with open(p, "rb") as f: for chunk in iter(lambda: f.read(65536), b""): for b in chunk: freq[b] += 1; total += 1 if total == 0: return 0.0 e = 0.0 for c in freq: if c: px = c / total e -= px * math.log2(px) return round(e, 4) def ssdeep_hash(p): rc, out, err = run([TOOL_PATHS.get("ssdeep", "ssdeep"), "-b", p], timeout=60) m = re.search(r"\d+:[A-Za-z0-9+/]+:[A-Za-z0-9+/]+", out) return m.group(0) if m else None def new_key(): return "sk-lynx-" + secrets.token_hex(24) def get_key(): return request.headers.get("X-API-Key") or request.args.get("api_key") or "" def auth(gate=True): key = get_key() if not key: return None, (jsonify({"error": "API key required. Sign up at /api/signup"}), 401) row = db().execute("SELECT * FROM users WHERE api_key=?", (key,)).fetchone() if not row: return None, (jsonify({"error": "Invalid API key"}), 401) if gate and not row["is_admin"] and row["credits"] <= 0: return None, (jsonify({"error": "No credits. Buy at /api/order"}), 402) return row, None _rate = collections.defaultdict(list) def rate_limited(ident, limit=10, window=60): now = time.time() _rate[ident] = [t for t in _rate[ident] if now - t < window] if len(_rate[ident]) >= limit: return True _rate[ident].append(now) return False def log_usage(key, tool, target, credits): db().execute("INSERT INTO usage_log (api_key, tool, target, credits, created_at) VALUES (?,?,?,?,?)", (key, tool, str(target)[:200], credits, int(time.time()))) db().execute("UPDATE users SET total_calls=total_calls+1, credits=credits-?, last_used_at=? WHERE api_key=?", (credits, int(time.time()), key)) db().commit() # ---------------------------------------------------------------------------- # Kali recon dispatch (isolated egress) # ---------------------------------------------------------------------------- _btc_ctx = ssl.create_default_context(); _btc_ctx.check_hostname = False; _btc_ctx.verify_mode = ssl.CERT_NONE def kali_call(tool, params, timeout=300): req = urllib.request.Request(KALI_API + "/api/tools/" + tool, data=json.dumps(params).encode(), headers={"Content-Type": "application/json"}) try: resp = urllib.request.urlopen(req, timeout=timeout) return json.loads(resp.read().decode()) except urllib.error.HTTPError as e: return {"error": f"kali http {e.code}", "body": e.read().decode()[:500]} except Exception as e: return {"error": f"kali unreachable: {e}"} SAFE_TARGET = re.compile(r"^[A-Za-z0-9.\-_:/]+$") def clean_target(t): if not t or not SAFE_TARGET.match(t) or len(t) > 200: return None return t # ---------------------------------------------------------------------------- # File analysis # ---------------------------------------------------------------------------- def analyze_file(path, name): """Full inspection report for a file.""" md5, sha1, sha256 = file_hashes(path) size = os.path.getsize(path) rep = { "file": name, "size": size, "hashes": {"md5": md5, "sha1": sha1, "sha256": sha256}, "ssdeep": ssdeep_hash(path), "entropy": entropy(path), } rc, out, _ = run([TOOL_PATHS.get("file", "file"), "-b", path]) rep["type"] = out.strip() if rc == 0 else None rc, out, _ = run([TOOL_PATHS.get("strings", "strings"), "-n", "6", path]) rep["strings"] = [s for s in out.splitlines() if s.strip()][:40] if rc == 0 else [] # exiftool if "exiftool" in TOOL_PATHS: rc, out, _ = run([TOOL_PATHS["exiftool"], "-j", path], timeout=90) if rc == 0: try: rep["metadata"] = json.loads(out)[0] if out.strip() else {} except Exception: rep["metadata"] = {"raw": out[:2000]} # binwalk signatures if "binwalk" in TOOL_PATHS: rc, out, _ = run([TOOL_PATHS["binwalk"], "--signature", "--quiet", path], timeout=120) if rc == 0 and out.strip(): rep["embedded_signatures"] = [l.strip() for l in out.splitlines() if l.strip()][:50] # PE analysis try: import pefile pe = pefile.PE(path) rep["pe"] = { "machine": hex(pe.FILE_HEADER.Machine), "sections": len(pe.sections), "imports": len(getattr(pe, "DIRECTORY_ENTRY_IMPORT", []) or []), "compile_time": pe.FILE_HEADER.TimeDateStamp, "is_packed": bool(pe.sections and any(getattr(s, "SizeOfRawData", 0) == 0 for s in pe.sections)), } pe.close() except Exception: pass # OLE / Office macro scan if "olevba" in TOOL_PATHS and name.lower().split(".")[-1] in ("doc", "docx", "xls", "xlsx", "ppt", "pptx", "docm", "xlsm", "pptm"): rc, out, _ = run([TOOL_PATHS["olevba"], path], timeout=90) if rc == 0 and out.strip(): rep["office_macros"] = out[:4000] # PDF analysis if name.lower().endswith(".pdf"): try: with open(path, "rb") as f: raw = f.read() rep["pdf"] = { "has_js": b"/JavaScript" in raw or b"/JS" in raw, "has_openaction": b"/OpenAction" in raw, "has_launch": b"/Launch" in raw, "has_embedded_file": b"/EmbeddedFile" in raw, "pages": raw.count(b"/Type /Page") or None, } except Exception: pass return rep def steg_extract(path, name, passphrase=""): """steghide + zsteg extraction.""" out = {} if "steghide" in TOOL_PATHS: ext = os.path.join(os.path.dirname(path), "extracted.bin") cmd = [TOOL_PATHS["steghide"], "extract", "-sf", path, "-xf", ext, "-p", passphrase or "", "-f"] rc, so, se = run(cmd, timeout=90) if rc == 0: with open(ext, "rb") as f: data = f.read() out["steghide"] = {"extracted_bytes": len(data), "content": data.decode("utf-8", "replace")[:2000]} os.remove(ext) else: out["steghide"] = {"error": (se or so).strip()[:500] or "no embedded data / wrong passphrase"} if "zsteg" in TOOL_PATHS: rc, so, se = run([TOOL_PATHS["zsteg"], path], timeout=90) out["zsteg"] = {"result": (so or se).strip()[:4000]} if rc == 0 else {"error": (se or "no LSB data").strip()[:400]} return out STEG_WORDLIST = "/opt/lynx/wordlist.txt" def steg_crack(path, name): if not os.path.exists(STEG_WORDLIST): return {"error": "wordlist missing"} rc, so, se = run([TOOL_PATHS.get("stegcracker", "stegcracker"), path, STEG_WORDLIST], timeout=600) return {"result": (so or se).strip()[:4000]} def forensics_disk(path, name): """Sleuth Kit analysis of a disk image.""" out = {} if "mmls" in TOOL_PATHS: rc, so, se = run([TOOL_PATHS["mmls"], path], timeout=90) out["partitions"] = (so or se).strip()[:2000] if rc == 0 else {"error": se.strip()[:300]} if "fsstat" in TOOL_PATHS: rc, so, se = run([TOOL_PATHS["fsstat"], path], timeout=90) out["fsstat"] = (so or se).strip()[:3000] if rc == 0 else {"error": se.strip()[:300]} if "fls" in TOOL_PATHS: rc, so, se = run([TOOL_PATHS["fls"], "-r", path], timeout=120) out["file_listing"] = [l for l in (so or "").splitlines()][:200] if rc == 0 else {"error": se.strip()[:300]} return out # ---------------------------------------------------------------------------- # App # ---------------------------------------------------------------------------- app = Flask(__name__) app.config["MAX_CONTENT_LENGTH"] = MAX_UPLOAD + 1024 * 1024 @app.teardown_appcontext def close_db(exc): c = getattr(g, "_db", None) if c is not None: c.close() def save_upload(): if "file" not in request.files: return None, (jsonify({"error": "file field required (multipart/form-data)"}), 400) f = request.files["file"] if not f or not f.filename: return None, (jsonify({"error": "empty file"}), 400) d = os.path.join(WORK_DIR, uuid.uuid4().hex) os.makedirs(d, exist_ok=True) p = os.path.join(d, os.path.basename(f.filename) or "upload.bin") f.save(p) return p, None # ----------------------------- public ----------------------------- @app.route("/beacon") def beacon(): return jsonify({"service": "lynx", "status": "ok", "time": int(time.time())}) @app.route("/health") def health(): try: kali = urllib.request.urlopen(KALI_API + "/health", timeout=6).read().decode() kali_ok = "healthy" in kali except Exception: kali_ok = False return jsonify({"status": "ok", "kali": kali_ok, "tools": len(TOOL_PATHS), "time": int(time.time())}) @app.route("/stats") def stats(): users = db().execute("SELECT COUNT(*) c FROM users").fetchone()["c"] calls = db().execute("SELECT COALESCE(SUM(total_calls),0) c FROM users").fetchone()["c"] return jsonify({"users": users, "total_calls": calls, "tools": list(COSTS.keys())}) # ----------------------------- auth / billing ----------------------------- @app.route("/api/signup", methods=["POST"]) def signup(): if rate_limited("signup_" + request.remote_addr, limit=5, window=300): return jsonify({"error": "rate limited"}), 429 d = request.json or {} email = (d.get("email") or "").strip().lower() if not email or "@" not in email or "." not in email: return jsonify({"error": "valid email required"}), 400 key = new_key() try: db().execute("INSERT INTO users (email, api_key, credits, created_at) VALUES (?,?,?,?)", (email, key, 0, int(time.time()))) db().commit() except sqlite3.IntegrityError: row = db().execute("SELECT api_key FROM users WHERE email=?", (email,)).fetchone() key = row["api_key"] return jsonify({"email": email, "api_key": key, "credits": 0, "note": "No KYC. Your key has 0 credits — buy at /api/order."}) @app.route("/api/pricing") def pricing(): return jsonify({"tiers": TIERS, "costs": COSTS, "currency": "USD", "note": "No KYC, no subscription."}) @app.route("/api/order", methods=["POST"]) def order(): if rate_limited("order_" + request.remote_addr, limit=10, window=60): return jsonify({"error": "rate limited"}), 429 d = request.json or {} key = d.get("api_key") or get_key() usd = float(d.get("usd", 5)) if usd not in TIERS: return jsonify({"error": "usd must be one of " + str(sorted(TIERS.keys()))}), 400 row = db().execute("SELECT * FROM users WHERE api_key=?", (key,)).fetchone() if not row: return jsonify({"error": "invalid api_key"}), 401 credits = TIERS[usd] order_id = "lynx-" + secrets.token_hex(8) # BTCPay invoice inv = { "amount": str(usd), "currency": "USD", "checkout": {"redirectURL": f"https://{PUBLIC_DOMAIN}/api/order/{order_id}"}, "metadata": {"orderId": order_id, "api_key": key, "credits": credits}, } req = urllib.request.Request(f"{BTCPAY_URL}/api/v1/stores/{BTCPAY_STORE}/invoices", data=json.dumps(inv).encode(), headers={"Content-Type": "application/json", "Authorization": "token " + BTCPAY_KEY}) try: resp = urllib.request.urlopen(req, timeout=20, context=_btc_ctx) inv_body = json.loads(resp.read().decode()) except Exception as e: return jsonify({"error": f"invoice failed: {e}"}), 502 db().execute("INSERT INTO orders (order_id, api_key, invoice_id, credits, created_at) VALUES (?,?,?,?,?)", (order_id, key, inv_body.get("id"), credits, int(time.time()))) db().commit() return jsonify({"order_id": order_id, "invoice_id": inv_body.get("id"), "credits": credits, "usd": usd, "checkout_link": inv_body.get("checkoutLink", "").replace("https://10.30.20.140", BTCPAY_PUBLIC).replace("http://10.30.20.140", BTCPAY_PUBLIC)}) @app.route("/api/order/") def order_status(order_id): o = db().execute("SELECT * FROM orders WHERE order_id=?", (order_id,)).fetchone() if not o: return jsonify({"error": "order not found"}), 404 if o["status"] == "settled": return jsonify({"order_id": order_id, "status": "settled", "credits": o["credits"]}) return jsonify({"order_id": order_id, "status": o["status"], "note": "awaiting payment"}) @app.route("/webhook/btcpay", methods=["POST"]) def btcpay_webhook(): body = request.get_json(silent=True) or {} # optional HMAC verify if request.headers.get("BTCPay-Sig"): import hmac as _hmac sig = "sha256=" + _hmac.new(BTCPAY_WEBHOOK_SECRET.encode(), request.get_data(), hashlib.sha256).hexdigest() if not _hmac.compare_digest(sig, request.headers.get("BTCPay-Sig", "")): return jsonify({"error": "bad sig"}), 401 etype = body.get("type", "") meta = body.get("metadata", {}) if isinstance(meta, dict): order_id = meta.get("orderId"); key = meta.get("api_key"); credits = meta.get("credits") else: order_id = key = credits = None if etype in ("InvoiceSettled", "InvoiceProcessing") and order_id: o = db().execute("SELECT * FROM orders WHERE order_id=? AND status='pending'", (order_id,)).fetchone() if o: db().execute("UPDATE orders SET status='settled', settled_at=? WHERE order_id=?", (int(time.time()), order_id)) db().execute("UPDATE users SET credits=credits+? WHERE api_key=?", (credits, key)) db().commit() return jsonify({"status": "ok"}) @app.route("/api/me") def me(): row, err = auth(gate=False) if not row: return err return jsonify({"email": row["email"], "api_key": row["api_key"], "credits": row["credits"], "total_calls": row["total_calls"]}) # ----------------------------- metered tools ----------------------------- def metered(tool): def deco(fn): @wraps(fn) def wrapper(*a, **kw): row, err = auth(gate=True) if not row: return err cost = COSTS[tool] if not row["is_admin"] and row["credits"] < cost: return jsonify({"error": f"insufficient credits ({row['credits']} < {cost}). Buy at /api/order"}), 402 try: result = fn(*a, **kw) finally: pass log_usage(row["api_key"], tool, kw.get("target", ""), cost) return result return wrapper return deco @app.route("/api/inspect/file", methods=["POST"]) @metered("inspect/file") def inspect_file(): p, err = save_upload() if err: return err name = request.files["file"].filename try: rep = analyze_file(p, name) return jsonify({"tool": "inspect/file", "credits": COSTS["inspect/file"], "result": rep}) finally: shutil.rmtree(os.path.dirname(p), ignore_errors=True) @app.route("/api/steg/extract", methods=["POST"]) @metered("steg/extract") def steg_extract_route(): p, err = save_upload() if err: return err name = request.files["file"].filename passphrase = (request.form.get("passphrase") or "") try: rep = steg_extract(p, name, passphrase) return jsonify({"tool": "steg/extract", "credits": COSTS["steg/extract"], "result": rep}) finally: shutil.rmtree(os.path.dirname(p), ignore_errors=True) @app.route("/api/steg/crack", methods=["POST"]) @metered("steg/crack") def steg_crack_route(): p, err = save_upload() if err: return err try: rep = steg_crack(p, request.files["file"].filename) return jsonify({"tool": "steg/crack", "credits": COSTS["steg/crack"], "result": rep}) finally: shutil.rmtree(os.path.dirname(p), ignore_errors=True) @app.route("/api/forensics/disk", methods=["POST"]) @metered("forensics/disk") def forensics_route(): p, err = save_upload() if err: return err try: rep = forensics_disk(p, request.files["file"].filename) return jsonify({"tool": "forensics/disk", "credits": COSTS["forensics/disk"], "result": rep}) finally: shutil.rmtree(os.path.dirname(p), ignore_errors=True) # ----------------------------- recon (Kali) ----------------------------- RECON_MAP = { "nmap": {"params": lambda t: {"target": t, "scan_type": "-sV"}}, "subfinder": {"params": lambda t: {"domain": t}}, "nuclei": {"params": lambda t: {"target": t, "timeout": 240}}, "theharvester": {"params": None}, "dnsrecon": {"params": None}, } @app.route("/api/recon/", methods=["POST"]) def recon_route(tool): if tool not in RECON_MAP: return jsonify({"error": "unknown recon tool", "available": list(RECON_MAP.keys())}), 400 d = request.json or {} target = clean_target(d.get("target", "")) if not target: return jsonify({"error": "valid target required"}), 400 # manual metering (cost depends on dynamic tool) row, err = auth(gate=True) if not row: return err cost = COSTS["recon/" + tool] if not row["is_admin"] and row["credits"] < cost: return jsonify({"error": f"insufficient credits ({row['credits']} < {cost}). Buy at /api/order"}), 402 if tool in ("theharvester", "dnsrecon"): # use Kali shell with sanitized fixed command if tool == "theharvester": cmd = f"theHarvester -d {target} -b all -l 100 2>&1 | head -80" else: cmd = f"dnsrecon -d {target} 2>&1 | head -100" res = kali_call("shell", {"command": cmd, "timeout": 240}) out = res.get("stdout", "") if isinstance(res, dict) else str(res) else: res = kali_call(tool, RECON_MAP[tool]["params"](target), timeout=360) out = res.get("stdout", "") if isinstance(res, dict) else str(res) log_usage(row["api_key"], f"recon/{tool}", target, cost) return jsonify({"tool": f"recon/{tool}", "target": target, "credits": cost, "result": out[:8000]}) # ----------------------------- MCP (agent-native, JSON-RPC 2.0) ----------------------------- MCP_TOOLS = { "lynx_signup": {"email": "string"}, "lynx_me": {"api_key": "string"}, "lynx_order": {"api_key": "string", "usd": "number"}, "lynx_inspect_file": {"api_key": "string", "filename": "string", "content_base64": "string"}, "lynx_steg_extract": {"api_key": "string", "filename": "string", "content_base64": "string", "passphrase": "string"}, "lynx_recon": {"api_key": "string", "tool": "string", "target": "string"}, } MCP_TOOL_DESCS = { "lynx_signup": "Create a no-KYC API key with an email. Returns sk-lynx-... key (0 credits).", "lynx_me": "Check credits + usage for a key.", "lynx_order": "Create a BTCPay invoice for credits. usd in {2,5,10,20}. Returns a checkout_link to pay in bitcoin.", "lynx_inspect_file": "Inspect a file (hashes, entropy, strings, metadata, PE/PDF/Office analysis). Pass file bytes as base64.", "lynx_steg_extract": "Extract hidden data (steghide + zsteg LSB). Optional passphrase.", "lynx_recon": "Run recon against an external target. tool in {nmap,subfinder,nuclei,theharvester,dnsrecon}.", } def _mcp_tool_call(name, args): key = args.get("api_key", "") if name == "lynx_signup": email = (args.get("email") or "").strip().lower() k = new_key() try: db().execute("INSERT INTO users (email, api_key, credits, created_at) VALUES (?,?,?,?)", (email, k, 0, int(time.time()))) db().commit() except sqlite3.IntegrityError: k = db().execute("SELECT api_key FROM users WHERE email=?", (email,)).fetchone()["api_key"] return {"api_key": k, "credits": 0} if name == "lynx_me": row = db().execute("SELECT * FROM users WHERE api_key=?", (key,)).fetchone() if not row: return {"error": "invalid api_key"} return {"email": row["email"], "credits": row["credits"], "total_calls": row["total_calls"]} if name == "lynx_order": usd = float(args.get("usd", 5)) if usd not in TIERS: return {"error": "usd in " + str(sorted(TIERS.keys()))} row = db().execute("SELECT * FROM users WHERE api_key=?", (key,)).fetchone() if not row: return {"error": "invalid api_key"} credits = TIERS[usd] order_id = "lynx-" + secrets.token_hex(8) inv = {"amount": str(usd), "currency": "USD", "checkout": {"redirectURL": f"https://{PUBLIC_DOMAIN}/api/order/{order_id}"}, "metadata": {"orderId": order_id, "api_key": key, "credits": credits}} req = urllib.request.Request(f"{BTCPAY_URL}/api/v1/stores/{BTCPAY_STORE}/invoices", data=json.dumps(inv).encode(), headers={"Content-Type": "application/json", "Authorization": "token " + BTCPAY_KEY}) try: resp = urllib.request.urlopen(req, timeout=20, context=_btc_ctx) inv_body = json.loads(resp.read().decode()) except Exception as e: return {"error": f"invoice failed: {e}"} db().execute("INSERT INTO orders (order_id, api_key, invoice_id, credits, created_at) VALUES (?,?,?,?,?)", (order_id, key, inv_body.get("id"), credits, int(time.time()))) db().commit() return {"order_id": order_id, "credits": credits, "usd": usd, "checkout_link": inv_body.get("checkoutLink", "").replace("https://10.30.20.140", BTCPAY_PUBLIC)} # file tools (base64 content) if name in ("lynx_inspect_file", "lynx_steg_extract"): row = db().execute("SELECT * FROM users WHERE api_key=?", (key,)).fetchone() if not row: return {"error": "invalid api_key"} tool = "inspect/file" if name == "lynx_inspect_file" else "steg/extract" cost = COSTS[tool] if not row["is_admin"] and row["credits"] < cost: return {"error": f"insufficient credits ({row['credits']} < {cost})"} try: content = base64.b64decode(args.get("content_base64", "")) except Exception: return {"error": "invalid content_base64"} d = os.path.join(WORK_DIR, uuid.uuid4().hex) os.makedirs(d, exist_ok=True) fn = os.path.basename(args.get("filename", "upload.bin")) or "upload.bin" p = os.path.join(d, fn) with open(p, "wb") as f: f.write(content) try: if name == "lynx_inspect_file": rep = analyze_file(p, fn) else: rep = steg_extract(p, fn, args.get("passphrase", "")) finally: shutil.rmtree(d, ignore_errors=True) log_usage(key, tool, fn, cost) return {"credits_used": cost, "result": rep} if name == "lynx_recon": row = db().execute("SELECT * FROM users WHERE api_key=?", (key,)).fetchone() if not row: return {"error": "invalid api_key"} tool = args.get("tool", "") if tool not in RECON_MAP: return {"error": "tool in " + str(list(RECON_MAP.keys()))} target = clean_target(args.get("target", "")) if not target: return {"error": "valid target required"} cost = COSTS["recon/" + tool] if not row["is_admin"] and row["credits"] < cost: return {"error": f"insufficient credits ({row['credits']} < {cost})"} if tool in ("theharvester", "dnsrecon"): cmd = f"theHarvester -d {target} -b all -l 100 2>&1 | head -80" if tool == "theharvester" else f"dnsrecon -d {target} 2>&1 | head -100" res = kali_call("shell", {"command": cmd, "timeout": 240}) out = res.get("stdout", "") if isinstance(res, dict) else str(res) else: res = kali_call(tool, RECON_MAP[tool]["params"](target), timeout=360) out = res.get("stdout", "") if isinstance(res, dict) else str(res) log_usage(key, "recon/" + tool, target, cost) return {"credits_used": cost, "result": out[:8000]} return {"error": "unknown tool"} @app.route("/mcp", methods=["POST"]) def mcp_endpoint(): req_json = request.get_json(silent=True) if not req_json: return jsonify({"error": {"code": -32700, "message": "invalid JSON"}}), 400 method = req_json.get("method") rid = req_json.get("id") params = req_json.get("params", {}) or {} if method == "initialize": return jsonify({"jsonrpc": "2.0", "id": rid, "result": { "protocolVersion": "2024-11-05", "capabilities": {"tools": {}}, "serverInfo": {"name": "lynx", "version": "1.0.0"}}}) if method == "ping": return jsonify({"jsonrpc": "2.0", "id": rid, "result": {}}) if method == "notifications/initialized": return jsonify({"jsonrpc": "2.0", "id": rid, "result": {}}) if method == "tools/list": tools = [{"name": n, "description": MCP_TOOL_DESCS[n], "inputSchema": {"type": "object", "properties": {k: {"type": v} for k, v in MCP_TOOLS[n].items()}, "required": list(MCP_TOOLS[n].keys())}} for n in MCP_TOOLS] return jsonify({"jsonrpc": "2.0", "id": rid, "result": {"tools": tools}}) if method == "tools/call": name = params.get("name", "") if name not in MCP_TOOLS: return jsonify({"jsonrpc": "2.0", "id": rid, "error": {"code": -32000, "message": "unknown tool"}}) args = params.get("arguments", {}) or {} try: result = _mcp_tool_call(name, args) except Exception as e: return jsonify({"jsonrpc": "2.0", "id": rid, "error": {"code": -32603, "message": str(e)}}) return jsonify({"jsonrpc": "2.0", "id": rid, "result": {"content": [{"type": "text", "text": json.dumps(result)}]}}) return jsonify({"jsonrpc": "2.0", "id": rid, "error": {"code": -32601, "message": "method not found"}}) # ----------------------------- admin ----------------------------- @app.route("/admin") def admin(): key = get_key() if key != ADMIN_KEY: return jsonify({"error": "admin only"}), 403 users = [dict(r) for r in db().execute("SELECT * FROM users ORDER BY id DESC LIMIT 50").fetchall()] return jsonify({"admin": True, "users": users, "admin_key": ADMIN_KEY}) @app.route("/admin/revoke", methods=["POST"]) def admin_revoke(): if get_key() != ADMIN_KEY: return jsonify({"error": "admin only"}), 403 key = (request.json or {}).get("api_key") db().execute("DELETE FROM users WHERE api_key=?", (key,)) db().commit() return jsonify({"revoked": key}) # ----------------------------- docs ----------------------------- @app.route("/openapi.json") def openapi(): spec = {"openapi": "3.0.0", "info": {"title": "LYNX", "version": "1.0.0", "description": "No-KYC inspection, steg, forensics, and recon API — paid in Bitcoin."}, "servers": [{"url": f"https://{PUBLIC_DOMAIN}"}]} return jsonify(spec) @app.route("/docs") def docs(): return Response(render_docs(), mimetype="text/html") @app.route("/") def home(): return Response(render_home(), mimetype="text/html") # ---------------------------------------------------------------------------- # HTML (spooky dark landing + docs) # ---------------------------------------------------------------------------- def render_home(): return _HTML_HEAD + HOME_BODY + _HTML_FOOT def render_docs(): return _HTML_HEAD + DOCS_BODY + _HTML_FOOT _HTML_HEAD = """ LYNX — the sharp-eyed inspector
""" HOME_BODY = """

LYNX

the sharp-eyed inspector — see what's hidden.

…agents
…inspections
20tools
no-KYCbitcoin

Inspect files · decode steganography · carve forensics · run recon — paid in sats.

How it works

# 1. get a key (no KYC — just an email)
curl -X POST https://lynx.thetempleofdoom.com/api/signup -H 'Content-Type: application/json' -d '{"email":"you@proton.me"}'

# 2. buy credits (bitcoin)
curl -X POST https://lynx.thetempleofdoom.com/api/order -H 'Content-Type: application/json' -d '{"usd":5}'

# 3. inspect a file
curl -X POST https://lynx.thetempleofdoom.com/api/inspect/file -H 'X-API-Key: sk-lynx-…' -F 'file=@suspicious.pdf'

The toolchain

🔬 File inspection

hashes (md5/sha1/sha256/ssdeep), entropy, strings, EXIF metadata, PE analysis, Office macro scan, PDF object analysis, binwalk firmware signatures.

1 credit

🕵️ Steganography

steghide extraction, zsteg LSB detection, stegcracker passphrase recovery — decode data hidden in images and files.

2–4 credits

🧬 Forensics

Sleuth Kit disk carving (mmls/fls/fsstat/tsk_recover), volatility3 memory analysis.

3 credits

📡 Recon

nmap, subfinder, nuclei, theHarvester, dnsrecon — isolated egress, never touches the operator network.

2–4 credits

Pricing — no subscription

USDcredits
$220
$560
$10150
$20400

Credits never expire. No KYC, no tracking, no bullshit. Pay on-chain or via Lightning.

Built for agents

Every tool is machine-callable. Pull /openapi.json for a full spec, or drive LYNX straight from your agent's MCP client. Programmatic keys, Bitcoin-settled metered billing — the API is the product.

Read the API docs →

""" DOCS_BODY = """

API reference

All metered endpoints accept the key via X-API-Key header or ?api_key= query param.

POST /api/signup

No-KYC key. Returns api_key (0 credits).

curl -X POST https://lynx.thetempleofdoom.com/api/signup -H 'Content-Type: application/json' -d '{"email":"you@proton.me"}'

POST /api/order

Create a BTCPay invoice. usd ∈ {2,5,10,20}. Returns checkout_link.

curl -X POST https://lynx.thetempleofdoom.com/api/order -H 'X-API-Key: sk-lynx-…' -H 'Content-Type: application/json' -d '{"usd":5}'

GET /api/me

Credits + usage for your key.

curl https://lynx.thetempleofdoom.com/api/me -H 'X-API-Key: sk-lynx-…'

POST /api/inspect/file

multipart upload → full inspection report. 1 credit

curl -X POST https://lynx.thetempleofdoom.com/api/inspect/file -H 'X-API-Key: sk-lynx-…' -F 'file=@sample.pdf'

POST /api/steg/extract

steghide + zsteg extraction. Optional passphrase form field. 2 credits

curl -X POST https://lynx.thetempleofdoom.com/api/steg/extract -H 'X-API-Key: sk-lynx-…' -F 'file=@hidden.png' -F 'passphrase=secret'

POST /api/steg/crack

stegcracker passphrase recovery. 4 credits

curl -X POST https://lynx.thetempleofdoom.com/api/steg/crack -H 'X-API-Key: sk-lynx-…' -F 'file=@hidden.jpg'

POST /api/forensics/disk

Sleuth Kit on a disk image (mmls/fsstat/fls). 3 credits

curl -X POST https://lynx.thetempleofdoom.com/api/forensics/disk -H 'X-API-Key: sk-lynx-…' -F 'file=@disk.img'

POST /api/recon/<tool>

nmap, subfinder, nuclei, theharvester, dnsrecon. Body {"target":"example.com"}.

curl -X POST https://lynx.thetempleofdoom.com/api/recon/nmap -H 'X-API-Key: sk-lynx-…' -H 'Content-Type: application/json' -d '{"target":"example.com"}'
""" _HTML_FOOT = """
""" # ---------------------------------------------------------------------------- # Bootstrap (runs at import — gunicorn needs this, not just __main__) # ---------------------------------------------------------------------------- init_db() resolve_tools() os.makedirs(WORK_DIR, exist_ok=True) if not os.path.exists(STEG_WORDLIST): with open(STEG_WORDLIST, "w") as f: f.write("\n".join(["password","123456","letmein","secret","admin","root","hidden","steg","changeme","dragon","monkey","qwerty","abc123","iloveyou","trustno1","hunter2","welcome","shadow","baseball","football","superman","batman","matrix","pokemon","starwars","joshua","master","passw0rd","password1","default","guest"]) + "\n") # Seed admin user (is_admin bypasses credit gate on metered tools) — use direct conn (no app context at import) _c = sqlite3.connect(DB_PATH, timeout=15) if not _c.execute("SELECT 1 FROM users WHERE api_key=?", (ADMIN_KEY,)).fetchone(): _c.execute("INSERT INTO users (email, api_key, credits, is_admin, created_at) VALUES (?,?,?,?,?)", ("admin@lynx.local", ADMIN_KEY, 999999, 1, int(time.time()))) _c.commit() _c.close() if __name__ == "__main__": app.run(host="0.0.0.0", port=PORT)