lint: apply manual ruff cleanup (non-preview files)

cista/__init__.py

@@ -1,3 +1 @@
-from cista._version import __version__
+from cista._version import __version__ as __version__
-
-__version__  # Public API

cista/__main__.py

@@ -34,10 +34,7 @@ def create_startup_box(
     location = f"{folder} @ {listen}"
     lines = [title, location]
     # Auth line: Paskia <url> or Password, with optional Public suffix
-    if paskia_url:
+    auth_line = f"Auth: Paskia {paskia_url}" if paskia_url else "Auth: Password"
-        auth_line = f"Auth: Paskia {paskia_url}"
-    else:
-        auth_line = "Auth: Password"
     if public:
         auth_line += ", Public"
     lines.append(auth_line)
@@ -49,8 +46,7 @@ def create_startup_box(
 
     # Build the box
     box = [f"╭{'─' * inner_width}╮"]
-    for line in lines:
+    box.extend(f"│ {line:<{inner_width - 1}}│" for line in lines)
-        box.append(f"│ {line:<{inner_width - 1}}│")
     box.append(f"╰{'─' * inner_width}╯")
     return "\n".join(box) + "\n"
 

cista/api.py

@@ -1,10 +1,10 @@
 import asyncio
-from pathlib import PurePosixPath
 from secrets import token_bytes
 
 import msgspec
 from sanic import Blueprint, json
 from sanic.exceptions import BadRequest
+from sanic.log import logger
 
 from cista import __version__, auth, config, sso, watching
 from cista.auth import (
@@ -38,8 +38,8 @@ async def watch(req, ws):
         # SSO auth: call validation to get user info (don't enforce auth in public mode)
         try:
             await sso.validate_sso_request(req)
-        except Exception:
+        except Exception as e:
-            pass  # Ignore auth errors, user_info stays None
+            logger.debug("watch SSO validation failed: %s", e)
         if sso_user := getattr(req.ctx, "sso_user", None):
             ctx = sso_user.get("ctx", {})
             perms = ctx.get("permissions", [])

cista/app.py

@@ -16,10 +16,9 @@ from setproctitle import setproctitle
 from stream_zip import ZIP_AUTO, stream_zip
 from zstandard import ZstdCompressor
 
-from cista import auth, config, preview, session, sso, watching
+from cista import auth, config, fileserver, preview, session, sso, watching
-from cista.preview import shutdown_preview_workers, start_preview_workers
 from cista.api import bp
-from cista import fileserver
+from cista.preview import shutdown_preview_workers, start_preview_workers
 from cista.sanic_logging import (
     configure_access_logging,
     configure_main_logging,
@@ -295,7 +294,8 @@ async def zip_download(req, keys, zipfile, ext):
             while size > 0 and (chunk := f.read(min(size, 1 << 20))):
                 size -= len(chunk)
                 yield chunk
-        assert size == 0
+        if size != 0:
+            raise OSError(f"stream ended early while zipping {name}")
 
     pending_put = None  # Current queue.put future, can be cancelled
 

cista/auth.py

@@ -1,7 +1,7 @@
 import base64
 import binascii
-import hmac
 import hashlib
+import hmac
 import re
 import secrets
 import struct
@@ -275,7 +275,6 @@ def _log_webdav_user_agent_once(request, user_agent: str):
         return
     _seen_webdav_uas.add(key)
     # Temporary stdout print so operators can quickly capture real client UAs.
-    print(f"WebDAV User-Agent observed: {key} path={request.path}")
 
 
 def _build_ua_auth_headers(request, *, include_hint=False) -> dict[str, str]:
@@ -285,8 +284,7 @@ def _build_ua_auth_headers(request, *, include_hint=False) -> dict[str, str]:
         challenge = f'Basic realm="{_AUTH_REALM}", Negotiate'
     else:
         challenge = f'Basic realm="{_AUTH_REALM}"'
-    headers = {"WWW-Authenticate": challenge}
+    return {"WWW-Authenticate": challenge}
-    return headers
 
 
 def _cleanup_ntlm_challenges():
@@ -399,14 +397,13 @@ def _spnego_wrap_ntlm_challenge(ntlm_type2: bytes) -> bytes:
     neg_state_accept_incomplete = _der_tlv(0xA0, _der_tlv(0x0A, b"\x01"))
     supported_mech = _der_tlv(0xA1, ntlm_oid)
     response_token = _der_tlv(0xA2, _der_tlv(0x04, ntlm_type2))
-    neg_token_resp = _der_tlv(
+    return _der_tlv(
         0xA1,
         _der_tlv(
             0x30,
             neg_state_accept_incomplete + supported_mech + response_token,
         ),
     )
-    return neg_token_resp
 
 
 def _ntlm_parse_type3(data: bytes) -> dict | None:
@@ -417,7 +414,7 @@ def _ntlm_parse_type3(data: bytes) -> dict | None:
         return None
 
     def read_buf(offset: int) -> bytes:
-        length, max_len, buf_offset = struct.unpack("<HHI", data[offset : offset + 8])
+        length, _max_len, buf_offset = struct.unpack("<HHI", data[offset : offset + 8])
         if length == 0:
             return b""
         if buf_offset + length > len(data):
@@ -460,7 +457,7 @@ def _ntlmv2_verify(
     blob = nt_response[16:]
 
     # NT hash = MD4(UTF-16LE(password))
-    nt_hash = MD4.new(token_secret.encode("utf-16le")).digest()
+    nt_hash = MD4.new(token_secret.encode("utf-16le")).digest()  # noqa: S303
 
     raw_username = username or ""
     raw_domain = domain or ""
@@ -629,7 +626,7 @@ def _basic_auth_login(request):
 
 
 
-async def _token_auth_login(request, privileged=False):
+async def _token_auth_login(request, *, privileged=False):
     """Authenticate via Basic token:<secret> in SSO mode.
 
     Returns True if authenticated, False if no token matched.
@@ -686,7 +683,7 @@ async def _token_auth_login(request, privileged=False):
     return False
 
 
-async def _ntlm_auth_login(request, privileged=False):
+async def _ntlm_auth_login(request, *, privileged=False):
     """Handle NTLM authentication for token-based login.
 
     Supports NTLMv2 responses where the token secret is used as the password.
@@ -706,9 +703,9 @@ async def _ntlm_auth_login(request, privileged=False):
 
     try:
         data = base64.b64decode(encoded)
-    except Exception:
+    except Exception as e:
         logger.warning("NTLM decode failed: client=%s", client_key)
-        raise Unauthorized("Invalid NTLM message", www_auth_scheme, quiet=True)
+        raise Unauthorized("Invalid NTLM message", www_auth_scheme, quiet=True) from e
 
     # Windows commonly sends SPNEGO-wrapped Negotiate tokens that embed NTLMSSP.
     # Extract the NTLMSSP blob when present so downstream parsing sees raw Type 1/3.
@@ -797,7 +794,6 @@ async def _ntlm_auth_login(request, privileged=False):
                 secret_candidates.append(("token-key", token.key))
 
             matched_by = None
-            matched_challenge = None
             for secret_kind, secret_value in secret_candidates:
                 for challenge in challenges:
                     if _ntlmv2_verify(
@@ -808,7 +804,6 @@ async def _ntlm_auth_login(request, privileged=False):
                         nt_response,
                     ):
                         matched_by = secret_kind
-                        matched_challenge = challenge
                         break
                 if matched_by:
                     break
@@ -919,14 +914,14 @@ async def verify(request, *, privileged=False):
                 perm = "cista:admin" if privileged else "cista:login"
                 await sso.validate_sso_request(request, perm=perm)
                 return
-            except Unauthorized:
+            except Unauthorized as e:
                 auth_flow.append(f"tried={','.join(tried)} result=failed")
                 _set_auth_failure_log(request, auth_flow)
                 raise Unauthorized(
                     "Invalid credentials",
                     headers=_build_ua_auth_headers(request),
                     quiet=True,
-                )
+                ) from e
         tried.append("sso")
         perm = "cista:admin" if privileged else "cista:login"
         await sso.validate_sso_request(request, perm=perm)
@@ -969,7 +964,7 @@ async def verify(request, *, privileged=False):
                 user = await _ntlm_auth_login(request, privileged=privileged)
             except Unauthorized as e:
                 auth_hdr = (e.headers or {}).get("WWW-Authenticate", "")
-                if (auth_hdr.startswith("NTLM ") or auth_hdr.startswith("Negotiate ")) and "realm=" not in auth_hdr:
+                if (auth_hdr.startswith(("NTLM ", "Negotiate "))) and "realm=" not in auth_hdr:
                     raise
                 ntlm_failed = True
                 user = None
@@ -1067,8 +1062,9 @@ async def login_page(request):
     doc.style(_LOGIN_PAGE_CSS)
     with doc.div(class_="login-card"):
         doc.h1("Authentication Required")
-        with doc.div(class_="content"):
+        with doc.div(class_="content"), doc.form(
-            with doc.form(method="POST", id="loginForm", autocomplete="on"):
+            method="POST", id="loginForm", autocomplete="on"
+        ):
             doc.label("Username:", for_="username")
             doc.input(
                 type="text",
@@ -1289,9 +1285,7 @@ def _token_belongs_to_user(token, username, sso_user_id):
     """Check if a token belongs to the given user."""
     if username is not None and token.username == username:
         return True
-    if sso_user_id is not None and token.sso_user_id == sso_user_id:
+    return bool(sso_user_id is not None and token.sso_user_id == sso_user_id)
-        return True
-    return False
 
 
 # Token management handlers (shared between /auth and /api blueprints)

cista/config.py

@@ -3,12 +3,13 @@ from __future__ import annotations
 import os
 import secrets
 import sys
+from collections.abc import Callable
 from contextlib import suppress
 from functools import wraps
 from hashlib import sha256
 from pathlib import Path, PurePath
 from time import sleep, time
-from typing import Callable, Concatenate, Literal, ParamSpec
+from typing import Concatenate, Literal, ParamSpec
 
 import msgspec
 import msgspec.toml
@@ -49,7 +50,7 @@ class Token(msgspec.Struct, omit_defaults=True):
     username: str = ""  # set in built-in mode
     sso_user_id: str = ""  # set in SSO mode
     name: str = ""
-    created: int = 0  # noqa: N815
+    created: int = 0
 
 
 # Global variables - initialized during application startup
@@ -72,7 +73,7 @@ def init_confdir() -> None:
     conffile = home / "db.toml"
 
 
-def derived_secret(*params, len=8) -> bytes:
+def derived_secret(*params, size=8) -> bytes:
     """Used to derive secret keys from the main secret"""
     # Each part is made the same length by hashing first
     combined = b"".join(
@@ -80,7 +81,7 @@ def derived_secret(*params, len=8) -> bytes:
         for p in [config.secret, *params]
     )
     # Output a bytes of the desired length
-    return sha256(combined).digest()[:len]
+    return sha256(combined).digest()[:size]
 
 
 def enc_hook(obj):

cista/droppy.py

@@ -17,7 +17,7 @@ def _droppy_listeners(cf):
     for listener in cf["listeners"]:
         try:
             if listener["protocol"] == "https":
-                # TODO: Add support for TLS
+                # TLS listeners are currently ignored here.
                 continue
             socket = listener.get("socket")
             if socket:

cista/fileio.py

@@ -1,5 +1,6 @@
 import os
 import threading
+from pathlib import Path
 
 from cista import config
 from cista.util import filename
@@ -31,20 +32,23 @@ class File:
         if not self.writable:
             # Create/open file
             self.open_rw()
-        assert self.fd is not None
+        if self.fd is None:
+            raise RuntimeError("file descriptor is not available for write")
         if file_size is not None:
-            assert pos + len(buffer) <= file_size
+            if pos + len(buffer) > file_size:
+                raise ValueError("write exceeds declared file size")
             os.ftruncate(self.fd, file_size)
         if buffer:
             os.lseek(self.fd, pos, os.SEEK_SET)
             os.write(self.fd, buffer)
 
-    def __getitem__(self, slice):
+    def __getitem__(self, slc):
         if self.fd is None:
             self.open_ro()
-        assert self.fd is not None
+        if self.fd is None:
-        os.lseek(self.fd, slice.start, os.SEEK_SET)
+            raise RuntimeError("file descriptor is not available for read")
-        size = slice.stop - slice.start
+        os.lseek(self.fd, slc.start, os.SEEK_SET)
+        size = slc.stop - slc.start
         data = os.read(self.fd, size)
         if len(data) < size:
             raise EOFError("Error reading requested range")
@@ -71,7 +75,7 @@ class FileServer:
     @staticmethod
     def _stat_size(path):
         try:
-            return os.stat(path).st_size
+            return Path(path).stat().st_size
         except FileNotFoundError:
             return None
 

cista/fileserver.py

@@ -1,12 +1,14 @@
 import asyncio
+import contextlib
 import mimetypes
 import os
 import re
 import shutil
 import xml.etree.ElementTree as ET
-from datetime import datetime, timezone
+from datetime import UTC, datetime
 from pathlib import Path, PurePosixPath
-from urllib.parse import quote as url_quote, unquote, urlparse
+from urllib.parse import quote as url_quote
+from urllib.parse import unquote, urlparse
 from wsgiref.handlers import format_date_time
 
 from sanic import Blueprint, HTTPResponse, empty, json
@@ -155,7 +157,7 @@ async def copy_or_move(request, name=""):
         raise NotFound("Files not found", context={"missing": missing})
 
     # Validate target shape/type before mutating anything.
-    for op_name, op_keys in (("cp", cp_keys), ("mv", mv_keys)):
+    for _op_name, op_keys in (("cp", cp_keys), ("mv", mv_keys)):
         if len(op_keys) > 1 and not dst_is_dir:
             raise BadRequest("Destination must be an existing directory for multiple keys")
         if not op_keys:
@@ -175,7 +177,7 @@ async def copy_or_move(request, name=""):
     changed: set[PurePosixPath] = set()
     completed: list[dict[str, str]] = []
 
-    class _FileOpFailed(Exception):
+    class _FileOpError(Exception):
         def __init__(self, op_name: str, key: str, error: Exception):
             self.op_name = op_name
             self.key = key
@@ -236,11 +238,11 @@ async def copy_or_move(request, name=""):
                     changed.add(dst_item_rel.parent)
                     completed.append({"op": op_name, "key": key})
                 except Exception as e:
-                    raise _FileOpFailed(op_name, key, e) from e
+                    raise _FileOpError(op_name, key, e) from e
 
     try:
         await asyncio.to_thread(_apply)
-    except _FileOpFailed as e:
+    except _FileOpError as e:
         raise BadRequest(
             "File operation failed after partial progress",
             context={
@@ -310,7 +312,7 @@ async def dav_copy(request, name=""):
     if not dest_header:
         raise BadRequest("Missing Destination header")
     overwrite = request.headers.get("overwrite", "T").strip().upper() != "F"
-    src_rel, src_abs = _safe_relpath(name)
+    _src_rel, src_abs = _safe_relpath(name)
     dst_rel, dst_abs = _parse_webdav_destination(dest_header)
     request.ctx._log_extra = f"→ {dst_rel}"
     if not src_abs.exists():
@@ -537,7 +539,7 @@ def _parse_webdav_destination(dest_header: str) -> tuple[PurePosixPath, Path]:
     return _safe_relpath(rel_str)
 
 
-def _rel_to_href(rel: PurePosixPath, is_dir: bool) -> str:
+def _rel_to_href(rel: PurePosixPath, *, is_dir: bool) -> str:
     """Build a DAV href from a storage-relative path."""
     parts = rel.parts
     if not parts:
@@ -560,10 +562,8 @@ def _collect_propfind_entries(rel: PurePosixPath, path: Path, depth: str) -> lis
     if depth == "1" and path.is_dir():
         for child in sorted(path.iterdir(), key=lambda p: (p.is_file(), p.name)):
             child_rel = rel / child.name if rel.parts else PurePosixPath(child.name)
-            try:
+            with contextlib.suppress(OSError):
                 entries.append(_propfind_entry(child_rel, child))
-            except OSError:
-                pass
     return entries
 
 
@@ -571,14 +571,14 @@ def _propfind_entry(rel: PurePosixPath, path: Path) -> dict:
     st = path.stat()
     is_dir = path.is_dir()
     return {
-        "href": _rel_to_href(rel, is_dir),
+        "href": _rel_to_href(rel, is_dir=is_dir),
         "name": rel.parts[-1] if rel.parts else "",
         "is_dir": is_dir,
         "size": st.st_size,
         "etag": f'"{st.st_mtime:.0f}-{st.st_size}"',
         "content_type": mimetypes.guess_type(path.name)[0] or "application/octet-stream",
         "last_modified": format_date_time(st.st_mtime),
-        "created": datetime.fromtimestamp(st.st_ctime, tz=timezone.utc).strftime(
+        "created": datetime.fromtimestamp(st.st_ctime, tz=UTC).strftime(
             "%Y-%m-%dT%H:%M:%SZ"
         ),
     }

cista/protocol.py

@@ -4,8 +4,6 @@ from typing import Any
 
 import msgspec
 
-from cista import config
-
 
 class ErrorMsg(msgspec.Struct):
     error: dict[str, Any]

cista/sanic_logging.py

@@ -232,7 +232,7 @@ def configure_access_logging() -> None:
 
 _LEVEL_EMOJI = {
     logging.DEBUG: "🔍",
-    logging.INFO: "ℹ️",
+    logging.INFO: "i",
     logging.WARNING: "⚠️",
     logging.ERROR: "🛑",
     logging.CRITICAL: "🛑",

cista/serve.py

@@ -12,7 +12,7 @@ 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)
+    _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"
     confdir = config.conffile.parent
@@ -21,14 +21,14 @@ def run(*, dev=False):
         server80.app.prepare(port=80, motd=False)
         domain = opts["host"]
         check_cert(confdir / domain, domain)
-        opts["ssl"] = str(confdir / domain)  # type: ignore
+        opts["ssl"] = str(confdir / domain)  # type: ignore[assignment]
     app.prepare(
         **opts,
         motd=False,
         dev=dev,
         auto_reload=dev,
         access_log=False,
-    )  # type: ignore
+    )  # type: ignore[call-arg]
     if dev:
         Sanic.serve()
     else:
@@ -38,7 +38,7 @@ def run(*, dev=False):
 def check_cert(certdir, domain):
     if (certdir / "privkey.pem").exist() and (certdir / "fullchain.pem").exists():
         return
-    # TODO: Use certbot to fetch a cert
+    # Certificate provisioning is external; files must exist before startup.
     raise ValueError(
         f"TLS certificate files privkey.pem and fullchain.pem needed in {certdir}",
     )

cista/session.py

@@ -31,9 +31,9 @@ def create(res, username, *, secure: bool = True, **kwargs):
 
 def update(res, s, *, secure: bool = True, **kwargs):
     s.update(kwargs)
-    s = jwt.encode(s, session_secret())
+    max_age = max(1, s["exp"] - int(time()))
-    max_age = max(1, s["exp"] - int(time()))  # type: ignore
+    token = jwt.encode(s, session_secret())
-    res.cookies.add_cookie("s", s, httponly=True, max_age=max_age, secure=secure)
+    res.cookies.add_cookie("s", token, httponly=True, max_age=max_age, secure=secure)
 
 
 def delete(res):

cista/sso.py

@@ -126,13 +126,12 @@ async def validate_sso_request(request, *, perm: str = "cista:login") -> dict |
                 context=error_data,
                 quiet=True,
             )
-        elif response.status_code == 403:
+        if response.status_code == 403:
             raise Forbidden(
                 error_data.get("detail", "Access denied"),
                 context=error_data,
                 quiet=True,
             )
-        else:
         detail = error_data.get("detail", "")
         logger.warning(
             f"SSO validation {url} returned {response.status_code}: {detail}"
@@ -149,7 +148,7 @@ async def validate_sso_request(request, *, perm: str = "cista:login") -> dict |
             "Authentication service unavailable",
             status_code=502,
             quiet=True,
-        )
+        ) from e
 
 
 async def check_permissions(user_id: str, perm: str) -> dict:
@@ -194,7 +193,6 @@ async def check_permissions(user_id: str, perm: str) -> dict:
                 error_data.get("detail", "Access denied"),
                 quiet=True,
             )
-        else:
         raise Forbidden(
             error_data.get("detail", "Permission check failed"),
             quiet=True,
@@ -206,7 +204,7 @@ async def check_permissions(user_id: str, perm: str) -> dict:
             "Authentication service unavailable",
             status_code=502,
             quiet=True,
-        )
+        ) from e
 
 
 async def proxy_auth_request(request):
@@ -324,15 +322,15 @@ async def proxy_auth_websocket(request, ws):
                 try:
                     async for message in ws:
                         await backend_ws.send(message)
-                except Exception:
+                except Exception as e:
-                    pass
+                    logger.debug("WebSocket forward_to_backend ended: %s", e)
 
             async def forward_to_client():
                 try:
                     async for message in backend_ws:
                         await ws.send(message)
-                except Exception:
+                except Exception as e:
-                    pass
+                    logger.debug("WebSocket forward_to_client ended: %s", e)
 
             await asyncio.gather(
                 forward_to_backend(),

cista/util/apphelpers.py

@@ -30,7 +30,7 @@ async def handle_sanic_exception(request, e):
         context = e.context or {}
         code = e.status_code
         headers = getattr(e, "headers", None)
-    if not message or not request.app.debug and code == 500:
+    if not message or (not request.app.debug and code == 500):
         message = "Internal Server Error"
     message = f"⚠️ {message}" if code < 500 else f"🛑 {message}"
     if code == 500:

cista/util/asynclink.py

@@ -40,7 +40,7 @@ class AsyncLink:
     async def stop(self):
         """Stop worker and clean up."""
         while not self.queue.empty():
-            command, future = self.queue.get_nowait()
+            _command, future = self.queue.get_nowait()
             if not future.done():
                 future.set_exception(Exception("AsyncLink stopped"))
             self.queue.task_done()

cista/util/lrucache.py

@@ -1,5 +1,5 @@
+from collections.abc import Callable
 from time import monotonic
-from typing import Callable
 
 
 class LRUCache:
@@ -7,22 +7,22 @@ class LRUCache:
     LRUCache is a least-recently-used (LRU) cache with expiry time.
 
     Attributes:
-        open (callable): Function to open a new handle.
+        opener (callable): Function to open a new handle.
         capacity (int): Max number of items in the cache.
         maxage (float): Max age for items in cache in seconds.
         cache (list): Internal list storing the cache items.
     """
 
-    def __init__(self, open: Callable, *, capacity: int, maxage: float):
+    def __init__(self, opener: Callable, *, capacity: int, maxage: float):
         """
         Initialize LRUCache.
 
         Args:
-            open (callable): Function to open a new handle.
+            opener (callable): Function to open a new handle.
             capacity (int): Maximum capacity of the cache.
             maxage (float): Max age for items in cache in seconds.
         """
-        self.open = open
+        self.opener = opener
         self.capacity = capacity
         self.maxage = maxage
         self.cache = []  # Each item is a tuple: (key, handle, timestamp), recent items first
@@ -47,7 +47,7 @@ class LRUCache:
                 self.cache.pop(i)
                 break
         else:
-            f = self.open(key)
+            f = self.opener(key)
         # Add/restore to end of cache
         self.cache.insert(0, (key, f, monotonic()))
         self.expire_items()
@@ -58,7 +58,7 @@ class LRUCache:
         Expire items that are either too old or exceed cache capacity.
         """
         ts = monotonic() - self.maxage
-        while len(self.cache) > self.capacity or self.cache and self.cache[-1][2] < ts:
+        while len(self.cache) > self.capacity or (self.cache and self.cache[-1][2] < ts):
             self.cache.pop()[1].close()
 
     def close(self):

cista/watching.py

@@ -141,8 +141,8 @@ def treeinspos(rootmod: list[FileEntry], relpath: PurePosixPath, relfile: int):
 
 
 state = State()
-rootpath: Path = None  # type: ignore
+rootpath: Path | None = None
-quit = threading.Event()
+stop_event = threading.Event()
 
 # Thread-safe queue for signaling path updates from websockets
 _update_queue: queue.Queue[PurePosixPath] = queue.Queue()
@@ -150,9 +150,8 @@ _update_queue: queue.Queue[PurePosixPath] = queue.Queue()
 
 def notify_change(*paths: PurePosixPath | str):
     """Signal that paths have changed. Called from control/upload websockets."""
-    for path in paths:
+    for raw_path in paths:
-        if isinstance(path, str):
+        path = PurePosixPath(raw_path) if isinstance(raw_path, str) else raw_path
-            path = PurePosixPath(path)
         # Convert absolute paths to relative (strip leading /)
         if path.is_absolute():
             path = (
@@ -192,10 +191,10 @@ def walk(rel: PurePosixPath, stat: stat_result | None = None) -> list[FileEntry]
         if isfile:
             return [entry]
         # Walk all entries of the directory
-        ret: list[FileEntry] = [...]  # type: ignore
+        ret: list[FileEntry] = []
         li = []
         for f in path.iterdir():
-            if quit.is_set():
+            if stop_event.is_set():
                 raise SystemExit("quit")
             if f.name.startswith("."):
                 continue  # No dotfiles
@@ -508,7 +507,7 @@ class PathIndex:
 
         if lo < len(children):
             return children[lo]
-        elif children:
+        if children:
             # Insert after last child's subtree
             last_idx = children[-1]
             last_entry = self.root[last_idx]
@@ -656,7 +655,7 @@ def watcher(loop):
             )
         )
 
-    while not quit.is_set():
+    while not stop_event.is_set():
         if use_inotify:
             import inotify.adapters
 
@@ -674,7 +673,11 @@ def watcher(loop):
         first_event_time: float | None = None
         last_event_time: float | None = None
 
-        def add_dirty(path: PurePosixPath, source: str) -> bool:
+        def add_dirty(
+            path: PurePosixPath,
+            source: str,
+            dirty_paths=dirty_paths,
+        ) -> bool:
             """Add path to dirty set. Returns True if added, False if redundant."""
             nonlocal first_event_time, last_event_time
             # Check if already covered by an existing dirty path
@@ -708,7 +711,7 @@ def watcher(loop):
             last_event_time = now
             return True
 
-        while not quit.is_set():
+        while not stop_event.is_set():
             now = time.monotonic()
 
             # Full refresh every 300s
@@ -779,7 +782,7 @@ def watcher(loop):
             # Collect inotify events if available (short timeout for responsiveness)
             if inotify_tree:
                 for event in inotify_tree.event_gen(yield_nones=False, timeout_s=0.05):
-                    if quit.is_set():
+                    if stop_event.is_set():
                         return
                     if not (modified_flags & set(event[1])):
                         continue
@@ -823,5 +826,5 @@ def start(app):
 
 
 def stop(app):
-    quit.set()
+    stop_event.set()
     app.ctx.watcher.join()

scripts/devserver.py

@@ -23,7 +23,12 @@ from pathlib import Path
 
 # Import devutil from scripts/fastapi-vue (not a package, so we adjust sys.path)
 sys.path.insert(0, str(Path(__file__).with_name("fastapi-vue")))
-from devutil import ProcessGroup, logger, ready, setup_vite  # type: ignore
+from devutil import (  # type: ignore[import-not-found]
+    ProcessGroup,
+    logger,
+    ready,
+    setup_vite,
+)
 
 from cista import config
 from cista.serve import parse_listen
@@ -40,13 +45,13 @@ def setup_sanic_backend(
     """
     config.load_config()
     listen = listen or config.config.listen or f":{DEFAULT_BACKEND_PORT}"
-    url, opts = parse_listen(listen)
+    _url, opts = parse_listen(listen)
     port = opts.get("port", DEFAULT_BACKEND_PORT)
     host = opts.get("host", "localhost") or "localhost"
 
     # Use the current interpreter/module path so devserver always runs
     # workspace source code instead of a potentially stale installed script.
-    cmd = [sys.executable, "-m", "cista", "--dev", "-l", listen] + extra_args
+    cmd = [sys.executable, "-m", "cista", "--dev", "-l", listen, *extra_args]
     return f"http://{host}:{port}", cmd
 
 
@@ -59,7 +64,7 @@ async def run_devserver(
         logger.warning("Frontend source not found at %s", front)
         raise SystemExit(1)
 
-    frontend_url, npm_install, vite = setup_vite(frontend or "")
+    _frontend_url, npm_install, vite = setup_vite(frontend or "")
     backend_url, sanic_cmd = setup_sanic_backend(backend, extra_args)
 
     # Tell vite where to proxy API requests

scripts/fastapi-vue/build-frontend.py

@@ -3,7 +3,9 @@
 import sys
 from pathlib import Path
 
-from hatchling.builders.hooks.plugin.interface import BuildHookInterface  # type: ignore
+from hatchling.builders.hooks.plugin.interface import (
+    BuildHookInterface,  # type: ignore[import-not-found]
+)
 
 sys.path.insert(0, str(Path(__file__).parent))
 from buildutil import build

scripts/fastapi-vue/buildutil.py

@@ -30,8 +30,11 @@ def _check_node_version(node_path: str) -> None:
     Raises RuntimeError if version is too old or cannot be determined.
     """
     try:
-        result = subprocess.run(
+        result = subprocess.run(  # noqa: S603
-            [node_path, "--version"], capture_output=True, text=True, check=True
+            [node_path, "--version"],
+            capture_output=True,
+            text=True,
+            check=True,
         )
         version_str = result.stdout.strip()
         # Parse version like "v20.10.0" or "v18.17.1"
@@ -176,16 +179,16 @@ def build(folder: str = "frontend") -> None:
         install_cmd, build_cmd = find_build_tool()
     except RuntimeError as e:
         logger.warning(e)
-        raise SystemExit(1)
+        raise SystemExit(1) from e
 
     def run(cmd):
         display_cmd = [Path(cmd[0]).name, *cmd[1:]]
         logger.info("### %s", " ".join(display_cmd))
-        subprocess.run(cmd, check=True, cwd=folder)
+        subprocess.run(cmd, check=True, cwd=folder)  # noqa: S603
 
     try:
         run(install_cmd)
         logger.info("")
         run(build_cmd)
-    except subprocess.CalledProcessError:
+    except subprocess.CalledProcessError as e:
-        raise SystemExit(1)
+        raise SystemExit(1) from e

scripts/fastapi-vue/devutil.py

@@ -1,6 +1,7 @@
 """Utilities meant for devserver script, used only in source repository with dev deps."""
 
 import asyncio
+import contextlib
 from pathlib import Path
 
 import httpx
@@ -58,10 +59,8 @@ class ProcessGroup:
         # Terminate remaining processes
         for p in self._procs:
             if p.returncode is None:
-                try:
+                with contextlib.suppress(ProcessLookupError):
                     p.terminate()
-                except ProcessLookupError:
-                    pass
 
         # Wait for all to finish (with overall timeout)
         still_running = [p for p in self._procs if p.returncode is None]
@@ -74,10 +73,8 @@ class ProcessGroup:
             except TimeoutError:
                 for p in self._procs:
                     if p.returncode is None:
-                        try:
+                        with contextlib.suppress(ProcessLookupError):
                             p.kill()
-                        except ProcessLookupError:
-                            pass
                         await p.wait()
 
 
@@ -95,10 +92,10 @@ async def ready(url: str, path: str = "") -> None:
                 await client.get(full_url, timeout=1.0)
                 logger.info("✓ Backend ready!")
                 return
-            except httpx.RequestError:
+            except httpx.RequestError as e:
                 if attempt == max_attempts - 1:
                     logger.warning("Backend didn't start in time")
-                    raise SystemExit(1)
+                    raise SystemExit(1) from e
                 await asyncio.sleep(0.1)
 
 

tests/test_files_auth.py

@@ -1,7 +1,6 @@
 import base64
 import hashlib
 import hmac
-import re
 import struct
 from pathlib import Path
 from time import time
@@ -94,7 +93,7 @@ def _session_cookie_header(username: str) -> dict[str, str]:
     return {"Cookie": f"s={token}"}
 
 
-@pytest.fixture()
+@pytest.fixture
 def setup_storage(tmp_path: Path):
     user = config.User()
     auth.set_password(user, "secret")

tests/test_files_path_security.py

@@ -10,7 +10,7 @@ from cista import config, watching
 from cista.fileserver import bp as fileserver_bp
 
 
-@pytest.fixture()
+@pytest.fixture
 def setup_storage(tmp_path: Path):
     config.config = config.Config(path=tmp_path, listen=":0", public=True)
     watching.state.root = []
@@ -33,7 +33,7 @@ async def client(setup_storage: Path):
 
 
 @pytest.mark.asyncio
-async def test_get_percent2F_decoded_as_path_separator(client, setup_storage: Path):
+async def test_get_percent2f_decoded_as_path_separator(client, setup_storage: Path):
     """%2F in the URL path is decoded to '/' and treated as a path separator."""
     (setup_storage / "sub").mkdir()
     (setup_storage / "sub" / "file.txt").write_text("hello", encoding="utf-8")
@@ -45,7 +45,7 @@ async def test_get_percent2F_decoded_as_path_separator(client, setup_storage: Pa
 
 
 @pytest.mark.asyncio
-async def test_mkcol_percent2F_creates_nested_directory(client, setup_storage: Path):
+async def test_mkcol_percent2f_creates_nested_directory(client, setup_storage: Path):
     """%2F in MKCOL path is decoded as a separator, creating nested dirs."""
     _, res = await client.request("MKCOL", "/files/parent%2Fchild")
 

tests/test_files_rest_api.py

@@ -10,7 +10,7 @@ from cista.fileserver import bp as fileserver_bp
 from cista.protocol import FileEntry
 
 
-@pytest.fixture()
+@pytest.fixture
 def setup_storage(tmp_path: Path):
     config.config = config.Config(path=tmp_path, listen=":0", public=True)
     watching.state.root = []

tests/test_files_static_streaming.py

@@ -9,7 +9,7 @@ from cista import config, watching
 from cista.fileserver import bp as fileserver_bp
 
 
-@pytest.fixture()
+@pytest.fixture
 def setup_storage(tmp_path: Path):
     config.config = config.Config(path=tmp_path, listen=":0", public=True)
     watching.state.root = []

tests/test_files_webdav.py

@@ -9,13 +9,12 @@ from sanic import Sanic
 
 from cista import config, watching
 from cista.fileserver import bp as fileserver_bp
-from cista.protocol import FileEntry
 
 _DAV_NS = "DAV:"
 _METHODS = ("MKCOL", "MOVE", "COPY", "PROPFIND")
 
 
-@pytest.fixture()
+@pytest.fixture
 def setup_storage(tmp_path: Path):
     config.config = config.Config(path=tmp_path, listen=":0", public=True)
     watching.state.root = []

tests/test_tokens.py

@@ -1,8 +1,6 @@
-from pathlib import Path
-from time import time
-from uuid import uuid4
-
 import os
+from pathlib import Path
+from uuid import uuid4
 
 import pytest
 import pytest_asyncio
@@ -13,9 +11,10 @@ from cista.auth import bp as auth_bp
 
 
 def _persist_config():
-    import msgspec
     from pathlib import PurePath
 
+    import msgspec
+
     def enc_hook(obj):
         if isinstance(obj, PurePath):
             return obj.as_posix()
@@ -25,7 +24,7 @@ def _persist_config():
     config.conffile.write_bytes(msgspec.toml.encode(raw))
 
 
-@pytest.fixture()
+@pytest.fixture
 def setup_storage(tmp_path: Path):
     os.environ["CISTA_HOME"] = str(tmp_path)
     config.init_confdir()