feat(workspace): add file upload + drag-drop with archive extraction (#3104, @antoniocarlos97ss)
This commit is contained in:
nesquena-hermes
parent
383e0501db
commit
ff81591e8e
7 changed files with 619 additions and 3 deletions
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
138
api/upload.py
138
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)
|
||||
|
|
|
|||
|
|
@ -1334,11 +1334,13 @@
|
|||
<button class="panel-icon-btn has-tooltip has-tooltip--bottom" id="btnUpDir" data-tooltip="Parent directory" onclick="navigateUp()" style="display:none"><svg width="14" height="14" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2" stroke-linecap="round" stroke-linejoin="round" aria-hidden="true"><line x1="12" y1="19" x2="12" y2="5"/><polyline points="5 12 12 5 19 12"/></svg></button>
|
||||
<button class="panel-icon-btn has-tooltip has-tooltip--bottom" id="btnNewFile" data-tooltip="New file" onclick="promptNewFile()"><svg width="14" height="14" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2" stroke-linecap="round" stroke-linejoin="round" aria-hidden="true"><line x1="12" y1="5" x2="12" y2="19"/><line x1="5" y1="12" x2="19" y2="12"/></svg></button>
|
||||
<button class="panel-icon-btn has-tooltip has-tooltip--bottom" id="btnNewFolder" data-tooltip="New folder" onclick="promptNewFolder()"><svg width="14" height="14" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2" stroke-linecap="round" stroke-linejoin="round" aria-hidden="true"><path d="M22 19a2 2 0 0 1-2 2H4a2 2 0 0 1-2-2V5a2 2 0 0 1 2-2h5l2 3h9a2 2 0 0 1 2 2z"/></svg></button>
|
||||
<button class="panel-icon-btn has-tooltip has-tooltip--bottom" id="btnRefreshPanel" data-tooltip="Refresh" onclick="if(S.session)loadDir(S.currentDir)"><svg width="14" height="14" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2" stroke-linecap="round" stroke-linejoin="round" aria-hidden="true"><polyline points="23 4 23 10 17 10"/><polyline points="1 20 1 14 7 14"/><path d="M3.51 9a9 9 0 0 1 14.85-3.36L23 10M1 14l4.64 4.36A9 9 0 0 0 20.49 15"/></svg></button>
|
||||
<button class="panel-icon-btn has-tooltip has-tooltip--bottom" id="btnRefreshPanel" data-tooltip="Refresh" onclick="if(S.session)loadDir(S.currentDir)"><svg width="14" height="14" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2" stroke-linecap="round" stroke-linejoin="round" aria-hidden="true"><polyline points="23 4 23 10 17 10" /><polyline points="1 20 1 14 7 14" /><path d="M3.51 9a9 9 0 0 1 14.85-3.36L23 10M1 14l4.64 4.36A9 9 0 0 0 20.49 15" /></svg></button>
|
||||
<button class="panel-icon-btn has-tooltip has-tooltip--bottom" id="btnUploadWorkspace" data-tooltip="Upload file" onclick="triggerWorkspaceUpload()"><svg width="14" height="14" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2" stroke-linecap="round" stroke-linejoin="round" aria-hidden="true"><path d="M21 15v4a2 2 0 0 1-2 2H5a2 2 0 0 1-2-2v-4"/><polyline points="17 8 12 3 7 8"/><line x1="12" y1="3" x2="12" y2="15"/></svg></button>
|
||||
<button class="panel-icon-btn has-tooltip has-tooltip--bottom" id="btnWorkspacePrefs" data-tooltip="Workspace options" data-i18n-title="workspace_options" aria-haspopup="true" aria-expanded="false" onclick="toggleWorkspacePrefsMenu(event)"><svg width="14" height="14" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2" stroke-linecap="round" stroke-linejoin="round" aria-hidden="true"><circle cx="12" cy="5" r="1.5"/><circle cx="12" cy="12" r="1.5"/><circle cx="12" cy="19" r="1.5"/></svg><span class="workspace-prefs-dot" id="workspacePrefsDot" hidden></span></button>
|
||||
<button class="panel-icon-btn close-preview has-tooltip has-tooltip--bottom" id="btnClearPreview" data-tooltip="Close preview"><svg width="14" height="14" viewBox="0 0 24 24" fill="none" stroke="currentColor" stroke-width="2" stroke-linecap="round" stroke-linejoin="round" aria-hidden="true"><line x1="18" y1="6" x2="6" y2="18"/><line x1="6" y1="6" x2="18" y2="18"/></svg></button>
|
||||
</div>
|
||||
</div>
|
||||
<input type="file" id="workspaceFileInput" class="file-input-visually-hidden" multiple accept="image/*,text/*,.pdf,.json,.csv,.md,.py,.js,.ts,.yaml,.yml,.toml,.zip,.tar,.gz,.tgz,.bz2,.xz">
|
||||
<div class="workspace-panel-tabs" role="tablist" aria-label="Workspace panel views">
|
||||
<button class="workspace-panel-tab active" id="workspaceFilesTab" type="button" onclick="switchWorkspacePanelTab('files')" role="tab" aria-selected="true">Files</button>
|
||||
<button class="workspace-panel-tab" id="workspaceArtifactsTab" type="button" onclick="switchWorkspacePanelTab('artifacts')" role="tab" aria-selected="false">Artifacts <span id="workspaceArtifactsCount" class="workspace-artifacts-count">0</span></button>
|
||||
|
|
|
|||
|
|
@ -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);}
|
||||
|
|
|
|||
|
|
@ -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();
|
||||
}
|
||||
}
|
||||
|
|
|
|||
400
tests/test_workspace_upload.py
Normal file
400
tests/test_workspace_upload.py
Normal file
|
|
@ -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()
|
||||
Loading…
Reference in a new issue