Cosmo Pay v1: tariffs, YooKassa payments with receipt, polling, Remnawave expireAt + SoloBot keys.expiry_time renewal

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 """<!doctype html><html lang="ru"><head><meta charset="utf-8">
+<meta name="viewport" content="width=device-width,initial-scale=1"><title>Cosmo VPN</title>
+<style>body{font-family:system-ui,sans-serif;background:#17121c;color:#f2ddf9;display:flex;min-height:100vh;
+align-items:center;justify-content:center;margin:0;text-align:center;padding:24px}
+a{display:inline-block;margin-top:20px;padding:14px 22px;border-radius:14px;background:#d8468e;color:#fff;text-decoration:none}</style>
+</head><body><div><h2>Оплата обрабатывается</h2><p>Вернитесь в приложение Cosmo VPN — подписка продлится автоматически.</p>
+<a href="cosmovpn://payment-return">Открыть Cosmo VPN</a></div></body></html>"""

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

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