Add Postgres-to-CSV export script and refactor shared export utilities

[rudokemper]

Goal

Add a Postgres-to-CSV export script to our script library, enabling Windmill operators to generate CSV files directly—without needing to query the Postgres database manually.

What I changed

• Refactored f/common_logic/db_connection.py ➜ f/common_logic/db_operations.py to include a new function for fetching data from Postgres, in addition to connection logic.
• Created f/common_logic/save-disk.py, which provides a save_export_file function that supports different file types (currently: csv, json, and geojson).
• postgres_to_geojson.py was updated to use the shared utility functions, and a new script postgres_to_csv.py was added using the same pattern.
• Added a test for postgres_to_csv using the same CoMapeo mock data as the GeoJSON script.

👀 @nicopace

[nicopace]

It looks great Rudo.
Have not run it, did an overview and left some comments.

[nicopace]

this line can be vulnerable to path traversal attacks if the input strings (db_table_name and file_type) come from an untrusted source and are not properly sanitized.

like:

file_path = storage_path / "../../etc/passwd.txt"

You mitigate this by:

1. Ensure storage_path is absolute and resolved: storage_path = storage_path.resolve() (double check, ChatGPT code)
2. Sanitize inputs

import re

def sanitize_filename(name):
    return re.sub(r"[^a-zA-Z0-9_-]", "_", name)

safe_name = sanitize_filename(db_table_name)
safe_type = sanitize_filename(file_type)
file_path = storage_path / f"{safe_name}.{safe_type}"

3. Check final path is within allowed directory

final_path = (storage_path / f"{safe_name}.{safe_type}").resolve()

# Ensure the path is within the storage directory
if not str(final_path).startswith(str(storage_path)):
    raise ValueError("Attempted path traversal detected!")

[nicopace]

A simpler version of this would be:

from pathlib import Path

def get_safe_file_path(storage_path, db_table_name, file_type):
    # Build path safely
    storage_path = Path(storage_path).resolve()
    file_path = (storage_path / f"{db_table_name}.{file_type}").resolve()

    # Check the resolved path stays within the storage directory
    if not file_path.is_relative_to(storage_path):
        raise ValueError("Invalid path: possible path traversal detected.")

    return file_path

[IamJeffG]

He does already validate file_type, just below.

And also I do like get_safe_file_path ⬆️

[rudokemper]

Thanks. I think at some point we could benefit from a repo-wide security audit, informed by any guardrails that Windmill has in place to prevent this kind of thing from happening already. But this being a shared code module, I opted to add this as a best practice 👍

[IamJeffG]

He does already validate file_type, just below.

And also I do like get_safe_file_path ⬆️