feat(booth): upload-for-pickup with human-readable ids (v0.1.1)
Add a reverse direction to the Booth: the operator (or any client via `curl -F`) can upload files through the browser and pick them up by a human-readable id. - POST /upload — streams files to a new booth named with a human-readable id (e.g. 4-wombat / star-84), 303-redirects to /b/<id>/ (id in the Location header so curl clients can read it). Uploads reuse the whole booth machinery (render, per-file download links, 24h TTL sweep, delete). - Human-readable ids: word+number in either order, collision-checked, from a curated 140-word friendly list; secrets-based selection. - Safety: filenames reduced to a safe basename (no traversal), streaming size cap (BOOTH_MAX_UPLOAD_MB, default 1024) + file-count cap (BOOTH_MAX_FILES, default 50), partial-write cleanup on any failure. - UI: Australis-themed upload/drop panel (drag-drop, progressive-enhancement JS, degrades to a native file input), a "⬆ pickup" badge on upload booths, a pickup banner, and a ⬇ download link on every gallery item. - python-multipart dependency; homepage tile description updated; 9 new tests (24 total, all green).
This commit is contained in:
@@ -1,4 +1,5 @@
|
||||
import os
|
||||
import re
|
||||
import time
|
||||
|
||||
import pytest
|
||||
@@ -8,11 +9,15 @@ from booth.app import (
|
||||
build_gallery,
|
||||
classify,
|
||||
create_app,
|
||||
generate_pickup_id,
|
||||
human_dur,
|
||||
is_expired,
|
||||
safe_upload_name,
|
||||
sweep_once,
|
||||
)
|
||||
|
||||
PICKUP_RE = re.compile(r"^(\d{1,2}-[a-z]+|[a-z]+-\d{1,2})$")
|
||||
|
||||
|
||||
# ---- pure helpers -----------------------------------------------------------
|
||||
|
||||
@@ -182,3 +187,102 @@ def test_healthz(client):
|
||||
assert r.status_code == 200
|
||||
body = r.json()
|
||||
assert body["ok"] is True and body["booths"] == 1
|
||||
|
||||
|
||||
# ---- uploads / pickup -------------------------------------------------------
|
||||
|
||||
|
||||
def test_generate_pickup_id_format():
|
||||
for _ in range(100):
|
||||
pid = generate_pickup_id(lambda n: False)
|
||||
assert PICKUP_RE.match(pid), pid
|
||||
|
||||
|
||||
def test_generate_pickup_id_avoids_collision():
|
||||
taken = {"4-wombat", "star-84"}
|
||||
for _ in range(50):
|
||||
pid = generate_pickup_id(lambda n: n in taken)
|
||||
assert pid not in taken
|
||||
|
||||
|
||||
def test_safe_upload_name():
|
||||
assert safe_upload_name("../../etc/passwd", "fb") == "passwd"
|
||||
assert safe_upload_name("C:\\Users\\x\\shot.png", "fb") == "shot.png"
|
||||
assert safe_upload_name("", "fb") == "fb"
|
||||
assert safe_upload_name(" ", "fb") == "fb"
|
||||
assert safe_upload_name(".hidden", "fb") == "hidden"
|
||||
assert safe_upload_name("...", "fb") == "fb"
|
||||
|
||||
|
||||
def _upload(client, files):
|
||||
return client.post("/upload", files=files, follow_redirects=False)
|
||||
|
||||
|
||||
def test_upload_creates_pickup_booth(client):
|
||||
c, data = client
|
||||
r = _upload(c, [
|
||||
("files", ("a.png", b"\x89PNG\r\n\x1a\n" + b"0" * 20, "image/png")),
|
||||
("files", ("notes.txt", b"pick me up", "text/plain")),
|
||||
])
|
||||
assert r.status_code == 303
|
||||
loc = r.headers["location"]
|
||||
pid = loc.split("/b/")[1].rstrip("/")
|
||||
assert PICKUP_RE.match(pid), pid
|
||||
booth = data / pid
|
||||
assert (booth / "a.png").is_file()
|
||||
assert (booth / "notes.txt").read_bytes() == b"pick me up"
|
||||
assert (booth / ".uploaded").is_file() # marker present
|
||||
|
||||
|
||||
def test_upload_booth_renders_pickup_ui(client):
|
||||
c, data = client
|
||||
r = _upload(c, [("files", ("shot.png", b"\x89PNG" + b"0" * 30, "image/png"))])
|
||||
pid = r.headers["location"].split("/b/")[1].rstrip("/")
|
||||
page = c.get(f"/b/{pid}/")
|
||||
assert page.status_code == 200
|
||||
assert "pickup" in page.text.lower() # badge / note
|
||||
assert 'download' in page.text # per-item download link
|
||||
# and it shows up flagged as an upload on the index
|
||||
assert pid in c.get("/").text
|
||||
|
||||
|
||||
def test_upload_sanitizes_traversal(client):
|
||||
c, data = client
|
||||
r = _upload(c, [("files", ("../../../etc/passwd", b"x", "text/plain"))])
|
||||
pid = r.headers["location"].split("/b/")[1].rstrip("/")
|
||||
booth = data / pid
|
||||
assert (booth / "passwd").is_file() # basename only
|
||||
assert not (data.parent / "passwd").exists() # nothing escaped upward
|
||||
|
||||
|
||||
def test_upload_rejects_too_many_files(tmp_path):
|
||||
app = create_app(tmp_path, start_sweeper=False, max_files=2)
|
||||
c = TestClient(app)
|
||||
files = [("files", (f"f{i}.txt", b"x", "text/plain")) for i in range(3)]
|
||||
r = c.post("/upload", files=files, follow_redirects=False)
|
||||
assert r.status_code == 413
|
||||
# no partial booth left behind
|
||||
assert list(tmp_path.iterdir()) == []
|
||||
|
||||
|
||||
def test_upload_rejects_too_large(tmp_path):
|
||||
app = create_app(tmp_path, start_sweeper=False, max_upload_mb=0.0001) # ~104 bytes
|
||||
c = TestClient(app)
|
||||
r = c.post(
|
||||
"/upload",
|
||||
files=[("files", ("big.bin", b"0" * 500, "application/octet-stream"))],
|
||||
follow_redirects=False,
|
||||
)
|
||||
assert r.status_code == 413
|
||||
assert list(tmp_path.iterdir()) == [] # partial write cleaned up
|
||||
|
||||
|
||||
def test_upload_dedupes_repeated_names(client):
|
||||
c, data = client
|
||||
r = _upload(c, [
|
||||
("files", ("shot.png", b"a" * 10, "image/png")),
|
||||
("files", ("shot.png", b"b" * 10, "image/png")),
|
||||
])
|
||||
pid = r.headers["location"].split("/b/")[1].rstrip("/")
|
||||
names = sorted(p.name for p in (data / pid).iterdir() if not p.name.startswith("."))
|
||||
assert names == ["shot-1.png", "shot.png"]
|
||||
|
||||
Reference in New Issue
Block a user