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

Merge pull request #2800 from BEDOLAGA-DEV/dev

Browse Source 
Dev
This commit is contained in:
Egor
2026-03-22 10:55:46 +03:00
committed by GitHub
parent 4a48818bc3 0335f40b47
commit 8a1da85f3e
3 changed files with 214 additions and 81 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/cabinet/routes/admin_referral_network.py
+103 -15
View File
@@ -2,6 +2,7 @@
import re
from collections import defaultdict
from datetime import UTC, datetime
import structlog
from fastapi import APIRouter, Depends, HTTPException, Query, status
@@ -16,6 +17,7 @@ from app.database.models import (
PartnerStatus,
ReferralEarning,
Subscription,
SubscriptionStatus,
Tariff,
Transaction,
TransactionType,
@@ -78,6 +80,7 @@ class NetworkUserNode(BaseModel):
personal_spent_kopeks: int
subscription_name: str | None
subscription_end: str | None
subscription_status: str | None
registered_at: str | None
@@ -114,6 +117,7 @@ class NetworkGraphResponse(BaseModel):
total_referrers: int
total_campaigns: int
total_earnings_kopeks: int
total_subscription_revenue_kopeks: int
class NetworkUserDetail(BaseModel):
@@ -134,6 +138,7 @@ class NetworkUserDetail(BaseModel):
personal_spent_kopeks: int
subscription_name: str | None
subscription_end: str | None
subscription_status: str | None
registered_at: str | None
@@ -215,6 +220,7 @@ def _build_user_node(
campaign_id: int | None,
subscription_name: str | None,
subscription_end_str: str | None,
subscription_status: str | None,
) -> NetworkUserNode:
return NetworkUserNode(
id=user.id,
@@ -232,6 +238,7 @@ def _build_user_node(
personal_spent_kopeks=personal_spent,
subscription_name=subscription_name,
subscription_end=subscription_end_str,
subscription_status=subscription_status,
registered_at=_format_datetime(user.created_at),
)
@@ -312,7 +319,7 @@ async def _fetch_branch_revenue(db: AsyncSession, user_ids: set[int]) -> dict[in
stmt = (
select(
referred_user.c.referred_by_id,
func.coalesce(func.sum(Transaction.amount_kopeks), 0),
func.coalesce(func.sum(func.abs(Transaction.amount_kopeks)), 0),
)
.join(referred_user, Transaction.user_id == referred_user.c.id)
.where(
@@ -333,7 +340,7 @@ async def _fetch_personal_spent(db: AsyncSession, user_ids: set[int]) -> dict[in
return {}
stmt = (
select(Transaction.user_id, func.coalesce(func.sum(Transaction.amount_kopeks), 0))
select(Transaction.user_id, func.coalesce(func.sum(func.abs(Transaction.amount_kopeks)), 0))
.where(
and_(
Transaction.user_id.in_(user_ids),
@@ -376,18 +383,81 @@ async def _fetch_campaign_registrations(db: AsyncSession, user_ids: set[int] | N
return {row[0]: row[1] for row in result}
async def _fetch_subscription_info(db: AsyncSession, user_ids: set[int]) -> dict[int, tuple[str | None, str | None]]:
"""Return {user_id: (tariff_name, end_date_iso)} for given users."""
def _compute_subscription_status(
is_trial: bool | None,
db_status: str | None,
end_date: datetime | None,
now: datetime,
) -> str | None:
"""Map subscription fields to a frontend status label.
Returns one of: 'trial_active', 'trial_expired', 'paid_active', 'paid_expired', or None.
Statuses DISABLED, PENDING, EXPIRED, LIMITED are treated as inactive regardless of end_date.
ACTIVE and TRIAL fall through to a date-based check.
"""
if is_trial is None:
return None
if db_status in (
SubscriptionStatus.DISABLED.value,
SubscriptionStatus.PENDING.value,
SubscriptionStatus.EXPIRED.value,
SubscriptionStatus.LIMITED.value,
):
return 'trial_expired' if is_trial else 'paid_expired'
if is_trial:
return 'trial_active' if (end_date and end_date > now) else 'trial_expired'
return 'paid_active' if (end_date and end_date > now) else 'paid_expired'
async def _fetch_subscription_info(
db: AsyncSession,
user_ids: set[int],
) -> dict[int, tuple[str | None, str | None, str | None]]:
"""Return {user_id: (tariff_name, end_date_iso, subscription_status)} for given users."""
if not user_ids:
return {}
stmt = (
select(Subscription.user_id, Tariff.name, Subscription.end_date)
row_num = (
func.row_number()
.over(
partition_by=Subscription.user_id,
order_by=Subscription.end_date.desc().nullslast(),
)
.label('rn')
)
inner = (
select(
Subscription.user_id,
Tariff.name,
Subscription.end_date,
Subscription.is_trial,
Subscription.status,
row_num,
)
.outerjoin(Tariff, Subscription.tariff_id == Tariff.id)
.where(Subscription.user_id.in_(user_ids))
)
subq = inner.subquery()
stmt = select(
subq.c.user_id,
subq.c.name,
subq.c.end_date,
subq.c.is_trial,
subq.c.status,
).where(subq.c.rn == 1)
result = await db.execute(stmt)
return {row[0]: (row[1], _format_datetime(row[2]) if row[2] else None) for row in result}
now = datetime.now(UTC)
out: dict[int, tuple[str | None, str | None, str | None]] = {}
for row in result:
user_id, tariff_name, end_date, is_trial, db_status = row
end_date_iso = _format_datetime(end_date) if end_date else None
sub_status = _compute_subscription_status(is_trial, db_status, end_date, now)
out[user_id] = (tariff_name, end_date_iso, sub_status)
return out
async def _fetch_campaign_stats(
@@ -428,7 +498,7 @@ async def _fetch_campaign_stats(
user_spent: dict[int, int] = {}
if all_campaign_users:
spent_stmt = (
select(Transaction.user_id, func.coalesce(func.sum(Transaction.amount_kopeks), 0))
select(Transaction.user_id, func.coalesce(func.sum(func.abs(Transaction.amount_kopeks)), 0))
.where(
and_(
Transaction.user_id.in_(all_campaign_users),
@@ -540,6 +610,7 @@ async def get_referral_network(
total_referrers=0,
total_campaigns=0,
total_earnings_kopeks=0,
total_subscription_revenue_kopeks=0,
)
# Cap to prevent excessive response sizes (deterministic: keep lowest IDs for stability)
@@ -567,7 +638,7 @@ async def get_referral_network(
# Build user nodes
user_nodes: list[NetworkUserNode] = []
for user in users:
sub = sub_info.get(user.id, (None, None))
sub = sub_info.get(user.id, (None, None, None))
user_nodes.append(
_build_user_node(
user,
@@ -578,6 +649,7 @@ async def get_referral_network(
campaign_id=campaign_regs.get(user.id),
subscription_name=sub[0],
subscription_end_str=sub[1],
subscription_status=sub[2],
)
)
@@ -629,6 +701,7 @@ async def get_referral_network(
total_referrers = len([u for u in user_nodes if u.direct_referrals > 0])
total_earnings = sum(personal_revenue.values())
total_subscription_revenue = sum(personal_spent.values())
return NetworkGraphResponse(
users=user_nodes,
@@ -638,6 +711,7 @@ async def get_referral_network(
total_referrers=total_referrers,
total_campaigns=len(campaign_nodes),
total_earnings_kopeks=total_earnings,
total_subscription_revenue_kopeks=total_subscription_revenue,
)
@@ -784,6 +858,7 @@ async def _build_scoped_graph(
total_referrers=0,
total_campaigns=len(campaign_nodes),
total_earnings_kopeks=0,
total_subscription_revenue_kopeks=0,
)
return NetworkGraphResponse(
users=[],
@@ -793,6 +868,7 @@ async def _build_scoped_graph(
total_referrers=0,
total_campaigns=0,
total_earnings_kopeks=0,
total_subscription_revenue_kopeks=0,
)
# Cap to prevent excessive response sizes
@@ -816,7 +892,7 @@ async def _build_scoped_graph(
user_nodes: list[NetworkUserNode] = []
for user in users:
sub = sub_info.get(user.id, (None, None))
sub = sub_info.get(user.id, (None, None, None))
user_nodes.append(
_build_user_node(
user,
@@ -827,6 +903,7 @@ async def _build_scoped_graph(
campaign_id=campaign_regs.get(user.id),
subscription_name=sub[0],
subscription_end_str=sub[1],
subscription_status=sub[2],
)
)
@@ -878,6 +955,7 @@ async def _build_scoped_graph(
total_referrers = len([u for u in user_nodes if u.direct_referrals > 0])
total_earnings = sum(personal_revenue.values())
total_subscription_revenue = sum(personal_spent.values())
return NetworkGraphResponse(
users=user_nodes,
@@ -887,6 +965,7 @@ async def _build_scoped_graph(
total_referrers=total_referrers,
total_campaigns=len(campaign_nodes),
total_earnings_kopeks=total_earnings,
total_subscription_revenue_kopeks=total_subscription_revenue,
)
@@ -1057,7 +1136,7 @@ async def get_network_user_detail(
branch_revenue = 0
# Personal spent
spent_stmt = select(func.coalesce(func.sum(Transaction.amount_kopeks), 0)).where(
spent_stmt = select(func.coalesce(func.sum(func.abs(Transaction.amount_kopeks)), 0)).where(
and_(
Transaction.user_id == user_id,
Transaction.type.in_(SPENT_TRANSACTION_TYPES),
@@ -1102,7 +1181,7 @@ async def get_network_user_detail(
# Branch revenue: total spent by all users in the branch
branch_user_ids_stmt = select(branch_cte.c.id)
branch_rev_stmt = select(func.coalesce(func.sum(Transaction.amount_kopeks), 0)).where(
branch_rev_stmt = select(func.coalesce(func.sum(func.abs(Transaction.amount_kopeks)), 0)).where(
and_(
Transaction.user_id.in_(branch_user_ids_stmt),
Transaction.type.in_(SPENT_TRANSACTION_TYPES),
@@ -1124,10 +1203,17 @@ async def get_network_user_detail(
# Subscription info
subscription_name: str | None = None
subscription_end: str | None = None
subscription_status: str | None = None
if user.subscription is not None:
if user.subscription.tariff is not None:
subscription_name = user.subscription.tariff.name
subscription_end = _format_datetime(user.subscription.end_date)
subscription_status = _compute_subscription_status(
user.subscription.is_trial,
user.subscription.status,
user.subscription.end_date,
datetime.now(UTC),
)
return NetworkUserDetail(
id=user.id,
@@ -1147,6 +1233,7 @@ async def get_network_user_detail(
personal_spent_kopeks=personal_spent,
subscription_name=subscription_name,
subscription_end=subscription_end,
subscription_status=subscription_status,
registered_at=_format_datetime(user.created_at),
)
@@ -1215,7 +1302,7 @@ async def get_network_campaign_detail(
total_spent = 0
if campaign_user_ids:
spent_stmt = (
select(Transaction.user_id, func.coalesce(func.sum(Transaction.amount_kopeks), 0))
select(Transaction.user_id, func.coalesce(func.sum(func.abs(Transaction.amount_kopeks)), 0))
.where(
and_(
Transaction.user_id.in_(campaign_user_ids),
@@ -1336,7 +1423,7 @@ async def search_referral_network(
sub_info = await _fetch_subscription_info(db, matched_ids)
for user in matched_users:
sub = sub_info.get(user.id, (None, None))
sub = sub_info.get(user.id, (None, None, None))
user_nodes.append(
_build_user_node(
user,
@@ -1347,6 +1434,7 @@ async def search_referral_network(
campaign_id=campaign_regs.get(user.id),
subscription_name=sub[0],
subscription_end_str=sub[1],
subscription_status=sub[2],
)
)
@@ -1400,7 +1488,7 @@ async def search_referral_network(
campaign_user_spent: dict[int, int] = {}
if all_campaign_user_ids:
spent_stmt = (
select(Transaction.user_id, func.coalesce(func.sum(Transaction.amount_kopeks), 0))
select(Transaction.user_id, func.coalesce(func.sum(func.abs(Transaction.amount_kopeks)), 0))
.where(
and_(
Transaction.user_id.in_(all_campaign_user_ids),
app/cabinet/routes/admin_roles.py
+18 -44
View File
@@ -441,6 +441,14 @@ async def assign_role(
detail='Role not found',
)
# Superadmin role is managed exclusively via ADMIN_IDS/ADMIN_EMAILS env config
if role.level >= SUPERADMIN_LEVEL:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail='Superadmin role is managed via ADMIN_IDS/ADMIN_EMAILS environment variables. '
'Add the user there and restart the bot.',
)
admin_level = await _get_admin_level(db, admin)
# Cannot assign a role with level >= own level
@@ -450,13 +458,6 @@ async def assign_role(
detail='Cannot assign a role with level >= your own role level',
)
# Superadmin assignments must be permanent — expiry would cause silent lockout
if role.level == SUPERADMIN_LEVEL and payload.expires_at is not None:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail='Superadmin role assignments cannot be time-limited',
)
# Verify target user exists
from app.database.crud.user import get_user_by_id
@@ -505,9 +506,7 @@ async def revoke_role(
admin: User = Depends(require_permission('roles:assign')),
db: AsyncSession = Depends(get_cabinet_db),
):
"""Revoke a role assignment. Cannot remove the last superadmin."""
from app.config import settings
from app.database.crud.user import get_user_by_id
"""Revoke a role assignment. Superadmin roles are managed via env config."""
from app.database.models import UserRole
# Lock the assignment row (FOR UPDATE held until commit)
@@ -526,6 +525,14 @@ async def revoke_role(
detail='Associated role not found',
)
# Superadmin role is managed exclusively via env config
if role.level >= SUPERADMIN_LEVEL:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail='Superadmin role is managed via ADMIN_IDS/ADMIN_EMAILS environment variables. '
'Remove the user from env and restart the bot.',
)
admin_level = await _get_admin_level(db, admin)
# Cannot revoke a role at or above own level
@@ -535,33 +542,6 @@ async def revoke_role(
detail='Cannot revoke a role at or above your own level',
)
# Block self-revocation of superadmin role
if role.level == SUPERADMIN_LEVEL and user_role.user_id == admin.id:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail='Cannot revoke your own superadmin role',
)
# Protect last superadmin (level 999).
# Advisory lock serializes concurrent superadmin revocations so two requests
# cannot both read count=2 and then both proceed to revoke.
if role.level == SUPERADMIN_LEVEL:
if not settings.is_sqlite():
await db.execute(sa.text('SELECT pg_advisory_xact_lock(736453)'))
superadmin_count = await UserRoleCRUD.get_superadmin_count(db)
if superadmin_count <= 1:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail='Cannot remove the last superadmin',
)
# Warn if target user is a legacy admin — RBAC revocation won't actually block access
target_user = await get_user_by_id(db, user_role.user_id)
is_target_legacy = target_user and settings.is_admin(
telegram_id=target_user.telegram_id,
email=target_user.email if target_user.email_verified else None,
)
# Revoke directly on the locked object (avoid CRUD re-fetch without FOR UPDATE)
user_role.is_active = False
await db.flush()
@@ -575,10 +555,4 @@ async def revoke_role(
role_name=role.name,
)
result_msg = {'message': 'Role revoked', 'assignment_id': assignment_id}
if is_target_legacy:
result_msg['warning'] = (
'This user is still listed in ADMIN_IDS/ADMIN_EMAILS env config. '
'They retain full access until removed from those settings and the bot is restarted.'
)
return result_msg
return {'message': 'Role revoked', 'assignment_id': assignment_id}
app/services/rbac_bootstrap_service.py
+93 -22
View File
@@ -10,6 +10,7 @@ from typing import Final
import structlog
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload
from app.config import settings
from app.database.crud.rbac import SUPERADMIN_LEVEL, UserRoleCRUD
@@ -215,19 +216,28 @@ async def bootstrap_superadmins(db: AsyncSession) -> None:
if assigned:
assigned_count += 1
# ── 4. Commit all changes ──────────────────────────────────────
# ── 4. Revoke superadmin from users NOT in env ───────────────
revoked_count = await _revoke_stale_superadmins(
db,
role_id=role_id,
admin_ids=admin_ids,
admin_emails=admin_emails,
)
# ── 5. Commit all changes ──────────────────────────────────────
await db.commit()
if assigned_count > 0:
if assigned_count > 0 or revoked_count > 0:
logger.info(
'Superadmin bootstrap completed',
assigned_count=assigned_count,
revoked_count=revoked_count,
role_id=role_id,
)
else:
logger.debug('Superadmin bootstrap: no new assignments needed')
logger.debug('Superadmin bootstrap: no changes needed')
# ── 5. Safety: warn if no active superadmins exist ────────────
# ── 6. Safety: warn if no active superadmins exist ────────────
await _warn_if_no_superadmins(db, admin_ids, admin_emails)
except Exception:
@@ -235,6 +245,60 @@ async def bootstrap_superadmins(db: AsyncSession) -> None:
logger.exception('Failed to bootstrap superadmins, continuing startup')
async def _revoke_stale_superadmins(
db: AsyncSession,
*,
role_id: int,
admin_ids: list[int],
admin_emails: list[str],
) -> int:
"""Revoke superadmin from users who are no longer in env config.
Env config (ADMIN_IDS / ADMIN_EMAILS) is the single source of truth.
If a user was removed from env, their superadmin DB role is deactivated
on the next bot restart.
Returns the number of revoked assignments.
"""
result = await db.execute(
select(UserRole)
.options(selectinload(UserRole.user))
.where(
UserRole.role_id == role_id,
UserRole.is_active.is_(True),
)
)
active_assignments = result.scalars().all()
admin_ids_set = set(admin_ids)
admin_emails_set = {e.lower() for e in admin_emails}
revoked = 0
for assignment in active_assignments:
user = assignment.user
if user is None:
continue
# Check if user is still in env config.
# email_verified is required — symmetric with _ensure_role_by_email.
in_env_by_id = user.telegram_id is not None and user.telegram_id in admin_ids_set
in_env_by_email = user.email is not None and user.email_verified and user.email.lower() in admin_emails_set
if not in_env_by_id and not in_env_by_email:
assignment.is_active = False
await db.flush()
revoked += 1
logger.warning(
'Revoked Superadmin role: user removed from env config',
user_id=user.id,
telegram_id=user.telegram_id,
email=user.email,
user_role_id=assignment.id,
)
return revoked
async def _warn_if_no_superadmins(
db: AsyncSession,
admin_ids: list[int],
@@ -281,13 +345,18 @@ async def _ensure_role_by_email(
email: str,
role_id: int,
) -> bool:
"""Assign Superadmin role to user found by email (case-insensitive). Returns True if assigned."""
result = await db.execute(select(User).where(func.lower(User.email) == email.lower()))
"""Assign Superadmin role to user found by verified email (case-insensitive). Returns True if assigned."""
result = await db.execute(
select(User).where(
func.lower(User.email) == email.lower(),
User.email_verified.is_(True),
)
)
user = result.scalar_one_or_none()
if user is None:
logger.debug(
'Admin user (email) not yet registered, skipping',
'Admin user (email) not yet registered or not verified, skipping',
email=email,
)
return False
@@ -302,13 +371,13 @@ async def _assign_if_missing(
role_id: int,
identifier: str,
) -> bool:
"""Create a UserRole row if none exists for this user/role pair.
"""Create or reactivate a UserRole row for this user/role pair.
If an assignment already exists (active or revoked), it is left as-is.
This ensures that an admin-revoked role is NOT silently reactivated
on every bot restart.
Env config (ADMIN_IDS / ADMIN_EMAILS) is the source of truth for
Superadmin assignments. If a previously revoked assignment exists,
it is reactivated — the env config always wins.
Returns True only if a brand-new assignment was created.
Returns True if a new assignment was created or an inactive one was reactivated.
"""
result = await db.execute(
select(UserRole).where(
@@ -325,16 +394,18 @@ async def _assign_if_missing(
user_id=user_id,
identifier=identifier,
)
else:
logger.info(
'Superadmin role was previously revoked, not reactivating '
'(remove user from ADMIN_IDS to stop this warning, '
'or re-assign via cabinet)',
user_id=user_id,
identifier=identifier,
user_role_id=existing.id,
)
return False
return False
# Reactivate: env config is the source of truth
existing.is_active = True
await db.flush()
logger.info(
'Reactivated Superadmin role (user is in env config)',
user_id=user_id,
identifier=identifier,
user_role_id=existing.id,
)
return True
user_role = UserRole(
user_id=user_id,