Upload a backup file for restoration.
(request)
| 258 | @permission_classes([IsAdmin]) |
| 259 | @parser_classes([MultiPartParser, FormParser]) |
| 260 | def upload_backup(request): |
| 261 | """Upload a backup file for restoration.""" |
| 262 | uploaded = request.FILES.get("file") |
| 263 | if not uploaded: |
| 264 | return Response( |
| 265 | {"detail": "No file uploaded"}, |
| 266 | status=status.HTTP_400_BAD_REQUEST, |
| 267 | ) |
| 268 | |
| 269 | try: |
| 270 | backup_dir = services.get_backup_dir() |
| 271 | # Sanitize filename: strip directory components to prevent path traversal |
| 272 | filename = Path(uploaded.name or "uploaded-backup.zip").name |
| 273 | if not filename: |
| 274 | filename = "uploaded-backup.zip" |
| 275 | |
| 276 | try: |
| 277 | safe_upload_path(filename, str(backup_dir)) |
| 278 | except ValueError: |
| 279 | return Response({"detail": "Invalid filename."}, status=status.HTTP_400_BAD_REQUEST) |
| 280 | |
| 281 | # Ensure unique filename |
| 282 | backup_file = (backup_dir / filename).resolve() |
| 283 | counter = 1 |
| 284 | while backup_file.exists(): |
| 285 | name_parts = filename.rsplit(".", 1) |
| 286 | if len(name_parts) == 2: |
| 287 | backup_file = backup_dir / f"{name_parts[0]}-{counter}.{name_parts[1]}" |
| 288 | else: |
| 289 | backup_file = backup_dir / f"{filename}-{counter}" |
| 290 | counter += 1 |
| 291 | |
| 292 | # Save uploaded file |
| 293 | with backup_file.open("wb") as f: |
| 294 | for chunk in uploaded.chunks(): |
| 295 | f.write(chunk) |
| 296 | |
| 297 | return Response( |
| 298 | { |
| 299 | "detail": "Backup uploaded successfully", |
| 300 | "filename": backup_file.name, |
| 301 | }, |
| 302 | status=status.HTTP_201_CREATED, |
| 303 | ) |
| 304 | except Exception as e: |
| 305 | return Response( |
| 306 | {"detail": f"Upload failed: {str(e)}"}, |
| 307 | status=status.HTTP_500_INTERNAL_SERVER_ERROR, |
| 308 | ) |
| 309 | |
| 310 | |
| 311 | @api_view(["POST"]) |
nothing calls this directly
no test coverage detected