fix: code review fixes — auth gap, path traversal, dead code, health-check timeout
- 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>
This commit is contained in:
+19
-6
@@ -9,9 +9,12 @@ 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.
|
||||
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
|
||||
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(
|
||||
@@ -29,9 +32,15 @@ def list_artifacts(session_id: str) -> list[dict]:
|
||||
|
||||
|
||||
def get_artifact(session_id: str, name: str) -> str | None:
|
||||
"""Return file content (utf-8) or None if the artifact does not exist."""
|
||||
artifact_path = ARTIFACTS_DIR / session_id / name
|
||||
if not artifact_path.exists():
|
||||
"""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")
|
||||
|
||||
@@ -40,9 +49,13 @@ 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 = session_dir / name
|
||||
artifact_path.write_text(content, encoding="utf-8")
|
||||
return artifact_path
|
||||
|
||||
Reference in New Issue
Block a user