Replace WebSocket control API with REST file operations
- Add cista/fileserver.py: REST blueprint at /files with PUT upload, DELETE, MKCOL, POST cp/mv (combined), GET/HEAD static serving - Remove WS control handler and all ControlBase/Cmd protocol types - Frontend: SelectionToolbar, FileExplorer, Gallery now POST to /files instead of opening a control WebSocket per operation - Remove controlUrl export from WS.ts - Add tests: REST API, static streaming, path/escaping security - Catch ValueError from filename.sanitize and return 400 Bad Request
This commit is contained in:
Leo Vasanko
+1
-13
@@ -8,8 +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, StatusMsg
|
||||
from cista.util.apphelpers import asend, websocket_wrapper
|
||||
from cista.util.apphelpers import websocket_wrapper
|
||||
|
||||
bp = Blueprint("api", url_prefix="/api")
|
||||
fileserver = FileServer()
|
||||
@@ -25,17 +24,6 @@ async def stop_fileserver(app):
|
||||
await fileserver.stop()
|
||||
|
||||
|
||||
@bp.websocket("control")
|
||||
@websocket_wrapper
|
||||
async def control(req, ws):
|
||||
while True:
|
||||
cmd = msgspec.json.decode(await ws.recv(), type=ControlTypes)
|
||||
await asyncio.to_thread(cmd)
|
||||
# Signal the watcher about affected paths
|
||||
watching.notify_change(*cmd.affected_paths())
|
||||
await asend(ws, StatusMsg(status="ack", req=cmd))
|
||||
|
||||
|
||||
@bp.websocket("watch")
|
||||
@websocket_wrapper
|
||||
async def watch(req, ws):
|
||||
|
||||
+24
-116
@@ -1,19 +1,16 @@
|
||||
import asyncio
|
||||
import datetime
|
||||
import mimetypes
|
||||
import re
|
||||
import time
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
from multiprocessing import cpu_count
|
||||
from pathlib import Path, PurePath, PurePosixPath
|
||||
from stat import S_IFDIR, S_IFREG
|
||||
from urllib.parse import unquote
|
||||
from wsgiref.handlers import format_date_time
|
||||
|
||||
import sanic.helpers
|
||||
from blake3 import blake3
|
||||
from sanic import Blueprint, Sanic, empty, json, raw, redirect
|
||||
from sanic.exceptions import BadRequest, Forbidden, NotFound
|
||||
from sanic import Sanic, empty, raw, redirect
|
||||
from sanic.exceptions import Forbidden, NotFound
|
||||
from sanic.log import logger
|
||||
from setproctitle import setproctitle
|
||||
from stream_zip import ZIP_AUTO, stream_zip
|
||||
@@ -21,17 +18,30 @@ 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, fileserver
|
||||
from cista.sanic_logging import configure_access_logging, configure_main_logging, format_access_log
|
||||
from cista.api import bp
|
||||
from cista import 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
|
||||
|
||||
# Workaround until Sanic PR #2824 is merged
|
||||
sanic.helpers._ENTITY_HEADERS = frozenset()
|
||||
|
||||
configure_access_logging()
|
||||
|
||||
app = Sanic("cista", strict_slashes=True)
|
||||
app.router.ALLOWED_METHODS = (
|
||||
*app.router.ALLOWED_METHODS,
|
||||
"MKCOL",
|
||||
"MOVE",
|
||||
"COPY",
|
||||
"PROPFIND",
|
||||
"PROPPATCH",
|
||||
"LOCK",
|
||||
"UNLOCK",
|
||||
)
|
||||
|
||||
configure_main_logging()
|
||||
# Register either SSO proxy or built-in auth routes based on PASKIA_BACKEND_URL
|
||||
if sso.paskia_enabled():
|
||||
@@ -40,6 +50,7 @@ else:
|
||||
app.blueprint(auth.bp) # Built-in auth routes
|
||||
app.blueprint(preview.bp)
|
||||
app.blueprint(bp)
|
||||
app.blueprint(fileserver.bp)
|
||||
app.exception(Exception)(handle_sanic_exception)
|
||||
|
||||
|
||||
@@ -106,7 +117,9 @@ async def log_access(req, res):
|
||||
qs = qs.decode(errors="replace")
|
||||
path = f"{path}?{qs}"
|
||||
extra = getattr(req.ctx, "_log_extra", None)
|
||||
line = format_access_log(client, res.status, req.method, host, path, duration_ms, extra=extra)
|
||||
line = format_access_log(
|
||||
client, res.status, req.method, host, path, duration_ms, extra=extra
|
||||
)
|
||||
access_logger.info(line)
|
||||
return res
|
||||
|
||||
@@ -119,112 +132,7 @@ async def forward_sso_cookies(req, res):
|
||||
res.headers.add("set-cookie", cookie)
|
||||
|
||||
|
||||
@app.before_server_start
|
||||
def http_fileserver(app):
|
||||
bp = Blueprint("fileserver")
|
||||
|
||||
@bp.on_request
|
||||
async def verify_fileserver(request):
|
||||
"""Verify access to file server routes."""
|
||||
await auth.verify(request)
|
||||
|
||||
@bp.put("/files/<name:path>")
|
||||
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,
|
||||
use_content_range=True,
|
||||
stream_large_files=True,
|
||||
directory_view=True,
|
||||
)
|
||||
app.blueprint(bp)
|
||||
|
||||
|
||||
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):
|
||||
|
||||
@@ -0,0 +1,415 @@
|
||||
import asyncio
|
||||
import mimetypes
|
||||
import os
|
||||
import re
|
||||
import shutil
|
||||
from pathlib import Path, PurePosixPath
|
||||
from urllib.parse import unquote
|
||||
from wsgiref.handlers import format_date_time
|
||||
|
||||
from sanic import Blueprint, empty, json
|
||||
from sanic.exceptions import BadRequest, NotFound
|
||||
|
||||
from cista import auth, config, watching
|
||||
from cista.api import fileserver
|
||||
from cista.util import filename
|
||||
|
||||
bp = Blueprint("fileserver", url_prefix="/files")
|
||||
|
||||
_CONTENT_RANGE_RE = re.compile(r"^bytes (\d+)-(\d+)/(\d+)$")
|
||||
_RANGE_RE = re.compile(r"^bytes=(\d*)-(\d*)$")
|
||||
_FILE_CHUNK_SIZE = 1 << 20
|
||||
|
||||
|
||||
@bp.on_request
|
||||
async def verify_fileserver(request):
|
||||
"""Verify access to file server routes."""
|
||||
await auth.verify(request)
|
||||
|
||||
|
||||
@bp.put("/<name:path>")
|
||||
async def upload_file_chunk(request, name):
|
||||
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
|
||||
|
||||
rel, _ = _safe_relpath(name)
|
||||
rel_name = rel.as_posix()
|
||||
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
|
||||
watching.notify_change(rel, *rel.parents)
|
||||
return json(
|
||||
{
|
||||
"status": "ack",
|
||||
"req": {
|
||||
"name": rel_name,
|
||||
"size": total,
|
||||
"start": start,
|
||||
"end": end,
|
||||
},
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
@bp.delete("/<name:path>")
|
||||
async def delete_file(request, name):
|
||||
rel, path = _safe_relpath(name)
|
||||
if not rel.parts:
|
||||
raise BadRequest("Refusing to delete root folder")
|
||||
|
||||
def _delete():
|
||||
if not path.exists():
|
||||
raise NotFound(f"File not found: {name}")
|
||||
if path.is_dir():
|
||||
shutil.rmtree(path)
|
||||
else:
|
||||
path.unlink()
|
||||
|
||||
await asyncio.to_thread(_delete)
|
||||
watching.notify_change(rel, *rel.parents)
|
||||
return empty(status=204)
|
||||
|
||||
|
||||
@bp.route("/<name:path>", methods=["MKCOL"])
|
||||
async def create_folder(request, name):
|
||||
rel, path = _safe_relpath(name)
|
||||
if not rel.parts:
|
||||
raise BadRequest("Refusing to create root folder")
|
||||
await asyncio.to_thread(path.mkdir, parents=True, exist_ok=False)
|
||||
watching.notify_change(rel, *rel.parents)
|
||||
return empty(status=201)
|
||||
|
||||
|
||||
@bp.post("/<name:path>")
|
||||
async def copy_or_move(request, name=""):
|
||||
provided_args = set(request.args.keys())
|
||||
if not provided_args:
|
||||
raise BadRequest("No query arguments passed")
|
||||
|
||||
allowed_args = {"cp", "mv"}
|
||||
unknown_args = sorted(provided_args - allowed_args)
|
||||
if unknown_args:
|
||||
raise BadRequest(f"Unknown query parameter(s): {', '.join(unknown_args)}")
|
||||
|
||||
mv_vals = request.args.getlist("mv")
|
||||
cp_vals = request.args.getlist("cp")
|
||||
|
||||
mv_keys: list[str] = []
|
||||
for value in mv_vals:
|
||||
mv_keys.extend(k for k in value.split() if k)
|
||||
|
||||
cp_keys: list[str] = []
|
||||
for value in cp_vals:
|
||||
cp_keys.extend(k for k in value.split() if k)
|
||||
|
||||
if not mv_keys and not cp_keys:
|
||||
raise BadRequest("No keys given")
|
||||
|
||||
dst_rel, dst_abs = _safe_relpath(name)
|
||||
|
||||
dst_exists = dst_abs.exists()
|
||||
dst_is_dir = dst_exists and dst_abs.is_dir()
|
||||
|
||||
ordered_keys = cp_keys + mv_keys
|
||||
key_paths = _get_key_paths(set(ordered_keys))
|
||||
missing = [key for key in ordered_keys if key not in key_paths]
|
||||
if missing:
|
||||
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)):
|
||||
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:
|
||||
continue
|
||||
if not dst_is_dir:
|
||||
if not dst_rel.parts:
|
||||
raise BadRequest("Destination file path is required")
|
||||
parent_abs = dst_abs.parent
|
||||
if not parent_abs.is_dir():
|
||||
raise BadRequest("Destination parent folder does not exist")
|
||||
if dst_exists and dst_abs.is_file():
|
||||
for key in op_keys:
|
||||
src_abs = _resolve_from_relpath(key_paths[key])
|
||||
if src_abs.is_dir():
|
||||
raise BadRequest("Cannot move/copy a directory to an existing file")
|
||||
|
||||
changed: set[PurePosixPath] = set()
|
||||
completed: list[dict[str, str]] = []
|
||||
|
||||
class _FileOpFailed(Exception):
|
||||
def __init__(self, op_name: str, key: str, error: Exception):
|
||||
self.op_name = op_name
|
||||
self.key = key
|
||||
self.error = error
|
||||
super().__init__(str(error))
|
||||
|
||||
def _apply():
|
||||
for op_name, op_keys in (("cp", cp_keys), ("mv", mv_keys)):
|
||||
op_multi = len(op_keys) > 1
|
||||
for key in op_keys:
|
||||
try:
|
||||
src_rel = key_paths[key]
|
||||
src_abs = _resolve_from_relpath(src_rel)
|
||||
|
||||
if op_multi:
|
||||
if not dst_is_dir:
|
||||
raise BadRequest(
|
||||
"Destination must be an existing directory for multiple keys"
|
||||
)
|
||||
dst_item_rel = (
|
||||
dst_rel / src_rel.name if dst_rel.parts else PurePosixPath(src_rel.name)
|
||||
)
|
||||
elif dst_is_dir:
|
||||
dst_item_rel = (
|
||||
dst_rel / src_rel.name if dst_rel.parts else PurePosixPath(src_rel.name)
|
||||
)
|
||||
else:
|
||||
if not dst_rel.parts:
|
||||
raise BadRequest("Destination file path is required")
|
||||
parent_abs = dst_abs.parent
|
||||
if not parent_abs.is_dir():
|
||||
raise BadRequest("Destination parent folder does not exist")
|
||||
if src_abs.is_dir() and dst_exists and dst_abs.is_file():
|
||||
raise BadRequest(
|
||||
"Cannot move/copy a directory to an existing file"
|
||||
)
|
||||
dst_item_rel = dst_rel
|
||||
|
||||
dst_item_abs = _resolve_from_relpath(dst_item_rel)
|
||||
|
||||
if op_name == "mv":
|
||||
# A no-op rename should still return success.
|
||||
if src_abs != dst_item_abs:
|
||||
shutil.move(src_abs, dst_item_abs)
|
||||
changed.add(src_rel)
|
||||
changed.add(src_rel.parent)
|
||||
elif src_abs.is_dir():
|
||||
shutil.copytree(
|
||||
src_abs,
|
||||
dst_item_abs,
|
||||
dirs_exist_ok=True,
|
||||
ignore_dangling_symlinks=True,
|
||||
)
|
||||
else:
|
||||
shutil.copy2(src_abs, dst_item_abs)
|
||||
|
||||
changed.add(dst_item_rel)
|
||||
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
|
||||
|
||||
try:
|
||||
await asyncio.to_thread(_apply)
|
||||
except _FileOpFailed as e:
|
||||
raise BadRequest(
|
||||
"File operation failed after partial progress",
|
||||
context={
|
||||
"failed_op": e.op_name,
|
||||
"failed_key": e.key,
|
||||
"error": str(e.error),
|
||||
"completed": completed,
|
||||
},
|
||||
) from e
|
||||
|
||||
notify_paths = [p for p in changed if p.parts]
|
||||
if notify_paths:
|
||||
watching.notify_change(*notify_paths)
|
||||
|
||||
return json(
|
||||
{
|
||||
"status": "ack",
|
||||
"counts": {"cp": len(cp_keys), "mv": len(mv_keys)},
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
@bp.get("/<name:path>")
|
||||
async def get_file(request, name=""):
|
||||
return await _send_static_file(request, name, head_only=False)
|
||||
|
||||
|
||||
@bp.head("/<name:path>")
|
||||
async def head_file(request, name=""):
|
||||
return await _send_static_file(request, name, head_only=True)
|
||||
|
||||
|
||||
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 _safe_relpath(path: str) -> tuple[PurePosixPath, Path]:
|
||||
"""Resolve a user path under storage root and enforce containment."""
|
||||
base = config.config.path.resolve()
|
||||
try:
|
||||
sanitized = filename.sanitize(unquote(path))
|
||||
except ValueError as e:
|
||||
raise BadRequest(f"Invalid path: {e}") from e
|
||||
resolved = (base / sanitized).resolve()
|
||||
if not resolved.is_relative_to(base):
|
||||
raise BadRequest("Invalid path")
|
||||
rel = PurePosixPath(resolved.relative_to(base).as_posix())
|
||||
return rel, resolved
|
||||
|
||||
|
||||
def _resolve_from_relpath(rel: PurePosixPath) -> Path:
|
||||
"""Resolve a relative path under storage root and enforce containment."""
|
||||
base = config.config.path.resolve()
|
||||
resolved = (base / rel).resolve()
|
||||
if not resolved.is_relative_to(base):
|
||||
raise BadRequest("Invalid path")
|
||||
return resolved
|
||||
|
||||
|
||||
async def _send_static_file(request, name: str, *, head_only: bool):
|
||||
_, path = _safe_relpath(name)
|
||||
|
||||
st = await asyncio.to_thread(path.stat)
|
||||
if path.is_dir():
|
||||
raise NotFound(f"Not a file: {name}")
|
||||
|
||||
size = st.st_size
|
||||
start = 0
|
||||
end_excl = size
|
||||
status = 200
|
||||
|
||||
range_header = request.headers.get("range")
|
||||
if range_header is not None:
|
||||
parsed = _parse_range_header(range_header, size)
|
||||
if parsed is None:
|
||||
return empty(
|
||||
status=416,
|
||||
headers={
|
||||
"accept-ranges": "bytes",
|
||||
"content-range": f"bytes */{size}",
|
||||
},
|
||||
)
|
||||
start, end_excl = parsed
|
||||
status = 206
|
||||
|
||||
length = end_excl - start
|
||||
mime = mimetypes.guess_type(path.name)[0] or "application/octet-stream"
|
||||
headers = {
|
||||
"accept-ranges": "bytes",
|
||||
"cache-control": "no-cache",
|
||||
"content-length": str(length),
|
||||
"content-type": mime,
|
||||
"last-modified": format_date_time(st.st_mtime),
|
||||
}
|
||||
if status == 206:
|
||||
headers["content-range"] = f"bytes {start}-{end_excl - 1}/{size}"
|
||||
|
||||
if head_only:
|
||||
return empty(status=status, headers=headers)
|
||||
|
||||
res = await request.respond(status=status, headers=headers)
|
||||
fd = await asyncio.to_thread(os.open, path, os.O_RDONLY)
|
||||
try:
|
||||
pos = start
|
||||
while pos < end_excl:
|
||||
chunk = await asyncio.to_thread(
|
||||
os.pread,
|
||||
fd,
|
||||
min(_FILE_CHUNK_SIZE, end_excl - pos),
|
||||
pos,
|
||||
)
|
||||
if not chunk:
|
||||
break
|
||||
pos += len(chunk)
|
||||
await res.send(chunk)
|
||||
finally:
|
||||
await asyncio.to_thread(os.close, fd)
|
||||
|
||||
|
||||
def _parse_range_header(header: str, size: int) -> tuple[int, int] | None:
|
||||
value = header.strip()
|
||||
if "," in value:
|
||||
return None
|
||||
m = _RANGE_RE.fullmatch(value)
|
||||
if m is None:
|
||||
return None
|
||||
|
||||
start_s, end_s = m.groups()
|
||||
if not start_s and not end_s:
|
||||
return None
|
||||
|
||||
if start_s:
|
||||
start = int(start_s)
|
||||
if start >= size:
|
||||
return None
|
||||
end_inclusive = int(end_s) if end_s else (size - 1)
|
||||
if end_inclusive < start:
|
||||
return None
|
||||
end_inclusive = min(end_inclusive, size - 1)
|
||||
return start, end_inclusive + 1
|
||||
|
||||
suffix_len = int(end_s)
|
||||
if suffix_len <= 0:
|
||||
return None
|
||||
if suffix_len >= size:
|
||||
return 0, size
|
||||
start = size - suffix_len
|
||||
return start, size
|
||||
|
||||
|
||||
def _get_key_paths(wanted: set[str]) -> dict[str, PurePosixPath]:
|
||||
"""Map file keys to their current relative filesystem paths."""
|
||||
loc = PurePosixPath()
|
||||
ret: dict[str, PurePosixPath] = {}
|
||||
with watching.state.lock:
|
||||
root = watching.state.root
|
||||
for f in root:
|
||||
loc = PurePosixPath(*loc.parts[: f.level - 1]) / f.name
|
||||
if f.key in wanted and f.key not in ret:
|
||||
ret[f.key] = loc
|
||||
if len(ret) == len(wanted):
|
||||
break
|
||||
return ret
|
||||
+2
-6
@@ -221,9 +221,7 @@ class _PreviewWorkerPool:
|
||||
logger.warning(
|
||||
"Preview worker protocol failure for %s: %s", filepath.name, e
|
||||
)
|
||||
raise PreviewError(
|
||||
f"worker protocol failure for {filepath.name}: {e}"
|
||||
)
|
||||
raise PreviewError(f"worker protocol failure for {filepath.name}: {e}")
|
||||
finally:
|
||||
if replace:
|
||||
await self._replace_worker(worker)
|
||||
@@ -470,9 +468,7 @@ def process_pdf(path, *, maxsize, maxzoom, quality, page_number=0):
|
||||
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
|
||||
)
|
||||
ret = img.write_to_buffer(".avif", Q=quality, effort=AVIF_FAST_EFFORT, strip=True)
|
||||
backend = "pdf+pyvips"
|
||||
t_save_end = perf_counter()
|
||||
|
||||
|
||||
@@ -1,125 +1,10 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import shutil
|
||||
from pathlib import PurePosixPath
|
||||
from typing import Any
|
||||
|
||||
import msgspec
|
||||
from sanic import BadRequest
|
||||
|
||||
from cista import config
|
||||
from cista.util import filename
|
||||
|
||||
## Control commands
|
||||
|
||||
class ControlBase(msgspec.Struct, tag_field="op", tag=str.lower):
|
||||
def __call__(self):
|
||||
raise NotImplementedError
|
||||
|
||||
def affected_paths(self) -> list[str]:
|
||||
"""Return list of paths affected by this operation for change notification."""
|
||||
return []
|
||||
|
||||
|
||||
class MkDir(ControlBase):
|
||||
path: str
|
||||
|
||||
def __call__(self):
|
||||
path = config.config.path / filename.sanitize(self.path)
|
||||
path.mkdir(parents=True, exist_ok=False)
|
||||
|
||||
def affected_paths(self) -> list[str]:
|
||||
return [filename.sanitize(self.path)]
|
||||
|
||||
|
||||
class Rename(ControlBase):
|
||||
path: str
|
||||
to: str
|
||||
|
||||
def __call__(self):
|
||||
to = filename.sanitize(self.to)
|
||||
if "/" in to:
|
||||
raise BadRequest("Rename 'to' name should only contain filename, not path")
|
||||
path = config.config.path / filename.sanitize(self.path)
|
||||
path.rename(path.with_name(to))
|
||||
|
||||
def affected_paths(self) -> list[str]:
|
||||
sanitized = filename.sanitize(self.path)
|
||||
new_path = str(PurePosixPath(sanitized).with_name(filename.sanitize(self.to)))
|
||||
return [sanitized, new_path]
|
||||
|
||||
|
||||
class Rm(ControlBase):
|
||||
sel: list[str]
|
||||
|
||||
def __call__(self):
|
||||
root = config.config.path
|
||||
sel = [root / filename.sanitize(p) for p in self.sel]
|
||||
for p in sel:
|
||||
if p.is_dir():
|
||||
shutil.rmtree(p)
|
||||
else:
|
||||
p.unlink()
|
||||
|
||||
def affected_paths(self) -> list[str]:
|
||||
return [filename.sanitize(p) for p in self.sel]
|
||||
|
||||
|
||||
class Mv(ControlBase):
|
||||
sel: list[str]
|
||||
dst: str
|
||||
|
||||
def __call__(self):
|
||||
root = config.config.path
|
||||
sel = [root / filename.sanitize(p) for p in self.sel]
|
||||
dst = root / filename.sanitize(self.dst)
|
||||
if not dst.is_dir():
|
||||
raise BadRequest("The destination must be a directory")
|
||||
for p in sel:
|
||||
shutil.move(p, dst)
|
||||
|
||||
def affected_paths(self) -> list[str]:
|
||||
dst = filename.sanitize(self.dst)
|
||||
paths = [filename.sanitize(p) for p in self.sel]
|
||||
# Include new locations in dst
|
||||
paths.extend(f"{dst}/{PurePosixPath(p).name}" for p in self.sel)
|
||||
return paths
|
||||
|
||||
|
||||
class Cp(ControlBase):
|
||||
sel: list[str]
|
||||
dst: str
|
||||
|
||||
def __call__(self):
|
||||
root = config.config.path
|
||||
sel = [root / filename.sanitize(p) for p in self.sel]
|
||||
dst = root / filename.sanitize(self.dst)
|
||||
if not dst.is_dir():
|
||||
raise BadRequest("The destination must be a directory")
|
||||
for p in sel:
|
||||
if p.is_dir():
|
||||
# Note: copies as dst rather than in dst unless name is appended.
|
||||
shutil.copytree(
|
||||
p,
|
||||
dst / p.name,
|
||||
dirs_exist_ok=True,
|
||||
ignore_dangling_symlinks=True,
|
||||
)
|
||||
else:
|
||||
shutil.copy2(p, dst)
|
||||
|
||||
def affected_paths(self) -> list[str]:
|
||||
dst = filename.sanitize(self.dst)
|
||||
# Only destinations are new (sources unchanged)
|
||||
return [f"{dst}/{PurePosixPath(filename.sanitize(p)).name}" for p in self.sel]
|
||||
|
||||
|
||||
ControlTypes = MkDir | Rename | Rm | Mv | Cp
|
||||
|
||||
|
||||
class StatusMsg(msgspec.Struct):
|
||||
status: str
|
||||
req: Any
|
||||
|
||||
|
||||
class ErrorMsg(msgspec.Struct):
|
||||
|
||||
+20
-13
@@ -8,18 +8,18 @@ from ipaddress import IPv6Address
|
||||
logger = logging.getLogger("cista.access")
|
||||
|
||||
_RESET = "\033[0m"
|
||||
_STATUS_INFO = "\033[32m" # 1xx (green)
|
||||
_STATUS_OK = "\033[1;92m" # 2xx (bright green)
|
||||
_STATUS_INFO = "\033[32m" # 1xx (green)
|
||||
_STATUS_OK = "\033[1;92m" # 2xx (bright green)
|
||||
_STATUS_REDIRECT = "\033[32m" # 3xx (green)
|
||||
_STATUS_CLIENT_ERR = "\033[0;31m" # 4xx (red)
|
||||
_STATUS_SERVER_ERR = "\033[1;91m" # 5xx (bold bright red)
|
||||
_METHOD_READ = "\033[0;34m" # GET, HEAD, OPTIONS (blue)
|
||||
_METHOD_WRITE = "\033[1;94m" # POST, PUT, DELETE, PATCH (bold bright blue)
|
||||
_HOST = "\033[38;5;242m" # hostname (dark grey)
|
||||
_PATH = "\033[38;5;250m" # path (light grey)
|
||||
_TIMING = "\033[38;5;242m" # timing (dark grey)
|
||||
_WS_OPEN = "\033[1;93m" # WebSocket connect (bold bright yellow)
|
||||
_WS_CLOSE = "\033[33m" # WebSocket disconnect (yellow)
|
||||
_STATUS_CLIENT_ERR = "\033[0;31m" # 4xx (red)
|
||||
_STATUS_SERVER_ERR = "\033[1;91m" # 5xx (bold bright red)
|
||||
_METHOD_READ = "\033[0;34m" # GET, HEAD, OPTIONS (blue)
|
||||
_METHOD_WRITE = "\033[1;94m" # POST, PUT, DELETE, PATCH (bold bright blue)
|
||||
_HOST = "\033[38;5;242m" # hostname (dark grey)
|
||||
_PATH = "\033[38;5;250m" # path (light grey)
|
||||
_TIMING = "\033[38;5;242m" # timing (dark grey)
|
||||
_WS_OPEN = "\033[1;93m" # WebSocket connect (bold bright yellow)
|
||||
_WS_CLOSE = "\033[33m" # WebSocket disconnect (yellow)
|
||||
_WS_STATUS = "\033[38;5;250m" # WebSocket close status (normal white)
|
||||
|
||||
|
||||
@@ -113,7 +113,12 @@ def _format_method_label(label: str, *, color: str | None = None) -> str:
|
||||
|
||||
|
||||
def format_access_log(
|
||||
client: str, status: int, method: str, host: str, path: str, duration_ms: float,
|
||||
client: str,
|
||||
status: int,
|
||||
method: str,
|
||||
host: str,
|
||||
path: str,
|
||||
duration_ms: float,
|
||||
extra: str | None = None,
|
||||
) -> str:
|
||||
ip = _format_left(format_client_ip(client))
|
||||
@@ -123,7 +128,9 @@ def format_access_log(
|
||||
path_str = f"{_PATH}{path}{_RESET}"
|
||||
timing_str = f"{_TIMING}{format_duration_ms(duration_ms)}{_RESET}"
|
||||
extra_str = f" {_TIMING}{extra}{_RESET}" if extra else ""
|
||||
return f"{ip} {status_str} {method_str} {host_str}{path_str}{extra_str} {timing_str}"
|
||||
return (
|
||||
f"{ip} {status_str} {method_str} {host_str}{path_str}{extra_str} {timing_str}"
|
||||
)
|
||||
|
||||
|
||||
_ws_counter = 1
|
||||
|
||||
@@ -76,7 +76,7 @@ import { ref, computed, watchEffect, shallowRef, onMounted, onUnmounted, nextTic
|
||||
import { useMainStore } from '@/stores/main'
|
||||
import { Doc } from '@/repositories/Document'
|
||||
import FileRenameInput from './FileRenameInput.vue'
|
||||
import { connect, controlUrl } from '@/repositories/WS'
|
||||
import { apiFetch } from '@/repositories/Client'
|
||||
import { formatSize } from '@/utils'
|
||||
import { useRouter } from 'vue-router'
|
||||
import ContextMenu from '@imengyu/vue3-context-menu'
|
||||
@@ -87,31 +87,36 @@ const props = defineProps<{
|
||||
}>()
|
||||
const store = useMainStore()
|
||||
const router = useRouter()
|
||||
|
||||
const filesUrl = (path: string) =>
|
||||
'/files/' + path.split('/').map(part => encodeURIComponent(part)).join('/')
|
||||
|
||||
const parseErrorMessage = async (res: Response) => {
|
||||
try {
|
||||
const data = await res.json()
|
||||
return data.message || data.detail || `${res.status} ${res.statusText}`
|
||||
} catch {
|
||||
return `${res.status} ${res.statusText}`
|
||||
}
|
||||
}
|
||||
|
||||
// File rename
|
||||
const editing = shallowRef<Doc | null>(null)
|
||||
const rename = (doc: Doc, newName: string) => {
|
||||
const rename = async (doc: Doc, newName: string) => {
|
||||
const oldName = doc.name
|
||||
const control = connect(controlUrl, {
|
||||
message(ev: MessageEvent) {
|
||||
const msg = JSON.parse(ev.data)
|
||||
if ('error' in msg) {
|
||||
console.error('Rename failed', msg.error.message, msg.error)
|
||||
doc.name = oldName
|
||||
} else {
|
||||
console.log('Rename succeeded', msg)
|
||||
}
|
||||
}
|
||||
})
|
||||
control.onopen = () => {
|
||||
control.send(
|
||||
JSON.stringify({
|
||||
op: 'rename',
|
||||
path: `${doc.loc}/${oldName}`,
|
||||
to: newName
|
||||
})
|
||||
)
|
||||
}
|
||||
doc.name = newName // We should get an update from watch but this is quicker
|
||||
try {
|
||||
const dstUrl = doc.loc ? filesUrl(doc.loc) : '/files/'
|
||||
const res = await apiFetch(
|
||||
`${dstUrl}?mv=${doc.key}&to=${encodeURIComponent(newName)}`,
|
||||
{ method: 'POST' }
|
||||
)
|
||||
if (!res.ok) throw new Error(await parseErrorMessage(res))
|
||||
} catch (err) {
|
||||
console.error('Rename failed', err)
|
||||
doc.name = oldName
|
||||
store.showToast(err instanceof Error ? err.message : 'Rename failed')
|
||||
}
|
||||
}
|
||||
defineExpose({
|
||||
newFolder() {
|
||||
@@ -253,31 +258,20 @@ onMounted(() => {
|
||||
}
|
||||
})
|
||||
onUnmounted(() => { clearInterval(modifiedTimer) })
|
||||
const mkdir = (doc: Doc, name: string) => {
|
||||
const control = connect(controlUrl, {
|
||||
open() {
|
||||
control.send(
|
||||
JSON.stringify({
|
||||
op: 'mkdir',
|
||||
path: `${doc.loc}/${name}`
|
||||
})
|
||||
)
|
||||
},
|
||||
message(ev: MessageEvent) {
|
||||
const msg = JSON.parse(ev.data)
|
||||
if ('error' in msg) {
|
||||
console.error('Mkdir failed', msg.error.message, msg.error)
|
||||
editing.value = null
|
||||
} else {
|
||||
console.log('mkdir', msg)
|
||||
router.push(doc.urlrouter)
|
||||
}
|
||||
}
|
||||
})
|
||||
const mkdir = async (doc: Doc, name: string) => {
|
||||
doc.name = name
|
||||
doc.key = crypto.randomUUID()
|
||||
store.addGhost(doc)
|
||||
editing.value = null
|
||||
const path = doc.loc ? `${doc.loc}/${name}` : name
|
||||
try {
|
||||
const res = await apiFetch(filesUrl(path), { method: 'MKCOL' })
|
||||
if (!res.ok) throw new Error(await parseErrorMessage(res))
|
||||
router.push(doc.urlrouter)
|
||||
} catch (err) {
|
||||
console.error('Mkdir failed', err)
|
||||
store.showToast(err instanceof Error ? err.message : 'Mkdir failed')
|
||||
}
|
||||
}
|
||||
const showFolderBreadcrumb = (i: number) => {
|
||||
const docs = props.documents
|
||||
@@ -373,24 +367,17 @@ const copyImage = async (doc: Doc) => {
|
||||
}
|
||||
}
|
||||
|
||||
const deleteFile = (doc: Doc) => {
|
||||
const deleteFile = async (doc: Doc) => {
|
||||
const path = doc.loc ? `${doc.loc}/${doc.name}` : doc.name
|
||||
store.hideDoc(path)
|
||||
const control = connect(controlUrl, {
|
||||
message(ev: MessageEvent) {
|
||||
const res = JSON.parse(ev.data)
|
||||
if ('error' in res) {
|
||||
console.error('Delete failed', res.error)
|
||||
store.unhideDoc(path)
|
||||
store.showToast(res.error.message || 'Delete failed')
|
||||
} else if (res.status === 'ack') {
|
||||
store.showToast(`🗑️ Deleted ${doc.name}`)
|
||||
control.close()
|
||||
}
|
||||
}
|
||||
})
|
||||
control.onopen = () => {
|
||||
control.send(JSON.stringify({ op: 'rm', sel: [path] }))
|
||||
try {
|
||||
const res = await apiFetch(filesUrl(path), { method: 'DELETE' })
|
||||
if (!res.ok) throw new Error(await parseErrorMessage(res))
|
||||
store.showToast(`🗑️ Deleted ${doc.name}`)
|
||||
} catch (err) {
|
||||
console.error('Delete failed', err)
|
||||
store.unhideDoc(path)
|
||||
store.showToast(err instanceof Error ? err.message : 'Delete failed')
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -12,7 +12,7 @@
|
||||
import { ref, computed, watchEffect, shallowRef, onMounted, onUnmounted, nextTick } from 'vue'
|
||||
import { useMainStore } from '@/stores/main'
|
||||
import { Doc } from '@/repositories/Document'
|
||||
import { connect, controlUrl } from '@/repositories/WS'
|
||||
import { apiFetch } from '@/repositories/Client'
|
||||
import { useRouter } from 'vue-router'
|
||||
import ContextMenu from '@imengyu/vue3-context-menu'
|
||||
import type { SortOrder } from '@/utils/docsort'
|
||||
@@ -23,32 +23,37 @@ const props = defineProps<{
|
||||
}>()
|
||||
const store = useMainStore()
|
||||
const router = useRouter()
|
||||
|
||||
const filesUrl = (path: string) =>
|
||||
'/files/' + path.split('/').map(part => encodeURIComponent(part)).join('/')
|
||||
|
||||
const parseErrorMessage = async (res: Response) => {
|
||||
try {
|
||||
const data = await res.json()
|
||||
return data.message || data.detail || `${res.status} ${res.statusText}`
|
||||
} catch {
|
||||
return `${res.status} ${res.statusText}`
|
||||
}
|
||||
}
|
||||
|
||||
// File rename
|
||||
const editing = shallowRef<Doc | null>(null)
|
||||
const exit = () => { editing.value = null }
|
||||
const rename = (doc: Doc, newName: string) => {
|
||||
const rename = async (doc: Doc, newName: string) => {
|
||||
const oldName = doc.name
|
||||
const control = connect(controlUrl, {
|
||||
message(ev: MessageEvent) {
|
||||
const msg = JSON.parse(ev.data)
|
||||
if ('error' in msg) {
|
||||
console.error('Rename failed', msg.error.message, msg.error)
|
||||
doc.name = oldName
|
||||
} else {
|
||||
console.log('Rename succeeded', msg)
|
||||
}
|
||||
}
|
||||
})
|
||||
control.onopen = () => {
|
||||
control.send(
|
||||
JSON.stringify({
|
||||
op: 'rename',
|
||||
path: `${doc.loc}/${oldName}`,
|
||||
to: newName
|
||||
})
|
||||
)
|
||||
}
|
||||
doc.name = newName // We should get an update from watch but this is quicker
|
||||
try {
|
||||
const dstUrl = doc.loc ? filesUrl(doc.loc) : '/files/'
|
||||
const res = await apiFetch(
|
||||
`${dstUrl}?mv=${doc.key}&to=${encodeURIComponent(newName)}`,
|
||||
{ method: 'POST' }
|
||||
)
|
||||
if (!res.ok) throw new Error(await parseErrorMessage(res))
|
||||
} catch (err) {
|
||||
console.error('Rename failed', err)
|
||||
doc.name = oldName
|
||||
store.showToast(err instanceof Error ? err.message : 'Rename failed')
|
||||
}
|
||||
}
|
||||
const gallery = ref<HTMLElement>()
|
||||
const columnCount = ref(1)
|
||||
@@ -202,31 +207,20 @@ onMounted(() => {
|
||||
onUnmounted(() => {
|
||||
resizeObserver?.disconnect()
|
||||
})
|
||||
const mkdir = (doc: Doc, name: string) => {
|
||||
const control = connect(controlUrl, {
|
||||
open() {
|
||||
control.send(
|
||||
JSON.stringify({
|
||||
op: 'mkdir',
|
||||
path: `${doc.loc}/${name}`
|
||||
})
|
||||
)
|
||||
},
|
||||
message(ev: MessageEvent) {
|
||||
const msg = JSON.parse(ev.data)
|
||||
if ('error' in msg) {
|
||||
console.error('Mkdir failed', msg.error.message, msg.error)
|
||||
editing.value = null
|
||||
} else {
|
||||
console.log('mkdir', msg)
|
||||
router.push(doc.urlrouter)
|
||||
}
|
||||
}
|
||||
})
|
||||
const mkdir = async (doc: Doc, name: string) => {
|
||||
doc.name = name
|
||||
doc.key = crypto.randomUUID()
|
||||
store.addGhost(doc)
|
||||
editing.value = null
|
||||
const path = doc.loc ? `${doc.loc}/${name}` : name
|
||||
try {
|
||||
const res = await apiFetch(filesUrl(path), { method: 'MKCOL' })
|
||||
if (!res.ok) throw new Error(await parseErrorMessage(res))
|
||||
router.push(doc.urlrouter)
|
||||
} catch (err) {
|
||||
console.error('Mkdir failed', err)
|
||||
store.showToast(err instanceof Error ? err.message : 'Mkdir failed')
|
||||
}
|
||||
}
|
||||
const showFolderBreadcrumb = (i: number) => {
|
||||
const docs = props.documents
|
||||
@@ -312,24 +306,17 @@ const copyImage = async (doc: Doc) => {
|
||||
}
|
||||
}
|
||||
|
||||
const deleteFile = (doc: Doc) => {
|
||||
const deleteFile = async (doc: Doc) => {
|
||||
const path = doc.loc ? `${doc.loc}/${doc.name}` : doc.name
|
||||
store.hideDoc(path)
|
||||
const control = connect(controlUrl, {
|
||||
message(ev: MessageEvent) {
|
||||
const res = JSON.parse(ev.data)
|
||||
if ('error' in res) {
|
||||
console.error('Delete failed', res.error)
|
||||
store.unhideDoc(path)
|
||||
store.showToast(res.error.message || 'Delete failed')
|
||||
} else if (res.status === 'ack') {
|
||||
store.showToast(`🗑️ Deleted ${doc.name}`)
|
||||
control.close()
|
||||
}
|
||||
}
|
||||
})
|
||||
control.onopen = () => {
|
||||
control.send(JSON.stringify({ op: 'rm', sel: [path] }))
|
||||
try {
|
||||
const res = await apiFetch(filesUrl(path), { method: 'DELETE' })
|
||||
if (!res.ok) throw new Error(await parseErrorMessage(res))
|
||||
store.showToast(`🗑️ Deleted ${doc.name}`)
|
||||
} catch (err) {
|
||||
console.error('Delete failed', err)
|
||||
store.unhideDoc(path)
|
||||
store.showToast(err instanceof Error ? err.message : 'Delete failed')
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -29,7 +29,7 @@
|
||||
</template>
|
||||
|
||||
<script setup lang="ts">
|
||||
import {connect, controlUrl} from '@/repositories/WS'
|
||||
import { apiFetch } from '@/repositories/Client'
|
||||
import { useMainStore } from '@/stores/main'
|
||||
import { computed, ref } from 'vue'
|
||||
import { formatSize } from '@/utils'
|
||||
@@ -49,6 +49,18 @@ const navigateTo = (path: string) => {
|
||||
router.push('/' + path)
|
||||
}
|
||||
|
||||
const filesUrl = (path: string) =>
|
||||
'/files/' + path.split('/').map(part => encodeURIComponent(part)).join('/')
|
||||
|
||||
const parseErrorMessage = async (res: Response) => {
|
||||
try {
|
||||
const data = await res.json()
|
||||
return data.message || data.detail || `${res.status} ${res.statusText}`
|
||||
} catch {
|
||||
return `${res.status} ${res.statusText}`
|
||||
}
|
||||
}
|
||||
|
||||
// Truncate long names to reasonable length
|
||||
const truncateName = (name: string, maxLen = 20): string => {
|
||||
if (name.length <= maxLen) return name
|
||||
@@ -115,43 +127,43 @@ const selectionDisplay = computed<SelectionDisplay>(() => {
|
||||
}
|
||||
})
|
||||
|
||||
const op = (opName: string, dst?: string) => {
|
||||
const op = async (opName: string, dst?: string) => {
|
||||
const sel = store.selectedFiles
|
||||
const keys = sel.keys
|
||||
const paths = sel.keys.map(key => {
|
||||
const doc = sel.docs[key]!
|
||||
return doc.loc ? `${doc.loc}/${doc.name}` : doc.name
|
||||
})
|
||||
const msg = {
|
||||
op: opName,
|
||||
sel: paths
|
||||
}
|
||||
// @ts-ignore
|
||||
if (dst !== undefined) msg.dst = dst
|
||||
|
||||
// Hide items being deleted or moved (optimistic update)
|
||||
if (opName === 'rm' || opName === 'mv') {
|
||||
for (const path of paths) store.hideDoc(path)
|
||||
}
|
||||
const control = connect(controlUrl, {
|
||||
message(ev: MessageEvent) {
|
||||
const res = JSON.parse(ev.data)
|
||||
if ('error' in res) {
|
||||
console.error('Control socket error', msg, res.error)
|
||||
store.error = res.error.message
|
||||
// Restore hidden items on error
|
||||
if (opName === 'rm' || opName === 'mv') {
|
||||
for (const path of paths) store.unhideDoc(path)
|
||||
}
|
||||
return
|
||||
} else if (res.status === 'ack') {
|
||||
console.log('Control ack OK', res)
|
||||
control.close()
|
||||
store.selected.clear()
|
||||
return
|
||||
} else console.log('Unknown control response', msg, res)
|
||||
|
||||
try {
|
||||
if (opName === 'rm') {
|
||||
for (const path of paths) {
|
||||
const res = await apiFetch(filesUrl(path), { method: 'DELETE' })
|
||||
if (!res.ok) throw new Error(await parseErrorMessage(res))
|
||||
}
|
||||
} else if (opName === 'mv' || opName === 'cp') {
|
||||
if (keys.length === 0) throw new Error('No selected files')
|
||||
const dstUrl = dst ? filesUrl(dst) : '/files/'
|
||||
const query = `${opName}=${keys.join('+')}`
|
||||
const res = await apiFetch(`${dstUrl}?${query}`, { method: 'POST' })
|
||||
if (!res.ok) throw new Error(await parseErrorMessage(res))
|
||||
} else {
|
||||
throw new Error(`Unsupported operation: ${opName}`)
|
||||
}
|
||||
|
||||
store.selected.clear()
|
||||
} catch (err) {
|
||||
const message = err instanceof Error ? err.message : String(err)
|
||||
console.error('REST file operation failed', opName, err)
|
||||
store.error = message
|
||||
if (opName === 'rm' || opName === 'mv') {
|
||||
for (const path of paths) store.unhideDoc(path)
|
||||
}
|
||||
})
|
||||
control.onopen = () => {
|
||||
control.send(JSON.stringify(msg))
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -2,7 +2,6 @@ import { useMainStore } from "@/stores/main"
|
||||
import { showAuthIframe, AuthCancelledError, isAuthIframeOpen } from 'paskia'
|
||||
import type { FileEntry, UpdateEntry, errorEvent } from "./Document"
|
||||
|
||||
export const controlUrl = '/api/control'
|
||||
export const watchUrl = '/api/watch'
|
||||
|
||||
let tree = [] as FileEntry[]
|
||||
|
||||
@@ -130,6 +130,7 @@ dev = [
|
||||
"mypy>=1.13.0",
|
||||
"pre-commit>=4.0.0",
|
||||
"httpx>=0.28.1",
|
||||
"sanic-testing>=24.6.0",
|
||||
]
|
||||
|
||||
[tool.coverage.run]
|
||||
|
||||
@@ -1,53 +0,0 @@
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from cista import config
|
||||
from cista.protocol import Cp, MkDir, Mv, Rename, Rm
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def setup_temp_dir():
|
||||
with tempfile.TemporaryDirectory() as tmpdirname:
|
||||
config.config = config.Config(path=Path(tmpdirname), listen=":0")
|
||||
yield Path(tmpdirname)
|
||||
|
||||
|
||||
def test_mkdir(setup_temp_dir):
|
||||
cmd = MkDir(path="new_folder")
|
||||
cmd()
|
||||
assert (setup_temp_dir / "new_folder").is_dir()
|
||||
|
||||
|
||||
def test_rename(setup_temp_dir):
|
||||
(setup_temp_dir / "old_name").mkdir()
|
||||
cmd = Rename(path="old_name", to="new_name")
|
||||
cmd()
|
||||
assert not (setup_temp_dir / "old_name").exists()
|
||||
assert (setup_temp_dir / "new_name").is_dir()
|
||||
|
||||
|
||||
def test_rm(setup_temp_dir):
|
||||
(setup_temp_dir / "folder_to_remove").mkdir()
|
||||
cmd = Rm(sel=["folder_to_remove"])
|
||||
cmd()
|
||||
assert not (setup_temp_dir / "folder_to_remove").exists()
|
||||
|
||||
|
||||
def test_mv(setup_temp_dir):
|
||||
(setup_temp_dir / "folder_to_move").mkdir()
|
||||
(setup_temp_dir / "destination").mkdir()
|
||||
cmd = Mv(sel=["folder_to_move"], dst="destination")
|
||||
cmd()
|
||||
assert not (setup_temp_dir / "folder_to_move").exists()
|
||||
assert (setup_temp_dir / "destination" / "folder_to_move").is_dir()
|
||||
|
||||
|
||||
def test_cp(setup_temp_dir):
|
||||
(setup_temp_dir / "folder_to_copy").mkdir()
|
||||
(setup_temp_dir / "destination").mkdir()
|
||||
cmd = Cp(sel=["folder_to_copy"], dst="destination")
|
||||
cmd()
|
||||
assert (setup_temp_dir / "folder_to_copy").is_dir()
|
||||
assert (setup_temp_dir / "destination" / "folder_to_copy").is_dir()
|
||||
@@ -0,0 +1,169 @@
|
||||
"""Path traversal and percent-encoding security tests for the fileserver."""
|
||||
from pathlib import Path
|
||||
from uuid import uuid4
|
||||
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
from sanic import Sanic
|
||||
|
||||
from cista import config, watching
|
||||
from cista.fileserver import bp as fileserver_bp
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def setup_storage(tmp_path: Path):
|
||||
config.config = config.Config(path=tmp_path, listen=":0", public=True)
|
||||
watching.state.root = []
|
||||
watching.rootpath = tmp_path
|
||||
yield tmp_path
|
||||
watching.state.root = []
|
||||
|
||||
|
||||
@pytest_asyncio.fixture()
|
||||
async def client(setup_storage: Path):
|
||||
app = Sanic(f"files-path-sec-test-{uuid4().hex}", strict_slashes=True)
|
||||
app.router.ALLOWED_METHODS = (*app.router.ALLOWED_METHODS, "MKCOL")
|
||||
app.blueprint(fileserver_bp)
|
||||
yield app.asgi_client
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# %2F — encoded slash should be decoded as a path separator
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
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")
|
||||
|
||||
_, res = await client.get("/files/sub%2Ffile.txt")
|
||||
|
||||
assert res.status_code == 200
|
||||
assert res.text == "hello"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
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")
|
||||
|
||||
assert res.status_code == 201
|
||||
assert (setup_storage / "parent" / "child").is_dir()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# %20 — encoded space in filename
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_percent20_in_filename(client, setup_storage: Path):
|
||||
(setup_storage / "my file.txt").write_text("spaced", encoding="utf-8")
|
||||
|
||||
_, res = await client.get("/files/my%20file.txt")
|
||||
|
||||
assert res.status_code == 200
|
||||
assert res.text == "spaced"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_mkcol_percent20_in_folder_name(client, setup_storage: Path):
|
||||
_, res = await client.request("MKCOL", "/files/my%20folder")
|
||||
|
||||
assert res.status_code == 201
|
||||
assert (setup_storage / "my folder").is_dir()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Path traversal — .. and encoded variants
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_dotdot_rejected(client):
|
||||
""".. is path-normalised by the router before reaching the handler."""
|
||||
_, res = await client.get("/files/..")
|
||||
assert res.status_code in (400, 404)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_dotdot_segment_rejected(client):
|
||||
"""Traversal via sub/../.. is path-normalised by the router."""
|
||||
_, res = await client.get("/files/sub/../..")
|
||||
assert res.status_code in (400, 404)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_encoded_dotdot_rejected(client):
|
||||
"""%2E%2E (encoded ..) must be rejected."""
|
||||
_, res = await client.get("/files/%2E%2E")
|
||||
assert res.status_code == 400
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_encoded_dotdot_segment_rejected(client):
|
||||
"""%2E%2E used as a segment in a longer path must be rejected."""
|
||||
_, res = await client.get("/files/sub%2F%2E%2E%2F..%2Fetc%2Fpasswd")
|
||||
assert res.status_code == 400
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_mkcol_dotdot_rejected(client):
|
||||
_, res = await client.request("MKCOL", "/files/..")
|
||||
assert res.status_code in (400, 404)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_delete_dotdot_rejected(client):
|
||||
_, res = await client.delete("/files/..")
|
||||
assert res.status_code in (400, 404)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Dot-prefixed filenames (.hidden, ...)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_hidden_file_rejected(client):
|
||||
"""Names starting with '.' are not allowed."""
|
||||
_, res = await client.get("/files/.hidden")
|
||||
assert res.status_code == 400
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_mkcol_hidden_folder_rejected(client):
|
||||
_, res = await client.request("MKCOL", "/files/.secret")
|
||||
assert res.status_code == 400
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Windows-style drive paths (c:/) — safe on Linux, stays inside storage root
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_mkcol_windows_drive_path_stays_within_root(client, setup_storage: Path):
|
||||
"""A Windows-style drive path like 'c:/foo' is treated as a relative path
|
||||
on Linux and resolves safely inside the storage root."""
|
||||
_, res = await client.request("MKCOL", "/files/c:/secret")
|
||||
|
||||
# Either created inside the storage root (201) or sanitised away (400/404).
|
||||
# The important assertion: nothing was created outside the storage root.
|
||||
assert not (Path("/c:") / "secret").exists()
|
||||
assert not (Path("c:/secret")).exists()
|
||||
if res.status_code == 201:
|
||||
# Created safely inside tmp storage
|
||||
assert (setup_storage / "c:" / "secret").is_dir()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_mkcol_backslash_in_path_sanitised(client, setup_storage: Path):
|
||||
"""Backslashes are replaced with dashes, not treated as path separators."""
|
||||
_, res = await client.request("MKCOL", "/files/foo\\..\\bar")
|
||||
|
||||
assert res.status_code in (201, 400)
|
||||
# Must not escape storage root
|
||||
assert not (setup_storage.parent / "bar").exists()
|
||||
@@ -0,0 +1,234 @@
|
||||
from pathlib import Path
|
||||
from uuid import uuid4
|
||||
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
from sanic import Sanic
|
||||
|
||||
from cista import config, watching
|
||||
from cista.fileserver import bp as fileserver_bp
|
||||
from cista.protocol import FileEntry
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def setup_storage(tmp_path: Path):
|
||||
config.config = config.Config(path=tmp_path, listen=":0", public=True)
|
||||
watching.state.root = []
|
||||
watching.rootpath = tmp_path
|
||||
yield tmp_path
|
||||
watching.state.root = []
|
||||
|
||||
|
||||
@pytest_asyncio.fixture()
|
||||
async def client(setup_storage: Path):
|
||||
app = Sanic(f"files-rest-test-{uuid4().hex}", strict_slashes=True)
|
||||
app.router.ALLOWED_METHODS = (*app.router.ALLOWED_METHODS, "MKCOL")
|
||||
app.blueprint(fileserver_bp)
|
||||
yield app.asgi_client
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_mkcol_creates_directory(client, setup_storage: Path):
|
||||
_, res = await client.request("MKCOL", "/files/new-folder")
|
||||
|
||||
assert res.status_code == 201
|
||||
assert (setup_storage / "new-folder").is_dir()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_delete_removes_file(client, setup_storage: Path):
|
||||
file_path = setup_storage / "delete-me.txt"
|
||||
file_path.write_text("hello", encoding="utf-8")
|
||||
|
||||
_, res = await client.delete("/files/delete-me.txt")
|
||||
|
||||
assert res.status_code == 204
|
||||
assert not file_path.exists()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_post_mv_moves_keys_to_target(client, setup_storage: Path):
|
||||
(setup_storage / "target").mkdir()
|
||||
(setup_storage / "alpha.txt").write_text("alpha", encoding="utf-8")
|
||||
(setup_storage / "beta.txt").write_text("beta", encoding="utf-8")
|
||||
|
||||
watching.state.root = [
|
||||
FileEntry(1, "target", "k-target", 0, 0, 0, 0),
|
||||
FileEntry(1, "alpha.txt", "k-alpha", 0, 5, 0, 1),
|
||||
FileEntry(1, "beta.txt", "k-beta", 0, 4, 0, 1),
|
||||
]
|
||||
|
||||
_, res = await client.post("/files/target?mv=k-alpha+k-beta")
|
||||
|
||||
assert res.status_code == 200
|
||||
assert res.json["status"] == "ack"
|
||||
assert not (setup_storage / "alpha.txt").exists()
|
||||
assert not (setup_storage / "beta.txt").exists()
|
||||
assert (setup_storage / "target" / "alpha.txt").is_file()
|
||||
assert (setup_storage / "target" / "beta.txt").is_file()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_post_cp_copies_keys_to_target(client, setup_storage: Path):
|
||||
(setup_storage / "target").mkdir()
|
||||
(setup_storage / "copy-me.txt").write_text("copy", encoding="utf-8")
|
||||
|
||||
watching.state.root = [
|
||||
FileEntry(1, "target", "k-target", 0, 0, 0, 0),
|
||||
FileEntry(1, "copy-me.txt", "k-copy", 0, 4, 0, 1),
|
||||
]
|
||||
|
||||
_, res = await client.post("/files/target?cp=k-copy")
|
||||
|
||||
assert res.status_code == 200
|
||||
assert res.json["counts"] == {"cp": 1, "mv": 0}
|
||||
assert (setup_storage / "copy-me.txt").is_file()
|
||||
assert (setup_storage / "target" / "copy-me.txt").is_file()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_post_cp_repeated_params_and_plus_form_are_equivalent(
|
||||
client,
|
||||
setup_storage: Path,
|
||||
):
|
||||
(setup_storage / "target").mkdir()
|
||||
(setup_storage / "one.txt").write_text("one", encoding="utf-8")
|
||||
(setup_storage / "two.txt").write_text("two", encoding="utf-8")
|
||||
|
||||
watching.state.root = [
|
||||
FileEntry(1, "target", "k-target", 0, 0, 0, 0),
|
||||
FileEntry(1, "one.txt", "k-one", 0, 3, 0, 1),
|
||||
FileEntry(1, "two.txt", "k-two", 0, 3, 0, 1),
|
||||
]
|
||||
|
||||
_, res1 = await client.post("/files/target?cp=k-one&cp=k-two")
|
||||
|
||||
assert res1.status_code == 200
|
||||
assert (setup_storage / "target" / "one.txt").is_file()
|
||||
assert (setup_storage / "target" / "two.txt").is_file()
|
||||
|
||||
(setup_storage / "target" / "one.txt").unlink()
|
||||
(setup_storage / "target" / "two.txt").unlink()
|
||||
|
||||
_, res2 = await client.post("/files/target?cp=k-one+k-two")
|
||||
|
||||
assert res2.status_code == 200
|
||||
assert (setup_storage / "target" / "one.txt").is_file()
|
||||
assert (setup_storage / "target" / "two.txt").is_file()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_post_mv_with_to_renames_single_key(
|
||||
client,
|
||||
setup_storage: Path,
|
||||
):
|
||||
(setup_storage / "dst").mkdir()
|
||||
(setup_storage / "old-name.txt").write_text("x", encoding="utf-8")
|
||||
|
||||
watching.state.root = [
|
||||
FileEntry(1, "dst", "k-dst", 0, 0, 0, 0),
|
||||
FileEntry(1, "old-name.txt", "k-old", 0, 1, 0, 1),
|
||||
]
|
||||
|
||||
_, res = await client.post("/files/dst/new-name.txt?mv=k-old")
|
||||
|
||||
assert res.status_code == 200
|
||||
assert not (setup_storage / "old-name.txt").exists()
|
||||
assert (setup_storage / "dst" / "new-name.txt").is_file()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_post_cp_single_key_to_file_path(client, setup_storage: Path):
|
||||
(setup_storage / "dst").mkdir()
|
||||
(setup_storage / "src.txt").write_text("copy", encoding="utf-8")
|
||||
|
||||
watching.state.root = [
|
||||
FileEntry(1, "dst", "k-dst", 0, 0, 0, 0),
|
||||
FileEntry(1, "src.txt", "k-src", 0, 4, 0, 1),
|
||||
]
|
||||
|
||||
_, res = await client.post("/files/dst/copied.txt?cp=k-src")
|
||||
|
||||
assert res.status_code == 200
|
||||
assert (setup_storage / "src.txt").is_file()
|
||||
assert (setup_storage / "dst" / "copied.txt").is_file()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_post_supports_combined_cp_then_mv(client, setup_storage: Path):
|
||||
(setup_storage / "target").mkdir()
|
||||
(setup_storage / "copy-me.txt").write_text("copy", encoding="utf-8")
|
||||
(setup_storage / "move-me.txt").write_text("move", encoding="utf-8")
|
||||
|
||||
watching.state.root = [
|
||||
FileEntry(1, "target", "k-target", 0, 0, 0, 0),
|
||||
FileEntry(1, "copy-me.txt", "k-copy", 0, 4, 0, 1),
|
||||
FileEntry(1, "move-me.txt", "k-move", 0, 4, 0, 1),
|
||||
]
|
||||
|
||||
_, res = await client.post("/files/target?cp=k-copy&mv=k-move")
|
||||
|
||||
assert res.status_code == 200
|
||||
assert res.json["counts"] == {"cp": 1, "mv": 1}
|
||||
assert (setup_storage / "copy-me.txt").is_file()
|
||||
assert not (setup_storage / "move-me.txt").exists()
|
||||
assert (setup_storage / "target" / "copy-me.txt").is_file()
|
||||
assert (setup_storage / "target" / "move-me.txt").is_file()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_post_rejects_unknown_query_args(client):
|
||||
_, res = await client.post("/files/?cp=k1&wat=1")
|
||||
|
||||
assert res.status_code == 400
|
||||
assert "unknown query parameter" in res.json["message"].lower()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_post_requires_query_args(client):
|
||||
_, res = await client.post("/files/")
|
||||
|
||||
assert res.status_code == 400
|
||||
assert "no query arguments" in res.json["message"].lower()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_post_rejects_multiple_keys_to_file_target(client, setup_storage: Path):
|
||||
(setup_storage / "a.txt").write_text("a", encoding="utf-8")
|
||||
(setup_storage / "b.txt").write_text("b", encoding="utf-8")
|
||||
(setup_storage / "target.txt").write_text("x", encoding="utf-8")
|
||||
|
||||
watching.state.root = [
|
||||
FileEntry(1, "a.txt", "k-a", 0, 1, 0, 1),
|
||||
FileEntry(1, "b.txt", "k-b", 0, 1, 0, 1),
|
||||
FileEntry(1, "target.txt", "k-target", 0, 1, 0, 1),
|
||||
]
|
||||
|
||||
_, cp_res = await client.post("/files/target.txt?cp=k-a+k-b")
|
||||
_, mv_res = await client.post("/files/target.txt?mv=k-a+k-b")
|
||||
|
||||
assert cp_res.status_code == 400
|
||||
assert "existing directory" in cp_res.json["message"].lower()
|
||||
assert mv_res.status_code == 400
|
||||
assert "existing directory" in mv_res.json["message"].lower()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_post_rejects_directory_to_existing_file_target(client, setup_storage: Path):
|
||||
(setup_storage / "folder").mkdir()
|
||||
(setup_storage / "folder" / "nested.txt").write_text("n", encoding="utf-8")
|
||||
(setup_storage / "existing.txt").write_text("e", encoding="utf-8")
|
||||
|
||||
watching.state.root = [
|
||||
FileEntry(1, "folder", "k-folder", 0, 0, 0, 0),
|
||||
FileEntry(2, "nested.txt", "k-nested", 0, 1, 0, 1),
|
||||
FileEntry(1, "existing.txt", "k-existing", 0, 1, 0, 1),
|
||||
]
|
||||
|
||||
_, cp_res = await client.post("/files/existing.txt?cp=k-folder")
|
||||
_, mv_res = await client.post("/files/existing.txt?mv=k-folder")
|
||||
|
||||
assert cp_res.status_code == 400
|
||||
assert "directory to an existing file" in cp_res.json["message"].lower()
|
||||
assert mv_res.status_code == 400
|
||||
assert "directory to an existing file" in mv_res.json["message"].lower()
|
||||
@@ -0,0 +1,100 @@
|
||||
from pathlib import Path
|
||||
from uuid import uuid4
|
||||
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
from sanic import Sanic
|
||||
|
||||
from cista import config, watching
|
||||
from cista.fileserver import bp as fileserver_bp
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def setup_storage(tmp_path: Path):
|
||||
config.config = config.Config(path=tmp_path, listen=":0", public=True)
|
||||
watching.state.root = []
|
||||
watching.rootpath = tmp_path
|
||||
yield tmp_path
|
||||
watching.state.root = []
|
||||
|
||||
|
||||
@pytest_asyncio.fixture()
|
||||
async def client(setup_storage: Path):
|
||||
app = Sanic(f"files-static-test-{uuid4().hex}", strict_slashes=True)
|
||||
app.router.ALLOWED_METHODS = (*app.router.ALLOWED_METHODS, "MKCOL")
|
||||
app.blueprint(fileserver_bp)
|
||||
yield app.asgi_client
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_file_full_content(client, setup_storage: Path):
|
||||
path = setup_storage / "hello.txt"
|
||||
path.write_bytes(b"hello world")
|
||||
|
||||
_, res = await client.get("/files/hello.txt")
|
||||
|
||||
assert res.status_code == 200
|
||||
assert res.body == b"hello world"
|
||||
assert res.headers.get("accept-ranges") == "bytes"
|
||||
assert res.headers.get("content-length") == "11"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_head_file_returns_headers_without_body(client, setup_storage: Path):
|
||||
path = setup_storage / "hello.txt"
|
||||
path.write_bytes(b"hello world")
|
||||
|
||||
_, res = await client.head("/files/hello.txt")
|
||||
|
||||
assert res.status_code == 200
|
||||
assert not res.body
|
||||
assert res.headers.get("content-length") == "11"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_file_range_start_end(client, setup_storage: Path):
|
||||
path = setup_storage / "hello.txt"
|
||||
path.write_bytes(b"hello world")
|
||||
|
||||
_, res = await client.get("/files/hello.txt", headers={"Range": "bytes=1-4"})
|
||||
|
||||
assert res.status_code == 206
|
||||
assert res.body == b"ello"
|
||||
assert res.headers.get("content-range") == "bytes 1-4/11"
|
||||
assert res.headers.get("content-length") == "4"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_file_suffix_range(client, setup_storage: Path):
|
||||
path = setup_storage / "hello.txt"
|
||||
path.write_bytes(b"hello world")
|
||||
|
||||
_, res = await client.get("/files/hello.txt", headers={"Range": "bytes=-5"})
|
||||
|
||||
assert res.status_code == 206
|
||||
assert res.body == b"world"
|
||||
assert res.headers.get("content-range") == "bytes 6-10/11"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_head_file_with_range(client, setup_storage: Path):
|
||||
path = setup_storage / "hello.txt"
|
||||
path.write_bytes(b"hello world")
|
||||
|
||||
_, res = await client.head("/files/hello.txt", headers={"Range": "bytes=0-4"})
|
||||
|
||||
assert res.status_code == 206
|
||||
assert not res.body
|
||||
assert res.headers.get("content-range") == "bytes 0-4/11"
|
||||
assert res.headers.get("content-length") == "5"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_file_unsatisfiable_range_returns_416(client, setup_storage: Path):
|
||||
path = setup_storage / "hello.txt"
|
||||
path.write_bytes(b"hello world")
|
||||
|
||||
_, res = await client.get("/files/hello.txt", headers={"Range": "bytes=99-100"})
|
||||
|
||||
assert res.status_code == 416
|
||||
assert res.headers.get("content-range") == "bytes */11"
|
||||
Reference in New Issue
Block a user