207 lines
7.4 KiB
Python
207 lines
7.4 KiB
Python
"""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`` / ``<img src>``."""
|
|
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()]
|