"""Signed public URLs for raster images referenced by generated HTML. The HTML preview is a blob iframe, so virtual sandbox paths and auth-gated artifact URLs cannot be used as CSS ``background-image`` values. These tokens grant read-only access to one image under a thread's ``uploads/`` or ``outputs/`` directory. They do not expire: the URL is written into durable HTML and must keep working when the page is reopened. """ from __future__ import annotations import base64 import hashlib import hmac import os import secrets import stat from pathlib import Path from deerflow.config import get_paths from deerflow.config.paths import Paths from deerflow.config.runtime_paths import runtime_home from deerflow.runtime.thread_paths import remember_thread_owner, resolve_path_user_id from deerflow.uploads.manager import normalize_filename, validate_path_traversal, validate_thread_id IMAGE_EXTENSIONS = frozenset({".jpg", ".jpeg", ".png", ".webp", ".gif", ".bmp"}) ALLOWED_KINDS = frozenset({"uploads", "outputs"}) IMAGE_MIME_TYPES = { ".jpg": "image/jpeg", ".jpeg": "image/jpeg", ".png": "image/png", ".webp": "image/webp", ".gif": "image/gif", ".bmp": "image/bmp", } SECRET_FILE_NAME = ".html-image-secret" _secret_cache: bytes | None = None def reset_html_image_secret_cache() -> None: """Clear the process-wide secret cache (tests only).""" global _secret_cache _secret_cache = None def _secret_path() -> Path: return runtime_home() / SECRET_FILE_NAME def get_html_image_secret() -> bytes: """Load or create the instance HMAC key used to sign public image tokens.""" global _secret_cache if _secret_cache is not None: return _secret_cache path = _secret_path() try: if path.is_file(): data = path.read_bytes().strip() if len(data) >= 32: _secret_cache = data return _secret_cache path.parent.mkdir(parents=True, exist_ok=True) data = secrets.token_hex(32).encode("ascii") path.write_bytes(data) try: path.chmod(0o600) except OSError: pass _secret_cache = data return _secret_cache except OSError: _secret_cache = secrets.token_hex(32).encode("ascii") return _secret_cache def _normalize_kind(kind: str) -> str: normalized = (kind or "").strip().lower() if normalized not in ALLOWED_KINDS: raise ValueError("unsupported image location") return normalized def _normalize_image_filename(filename: str) -> str: safe = normalize_filename(filename) if Path(safe).suffix.lower() not in IMAGE_EXTENSIONS: raise ValueError("not a public HTML image") return safe def mint_html_image_token(thread_id: str, kind: str, filename: str) -> str: """Sign a capability token for one thread image file.""" validate_thread_id(thread_id) kind = _normalize_kind(kind) filename = _normalize_image_filename(filename) payload = f"{thread_id}|{kind}|{filename}" encoded = base64.urlsafe_b64encode(payload.encode("utf-8")).decode("ascii").rstrip("=") signature = hmac.new(get_html_image_secret(), payload.encode("utf-8"), hashlib.sha256).hexdigest() return f"{encoded}.{signature}" def verify_html_image_token(token: str) -> tuple[str, str, str]: """Return ``(thread_id, kind, filename)`` or raise ``ValueError``.""" if not token or token.count(".") != 1: raise ValueError("malformed token") encoded, signature = token.split(".", 1) if not encoded or not signature: raise ValueError("malformed token") padding = "=" * (-len(encoded) % 4) try: payload = base64.urlsafe_b64decode(encoded + padding).decode("utf-8") except (ValueError, UnicodeDecodeError) as exc: raise ValueError("malformed token") from exc parts = payload.split("|") if len(parts) != 3: raise ValueError("malformed token") thread_id, kind, filename = parts expected = hmac.new(get_html_image_secret(), payload.encode("utf-8"), hashlib.sha256).hexdigest() if not hmac.compare_digest(expected, signature): raise ValueError("bad signature") validate_thread_id(thread_id) return thread_id, _normalize_kind(kind), _normalize_image_filename(filename) def html_images_base_url(runtime_base: str | None = None) -> str: """Browser-reachable prefix for signed HTML images. Derived from the HTML runtime address saved in the settings dialog, so intranet deployments keep the same Gateway origin as G6/ECharts. """ if runtime_base is None: from deerflow.html_runtime.addresses import html_runtime_base_url runtime_base = html_runtime_base_url() base = runtime_base.rstrip("/") if base.endswith("/html-runtime"): return f"{base[: -len('html-runtime')]}html-images" parent, _sep, _name = base.rpartition("/") return f"{parent}/html-images" if parent else "/api/public/html-images" def public_image_url(thread_id: str, filename: str, *, kind: str = "uploads") -> str: """Absolute URL suitable for CSS ``background-image`` / ````.""" token = mint_html_image_token(thread_id, kind, filename) return f"{html_images_base_url()}/{token}" def _image_base_dir(paths: Paths, thread_id: str, kind: str, user_id: str | None) -> Path: if kind == "uploads": return paths.sandbox_uploads_dir(thread_id, user_id=user_id) return paths.sandbox_outputs_dir(thread_id, user_id=user_id) def _resolve_regular_image_at(base: Path, filename: str) -> Path: target = (base / filename).resolve() validate_path_traversal(target, base) opened_stat = os.lstat(target) if not stat.S_ISREG(opened_stat.st_mode): raise ValueError("not a regular image file") return target def _scan_user_buckets_for_html_image( paths: Paths, thread_id: str, kind: str, filename: str, ) -> Path: """Find a thread image when metadata lookup resolves the wrong user bucket.""" users_root = paths.base_dir / "users" if users_root.is_dir(): for user_dir in sorted(users_root.iterdir()): if not user_dir.is_dir(): continue base = _image_base_dir(paths, thread_id, kind, user_dir.name) try: target = _resolve_regular_image_at(base, filename) except (FileNotFoundError, ValueError, OSError): continue remember_thread_owner(thread_id, user_dir.name) return target legacy_base = _image_base_dir(paths, thread_id, kind, None) return _resolve_regular_image_at(legacy_base, filename) def resolve_html_image_path(thread_id: str, kind: str, filename: str) -> Path: """Resolve a signed image reference to a regular host file.""" validate_thread_id(thread_id) kind = _normalize_kind(kind) filename = _normalize_image_filename(filename) user_id = resolve_path_user_id(thread_id) paths = get_paths() base = _image_base_dir(paths, thread_id, kind, user_id) try: return _resolve_regular_image_at(base, filename) except FileNotFoundError: try: return _scan_user_buckets_for_html_image(paths, thread_id, kind, filename) except (FileNotFoundError, ValueError, OSError): raise FileNotFoundError(filename) from None def image_media_type(filename: str) -> str: """Return the MIME type for a supported raster image filename.""" return IMAGE_MIME_TYPES[Path(filename).suffix.lower()]