Add durable control-plane state storage

[arbazkhan971]

/claim #161

Summary

• Adds a shared Mongo-backed control-plane store for MCP temp tokens, OAuth authorization codes, admin sessions, and per-identity rate limits.
• Keeps in-memory fallback for dev/test, but production mode now refuses to silently fall back when Mongo is unavailable.
• Tightens API-key handling so the bearer-token path fails cleanly instead of silently degrading in production.

Validation

• uv run --python /Users/arbaz/.local/bin/python3.11 pytest -q tests/unit/test_control_plane_store.py tests/unit/test_database_stores.py tests/api/test_dependencies_and_routes.py
• git diff --check

Notes

• This is a first pass at the control-plane durability issue. The repo does not currently have a Redis/shared-cache layer, so the shared Mongo-backed store is the smallest viable production-safe path here.

[gemini-code-assist]

Code Review

This pull request replaces various in-memory storage mechanisms (for rate limiting, admin sessions, and OAuth tokens) with a centralized, MongoDB-backed ControlPlaneStore to support multi-instance deployments. It also introduces stricter error handling in production, ensuring that durable storage is used. Review feedback focused on improving timestamp consistency by capturing the current time once per operation and adopting datetime.now(timezone.utc) over the deprecated utcnow(). Additionally, suggestions were made to pass the database-generated expiration timestamps back to the API responses to avoid redundant calculations and potential drift.

[gemini-code-assist]

The expires_at timestamp returned by control_plane_store.create_temp_token is currently ignored. Returning it from this helper allows the caller to use the exact timestamp stored in the database, avoiding redundant calculations and potential drift.

Suggested change

	def _create_mcp_temp_token(user_id: str) -> str:
	"""Create and store a temporary token for the user."""
	token = _generate_mcp_temp_token()
	expires_at = datetime.utcnow() + timedelta(minutes=TEMP_TOKEN_TTL_MINUTES)
	_mcp_temp_tokens[token] = {
	"user_id": user_id,
	"created_at": datetime.utcnow(),
	"expires_at": expires_at,
	"exchanged": False,
	}
	token, _ = control_plane_store.create_temp_token(
	user_id=user_id,
	ttl_minutes=TEMP_TOKEN_TTL_MINUTES,
	prefix=TEMP_TOKEN_PREFIX,
	)
	return token
	def _create_mcp_temp_token(user_id: str) -> tuple[str, datetime]:
	"""Create and store a temporary token for the user."""
	return control_plane_store.create_temp_token(
	user_id=user_id,
	ttl_minutes=TEMP_TOKEN_TTL_MINUTES,
	prefix=TEMP_TOKEN_PREFIX,
	)

[gemini-code-assist]

Use the expires_at timestamp returned by _create_mcp_temp_token instead of recalculating it. This ensures consistency between the response and the stored state.

Suggested change

	try:
	temp_token = _create_mcp_temp_token(user_id)
	except RuntimeError:
	raise HTTPException(
	status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
	detail="Authentication storage unavailable",
	)
	return MCPTempTokenResponse(
	temp_token=temp_token,
	expires_in=TEMP_TOKEN_TTL_MINUTES * 60,
	expires_at=_mcp_temp_tokens[temp_token]["expires_at"]
	expires_at=datetime.utcnow() + timedelta(minutes=TEMP_TOKEN_TTL_MINUTES)
	)
	try:
	temp_token, expires_at = _create_mcp_temp_token(user_id)
	except RuntimeError:
	raise HTTPException(
	status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
	detail="Authentication storage unavailable",
	)
	return MCPTempTokenResponse(
	temp_token=temp_token,
	expires_in=TEMP_TOKEN_TTL_MINUTES * 60,
	expires_at=expires_at
	)

[gemini-code-assist]

Include timezone in the imports to support creating timezone-aware datetime objects, which is the recommended approach over the deprecated utcnow().

Suggested change

	from datetime import datetime, timedelta
	from datetime import datetime, timedelta, timezone

[gemini-code-assist]

Capture the current time once at the start of the method. This ensures that created_at and expires_at are perfectly consistent and avoids multiple calls to the clock. Additionally, prefer datetime.now(timezone.utc) as utcnow() is deprecated.

Suggested change

	def create_temp_token(self, user_id: str, ttl_minutes: int, prefix: str = "xm-temp-") -> Tuple[str, datetime]:
	token = _random_token(prefix, 32)
	expires_at = datetime.utcnow() + timedelta(minutes=ttl_minutes)
	token_hash = _hash_secret(token)
	doc = {
	"token_hash": token_hash,
	"user_id": user_id,
	"created_at": datetime.utcnow(),
	"expires_at": expires_at,
	}
	def create_temp_token(self, user_id: str, ttl_minutes: int, prefix: str = "xm-temp-") -> Tuple[str, datetime]:
	now = datetime.now(timezone.utc)
	token = _random_token(prefix, 32)
	expires_at = now + timedelta(minutes=ttl_minutes)
	token_hash = _hash_secret(token)
	doc = {
	"token_hash": token_hash,
	"user_id": user_id,
	"created_at": now,
	"expires_at": expires_at,
	}

[gemini-code-assist]

Capture the current time once to ensure consistency between the session token generation, created_at, and expires_at fields.

Suggested change

	def create_admin_session(self, user: Dict[str, Any], ttl_hours: int = 24) -> str:
	token = _hash_secret(f"{user.get('username', 'admin')}:{secrets.token_hex(32)}:{time.time()}")
	expires_at = datetime.utcnow() + timedelta(hours=ttl_hours)
	doc = {
	"session_hash": token,
	"user": user,
	"created_at": datetime.utcnow(),
	"expires_at": expires_at,
	}
	def create_admin_session(self, user: Dict[str, Any], ttl_hours: int = 24) -> str:
	now = datetime.now(timezone.utc)
	token = _hash_secret(f"{user.get('username', 'admin')}:{secrets.token_hex(32)}:{now.timestamp()}")
	expires_at = now + timedelta(hours=ttl_hours)
	doc = {
	"session_hash": token,
	"user": user,
	"created_at": now,
	"expires_at": expires_at,
	}