Create auth.py

bot/integrations/auth.py

@@ -0,0 +1,85 @@
+# PATH: bot/integrations/auth.py
+from __future__ import annotations
+
+from typing import Any, Awaitable, Callable, Dict, Optional
+
+
+# We KNOW cf_worker2 has is_allowed (your bot/core/auth.py already imports it),
+# so we treat that as the primary source.
+try:
+    from bot.integrations.cf_worker2 import is_allowed as _is_allowed  # type: ignore
+except Exception:
+    _is_allowed = None  # type: ignore
+
+
+# Optional admin helpers might exist in cf_worker2 (or cf_worker1 in some setups).
+_allow_user: Optional[Callable[[int], Awaitable[Any]]] = None
+_disallow_user: Optional[Callable[[int], Awaitable[Any]]] = None
+_get_stats: Optional[Callable[[], Awaitable[Any]]] = None
+
+for _modname in ("bot.integrations.cf_worker2", "bot.integrations.cf_worker1"):
+    try:
+        _mod = __import__(_modname, fromlist=["*"])
+        if _allow_user is None and hasattr(_mod, "allow_user"):
+            _allow_user = getattr(_mod, "allow_user")
+        if _disallow_user is None and hasattr(_mod, "disallow_user"):
+            _disallow_user = getattr(_mod, "disallow_user")
+        if _get_stats is None and hasattr(_mod, "get_stats"):
+            _get_stats = getattr(_mod, "get_stats")
+    except Exception:
+        continue
+
+
+def _coerce_allowed(v: Any) -> bool:
+    if isinstance(v, bool):
+        return v
+    if isinstance(v, dict):
+        # common patterns from worker APIs
+        for k in ("allowed", "is_allowed", "result"):
+            if k in v:
+                return bool(v.get(k))
+        # sometimes {"ok": True} means allowed; sometimes not — be conservative
+        if "ok" in v and ("allowed" not in v and "is_allowed" not in v):
+            return bool(v.get("ok"))
+        return False
+    return bool(v)
+
+
+async def is_allowed(uid: int) -> bool:
+    """Return True/False for allowlist check."""
+    if _is_allowed is None:
+        # If backend isn't available, default deny (owners/admins bypass in handlers anyway)
+        return False
+    out = await _is_allowed(uid)  # may be bool or dict depending on your worker
+    return _coerce_allowed(out)
+
+
+async def allow_user(uid: int) -> Dict[str, Any]:
+    """Admin: allow a user. Returns dict (ok/err)."""
+    if _allow_user is None:
+        return {"ok": False, "err": "allow_user_not_implemented"}
+    out = await _allow_user(uid)
+    return out if isinstance(out, dict) else {"ok": True}
+
+
+async def disallow_user(uid: int) -> Dict[str, Any]:
+    """Admin: disallow a user. Returns dict (ok/err)."""
+    if _disallow_user is None:
+        return {"ok": False, "err": "disallow_user_not_implemented"}
+    out = await _disallow_user(uid)
+    return out if isinstance(out, dict) else {"ok": True}
+
+
+async def get_stats() -> Dict[str, Any]:
+    """Admin: return stats dict used by /stats."""
+    if _get_stats is None:
+        # Keep handlers.py happy (it expects a dict)
+        return {
+            "ok": False,
+            "err": "stats_not_implemented",
+            "allowed_users": 0,
+            "profiles": 0,
+            "uploads_today": 0,
+        }
+    out = await _get_stats()
+    return out if isinstance(out, dict) else {"ok": True}