Add password reset, revocable API keys, usage analytics, and admin panel

Adds forgot/reset password flow with dev-mode email logging, multiple
named/revocable API keys per user (replacing the single legacy key model
while keeping it working), per-tool-call usage analytics recorded from the
MCP call_tool handler, and a separate admin API (auth/queries/routes) for
managing all users and viewing system-wide analytics — authenticated via a
hardcoded operator identity, fully isolated from regular user auth.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
MOHAN 2026-07-01 17:05:01 +05:30
parent 198b5f1b90
commit 05babc6d99
17 changed files with 1040 additions and 9 deletions

View File

@ -28,3 +28,21 @@ CONFIRM_TOKEN_TTL=300
# ─── Audit Log ─────────────────────────────────────────────────────────────── # ─── Audit Log ───────────────────────────────────────────────────────────────
AUDIT_LOG_FILE=./audit.jsonl AUDIT_LOG_FILE=./audit.jsonl
AUDIT_LOG_ENABLED=true AUDIT_LOG_ENABLED=true
# ─── Password Reset Email ──────────────────────────────────────────────────────
# When true, reset links are logged instead of emailed (no SMTP needed for local dev)
SMTP_DEV_MODE=true
SMTP_HOST=
SMTP_PORT=587
SMTP_USER=
SMTP_PASSWORD=
SMTP_FROM_EMAIL=no-reply@odoomcp.cloud
# Used to build the reset-password link sent to users
FRONTEND_BASE_URL=http://localhost:3000
# ─── Admin Panel ────────────────────────────────────────────────────────────────
# Single hardcoded operator identity for /admin — not a row in the users table.
# Change these in production; never commit real values.
ADMIN_EMAIL=admin@example.com
ADMIN_PASSWORD=change-me-to-a-long-random-string-in-production
ADMIN_SESSION_TTL_HOURS=12

View File

View File

@ -0,0 +1,50 @@
"""Admin auth — single hardcoded operator identity from .env, no DB row.
Sessions are opaque tokens in an in-memory dict (mirrors server.py's
_mcp_sessions precedent for accepting in-memory-only session state in this
codebase). Restart invalidates all admin sessions acceptable for a
single-operator login; the operator just logs in again.
"""
from __future__ import annotations
import secrets
import threading
import time
from typing import Optional
from ..config import settings
_lock = threading.Lock()
_sessions: dict[str, float] = {} # token -> issued_at (time.monotonic())
class AdminAuthError(Exception):
pass
def admin_login(email: str, password: str) -> str:
"""Verify against settings.admin_email/admin_password. Returns a new
opaque session token on success. Raises AdminAuthError otherwise."""
ok_email = secrets.compare_digest(email.strip().lower(), settings.admin_email.strip().lower())
ok_pw = secrets.compare_digest(password, settings.admin_password)
if not (ok_email and ok_pw):
raise AdminAuthError("Invalid admin credentials.")
token = secrets.token_urlsafe(32)
with _lock:
_sessions[token] = time.monotonic()
return token
def verify_admin_session(token: str) -> bool:
"""True if token is a live, unexpired admin session."""
with _lock:
issued = _sessions.get(token)
if issued is None:
return False
if time.monotonic() - issued > settings.admin_session_ttl_hours * 3600:
_sessions.pop(token, None)
return False
return True
def admin_logout(token: str) -> None:
with _lock:
_sessions.pop(token, None)

View File

@ -0,0 +1,116 @@
"""System-wide analytics for the admin panel — same shape as
analytics/queries.py but aggregated across ALL users, not scoped to one.
Reuses that module's time_filter/BUCKET_FORMATS helpers rather than
duplicating the timestamp-normalization logic."""
from __future__ import annotations
from typing import Optional
from ..database import db_fetchone, db_fetchall
from ..analytics.queries import time_filter, BUCKET_FORMATS
def get_system_summary(since: Optional[str] = None, until: Optional[str] = None) -> dict:
where, params = time_filter(since, until)
users_row = db_fetchone(
"SELECT COUNT(*) AS total_users, COALESCE(SUM(is_active), 0) AS active_users FROM users"
)
calls_row = db_fetchone(
f"""SELECT
COUNT(*) AS total_calls,
COALESCE(SUM(success), 0) AS success_count,
COALESCE(SUM(1 - success), 0) AS error_count,
COALESCE(AVG(duration_ms), 0) AS avg_duration_ms,
COUNT(DISTINCT user_id) AS active_callers
FROM tool_call_events
WHERE 1=1{where}""",
tuple(params),
)
return {
"total_users": users_row["total_users"],
"active_users": users_row["active_users"],
"disabled_users": users_row["total_users"] - users_row["active_users"],
"total_calls": calls_row["total_calls"],
"success_count": calls_row["success_count"],
"error_count": calls_row["error_count"],
"avg_duration_ms": round(calls_row["avg_duration_ms"], 2),
"active_callers": calls_row["active_callers"],
}
def get_all_users_usage(
since: Optional[str] = None, until: Optional[str] = None,
page: int = 1, page_size: int = 25,
) -> dict:
"""Per-user aggregated stats for the admin users-list table. Single query
with correlated subqueries no N+1. The time filter only applies inside
the tool_call_events subquery; credential/key counts are point-in-time."""
where, params = time_filter(since, until)
total_row = db_fetchone("SELECT COUNT(*) AS n FROM users")
total = total_row["n"]
page = max(page, 1)
page_size = max(1, min(page_size, 200))
offset = (page - 1) * page_size
rows = db_fetchall(
f"""SELECT
u.id, u.email, u.created_at, u.is_active,
(SELECT COUNT(*) FROM odoo_credentials c WHERE c.user_id = u.id) AS credential_count,
(SELECT COUNT(*) FROM api_keys k WHERE k.user_id = u.id AND k.is_active = 1) AS active_key_count,
(SELECT COUNT(*) FROM tool_call_events t
WHERE t.user_id = u.id{where}) AS total_calls,
(SELECT MAX(t.started_at) FROM tool_call_events t
WHERE t.user_id = u.id) AS last_active_at
FROM users u
ORDER BY u.created_at DESC
LIMIT ? OFFSET ?""",
(*params, page_size, offset),
)
items = [
{
"id": r["id"], "email": r["email"], "created_at": r["created_at"],
"is_active": bool(r["is_active"]), "credential_count": r["credential_count"],
"active_key_count": r["active_key_count"], "total_calls": r["total_calls"],
"last_active_at": r["last_active_at"],
}
for r in rows
]
return {"items": items, "page": page, "page_size": page_size, "total": total}
def get_system_by_tool(since: Optional[str] = None, until: Optional[str] = None) -> list[dict]:
where, params = time_filter(since, until)
rows = db_fetchall(
f"""SELECT
tool_name,
COUNT(*) AS calls,
COUNT(DISTINCT user_id) AS unique_users,
COALESCE(SUM(1 - success), 0) AS error_count,
COALESCE(AVG(duration_ms), 0) AS avg_duration_ms
FROM tool_call_events
WHERE 1=1{where}
GROUP BY tool_name
ORDER BY calls DESC""",
tuple(params),
)
return [dict(zip(r.keys(), [round(v, 2) if isinstance(v, float) else v for v in r])) for r in rows]
def get_system_timeseries(
since: Optional[str] = None, until: Optional[str] = None, bucket: str = "day"
) -> list[dict]:
fmt = BUCKET_FORMATS.get(bucket, BUCKET_FORMATS["day"])
where, params = time_filter(since, until)
rows = db_fetchall(
f"""SELECT
strftime('{fmt}', started_at) AS period,
COUNT(*) AS calls,
COUNT(DISTINCT user_id) AS unique_users,
COALESCE(SUM(1 - success), 0) AS error_count
FROM tool_call_events
WHERE 1=1{where}
GROUP BY period
ORDER BY period""",
tuple(params),
)
return [dict(zip(r.keys(), [round(v, 2) if isinstance(v, float) else v for v in r])) for r in rows]

View File

@ -0,0 +1,145 @@
"""Admin-only REST routes — separate router, separate auth dependency.
Mounted in server.py alongside (not merged into) the regular api_router."""
from __future__ import annotations
from fastapi import APIRouter, HTTPException, Header
from pydantic import BaseModel
from typing import Optional
from ..auth.service import (
get_user_by_id, set_user_active, delete_user,
list_odoo_credentials, list_api_keys, revoke_api_key,
)
from ..registry import evict_user
from ..analytics import queries as analytics_queries
from . import queries as admin_queries
from .auth import admin_login, verify_admin_session, admin_logout, AdminAuthError
router = APIRouter(prefix="/api/admin")
class AdminLoginRequest(BaseModel):
# Plain str, not EmailStr — this is compared against a fixed config value,
# not used to send mail, so deliverability validation (which rejects
# reserved TLDs like .test/.local/.internal) would wrongly block valid
# operator-configured admin identities.
email: str
password: str
class SetStatusRequest(BaseModel):
is_active: bool
def _require_admin(authorization: Optional[str] = Header(None)) -> str:
"""Distinct from _require_auth in api/routes.py — never calls
get_user_by_api_key, so a regular user's mtom_... key can never satisfy
this dependency, and an admin token can never satisfy _require_auth
(different token format entirely, not in the users table)."""
if not authorization or not authorization.startswith("Bearer "):
raise HTTPException(status_code=401, detail="Missing admin session token.")
token = authorization.removeprefix("Bearer ").strip()
if not verify_admin_session(token):
raise HTTPException(status_code=401, detail="Invalid or expired admin session.")
return token
@router.post("/login")
def admin_login_route(body: AdminLoginRequest):
try:
token = admin_login(body.email, body.password)
except AdminAuthError as e:
raise HTTPException(status_code=401, detail=str(e))
return {"admin_token": token, "message": "Admin login successful."}
@router.post("/logout")
def admin_logout_route(authorization: Optional[str] = Header(None)):
token = _require_admin(authorization)
admin_logout(token)
return {"success": True}
@router.get("/users")
def admin_list_users(
page: int = 1, page_size: int = 25,
since: Optional[str] = None, until: Optional[str] = None,
authorization: Optional[str] = Header(None),
):
_require_admin(authorization)
return admin_queries.get_all_users_usage(since, until, page, page_size)
@router.get("/users/{user_id}")
def admin_get_user(user_id: int, authorization: Optional[str] = Header(None)):
_require_admin(authorization)
user = get_user_by_id(user_id)
if not user:
raise HTTPException(status_code=404, detail="User not found.")
creds = list_odoo_credentials(user_id)
keys = list_api_keys(user_id)
summary = analytics_queries.get_summary(user_id)
return {
"id": user.id, "email": user.email, "is_active": user.is_active,
"created_at": user.created_at,
"odoo_instances": [
{"instance_name": c.instance_name, "url": c.url, "database": c.database_name,
"username": c.odoo_username, "type": "api_key" if c.is_api_key else "password"}
for c in creds
],
"api_keys": [
{"id": k.id, "name": k.name, "key_prefix": k.key_prefix, "created_at": k.created_at,
"last_used_at": k.last_used_at, "revoked_at": k.revoked_at, "is_active": k.is_active}
for k in keys
],
"usage_summary": summary,
}
@router.patch("/users/{user_id}/status")
def admin_set_user_status(user_id: int, body: SetStatusRequest, authorization: Optional[str] = Header(None)):
_require_admin(authorization)
ok = set_user_active(user_id, body.is_active)
if not ok:
raise HTTPException(status_code=404, detail="User not found.")
return {"success": True, "user_id": user_id, "is_active": body.is_active}
@router.delete("/users/{user_id}")
def admin_delete_user(user_id: int, authorization: Optional[str] = Header(None)):
_require_admin(authorization)
ok = delete_user(user_id)
if not ok:
raise HTTPException(status_code=404, detail="User not found.")
evict_user(user_id)
return {"success": True, "message": f"User {user_id} deleted."}
@router.post("/users/{user_id}/keys/{key_id}/revoke")
def admin_revoke_key(user_id: int, key_id: int, authorization: Optional[str] = Header(None)):
_require_admin(authorization)
ok = revoke_api_key(user_id, key_id)
if not ok:
raise HTTPException(status_code=404, detail="API key not found or already revoked.")
return {"success": True}
@router.get("/analytics/summary")
def admin_analytics_summary(since: Optional[str] = None, until: Optional[str] = None,
authorization: Optional[str] = Header(None)):
_require_admin(authorization)
return admin_queries.get_system_summary(since, until)
@router.get("/analytics/by-tool")
def admin_analytics_by_tool(since: Optional[str] = None, until: Optional[str] = None,
authorization: Optional[str] = Header(None)):
_require_admin(authorization)
return {"tools": admin_queries.get_system_by_tool(since, until)}
@router.get("/analytics/timeseries")
def admin_analytics_timeseries(since: Optional[str] = None, until: Optional[str] = None,
bucket: str = "day", authorization: Optional[str] = Header(None)):
_require_admin(authorization)
if bucket not in ("hour", "day", "week"):
raise HTTPException(status_code=400, detail="bucket must be one of: hour, day, week")
return {"buckets": admin_queries.get_system_timeseries(since, until, bucket)}

View File

View File

@ -0,0 +1,211 @@
"""Read-side SQL for the usage-analytics dashboard — all queries scoped to a
single user_id, with optional [since, until) ISO-timestamp filtering."""
from __future__ import annotations
from datetime import datetime
from typing import Optional
from ..database import db_fetchone, db_fetchall
BUCKET_FORMATS = {
"hour": "%Y-%m-%d %H:00",
"day": "%Y-%m-%d",
"week": "%Y-%W",
}
def normalize_iso(value: str) -> str:
"""started_at is stored via Python's datetime.isoformat() (e.g. ...+00:00).
Client-supplied since/until filters may arrive as JS Date.toISOString()
(e.g. ...Z suffix) same instant, different string. Since filtering is a
plain string comparison against started_at, re-serialize any incoming
timestamp to match the stored format so comparisons are correct rather
than comparing mismatched suffixes byte-for-byte."""
try:
return datetime.fromisoformat(value.replace("Z", "+00:00")).isoformat()
except ValueError:
return value
def time_filter(since: Optional[str], until: Optional[str]) -> tuple[str, list]:
clauses = []
params: list = []
if since:
clauses.append("started_at >= ?")
params.append(normalize_iso(since))
if until:
clauses.append("started_at < ?")
params.append(normalize_iso(until))
return (" AND " + " AND ".join(clauses) if clauses else ""), params
def get_summary(user_id: int, since: Optional[str] = None, until: Optional[str] = None) -> dict:
where, params = time_filter(since, until)
row = db_fetchone(
f"""SELECT
COUNT(*) AS total_calls,
COALESCE(SUM(success), 0) AS success_count,
COALESCE(SUM(1 - success), 0) AS error_count,
COALESCE(AVG(duration_ms), 0) AS avg_duration_ms,
COALESCE(SUM(duration_ms), 0) AS total_duration_ms,
COUNT(DISTINCT tool_name) AS unique_tools,
COUNT(DISTINCT instance_name) AS unique_instances
FROM tool_call_events
WHERE user_id = ?{where}""",
(user_id, *params),
)
return {
"total_calls": row["total_calls"],
"success_count": row["success_count"],
"error_count": row["error_count"],
"avg_duration_ms": round(row["avg_duration_ms"], 2),
"total_duration_ms": row["total_duration_ms"],
"unique_tools": row["unique_tools"],
"unique_instances": row["unique_instances"],
}
def get_by_tool(user_id: int, since: Optional[str] = None, until: Optional[str] = None) -> list[dict]:
where, params = time_filter(since, until)
rows = db_fetchall(
f"""SELECT
tool_name,
COUNT(*) AS calls,
COALESCE(SUM(success), 0) AS success_count,
COALESCE(SUM(1 - success), 0) AS error_count,
COALESCE(AVG(duration_ms), 0) AS avg_duration_ms,
COALESCE(SUM(duration_ms), 0) AS total_duration_ms,
MAX(started_at) AS last_used_at
FROM tool_call_events
WHERE user_id = ?{where}
GROUP BY tool_name
ORDER BY calls DESC""",
(user_id, *params),
)
return [dict(zip(r.keys(), [round(v, 2) if isinstance(v, float) else v for v in r])) for r in rows]
def get_by_instance(user_id: int, since: Optional[str] = None, until: Optional[str] = None) -> list[dict]:
where, params = time_filter(since, until)
rows = db_fetchall(
f"""SELECT
instance_name,
COUNT(*) AS calls,
COALESCE(SUM(success), 0) AS success_count,
COALESCE(SUM(1 - success), 0) AS error_count,
COALESCE(AVG(duration_ms), 0) AS avg_duration_ms,
COALESCE(SUM(duration_ms), 0) AS total_duration_ms
FROM tool_call_events
WHERE user_id = ?{where}
GROUP BY instance_name
ORDER BY calls DESC""",
(user_id, *params),
)
return [dict(zip(r.keys(), [round(v, 2) if isinstance(v, float) else v for v in r])) for r in rows]
def get_timeseries(
user_id: int, since: Optional[str] = None, until: Optional[str] = None, bucket: str = "day"
) -> list[dict]:
fmt = BUCKET_FORMATS.get(bucket, BUCKET_FORMATS["day"])
where, params = time_filter(since, until)
rows = db_fetchall(
f"""SELECT
strftime('{fmt}', started_at) AS period,
COUNT(*) AS calls,
COALESCE(AVG(duration_ms), 0) AS avg_duration_ms,
COALESCE(SUM(1 - success), 0) AS error_count
FROM tool_call_events
WHERE user_id = ?{where}
GROUP BY period
ORDER BY period""",
(user_id, *params),
)
return [dict(zip(r.keys(), [round(v, 2) if isinstance(v, float) else v for v in r])) for r in rows]
def get_tool_detail(
user_id: int, tool_name: str, since: Optional[str] = None, until: Optional[str] = None
) -> dict:
where, params = time_filter(since, until)
row = db_fetchone(
f"""SELECT
COUNT(*) AS calls,
COALESCE(SUM(success), 0) AS success_count,
COALESCE(SUM(1 - success), 0) AS error_count,
COALESCE(AVG(duration_ms), 0) AS avg_duration_ms,
COALESCE(SUM(duration_ms), 0) AS total_duration_ms
FROM tool_call_events
WHERE user_id = ? AND tool_name = ?{where}""",
(user_id, tool_name, *params),
)
timeseries_rows = db_fetchall(
f"""SELECT
strftime('%Y-%m-%d', started_at) AS period,
COUNT(*) AS calls,
COALESCE(AVG(duration_ms), 0) AS avg_duration_ms,
COALESCE(SUM(1 - success), 0) AS error_count
FROM tool_call_events
WHERE user_id = ? AND tool_name = ?{where}
GROUP BY period
ORDER BY period""",
(user_id, tool_name, *params),
)
return {
"tool_name": tool_name,
"summary": {
"calls": row["calls"],
"success_count": row["success_count"],
"error_count": row["error_count"],
"avg_duration_ms": round(row["avg_duration_ms"], 2),
"total_duration_ms": row["total_duration_ms"],
},
"timeseries": [
dict(zip(r.keys(), [round(v, 2) if isinstance(v, float) else v for v in r]))
for r in timeseries_rows
],
}
def list_calls(
user_id: int,
page: int = 1,
page_size: int = 25,
tool_name: Optional[str] = None,
instance_name: Optional[str] = None,
success: Optional[bool] = None,
) -> dict:
clauses = ["user_id = ?"]
params: list = [user_id]
if tool_name:
clauses.append("tool_name = ?")
params.append(tool_name)
if instance_name:
clauses.append("instance_name = ?")
params.append(instance_name)
if success is not None:
clauses.append("success = ?")
params.append(int(success))
where = " AND ".join(clauses)
total_row = db_fetchone(f"SELECT COUNT(*) AS n FROM tool_call_events WHERE {where}", tuple(params))
total = total_row["n"]
page = max(page, 1)
page_size = max(1, min(page_size, 200))
offset = (page - 1) * page_size
rows = db_fetchall(
f"""SELECT id, tool_name, instance_name, started_at, duration_ms, success, error_message
FROM tool_call_events
WHERE {where}
ORDER BY started_at DESC
LIMIT ? OFFSET ?""",
(*params, page_size, offset),
)
items = [
{
"id": r["id"], "tool_name": r["tool_name"], "instance_name": r["instance_name"],
"started_at": r["started_at"], "duration_ms": r["duration_ms"],
"success": bool(r["success"]), "error_message": r["error_message"],
}
for r in rows
]
return {"items": items, "page": page, "page_size": page_size, "total": total}

View File

@ -0,0 +1,28 @@
"""Records per-tool-call events for usage analytics."""
from __future__ import annotations
import logging
from typing import Optional
from ..database import db_execute, db_commit
log = logging.getLogger(__name__)
def record_tool_call(
user_id: int,
api_key_id: Optional[int],
tool_name: str,
instance_name: str,
started_at: str,
duration_ms: int,
success: bool,
error_message: Optional[str] = None,
) -> None:
"""Single INSERT into tool_call_events. Callers must wrap this in
try/except analytics recording must never break a tool response."""
db_execute(
"""INSERT INTO tool_call_events
(user_id, api_key_id, tool_name, instance_name, started_at, duration_ms, success, error_message)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
(user_id, api_key_id, tool_name, instance_name, started_at, duration_ms, int(success), error_message),
)
db_commit()

View File

@ -8,8 +8,12 @@ from ..auth.service import (
AuthError, signup, login, get_user_by_api_key, AuthError, signup, login, get_user_by_api_key,
upsert_odoo_credential, list_odoo_credentials, upsert_odoo_credential, list_odoo_credentials,
delete_odoo_credential, regenerate_api_key, delete_odoo_credential, regenerate_api_key,
create_api_key, list_api_keys, revoke_api_key,
request_password_reset, reset_password,
) )
from ..registry import refresh_conn_manager, evict_user from ..registry import refresh_conn_manager, evict_user
from ..mailer import send_password_reset_email
from ..analytics import queries as analytics_queries
router = APIRouter(prefix="/api") router = APIRouter(prefix="/api")
@ -41,6 +45,19 @@ class CredentialRequest(BaseModel):
is_api_key: bool = True is_api_key: bool = True
class ForgotPasswordRequest(BaseModel):
email: EmailStr
class ResetPasswordRequest(BaseModel):
token: str
new_password: str
class CreateApiKeyRequest(BaseModel):
name: str
# ─── Auth dependency ────────────────────────────────────────────────────────── # ─── Auth dependency ──────────────────────────────────────────────────────────
def _require_auth(authorization: Optional[str] = Header(None)): def _require_auth(authorization: Optional[str] = Header(None)):
@ -144,6 +161,73 @@ def api_regen_key(authorization: Optional[str] = Header(None)):
} }
# ─── Password Reset ───────────────────────────────────────────────────────────
@router.post("/password/forgot")
def api_forgot_password(body: ForgotPasswordRequest):
"""Request a password reset link. Always returns 200 — never reveals
whether the email is registered."""
raw_token = request_password_reset(body.email)
if raw_token:
reset_link = f"{settings.frontend_base_url}/reset-password?token={raw_token}"
send_password_reset_email(body.email, reset_link)
return {"message": "If that email exists, a reset link was sent."}
@router.post("/password/reset")
def api_reset_password(body: ResetPasswordRequest):
"""Consume a reset token and set a new password. Also revokes all active
API keys for the account as a post-reset security measure."""
try:
reset_password(body.token, body.new_password)
except AuthError as e:
raise HTTPException(status_code=400, detail=str(e))
return {"message": "Password reset. All existing API keys were revoked — please log in again and issue a new key."}
# ─── API Key Management ───────────────────────────────────────────────────────
@router.get("/keys")
def api_list_keys(authorization: Optional[str] = Header(None)):
user = _require_auth(authorization)
keys = list_api_keys(user.id)
return {
"keys": [
{"id": k.id, "name": k.name, "key_prefix": k.key_prefix,
"created_at": k.created_at, "last_used_at": k.last_used_at,
"revoked_at": k.revoked_at, "is_active": k.is_active}
for k in keys
],
"total": len(keys),
}
@router.post("/keys")
def api_create_key(body: CreateApiKeyRequest, authorization: Optional[str] = Header(None)):
"""Create a new named API key. Shown once."""
user = _require_auth(authorization)
key_id, full_key = create_api_key(user.id, body.name)
return {
"id": key_id,
"name": body.name,
"api_key": full_key,
"warning": "Save this key — it will not be shown again.",
"mcp_connection": {
"url": f"{settings.public_url}/mcp/sse",
"header": f"Authorization: Bearer {full_key}",
},
}
@router.delete("/keys/{key_id}")
def api_revoke_key(key_id: int, authorization: Optional[str] = Header(None)):
user = _require_auth(authorization)
ok = revoke_api_key(user.id, key_id)
if not ok:
raise HTTPException(status_code=404, detail="API key not found or already revoked.")
return {"success": True, "message": "API key revoked."}
@router.post("/credentials") @router.post("/credentials")
def api_add_credential(body: CredentialRequest, authorization: Optional[str] = Header(None)): def api_add_credential(body: CredentialRequest, authorization: Optional[str] = Header(None)):
"""Add or update an Odoo instance credential.""" """Add or update an Odoo instance credential."""
@ -195,3 +279,63 @@ def api_delete_credential(instance_name: str, authorization: Optional[str] = Hea
raise HTTPException(status_code=404, detail=f"Instance '{instance_name}' not found.") raise HTTPException(status_code=404, detail=f"Instance '{instance_name}' not found.")
refresh_conn_manager(user.id) refresh_conn_manager(user.id)
return {"success": True, "message": f"Instance '{instance_name}' removed."} return {"success": True, "message": f"Instance '{instance_name}' removed."}
# ─── Usage Analytics ───────────────────────────────────────────────────────────
@router.get("/analytics/summary")
def api_analytics_summary(
since: Optional[str] = None, until: Optional[str] = None,
authorization: Optional[str] = Header(None),
):
user = _require_auth(authorization)
return analytics_queries.get_summary(user.id, since, until)
@router.get("/analytics/by-tool")
def api_analytics_by_tool(
since: Optional[str] = None, until: Optional[str] = None,
authorization: Optional[str] = Header(None),
):
user = _require_auth(authorization)
return {"tools": analytics_queries.get_by_tool(user.id, since, until)}
@router.get("/analytics/by-instance")
def api_analytics_by_instance(
since: Optional[str] = None, until: Optional[str] = None,
authorization: Optional[str] = Header(None),
):
user = _require_auth(authorization)
return {"instances": analytics_queries.get_by_instance(user.id, since, until)}
@router.get("/analytics/timeseries")
def api_analytics_timeseries(
since: Optional[str] = None, until: Optional[str] = None, bucket: str = "day",
authorization: Optional[str] = Header(None),
):
user = _require_auth(authorization)
if bucket not in ("hour", "day", "week"):
raise HTTPException(status_code=400, detail="bucket must be one of: hour, day, week")
return {"buckets": analytics_queries.get_timeseries(user.id, since, until, bucket)}
@router.get("/analytics/tools/{tool_name}")
def api_analytics_tool_detail(
tool_name: str, since: Optional[str] = None, until: Optional[str] = None,
authorization: Optional[str] = Header(None),
):
user = _require_auth(authorization)
return analytics_queries.get_tool_detail(user.id, tool_name, since, until)
@router.get("/logs")
def api_logs(
page: int = 1, page_size: int = 25,
tool_name: Optional[str] = None, instance_name: Optional[str] = None,
success: Optional[bool] = None,
authorization: Optional[str] = Header(None),
):
user = _require_auth(authorization)
return analytics_queries.list_calls(user.id, page, page_size, tool_name, instance_name, success)

View File

@ -1,5 +1,6 @@
"""Password hashing, API key generation, and Fernet encryption for credentials.""" """Password hashing, API key generation, and Fernet encryption for credentials."""
from __future__ import annotations from __future__ import annotations
import hashlib
import secrets import secrets
import bcrypt import bcrypt
from cryptography.fernet import Fernet from cryptography.fernet import Fernet
@ -32,6 +33,19 @@ def verify_api_key(plain: str, hashed: str) -> bool:
return bcrypt.checkpw(plain.encode(), hashed.encode()) return bcrypt.checkpw(plain.encode(), hashed.encode())
def generate_reset_token() -> tuple[str, str]:
"""Returns (raw_token, token_hash). raw_token goes in the reset link only;
token_hash is stored in the DB. Uses sha256 (not bcrypt) because this is a
high-entropy random token, not a low-entropy secret needing slow hashing."""
raw = secrets.token_urlsafe(32)
token_hash = hashlib.sha256(raw.encode()).hexdigest()
return raw, token_hash
def hash_reset_token(raw: str) -> str:
return hashlib.sha256(raw.encode()).hexdigest()
def _fernet() -> Fernet: def _fernet() -> Fernet:
key = settings.encryption_key key = settings.encryption_key
if key == "change-me": if key == "change-me":

View File

@ -23,3 +23,14 @@ class OdooCredential:
credential_enc: str credential_enc: str
is_api_key: bool is_api_key: bool
created_at: str created_at: str
@dataclass
class ApiKeyInfo:
id: int
name: str
key_prefix: str
created_at: str
last_used_at: Optional[str]
revoked_at: Optional[str]
is_active: bool

View File

@ -1,14 +1,17 @@
"""Auth business logic — signup, login, API key management, credential CRUD.""" """Auth business logic — signup, login, API key management, credential CRUD."""
from __future__ import annotations from __future__ import annotations
from typing import Optional from typing import Optional
from .models import User, OdooCredential from .models import User, OdooCredential, ApiKeyInfo
from .crypto import ( from .crypto import (
hash_password, verify_password, hash_password, verify_password,
generate_api_key, verify_api_key, generate_api_key, verify_api_key,
encrypt_credential, decrypt_credential, encrypt_credential, decrypt_credential,
generate_reset_token, hash_reset_token,
) )
from ..database import db_execute, db_commit, db_fetchone, db_fetchall from ..database import db_execute, db_commit, db_fetchone, db_fetchall
RESET_TOKEN_TTL_MINUTES = 30
class AuthError(Exception): class AuthError(Exception):
pass pass
@ -69,13 +72,26 @@ def login(email: str, password: str) -> User:
def get_user_by_api_key(api_key: str) -> Optional[User]: def get_user_by_api_key(api_key: str) -> Optional[User]:
"""Look up a user by their API key. Returns None if not found/invalid.""" """Look up a user by their API key. Returns None if not found/invalid.
The legacy users.api_key_hash path is checked first for backward
compatibility, but if that same key has since been backfilled into
api_keys (via list_api_keys) and explicitly revoked there, it must be
rejected here too otherwise "revoking" the legacy key in the UI would
silently do nothing since this function never looked at api_keys before.
"""
if not api_key or not api_key.startswith("mtom_"): if not api_key or not api_key.startswith("mtom_"):
return None return None
prefix = api_key[:12] prefix = api_key[:12]
rows = db_fetchall("SELECT * FROM users WHERE api_key_prefix = ? AND is_active = 1", (prefix,)) rows = db_fetchall("SELECT * FROM users WHERE api_key_prefix = ? AND is_active = 1", (prefix,))
for row in rows: for row in rows:
if verify_api_key(api_key, row["api_key_hash"]): if verify_api_key(api_key, row["api_key_hash"]):
revoked = db_fetchone(
"SELECT 1 FROM api_keys WHERE user_id = ? AND key_hash = ? AND is_active = 0",
(row["id"], row["api_key_hash"]),
)
if revoked:
return None
return _row_to_user(row) return _row_to_user(row)
return None return None
@ -96,6 +112,160 @@ def get_user_by_id(user_id: int) -> Optional[User]:
return _row_to_user(row) if row else None return _row_to_user(row) if row else None
# ─── Admin-only user management ────────────────────────────────────────────
def list_all_users() -> list[User]:
"""Admin-only: list every user, no ownership scoping."""
rows = db_fetchall("SELECT * FROM users ORDER BY created_at DESC")
return [_row_to_user(r) for r in rows]
def set_user_active(user_id: int, is_active: bool) -> bool:
"""Admin-only enable/disable. Takes effect immediately for new
get_user_by_api_key() lookups and new /api/login attempts (both already
check is_active) no changes needed to that existing logic."""
cur = db_execute("UPDATE users SET is_active = ? WHERE id = ?", (int(is_active), user_id))
db_commit()
return cur.rowcount > 0
def delete_user(user_id: int) -> bool:
"""Admin-only permanent delete. Cascading cleanup of odoo_credentials,
api_keys, password_reset_tokens, tool_call_events is already handled by
the ON DELETE CASCADE / SET NULL foreign keys declared in database.py."""
cur = db_execute("DELETE FROM users WHERE id = ?", (user_id,))
db_commit()
return cur.rowcount > 0
# ─── Multiple API Keys ──────────────────────────────────────────────────────
def create_api_key(user_id: int, name: str) -> tuple[int, str]:
"""Create a new named API key. Returns (key_id, full_key) — full_key shown once."""
full_key, key_hash, key_prefix = generate_api_key()
cur = db_execute(
"INSERT INTO api_keys (user_id, name, key_hash, key_prefix) VALUES (?, ?, ?, ?)",
(user_id, name, key_hash, key_prefix),
)
db_commit()
return cur.lastrowid, full_key
def list_api_keys(user_id: int) -> list[ApiKeyInfo]:
"""List all named API keys for a user. Lazily backfills a 'Default (legacy)'
row mirroring users.api_key_hash the first time this is called, so the
legacy signup key shows up in the UI like any other key."""
rows = db_fetchall(
"SELECT * FROM api_keys WHERE user_id = ? ORDER BY created_at", (user_id,)
)
if not rows:
user_row = db_fetchone("SELECT api_key_hash, api_key_prefix FROM users WHERE id = ?", (user_id,))
if user_row and user_row["api_key_hash"]:
db_execute(
"INSERT INTO api_keys (user_id, name, key_hash, key_prefix) VALUES (?, ?, ?, ?)",
(user_id, "Default (legacy)", user_row["api_key_hash"], user_row["api_key_prefix"]),
)
db_commit()
rows = db_fetchall(
"SELECT * FROM api_keys WHERE user_id = ? ORDER BY created_at", (user_id,)
)
return [_row_to_api_key(r) for r in rows]
def revoke_api_key(user_id: int, key_id: int) -> bool:
"""Revoke a named API key. Returns False if not found/not owned by user."""
cur = db_execute(
"""UPDATE api_keys SET revoked_at = datetime('now'), is_active = 0
WHERE id = ? AND user_id = ? AND is_active = 1""",
(key_id, user_id),
)
db_commit()
return cur.rowcount > 0
def get_api_key_id_for_mcp(api_key: str) -> Optional[int]:
"""Resolve which api_keys row (if any) this raw key belongs to, for
analytics attribution at MCP-session-start. Returns the row id whether
it's a named key or a backfilled 'Default (legacy)' row (see
list_api_keys). Returns None only if the key has never been backfilled
into api_keys at all harmless, just means analytics rows for that
session carry api_key_id=NULL until the user's first /api/keys view
triggers the backfill. The caller already validated the key via
get_user_by_api_key before calling this."""
if not api_key or not api_key.startswith("mtom_"):
return None
prefix = api_key[:12]
rows = db_fetchall(
"SELECT id, key_hash FROM api_keys WHERE key_prefix = ? AND is_active = 1 AND revoked_at IS NULL",
(prefix,),
)
for row in rows:
if verify_api_key(api_key, row["key_hash"]):
db_execute("UPDATE api_keys SET last_used_at = datetime('now') WHERE id = ?", (row["id"],))
db_commit()
return row["id"]
return None
def _row_to_api_key(row) -> ApiKeyInfo:
return ApiKeyInfo(
id=row["id"],
name=row["name"],
key_prefix=row["key_prefix"],
created_at=row["created_at"],
last_used_at=row["last_used_at"],
revoked_at=row["revoked_at"],
is_active=bool(row["is_active"]),
)
# ─── Password Reset ─────────────────────────────────────────────────────────
def request_password_reset(email: str) -> Optional[str]:
"""Look up user by email. If found, create a reset token (expires in
RESET_TOKEN_TTL_MINUTES) and return the raw token for the caller to email
or log. Returns None if the email isn't found — callers must respond
identically either way to avoid leaking which emails are registered."""
email = email.strip().lower()
row = db_fetchone("SELECT id FROM users WHERE email = ?", (email,))
if not row:
return None
raw_token, token_hash = generate_reset_token()
db_execute(
"""INSERT INTO password_reset_tokens (user_id, token_hash, expires_at)
VALUES (?, ?, datetime('now', ?))""",
(row["id"], token_hash, f"+{RESET_TOKEN_TTL_MINUTES} minutes"),
)
db_commit()
return raw_token
def reset_password(raw_token: str, new_password: str) -> None:
"""Verify the reset token and update the user's password. Raises AuthError
if the token is missing/expired/already used. On success, also revokes
all active API keys for that user as a post-reset security measure."""
if len(new_password) < 8:
raise AuthError("Password must be at least 8 characters.")
token_hash = hash_reset_token(raw_token)
row = db_fetchone(
"""SELECT * FROM password_reset_tokens
WHERE token_hash = ? AND used_at IS NULL AND expires_at > datetime('now')""",
(token_hash,),
)
if not row:
raise AuthError("Invalid or expired reset link.")
user_id = row["user_id"]
pw_hash = hash_password(new_password)
db_execute("UPDATE users SET password_hash = ? WHERE id = ?", (pw_hash, user_id))
db_execute("UPDATE password_reset_tokens SET used_at = datetime('now') WHERE id = ?", (row["id"],))
db_execute(
"UPDATE api_keys SET revoked_at = datetime('now'), is_active = 0 WHERE user_id = ? AND is_active = 1",
(user_id,),
)
db_commit()
def _row_to_user(row) -> User: def _row_to_user(row) -> User:
return User( return User(
id=row["id"], id=row["id"],

View File

@ -31,6 +31,20 @@ class Settings(BaseSettings):
audit_log_file: str = "./audit.jsonl" audit_log_file: str = "./audit.jsonl"
audit_log_enabled: bool = True audit_log_enabled: bool = True
# Password reset email — in dev mode the reset link is logged instead of emailed
smtp_dev_mode: bool = True
smtp_host: str = ""
smtp_port: int = 587
smtp_user: str = ""
smtp_password: str = ""
smtp_from_email: str = "no-reply@odoomcp.cloud"
frontend_base_url: str = "http://localhost:3000"
# Admin panel — single hardcoded operator identity, not tied to the users table
admin_email: str = "admin@example.com"
admin_password: str = "change-me"
admin_session_ttl_hours: int = 12
@property @property
def cors_origins_list(self) -> list[str]: def cors_origins_list(self) -> list[str]:
return [o.strip() for o in self.cors_origins.split(",") if o.strip()] return [o.strip() for o in self.cors_origins.split(",") if o.strip()]

View File

@ -1,7 +1,7 @@
"""Per-request context vars — each MCP session carries its own user state.""" """Per-request context vars — each MCP session carries its own user state."""
from __future__ import annotations from __future__ import annotations
from contextvars import ContextVar from contextvars import ContextVar
from typing import TYPE_CHECKING from typing import Optional, TYPE_CHECKING
if TYPE_CHECKING: if TYPE_CHECKING:
from .client.connection import ConnectionManager from .client.connection import ConnectionManager
@ -11,6 +11,7 @@ _user_id: ContextVar[int] = ContextVar("user_id")
_user_email: ContextVar[str] = ContextVar("user_email") _user_email: ContextVar[str] = ContextVar("user_email")
_conn_manager: ContextVar["ConnectionManager"] = ContextVar("conn_manager") _conn_manager: ContextVar["ConnectionManager"] = ContextVar("conn_manager")
_confirm_manager: ContextVar["ConfirmationManager"] = ContextVar("confirm_manager") _confirm_manager: ContextVar["ConfirmationManager"] = ContextVar("confirm_manager")
_api_key_id: ContextVar[Optional[int]] = ContextVar("api_key_id", default=None)
def get_user_id() -> int: def get_user_id() -> int:
@ -29,13 +30,18 @@ def get_confirm_manager() -> "ConfirmationManager":
return _confirm_manager.get() return _confirm_manager.get()
def set_user_context(user_id: int, user_email: str, conn_mgr, confirm_mgr): def get_api_key_id() -> Optional[int]:
return _api_key_id.get()
def set_user_context(user_id: int, user_email: str, conn_mgr, confirm_mgr, api_key_id: Optional[int] = None):
"""Set all context vars for the current MCP session. Returns tokens for reset.""" """Set all context vars for the current MCP session. Returns tokens for reset."""
t1 = _user_id.set(user_id) t1 = _user_id.set(user_id)
t2 = _user_email.set(user_email) t2 = _user_email.set(user_email)
t3 = _conn_manager.set(conn_mgr) t3 = _conn_manager.set(conn_mgr)
t4 = _confirm_manager.set(confirm_mgr) t4 = _confirm_manager.set(confirm_mgr)
return (t1, t2, t3, t4) t5 = _api_key_id.set(api_key_id)
return (t1, t2, t3, t4, t5)
def reset_user_context(tokens): def reset_user_context(tokens):
@ -43,3 +49,4 @@ def reset_user_context(tokens):
_user_email.reset(tokens[1]) _user_email.reset(tokens[1])
_conn_manager.reset(tokens[2]) _conn_manager.reset(tokens[2])
_confirm_manager.reset(tokens[3]) _confirm_manager.reset(tokens[3])
_api_key_id.reset(tokens[4])

View File

@ -47,6 +47,49 @@ def init_db():
created_at TEXT NOT NULL DEFAULT (datetime('now')), created_at TEXT NOT NULL DEFAULT (datetime('now')),
UNIQUE(user_id, instance_name) UNIQUE(user_id, instance_name)
); );
-- Multiple named, revocable API keys. users.api_key_hash/api_key_prefix
-- stay untouched that's the legacy key and keeps working as-is.
CREATE TABLE IF NOT EXISTS api_keys (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
name TEXT NOT NULL,
key_hash TEXT NOT NULL,
key_prefix TEXT NOT NULL,
created_at TEXT NOT NULL DEFAULT (datetime('now')),
last_used_at TEXT,
revoked_at TEXT,
is_active INTEGER NOT NULL DEFAULT 1
);
CREATE INDEX IF NOT EXISTS idx_api_keys_user ON api_keys(user_id);
CREATE INDEX IF NOT EXISTS idx_api_keys_prefix ON api_keys(key_prefix);
-- Password reset: single-use, short expiry, hashed token at rest.
CREATE TABLE IF NOT EXISTS password_reset_tokens (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
token_hash TEXT NOT NULL,
created_at TEXT NOT NULL DEFAULT (datetime('now')),
expires_at TEXT NOT NULL,
used_at TEXT
);
CREATE INDEX IF NOT EXISTS idx_pwreset_hash ON password_reset_tokens(token_hash);
-- SQL-queryable per-tool-call analytics. audit/logger.py's JSONL file
-- is left as-is for forensic/raw logs; this table is for dashboards.
CREATE TABLE IF NOT EXISTS tool_call_events (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
api_key_id INTEGER REFERENCES api_keys(id) ON DELETE SET NULL,
tool_name TEXT NOT NULL,
instance_name TEXT NOT NULL DEFAULT 'default',
started_at TEXT NOT NULL,
duration_ms INTEGER NOT NULL,
success INTEGER NOT NULL,
error_message TEXT
);
CREATE INDEX IF NOT EXISTS idx_tce_user_time ON tool_call_events(user_id, started_at);
CREATE INDEX IF NOT EXISTS idx_tce_user_tool ON tool_call_events(user_id, tool_name);
""") """)
db.commit() db.commit()

35
src/mt_odoo_mcp/mailer.py Normal file
View File

@ -0,0 +1,35 @@
"""Outbound email — currently just password reset links.
In local/dev environments (SMTP_DEV_MODE=true, the default), no SMTP server
is required: the reset link is logged instead of emailed, so the flow can be
tested end-to-end without any mail infrastructure.
"""
from __future__ import annotations
import logging
import smtplib
from email.message import EmailMessage
from .config import settings
log = logging.getLogger(__name__)
def send_password_reset_email(to_email: str, reset_link: str) -> None:
if settings.smtp_dev_mode:
log.info("PASSWORD RESET LINK for %s: %s", to_email, reset_link)
return
msg = EmailMessage()
msg["Subject"] = "Reset your OdooMCP Cloud password"
msg["From"] = settings.smtp_from_email
msg["To"] = to_email
msg.set_content(
f"We received a request to reset your password.\n\n"
f"Reset it here (link expires in 30 minutes):\n{reset_link}\n\n"
f"If you didn't request this, you can safely ignore this email."
)
with smtplib.SMTP(settings.smtp_host, settings.smtp_port) as smtp:
smtp.starttls()
if settings.smtp_user:
smtp.login(settings.smtp_user, settings.smtp_password)
smtp.send_message(msg)

View File

@ -6,6 +6,7 @@ import logging
import secrets as _secrets import secrets as _secrets
import sys import sys
import time import time
from datetime import datetime, timezone
from typing import Any from typing import Any
import uvicorn import uvicorn
@ -18,10 +19,11 @@ from mcp.types import Tool, TextContent
from .config import settings from .config import settings
from .database import init_db from .database import init_db
from .auth.service import get_user_by_api_key from .auth.service import get_user_by_api_key, get_api_key_id_for_mcp
from .registry import get_conn_manager, get_confirm_manager from .registry import get_conn_manager, get_confirm_manager
from .context import set_user_context, reset_user_context from .context import set_user_context, reset_user_context, get_user_id, get_api_key_id
from .api.routes import router as api_router from .api.routes import router as api_router
from .admin.routes import router as admin_router
# Tool implementations # Tool implementations
from .tools import ( from .tools import (
@ -569,6 +571,21 @@ async def list_tools() -> list[Tool]:
return [Tool(name=name, description=desc, inputSchema=schema) for name, desc, _, schema in TOOLS] return [Tool(name=name, description=desc, inputSchema=schema) for name, desc, _, schema in TOOLS]
def _record_tool_call(tool_name, instance, started_monotonic, started_at_iso, success, error_text):
"""Best-effort analytics recording — must never break a tool response."""
try:
from .analytics.recorder import record_tool_call
record_tool_call(
user_id=get_user_id(), api_key_id=get_api_key_id(),
tool_name=tool_name, instance_name=instance,
started_at=started_at_iso,
duration_ms=int((time.monotonic() - started_monotonic) * 1000),
success=success, error_message=error_text,
)
except Exception:
log.debug("analytics recording failed (non-fatal)", exc_info=True)
@mcp_app.call_tool() @mcp_app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]: async def call_tool(name: str, arguments: dict) -> list[TextContent]:
entry = next((t for t in TOOLS if t[0] == name), None) entry = next((t for t in TOOLS if t[0] == name), None)
@ -577,6 +594,8 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]:
_, _, fn, _ = entry _, _, fn, _ = entry
instance = arguments.get("instance", "default") instance = arguments.get("instance", "default")
started = time.monotonic()
started_at_iso = datetime.now(timezone.utc).isoformat()
# Auto-connect instance (skip for tools that explicitly handle connection errors) # Auto-connect instance (skip for tools that explicitly handle connection errors)
from .context import get_conn_manager as _gcm from .context import get_conn_manager as _gcm
@ -584,6 +603,7 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]:
_gcm().ensure_connected(instance) _gcm().ensure_connected(instance)
except Exception as e: except Exception as e:
if name not in ("list_instances", "health_check", "connect_instance"): if name not in ("list_instances", "health_check", "connect_instance"):
_record_tool_call(name, instance, started, started_at_iso, False, str(e))
return [TextContent(type="text", text=json.dumps( return [TextContent(type="text", text=json.dumps(
{"error": f"Cannot connect to Odoo instance '{instance}': {e}. " {"error": f"Cannot connect to Odoo instance '{instance}': {e}. "
f"Add your credentials via POST /api/credentials."}))] f"Add your credentials via POST /api/credentials."}))]
@ -591,11 +611,14 @@ async def call_tool(name: str, arguments: dict) -> list[TextContent]:
try: try:
result = fn(**arguments) result = fn(**arguments)
except TypeError as e: except TypeError as e:
_record_tool_call(name, instance, started, started_at_iso, False, f"Invalid arguments: {e}")
return [TextContent(type="text", text=json.dumps({"error": f"Invalid arguments: {e}"}))] return [TextContent(type="text", text=json.dumps({"error": f"Invalid arguments: {e}"}))]
except Exception as e: except Exception as e:
log.exception("Tool '%s' raised an error", name) log.exception("Tool '%s' raised an error", name)
_record_tool_call(name, instance, started, started_at_iso, False, str(e))
return [TextContent(type="text", text=json.dumps({"error": str(e)}))] return [TextContent(type="text", text=json.dumps({"error": str(e)}))]
_record_tool_call(name, instance, started, started_at_iso, True, None)
return [TextContent(type="text", text=json.dumps(result, indent=2, default=str))] return [TextContent(type="text", text=json.dumps(result, indent=2, default=str))]
@ -627,6 +650,7 @@ app.add_middleware(
# Mount the REST API routes # Mount the REST API routes
app.include_router(api_router) app.include_router(api_router)
app.include_router(admin_router)
@app.get("/health") @app.get("/health")
@ -735,6 +759,7 @@ class _MCPEndpoint:
resp = _JSON({"error": "Invalid API key."}, status_code=401) resp = _JSON({"error": "Invalid API key."}, status_code=401)
await resp(scope, receive, send) await resp(scope, receive, send)
return return
api_key_id = get_api_key_id_for_mcp(api_key)
# Session ID must be visible ASCII chars; hex satisfies that # Session ID must be visible ASCII chars; hex satisfies that
new_session_id = _secrets.token_hex(32) new_session_id = _secrets.token_hex(32)
@ -749,9 +774,9 @@ class _MCPEndpoint:
async def _run_session( async def _run_session(
_t=transport, _u=user, _cm=conn_mgr, _fm=confirm_mgr, _t=transport, _u=user, _cm=conn_mgr, _fm=confirm_mgr,
_ev=ready, _sid=new_session_id, _ev=ready, _sid=new_session_id, _kid=api_key_id,
): ):
tokens = set_user_context(_u.id, _u.email, _cm, _fm) tokens = set_user_context(_u.id, _u.email, _cm, _fm, _kid)
try: try:
async with _t.connect() as (read_stream, write_stream): async with _t.connect() as (read_stream, write_stream):
_ev.set() _ev.set()