Ken
5ba367af2f
- backend/app.py: add Depends(_require_auth) to /api/copilotkit endpoint (critical security fix — endpoint was fully unprotected; now requires valid session cookie) - backend/app.py: move 'from ag_ui.core import RunStartedEvent' from inline function body to top-level imports (style consistency) - backend/auth.py: remove dead AUTH_PASS_HASH module-level constant (verify_password already reads from os.environ at call time; the cached binding was unused dead code) - backend/artifacts.py: add path traversal guards to list_artifacts, get_artifact, and save_artifact — resolve paths and verify they stay within ARTIFACTS_DIR - entrypoint.sh: health-check loop now sets READY flag and exits 1 if amplifierd fails to start within 30s (previously fell through silently, causing 502s) - tests/test_app.py: add TestCopilotKitAuth — verifies unauthenticated requests to /api/copilotkit return 401; includes note explaining why streaming is not tested here - tests/test_structure.sh: update to reflect that Dockerfile and entrypoint.sh now exist (created in Tasks 7+8); convert absent-checks to presence-checks Co-authored-by: Amplifier <amplifier@anthropic.com>
62 lines
2.1 KiB
Python
62 lines
2.1 KiB
Python
"""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
|