Validate a remote media URL to reduce SSRF risk. We block localhost/private/link-local/reserved IPs (including via DNS), and we only allow http(s) URLs.
(url: str)
| 12674 | for block in aesgcm_decrypt_generator(fp): out.write(block) |
| 12675 | else: archive.write(fp, arcname=arcname) |
| 12676 | return zip_path |
| 12677 | |
| 12678 | |
| 12679 | def _send_temporary_zip(zip_path: str, download_name: str): |
| 12680 | response = send_file(zip_path, as_attachment=True, download_name=download_name, conditional=True, max_age=0) |
| 12681 | response.headers["Cache-Control"] = "no-store" |
| 12682 | def cleanup(): |
| 12683 | try: |
| 12684 | if os.path.exists(zip_path): |
| 12685 | os.remove(zip_path) |
| 12686 | except Exception: |
| 12687 | pass |
| 12688 | response.call_on_close(cleanup) |
| 12689 | return response |
| 12690 | |
| 12691 | |
| 12692 | @app.route("/files/bulk", methods=["POST"]) |
| 12693 | @login_required |
| 12694 | def file_bulk_action(): |
| 12695 | entries = list(dict.fromkeys(request.form.getlist("entry")))[:500]; action = request.form.get("action", "download"); return_p = safe_relpath(request.form.get("return_p", "") or "") |
| 12696 | if not entries: flash("Select at least one entry."); return redirect(url_for("files", p=return_p)) |
| 12697 | if action == "download": |
| 12698 | try: |
| 12699 | path = _zip_paths(current_user(), entries, "selected") |
| 12700 | except Exception: |
| 12701 | flash("Not enough temporary storage to create the ZIP archive.") |
| 12702 | return redirect(url_for("files", p=return_p)) |
| 12703 | _log_file_activity(current_user(), "bulk_download", return_p, f"{len(entries)} entries") |
| 12704 | return _send_temporary_zip(path, "ButSystem-selected-files.zip") |
| 12705 | if action == "delete": |
| 12706 | count = 0 |
| 12707 | for rel in entries: |
| 12708 | try: |
| 12709 | rel = safe_relpath(rel); target = abs_user_path(current_user(), rel) |
| 12710 | if os.path.isdir(target): shutil.rmtree(target) |
| 12711 | else: os.remove(target) |
| 12712 | _delete_file_metadata(current_user(), rel); count += 1 |
| 12713 | except Exception: pass |
| 12714 | _log_file_activity(current_user(), "bulk_delete", return_p, f"{count} entries"); flash(f"Deleted {count} entries."); return redirect(url_for("files", p=return_p)) |
| 12715 | if action == "move": |
| 12716 | destination_rel = safe_relpath(request.form.get("destination", "") or ""); destination_dir = abs_user_path(current_user(), destination_rel); count = 0 |
| 12717 | if not os.path.isdir(destination_dir): flash("Invalid destination."); return redirect(url_for("files", p=return_p)) |
| 12718 | for rel in entries: |
| 12719 | try: |
| 12720 | rel = safe_relpath(rel); source = abs_user_path(current_user(), rel) |
| 12721 | if os.path.isdir(source) and (destination_dir == source or destination_dir.startswith(source + os.sep)): continue |
| 12722 | destination = _unique_destination(destination_dir, os.path.basename(source)); shutil.move(source, destination); new_rel = os.path.relpath(destination, user_root(current_user())).replace(os.sep, "/"); _remap_file_metadata(current_user(), rel, new_rel); count += 1 |
| 12723 | except Exception: pass |
| 12724 | _log_file_activity(current_user(), "bulk_move", destination_rel, f"{count} entries"); flash(f"Moved {count} entries."); return redirect(url_for("files", p=destination_rel)) |
| 12725 | flash("Unknown bulk action."); return redirect(url_for("files", p=return_p)) |
no test coverage detected