# middlewares/auth_check.py
from __future__ import annotations
import asyncio
import inspect
import logging
import sqlite3
import time
from collections import defaultdict, deque
from html import escape
from pathlib import Path
from typing import (
Any,
Awaitable,
Callable,
Dict,
Final,
Optional,
)
from aiogram import BaseMiddleware
from aiogram.exceptions import TelegramBadRequest
from aiogram.types import (
CallbackQuery,
Message,
TelegramObject,
User,
)
from core.config import settings
from core.ui_emojis import (
ce,
premiumize_html,
)
logger = logging.getLogger(__name__)
EVENT_MESSAGE: Final[str] = "message"
EVENT_CALLBACK: Final[str] = "callback"
EVENT_OTHER: Final[str] = "other"
RateKey = tuple[int, str]
class AuthCheckMiddleware(BaseMiddleware):
"""
Authentication / Authorization Middleware.
مسئولیتها:
- استخراج Telegram user
- اتصال خودکار به db موجود در Dispatcher data
- تشخیص is_blocked / is_admin / kyc_status دیتابیس فعلی
- پشتیبانی از DB interfaceهای قدیمی
- تشخیص مالک از settings.ADMIN_IDS
- تزریق Security Context به Handler
- KYC restriction اختیاری
- ثبت Activity به شکل best-effort
- Rate limiter داخلی اختیاری
- جلوگیری از اجرای دوباره Handler در زمان Exception
نکته مهم:
پروژه ThrottlingMiddleware جداگانه دارد؛ بنابراین
rate_limit پیشفرض این Middleware برابر صفر است.
در صورت نیاز میتوان Rate Limiter داخلی Auth را
صراحتاً فعال کرد.
Security:
خطای Handler هیچوقت توسط این Middleware به عنوان
خطای Authentication تفسیر نمیشود و Handler مجدداً
اجرا نخواهد شد.
"""
def __init__(
self,
db: Any = None,
*,
rate_limit: int = 0,
rate_window: int = 60,
default_role: str = "user",
default_tier: str = "standard",
require_kyc: bool = False,
fail_closed: bool = False,
skip_private_commands: bool = False,
whitelist_admins_from_rate_limit: bool = True,
max_rate_users: int = 20_000,
) -> None:
super().__init__()
self.db = db
try:
rate_limit = int(rate_limit)
rate_window = int(rate_window)
max_rate_users = int(max_rate_users)
except (TypeError, ValueError) as exc:
raise ValueError(
"Invalid AuthCheckMiddleware numeric configuration."
) from exc
if rate_limit < 0:
raise ValueError(
"rate_limit must be >= 0."
)
if rate_window <= 0:
raise ValueError(
"rate_window must be > 0."
)
if max_rate_users <= 0:
raise ValueError(
"max_rate_users must be > 0."
)
self.rate_limit = rate_limit
self.rate_window = rate_window
self.max_rate_users = max_rate_users
self.default_role = self._normalize_role(
default_role
)
self.default_tier = (
str(default_tier or "standard")
.strip()
.lower()
or "standard"
)
self.require_kyc = bool(
require_kyc
)
self.fail_closed = bool(
fail_closed
)
self.skip_private_commands = bool(
skip_private_commands
)
self.whitelist_admins_from_rate_limit = bool(
whitelist_admins_from_rate_limit
)
# Message و Callback مستقل هستند.
self._rate_buckets: Dict[
RateKey,
deque[float],
] = defaultdict(deque)
self._rate_lock = asyncio.Lock()
logger.info(
"AuthCheckMiddleware initialized | "
"internal_rate=%s/%ss | "
"fail_closed=%s | require_kyc=%s",
self.rate_limit,
self.rate_window,
self.fail_closed,
self.require_kyc,
)
# ========================================================
# Main Middleware
# ========================================================
async def __call__(
self,
handler: Callable[
[
TelegramObject,
Dict[str, Any],
],
Awaitable[Any],
],
event: TelegramObject,
data: Dict[str, Any],
) -> Any:
user = self._extract_user(
event
)
# Updateهای بدون User را دستکاری نمیکنیم.
if user is None:
return await handler(
event,
data,
)
if getattr(
user,
"is_bot",
False,
):
return await handler(
event,
data,
)
user_id = int(
user.id
)
username = (
user.username
or ""
)
full_name = (
user.full_name
or ""
)
db = self._resolve_db(
data
)
# اطلاعات پایه همیشه برای Handler قابل دسترساند.
data[
"authenticated_user"
] = user
data[
"telegram_user"
] = user
data[
"user_id"
] = user_id
data[
"username"
] = username
data[
"full_name"
] = full_name
# ----------------------------------------------------
# فقط Authentication/Authorization داخل try است.
#
# Handler اصلی عمداً خارج از این try اجرا میشود
# تا Exception خود Handler باعث اجرای دوباره آن نشود.
# ----------------------------------------------------
try:
configured_admin = (
self._is_config_admin(
user_id
)
)
# -----------------------------------------------
# Optional internal rate limit
# -----------------------------------------------
if (
self.rate_limit > 0
and not (
configured_admin
and self.whitelist_admins_from_rate_limit
)
and not self._should_skip_rate_limit(
event
)
):
allowed = (
await self._check_rate_limit(
user_id,
event,
)
)
if not allowed:
logger.info(
"Auth internal rate limit exceeded | "
"user_id=%s event=%s",
user_id,
self._event_type(
event
),
)
await self._reject_rate_limit(
event
)
return None
# -----------------------------------------------
# Account
# -----------------------------------------------
account = await self._load_account(
user_id,
db,
)
if account is None:
if (
self.fail_closed
and db is not None
):
logger.error(
"Account lookup failed with fail_closed=True | "
"user_id=%s",
user_id,
)
await self._reject_system_error(
event
)
return None
account = {}
account = self._normalize_account(
account,
user_id=user_id,
)
# -----------------------------------------------
# Block / ban
# -----------------------------------------------
is_banned = self._parse_bool(
account.get(
"is_banned"
)
)
ban_reason = (
account.get(
"ban_reason"
)
or account.get(
"reason"
)
or account.get(
"block_reason"
)
or "محدودیت امنیتی"
)
# ADMIN_IDS هم از Ban دیتابیس معاف نیست.
# اگر مالک عمداً در DB مسدود شود، وضعیت DB محترم است.
if is_banned:
logger.warning(
"Blocked user attempted access | "
"user_id=%s",
user_id,
)
await self._reject_banned(
event,
ban_reason,
)
return None
# -----------------------------------------------
# Role / Admin
# -----------------------------------------------
role = self._normalize_role(
account.get(
"role",
self.default_role,
)
)
is_admin = bool(
configured_admin
or self._parse_bool(
account.get(
"is_admin"
)
)
or role
in {
"admin",
"superadmin",
"owner",
}
)
if configured_admin:
role = "owner"
elif (
self._parse_bool(
account.get(
"is_admin"
)
)
and role == "user"
):
role = "admin"
is_superadmin = (
role
in {
"superadmin",
"owner",
}
)
# -----------------------------------------------
# Tier
# -----------------------------------------------
user_tier = str(
account.get(
"tier",
self.default_tier,
)
or self.default_tier
).strip().lower()
# -----------------------------------------------
# KYC
# -----------------------------------------------
kyc_raw = account.get(
"kyc_value"
)
is_kyc_verified = (
self._parse_kyc_verified(
kyc_raw
)
)
kyc_status = (
self._normalize_kyc_status(
kyc_raw
)
)
# -----------------------------------------------
# Security Context
# -----------------------------------------------
data[
"role"
] = role
data[
"user_role"
] = role
data[
"is_admin"
] = is_admin
data[
"is_superadmin"
] = is_superadmin
data[
"user_tier"
] = user_tier
data[
"kyc_status"
] = kyc_status
data[
"is_kyc_verified"
] = is_kyc_verified
data[
"is_banned"
] = False
data[
"auth_context"
] = {
"user_id":
user_id,
"username":
username,
"role":
role,
"tier":
user_tier,
"is_admin":
is_admin,
"is_superadmin":
is_superadmin,
"kyc_status":
kyc_status,
"is_kyc_verified":
is_kyc_verified,
"configured_admin":
configured_admin,
}
# -----------------------------------------------
# KYC Restriction
# -----------------------------------------------
# مالک/ادمین نباید به خاطر KYC از پنل مدیریت
# خودش قفل شود.
if (
self.require_kyc
and not is_admin
and not is_kyc_verified
):
logger.info(
"KYC required | user_id=%s",
user_id,
)
await self._reject_kyc(
event
)
return None
# -----------------------------------------------
# Activity - best effort
# -----------------------------------------------
await self._update_activity(
db=db,
user=user,
)
except Exception:
logger.exception(
"Auth middleware security-context error | "
"user_id=%s",
user_id,
)
if self.fail_closed:
await self._reject_system_error(
event
)
return None
# Fail-open فقط روی خود Auth اعمال میشود.
#
# نکته:
# Handler اینجا اجرا نمیشود؛ فقط اجازه میدهیم
# پس از خروج از try یک بار اجرا شود.
self._inject_fail_open_context(
data,
user_id=user_id,
username=username,
)
# ====================================================
# Handler is executed EXACTLY ONCE.
# ====================================================
return await handler(
event,
data,
)
# ========================================================
# DB Resolution
# ========================================================
def _resolve_db(
self,
data: Dict[str, Any],
) -> Any:
"""
main.py فعلی db را در Dispatcher workflow data میگذارد:
dp["db"] = db_manager
پس حتی اگر AuthCheckMiddleware() بدون db ساخته شود،
از data همان instance اصلی دریافت میشود.
"""
if self.db is not None:
return self.db
return data.get(
"db"
)
# ========================================================
# User / Event
# ========================================================
@staticmethod
def _extract_user(
event: TelegramObject,
) -> Optional[User]:
if isinstance(
event,
Message,
):
return event.from_user
if isinstance(
event,
CallbackQuery,
):
return event.from_user
return None
@staticmethod
def _event_type(
event: TelegramObject,
) -> str:
if isinstance(
event,
CallbackQuery,
):
return EVENT_CALLBACK
if isinstance(
event,
Message,
):
return EVENT_MESSAGE
return EVENT_OTHER
def _should_skip_rate_limit(
self,
event: TelegramObject,
) -> bool:
if not self.skip_private_commands:
return False
if not isinstance(
event,
Message,
):
return False
chat = getattr(
event,
"chat",
None,
)
text = (
event.text
or ""
).strip()
return bool(
chat
and getattr(
chat,
"type",
None,
) == "private"
and text.startswith("/")
)
# ========================================================
# Config Admin
# ========================================================
@staticmethod
def _admin_ids() -> frozenset[int]:
method = getattr(
settings,
"get_admin_list",
None,
)
if callable(method):
try:
return frozenset(
int(user_id)
for user_id
in method()
if int(user_id) > 0
)
except Exception:
logger.exception(
"Could not parse settings.get_admin_list()."
)
raw = str(
getattr(
settings,
"ADMIN_IDS",
"",
)
or ""
)
result: set[int] = set()
for item in raw.split(","):
item = item.strip()
if not item:
continue
try:
user_id = int(
item
)
if user_id > 0:
result.add(
user_id
)
except ValueError:
logger.warning(
"Invalid ADMIN_IDS item ignored."
)
return frozenset(
result
)
@classmethod
def _is_config_admin(
cls,
user_id: int,
) -> bool:
return (
int(user_id)
in cls._admin_ids()
)
# ========================================================
# Optional Internal Rate Limiter
# ========================================================
async def _check_rate_limit(
self,
user_id: int,
event: TelegramObject,
) -> bool:
"""
Sliding Window.
Message و Callback bucket جدا دارند.
"""
if self.rate_limit <= 0:
return True
key = (
int(user_id),
self._event_type(
event
),
)
now = time.monotonic()
cutoff = (
now - self.rate_window
)
async with self._rate_lock:
bucket = self._rate_buckets[
key
]
while (
bucket
and bucket[0] <= cutoff
):
bucket.popleft()
if (
len(bucket)
>= self.rate_limit
):
return False
bucket.append(
now
)
self._cleanup_rate_buckets_locked(
now
)
return True
def _cleanup_rate_buckets_locked(
self,
now: float,
) -> None:
if (
len(self._rate_buckets)
<= self.max_rate_users
):
return
cutoff = (
now - self.rate_window
)
empty_keys: list[
RateKey
] = []
for key, bucket in (
self._rate_buckets.items()
):
while (
bucket
and bucket[0] <= cutoff
):
bucket.popleft()
if not bucket:
empty_keys.append(
key
)
for key in empty_keys:
self._rate_buckets.pop(
key,
None,
)
if (
len(self._rate_buckets)
<= self.max_rate_users
):
return
# آخرین محافظ حافظه:
# bucketهایی که قدیمیترین آخرین timestamp دارند حذف شوند.
ordered = sorted(
self._rate_buckets.items(),
key=lambda item: (
item[1][-1]
if item[1]
else 0.0
),
)
overflow = (
len(self._rate_buckets)
- self.max_rate_users
)
for key, _ in ordered[
:overflow
]:
self._rate_buckets.pop(
key,
None,
)
def clear_rate_limit(
self,
user_id: Optional[int] = None,
) -> None:
"""
API سازگار با نسخه قبلی.
چون عملیات بدون await و روی event-loop انجام میشود،
برای reset مدیریتی کافی است.
"""
if user_id is None:
self._rate_buckets.clear()
return
user_id = int(
user_id
)
for key in list(
self._rate_buckets
):
if key[0] == user_id:
self._rate_buckets.pop(
key,
None,
)
# ========================================================
# Account Loading
# ========================================================
async def _load_account(
self,
user_id: int,
db: Any,
) -> Optional[Dict[str, Any]]:
"""
Account lookup order for the CURRENT project:
1) Direct SQLite users.user_id
2) get_user_auth_context (legacy/richer fallback)
3) get_user (legacy fallback)
4) granular legacy methods
Why SQLite first?
The current DatabaseManager.get_user() queries a non-existent
users.id column, while the real primary key is users.user_id.
Calling that method on every Telegram update creates noisy
OperationalError logs and unnecessary latency.
None:
lookup genuinely failed.
{}:
database is available but the user has no row yet.
"""
if db is None:
return {}
had_error = False
# ----------------------------------------------------
# 1. Direct SQLite FIRST
# ----------------------------------------------------
try:
sqlite_account = (
await self._load_account_from_sqlite(
db,
user_id,
)
)
if sqlite_account is not None:
return sqlite_account
except Exception:
had_error = True
logger.exception(
"SQLite auth lookup failed | "
"user_id=%s",
user_id,
)
# ----------------------------------------------------
# 2. Rich auth context fallback
# ----------------------------------------------------
method = getattr(
db,
"get_user_auth_context",
None,
)
if callable(method):
try:
result = await self._call_method(
method,
user_id,
)
if result is None:
return {}
if isinstance(
result,
dict,
):
return result
return self._object_to_account(
result
)
except Exception:
had_error = True
# Legacy DB methods are fallback-only now.
logger.debug(
"get_user_auth_context fallback failed | "
"user_id=%s",
user_id,
exc_info=True,
)
# ----------------------------------------------------
# 3. get_user legacy fallback
# ----------------------------------------------------
method = getattr(
db,
"get_user",
None,
)
if callable(method):
try:
result = await self._call_method(
method,
user_id,
)
if result is None:
return {}
if isinstance(
result,
dict,
):
return result
return self._object_to_account(
result
)
except Exception:
had_error = True
# Do not fill normal ERROR logs with the known
# users.id/users.user_id legacy mismatch.
logger.debug(
"db.get_user legacy fallback failed | "
"user_id=%s",
user_id,
exc_info=True,
)
# ----------------------------------------------------
# 4. Granular legacy methods
# ----------------------------------------------------
account: Dict[str, Any] = {}
ban_method = getattr(
db,
"check_user_ban_status",
None,
)
if callable(ban_method):
try:
result = await self._call_method(
ban_method,
user_id,
)
if isinstance(
result,
tuple,
):
account[
"is_banned"
] = bool(
result[0]
)
if len(result) > 1:
account[
"ban_reason"
] = result[1]
else:
account[
"is_banned"
] = bool(
result
)
except Exception:
had_error = True
logger.debug(
"check_user_ban_status legacy fallback failed | "
"user_id=%s",
user_id,
exc_info=True,
)
role_method = getattr(
db,
"get_user_role",
None,
)
if callable(role_method):
try:
account[
"role"
] = await self._call_method(
role_method,
user_id,
)
except Exception:
had_error = True
logger.debug(
"get_user_role legacy fallback failed | "
"user_id=%s",
user_id,
exc_info=True,
)
tier_method = getattr(
db,
"get_user_tier",
None,
)
if callable(tier_method):
try:
account[
"tier"
] = await self._call_method(
tier_method,
user_id,
)
except Exception:
had_error = True
logger.debug(
"get_user_tier legacy fallback failed | "
"user_id=%s",
user_id,
exc_info=True,
)
kyc_method = getattr(
db,
"get_kyc_status",
None,
)
if callable(kyc_method):
try:
account[
"kyc_status"
] = await self._call_method(
kyc_method,
user_id,
)
except Exception:
had_error = True
logger.debug(
"get_kyc_status legacy fallback failed | "
"user_id=%s",
user_id,
exc_info=True,
)
if account:
return account
if had_error:
return None
return {}
# ========================================================
# SQLite Fallback
# ========================================================
@staticmethod
def _resolve_db_path(
db: Any,
) -> Optional[str]:
for attribute in (
"db_path",
"database_path",
"path",
):
value = getattr(
db,
attribute,
None,
)
if value:
return str(
value
)
value = (
getattr(
settings,
"DB_PATH",
None,
)
or getattr(
settings,
"DATABASE_PATH",
None,
)
)
if value:
return str(
value
)
return None
async def _load_account_from_sqlite(
self,
db: Any,
user_id: int,
) -> Optional[Dict[str, Any]]:
path = self._resolve_db_path(
db
)
if not path:
return None
return await asyncio.to_thread(
self._load_account_from_sqlite_sync,
path,
int(user_id),
)
@staticmethod
def _load_account_from_sqlite_sync(
db_path: str,
user_id: int,
) -> Optional[Dict[str, Any]]:
conn = sqlite3.connect(
db_path,
timeout=5,
)
conn.row_factory = sqlite3.Row
try:
table = conn.execute(
"""
SELECT 1
FROM sqlite_master
WHERE type='table'
AND name='users'
LIMIT 1
"""
).fetchone()
if not table:
return None
row = conn.execute(
"""
SELECT *
FROM users
WHERE user_id = ?
LIMIT 1
""",
(
user_id,
),
).fetchone()
if not row:
return {}
return dict(
row
)
finally:
conn.close()
# ========================================================
# Account Normalization
# ========================================================
def _normalize_account(
self,
account: Dict[str, Any],
*,
user_id: int,
) -> Dict[str, Any]:
raw = dict(
account or {}
)
blocked_value = self._first_present(
raw,
"is_banned",
"banned",
"blocked",
"is_blocked",
)
is_admin_value = self._first_present(
raw,
"is_admin",
"admin",
)
role_value = self._first_present(
raw,
"role",
"user_role",
)
tier_value = self._first_present(
raw,
"tier",
"user_tier",
)
kyc_value = self._first_present(
raw,
"is_kyc_verified",
"kyc_verified",
"kyc_status",
)
result = dict(
raw
)
result[
"is_banned"
] = self._parse_bool(
blocked_value
)
result[
"is_admin"
] = self._parse_bool(
is_admin_value
)
result[
"role"
] = self._normalize_role(
role_value
or self.default_role
)
result[
"tier"
] = (
str(
tier_value
or self.default_tier
)
.strip()
.lower()
)
result[
"kyc_value"
] = kyc_value
if self._is_config_admin(
user_id
):
result[
"is_admin"
] = True
result[
"role"
] = "owner"
elif (
result[
"is_admin"
]
and result[
"role"
] == "user"
):
result[
"role"
] = "admin"
return result
@staticmethod
def _first_present(
data: Dict[str, Any],
*keys: str,
) -> Any:
for key in keys:
if (
key in data
and data[key]
is not None
):
return data[
key
]
return None
@staticmethod
def _object_to_account(
obj: Any,
) -> Dict[str, Any]:
fields = (
"user_id",
"is_banned",
"banned",
"blocked",
"is_blocked",
"ban_reason",
"block_reason",
"reason",
"role",
"user_role",
"is_admin",
"tier",
"user_tier",
"is_kyc_verified",
"kyc_verified",
"kyc_status",
)
result: Dict[str, Any] = {}
for field in fields:
if hasattr(
obj,
field,
):
result[
field
] = getattr(
obj,
field,
)
return result
# ========================================================
# Activity
# ========================================================
async def _update_activity(
self,
*,
db: Any,
user: User,
) -> None:
"""
خطای Activity هیچوقت درخواست اصلی را متوقف نمیکند.
ابتدا API خود DatabaseManager را امتحان میکنیم.
اگر وجود نداشت/خراب بود، SQLite fallback فقط ستونهایی
را که واقعاً در users وجود دارند بروزرسانی میکند.
"""
if db is None:
return
user_id = int(
user.id
)
method = getattr(
db,
"upsert_user_activity",
None,
)
if callable(method):
try:
await self._call_with_supported_kwargs(
method,
{
"user_id":
user_id,
"username":
user.username
or "",
"full_name":
user.full_name
or "",
"first_name":
user.first_name
or "",
"last_name":
user.last_name
or "",
},
)
return
except Exception:
logger.warning(
"upsert_user_activity failed; "
"using SQLite fallback | user_id=%s",
user_id,
exc_info=True,
)
try:
await self._update_activity_sqlite(
db,
user,
)
except Exception:
logger.warning(
"Activity SQLite fallback failed | user_id=%s",
user_id,
exc_info=True,
)
async def _update_activity_sqlite(
self,
db: Any,
user: User,
) -> None:
path = self._resolve_db_path(
db
)
if not path:
return
await asyncio.to_thread(
self._update_activity_sqlite_sync,
path,
int(user.id),
user.username or "",
user.first_name or "",
user.last_name or "",
user.full_name or "",
)
@staticmethod
def _update_activity_sqlite_sync(
db_path: str,
user_id: int,
username: str,
first_name: str,
last_name: str,
full_name: str,
) -> None:
conn = sqlite3.connect(
db_path,
timeout=5,
)
conn.row_factory = sqlite3.Row
try:
table = conn.execute(
"""
SELECT 1
FROM sqlite_master
WHERE type='table'
AND name='users'
LIMIT 1
"""
).fetchone()
if not table:
return
columns = {
str(
row["name"]
)
for row in conn.execute(
"PRAGMA table_info(users)"
).fetchall()
}
if "user_id" not in columns:
return
values: Dict[
str,
Any,
] = {
"username":
username or None,
"first_name":
first_name or None,
"last_name":
last_name or None,
"full_name":
full_name or None,
}
existing = conn.execute(
"""
SELECT 1
FROM users
WHERE user_id = ?
LIMIT 1
""",
(
user_id,
),
).fetchone()
if existing:
assignments: list[
str
] = []
params: list[
Any
] = []
for column, value in (
values.items()
):
if column in columns:
assignments.append(
f"{column} = ?"
)
params.append(
value
)
if (
"updated_at"
in columns
):
assignments.append(
"updated_at = CURRENT_TIMESTAMP"
)
if assignments:
params.append(
user_id
)
conn.execute(
(
"UPDATE users SET "
+ ", ".join(
assignments
)
+ " WHERE user_id = ?"
),
tuple(
params
),
)
else:
insert_columns = [
"user_id"
]
insert_values: list[
Any
] = [
user_id
]
for column, value in (
values.items()
):
if column in columns:
insert_columns.append(
column
)
insert_values.append(
value
)
placeholders = ", ".join(
"?"
for _ in insert_columns
)
conn.execute(
(
"INSERT INTO users ("
+ ", ".join(
insert_columns
)
+ ") VALUES ("
+ placeholders
+ ")"
),
tuple(
insert_values
),
)
conn.commit()
finally:
conn.close()
# ========================================================
# Sync / Async DB Compatibility
# ========================================================
@staticmethod
async def _call_method(
method: Callable[..., Any],
*args,
**kwargs,
) -> Any:
if inspect.iscoroutinefunction(
method
):
return await method(
*args,
**kwargs,
)
result = await asyncio.to_thread(
method,
*args,
**kwargs,
)
if inspect.isawaitable(
result
):
return await result
return result
@classmethod
async def _call_with_supported_kwargs(
cls,
method: Callable[..., Any],
values: Dict[str, Any],
) -> Any:
"""
Signatureهای مختلف DatabaseManager را بدون حدس خطرناک
تا حد ممکن پشتیبانی میکند.
"""
kwargs = dict(
values
)
try:
signature = inspect.signature(
method
)
accepts_kwargs = any(
parameter.kind
== inspect.Parameter.VAR_KEYWORD
for parameter
in signature.parameters.values()
)
if not accepts_kwargs:
kwargs = {
key: value
for key, value
in kwargs.items()
if key
in signature.parameters
}
except (
TypeError,
ValueError,
):
pass
return await cls._call_method(
method,
**kwargs,
)
# ========================================================
# Role / KYC Helpers
# ========================================================
@staticmethod
def _normalize_role(
role: Any,
) -> str:
if role is None:
return "user"
value = (
str(role)
.strip()
.lower()
)
aliases = {
"administrator":
"admin",
"administrator_user":
"admin",
"super_admin":
"superadmin",
"super-admin":
"superadmin",
"root":
"owner",
}
return aliases.get(
value,
value or "user",
)
@staticmethod
def _parse_bool(
value: Any,
) -> bool:
if isinstance(
value,
bool,
):
return value
if isinstance(
value,
(int, float),
):
return value == 1
if isinstance(
value,
str,
):
return (
value
.strip()
.lower()
in {
"1",
"true",
"yes",
"on",
"active",
"enabled",
"verified",
"approved",
"admin",
"owner",
}
)
return False
@classmethod
def _parse_kyc_verified(
cls,
value: Any,
) -> bool:
if cls._parse_bool(
value
):
return True
if isinstance(
value,
str,
):
return (
value
.strip()
.lower()
in {
"passed",
"accepted",
"complete",
"completed",
}
)
return False
@classmethod
def _normalize_kyc_status(
cls,
value: Any,
) -> str:
if cls._parse_kyc_verified(
value
):
return "approved"
if value is None:
return "none"
raw = (
str(value)
.strip()
.lower()
)
aliases = {
"":
"none",
"0":
"none",
"false":
"none",
"none":
"none",
"not_submitted":
"none",
"waiting":
"pending",
"review":
"pending",
"under_review":
"pending",
"declined":
"rejected",
"denied":
"rejected",
}
return aliases.get(
raw,
raw,
)
# ========================================================
# Fail-open Context
# ========================================================
def _inject_fail_open_context(
self,
data: Dict[str, Any],
*,
user_id: int,
username: str,
) -> None:
configured_admin = (
self._is_config_admin(
user_id
)
)
role = (
"owner"
if configured_admin
else self.default_role
)
data.setdefault(
"role",
role,
)
data.setdefault(
"user_role",
role,
)
data.setdefault(
"is_admin",
configured_admin,
)
data.setdefault(
"is_superadmin",
configured_admin,
)
data.setdefault(
"user_tier",
self.default_tier,
)
data.setdefault(
"kyc_status",
"unknown",
)
data.setdefault(
"is_kyc_verified",
False,
)
data.setdefault(
"is_banned",
False,
)
data.setdefault(
"auth_context",
{
"user_id":
user_id,
"username":
username,
"role":
role,
"tier":
self.default_tier,
"is_admin":
configured_admin,
"is_superadmin":
configured_admin,
"kyc_status":
"unknown",
"is_kyc_verified":
False,
"auth_degraded":
True,
},
)
# ========================================================
# Telegram Rejections
# ========================================================
async def _reject_banned(
self,
event: TelegramObject,
reason: Any,
) -> None:
safe_reason = self._safe_reason(
reason
)
text = (
f"{ce('error')} "
"حساب شما مسدود است.\n\n"
"دسترسی شما به سرویسهای ربات محدود شده است.\n\n"
f"{ce('chat')} دلیل: "
f"{safe_reason}\n\n"
f"{ce('support')} "
"برای پیگیری با پشتیبانی تماس بگیرید."
)
await self._answer_event(
event,
text,
callback_text=(
"⛔ حساب شما مسدود است."
),
alert=True,
)
async def _reject_rate_limit(
self,
event: TelegramObject,
) -> None:
text = (
f"{ce('loading')} "
"تعداد درخواستها بیش از حد مجاز است.\n\n"
"لطفاً کمی صبر کنید و دوباره تلاش کنید."
)
await self._answer_event(
event,
text,
callback_text=(
"⏳ لطفاً کمی آهستهتر تلاش کنید."
),
alert=True,
)
async def _reject_kyc(
self,
event: TelegramObject,
) -> None:
text = (
f"{ce('warning')} "
"احراز هویت مورد نیاز است.\n\n"
"برای استفاده از این بخش ابتدا مراحل "
"احراز هویت را تکمیل کنید."
)
await self._answer_event(
event,
text,
callback_text=(
"⚠️ ابتدا احراز هویت را تکمیل کنید."
),
alert=True,
)
async def _reject_system_error(
self,
event: TelegramObject,
) -> None:
text = (
f"{ce('warning')} "
"خطای موقت سیستم\n\n"
"امکان بررسی حساب شما در حال حاضر وجود ندارد. "
"لطفاً کمی بعد دوباره تلاش کنید."
)
await self._answer_event(
event,
text,
callback_text=(
"⚠️ خطای موقت سیستم. کمی بعد تلاش کنید."
),
alert=True,
)
async def _answer_event(
self,
event: TelegramObject,
text: str,
*,
callback_text: Optional[str] = None,
alert: bool = False,
) -> None:
"""
CallbackQuery از HTML entity / tg-emoji پشتیبانی نمیکند،
پس popup یک متن ساده جدا دریافت میکند.
"""
try:
if isinstance(
event,
Message,
):
await event.answer(
premiumize_html(
text
),
parse_mode="HTML",
)
elif isinstance(
event,
CallbackQuery,
):
await event.answer(
(
callback_text
or self._plain_text(
text
)
)[:200],
show_alert=alert,
)
except TelegramBadRequest as exc:
logger.debug(
"Telegram rejected auth response: %s",
exc,
)
except Exception:
logger.exception(
"Failed to send auth middleware response."
)
@staticmethod
def _safe_reason(
reason: Any,
) -> str:
if reason is None:
return "محدودیت امنیتی"
text = (
str(reason)
.strip()
)
if not text:
return "محدودیت امنیتی"
# جلوگیری از HTML injection.
return escape(
text[:500]
)
@staticmethod
def _plain_text(
text: str,
) -> str:
# برای fallback callback؛ فقط tagهای ساده حذف میشوند.
import re
value = re.sub(
r"]*>.*?",
"",
str(text),
flags=re.DOTALL,
)
value = re.sub(
r"<[^>]+>",
"",
value,
)
return (
value
.replace("<", "<")
.replace(">", ">")
.replace("&", "&")
.strip()
)
# ============================================================
# Factory
# ============================================================
def create_auth_middleware(
db: Any = None,
*,
rate_limit: int = 0,
rate_window: int = 60,
default_role: str = "user",
default_tier: str = "standard",
require_kyc: bool = False,
fail_closed: bool = False,
skip_private_commands: bool = False,
whitelist_admins_from_rate_limit: bool = True,
max_rate_users: int = 20_000,
) -> AuthCheckMiddleware:
return AuthCheckMiddleware(
db=db,
rate_limit=rate_limit,
rate_window=rate_window,
default_role=default_role,
default_tier=default_tier,
require_kyc=require_kyc,
fail_closed=fail_closed,
skip_private_commands=skip_private_commands,
whitelist_admins_from_rate_limit=(
whitelist_admins_from_rate_limit
),
max_rate_users=max_rate_users,
)
__all__ = [
"AuthCheckMiddleware",
"create_auth_middleware",
]