from __future__ import annotations import base64 import hashlib import hmac import json import mimetypes import os import re import secrets import shutil import struct import threading import uuid import zipfile from dataclasses import dataclass from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any, Iterable from cryptography.hazmat.primitives.ciphers.aead import AESGCM, ChaCha20Poly1305 from huggingface_hub import HfApi, hf_hub_download MB = 1024 * 1024 GB = 1024 * MB STATE_REMOTE_PATH = "state/state.json" MAGIC = b"CVLT3" CHUNK_SIZE = 4 * MB def utcnow() -> datetime: return datetime.now(timezone.utc) def iso(value: datetime | None = None) -> str: return (value or utcnow()).isoformat() def parse_dt(value: str | None) -> datetime | None: if not value: return None try: parsed = datetime.fromisoformat(value) return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc) except (TypeError, ValueError): return None @dataclass(frozen=True) class Plan: key: str name: str quota_bytes: int reserved_bytes: int retention_days: int delete_after_age_even_if_downloaded: bool concurrent_downloads: int concurrent_uploads: int analytics_delay_hours: int analytics_history_days: int token_length: int folder_upload: bool encryption: str display_price: str old_price: str | None chat_access: bool priority: str PLANS: dict[str, Plan] = { "free": Plan( key="free", name="Free", quota_bytes=250 * MB, reserved_bytes=250 * MB, retention_days=2, delete_after_age_even_if_downloaded=True, concurrent_downloads=1, concurrent_uploads=1, analytics_delay_hours=24, analytics_history_days=30, token_length=24, folder_upload=True, encryption="none", display_price="$0", old_price=None, chat_access=False, priority="Low", ), "plus": Plan( key="plus", name="Plus", quota_bytes=int(3.5 * GB), reserved_bytes=int(3.5 * GB), retention_days=7, delete_after_age_even_if_downloaded=False, concurrent_downloads=3, concurrent_uploads=3, analytics_delay_hours=12, analytics_history_days=365, token_length=8, folder_upload=True, encryption="aes256-gcm", display_price="$2.49", old_price="$5.00", chat_access=True, priority="Medium", ), "pro": Plan( key="pro", name="Pro", quota_bytes=int(6.5 * GB), reserved_bytes=int(6.5 * GB), retention_days=30, delete_after_age_even_if_downloaded=False, concurrent_downloads=6, concurrent_uploads=6, analytics_delay_hours=6, analytics_history_days=365, token_length=12, folder_upload=True, encryption="chacha20-poly1305", display_price="$4.99", old_price="$10.00", chat_access=True, priority="High", ), "admin": Plan( key="admin", name="Developer", quota_bytes=10**18, reserved_bytes=0, retention_days=36500, delete_after_age_even_if_downloaded=False, concurrent_downloads=10**9, concurrent_uploads=10**9, analytics_delay_hours=0, analytics_history_days=3650, token_length=16, folder_upload=True, encryption="chacha20-poly1305", display_price="Developer", old_price=None, chat_access=True, priority="Developer", ), } @dataclass(frozen=True) class Settings: hf_token: str | None dataset_repo: str | None master_key: str admin_password: str public_base_url: str total_storage_capacity_bytes: int system_storage_reserve_bytes: int app_storage_limit_bytes: int purchase_safety_bytes: int temp_dir: Path local_root: Path local_mode: bool def get_settings() -> Settings: temp_dir = Path(os.getenv("TEMP_DIR", "/tmp/cloudvault-temp")) local_root = Path(os.getenv("LOCAL_STORAGE_DIR", "/tmp/cloudvault-local")) temp_dir.mkdir(parents=True, exist_ok=True) local_root.mkdir(parents=True, exist_ok=True) token = os.getenv("HF_TOKEN") or None repo = os.getenv("HF_DATASET_REPO") or None forced_local = os.getenv("LOCAL_MODE", "").lower() in {"1", "true", "yes"} space_host = os.getenv("SPACE_HOST", "").strip() public_url = os.getenv("PUBLIC_BASE_URL", "").strip().rstrip("/") if not public_url and space_host: public_url = f"https://{space_host}" total_storage_capacity_bytes = int(float(os.getenv("TOTAL_STORAGE_GB", "95")) * GB) system_storage_reserve_bytes = int(float(os.getenv("SYSTEM_STORAGE_RESERVE_GB", "5")) * GB) system_storage_reserve_bytes = min(max(0, system_storage_reserve_bytes), total_storage_capacity_bytes) package_pool_bytes = max(0, total_storage_capacity_bytes - system_storage_reserve_bytes) configured_package_pool = int(float(os.getenv("APP_STORAGE_LIMIT_GB", "90")) * GB) return Settings( hf_token=token, dataset_repo=repo, master_key=os.getenv("APP_MASTER_KEY", "CHANGE-ME-IN-SPACE-SECRETS"), admin_password="eses1428", public_base_url=public_url, total_storage_capacity_bytes=total_storage_capacity_bytes, system_storage_reserve_bytes=system_storage_reserve_bytes, app_storage_limit_bytes=min(package_pool_bytes, max(0, configured_package_pool)), purchase_safety_bytes=int(float(os.getenv("PURCHASE_SAFETY_GB", "0")) * GB), temp_dir=temp_dir, local_root=local_root, local_mode=forced_local or not (token and repo), ) SETTINGS = get_settings() def human_bytes(value: int) -> str: size = float(max(0, value)) for unit in ("B", "KB", "MB", "GB", "TB"): if size < 1024 or unit == "TB": return f"{size:.0f} {unit}" if unit in {"B", "KB"} else f"{size:.2f} {unit}" size /= 1024 return f"{size:.2f} TB" def safe_filename(name: str) -> str: clean = Path(name or "file.bin").name.replace("\x00", "") clean = re.sub(r"[\r\n]", "", clean).strip() return clean[:180] or "file.bin" def validate_registration(username: str, password: str) -> list[str]: username = (username or "").strip() password = password or "" errors: list[str] = [] if len(username) < 3: errors.append("Username must contain at least 3 characters.") if len(username) > 32: errors.append("Username may contain at most 32 characters.") if not re.fullmatch(r"[A-Za-z0-9_.-]+", username): errors.append("Username may contain letters, numbers, dot, underscore and dash only.") if username.casefold() == "admin": errors.append("This username is reserved.") if len(password) < 6: errors.append("Password must contain at least 6 characters.") if not re.search(r"[A-Za-z]", password): errors.append("Password must contain at least one letter.") if not re.search(r"\d", password): errors.append("Password must contain at least one number.") return errors def hash_password(password: str) -> str: salt = secrets.token_bytes(16) digest = hashlib.scrypt(password.encode("utf-8"), salt=salt, n=2**14, r=8, p=1, dklen=32) return "scrypt$" + base64.urlsafe_b64encode(salt).decode() + "$" + base64.urlsafe_b64encode(digest).decode() def verify_password(password: str, stored: str | None) -> bool: try: algorithm, salt_text, digest_text = (stored or "").split("$", 2) if algorithm != "scrypt": return False salt = base64.urlsafe_b64decode(salt_text.encode()) expected = base64.urlsafe_b64decode(digest_text.encode()) actual = hashlib.scrypt(password.encode("utf-8"), salt=salt, n=2**14, r=8, p=1, dklen=32) return hmac.compare_digest(actual, expected) except (ValueError, TypeError): return False def random_share_id(length: int) -> str: alphabet = "ABCDEFGHJKLMNPQRSTUVWXYZabcdefghijkmnopqrstuvwxyz23456789" return "".join(secrets.choice(alphabet) for _ in range(length)) class StorageBackend: def __init__(self, settings: Settings) -> None: self.settings = settings self.api = HfApi(token=settings.hf_token) if not settings.local_mode else None if self.api and settings.dataset_repo: self.api.create_repo(repo_id=settings.dataset_repo, repo_type="dataset", private=True, exist_ok=True) @property def persistent(self) -> bool: return not self.settings.local_mode def upload(self, local_path: Path, remote_path: str, message: str) -> None: if self.settings.local_mode: target = self.settings.local_root / remote_path target.parent.mkdir(parents=True, exist_ok=True) shutil.copy2(local_path, target) return assert self.api and self.settings.dataset_repo self.api.upload_file( path_or_fileobj=str(local_path), path_in_repo=remote_path, repo_id=self.settings.dataset_repo, repo_type="dataset", commit_message=message, ) def download(self, remote_path: str, local_path: Path) -> bool: local_path.parent.mkdir(parents=True, exist_ok=True) if self.settings.local_mode: source = self.settings.local_root / remote_path if not source.exists(): return False shutil.copy2(source, local_path) return True assert self.settings.dataset_repo try: cached = hf_hub_download( repo_id=self.settings.dataset_repo, repo_type="dataset", filename=remote_path, token=self.settings.hf_token, ) shutil.copy2(cached, local_path) return True except Exception: return False def delete_many(self, remote_paths: Iterable[str], message: str = "Delete CloudVaultHub files") -> None: paths = [p for p in dict.fromkeys(remote_paths) if p] if not paths: return if self.settings.local_mode: for remote in paths: (self.settings.local_root / remote).unlink(missing_ok=True) return assert self.api and self.settings.dataset_repo from huggingface_hub import CommitOperationDelete operations = [CommitOperationDelete(path_in_repo=path) for path in paths] try: self.api.create_commit( repo_id=self.settings.dataset_repo, repo_type="dataset", operations=operations, commit_message=message, ) except Exception: pass BACKEND = StorageBackend(SETTINGS) class StateStore: def __init__(self, backend: StorageBackend, settings: Settings) -> None: self.backend = backend self.settings = settings self.lock = threading.RLock() self.state: dict[str, Any] = { "version": 3, "users": {}, "files": {}, "events": [], "chat": [], "chat_rooms": {}, "support_tickets": [], "config": {}, } self._load() def _load(self) -> None: changed = False loaded_existing = False with self.lock: temp = self.settings.temp_dir / f"state-load-{uuid.uuid4().hex}.json" if self.backend.download(STATE_REMOTE_PATH, temp): loaded_existing = True try: loaded = json.loads(temp.read_text(encoding="utf-8")) if isinstance(loaded, dict): self.state.update(loaded) except (OSError, json.JSONDecodeError): pass finally: temp.unlink(missing_ok=True) changed |= self._normalize_unlocked() changed |= self._ensure_admin_unlocked() if changed or not loaded_existing: self._save_unlocked("Initialize or upgrade CloudVaultHub state") def _normalize_unlocked(self) -> bool: changed = False defaults = { "system_enabled": True, "chat_enabled": True, "plus_chat_enabled": True, "pro_chat_enabled": False, "support_enabled": True, "support_force_open": False, "support_auto_paused": False, "purchases_enabled": True, "registrations_enabled": True, "delete_paid_on_expiry": False, "shutdown_mode": False, "chat_mode": "single", "chat_room_capacity": 0, } if not isinstance(self.state.get("users"), dict): self.state["users"] = {}; changed = True if not isinstance(self.state.get("files"), dict): self.state["files"] = {}; changed = True if not isinstance(self.state.get("events"), list): self.state["events"] = []; changed = True if not isinstance(self.state.get("notifications"), list): self.state["notifications"] = []; changed = True if not isinstance(self.state.get("notification_reads"), dict): self.state["notification_reads"] = {}; changed = True if not isinstance(self.state.get("plan_features"), dict): self.state["plan_features"] = {}; changed = True if not isinstance(self.state.get("websites"), dict): self.state["websites"] = {}; changed = True if not isinstance(self.state.get("chat"), list): self.state["chat"] = []; changed = True if not isinstance(self.state.get("chat_rooms"), dict): self.state["chat_rooms"] = {}; changed = True if not isinstance(self.state.get("support_tickets"), list): self.state["support_tickets"] = []; changed = True for ticket in self.state["support_tickets"]: ticket.setdefault("id", uuid.uuid4().hex) ticket.setdefault("user_id", "") ticket.setdefault("username", "User") ticket.setdefault("plan", "plus") ticket.setdefault("subject", "") ticket.setdefault("message", "") ticket.setdefault("created_at", iso()) ticket.setdefault("status", "open") ticket.setdefault("reply", "") ticket.setdefault("replied_at", None) ticket.setdefault("read_by_user", False) if not isinstance(self.state.get("config"), dict): self.state["config"] = {}; changed = True for key, value in defaults.items(): if key not in self.state["config"]: self.state["config"][key] = value; changed = True # CloudVaultHub uses exactly one unlimited Global Chat room. if self.state["config"].get("chat_mode") != "single": self.state["config"]["chat_mode"] = "single"; changed = True if self.state["config"].get("chat_room_capacity") != 0: self.state["config"]["chat_room_capacity"] = 0; changed = True if "global-1" not in self.state["chat_rooms"]: self.state["chat_rooms"]["global-1"] = { "id": "global-1", "name": "Global 1", "paused": False, "created_at": iso() } changed = True if set(self.state["chat_rooms"]) != {"global-1"}: room1 = self.state["chat_rooms"].get("global-1", { "id": "global-1", "name": "Global Chat", "paused": False, "created_at": iso() }) room1["name"] = "Global Chat" self.state["chat_rooms"] = {"global-1": room1} changed = True for message in self.state["chat"]: if message.get("room_id") != "global-1": message["room_id"] = "global-1"; changed = True for user in self.state["users"].values(): if "role" not in user: user["role"] = "user"; changed = True normalized_role = str(user.get("role") or "user").strip().lower() if normalized_role not in {"user", "admin"}: normalized_role = "user" if user.get("role") != normalized_role: user["role"] = normalized_role; changed = True normalized_plan = str(user.get("plan") or "free").strip().lower() if normalized_role == "admin": normalized_plan = "admin" elif normalized_plan not in {"free", "plus", "pro"}: normalized_plan = "free" if user.get("plan") != normalized_plan: user["plan"] = normalized_plan; changed = True if "is_banned" not in user: user["is_banned"] = False; changed = True if "ban_until" not in user: user["ban_until"] = None; changed = True if "delete_on_expiry" not in user: user["delete_on_expiry"] = False; changed = True if user.get("chat_room_id") != "global-1": user["chat_room_id"] = "global-1"; changed = True billing_defaults = { "pending_purchase_plan": None, "pending_purchase_expires_at": None, "billing_subscription_id": None, "billing_customer_id": None, "billing_plan": None, "billing_status": None, "billing_email": None, "billing_auto_renew": False, "billing_renews_at": None, "billing_ends_at": None, "billing_anchor": None, "billing_card_brand": None, "billing_card_last_four": None, "billing_payment_processor": None, "billing_urls": {}, "billing_updated_at": None, } for billing_key, billing_value in billing_defaults.items(): if billing_key not in user: user[billing_key] = billing_value changed = True # Keep only the newest 400 messages in the single Global Chat room. normalized_chat = sorted(self.state["chat"], key=lambda m: m.get("created_at", ""))[-400:] for message in normalized_chat: message["room_id"] = "global-1" if len(normalized_chat) != len(self.state["chat"]): changed = True self.state["chat"] = normalized_chat self.state["version"] = 4 return changed def _ensure_admin_unlocked(self) -> bool: """Guarantee exactly one canonical backend admin account.""" users = self.state["users"] exact = next( (u for u in users.values() if str(u.get("username", "")).strip().casefold() == "admin"), None, ) legacy_admins = [ u for u in users.values() if str(u.get("role", "")).strip().lower() == "admin" ] admin = exact or (legacy_admins[0] if legacy_admins else None) now = utcnow() if admin is None: uid = "cloudvault-admin" admin = { "id": uid, "username": "admin", "password_hash": hash_password(self.settings.admin_password), "plan": "admin", "role": "admin", "plan_started_at": iso(now), "plan_expires_at": None, "created_at": iso(now), "is_banned": False, "ban_until": None, "delete_on_expiry": False, "last_chat_at": None, "chat_room_id": "global-1", "pending_purchase_plan": None, "pending_purchase_expires_at": None, "billing_subscription_id": None, "billing_customer_id": None, "billing_plan": None, "billing_status": None, "billing_email": None, "billing_auto_renew": False, "billing_renews_at": None, "billing_ends_at": None, "billing_anchor": None, "billing_card_brand": None, "billing_card_last_four": None, "billing_payment_processor": None, "billing_urls": {}, "billing_updated_at": None, } users[uid] = admin return True changed = False # Migrate an old role=admin account to the canonical username instead of # creating a second hidden admin account. if str(admin.get("username", "")).strip().casefold() != "admin": admin["username"] = "admin" changed = True required = { "plan": "admin", "role": "admin", "is_banned": False, "ban_until": None, "plan_expires_at": None, "delete_on_expiry": False, "chat_room_id": admin.get("chat_room_id") or "global-1", } for key, value in required.items(): if admin.get(key) != value: admin[key] = value changed = True if not verify_password(self.settings.admin_password, admin.get("password_hash")): admin["password_hash"] = hash_password(self.settings.admin_password) changed = True # No other account may retain backend admin privileges. for other in users.values(): if other is admin: continue if str(other.get("role", "")).strip().lower() == "admin": other["role"] = "user" other["plan"] = "free" other["plan_expires_at"] = None other["delete_on_expiry"] = False changed = True return changed def _save_unlocked(self, message: str) -> None: temp = self.settings.temp_dir / f"state-save-{uuid.uuid4().hex}.json" temp.write_text(json.dumps(self.state, ensure_ascii=False, separators=(",", ":")), encoding="utf-8") self.backend.upload(temp, STATE_REMOTE_PATH, message) temp.unlink(missing_ok=True) def _username_exists_unlocked(self, username: str) -> bool: wanted = username.strip().casefold() return any(str(u.get("username", "")).casefold() == wanted for u in self.state["users"].values()) def _is_admin_unlocked(self, user_id: str | None) -> bool: user = self.state["users"].get(user_id or "") return bool(user and user.get("role") == "admin") def _assert_admin_unlocked(self, user_id: str | None) -> None: if not self._is_admin_unlocked(user_id): raise PermissionError("Developer access is required.") def _effective_reserved_for_user_unlocked( self, user: dict[str, Any], now: datetime | None = None, ) -> int: if user.get("role") == "admin": return 0 current_plan = PLANS.get( str(user.get("plan", "free")), PLANS["free"], ) reserved = current_plan.reserved_bytes check_time = now or utcnow() pending_plan_key = str( user.get("pending_purchase_plan") or "" ) pending_expires = parse_dt( user.get("pending_purchase_expires_at") ) if ( pending_plan_key in {"plus", "pro"} and pending_expires and pending_expires > check_time ): reserved = max( reserved, PLANS[pending_plan_key].reserved_bytes, ) return reserved def _reserved_unlocked( self, exclude_user_id: str | None = None, ) -> int: now = utcnow() total = 0 for uid, user in self.state["users"].items(): if uid == exclude_user_id or user.get("role") == "admin": continue total += self._effective_reserved_for_user_unlocked( user, now, ) return total def _used_unlocked(self, user_id: str | None = None) -> int: return sum( int(f.get("original_size", 0)) for f in self.state["files"].values() if user_id is None or f.get("owner_id") == user_id ) def _admin_used_unlocked(self) -> int: admin_ids = {uid for uid, u in self.state["users"].items() if u.get("role") == "admin"} return sum(int(f.get("original_size", 0)) for f in self.state["files"].values() if f.get("owner_id") in admin_ids) def _package_used_unlocked(self) -> int: admin_ids = {uid for uid, u in self.state["users"].items() if u.get("role") == "admin"} return sum( int(f.get("original_size", 0)) for f in self.state["files"].values() if f.get("owner_id") not in admin_ids ) def _capacity_for_new_reservation_unlocked( self, bytes_needed: int, exclude_user_id: str | None = None, safety_bytes: int = 0, ) -> bool: projected = ( self._reserved_unlocked( exclude_user_id=exclude_user_id ) + max(0, int(bytes_needed)) + max(0, int(safety_bytes)) ) return projected <= self.settings.app_storage_limit_bytes def _capacity_for_paid_plan_unlocked( self, user_id: str, plan_key: str, ) -> bool: return self._capacity_for_new_reservation_unlocked( PLANS[plan_key].reserved_bytes, exclude_user_id=user_id, safety_bytes=self.settings.purchase_safety_bytes, ) def purchase_capacity( self, user_id: str, plan_key: str, ) -> dict[str, Any]: if plan_key not in {"plus", "pro"}: raise ValueError("Invalid paid package.") with self.lock: if user_id not in self.state["users"]: raise ValueError("Account not found.") projected_without_safety = ( self._reserved_unlocked( exclude_user_id=user_id ) + PLANS[plan_key].reserved_bytes ) remaining = max( 0, self.settings.app_storage_limit_bytes - projected_without_safety, ) return { "allowed": remaining >= self.settings.purchase_safety_bytes, "remaining_after_reservation": remaining, "required_safety": self.settings.purchase_safety_bytes, "target_reserved": PLANS[plan_key].reserved_bytes, } def config(self) -> dict[str, Any]: with self.lock: return dict(self.state["config"]) def username_exists(self, username: str) -> bool: with self.lock: return self._username_exists_unlocked(username) def is_admin(self, user_id: str | None) -> bool: with self.lock: return self._is_admin_unlocked(user_id) def create_user(self, username: str, password_hash: str, plan_key: str) -> dict[str, Any]: if plan_key not in {"free", "plus", "pro"}: raise ValueError("Invalid package.") plan = PLANS[plan_key] with self.lock: cfg = self.state["config"] if not cfg.get("system_enabled", True): raise ValueError("The system is temporarily paused.") if not cfg.get("registrations_enabled", True): raise ValueError("New account registration is currently closed.") if plan_key in {"plus", "pro"} and not cfg.get("purchases_enabled", True): raise ValueError("Package activation is currently paused.") if self._username_exists_unlocked(username): raise ValueError("This username is already in use.") if not self._capacity_for_new_reservation_unlocked(plan.reserved_bytes): raise ValueError("The server does not have enough reserved storage to create this account.") user_id = uuid.uuid4().hex now = utcnow() user = { "id": user_id, "username": username.strip(), "password_hash": password_hash, "plan": plan_key, "role": "user", "plan_started_at": iso(now), "plan_expires_at": iso(now + timedelta(days=28)) if plan_key != "free" else None, "plan_expiry_notice_for": None, "created_at": iso(now), "is_banned": False, "ban_until": None, "delete_on_expiry": bool(cfg.get("delete_paid_on_expiry", False) and plan_key != "free"), "last_chat_at": None, "chat_room_id": None, "pending_purchase_plan": None, "pending_purchase_expires_at": None, "billing_subscription_id": None, "billing_customer_id": None, "billing_plan": None, "billing_status": None, "billing_email": None, "billing_auto_renew": False, "billing_renews_at": None, "billing_ends_at": None, "billing_anchor": None, "billing_card_brand": None, "billing_card_last_four": None, "billing_payment_processor": None, "billing_urls": {}, "billing_updated_at": None, } self.state["users"][user_id] = user expiry_text = ( f" It expires on {parse_dt(user['plan_expires_at']).strftime('%d %b %Y · %H:%M UTC')}." if user.get("plan_expires_at") else "" ) self._notify_user_unlocked( user_id, "Welcome to CloudVaultHub", f"Your account was created on the {plan.name} package.{expiry_text}", "account", ) self._save_unlocked("Create CloudVaultHub account") return dict(user) def authenticate(self, username: str, password: str) -> dict[str, Any] | None: with self.lock: wanted = (username or "").strip().casefold() user = next((u for u in self.state["users"].values() if str(u.get("username", "")).casefold() == wanted), None) if not user or not verify_password(password, user.get("password_hash")): return None now = utcnow() ban_until = parse_dt(user.get("ban_until")) if user.get("is_banned"): if ban_until and ban_until <= now: user["is_banned"] = False user["ban_until"] = None self._save_unlocked("Automatically remove expired ban") else: when = "permanently" if not ban_until else f"until {ban_until.strftime('%d %b %Y %H:%M UTC')}" raise PermissionError(f"This account is banned {when}.") if user.get("role") != "admin" and not self.state["config"].get("system_enabled", True): raise PermissionError("CloudVaultHub is temporarily paused by the developer.") return dict(user) def get_user(self, user_id: str | None) -> dict[str, Any] | None: if not user_id: return None self.cleanup(user_id) with self.lock: user = self.state["users"].get(user_id) return dict(user) if user else None def list_users(self, admin_id: str) -> list[dict[str, Any]]: with self.lock: self._assert_admin_unlocked(admin_id) users = [dict(u) for u in self.state["users"].values() if u.get("role") != "admin"] return sorted(users, key=lambda u: u.get("created_at", ""), reverse=True) def change_plan(self, user_id: str, plan_key: str) -> dict[str, Any]: if plan_key not in {"free", "plus", "pro"}: raise ValueError("Invalid package.") plan = PLANS[plan_key] with self.lock: user = self.state["users"].get(user_id) if not user: raise ValueError("Account not found.") if user.get("role") == "admin": raise ValueError("The developer account does not use packages.") cfg = self.state["config"] if not cfg.get("system_enabled", True): raise ValueError("The system is temporarily paused.") if plan_key in {"plus", "pro"} and not cfg.get("purchases_enabled", True): raise ValueError("Package activation is currently paused.") if ( plan_key in {"plus", "pro"} and not self._capacity_for_paid_plan_unlocked( user_id, plan_key, ) ): raise ValueError( "The server cannot reserve this package while keeping " f"{human_bytes(self.settings.purchase_safety_bytes)} free." ) now = utcnow() previous_plan = user.get("plan", "free") user["plan"] = plan_key if previous_plan == "free" and plan_key in {"plus", "pro"}: user["chat_room_id"] = None user["plan_started_at"] = iso(now) user["plan_expires_at"] = iso(now + timedelta(days=28)) if plan_key != "free" else None user["plan_expiry_notice_for"] = None user["delete_on_expiry"] = bool(cfg.get("delete_paid_on_expiry", False) and plan_key != "free") removed = self._trim_user_unlocked(user_id, plan.quota_bytes) expiry_text = ( f" Your package expires on {parse_dt(user['plan_expires_at']).strftime('%d %b %Y · %H:%M UTC')}." if user.get("plan_expires_at") else "" ) self._notify_user_unlocked( user_id, "Your package changed", f"Your package changed from {PLANS.get(previous_plan, PLANS['free']).name} to {plan.name}.{expiry_text}", "account", ) self._save_unlocked("Change CloudVaultHub package") self.backend.delete_many(removed, "Trim files after package change") return self.get_user(user_id) or {} def admin_change_user_plan(self, admin_id: str, user_id: str, plan_key: str, duration_days: int = 28) -> dict[str, Any]: """Developer-only package assignment that does not require a checkout.""" if plan_key not in {"free", "plus", "pro"}: raise ValueError("Invalid package.") with self.lock: self._assert_admin_unlocked(admin_id) user = self.state["users"].get(user_id) if not user: raise ValueError("Account not found.") if user.get("role") == "admin": raise ValueError("The developer account cannot be assigned a customer package.") plan = PLANS[plan_key] if plan_key in {"plus", "pro"} and not self._capacity_for_paid_plan_unlocked(user_id, plan_key): raise ValueError("There is not enough reservable server storage for this package.") now = utcnow() previous_plan = str(user.get("plan") or "free") user["plan"] = plan_key user["plan_started_at"] = iso(now) user["plan_expires_at"] = iso(now + timedelta(days=max(1, int(duration_days)))) if plan_key != "free" else None user["plan_expiry_notice_for"] = None user["delete_on_expiry"] = False user["pending_purchase_plan"] = None user["pending_purchase_expires_at"] = None removed = self._trim_user_unlocked(user_id, plan.quota_bytes) expiry_text = ( f" It expires on {parse_dt(user['plan_expires_at']).strftime('%d %b %Y · %H:%M UTC')}." if user.get("plan_expires_at") else " It does not have a paid-package expiration date." ) self._notify_user_unlocked( user_id, "Admin changed your package", f"Your package changed from {PLANS.get(previous_plan, PLANS['free']).name} to {plan.name}.{expiry_text}", "account", ) self._save_unlocked("Developer changed CloudVaultHub user package") self.backend.delete_many(removed, "Trim files after developer package change") return self.get_user(user_id) or {} def begin_purchase( self, user_id: str, plan_key: str, hold_hours: int = 24, ) -> dict[str, Any]: if plan_key not in {"plus", "pro"}: raise ValueError("Invalid paid package.") with self.lock: user = self.state["users"].get(user_id) if not user: raise ValueError("Account not found.") if user.get("role") == "admin": raise ValueError( "The developer account does not use paid packages." ) cfg = self.state["config"] if not cfg.get("system_enabled", True): raise ValueError("The system is temporarily paused.") if not cfg.get("purchases_enabled", True): raise ValueError( "Package purchases are currently paused." ) subscription_id = str( user.get("billing_subscription_id") or "" ) billing_status = str( user.get("billing_status") or "" ) if ( subscription_id and billing_status in { "active", "on_trial", "past_due", "paused", "cancelled", } ): raise ValueError( "This account already has a subscription. " "Use Billing Settings or the Customer Portal." ) if not self._capacity_for_paid_plan_unlocked( user_id, plan_key, ): raise ValueError( "The server cannot reserve this package while keeping " f"{human_bytes(self.settings.purchase_safety_bytes)} free." ) expires_at = utcnow() + timedelta( hours=max(1, min(int(hold_hours), 72)) ) user["pending_purchase_plan"] = plan_key user["pending_purchase_expires_at"] = iso(expires_at) self._save_unlocked( f"Reserve CloudVaultHub {plan_key} checkout capacity" ) return dict(user) def billing_info( self, user_id: str | None, ) -> dict[str, Any]: if not user_id: return {} self.cleanup(user_id) with self.lock: user = self.state["users"].get(user_id) if not user: return {} keys = [ "id", "username", "plan", "plan_expires_at", "pending_purchase_plan", "pending_purchase_expires_at", "billing_subscription_id", "billing_customer_id", "billing_plan", "billing_status", "billing_email", "billing_auto_renew", "billing_renews_at", "billing_ends_at", "billing_anchor", "billing_card_brand", "billing_card_last_four", "billing_payment_processor", "billing_urls", "billing_updated_at", ] return {key: user.get(key) for key in keys} def find_user_by_billing_email( self, email: str | None, ) -> dict[str, Any] | None: wanted = str(email or "").strip().casefold() if not wanted: return None with self.lock: user = next( ( account for account in self.state["users"].values() if str( account.get("billing_email") or "" ).strip().casefold() == wanted ), None, ) return dict(user) if user else None def apply_subscription_snapshot( self, user_id: str, plan_key: str, subscription_id: str, attributes: dict[str, Any], event_name: str = "subscription_updated", ) -> dict[str, Any]: if plan_key not in {"plus", "pro"}: raise ValueError("Unsupported subscription package.") removed: list[str] = [] with self.lock: user = self.state["users"].get(user_id) if not user: raise ValueError("Account not found.") if user.get("role") == "admin": raise ValueError( "The developer account cannot receive a subscription." ) previous_plan = str(user.get("plan") or "free") status = str( attributes.get("status") or "" ).strip().lower() cancelled = bool( attributes.get("cancelled") or status == "cancelled" ) renews_at = attributes.get("renews_at") ends_at = attributes.get("ends_at") urls = attributes.get("urls") if not isinstance(urls, dict): urls = {} user["billing_subscription_id"] = ( str(subscription_id or "") or user.get("billing_subscription_id") ) user["billing_customer_id"] = attributes.get( "customer_id", user.get("billing_customer_id"), ) user["billing_plan"] = plan_key user["billing_status"] = status or event_name user["billing_email"] = attributes.get( "user_email", user.get("billing_email"), ) user["billing_auto_renew"] = ( not cancelled and status not in {"expired", "unpaid"} ) user["billing_renews_at"] = renews_at user["billing_ends_at"] = ends_at user["billing_anchor"] = attributes.get("billing_anchor") user["billing_card_brand"] = attributes.get("card_brand") user["billing_card_last_four"] = attributes.get("card_last_four") user["billing_payment_processor"] = attributes.get( "payment_processor" ) if urls: user["billing_urls"] = urls user["billing_updated_at"] = iso() user["pending_purchase_plan"] = None user["pending_purchase_expires_at"] = None now = utcnow() if status == "expired": user["plan"] = "free" user["plan_started_at"] = iso(now) user["plan_expires_at"] = None user["plan_expiry_notice_for"] = None user["billing_auto_renew"] = False removed.extend( self._trim_user_unlocked( user_id, PLANS["free"].quota_bytes, ) ) else: user["plan"] = plan_key user["plan_started_at"] = ( user.get("plan_started_at") or iso(now) ) expiry_value = ( ends_at if cancelled and ends_at else renews_at ) expiry = parse_dt(expiry_value) if expiry is None: expiry = now + timedelta(days=28) user["plan_expires_at"] = iso(expiry) user["plan_expiry_notice_for"] = None user["delete_on_expiry"] = False current_plan = str(user.get("plan") or "free") if current_plan != previous_plan or status == "expired": expiry_text = ( f" It expires or renews on {parse_dt(user['plan_expires_at']).strftime('%d %b %Y · %H:%M UTC')}." if user.get("plan_expires_at") else "" ) self._notify_user_unlocked( user_id, "Subscription package updated", f"Your package changed from {PLANS.get(previous_plan, PLANS['free']).name} to {PLANS.get(current_plan, PLANS['free']).name}.{expiry_text}", "account", ) self._save_unlocked( f"Apply Lemon Squeezy event {event_name}" ) self.backend.delete_many( removed, "Trim files after subscription expiry", ) return self.get_user(user_id) or {} def used_bytes(self, user_id: str) -> int: with self.lock: return self._used_unlocked(user_id) def total_used_bytes(self) -> int: with self.lock: return self._used_unlocked() def total_reserved_bytes(self) -> int: with self.lock: return self._reserved_unlocked() def available_for_admin(self) -> int: with self.lock: return max(0, self.settings.app_storage_limit_bytes - self._used_unlocked()) def list_files(self, user_id: str) -> list[dict[str, Any]]: self.cleanup(user_id) with self.lock: files = [dict(f) for f in self.state["files"].values() if f.get("owner_id") == user_id] return sorted(files, key=lambda item: item.get("created_at", ""), reverse=True) def _unique_token_unlocked(self, length: int) -> str: existing = {f.get("share_token") for f in self.state["files"].values()} for _ in range(300): token = random_share_id(length) if token not in existing: return token raise RuntimeError("Could not create a unique sharing ID.") def add_file(self, user_id: str, metadata: dict[str, Any]) -> dict[str, Any]: with self.lock: user = self.state["users"].get(user_id) if not user: raise ValueError("Account not found.") cfg = self.state["config"] if user.get("role") != "admin" and not cfg.get("system_enabled", True): raise ValueError("The system is temporarily paused.") size = int(metadata["original_size"]) is_admin = str(user.get("role", "")).lower() == "admin" plan_key = str(user.get("plan", "free")).lower() plan = PLANS["admin"] if is_admin else PLANS.get(plan_key, PLANS["free"]) if is_admin: remaining = max(0, self.settings.app_storage_limit_bytes - self._used_unlocked()) if size > remaining: raise ValueError( f"Upload failed: developer storage is full. Available {human_bytes(remaining)}, file size {human_bytes(size)}." ) else: used_by_user = self._used_unlocked(user_id) remaining = max(0, plan.quota_bytes - used_by_user) if size > remaining: raise ValueError( f"Upload failed: your {plan.name} storage is full. Available {human_bytes(remaining)}, file size {human_bytes(size)}." ) server_remaining = max(0, self.settings.app_storage_limit_bytes - self._used_unlocked()) if size > server_remaining: raise ValueError( f"Upload failed: CloudVaultHub server storage is full. Available {human_bytes(server_remaining)}, file size {human_bytes(size)}." ) metadata = dict(metadata) metadata["share_token"] = self._unique_token_unlocked(plan.token_length) metadata["share_password_hash"] = None self.state["files"][metadata["id"]] = metadata self._save_unlocked("Save CloudVaultHub file metadata") return dict(metadata) def get_file_by_token(self, token: str) -> dict[str, Any] | None: self.cleanup() with self.lock: for item in self.state["files"].values(): if item.get("share_token") == token: return dict(item) return None def delete_file(self, user_id: str, file_id: str) -> bool: remote = None with self.lock: item = self.state["files"].get(file_id) if not item or item.get("owner_id") != user_id: return False remote = item.get("remote_path") del self.state["files"][file_id] self.state["events"] = [e for e in self.state["events"] if e.get("file_id") != file_id] self._save_unlocked("Delete CloudVaultHub file") self.backend.delete_many([remote], "Delete CloudVaultHub file object") return True def _delete_user_unlocked(self, user_id: str) -> list[str]: user = self.state["users"].get(user_id) if not user or user.get("role") == "admin": return [] remotes: list[str] = [] for file_id, item in list(self.state["files"].items()): if item.get("owner_id") == user_id: remotes.append(item.get("remote_path", "")) del self.state["files"][file_id] self.state["events"] = [e for e in self.state["events"] if e.get("owner_id") != user_id] self.state["chat"] = [m for m in self.state["chat"] if m.get("sender_id") != user_id] self.state["support_tickets"] = [t for t in self.state["support_tickets"] if t.get("user_id") != user_id] del self.state["users"][user_id] return remotes def delete_account(self, user_id: str) -> bool: with self.lock: if user_id not in self.state["users"] or self._is_admin_unlocked(user_id): return False remotes = self._delete_user_unlocked(user_id) self._save_unlocked("Delete CloudVaultHub account") self.backend.delete_many(remotes, "Delete account objects") return True def record_download(self, file_id: str, country: str, device: str) -> None: with self.lock: item = self.state["files"].get(file_id) if not item: return now = iso() item["download_count"] = int(item.get("download_count", 0)) + 1 item["last_download_at"] = now self.state["events"].append({ "id": uuid.uuid4().hex, "file_id": file_id, "owner_id": item.get("owner_id"), "created_at": now, "country": (country or "Unknown")[:100], "device": (device or "Unknown")[:60], }) cutoff = utcnow() - timedelta(days=400) self.state["events"] = [ e for e in self.state["events"] if (parse_dt(e.get("created_at")) or utcnow()) >= cutoff ][-50000:] self._save_unlocked("Record CloudVaultHub download") def analytics(self, user_id: str, requested_days: int) -> dict[str, Any]: user = self.get_user(user_id) if not user: return {"events": [], "total": 0, "unique_files": 0, "countries": {}, "devices": {}, "days": 0, "delay": 0} is_admin = str(user.get("role", "")).lower() == "admin" plan_key = str(user.get("plan", "free")).lower() plan = PLANS["admin"] if is_admin else PLANS.get(plan_key, PLANS["free"]) days = max(1, min(int(requested_days), plan.analytics_history_days)) now = utcnow() start = now - timedelta(days=days) visible_until = now - timedelta(hours=plan.analytics_delay_hours) with self.lock: events = [] for event in self.state["events"]: if event.get("owner_id") != user_id: continue timestamp = parse_dt(event.get("created_at")) if timestamp and start <= timestamp <= visible_until: events.append(dict(event)) countries: dict[str, int] = {} devices: dict[str, int] = {} for event in events: country = event.get("country", "Unknown") device = event.get("device", "Unknown") countries[country] = countries.get(country, 0) + 1 devices[device] = devices.get(device, 0) + 1 return { "events": sorted(events, key=lambda e: e.get("created_at", ""), reverse=True), "total": len(events), "unique_files": len({e.get("file_id") for e in events}), "countries": countries, "devices": devices, "days": days, "delay": plan.analytics_delay_hours, } def _trim_user_unlocked(self, user_id: str, max_bytes: int) -> list[str]: owned = [f for f in self.state["files"].values() if f.get("owner_id") == user_id] owned.sort(key=lambda f: f.get("created_at", "")) used = sum(int(f.get("original_size", 0)) for f in owned) remotes: list[str] = [] for item in owned: if used <= max_bytes: break used -= int(item.get("original_size", 0)) remotes.append(item.get("remote_path", "")) self.state["files"].pop(item["id"], None) self.state["events"] = [e for e in self.state["events"] if e.get("file_id") != item["id"]] return remotes def _notify_user_unlocked( self, user_id: str, title: str, message: str, kind: str = "system", ) -> None: """Append a targeted notification while the store lock is already held.""" if user_id not in self.state.get("users", {}): return notifications = self.state.setdefault("notifications", []) notifications.append({ "id": uuid.uuid4().hex, "title": str(title or "CloudVaultHub update")[:120], "message": str(message or "")[:3000], "audience": [], "kind": str(kind or "system"), "user_id": user_id, "recipient_ids": [user_id], "created_at": iso(), }) if len(notifications) > 2000: del notifications[:-2000] def cleanup(self, only_user_id: str | None = None) -> None: remotes: list[str] = [] changed = False now = utcnow() with self.lock: user_ids = [only_user_id] if only_user_id else list(self.state["users"].keys()) for uid in user_ids: user = self.state["users"].get(uid) if not user or user.get("role") == "admin": continue ban_until = parse_dt(user.get("ban_until")) if user.get("is_banned") and ban_until and ban_until <= now: user["is_banned"] = False user["ban_until"] = None changed = True pending_expires = parse_dt( user.get("pending_purchase_expires_at") ) if ( user.get("pending_purchase_plan") and pending_expires and pending_expires <= now ): user["pending_purchase_plan"] = None user["pending_purchase_expires_at"] = None changed = True expires = parse_dt(user.get("plan_expires_at")) if ( user.get("plan") in {"plus", "pro"} and expires and now < expires <= now + timedelta(days=7) ): expiry_marker = iso(expires) if user.get("plan_expiry_notice_for") != expiry_marker: remaining_days = max(1, (expires.date() - now.date()).days) self._notify_user_unlocked( uid, "Your package expires soon", f"Your {PLANS.get(str(user.get('plan')), PLANS['free']).name} package has {remaining_days} day(s) remaining and expires on {expires.strftime('%d %b %Y · %H:%M UTC')}.", "account", ) user["plan_expiry_notice_for"] = expiry_marker changed = True if user.get("plan") != "free" and expires and expires <= now: should_delete = bool(user.get("delete_on_expiry") or self.state["config"].get("delete_paid_on_expiry", False)) if should_delete: remotes.extend(self._delete_user_unlocked(uid)) else: user["plan"] = "free" user["plan_started_at"] = iso(now) user["plan_expires_at"] = None user["plan_expiry_notice_for"] = None user["delete_on_expiry"] = False user["billing_status"] = "expired" user["billing_auto_renew"] = False user["billing_ends_at"] = iso(now) user["pending_purchase_plan"] = None user["pending_purchase_expires_at"] = None remotes.extend(self._trim_user_unlocked(uid, PLANS["free"].quota_bytes)) self._notify_user_unlocked( uid, "Your package expired", "Your paid package ended and the account returned to Free. Files above the Free storage limit may have been removed.", "account", ) changed = True for file_id, item in list(self.state["files"].items()): if only_user_id and item.get("owner_id") != only_user_id: continue user = self.state["users"].get(item.get("owner_id")) if not user: remotes.append(item.get("remote_path", "")) del self.state["files"][file_id] changed = True continue if user.get("role") == "admin": continue uploaded_plan = PLANS.get(item.get("uploaded_under_plan", user.get("plan", "free")), PLANS["free"]) created = parse_dt(item.get("created_at")) or now age = now - created delete = False if uploaded_plan.delete_after_age_even_if_downloaded and age >= timedelta(days=uploaded_plan.retention_days): delete = True elif not uploaded_plan.delete_after_age_even_if_downloaded and int(item.get("download_count", 0)) == 0 and age >= timedelta(days=uploaded_plan.retention_days): delete = True if delete: display_name = str(item.get("display_name") or "A file") retention_days = max(0, int(uploaded_plan.retention_days)) remotes.append(item.get("remote_path", "")) del self.state["files"][file_id] self.state["events"] = [e for e in self.state["events"] if e.get("file_id") != file_id] self._notify_user_unlocked( str(item.get("owner_id") or ""), "File removed automatically", f"{display_name} was deleted automatically after {retention_days} day(s) under the {uploaded_plan.name} retention rule.", "file_deleted", ) changed = True if changed: self._save_unlocked("Automatic CloudVaultHub cleanup") self.backend.delete_many(remotes, "Delete expired CloudVaultHub objects") # ---------------- Developer administration ---------------- def admin_overview(self, admin_id: str) -> dict[str, Any]: self.cleanup() with self.lock: self._assert_admin_unlocked(admin_id) counts = {"free": 0, "plus": 0, "pro": 0} used_by_plan = {"free": 0, "plus": 0, "pro": 0, "admin": 0} for uid, user in self.state["users"].items(): is_admin = str(user.get("role", "")).strip().lower() == "admin" plan_key = str(user.get("plan") or "free").strip().lower() if not is_admin and plan_key not in counts: plan_key = "free" if not is_admin: counts[plan_key] += 1 bucket = "admin" if is_admin else plan_key used_by_plan[bucket] = used_by_plan.get(bucket, 0) + self._used_unlocked(uid) total_used = self._used_unlocked() package_used = self._package_used_unlocked() system_used = self._admin_used_unlocked() return { "counts": counts, "registered": sum(counts.values()), "used_by_plan": used_by_plan, "total_used": total_used, "remaining": max(0, self.settings.total_storage_capacity_bytes - total_used), "limit": self.settings.app_storage_limit_bytes, "total_capacity": self.settings.total_storage_capacity_bytes, "system_reserve": self.settings.system_storage_reserve_bytes, "system_used": system_used, "system_remaining": max(0, self.settings.system_storage_reserve_bytes - system_used), "package_capacity": self.settings.app_storage_limit_bytes, "package_used": package_used, "package_remaining": max(0, self.settings.app_storage_limit_bytes - package_used), "storage_pool_remaining": max(0, self.settings.app_storage_limit_bytes - total_used), "reserved": self._reserved_unlocked(), "config": dict(self.state["config"]), "chat_messages": len(self.state["chat"]), "support_tickets": len(self.state["support_tickets"]), "support_open": sum(1 for t in self.state["support_tickets"] if t.get("status") != "answered"), } def ban_user(self, admin_id: str, target_id: str, hours: float | None = None) -> None: with self.lock: self._assert_admin_unlocked(admin_id) target = self.state["users"].get(target_id) if not target or target.get("role") == "admin": raise ValueError("The selected account cannot be banned.") target["is_banned"] = True target["ban_until"] = iso(utcnow() + timedelta(hours=float(hours))) if hours and float(hours) > 0 else None self._save_unlocked("Ban CloudVaultHub account") def unban_user(self, admin_id: str, target_id: str) -> None: with self.lock: self._assert_admin_unlocked(admin_id) target = self.state["users"].get(target_id) if not target or target.get("role") == "admin": raise ValueError("Account not found.") target["is_banned"] = False target["ban_until"] = None self._save_unlocked("Unban CloudVaultHub account") def admin_delete_user(self, admin_id: str, target_id: str) -> None: with self.lock: self._assert_admin_unlocked(admin_id) remotes = self._delete_user_unlocked(target_id) if not remotes and target_id not in self.state["users"]: pass self._save_unlocked("Developer deleted account") self.backend.delete_many(remotes, "Developer delete account objects") def factory_reset(self, admin_id: str, confirmation: str) -> None: """Delete all user data and restore the service to first-install state.""" if str(confirmation or "").strip() != "RESET CLOUDVAULTHUB": raise ValueError("Type RESET CLOUDVAULTHUB exactly.") with self.lock: self._assert_admin_unlocked(admin_id) # Delete every persisted object except the state file itself. remote_paths: list[str] = [] if self.settings.local_mode: root = self.settings.local_root if root.exists(): for item in root.rglob("*"): if item.is_file() and item.relative_to(root).as_posix() != STATE_REMOTE_PATH: remote_paths.append(item.relative_to(root).as_posix()) else: try: assert self.backend.api and self.settings.dataset_repo remote_paths = [ name for name in self.backend.api.list_repo_files( repo_id=self.settings.dataset_repo, repo_type="dataset" ) if name != STATE_REMOTE_PATH and name != ".gitattributes" ] except Exception: # Fall back to paths referenced by the current state. for item in self.state.get("files", {}).values(): if isinstance(item, dict) and item.get("remote_path"): remote_paths.append(str(item["remote_path"])) for project in self.state.get("websites", {}).values(): for item in (project.get("entries", {}) if isinstance(project, dict) else {}).values(): if isinstance(item, dict) and item.get("remote_path"): remote_paths.append(str(item["remote_path"])) self.backend.delete_many(remote_paths, "Factory reset CloudVaultHub storage") self.state = { "version": 4, "users": {}, "files": {}, "events": [], "chat": [], "chat_rooms": {}, "support_tickets": [], "config": {}, "websites": {}, "notifications": [], "notification_reads": {}, "plan_features": {}, } self._normalize_unlocked() self._ensure_admin_unlocked() self._save_unlocked("Factory reset CloudVaultHub") def delete_all_free(self, admin_id: str) -> int: with self.lock: self._assert_admin_unlocked(admin_id) targets = [uid for uid, u in self.state["users"].items() if u.get("role") != "admin" and u.get("plan") == "free"] remotes: list[str] = [] for uid in targets: remotes.extend(self._delete_user_unlocked(uid)) self._save_unlocked("Delete all Free accounts") self.backend.delete_many(remotes, "Delete all Free account objects") return len(targets) def delete_all_users(self, admin_id: str) -> int: with self.lock: self._assert_admin_unlocked(admin_id) targets = [uid for uid, u in self.state["users"].items() if u.get("role") != "admin"] remotes: list[str] = [] for uid in targets: remotes.extend(self._delete_user_unlocked(uid)) self._save_unlocked("Delete all CloudVaultHub user accounts") self.backend.delete_many(remotes, "Delete all user objects") return len(targets) def set_config(self, admin_id: str, **updates: Any) -> dict[str, Any]: allowed = { "system_enabled", "chat_enabled", "plus_chat_enabled", "pro_chat_enabled", "support_enabled", "support_force_open", "support_auto_paused", "purchases_enabled", "registrations_enabled", "delete_paid_on_expiry", "shutdown_mode", } with self.lock: self._assert_admin_unlocked(admin_id) for key, value in updates.items(): if key in allowed: self.state["config"][key] = bool(value) self._save_unlocked("Update CloudVaultHub developer controls") return dict(self.state["config"]) def begin_shutdown(self, admin_id: str) -> int: with self.lock: self._assert_admin_unlocked(admin_id) self.state["config"].update({ "system_enabled": False, "chat_enabled": False, "plus_chat_enabled": False, "pro_chat_enabled": False, "purchases_enabled": False, "registrations_enabled": False, "delete_paid_on_expiry": True, "shutdown_mode": True, }) free_ids = [uid for uid, u in self.state["users"].items() if u.get("role") != "admin" and u.get("plan") == "free"] remotes: list[str] = [] for uid in free_ids: remotes.extend(self._delete_user_unlocked(uid)) for user in self.state["users"].values(): if user.get("role") != "admin" and user.get("plan") in {"plus", "pro"}: user["delete_on_expiry"] = True self._save_unlocked("Begin CloudVaultHub shutdown mode") self.backend.delete_many(remotes, "Shutdown mode delete Free objects") return len(free_ids) def cancel_shutdown(self, admin_id: str) -> None: with self.lock: self._assert_admin_unlocked(admin_id) self.state["config"].update({ "system_enabled": True, "chat_enabled": True, "plus_chat_enabled": True, "pro_chat_enabled": True, "purchases_enabled": True, "registrations_enabled": True, "delete_paid_on_expiry": False, "shutdown_mode": False, }) for user in self.state["users"].values(): if user.get("role") != "admin": user["delete_on_expiry"] = False self._save_unlocked("Cancel CloudVaultHub shutdown mode") # ---------------- Admin support tickets ---------------- def _require_support_user_unlocked(self, user_id: str) -> dict[str, Any]: user = self.state["users"].get(user_id) if not user: raise PermissionError("Login required.") if user.get("role") == "admin": return user if user.get("plan") not in {"plus", "pro"}: raise PermissionError("Admin Support is available only for Plus and Pro accounts.") return user def support_runtime(self) -> dict[str, Any]: with self.lock: cfg = self.state["config"] enabled = bool(cfg.get("support_enabled", True)) forced = bool(cfg.get("support_force_open", False)) auto_paused = bool(cfg.get("support_auto_paused", False)) return {"enabled": bool(enabled and (forced or not auto_paused)), "manual_enabled": enabled, "force_open": forced, "auto_paused": auto_paused} def set_support_memory_state(self, paused: bool) -> None: with self.lock: paused = bool(paused) if self.state["config"].get("support_auto_paused", False) != paused: self.state["config"]["support_auto_paused"] = paused self._save_unlocked("Update CloudVaultHub support memory protection") def submit_support_ticket(self, user_id: str, subject: str, message: str) -> dict[str, Any]: subject = (subject or "").strip(); message = (message or "").strip() if not subject: raise ValueError("Subject is required.") if len(subject) > 200: raise ValueError("Subject may contain at most 200 characters.") if not message: raise ValueError("Message is required.") if len(message) > 5000: raise ValueError("Message may contain at most 5000 characters.") with self.lock: user = self._require_support_user_unlocked(user_id) cfg = self.state["config"] if not cfg.get("support_enabled", True): raise PermissionError("Admin Support is currently paused.") if cfg.get("support_auto_paused", False) and not cfg.get("support_force_open", False): raise PermissionError("Admin Support is temporarily paused by server protection.") today = utcnow().date() for ticket in self.state["support_tickets"]: created = parse_dt(ticket.get("created_at")) if ticket.get("user_id") == user_id and created and created.date() == today: raise ValueError("You can send only one Admin Support message per day.") ticket = {"id": uuid.uuid4().hex, "user_id": user_id, "username": user.get("username", "User"), "plan": user.get("plan", "plus"), "subject": subject, "message": message, "created_at": iso(), "status": "open", "reply": "", "replied_at": None, "read_by_user": False} self.state["support_tickets"].append(ticket) self._save_unlocked("Create CloudVaultHub support ticket") return dict(ticket) def user_support_tickets(self, user_id: str) -> list[dict[str, Any]]: with self.lock: self._require_support_user_unlocked(user_id) return sorted([dict(t) for t in self.state["support_tickets"] if t.get("user_id") == user_id], key=lambda t: t.get("created_at", ""), reverse=True) def admin_support_tickets(self, admin_id: str) -> list[dict[str, Any]]: with self.lock: self._assert_admin_unlocked(admin_id) priority = {"pro": 0, "plus": 1} tickets = [dict(t) for t in self.state["support_tickets"]] tickets.sort(key=lambda t: (priority.get(str(t.get("plan", "plus")), 2), -(parse_dt(t.get("created_at")).timestamp() if parse_dt(t.get("created_at")) else 0))) return tickets def get_support_ticket(self, actor_id: str, ticket_id: str) -> dict[str, Any] | None: with self.lock: actor = self.state["users"].get(actor_id) if not actor: return None for ticket in self.state["support_tickets"]: if ticket.get("id") == ticket_id and (actor.get("role") == "admin" or ticket.get("user_id") == actor_id): return dict(ticket) return None def reply_support_ticket(self, admin_id: str, ticket_id: str, reply: str) -> dict[str, Any]: reply = (reply or "").strip() if not reply: raise ValueError("Reply is required.") if len(reply) > 5000: raise ValueError("Reply may contain at most 5000 characters.") with self.lock: self._assert_admin_unlocked(admin_id) for ticket in self.state["support_tickets"]: if ticket.get("id") == ticket_id: ticket.update({"reply": reply, "status": "answered", "replied_at": iso(), "read_by_user": False}) self._save_unlocked("Reply to CloudVaultHub support ticket") return dict(ticket) raise ValueError("Support message not found.") def mark_support_read(self, user_id: str, ticket_id: str) -> dict[str, Any]: with self.lock: for ticket in self.state["support_tickets"]: if ticket.get("id") == ticket_id and ticket.get("user_id") == user_id: ticket["read_by_user"] = True self._save_unlocked("Mark CloudVaultHub support reply as read") return dict(ticket) raise ValueError("Support message not found.") # ---------------- Legacy global chat (disabled) ---------------- def _chat_eligible_unlocked(self, user: dict[str, Any] | None) -> bool: if not user: return False if user.get("role") == "admin": return True return PLANS.get(user.get("plan", "free"), PLANS["free"]).chat_access def _room_sort_key_unlocked(self, room_id: str) -> tuple[int, str]: try: return int(room_id.rsplit("-", 1)[-1]), room_id except (TypeError, ValueError): return 10**9, room_id def _ensure_room_unlocked(self, room_id: str) -> dict[str, Any]: room = self.state["chat_rooms"].get(room_id) if room: return room number, _ = self._room_sort_key_unlocked(room_id) if number >= 10**9: number = max([self._room_sort_key_unlocked(rid)[0] for rid in self.state["chat_rooms"]] or [0]) + 1 room_id = f"global-{number}" room = {"id": room_id, "name": f"Global {number}", "paused": False, "created_at": iso()} self.state["chat_rooms"][room_id] = room return room def _room_member_count_unlocked(self, room_id: str, exclude_user_id: str | None = None) -> int: return sum( 1 for uid, user in self.state["users"].items() if uid != exclude_user_id and self._chat_eligible_unlocked(user) and user.get("chat_room_id", "global-1") == room_id ) def _new_room_unlocked(self) -> dict[str, Any]: numbers = [self._room_sort_key_unlocked(room_id)[0] for room_id in self.state["chat_rooms"]] number = max(numbers or [0]) + 1 room_id = f"global-{number}" room = {"id": room_id, "name": f"Global {number}", "paused": False, "created_at": iso()} self.state["chat_rooms"][room_id] = room return room def _assign_room_unlocked(self, user_id: str, force: bool = False) -> str: user = self.state["users"].get(user_id) if not self._chat_eligible_unlocked(user): return "" cfg = self.state["config"] if cfg.get("chat_mode", "single") == "single": user["chat_room_id"] = "global-1" return "global-1" current = user.get("chat_room_id") if not force and current in self.state["chat_rooms"]: return str(current) capacity = int(cfg.get("chat_room_capacity", 200)) available = [ rid for rid, room in self.state["chat_rooms"].items() if not room.get("paused") and self._room_member_count_unlocked(rid, exclude_user_id=user_id) < capacity ] if not available: available = [self._new_room_unlocked()["id"]] chosen = secrets.choice(sorted(available, key=self._room_sort_key_unlocked)) user["chat_room_id"] = chosen return chosen def chat_room_state(self, user_id: str) -> dict[str, Any]: with self.lock: user = self.state["users"].get(user_id) if not self._chat_eligible_unlocked(user): return {"mode": "single", "capacity": 0, "current": None, "can_choose": False, "rooms": []} user["chat_room_id"] = "global-1" room = self.state["chat_rooms"].setdefault( "global-1", {"id": "global-1", "name": "Global Chat", "paused": False, "created_at": iso()}, ) room["name"] = "Global Chat" return { "mode": "single", "capacity": 0, "current": "global-1", "can_choose": False, "rooms": [{ "id": "global-1", "name": "Global Chat", "paused": bool(room.get("paused", False)), "members": self._room_member_count_unlocked("global-1"), "messages": len(self.state["chat"]), }], } def change_chat_room(self, user_id: str, room_id: str) -> str: with self.lock: user = self.state["users"].get(user_id) if not self._chat_eligible_unlocked(user): raise PermissionError("Global Chat is available only to Plus and Pro accounts.") if self.state["config"].get("chat_mode", "single") != "multi": user["chat_room_id"] = "global-1" return "global-1" if user.get("role") != "admin" and user.get("plan") != "pro": raise PermissionError("Plus accounts are assigned to a room automatically.") room = self.state["chat_rooms"].get(room_id) if not room: raise ValueError("The selected chat room does not exist.") if room.get("paused") and user.get("role") != "admin": raise ValueError("The selected chat room is paused.") current = user.get("chat_room_id") capacity = int(self.state["config"].get("chat_room_capacity", 200)) if current != room_id and user.get("role") != "admin" and self._room_member_count_unlocked(room_id) >= capacity: raise ValueError("The selected chat room is full.") user["chat_room_id"] = room_id self._save_unlocked("Change CloudVaultHub chat room") return room_id def set_chat_configuration(self, admin_id: str, mode: str, capacity: int) -> dict[str, Any]: with self.lock: self._assert_admin_unlocked(admin_id) self.state["config"]["chat_mode"] = "single" self.state["config"]["chat_room_capacity"] = 0 room = self.state["chat_rooms"].setdefault( "global-1", {"id": "global-1", "name": "Global Chat", "paused": False, "created_at": iso()}, ) room["name"] = "Global Chat" self.state["chat_rooms"] = {"global-1": room} for user in self.state["users"].values(): if self._chat_eligible_unlocked(user): user["chat_room_id"] = "global-1" for message in self.state["chat"]: message["room_id"] = "global-1" self.state["chat"] = sorted(self.state["chat"], key=lambda m: m.get("created_at", ""))[-400:] self._save_unlocked("Use one unlimited Global Chat room") return dict(self.state["config"]) def toggle_chat_room(self, admin_id: str, room_id: str) -> bool: with self.lock: self._assert_admin_unlocked(admin_id) room = self.state["chat_rooms"].get(room_id) if not room: raise ValueError("Chat room not found.") room["paused"] = not bool(room.get("paused")) self._save_unlocked("Toggle CloudVaultHub chat room") return bool(room["paused"]) def clear_chat_room(self, admin_id: str, room_id: str) -> int: with self.lock: self._assert_admin_unlocked(admin_id) if room_id not in self.state["chat_rooms"]: raise ValueError("Chat room not found.") before = len(self.state["chat"]) self.state["chat"] = [m for m in self.state["chat"] if m.get("room_id", "global-1") != room_id] removed = before - len(self.state["chat"]) self._save_unlocked("Clear CloudVaultHub chat room") return removed def send_chat(self, user_id: str, text: str, room_id: str | None = None) -> dict[str, Any]: message = (text or "").strip() if not message: raise ValueError("Write a message first.") if len(message) > 1000: raise ValueError("Messages may contain at most 1000 characters.") with self.lock: user = self.state["users"].get(user_id) if not user: raise ValueError("Please log in again.") plan = PLANS.get(user.get("plan", "free"), PLANS["free"]) if not self._chat_eligible_unlocked(user): raise PermissionError("Global Chat is available only to Plus and Pro accounts.") if user.get("is_banned"): raise PermissionError("This account is banned.") cfg = self.state["config"] if not cfg.get("system_enabled", True): raise PermissionError("The system is paused.") if not cfg.get("chat_enabled", True): raise PermissionError("Global Chat is paused by the developer.") if user.get("role") != "admin" and user.get("plan") == "plus" and not cfg.get("plus_chat_enabled", True): raise PermissionError("Global Chat is paused for Plus accounts.") if user.get("role") != "admin" and user.get("plan") == "pro" and not cfg.get("pro_chat_enabled", True): raise PermissionError("Global Chat is paused for Pro accounts.") user["chat_room_id"] = "global-1" assigned_room = "global-1" room = self.state["chat_rooms"].get(assigned_room) if room and room.get("paused") and user.get("role") != "admin": raise PermissionError("This chat room is paused by the developer.") now = utcnow() last = parse_dt(user.get("last_chat_at")) if last and (now - last).total_seconds() < 3: wait = max(1, 3 - int((now - last).total_seconds())) raise ValueError(f"Wait {wait} second(s) before sending another message.") record = { "id": uuid.uuid4().hex, "room_id": assigned_room, "sender_id": user_id, "sender_name": "admin" if user.get("role") == "admin" else user.get("username", "User"), "sender_plan": "admin" if user.get("role") == "admin" else user.get("plan", "free"), "text": message, "created_at": iso(now), "visible_at": iso(now + timedelta(seconds=2)), } user["last_chat_at"] = iso(now) self.state["chat"].append(record) self.state["chat"] = sorted(self.state["chat"], key=lambda m: m.get("created_at", ""))[-400:] self._save_unlocked("Send CloudVaultHub Global Chat message") return dict(record) def chat_messages(self, user_id: str, room_id: str | None = None) -> list[dict[str, Any]]: with self.lock: user = self.state["users"].get(user_id) if not self._chat_eligible_unlocked(user): return [] assigned = self._assign_room_unlocked(user_id) now = utcnow() limit = 400 if user.get("role") == "admin" or user.get("plan") == "pro" else 200 return [ dict(m) for m in self.state["chat"] if m.get("room_id", "global-1") == assigned and (parse_dt(m.get("visible_at")) or now) <= now ][-limit:] def clear_chat(self, admin_id: str) -> None: with self.lock: self._assert_admin_unlocked(admin_id) self.state["chat"] = [] self._save_unlocked("Clear CloudVaultHub Global Chat") STORE = StateStore(BACKEND, SETTINGS) def derive_file_key(user_id: str, file_id: str, algorithm: str) -> bytes: key_material = hashlib.sha256(SETTINGS.master_key.encode("utf-8")).digest() return hmac.new(key_material, f"{user_id}:{file_id}:{algorithm}".encode("utf-8"), hashlib.sha256).digest() def encrypt_file(source: Path, destination: Path, user_id: str, file_id: str, algorithm: str) -> int: if algorithm == "none": shutil.copy2(source, destination) return destination.stat().st_size algorithm_id = 1 if algorithm == "aes256-gcm" else 2 key = derive_file_key(user_id, file_id, algorithm) cipher = AESGCM(key) if algorithm_id == 1 else ChaCha20Poly1305(key) with source.open("rb") as fin, destination.open("wb") as fout: fout.write(MAGIC + bytes([algorithm_id]) + struct.pack(">I", CHUNK_SIZE)) index = 0 while True: chunk = fin.read(CHUNK_SIZE) if not chunk: break nonce = os.urandom(12) encrypted = cipher.encrypt(nonce, chunk, struct.pack(">Q", index)) fout.write(nonce) fout.write(struct.pack(">I", len(encrypted))) fout.write(encrypted) index += 1 return destination.stat().st_size def decrypt_file(source: Path, destination: Path, user_id: str, file_id: str, algorithm: str) -> None: if algorithm == "none": shutil.copy2(source, destination) return with source.open("rb") as fin, destination.open("wb") as fout: magic = fin.read(len(MAGIC)) if magic not in {MAGIC, b"CVLT2"}: raise ValueError("The encrypted file header is invalid.") algorithm_id_raw = fin.read(1) if not algorithm_id_raw: raise ValueError("The encrypted file is incomplete.") algorithm_id = algorithm_id_raw[0] chunk_raw = fin.read(4) if len(chunk_raw) != 4: raise ValueError("The encrypted file is incomplete.") _chunk_size = struct.unpack(">I", chunk_raw)[0] expected = 1 if algorithm == "aes256-gcm" else 2 if algorithm_id != expected: raise ValueError("The encryption mode does not match the file metadata.") key = derive_file_key(user_id, file_id, algorithm) cipher = AESGCM(key) if algorithm_id == 1 else ChaCha20Poly1305(key) index = 0 while True: nonce = fin.read(12) if not nonce: break if len(nonce) != 12: raise ValueError("The encrypted file is incomplete.") length_raw = fin.read(4) if len(length_raw) != 4: raise ValueError("The encrypted file is incomplete.") length = struct.unpack(">I", length_raw)[0] encrypted = fin.read(length) if len(encrypted) != length: raise ValueError("The encrypted file is incomplete.") fout.write(cipher.decrypt(nonce, encrypted, struct.pack(">Q", index))) index += 1 def zip_directory_files(paths: list[str], destination: Path) -> tuple[int, str]: valid = [Path(p) for p in paths if p and Path(p).is_file()] if not valid: raise ValueError("Choose a folder first.") total = 0 with zipfile.ZipFile(destination, "w", compression=zipfile.ZIP_DEFLATED, allowZip64=True) as archive: used_names: set[str] = set() for index, path in enumerate(valid): name = safe_filename(path.name) if name in used_names: name = f"{index + 1}-{name}" used_names.add(name) archive.write(path, arcname=name) total += path.stat().st_size return total, "uploaded-folder.zip" def estimate_upload_seconds(size_bytes: int, plan_key: str) -> int: # Approximate end-to-end speed. Real time depends on network and Hugging Face load. mb_per_second = {"free": 2.0, "plus": 5.0, "pro": 8.0, "admin": 12.0}.get(plan_key, 2.0) processing_overhead = {"free": 3, "plus": 5, "pro": 6, "admin": 3}.get(plan_key, 4) return max(2, int(size_bytes / MB / mb_per_second) + processing_overhead) def format_eta(seconds: int) -> str: if seconds < 60: return f"about {seconds} seconds" minutes = max(1, round(seconds / 60)) return f"about {minutes} minute{'s' if minutes != 1 else ''}" def prepare_upload(source_path: str, folder_paths: list[str] | None, user_id: str, progress=None) -> dict[str, Any]: user = STORE.get_user(user_id) if not user: raise ValueError("Please log in again.") cfg = STORE.config() if user.get("role") != "admin" and not cfg.get("system_enabled", True): raise ValueError("The system is temporarily paused.") is_admin = str(user.get("role", "")).lower() == "admin" plan_key = str(user.get("plan", "free")).lower() plan = PLANS["admin"] if is_admin else PLANS.get(plan_key, PLANS["free"]) file_id = uuid.uuid4().hex raw_temp: Path | None = None encrypted_temp: Path | None = None try: if folder_paths: if not plan.folder_upload: raise ValueError("Folder upload is not available on the Free package.") raw_temp = SETTINGS.temp_dir / f"folder-{file_id}.zip" original_size, display_name = zip_directory_files(folder_paths, raw_temp) content_type = "application/zip" is_folder = True else: if not source_path: raise ValueError("Choose a file or folder first.") source = Path(source_path) if not source.is_file(): raise ValueError("The selected file could not be read.") raw_temp = SETTINGS.temp_dir / f"raw-{file_id}-{safe_filename(source.name)}" shutil.copy2(source, raw_temp) original_size = raw_temp.stat().st_size display_name = safe_filename(source.name) content_type = mimetypes.guess_type(display_name)[0] or "application/octet-stream" is_folder = False if original_size <= 0: raise ValueError("Empty files cannot be uploaded.") if is_admin: remaining = STORE.available_for_admin() if original_size > remaining: raise ValueError( f"Upload failed: developer storage is full. Available {human_bytes(remaining)}, file size {human_bytes(original_size)}." ) else: remaining = max(0, plan.quota_bytes - STORE.used_bytes(user_id)) if original_size > remaining: raise ValueError( f"Upload failed: your {plan.name} storage is full. Available {human_bytes(remaining)}, file size {human_bytes(original_size)}." ) estimated_seconds = estimate_upload_seconds(original_size, plan.key) estimated_text = format_eta(estimated_seconds) if progress: progress(0.12, desc=f"Estimated completion: {estimated_text}") progress(0.25, desc=f"Preparing file · {estimated_text}") upload_path = raw_temp stored_size = original_size if plan.encryption != "none": encrypted_temp = SETTINGS.temp_dir / f"encrypted-{file_id}.bin" stored_size = encrypt_file(raw_temp, encrypted_temp, user_id, file_id, plan.encryption) upload_path = encrypted_temp if progress: progress(0.60, desc="Encrypting file") remote_path = f"objects/{user_id}/{file_id}.blob" if progress: progress(0.78, desc="Saving to persistent storage") BACKEND.upload(upload_path, remote_path, "Upload CloudVaultHub user file") metadata = { "id": file_id, "owner_id": user_id, "display_name": display_name, "remote_path": remote_path, "original_size": original_size, "stored_size": stored_size, "is_folder": is_folder, "content_type": content_type, "encryption": plan.encryption, "share_password_hash": None, "download_count": 0, "last_download_at": None, "created_at": iso(), "uploaded_under_plan": plan.key, "estimated_upload_seconds": estimated_seconds, } try: saved = STORE.add_file(user_id, metadata) except Exception: BACKEND.delete_many([remote_path], "Rollback incomplete CloudVaultHub upload") raise if progress: progress(1.0, desc="Upload completed") return saved finally: if raw_temp: raw_temp.unlink(missing_ok=True) if encrypted_temp: encrypted_temp.unlink(missing_ok=True) def prepare_generated_folder_archive( archive_path: str | Path, display_name: str, user_id: str, progress=None, ) -> dict[str, Any]: user = STORE.get_user(user_id) if not user: raise ValueError("Please log in again.") cfg = STORE.config() if user.get("role") != "admin" and not cfg.get("system_enabled", True): raise ValueError("The system is temporarily paused.") if user.get("role") != "admin" and user.get("plan") != "pro": raise PermissionError("Bu özellik yalnızca Pro hesaplarda kullanılabilir.") source = Path(archive_path) if not source.is_file() or source.suffix.lower() != ".zip": raise ValueError("Hazırlanan web sitesi klasörü bulunamadı.") original_size = source.stat().st_size if original_size <= 0: raise ValueError("Boş klasör buluta eklenemez.") is_admin = str(user.get("role", "")).lower() == "admin" plan_key = str(user.get("plan", "free")).lower() plan = PLANS["admin"] if is_admin else PLANS.get(plan_key, PLANS["free"]) if user.get("role") == "admin": if original_size > STORE.available_for_admin(): raise ValueError("The remaining unreserved server storage is not enough for this upload.") elif STORE.used_bytes(user_id) + original_size > plan.quota_bytes: raise ValueError("There is not enough cloud storage space for this folder.") file_id = uuid.uuid4().hex encrypted_temp: Path | None = None try: upload_path = source stored_size = original_size if progress: progress(0.20, desc="Web sitesi klasörü hazırlanıyor") if plan.encryption != "none": encrypted_temp = SETTINGS.temp_dir / f"generated-folder-{file_id}.bin" stored_size = encrypt_file(source, encrypted_temp, user_id, file_id, plan.encryption) upload_path = encrypted_temp if progress: progress(0.60, desc="Klasör şifreleniyor") remote_path = f"objects/{user_id}/{file_id}.blob" BACKEND.upload(upload_path, remote_path, "Upload generated website folder") metadata = { "id": file_id, "owner_id": user_id, "display_name": safe_filename(display_name), "remote_path": remote_path, "original_size": original_size, "stored_size": stored_size, "is_folder": True, "content_type": "application/zip", "encryption": plan.encryption, "share_password_hash": None, "download_count": 0, "last_download_at": None, "created_at": iso(), "uploaded_under_plan": plan.key, "estimated_upload_seconds": estimate_upload_seconds(original_size, plan.key), } saved = STORE.add_file(user_id, metadata) if progress: progress(1.0, desc="Klasör bulut depolamaya eklendi") return saved finally: if encrypted_temp: encrypted_temp.unlink(missing_ok=True) def materialize_download(item: dict[str, Any]) -> Path: file_id = item["id"] remote_temp = SETTINGS.temp_dir / f"download-remote-{file_id}-{uuid.uuid4().hex}.bin" output = SETTINGS.temp_dir / f"download-{uuid.uuid4().hex}-{safe_filename(item.get('display_name', 'file.bin'))}" if not BACKEND.download(item["remote_path"], remote_temp): raise FileNotFoundError("The stored object could not be downloaded.") try: decrypt_file(remote_temp, output, item["owner_id"], file_id, item.get("encryption", "none")) finally: remote_temp.unlink(missing_ok=True) return output def guess_device(user_agent: str) -> str: value = (user_agent or "").lower() if "ipad" in value or "tablet" in value: return "Tablet" if "android" in value or "iphone" in value or "mobile" in value: return "Mobile" return "Desktop"