fix: sanitize nasa_id and filenames to prevent path traversal (#1)

- Add sanitize_filename() to strip path separators and parent refs
- Apply sanitization to nasa_id used in metadata filenames
- Apply sanitization to image filenames derived from asset_href
- Preserve raw nasa_id for API calls to avoid breaking asset lookup
This commit is contained in:
Jarian Cottingham 2026-07-05 07:43:07 +00:00
parent 8e7f5bc920
commit 668f48897f

View File

@ -44,6 +44,7 @@ import argparse
import datetime import datetime
import json import json
import os import os
import re
import sys import sys
import time import time
from pathlib import Path from pathlib import Path
@ -180,6 +181,16 @@ def download_file(url: str, dest: Path) -> bool:
raise RuntimeError(f"Download error for {url}: {e}") from e raise RuntimeError(f"Download error for {url}: {e}") from e
def sanitize_filename(name: str) -> str:
"""Sanitize a string for use as a filename, preventing path traversal."""
# Strip path separators and parent directory references
sanitized = re.sub(r'[\/\\:\*\?"<>|]', "_", name)
sanitized = re.sub(r"^\.{1,2}($|_)", "item_", sanitized)
# Collapse multiple underscores
sanitized = re.sub(r"_+", "_", sanitized)
return sanitized.strip(".") or "item"
def save_metadata(item: Dict, dest: Path) -> None: def save_metadata(item: Dict, dest: Path) -> None:
"""Persist the full API item as formatted JSON.""" """Persist the full API item as formatted JSON."""
with dest.open("w", encoding="utf-8") as f: with dest.open("w", encoding="utf-8") as f:
@ -224,12 +235,13 @@ def main() -> None:
for idx, item in enumerate(tqdm(items, desc="Downloading"), start=1): for idx, item in enumerate(tqdm(items, desc="Downloading"), start=1):
data = item.get("data", [{}])[0] data = item.get("data", [{}])[0]
nasa_id = data.get("nasa_id") or data.get("title", f"item_{idx}") raw_nasa_id = data.get("nasa_id") or data.get("title", f"item_{idx}")
nasa_id = sanitize_filename(str(raw_nasa_id))
# ------------------------------------------------------------------ # ------------------------------------------------------------------
# Asset lookup: fetch highresolution image URL # Asset lookup: fetch highresolution image URL
# ------------------------------------------------------------------ # ------------------------------------------------------------------
asset_url = f"{API_ASSET_URL}/{nasa_id}" asset_url = f"{API_ASSET_URL}/{raw_nasa_id}"
try: try:
asset_resp = requests.get(asset_url, timeout=20) asset_resp = requests.get(asset_url, timeout=20)
asset_resp.raise_for_status() asset_resp.raise_for_status()
@ -243,7 +255,7 @@ def main() -> None:
print(f"[{idx}] Failed asset lookup for {nasa_id}: {e}", file=sys.stderr) print(f"[{idx}] Failed asset lookup for {nasa_id}: {e}", file=sys.stderr)
continue continue
filename = Path(asset_href).name filename = sanitize_filename(Path(asset_href).name)
image_path = IMG_DIR / filename image_path = IMG_DIR / filename
meta_path = META_DIR / f"index-{nasa_id}.json" meta_path = META_DIR / f"index-{nasa_id}.json"