1044 lines
46 KiB
Python
1044 lines
46 KiB
Python
"""
|
|
Cosmo Pay v2 — in-app subscription renewal / trial conversion 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 (same class for paid keys; standard + VIP for trials) with SoloBot's own
|
|
device pricing -> create a YooKassa payment (owner's shop, receipt as SoloBot does) -> poll YooKassa ->
|
|
on success apply exactly what SoloBot's own renewal does (handlers/keys/key_renew.complete_key_renewal):
|
|
Remnawave: expireAt, hwidDeviceLimit (0 = unlimited), trafficLimitBytes; clear HWID devices if the
|
|
device limit shrinks (renewal.renew_on_remnawave);
|
|
SoloBot DB: expiry_time, tariff_id, selected/current device & traffic limits, selected_price_rub,
|
|
expiry notification flags reset.
|
|
SoloBot's API PATCH /keys/edit is NOT used (it zeroes the device limit).
|
|
|
|
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 base64
|
|
import hashlib
|
|
import hmac
|
|
import html
|
|
import importlib.util
|
|
import json
|
|
from urllib.parse import urlparse
|
|
import logging
|
|
import os
|
|
import re
|
|
import secrets
|
|
import socket
|
|
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, RedirectResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
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})/?$")
|
|
OFFER_GROUP = "cosmo"
|
|
PAID_GROUPS = {"cosmo", "vip"} # renew within the same device class
|
|
TRIAL_GROUPS = {"trial"} # convert: standard (configurable) + VIP
|
|
PENDING_TTL_S = 3 * 3600
|
|
# App users have no Telegram account: SoloBot needs a users row (keys.tg_id FK), so they get synthetic negative
|
|
# ids far below any real chat id. The bot finds them by key name ("email") like everyone else.
|
|
SYNTH_TG_BASE = -1_000_000_000_000
|
|
APP_SOURCE = "utm_CosmoApp"
|
|
APP_CLUSTER = "CosmoCluster"
|
|
TRIALS_PER_IP_DAY = 3
|
|
TRIALS_PER_DAY = 300
|
|
YK_TIMEOUT = httpx.Timeout(20, connect=3)
|
|
DAY_MS = 86_400_000
|
|
GB = 1024 ** 3
|
|
|
|
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
|
|
RW_HEADERS = {"Authorization": f"Bearer {RW_TOKEN}"}
|
|
SB_PARAMS = {"tg_id": ENV["SOLOBOT_ADMIN_TG_ID"]}
|
|
SB_HEADERS = {"X-Token": ENV["SOLOBOT_API_TOKEN"]}
|
|
|
|
|
|
# ------------------------------------------------------------------ storage
|
|
PAY_COLUMNS = {
|
|
"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",
|
|
"devices": "integer", "old_tariff_id": "integer",
|
|
"new_key": "integer", "hwid_hash": "text", "link": "text",
|
|
}
|
|
|
|
|
|
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(" + ", ".join(f"{k} {v}" for k, v in PAY_COLUMNS.items()) + ")")
|
|
have = {r["name"] for r in con.execute("pragma table_info(payments)")}
|
|
for k, v in PAY_COLUMNS.items():
|
|
if k not in have:
|
|
con.execute(f"alter table payments add column {k} {v.replace(' primary key', '')}")
|
|
# One app device (HWID) -> its subscription; one trial per device.
|
|
con.execute("create table if not exists devices(hwid_hash text primary key, client_id text, email text, "
|
|
"tg_id integer, trial integer, ip text, created integer, updated integer)")
|
|
# Bot messages (admin mailings) for app users without Telegram; the app polls them.
|
|
con.execute("create table if not exists inbox(id integer primary key autoincrement, tg_id integer, text text, created integer)")
|
|
con.execute("create index if not exists inbox_tg on inbox(tg_id, id)")
|
|
# Web Push subscriptions of the PWA (iPhone and browsers): mailings + expiry reminders are pushed by the server.
|
|
con.execute("create table if not exists push_subs(endpoint text primary key, sub_json text, hwid_hash text, "
|
|
"sub_token text, reminded text, created integer, updated integer)")
|
|
return con
|
|
|
|
|
|
def token_hash(token: str) -> str:
|
|
return hashlib.sha256(("cosmo-pay:" + token).encode()).hexdigest()[:24]
|
|
|
|
|
|
HWID_RE = re.compile(r"^[0-9a-f]{32}$")
|
|
|
|
|
|
def hwid_hash(hwid: str) -> str:
|
|
return hashlib.sha256(("cosmo-pay-hwid:" + hwid).encode()).hexdigest()[:32]
|
|
|
|
|
|
# ------------------------------------------------------------------ SoloBot (read-only API)
|
|
_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:
|
|
if (k.get("remnawave_link") or "").rstrip("/").endswith("/" + token):
|
|
return k
|
|
return None
|
|
|
|
|
|
def tariff_title(name: str) -> str:
|
|
return (name or "").split("•")[0].strip() or name
|
|
|
|
|
|
# ------------------------------------------------------------------ pricing (mirrors SoloBot resolve_price_to_charge)
|
|
def _int_list(values):
|
|
out = []
|
|
for v in values or []:
|
|
try:
|
|
out.append(int(v))
|
|
except (TypeError, ValueError):
|
|
pass
|
|
return sorted(set(out))
|
|
|
|
|
|
def device_options(t) -> list[int]:
|
|
return _int_list(t.get("device_options")) if t.get("configurable") else []
|
|
|
|
|
|
def base_devices(t) -> int | None:
|
|
opts = device_options(t)
|
|
if opts:
|
|
return min(opts)
|
|
return t.get("device_limit") # None = unlimited
|
|
|
|
|
|
def price_for(t, devices: int | None) -> tuple[int, int | None]:
|
|
"""(price_rub, device_limit to apply; None = unlimited) for tariff t and requested devices."""
|
|
base_price = int(t.get("price_rub") or 0)
|
|
opts = device_options(t)
|
|
if not opts:
|
|
return base_price, t.get("device_limit")
|
|
base = min(opts)
|
|
target = devices if devices in opts else base
|
|
add = 0
|
|
if target > base:
|
|
overrides = t.get("device_overrides") or {}
|
|
ov = overrides.get(str(target), overrides.get(target)) if isinstance(overrides, dict) else None
|
|
add = int(ov) if ov is not None else (target - base) * int(t.get("device_step_rub") or 0)
|
|
return base_price + add, target
|
|
|
|
|
|
def default_devices_for(key, t) -> int | None:
|
|
opts = device_options(t)
|
|
if not opts:
|
|
return None
|
|
sel = key.get("selected_device_limit")
|
|
try:
|
|
sel = int(sel) if sel is not None else None
|
|
except (TypeError, ValueError):
|
|
sel = None
|
|
return sel if sel in opts else min(opts)
|
|
|
|
|
|
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"))
|
|
grp = cur.get("group_code") if cur else None
|
|
pool = [t for t in tariffs if t.get("is_active") and t.get("group_code") == OFFER_GROUP
|
|
and (t.get("price_rub") or 0) > 0 and (t.get("duration_days") or 0) > 0]
|
|
if grp in PAID_GROUPS:
|
|
cls_unlimited = cur.get("device_limit") is None
|
|
offer = [t for t in pool if (t.get("device_limit") is None) == cls_unlimited]
|
|
elif grp in TRIAL_GROUPS:
|
|
offer = pool
|
|
else:
|
|
offer = []
|
|
offer.sort(key=lambda t: (t.get("device_limit") is None, t.get("sort_order") is None, t.get("sort_order") or 0, t["duration_days"]))
|
|
return cur, offer
|
|
|
|
|
|
def tariff_payload(key, t):
|
|
opts = device_options(t)
|
|
d = default_devices_for(key, t)
|
|
price, _ = price_for(t, d)
|
|
return {"id": t["id"], "title": tariff_title(t["name"]), "days": t["duration_days"],
|
|
"price_rub": price, "base_price_rub": int(t.get("price_rub") or 0),
|
|
"unlimited_devices": t.get("device_limit") is None and not opts,
|
|
"device_options": opts, "default_devices": d,
|
|
"device_step_rub": int(t.get("device_step_rub") or 0) if opts else 0,
|
|
"device_overrides": t.get("device_overrides") or {} if opts else {}}
|
|
|
|
|
|
# ------------------------------------------------------------------ SoloBot DB write (same fields as SoloBot renewal)
|
|
def db_apply_renewal(client_id: str, expiry_ms: int, tariff_id: int, devices: int | None, price_rub: int):
|
|
if not re.fullmatch(r"[0-9a-fA-F-]{32,40}", client_id):
|
|
raise ValueError("bad client_id")
|
|
dev = "null" if devices is None else str(int(devices))
|
|
sql = ("update keys set expiry_time = :'exp', tariff_id = :'tid', "
|
|
f"selected_device_limit = {dev}, current_device_limit = {dev}, "
|
|
"selected_traffic_limit = null, current_traffic_limit = null, selected_price_rub = :'price', "
|
|
"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"tid={int(tariff_id)}", "-v", f"price={int(price_rub)}",
|
|
"-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=RW_HEADERS, timeout=15)
|
|
r.raise_for_status()
|
|
return r.json().get("response", {})
|
|
|
|
|
|
async def rw_apply(client, client_id: str, expiry_ms: int, devices: int | None, traffic_gb: int | None):
|
|
iso = datetime.fromtimestamp(expiry_ms / 1000, tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.000Z")
|
|
body = {"uuid": client_id, "expireAt": iso,
|
|
"hwidDeviceLimit": int(devices or 0), # SoloBot convention: 0 = unlimited
|
|
"trafficLimitBytes": int(traffic_gb) * GB if traffic_gb else 0}
|
|
r = await client.patch(f"{REMNAWAVE_API}/users", headers=RW_HEADERS, json=body, timeout=20)
|
|
if r.status_code >= 400:
|
|
raise RuntimeError(f"remnawave PATCH {r.status_code}: {r.text[:150]}")
|
|
resp = r.json().get("response") or {}
|
|
if not str(resp.get("expireAt", "")).startswith(iso[:16]):
|
|
raise RuntimeError(f"remnawave expireAt mismatch: {resp.get('expireAt')}")
|
|
|
|
|
|
def psql(sql: str, **vars) -> str:
|
|
args = ["sudo", "-u", "postgres", "psql", "-d", "cosmonetdb", "-Atq", "-v", "ON_ERROR_STOP=1"]
|
|
for k, v in vars.items():
|
|
args += ["-v", f"{k}={v}"]
|
|
p = subprocess.run(args, input=sql, capture_output=True, text=True, timeout=20)
|
|
if p.returncode != 0:
|
|
raise RuntimeError(f"psql rc={p.returncode}: {p.stderr.strip()}"[:200])
|
|
return p.stdout.strip()
|
|
|
|
|
|
def gen_key_email() -> str:
|
|
alphabet = "abcdefghijklmnopqrstuvwxyz0123456789"
|
|
return "".join(secrets.choice(alphabet) for _ in range(8))
|
|
|
|
|
|
def app_squads() -> list[str]:
|
|
out = psql("select inbound_id from servers where cluster_name = :'c' and enabled "
|
|
"and lower(panel_type) = 'remnawave' and inbound_id is not null;\n", c=APP_CLUSTER)
|
|
squads = [s for s in out.splitlines() if re.fullmatch(r"[0-9a-f-]{36}", s)]
|
|
if not squads:
|
|
raise RuntimeError("no remnawave squads for app cluster")
|
|
return squads
|
|
|
|
|
|
def db_key_exists(client_id: str) -> bool:
|
|
return psql("select count(*) from keys where client_id = :'cid';\n", cid=client_id) == "1"
|
|
|
|
|
|
def db_insert_app_key(client_id: str, email: str, link: str, expiry_ms: int, tariff_id: int,
|
|
devices: int | None, price_rub: int | None, trial: bool) -> int:
|
|
"""New SoloBot user (synthetic tg_id, source Cosmo App) + key, in one transaction. Returns tg_id."""
|
|
dev = "null" if devices is None else str(int(devices))
|
|
price = "null" if price_rub is None else str(int(price_rub))
|
|
sql = (
|
|
"begin;\n"
|
|
"select pg_advisory_xact_lock(7710301);\n"
|
|
"with n as (select coalesce(min(tg_id), :'base') - 1 as tg from users where tg_id <= :'base'),\n"
|
|
f" u as (insert into users (tg_id, first_name, language_code, trial, source_code) "
|
|
"select tg, 'Cosmo App', 'ru', 1, :'src' from n returning tg_id)\n" # like SoloBot: trial=1 once a key exists
|
|
"insert into keys (tg_id, client_id, email, created_at, expiry_time, key, server_id, notified, notified_24h, "
|
|
"remnawave_link, is_frozen, tariff_id, selected_device_limit, current_device_limit, selected_price_rub) "
|
|
f"select tg_id, :'cid', :'email', :'now', :'exp', :'link', :'cluster', false, false, :'link', false, :'tid', "
|
|
f"{dev}, {dev}, {price} from u returning tg_id;\n"
|
|
"commit;\n"
|
|
)
|
|
out = psql(sql, base=SYNTH_TG_BASE, src=APP_SOURCE, cid=client_id, email=email, now=int(time.time() * 1000),
|
|
exp=int(expiry_ms), link=link, cluster=APP_CLUSTER, tid=int(tariff_id))
|
|
tg = [l for l in out.splitlines() if re.fullmatch(r"-\d+", l)]
|
|
if not tg:
|
|
raise RuntimeError("db insert returned no tg_id")
|
|
return int(tg[-1])
|
|
|
|
|
|
async def rw_create_user(client, client_id: str, email: str, expiry_ms: int, devices: int | None, traffic_gb: int | None):
|
|
iso = datetime.fromtimestamp(expiry_ms / 1000, tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.000Z")
|
|
body = {"username": email, "uuid": client_id, "trafficLimitStrategy": "NO_RESET", "expireAt": iso,
|
|
"activeInternalSquads": app_squads(), "hwidDeviceLimit": int(devices or 0), "description": "Cosmo App"}
|
|
if traffic_gb:
|
|
body["trafficLimitBytes"] = int(traffic_gb) * GB
|
|
r = await client.post(f"{REMNAWAVE_API}/users", headers=RW_HEADERS, json=body, timeout=20)
|
|
if r.status_code >= 400:
|
|
raise RuntimeError(f"remnawave create {r.status_code}: {r.text[:150]}")
|
|
return r.json().get("response") or {}
|
|
|
|
|
|
async def rw_delete_user(client, client_id: str):
|
|
r = await client.delete(f"{REMNAWAVE_API}/users/{client_id}", headers=RW_HEADERS, timeout=20)
|
|
if r.status_code >= 400:
|
|
log.error("remnawave delete %s.. failed %s", client_id[:8], r.status_code)
|
|
|
|
|
|
async def ensure_app_key(client, client_id: str, email: str, expiry_ms: int, t, devices: int | None,
|
|
price_rub: int | None, trial: bool) -> str:
|
|
"""Idempotent: Remnawave user + SoloBot user/key for an app device. Returns the subscription link."""
|
|
r = await client.get(f"{REMNAWAVE_API}/users/{client_id}", headers=RW_HEADERS, timeout=15)
|
|
created = False
|
|
if r.status_code == 404 or (r.status_code < 400 and not (r.json().get("response") or {}).get("uuid")):
|
|
user = await rw_create_user(client, client_id, email, expiry_ms, devices, t.get("traffic_limit"))
|
|
created = True
|
|
elif r.status_code >= 400:
|
|
raise RuntimeError(f"remnawave get {r.status_code}")
|
|
else:
|
|
user = r.json()["response"]
|
|
await rw_apply(client, client_id, expiry_ms, devices, t.get("traffic_limit"))
|
|
link = (user.get("subscriptionUrl") or "").strip()
|
|
if not SUB_URL_RE.match(link):
|
|
raise RuntimeError("unexpected subscription url from remnawave")
|
|
if not db_key_exists(client_id):
|
|
try:
|
|
db_insert_app_key(client_id, email, link, expiry_ms, t["id"], devices, price_rub, trial)
|
|
except Exception:
|
|
if created:
|
|
await rw_delete_user(client, client_id)
|
|
raise
|
|
_cache["keys"] = (0.0, [])
|
|
return link
|
|
|
|
|
|
async def rw_clear_hwid(client, client_id: str):
|
|
r = await client.post(f"{REMNAWAVE_API}/hwid/devices/delete-all", headers=RW_HEADERS,
|
|
json={"userUuid": client_id}, timeout=20)
|
|
if r.status_code >= 400:
|
|
log.warning("remnawave hwid delete-all %s: %s", r.status_code, r.text[:150])
|
|
|
|
|
|
# ------------------------------------------------------------------ YooKassa
|
|
async def yk_create(client, amount: float, description: str, metadata: dict, return_url: str | None = None):
|
|
value = f"{amount:.2f}"
|
|
body = {
|
|
"amount": {"value": value, "currency": "RUB"},
|
|
"capture": True,
|
|
"confirmation": {"type": "redirect", "return_url": return_url or 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"}
|
|
try:
|
|
r = await yk_request(client, "POST", "/payments", json=body, headers={"Idempotence-Key": str(uuid.uuid4())})
|
|
except httpx.HTTPError as e:
|
|
log.error("yookassa create failed: %s", type(e).__name__)
|
|
raise HTTPException(502, "payment provider unreachable")
|
|
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 yk_request(client, "GET", f"/payments/{yk_id}")
|
|
r.raise_for_status()
|
|
return r.json()
|
|
|
|
|
|
# One of api.yookassa.ru's addresses intermittently drops connections from this host, so we connect to each
|
|
# resolved address ourselves (TLS still verified for api.yookassa.ru via SNI) and stick to the one that works.
|
|
# Only failed connects move on to the next address, so a request is never sent twice.
|
|
YK_HOST = "api.yookassa.ru"
|
|
_yk_ips = {"ts": 0.0, "ips": [], "good": None}
|
|
|
|
|
|
async def _yk_addresses():
|
|
if time.time() - _yk_ips["ts"] > 300 or not _yk_ips["ips"]:
|
|
infos = await asyncio.get_running_loop().getaddrinfo(YK_HOST, 443, family=socket.AF_INET, type=socket.SOCK_STREAM)
|
|
_yk_ips["ips"] = sorted({i[4][0] for i in infos})
|
|
_yk_ips["ts"] = time.time()
|
|
good = _yk_ips["good"]
|
|
return ([good] if good in _yk_ips["ips"] else []) + [ip for ip in _yk_ips["ips"] if ip != good]
|
|
|
|
|
|
async def yk_request(client, method: str, path: str, **kw):
|
|
last = None
|
|
headers = {"Host": YK_HOST, **kw.pop("headers", {})} # same Idempotence-Key on every attempt
|
|
for _ in range(2):
|
|
for ip in await _yk_addresses():
|
|
try:
|
|
r = await client.request(method, f"https://{ip}/v3{path}", auth=YK_AUTH, timeout=YK_TIMEOUT,
|
|
headers=headers, extensions={"sni_hostname": YK_HOST}, **kw)
|
|
_yk_ips["good"] = ip
|
|
return r
|
|
except (httpx.ConnectError, httpx.ConnectTimeout) as e:
|
|
last = e
|
|
if _yk_ips["good"] == ip:
|
|
_yk_ips["good"] = None
|
|
log.warning("yookassa %s unreachable (%s), trying next address", ip, type(e).__name__)
|
|
raise last
|
|
|
|
|
|
# ------------------------------------------------------------------ app devices (HWID -> subscription)
|
|
def device_bind(con, hh: str, client_id: str, email: str, trial: bool, ip: str | None):
|
|
now = int(time.time())
|
|
con.execute("insert into devices(hwid_hash, client_id, email, trial, ip, created, updated) values(?,?,?,?,?,?,?) "
|
|
"on conflict(hwid_hash) do update set client_id=excluded.client_id, email=excluded.email, "
|
|
"trial=max(devices.trial, excluded.trial), updated=excluded.updated",
|
|
(hh, client_id, email, 1 if trial else 0, ip, now, now))
|
|
con.commit()
|
|
|
|
|
|
async def device_key(client, con, hh: str):
|
|
"""(devices row, live SoloBot key or None). The bot deletes expired keys, so the row may outlive its key."""
|
|
row = con.execute("select * from devices where hwid_hash=?", (hh,)).fetchone()
|
|
if not row or not row["client_id"]:
|
|
return row, None
|
|
keys = await sb_get(client, "keys", ttl=20)
|
|
return row, next((k for k in keys if k["client_id"] == row["client_id"]), None)
|
|
|
|
|
|
async def trial_tariff(client):
|
|
if bool(getattr(CFG, "TRIAL_TIME_DISABLE", False)):
|
|
return None
|
|
tariffs = await sb_get(client, "tariffs", ttl=300)
|
|
ts = [t for t in tariffs if t.get("is_active") and t.get("group_code") in TRIAL_GROUPS and (t.get("duration_days") or 0) > 0]
|
|
return sorted(ts, key=lambda t: (t.get("sort_order") or 0, t["id"]))[0] if ts else None
|
|
|
|
|
|
def new_user_offer(tariffs):
|
|
pool = [t for t in tariffs if t.get("is_active") and t.get("group_code") == OFFER_GROUP
|
|
and (t.get("price_rub") or 0) > 0 and (t.get("duration_days") or 0) > 0]
|
|
pool.sort(key=lambda t: (t.get("device_limit") is None, t.get("sort_order") is None, t.get("sort_order") or 0, t["duration_days"]))
|
|
return pool
|
|
|
|
|
|
# ------------------------------------------------------------------ Web Push (PWA)
|
|
VAPID_KEY_FILE = "/etc/cosmo-pay-vapid.pem"
|
|
VAPID_SUB = "https://bot.cosmonet.online"
|
|
PUSH_HOSTS = ("push.apple.com", "fcm.googleapis.com", "push.services.mozilla.com", "notify.windows.com")
|
|
REMIND_TEXT = {
|
|
"3": "Подписка заканчивается — осталось меньше 3 дней. Продлите её в Cosmo VPN.",
|
|
"1": "Подписка закончится меньше чем через сутки. Продлите её, чтобы VPN не отключился.",
|
|
"0": "Подписка закончилась. Продлите её в Cosmo VPN, чтобы снова подключиться.",
|
|
}
|
|
_vapid_pub = None
|
|
|
|
|
|
def vapid_public_key() -> str:
|
|
global _vapid_pub
|
|
if _vapid_pub is None:
|
|
from cryptography.hazmat.primitives import serialization
|
|
with open(VAPID_KEY_FILE, "rb") as f:
|
|
key = serialization.load_pem_private_key(f.read(), None)
|
|
raw = key.public_key().public_bytes(serialization.Encoding.X962, serialization.PublicFormat.UncompressedPoint)
|
|
_vapid_pub = base64.urlsafe_b64encode(raw).rstrip(b"=").decode()
|
|
return _vapid_pub
|
|
|
|
|
|
def _webpush_sync(sub_json: str, payload: dict) -> int:
|
|
from pywebpush import WebPushException, webpush
|
|
try:
|
|
r = webpush(subscription_info=json.loads(sub_json), data=json.dumps(payload, ensure_ascii=False),
|
|
vapid_private_key=VAPID_KEY_FILE, vapid_claims={"sub": VAPID_SUB}, ttl=86400, timeout=10)
|
|
return r.status_code
|
|
except WebPushException as e:
|
|
return e.response.status_code if e.response is not None else 0
|
|
except Exception as e:
|
|
log.warning("webpush failed: %s", type(e).__name__)
|
|
return 0
|
|
|
|
|
|
async def push_send(con, row, text: str, url: str) -> bool:
|
|
st = await asyncio.to_thread(_webpush_sync, row["sub_json"], {"title": "Cosmo VPN", "body": text, "url": url})
|
|
if st in (404, 410): # subscription gone (app removed / permission revoked)
|
|
con.execute("delete from push_subs where endpoint=?", (row["endpoint"],)); con.commit()
|
|
return 200 <= st < 300
|
|
|
|
|
|
async def push_key(client, con, row):
|
|
if row["sub_token"]:
|
|
return await find_key(client, row["sub_token"])
|
|
_, key = await device_key(client, con, row["hwid_hash"])
|
|
return key
|
|
|
|
|
|
async def push_to_tg(client, tg_id: int, text: str):
|
|
try:
|
|
con = db()
|
|
for row in con.execute("select * from push_subs").fetchall():
|
|
key = await push_key(client, con, row)
|
|
if key and key.get("tg_id") == tg_id:
|
|
await push_send(con, row, text, f"{PUBLIC_BASE}/app/")
|
|
except Exception as e:
|
|
log.error("push mailing: %s", e)
|
|
|
|
|
|
def remind_stage(expiry_ms: int | None) -> str | None:
|
|
if not expiry_ms:
|
|
return None
|
|
left = expiry_ms - int(time.time() * 1000)
|
|
if left > 3650 * DAY_MS:
|
|
return None
|
|
if left <= 0:
|
|
return "0"
|
|
if left <= DAY_MS:
|
|
return "1"
|
|
if left <= 3 * DAY_MS:
|
|
return "3"
|
|
return None
|
|
|
|
|
|
async def push_reminders(client):
|
|
while True:
|
|
await asyncio.sleep(1800)
|
|
try:
|
|
if not os.path.exists(VAPID_KEY_FILE):
|
|
continue
|
|
con = db()
|
|
for row in con.execute("select * from push_subs").fetchall():
|
|
key = await push_key(client, con, row)
|
|
exp = key.get("expiry_time") if key else None
|
|
stage = remind_stage(exp)
|
|
mark = f"{exp}|{stage}"
|
|
if stage and row["reminded"] != mark and await push_send(con, row, REMIND_TEXT[stage], f"{PUBLIC_BASE}/app/?renew=1"):
|
|
con.execute("update push_subs set reminded=?, updated=? where endpoint=?",
|
|
(mark, int(time.time()), row["endpoint"])); con.commit()
|
|
except Exception as e:
|
|
log.error("push reminders: %s", e)
|
|
|
|
|
|
# ------------------------------------------------------------------ renewal (serialized, idempotent)
|
|
renew_lock = asyncio.Lock()
|
|
|
|
|
|
def _row(con, yk_id):
|
|
return con.execute("select * from payments where yk_id=?", (yk_id,)).fetchone()
|
|
|
|
|
|
async def process_payment(client, yk_id: str):
|
|
"""pending -> succeeded (absolute target fixed) -> renewed. Safe to call repeatedly."""
|
|
async with renew_lock:
|
|
con = db()
|
|
row = _row(con, yk_id)
|
|
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 _row(con, yk_id)
|
|
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 _row(con, yk_id)
|
|
if row["new_key"]:
|
|
cur_ms = 0 # subscription is created on success
|
|
else:
|
|
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
|
|
target = max(int(time.time() * 1000), cur_ms, other) + 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 = _row(con, yk_id)
|
|
if row["status"] in ("succeeded", "error") and row["target_expiry"]:
|
|
try:
|
|
tariffs = await sb_get(client, "tariffs", ttl=300)
|
|
t = next(x for x in tariffs if x["id"] == row["tariff_id"])
|
|
devices = row["devices"] if device_options(t) else t.get("device_limit")
|
|
if row["new_key"]:
|
|
link = await ensure_app_key(client, row["client_id"], row["email"], row["target_expiry"], t,
|
|
devices, int(row["amount"]), trial=False)
|
|
device_bind(con, row["hwid_hash"], row["client_id"], row["email"], trial=False, ip=None)
|
|
con.execute("update payments set status='renewed', link=?, error=null, updated=? where yk_id=?",
|
|
(link, now, yk_id)); con.commit()
|
|
log.info("payment %s created app key %s.. tariff %s devices %s", yk_id, row["client_id"][:8],
|
|
row["tariff_id"], devices if devices is not None else "unlimited")
|
|
return _row(con, yk_id)
|
|
# SoloBot clears HWID devices when the limit shrinks (renew_on_remnawave).
|
|
keys = await sb_get(client, "keys", ttl=0)
|
|
k = next((x for x in keys if x["client_id"] == row["client_id"]), {}) or {}
|
|
old_limit = k.get("current_device_limit") or k.get("selected_device_limit")
|
|
if old_limit is None and row["old_tariff_id"]:
|
|
old_limit = next((x.get("device_limit") for x in tariffs if x["id"] == row["old_tariff_id"]), None)
|
|
shrink = devices is not None and old_limit is not None and int(devices) < int(old_limit)
|
|
await rw_apply(client, row["client_id"], row["target_expiry"], devices, t.get("traffic_limit"))
|
|
if shrink:
|
|
await rw_clear_hwid(client, row["client_id"])
|
|
db_apply_renewal(row["client_id"], row["target_expiry"], row["tariff_id"], devices, int(row["amount"]))
|
|
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.. tariff %s devices %s until %s", yk_id, row["client_id"][:8],
|
|
row["tariff_id"], devices if devices is not None else "unlimited",
|
|
datetime.fromtimestamp(row["target_expiry"] / 1000, tz=timezone.utc).isoformat())
|
|
except Exception as e: # 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 _row(con, yk_id)
|
|
|
|
|
|
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))
|
|
reminders = asyncio.create_task(push_reminders(app.state.client))
|
|
yield
|
|
task.cancel()
|
|
reminders.cancel()
|
|
await app.state.client.aclose()
|
|
|
|
|
|
app = FastAPI(lifespan=lifespan, docs_url=None, redoc_url=None, openapi_url=None)
|
|
_hits: dict[str, list[float]] = {}
|
|
WEB_DIR = "/opt/cosmo-pay/web"
|
|
|
|
|
|
@app.middleware("http")
|
|
async def web_cache_headers(request: Request, call_next):
|
|
resp = await call_next(request)
|
|
path = request.url.path
|
|
if path.startswith("/app"):
|
|
# The PWA shell and service worker must update promptly; icons may be cached.
|
|
resp.headers["Cache-Control"] = "public, max-age=86400" if "/icons/" in path else "no-cache"
|
|
return resp
|
|
|
|
|
|
def rate_limit(request: Request, per_min: int = 30):
|
|
# Separate bucket per endpoint, so status polling can't starve e.g. trial activation.
|
|
ip = request.headers.get("x-real-ip") or (request.client.host if request.client else "?")
|
|
bucket = f"{ip} {request.url.path}"
|
|
now = time.time()
|
|
lst = [t for t in _hits.get(bucket, []) if now - t < 60]
|
|
if len(lst) >= per_min:
|
|
raise HTTPException(429, "too many requests")
|
|
lst.append(now)
|
|
_hits[bucket] = lst
|
|
if len(_hits) > 50_000:
|
|
_hits.clear()
|
|
|
|
|
|
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 | None = None
|
|
hwid: str | None = None
|
|
|
|
|
|
class PayReq(SubReq):
|
|
tariff_id: int
|
|
devices: int | None = None
|
|
return_to: str | None = None # "web": come back to the PWA instead of the Android deep-link page
|
|
|
|
|
|
class StatusReq(SubReq):
|
|
payment_id: str
|
|
|
|
|
|
class DeviceReq(BaseModel):
|
|
hwid: str
|
|
|
|
|
|
class InboxReq(SubReq):
|
|
since: int | None = None
|
|
|
|
|
|
class InternalMsg(BaseModel):
|
|
tg_id: int
|
|
text: str
|
|
|
|
|
|
INTERNAL_TOKEN_FILE = "/etc/cosmo-pay-internal.token"
|
|
INBOX_KEEP_S = 30 * 86400
|
|
|
|
|
|
def plain_text(html_text: str) -> str:
|
|
t = re.sub(r"<br\s*/?>", "\n", html_text or "", flags=re.I)
|
|
t = re.sub(r"<[^>]+>", "", t)
|
|
return html.unescape(t).strip()[:2000]
|
|
|
|
|
|
def client_ip(request: Request) -> str:
|
|
return request.headers.get("x-real-ip") or (request.client.host if request.client else "?")
|
|
|
|
|
|
def parse_hwid(hwid: str | None) -> str:
|
|
h = (hwid or "").strip().lower()
|
|
if not HWID_RE.match(h):
|
|
raise HTTPException(400, "bad device id")
|
|
return h
|
|
|
|
|
|
async def resolve(client, req: SubReq):
|
|
"""-> (key or None, identity hashes accepted for this caller, hwid hash or None).
|
|
A subscription URL identifies an existing key; a bare HWID identifies the app device (maybe without a key)."""
|
|
if req.subscription_url:
|
|
token = parse_token(req.subscription_url)
|
|
key = await find_key(client, token)
|
|
if not key:
|
|
raise HTTPException(404, "subscription not found")
|
|
return key, {token_hash(token)}, None
|
|
hw = parse_hwid(req.hwid)
|
|
hh = hwid_hash(hw)
|
|
_, key = await device_key(client, db(), hh)
|
|
ids = {token_hash("hwid:" + hw)}
|
|
if key:
|
|
m = SUB_URL_RE.match(key.get("remnawave_link") or "")
|
|
if m:
|
|
ids.add(token_hash(m.group(1)))
|
|
return key, ids, hh
|
|
|
|
|
|
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)
|
|
client = request.app.state.client
|
|
key, _, _ = await resolve(client, req)
|
|
if not key: # app device without a subscription (or the bot already deleted its expired one): new purchase
|
|
offer = new_user_offer(await sb_get(client, "tariffs", ttl=300))
|
|
return {"renewable": bool(offer), "expire_at": None, "current_tariff": None, "current_devices": None,
|
|
"is_trial": False, "is_new": True, "bot_url": BOT_LINK, "tariffs": [tariff_payload({}, t) for t in offer]}
|
|
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,
|
|
"current_devices": key.get("selected_device_limit"),
|
|
"is_trial": bool(cur and cur.get("group_code") in TRIAL_GROUPS),
|
|
"bot_url": BOT_LINK,
|
|
"tariffs": [tariff_payload(key, t) for t in offer],
|
|
}
|
|
|
|
|
|
@app.post("/api/v1/payments")
|
|
async def create_payment(req: PayReq, request: Request):
|
|
rate_limit(request, per_min=10)
|
|
client = request.app.state.client
|
|
key, ids, hh = await resolve(client, req)
|
|
ident = sorted(ids)[0] if len(ids) == 1 else token_hash("hwid:" + parse_hwid(req.hwid))
|
|
if key:
|
|
_, offer = await offer_for_key(client, key)
|
|
else:
|
|
offer = new_user_offer(await sb_get(client, "tariffs", ttl=300))
|
|
key = {}
|
|
t = next((x for x in offer if x["id"] == req.tariff_id), None)
|
|
if not t:
|
|
raise HTTPException(400, "tariff not available")
|
|
opts = device_options(t)
|
|
if opts and req.devices is not None and req.devices not in opts:
|
|
raise HTTPException(400, "devices not available")
|
|
devices = (req.devices if req.devices is not None else default_devices_for(key, t)) if opts else None
|
|
price, devices = price_for(t, devices)
|
|
title = tariff_title(t["name"])
|
|
dev_txt = f", {devices} устр." if opts else ""
|
|
web = req.return_to == "web"
|
|
p = await yk_create(client, float(price), f"Cosmo VPN: {title}{dev_txt}",
|
|
{"source": "cosmo_web" if web else "cosmo_app", "tariff_id": str(t["id"])},
|
|
return_url=f"{PUBLIC_BASE}/app/?paid=1" if web else None)
|
|
new_key = not key
|
|
client_id = key.get("client_id") or str(uuid.uuid4()) # new key: ids fixed now so creation retries are idempotent
|
|
email = key.get("email") or gen_key_email()
|
|
con = db()
|
|
now = int(time.time())
|
|
con.execute(f"insert into payments({','.join(PAY_COLUMNS)}) values({','.join('?' * len(PAY_COLUMNS))})",
|
|
(p["id"], ident, client_id, email, key.get("tg_id"), t["id"], title,
|
|
t["duration_days"], float(price), "pending", None, None, now, now, devices, key.get("tariff_id"),
|
|
1 if new_key else 0, hh, None))
|
|
con.commit()
|
|
log.info("payment %s created: tariff %s, devices %s, %s RUB, %s %s..", p["id"], t["id"],
|
|
devices if devices is not None else "unlimited", price, "new key" if new_key else "key", client_id[:8])
|
|
return {"payment_id": p["id"], "confirmation_url": p["confirmation"]["confirmation_url"], "amount_rub": price}
|
|
|
|
|
|
@app.post("/api/v1/payments/status")
|
|
async def payment_status(req: StatusReq, request: Request):
|
|
rate_limit(request, per_min=60)
|
|
if req.subscription_url:
|
|
ids = {token_hash(parse_token(req.subscription_url))}
|
|
else:
|
|
ids = {token_hash("hwid:" + parse_hwid(req.hwid))}
|
|
con = db()
|
|
row = con.execute("select * from payments where yk_id=?", (req.payment_id,)).fetchone()
|
|
if not row or row["token_hash"] not in ids:
|
|
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,
|
|
"subscription_url": row["link"] if status == "renewed" and row["new_key"] else None}
|
|
|
|
|
|
@app.post("/internal/inbox")
|
|
async def internal_inbox(msg: InternalMsg, request: Request):
|
|
"""SoloBot (patched sender) hands over a mailing for an app user. Local calls only: requests coming through
|
|
nginx carry X-Real-IP and are refused, and the shared token must match."""
|
|
if request.headers.get("x-real-ip") or (request.client and request.client.host != "127.0.0.1"):
|
|
raise HTTPException(404)
|
|
try:
|
|
token = open(INTERNAL_TOKEN_FILE).read().strip()
|
|
except OSError:
|
|
raise HTTPException(503)
|
|
if not token or not hmac.compare_digest(token, request.headers.get("x-internal-token", "")):
|
|
raise HTTPException(403)
|
|
if msg.tg_id > SYNTH_TG_BASE:
|
|
raise HTTPException(400, "not an app user")
|
|
text = plain_text(msg.text)
|
|
if not text:
|
|
return {"ok": True, "stored": False}
|
|
con = db()
|
|
now = int(time.time())
|
|
con.execute("insert into inbox(tg_id, text, created) values(?,?,?)", (msg.tg_id, text, now))
|
|
con.execute("delete from inbox where created < ?", (now - INBOX_KEEP_S,))
|
|
con.commit()
|
|
asyncio.create_task(push_to_tg(request.app.state.client, msg.tg_id, text)) # PWA users get it pushed right away
|
|
return {"ok": True, "stored": True}
|
|
|
|
|
|
class PushSubReq(SubReq):
|
|
subscription: dict
|
|
|
|
|
|
@app.get("/api/v1/push/key")
|
|
async def push_key_endpoint():
|
|
try:
|
|
return {"key": vapid_public_key()}
|
|
except OSError:
|
|
raise HTTPException(503, "push not configured")
|
|
|
|
|
|
@app.post("/api/v1/push/subscribe")
|
|
async def push_subscribe(req: PushSubReq, request: Request):
|
|
rate_limit(request, per_min=10)
|
|
ep = str(req.subscription.get("endpoint") or "")
|
|
keys = req.subscription.get("keys") or {}
|
|
host = urlparse(ep).hostname or ""
|
|
# Only real push services (the server posts to this URL).
|
|
if not ep.startswith("https://") or len(ep) > 1000 or not any(host == h or host.endswith("." + h) for h in PUSH_HOSTS) \
|
|
or not isinstance(keys, dict) or not keys.get("p256dh") or not keys.get("auth"):
|
|
raise HTTPException(400, "bad push subscription")
|
|
if req.subscription_url:
|
|
token, hh = parse_token(req.subscription_url), None
|
|
else:
|
|
token, hh = None, hwid_hash(parse_hwid(req.hwid))
|
|
sub_json = json.dumps({"endpoint": ep, "keys": {"p256dh": str(keys["p256dh"]), "auth": str(keys["auth"])}})
|
|
now = int(time.time())
|
|
con = db()
|
|
con.execute("insert into push_subs(endpoint, sub_json, hwid_hash, sub_token, created, updated) values(?,?,?,?,?,?) "
|
|
"on conflict(endpoint) do update set sub_json=excluded.sub_json, hwid_hash=excluded.hwid_hash, "
|
|
"sub_token=excluded.sub_token, updated=excluded.updated", (ep, sub_json, hh, token, now, now))
|
|
con.commit()
|
|
return {"ok": True}
|
|
|
|
|
|
@app.post("/api/v1/inbox")
|
|
async def inbox(req: InboxReq, request: Request):
|
|
"""Polled by the app: current expiry (for local reminders) and new messages for this subscription.
|
|
since=None -> no messages, just the latest id (the app starts from 'now', not from old mailings)."""
|
|
rate_limit(request)
|
|
client = request.app.state.client
|
|
try:
|
|
key, _, _ = await resolve(client, req)
|
|
except HTTPException as e:
|
|
if e.status_code == 404:
|
|
return {"expire_at": None, "last_id": 0, "messages": []}
|
|
raise
|
|
if not key:
|
|
return {"expire_at": None, "last_id": 0, "messages": []}
|
|
con = db()
|
|
tg = key.get("tg_id")
|
|
last = con.execute("select coalesce(max(id), 0) from inbox where tg_id=?", (tg,)).fetchone()[0]
|
|
msgs = []
|
|
if req.since is not None:
|
|
msgs = [{"id": r["id"], "text": r["text"], "created": r["created"]} for r in con.execute(
|
|
"select id, text, created from inbox where tg_id=? and id>? order by id limit 20", (tg, int(req.since)))]
|
|
return {"expire_at": iso_ms(key.get("expiry_time")), "last_id": last, "messages": msgs}
|
|
|
|
|
|
@app.post("/api/v1/device")
|
|
async def device(req: DeviceReq, request: Request):
|
|
"""App start without a subscription: restore this device's subscription, or say whether a trial is available."""
|
|
rate_limit(request)
|
|
client = request.app.state.client
|
|
hh = hwid_hash(parse_hwid(req.hwid))
|
|
row, key = await device_key(client, db(), hh)
|
|
t = await trial_tariff(client)
|
|
return {
|
|
"subscription_url": key.get("remnawave_link") if key else None,
|
|
"trial_available": bool(t) and not key and not (row and row["trial"]),
|
|
"trial_days": t["duration_days"] if t else None,
|
|
"trial_devices": t.get("device_limit") if t else None,
|
|
}
|
|
|
|
|
|
@app.post("/api/v1/trial")
|
|
async def trial(req: DeviceReq, request: Request):
|
|
rate_limit(request, per_min=5)
|
|
client = request.app.state.client
|
|
hh = hwid_hash(parse_hwid(req.hwid))
|
|
ip = client_ip(request)
|
|
async with renew_lock:
|
|
con = db()
|
|
row, key = await device_key(client, con, hh)
|
|
if key:
|
|
return {"subscription_url": key.get("remnawave_link")}
|
|
if row and row["trial"]:
|
|
raise HTTPException(409, "trial already used on this device")
|
|
day = int(time.time()) - 86400
|
|
if con.execute("select count(*) from devices where trial=1 and ip=? and created>?", (ip, day)).fetchone()[0] >= TRIALS_PER_IP_DAY \
|
|
or con.execute("select count(*) from devices where trial=1 and created>?", (day,)).fetchone()[0] >= TRIALS_PER_DAY:
|
|
log.warning("trial limit reached for %s", ip)
|
|
raise HTTPException(429, "too many trials, try later")
|
|
t = await trial_tariff(client)
|
|
if not t:
|
|
raise HTTPException(403, "trial disabled")
|
|
client_id, email = str(uuid.uuid4()), gen_key_email()
|
|
expiry = int(time.time() * 1000) + int(t["duration_days"]) * DAY_MS
|
|
try:
|
|
link = await ensure_app_key(client, client_id, email, expiry, t, t.get("device_limit"), None, trial=True)
|
|
except Exception as e:
|
|
log.error("trial creation failed: %s", e)
|
|
raise HTTPException(502, "could not create subscription")
|
|
device_bind(con, hh, client_id, email, trial=True, ip=ip)
|
|
log.info("trial key %s.. (%s) created for app device, %s days", client_id[:8], email, t["duration_days"])
|
|
return {"subscription_url": link}
|
|
|
|
|
|
@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>"""
|
|
|
|
|
|
@app.get("/app")
|
|
async def web_app_redirect():
|
|
return RedirectResponse(f"{PUBLIC_BASE}/app/", status_code=301)
|
|
|
|
|
|
# PWA (iPhone and any browser): static shell served from /opt/cosmo-pay/web, same API as the Android app.
|
|
if os.path.isdir(WEB_DIR):
|
|
app.mount("/app", StaticFiles(directory=WEB_DIR, html=True), name="web")
|