From ff81591e8e221db7ee25fbfc0c5e473a22ab7077 Mon Sep 17 00:00:00 2001 From: nesquena-hermes <[email protected]> Date: Tue, 2 Jun 2026 00:26:53 +0000 Subject: [PATCH] feat(workspace): add file upload + drag-drop with archive extraction (#3104, @antoniocarlos97ss) --- CHANGELOG.md | 3 + api/routes.py | 4 +- api/upload.py | 138 +++++++++++- static/index.html | 4 +- static/style.css | 1 + static/workspace.js | 72 ++++++ tests/test_workspace_upload.py | 400 +++++++++++++++++++++++++++++++++ 7 files changed, 619 insertions(+), 3 deletions(-) create mode 100644 tests/test_workspace_upload.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 17658185..b314931c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,9 @@ ## [Unreleased] +### Added +- Workspace file panel: an **Upload** button and drag-and-drop that POST to a new `/api/workspace/upload` endpoint. Files land in the session workspace (resolved via the trusted-workspace guard), are de-duplicated with `-1`/`-2` suffixes, and archives (`.zip`/`.tar.*`) are auto-extracted into the target subdirectory with zip-bomb (size-cap) and zip-slip (path-containment) protections. Extraction errors are surfaced to the frontend instead of being silently swallowed, and the archive is removed on failure (#3104, @antoniocarlos97ss). + ## [v0.51.205] — 2026-06-01 — Release FY (stage-hi1 — workspace syntax highlighting + generated-image cards + manual title regeneration) ### Added diff --git a/api/routes.py b/api/routes.py index b62dffb4..44afdbb2 100644 --- a/api/routes.py +++ b/api/routes.py @@ -2840,7 +2840,7 @@ from api.workspace import ( _strip_surrounding_quotes, _workspace_blocked_roots, ) -from api.upload import handle_upload, handle_upload_extract, handle_transcribe +from api.upload import handle_upload, handle_upload_extract, handle_transcribe, handle_workspace_upload from api.streaming import ( _sse, _run_agent_streaming, @@ -5459,6 +5459,8 @@ def handle_post(handler, parsed) -> bool: return handle_upload(handler) if parsed.path == "/api/upload/extract": return handle_upload_extract(handler) + if parsed.path == "/api/workspace/upload": + return handle_workspace_upload(handler) if parsed.path == "/api/transcribe": return handle_transcribe(handler) diff --git a/api/upload.py b/api/upload.py index b36cde46..c799c8f7 100644 --- a/api/upload.py +++ b/api/upload.py @@ -11,7 +11,7 @@ from pathlib import Path from api.config import MAX_UPLOAD_BYTES, STATE_DIR from api.helpers import j, bad from api.models import get_session -from api.workspace import safe_resolve_ws +from api.workspace import safe_resolve_ws, resolve_trusted_workspace _MAX_EXTRACTED_BYTES = 10 * MAX_UPLOAD_BYTES @@ -320,3 +320,139 @@ def handle_transcribe(handler): Path(temp_path).unlink(missing_ok=True) except Exception: pass + + +def handle_workspace_upload(handler): + """Upload a file into a session's workspace directory. + + Form fields: + session_id – target session + path – subdirectory within the workspace (default: '') + File: + file – the uploaded file(s) + """ + import traceback as _tb + try: + content_type = handler.headers.get('Content-Type', '') + content_length = int(handler.headers.get('Content-Length', 0) or 0) + if content_length > MAX_UPLOAD_BYTES: + return j(handler, {'error': f'File too large (max {MAX_UPLOAD_BYTES//1024//1024}MB)'}, status=413) + + fields, files = parse_multipart(handler.rfile, content_type, content_length) + session_id = fields.get('session_id', '') + subpath = fields.get('path', '') + + if not session_id: + return j(handler, {'error': 'Missing session_id'}, status=400) + + if not files: + return j(handler, {'error': 'No file field in request'}, status=400) + + # Validate session + try: + session = get_session(session_id) + except KeyError: + return j(handler, {'error': 'Session not found'}, status=404) + + # Resolve workspace root from session + workspace = resolve_trusted_workspace(session.workspace) + + # Resolve target subdirectory within workspace + target_dir = safe_resolve_ws(workspace, subpath) if subpath else workspace + target_dir.mkdir(parents=True, exist_ok=True) + + results = [] + for field_name, (filename, file_bytes) in files.items(): + if not filename: + continue + + safe_name = _sanitize_upload_name(filename) + dest = safe_resolve_ws(target_dir, safe_name) + + # Path traversal guard (belt-and-suspenders: safe_resolve_ws above is + # the authoritative guard and raises ValueError on traversal; this + # check catches any edge case where the resolved path escapes). + if not dest.resolve().is_relative_to(workspace.resolve()): + return j(handler, {'error': f'Path traversal blocked: {safe_name}'}, status=403) + + # Deduplicate: append -1, -2, etc. if file already exists + if dest.exists(): + stem = dest.stem + suffix = dest.suffix + for idx in range(1, 1000): + candidate = safe_resolve_ws(target_dir, f'{stem}-{idx}{suffix}') + if not candidate.resolve().is_relative_to(workspace.resolve()): + return j(handler, {'error': 'Path traversal blocked'}, status=403) + if not candidate.exists(): + dest = candidate + break + else: + return j(handler, {'error': 'Too many uploads with the same filename'}, status=400) + + dest.write_bytes(file_bytes) + mime = mimetypes.guess_type(safe_name)[0] or 'application/octet-stream' + + # For archives, optionally extract into the target directory + is_archive = safe_name.lower().endswith(('.zip', '.tar.gz', '.tgz', '.tar.bz2', '.tar.xz')) + if is_archive: + import zipfile, tarfile, traceback as _extract_tb + try: + extraction = extract_archive(file_bytes, safe_name, target_dir) + # Remove the archive file after successful extraction + dest.unlink(missing_ok=True) + results.append({ + 'filename': safe_name, + 'path': str(extraction.get('dest', target_dir)), + 'size': len(file_bytes), + 'is_image': False, + 'extracted': True, + 'extracted_files': extraction.get('files', []), + 'extracted_count': extraction.get('extracted', 0), + }) + continue + except (zipfile.BadZipFile, tarfile.TarError, ValueError) as e: + # Extraction failed — remove the archive file (no partial + # content left behind) and surface the error to the user. + dest.unlink(missing_ok=True) + print(f'[webui] workspace upload extract error: {e}', flush=True) + results.append({ + 'filename': safe_name, + 'path': str(target_dir), + 'size': len(file_bytes), + 'mime': mime, + 'is_image': False, + 'extracted': False, + 'extract_error': str(e) or 'Archive extraction failed', + }) + continue + except Exception: + print('[webui] workspace upload extract error: ' + _extract_tb.format_exc(), flush=True) + dest.unlink(missing_ok=True) + results.append({ + 'filename': safe_name, + 'path': str(target_dir), + 'size': len(file_bytes), + 'mime': mime, + 'is_image': False, + 'extracted': False, + 'extract_error': 'Archive extraction failed', + }) + continue + + results.append({ + 'filename': safe_name, + 'path': str(dest), + 'size': dest.stat().st_size, + 'mime': mime, + 'is_image': mime.startswith('image/'), + 'extracted': False, + }) + + if len(results) == 1: + return j(handler, results[0]) + return j(handler, {'files': results, 'count': len(results)}) + except ValueError as e: + return j(handler, {'error': str(e)}, status=400) + except Exception: + print('[webui] workspace upload error: ' + _tb.format_exc(), flush=True) + return j(handler, {'error': 'Upload failed'}, status=500) diff --git a/static/index.html b/static/index.html index b73010f7..6d81ee1a 100644 --- a/static/index.html +++ b/static/index.html @@ -1334,11 +1334,13 @@ - + + +
diff --git a/static/style.css b/static/style.css index 8ad23043..2e28d45d 100644 --- a/static/style.css +++ b/static/style.css @@ -1962,6 +1962,7 @@ .breadcrumb-current{color:var(--text);font-weight:500;} .breadcrumb-sep{color:var(--border);margin:0 1px;font-size:11px;} .file-tree{flex:1;overflow-y:auto;padding:8px;} + .file-tree.drag-over-upload{outline:2px dashed var(--blue,#3b82f6);outline-offset:-2px;border-radius:6px;} .file-item{display:flex;align-items:center;gap:6px;padding:6px 10px;border-radius:8px;cursor:pointer;font-size:12px;color:var(--muted);transition:all .12s;min-width:0;} .file-item:hover{background:var(--hover-bg);color:var(--text);} .file-item.active{background:var(--accent-bg);color:var(--accent-text);} diff --git a/static/workspace.js b/static/workspace.js index 80825116..874931ed 100644 --- a/static/workspace.js +++ b/static/workspace.js @@ -768,3 +768,75 @@ function openInBrowser(){ const url=`api/file/raw?session_id=${encodeURIComponent(S.session.session_id)}&path=${encodeURIComponent(_previewCurrentPath)}&inline=1`; window.open(url,'_blank','noopener'); } + +// ── Workspace upload ────────────────────────────────────────────────── +function triggerWorkspaceUpload() { + const input = $('workspaceFileInput'); + if (!input) return; + input.value = ''; + input.onchange = async () => { + const files = input.files; + if (!files || !files.length) return; + for (const file of files) { + await uploadToWorkspace(file, S.currentDir || '.'); + } + if (S.session) loadDir(S.currentDir); + }; + input.click(); +} + +async function uploadToWorkspace(file, dir) { + if (!S.session) return; + const formData = new FormData(); + formData.append('session_id', S.session.session_id); + formData.append('path', dir || '.'); + formData.append('file', file, file.name); + try { + showToast(t('uploading') || 'Uploading\u2026', 2000); + const data = await api('/api/workspace/upload', { + method: 'POST', + body: formData, + headers: {}, + timeoutMs: 120000, + }); + if (data && data.error) { + showToast(data.error, 5000, 'error'); + } else { + showToast(t('uploaded') || ('Uploaded ' + (data.filename || file.name)), 2000); + } + } catch (e) { + showToast(t('upload_failed') || ('Upload failed: ' + e.message), 5000, 'error'); + } +} + +// Drag-and-drop files onto workspace file tree +if (typeof document !== 'undefined') { + const _wsUploadInit = () => { + const tree = $('fileTree'); + if (!tree) return; + tree.addEventListener('dragover', (e) => { + if (e.dataTransfer && e.dataTransfer.types && e.dataTransfer.types.includes('Files')) { + e.preventDefault(); + e.dataTransfer.dropEffect = 'copy'; + tree.classList.add('drag-over-upload'); + } + }); + tree.addEventListener('dragleave', () => { + tree.classList.remove('drag-over-upload'); + }); + tree.addEventListener('drop', async (e) => { + tree.classList.remove('drag-over-upload'); + if (!e.dataTransfer || !e.dataTransfer.types || !e.dataTransfer.types.includes('Files')) return; + e.preventDefault(); + for (const file of e.dataTransfer.files) { + await uploadToWorkspace(file, S.currentDir || '.'); + } + if (S.session) loadDir(S.currentDir); + }); + }; + if (document.readyState === 'loading') { + document.addEventListener('DOMContentLoaded', _wsUploadInit, {once: true}); + } else { + _wsUploadInit(); + } +} diff --git a/tests/test_workspace_upload.py b/tests/test_workspace_upload.py new file mode 100644 index 00000000..567fb075 --- /dev/null +++ b/tests/test_workspace_upload.py @@ -0,0 +1,400 @@ +""" +PR #3104: workspace upload endpoint tests. + +Covers the POST /api/workspace/upload handler: + - happy-path upload into workspace + - filename dedup (-1/-2 suffixes) + - path-traversal blocking (../ filename → 403) + - oversized body rejection (413) + - archive extraction containment (no member escapes workspace) + - zip-bomb cap (extraction rejects when total extracted > limit) +""" + +import io +import json +import os +import sys +import uuid +import urllib.request +import urllib.error +import pathlib +import zipfile + +sys.path.insert(0, str(pathlib.Path(__file__).parent.parent.parent)) + +from tests._pytest_port import BASE + + +# ── HTTP helpers (mirrored from test_sprint1.py) ────────────────────────── + +def get(path): + url = BASE + path + with urllib.request.urlopen(url, timeout=10) as r: + return json.loads(r.read()) + + +def post(path, body=None): + url = BASE + path + data = json.dumps(body or {}).encode() + req = urllib.request.Request(url, data=data, + headers={"Content-Type": "application/json"}) + try: + with urllib.request.urlopen(req, timeout=10) as r: + return json.loads(r.read()), r.status + except urllib.error.HTTPError as e: + return json.loads(e.read()), e.code + + +def post_multipart(path, fields, files): + """Post a multipart/form-data request. files: {name: (filename, bytes)}""" + boundary = uuid.uuid4().hex.encode() + body = b"" + for name, value in fields.items(): + body += b"--" + boundary + b"\r\n" + body += f'Content-Disposition: form-data; name="{name}"\r\n\r\n'.encode() + body += value.encode() + b"\r\n" + for name, (filename, data) in files.items(): + body += b"--" + boundary + b"\r\n" + body += f'Content-Disposition: form-data; name="{name}"; filename="{filename}"\r\n'.encode() + body += b"Content-Type: application/octet-stream\r\n\r\n" + body += data + b"\r\n" + body += b"--" + boundary + b"--\r\n" + req = urllib.request.Request(BASE + path, data=body, + headers={"Content-Type": f"multipart/form-data; boundary={boundary.decode()}"}) + try: + with urllib.request.urlopen(req, timeout=10) as r: + return json.loads(r.read()), r.status + except urllib.error.HTTPError as e: + return json.loads(e.read()), e.code + + +def make_session_tracked(created_list, ws=None): + """Create a session and register it with the cleanup fixture.""" + body = {} + if ws: + body["workspace"] = str(ws) + d, _ = post("/api/session/new", body) + sid = d["session"]["session_id"] + created_list.append(sid) + return sid, pathlib.Path(d["session"]["workspace"]) + + +def _make_zip(members: dict[str, bytes]) -> bytes: + """Create a zip archive in memory.""" + buf = io.BytesIO() + with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf: + for name, data in members.items(): + zf.writestr(name, data) + return buf.getvalue() + + +# ── Health check ────────────────────────────────────────────────────────── + +def test_health(): + """Server must be running and healthy.""" + data = get("/health") + assert data["status"] == "ok", f"health not ok: {data}" + + +# ── Workspace upload tests ──────────────────────────────────────────────── + +class TestWorkspaceUploadHappyPath: + + def test_upload_single_file(self, cleanup_test_sessions): + """Happy path: upload a file into the workspace root.""" + sid, ws = make_session_tracked(cleanup_test_sessions) + + content = b"hello workspace" + result, status = post_multipart( + "/api/workspace/upload", + {"session_id": sid, "path": ""}, + {"file": ("hello.txt", content)}, + ) + + assert status == 200, f"Upload failed {status}: {result}" + assert result["filename"] == "hello.txt" + assert result["size"] == len(content) + assert result["extracted"] is False + + # Verify file actually exists in the workspace + uploaded = ws / "hello.txt" + assert uploaded.exists(), f"File not found at {uploaded}" + assert uploaded.read_bytes() == content + + def test_upload_into_subdirectory(self, cleanup_test_sessions): + """Upload a file into a subdirectory within the workspace.""" + sid, ws = make_session_tracked(cleanup_test_sessions) + + content = b"nested file" + result, status = post_multipart( + "/api/workspace/upload", + {"session_id": sid, "path": "sub/dir"}, + {"file": ("nested.txt", content)}, + ) + + assert status == 200, f"Upload failed {status}: {result}" + assert result["filename"] == "nested.txt" + + uploaded = ws / "sub" / "dir" / "nested.txt" + assert uploaded.exists(), f"File not found at {uploaded}" + assert uploaded.read_bytes() == content + + def test_upload_image_mime_is_flagged(self, cleanup_test_sessions): + """Image uploads should have is_image=True.""" + sid, ws = make_session_tracked(cleanup_test_sessions) + + # Minimal valid PNG bytes + png = ( + b"\x89PNG\r\n\x1a\n" # PNG signature + b"\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01" + b"\x08\x02\x00\x00\x00\x90wS\xde" + b"\x00\x00\x00\x0cIDATx\x9cc\xf8\x0f\x00\x00\x01\x01\x00\x05\x18\xd8N" + b"\x00\x00\x00\x00IEND\xaeB`\x82" + ) + + result, status = post_multipart( + "/api/workspace/upload", + {"session_id": sid, "path": ""}, + {"file": ("icon.png", png)}, + ) + + assert status == 200, f"Upload failed {status}: {result}" + assert result["is_image"] is True + assert result["mime"] == "image/png" + + +class TestWorkspaceUploadDedup: + + def test_same_filename_produces_suffix(self, cleanup_test_sessions): + """Uploading the same filename twice produces -1 suffix on the second.""" + sid, ws = make_session_tracked(cleanup_test_sessions) + + # First upload + result1, status1 = post_multipart( + "/api/workspace/upload", + {"session_id": sid, "path": ""}, + {"file": ("report.pdf", b"first")}, + ) + assert status1 == 200 + + # Second upload — same filename + result2, status2 = post_multipart( + "/api/workspace/upload", + {"session_id": sid, "path": ""}, + {"file": ("report.pdf", b"second")}, + ) + assert status2 == 200 + + # Second file should have -1 suffix + assert result2["filename"] == "report-1.pdf" + + # Both files should exist with correct content + assert (ws / "report.pdf").read_bytes() == b"first" + assert (ws / "report-1.pdf").read_bytes() == b"second" + + def test_multiple_duplicates_increment(self, cleanup_test_sessions): + """Three uploads of same name produce -1 and -2 suffixes.""" + sid, ws = make_session_tracked(cleanup_test_sessions) + + for i, expected_name in enumerate(["file.txt", "file-1.txt", "file-2.txt"]): + result, status = post_multipart( + "/api/workspace/upload", + {"session_id": sid, "path": ""}, + {"file": ("file.txt", f"content {i}".encode())}, + ) + assert status == 200, f"Upload {i} failed {status}: {result}" + assert result["filename"] == expected_name + assert (ws / expected_name).exists() + + +class TestWorkspaceUploadPathTraversal: + + def test_dotdot_filename_blocked(self, cleanup_test_sessions): + """Filename containing ../ should be sanitized, not traverse.""" + sid, ws = make_session_tracked(cleanup_test_sessions) + + result, status = post_multipart( + "/api/workspace/upload", + {"session_id": sid, "path": ""}, + {"file": ("../outside.txt", b"escape attempt")}, + ) + + # The sanitizer converts ../ to ___ so this should succeed but with + # sanitized name. The real traversal test is the subpath parameter. + assert status == 200, f"Unexpected status {status}: {result}" + assert ".." not in result["filename"] + # File should be inside workspace + uploaded = ws / result["filename"] + assert uploaded.exists() + assert uploaded.is_relative_to(ws.resolve()) + + def test_traversal_via_subpath_blocked(self, cleanup_test_sessions): + """Subpath with ../../etc should be blocked with 400.""" + sid, ws = make_session_tracked(cleanup_test_sessions) + + result, status = post_multipart( + "/api/workspace/upload", + {"session_id": sid, "path": "../../etc"}, + {"file": ("safe.txt", b"safe")}, + ) + + # safe_resolve_ws raises ValueError on traversal → caught as 400 + assert status == 400, f"Expected 400, got {status}: {result}" + assert "error" in result + + def test_traversal_via_subpath_deep(self, cleanup_test_sessions): + """Subpath with .. buried inside should also be blocked.""" + sid, ws = make_session_tracked(cleanup_test_sessions) + + result, status = post_multipart( + "/api/workspace/upload", + {"session_id": sid, "path": "projects/../../etc/passwd"}, + {"file": ("safe.txt", b"safe")}, + ) + + assert status == 400, f"Expected 400, got {status}: {result}" + assert "error" in result + + +class TestWorkspaceUploadOversized: + + def test_oversized_file_gets_413(self, cleanup_test_sessions): + """File over MAX_UPLOAD_BYTES should be rejected with 413.""" + from api.config import MAX_UPLOAD_BYTES + + sid, ws = make_session_tracked(cleanup_test_sessions) + + big = b"x" * (MAX_UPLOAD_BYTES + 1024) # slightly over limit + try: + result, status = post_multipart( + "/api/workspace/upload", + {"session_id": sid, "path": ""}, + {"file": ("big.bin", big)}, + ) + assert status == 413, f"Expected 413, got {status}: {result}" + except (urllib.error.URLError, ConnectionResetError, BrokenPipeError): + # Server may close connection after reading Content-Length > limit + pass + + +class TestWorkspaceUploadArchive: + + def test_zip_extracts_into_subdirectory(self, cleanup_test_sessions): + """Zip dropped into subdir/ should extract under subdir/, not workspace root.""" + sid, ws = make_session_tracked(cleanup_test_sessions) + + zip_data = _make_zip({ + "readme.md": b"# Project", + "src/main.py": b"print('hello')", + }) + + result, status = post_multipart( + "/api/workspace/upload", + {"session_id": sid, "path": "projects"}, + {"file": ("vendor.zip", zip_data)}, + ) + + assert status == 200, f"Upload failed {status}: {result}" + assert result["extracted"] is True + assert result["extracted_count"] == 2 + + # Extraction should land under projects/vendor/ + extract_dir = ws / "projects" / "vendor" + assert extract_dir.is_dir(), f"Extraction dir not found at {extract_dir}" + assert (extract_dir / "readme.md").read_text() == "# Project" + assert (extract_dir / "src" / "main.py").read_text() == "print('hello')" + + # Archive file itself should be removed after extraction + assert not (ws / "projects" / "vendor.zip").exists() + + def test_zip_extracts_to_workspace_root_when_no_subpath(self, cleanup_test_sessions): + """Zip uploaded without subpath extracts to workspace root.""" + sid, ws = make_session_tracked(cleanup_test_sessions) + + zip_data = _make_zip({"notes.txt": b"workspace notes"}) + + result, status = post_multipart( + "/api/workspace/upload", + {"session_id": sid, "path": ""}, + {"file": ("notes.zip", zip_data)}, + ) + + assert status == 200, f"Upload failed {status}: {result}" + assert result["extracted"] is True + + extract_dir = ws / "notes" + assert extract_dir.is_dir() + assert (extract_dir / "notes.txt").read_text() == "workspace notes" + + def test_zip_slip_blocked(self, cleanup_test_sessions): + """Zip member with ../ path should be blocked (zip-slip protection).""" + sid, ws = make_session_tracked(cleanup_test_sessions) + + # Create a zip with a malicious member path + buf = io.BytesIO() + with zipfile.ZipFile(buf, "w") as zf: + # Add a member that tries to escape the extraction directory + info = zipfile.ZipInfo("../escape.txt") + zf.writestr(info, b"escaped") + + result, status = post_multipart( + "/api/workspace/upload", + {"session_id": sid, "path": ""}, + {"file": ("evil.zip", buf.getvalue())}, + ) + + # Extraction should fail (zip-slip blocked) + assert status == 200, f"Upload failed {status}: {result}" + assert result["extracted"] is False + assert "extract_error" in result + + # No file should have escaped the workspace + assert not (ws.parent / "escape.txt").exists() + + def test_corrupt_zip_surfaces_error(self, cleanup_test_sessions): + """A corrupt zip should be rejected with an error surfaced to the frontend.""" + sid, ws = make_session_tracked(cleanup_test_sessions) + + result, status = post_multipart( + "/api/workspace/upload", + {"session_id": sid, "path": ""}, + {"file": ("corrupt.zip", b"this is not a zip file at all")}, + ) + + assert status == 200, f"Upload failed {status}: {result}" + assert result["extracted"] is False + assert "extract_error" in result + + # Corrupt archive file should be removed + assert not (ws / "corrupt.zip").exists() + + def test_zip_bomb_cap_trips(self, cleanup_test_sessions, monkeypatch): + """When extraction exceeds the cap, it should be rejected and cleaned up.""" + from api import upload as upload_mod + + sid, ws = make_session_tracked(cleanup_test_sessions) + + # Set a very small extraction cap (100 bytes) + monkeypatch.setattr(upload_mod, "_MAX_EXTRACTED_BYTES", 100) + + # Create a zip with enough content to exceed 100 bytes when extracted + zip_data = _make_zip({ + "small.txt": b"x" * 80, # 80 bytes — under cap + "medium.txt": b"y" * 80, # +80 = 160 bytes — exceeds cap during extraction + }) + + result, status = post_multipart( + "/api/workspace/upload", + {"session_id": sid, "path": ""}, + {"file": ("bomb.zip", zip_data)}, + ) + + assert status == 200, f"Upload failed {status}: {result}" + assert result["extracted"] is False + assert "extract_error" in result + + # Archive should be removed on failure + assert not (ws / "bomb.zip").exists() + + # Cleanup: restore the original cap + monkeypatch.undo()