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>
53 lines
1.6 KiB
Python
53 lines
1.6 KiB
Python
"""Auth module — bcrypt password verification and session token management."""
|
|
|
|
import os
|
|
import secrets
|
|
|
|
import bcrypt
|
|
|
|
# AUTH_USER is read at import time (used by app.py to compare against the submitted email).
|
|
# AUTH_PASS_HASH is intentionally NOT cached here — verify_password() reads it fresh from
|
|
# os.environ at call time so runtime env changes (e.g. secret rotation) are respected.
|
|
AUTH_USER: str = os.environ.get("AUTH_USER", "admin@localhost")
|
|
|
|
# In-memory session store: token -> email
|
|
_sessions: dict[str, str] = {}
|
|
|
|
|
|
def verify_password(password: str) -> bool:
|
|
"""Return True if password matches the bcrypt hash in AUTH_PASS_HASH env var.
|
|
|
|
Reads AUTH_PASS_HASH from env at call time so runtime env changes are respected.
|
|
Returns False if no hash is configured, if password is empty, or on any error.
|
|
"""
|
|
if not password:
|
|
return False
|
|
|
|
hash_str = os.environ.get("AUTH_PASS_HASH", "")
|
|
if not hash_str:
|
|
return False
|
|
|
|
try:
|
|
return bcrypt.checkpw(password.encode(), hash_str.encode())
|
|
except (ValueError, TypeError):
|
|
return False
|
|
|
|
|
|
def create_session_token(email: str) -> str:
|
|
"""Generate a session token, store it mapped to email, and return the token."""
|
|
token = secrets.token_hex(32)
|
|
_sessions[token] = email
|
|
return token
|
|
|
|
|
|
def validate_session_token(token: str) -> str | None:
|
|
"""Return the email for a valid token, or None if token is empty/unknown."""
|
|
if not token:
|
|
return None
|
|
return _sessions.get(token)
|
|
|
|
|
|
def invalidate_session_token(token: str) -> None:
|
|
"""Remove a session token from the store."""
|
|
_sessions.pop(token, None)
|