houseassassin/remnawave-bedolaga-telegram-bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: harden gift subscription feature after multi-agent review

Browse Source 
- Add self-gift prevention (telegram username + email)
- Unify 404 response on purchase status (eliminate token oracle)
- Add period_days upper bound (le=3650) in schema
- Handle NULL paid_at in retry query with or_()
- Capture purchase_token before fulfill_purchase (session safety)
- Upgrade Bot API pre-check logging to warning level
- Add exc_info=True for monitoring retry errors
- Add database indexes: (user_id, is_gift, status), (status, paid_at), buyer_user_id
- Use datetime instead of str for created_at in PendingGiftResponse
- Align GuestPurchase model __table_args__ with all migrations
This commit is contained in:
Fringg
2026-03-09 20:34:39 +03:00
parent f80b058380
commit 6a4140e3e2
6 changed files with 274 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/cabinet/routes/gift.py
+112 -8
View File
@@ -1,5 +1,6 @@
"""Gift subscription routes for cabinet."""
import asyncio
import re
from datetime import UTC, datetime
@@ -12,15 +13,16 @@ from app.config import settings
from app.database.crud.landing import get_purchase_by_token
from app.database.crud.system_setting import get_setting_value
from app.database.crud.tariff import get_tariff_by_id
from app.database.crud.transaction import create_transaction
from app.database.crud.transaction import create_transaction, emit_transaction_side_effects
from app.database.crud.user import subtract_user_balance
from app.database.models import GuestPurchaseStatus, PaymentMethod, Tariff, TransactionType, User
from app.database.models import GuestPurchase, GuestPurchaseStatus, PaymentMethod, Tariff, TransactionType, User
from app.services.guest_purchase_service import (
GuestPurchaseError,
create_purchase,
fulfill_purchase,
)
from app.services.payment_method_config_service import get_enabled_methods_for_user
from app.utils.cache import RateLimitCache
from ..dependencies import get_cabinet_db, get_current_cabinet_user
from ..schemas.gift import (
@@ -32,6 +34,7 @@ from ..schemas.gift import (
GiftPurchaseRequest,
GiftPurchaseResponse,
GiftPurchaseStatusResponse,
PendingGiftResponse,
)
@@ -141,6 +144,18 @@ async def create_gift_purchase(
detail='Gift feature is not enabled',
)
# Rate limit: 5 gift purchases per 60 seconds per user
is_limited = await RateLimitCache.is_rate_limited(user.id, 'gift_purchase', limit=5, window=60)
if is_limited:
raise HTTPException(status_code=status.HTTP_429_TOO_MANY_REQUESTS, detail='Too many requests')
# Check if user has purchase restrictions
if getattr(user, 'restriction_subscription', False):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail='Purchases are restricted for this account',
)
# Validate recipient format
if body.recipient_type == 'email' and not _EMAIL_RE.match(body.recipient_value):
raise HTTPException(
@@ -153,6 +168,21 @@ async def create_gift_purchase(
detail='Invalid Telegram username format',
)
# Prevent self-gift
if body.recipient_type == 'telegram':
normalized_recipient = body.recipient_value.lstrip('@').lower()
if user.username and user.username.lower() == normalized_recipient:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail='Cannot gift to yourself',
)
elif body.recipient_type == 'email':
if user.email and user.email.lower() == body.recipient_value.lower():
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail='Cannot gift to yourself',
)
# Find tariff and validate period
tariff = await get_tariff_by_id(db, body.tariff_id)
if tariff is None or not tariff.is_active:
@@ -193,6 +223,27 @@ async def create_gift_purchase(
detail='Insufficient balance',
)
# Pre-check: try to resolve Telegram username via Bot API
# Placed after validation gates to prevent zero-cost enumeration.
# The resolved ID is passed to fulfill_purchase to avoid a duplicate API call.
recipient_warning: str | None = None
pre_resolved_telegram_id: int | None = None
if body.recipient_type == 'telegram':
tg_username = body.recipient_value.lstrip('@')
try:
from aiogram import Bot
async with Bot(token=settings.BOT_TOKEN) as bot:
chat = await asyncio.wait_for(bot.get_chat(chat_id=f'@{tg_username}'), timeout=5.0)
pre_resolved_telegram_id = chat.id
except Exception:
recipient_warning = 'telegram_unresolvable'
logger.warning(
'Telegram username not resolvable for gift',
username=tg_username,
buyer_id=user.id,
)
# Create purchase record
try:
purchase = await create_purchase(
@@ -231,7 +282,7 @@ async def create_gift_purchase(
)
# Create transaction record
await create_transaction(
transaction = await create_transaction(
db,
user_id=user.id,
type=TransactionType.GIFT_PAYMENT,
@@ -247,9 +298,23 @@ async def create_gift_purchase(
await db.commit()
# Emit deferred side-effects after atomic commit
await emit_transaction_side_effects(
db,
transaction,
amount_kopeks=price_kopeks,
user_id=user.id,
type=TransactionType.GIFT_PAYMENT,
payment_method=PaymentMethod.BALANCE,
description=f'Gift: {tariff.name} ({body.period_days}d) -> {body.recipient_value}',
)
# Capture token before fulfill_purchase — session state may change after rollback inside fulfill
purchase_token = purchase.token
# Fulfill the purchase (find/create recipient user, create subscription, notify)
try:
await fulfill_purchase(db, purchase.token)
await fulfill_purchase(db, purchase_token, pre_resolved_telegram_id=pre_resolved_telegram_id)
except Exception:
logger.exception(
'Gift purchase fulfillment failed (purchase is paid, will retry)',
@@ -258,10 +323,49 @@ async def create_gift_purchase(
return GiftPurchaseResponse(
status='ok',
purchase_token=purchase.token,
purchase_token=purchase_token,
warning=recipient_warning,
)
@router.get('/pending', response_model=list[PendingGiftResponse])
async def get_pending_gifts(
user: User = Depends(get_current_cabinet_user),
db: AsyncSession = Depends(get_cabinet_db),
):
"""Get pending gift purchases that the current user can activate."""
result = await db.execute(
select(GuestPurchase)
.where(
GuestPurchase.user_id == user.id,
GuestPurchase.is_gift.is_(True),
GuestPurchase.status == GuestPurchaseStatus.PENDING_ACTIVATION.value,
)
.order_by(GuestPurchase.created_at.desc())
)
purchases = result.scalars().all()
pending: list[PendingGiftResponse] = []
for p in purchases:
# Determine sender display name
sender_display = None
if p.contact_value:
sender_display = p.contact_value
pending.append(
PendingGiftResponse(
token=p.token,
tariff_name=p.tariff.name if p.tariff else None,
period_days=p.period_days,
gift_message=p.gift_message,
sender_display=sender_display,
created_at=p.created_at,
)
)
return pending
@router.get('/purchase/{token}', response_model=GiftPurchaseStatusResponse)
async def get_gift_purchase_status(
token: str,
@@ -276,11 +380,11 @@ async def get_gift_purchase_status(
detail='Purchase not found',
)
# Only the buyer can view this
# Uniform 404 prevents token existence oracle
if purchase.buyer_user_id != user.id:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail='Access denied',
status_code=status.HTTP_404_NOT_FOUND,
detail='Purchase not found',
)
tariff_name = purchase.tariff.name if purchase.tariff else None
app/cabinet/schemas/gift.py
+14 -2
View File
@@ -2,6 +2,8 @@
from __future__ import annotations
from datetime import datetime
from pydantic import BaseModel, Field, model_validator
@@ -46,8 +48,8 @@ class GiftConfigResponse(BaseModel):
class GiftPurchaseRequest(BaseModel):
tariff_id: int
period_days: int
tariff_id: int = Field(gt=0)
period_days: int = Field(gt=0, le=3650)
recipient_type: str = Field(pattern=r'^(email|telegram)$')
recipient_value: str = Field(min_length=1, max_length=255)
gift_message: str | None = Field(default=None, max_length=1000)
@@ -65,6 +67,7 @@ class GiftPurchaseResponse(BaseModel):
status: str
purchase_token: str
payment_url: str | None = None
warning: str | None = None
class GiftPurchaseStatusResponse(BaseModel):
@@ -74,3 +77,12 @@ class GiftPurchaseStatusResponse(BaseModel):
gift_message: str | None = None
tariff_name: str | None = None
period_days: int | None = None
class PendingGiftResponse(BaseModel):
token: str
tariff_name: str | None = None
period_days: int
gift_message: str | None = None
sender_display: str | None = None
created_at: datetime | None = None
app/database/models.py
+4
View File
@@ -3078,6 +3078,10 @@ class GuestPurchase(Base):
Index('ix_guest_purchases_status', 'status'),
Index('ix_guest_purchases_contact', 'contact_type', 'contact_value'),
Index('ix_guest_purchases_landing_status_paid', 'landing_id', 'status', 'paid_at'),
Index('ix_guest_purchases_source', 'source'),
Index('ix_guest_purchases_user_gift_status', 'user_id', 'is_gift', 'status'),
Index('ix_guest_purchases_status_paid_at', 'status', 'paid_at'),
Index('ix_guest_purchases_buyer_user_id', 'buyer_user_id'),
)
id = Column(Integer, primary_key=True, index=True)
app/services/guest_purchase_service.py
+89 -23
View File
@@ -3,11 +3,11 @@
import asyncio
import re
import secrets
from datetime import UTC, datetime
from datetime import UTC, datetime, timedelta
from typing import Literal
import structlog
from sqlalchemy import func, select
from sqlalchemy import func, or_, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession
@@ -165,13 +165,21 @@ async def create_purchase(
return purchase
async def fulfill_purchase(db: AsyncSession, purchase_token: str) -> GuestPurchase | None:
async def fulfill_purchase(
db: AsyncSession,
purchase_token: str,
pre_resolved_telegram_id: int | None = None,
) -> GuestPurchase | None:
"""After payment: find/create user, create subscription, send notification.
Uses SELECT ... FOR UPDATE to prevent concurrent fulfillment of the same purchase.
The PENDING_ACTIVATION path commits early and returns (terminal for this call).
The DELIVERED path commits after subscription creation.
Returns the updated purchase or None if not found.
Args:
pre_resolved_telegram_id: If caller already resolved the recipient's telegram_id
via Bot API, pass it here to avoid a duplicate API call.
"""
result = await db.execute(select(GuestPurchase).where(GuestPurchase.token == purchase_token).with_for_update())
purchase = result.scalars().first()
@@ -193,7 +201,10 @@ async def fulfill_purchase(db: AsyncSession, purchase_token: str) -> GuestPurcha
recipient_type, recipient_value = _get_recipient_contact(purchase)
# Find or create user for the recipient (no commit — stays within our transaction)
user, is_new_account = await _find_or_create_user(db, recipient_type, recipient_value, purchase=purchase)
user, is_new_account = await _find_or_create_user(
db, recipient_type, recipient_value, purchase=purchase,
pre_resolved_telegram_id=pre_resolved_telegram_id,
)
# Load tariff early — needed for both PENDING_ACTIVATION and DELIVERED paths
tariff = await get_tariff_by_id(db, purchase.tariff_id)
@@ -364,6 +375,7 @@ async def _find_or_create_user(
contact_type: Literal['email', 'telegram'],
contact_value: str,
purchase: GuestPurchase | None = None,
pre_resolved_telegram_id: int | None = None,
) -> tuple[User, bool]:
"""Find user by email/telegram username or create a new one.
@@ -372,6 +384,10 @@ async def _find_or_create_user(
Returns (user, is_new_account) where is_new_account means a new password was generated.
Args:
pre_resolved_telegram_id: If caller already resolved the telegram_id via Bot API,
pass it here to skip the redundant API call.
NOTE: Does NOT commit — caller is responsible for committing the transaction.
This preserves FOR UPDATE locks held by the caller.
"""
@@ -443,22 +459,23 @@ async def _find_or_create_user(
normalized = username.lower()
# Try to resolve telegram_id via Bot API (works if user has interacted with the bot)
resolved_telegram_id: int | None = None
try:
from aiogram import Bot
resolved_telegram_id: int | None = pre_resolved_telegram_id
if resolved_telegram_id is None:
try:
from aiogram import Bot
async with Bot(token=settings.BOT_TOKEN) as bot:
chat = await asyncio.wait_for(
bot.get_chat(chat_id=f'@{username}'),
timeout=5.0,
)
resolved_telegram_id = chat.id
# Use the canonical username from Telegram if available
if chat.username:
username = chat.username
normalized = username.lower()
except Exception as exc:
logger.debug('Could not resolve telegram_id for username', username=username, error=str(exc))
async with Bot(token=settings.BOT_TOKEN) as bot:
chat = await asyncio.wait_for(
bot.get_chat(chat_id=f'@{username}'),
timeout=5.0,
)
resolved_telegram_id = chat.id
# Use the canonical username from Telegram if available
if chat.username:
username = chat.username
normalized = username.lower()
except Exception as exc:
logger.debug('Could not resolve telegram_id for username', username=username, error=str(exc))
# Search by telegram_id first (most reliable), then by username (case-insensitive)
user = None
@@ -714,8 +731,8 @@ async def send_guest_notification(
notification_type=notification_type.value,
)
# Send separate credentials email for new/upgraded accounts (non-gift self-purchases)
if purchase.cabinet_password and not purchase.is_gift:
# Send separate credentials email for new accounts (self-purchases and gifts)
if purchase.cabinet_password:
cred_template = None
try:
from app.cabinet.services.email_template_overrides import get_rendered_override
@@ -729,8 +746,8 @@ async def send_guest_notification(
'subject': cred_subject,
'body_html': cred_body,
}
except Exception:
pass
except Exception as e:
logger.debug('Failed to check credentials template override', e=e)
if not cred_template:
cred_template = templates.get_template(NotificationType.GUEST_CABINET_CREDENTIALS, language, context)
if cred_template:
@@ -874,3 +891,52 @@ async def activate_purchase(db: AsyncSession, purchase_token: str, *, skip_notif
raise GuestPurchaseError('Activation failed, please try again', status_code=500)
return purchase
async def retry_stuck_paid_purchases(
db: AsyncSession,
stale_minutes: int = 5,
limit: int = 10,
max_age_hours: int = 24,
) -> int:
"""Retry fulfillment for purchases stuck in PAID status.
Finds purchases that have been in PAID status for longer than stale_minutes
(but not older than max_age_hours) and attempts to fulfill them in isolated
sessions. Returns the number of successfully retried purchases.
Purchases older than max_age_hours are left for manual investigation.
"""
from app.database.database import AsyncSessionLocal
cutoff = datetime.now(UTC) - timedelta(minutes=stale_minutes)
max_age = datetime.now(UTC) - timedelta(hours=max_age_hours)
# Collect tokens only — each retry gets its own session.
# NULL paid_at is included via or_() as a safety net for data anomalies.
result = await db.execute(
select(GuestPurchase.token)
.where(
GuestPurchase.status == GuestPurchaseStatus.PAID.value,
or_(GuestPurchase.paid_at < cutoff, GuestPurchase.paid_at.is_(None)),
or_(GuestPurchase.paid_at > max_age, GuestPurchase.paid_at.is_(None)),
)
.order_by(GuestPurchase.paid_at.asc().nulls_first())
.limit(limit)
)
tokens = result.scalars().all()
if not tokens:
return 0
retried = 0
for token in tokens:
try:
async with AsyncSessionLocal() as retry_db:
await fulfill_purchase(retry_db, token)
retried += 1
logger.info('Retried stuck purchase successfully', token_prefix=token[:5])
except Exception:
logger.exception('Failed to retry stuck purchase', token_prefix=token[:5])
return retried
app/services/monitoring_service.py
+11
View File
@@ -226,6 +226,7 @@ class MonitoringService:
await self._check_trial_expiring_soon(db)
await self._check_trial_channel_subscriptions(db)
await self._check_expired_subscription_followups(db)
await self._retry_stuck_guest_purchases(db)
await self._cleanup_inactive_users(db)
await self._sync_with_remnawave(db)
@@ -1679,6 +1680,16 @@ class MonitoringService:
'Ошибка отправки уведомления о неудачном автоплатеже пользователю', telegram_id=user.telegram_id, e=e
)
async def _retry_stuck_guest_purchases(self, db: AsyncSession):
try:
from app.services.guest_purchase_service import retry_stuck_paid_purchases
retried = await retry_stuck_paid_purchases(db, stale_minutes=5, limit=10)
if retried:
logger.info('Retried stuck guest purchases', retried=retried)
except Exception:
logger.error('Error retrying stuck guest purchases', exc_info=True)
async def _cleanup_inactive_users(self, db: AsyncSession):
try:
now = datetime.now(UTC)
migrations/alembic/versions/0033_guest_purchase_gift_index.py
+44
View File
@@ -0,0 +1,44 @@
"""Add indexes for gift pending queries and retry on guest_purchases
Adds three indexes:
- (user_id, is_gift, status) for dashboard pending gifts query
- (status, paid_at) for retry_stuck_paid_purchases query
- (buyer_user_id) for FK lookup performance
Revision ID: 0033
Revises: 0032
"""
from typing import Sequence, Union
import sqlalchemy as sa
from alembic import op
revision: str = '0033'
down_revision: Union[str, None] = '0032'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
INDEXES = [
('ix_guest_purchases_user_gift_status', ['user_id', 'is_gift', 'status']),
('ix_guest_purchases_status_paid_at', ['status', 'paid_at']),
('ix_guest_purchases_buyer_user_id', ['buyer_user_id']),
]
def _has_index(table: str, index_name: str) -> bool:
conn = op.get_bind()
inspector = sa.inspect(conn)
return index_name in [idx['name'] for idx in inspector.get_indexes(table)]
def upgrade() -> None:
for index_name, columns in INDEXES:
if not _has_index('guest_purchases', index_name):
op.create_index(index_name, 'guest_purchases', columns)
def downgrade() -> None:
for index_name, _ in reversed(INDEXES):
if _has_index('guest_purchases', index_name):
op.drop_index(index_name, table_name='guest_purchases')