#!/usr/bin/env python3 """ sms.py — Temporary inbound-SMS tool for Pleiades (Clean Proxys). A NumberProvider abstraction over SMSPool, plus Flask routes, pricing, the number lifecycle, code detection, and realtime polling for the temp-number marketplace. Reuses the existing wallet (balance_sats + transactions ledger) and BTCPay deposit flow — customers top up BTC once, then rent numbers on demand against their wallet balance. Provider facts (verified live 2026-09-20): GET /request/balance -> {"balance":"7.45"} GET /country/retrieve_all -> [{ID,name,short_name,cc,region}] GET /service/retrieve_all -> [{ID,name,favourite}] GET /pool/retrieve_all -> [{ID,name}] GET /purchase/sms?service=&country= -> {success,number,phonenumber,order_id, country,service,pool,expires_in, expiration,cost,cost_in_cents, current_balance} GET /sms/check?orderid= -> {status,message,resend,expiration,time_left} GET /sms/cancel?orderid= -> {success,message} (refunds unused rental) GET /request/active -> [] (active orders) SMSPool /sms/check status codes: 1=SMS received, 2/3=waiting/pending, 4=cancelled/expired, 5/6=refunded. Rental cost is charged at purchase and refunded on /sms/cancel if no SMS was received -> float exposure is only `concurrent active sessions * max rental cost`. """ import re import time import requests from flask import (abort, flash, jsonify, redirect, render_template, request, session, url_for) import db CFG = db.CFG SMSPOOL_KEY = (CFG.get("smspool") or {}).get("key", "") # -------------------------------------------------------------------------- # Provider abstraction # -------------------------------------------------------------------------- class NumberProvider: """Swap the upstream by implementing this interface and changing PROVIDER.""" def list_countries(self): raise NotImplementedError def list_services(self): raise NotImplementedError def provision_number(self, service, country, expiry_min): raise NotImplementedError def check_sms(self, order_id): raise NotImplementedError def release_number(self, order_id): raise NotImplementedError def get_balance(self): raise NotImplementedError class SMSPoolProvider(NumberProvider): BASE = "https://api.smspool.net" def __init__(self, key): self.key = key def _get(self, path, params): p = dict(params) p["key"] = self.key r = requests.get(f"{self.BASE}/{path}", params=p, timeout=30) ct = r.headers.get("content-type", "") if "json" in ct: try: return r.json() except ValueError: return {"success": 0, "message": r.text[:300]} return {"success": 0, "message": r.text[:300]} def list_countries(self): return self._get("country/retrieve_all", {}) def list_services(self): return self._get("service/retrieve_all", {}) def provision_number(self, service, country, expiry_min): # `expiry` is passed for API compatibility but SMSPool ignores it — # every rental hard-caps at 20 min (see MAX_RENTAL_MIN). return self._get("purchase/sms", {"service": service, "country": country}) def check_sms(self, order_id): return self._get("sms/check", {"orderid": order_id}) def release_number(self, order_id): return self._get("sms/cancel", {"orderid": order_id}) def get_balance(self): r = self._get("request/balance", {}) try: return float(r.get("balance", 0)) except (TypeError, ValueError): return None PROVIDER = SMSPoolProvider(SMSPOOL_KEY) # -------------------------------------------------------------------------- # Curated popular services (SMSPool IDs, verified 2026-09-20) # -------------------------------------------------------------------------- POPULAR_SERVICES = [ (395, "Google / Gmail"), (907, "Telegram"), (1012, "WhatsApp"), (948, "Twitter / X"), (457, "Instagram / Threads"), (273, "Discord"), (329, "Facebook / Meta"), (671, "OpenAI / ChatGPT"), (924, "TikTok / Douyin"), (951, "Uber / Postmates"), (39, "Amazon / AWS"), (1072, "Microsoft / Outlook / Bing"), (48, "Apple"), (630, "Netflix"), (926, "Tinder"), (846, "Snapchat"), (829, "Signal"), (523, "LinkedIn"), (1554, "Reddit"), (28, "Airbnb"), (280, "DoorDash"), (233, "Craigslist"), (1034, "Yahoo"), ] DEFAULT_PRICING = [ {"duration_min": 5, "price_sats": 1500}, {"duration_min": 10, "price_sats": 2000}, {"duration_min": 15, "price_sats": 2500}, {"duration_min": 20, "price_sats": 3000}, ] MAX_RENTAL_MIN = 20 # VERIFIED 2026-09-20: SMSPool ignores `expiry` param — rental numbers # hard-cap at 1200s (20 min) regardless of what's requested. Longer # sessions require a re-rent loop (not yet implemented). POLL_THROTTLE_SEC = 3 # min seconds between SMSPool /sms/check calls per session — # caps upstream API load + spend regardless of client poll rate. # -------------------------------------------------------------------------- # Schema # -------------------------------------------------------------------------- SCHEMA = """ CREATE TABLE IF NOT EXISTS sms_sessions ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER NOT NULL, service_id INTEGER, service_name TEXT, country_id INTEGER, country_name TEXT, country_cc TEXT, provider_order_id TEXT, number TEXT, status TEXT DEFAULT 'active', price_sats INTEGER, provider_cost TEXT, created_at INTEGER, expires_at INTEGER, sms_received_at INTEGER, last_check INTEGER ); CREATE TABLE IF NOT EXISTS sms_messages ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id INTEGER NOT NULL, from_number TEXT, body TEXT, detected_code TEXT, received_at INTEGER ); CREATE TABLE IF NOT EXISTS sms_pricing ( id INTEGER PRIMARY KEY AUTOINCREMENT, duration_min INTEGER UNIQUE, price_sats INTEGER, active INTEGER DEFAULT 1 ); CREATE TABLE IF NOT EXISTS sms_countries_cache ( id INTEGER PRIMARY KEY, name TEXT, short_name TEXT, cc TEXT, region TEXT ); CREATE TABLE IF NOT EXISTS sms_services_cache ( id INTEGER PRIMARY KEY, name TEXT ); """ def init_sms(): d = db.get_db() d.executescript(SCHEMA) # migrations cols = [r[1] for r in d.execute("PRAGMA table_info(sms_sessions)")] if "last_check" not in cols: d.execute("ALTER TABLE sms_sessions ADD COLUMN last_check INTEGER") if d.execute("SELECT COUNT(*) c FROM sms_pricing").fetchone()["c"] == 0: for p in DEFAULT_PRICING: d.execute("INSERT INTO sms_pricing (duration_min, price_sats, active) VALUES (?,?,1)", (p["duration_min"], p["price_sats"])) d.commit() d.close() def get_pricing(): d = db.get_db() rows = [dict(r) for r in d.execute( "SELECT * FROM sms_pricing WHERE active=1 ORDER BY duration_min")] d.close() return rows def price_for(duration_min): d = db.get_db() r = d.execute("SELECT price_sats FROM sms_pricing WHERE duration_min=? AND active=1", (duration_min,)).fetchone() d.close() return r["price_sats"] if r else None # -------------------------------------------------------------------------- # Code detection (convenience only — never alters the raw SMS) # -------------------------------------------------------------------------- _CODE_CONTEXT = re.compile(r"(code|verification|verify|otp|pin|confirm|security|login|auth)", re.I) def detect_code(body): if not body: return None for m in re.finditer(r"\b(\d{4,8})\b", body): s = max(0, m.start() - 40) ctx = body[s:m.end() + 20] if _CODE_CONTEXT.search(ctx): return m.group(1) m = re.search(r"\b(\d{4,8})\b", body) return m.group(1) if m else None # -------------------------------------------------------------------------- # Helpers # -------------------------------------------------------------------------- def _client_ip(): xff = request.headers.get("X-Forwarded-For", "") return xff.split(",")[0].strip() if xff else (request.remote_addr or "?") def _rate_ok(ip, limit, window): now = time.time() d = db.get_db() d.execute("DELETE FROM rate_limits WHERE ts < ?", (now - window,)) c = d.execute("SELECT COUNT(*) c FROM rate_limits WHERE ip=?", (ip,)).fetchone()["c"] if c >= limit: d.close() return False d.execute("INSERT INTO rate_limits (ip, ts) VALUES (?,?)", (ip, now)) d.commit() d.close() return True def _active_session(user_id): d = db.get_db() r = d.execute( "SELECT * FROM sms_sessions WHERE user_id=? AND status IN ('active','sms_received') " "ORDER BY id DESC LIMIT 1", (user_id,)).fetchone() d.close() return dict(r) if r else None def _session_messages(session_id): d = db.get_db() rows = [dict(r) for r in d.execute( "SELECT * FROM sms_messages WHERE session_id=? ORDER BY id", (session_id,)).fetchall()] d.close() return rows def _debit(user_id, amount_sats, type_, ref): """Atomically deduct a wallet balance, returning True on success. The `AND balance_sats >= ?` guard makes the check+deduct a single atomic UPDATE — no double-spend window under concurrent rent requests. Inserts the matching ledger entry only when the debit actually landed. """ d = db.get_db() cur = d.execute( "UPDATE users SET balance_sats = balance_sats - ? WHERE id=? AND balance_sats >= ?", (amount_sats, user_id, amount_sats)) if cur.rowcount == 0: d.close() return False d.execute("INSERT INTO transactions (user_id, amount_sats, type, ref, ts) VALUES (?,?,?,?,?)", (user_id, -amount_sats, type_, ref, int(time.time()))) d.commit() d.close() return True def _service_name(service_id): for sid, name in POPULAR_SERVICES: if sid == service_id: return name d = db.get_db() r = d.execute("SELECT name FROM sms_services_cache WHERE id=?", (service_id,)).fetchone() d.close() return r["name"] if r else ("Service %d" % service_id) def _enabled_countries(): d = db.get_db() rows = [dict(r) for r in d.execute( "SELECT * FROM sms_countries_cache ORDER BY (id=1) DESC, name").fetchall()] d.close() if rows: return rows try: data = PROVIDER.list_countries() except Exception: data = [] if not isinstance(data, list): data = [] d = db.get_db() for c in data: try: d.execute("INSERT OR REPLACE INTO sms_countries_cache (id,name,short_name,cc,region) " "VALUES (?,?,?,?,?)", (c.get("ID"), c.get("name"), c.get("short_name"), c.get("cc"), c.get("region"))) except Exception: continue d.commit() d.close() return [{"id": c.get("ID"), "name": c.get("name"), "short_name": c.get("short_name"), "cc": c.get("cc"), "region": c.get("region")} for c in data] def _all_services(): d = db.get_db() rows = [dict(r) for r in d.execute("SELECT * FROM sms_services_cache").fetchall()] d.close() if rows: return rows try: data = PROVIDER.list_services() except Exception: data = [] if not isinstance(data, list): data = [] d = db.get_db() for s in data: try: d.execute("INSERT OR REPLACE INTO sms_services_cache (id,name) VALUES (?,?)", (s.get("ID"), s.get("name"))) except Exception: continue d.commit() d.close() return [{"id": s.get("ID"), "name": s.get("name")} for s in data] def _extract_sms(chk): """Return (body, from_number) from a /sms/check response, or (None, None). Defensive: SMSPool's exact "SMS received" field names aren't fully documented, so check every plausible location. Only treat status==1 (or an explicit sms/text payload) as a received message — never the top-level `message` field (which carries status text like "This order has been refunded"). """ if not isinstance(chk, dict): return None, None status = chk.get("status") sms = chk.get("sms") if isinstance(sms, dict): body = (sms.get("text") or sms.get("body") or sms.get("message") or sms.get("content") or "") frm = (sms.get("number") or sms.get("sender") or sms.get("from") or "") if body: return body, frm if status == 1: body = chk.get("text") or chk.get("body") or chk.get("content") or "" frm = chk.get("number") or chk.get("sender") or chk.get("from") or "" if body: return body, frm return None, None # -------------------------------------------------------------------------- # Route registration # -------------------------------------------------------------------------- def init_sms_app(app, login_required, current_user): init_sms() @app.route("/sms") @login_required def sms_home(): u = current_user() pricing = get_pricing() countries = _enabled_countries() sess = _active_session(u["id"]) bal = db.get_balance_sats(u["id"]) return render_template("sms.html", user=u, pricing=pricing, countries=countries, session=sess, messages=_session_messages(sess["id"]) if sess else [], popular=POPULAR_SERVICES, balance=bal) @app.route("/sms/services") @login_required def sms_services(): q = (request.args.get("q") or "").strip().lower() if q: out = [s for s in _all_services() if q in s["name"].lower()][:30] return jsonify({"services": out}) return jsonify({"services": [{"id": i, "name": n} for i, n in POPULAR_SERVICES]}) @app.route("/sms/rent", methods=["POST"]) @login_required def sms_rent(): u = current_user() if not _rate_ok(_client_ip(), limit=10, window=300): flash("Too many attempts — slow down.", "warn") return redirect(url_for("sms_home")) if _active_session(u["id"]): flash("You already have an active number. Release it first.", "warn") return redirect(url_for("sms_home")) try: service_id = int(request.form.get("service_id")) country_id = int(request.form.get("country_id")) duration = int(request.form.get("duration")) except (TypeError, ValueError): flash("Bad request.", "warn") return redirect(url_for("sms_home")) price = price_for(duration) if price is None: flash("Invalid duration.", "warn") return redirect(url_for("sms_home")) country = next((c for c in _enabled_countries() if c["id"] == country_id), None) if not country: flash("Country unavailable.", "warn") return redirect(url_for("sms_home")) # debit BEFORE provisioning — atomic check+deduct, refunded if provision fails ref = db.random_token(12) if not _debit(u["id"], price, "sms", ref): flash("Insufficient balance. Deposit BTC in your wallet first.", "warn") return redirect(url_for("wallet")) try: res = PROVIDER.provision_number(service_id, country_id, min(duration, MAX_RENTAL_MIN)) except Exception as e: db.add_transaction(u["id"], price, "sms_refund", ref) flash(f"Provider error: {e}", "warn") return redirect(url_for("sms_home")) if not res.get("success"): msg = (res.get("errors") or [{}])[0].get("message") or res.get("message") or "No numbers available." db.add_transaction(u["id"], price, "sms_refund", ref) flash(f"Number unavailable: {msg}", "warn") return redirect(url_for("sms_home")) order_id = res.get("order_id") or res.get("orderid") number = res.get("number") or res.get("phonenumber") try: cost = float(res.get("cost", 0) or 0) except (TypeError, ValueError): cost = 0.0 service_name = _service_name(service_id) d = db.get_db() d.execute( "INSERT INTO sms_sessions (user_id, service_id, service_name, country_id, " "country_name, country_cc, provider_order_id, number, status, price_sats, " "provider_cost, created_at, expires_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)", (u["id"], service_id, service_name, country_id, country["name"], country["cc"], order_id, number, "active", price, cost, int(time.time()), int(time.time()) + duration * 60)) d.commit() d.close() flash(f"Number {number} reserved for {duration} min.", "ok") return redirect(url_for("sms_home")) @app.route("/sms/session//poll") @login_required def sms_poll(sid): u = current_user() d = db.get_db() row = d.execute("SELECT * FROM sms_sessions WHERE id=? AND user_id=?", (sid, u["id"])).fetchone() if not row: d.close() return jsonify({"error": "not found"}), 404 sess = dict(row) now = int(time.time()) # lazy expiry -> release upstream if sess["status"] in ("active", "sms_received") and now >= sess["expires_at"]: try: PROVIDER.release_number(sess["provider_order_id"]) except Exception: pass d.execute("UPDATE sms_sessions SET status='expired' WHERE id=?", (sid,)) d.commit() sess["status"] = "expired" if sess["status"] == "active" and now - (sess.get("last_check") or 0) >= POLL_THROTTLE_SEC: d.execute("UPDATE sms_sessions SET last_check=? WHERE id=?", (now, sid)) d.commit() try: chk = PROVIDER.check_sms(sess["provider_order_id"]) except Exception: chk = {} body, from_num = _extract_sms(chk) status = chk.get("status") if body: code = detect_code(body) d.execute( "INSERT INTO sms_messages (session_id, from_number, body, detected_code, received_at) " "VALUES (?,?,?,?,?)", (sid, from_num, body, code, now)) d.execute("UPDATE sms_sessions SET status='sms_received', sms_received_at=? WHERE id=?", (now, sid)) d.commit() sess["status"] = "sms_received" elif status in (4, 5, 6): d.execute("UPDATE sms_sessions SET status='released' WHERE id=?", (sid,)) d.commit() sess["status"] = "released" msgs = [dict(r) for r in d.execute( "SELECT * FROM sms_messages WHERE session_id=? ORDER BY id", (sid,)).fetchall()] d.close() return jsonify({ "id": sess["id"], "number": sess["number"], "service": sess["service_name"], "country": sess["country_name"], "status": sess["status"], "expires_at": sess["expires_at"], "time_remaining": max(0, sess["expires_at"] - now), "messages": msgs, "code": msgs[-1]["detected_code"] if msgs and msgs[-1]["detected_code"] else None, }) @app.route("/sms/session//release", methods=["POST"]) @login_required def sms_release(sid): u = current_user() d = db.get_db() row = d.execute("SELECT * FROM sms_sessions WHERE id=? AND user_id=?", (sid, u["id"])).fetchone() if not row: d.close() abort(404) if row["status"] in ("active", "sms_received"): try: PROVIDER.release_number(row["provider_order_id"]) except Exception: pass d.execute("UPDATE sms_sessions SET status='released' WHERE id=?", (sid,)) d.commit() flash("Number released.", "ok") d.close() return redirect(url_for("sms_home")) # ----- admin ----- @app.route("/admin/sms", methods=["GET", "POST"]) def sms_admin(): if not session.get("admin"): abort(403) if request.method == "POST": changed = 0 for p in get_pricing(): val = (request.form.get("price_%d" % p["duration_min"]) or "").strip() if val: try: price = int(val) d = db.get_db() d.execute("UPDATE sms_pricing SET price_sats=? WHERE duration_min=?", (price, p["duration_min"])) d.commit() d.close() changed += 1 except ValueError: continue flash("Updated %d pricing rows." % changed, "ok") return redirect(url_for("sms_admin")) d = db.get_db() stats = { "active": d.execute("SELECT COUNT(*) c FROM sms_sessions WHERE status IN ('active','sms_received')").fetchone()["c"], "total_sessions": d.execute("SELECT COUNT(*) c FROM sms_sessions").fetchone()["c"], "sms_received": d.execute("SELECT COUNT(*) c FROM sms_messages").fetchone()["c"], "revenue_sats": d.execute("SELECT COALESCE(SUM(amount_sats),0) s FROM transactions WHERE type='sms'").fetchone()["s"], "provider_cost_usd": d.execute("SELECT COALESCE(SUM(CAST(provider_cost AS REAL)),0) s FROM sms_sessions").fetchone()["s"], } sessions = [dict(r) for r in d.execute( "SELECT s.*, u.username FROM sms_sessions s LEFT JOIN users u ON u.id=s.user_id " "ORDER BY s.id DESC LIMIT 50").fetchall()] pricing = get_pricing() d.close() try: smspool_balance = PROVIDER.get_balance() except Exception: smspool_balance = None return render_template("sms_admin.html", stats=stats, sessions=sessions, pricing=pricing, smspool_balance=smspool_balance)