diff --git a/memanto/app/routes/auth_deps.py b/memanto/app/routes/auth_deps.py index 7cdb55e64..5ffd5c9ee 100644 --- a/memanto/app/routes/auth_deps.py +++ b/memanto/app/routes/auth_deps.py @@ -2,8 +2,15 @@ Authentication Dependencies for V2 API Shared authentication utilities to avoid circular imports. + +Security notes: +- Management endpoints require an explicit management credential. +- Session endpoints require a validated X-Session-Token or HttpOnly cookie. +- Management authorization does not trust client IP/loopback status. """ +import secrets + from fastapi import Cookie, Header, HTTPException, Request, Response from memanto.app.models.session import Session @@ -19,15 +26,11 @@ def set_session_cookie( - response: Response, session_token: str, request: Request + response: Response, + session_token: str, + request: Request, ) -> None: - """Store the browser UI session token outside JavaScript-readable state. - - MEMANTO defaults to binding 0.0.0.0 with no built-in TLS (see docker-compose.yml - and Settings.HOST), so a hardcoded Secure=True would silently stop browsers from - ever sending the cookie back over the plain-HTTP deployment this ships with by - default. Mark it Secure only when the current request actually arrived over HTTPS. - """ + """Set the authenticated browser session cookie.""" response.set_cookie( SESSION_COOKIE_NAME, session_token, @@ -39,195 +42,219 @@ def set_session_cookie( def clear_session_cookie(response: Response) -> None: - """Clear the browser UI session cookie.""" - response.delete_cookie(SESSION_COOKIE_NAME, path="/") + """Remove the authenticated browser session cookie.""" + response.delete_cookie( + SESSION_COOKIE_NAME, + path="/", + ) def get_moorcheh_api_key() -> str: - """ - Get Moorcheh API key from server configuration. - - Returns: - API key (or a placeholder string when running against the on-prem - backend, which does not require an API key). - - Raises: - HTTPException: If cloud is selected and no key is configured. - """ + """Return the configured server-side Moorcheh credential.""" from memanto.app.clients.backend import Backend, parse_backend from memanto.app.config import settings - if parse_backend(settings.MEMANTO_BACKEND) == Backend.ON_PREM: - # On-prem talks to localhost; routes that take ``moorcheh_api_key`` as - # a dependency no longer use it for outbound calls (they go through - # ``get_moorcheh_client()``), but the FastAPI signatures still need a - # string. Return a placeholder so the dependency resolves. + backend = parse_backend(settings.MEMANTO_BACKEND) + + if backend == Backend.ON_PREM: return "on-prem" - if settings.MOORCHEH_API_KEY: - return settings.MOORCHEH_API_KEY + api_key = settings.MOORCHEH_API_KEY - raise HTTPException( - status_code=500, - detail="Server misconfigured: MOORCHEH_API_KEY is not set", - ) + if not api_key: + raise HTTPException( + status_code=500, + detail="Server misconfigured: MOORCHEH_API_KEY is not set", + ) + + return api_key def _extract_presented_credential( authorization: str | None, x_api_key: str | None, ) -> str | None: - """Extract a client-presented management credential from request headers.""" - if x_api_key and x_api_key.strip(): - return x_api_key.strip() + """Extract a management credential from supported headers.""" + + if x_api_key: + value = x_api_key.strip() + + if value: + return value + if authorization: - parts = authorization.split(None, 1) - if len(parts) == 2 and parts[0].lower() == "bearer" and parts[1].strip(): - return parts[1].strip() + parts = authorization.strip().split(None, 1) + + if len(parts) == 2: + scheme, credential = parts + + if ( + scheme.lower() == "bearer" + and credential.strip() + ): + return credential.strip() + return None -def _is_loopback_host(host: str | None) -> bool: - """Return True when *host* is a loopback address (IPv4/IPv6/mapped).""" - if not host: - return False - import ipaddress +def _get_expected_management_credential() -> str | None: + """Return the credential authorized to manage agents.""" - try: - addr = ipaddress.ip_address(host) - except ValueError: - return False - if addr.is_loopback: - return True - ipv4_mapped = getattr(addr, "ipv4_mapped", None) - return ipv4_mapped is not None and ipv4_mapped.is_loopback + from memanto.app.clients.backend import Backend, parse_backend + from memanto.app.config import settings + + backend = parse_backend(settings.MEMANTO_BACKEND) + + if backend == Backend.ON_PREM: + secret = (settings.MEMANTO_SECRET_KEY or "").strip() + return secret or None + + api_key = (settings.MOORCHEH_API_KEY or "").strip() + return api_key or None def require_management_access( request: Request, authorization: str | None = Header(None), - x_api_key: str | None = Header(None, alias="X-Api-Key"), + x_api_key: str | None = Header( + None, + alias="X-Api-Key", + ), ) -> str: - """Authorize agent-lifecycle / management endpoints. - - MEMANTO is a single-tenant companion service. Agent create/list/delete/ - activate endpoints previously only checked that the *server* had a - configured API key, not that the *caller* was authorized. Combined with - the default ``HOST=0.0.0.0`` bind (see Settings / docker-compose), any - network peer could create agents, activate sessions, and obtain - ``session_token`` values for memory read/write. - - Access is granted when either: - - 1. The caller presents the server management credential - (``Authorization: Bearer `` or ``X-Api-Key``), matched with - ``secrets.compare_digest`` against the configured cloud API key, or - against ``MEMANTO_SECRET_KEY`` for on-prem; or - 2. The request originates from the loopback interface (local desktop - CLI / browser UX without forcing every local call to attach a key). - - Returns the server-side Moorcheh credential string used by downstream - service calls (same contract as ``get_moorcheh_api_key``). + """Authorize agent-management endpoints. + + Management access requires an explicit configured credential. + + Supported authentication: + + Authorization: Bearer + + or: + + X-Api-Key: + + No request IP address is treated as automatically trusted. """ - import secrets - from memanto.app.clients.backend import Backend, parse_backend - from memanto.app.config import settings + del request server_key = get_moorcheh_api_key() - presented = _extract_presented_credential(authorization, x_api_key) - backend = parse_backend(settings.MEMANTO_BACKEND) + expected = _get_expected_management_credential() - expected: str | None - if backend == Backend.ON_PREM: - # On-prem has no cloud API key; use the JWT/session secret as the - # management shared secret when one is configured. - expected = (settings.MEMANTO_SECRET_KEY or "").strip() or None - else: - expected = server_key if server_key and server_key != "on-prem" else None - - if presented and expected and secrets.compare_digest(presented, expected): - return server_key - - client_host = request.client.host if request.client else None - if _is_loopback_host(client_host): - return server_key - - raise HTTPException( - status_code=401, - detail=( - "Unauthorized. Agent management endpoints require either a " - "loopback client or a valid management credential " - "(Authorization: Bearer or X-Api-Key)." - ), + if expected is None: + raise HTTPException( + status_code=500, + detail=( + "Server misconfigured: management credential " + "is unavailable" + ), + ) + + presented = _extract_presented_credential( + authorization, + x_api_key, ) + if presented is None: + raise HTTPException( + status_code=401, + detail=( + "Unauthorized. Management credential required. " + "Use Authorization: Bearer or X-Api-Key." + ), + ) + + if not secrets.compare_digest( + presented, + expected, + ): + raise HTTPException( + status_code=401, + detail="Unauthorized. Invalid management credential.", + ) + + return server_key + def verify_moorcheh_api_key( request: Request, authorization: str | None = Header(None), - x_api_key: str | None = Header(None, alias="X-Api-Key"), + x_api_key: str | None = Header( + None, + alias="X-Api-Key", + ), ) -> str: - """Authorize management access and return the server Moorcheh credential. + """Compatibility wrapper for existing FastAPI dependencies.""" - Kept as a thin wrapper so existing ``Depends(verify_moorcheh_api_key)`` - call sites pick up the new authorization rules without signature churn - at every route. - """ - return require_management_access(request, authorization, x_api_key) + return require_management_access( + request=request, + authorization=authorization, + x_api_key=x_api_key, + ) def get_current_session( request: Request, response: Response, x_session_token: str | None = Header(None), - session_cookie: str | None = Cookie(None, alias=SESSION_COOKIE_NAME), + session_cookie: str | None = Cookie( + None, + alias=SESSION_COOKIE_NAME, + ), ) -> Session: - """ - Get and validate current session - - Args: - x_session_token: Session token header + """Validate and return the authenticated session. - Returns: - Validated Session - - Raises: - HTTPException: If session is invalid or expired + The agent identity is always obtained from the validated session token. + A caller cannot supply a separate agent_id through this dependency. """ + session_token = x_session_token or session_cookie + if not session_token: raise HTTPException( - status_code=401, detail="Missing session token. Use X-Session-Token header." + status_code=401, + detail=( + "Missing session token. " + "Use X-Session-Token header." + ), ) session_service = get_session_service() try: - token_payload = session_service.validate_session(session_token) + token_payload = session_service.validate_session( + session_token + ) - # Get session from storage - session = session_service.get_session(token_payload.agent_id) - if not session: + # Identity comes only from the validated session. + agent_id = token_payload.agent_id + + session = session_service.get_session(agent_id) + + if session is None: raise SessionNotFoundError( - f"Session for agent {token_payload.agent_id} not found" + f"Session for agent {agent_id} not found" ) - # Auto-renew session if near expiry renewed = session_service.check_and_auto_renew( - agent_id=token_payload.agent_id, + agent_id=agent_id, ) - if renewed: + + if renewed is not None: session = renewed - # The renewed session gets a new session_id/token, invalidating - # the one the caller just presented. Browser callers authenticate - # via the HttpOnly cookie (never re-read the token in JS), so - # without this the cookie goes stale and the very next request - # fails signature/session_id validation. + if session_cookie: - set_session_cookie(response, renewed.session_token, request) + set_session_cookie( + response=response, + session_token=renewed.session_token, + request=request, + ) return session - except (SessionExpiredError, SessionNotFoundError, InvalidSessionTokenError) as e: - raise map_error_to_http_exception(e) + except ( + SessionExpiredError, + SessionNotFoundError, + InvalidSessionTokenError, + ) as exc: + raise map_error_to_http_exception(exc) from exc \ No newline at end of file