admin_access_manager.py

# admin_access_manager.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""
Divid Teacher — مدير دخول المدراء والصلاحيات.

- كلمات المرور تبقى في Hugging Face Secrets فقط.
- المدير الرئيسي: ADMIN_PANEL_KEY
- المدير 1: ADMIN_MANAGER_1_KEY
- المدير 2: ADMIN_MANAGER_2_KEY
- الصلاحيات تحفظ في system/admin_accounts.json بدون أي قيمة سرية.
- كل عملية حساسة تفحص الصلاحية على السيرفر.
"""

from __future__ import annotations

import copy
import hmac
import json
import os
import threading
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, Iterable, List, Optional

VERSION = "admin_access_manager_v1_multi_admin_button_permissions"
SCHEMA_VERSION = "admin_access_schema_v1"

PERMISSIONS: Dict[str, str] = {
    "dashboard.view": "عرض لوحة الإدارة",
    "students.view": "عرض الطلاب والبحث فيهم",
    "students.edit": "تعديل بيانات الطالب",
    "students.status": "تغيير حالة حساب الطالب",
    "students.reset_access_key": "تغيير مفتاح دخول الطالب",
    "students.reset_parent_password": "إعادة ضبط كلمة مرور ولي الأمر",
    "students.devices": "إدارة أجهزة الطالب",
    "students.delete": "حذف حساب طالب",
    "students.academic.view": "عرض الملف الأكاديمي للطالب",
    "students.reports.view": "عرض تقارير الطالب",
    "question_reports.view": "عرض بلاغات الأسئلة",
    "question_reports.update": "تحديث حالة بلاغات الأسئلة",
    "repositories.view": "عرض المستودعات وملفاتها",
    "repositories.open": "فتح ومعاينة ملفات المستودع",
    "repositories.download": "تنزيل ملفات المستودع",
    "repositories.edit": "تحرير وحفظ الملفات النصية",
    "repositories.upload": "رفع أو استبدال ملفات المستودع",
    "repositories.delete": "حذف ملفات المستودع",
    "questions.audit.scan": "فحص بنوك الأسئلة",
    "questions.audit.create": "إنشاء أو استئناف مهمة تدقيق",
    "questions.audit.run": "تشغيل Phi/Qwen لتدقيق الأسئلة",
    "questions.audit.outputs": "عرض مخرجات وسجل التدقيق",
    "questions.repair.view": "عرض الأسئلة المستبعدة وأسبابها",
    "questions.repair.create": "إنشاء أو استئناف مهمة إصلاح",
    "questions.repair.run": "تشغيل إصلاح الأسئلة وإعادة تدقيقها",
    "system.snapshot": "عرض لقطة النظام",
    "audit.view": "عرض سجل العمليات الإدارية",
    "admins.view": "عرض المديرين والصلاحيات",
    "admins.permissions.manage": "تعديل صلاحيات المديرين",
}

MANAGER_1_DEFAULT = [
    "dashboard.view", "students.view", "students.academic.view",
    "students.reports.view", "question_reports.view",
    "repositories.view", "repositories.open", "repositories.download",
    "questions.audit.scan", "questions.audit.create", "questions.audit.run",
    "questions.audit.outputs", "questions.repair.view",
    "questions.repair.create", "questions.repair.run", "system.snapshot",
]

MANAGER_2_DEFAULT = [
    "dashboard.view", "students.view", "students.academic.view",
    "students.reports.view", "question_reports.view",
]


def _now() -> str:
    return datetime.now(timezone.utc).isoformat()


def _text(value: Any) -> str:
    return str(value if value is not None else "").strip()


class AdminAccessManager:
    def __init__(self) -> None:
        self.version = VERSION
        self.schema_version = SCHEMA_VERSION
        self._lock = threading.RLock()
        system_folder = Path(os.getenv("SYSTEM_FOLDER", "system"))
        system_folder.mkdir(parents=True, exist_ok=True)
        self.accounts_file = system_folder / "admin_accounts.json"
        self.audit_file = system_folder / "admin_access_audit.jsonl"
        self._ensure_accounts_file()

    def default_accounts(self) -> Dict[str, Any]:
        return {
            "schema_version": SCHEMA_VERSION,
            "updated_at": _now(),
            "accounts": [
                {
                    "admin_id": "owner",
                    "display_name": "المدير الرئيسي",
                    "secret_name": "ADMIN_PANEL_KEY",
                    "enabled": True,
                    "permissions": ["*"],
                    "protected": True,
                },
                {
                    "admin_id": "manager_1",
                    "display_name": "مدير 1",
                    "secret_name": "ADMIN_MANAGER_1_KEY",
                    "enabled": True,
                    "permissions": list(MANAGER_1_DEFAULT),
                    "protected": False,
                },
                {
                    "admin_id": "manager_2",
                    "display_name": "مدير 2",
                    "secret_name": "ADMIN_MANAGER_2_KEY",
                    "enabled": True,
                    "permissions": list(MANAGER_2_DEFAULT),
                    "protected": False,
                },
            ],
        }

    def _read_config(self) -> Dict[str, Any]:
        try:
            with self._lock:
                if not self.accounts_file.exists():
                    return {}
                value = json.loads(self.accounts_file.read_text(encoding="utf-8"))
            return value if isinstance(value, dict) else {}
        except Exception:
            return {}

    def _write_config(self, data: Dict[str, Any]) -> None:
        safe = copy.deepcopy(data if isinstance(data, dict) else {})
        safe["schema_version"] = SCHEMA_VERSION
        safe["updated_at"] = _now()
        for item in safe.get("accounts", []):
            if not isinstance(item, dict):
                continue
            for forbidden in ("password", "key", "secret", "secret_value", "token"):
                item.pop(forbidden, None)
        with self._lock:
            self.accounts_file.parent.mkdir(parents=True, exist_ok=True)
            tmp = self.accounts_file.with_suffix(".tmp")
            tmp.write_text(json.dumps(safe, ensure_ascii=False, indent=2), encoding="utf-8")
            tmp.replace(self.accounts_file)

    def _ensure_accounts_file(self) -> None:
        if not self.accounts_file.exists():
            self._write_config(self.default_accounts())
            return
        data = self._read_config()
        if not isinstance(data, dict) or not isinstance(data.get("accounts"), list):
            self._write_config(self.default_accounts())
            return
        accounts = data["accounts"]
        by_id = {
            _text(item.get("admin_id")): item
            for item in accounts
            if isinstance(item, dict) and _text(item.get("admin_id"))
        }
        changed = False
        for default in self.default_accounts()["accounts"]:
            if default["admin_id"] not in by_id:
                accounts.append(copy.deepcopy(default))
                changed = True
        owner = next((x for x in accounts if isinstance(x, dict) and _text(x.get("admin_id")) == "owner"), None)
        if owner is not None:
            fixed = {
                "secret_name": "ADMIN_PANEL_KEY",
                "enabled": True,
                "permissions": ["*"],
                "protected": True,
            }
            for key, value in fixed.items():
                if owner.get(key) != value:
                    owner[key] = copy.deepcopy(value)
                    changed = True
        if changed:
            data["accounts"] = accounts
            self._write_config(data)

    def _accounts(self) -> List[Dict[str, Any]]:
        data = self._read_config()
        return [
            item for item in (data.get("accounts") or [])
            if isinstance(item, dict) and _text(item.get("admin_id"))
        ]

    def normalize_permissions(self, values: Iterable[Any]) -> List[str]:
        result: List[str] = []
        for value in values or []:
            permission = _text(value)
            if permission == "*":
                return ["*"]
            if permission in PERMISSIONS and permission not in result:
                result.append(permission)
        return result

    def _public_account(self, account: Dict[str, Any]) -> Dict[str, Any]:
        secret_name = _text(account.get("secret_name"))
        return {
            "admin_id": _text(account.get("admin_id")),
            "display_name": _text(account.get("display_name") or account.get("admin_id")),
            "secret_name": secret_name,
            "secret_configured": bool(_text(os.getenv(secret_name, ""))) if secret_name else False,
            "enabled": bool(account.get("enabled", True)),
            "permissions": self.normalize_permissions(account.get("permissions") or []),
            "protected": bool(account.get("protected", False)),
        }

    def public_config(self) -> Dict[str, Any]:
        return {
            "success": True,
            "version": self.version,
            "schema_version": self.schema_version,
            "permissions": dict(PERMISSIONS),
            "accounts": [self._public_account(x) for x in self._accounts()],
        }

    def authenticate(self, password: Any) -> Dict[str, Any]:
        supplied = _text(password)
        if not supplied:
            return {"success": False, "authorized": False, "message": "أدخل كلمة مرور المدير.", "version": self.version}
        configured = 0
        for account in self._accounts():
            if not bool(account.get("enabled", True)):
                continue
            secret_name = _text(account.get("secret_name"))
            expected = _text(os.getenv(secret_name, "")) if secret_name else ""
            if not expected:
                continue
            configured += 1
            if hmac.compare_digest(supplied, expected):
                public = self._public_account(account)
                self._audit("admin.login", actor=public["admin_id"], success=True)
                return {
                    "success": True,
                    "authorized": True,
                    "admin": public,
                    "admin_id": public["admin_id"],
                    "display_name": public["display_name"],
                    "permissions": public["permissions"],
                    "version": self.version,
                }
        self._audit("admin.login_failed", actor="unknown", success=False)
        return {
            "success": False,
            "authorized": False,
            "message": "كلمة مرور المدير غير صحيحة." if configured else "لا يوجد Secret إداري مضبوط. اضبط ADMIN_PANEL_KEY على الأقل.",
            "version": self.version,
        }

    def has_permission(self, admin: Dict[str, Any], permission: str) -> bool:
        permissions = admin.get("permissions") or [] if isinstance(admin, dict) else []
        return "*" in permissions or _text(permission) in permissions

    def guard(self, password: Any, permission: str, action: str = "", target: str = "") -> Dict[str, Any]:
        auth = self.authenticate(password)
        if not auth.get("success"):
            return auth
        admin = auth.get("admin") or {}
        if not self.has_permission(admin, permission):
            self._audit(action or "permission.denied", actor=_text(admin.get("admin_id")), target=target, success=False, details={"permission": permission})
            return {
                "success": False,
                "authorized": False,
                "permission": permission,
                "admin_id": admin.get("admin_id", ""),
                "display_name": admin.get("display_name", ""),
                "message": "هذا المدير لا يملك صلاحية: " + PERMISSIONS.get(permission, permission),
                "version": self.version,
            }
        return {
            "success": True,
            "authorized": True,
            "permission": permission,
            "admin": admin,
            "admin_id": admin.get("admin_id", ""),
            "display_name": admin.get("display_name", ""),
            "version": self.version,
        }

    def list_admins(self) -> Dict[str, Any]:
        data = self.public_config()
        return {"success": True, "admins": data["accounts"], "permissions": data["permissions"], "version": self.version}

    def get_admin(self, admin_id: Any) -> Dict[str, Any]:
        wanted = _text(admin_id)
        for item in self._accounts():
            if _text(item.get("admin_id")) == wanted:
                return {"success": True, "admin": self._public_account(item), "version": self.version}
        return {"success": False, "message": "المدير غير موجود.", "version": self.version}

    def update_admin_profile(
        self,
        actor_password: Any,
        target_admin_id: Any,
        display_name: Any = "",
        enabled: bool = True,
        permissions: Optional[Iterable[Any]] = None,
    ) -> Dict[str, Any]:
        auth = self.guard(actor_password, "admins.permissions.manage", action="admins.permissions.update", target=_text(target_admin_id))
        if not auth.get("success"):
            return auth
        target_id = _text(target_admin_id)
        data = self._read_config()
        accounts = data.get("accounts") or []
        target = next((x for x in accounts if isinstance(x, dict) and _text(x.get("admin_id")) == target_id), None)
        if target is None:
            return {"success": False, "message": "المدير المطلوب غير موجود.", "version": self.version}
        if _text(display_name):
            target["display_name"] = _text(display_name)
        if target_id == "owner":
            target["enabled"] = True
            target["permissions"] = ["*"]
            target["secret_name"] = "ADMIN_PANEL_KEY"
            target["protected"] = True
        else:
            target["enabled"] = bool(enabled)
            target["permissions"] = self.normalize_permissions(permissions or [])
        data["accounts"] = accounts
        self._write_config(data)
        actor = auth.get("admin") or {}
        self._audit("admins.permissions.updated", actor=_text(actor.get("admin_id")), target=target_id, success=True, details={"enabled": target.get("enabled"), "permissions": target.get("permissions")})
        return {"success": True, "message": "تم حفظ صلاحيات المدير.", "admin": self._public_account(target), "version": self.version}

    def permission_choices(self) -> List[tuple]:
        return [(label, permission) for permission, label in PERMISSIONS.items()]

    def _audit(self, action: str, actor: str = "", target: str = "", success: bool = True, details: Optional[Dict[str, Any]] = None) -> None:
        event = {"created_at": _now(), "action": _text(action), "actor": _text(actor) or "unknown", "target": _text(target), "success": bool(success), "details": details if isinstance(details, dict) else {}, "version": self.version}
        try:
            with self._lock:
                self.audit_file.parent.mkdir(parents=True, exist_ok=True)
                with self.audit_file.open("a", encoding="utf-8") as file:
                    file.write(json.dumps(event, ensure_ascii=False) + "\n")
        except Exception:
            pass

    def log_authorized_action(self, password: Any, permission: str, action: str, target: str = "", details: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
        auth = self.guard(password, permission, action=action, target=target)
        if not auth.get("success"):
            return auth
        admin = auth.get("admin") or {}
        self._audit(action, actor=_text(admin.get("admin_id")), target=target, success=True, details=details)
        return auth

    def summary(self) -> Dict[str, Any]:
        admins = self.list_admins().get("admins") or []
        return {
            "success": True,
            "ready": True,
            "version": self.version,
            "schema_version": self.schema_version,
            "admins": len(admins),
            "configured_secrets": sum(1 for x in admins if x.get("secret_configured")),
            "permission_count": len(PERMISSIONS),
            "passwords_saved_to_file": False,
            "owner_secret": "ADMIN_PANEL_KEY",
            "manager_1_secret": "ADMIN_MANAGER_1_KEY",
            "manager_2_secret": "ADMIN_MANAGER_2_KEY",
        }


_MANAGER: Optional[AdminAccessManager] = None
_MANAGER_LOCK = threading.Lock()


def get_manager() -> AdminAccessManager:
    global _MANAGER
    if _MANAGER is None:
        with _MANAGER_LOCK:
            if _MANAGER is None:
                _MANAGER = AdminAccessManager()
    return _MANAGER


def authenticate_admin(password: Any) -> Dict[str, Any]:
    return get_manager().authenticate(password)


def summary() -> Dict[str, Any]:
    return get_manager().summary()