Skip to content

Dre #1876

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
adyogyy-creator wants to merge 2 commits into
base: main
Choose a base branch
from
Open

Dre #1876

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
1222366
Update auth_deps.py
adyogyy-creator Aug 19, 2026
a05f444
Update auth_deps.py
adyogyy-creator Aug 19, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
293 changes: 160 additions & 133 deletions memanto/app/routes/auth_deps.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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,
Expand All @@ -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 <key>`` 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 <key>

or:

X-Api-Key: <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 <key> 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 <key> 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
Loading