Add authorization and runtime settings services

services/auth.py

@@ -0,0 +1,105 @@
+import json
+import os
+from config import AUTH_FILE, OWNER_ID, OWNER_USERNAME
+
+_authorized: set[int] = set()
+_banned: set[int] = set()
+_known_users: set[int] = set()
+_public_mode: bool = False
+
+
+def load():
+    global _authorized, _banned
+    if os.path.exists(AUTH_FILE):
+        with open(AUTH_FILE) as f:
+            data = json.load(f)
+        _authorized = {int(x) for x in data.get("authorized", [])}
+        _banned = {int(x) for x in data.get("banned", [])}
+    if OWNER_ID:
+        _authorized.add(OWNER_ID)
+
+
+def save():
+    with open(AUTH_FILE, "w") as f:
+        json.dump(
+            {"authorized": sorted(_authorized), "banned": sorted(_banned)},
+            f,
+            indent=2,
+        )
+
+
+def is_owner(user_id: int, username: str = "") -> bool:
+    if OWNER_ID > 0 and user_id == OWNER_ID:
+        return True
+    if OWNER_USERNAME and username and username.lower() == OWNER_USERNAME.lower():
+        return True
+    return False
+
+
+def is_strictly_authorized(user_id: int, username: str = "") -> bool:
+    """Always checks the actual auth list — ignores public mode. Used for sensitive commands (VC)."""
+    if is_owner(user_id, username):
+        return True
+    return user_id in _authorized and user_id not in _banned
+
+
+def is_authorized(user_id: int, username: str = "") -> bool:
+    if is_owner(user_id, username):
+        return True
+    if user_id in _banned:
+        return False
+    if _public_mode:
+        return True
+    return user_id in _authorized
+
+
+def set_public(enabled: bool):
+    global _public_mode
+    _public_mode = enabled
+    try:
+        from services import settings as settings_store
+        settings_store.save()
+    except Exception:
+        pass
+
+
+def is_public() -> bool:
+    return _public_mode
+
+
+def add_user(user_id: int):
+    _authorized.add(user_id)
+    _banned.discard(user_id)
+    save()
+
+
+def remove_user(user_id: int):
+    _authorized.discard(user_id)
+    save()
+
+
+def ban_user(user_id: int):
+    _banned.add(user_id)
+    _authorized.discard(user_id)
+    save()
+
+
+def unban_user(user_id: int):
+    _banned.discard(user_id)
+    save()
+
+
+def register_known(user_id: int, username: str = ""):
+    _known_users.add(user_id)
+
+
+def get_authorized() -> list[int]:
+    return sorted(_authorized)
+
+
+def get_banned() -> list[int]:
+    return sorted(_banned)
+
+
+def get_known_users() -> list[int]:
+    return sorted(_known_users)

services/settings.py

@@ -0,0 +1,41 @@
+import json
+import os
+from config import SETTINGS_FILE
+from services import ai_client, auth
+from utils import context
+
+_loading = False
+
+
+def load():
+    global _loading
+    if not os.path.exists(SETTINGS_FILE):
+        save()
+        return
+
+    _loading = True
+    try:
+        with open(SETTINGS_FILE, "r", encoding="utf-8") as f:
+            data = json.load(f)
+        if isinstance(data, dict):
+            ai_client.import_provider_settings(data.get("ai", {}))
+            context.import_settings(data.get("context", {}))
+            auth.set_public(bool(data.get("public_mode", False)))
+    finally:
+        _loading = False
+
+
+def save():
+    if _loading:
+        return
+
+    data = {
+        "ai": ai_client.export_provider_settings(),
+        "public_mode": auth.is_public(),
+        "context": context.export_settings(),
+    }
+    tmp = f"{SETTINGS_FILE}.tmp"
+    with open(tmp, "w", encoding="utf-8") as f:
+        json.dump(data, f, ensure_ascii=False, indent=2)
+        f.write("\n")
+    os.replace(tmp, SETTINGS_FILE)