prog1764
896c26bd57
- IDOR: отмена/редактирование брони только владельцем (проверка user_id) - пароль админа: убран дефолт admin123, обязателен из env; сравнение через hmac.compare_digest; TTL сессии 1ч - callback_data: безопасный парсинг int + .get() вместо прямого индексирования (KeyError/ValueError)
95 lines
3.1 KiB
Python
95 lines
3.1 KiB
Python
"""Admin handler for spa-telegram-bot — secure with password."""
|
|
import hmac
|
|
import time
|
|
|
|
from aiogram import Router, F
|
|
from aiogram.types import Message, CallbackQuery
|
|
from aiogram.filters import Command, CommandObject
|
|
from app.db.memory import load_bookings
|
|
from app.config import ADMIN_PASSWORD
|
|
from app.logger import logger
|
|
|
|
router = Router()
|
|
|
|
ADMIN_IDS = {991309145}
|
|
# Store approved sessions — user_ids that passed password check
|
|
_approved: dict[int, float] = {}
|
|
APPROVED_TTL_SECONDS = 60 * 60 # 1 hour
|
|
|
|
|
|
def is_admin(user_id: int, password: str | None = None) -> bool:
|
|
"""Check if user is admin: by user_id, by password, or by approved session."""
|
|
if user_id in ADMIN_IDS:
|
|
return True
|
|
now = time.time()
|
|
expires_at = _approved.get(user_id)
|
|
if expires_at and expires_at > now:
|
|
return True
|
|
if expires_at:
|
|
_approved.pop(user_id, None) # expired session
|
|
if password and hmac.compare_digest(password.encode("utf-8"), ADMIN_PASSWORD.encode("utf-8")):
|
|
_approved[user_id] = now + APPROVED_TTL_SECONDS
|
|
return True
|
|
return False
|
|
|
|
|
|
@router.message(Command("admin"))
|
|
async def admin_panel(message: Message, command: CommandObject):
|
|
uid = message.from_user.id
|
|
password = command.args
|
|
|
|
if not is_admin(uid, password=password):
|
|
return # Silent
|
|
|
|
logger.info("Admin {uid} opened admin panel", uid=uid)
|
|
|
|
bookings = load_bookings()
|
|
active = [b for b in bookings if b.get("status") == "active"]
|
|
|
|
if not active:
|
|
await message.answer("📋 Активных записей нет")
|
|
return
|
|
|
|
from aiogram.types import InlineKeyboardMarkup, InlineKeyboardButton
|
|
|
|
for booking in active:
|
|
buttons = [[
|
|
InlineKeyboardButton(
|
|
text="❌ Отменить",
|
|
callback_data=f"admin_cancel_{booking['id']}"
|
|
)
|
|
]]
|
|
user_name = booking.get("user_name") or f"id{booking['user_id']}"
|
|
await message.answer(
|
|
f"📋 Запись #{booking['id']}\n"
|
|
f"👤 Клиент: {user_name}\n"
|
|
f"💆 Услуги: {booking.get('service', '?')}\n"
|
|
f"👤 Мастер: {booking.get('master', '?')}\n"
|
|
f"📅 Дата: {booking.get('date', '?')}\n"
|
|
f"🕒 Время: {booking.get('time', '?')}",
|
|
reply_markup=InlineKeyboardMarkup(inline_keyboard=buttons)
|
|
)
|
|
|
|
|
|
@router.callback_query(F.data.startswith("admin_cancel_"))
|
|
async def admin_cancel_booking(callback: CallbackQuery):
|
|
if not is_admin(callback.from_user.id):
|
|
await callback.answer("⛔ Нет доступа")
|
|
return
|
|
|
|
try:
|
|
booking_id = int(callback.data.split("_", 2)[2])
|
|
except (ValueError, IndexError):
|
|
await callback.answer("Некорректные данные")
|
|
return
|
|
|
|
logger.info("Admin {uid} cancelled booking #{bid}", uid=callback.from_user.id, bid=booking_id)
|
|
|
|
from app.db.memory import cancel_booking
|
|
cancel_booking(booking_id)
|
|
|
|
await callback.message.edit_text(
|
|
f"✅ Запись #{booking_id} отменена админом"
|
|
)
|
|
await callback.answer("Запись отменена")
|