dbadmin.py 17 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592
  1. from __future__ import annotations
  2. import asyncio
  3. import re
  4. from datetime import UTC, datetime
  5. from secrets import token_hex
  6. from typing import Any
  7. from pymongo import ASCENDING, DESCENDING
  8. from wbb import BOT_PROFILE_ID, control_db, db
  9. admin_usersdb = control_db.admin_users
  10. admin_sessionsdb = control_db.admin_sessions
  11. auditdb = db.admin_audit_logs
  12. managed_chatsdb = db.managed_chats
  13. managed_chat_settingsdb = db.managed_chat_settings
  14. invite_linksdb = db.admin_invite_links
  15. recent_chat_membersdb = db.recent_chat_members
  16. member_identity_changesdb = db.member_identity_changes
  17. _index_lock = asyncio.Lock()
  18. _indexes_ready = False
  19. def utc_now() -> datetime:
  20. return datetime.now(UTC)
  21. async def ensure_admin_indexes() -> None:
  22. global _indexes_ready
  23. if _indexes_ready:
  24. return
  25. async with _index_lock:
  26. if _indexes_ready:
  27. return
  28. await admin_usersdb.create_index([("username", ASCENDING)], unique=True)
  29. await admin_sessionsdb.create_index([("token_hash", ASCENDING)], unique=True)
  30. await admin_sessionsdb.create_index("expires_at", expireAfterSeconds=0)
  31. await auditdb.create_index([("created_at", DESCENDING)])
  32. await auditdb.create_index(
  33. [("chat_id", ASCENDING), ("created_at", DESCENDING)]
  34. )
  35. await managed_chatsdb.create_index(
  36. [("bot_id", ASCENDING), ("chat_id", ASCENDING)], unique=True
  37. )
  38. await managed_chatsdb.create_index([("last_seen_at", DESCENDING)])
  39. await managed_chat_settingsdb.create_index(
  40. [("bot_id", ASCENDING), ("chat_id", ASCENDING)], unique=True
  41. )
  42. await invite_linksdb.create_index(
  43. [("bot_id", ASCENDING), ("chat_id", ASCENDING), ("created_at", DESCENDING)]
  44. )
  45. await recent_chat_membersdb.create_index(
  46. [
  47. ("bot_id", ASCENDING),
  48. ("chat_id", ASCENDING),
  49. ("user_id", ASCENDING),
  50. ],
  51. unique=True,
  52. )
  53. await recent_chat_membersdb.create_index(
  54. [("bot_id", ASCENDING), ("chat_id", ASCENDING), ("last_seen_at", DESCENDING)]
  55. )
  56. await member_identity_changesdb.create_index(
  57. [("bot_id", ASCENDING), ("chat_id", ASCENDING), ("observed_at", DESCENDING)]
  58. )
  59. await member_identity_changesdb.create_index(
  60. [
  61. ("bot_id", ASCENDING),
  62. ("chat_id", ASCENDING),
  63. ("user_id", ASCENDING),
  64. ("observed_at", DESCENDING),
  65. ]
  66. )
  67. _indexes_ready = True
  68. async def ensure_default_admin(
  69. *, username: str, password_hash: str
  70. ) -> dict[str, Any]:
  71. await ensure_admin_indexes()
  72. now = utc_now()
  73. await admin_usersdb.update_one(
  74. {"username": username},
  75. {
  76. "$setOnInsert": {
  77. "username": username,
  78. "password_hash": password_hash,
  79. "must_change_password": True,
  80. "failed_login_count": 0,
  81. "created_at": now,
  82. "updated_at": now,
  83. }
  84. },
  85. upsert=True,
  86. )
  87. return await admin_usersdb.find_one({"username": username})
  88. async def get_admin_user(username: str) -> dict[str, Any] | None:
  89. await ensure_admin_indexes()
  90. return await admin_usersdb.find_one({"username": username})
  91. async def record_login_failure(username: str) -> None:
  92. await ensure_admin_indexes()
  93. await admin_usersdb.update_one(
  94. {"username": username},
  95. {
  96. "$inc": {"failed_login_count": 1},
  97. "$set": {"last_failed_login_at": utc_now(), "updated_at": utc_now()},
  98. },
  99. )
  100. async def record_login_success(username: str) -> None:
  101. await ensure_admin_indexes()
  102. await admin_usersdb.update_one(
  103. {"username": username},
  104. {
  105. "$set": {
  106. "failed_login_count": 0,
  107. "last_login_at": utc_now(),
  108. "updated_at": utc_now(),
  109. }
  110. },
  111. )
  112. async def update_admin_password(username: str, password_hash: str) -> bool:
  113. await ensure_admin_indexes()
  114. result = await admin_usersdb.update_one(
  115. {"username": username},
  116. {
  117. "$set": {
  118. "password_hash": password_hash,
  119. "must_change_password": False,
  120. "password_updated_at": utc_now(),
  121. "updated_at": utc_now(),
  122. }
  123. },
  124. )
  125. if result.modified_count:
  126. await admin_sessionsdb.delete_many({"username": username})
  127. return result.modified_count == 1
  128. async def create_admin_session(
  129. *,
  130. username: str,
  131. token_hash: str,
  132. csrf_token: str,
  133. expires_at: datetime,
  134. remote_address: str,
  135. user_agent: str,
  136. ) -> None:
  137. await ensure_admin_indexes()
  138. now = utc_now()
  139. await admin_sessionsdb.insert_one(
  140. {
  141. "username": username,
  142. "token_hash": token_hash,
  143. "csrf_token": csrf_token,
  144. "remote_address": remote_address[:200],
  145. "user_agent": user_agent[:500],
  146. "created_at": now,
  147. "last_seen_at": now,
  148. "expires_at": expires_at,
  149. }
  150. )
  151. async def get_admin_session(token_hash: str) -> dict[str, Any] | None:
  152. await ensure_admin_indexes()
  153. session = await admin_sessionsdb.find_one(
  154. {"token_hash": token_hash, "expires_at": {"$gt": utc_now()}}
  155. )
  156. if not session:
  157. return None
  158. user = await get_admin_user(session["username"])
  159. if not user:
  160. return None
  161. await admin_sessionsdb.update_one(
  162. {"_id": session["_id"]}, {"$set": {"last_seen_at": utc_now()}}
  163. )
  164. return {**session, "user": user}
  165. async def revoke_admin_session(token_hash: str) -> None:
  166. await ensure_admin_indexes()
  167. await admin_sessionsdb.delete_one({"token_hash": token_hash})
  168. async def revoke_admin_sessions(username: str) -> None:
  169. await ensure_admin_indexes()
  170. await admin_sessionsdb.delete_many({"username": username})
  171. async def upsert_managed_chat(
  172. *,
  173. chat_id: int,
  174. title: str | None,
  175. username: str | None,
  176. chat_type: str,
  177. member_count: int | None = None,
  178. accessible: bool = True,
  179. bot_status: str | None = None,
  180. bot_privileges: list[str] | None = None,
  181. ) -> None:
  182. await ensure_admin_indexes()
  183. now = utc_now()
  184. values: dict[str, Any] = {
  185. "title": title or str(chat_id),
  186. "username": username,
  187. "type": chat_type,
  188. "accessible": accessible,
  189. "last_seen_at": now,
  190. "updated_at": now,
  191. }
  192. if member_count is not None:
  193. values["member_count"] = int(member_count)
  194. if bot_status is not None:
  195. values["bot_status"] = bot_status
  196. if bot_privileges is not None:
  197. values["bot_privileges"] = bot_privileges
  198. await managed_chatsdb.update_one(
  199. {"bot_id": BOT_PROFILE_ID, "chat_id": chat_id},
  200. {
  201. "$set": values,
  202. "$setOnInsert": {"bot_id": BOT_PROFILE_ID, "created_at": now},
  203. },
  204. upsert=True,
  205. )
  206. async def mark_managed_chat_unavailable(chat_id: int, reason: str) -> None:
  207. await ensure_admin_indexes()
  208. await managed_chatsdb.update_one(
  209. {"bot_id": BOT_PROFILE_ID, "chat_id": chat_id},
  210. {
  211. "$set": {
  212. "accessible": False,
  213. "last_error": reason[:500],
  214. "updated_at": utc_now(),
  215. }
  216. },
  217. upsert=True,
  218. )
  219. async def delete_unavailable_managed_chat(chat_id: int) -> bool:
  220. """Remove an unavailable chat from this bot's management index only."""
  221. await ensure_admin_indexes()
  222. result = await managed_chatsdb.delete_one(
  223. {
  224. "bot_id": BOT_PROFILE_ID,
  225. "chat_id": int(chat_id),
  226. "accessible": False,
  227. }
  228. )
  229. return result.deleted_count == 1
  230. async def managed_chat_exists(chat_id: int) -> bool:
  231. await ensure_admin_indexes()
  232. return (
  233. await managed_chatsdb.find_one(
  234. {"bot_id": BOT_PROFILE_ID, "chat_id": int(chat_id)}, {"_id": 1}
  235. )
  236. is not None
  237. )
  238. async def list_managed_chats(
  239. *, query: str = "", page: int = 1, page_size: int = 20
  240. ) -> tuple[list[dict[str, Any]], int]:
  241. await ensure_admin_indexes()
  242. filters: dict[str, Any] = {"bot_id": BOT_PROFILE_ID}
  243. query = query.strip()
  244. if query:
  245. if query.lstrip("-").isdigit():
  246. filters["chat_id"] = int(query)
  247. else:
  248. value = re.escape(query.lstrip("@"))
  249. filters["$or"] = [
  250. {"title": {"$regex": value, "$options": "i"}},
  251. {"username": {"$regex": value, "$options": "i"}},
  252. ]
  253. page = max(1, page)
  254. page_size = max(1, min(page_size, 100))
  255. total = await managed_chatsdb.count_documents(filters)
  256. cursor = (
  257. managed_chatsdb.find(filters)
  258. .sort([("accessible", DESCENDING), ("last_seen_at", DESCENDING)])
  259. .skip((page - 1) * page_size)
  260. .limit(page_size)
  261. )
  262. return [doc async for doc in cursor], total
  263. async def record_audit(
  264. *,
  265. source: str,
  266. actor_id: int | str | None,
  267. actor_name: str,
  268. action: str,
  269. chat_id: int | None = None,
  270. target_id: int | str | None = None,
  271. summary: str = "",
  272. success: bool = True,
  273. error: str = "",
  274. metadata: dict[str, Any] | None = None,
  275. ) -> None:
  276. await ensure_admin_indexes()
  277. await auditdb.insert_one(
  278. {
  279. "source": source,
  280. "actor_id": actor_id,
  281. "actor_name": actor_name,
  282. "action": action,
  283. "chat_id": chat_id,
  284. "target_id": target_id,
  285. "summary": summary[:1000],
  286. "success": bool(success),
  287. "error": error[:1000],
  288. "metadata": metadata or {},
  289. "bot_id": BOT_PROFILE_ID,
  290. "created_at": utc_now(),
  291. }
  292. )
  293. async def list_audit_logs(
  294. *,
  295. chat_id: int | None = None,
  296. action: str | None = None,
  297. page: int = 1,
  298. page_size: int = 20,
  299. ) -> tuple[list[dict[str, Any]], int]:
  300. await ensure_admin_indexes()
  301. filters: dict[str, Any] = {}
  302. if chat_id is not None:
  303. filters["chat_id"] = chat_id
  304. if action:
  305. filters["action"] = action
  306. page = max(1, page)
  307. page_size = max(1, min(page_size, 100))
  308. total = await auditdb.count_documents(filters)
  309. cursor = (
  310. auditdb.find(filters)
  311. .sort("created_at", DESCENDING)
  312. .skip((page - 1) * page_size)
  313. .limit(page_size)
  314. )
  315. return [doc async for doc in cursor], total
  316. async def get_managed_chat_settings(chat_id: int) -> dict[str, Any]:
  317. await ensure_admin_indexes()
  318. doc = await managed_chat_settingsdb.find_one(
  319. {"bot_id": BOT_PROFILE_ID, "chat_id": int(chat_id)}
  320. )
  321. if doc is not None:
  322. doc.setdefault("captcha_enabled", False)
  323. doc.setdefault("interaction_settings", {
  324. "cleanup_seconds": 10,
  325. "commands": {key: True for key in ("giveaway", "checkin", "points", "history", "shortcuts")},
  326. })
  327. return doc
  328. return {
  329. "chat_id": int(chat_id),
  330. "auto_replies": [],
  331. "blacklist_words": [],
  332. "welcome": {"enabled": False, "text": "", "media": None},
  333. "identity_monitor": {"enabled": False, "notify_in_chat": False},
  334. "captcha_enabled": False,
  335. "chatbot_enabled": False,
  336. "antiflood_enabled": True,
  337. "interaction_settings": {
  338. "cleanup_seconds": 10,
  339. "commands": {key: True for key in ("giveaway", "checkin", "points", "history", "shortcuts")},
  340. },
  341. }
  342. async def update_managed_chat_settings(
  343. chat_id: int, values: dict[str, Any]
  344. ) -> dict[str, Any]:
  345. await ensure_admin_indexes()
  346. allowed = {
  347. "auto_replies",
  348. "blacklist_words",
  349. "risk_control",
  350. "risk_rules",
  351. "identity_monitor",
  352. "welcome",
  353. "captcha_enabled",
  354. "chatbot_enabled",
  355. "antiflood_enabled",
  356. "interaction_settings",
  357. }
  358. update = {key: value for key, value in values.items() if key in allowed}
  359. now = utc_now()
  360. await managed_chat_settingsdb.update_one(
  361. {"bot_id": BOT_PROFILE_ID, "chat_id": int(chat_id)},
  362. {
  363. "$set": {**update, "updated_at": now},
  364. "$setOnInsert": {
  365. "bot_id": BOT_PROFILE_ID,
  366. "chat_id": int(chat_id),
  367. "created_at": now,
  368. },
  369. },
  370. upsert=True,
  371. )
  372. return await get_managed_chat_settings(chat_id)
  373. async def upsert_recent_chat_member(
  374. *,
  375. chat_id: int,
  376. user_id: int,
  377. username: str | None,
  378. first_name: str | None,
  379. last_name: str | None,
  380. is_bot: bool,
  381. ) -> None:
  382. await ensure_admin_indexes()
  383. now = utc_now()
  384. await recent_chat_membersdb.update_one(
  385. {
  386. "bot_id": BOT_PROFILE_ID,
  387. "chat_id": int(chat_id),
  388. "user_id": int(user_id),
  389. },
  390. {
  391. "$set": {
  392. "username": username,
  393. "first_name": first_name,
  394. "last_name": last_name,
  395. "is_bot": bool(is_bot),
  396. "last_seen_at": now,
  397. },
  398. "$inc": {"message_count": 1},
  399. "$setOnInsert": {
  400. "bot_id": BOT_PROFILE_ID,
  401. "chat_id": int(chat_id),
  402. "user_id": int(user_id),
  403. "created_at": now,
  404. },
  405. },
  406. upsert=True,
  407. )
  408. async def get_recent_chat_member(chat_id: int, user_id: int) -> dict[str, Any] | None:
  409. await ensure_admin_indexes()
  410. return await recent_chat_membersdb.find_one(
  411. {
  412. "bot_id": BOT_PROFILE_ID,
  413. "chat_id": int(chat_id),
  414. "user_id": int(user_id),
  415. }
  416. )
  417. async def record_member_identity_change(
  418. *,
  419. chat_id: int,
  420. user_id: int,
  421. before: dict[str, Any],
  422. after: dict[str, Any],
  423. changed_fields: list[str],
  424. ) -> dict[str, Any]:
  425. await ensure_admin_indexes()
  426. event = {
  427. "change_id": token_hex(10),
  428. "bot_id": BOT_PROFILE_ID,
  429. "chat_id": int(chat_id),
  430. "user_id": int(user_id),
  431. "before": before,
  432. "after": after,
  433. "changed_fields": changed_fields,
  434. "observed_at": utc_now(),
  435. }
  436. await member_identity_changesdb.insert_one(event)
  437. return event
  438. async def list_member_identity_changes(
  439. chat_id: int,
  440. *,
  441. page: int = 1,
  442. page_size: int = 20,
  443. ) -> tuple[list[dict[str, Any]], int]:
  444. await ensure_admin_indexes()
  445. filters = {"bot_id": BOT_PROFILE_ID, "chat_id": int(chat_id)}
  446. page = max(1, int(page))
  447. page_size = max(1, min(int(page_size), 100))
  448. total = await member_identity_changesdb.count_documents(filters)
  449. cursor = (
  450. member_identity_changesdb.find(filters)
  451. .sort("observed_at", DESCENDING)
  452. .skip((page - 1) * page_size)
  453. .limit(page_size)
  454. )
  455. return [item async for item in cursor], total
  456. async def list_recent_chat_members(
  457. chat_id: int,
  458. *,
  459. query: str = "",
  460. limit: int = 30,
  461. ) -> list[dict[str, Any]]:
  462. await ensure_admin_indexes()
  463. filters: dict[str, Any] = {
  464. "bot_id": BOT_PROFILE_ID,
  465. "chat_id": int(chat_id),
  466. "is_bot": {"$ne": True},
  467. }
  468. normalized_query = query.strip().lstrip("@")
  469. if normalized_query:
  470. if normalized_query.isdigit():
  471. filters["user_id"] = int(normalized_query)
  472. else:
  473. pattern = re.escape(normalized_query)
  474. filters["$or"] = [
  475. {"username": {"$regex": pattern, "$options": "i"}},
  476. {"first_name": {"$regex": pattern, "$options": "i"}},
  477. {"last_name": {"$regex": pattern, "$options": "i"}},
  478. ]
  479. cursor = (
  480. recent_chat_membersdb.find(filters)
  481. .sort("last_seen_at", DESCENDING)
  482. .limit(max(1, min(int(limit), 50)))
  483. )
  484. return [item async for item in cursor]
  485. async def store_invite_link(
  486. *, chat_id: int, invite_link: str, name: str, expires_at: datetime | None
  487. ) -> None:
  488. await ensure_admin_indexes()
  489. await invite_linksdb.update_one(
  490. {
  491. "bot_id": BOT_PROFILE_ID,
  492. "chat_id": int(chat_id),
  493. "invite_link": invite_link,
  494. },
  495. {
  496. "$set": {
  497. "name": name,
  498. "expires_at": expires_at,
  499. "revoked": False,
  500. "updated_at": utc_now(),
  501. },
  502. "$setOnInsert": {"bot_id": BOT_PROFILE_ID, "created_at": utc_now()},
  503. },
  504. upsert=True,
  505. )
  506. async def revoke_stored_invite_link(chat_id: int, invite_link: str) -> None:
  507. await ensure_admin_indexes()
  508. await invite_linksdb.update_one(
  509. {
  510. "bot_id": BOT_PROFILE_ID,
  511. "chat_id": int(chat_id),
  512. "invite_link": invite_link,
  513. },
  514. {"$set": {"revoked": True, "revoked_at": utc_now()}},
  515. )
  516. async def list_stored_invite_links(chat_id: int) -> list[dict[str, Any]]:
  517. await ensure_admin_indexes()
  518. cursor = invite_linksdb.find(
  519. {"bot_id": BOT_PROFILE_ID, "chat_id": int(chat_id)}
  520. ).sort(
  521. "created_at", DESCENDING
  522. )
  523. return [doc async for doc in cursor]
  524. async def dashboard_counts() -> dict[str, int]:
  525. await ensure_admin_indexes()
  526. return {
  527. "chats": await managed_chatsdb.count_documents({"accessible": {"$ne": False}}),
  528. "audit_events": await auditdb.count_documents({}),
  529. }