Merge GitHub dev into GitLab dev
This commit is contained in:
@@ -348,6 +348,16 @@ SETTINGS_MANIFEST: List[SettingField] = [
|
||||
"Английская версия текста на этапе оплаты.",
|
||||
subsection="checkout",
|
||||
),
|
||||
SettingField(
|
||||
"PAYMENT_REQUEST_TIMEOUT_SECONDS",
|
||||
"float",
|
||||
"payments",
|
||||
"Таймаут запроса к провайдеру",
|
||||
"Максимальное общее время одного API-запроса к платёжному провайдеру, в секундах.",
|
||||
optional=False,
|
||||
min=1,
|
||||
subsection="checkout",
|
||||
),
|
||||
# ─── Payment providers (toggles) ───────────────────────────────
|
||||
# Common
|
||||
SettingField("STARS_ENABLED", "bool", "payments", "Telegram Stars", subsection="common"),
|
||||
|
||||
@@ -1,13 +1,16 @@
|
||||
import asyncio
|
||||
import functools
|
||||
import hmac
|
||||
import logging
|
||||
|
||||
from aiogram import Bot, Dispatcher
|
||||
from aiogram.webhook.aiohttp_server import SimpleRequestHandler, setup_application
|
||||
from aiohttp import web
|
||||
from aiohttp.web_log import AccessLogger, KeyMethod
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
|
||||
from bot.payment_providers import iter_provider_specs, iter_service_keys
|
||||
from bot.utils.request_security import request_client_ip
|
||||
from config.settings import Settings
|
||||
|
||||
|
||||
@@ -18,6 +21,39 @@ class SecureSimpleRequestHandler(SimpleRequestHandler):
|
||||
return hmac.compare_digest(telegram_secret_token, self.secret_token)
|
||||
|
||||
|
||||
class TrustedProxyAccessLogger(AccessLogger):
|
||||
"""Aiohttp access logger that respects trusted X-Forwarded-For headers."""
|
||||
|
||||
def compile_format(self, log_format):
|
||||
methods = []
|
||||
for atom in self.FORMAT_RE.findall(log_format):
|
||||
if atom[1] == "":
|
||||
format_key = self.LOG_FORMAT_MAP[atom[0]]
|
||||
method = getattr(type(self), f"_format_{atom[0]}", None)
|
||||
if method is None:
|
||||
method = getattr(AccessLogger, f"_format_{atom[0]}")
|
||||
methods.append(KeyMethod(format_key, method))
|
||||
else:
|
||||
format_key = (self.LOG_FORMAT_MAP[atom[2]], atom[1])
|
||||
method = getattr(type(self), f"_format_{atom[2]}", None)
|
||||
if method is None:
|
||||
method = getattr(AccessLogger, f"_format_{atom[2]}")
|
||||
methods.append(KeyMethod(format_key, functools.partial(method, atom[1])))
|
||||
|
||||
compiled = self.FORMAT_RE.sub(r"%s", log_format)
|
||||
compiled = self.CLEANUP_RE.sub(r"%\1", compiled)
|
||||
return compiled, methods
|
||||
|
||||
@staticmethod
|
||||
def _format_a(request, response, time):
|
||||
if request is None:
|
||||
return "-"
|
||||
settings = request.app.get("settings") if hasattr(request, "app") else None
|
||||
trusted_proxies = getattr(settings, "trusted_proxies", None)
|
||||
client_ip = request_client_ip(request, trusted_proxies=trusted_proxies)
|
||||
return client_ip or "-"
|
||||
|
||||
|
||||
def _inject_shared_instances(
|
||||
app: web.Application,
|
||||
dp: Dispatcher,
|
||||
@@ -110,7 +146,7 @@ async def build_and_start_web_app(
|
||||
|
||||
runners = []
|
||||
|
||||
webhooks_runner = web.AppRunner(app)
|
||||
webhooks_runner = web.AppRunner(app, access_log_class=TrustedProxyAccessLogger)
|
||||
await webhooks_runner.setup()
|
||||
runners.append(webhooks_runner)
|
||||
site = web.TCPSite(
|
||||
@@ -133,7 +169,10 @@ async def build_and_start_web_app(
|
||||
settings,
|
||||
async_session_factory,
|
||||
)
|
||||
subscription_runner = web.AppRunner(subscription_app)
|
||||
subscription_runner = web.AppRunner(
|
||||
subscription_app,
|
||||
access_log_class=TrustedProxyAccessLogger,
|
||||
)
|
||||
await subscription_runner.setup()
|
||||
runners.append(subscription_runner)
|
||||
subscription_site = web.TCPSite(
|
||||
|
||||
@@ -153,7 +153,7 @@ class FreeKassaService(HttpClientMixin):
|
||||
self.default_currency: str = default_payment_currency_code_for_settings(settings).upper()
|
||||
|
||||
self.api_base_url: str = "https://api.fk.life/v1"
|
||||
self._init_http_client(total_timeout=15)
|
||||
self._init_http_client(total_timeout=self.settings.PAYMENT_REQUEST_TIMEOUT_SECONDS)
|
||||
self._nonce_lock = asyncio.Lock()
|
||||
self._last_nonce = int(time.time() * 1000)
|
||||
|
||||
|
||||
@@ -243,7 +243,7 @@ class HeleketService(HttpClientMixin):
|
||||
self.referral_service = referral_service
|
||||
self._default_return_url = default_return_url
|
||||
|
||||
self._init_http_client(total_timeout=20)
|
||||
self._init_http_client(total_timeout=self.settings.PAYMENT_REQUEST_TIMEOUT_SECONDS)
|
||||
if not self.configured:
|
||||
logging.warning(
|
||||
"HeleketService initialized but not fully configured. Payments disabled."
|
||||
|
||||
@@ -546,7 +546,7 @@ class PaykillaService(HttpClientMixin):
|
||||
self._exchange_rate_cache: Dict[tuple[str, str], tuple[float, Decimal]] = {}
|
||||
self._currency_cache: tuple[float, List[Dict[str, Any]]] = (0, [])
|
||||
|
||||
self._init_http_client(total_timeout=20)
|
||||
self._init_http_client(total_timeout=self.settings.PAYMENT_REQUEST_TIMEOUT_SECONDS)
|
||||
if not self.configured:
|
||||
logging.warning(
|
||||
"PaykillaService initialized but not fully configured. Payments disabled."
|
||||
|
||||
@@ -157,7 +157,7 @@ class PlategaService(HttpClientMixin):
|
||||
self.referral_service = referral_service
|
||||
self._default_return_url = default_return_url
|
||||
|
||||
self._init_http_client(total_timeout=20)
|
||||
self._init_http_client(total_timeout=self.settings.PAYMENT_REQUEST_TIMEOUT_SECONDS)
|
||||
if not self.configured:
|
||||
logging.warning(
|
||||
"PlategaService initialized but not fully configured. Payments disabled."
|
||||
|
||||
@@ -136,7 +136,7 @@ class SeverPayService(HttpClientMixin):
|
||||
self.referral_service = referral_service
|
||||
self._default_return_url = default_return_url
|
||||
|
||||
self._init_http_client(total_timeout=15)
|
||||
self._init_http_client(total_timeout=self.settings.PAYMENT_REQUEST_TIMEOUT_SECONDS)
|
||||
|
||||
if not self.configured:
|
||||
logging.warning(
|
||||
|
||||
@@ -1,12 +1,14 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
from typing import Any, Callable, Dict, Mapping, Optional, Tuple
|
||||
|
||||
from aiohttp import ClientSession, ClientTimeout
|
||||
from aiohttp import ClientError, ClientSession, ClientTimeout, TraceConfig
|
||||
|
||||
SuccessCheck = Callable[[int, Any], bool]
|
||||
_TRANSPORT_ATTEMPTS = 2
|
||||
|
||||
|
||||
def http_ok(status: int, _body: Any) -> bool:
|
||||
@@ -14,6 +16,29 @@ def http_ok(status: int, _body: Any) -> bool:
|
||||
return status == 200
|
||||
|
||||
|
||||
def _trace_request_ctx(trace_config_ctx: Any) -> Optional[dict]:
|
||||
ctx = getattr(trace_config_ctx, "trace_request_ctx", None)
|
||||
return ctx if isinstance(ctx, dict) else None
|
||||
|
||||
|
||||
async def _mark_request_headers_sent(session, trace_config_ctx, params) -> None:
|
||||
ctx = _trace_request_ctx(trace_config_ctx)
|
||||
if ctx is not None:
|
||||
ctx["headers_sent"] = True
|
||||
|
||||
|
||||
def _payment_trace_config() -> TraceConfig:
|
||||
trace_config = TraceConfig()
|
||||
trace_config.on_request_headers_sent.append(_mark_request_headers_sent)
|
||||
return trace_config
|
||||
|
||||
|
||||
def _should_retry_transport_error(exc: Exception, trace_ctx: Mapping[str, Any]) -> bool:
|
||||
if trace_ctx.get("headers_sent"):
|
||||
return False
|
||||
return isinstance(exc, (asyncio.TimeoutError, ClientError, OSError))
|
||||
|
||||
|
||||
async def post_json_request(
|
||||
session: ClientSession,
|
||||
url: str,
|
||||
@@ -29,34 +54,47 @@ async def post_json_request(
|
||||
returns ``(False, {"status": ..., "message": ..., "raw": ...?})`` so callers
|
||||
can decide what to do (typically: mark the payment as ``failed_creation``).
|
||||
"""
|
||||
try:
|
||||
async with session.post(
|
||||
url,
|
||||
json=body,
|
||||
headers=dict(headers) if headers else None,
|
||||
) as response:
|
||||
response_text = await response.text()
|
||||
try:
|
||||
response_data = json.loads(response_text) if response_text else {}
|
||||
except json.JSONDecodeError:
|
||||
logging.error("%s: invalid JSON response: %s", log_prefix, response_text)
|
||||
return False, {
|
||||
"status": response.status,
|
||||
"message": "invalid_json",
|
||||
"raw": response_text,
|
||||
}
|
||||
if not is_success(response.status, response_data):
|
||||
logging.error(
|
||||
"%s: API returned error (status=%s, body=%s)",
|
||||
for attempt in range(1, _TRANSPORT_ATTEMPTS + 1):
|
||||
trace_ctx: dict[str, Any] = {"headers_sent": False}
|
||||
try:
|
||||
async with session.post(
|
||||
url,
|
||||
json=body,
|
||||
headers=dict(headers) if headers else None,
|
||||
trace_request_ctx=trace_ctx,
|
||||
) as response:
|
||||
response_text = await response.text()
|
||||
try:
|
||||
response_data = json.loads(response_text) if response_text else {}
|
||||
except json.JSONDecodeError:
|
||||
logging.error("%s: invalid JSON response: %s", log_prefix, response_text)
|
||||
return False, {
|
||||
"status": response.status,
|
||||
"message": "invalid_json",
|
||||
"raw": response_text,
|
||||
}
|
||||
if not is_success(response.status, response_data):
|
||||
logging.error(
|
||||
"%s: API returned error (status=%s, body=%s)",
|
||||
log_prefix,
|
||||
response.status,
|
||||
response_data,
|
||||
)
|
||||
return False, {"status": response.status, "message": response_data}
|
||||
return True, response_data
|
||||
except Exception as exc:
|
||||
if attempt < _TRANSPORT_ATTEMPTS and _should_retry_transport_error(exc, trace_ctx):
|
||||
logging.warning(
|
||||
"%s: transport failed before request headers were sent; retrying (%s/%s): %s", # noqa: E501
|
||||
log_prefix,
|
||||
response.status,
|
||||
response_data,
|
||||
attempt + 1,
|
||||
_TRANSPORT_ATTEMPTS,
|
||||
exc,
|
||||
)
|
||||
return False, {"status": response.status, "message": response_data}
|
||||
return True, response_data
|
||||
except Exception as exc:
|
||||
logging.exception("%s: request failed.", log_prefix)
|
||||
return False, {"message": str(exc)}
|
||||
continue
|
||||
logging.exception("%s: request failed.", log_prefix)
|
||||
return False, {"message": str(exc)}
|
||||
return False, {"message": "request_failed"}
|
||||
|
||||
|
||||
def first_value(data: Optional[Mapping[str, Any]], *keys: str) -> Optional[str]:
|
||||
@@ -76,6 +114,9 @@ class HttpClientMixin:
|
||||
Each subclass calls ``self._init_http_client(total_timeout=...)`` from
|
||||
``__init__`` and inherits ``_get_session`` / ``close``. The session is
|
||||
created on first use and recreated transparently if it was closed.
|
||||
|
||||
Provider API calls are traced so callers can retry transport failures only
|
||||
when aiohttp has not sent request headers yet.
|
||||
"""
|
||||
|
||||
_timeout: ClientTimeout
|
||||
@@ -87,7 +128,10 @@ class HttpClientMixin:
|
||||
|
||||
async def _get_session(self) -> ClientSession:
|
||||
if self._session is None or self._session.closed:
|
||||
self._session = ClientSession(timeout=self._timeout)
|
||||
self._session = ClientSession(
|
||||
timeout=self._timeout,
|
||||
trace_configs=[_payment_trace_config()],
|
||||
)
|
||||
return self._session
|
||||
|
||||
async def close(self) -> None:
|
||||
|
||||
@@ -206,7 +206,7 @@ class WataService(HttpClientMixin):
|
||||
self._default_return_url = default_return_url
|
||||
self._cached_public_key_pem = None # populated by webhook on first verify
|
||||
|
||||
self._init_http_client(total_timeout=10)
|
||||
self._init_http_client(total_timeout=self.settings.PAYMENT_REQUEST_TIMEOUT_SECONDS)
|
||||
if not self.configured:
|
||||
logging.warning("WataService initialized but not fully configured. Payments disabled.")
|
||||
|
||||
|
||||
Reference in New Issue
Block a user