feat: security hardening (filename sanitization, bucket validation, file size enforcement)

.env.example

@@ -21,6 +21,7 @@ ENCRYPTION_KEY=replace_with_fernet_key
 
 # Upload behavior
 PRESIGNED_URL_EXPIRY=3600
+MAX_FILE_SIZE_BYTES=5368709120
 UPLOAD_CLEANUP_INTERVAL_SECONDS=300
 
 # Optional mock mode

backend/app/config.py

@@ -1,8 +1,17 @@
 from pydantic_settings import BaseSettings
 from functools import lru_cache
+from pathlib import Path
 from pydantic import Field, field_validator
 
 
+_BACKEND_ROOT = Path(__file__).resolve().parents[1]
+_REPO_ROOT = Path(__file__).resolve().parents[2]
+_ENV_FILES = (
+    str(_REPO_ROOT / ".env"),
+    str(_BACKEND_ROOT / ".env"),
+)
+
+
 class Settings(BaseSettings):
     # Legacy-only fallbacks; multipart upload routes resolve per-user bucket credentials from MongoDB.
     AWS_ACCESS_KEY_ID: str = ""
@@ -15,6 +24,7 @@ class Settings(BaseSettings):
     USE_MOCK_S3: bool = False
     MOCK_S3_STATE_FILE: str = "tmp/mock_s3_state.json"
     MOCK_S3_PART_FAILURE_RATE: float = Field(default=0.0, ge=0.0, le=1.0)
+    MAX_FILE_SIZE_BYTES: int = Field(default=5 * 1024 * 1024 * 1024, ge=1)
     UPLOAD_CLEANUP_INTERVAL_SECONDS: int = Field(default=300, ge=60, le=86400)
 
     # MongoDB + Auth Config
@@ -49,7 +59,7 @@ def validate_mongo_uri(cls, value: str) -> str:
     def cors_origins(self) -> list[str]:
         return [origin.strip() for origin in self.CORS_ALLOW_ORIGINS.split(",") if origin.strip()]
 
-    model_config = {"env_file": ".env", "env_file_encoding": "utf-8", "extra": "allow"}
+    model_config = {"env_file": _ENV_FILES, "env_file_encoding": "utf-8", "extra": "allow"}
 
 
 @lru_cache()

backend/app/mock_s3_service.py

@@ -190,6 +190,62 @@ def generate_presigned_get_url(file_key: str) -> str:
     return f"mock://{file_key}"
 
 
+def delete_object(file_key: str) -> dict:
+    with _state_lock:
+        state = _load_state()
+        removed = False
+
+        for upload_id in list(state.get("completed", {}).keys()):
+            completed_entry = state["completed"].get(upload_id) or {}
+            if completed_entry.get("file_key") == file_key:
+                state["completed"].pop(upload_id, None)
+                removed = True
+
+        for upload_id in list(state.get("uploads", {}).keys()):
+            upload_entry = state["uploads"].get(upload_id) or {}
+            if upload_entry.get("file_key") == file_key:
+                state["uploads"].pop(upload_id, None)
+                removed = True
+
+        if removed:
+            _save_state(state)
+
+    return {"message": "File deleted successfully", "file_key": file_key}
+
+
+def delete_prefix(prefix: str) -> dict:
+    normalized_prefix = (prefix or "").strip()
+    if not normalized_prefix:
+        raise ValueError("Invalid prefix")
+
+    with _state_lock:
+        state = _load_state()
+        deleted_count = 0
+
+        for upload_id in list(state.get("completed", {}).keys()):
+            completed_entry = state["completed"].get(upload_id) or {}
+            file_key = completed_entry.get("file_key") or ""
+            if file_key.startswith(normalized_prefix):
+                state["completed"].pop(upload_id, None)
+                deleted_count += 1
+
+        for upload_id in list(state.get("uploads", {}).keys()):
+            upload_entry = state["uploads"].get(upload_id) or {}
+            file_key = upload_entry.get("file_key") or ""
+            if file_key.startswith(normalized_prefix):
+                state["uploads"].pop(upload_id, None)
+                deleted_count += 1
+
+        if deleted_count > 0:
+            _save_state(state)
+
+    return {
+        "message": "Prefix deleted successfully",
+        "prefix": normalized_prefix,
+        "deleted_count": deleted_count,
+    }
+
+
 def get_upload(upload_id: str):
     with _state_lock:
         state = _load_state()