import asyncio
import datetime
import json
import os
import re
import secrets
import subprocess
import urllib.request
from fastapi import FastAPI, HTTPException, Request, Response, File, UploadFile
from fastapi.middleware.cors import CORSMiddleware
from fastapi import Body
from fastapi.responses import HTMLResponse, PlainTextResponse, FileResponse
import backup
import db
import legal
import links
import nodeprov
import payments
import settings
import totp
import webhooks
import xray_manager
from config import SITE_DOMAIN, SUB_DOMAIN, PANEL_DOMAIN, ADMIN_PATH, BASE_DIR
HWID_RE = re.compile(r"^[a-zA-Z0-9=-]{10,64}$")
ENV_PATH = os.path.join(BASE_DIR, ".env")
db.init_db()
def _update_env_var(key: str, value: str):
legal.update_env_var(key, value)
app = FastAPI(title="mbs-api")
app.add_middleware(
CORSMiddleware,
allow_origins=[f"https://{SITE_DOMAIN}", f"https://www.{SITE_DOMAIN}", f"https://{PANEL_DOMAIN}"],
allow_methods=["GET", "POST", "PATCH", "DELETE"],
allow_headers=["*"],
allow_credentials=True,
)
ADMIN_COOKIE = "mbs_admin"
def require_admin(request: Request):
token = request.cookies.get(ADMIN_COOKIE)
if not db.validate_admin_session(token):
raise HTTPException(401, "unauthorized")
def _days_left(sub: dict) -> int:
exp = datetime.datetime.fromisoformat(sub["expires_at"])
reference = datetime.datetime.fromisoformat(sub["held_at"]) if sub.get("held_at") else datetime.datetime.utcnow()
delta = exp - reference
return max(0, delta.days)
CLIENT_UA_MARKERS = (
"happ", "v2ray", "v2box", "nekoray", "nekobox", "clash", "hiddify",
"streisand", "shadowrocket", "sing-box", "singbox", "karing", "loon",
"quantumult", "surge", "stash",
)
def _is_app_client(user_agent: str) -> bool:
ua = (user_agent or "").lower()
return any(m in ua for m in CLIENT_UA_MARKERS)
SUB_PAGE_TEMPLATE = """
{brand_name} — подписка
"""
SUB_PAGE_EXPIRED_TEMPLATE = """
{brand_name} — подписка
{brand_name}
Подписка истекла
Доступ по этой ссылке закончился. Продли подписку в боте — ссылка останется той же, ничего заново настраивать не нужно.
Продлить в боте
"""
@app.get("/healthz")
def healthz():
return {"ok": True}
@app.get("/sub/{token}")
def get_subscription(token: str, request: Request):
user = db.get_user_by_token(token)
if not user:
raise HTTPException(404, "not found")
subs = db.list_active_subscriptions(tg_id=user["tg_id"])
ua = request.headers.get("user-agent", "")
if not _is_app_client(ua):
brand_name = settings.get_brand_name()
if not subs:
_, bot_username = settings.bot_credentials()
return HTMLResponse(SUB_PAGE_EXPIRED_TEMPLATE.format(bot_username=bot_username, brand_name=brand_name))
sub_url = f"https://{SUB_DOMAIN}/sub/{token}"
return HTMLResponse(SUB_PAGE_TEMPLATE.format(sub_url=sub_url, brand_name=brand_name))
hwid_cfg = settings.get_hwid_settings()
if hwid_cfg["enabled"]:
hwid = request.headers.get("x-hwid", "")
if not HWID_RE.match(hwid):
raise HTTPException(404, "hwid required")
limit = user["hwid_limit"] if user["hwid_limit"] is not None else hwid_cfg["fallback_limit"]
_, allowed = db.add_device_if_under_limit(
user["tg_id"], hwid, limit,
request.headers.get("x-device-os"),
request.headers.get("x-device-model"),
ua,
)
if not allowed:
raise HTTPException(404, "device limit reached", headers={"x-hwid-max-devices-reached": "true"})
content = links.build_subscription_text(subs)
return Response(content=content, media_type="text/plain")
@app.get("/api/cabinet/{token}")
def cabinet(token: str):
user = db.get_user_by_token(token)
if not user:
raise HTTPException(404, "not found")
subs = db.list_active_subscriptions(tg_id=user["tg_id"])
plans_by_code = settings.get_plans_by_code()
out = []
for s in subs:
plan = plans_by_code.get(s["plan"])
out.append({
"node": s["node"],
"plan": s["plan"],
"plan_label": plan["label"] if plan else s["plan"],
"expires_at": s["expires_at"],
"days_left": _days_left(s),
})
return {
"username": user["username"],
"subscriptions": out,
"sub_link": f"https://{SUB_DOMAIN}/sub/{token}",
"nodes_available": [n["label"] for n in db.list_nodes(enabled_only=True)],
}
@app.get("/mgmt-pubkey.txt", response_class=PlainTextResponse)
def mgmt_pubkey():
return nodeprov.ensure_mgmt_key() + "\n"
@app.get("/install/{token}.sh", response_class=PlainTextResponse)
def install_script(token: str):
node = db.get_node_by_token(token)
if not node:
raise HTTPException(404, "unknown token")
return nodeprov.render_install_script(node)
def _tg_send_message(tg_id: int, text: str):
bot_token, _ = settings.bot_credentials()
url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
data = json.dumps({"chat_id": tg_id, "text": text, "parse_mode": "HTML"}).encode()
req = urllib.request.Request(url, data=data, method="POST", headers={"Content-Type": "application/json"})
try:
urllib.request.urlopen(req, timeout=10)
except Exception:
pass
def _grant_paid_subscription(payment_id: str):
payment = db.get_payment(payment_id)
if not payment or payment["status"] == "paid":
return
plan = settings.get_plans_by_code().get(payment["plan"])
node = db.get_node(payment["node"])
if not plan or not node:
return
payment = db.mark_payment_paid(payment_id)
if not payment:
return
sub = db.create_subscription(payment["tg_id"], payment["node"], plan["days"], payment["plan"], source="payment")
xray_manager.add_client_to_node(node, sub["uuid"], email=sub["uuid"])
user = db.get_or_create_user(payment["tg_id"], None)
_tg_send_message(
payment["tg_id"],
f"Оплата получена\n\n"
f"Сервер: {node['label']}\n"
f"Срок: {plan['label']} — до {sub['expires_at'][:10]}\n\n"
f"Ссылка-подписка:\nhttps://{SUB_DOMAIN}/sub/{user['token']}",
)
webhooks.send("payment.paid", {
"tg_id": payment["tg_id"],
"amount": payment["amount"],
"provider": payment["provider"],
"node": payment["node"],
"plan": payment["plan"],
"subscription_uuid": sub["uuid"],
"expires_at": sub["expires_at"],
})
def _check_and_reconcile_payment(payment: dict) -> str:
if not payment.get("external_id"):
return payment["status"]
try:
status = payments.check_payment_status(payment["provider"], payment["external_id"])
except Exception:
return payment["status"]
if status in payments.PAID_STATUSES:
_grant_paid_subscription(payment["id"])
return "paid"
if status in payments.FAILED_STATUSES:
db.mark_payment_failed(payment["id"])
return "failed"
return payment["status"]
@app.get("/admin/api/payments")
def admin_list_payments(request: Request):
require_admin(request)
nodes_by_code = {n["code"]: n for n in db.list_nodes()}
plans_by_code = settings.get_plans_by_code()
out = []
for p in db.list_payments():
node = nodes_by_code.get(p["node"])
plan = plans_by_code.get(p["plan"])
out.append({
**p,
"node_label": node["label"] if node else p["node"],
"plan_label": plan["label"] if plan else p["plan"],
"provider_label": payments.PROVIDER_NAMES.get(p["provider"], p["provider"]),
})
return out
@app.post("/admin/api/payments/{payment_id}/check")
def admin_check_payment(payment_id: str, request: Request):
require_admin(request)
payment = db.get_payment(payment_id)
if not payment:
raise HTTPException(404, "not found")
if payment["status"] != "pending":
return {"status": payment["status"]}
status = _check_and_reconcile_payment(payment)
return {"status": status}
@app.get("/admin/api/payments/legal-settings")
def admin_get_legal_settings(request: Request):
require_admin(request)
return legal.get_settings()
@app.post("/admin/api/payments/legal-settings")
def admin_set_legal_settings(request: Request, body: dict = Body(...)):
require_admin(request)
for key in ("LEGAL_NAME", "LEGAL_INN", "REFUND_HOURS", "SUPPORT_CONTACT", "SUPPORT_EMAIL"):
if key in body:
_update_env_var(key, str(body[key]).strip())
if not legal.read_env_var("OFFER_EFFECTIVE_DATE"):
_update_env_var("OFFER_EFFECTIVE_DATE", datetime.date.today().strftime("%d.%m.%Y"))
return {"ok": True, "settings": legal.get_settings()}
@app.get("/admin/api/payments/yookassa-settings")
def admin_get_yookassa_settings(request: Request):
require_admin(request)
shop_id, secret_key = settings.yookassa_credentials()
return {
"enabled": settings.get_payment_settings()["yookassa_enabled"],
"shop_id": shop_id,
"has_secret": bool(secret_key),
}
@app.post("/admin/api/payments/yookassa-settings")
def admin_set_yookassa_settings(request: Request, body: dict = Body(...)):
require_admin(request)
shop_id = (body.get("shop_id") or "").strip()
secret_key = (body.get("secret_key") or "").strip()
if not shop_id or not secret_key:
raise HTTPException(400, "shop_id и secret_key обязательны")
try:
payments.validate_yookassa_credentials(shop_id, secret_key)
except Exception:
raise HTTPException(401, "ЮKassa не приняла эти ключи — проверь shop_id и секретный ключ")
_update_env_var("YOOKASSA_SHOP_ID", shop_id)
_update_env_var("YOOKASSA_SECRET_KEY", secret_key)
_update_env_var("YOOKASSA_ENABLED", "true")
_update_env_var("PAYMENTS_ENABLED", "true")
restarted = False
try:
subprocess.run(["systemctl", "restart", "mbs-bot"], check=True, timeout=15)
restarted = True
except Exception:
restarted = False
return {"ok": True, "restarted_bot": restarted}
@app.get("/admin/api/payments/platega-settings")
def admin_get_platega_settings(request: Request):
require_admin(request)
merchant_id, secret = settings.platega_credentials()
return {
"enabled": settings.get_payment_settings()["platega_enabled"],
"merchant_id": merchant_id,
"has_secret": bool(secret),
}
@app.post("/admin/api/payments/platega-settings")
def admin_set_platega_settings(request: Request, body: dict = Body(...)):
require_admin(request)
merchant_id = (body.get("merchant_id") or "").strip()
secret = (body.get("secret") or "").strip()
if not merchant_id or not secret:
raise HTTPException(400, "merchant_id и secret обязательны")
_update_env_var("PLATEGA_MERCHANT_ID", merchant_id)
_update_env_var("PLATEGA_SECRET", secret)
_update_env_var("PLATEGA_ENABLED", "true")
_update_env_var("PAYMENTS_ENABLED", "true")
restarted = False
try:
subprocess.run(["systemctl", "restart", "mbs-bot"], check=True, timeout=15)
restarted = True
except Exception:
restarted = False
return {"ok": True, "restarted_bot": restarted}
@app.get("/admin/api/payments/plan-settings")
def admin_get_plan_settings(request: Request):
require_admin(request)
return {
"payments_enabled": settings.get_payment_settings()["payments_enabled"],
"plans": settings.get_plans(),
}
@app.post("/admin/api/payments/plan-settings")
def admin_set_plan_settings(request: Request, body: dict = Body(...)):
require_admin(request)
prices = body.get("prices") or {}
known_codes = settings.PRICE_ENV_KEYS.keys()
clean_prices = {}
for code, value in prices.items():
if code not in known_codes:
continue
try:
price = int(value)
except (TypeError, ValueError):
raise HTTPException(400, f"цена для тарифа {code} должна быть целым числом")
if price < 0:
raise HTTPException(400, f"цена для тарифа {code} не может быть отрицательной")
clean_prices[code] = price
settings.set_plan_prices(clean_prices)
if "payments_enabled" in body:
_update_env_var("PAYMENTS_ENABLED", "true" if body["payments_enabled"] else "false")
return {
"payments_enabled": settings.get_payment_settings()["payments_enabled"],
"plans": settings.get_plans(),
}
@app.get("/admin/api/hwid-settings")
def admin_get_hwid_settings(request: Request):
require_admin(request)
return settings.get_hwid_settings()
@app.post("/admin/api/hwid-settings")
def admin_set_hwid_settings(request: Request, body: dict = Body(...)):
require_admin(request)
if "enabled" in body:
_update_env_var("HWID_LIMIT_ENABLED", "true" if body["enabled"] else "false")
if "fallback_limit" in body:
try:
limit = int(body["fallback_limit"])
except (TypeError, ValueError):
raise HTTPException(400, "лимит устройств должен быть целым числом")
if not (1 <= limit <= 1000):
raise HTTPException(400, "лимит устройств должен быть от 1 до 1000")
_update_env_var("HWID_FALLBACK_LIMIT", str(limit))
return settings.get_hwid_settings()
@app.get("/admin/api/webhook-settings")
def admin_get_webhook_settings(request: Request):
require_admin(request)
return {
"url": legal.read_env_var("WEBHOOK_URL", ""),
"secret": legal.read_env_var("WEBHOOK_SECRET", ""),
}
@app.post("/admin/api/webhook-settings")
def admin_set_webhook_settings(request: Request, body: dict = Body(...)):
require_admin(request)
url = (body.get("url") or "").strip()
_update_env_var("WEBHOOK_URL", url)
if url and not legal.read_env_var("WEBHOOK_SECRET", ""):
_update_env_var("WEBHOOK_SECRET", secrets.token_hex(24))
return {"url": legal.read_env_var("WEBHOOK_URL", ""), "secret": legal.read_env_var("WEBHOOK_SECRET", "")}
@app.post("/payments/webhook/yookassa")
async def yookassa_webhook(request: Request):
body = await request.json()
if not payments.verify_yookassa_notification(body):
raise HTTPException(400, "unexpected event")
obj = body.get("object", {})
payment_id = (obj.get("metadata") or {}).get("payment_id")
if not payment_id:
raise HTTPException(400, "missing payment_id")
payment = db.get_payment(payment_id)
if not payment or not payment.get("external_id"):
raise HTTPException(400, "unknown payment")
status = await asyncio.to_thread(payments.check_yookassa_payment, payment["external_id"])
if status not in payments.PAID_STATUSES:
return {"ok": True}
await asyncio.to_thread(_grant_paid_subscription, payment_id)
return {"ok": True}
@app.post("/payments/webhook/platega")
async def platega_webhook(request: Request):
raw = await request.body()
signature = request.headers.get("x-signature") or request.headers.get("signature") or ""
if not payments.verify_platega_signature(raw, signature):
raise HTTPException(401, "bad signature")
body = json.loads(raw)
status = (body.get("status") or "").lower()
payment_id = body.get("id") or body.get("paymentId")
if status not in ("succeeded", "success", "paid") or not payment_id:
return {"ok": True}
await asyncio.to_thread(_grant_paid_subscription, payment_id)
return {"ok": True}
@app.get("/pay/done", response_class=HTMLResponse)
def pay_done():
return (
""
""
"Оплата"
""
"Спасибо!
Возвращайся в Telegram — подписка придёт туда автоматически "
"в течение минуты после подтверждения оплаты.