"""سجل الموديولات السحابي (VDE) — يوقّع حزم الكود ويسلّمها OTA للأجهزة المعتمدة. - المفتاح الخاص للتوقيع على السحابة؛ الأجهزة تتحقق بالمفتاح العام (يُزوّد مرة). - رفع موديول (أدمن) → تعبئة + توقيع + تخزين .vmod + رفع modules_version. - تنزيل (جهاز معتمد فقط) عبر بوابة الحماية نفسها. صيغة الحزمة والهاش **مطابقان** لـ edge/vde/loader.py (نشرتان منفصلتان — تُبقيان متطابقتين). """ import base64 import hashlib import io import json import os import zipfile from cryptography.hazmat.primitives import serialization from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey from fastapi import (APIRouter, Depends, File, Form, Header, HTTPException, UploadFile) from fastapi.responses import FileResponse from .db import DATA_DIR from .device import gate, get_branch from .models import Branch REG_DIR = os.path.join(DATA_DIR, "module_registry") os.makedirs(REG_DIR, exist_ok=True) INDEX = os.path.join(REG_DIR, "index.json") VERSION_FILE = os.path.join(REG_DIR, "modules_version") KEY_FILE = os.path.join(REG_DIR, "signing_key.bin") ADMIN_TOKEN = os.environ.get("VEEDON_ADMIN_TOKEN", "veedon-admin-dev") device_router = APIRouter(prefix="/api/device", tags=["modules-device"]) admin_router = APIRouter(prefix="/api/admin/modules", tags=["modules-admin"]) # ---------- المفاتيح ---------- def _priv() -> bytes: if not os.path.exists(KEY_FILE): sk = Ed25519PrivateKey.generate() with open(KEY_FILE, "wb") as f: f.write(sk.private_bytes(serialization.Encoding.Raw, serialization.PrivateFormat.Raw, serialization.NoEncryption())) with open(KEY_FILE, "rb") as f: return f.read() def _pub_b64() -> str: sk = Ed25519PrivateKey.from_private_bytes(_priv()) pub = sk.public_key().public_bytes(serialization.Encoding.Raw, serialization.PublicFormat.Raw) return base64.b64encode(pub).decode() # ---------- التعبئة والتوقيع (مطابق لـ loader._digest) ---------- def _digest(manifest_no_sig: dict, files: dict) -> bytes: h = hashlib.sha256() h.update(json.dumps(manifest_no_sig, sort_keys=True, ensure_ascii=False).encode()) for name in sorted(files): h.update(name.encode()) h.update(files[name]) return h.digest() def _pack_signed(manifest: dict, files: dict) -> bytes: man = {k: v for k, v in manifest.items() if k != "signature"} sig = Ed25519PrivateKey.from_private_bytes(_priv()).sign(_digest(man, files)) man["signature"] = base64.b64encode(sig).decode() buf = io.BytesIO() with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as z: z.writestr("manifest.json", json.dumps(man, ensure_ascii=False)) for name, data in files.items(): z.writestr(name, data) return buf.getvalue() def _load_index() -> dict: if os.path.exists(INDEX): with open(INDEX, encoding="utf-8") as f: return json.load(f) return {} def _bump_version() -> int: v = 0 if os.path.exists(VERSION_FILE): v = int(open(VERSION_FILE).read() or "0") v += 1 with open(VERSION_FILE, "w") as f: f.write(str(v)) return v # ---------- أدمن: رفع موديول ---------- @admin_router.get("/pubkey") def pubkey(x_admin_token: str = Header(default="")): if x_admin_token != ADMIN_TOKEN: raise HTTPException(403, "admin token required") return {"public_key": _pub_b64()} @admin_router.get("") def list_admin(x_admin_token: str = Header(default="")): if x_admin_token != ADMIN_TOKEN: raise HTTPException(403, "admin token required") mv = int(open(VERSION_FILE).read()) if os.path.exists(VERSION_FILE) else 0 return {"modules_version": mv, "modules": _load_index()} @admin_router.post("") def publish(x_admin_token: str = Header(default=""), key: str = Form(...), version: int = Form(...), kind: str = Form("custom"), triggers: str = Form("motion"), detector: UploadFile = File(...), model: UploadFile = File(None)): """يرفع detector.py (+نموذج اختياري) → يوقّع → يخزّن → يرفع modules_version.""" if x_admin_token != ADMIN_TOKEN: raise HTTPException(403, "admin token required") files = {"detector.py": detector.file.read()} manifest = {"key": key, "version": int(version), "kind": kind, "entry": "detector.py", "triggers": [t.strip() for t in triggers.split(",") if t.strip()]} if model is not None: mname = os.path.basename(model.filename or f"{key}.onnx") files[mname] = model.file.read() manifest["model"] = mname vmod = _pack_signed(manifest, files) kdir = os.path.join(REG_DIR, key) os.makedirs(kdir, exist_ok=True) with open(os.path.join(kdir, f"v{version}.vmod"), "wb") as f: f.write(vmod) idx = _load_index() entry = idx.setdefault(key, {"kind": kind, "versions": []}) entry.update(kind=kind, triggers=manifest["triggers"], latest=int(version)) if int(version) not in entry["versions"]: entry["versions"].append(int(version)) with open(INDEX, "w", encoding="utf-8") as f: json.dump(idx, f, ensure_ascii=False) mv = _bump_version() return {"key": key, "version": version, "modules_version": mv, "size": len(vmod), "signed": True} # ---------- جهاز: قائمة + تنزيل (خلف بوابة الحماية) ---------- @device_router.get("/module_pubkey") def module_pubkey(): """المفتاح العام لتوقيع الموديولات — **عام** (يُزوّده install.sh للجهاز للتحقق).""" return {"public_key": _pub_b64()} @device_router.get("/modules") def list_modules(branch: Branch = Depends(get_branch), x_device_fingerprint: str = Header(default="")): allowed, reason = gate(branch, x_device_fingerprint) if not allowed: raise HTTPException(403, reason) mv = int(open(VERSION_FILE).read()) if os.path.exists(VERSION_FILE) else 0 return {"modules_version": mv, "modules": _load_index()} @device_router.get("/modules/{key}/{version}") def download_module(key: str, version: int, branch: Branch = Depends(get_branch), x_device_fingerprint: str = Header(default="")): allowed, reason = gate(branch, x_device_fingerprint) if not allowed: raise HTTPException(403, reason) path = os.path.join(REG_DIR, os.path.basename(key), f"v{int(version)}.vmod") if not os.path.isfile(path): raise HTTPException(404, "module version not published") return FileResponse(path, filename=f"{key}-v{version}.vmod", media_type="application/zip")