"""Artifact storage module — per-session CRUD operations on the filesystem.""" from pathlib import Path ARTIFACTS_DIR = Path("/app/artifacts") def list_artifacts(session_id: str) -> list[dict]: """Return sorted list of artifact dicts for a session. Each dict contains: name, session_id, size. Returns [] if the session directory does not exist or if session_id would resolve outside ARTIFACTS_DIR (path traversal guard). """ session_dir = (ARTIFACTS_DIR / session_id).resolve() if not session_dir.is_relative_to(ARTIFACTS_DIR.resolve()): return [] if not session_dir.exists(): return [] return sorted( [ { "name": f.name, "session_id": session_id, "size": f.stat().st_size, } for f in session_dir.iterdir() if f.is_file() ], key=lambda d: d["name"], ) def get_artifact(session_id: str, name: str) -> str | None: """Return file content (utf-8) or None if artifact does not exist. Returns None (silently) if the resolved path would escape ARTIFACTS_DIR (path traversal guard). """ artifact_path = (ARTIFACTS_DIR / session_id / name).resolve() if not artifact_path.is_relative_to(ARTIFACTS_DIR.resolve()): return None if not artifact_path.is_file(): return None return artifact_path.read_text(encoding="utf-8") def save_artifact(session_id: str, name: str, content: str) -> Path: """Write content to the artifact file, creating parent dirs as needed. Returns the Path of the written file. Raises ValueError if the resolved path would escape ARTIFACTS_DIR (path traversal guard). """ session_dir = ARTIFACTS_DIR / session_id artifact_path = (session_dir / name).resolve() if not artifact_path.is_relative_to(ARTIFACTS_DIR.resolve()): raise ValueError(f"Path traversal detected in artifact name: {name!r}") session_dir.mkdir(parents=True, exist_ok=True) artifact_path.write_text(content, encoding="utf-8") return artifact_path