# 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()