diff --git a/cista/api.py b/cista/api.py index 93aaa37..aa764b9 100644 --- a/cista/api.py +++ b/cista/api.py @@ -1,5 +1,4 @@ import asyncio -import typing from pathlib import PurePosixPath from secrets import token_bytes @@ -9,7 +8,7 @@ from sanic.exceptions import BadRequest from cista import __version__, auth, config, sso, watching from cista.fileio import FileServer -from cista.protocol import ControlTypes, FileRange, StatusMsg +from cista.protocol import ControlTypes, StatusMsg from cista.util.apphelpers import asend, websocket_wrapper bp = Blueprint("api", url_prefix="/api") @@ -26,65 +25,6 @@ async def stop_fileserver(app): await fileserver.stop() -@bp.websocket("upload") -@websocket_wrapper -async def upload(req, ws): - alink = fileserver.alink - while True: - req = None - text = await ws.recv() - if not isinstance(text, str): - raise ValueError( - f"Expected JSON control, got binary len(data) = {len(text)}", - ) - req = msgspec.json.decode(text, type=FileRange) - pos = req.start - while True: - data = await ws.recv() - if not isinstance(data, bytes): - break - if len(data) > req.end - pos: - raise ValueError( - f"Expected up to {req.end - pos} bytes, got {len(data)} bytes" - ) - sentsize = await alink(("upload", req.name, pos, data, req.size)) - pos += typing.cast(int, sentsize) - if pos >= req.end: - break - if pos != req.end: - d = f"{len(data)} bytes" if isinstance(data, bytes) else data - raise ValueError(f"Expected {req.end - pos} more bytes, got {d}") - # Signal the watcher about the uploaded file and its parent directories - path = PurePosixPath(req.name) - watching.notify_change(path, *path.parents) - # Report success - res = StatusMsg(status="ack", req=req) - await asend(ws, res) - - -@bp.websocket("download") -@websocket_wrapper -async def download(req, ws): - alink = fileserver.alink - while True: - req = None - text = await ws.recv() - if not isinstance(text, str): - raise ValueError( - f"Expected JSON control, got binary len(data) = {len(text)}", - ) - req = msgspec.json.decode(text, type=FileRange) - pos = req.start - while pos < req.end: - end = min(req.end, pos + (1 << 20)) - data = typing.cast(bytes, await alink(("download", req.name, pos, end))) - await asend(ws, data) - pos += len(data) - # Report success - res = StatusMsg(status="ack", req=req) - await asend(ws, res) - - @bp.websocket("control") @websocket_wrapper async def control(req, ws): diff --git a/cista/app.py b/cista/app.py index a2e8897..3c64b13 100644 --- a/cista/app.py +++ b/cista/app.py @@ -1,6 +1,7 @@ import asyncio import datetime import mimetypes +import re import time from concurrent.futures import ThreadPoolExecutor from multiprocessing import cpu_count @@ -11,8 +12,8 @@ from wsgiref.handlers import format_date_time import sanic.helpers from blake3 import blake3 -from sanic import Blueprint, Sanic, empty, raw, redirect -from sanic.exceptions import Forbidden, NotFound +from sanic import Blueprint, Sanic, empty, json, raw, redirect +from sanic.exceptions import BadRequest, Forbidden, NotFound from sanic.log import logger from setproctitle import setproctitle from stream_zip import ZIP_AUTO, stream_zip @@ -20,7 +21,7 @@ from zstandard import ZstdCompressor from cista import auth, config, preview, session, sso, watching from cista.preview import shutdown_preview_workers, start_preview_workers -from cista.api import bp +from cista.api import bp, fileserver from cista.sanic_logging import configure_access_logging, configure_main_logging, format_access_log from cista.sanic_logging import logger as access_logger from cista.util.apphelpers import handle_sanic_exception @@ -127,6 +128,68 @@ def http_fileserver(app): """Verify access to file server routes.""" await auth.verify(request) + @bp.put("/files/") + async def upload_file_chunk(request, *args, **kwargs): + body = request.body + header = request.headers.get("content-range") + if header: + start, end, total = _parse_content_range(header, len(body)) + else: + start = 0 + end = len(body) + total = end + raw_name = kwargs.get("name") + if raw_name is None and args: + raw_name = args[0] + if not isinstance(raw_name, str) or not raw_name: + prefix = "/files/" + if not request.path.startswith(prefix): + raise BadRequest("Invalid upload path") + raw_name = request.path[len(prefix) :] + rel_name = unquote(raw_name) + upload_info = await asyncio.to_thread( + fileserver.upload_info, + rel_name, + start, + body, + total, + ) + extras = [] + chunk_len = end - start + whole_file = start == 0 and end == total + if not whole_file: + start_mib = _to_mib_int(start) + chunk_mib = _to_mib_int(chunk_len) + # Keep range logs compact for fixed-size upload blocks. + if chunk_mib == 16: + extras.append(f"{start_mib}MiB") + else: + extras.append(f"{start_mib}+{chunk_mib}MiB") + if upload_info.get("created"): + extras.append(f"created {_to_mib_int(total)}MiB") + size_before = upload_info.get("size_before") + size_after = upload_info.get("size_after") + if ( + size_before is not None + and size_after is not None + and size_before != size_after + ): + extras.append("resized") + request.ctx._log_extra = " ".join(extras) if extras else None + path = PurePosixPath(rel_name) + watching.notify_change(path, *path.parents) + return json( + { + "status": "ack", + "req": { + "name": rel_name, + "size": total, + "start": start, + "end": end, + }, + } + ) + bp.static( "/files/", config.config.path, @@ -138,6 +201,30 @@ def http_fileserver(app): www = {} +_CONTENT_RANGE_RE = re.compile(r"^bytes (\d+)-(\d+)/(\d+)$") + + +def _parse_content_range(header: str, body_len: int) -> tuple[int, int, int]: + m = _CONTENT_RANGE_RE.fullmatch(header.strip()) + if m is None: + raise BadRequest("Invalid Content-Range format") + start, end_inclusive, total = (int(v) for v in m.groups()) + if total <= 0: + raise BadRequest("Invalid Content-Range total size") + if start > end_inclusive: + raise BadRequest("Invalid Content-Range range") + if end_inclusive >= total: + raise BadRequest("Content-Range exceeds total size") + expected_len = end_inclusive - start + 1 + if expected_len != body_len: + raise BadRequest( + f"Content length mismatch for range: expected {expected_len}, got {body_len}" + ) + return start, end_inclusive + 1, total + + +def _to_mib_int(value_bytes: int) -> int: + return round(value_bytes / (1 << 20)) def _load_wwwroot(www): diff --git a/cista/fileio.py b/cista/fileio.py index b9052f8..8b340da 100644 --- a/cista/fileio.py +++ b/cista/fileio.py @@ -1,9 +1,8 @@ -import asyncio import os +import threading from cista import config from cista.util import filename -from cista.util.asynclink import AsyncLink from cista.util.lrucache import LRUCache @@ -62,38 +61,32 @@ class File: class FileServer: async def start(self): - self.alink = AsyncLink() - self.worker = asyncio.get_event_loop().run_in_executor( - None, - self.worker_thread, - self.alink.to_sync, - ) self.cache = LRUCache(File, capacity=10, maxage=5.0) + self.cache_lock = threading.Lock() + self.file_locks: dict[str, threading.Lock] = {} async def stop(self): - await self.alink.stop() - await self.worker + self.cache.close() - def worker_thread(self, slink): + @staticmethod + def _stat_size(path): try: - for req in slink: - with req as (command, *args): - if command == "upload": - req.set_result(self.upload(*args)) - elif command == "download": - req.set_result(self.download(*args)) - else: - raise NotImplementedError(f"Unhandled {command=} {args}") - finally: - self.cache.close() + return os.stat(path).st_size + except FileNotFoundError: + return None - def upload(self, name, pos, data, file_size): + def upload_info(self, name, pos, data, file_size): name = filename.sanitize(name) - f = self.cache[name] - f.write(pos, data, file_size=file_size) - return len(data) - - def download(self, name, start, end): - name = filename.sanitize(name) - f = self.cache[name] - return f[start:end] + with self.cache_lock: + f = self.cache[name] + lock = self.file_locks.setdefault(name, threading.Lock()) + with lock: + size_before = self._stat_size(f.path) + f.write(pos, data, file_size=file_size) + size_after = self._stat_size(f.path) + return { + "written": len(data), + "created": size_before is None, + "size_before": size_before, + "size_after": size_after, + } diff --git a/cista/protocol.py b/cista/protocol.py index c04147a..220e887 100644 --- a/cista/protocol.py +++ b/cista/protocol.py @@ -12,7 +12,6 @@ from cista.util import filename ## Control commands - class ControlBase(msgspec.Struct, tag_field="op", tag=str.lower): def __call__(self): raise NotImplementedError @@ -118,19 +117,9 @@ class Cp(ControlBase): ControlTypes = MkDir | Rename | Rm | Mv | Cp -## File uploads and downloads - - -class FileRange(msgspec.Struct): - name: str - size: int - start: int - end: int - - class StatusMsg(msgspec.Struct): status: str - req: FileRange + req: Any class ErrorMsg(msgspec.Struct): diff --git a/frontend/src/components/UploadButton.vue b/frontend/src/components/UploadButton.vue index cce59e3..795e3e2 100644 --- a/frontend/src/components/UploadButton.vue +++ b/frontend/src/components/UploadButton.vue @@ -8,12 +8,11 @@