Imports to top of file (ruff rule now). Refactor a couple of import cycles by implementing clear hierarchy of modules.

This commit is contained in:
2026-05-02 04:46:34 +00:00
parent 5c7c7343ad
commit 922069c603
15 changed files with 480 additions and 457 deletions
+1 -3
View File
@@ -6,6 +6,7 @@ from docopt import docopt
import cista
from cista import app, config, droppy, onlyoffice, serve, server80
from cista.sso import PASKIA_BACKEND_URL
from cista.util import pwgen
del app, server80.app # Only import needed, for Sanic multiprocessing
@@ -154,9 +155,6 @@ def _main():
if not config.config.path.is_dir():
raise ValueError(f"No such directory: {config.config.path}")
dev = args["--dev"]
# Check for Paskia SSO
from cista.sso import PASKIA_BACKEND_URL
# Print startup box
startup_box = create_startup_box(
folder=config.config.path,
+1 -3
View File
@@ -6,7 +6,7 @@ from sanic import Blueprint, json
from sanic.exceptions import BadRequest
from sanic.log import logger
from cista import __version__, auth, config, sharefs, sso, watching
from cista import __version__, auth, config, onlyoffice, sharefs, sso, watching
from cista.auth import (
create_share_token_handler,
create_token_handler,
@@ -55,8 +55,6 @@ async def watch(req, ws):
"privileged": req.ctx.user.privileged,
}
from cista import onlyoffice
await ws.send(
msgspec.json.encode(
{
+19 -54
View File
@@ -2,22 +2,21 @@ import base64
import binascii
import hashlib
import hmac
import re
import secrets
import struct
from pathlib import PurePosixPath
from time import time
from unicodedata import normalize
import argon2
import msgspec
from Crypto.Hash import MD4
from html5tagger import Document
from sanic import Blueprint, html, json, redirect
from sanic.exceptions import BadRequest, Forbidden, Unauthorized
from sanic.log import logger
from cista import config, session, sharefs
from cista.util import pwgen
from cista import sso as _sso_module
from cista.util import pwgen, pwhash
from cista.util.filename import sanitize
_LOGIN_PAGE_CSS = """\
@@ -175,16 +174,8 @@ form.onsubmit = async (e) => {
};
"""
# Import for SSO validation (lazily loaded to avoid circular imports)
_sso_module = None
def _get_sso():
global _sso_module
if _sso_module is None:
from cista import sso
_sso_module = sso
return _sso_module
@@ -234,9 +225,6 @@ def hydrate_request_auth_context(request, *, source: str) -> None:
auth_flow.append(f"session:{source}(bad-jwt)")
_argon = argon2.PasswordHasher()
_droppyhash = re.compile(r"^([a-f0-9]{64})\$([a-f0-9]{8})$")
_AUTH_REALM = "cista"
_AUTH_CACHE_TTL = 10
_auth_cache: dict[str, tuple[float, config.User]] = {}
@@ -448,12 +436,6 @@ def _ntlmv2_verify(
nt_response: bytes,
) -> bool:
"""Verify an NTLMv2 response using the plaintext token secret as the password."""
try:
from Crypto.Hash import MD4
except ImportError:
logger.error("pycryptodome MD4 not available, cannot verify NTLM")
return False
if len(nt_response) < 16:
return False
@@ -517,47 +499,30 @@ def _ntlmv2_verify(
return False
def _pwnorm(password):
return normalize("NFC", password).strip().encode()
def _cache_key(username: str, password: str) -> str:
return hashlib.sha256(f"{username}\x00{password}".encode()).hexdigest()
def login(username: str, password: str):
normalized_username = pwhash.normalize_secret(username).decode()
cache_key = _cache_key(username, password)
cached = _auth_cache.get(cache_key)
if cached:
ts, user = cached
if time() - ts < _AUTH_CACHE_TTL:
return user
current = config.config.users.get(normalized_username)
if current and current.hash == user.hash:
return current
del _auth_cache[cache_key]
un = _pwnorm(username)
pw = _pwnorm(password)
try:
u = config.config.users[un.decode()]
u = config.config.users[normalized_username]
except KeyError:
raise ValueError("Invalid username") from None
# Verify password
need_rehash = False
if not u.hash:
raise ValueError("Account disabled")
if (m := _droppyhash.match(u.hash)) is not None:
h, s = m.groups()
h2 = hmac.digest(pw + s.encode() + un, b"", "sha256").hex()
if not hmac.compare_digest(h, h2):
raise ValueError("Invalid password")
# Droppy hashes are weak, do a hash update
need_rehash = True
else:
try:
_argon.verify(u.hash, pw)
except Exception:
raise ValueError("Invalid password") from None
if _argon.check_needs_rehash(u.hash):
need_rehash = True
need_rehash = pwhash.verify_hash(
u.hash, username=normalized_username, password=password
)
# Login successful
if need_rehash:
set_password(u, password)
@@ -568,7 +533,7 @@ def login(username: str, password: str):
def set_password(user: config.User, password: str):
user.hash = _argon.hash(_pwnorm(password))
pwhash.set_password(user, password)
_auth_cache.clear()
@@ -670,11 +635,12 @@ async def _token_auth_login(request, *, privileged=False):
ctx = data.get("ctx", {}) if isinstance(data, dict) else {}
user_info = ctx.get("user", {}) if isinstance(ctx, dict) else {}
request.ctx.username = user_info.get("display_name", "")
return True
except Forbidden:
raise
except Exception:
return False
else:
return True
if token.username:
user = config.config.users.get(token.username)
@@ -840,12 +806,13 @@ async def _ntlm_auth_login(request, *, privileged=False):
token.sso_user_id,
tid[:8],
)
return True
except Forbidden:
raise
except Exception as e:
logger.warning("NTLM SSO check failed: %s", e)
continue
else:
return True
if token.username:
user = config.config.users.get(token.username)
@@ -927,7 +894,6 @@ async def verify(request, *, privileged=False):
try:
perm = "cista:admin" if privileged else "cista:login"
await sso.validate_sso_request(request, perm=perm)
return
except Unauthorized as e:
auth_flow.append(f"tried={','.join(tried)} result=failed")
_set_auth_failure_log(request, auth_flow)
@@ -936,6 +902,8 @@ async def verify(request, *, privileged=False):
headers=_build_ua_auth_headers(request),
quiet=True,
) from e
else:
return
tried.append("sso")
perm = "cista:admin" if privileged else "cista:login"
await sso.validate_sso_request(request, perm=perm)
@@ -1228,8 +1196,7 @@ async def create_user(request):
raise BadRequest("User already exists")
if not password:
password = pwgen.generate()
changes = {"privileged": privileged}
changes["hash"] = _argon.hash(_pwnorm(password))
changes = {"privileged": privileged, "password": password}
try:
config.update_user(username, changes)
except Exception as e:
@@ -1256,8 +1223,6 @@ async def update_user(request, username):
if changes["password"] == "":
changes["password"] = pwgen.generate()
password_response = changes["password"]
changes["hash"] = _argon.hash(_pwnorm(changes["password"]))
del changes["password"]
if not changes:
return json({"message": "No changes"})
try:
+3 -3
View File
@@ -14,6 +14,8 @@ from typing import Concatenate, Literal, ParamSpec
import msgspec
import msgspec.toml
from .util import pwhash
class Config(msgspec.Struct):
path: Path
@@ -199,9 +201,7 @@ def update_user(conf: Config, name: str, changes: dict) -> Config:
except KeyError:
u = User()
if "password" in changes:
from . import auth
auth.set_password(u, changes["password"])
pwhash.set_password(u, changes["password"])
del changes["password"]
udict = msgspec.to_builtins(u, enc_hook=enc_hook)
udict.update(changes)
+2 -1
View File
@@ -186,9 +186,10 @@ async def is_available_async(request_timeout: float = 2.0) -> bool:
client = get_httpx_client()
try:
response = await client.get(url, timeout=request_timeout)
return response.status_code in (200, 405)
except Exception:
return False
else:
return response.status_code in (200, 405)
_oo_available_cache: tuple[bool, float] | None = None
+28 -366
View File
@@ -1,12 +1,8 @@
import asyncio
import contextlib
import gc
import io
import mimetypes
import struct
import subprocess
import sys
import tempfile
import threading
import urllib.parse
from collections import OrderedDict
@@ -17,19 +13,20 @@ from time import perf_counter
from urllib.parse import unquote
from wsgiref.handlers import format_date_time
import av
import fitz # PyMuPDF
import httpx
import msgspec
import numpy as np
import pyvips
from blake3 import blake3
from sanic import Blueprint, empty, raw, redirect
from sanic.exceptions import NotFound
from sanic.log import logger
from cista import auth, config, onlyoffice, sharefs
from cista.preview_worker import PreviewRequest, PreviewResponse
from cista.preview_worker import (
DOC_PREVIEW_SUFFIXES,
OFFICE_PREVIEW_SUFFIXES,
PreviewRequest,
PreviewResponse,
)
from cista.util.filename import sanitize
bp = Blueprint("preview", url_prefix="/preview")
@@ -165,6 +162,14 @@ class _PreviewWorkerPool:
self._seq = 0
self._closed = False
async def _read_startup_stderr(self, proc: asyncio.subprocess.Process) -> str:
if proc.stderr is None:
return ""
with contextlib.suppress(TimeoutError):
data = await asyncio.wait_for(proc.stderr.read(), timeout=0.5)
return data.decode(errors="replace").strip()
return ""
async def _spawn_worker(self) -> _PreviewWorker:
proc = await asyncio.create_subprocess_exec(
sys.executable,
@@ -172,7 +177,7 @@ class _PreviewWorkerPool:
"cista.preview_worker",
stdin=asyncio.subprocess.PIPE,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.DEVNULL,
stderr=asyncio.subprocess.PIPE,
start_new_session=True,
)
_active_procs.add(proc)
@@ -181,8 +186,21 @@ class _PreviewWorkerPool:
except TimeoutError as err:
with contextlib.suppress(ProcessLookupError):
proc.kill()
with contextlib.suppress(Exception):
await proc.wait()
stderr = await self._read_startup_stderr(proc)
if stderr:
raise WorkerProtocolError(
"preview worker failed to become ready: " + stderr.splitlines()[-1]
) from err
raise WorkerProtocolError("preview worker failed to become ready") from err
except asyncio.IncompleteReadError as err:
stderr = await self._read_startup_stderr(proc)
if stderr:
raise WorkerProtocolError(
"preview worker exited before signalling readiness: "
+ stderr.splitlines()[-1]
) from err
raise WorkerProtocolError(
"preview worker exited before signalling readiness"
) from err
@@ -498,50 +516,6 @@ async def _run_preview_process(
return await _preview_pool.run(filepath, quality, maxsize, maxzoom, data)
DOC_PREVIEW_SUFFIXES = {".pdf", ".xps", ".epub", ".mobi"}
OFFICE_PREVIEW_SUFFIXES = {
".doc",
".dot",
".docx",
".docm",
".dotx",
".dotm",
".rtf",
".odt",
".ott",
".txt",
".md",
".mhtml",
".mht",
".html",
".htm",
".xml",
".wps",
".wri",
# Spreadsheets
".xls",
".xlsx",
".xlsm",
".xlsb",
".xltx",
".xltm",
".ods",
".ots",
".csv",
# Presentations
".ppt",
".pptx",
".pptm",
".pps",
".ppsx",
".pot",
".potx",
".odp",
".otp",
}
def _onlyoffice_error_short_text(detail: str) -> str:
if detail.startswith("OnlyOffice conversion error:"):
code = detail.rsplit(":", 1)[-1].strip()
@@ -698,315 +672,3 @@ async def preview(req, path):
_preview_cache.set(etag, CachedPreview(headers=headers, body=img))
return raw(img, headers=headers)
def dispatch(path, quality, maxsize, maxzoom, data=None):
backend = "unknown"
try:
if data:
backend = "pyvips"
return process_image_buffer(
data, quality=quality, maxsize=maxsize, maxzoom=maxzoom
)
suffix = path.suffix.lower()
if suffix in DOC_PREVIEW_SUFFIXES:
backend = "pdf"
return process_pdf(path, quality=quality, maxsize=maxsize, maxzoom=maxzoom)
mime_type, _ = mimetypes.guess_type(path.name)
if mime_type and mime_type.startswith("video/"):
backend = "video"
return process_video(path, quality=quality, maxsize=maxsize)
if mime_type and mime_type.startswith("image/"):
backend = "pyvips"
return process_image(path, quality=quality, maxsize=maxsize)
except ValueError as e:
return None, PreviewResponse(ok=False, backend=backend, error=str(e))
except Exception as e:
logger.exception("Preview dispatch failed for %s", path)
return None, PreviewResponse(ok=False, backend=backend, error=str(e))
return None, PreviewResponse(ok=False, backend=backend, error="preview unsupported")
def process_image(path, *, maxsize, quality):
return process_image_pyvips(path, maxsize=maxsize, quality=quality)
def _get_image_dimensions(path: Path) -> tuple[int, int] | None:
"""Probe image dimensions.
pyvips can read the header of most formats (including HEIC) without
fully decoding the image.
"""
try:
img = pyvips.Image.new_from_file(str(path))
return img.width, img.height
except pyvips.error.Error:
return None
def _image_via_ffmpeg(path: Path, maxsize: int, quality: int) -> bytes:
"""Convert any image to AVIF using ffmpeg CLI.
ffmpeg handles HEIC tile assembly, EXIF rotation, HDR metadata and
ICC profile embedding automatically.
"""
dims = _get_image_dimensions(path)
crf = int(63 * (1 - quality / 100) ** 2)
with tempfile.NamedTemporaryFile(suffix=".avif", delete=False) as tmp_f:
tmp_path = tmp_f.name
cmd = [
"ffmpeg",
"-y",
"-i",
str(path),
"-frames:v",
"1",
"-c:v",
"av1",
"-crf",
str(crf),
"-cpu-used",
"8",
tmp_path,
]
if dims is not None:
w, h = dims
if max(w, h) > maxsize:
scale = min(maxsize / w, maxsize / h)
new_w = int(w * scale)
new_h = int(h * scale)
# insert -s <wxh> right after the input file
cmd.insert(4, "-s")
cmd.insert(5, f"{new_w}x{new_h}")
try:
subprocess.run(cmd, capture_output=True, check=True, shell=False) # noqa: S603
with Path(tmp_path).open("rb") as f:
return f.read()
finally:
Path(tmp_path).unlink(missing_ok=True)
def process_image_pyvips(path, *, maxsize, quality):
t_start = perf_counter()
suffix = path.suffix.lower()
# HEIC/HEIF: ffmpeg handles tile assembly and HDR correctly;
# skip pyvips entirely.
if suffix in (".heic", ".heif"):
ret = _image_via_ffmpeg(path, maxsize, quality)
t_end = perf_counter()
return ret, PreviewResponse(
ok=True,
mime="image/avif",
backend="ffmpeg",
timings=[round((t_end - t_start) * 1000, 1)],
)
# Other image formats: pyvips first, ffmpeg fallback.
load_opts = {"access": "sequential"}
try:
img = pyvips.Image.new_from_file(str(path), **load_opts)
img = img.autorot()
scale = min(maxsize / img.width, maxsize / img.height, 1.0)
if scale < 1.0:
img = img.resize(scale)
ret = img.write_to_buffer(
".avif",
Q=quality,
effort=AVIF_FAST_EFFORT,
strip=True,
)
backend = "pyvips"
except pyvips.error.Error:
ret = _image_via_ffmpeg(path, maxsize, quality)
backend = "ffmpeg"
t_end = perf_counter()
return ret, PreviewResponse(
ok=True,
mime="image/avif",
backend=backend,
timings=[round((t_end - t_start) * 1000, 1)],
)
def process_image_buffer(data: bytes, *, quality, maxsize, maxzoom):
t_start = perf_counter()
img = pyvips.Image.new_from_buffer(data, "")
img = img.autorot()
scale = min(maxsize / img.width, maxsize / img.height, 1.0)
if scale < 1.0:
img = img.resize(scale)
ret = img.write_to_buffer(
".avif",
Q=quality,
effort=AVIF_FAST_EFFORT,
strip=True,
)
t_end = perf_counter()
return ret, PreviewResponse(
ok=True,
mime="image/avif",
backend="pyvips",
timings=[round((t_end - t_start) * 1000, 1)],
)
def process_pdf(path, *, maxsize, maxzoom, quality, page_number=0):
t_load_start = perf_counter()
pdf = fitz.open(path)
page = pdf.load_page(page_number)
w, h = page.rect[2:4]
zoom = min(maxsize / w, maxsize / h, maxzoom)
mat = fitz.Matrix(zoom, zoom)
pix = page.get_pixmap(matrix=mat)
t_load_end = perf_counter()
t_save_start = perf_counter()
img = pyvips.Image.new_from_memory(
pix.samples_mv, pix.width, pix.height, pix.n, "uchar"
)
ret = img.write_to_buffer(".avif", Q=quality, effort=AVIF_FAST_EFFORT, strip=True)
backend = "pdf+pyvips"
t_save_end = perf_counter()
return ret, PreviewResponse(
ok=True,
mime="image/avif",
backend=backend,
timings=[
round((t_load_end - t_load_start) * 1000, 1),
round((t_save_end - t_save_start) * 1000, 1),
],
)
def process_video(path, *, maxsize, quality):
frame = None
imgdata = io.BytesIO()
istream = ostream = icc = occ = frame = None
t_load_start = perf_counter()
# Initialize to avoid "possibly unbound" in static analysis when exceptions occur
t_load_end = t_load_start
t_save_start = t_load_start
t_save_end = t_load_start
with (
av.open(
str(path),
options={
"analyzeduration": "1000000", # 1 second (in microseconds)
"fflags": "fastseek",
},
) as icontainer,
av.open(imgdata, "w", format="avif") as ocontainer,
):
istream = icontainer.streams.video[0]
istream.codec_context.skip_frame = "NONKEY"
icontainer.seek((icontainer.duration or 0) // 8)
for frame in icontainer.decode(istream):
if frame.dts is not None:
break
else:
raise RuntimeError("No frames found in video")
# Resize frame to thumbnail size
if frame.width > maxsize or frame.height > maxsize:
scale_factor = min(maxsize / frame.width, maxsize / frame.height)
new_width = int(frame.width * scale_factor)
new_height = int(frame.height * scale_factor)
frame = frame.reformat(width=new_width, height=new_height)
# Apply EXIF rotation if present
if frame.rotation:
# frame.rotation indicates clockwise rotation needed to display correctly
# np.rot90 rotates counter-clockwise, so we negate k
k = (frame.rotation // 90) % 4 # Convert to counter-clockwise rotations
if k == 2:
# 180° rotation can be done in YUV420p, preserving HDR
try:
fplanes = frame.to_ndarray()
# Split into Y, U, V planes of proper dimensions
planes = [
fplanes[: frame.height],
fplanes[
frame.height : frame.height + frame.height // 4
].reshape(frame.height // 2, frame.width // 2),
fplanes[frame.height + frame.height // 4 :].reshape(
frame.height // 2, frame.width // 2
),
]
# Rotate each plane by 180°
planes = [np.rot90(p, 2) for p in planes]
# Restore PyAV format
planes = np.hstack([p.flat for p in planes]).reshape(
-1, planes[0].shape[1]
)
frame = av.VideoFrame.from_ndarray(planes, format=frame.format.name)
del planes, fplanes
except Exception as e:
logger.exception(f"Error rotating video frame by 180°: {e}")
elif k in (1, 3):
# 90° or 270° rotation requires RGB conversion (loses HDR)
try:
rgb = frame.to_ndarray(format="rgb24")
rgb = np.rot90(rgb, k)
frame = av.VideoFrame.from_ndarray(rgb, format="rgb24")
frame = frame.reformat(
format="yuv420p"
) # Convert back for encoding
del rgb
except Exception as e:
logger.exception(
f"Error rotating video frame by {frame.rotation}°: {e}"
)
# libsvtav1 rejects full-range JPEG-style YUV pixel formats such as
# yuvj420p, so normalize them before opening the encoder.
if frame.format.name.startswith("yuvj"):
frame = frame.reformat(format="yuv420p")
t_load_end = perf_counter()
t_save_start = perf_counter()
crf = str(int(63 * (1 - quality / 100) ** 2)) # Closely matching PIL quality-%
ostream = ocontainer.add_stream(
"av1",
options={
"crf": crf,
"usage": "realtime",
"cpu-used": "8",
"threads": "1",
},
)
if not isinstance(ostream, av.VideoStream):
raise PreviewError("failed to initialize AV1 video stream")
ostream.width = frame.width
ostream.height = frame.height
ostream.pix_fmt = frame.format.name
icc = istream.codec_context
occ = ostream.codec_context
# Copy HDR metadata from input video stream
occ.color_primaries = icc.color_primaries
occ.color_trc = icc.color_trc
occ.colorspace = icc.colorspace
occ.color_range = icc.color_range
ocontainer.mux(ostream.encode(frame))
ocontainer.mux(ostream.encode(None)) # Flush the stream
t_save_end = perf_counter()
# Capture result before cleanup
ret = imgdata.getvalue()
resp = PreviewResponse(
ok=True,
mime="image/avif",
backend="video",
timings=[
round((t_load_end - t_load_start) * 1000, 1),
round((t_save_end - t_save_start) * 1000, 1),
],
)
del imgdata, istream, ostream, icc, occ, frame
gc.collect()
return ret, resp
+370 -9
View File
@@ -1,4 +1,4 @@
"""Preview generation worker subprocess.
"""Preview generation worker subprocess and synchronous preview engine.
Two modes are supported:
1) Legacy one-shot mode: argv has path/quality/maxsize/maxzoom.
@@ -13,17 +13,73 @@ where packet = (uint32 json size)(uint32 payload size)(json)(binary payload).
"""
import contextlib
import gc
import io
import logging
import mimetypes
import struct
import subprocess
import sys
import tempfile
from pathlib import Path
from time import perf_counter
import av
import fitz # PyMuPDF
import msgspec
import numpy as np
import pyvips
from blake3 import blake3
from cista import config
logger = logging.getLogger(__name__)
AVIF_FAST_EFFORT = 0
DOC_PREVIEW_SUFFIXES = {".pdf", ".xps", ".epub", ".mobi"}
OFFICE_PREVIEW_SUFFIXES = {
".doc",
".dot",
".docx",
".docm",
".dotx",
".dotm",
".rtf",
".odt",
".ott",
".txt",
".md",
".mhtml",
".mht",
".html",
".htm",
".xml",
".wps",
".wri",
# Spreadsheets
".xls",
".xlsx",
".xlsm",
".xlsb",
".xltx",
".xltm",
".ods",
".ots",
".csv",
# Presentations
".ppt",
".pptx",
".pptm",
".pps",
".ppsx",
".pot",
".potx",
".odp",
".otp",
}
class PreviewRequest(msgspec.Struct, omit_defaults=True):
path: str
@@ -78,13 +134,324 @@ def _write_response(resp: PreviewResponse, payload: bytes) -> None:
sys.stdout.buffer.flush()
def dispatch(path, quality, maxsize, maxzoom, data=None):
backend = "unknown"
try:
if data:
backend = "pyvips"
return process_image_buffer(
data, quality=quality, maxsize=maxsize, maxzoom=maxzoom
)
suffix = path.suffix.lower()
if suffix in DOC_PREVIEW_SUFFIXES:
backend = "pdf"
return process_pdf(path, quality=quality, maxsize=maxsize, maxzoom=maxzoom)
mime_type, _ = mimetypes.guess_type(path.name)
if mime_type and mime_type.startswith("video/"):
backend = "video"
return process_video(path, quality=quality, maxsize=maxsize)
if mime_type and mime_type.startswith("image/"):
backend = "pyvips"
return process_image(path, quality=quality, maxsize=maxsize)
except ValueError as e:
return None, PreviewResponse(ok=False, backend=backend, error=str(e))
except Exception as e:
logger.exception("Preview dispatch failed for %s", path)
return None, PreviewResponse(ok=False, backend=backend, error=str(e))
return None, PreviewResponse(ok=False, backend=backend, error="preview unsupported")
def process_image(path, *, maxsize, quality):
return process_image_pyvips(path, maxsize=maxsize, quality=quality)
def _get_image_dimensions(path: Path) -> tuple[int, int] | None:
"""Probe image dimensions.
pyvips can read the header of most formats (including HEIC) without
fully decoding the image.
"""
try:
img = pyvips.Image.new_from_file(str(path))
except pyvips.error.Error:
return None
else:
return img.width, img.height
def _image_via_ffmpeg(path: Path, maxsize: int, quality: int) -> bytes:
"""Convert any image to AVIF using ffmpeg CLI.
ffmpeg handles HEIC tile assembly, EXIF rotation, HDR metadata and
ICC profile embedding automatically.
"""
dims = _get_image_dimensions(path)
crf = int(63 * (1 - quality / 100) ** 2)
with tempfile.NamedTemporaryFile(suffix=".avif", delete=False) as tmp_f:
tmp_path = tmp_f.name
cmd = [
"ffmpeg",
"-y",
"-i",
str(path),
"-frames:v",
"1",
"-c:v",
"av1",
"-crf",
str(crf),
"-cpu-used",
"8",
tmp_path,
]
if dims is not None:
w, h = dims
if max(w, h) > maxsize:
scale = min(maxsize / w, maxsize / h)
new_w = int(w * scale)
new_h = int(h * scale)
# insert -s <wxh> right after the input file
cmd.insert(4, "-s")
cmd.insert(5, f"{new_w}x{new_h}")
try:
subprocess.run(cmd, capture_output=True, check=True, shell=False) # noqa: S603
with Path(tmp_path).open("rb") as f:
return f.read()
finally:
Path(tmp_path).unlink(missing_ok=True)
def process_image_pyvips(path, *, maxsize, quality):
t_start = perf_counter()
suffix = path.suffix.lower()
# HEIC/HEIF: ffmpeg handles tile assembly and HDR correctly;
# skip pyvips entirely.
if suffix in (".heic", ".heif"):
ret = _image_via_ffmpeg(path, maxsize, quality)
t_end = perf_counter()
return ret, PreviewResponse(
ok=True,
mime="image/avif",
backend="ffmpeg",
timings=[round((t_end - t_start) * 1000, 1)],
)
# Other image formats: pyvips first, ffmpeg fallback.
load_opts = {"access": "sequential"}
try:
img = pyvips.Image.new_from_file(str(path), **load_opts)
img = img.autorot()
scale = min(maxsize / img.width, maxsize / img.height, 1.0)
if scale < 1.0:
img = img.resize(scale)
ret = img.write_to_buffer(
".avif",
Q=quality,
effort=AVIF_FAST_EFFORT,
strip=True,
)
backend = "pyvips"
except pyvips.error.Error:
ret = _image_via_ffmpeg(path, maxsize, quality)
backend = "ffmpeg"
t_end = perf_counter()
return ret, PreviewResponse(
ok=True,
mime="image/avif",
backend=backend,
timings=[round((t_end - t_start) * 1000, 1)],
)
def process_image_buffer(data: bytes, *, quality, maxsize, maxzoom):
t_start = perf_counter()
img = pyvips.Image.new_from_buffer(data, "")
img = img.autorot()
scale = min(maxsize / img.width, maxsize / img.height, 1.0)
if scale < 1.0:
img = img.resize(scale)
ret = img.write_to_buffer(
".avif",
Q=quality,
effort=AVIF_FAST_EFFORT,
strip=True,
)
t_end = perf_counter()
return ret, PreviewResponse(
ok=True,
mime="image/avif",
backend="pyvips",
timings=[round((t_end - t_start) * 1000, 1)],
)
def process_pdf(path, *, maxsize, maxzoom, quality, page_number=0):
t_load_start = perf_counter()
pdf = fitz.open(path)
page = pdf.load_page(page_number)
w, h = page.rect[2:4]
zoom = min(maxsize / w, maxsize / h, maxzoom)
mat = fitz.Matrix(zoom, zoom)
pix = page.get_pixmap(matrix=mat)
t_load_end = perf_counter()
t_save_start = perf_counter()
img = pyvips.Image.new_from_memory(
pix.samples_mv, pix.width, pix.height, pix.n, "uchar"
)
ret = img.write_to_buffer(".avif", Q=quality, effort=AVIF_FAST_EFFORT, strip=True)
backend = "pdf+pyvips"
t_save_end = perf_counter()
return ret, PreviewResponse(
ok=True,
mime="image/avif",
backend=backend,
timings=[
round((t_load_end - t_load_start) * 1000, 1),
round((t_save_end - t_save_start) * 1000, 1),
],
)
def process_video(path, *, maxsize, quality):
frame = None
imgdata = io.BytesIO()
istream = ostream = icc = occ = frame = None
t_load_start = perf_counter()
# Initialize to avoid "possibly unbound" in static analysis when exceptions occur
t_load_end = t_load_start
t_save_start = t_load_start
t_save_end = t_load_start
with (
av.open(
str(path),
options={
"analyzeduration": "1000000", # 1 second (in microseconds)
"fflags": "fastseek",
},
) as icontainer,
av.open(imgdata, "w", format="avif") as ocontainer,
):
istream = icontainer.streams.video[0]
istream.codec_context.skip_frame = "NONKEY"
icontainer.seek((icontainer.duration or 0) // 8)
for frame in icontainer.decode(istream):
if frame.dts is not None:
break
else:
raise RuntimeError("No frames found in video")
# Resize frame to thumbnail size
if frame.width > maxsize or frame.height > maxsize:
scale_factor = min(maxsize / frame.width, maxsize / frame.height)
new_width = int(frame.width * scale_factor)
new_height = int(frame.height * scale_factor)
frame = frame.reformat(width=new_width, height=new_height)
# Apply EXIF rotation if present
if frame.rotation:
# frame.rotation indicates clockwise rotation needed to display correctly
# np.rot90 rotates counter-clockwise, so we negate k
k = (frame.rotation // 90) % 4 # Convert to counter-clockwise rotations
if k == 2:
# 180° rotation can be done in YUV420p, preserving HDR
try:
fplanes = frame.to_ndarray()
# Split into Y, U, V planes of proper dimensions
planes = [
fplanes[: frame.height],
fplanes[
frame.height : frame.height + frame.height // 4
].reshape(frame.height // 2, frame.width // 2),
fplanes[frame.height + frame.height // 4 :].reshape(
frame.height // 2, frame.width // 2
),
]
# Rotate each plane by 180°
planes = [np.rot90(p, 2) for p in planes]
# Restore PyAV format
planes = np.hstack([p.flat for p in planes]).reshape(
-1, planes[0].shape[1]
)
frame = av.VideoFrame.from_ndarray(planes, format=frame.format.name)
del planes, fplanes
except Exception:
logger.exception("Error rotating video frame by 180°")
elif k in (1, 3):
# 90° or 270° rotation requires RGB conversion (loses HDR)
try:
rgb = frame.to_ndarray(format="rgb24")
rgb = np.rot90(rgb, k)
frame = av.VideoFrame.from_ndarray(rgb, format="rgb24")
frame = frame.reformat(
format="yuv420p"
) # Convert back for encoding
del rgb
except Exception:
logger.exception(
"Error rotating video frame by %s°", frame.rotation
)
# libsvtav1 rejects full-range JPEG-style YUV pixel formats such as
# yuvj420p, so normalize them before opening the encoder.
if frame.format.name.startswith("yuvj"):
frame = frame.reformat(format="yuv420p")
t_load_end = perf_counter()
t_save_start = perf_counter()
crf = str(int(63 * (1 - quality / 100) ** 2)) # Closely matching PIL quality-%
ostream = ocontainer.add_stream(
"av1",
options={
"crf": crf,
"usage": "realtime",
"cpu-used": "8",
"threads": "1",
},
)
if not isinstance(ostream, av.VideoStream):
raise RuntimeError("failed to initialize AV1 video stream")
ostream.width = frame.width
ostream.height = frame.height
ostream.pix_fmt = frame.format.name
icc = istream.codec_context
occ = ostream.codec_context
# Copy HDR metadata from input video stream
occ.color_primaries = icc.color_primaries
occ.color_trc = icc.color_trc
occ.colorspace = icc.colorspace
occ.color_range = icc.color_range
ocontainer.mux(ostream.encode(frame))
ocontainer.mux(ostream.encode(None)) # Flush the stream
t_save_end = perf_counter()
# Capture result before cleanup
ret = imgdata.getvalue()
resp = PreviewResponse(
ok=True,
mime="image/avif",
backend="video",
timings=[
round((t_load_end - t_load_start) * 1000, 1),
round((t_save_end - t_save_start) * 1000, 1),
],
)
del imgdata, istream, ostream, icc, occ, frame
gc.collect()
return ret, resp
def _run_once() -> None:
if len(sys.argv) != 5:
sys.stderr.write(f"Usage: {sys.argv[0]} <path> <quality> <maxsize> <maxzoom>\n")
sys.exit(1)
from cista.preview import dispatch
path = Path(sys.argv[1])
quality = int(sys.argv[2])
maxsize = int(sys.argv[3])
@@ -96,8 +463,6 @@ def _run_once() -> None:
def _run_loop() -> None:
from cista.preview import dispatch
while True:
result = _read_request()
if result is None:
@@ -137,8 +502,6 @@ def main() -> None:
# Configure all log output to stderr before any imports that may emit logs.
logging.basicConfig(stream=sys.stderr, level=logging.INFO)
try:
from cista import config
config.load_config()
logger.warning(
"preview-worker config=%s master_secret=%s",
@@ -152,8 +515,6 @@ def main() -> None:
return
# Eagerly import heavy modules before signalling readiness so the parent
# does not hand us a request while we are still initialising.
from cista.preview import dispatch # noqa: F401
sys.stdout.buffer.write(b"\x01")
sys.stdout.buffer.flush()
_run_loop()
+2 -2
View File
@@ -5,6 +5,8 @@ import sys
import unicodedata
from ipaddress import IPv6Address
from sanic.log import LOGGING_CONFIG_DEFAULTS
logger = logging.getLogger("cista.access")
_RESET = "\033[0m"
@@ -269,8 +271,6 @@ def configure_main_logging() -> None:
Patches LOGGING_CONFIG_DEFAULTS so the formatter survives every dictConfig
call Sanic makes during serve_single() / serve().
"""
from sanic.log import LOGGING_CONFIG_DEFAULTS
LOGGING_CONFIG_DEFAULTS["formatters"]["generic"] = {
"class": "cista.sanic_logging._EmojiFormatter",
}
+1 -2
View File
@@ -6,12 +6,11 @@ from fastapi_vue.hostutil import parse_endpoint
from sanic import Sanic
from cista import config, server80
from cista.app import app
def run(*, dev=False):
"""Run Sanic main process that spawns worker processes to serve HTTP requests."""
from .app import app
_url, opts = parse_listen(config.config.listen)
# Silence Sanic's warning about running in production rather than debug
os.environ["SANIC_IGNORE_PRODUCTION_WARNING"] = "1"
+2 -1
View File
@@ -107,10 +107,11 @@ async def validate_sso_request(request, *, perm: str = "cista:login") -> dict |
request.ctx.sso_user = data
if "set-cookie" in response.headers:
request.ctx.sso_cookies = response.headers.get_list("set-cookie")
return data
except Exception:
request.ctx.sso_user = {}
return {}
else:
return data
try:
error_data = response.json()
+47
View File
@@ -0,0 +1,47 @@
import hmac
import re
from typing import Protocol
from unicodedata import normalize
import argon2
_argon = argon2.PasswordHasher()
_droppyhash = re.compile(r"^([a-f0-9]{64})\$([a-f0-9]{8})$")
class SupportsHash(Protocol):
hash: str
def normalize_secret(value: str) -> bytes:
return normalize("NFC", value).strip().encode()
def verify_hash(user_hash: str, *, username: str, password: str) -> bool:
"""Verify password hash and return whether the stored hash should be upgraded."""
if not user_hash:
raise ValueError("Account disabled")
normalized_username = normalize_secret(username)
normalized_password = normalize_secret(password)
if (match := _droppyhash.match(user_hash)) is not None:
expected_hash, salt = match.groups()
computed_hash = hmac.digest(
normalized_password + salt.encode() + normalized_username,
b"",
"sha256",
).hex()
if not hmac.compare_digest(expected_hash, computed_hash):
raise ValueError("Invalid password")
return True
try:
_argon.verify(user_hash, normalized_password)
except Exception:
raise ValueError("Invalid password") from None
return _argon.check_needs_rehash(user_hash)
def set_password(user: SupportsHash, password: str) -> None:
user.hash = _argon.hash(normalize_secret(password))
+1 -4
View File
@@ -9,6 +9,7 @@ from os import stat_result
from pathlib import Path, PurePosixPath
from stat import S_ISDIR, S_ISREG
import inotify.adapters
import msgspec
from natsort import humansorted, natsort_keygen, ns
from sanic.log import logger
@@ -641,8 +642,6 @@ def watcher(loop):
modified_flags = frozenset()
if use_inotify:
import inotify.adapters
modified_flags = frozenset(
(
"IN_CREATE",
@@ -657,8 +656,6 @@ def watcher(loop):
while not stop_event.is_set():
if use_inotify:
import inotify.adapters
inotify_tree = inotify.adapters.InotifyTree(rootpath.as_posix())
# Initialize the tree from filesystem
-2
View File
@@ -152,7 +152,6 @@ ignore = [
"EM101", # exception-message style; low signal for this project
"EM102", # exception-message style; low signal for this project
"INP001", # scripts folder intentionally lacks package markers
"PLC0415", # lazy imports used to avoid startup/circular import issues
"PLR0911", # legacy complexity; keep other correctness rules enabled
"PLR0912", # legacy complexity; keep other correctness rules enabled
"PLR0913", # legacy complexity; keep other correctness rules enabled
@@ -163,7 +162,6 @@ ignore = [
"TRY002", # exception-class strictness too noisy on legacy handlers
"TRY003", # exception-message strictness too noisy on legacy handlers
"TRY004", # type-check strictness too noisy on legacy handlers
"TRY300", # stylistic try/else preference
"TRY301", # stylistic raise-in-try preference
]
isort.known-first-party = ["cista"]
+1 -2
View File
@@ -8,6 +8,7 @@ from uuid import uuid4
import pytest
import pytest_asyncio
from Crypto.Hash import MD4
from sanic import Sanic
from cista import auth, config, session, watching
@@ -29,8 +30,6 @@ def _ntlm_type3(
username: str, password: str, domain: str, challenge: bytes
) -> dict[str, str]:
"""Build an NTLMv2 Type 3 message for testing."""
from Crypto.Hash import MD4
# NT hash
nt_hash = MD4.new(password.encode("utf-16le")).digest()
# NTLMv2 hash
+2 -5
View File
@@ -1,6 +1,7 @@
from pathlib import Path
from pathlib import Path, PurePath
from uuid import uuid4
import msgspec
import pytest
import pytest_asyncio
from sanic import Sanic
@@ -11,10 +12,6 @@ from cista.auth import bp as auth_bp
def _persist_config():
from pathlib import PurePath
import msgspec
def enc_hook(obj):
if isinstance(obj, PurePath):
return obj.as_posix()