commit ba282347bba2c0a8550d8c42e2ee1fa075d39cbd Author: drjones Date: Mon Sep 14 02:30:05 2026 -0700 LYNX — no-KYC inspection/steg/forensics/recon API (initial release) diff --git a/README.md b/README.md new file mode 100644 index 0000000..f73e416 --- /dev/null +++ b/README.md @@ -0,0 +1,70 @@ +# LYNX — the sharp-eyed inspector 🔭 + +No-KYC, Bitcoin-paid API for **file inspection, steganography, forensics, and recon**. +Public: https://lynx.thetempleofdoom.com + +## What it does + +| Class | Tools | Credits | +|---|---|---| +| 🔬 File inspection | md5/sha1/sha256/ssdeep, entropy, strings, exiftool, PE, OLE macros, PDF, binwalk | 1 | +| 🕵️ Steganography | steghide extract, stegcracker, zsteg LSB | 2–4 | +| 🧬 Forensics | Sleuth Kit (mmls/fls/fsstat/tsk_recover), volatility3 | 3 | +| 📡 Recon | nmap, subfinder, nuclei, theHarvester, dnsrecon (isolated Kali egress) | 2–4 | + +## Quick start + +```bash +# 1. no-KYC key +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 'X-API-Key: sk-lynx-…' -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' + +# 4. decode hidden data +curl -X POST https://lynx.thetempleofdoom.com/api/steg/extract -H 'X-API-Key: sk-lynx-…' -F 'file=@hidden.png' -F 'passphrase=secret' + +# 5. recon an external target +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"}' +``` + +## Architecture + +- **Host:** CT707 (Debian 12), Flask + gunicorn `:5059` +- **File/steg/forensics:** run locally on CT707 (fast, no binary-transfer issues) +- **Recon:** dispatched to Kali (`10.30.30.177`, isolated vmbr1 segment) — egress never touches the production network +- **Billing:** BTCPay store + webhook, SQLite credits, no subscriptions, credits never expire +- **Agent-native:** `/mcp` JSON-RPC 2.0 endpoint — 6 tools (`lynx_signup`, `lynx_me`, `lynx_order`, `lynx_inspect_file`, `lynx_steg_extract`, `lynx_recon`) + +## API surface + +REST: `/api/signup`, `/api/order`, `/api/order/`, `/webhook/btcpay`, `/api/me`, `/api/pricing`, +`/api/inspect/file`, `/api/steg/extract`, `/api/steg/crack`, `/api/forensics/disk`, `/api/recon/`, +`/stats`, `/admin`, `/docs`, `/openapi.json`. +MCP: `/mcp` (JSON-RPC 2.0). + +Auth: `X-API-Key` header or `?api_key=` query. 0 credits → 402. Rate-limited per key/IP. + +## Pricing (USD → credits) + +$2=20 · $5=60 · $10=150 · $20=400 + +## Deploy + +```bash +# from the repo root +tar czf /tmp/lynx.tar.gz app.py lynx.service +scp /tmp/lynx.tar.gz root@10.30.20.85:/tmp/ +ssh root@10.30.20.85 'pct push 707 /tmp/lynx.tar.gz /tmp/ && \ + pct exec 707 -- bash -c "cd /opt/lynx && tar xzf /tmp/lynx.tar.gz && \ + cp lynx.service /etc/systemd/system/ && systemctl daemon-reload && systemctl restart lynx"' +``` + +## Notes + +- Kali's `write_file` API is text-only (corrupts binaries) — that's why file tools run on CT707. +- Recon targets are sanitized (`[A-Za-z0-9.\-_:/]+`) and tool args are server-constructed (no shell injection). +- Cloudflare `enable_js` JS-challenge blocks bare `Python-urllib` UAs; curl/requests/MCP SDKs pass fine. diff --git a/app.py b/app.py new file mode 100644 index 0000000..94f803d --- /dev/null +++ b/app.py @@ -0,0 +1,897 @@ +#!/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) diff --git a/lynx.service b/lynx.service new file mode 100644 index 0000000..9bcf5ef --- /dev/null +++ b/lynx.service @@ -0,0 +1,17 @@ +[Unit] +Description=LYNX inspection API (no-KYC, bitcoin-paid) +After=network.target + +[Service] +Type=simple +WorkingDirectory=/opt/lynx +Environment=PATH=/usr/local/bin:/usr/bin:/bin:/usr/local/sbin:/usr/sbin:/sbin +Environment=LYNX_PORT=5059 +Environment=LYNX_DOMAIN=lynx.thetempleofdoom.com +Environment=LYNX_ADMIN_KEY=sk-lynx-admin-7c97d4db9eac0f3a +ExecStart=/usr/local/bin/gunicorn -w 2 -b 0.0.0.0:5059 --timeout 600 --access-logfile - app:app +Restart=always +RestartSec=3 + +[Install] +WantedBy=multi-user.target