commit 8439736b5471110f80c8755582664fb80b4fcc36 Author: Cosmo VPN Date: Wed Sep 30 16:02:18 2026 +0300 Cosmo Pay v1: tariffs, YooKassa payments with receipt, polling, Remnawave expireAt + SoloBot keys.expiry_time renewal diff --git a/app.py b/app.py new file mode 100644 index 0000000..fdda7cc --- /dev/null +++ b/app.py @@ -0,0 +1,387 @@ +""" +Cosmo Pay — in-app subscription renewal for Cosmo VPN. + +Flow: app sends its subscription URL -> we find the SoloBot key (keys.remnawave_link) via SoloBot's +read-only API -> offer tariffs of the same class -> create a YooKassa payment (owner's shop, receipt as +SoloBot does) -> poll YooKassa -> on success set the new absolute expiry in Remnawave (PATCH /users, +expireAt only — limits untouched) and in SoloBot's DB (keys.expiry_time + reset expiry-notification flags). + +Secrets are read on the server only: SoloBot config.py (YooKassa keys, receipt settings, Remnawave API +token) and /etc/cosmo-pay.env (SoloBot admin API token). Nothing secret is logged or returned. +""" +import asyncio +import hashlib +import importlib.util +import json +import logging +import os +import re +import sqlite3 +import subprocess +import time +import uuid +from contextlib import asynccontextmanager +from datetime import datetime, timezone + +import httpx +from fastapi import FastAPI, HTTPException, Request +from fastapi.responses import HTMLResponse +from pydantic import BaseModel + +BOT_DIR = "/root/cosmonetvpn_bot" +ENV_FILE = "/etc/cosmo-pay.env" +DB_FILE = "/var/lib/cosmo-pay/pay.db" +SOLOBOT_API = "http://127.0.0.1:3003/api" +REMNAWAVE_API = "https://rw.cosmonet.online/api" +YOOKASSA_API = "https://api.yookassa.ru/v3" +PUBLIC_BASE = "https://bot.cosmonet.online/cosmo-pay" +BOT_LINK = "https://t.me/cosmonetvpn_bot" +SUB_URL_RE = re.compile(r"^https://sub\.cosmonet\.online/([A-Za-z0-9_-]{8,64})/?$") +RENEWABLE_GROUPS = {"cosmo", "vip"} +OFFER_GROUP = "cosmo" +PENDING_TTL_S = 3 * 3600 +DAY_MS = 86_400_000 + +logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") +log = logging.getLogger("cosmo-pay") + + +# ------------------------------------------------------------------ configuration (server-side only) +def _load_bot_config(): + spec = importlib.util.spec_from_file_location("solobot_config", f"{BOT_DIR}/config.py") + mod = importlib.util.module_from_spec(spec) + cwd = os.getcwd() + try: + os.chdir(BOT_DIR) + spec.loader.exec_module(mod) + finally: + os.chdir(cwd) + return mod + + +def _load_env(): + env = {} + with open(ENV_FILE) as f: + for line in f: + line = line.strip() + if line and not line.startswith("#") and "=" in line: + k, v = line.split("=", 1) + env[k.strip()] = v.strip() + return env + + +CFG = _load_bot_config() +ENV = _load_env() +YK_AUTH = (str(CFG.YOOKASSA_SHOP_ID), str(CFG.YOOKASSA_SECRET_KEY)) +RECEIPT_EMAIL = getattr(CFG, "EMAIL", "") or "" +VAT_CODE = int(getattr(CFG, "VAT_CODE", 1) or 1) +SBP_ONLY = bool(getattr(CFG, "YOOKASSA_SBP_ONLY", False)) +RW_TOKEN = CFG.REMNAWAVE_ACCESS_TOKEN +SB_PARAMS = {"tg_id": ENV["SOLOBOT_ADMIN_TG_ID"]} +SB_HEADERS = {"X-Token": ENV["SOLOBOT_API_TOKEN"]} + + +# ------------------------------------------------------------------ storage +def db(): + os.makedirs(os.path.dirname(DB_FILE), exist_ok=True) + con = sqlite3.connect(DB_FILE, timeout=10) + con.row_factory = sqlite3.Row + con.execute("""create table if not exists payments( + yk_id text primary key, token_hash text, client_id text, email text, tg_id integer, + tariff_id integer, tariff_name text, days integer, amount real, + status text, target_expiry integer, error text, created integer, updated integer)""") + return con + + +def token_hash(token: str) -> str: + return hashlib.sha256(("cosmo-pay:" + token).encode()).hexdigest()[:24] + + +# ------------------------------------------------------------------ SoloBot (read-only API) + DB write +_cache = {"keys": (0.0, []), "tariffs": (0.0, [])} + + +async def sb_get(client: httpx.AsyncClient, path: str, ttl: float): + ts, data = _cache[path] + if time.time() - ts < ttl: + return data + r = await client.get(f"{SOLOBOT_API}/{path}/", params=SB_PARAMS, headers=SB_HEADERS, timeout=15) + r.raise_for_status() + data = r.json() + _cache[path] = (time.time(), data) + return data + + +async def find_key(client, token: str): + keys = await sb_get(client, "keys", ttl=20) + for k in keys: + link = (k.get("remnawave_link") or "").rstrip("/") + if link.endswith("/" + token): + return k + return None + + +def tariff_title(name: str) -> str: + return (name or "").split("•")[0].strip() or name + + +async def offer_for_key(client, key): + tariffs = await sb_get(client, "tariffs", ttl=300) + by_id = {t["id"]: t for t in tariffs} + cur = by_id.get(key.get("tariff_id")) + if not cur or cur.get("group_code") not in RENEWABLE_GROUPS: + return cur, [] + cls = cur.get("device_limit") + offer = [t for t in tariffs + if t.get("is_active") and t.get("group_code") == OFFER_GROUP + and t.get("device_limit") == cls and (t.get("price_rub") or 0) > 0 and (t.get("duration_days") or 0) > 0] + offer.sort(key=lambda t: (t.get("sort_order") is None, t.get("sort_order") or 0, t["duration_days"])) + return cur, offer + + +def db_set_expiry(client_id: str, expiry_ms: int): + if not re.fullmatch(r"[0-9a-fA-F-]{32,40}", client_id): + raise ValueError("bad client_id") + sql = ("update keys set expiry_time = :'exp', notified = false, notified_24h = false " + "where client_id = :'cid' returning client_id;\n") + p = subprocess.run(["sudo", "-u", "postgres", "psql", "-d", "cosmonetdb", "-Atq", "-v", "ON_ERROR_STOP=1", + "-v", f"exp={int(expiry_ms)}", "-v", f"cid={client_id}"], + input=sql, capture_output=True, text=True, timeout=20) + if p.returncode != 0 or p.stdout.strip() != client_id: + raise RuntimeError(f"db update failed: rc={p.returncode} {p.stderr.strip()}"[:200]) + + +# ------------------------------------------------------------------ Remnawave +async def rw_get_user(client, client_id: str): + r = await client.get(f"{REMNAWAVE_API}/users/{client_id}", headers={"Authorization": f"Bearer {RW_TOKEN}"}, timeout=15) + r.raise_for_status() + return r.json().get("response", {}) + + +async def rw_set_expiry(client, client_id: str, expiry_ms: int): + iso = datetime.fromtimestamp(expiry_ms / 1000, tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.000Z") + r = await client.patch(f"{REMNAWAVE_API}/users", headers={"Authorization": f"Bearer {RW_TOKEN}"}, + json={"uuid": client_id, "expireAt": iso}, timeout=20) + if r.status_code >= 400: + raise RuntimeError(f"remnawave PATCH {r.status_code}: {r.text[:150]}") + got = (r.json().get("response") or {}).get("expireAt", "") + if not got.startswith(iso[:16]): + raise RuntimeError(f"remnawave expireAt mismatch: {got}") + + +# ------------------------------------------------------------------ YooKassa +async def yk_create(client, amount: float, description: str, metadata: dict): + value = f"{amount:.2f}" + body = { + "amount": {"value": value, "currency": "RUB"}, + "capture": True, + "confirmation": {"type": "redirect", "return_url": f"{PUBLIC_BASE}/return"}, + "description": description[:128], + "metadata": metadata, + } + if RECEIPT_EMAIL: + body["receipt"] = { + "customer": {"email": RECEIPT_EMAIL}, + "items": [{"description": description[:128], "quantity": "1.00", + "amount": {"value": value, "currency": "RUB"}, "vat_code": VAT_CODE, + "payment_mode": "full_payment", "payment_subject": "service"}], + } + if SBP_ONLY: + body["payment_method_data"] = {"type": "sbp"} + r = await client.post(f"{YOOKASSA_API}/payments", auth=YK_AUTH, json=body, + headers={"Idempotence-Key": str(uuid.uuid4())}, timeout=20) + if r.status_code >= 400: + log.error("yookassa create failed %s: %s", r.status_code, r.text[:300]) + raise HTTPException(502, "payment provider error") + return r.json() + + +async def yk_get(client, yk_id: str): + r = await client.get(f"{YOOKASSA_API}/payments/{yk_id}", auth=YK_AUTH, timeout=15) + r.raise_for_status() + return r.json() + + +# ------------------------------------------------------------------ renewal (serialized, idempotent) +renew_lock = asyncio.Lock() + + +async def process_payment(client, yk_id: str): + """Advance one payment: pending -> succeeded(target fixed) -> renewed. Safe to call repeatedly.""" + async with renew_lock: + con = db() + row = con.execute("select * from payments where yk_id=?", (yk_id,)).fetchone() + if not row or row["status"] in ("renewed", "canceled"): + return row + now = int(time.time()) + if row["status"] == "pending": + p = await yk_get(client, yk_id) + st = p.get("status") + if st == "canceled": + con.execute("update payments set status='canceled', updated=? where yk_id=?", (now, yk_id)); con.commit() + return con.execute("select * from payments where yk_id=?", (yk_id,)).fetchone() + if st != "succeeded" or not p.get("paid"): + return row + if f"{float(p['amount']['value']):.2f}" != f"{row['amount']:.2f}": + con.execute("update payments set status='error', error='amount mismatch', updated=? where yk_id=?", (now, yk_id)); con.commit() + log.error("payment %s amount mismatch", yk_id) + return con.execute("select * from payments where yk_id=?", (yk_id,)).fetchone() + # Fix the absolute target once: max(now, current expiry in Remnawave, other renewals of this key) + days. + user = await rw_get_user(client, row["client_id"]) + cur_ms = int(datetime.fromisoformat(user["expireAt"].replace("Z", "+00:00")).timestamp() * 1000) + other = con.execute("select max(target_expiry) m from payments where client_id=? and status in ('succeeded','renewed')", + (row["client_id"],)).fetchone()["m"] or 0 + base = max(int(time.time() * 1000), cur_ms, other) + target = base + int(row["days"]) * DAY_MS + con.execute("update payments set status='succeeded', target_expiry=?, updated=? where yk_id=?", (target, now, yk_id)); con.commit() + row = con.execute("select * from payments where yk_id=?", (yk_id,)).fetchone() + if row["status"] in ("succeeded", "error") and row["target_expiry"]: + try: + await rw_set_expiry(client, row["client_id"], row["target_expiry"]) + db_set_expiry(row["client_id"], row["target_expiry"]) + con.execute("update payments set status='renewed', error=null, updated=? where yk_id=?", (now, yk_id)); con.commit() + _cache["keys"] = (0.0, []) + log.info("payment %s renewed key %s.. until %s", yk_id, row["client_id"][:8], + datetime.fromtimestamp(row["target_expiry"] / 1000, tz=timezone.utc).isoformat()) + except Exception as e: # stays retryable; target is absolute so retries are idempotent + con.execute("update payments set status='error', error=?, updated=? where yk_id=?", (str(e)[:200], now, yk_id)); con.commit() + log.error("payment %s renewal failed: %s", yk_id, e) + return con.execute("select * from payments where yk_id=?", (yk_id,)).fetchone() + + +async def poller(client): + while True: + try: + con = db() + ids = [r["yk_id"] for r in con.execute( + "select yk_id from payments where status in ('pending','succeeded','error') and created > ?", + (int(time.time()) - PENDING_TTL_S,))] + for yk_id in ids: + await process_payment(client, yk_id) + except Exception as e: + log.error("poller: %s", e) + await asyncio.sleep(15) + + +# ------------------------------------------------------------------ HTTP API +@asynccontextmanager +async def lifespan(app): + app.state.client = httpx.AsyncClient() + task = asyncio.create_task(poller(app.state.client)) + yield + task.cancel() + await app.state.client.aclose() + + +app = FastAPI(lifespan=lifespan, docs_url=None, redoc_url=None, openapi_url=None) + +_hits: dict[str, list[float]] = {} + + +def rate_limit(request: Request, per_min: int = 30): + ip = request.headers.get("x-real-ip") or (request.client.host if request.client else "?") + now = time.time() + lst = [t for t in _hits.get(ip, []) if now - t < 60] + if len(lst) >= per_min: + raise HTTPException(429, "too many requests") + lst.append(now) + _hits[ip] = lst + + +def parse_token(sub_url: str) -> str: + m = SUB_URL_RE.match((sub_url or "").strip()) + if not m: + raise HTTPException(400, "unknown subscription") + return m.group(1) + + +class SubReq(BaseModel): + subscription_url: str + + +class PayReq(BaseModel): + subscription_url: str + tariff_id: int + + +class StatusReq(BaseModel): + subscription_url: str + payment_id: str + + +def iso_ms(ms): + return datetime.fromtimestamp(ms / 1000, tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") if ms else None + + +@app.get("/health") +async def health(): + return {"ok": True} + + +@app.post("/api/v1/tariffs") +async def tariffs(req: SubReq, request: Request): + rate_limit(request) + token = parse_token(req.subscription_url) + client = request.app.state.client + key = await find_key(client, token) + if not key: + raise HTTPException(404, "subscription not found") + cur, offer = await offer_for_key(client, key) + return { + "renewable": bool(offer), + "expire_at": iso_ms(key.get("expiry_time")), + "current_tariff": tariff_title(cur.get("name")) if cur else None, + "bot_url": BOT_LINK, + "tariffs": [{"id": t["id"], "title": tariff_title(t["name"]), "days": t["duration_days"], + "price_rub": t["price_rub"]} for t in offer], + } + + +@app.post("/api/v1/payments") +async def create_payment(req: PayReq, request: Request): + rate_limit(request, per_min=10) + token = parse_token(req.subscription_url) + client = request.app.state.client + key = await find_key(client, token) + if not key: + raise HTTPException(404, "subscription not found") + _, offer = await offer_for_key(client, key) + t = next((x for x in offer if x["id"] == req.tariff_id), None) + if not t: + raise HTTPException(400, "tariff not available") + title = tariff_title(t["name"]) + p = await yk_create(client, float(t["price_rub"]), f"Cosmo VPN: продление подписки — {title}", + {"source": "cosmo_app", "tariff_id": str(t["id"])}) + con = db() + now = int(time.time()) + con.execute("insert into payments values(?,?,?,?,?,?,?,?,?,?,?,?,?,?)", + (p["id"], token_hash(token), key["client_id"], key.get("email"), key.get("tg_id"), t["id"], title, + t["duration_days"], float(t["price_rub"]), "pending", None, None, now, now)) + con.commit() + log.info("payment %s created: tariff %s, %s RUB, key %s..", p["id"], t["id"], t["price_rub"], key["client_id"][:8]) + return {"payment_id": p["id"], "confirmation_url": p["confirmation"]["confirmation_url"]} + + +@app.post("/api/v1/payments/status") +async def payment_status(req: StatusReq, request: Request): + rate_limit(request, per_min=60) + token = parse_token(req.subscription_url) + con = db() + row = con.execute("select * from payments where yk_id=? and token_hash=?", (req.payment_id, token_hash(token))).fetchone() + if not row: + raise HTTPException(404, "payment not found") + row = await process_payment(request.app.state.client, req.payment_id) + status = {"pending": "pending", "succeeded": "processing", "error": "processing", + "renewed": "renewed", "canceled": "canceled"}.get(row["status"], "pending") + return {"status": status, "expire_at": iso_ms(row["target_expiry"]) if status == "renewed" else None} + + +@app.get("/return", response_class=HTMLResponse) +async def return_page(): + return """ +Cosmo VPN + +

Оплата обрабатывается

Вернитесь в приложение Cosmo VPN — подписка продлится автоматически.

+Открыть Cosmo VPN
""" diff --git a/cosmo-pay.service b/cosmo-pay.service new file mode 100644 index 0000000..2be7c4e --- /dev/null +++ b/cosmo-pay.service @@ -0,0 +1,15 @@ +[Unit] +Description=Cosmo Pay (in-app renewal for Cosmo VPN) +After=network-online.target postgresql.service bot.service +Wants=network-online.target + +[Service] +Type=simple +WorkingDirectory=/opt/cosmo-pay +ExecStart=/opt/cosmo-pay/venv/bin/uvicorn app:app --host 127.0.0.1 --port 3010 --workers 1 --proxy-headers --forwarded-allow-ips 127.0.0.1 +Restart=always +RestartSec=5 +User=root + +[Install] +WantedBy=multi-user.target diff --git a/install.sh b/install.sh new file mode 100644 index 0000000..0544588 --- /dev/null +++ b/install.sh @@ -0,0 +1,21 @@ +true +# Cosmo Pay install on the bot server. Expects app.py and cosmo-pay.service in /tmp/cosmo-pay-src/. +set -e +export DEBIAN_FRONTEND=noninteractive +S=/tmp/cosmo-pay-src +[ -f $S/app.py ] && [ -f $S/cosmo-pay.service ] || { echo "sources missing"; exit 1; } +[ -r /etc/cosmo-pay.env ] || { echo "/etc/cosmo-pay.env missing"; exit 1; } +mkdir -p /opt/cosmo-pay /var/lib/cosmo-pay; chmod 700 /var/lib/cosmo-pay +[ -d /opt/cosmo-pay/venv ] || python3 -m venv /opt/cosmo-pay/venv +/opt/cosmo-pay/venv/bin/pip install -q --upgrade pip >/dev/null +/opt/cosmo-pay/venv/bin/pip install -q "fastapi>=0.110" "uvicorn>=0.29" "httpx>=0.27" >/dev/null +/opt/cosmo-pay/venv/bin/python -c "import fastapi, uvicorn, httpx; print('deps: fastapi', fastapi.__version__, '| uvicorn', uvicorn.__version__, '| httpx', httpx.__version__)" +[ -f /opt/cosmo-pay/app.py ] && cp -a /opt/cosmo-pay/app.py /opt/cosmo-pay/app.py.bak.$(date +%F-%H%M) +install -m 600 $S/app.py /opt/cosmo-pay/app.py +install -m 644 $S/cosmo-pay.service /etc/systemd/system/cosmo-pay.service +/opt/cosmo-pay/venv/bin/python -m py_compile /opt/cosmo-pay/app.py && echo "app compiles" +systemctl daemon-reload; systemctl enable cosmo-pay >/dev/null 2>&1; systemctl restart cosmo-pay; sleep 4 +echo "service: $(systemctl is-active cosmo-pay)"; journalctl -u cosmo-pay --since -30s --no-pager | tail -5 | cut -c1-200 +echo "local health: $(curl -s -m 5 http://127.0.0.1:3010/health)" +echo "listen: $(ss -tlnH '( sport = :3010 )' | awk '{print $4}')" +rm -rf $S