fix(automation): address code review - path traversal fix and test marker
- Sanitise uploaded filenames with os.path.basename() to prevent path traversal - Change TestWebhookDispatchIntegration marker from unit to integration Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
This commit is contained in:
@@ -274,12 +274,17 @@ def action_upload(
|
|||||||
if not file.filename:
|
if not file.filename:
|
||||||
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Filename is required")
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Filename is required")
|
||||||
|
|
||||||
|
# Sanitise filename to prevent path traversal attacks
|
||||||
|
safe_filename = os.path.basename(file.filename)
|
||||||
|
if not safe_filename:
|
||||||
|
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Filename is required")
|
||||||
|
|
||||||
owner_id = user.get("preferred_username") or user.get("email") or user.get("id", "automation")
|
owner_id = user.get("preferred_username") or user.get("email") or user.get("id", "automation")
|
||||||
workdir = settings.workdir or tempfile.gettempdir()
|
workdir = settings.workdir or tempfile.gettempdir()
|
||||||
upload_dir = os.path.join(workdir, "uploads")
|
upload_dir = os.path.join(workdir, "uploads")
|
||||||
os.makedirs(upload_dir, exist_ok=True)
|
os.makedirs(upload_dir, exist_ok=True)
|
||||||
|
|
||||||
dest_path = os.path.join(upload_dir, file.filename)
|
dest_path = os.path.join(upload_dir, safe_filename)
|
||||||
try:
|
try:
|
||||||
contents = file.file.read()
|
contents = file.file.read()
|
||||||
with open(dest_path, "wb") as f:
|
with open(dest_path, "wb") as f:
|
||||||
@@ -295,12 +300,12 @@ def action_upload(
|
|||||||
|
|
||||||
result = process_document.delay(dest_path, owner_id)
|
result = process_document.delay(dest_path, owner_id)
|
||||||
task_id = result.id
|
task_id = result.id
|
||||||
logger.info("Automation upload queued: file=%s, task=%s, owner=%s", file.filename, task_id, owner_id)
|
logger.info("Automation upload queued: file=%s, task=%s, owner=%s", safe_filename, task_id, owner_id)
|
||||||
except Exception as exc:
|
except Exception as exc:
|
||||||
logger.warning("Could not queue processing task (Celery may be unavailable): %s", exc)
|
logger.warning("Could not queue processing task (Celery may be unavailable): %s", exc)
|
||||||
|
|
||||||
return {
|
return {
|
||||||
"status": "accepted",
|
"status": "accepted",
|
||||||
"filename": file.filename,
|
"filename": safe_filename,
|
||||||
"task_id": task_id,
|
"task_id": task_id,
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -231,7 +231,7 @@ class TestDeliverAutomationHookTask:
|
|||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.unit
|
@pytest.mark.integration
|
||||||
class TestWebhookDispatchIntegration:
|
class TestWebhookDispatchIntegration:
|
||||||
"""Test that dispatch_webhook_event also triggers automation hooks."""
|
"""Test that dispatch_webhook_event also triggers automation hooks."""
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user