austnv/remnawave-minishop
1
0
Fork 0
Code Issues 1 Pull Requests Activity

refactor: move wata env-config into module

Browse Source
This commit is contained in:
3252a8
2026-05-18 16:18:04 +03:00
parent 15b3f9c084
commit 74ef9d31b4
4 changed files with 150 additions and 156 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/bot/app/web/admin_settings_manifest.py
-55
View File
@@ -336,61 +336,6 @@ SETTINGS_MANIFEST: List[SettingField] = [
SettingField("PLATEGA_FAILED_URL", "url", "payments", "Failed URL", subsection="Platega"),
*_payment_presentation_fields("PLATEGA_SBP", "Platega SBP", default_icon="CreditCard"),
*_payment_presentation_fields("PLATEGA_CRYPTO", "Platega Crypto", default_icon="Bitcoin"),
# Wata
SettingField("WATA_ENABLED", "bool", "payments", "Enabled", subsection="Wata"),
SettingField(
"WATA_API_TOKEN",
"string",
"payments",
"API token",
subsection="Wata",
secret=True,
),
SettingField(
"WATA_BASE_URL",
"url",
"payments",
"Base URL",
placeholder="https://api.wata.pro/api/h2h",
subsection="Wata",
),
SettingField("WATA_RETURN_URL", "url", "payments", "Return URL", subsection="Wata"),
SettingField("WATA_FAILED_URL", "url", "payments", "Failed URL", subsection="Wata"),
SettingField(
"WATA_PAYMENT_LINK_TTL_DAYS",
"int",
"payments",
"Payment link lifetime (days)",
"1..30; Wata defaults to 3 days and allows up to 30 days.",
subsection="Wata",
min=1,
max=30,
),
SettingField(
"WATA_WEBHOOK_VERIFY_SIGNATURE",
"bool",
"payments",
"Verify webhook signature",
subsection="Wata",
),
SettingField(
"WATA_PUBLIC_KEY",
"text",
"payments",
"Webhook public key",
"Optional. If empty, the backend fetches it from Wata.",
subsection="Wata",
secret=True,
),
SettingField(
"WATA_TRUSTED_IPS",
"string",
"payments",
"Trusted IPs",
"Comma-separated IP addresses accepted for Wata webhooks.",
subsection="Wata",
),
*_payment_presentation_fields("WATA", "Wata", default_icon="WalletCards"),
# CryptoPay
SettingField("CRYPTOPAY_ENABLED", "bool", "payments", "Включена", subsection="CryptoPay"),
SettingField(
backend/bot/payment_providers/wata.py
+142 -16
View File
@@ -2,13 +2,15 @@ import base64
import json
import logging
from datetime import datetime, timedelta, timezone
from typing import Any, Dict, Optional, Tuple
from typing import Any, Dict, List, Optional, Tuple
from aiogram import Bot, F, Router, types
from aiohttp import web
from cryptography.exceptions import InvalidSignature
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding
from pydantic import Field, field_validator
from pydantic_settings import SettingsConfigDict
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import sessionmaker
@@ -19,7 +21,13 @@ from bot.utils.request_security import ip_in_allowlist, request_client_ip
from config.settings import Settings
from db.dal import payment_dal
from .base import PaymentProviderSpec, ServiceFactoryContext, WebAppPaymentContext
from .base import (
PaymentProviderSpec,
ProviderEnvConfig,
ProviderManifestField,
ServiceFactoryContext,
WebAppPaymentContext,
)
from .shared import (
HttpClientMixin,
PaymentSuccessRequest,
@@ -48,12 +56,74 @@ router = Router(name="user_subscription_payments_wata_router")
_LOG = "wata"
class WataConfig(ProviderEnvConfig):
model_config = SettingsConfigDict(
env_file=".env",
env_file_encoding="utf-8",
env_prefix="WATA_",
extra="ignore",
)
ENABLED: bool = Field(default=False)
API_TOKEN: Optional[str] = None
BASE_URL: str = Field(default="https://api.wata.pro/api/h2h")
RETURN_URL: Optional[str] = None
FAILED_URL: Optional[str] = None
PAYMENT_LINK_TTL_DAYS: int = Field(default=3)
WEBHOOK_VERIFY_SIGNATURE: bool = Field(default=True)
PUBLIC_KEY: Optional[str] = None
TRUSTED_IPS: str = Field(default="62.84.126.140,51.250.106.150")
@field_validator("PAYMENT_LINK_TTL_DAYS", mode="before")
@classmethod
def _clamp_ttl(cls, v):
if isinstance(v, str):
v = v.strip()
try:
value = int(v)
except (TypeError, ValueError):
return 3
return min(30, max(1, value))
@field_validator("API_TOKEN", "RETURN_URL", "FAILED_URL", "PUBLIC_KEY", mode="before")
@classmethod
def _strip_optional(cls, v):
if isinstance(v, str) and not v.strip():
return None
return v
@property
def webhook_path(self) -> str:
return "/webhook/wata"
@property
def trusted_ips_list(self) -> List[str]:
return [item.strip() for item in (self.TRUSTED_IPS or "").split(",") if item.strip()]
class WataPresentation(ProviderEnvConfig):
model_config = SettingsConfigDict(
env_file=".env",
env_file_encoding="utf-8",
env_prefix="PAYMENT_WATA_",
extra="ignore",
)
WEBAPP_LABEL_RU: Optional[str] = None
WEBAPP_LABEL_EN: Optional[str] = None
WEBAPP_ICON: Optional[str] = None
TELEGRAM_LABEL_RU: Optional[str] = None
TELEGRAM_LABEL_EN: Optional[str] = None
TELEGRAM_EMOJI: Optional[str] = None
class WataService(HttpClientMixin):
def __init__(
self,
*,
bot: Bot,
settings: Settings,
config: WataConfig,
i18n: JsonI18n,
async_session_factory: sessionmaker,
subscription_service: SubscriptionService,
@@ -62,21 +132,22 @@ class WataService(HttpClientMixin):
):
self.bot = bot
self.settings = settings
self.config = config
self.i18n = i18n
self.async_session_factory = async_session_factory
self.subscription_service = subscription_service
self.referral_service = referral_service
self.base_url = (settings.WATA_BASE_URL or "https://api.wata.pro/api/h2h").rstrip("/")
self.api_token = settings.WATA_API_TOKEN or ""
self.return_url = settings.WATA_RETURN_URL or f"https://t.me/{default_return_url}"
self.failed_url = settings.WATA_FAILED_URL or self.return_url
self.payment_link_ttl_days = settings.WATA_PAYMENT_LINK_TTL_DAYS
self.verify_webhook_signature = settings.WATA_WEBHOOK_VERIFY_SIGNATURE
self._public_key_pem = settings.WATA_PUBLIC_KEY
self.base_url = (config.BASE_URL or "https://api.wata.pro/api/h2h").rstrip("/")
self.api_token = config.API_TOKEN or ""
self.return_url = config.RETURN_URL or f"https://t.me/{default_return_url}"
self.failed_url = config.FAILED_URL or self.return_url
self.payment_link_ttl_days = config.PAYMENT_LINK_TTL_DAYS
self.verify_webhook_signature = config.WEBHOOK_VERIFY_SIGNATURE
self._public_key_pem = config.PUBLIC_KEY
self._init_http_client(total_timeout=20)
self.configured: bool = bool(settings.WATA_ENABLED and self.api_token)
self.configured: bool = bool(config.ENABLED and self.api_token)
if not self.configured:
logging.warning("WataService initialized but not fully configured. Payments disabled.")
@@ -119,7 +190,8 @@ class WataService(HttpClientMixin):
async def _get_public_key_pem(self) -> Optional[str]:
if self._public_key_pem:
return self._public_key_pem.replace("\\n", "\n")
value = self._public_key_pem
return value.replace("\\n", "\n") if isinstance(value, str) else None
session = await self._get_session()
try:
@@ -159,9 +231,8 @@ class WataService(HttpClientMixin):
return web.Response(status=503, text="wata_disabled")
client_ip = request_client_ip(request, trusted_proxies=self.settings.trusted_proxies)
if self.settings.wata_trusted_ips and not ip_in_allowlist(
client_ip, self.settings.wata_trusted_ips
):
trusted = self.config.trusted_ips_list
if trusted and not ip_in_allowlist(client_ip, trusted):
logging.warning("Wata webhook denied from unauthorized IP source.")
return web.Response(status=403, text="forbidden")
@@ -417,9 +488,12 @@ async def wata_webhook_route(request: web.Request) -> web.Response:
def create_service(ctx: ServiceFactoryContext) -> WataService:
bundle = ctx.config_for("wata_service")
config = bundle.config if bundle and isinstance(bundle.config, WataConfig) else WataConfig()
return WataService(
bot=ctx.bot,
settings=ctx.settings,
config=config,
i18n=ctx.i18n,
async_session_factory=ctx.async_session_factory,
subscription_service=ctx.subscription_service,
@@ -428,6 +502,55 @@ def create_service(ctx: ServiceFactoryContext) -> WataService:
)
_PRESENTATION_MANIFEST = tuple(
ProviderManifestField(
key=key, type=type_, label=label, description=description,
placeholder=placeholder, subsection="Wata",
target="presentation", attr=attr,
)
for key, type_, label, description, placeholder, attr in (
("PAYMENT_WATA_WEBAPP_LABEL_RU", "string", "WebApp button text (RU)",
"Custom Russian text shown in the Web App payment method button.", "", "WEBAPP_LABEL_RU"),
("PAYMENT_WATA_WEBAPP_LABEL_EN", "string", "WebApp button text (EN)",
"Custom English text shown in the Web App payment method button.", "", "WEBAPP_LABEL_EN"),
("PAYMENT_WATA_WEBAPP_ICON", "icon", "WebApp button icon",
"Lucide icon name rendered inside the Web App payment method button.",
"WalletCards", "WEBAPP_ICON"),
("PAYMENT_WATA_TELEGRAM_LABEL_RU", "string", "Telegram button text (RU)",
"Custom Russian text shown in Telegram bot payment buttons.", "", "TELEGRAM_LABEL_RU"),
("PAYMENT_WATA_TELEGRAM_LABEL_EN", "string", "Telegram button text (EN)",
"Custom English text shown in Telegram bot payment buttons.", "", "TELEGRAM_LABEL_EN"),
("PAYMENT_WATA_TELEGRAM_EMOJI", "string", "Telegram button emoji",
"Emoji prepended to the Telegram bot payment button when customized.",
"💳", "TELEGRAM_EMOJI"),
)
)
_CONFIG_MANIFEST = (
ProviderManifestField("WATA_ENABLED", "bool", "Enabled", subsection="Wata", attr="ENABLED"),
ProviderManifestField("WATA_API_TOKEN", "string", "API token", subsection="Wata",
secret=True, attr="API_TOKEN"),
ProviderManifestField("WATA_BASE_URL", "url", "Base URL",
placeholder="https://api.wata.pro/api/h2h",
subsection="Wata", attr="BASE_URL"),
ProviderManifestField("WATA_RETURN_URL", "url", "Return URL",
subsection="Wata", attr="RETURN_URL"),
ProviderManifestField("WATA_FAILED_URL", "url", "Failed URL",
subsection="Wata", attr="FAILED_URL"),
ProviderManifestField("WATA_PAYMENT_LINK_TTL_DAYS", "int", "Payment link lifetime (days)",
description="1..30; Wata defaults to 3 days and allows up to 30 days.",
subsection="Wata", min=1, max=30, attr="PAYMENT_LINK_TTL_DAYS"),
ProviderManifestField("WATA_WEBHOOK_VERIFY_SIGNATURE", "bool", "Verify webhook signature",
subsection="Wata", attr="WEBHOOK_VERIFY_SIGNATURE"),
ProviderManifestField("WATA_PUBLIC_KEY", "text", "Webhook public key",
description="Optional. If empty, the backend fetches it from Wata.",
subsection="Wata", secret=True, attr="PUBLIC_KEY"),
ProviderManifestField("WATA_TRUSTED_IPS", "string", "Trusted IPs",
description="Comma-separated IP addresses accepted for Wata webhooks.",
subsection="Wata", attr="TRUSTED_IPS"),
)
SPEC = PaymentProviderSpec(
id="wata",
provider_key="wata",
@@ -438,12 +561,15 @@ SPEC = PaymentProviderSpec(
telegram_labels={"ru": "Wata", "en": "Wata"},
telegram_emoji="💳",
pending_status="pending_wata",
enabled=lambda settings: settings.WATA_ENABLED,
enabled=lambda config: bool(getattr(config, "ENABLED", False)),
service_key="wata_service",
callback_prefix="pay_wata",
router=router,
create_service=create_service,
webhook_path=lambda settings: settings.wata_webhook_path,
webhook_path=lambda source: "/webhook/wata",
webhook_route=wata_webhook_route,
create_webapp_payment=create_webapp_payment,
config_class=WataConfig,
presentation_class=WataPresentation,
manifest_fields=_CONFIG_MANIFEST + _PRESENTATION_MANIFEST,
)
backend/config/settings.py
-81
View File
@@ -56,15 +56,6 @@ class PaymentSettings(BaseModel):
platega_crypto_method: int
platega_return_url: Optional[str]
platega_failed_url: Optional[str]
wata_enabled: bool
wata_api_token: Optional[str]
wata_base_url: str
wata_return_url: Optional[str]
wata_failed_url: Optional[str]
wata_payment_link_ttl_days: int
wata_webhook_verify_signature: bool
wata_public_key: Optional[str]
wata_trusted_ips: List[str]
cryptopay_enabled: bool
cryptopay_token: Optional[str]
cryptopay_network: str
@@ -248,25 +239,6 @@ class Settings(BaseSettings):
description="Comma-separated FreeKassa webhook IP allowlist.",
)
WATA_ENABLED: bool = Field(default=False)
WATA_API_TOKEN: Optional[str] = None
WATA_BASE_URL: str = Field(default="https://api.wata.pro/api/h2h")
WATA_RETURN_URL: Optional[str] = None
WATA_FAILED_URL: Optional[str] = None
WATA_PAYMENT_LINK_TTL_DAYS: int = Field(
default=3,
description="Payment link lifetime in days (1-30).",
)
WATA_WEBHOOK_VERIFY_SIGNATURE: bool = Field(default=True)
WATA_PUBLIC_KEY: Optional[str] = Field(
default=None,
description="Optional cached Wata RSA public key for webhook signature verification.",
)
WATA_TRUSTED_IPS: str = Field(
default="62.84.126.140,51.250.106.150",
description="Comma-separated Wata webhook IP allowlist.",
)
YOOKASSA_ENABLED: bool = Field(default=True)
STARS_ENABLED: bool = Field(default=True)
PAYMENT_METHODS_ORDER: Optional[str] = Field(
@@ -291,12 +263,6 @@ class Settings(BaseSettings):
PAYMENT_PLATEGA_CRYPTO_TELEGRAM_LABEL_RU: Optional[str] = None
PAYMENT_PLATEGA_CRYPTO_TELEGRAM_LABEL_EN: Optional[str] = None
PAYMENT_PLATEGA_CRYPTO_TELEGRAM_EMOJI: Optional[str] = None
PAYMENT_WATA_WEBAPP_LABEL_RU: Optional[str] = None
PAYMENT_WATA_WEBAPP_LABEL_EN: Optional[str] = None
PAYMENT_WATA_WEBAPP_ICON: Optional[str] = None
PAYMENT_WATA_TELEGRAM_LABEL_RU: Optional[str] = None
PAYMENT_WATA_TELEGRAM_LABEL_EN: Optional[str] = None
PAYMENT_WATA_TELEGRAM_EMOJI: Optional[str] = None
PAYMENT_YOOKASSA_WEBAPP_LABEL_RU: Optional[str] = None
PAYMENT_YOOKASSA_WEBAPP_LABEL_EN: Optional[str] = None
PAYMENT_YOOKASSA_WEBAPP_ICON: Optional[str] = None
@@ -580,15 +546,6 @@ class Settings(BaseSettings):
platega_crypto_method=self.PLATEGA_CRYPTO_METHOD,
platega_return_url=self.PLATEGA_RETURN_URL,
platega_failed_url=self.PLATEGA_FAILED_URL,
wata_enabled=self.WATA_ENABLED,
wata_api_token=self.WATA_API_TOKEN,
wata_base_url=self.WATA_BASE_URL,
wata_return_url=self.WATA_RETURN_URL,
wata_failed_url=self.WATA_FAILED_URL,
wata_payment_link_ttl_days=self.WATA_PAYMENT_LINK_TTL_DAYS,
wata_webhook_verify_signature=self.WATA_WEBHOOK_VERIFY_SIGNATURE,
wata_public_key=self.WATA_PUBLIC_KEY,
wata_trusted_ips=self.wata_trusted_ips,
cryptopay_enabled=self.CRYPTOPAY_ENABLED,
cryptopay_token=self.CRYPTOPAY_TOKEN,
cryptopay_network=self.CRYPTOPAY_NETWORK,
@@ -763,24 +720,6 @@ class Settings(BaseSettings):
return f"{base.rstrip('/')}{self.freekassa_webhook_path}"
return None
@computed_field
@property
def wata_webhook_path(self) -> str:
return "/webhook/wata"
@computed_field
@property
def wata_full_webhook_url(self) -> Optional[str]:
base = self.WEBHOOK_BASE_URL
if base:
return f"{base.rstrip('/')}{self.wata_webhook_path}"
return None
@computed_field
@property
def wata_trusted_ips(self) -> List[str]:
return _split_csv(self.WATA_TRUSTED_IPS)
@computed_field
@property
def platega_webhook_path(self) -> str:
@@ -1123,10 +1062,6 @@ class Settings(BaseSettings):
"REQUIRED_CHANNEL_LINK",
"PLATEGA_RETURN_URL",
"PLATEGA_FAILED_URL",
"WATA_RETURN_URL",
"WATA_FAILED_URL",
"WATA_API_TOKEN",
"WATA_PUBLIC_KEY",
"CRYPT4_REDIRECT_URL",
"PRIVACY_POLICY_URL",
"USER_AGREEMENT_URL",
@@ -1156,17 +1091,6 @@ class Settings(BaseSettings):
return None
return v
@field_validator("WATA_PAYMENT_LINK_TTL_DAYS", mode="before")
@classmethod
def validate_wata_ttl_days(cls, v):
if isinstance(v, str):
v = v.strip()
try:
value = int(v)
except (TypeError, ValueError):
return 3
return min(30, max(1, value))
# Notification types
LOG_NEW_USERS: bool = Field(
default=True, description="Send notifications for new user registrations"
@@ -1253,11 +1177,6 @@ def get_settings() -> Settings:
logging.warning(
"CRITICAL: Platega is enabled but merchant credentials (PLATEGA_MERCHANT_ID/PLATEGA_SECRET) are missing. Platega payments will not work." # noqa: E501
)
if _settings_instance.WATA_ENABLED:
if not _settings_instance.WATA_API_TOKEN:
logging.warning(
"CRITICAL: Wata is enabled but WATA_API_TOKEN is missing. Wata payments will not work." # noqa: E501
)
except ValidationError as e:
logging.critical(f"Pydantic validation error while loading settings: {e}")
tests/test_payment_provider_registry.py
+8 -4
View File
@@ -198,7 +198,14 @@ def test_provider_presentation_ignores_cross_language_override():
)
def test_payment_method_keyboard_uses_custom_telegram_text_without_changing_callback():
def test_payment_method_keyboard_uses_custom_telegram_text_without_changing_callback(monkeypatch):
from bot.payment_providers import build_provider_configs
monkeypatch.setenv("WATA_ENABLED", "True")
monkeypatch.setenv("PAYMENT_WATA_TELEGRAM_LABEL_EN", "Wata custom")
monkeypatch.setenv("PAYMENT_WATA_TELEGRAM_EMOJI", "💸")
build_provider_configs()
settings = Settings(
_env_file=None,
BOT_TOKEN="token",
@@ -206,9 +213,6 @@ def test_payment_method_keyboard_uses_custom_telegram_text_without_changing_call
POSTGRES_PASSWORD="app_password",
TARIFFS_CONFIG_PATH="missing-tariffs.json",
PAYMENT_METHODS_ORDER="wata",
WATA_ENABLED=True,
PAYMENT_WATA_TELEGRAM_LABEL_EN="Wata custom",
PAYMENT_WATA_TELEGRAM_EMOJI="💸",
)
i18n = SimpleNamespace(gettext=lambda _lang, key, **_kwargs: key)