diff --git a/.gitignore b/.gitignore
index 8e4ddd7..1ea1171 100644
--- a/.gitignore
+++ b/.gitignore
@@ -159,3 +159,7 @@ pytest-cache-files-*/
# Project specific
# ====================================
# Add project-specific ignores here
+.worktrees/
+worktrees/
+/.agents
+pytest-out.txt
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 42f43e1..8734e5c 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -7,6 +7,43 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
---
+## [3.1.0] - 2026-02-23
+
+### Live Platform Foundation (Track E.1 - E.3)
+
+Major release introducing the Live Platform architecture — SQLite database, OAuth social login, site management, and per-user MCP endpoints. All features are included in the Community Edition.
+
+#### Added
+- **SQLite Database Backend** (E.1): Async SQLite via aiosqlite, WAL mode, schema versioning with migrations framework, CRUD for users/sites/API keys/connection tokens
+- **Credential Encryption** (E.1): AES-256-GCM with HKDF per-site key derivation, versioned wire format for future migration support
+- **OAuth Social Login** (E.2): GitHub + Google OAuth 2.0, CSRF state tokens, JWT session management, dual session types (admin + oauth_user)
+- **Site Management API** (E.3): Plugin credential definitions for all 9 plugins, connection validation, encrypted site CRUD, MAX_SITES_PER_USER=10
+- **User API Keys** (E.3): bcrypt-hashed keys with `mhu_` prefix, 8-char prefix for indexed lookup, 5-minute validation cache
+- **Per-User MCP Endpoints** (E.3): Direct JSON-RPC handler at `/u/{user_id}/{alias}/mcp`, per-user rate limiting (30/min, 500/hr)
+- **Config Snippets** (E.3): Auto-generated config for Claude Desktop, Claude Code, Cursor, VS Code, and ChatGPT
+- **Dashboard Pages**: My Sites (list/add/test/delete), Connect (API keys + config snippets), Profile, OAuth Login
+- **Dark/Light Mode Toggle**: Theme switcher across all dashboard pages
+- **RBAC**: Role-based access control for dashboard
+- **Active Health Checks**: Background health monitoring for connected services
+
+#### Fixed
+- CSRF middleware body consumption bug
+- OAuth log noise and DCR crash on startup
+- WordPress site connection validation (uses aiohttp to match plugin client)
+- Tenant isolation enforced on all site queries
+
+#### Security
+- All bare `except:` replaced with `except Exception:` across 12 files
+- Network error differentiation (DNS, SSL, timeout, connection refused)
+- Retry with exponential backoff for transient errors only
+- Auth/config errors never retried
+
+#### Tests
+- 452 total tests (up from 303), all passing
+- New test suites: database (37), encryption (27), user_auth (32), site_api (17), user_keys (14), user_endpoints (12), config_snippets (10)
+
+---
+
## [2.9.0] - 2026-02-14
### Project Revival - Dependency Updates & Documentation Sync
diff --git a/core/__init__.py b/core/__init__.py
index 7cca9fe..5b84546 100644
--- a/core/__init__.py
+++ b/core/__init__.py
@@ -45,12 +45,10 @@ from core.rate_limiter import RateLimitConfig, RateLimiter, get_rate_limiter
from core.site_manager import SiteConfig, SiteManager, get_site_manager
# Legacy (kept for backward compatibility, will be removed in v2.0)
-from core.site_registry import SiteInfo, SiteRegistry, get_site_registry
from core.tool_generator import ToolGenerator
# Tool Management (Option B architecture)
from core.tool_registry import ToolDefinition, ToolRegistry, get_tool_registry
-from core.unified_tools import UnifiedToolGenerator
__all__ = [
# Authentication
@@ -64,11 +62,6 @@ __all__ = [
"SiteManager",
"SiteConfig",
"get_site_manager",
- # Legacy (deprecated)
- "SiteRegistry",
- "SiteInfo",
- "get_site_registry",
- "UnifiedToolGenerator",
# Tool Management
"ToolRegistry",
"ToolDefinition",
diff --git a/core/auth.py b/core/auth.py
index 648d006..7877f0a 100644
--- a/core/auth.py
+++ b/core/auth.py
@@ -51,10 +51,13 @@ class AuthManager:
Returns:
bool: True if valid
"""
- is_valid = secrets.compare_digest(api_key, self.master_api_key)
+ if not self.master_api_key:
+ return False
+
+ is_valid = secrets.compare_digest(api_key.strip(), self.master_api_key.strip())
if not is_valid:
- logger.warning("Invalid API key attempt")
+ logger.debug("Master key validation failed for provided token")
return is_valid
diff --git a/core/config_snippets.py b/core/config_snippets.py
new file mode 100644
index 0000000..c05703f
--- /dev/null
+++ b/core/config_snippets.py
@@ -0,0 +1,115 @@
+"""MCP client configuration snippet generation (Track E.3).
+
+Generates copy-paste configuration snippets for connecting AI clients
+(Claude Desktop, Claude Code, Cursor, VS Code, ChatGPT) to per-user
+MCP endpoints.
+
+Usage:
+ from core.config_snippets import generate_config, get_supported_clients
+
+ snippet = generate_config(
+ base_url="https://mcp.example.com",
+ user_id="abc123",
+ alias="myblog",
+ api_key="mhu_...",
+ client_type="claude_desktop",
+ )
+"""
+
+import json
+
+# Supported MCP client types
+SUPPORTED_CLIENTS = [
+ {
+ "id": "claude_desktop",
+ "label": "Claude Desktop",
+ "description": "Anthropic's desktop app for Claude",
+ },
+ {
+ "id": "claude_code",
+ "label": "Claude Code",
+ "description": "Anthropic's CLI for Claude",
+ },
+ {
+ "id": "cursor",
+ "label": "Cursor",
+ "description": "AI-first code editor",
+ },
+ {
+ "id": "vscode",
+ "label": "VS Code",
+ "description": "Visual Studio Code with MCP extension",
+ },
+ {
+ "id": "chatgpt",
+ "label": "ChatGPT",
+ "description": "OpenAI ChatGPT (URL-based)",
+ },
+]
+
+
+def get_supported_clients() -> list[dict[str, str]]:
+ """Return the list of supported MCP client types."""
+ return SUPPORTED_CLIENTS
+
+
+def generate_config(
+ base_url: str,
+ user_id: str,
+ alias: str,
+ api_key: str,
+ client_type: str,
+) -> str:
+ """Generate a configuration snippet for the given MCP client.
+
+ Args:
+ base_url: Public URL of the MCP Hub instance (no trailing slash).
+ user_id: User UUID.
+ alias: Site alias.
+ api_key: User API key (``mhu_...``).
+ client_type: One of the supported client type IDs.
+
+ Returns:
+ JSON configuration string ready for copy-paste.
+
+ Raises:
+ ValueError: If client_type is not supported.
+ """
+ base_url = base_url.rstrip("/")
+ endpoint_url = f"{base_url}/u/{user_id}/{alias}/mcp"
+ server_name = f"mcphub-{alias}"
+
+ if client_type in ("claude_desktop", "claude_code"):
+ config = {
+ "mcpServers": {
+ server_name: {
+ "url": endpoint_url,
+ "headers": {
+ "Authorization": f"Bearer {api_key}",
+ },
+ }
+ }
+ }
+ return json.dumps(config, indent=2)
+
+ elif client_type in ("cursor", "vscode"):
+ config = {
+ "mcp": {
+ "servers": {
+ server_name: {
+ "url": endpoint_url,
+ "headers": {
+ "Authorization": f"Bearer {api_key}",
+ },
+ }
+ }
+ }
+ }
+ return json.dumps(config, indent=2)
+
+ elif client_type == "chatgpt":
+ return endpoint_url
+
+ else:
+ valid = [c["id"] for c in SUPPORTED_CLIENTS]
+ raise ValueError(f"Unsupported client type '{client_type}'. Valid: {valid}")
diff --git a/core/dashboard/auth.py b/core/dashboard/auth.py
index a62321c..7c23cff 100644
--- a/core/dashboard/auth.py
+++ b/core/dashboard/auth.py
@@ -89,8 +89,9 @@ class DashboardAuth:
if not api_key:
return False, "", None
+ api_key_clean = api_key.strip()
# Check master API key (from env var)
- if self.master_api_key and secrets.compare_digest(api_key, self.master_api_key):
+ if self.master_api_key and secrets.compare_digest(api_key_clean, self.master_api_key.strip()):
return True, "master", None
# Check AuthManager's master key (covers auto-generated temp keys)
@@ -202,6 +203,9 @@ class DashboardAuth:
user_type=payload["type"],
key_id=payload.get("kid"),
)
+ except KeyError as e:
+ logger.debug(f"Invalid dashboard session payload (missing key): {e}")
+ return None
except jwt.ExpiredSignatureError:
logger.debug("Dashboard session expired")
return None
@@ -224,6 +228,27 @@ class DashboardAuth:
return None
return self.validate_session(token)
+ def get_user_session_from_request(self, request: Request) -> dict | None:
+ """Extract and validate an OAuth user session from request.
+
+ Args:
+ request: Starlette request object.
+
+ Returns:
+ User session dict (user_id, email, name, role, type)
+ or None.
+ """
+ token = request.cookies.get(self.COOKIE_NAME)
+ if not token:
+ return None
+ try:
+ from core.user_auth import get_user_auth
+
+ user_auth = get_user_auth()
+ return user_auth.validate_user_session(token)
+ except (RuntimeError, Exception):
+ return None
+
def set_session_cookie(self, response: Response, token: str) -> Response:
"""
Set session cookie on response.
@@ -273,13 +298,15 @@ class DashboardAuth:
RedirectResponse to login page if not authenticated, None if OK.
"""
session = self.get_session_from_request(request)
- if not session:
+ user_session = self.get_user_session_from_request(request)
+
+ if not session and not user_session:
# Store original URL for redirect after login
next_url = str(request.url.path)
if request.url.query:
next_url += f"?{request.url.query}"
return RedirectResponse(
- url=f"/dashboard/login?next={next_url}",
+ url=f"/auth/login?next={next_url}",
status_code=303,
)
return None
@@ -291,3 +318,57 @@ def get_dashboard_auth() -> DashboardAuth:
if _dashboard_auth is None:
_dashboard_auth = DashboardAuth()
return _dashboard_auth
+
+
+# ── Role-checking helpers ──────────────────────────────────────
+
+
+def is_admin_session(session) -> bool:
+ """Check if session is admin (master key or API key with admin scope).
+
+ Args:
+ session: DashboardSession or OAuth user dict.
+
+ Returns:
+ True if admin session.
+ """
+ if isinstance(session, DashboardSession):
+ return session.user_type in ("master", "api_key")
+ if isinstance(session, dict):
+ return session.get("type") == "master" or session.get("role") == "admin"
+ return False
+
+
+def get_session_display_info(session) -> dict:
+ """Get display info for header/UI.
+
+ Args:
+ session: DashboardSession or OAuth user dict.
+
+ Returns:
+ Dict with name, type, email, avatar keys.
+ """
+ if isinstance(session, DashboardSession):
+ return {"name": "Admin", "type": "admin", "email": None, "avatar": None}
+ if isinstance(session, dict):
+ return {
+ "name": session.get("name") or session.get("email", "User"),
+ "type": "user",
+ "email": session.get("email"),
+ "avatar": None,
+ }
+ return {"name": "Unknown", "type": "unknown", "email": None, "avatar": None}
+
+
+def get_session_user_id(session) -> str | None:
+ """Get user_id for OAuth sessions, None for admin.
+
+ Args:
+ session: DashboardSession or OAuth user dict.
+
+ Returns:
+ User UUID string or None.
+ """
+ if isinstance(session, dict):
+ return session.get("user_id")
+ return None
diff --git a/core/dashboard/routes.py b/core/dashboard/routes.py
index 59333fa..f5d2a81 100644
--- a/core/dashboard/routes.py
+++ b/core/dashboard/routes.py
@@ -12,7 +12,12 @@ from starlette.requests import Request
from starlette.responses import HTMLResponse, JSONResponse, RedirectResponse, Response
from starlette.templating import Jinja2Templates
-from .auth import get_dashboard_auth
+from .auth import (
+ get_dashboard_auth,
+ get_session_display_info,
+ get_session_user_id,
+ is_admin_session,
+)
logger = logging.getLogger(__name__)
@@ -45,6 +50,10 @@ def get_plugin_display_name(plugin_type: str) -> str:
# Register custom Jinja filter
templates.env.filters["plugin_name"] = get_plugin_display_name
+# Register RBAC helpers as Jinja2 globals (available in all templates)
+templates.env.globals["is_admin_session"] = is_admin_session
+templates.env.globals["get_session_display_info"] = get_session_display_info
+
# Dashboard translations
DASHBOARD_TRANSLATIONS = {
"en": {
@@ -59,7 +68,7 @@ DASHBOARD_TRANSLATIONS = {
"logout": "Logout",
# Login page
"login_title": "Dashboard Login",
- "login_subtitle": "Enter your Master API Key to access the dashboard",
+ "login_subtitle": "Enter your Master API Key to access the MCP (Model Context Protocol) Hub",
"api_key_label": "API Key",
"api_key_placeholder": "sk-... or cmp_...",
"login_button": "Login",
@@ -99,6 +108,49 @@ DASHBOARD_TRANSLATIONS = {
"offline": "Offline",
"active": "Active",
"inactive": "Inactive",
+ # Sites (E.3)
+ "my_sites": "My Sites",
+ "add_site": "Add Site / Connect Service",
+ "connect": "Connect",
+ "plugin_type": "Plugin Type",
+ "site_url": "Site URL",
+ "site_alias": "Alias",
+ "site_alias_hint": "Friendly name (letters, numbers, hyphens)",
+ "status": "Status",
+ "actions": "Actions",
+ "test_connection": "Test Connection",
+ "no_sites": "No sites added yet",
+ "add_first_site": "Add your first site to get started",
+ "site_added": "Site added successfully",
+ "site_deleted": "Site deleted",
+ "connection_ok": "Connection OK",
+ "connection_failed": "Connection failed",
+ "credentials": "Credentials",
+ "select_plugin": "Select plugin type",
+ "adding_site": "Adding site...",
+ "testing": "Testing...",
+ "copy": "Copy",
+ "copied": "Copied!",
+ "api_key_name": "Key Name",
+ "generate_key": "Generate API Key",
+ "your_api_key": "Your API Key",
+ "key_shown_once": "This key will only be shown once. Copy it now!",
+ "no_api_keys": "No API keys yet",
+ "config_snippets": "Configuration Snippets",
+ "select_site": "Select a site",
+ "select_client": "Select client",
+ "max_sites_reached": "Maximum sites reached",
+ # User dashboard
+ "my_services": "My Services",
+ "active_connections": "Active Connections",
+ "user_api_keys": "API Keys",
+ "available_tools": "Available Tools",
+ "site_status": "Site Status",
+ "no_services_yet": "You haven't connected any services yet.",
+ "add_first_service": "Add your first service to start using MCP tools.",
+ "admin_login": "Admin Login with API Key",
+ "profile": "Profile",
+ "admin_badge": "Admin",
},
"fa": {
# Navigation
@@ -112,7 +164,7 @@ DASHBOARD_TRANSLATIONS = {
"logout": "خروج",
# Login page
"login_title": "ورود به داشبورد",
- "login_subtitle": "کلید API مستر خود را برای دسترسی وارد کنید",
+ "login_subtitle": "برای دسترسی به هاب MCP (Model Context Protocol)، کلید API مستر خود را وارد کنید",
"api_key_label": "کلید API",
"api_key_placeholder": "sk-... یا cmp_...",
"login_button": "ورود",
@@ -152,6 +204,49 @@ DASHBOARD_TRANSLATIONS = {
"offline": "آفلاین",
"active": "فعال",
"inactive": "غیرفعال",
+ # Sites (E.3)
+ "my_sites": "سایتهای من",
+ "add_site": "افزودن سایت / اتصال سرویس",
+ "connect": "اتصال",
+ "plugin_type": "نوع پلاگین",
+ "site_url": "آدرس سایت",
+ "site_alias": "نام مستعار",
+ "site_alias_hint": "نام دوستانه (حروف، اعداد، خط تیره)",
+ "status": "وضعیت",
+ "actions": "عملیات",
+ "test_connection": "تست اتصال",
+ "no_sites": "هنوز سایتی اضافه نشده",
+ "add_first_site": "اولین سایت خود را اضافه کنید",
+ "site_added": "سایت با موفقیت اضافه شد",
+ "site_deleted": "سایت حذف شد",
+ "connection_ok": "اتصال برقرار",
+ "connection_failed": "اتصال ناموفق",
+ "credentials": "مشخصات دسترسی",
+ "select_plugin": "نوع پلاگین را انتخاب کنید",
+ "adding_site": "در حال افزودن سایت...",
+ "testing": "در حال تست...",
+ "copy": "کپی",
+ "copied": "کپی شد!",
+ "api_key_name": "نام کلید",
+ "generate_key": "تولید کلید API",
+ "your_api_key": "کلید API شما",
+ "key_shown_once": "این کلید فقط یکبار نمایش داده میشود. الان کپی کنید!",
+ "no_api_keys": "هنوز کلید API ندارید",
+ "config_snippets": "نمونه کدهای پیکربندی",
+ "select_site": "انتخاب سایت",
+ "select_client": "انتخاب کلاینت",
+ "max_sites_reached": "حداکثر سایتها رسیده است",
+ # User dashboard
+ "my_services": "سرویسهای من",
+ "active_connections": "اتصالات فعال",
+ "user_api_keys": "کلیدهای API",
+ "available_tools": "ابزارهای موجود",
+ "site_status": "وضعیت سایتها",
+ "no_services_yet": "هنوز سرویسی متصل نکردهاید.",
+ "add_first_service": "اولین سرویس خود را اضافه کنید تا از ابزارهای MCP استفاده کنید.",
+ "admin_login": "ورود مدیر با کلید API",
+ "profile": "پروفایل",
+ "admin_badge": "مدیر",
},
}
@@ -221,6 +316,56 @@ async def get_dashboard_stats() -> dict:
return stats
+async def get_user_dashboard_stats(user_id: str) -> dict:
+ """Get dashboard statistics for an OAuth user."""
+ stats = {
+ "sites_count": 0,
+ "active_sites_count": 0,
+ "api_keys_count": 0,
+ "tools_count": 0,
+ }
+ try:
+ from core.site_api import get_user_sites
+
+ sites = await get_user_sites(user_id)
+ stats["sites_count"] = len(sites)
+ stats["active_sites_count"] = len(
+ [s for s in sites if s.get("status") == "active"]
+ )
+ except Exception as e:
+ logger.warning(f"Error getting user sites count: {e}")
+
+ try:
+ from core.user_keys import get_user_key_manager
+
+ key_mgr = get_user_key_manager()
+ keys = await key_mgr.list_keys(user_id)
+ stats["api_keys_count"] = len(keys)
+ except Exception as e:
+ logger.debug(f"Error getting user keys count: {e}")
+
+ try:
+ from core.tool_registry import get_tool_registry
+
+ tool_registry = get_tool_registry()
+ stats["tools_count"] = len(tool_registry.get_all())
+ except Exception:
+ stats["tools_count"] = 596 # Fallback
+
+ return stats
+
+
+async def get_user_sites_summary(user_id: str) -> list:
+ """Get a summary of user's sites with status for dashboard display."""
+ try:
+ from core.site_api import get_user_sites
+
+ return await get_user_sites(user_id)
+ except Exception as e:
+ logger.warning(f"Error getting user sites summary: {e}")
+ return []
+
+
async def get_projects_by_type() -> dict:
"""Get projects grouped by plugin type."""
projects_by_type = {}
@@ -313,7 +458,7 @@ async def dashboard_login_page(request: Request) -> Response:
auth = get_dashboard_auth()
# Check if already logged in
- session = auth.get_session_from_request(request)
+ session = auth.get_session_from_request(request) or auth.get_user_session_from_request(request)
if session:
return RedirectResponse(url="/dashboard", status_code=303)
@@ -408,19 +553,18 @@ async def dashboard_login_submit(request: Request) -> Response:
audit_logger.log_authentication(success=True, ip_address=client_ip)
except Exception as e:
logger.warning(f"Failed to log auth event: {e}")
-
- logger.info(f"Dashboard login successful: type={user_type}, ip={client_ip}")
+ logger.info(f"Dashboard login successful: type={user_type}, ip={client_ip}")
return response
-async def dashboard_logout(request: Request) -> Response:
- """Handle dashboard logout."""
+async def auth_logout(request: Request) -> Response:
+ """Clear OAuth user session cookie."""
+ from core.dashboard.auth import get_dashboard_auth
+
auth = get_dashboard_auth()
- client_ip = get_client_ip(request)
-
- response = RedirectResponse(url="/dashboard/login", status_code=303)
+ response = RedirectResponse(url="/auth/login", status_code=303)
auth.clear_session_cookie(response)
-
+ client_ip = request.client.host if request.client else "unknown"
# Log logout event
try:
from core.audit_log import get_audit_logger
@@ -436,6 +580,9 @@ async def dashboard_logout(request: Request) -> Response:
logger.info(f"Dashboard logout from {client_ip}")
return response
+# Alias for backwards compatibility with __init__.py and other routes
+dashboard_logout = auth_logout
+
async def dashboard_home(request: Request) -> Response:
"""Render dashboard home page."""
@@ -446,7 +593,10 @@ async def dashboard_home(request: Request) -> Response:
if redirect:
return redirect
- session = auth.get_session_from_request(request)
+ session = auth.get_session_from_request(request) or auth.get_user_session_from_request(request)
+ admin = is_admin_session(session)
+ display_info = get_session_display_info(session)
+ user_id = get_session_user_id(session)
# Get language
accept_language = request.headers.get("accept-language")
@@ -454,26 +604,38 @@ async def dashboard_home(request: Request) -> Response:
lang = detect_language(accept_language, query_lang)
t = get_translations(lang)
- # Get dashboard data
- stats = await get_dashboard_stats()
- projects_by_type = await get_projects_by_type()
- recent_activity = await get_recent_activity(limit=5)
- health_summary = await get_health_summary()
+ context = {
+ "request": request,
+ "lang": lang,
+ "t": t,
+ "session": session,
+ "is_admin": admin,
+ "display_info": display_info,
+ "current_page": "dashboard",
+ }
- return templates.TemplateResponse(
- "dashboard/index.html",
- {
- "request": request,
- "lang": lang,
- "t": t,
- "session": session,
+ if admin:
+ # Admin dashboard — full system stats
+ stats = await get_dashboard_stats()
+ projects_by_type = await get_projects_by_type()
+ recent_activity = await get_recent_activity(limit=5)
+ health_summary = await get_health_summary()
+ context.update({
"stats": stats,
"projects_by_type": projects_by_type,
"recent_activity": recent_activity,
"health_summary": health_summary,
- "current_page": "dashboard",
- },
- )
+ })
+ else:
+ # User dashboard — personal stats
+ user_stats = await get_user_dashboard_stats(user_id) if user_id else {}
+ user_sites = await get_user_sites_summary(user_id) if user_id else []
+ context.update({
+ "stats": user_stats,
+ "user_sites": user_sites,
+ })
+
+ return templates.TemplateResponse("dashboard/index.html", context)
async def dashboard_api_stats(request: Request) -> Response:
@@ -481,13 +643,19 @@ async def dashboard_api_stats(request: Request) -> Response:
auth = get_dashboard_auth()
# Check authentication
- session = auth.get_session_from_request(request)
+ session = auth.get_session_from_request(request) or auth.get_user_session_from_request(request)
if not session:
return JSONResponse({"error": "Unauthorized"}, status_code=401)
- stats = await get_dashboard_stats()
- projects_by_type = await get_projects_by_type()
- health_summary = await get_health_summary()
+ if is_admin_session(session):
+ stats = await get_dashboard_stats()
+ projects_by_type = await get_projects_by_type()
+ health_summary = await get_health_summary()
+ else:
+ user_id = get_session_user_id(session)
+ stats = await get_user_dashboard_stats(user_id) if user_id else {}
+ projects_by_type = {}
+ health_summary = {"status": "n/a"}
return JSONResponse(
{
@@ -518,13 +686,32 @@ def get_cached_health_status(project_id: str) -> dict:
health_monitor = get_health_monitor()
if not health_monitor:
- return {"status": "unknown", "last_check": None, "error_rate": 0}
+ return {"status": "unknown", "last_check": None, "error_rate": 0, "reason": None}
- # Get cached metrics (last 24 hours)
+ # First check the latest active health status
+ latest = health_monitor.latest_health_status.get(project_id)
+ if latest:
+ if latest.healthy:
+ status = "healthy"
+ elif 0 < latest.error_rate_percent <= 10:
+ # If recently slightly failing
+ status = "warning"
+ else:
+ status = "unhealthy"
+
+ reason = latest.recent_errors[-1] if latest.recent_errors else None
+ return {
+ "status": status,
+ "last_check": latest.last_check.isoformat() if latest.last_check else None,
+ "error_rate": latest.error_rate_percent,
+ "reason": reason
+ }
+
+ # Fallback to cached metrics (last 24 hours) if no active check exists yet
metrics = health_monitor.get_project_metrics(project_id, hours=24)
if not metrics or metrics.get("total_requests", 0) == 0:
- return {"status": "unknown", "last_check": None, "error_rate": 0}
+ return {"status": "unknown", "last_check": None, "error_rate": 0, "reason": None}
error_rate = metrics.get("error_rate_percent", 0)
@@ -543,7 +730,7 @@ def get_cached_health_status(project_id: str) -> dict:
if history:
last_check = history[-1].timestamp.isoformat()
- return {"status": status, "last_check": last_check, "error_rate": error_rate}
+ return {"status": status, "last_check": last_check, "error_rate": error_rate, "reason": None}
except Exception as e:
logger.warning(f"Error getting cached health for {project_id}: {e}")
return {"status": "unknown", "last_check": None, "error_rate": 0}
@@ -555,6 +742,7 @@ async def get_all_projects(
status_filter: str | None = None,
page: int = 1,
per_page: int = 20,
+ user_session: dict | object | None = None,
) -> dict:
"""Get all projects with optional filtering."""
projects = []
@@ -566,7 +754,21 @@ async def get_all_projects(
site_manager = get_site_manager()
sites = site_manager.list_all_sites()
+ is_master = False
+ current_user_id = None
+ if user_session:
+ if hasattr(user_session, "user_type") and user_session.user_type == "master":
+ is_master = True
+ elif isinstance(user_session, dict) and "user_id" in user_session:
+ current_user_id = user_session["user_id"]
+
for site in sites:
+ # Tenant isolation checks
+ site_user_id = site.get("user_id")
+ if not is_master:
+ if site_user_id != current_user_id:
+ continue
+
site_plugin_type = site.get("plugin_type", "unknown")
available_plugin_types.add(site_plugin_type)
@@ -605,6 +807,7 @@ async def get_all_projects(
"health_status": cached_health["status"],
"last_health_check": cached_health["last_check"],
"error_rate": cached_health["error_rate"],
+ "reason": cached_health.get("reason"),
}
)
@@ -743,16 +946,11 @@ async def get_project_detail(project_id: str) -> dict | None:
async def dashboard_projects_list(request: Request) -> Response:
- """Render projects list page."""
- auth = get_dashboard_auth()
-
- # Check authentication
- redirect = auth.require_auth(request)
+ """Render projects list page (admin only)."""
+ session, redirect = _require_admin_session(request)
if redirect:
return redirect
- session = auth.get_session_from_request(request)
-
# Get language
accept_language = request.headers.get("accept-language")
query_lang = request.query_params.get("lang")
@@ -771,6 +969,7 @@ async def dashboard_projects_list(request: Request) -> Response:
search=search if search else None,
status_filter=status if status else None,
page=page,
+ user_session=session,
)
return templates.TemplateResponse(
@@ -803,7 +1002,7 @@ async def dashboard_project_detail(request: Request) -> Response:
if redirect:
return redirect
- session = auth.get_session_from_request(request)
+ session = auth.get_session_from_request(request) or auth.get_user_session_from_request(request)
# Get language
accept_language = request.headers.get("accept-language")
@@ -850,7 +1049,7 @@ async def dashboard_api_projects(request: Request) -> Response:
auth = get_dashboard_auth()
# Check authentication
- session = auth.get_session_from_request(request)
+ session = auth.get_session_from_request(request) or auth.get_user_session_from_request(request)
if not session:
return JSONResponse({"error": "Unauthorized"}, status_code=401)
@@ -863,6 +1062,7 @@ async def dashboard_api_projects(request: Request) -> Response:
plugin_type=plugin_type,
search=search,
page=page,
+ user_session=session,
)
return JSONResponse(projects_data)
@@ -873,7 +1073,7 @@ async def dashboard_api_project_detail(request: Request) -> Response:
auth = get_dashboard_auth()
# Check authentication
- session = auth.get_session_from_request(request)
+ session = auth.get_session_from_request(request) or auth.get_user_session_from_request(request)
if not session:
return JSONResponse({"error": "Unauthorized"}, status_code=401)
@@ -893,7 +1093,7 @@ async def dashboard_project_health_check(request: Request) -> Response:
auth = get_dashboard_auth()
# Check authentication
- session = auth.get_session_from_request(request)
+ session = auth.get_session_from_request(request) or auth.get_user_session_from_request(request)
if not session:
logger.warning("Health check unauthorized - no session")
return JSONResponse({"error": "Unauthorized", "status": "error"}, status_code=401)
@@ -1022,16 +1222,11 @@ async def get_all_api_keys(
async def dashboard_api_keys_list(request: Request) -> Response:
- """Render API keys list page."""
- auth = get_dashboard_auth()
-
- # Check authentication
- redirect = auth.require_auth(request)
+ """Render API keys list page (admin only)."""
+ session, redirect = _require_admin_session(request)
if redirect:
return redirect
- session = auth.get_session_from_request(request)
-
# Get language
accept_language = request.headers.get("accept-language")
query_lang = request.query_params.get("lang")
@@ -1080,13 +1275,10 @@ async def dashboard_api_keys_list(request: Request) -> Response:
async def dashboard_api_keys_create(request: Request) -> Response:
- """API endpoint to create a new API key."""
- auth = get_dashboard_auth()
-
- # Check authentication
- session = auth.get_session_from_request(request)
- if not session:
- return JSONResponse({"error": "Unauthorized"}, status_code=401)
+ """API endpoint to create a new API key (admin only)."""
+ session, redirect = _require_admin_session(request)
+ if redirect:
+ return JSONResponse({"error": "Admin access required"}, status_code=403)
try:
data = await request.json()
@@ -1135,13 +1327,10 @@ async def dashboard_api_keys_create(request: Request) -> Response:
async def dashboard_api_keys_revoke(request: Request) -> Response:
- """API endpoint to revoke an API key."""
- auth = get_dashboard_auth()
-
- # Check authentication
- session = auth.get_session_from_request(request)
- if not session:
- return JSONResponse({"error": "Unauthorized"}, status_code=401)
+ """API endpoint to revoke an API key (admin only)."""
+ session, redirect = _require_admin_session(request)
+ if redirect:
+ return JSONResponse({"error": "Admin access required"}, status_code=403)
try:
key_id = request.path_params.get("key_id", "")
@@ -1169,13 +1358,10 @@ async def dashboard_api_keys_revoke(request: Request) -> Response:
async def dashboard_api_keys_delete(request: Request) -> Response:
- """API endpoint to delete an API key."""
- auth = get_dashboard_auth()
-
- # Check authentication
- session = auth.get_session_from_request(request)
- if not session:
- return JSONResponse({"error": "Unauthorized"}, status_code=401)
+ """API endpoint to delete an API key (admin only)."""
+ session, redirect = _require_admin_session(request)
+ if redirect:
+ return JSONResponse({"error": "Admin access required"}, status_code=403)
try:
key_id = request.path_params.get("key_id", "")
@@ -1241,16 +1427,11 @@ async def get_oauth_clients_data() -> dict:
async def dashboard_oauth_clients_list(request: Request) -> Response:
- """Render OAuth clients list page."""
- auth = get_dashboard_auth()
-
- # Check authentication
- redirect = auth.require_auth(request)
+ """Render OAuth clients list page (admin only)."""
+ session, redirect = _require_admin_session(request)
if redirect:
return redirect
- session = auth.get_session_from_request(request)
-
# Get language
accept_language = request.headers.get("accept-language")
query_lang = request.query_params.get("lang")
@@ -1275,13 +1456,10 @@ async def dashboard_oauth_clients_list(request: Request) -> Response:
async def dashboard_oauth_clients_create(request: Request) -> Response:
- """API endpoint to create OAuth client."""
- auth = get_dashboard_auth()
-
- # Check authentication
- session = auth.get_session_from_request(request)
- if not session:
- return JSONResponse({"error": "Unauthorized"}, status_code=401)
+ """API endpoint to create OAuth client (admin only)."""
+ session, redirect = _require_admin_session(request)
+ if redirect:
+ return JSONResponse({"error": "Admin access required"}, status_code=403)
try:
data = await request.json()
@@ -1325,13 +1503,10 @@ async def dashboard_oauth_clients_create(request: Request) -> Response:
async def dashboard_oauth_clients_delete(request: Request) -> Response:
- """API endpoint to delete OAuth client."""
- auth = get_dashboard_auth()
-
- # Check authentication
- session = auth.get_session_from_request(request)
- if not session:
- return JSONResponse({"error": "Unauthorized"}, status_code=401)
+ """API endpoint to delete OAuth client (admin only)."""
+ session, redirect = _require_admin_session(request)
+ if redirect:
+ return JSONResponse({"error": "Admin access required"}, status_code=403)
try:
client_id = request.path_params.get("client_id", "")
@@ -1483,16 +1658,11 @@ async def get_audit_logs_data(
async def dashboard_audit_logs_list(request: Request) -> Response:
- """Render audit logs list page."""
- auth = get_dashboard_auth()
-
- # Check authentication
- redirect = auth.require_auth(request)
+ """Render audit logs list page (admin only)."""
+ session, redirect = _require_admin_session(request)
if redirect:
return redirect
- session = auth.get_session_from_request(request)
-
# Get language
accept_language = request.headers.get("accept-language")
query_lang = request.query_params.get("lang")
@@ -1552,7 +1722,7 @@ async def dashboard_api_audit_logs(request: Request) -> Response:
auth = get_dashboard_auth()
# Check authentication
- session = auth.get_session_from_request(request)
+ session = auth.get_session_from_request(request) or auth.get_user_session_from_request(request)
if not session:
return JSONResponse({"error": "Unauthorized"}, status_code=401)
@@ -1817,21 +1987,16 @@ async def get_health_data(live_check: bool = True) -> dict:
async def dashboard_health_page(request: Request) -> Response:
"""
- Render health monitoring page.
+ Render health monitoring page (admin only).
By default, shows cached health data (fast load).
If ?refresh=true, performs live health checks (slower but accurate).
"""
try:
- auth = get_dashboard_auth()
-
- # Check authentication
- redirect = auth.require_auth(request)
+ session, redirect = _require_admin_session(request)
if redirect:
return redirect
- session = auth.get_session_from_request(request)
-
# Get language
accept_language = request.headers.get("accept-language")
query_lang = request.query_params.get("lang")
@@ -1871,15 +2036,11 @@ async def dashboard_health_page(request: Request) -> Response:
async def dashboard_health_projects_partial(request: Request) -> Response:
- """HTMX endpoint for projects health data (HTML partial)."""
+ """HTMX endpoint for projects health data (admin only, HTML partial)."""
logger.debug("Health projects partial endpoint called")
- auth = get_dashboard_auth()
-
- # Check authentication
- session = auth.get_session_from_request(request)
- if not session:
- logger.warning("Health projects partial: Unauthorized - no session")
+ session, redirect = _require_admin_session(request)
+ if redirect:
return HTMLResponse(
"
| Unauthorized |
",
status_code=401,
@@ -1927,7 +2088,7 @@ async def dashboard_api_health(request: Request) -> Response:
auth = get_dashboard_auth()
# Check authentication
- session = auth.get_session_from_request(request)
+ session = auth.get_session_from_request(request) or auth.get_user_session_from_request(request)
if not session:
return JSONResponse({"error": "Unauthorized"}, status_code=401)
@@ -2025,16 +2186,11 @@ def get_about_info() -> dict:
async def dashboard_settings_page(request: Request) -> Response:
- """Render settings page."""
- auth = get_dashboard_auth()
-
- # Check authentication
- redirect = auth.require_auth(request)
+ """Render settings page (admin only)."""
+ session, redirect = _require_admin_session(request)
if redirect:
return redirect
- session = auth.get_session_from_request(request)
-
# Get language
accept_language = request.headers.get("accept-language")
query_lang = request.query_params.get("lang")
@@ -2047,11 +2203,18 @@ async def dashboard_settings_page(request: Request) -> Response:
about = get_about_info()
# Format session info
- session_info = {
- "user_type": session.user_type if session else "unknown",
- "created_at": session.created_at.isoformat() if session and session.created_at else "",
- "expires_at": session.expires_at.isoformat() if session and session.expires_at else "",
- }
+ if isinstance(session, dict):
+ session_info = {
+ "user_type": session.get("type", "oauth_user"),
+ "created_at": "",
+ "expires_at": "",
+ }
+ else:
+ session_info = {
+ "user_type": session.user_type if session else "unknown",
+ "created_at": session.created_at.isoformat() if session and session.created_at else "",
+ "expires_at": session.expires_at.isoformat() if session and session.expires_at else "",
+ }
return templates.TemplateResponse(
"dashboard/settings/index.html",
@@ -2068,6 +2231,592 @@ async def dashboard_settings_page(request: Request) -> Response:
)
+# =============================================================================
+# E.2: OAuth Social Login Routes
+# =============================================================================
+
+
+async def auth_login_page(request: Request) -> Response:
+ """Render the OAuth login page with GitHub + Google buttons."""
+ auth = get_dashboard_auth()
+
+ # Check if already logged in (admin or user session)
+ session = auth.get_session_from_request(request) or auth.get_user_session_from_request(request)
+ if session:
+ return RedirectResponse(url="/dashboard", status_code=303)
+
+ # Get language
+ accept_language = request.headers.get("accept-language")
+ query_lang = request.query_params.get("lang")
+ lang = detect_language(accept_language, query_lang)
+ t = get_translations(lang)
+
+ error = request.query_params.get("error")
+
+ # Get available providers
+ providers = []
+ try:
+ from core.user_auth import get_user_auth
+
+ user_auth = get_user_auth()
+ providers = user_auth.available_providers()
+ except RuntimeError:
+ pass
+
+ return templates.TemplateResponse(
+ "dashboard/auth-login.html",
+ {
+ "request": request,
+ "lang": lang,
+ "t": t,
+ "error": error,
+ "providers": providers,
+ "version": _get_project_version(),
+ },
+ )
+
+
+async def auth_provider_redirect(request: Request) -> Response:
+ """Redirect to OAuth provider (GitHub or Google)."""
+ provider = request.path_params.get("provider", "")
+
+ if provider in ("login", "logout", "callback"):
+ return RedirectResponse(url=f"/auth/{provider}", status_code=303)
+
+ try:
+ from core.user_auth import get_user_auth
+
+ user_auth = get_user_auth()
+ auth_url, state = user_auth.get_authorization_url(provider)
+
+ response = RedirectResponse(url=auth_url, status_code=302)
+ response.set_cookie(
+ key="oauth_state",
+ value=state,
+ max_age=600,
+ httponly=True,
+ secure=os.environ.get("DASHBOARD_SECURE_COOKIE", "true").lower() == "true",
+ samesite="lax",
+ )
+ return response
+ except (RuntimeError, ValueError) as e:
+ logger.error("OAuth redirect failed for %s: %s", provider, e)
+ return RedirectResponse(
+ url="/auth/login?error=provider_unavailable",
+ status_code=303,
+ )
+
+
+async def auth_callback(request: Request) -> Response:
+ """Handle OAuth callback from provider."""
+ provider = request.path_params.get("provider", "")
+ code = request.query_params.get("code")
+ state = request.query_params.get("state")
+ error = request.query_params.get("error")
+
+ if error:
+ logger.warning("OAuth callback error from %s: %s", provider, error)
+ return RedirectResponse(url="/auth/login?error=oauth_denied", status_code=303)
+
+ if not code or not state:
+ return RedirectResponse(
+ url="/auth/login?error=missing_params",
+ status_code=303,
+ )
+
+ try:
+ from core.user_auth import get_user_auth
+
+ user_auth = get_user_auth()
+
+ # Validate state (CSRF protection)
+ if not user_auth.validate_state(state):
+ logger.warning("OAuth callback: invalid state for %s", provider)
+ return RedirectResponse(
+ url="/auth/login?error=invalid_state",
+ status_code=303,
+ )
+
+ # Exchange code for user info
+ user_info = await user_auth.exchange_code(provider, code)
+
+ if not user_info.get("email"):
+ return RedirectResponse(
+ url="/auth/login?error=no_email",
+ status_code=303,
+ )
+
+ # Get client IP for rate limiting
+ client_ip = get_client_ip(request)
+
+ # Look up or create user in database
+ from core.database import get_database
+
+ db = get_database()
+
+ existing_user = await db.get_user_by_provider(
+ provider=user_info["provider"],
+ provider_id=user_info["provider_id"],
+ )
+
+ if existing_user:
+ await db.update_user_last_login(existing_user["id"])
+ user = existing_user
+ else:
+ # Check if user exists by email (handle OAuth provider collision)
+ user_by_email = await db.get_user_by_email(user_info["email"])
+ if user_by_email:
+ # Merge login: user is recognized by email regardless of provider
+ await db.update_user_last_login(user_by_email["id"])
+ user = user_by_email
+ logger.info(
+ "User %s logged in with alternate provider %s (original: %s)",
+ user_info["email"], provider, user_by_email["provider"]
+ )
+ else:
+ # New registration -- check rate limit
+ if not user_auth.check_registration_rate(client_ip):
+ logger.warning(
+ "Registration rate limit hit for IP %s",
+ client_ip,
+ )
+ return RedirectResponse(
+ url="/auth/login?error=rate_limit",
+ status_code=303,
+ )
+
+ user = await db.create_user(
+ email=user_info["email"],
+ name=user_info.get("name"),
+ provider=user_info["provider"],
+ provider_id=user_info["provider_id"],
+ avatar_url=user_info.get("avatar_url"),
+ )
+ user_auth.record_registration(client_ip)
+ logger.info(
+ "New user registered: %s via %s",
+ user_info["email"],
+ provider,
+ )
+
+ # Create session
+ token = user_auth.create_user_session(
+ user_id=user["id"],
+ email=user["email"],
+ name=user.get("name"),
+ role=user.get("role", "user"),
+ )
+
+ response = RedirectResponse(url="/dashboard", status_code=303)
+ dashboard_auth = get_dashboard_auth()
+ dashboard_auth.set_session_cookie(response, token)
+ response.delete_cookie(key="oauth_state")
+
+ logger.info(
+ "OAuth login successful: %s via %s",
+ user["email"],
+ provider,
+ )
+ return response
+
+ except ValueError as e:
+ logger.error("OAuth callback failed for %s: %s", provider, e)
+ return RedirectResponse(
+ url="/auth/login?error=exchange_failed",
+ status_code=303,
+ )
+ except Exception as e:
+ logger.error(
+ "OAuth callback unexpected error for %s: %s",
+ provider,
+ e,
+ )
+ return RedirectResponse(
+ url="/auth/login?error=server_error",
+ status_code=303,
+ )
+
+
+async def auth_logout(request: Request) -> Response:
+ """Log out the current user (both admin and OAuth sessions)."""
+ auth = get_dashboard_auth()
+ response = RedirectResponse(url="/auth/login", status_code=303)
+ auth.clear_session_cookie(response)
+ return response
+
+
+async def dashboard_profile_page(request: Request) -> Response:
+ """Render the user profile page."""
+ auth = get_dashboard_auth()
+
+ # Check for OAuth user session
+ user_session = auth.get_user_session_from_request(request)
+
+ if not user_session:
+ admin_session = auth.get_session_from_request(request)
+ if not admin_session:
+ return RedirectResponse(url="/auth/login", status_code=303)
+ return RedirectResponse(url="/dashboard", status_code=303)
+
+ # Get language
+ accept_language = request.headers.get("accept-language")
+ query_lang = request.query_params.get("lang")
+ lang = detect_language(accept_language, query_lang)
+ t = get_translations(lang)
+
+ # Get full user from database
+ user_data = None
+ try:
+ from core.database import get_database
+
+ db = get_database()
+ user_data = await db.get_user_by_id(user_session["user_id"])
+ except Exception as e:
+ logger.warning("Failed to fetch user profile: %s", e)
+
+ return templates.TemplateResponse(
+ "dashboard/profile.html",
+ {
+ "request": request,
+ "lang": lang,
+ "t": t,
+ "session": user_session,
+ "user": user_data,
+ "current_page": "profile",
+ },
+ )
+
+
+# ======================================================================
+# Site Management Routes (E.3)
+# ======================================================================
+
+
+def _require_user_session(request: Request):
+ """Get OAuth user session or return redirect. Helper for E.3 routes."""
+ auth = get_dashboard_auth()
+ user_session = auth.get_user_session_from_request(request)
+ if not user_session:
+ return None, RedirectResponse(url="/auth/login", status_code=303)
+ return user_session, None
+
+
+def _require_admin_session(request: Request):
+ """Get admin session or redirect to dashboard. Helper for admin-only routes."""
+ auth = get_dashboard_auth()
+ session = auth.get_session_from_request(request)
+ if session and is_admin_session(session):
+ return session, None
+ # Not admin — redirect to dashboard home
+ return None, RedirectResponse(url="/dashboard", status_code=303)
+
+
+async def dashboard_sites_list(request: Request) -> Response:
+ """Render the My Sites page."""
+ auth = get_dashboard_auth()
+ # Allow both OAuth users and admin
+ admin_session = auth.get_session_from_request(request)
+ user_session = auth.get_user_session_from_request(request)
+ if not user_session and not admin_session:
+ return RedirectResponse(url="/auth/login", status_code=303)
+ # For admin, we don't have user sites — redirect to projects
+ if admin_session and not user_session:
+ return RedirectResponse(url="/dashboard/projects", status_code=303)
+
+ accept_language = request.headers.get("accept-language")
+ query_lang = request.query_params.get("lang")
+ lang = detect_language(accept_language, query_lang)
+ t = get_translations(lang)
+
+ from core.site_api import PLUGIN_DISPLAY_NAMES as SITE_PLUGIN_NAMES
+ from core.site_api import get_user_sites
+
+ # Flash messages from query params
+ msg = request.query_params.get("msg")
+ error = request.query_params.get("error")
+
+ try:
+ sites = await get_user_sites(user_session["user_id"])
+ except RuntimeError:
+ sites = []
+ error = error or (
+ "Site storage is not configured. Contact the administrator."
+ if lang != "fa"
+ else "ذخیرهسازی سایتها پیکربندی نشده. با مدیر تماس بگیرید."
+ )
+
+ return templates.TemplateResponse(
+ "dashboard/sites/list.html",
+ {
+ "request": request,
+ "lang": lang,
+ "t": t,
+ "session": user_session,
+ "sites": sites,
+ "plugin_names": SITE_PLUGIN_NAMES,
+ "current_page": "my_sites",
+ "msg": msg,
+ "error": error,
+ },
+ )
+
+
+async def dashboard_sites_add(request: Request) -> Response:
+ """Render the Add Site form."""
+ user_session, redirect = _require_user_session(request)
+ if redirect:
+ return redirect
+
+ accept_language = request.headers.get("accept-language")
+ query_lang = request.query_params.get("lang")
+ lang = detect_language(accept_language, query_lang)
+ t = get_translations(lang)
+
+ import json
+
+ from core.site_api import get_user_credential_fields, get_user_plugin_names
+
+ # Non-admin users get a filtered list (no wordpress_advanced)
+ plugin_fields = get_user_credential_fields()
+ plugin_names = get_user_plugin_names()
+
+ return templates.TemplateResponse(
+ "dashboard/sites/add.html",
+ {
+ "request": request,
+ "lang": lang,
+ "t": t,
+ "session": user_session,
+ "plugin_fields": plugin_fields,
+ "plugin_fields_json": json.dumps(plugin_fields),
+ "plugin_names": plugin_names,
+ "current_page": "my_sites",
+ },
+ )
+
+
+async def dashboard_connect_page(request: Request) -> Response:
+ """Render the Connect page (config snippets + API key management)."""
+ user_session, redirect = _require_user_session(request)
+ if redirect:
+ return redirect
+
+ accept_language = request.headers.get("accept-language")
+ query_lang = request.query_params.get("lang")
+ lang = detect_language(accept_language, query_lang)
+ t = get_translations(lang)
+
+ from core.config_snippets import get_supported_clients
+ from core.site_api import get_user_sites
+ from core.user_keys import get_user_key_manager
+
+ sites = await get_user_sites(user_session["user_id"])
+
+ keys = []
+ try:
+ key_mgr = get_user_key_manager()
+ keys = await key_mgr.list_keys(user_session["user_id"])
+ except RuntimeError:
+ pass
+
+ # Flash messages
+ new_key = request.query_params.get("new_key")
+
+ return templates.TemplateResponse(
+ "dashboard/connect.html",
+ {
+ "request": request,
+ "lang": lang,
+ "t": t,
+ "session": user_session,
+ "sites": sites,
+ "api_keys": keys,
+ "clients": get_supported_clients(),
+ "current_page": "connect",
+ "new_key": new_key,
+ },
+ )
+
+
+# ======================================================================
+# Site Management API Routes (E.3)
+# ======================================================================
+
+
+async def api_create_site(request: Request) -> Response:
+ """POST /api/sites — Create a new site."""
+ user_session, redirect = _require_user_session(request)
+ if redirect:
+ return JSONResponse({"error": "Unauthorized"}, status_code=401)
+
+ try:
+ body = await request.json()
+ except Exception:
+ return JSONResponse({"error": "Invalid JSON"}, status_code=400)
+
+ from core.site_api import create_user_site
+
+ try:
+ site = await create_user_site(
+ user_id=user_session["user_id"],
+ plugin_type=body.get("plugin_type", ""),
+ alias=body.get("alias", ""),
+ url=body.get("url", "").strip().rstrip("/"),
+ credentials=body.get("credentials", {}),
+ )
+ return JSONResponse({"site": site, "message": "Site created"})
+ except ValueError as e:
+ return JSONResponse({"error": str(e)}, status_code=400)
+ except RuntimeError as e:
+ # Encryption or database not configured
+ logger.error("Site storage not configured: %s", e)
+ return JSONResponse(
+ {"error": "Site storage is not configured. Contact the administrator."},
+ status_code=503,
+ )
+ except Exception as e:
+ logger.error("Failed to create site: %s", e, exc_info=True)
+ return JSONResponse({"error": "Internal error"}, status_code=500)
+
+
+async def api_list_sites(request: Request) -> Response:
+ """GET /api/sites — List user's sites."""
+ user_session, redirect = _require_user_session(request)
+ if redirect:
+ return JSONResponse({"error": "Unauthorized"}, status_code=401)
+
+ from core.site_api import get_user_sites
+
+ sites = await get_user_sites(user_session["user_id"])
+ return JSONResponse({"sites": sites})
+
+
+async def api_delete_site(request: Request) -> Response:
+ """DELETE /api/sites/{id} — Delete a site."""
+ user_session, redirect = _require_user_session(request)
+ if redirect:
+ return JSONResponse({"error": "Unauthorized"}, status_code=401)
+
+ site_id = request.path_params.get("id", "")
+ from core.site_api import delete_user_site
+
+ deleted = await delete_user_site(site_id, user_session["user_id"])
+ if deleted:
+ return JSONResponse({"message": "Site deleted"})
+ return JSONResponse({"error": "Site not found"}, status_code=404)
+
+
+async def api_test_site(request: Request) -> Response:
+ """POST /api/sites/{id}/test — Test site connection."""
+ user_session, redirect = _require_user_session(request)
+ if redirect:
+ return JSONResponse({"error": "Unauthorized"}, status_code=401)
+
+ site_id = request.path_params.get("id", "")
+ from core.site_api import test_site_connection
+
+ try:
+ ok, msg = await test_site_connection(site_id, user_session["user_id"])
+ return JSONResponse({"ok": ok, "message": msg})
+ except ValueError as e:
+ return JSONResponse({"error": str(e)}, status_code=404)
+ except Exception as e:
+ logger.error("Test connection failed: %s", e, exc_info=True)
+ return JSONResponse({"error": "Internal error"}, status_code=500)
+
+
+async def api_create_key(request: Request) -> Response:
+ """POST /api/keys — Create a new user API key."""
+ user_session, redirect = _require_user_session(request)
+ if redirect:
+ return JSONResponse({"error": "Unauthorized"}, status_code=401)
+
+ try:
+ body = await request.json()
+ except Exception:
+ return JSONResponse({"error": "Invalid JSON"}, status_code=400)
+
+ from core.user_keys import get_user_key_manager
+
+ try:
+ key_mgr = get_user_key_manager()
+ result = await key_mgr.create_key(
+ user_id=user_session["user_id"],
+ name=body.get("name", "Default"),
+ scopes=body.get("scopes", "read write"),
+ expires_in_days=body.get("expires_in_days"),
+ )
+ return JSONResponse({"key": result})
+ except RuntimeError as e:
+ return JSONResponse({"error": str(e)}, status_code=503)
+ except Exception as e:
+ logger.error("Failed to create API key: %s", e, exc_info=True)
+ return JSONResponse({"error": "Internal error"}, status_code=500)
+
+
+async def api_list_keys(request: Request) -> Response:
+ """GET /api/keys — List user's API keys."""
+ user_session, redirect = _require_user_session(request)
+ if redirect:
+ return JSONResponse({"error": "Unauthorized"}, status_code=401)
+
+ from core.user_keys import get_user_key_manager
+
+ try:
+ key_mgr = get_user_key_manager()
+ keys = await key_mgr.list_keys(user_session["user_id"])
+ return JSONResponse({"keys": keys})
+ except RuntimeError:
+ return JSONResponse({"keys": []})
+
+
+async def api_delete_key(request: Request) -> Response:
+ """DELETE /api/keys/{id} — Delete an API key."""
+ user_session, redirect = _require_user_session(request)
+ if redirect:
+ return JSONResponse({"error": "Unauthorized"}, status_code=401)
+
+ key_id = request.path_params.get("id", "")
+ from core.user_keys import get_user_key_manager
+
+ try:
+ key_mgr = get_user_key_manager()
+ deleted = await key_mgr.delete_key(key_id, user_session["user_id"])
+ if deleted:
+ return JSONResponse({"message": "Key deleted"})
+ return JSONResponse({"error": "Key not found"}, status_code=404)
+ except RuntimeError as e:
+ return JSONResponse({"error": str(e)}, status_code=503)
+
+
+async def api_get_config(request: Request) -> Response:
+ """GET /api/config/{alias} — Get config snippets for a site."""
+ user_session, redirect = _require_user_session(request)
+ if redirect:
+ return JSONResponse({"error": "Unauthorized"}, status_code=401)
+
+ alias = request.path_params.get("alias", "")
+ client_type = request.query_params.get("client", "claude_desktop")
+
+ import os
+
+ from core.config_snippets import generate_config
+
+ base_url = os.getenv("PUBLIC_URL", "http://localhost:8000")
+
+ try:
+ snippet = generate_config(
+ base_url=base_url,
+ user_id=user_session["user_id"],
+ alias=alias,
+ api_key="mhu_YOUR_API_KEY_HERE",
+ client_type=client_type,
+ )
+ return JSONResponse({"config": snippet, "client_type": client_type})
+ except ValueError as e:
+ return JSONResponse({"error": str(e)}, status_code=400)
+
+
def register_dashboard_routes(mcp):
"""
Register dashboard routes with the MCP server.
@@ -2077,6 +2826,15 @@ def register_dashboard_routes(mcp):
"""
logger.info("Registering dashboard routes...")
+ # Auth routes (E.2: OAuth Social Login)
+ mcp.custom_route("/auth/login", methods=["GET"])(auth_login_page)
+ mcp.custom_route("/auth/callback/{provider}", methods=["GET"])(auth_callback)
+ mcp.custom_route("/auth/logout", methods=["GET", "POST"])(auth_logout)
+ mcp.custom_route("/auth/{provider}", methods=["GET"])(auth_provider_redirect)
+
+ # Profile page
+ mcp.custom_route("/dashboard/profile", methods=["GET"])(dashboard_profile_page)
+
# Login routes
mcp.custom_route("/dashboard/login", methods=["GET"])(dashboard_login_page)
mcp.custom_route("/dashboard/login", methods=["POST"])(dashboard_login_submit)
@@ -2136,4 +2894,23 @@ def register_dashboard_routes(mcp):
dashboard_health_projects_partial
)
+ # Site Management routes (E.3)
+ mcp.custom_route("/dashboard/sites", methods=["GET"])(dashboard_sites_list)
+ mcp.custom_route("/dashboard/sites/add", methods=["GET"])(dashboard_sites_add)
+ mcp.custom_route("/dashboard/connect", methods=["GET"])(dashboard_connect_page)
+
+ # Site Management API (E.3)
+ mcp.custom_route("/api/sites", methods=["GET"])(api_list_sites)
+ mcp.custom_route("/api/sites", methods=["POST"])(api_create_site)
+ mcp.custom_route("/api/sites/{id}/test", methods=["POST"])(api_test_site)
+ mcp.custom_route("/api/sites/{id}", methods=["DELETE"])(api_delete_site)
+
+ # User API Key routes (E.3)
+ mcp.custom_route("/api/keys", methods=["GET"])(api_list_keys)
+ mcp.custom_route("/api/keys", methods=["POST"])(api_create_key)
+ mcp.custom_route("/api/keys/{id}", methods=["DELETE"])(api_delete_key)
+
+ # Config snippet API (E.3)
+ mcp.custom_route("/api/config/{alias}", methods=["GET"])(api_get_config)
+
logger.info("Dashboard routes registered successfully")
diff --git a/core/database.py b/core/database.py
new file mode 100644
index 0000000..1d6f878
--- /dev/null
+++ b/core/database.py
@@ -0,0 +1,700 @@
+"""SQLite database backend for the Live Platform (Track E).
+
+Manages the database connection, schema creation, migrations, and CRUD
+operations for the user system. Uses aiosqlite for async SQLite access
+with WAL mode and foreign key enforcement.
+
+This module is only for user-registered sites on the Live Platform.
+Admin endpoints continue to use env var sites via SiteManager.
+
+Usage:
+ db = await initialize_database()
+ user = await db.create_user(
+ email="user@example.com",
+ name="Jane",
+ provider="github",
+ provider_id="12345",
+ )
+ await db.close()
+
+ # Or as a context manager:
+ async with Database("/tmp/test.db") as db:
+ user = await db.get_user_by_id("some-uuid")
+"""
+
+import logging
+import os
+import uuid
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any
+
+import aiosqlite
+
+logger = logging.getLogger(__name__)
+
+# Default data directory: /app/data in Docker, ./data elsewhere
+_DEFAULT_DATA_DIR = "/app/data" if Path("/app").exists() else "./data"
+
+# Schema version — increment when adding migrations
+SCHEMA_VERSION = 3
+
+# Initial schema DDL
+_SCHEMA_SQL = """\
+-- Users (OAuth social login)
+CREATE TABLE IF NOT EXISTS users (
+ id TEXT PRIMARY KEY,
+ email TEXT UNIQUE NOT NULL,
+ name TEXT,
+ avatar_url TEXT,
+ provider TEXT NOT NULL,
+ provider_id TEXT NOT NULL,
+ role TEXT NOT NULL DEFAULT 'user',
+ created_at TEXT NOT NULL,
+ last_login TEXT,
+ UNIQUE(provider, provider_id)
+);
+
+-- User's registered sites (any plugin type)
+CREATE TABLE IF NOT EXISTS sites (
+ id TEXT PRIMARY KEY,
+ user_id TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+ plugin_type TEXT NOT NULL,
+ alias TEXT NOT NULL,
+ url TEXT NOT NULL,
+ credentials BLOB NOT NULL,
+ status TEXT NOT NULL DEFAULT 'pending',
+ status_msg TEXT,
+ last_health TEXT,
+ created_at TEXT NOT NULL,
+ UNIQUE(user_id, alias)
+);
+
+-- Per-user API keys (for MCP client authentication)
+CREATE TABLE IF NOT EXISTS user_api_keys (
+ id TEXT PRIMARY KEY,
+ user_id TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+ key_hash TEXT NOT NULL,
+ key_prefix TEXT,
+ name TEXT NOT NULL,
+ scopes TEXT NOT NULL DEFAULT 'read write',
+ last_used TEXT,
+ use_count INTEGER NOT NULL DEFAULT 0,
+ created_at TEXT NOT NULL,
+ expires_at TEXT
+);
+
+-- WP plugin connection tokens (short-lived, for MCP Connect plugin)
+CREATE TABLE IF NOT EXISTS connection_tokens (
+ id TEXT PRIMARY KEY,
+ user_id TEXT NOT NULL REFERENCES users(id) ON DELETE CASCADE,
+ token_hash TEXT NOT NULL,
+ created_at TEXT NOT NULL,
+ expires_at TEXT NOT NULL,
+ used INTEGER NOT NULL DEFAULT 0
+);
+
+-- Schema version tracking
+CREATE TABLE IF NOT EXISTS schema_version (
+ version INTEGER PRIMARY KEY,
+ applied_at TEXT NOT NULL
+);
+"""
+
+# Migration registry: version -> SQL string
+# Migrations are applied sequentially from the current version + 1 up to SCHEMA_VERSION.
+_MIGRATIONS: dict[int, str] = {
+ 2: (
+ "ALTER TABLE user_api_keys ADD COLUMN key_prefix TEXT;\n"
+ "CREATE INDEX IF NOT EXISTS idx_user_api_keys_prefix ON user_api_keys(key_prefix);\n"
+ ),
+}
+
+
+class Database:
+ """Async SQLite database for the Live Platform.
+
+ Manages connections, schema, migrations, and provides CRUD helpers
+ for users, sites, API keys, and connection tokens.
+
+ Attributes:
+ db_path: Resolved path to the SQLite database file.
+ """
+
+ def __init__(self, db_path: str | None = None) -> None:
+ """Initialize database configuration.
+
+ Args:
+ db_path: Path to the SQLite database file. If not provided,
+ reads DATABASE_PATH env var, defaulting to ``data/mcphub.db``.
+ """
+ if db_path is None:
+ db_file = os.getenv("DATABASE_PATH", None)
+ if db_file:
+ self.db_path = Path(db_file)
+ else:
+ self.db_path = Path(_DEFAULT_DATA_DIR) / "mcphub.db"
+ else:
+ self.db_path = Path(db_path)
+
+ self._conn: aiosqlite.Connection | None = None
+
+ async def initialize(self) -> None:
+ """Create data directory, connect, and set up schema.
+
+ Enables WAL mode and foreign keys, creates tables if they do
+ not exist, and runs any pending migrations.
+ """
+ # Ensure parent directory exists
+ self.db_path.parent.mkdir(parents=True, exist_ok=True)
+
+ self._conn = await aiosqlite.connect(str(self.db_path))
+ self._conn.row_factory = aiosqlite.Row
+
+ # Enable WAL mode for better concurrent read performance
+ await self._conn.execute("PRAGMA journal_mode=WAL")
+ # Enable foreign key enforcement
+ await self._conn.execute("PRAGMA foreign_keys=ON")
+
+ await self._create_schema()
+ await self._run_migrations()
+
+ logger.info("Database initialized at %s", self.db_path)
+
+ async def close(self) -> None:
+ """Close the database connection."""
+ if self._conn is not None:
+ await self._conn.close()
+ self._conn = None
+ logger.info("Database connection closed")
+
+ # ------------------------------------------------------------------
+ # Context manager
+ # ------------------------------------------------------------------
+
+ async def __aenter__(self) -> "Database":
+ """Enter async context — initialize and return self."""
+ await self.initialize()
+ return self
+
+ async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
+ """Exit async context — close the connection."""
+ await self.close()
+
+ # ------------------------------------------------------------------
+ # Low-level query helpers
+ # ------------------------------------------------------------------
+
+ def _require_conn(self) -> aiosqlite.Connection:
+ """Return the active connection or raise if not initialized."""
+ if self._conn is None:
+ raise RuntimeError("Database not initialized. Call initialize() first.")
+ return self._conn
+
+ async def execute(self, sql: str, params: tuple[Any, ...] = ()) -> aiosqlite.Cursor:
+ """Execute a single write SQL statement and commit.
+
+ Args:
+ sql: SQL statement.
+ params: Bind parameters.
+
+ Returns:
+ The aiosqlite Cursor.
+ """
+ conn = self._require_conn()
+ cursor = await conn.execute(sql, params)
+ await conn.commit()
+ return cursor
+
+ async def executemany(self, sql: str, params_list: list[tuple[Any, ...]]) -> aiosqlite.Cursor:
+ """Execute a SQL statement with multiple parameter sets and commit.
+
+ Args:
+ sql: SQL statement.
+ params_list: List of parameter tuples.
+
+ Returns:
+ The aiosqlite Cursor.
+ """
+ conn = self._require_conn()
+ cursor = await conn.executemany(sql, params_list)
+ await conn.commit()
+ return cursor
+
+ async def fetchone(self, sql: str, params: tuple[Any, ...] = ()) -> dict[str, Any] | None:
+ """Fetch a single row as a dictionary (read-only, no commit).
+
+ Args:
+ sql: SQL query.
+ params: Bind parameters.
+
+ Returns:
+ Row as a dict, or None if no result.
+ """
+ conn = self._require_conn()
+ cursor = await conn.execute(sql, params)
+ row = await cursor.fetchone()
+ if row is None:
+ return None
+ return dict(row)
+
+ async def fetchall(self, sql: str, params: tuple[Any, ...] = ()) -> list[dict[str, Any]]:
+ """Fetch all rows as a list of dictionaries (read-only, no commit).
+
+ Args:
+ sql: SQL query.
+ params: Bind parameters.
+
+ Returns:
+ List of rows, each as a dict.
+ """
+ conn = self._require_conn()
+ cursor = await conn.execute(sql, params)
+ rows = await cursor.fetchall()
+ return [dict(row) for row in rows]
+
+ # ------------------------------------------------------------------
+ # Schema management
+ # ------------------------------------------------------------------
+
+ async def _create_schema(self) -> None:
+ """Create all tables if they do not already exist."""
+ conn = self._require_conn()
+
+ # Check if it's a completely fresh DB (no users table)
+ row = await self.fetchone("SELECT name FROM sqlite_master WHERE type='table' AND name='users'")
+ is_fresh = row is None
+
+ await conn.executescript(_SCHEMA_SQL)
+
+ if is_fresh:
+ # Seed initial schema version
+ row = await self.fetchone("SELECT MAX(version) AS v FROM schema_version")
+ if row is None or row["v"] is None:
+ await self.execute(
+ "INSERT INTO schema_version (version, applied_at) VALUES (?, ?)",
+ (SCHEMA_VERSION, _utc_now()),
+ )
+ else:
+ # For existing v1 databases, if schema_version was just created
+ row = await self.fetchone("SELECT COUNT(*) as c FROM schema_version")
+ if row and row["c"] == 0:
+ await self.execute(
+ "INSERT INTO schema_version (version, applied_at) VALUES (?, ?)",
+ (1, _utc_now()),
+ )
+
+ async def _run_migrations(self) -> None:
+ """Apply any pending migrations sequentially."""
+ conn = self._require_conn()
+ row = await self.fetchone("SELECT MAX(version) AS v FROM schema_version")
+ current = row["v"] if row and row["v"] is not None else 0
+
+ for version in range(current + 1, SCHEMA_VERSION + 1):
+ if version == 3:
+ logger.info("Applying python migration for version 3")
+ # Idempotent repair for v1->v2 upgrade bug (some DBs stamped v2 but missed the column)
+ try:
+ await self.execute("ALTER TABLE user_api_keys ADD COLUMN key_prefix TEXT")
+ except Exception as e:
+ if "duplicate column name" not in str(e).lower():
+ raise
+ await self.execute("CREATE INDEX IF NOT EXISTS idx_user_api_keys_prefix ON user_api_keys(key_prefix)")
+ else:
+ migration_sql = _MIGRATIONS.get(version)
+ if migration_sql is not None:
+ logger.info("Applying migration to version %d", version)
+ await conn.executescript(migration_sql)
+ logger.info("Migration to version %d applied", version)
+ else:
+ logger.warning("No migration SQL for version %d, recording version only", version)
+
+ # Always record version to avoid infinite retry
+ await self.execute(
+ "INSERT INTO schema_version (version, applied_at) VALUES (?, ?)",
+ (version, _utc_now()),
+ )
+
+ # ------------------------------------------------------------------
+ # User CRUD
+ # ------------------------------------------------------------------
+
+ async def create_user(
+ self,
+ email: str,
+ name: str | None,
+ provider: str,
+ provider_id: str,
+ avatar_url: str | None = None,
+ role: str = "user",
+ ) -> dict[str, Any]:
+ """Create a new user.
+
+ Args:
+ email: User email (unique).
+ name: Display name.
+ provider: OAuth provider ('github' or 'google').
+ provider_id: Provider's unique user ID.
+ avatar_url: URL to user avatar.
+ role: User role ('user' or 'admin').
+
+ Returns:
+ The created user as a dict.
+
+ Raises:
+ aiosqlite.IntegrityError: If email or provider+provider_id already exists.
+ """
+ user_id = str(uuid.uuid4())
+ now = _utc_now()
+
+ await self.execute(
+ "INSERT INTO users (id, email, name, avatar_url, provider, provider_id, role, "
+ "created_at, last_login) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
+ (user_id, email, name, avatar_url, provider, provider_id, role, now, now),
+ )
+
+ result = await self.get_user_by_id(user_id)
+ if result is None:
+ raise RuntimeError(f"Failed to read back created user {user_id}")
+ return result
+
+ async def get_user_by_id(self, user_id: str) -> dict[str, Any] | None:
+ """Get a user by their UUID.
+
+ Args:
+ user_id: User UUID.
+
+ Returns:
+ User dict, or None if not found.
+ """
+ return await self.fetchone("SELECT * FROM users WHERE id = ?", (user_id,))
+
+ async def get_user_by_email(self, email: str) -> dict[str, Any] | None:
+ """Get a user by their email.
+
+ Args:
+ email: User email.
+
+ Returns:
+ User dict, or None if not found.
+ """
+ return await self.fetchone("SELECT * FROM users WHERE email = ?", (email,))
+
+ async def get_user_by_provider(self, provider: str, provider_id: str) -> dict[str, Any] | None:
+ """Get a user by OAuth provider and provider ID.
+
+ Args:
+ provider: OAuth provider name.
+ provider_id: Provider's unique user ID.
+
+ Returns:
+ User dict, or None if not found.
+ """
+ return await self.fetchone(
+ "SELECT * FROM users WHERE provider = ? AND provider_id = ?",
+ (provider, provider_id),
+ )
+
+ async def update_user_last_login(self, user_id: str) -> None:
+ """Update a user's last_login timestamp to now.
+
+ Args:
+ user_id: User UUID.
+ """
+ await self.execute(
+ "UPDATE users SET last_login = ? WHERE id = ?",
+ (_utc_now(), user_id),
+ )
+
+ # ------------------------------------------------------------------
+ # Site CRUD
+ # ------------------------------------------------------------------
+
+ async def create_site(
+ self,
+ user_id: str,
+ plugin_type: str,
+ alias: str,
+ url: str,
+ credentials: bytes,
+ status: str = "pending",
+ status_msg: str | None = None,
+ ) -> dict[str, Any]:
+ """Create a new site for a user.
+
+ Args:
+ user_id: Owner's UUID.
+ plugin_type: Plugin type (e.g. 'wordpress', 'gitea').
+ alias: User-chosen friendly name (unique per user).
+ url: Site URL.
+ credentials: AES-256-GCM encrypted credentials blob.
+ status: Initial status ('pending', 'active', 'error', 'disabled').
+ status_msg: Optional human-readable status message.
+
+ Returns:
+ The created site as a dict.
+
+ Raises:
+ aiosqlite.IntegrityError: If alias is duplicated for the same user.
+ """
+ site_id = str(uuid.uuid4())
+ now = _utc_now()
+
+ await self.execute(
+ "INSERT INTO sites (id, user_id, plugin_type, alias, url, credentials, "
+ "status, status_msg, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
+ (site_id, user_id, plugin_type, alias, url, credentials, status, status_msg, now),
+ )
+
+ result = await self.get_site(site_id, user_id)
+ if result is None:
+ raise RuntimeError(f"Failed to read back created site {site_id}")
+ return result
+
+ async def get_sites_by_user(self, user_id: str) -> list[dict[str, Any]]:
+ """Get all sites belonging to a user.
+
+ Args:
+ user_id: Owner's UUID.
+
+ Returns:
+ List of site dicts.
+ """
+ return await self.fetchall(
+ "SELECT * FROM sites WHERE user_id = ? ORDER BY created_at",
+ (user_id,),
+ )
+
+ async def get_site(self, site_id: str, user_id: str) -> dict[str, Any] | None:
+ """Get a single site, scoped to a specific user.
+
+ Args:
+ site_id: Site UUID.
+ user_id: Owner's UUID (for tenant isolation).
+
+ Returns:
+ Site dict, or None if not found or not owned by user.
+ """
+ return await self.fetchone(
+ "SELECT * FROM sites WHERE id = ? AND user_id = ?",
+ (site_id, user_id),
+ )
+
+ async def delete_site(self, site_id: str, user_id: str) -> bool:
+ """Delete a site, scoped to a specific user.
+
+ Args:
+ site_id: Site UUID.
+ user_id: Owner's UUID (for tenant isolation).
+
+ Returns:
+ True if a row was deleted, False otherwise.
+ """
+ cursor = await self.execute(
+ "DELETE FROM sites WHERE id = ? AND user_id = ?",
+ (site_id, user_id),
+ )
+ return cursor.rowcount > 0
+
+ async def update_site_status(
+ self,
+ site_id: str,
+ status: str,
+ status_msg: str | None = None,
+ user_id: str | None = None,
+ ) -> None:
+ """Update a site's status and optional status message.
+
+ Args:
+ site_id: Site UUID.
+ status: New status ('pending', 'active', 'error', 'disabled').
+ status_msg: Optional human-readable status message.
+ user_id: Optional user UUID for tenant-scoped update. When provided,
+ only updates if the site belongs to this user. When None,
+ performs system-level update (e.g., health checks).
+ """
+ if user_id is not None:
+ await self.execute(
+ "UPDATE sites SET status = ?, status_msg = ? WHERE id = ? AND user_id = ?",
+ (status, status_msg, site_id, user_id),
+ )
+ else:
+ await self.execute(
+ "UPDATE sites SET status = ?, status_msg = ? WHERE id = ?",
+ (status, status_msg, site_id),
+ )
+
+ async def get_site_by_alias(self, user_id: str, alias: str) -> dict[str, Any] | None:
+ """Get a site by user ID and alias.
+
+ Args:
+ user_id: Owner's UUID.
+ alias: Site alias (unique per user).
+
+ Returns:
+ Site dict, or None if not found.
+ """
+ return await self.fetchone(
+ "SELECT * FROM sites WHERE user_id = ? AND alias = ?",
+ (user_id, alias),
+ )
+
+ async def count_sites_by_user(self, user_id: str) -> int:
+ """Count the number of sites belonging to a user.
+
+ Args:
+ user_id: Owner's UUID.
+
+ Returns:
+ Number of sites.
+ """
+ row = await self.fetchone(
+ "SELECT COUNT(*) AS cnt FROM sites WHERE user_id = ?",
+ (user_id,),
+ )
+ return row["cnt"] if row else 0
+
+ # ------------------------------------------------------------------
+ # User API Key CRUD
+ # ------------------------------------------------------------------
+
+ async def create_api_key(
+ self,
+ user_id: str,
+ key_hash: str,
+ key_prefix: str,
+ name: str,
+ scopes: str = "read write",
+ expires_at: str | None = None,
+ ) -> dict[str, Any]:
+ """Create a new user API key.
+
+ Args:
+ user_id: Owner's UUID.
+ key_hash: bcrypt hash of the API key.
+ key_prefix: First 8 chars after ``mhu_`` prefix for fast lookup.
+ name: Human label (e.g. "Claude Desktop").
+ scopes: Space-separated scopes.
+ expires_at: Optional ISO 8601 expiry timestamp.
+
+ Returns:
+ The created API key row as a dict.
+ """
+ key_id = str(uuid.uuid4())
+ now = _utc_now()
+
+ await self.execute(
+ "INSERT INTO user_api_keys "
+ "(id, user_id, key_hash, key_prefix, name, scopes, created_at, expires_at) "
+ "VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
+ (key_id, user_id, key_hash, key_prefix, name, scopes, now, expires_at),
+ )
+
+ result = await self.fetchone(
+ "SELECT id, user_id, key_prefix, name, scopes, last_used, use_count, "
+ "created_at, expires_at FROM user_api_keys WHERE id = ?",
+ (key_id,),
+ )
+ if result is None:
+ raise RuntimeError(f"Failed to read back created API key {key_id}")
+ return result
+
+ async def get_api_keys_by_user(self, user_id: str) -> list[dict[str, Any]]:
+ """Get all API keys for a user (without key_hash).
+
+ Args:
+ user_id: Owner's UUID.
+
+ Returns:
+ List of API key dicts (key_hash excluded).
+ """
+ return await self.fetchall(
+ "SELECT id, user_id, key_prefix, name, scopes, last_used, use_count, "
+ "created_at, expires_at FROM user_api_keys WHERE user_id = ? ORDER BY created_at",
+ (user_id,),
+ )
+
+ async def get_api_key_by_prefix(self, key_prefix: str) -> dict[str, Any] | None:
+ """Get an API key by its prefix (includes key_hash for verification).
+
+ Args:
+ key_prefix: First 8 chars of the key after ``mhu_``.
+
+ Returns:
+ API key dict including key_hash, or None.
+ """
+ return await self.fetchone(
+ "SELECT * FROM user_api_keys WHERE key_prefix = ?",
+ (key_prefix,),
+ )
+
+ async def delete_api_key(self, key_id: str, user_id: str) -> bool:
+ """Delete an API key, scoped to a specific user.
+
+ Args:
+ key_id: API key UUID.
+ user_id: Owner's UUID (for tenant isolation).
+
+ Returns:
+ True if a row was deleted, False otherwise.
+ """
+ cursor = await self.execute(
+ "DELETE FROM user_api_keys WHERE id = ? AND user_id = ?",
+ (key_id, user_id),
+ )
+ return cursor.rowcount > 0
+
+ async def update_api_key_usage(self, key_id: str) -> None:
+ """Increment use_count and update last_used timestamp for an API key.
+
+ Args:
+ key_id: API key UUID.
+ """
+ await self.execute(
+ "UPDATE user_api_keys SET use_count = use_count + 1, last_used = ? WHERE id = ?",
+ (_utc_now(), key_id),
+ )
+
+
+# ======================================================================
+# Module-level helpers
+# ======================================================================
+
+
+def _utc_now() -> str:
+ """Return the current UTC time as an ISO 8601 string."""
+ return datetime.now(UTC).isoformat()
+
+
+# Singleton instance
+_database: Database | None = None
+
+
+def get_database() -> Database:
+ """Get the singleton Database instance.
+
+ Returns:
+ The Database singleton.
+
+ Raises:
+ RuntimeError: If initialize_database() has not been called.
+ """
+ if _database is None:
+ raise RuntimeError("Database not initialized. Call initialize_database() first.")
+ return _database
+
+
+async def initialize_database(db_path: str | None = None) -> Database:
+ """Create, initialize, and store the singleton Database instance.
+
+ Args:
+ db_path: Optional path to the SQLite database file.
+
+ Returns:
+ The initialized Database instance.
+ """
+ global _database
+ db = Database(db_path)
+ await db.initialize()
+ _database = db
+ return db
diff --git a/core/encryption.py b/core/encryption.py
new file mode 100644
index 0000000..16999c7
--- /dev/null
+++ b/core/encryption.py
@@ -0,0 +1,212 @@
+"""Credential encryption for the Live Platform.
+
+Provides AES-256-GCM encryption with HKDF key derivation for per-site
+credential storage. Credentials are encrypted as JSON blobs and stored
+in SQLite. Decryption happens only during tool execution and plaintext
+is never logged.
+
+Usage:
+ encryption = get_credential_encryption()
+ cipherdata = encryption.encrypt_credentials(
+ {"username": "admin", "app_password": "xxxx xxxx"},
+ site_id="site_abc123",
+ )
+ credentials = encryption.decrypt_credentials(cipherdata, site_id="site_abc123")
+"""
+
+import base64
+import json
+import logging
+import os
+
+from cryptography.hazmat.primitives import hashes
+from cryptography.hazmat.primitives.ciphers.aead import AESGCM
+from cryptography.hazmat.primitives.kdf.hkdf import HKDF
+
+logger = logging.getLogger(__name__)
+
+# Constants
+_NONCE_LENGTH = 12 # 96-bit nonce for AES-GCM
+_KEY_LENGTH = 32 # 256-bit key
+_HKDF_SALT = b"mcphub-v1"
+_FORMAT_VERSION = b"\x01" # Wire format version for future migration support
+
+
+class CredentialEncryption:
+ """AES-256-GCM encryption with per-site HKDF-derived keys.
+
+ The master key is read from the ENCRYPTION_KEY environment variable
+ (base64-encoded 32-byte key). Per-site keys are derived via HKDF
+ using the site_id as the info parameter, ensuring each site has a
+ unique encryption key.
+
+ Storage format: version (1 byte) || nonce (12 bytes) || ciphertext || tag (16 bytes)
+ """
+
+ def __init__(self, encryption_key: str | None = None) -> None:
+ """Initialize credential encryption.
+
+ Args:
+ encryption_key: Base64-encoded 32-byte key. If not provided,
+ reads from the ENCRYPTION_KEY environment variable.
+
+ Raises:
+ ValueError: If the encryption key is missing or invalid.
+ """
+ raw_key = encryption_key or os.getenv("ENCRYPTION_KEY")
+
+ if not raw_key:
+ raise ValueError(
+ "ENCRYPTION_KEY is required. Set it as an environment variable "
+ "or pass it directly. Generate one with: "
+ 'python -c "import os, base64; print(base64.b64encode(os.urandom(32)).decode())"'
+ )
+
+ try:
+ self._master_key = base64.b64decode(raw_key)
+ except Exception as exc:
+ raise ValueError("ENCRYPTION_KEY must be a valid base64-encoded string.") from exc
+
+ if len(self._master_key) != _KEY_LENGTH:
+ raise ValueError(
+ f"ENCRYPTION_KEY must decode to exactly {_KEY_LENGTH} bytes, "
+ f"got {len(self._master_key)} bytes."
+ )
+
+ logger.info("Credential encryption initialized")
+
+ def _derive_key(self, site_id: str) -> bytes:
+ """Derive a per-site encryption key using HKDF.
+
+ Args:
+ site_id: Unique identifier for the site.
+
+ Returns:
+ 32-byte derived key for the given site.
+ """
+ hkdf = HKDF(
+ algorithm=hashes.SHA256(),
+ length=_KEY_LENGTH,
+ salt=_HKDF_SALT,
+ info=site_id.encode("utf-8"),
+ )
+ return hkdf.derive(self._master_key)
+
+ def encrypt(self, plaintext: str, site_id: str) -> bytes:
+ """Encrypt a plaintext string for a specific site.
+
+ Args:
+ plaintext: The string to encrypt.
+ site_id: Site identifier used for key derivation.
+
+ Returns:
+ Encrypted bytes: version (1) || nonce (12) || ciphertext || tag (16).
+ """
+ derived_key = self._derive_key(site_id)
+ aesgcm = AESGCM(derived_key)
+ nonce = os.urandom(_NONCE_LENGTH)
+ ciphertext_with_tag = aesgcm.encrypt(nonce, plaintext.encode("utf-8"), None)
+ return _FORMAT_VERSION + nonce + ciphertext_with_tag
+
+ def decrypt(self, cipherdata: bytes, site_id: str) -> str:
+ """Decrypt cipherdata for a specific site.
+
+ Args:
+ cipherdata: Encrypted bytes (nonce || ciphertext || tag).
+ site_id: Site identifier used for key derivation.
+
+ Returns:
+ Original plaintext string.
+
+ Raises:
+ cryptography.exceptions.InvalidTag: If decryption fails
+ (wrong key, tampered data, or wrong site_id).
+ ValueError: If cipherdata is too short or has unsupported version.
+ """
+ # Minimum: 1 (version) + 12 (nonce) + 16 (tag) = 29 bytes
+ min_length = 1 + _NONCE_LENGTH + 16
+ if len(cipherdata) < min_length:
+ raise ValueError(
+ f"Cipherdata too short: expected at least {min_length} bytes, "
+ f"got {len(cipherdata)}."
+ )
+
+ version = cipherdata[:1]
+ if version != _FORMAT_VERSION:
+ raise ValueError(
+ f"Unsupported encryption format version: {version!r}. "
+ f"Expected {_FORMAT_VERSION!r}."
+ )
+
+ nonce = cipherdata[1 : 1 + _NONCE_LENGTH]
+ ciphertext_with_tag = cipherdata[1 + _NONCE_LENGTH :]
+
+ derived_key = self._derive_key(site_id)
+ aesgcm = AESGCM(derived_key)
+ plaintext_bytes = aesgcm.decrypt(nonce, ciphertext_with_tag, None)
+ return plaintext_bytes.decode("utf-8")
+
+ def encrypt_credentials(self, credentials: dict, site_id: str) -> bytes:
+ """Encrypt a credentials dictionary for a specific site.
+
+ The dictionary is serialized to JSON, then encrypted.
+
+ Args:
+ credentials: Dictionary of credentials (e.g., username, password).
+ site_id: Site identifier used for key derivation.
+
+ Returns:
+ Encrypted bytes.
+ """
+ json_str = json.dumps(credentials, separators=(",", ":"), sort_keys=True)
+ return self.encrypt(json_str, site_id)
+
+ def decrypt_credentials(self, cipherdata: bytes, site_id: str) -> dict:
+ """Decrypt cipherdata back to a credentials dictionary.
+
+ Args:
+ cipherdata: Encrypted bytes from encrypt_credentials.
+ site_id: Site identifier used for key derivation.
+
+ Returns:
+ Original credentials dictionary.
+
+ Raises:
+ cryptography.exceptions.InvalidTag: If decryption fails.
+ json.JSONDecodeError: If decrypted data is not valid JSON.
+ """
+ json_str = self.decrypt(cipherdata, site_id)
+ return json.loads(json_str)
+
+
+# Global credential encryption instance
+_credential_encryption: CredentialEncryption | None = None
+
+
+def initialize_credential_encryption(
+ encryption_key: str | None = None,
+) -> CredentialEncryption:
+ """Initialize the global credential encryption instance.
+
+ Args:
+ encryption_key: Base64-encoded 32-byte key. If not provided,
+ reads from the ENCRYPTION_KEY environment variable.
+
+ Returns:
+ The initialized CredentialEncryption instance.
+ """
+ global _credential_encryption
+ _credential_encryption = CredentialEncryption(encryption_key)
+ return _credential_encryption
+
+
+def get_credential_encryption() -> CredentialEncryption:
+ """Get the global credential encryption instance.
+
+ Lazily initializes from the ENCRYPTION_KEY environment variable
+ if not already initialized via initialize_credential_encryption().
+ """
+ global _credential_encryption
+ if _credential_encryption is None:
+ _credential_encryption = CredentialEncryption()
+ return _credential_encryption
diff --git a/core/health.py b/core/health.py
index 2f7aa6f..cbdac2b 100644
--- a/core/health.py
+++ b/core/health.py
@@ -12,6 +12,7 @@ This module provides comprehensive health monitoring capabilities including:
Author: MCP Hub Team
"""
+import asyncio
import json
import logging
import time
@@ -175,6 +176,11 @@ class HealthMonitor:
# Request rate tracking (for requests per minute)
self.request_timestamps: deque = deque(maxlen=1000)
+ # Active background checks
+ self.latest_health_status: dict[str, ProjectHealthStatus] = {}
+ self._bg_task: asyncio.Task | None = None
+ self._is_running = False
+
logger.info("HealthMonitor initialized")
def _setup_default_thresholds(self):
@@ -470,7 +476,7 @@ class HealthMonitor:
}
alerts = self._check_alerts(project_id, alert_check_data)
- return ProjectHealthStatus(
+ status = ProjectHealthStatus(
project_id=project_id,
healthy=is_healthy,
last_check=datetime.now(UTC),
@@ -480,6 +486,8 @@ class HealthMonitor:
alerts=alerts,
details=health_result,
)
+ self.latest_health_status[project_id] = status
+ return status
except Exception as e:
response_time_ms = (time.time() - start_time) * 1000
@@ -493,7 +501,7 @@ class HealthMonitor:
error_message=error_msg,
)
- return ProjectHealthStatus(
+ status = ProjectHealthStatus(
project_id=project_id,
healthy=False,
last_check=datetime.now(UTC),
@@ -502,6 +510,8 @@ class HealthMonitor:
recent_errors=[error_msg],
alerts=[f"CRITICAL: Health check failed - {error_msg}"],
)
+ self.latest_health_status[project_id] = status
+ return status
def _find_site_info(self, project_id: str) -> dict[str, Any] | None:
"""Find site info from SiteManager by full_id."""
@@ -762,7 +772,45 @@ class HealthMonitor:
self.failed_requests = 0
self.response_times.clear()
self.request_timestamps.clear()
+ self.latest_health_status.clear()
logger.warning("All metrics have been reset")
+
+ async def start_background_checks(self, interval_seconds: int = 60):
+ """Start background health checks for all projects."""
+ if self._is_running:
+ return
+
+ self._is_running = True
+ logger.info(f"Starting background health checks every {interval_seconds} seconds")
+
+ async def _loop():
+ # Initial wait to let server start up fully
+ await asyncio.sleep(5)
+ while self._is_running:
+ try:
+ await self.check_all_projects_health(include_metrics=True)
+ except Exception as e:
+ logger.error(f"Error in background health check loop: {e}")
+
+ # Sleep interval, check _is_running periodically
+ for _ in range(interval_seconds):
+ if not self._is_running:
+ break
+ await asyncio.sleep(1)
+
+ self._bg_task = asyncio.create_task(_loop())
+
+ async def stop_background_checks(self):
+ """Stop background health checks."""
+ self._is_running = False
+ if self._bg_task:
+ self._bg_task.cancel()
+ try:
+ await self._bg_task
+ except asyncio.CancelledError:
+ pass
+ self._bg_task = None
+ logger.info("Background health checks stopped")
# Singleton instance
diff --git a/core/site_api.py b/core/site_api.py
new file mode 100644
index 0000000..2b536b6
--- /dev/null
+++ b/core/site_api.py
@@ -0,0 +1,545 @@
+"""Site management logic for the Live Platform (Track E.3).
+
+Provides site CRUD operations, connection validation, and credential
+field definitions for all 9 plugin types. Coordinates between the
+database, encryption, and plugin health check layers.
+
+Usage:
+ from core.site_api import create_user_site, get_user_sites, validate_site_connection
+
+ ok, msg = await validate_site_connection("wordpress", "https://example.com", creds)
+ site = await create_user_site(user_id, "wordpress", "myblog", url, creds)
+"""
+
+import logging
+import os
+from typing import Any
+
+import aiohttp
+
+logger = logging.getLogger(__name__)
+
+# Maximum sites per user (configurable via env var)
+MAX_SITES_PER_USER = int(os.getenv("MAX_SITES_PER_USER", "10"))
+
+# Plugin credential field definitions — drives the dynamic "Add Site" form
+# and server-side validation. Each field has:
+# name: form input name (matches credential JSON key)
+# label: display label
+# type: "text" or "password"
+# required: whether the field is mandatory
+PLUGIN_CREDENTIAL_FIELDS: dict[str, list[dict[str, Any]]] = {
+ "wordpress": [
+ {
+ "name": "username",
+ "label": "Username",
+ "type": "text",
+ "required": True,
+ "hint": "Your WordPress admin username",
+ },
+ {
+ "name": "app_password",
+ "label": "Application Password",
+ "type": "password",
+ "required": True,
+ "hint": "WordPress Admin → Users → Profile → Application Passwords",
+ },
+ ],
+ "woocommerce": [
+ {
+ "name": "consumer_key",
+ "label": "Consumer Key",
+ "type": "text",
+ "required": True,
+ "hint": "WooCommerce → Settings → Advanced → REST API",
+ },
+ {
+ "name": "consumer_secret",
+ "label": "Consumer Secret",
+ "type": "password",
+ "required": True,
+ "hint": "Shown once when creating the REST API key",
+ },
+ ],
+ "wordpress_advanced": [
+ {
+ "name": "username",
+ "label": "Username",
+ "type": "text",
+ "required": True,
+ "hint": "Your WordPress admin username",
+ },
+ {
+ "name": "app_password",
+ "label": "Application Password",
+ "type": "password",
+ "required": True,
+ "hint": "WordPress Admin → Users → Profile → Application Passwords",
+ },
+ {
+ "name": "container",
+ "label": "Docker Container Name",
+ "type": "text",
+ "required": False,
+ "hint": "Docker container running WordPress (for WP-CLI access)",
+ },
+ ],
+ "gitea": [
+ {
+ "name": "token",
+ "label": "Access Token",
+ "type": "password",
+ "required": True,
+ "hint": "Gitea → Settings → Applications → Generate Token",
+ },
+ ],
+ "n8n": [
+ {
+ "name": "api_key",
+ "label": "API Key",
+ "type": "password",
+ "required": True,
+ "hint": "n8n → Settings → API → Create API Key",
+ },
+ ],
+ "supabase": [
+ {
+ "name": "service_role_key",
+ "label": "Service Role Key",
+ "type": "password",
+ "required": True,
+ "hint": "Supabase Dashboard → Settings → API → service_role key",
+ },
+ {
+ "name": "anon_key",
+ "label": "Anon Key",
+ "type": "password",
+ "required": True,
+ "hint": "Supabase Dashboard → Settings → API → anon key",
+ },
+ ],
+ "openpanel": [
+ {
+ "name": "client_id",
+ "label": "Client ID",
+ "type": "text",
+ "required": True,
+ "hint": "OpenPanel admin panel → API section",
+ },
+ {
+ "name": "client_secret",
+ "label": "Client Secret",
+ "type": "password",
+ "required": True,
+ "hint": "Generated with your Client ID",
+ },
+ ],
+ "appwrite": [
+ {
+ "name": "project_id",
+ "label": "Project ID",
+ "type": "text",
+ "required": True,
+ "hint": "Appwrite Console → Project Settings → Project ID",
+ },
+ {
+ "name": "api_key",
+ "label": "API Key",
+ "type": "password",
+ "required": True,
+ "hint": "Appwrite Console → Project Settings → API Keys → Create",
+ },
+ ],
+ "directus": [
+ {
+ "name": "token",
+ "label": "Static Token",
+ "type": "password",
+ "required": True,
+ "hint": "Directus → Settings → User → Static Token",
+ },
+ ],
+}
+
+# Plugin display names for UI
+PLUGIN_DISPLAY_NAMES: dict[str, str] = {
+ "wordpress": "WordPress",
+ "woocommerce": "WooCommerce",
+ "wordpress_advanced": "WordPress Advanced",
+ "gitea": "Gitea",
+ "n8n": "n8n",
+ "supabase": "Supabase",
+ "openpanel": "OpenPanel",
+ "appwrite": "Appwrite",
+ "directus": "Directus",
+}
+
+# Health check endpoints per plugin type
+_HEALTH_ENDPOINTS: dict[str, dict[str, Any]] = {
+ "wordpress": {"path": "/wp-json/wp/v2/users/me", "method": "GET"},
+ "woocommerce": {"path": "/wp-json/wc/v3/system_status", "method": "GET"},
+ "wordpress_advanced": {"path": "/wp-json/wp/v2/users/me", "method": "GET"},
+ "gitea": {"path": "/api/v1/user", "method": "GET"},
+ "n8n": {"path": "/healthz", "method": "GET"},
+ "supabase": {"path": "/rest/v1/", "method": "GET"},
+ "openpanel": {"path": "/api/v1/oauth/token", "method": "POST"},
+ "appwrite": {"path": "/v1/health", "method": "GET"},
+ "directus": {"path": "/server/health", "method": "GET"},
+}
+
+
+def get_credential_fields(plugin_type: str) -> list[dict[str, Any]]:
+ """Get credential field definitions for a plugin type.
+
+ Args:
+ plugin_type: Plugin type name.
+
+ Returns:
+ List of field definition dicts.
+
+ Raises:
+ ValueError: If plugin_type is unknown.
+ """
+ fields = PLUGIN_CREDENTIAL_FIELDS.get(plugin_type)
+ if fields is None:
+ raise ValueError(
+ f"Unknown plugin type '{plugin_type}'. "
+ f"Valid: {list(PLUGIN_CREDENTIAL_FIELDS.keys())}"
+ )
+ return fields
+
+
+def get_user_credential_fields() -> dict[str, list[dict[str, Any]]]:
+ """Get credential fields for regular (non-admin) users.
+
+ Excludes plugin types that require admin-level infrastructure
+ (e.g., wordpress_advanced needs Docker access).
+
+ Returns:
+ Filtered dict of plugin_type -> field definitions.
+ """
+ excluded = {"wordpress_advanced"}
+ return {k: v for k, v in PLUGIN_CREDENTIAL_FIELDS.items() if k not in excluded}
+
+
+def get_user_plugin_names() -> dict[str, str]:
+ """Get plugin display names for regular (non-admin) users.
+
+ Excludes plugins that require admin-level infrastructure.
+
+ Returns:
+ Filtered dict of plugin_type -> display name.
+ """
+ excluded = {"wordpress_advanced"}
+ return {k: v for k, v in PLUGIN_DISPLAY_NAMES.items() if k not in excluded}
+
+
+def validate_credentials(plugin_type: str, credentials: dict[str, str]) -> tuple[bool, list[str]]:
+ """Validate that all required credential fields are present and non-empty.
+
+ Args:
+ plugin_type: Plugin type name.
+ credentials: Dict of credential key→value.
+
+ Returns:
+ Tuple of (is_valid, list_of_error_messages).
+ """
+ fields = get_credential_fields(plugin_type)
+ errors: list[str] = []
+
+ for field in fields:
+ if field["required"]:
+ value = credentials.get(field["name"], "").strip()
+ if not value:
+ errors.append(f"'{field['label']}' is required")
+
+ return (len(errors) == 0, errors)
+
+
+async def validate_site_connection(
+ plugin_type: str, url: str, credentials: dict[str, str]
+) -> tuple[bool, str]:
+ """Test connectivity to a site using HTTP health check.
+
+ Args:
+ plugin_type: Plugin type name.
+ url: Site URL.
+ credentials: Plaintext credential dict.
+
+ Returns:
+ Tuple of (success, message). Message is "OK" on success or
+ a human-readable error description.
+ """
+ endpoint_info = _HEALTH_ENDPOINTS.get(plugin_type)
+ if endpoint_info is None:
+ return False, f"Unknown plugin type '{plugin_type}'"
+
+ check_url = url.rstrip("/") + endpoint_info["path"]
+ method = endpoint_info["method"]
+
+ # Build auth headers per plugin type
+ headers: dict[str, str] = {}
+ if plugin_type in ("wordpress", "wordpress_advanced"):
+ import base64
+
+ username = credentials.get("username", "")
+ app_password = credentials.get("app_password", "")
+ token = base64.b64encode(f"{username}:{app_password}".encode()).decode()
+ headers["Authorization"] = f"Basic {token}"
+ elif plugin_type == "woocommerce":
+ import base64
+
+ ck = credentials.get("consumer_key", "")
+ cs = credentials.get("consumer_secret", "")
+ token = base64.b64encode(f"{ck}:{cs}".encode()).decode()
+ headers["Authorization"] = f"Basic {token}"
+ elif plugin_type == "gitea":
+ headers["Authorization"] = f"token {credentials.get('token', '')}"
+ elif plugin_type == "n8n":
+ headers["X-N8N-API-KEY"] = credentials.get("api_key", "")
+ elif plugin_type == "supabase":
+ headers["apikey"] = credentials.get("service_role_key", "")
+ headers["Authorization"] = f"Bearer {credentials.get('service_role_key', '')}"
+ elif plugin_type == "appwrite":
+ headers["X-Appwrite-Project"] = credentials.get("project_id", "")
+ headers["X-Appwrite-Key"] = credentials.get("api_key", "")
+ elif plugin_type == "directus":
+ headers["Authorization"] = f"Bearer {credentials.get('token', '')}"
+ elif plugin_type == "openpanel":
+ # OpenPanel uses token exchange — just check that the URL is reachable
+ pass
+
+ try:
+ timeout = aiohttp.ClientTimeout(total=15)
+ async with aiohttp.ClientSession(timeout=timeout) as session:
+ if method == "POST" and plugin_type == "openpanel":
+ async with session.post(
+ check_url,
+ json={
+ "clientId": credentials.get("client_id", ""),
+ "clientSecret": credentials.get("client_secret", ""),
+ },
+ ) as resp:
+ status_code = resp.status
+ resp_text = await resp.text()
+ elif method == "POST":
+ async with session.post(check_url, headers=headers) as resp:
+ status_code = resp.status
+ resp_text = await resp.text()
+ else:
+ async with session.get(check_url, headers=headers) as resp:
+ status_code = resp.status
+ resp_text = await resp.text()
+
+ if status_code < 400:
+ return True, "OK"
+ elif status_code == 401:
+ return False, "Authentication failed — check credentials"
+ elif status_code == 403:
+ return False, "Access forbidden — check permissions or API may be disabled"
+ elif status_code == 404:
+ return False, f"Endpoint not found at {check_url} — check URL"
+ else:
+ return False, f"HTTP {status_code}: {resp_text[:200]}"
+
+ except aiohttp.ClientConnectorError:
+ return False, "Connection failed — check URL and ensure the site is reachable"
+ except TimeoutError:
+ return False, "Connection timed out (15s) — site may be slow or unreachable"
+ except aiohttp.InvalidURL:
+ return False, "Invalid URL protocol — use https:// or http://"
+ except Exception as e:
+ return False, f"Connection error: {type(e).__name__}: {e}"
+
+
+async def create_user_site(
+ user_id: str,
+ plugin_type: str,
+ alias: str,
+ url: str,
+ credentials: dict[str, str],
+ skip_validation: bool = False,
+) -> dict[str, Any]:
+ """Create a new site for a user.
+
+ Validates credentials, tests the connection, encrypts credentials,
+ and stores in the database.
+
+ Args:
+ user_id: Owner's UUID.
+ plugin_type: Plugin type name.
+ alias: User-chosen friendly name.
+ url: Site URL.
+ credentials: Plaintext credential dict.
+ skip_validation: If True, skip connection test (for testing).
+
+ Returns:
+ The created site dict (without decrypted credentials).
+
+ Raises:
+ ValueError: On validation errors (bad plugin type, missing fields,
+ alias taken, site limit reached, connection failed).
+ """
+ from core.database import get_database
+ from core.encryption import get_credential_encryption
+
+ # Validate plugin type
+ if plugin_type not in PLUGIN_CREDENTIAL_FIELDS:
+ raise ValueError(
+ f"Unknown plugin type '{plugin_type}'. "
+ f"Valid: {list(PLUGIN_CREDENTIAL_FIELDS.keys())}"
+ )
+
+ # Validate alias format
+ alias = alias.strip().lower()
+ if not alias or len(alias) < 2 or len(alias) > 50:
+ raise ValueError("Alias must be 2-50 characters")
+ if not alias.replace("-", "").replace("_", "").isalnum():
+ raise ValueError("Alias may only contain letters, numbers, hyphens, and underscores")
+
+ # Validate required credential fields
+ valid, errors = validate_credentials(plugin_type, credentials)
+ if not valid:
+ raise ValueError(f"Missing credentials: {', '.join(errors)}")
+
+ db = get_database()
+
+ # Check site limit
+ count = await db.count_sites_by_user(user_id)
+ if count >= MAX_SITES_PER_USER:
+ raise ValueError(f"Site limit reached ({MAX_SITES_PER_USER} sites per user)")
+
+ # Check alias uniqueness (DB constraint will also catch this)
+ existing = await db.get_site_by_alias(user_id, alias)
+ if existing is not None:
+ raise ValueError(f"Alias '{alias}' is already in use")
+
+ # Test connection
+ status = "active"
+ status_msg = "Connection verified"
+ if not skip_validation:
+ ok, msg = await validate_site_connection(plugin_type, url, credentials)
+ if not ok:
+ raise ValueError(f"Connection test failed: {msg}")
+
+ # Encrypt credentials
+ encryptor = get_credential_encryption()
+ # We need the site_id for encryption, but we don't have it yet.
+ # Use a pre-generated UUID as the site_id.
+ import uuid
+
+ site_id = str(uuid.uuid4())
+ encrypted = encryptor.encrypt_credentials(credentials, site_id)
+
+ # Store in database — we bypass db.create_site() to use our pre-generated ID
+ from core.database import _utc_now
+
+ now = _utc_now()
+ await db.execute(
+ "INSERT INTO sites (id, user_id, plugin_type, alias, url, credentials, "
+ "status, status_msg, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
+ (site_id, user_id, plugin_type, alias, url, encrypted, status, status_msg, now),
+ )
+
+ # Return the created site (without credentials blob)
+ site = await db.get_site(site_id, user_id)
+ if site is None:
+ raise RuntimeError(f"Failed to read back created site {site_id}")
+
+ result = dict(site)
+ result.pop("credentials", None)
+ logger.info("Created site %s (%s) for user %s", alias, plugin_type, user_id)
+ return result
+
+
+async def get_user_sites(user_id: str) -> list[dict[str, Any]]:
+ """Get all sites for a user (without credentials).
+
+ Args:
+ user_id: Owner's UUID.
+
+ Returns:
+ List of site dicts.
+ """
+ from core.database import get_database
+
+ db = get_database()
+ sites = await db.get_sites_by_user(user_id)
+ # Strip credentials blob from response
+ return [{k: v for k, v in site.items() if k != "credentials"} for site in sites]
+
+
+async def get_user_site(site_id: str, user_id: str) -> dict[str, Any] | None:
+ """Get a single site (without credentials).
+
+ Args:
+ site_id: Site UUID.
+ user_id: Owner's UUID.
+
+ Returns:
+ Site dict or None.
+ """
+ from core.database import get_database
+
+ db = get_database()
+ site = await db.get_site(site_id, user_id)
+ if site is None:
+ return None
+ result = dict(site)
+ result.pop("credentials", None)
+ return result
+
+
+async def delete_user_site(site_id: str, user_id: str) -> bool:
+ """Delete a user's site.
+
+ Args:
+ site_id: Site UUID.
+ user_id: Owner's UUID.
+
+ Returns:
+ True if deleted, False if not found.
+ """
+ from core.database import get_database
+
+ db = get_database()
+ deleted = await db.delete_site(site_id, user_id)
+ if deleted:
+ logger.info("Deleted site %s for user %s", site_id, user_id)
+ return deleted
+
+
+async def test_site_connection(site_id: str, user_id: str) -> tuple[bool, str]:
+ """Test connectivity to an existing site.
+
+ Decrypts credentials from the database, runs the health check,
+ and updates the site status.
+
+ Args:
+ site_id: Site UUID.
+ user_id: Owner's UUID.
+
+ Returns:
+ Tuple of (success, message).
+
+ Raises:
+ ValueError: If site not found.
+ """
+ from core.database import get_database
+ from core.encryption import get_credential_encryption
+
+ db = get_database()
+ site = await db.get_site(site_id, user_id)
+ if site is None:
+ raise ValueError("Site not found")
+
+ encryptor = get_credential_encryption()
+ credentials = encryptor.decrypt_credentials(site["credentials"], site_id)
+
+ ok, msg = await validate_site_connection(site["plugin_type"], site["url"], credentials)
+
+ # Update status
+ new_status = "active" if ok else "error"
+ await db.update_site_status(site_id, new_status, msg, user_id=user_id)
+
+ return ok, msg
diff --git a/core/site_manager.py b/core/site_manager.py
index 83afef4..4502d78 100644
--- a/core/site_manager.py
+++ b/core/site_manager.py
@@ -50,6 +50,7 @@ class SiteConfig(BaseModel):
site_id: str = Field(..., description="Unique site identifier")
plugin_type: str = Field(..., description="Plugin type (wordpress, gitea, etc)")
alias: str | None = Field(None, description="Friendly alias for the site")
+ user_id: str | None = Field(None, description="Owner user ID for the site")
# Common config fields (plugins may require additional fields)
url: str | None = Field(None, description="Site URL")
@@ -452,6 +453,7 @@ class SiteManager:
"site_id": config.site_id,
"alias": config.alias,
"full_id": config.get_full_id(),
+ "user_id": config.user_id,
}
)
diff --git a/core/site_registry.py b/core/site_registry.py
deleted file mode 100644
index 262eefa..0000000
--- a/core/site_registry.py
+++ /dev/null
@@ -1,371 +0,0 @@
-"""
-Site Registry
-
-Manages site configurations discovered from environment variables.
-Supports multi-site setups with aliases and dynamic discovery.
-"""
-
-import logging
-import os
-import re
-from typing import Any
-
-logger = logging.getLogger(__name__)
-
-
-class SiteInfo:
- """Information about a single site."""
-
- def __init__(
- self, plugin_type: str, site_id: str, config: dict[str, Any], alias: str | None = None
- ):
- """
- Initialize site information.
-
- Args:
- plugin_type: Type of plugin (e.g., 'wordpress')
- site_id: Site identifier (e.g., 'site1')
- config: Site configuration from environment
- alias: Optional friendly alias for the site
- """
- self.plugin_type = plugin_type
- self.site_id = site_id
- self.config = config
- self.alias = alias or site_id
-
- def get_full_id(self) -> str:
- """Get full site identifier: plugin_type_site_id"""
- return f"{self.plugin_type}_{self.site_id}"
-
- def get_display_name(self) -> str:
- """Get display name (alias if available, otherwise site_id)"""
- return self.alias
-
- def to_dict(self) -> dict[str, Any]:
- """Convert to dictionary for serialization."""
- return {
- "plugin_type": self.plugin_type,
- "site_id": self.site_id,
- "alias": self.alias,
- "full_id": self.get_full_id(),
- "config_keys": list(self.config.keys()),
- }
-
-
-class SiteRegistry:
- """
- Registry for managing site configurations across plugin types.
-
- Discovers sites from environment variables:
- - {PLUGIN_TYPE}_{SITE_ID}_{CONFIG_KEY}
- - {PLUGIN_TYPE}_{SITE_ID}_ALIAS (optional)
-
- Example:
- WORDPRESS_SITE1_URL=https://example.com
- WORDPRESS_SITE1_USERNAME=admin
- WORDPRESS_SITE1_APP_PASSWORD=xxxx
- WORDPRESS_SITE2_URL=https://myblog.com
- WORDPRESS_SITE2_ALIAS=myblog
- """
-
- def __init__(self):
- """Initialize site registry."""
- self.sites: dict[str, SiteInfo] = {} # full_id -> SiteInfo
- self.aliases: dict[str, str] = {} # alias -> full_id
- self.alias_conflicts: dict[str, list[str]] = {} # alias -> [full_ids that wanted it]
- self.logger = logging.getLogger("SiteRegistry")
-
- def discover_sites(self, plugin_types: list[str]) -> None:
- """
- Discover sites from environment variables.
-
- Args:
- plugin_types: List of plugin types to discover (e.g., ['wordpress'])
- """
- self.logger.info("Starting site discovery...")
-
- for plugin_type in plugin_types:
- self._discover_plugin_sites(plugin_type)
-
- self.logger.info(
- f"Discovery complete. Found {len(self.sites)} sites "
- f"with {len(self.aliases)} aliases."
- )
-
- # Log alias conflicts if any
- if self.alias_conflicts:
- self.logger.info("Duplicate alias conflicts detected:")
- for alias, full_ids in self.alias_conflicts.items():
- winner = self.aliases.get(alias)
- losers = [fid for fid in full_ids if fid != winner]
- self.logger.info(f" Alias '{alias}': {winner} (winner), {losers} (using full_id)")
-
- # Reserved words that should NOT be interpreted as site IDs
- RESERVED_SITE_WORDS = {
- "limit",
- "rate",
- "config",
- "debug",
- "log",
- "level",
- "mode",
- "timeout",
- "retry",
- "max",
- "min",
- "default",
- "global",
- "enabled",
- "disabled",
- "host",
- "port",
- "path",
- "key",
- "secret",
- "token",
- "advanced",
- "basic",
- "simple",
- "pro",
- "premium",
- "standard",
- }
-
- def _discover_plugin_sites(self, plugin_type: str) -> None:
- """
- Discover all sites for a specific plugin type.
-
- Args:
- plugin_type: Type of plugin (e.g., 'wordpress')
- """
- prefix = plugin_type.upper() + "_"
-
- # Find all site IDs for this plugin type
- site_ids = set()
- env_pattern = re.compile(f"^{prefix}([A-Z0-9_]+?)_(.+)$")
-
- for env_key in os.environ.keys():
- match = env_pattern.match(env_key)
- if match:
- site_id = match.group(1).lower()
- # Skip reserved words that are not real site IDs
- if site_id not in self.RESERVED_SITE_WORDS:
- site_ids.add(site_id)
-
- # Create SiteInfo for each discovered site
- for site_id in site_ids:
- try:
- config = self._load_site_config(plugin_type, site_id)
- if config:
- alias = config.pop("alias", None) # Extract alias if present
- site_info = SiteInfo(plugin_type, site_id, config, alias)
-
- full_id = site_info.get_full_id()
- self.sites[full_id] = site_info
-
- # Register alias with duplicate detection
- if alias:
- self._register_alias_safe(alias, full_id)
- # Also register with prefix
- prefixed_alias = f"{plugin_type}_{alias}"
- self._register_alias_safe(prefixed_alias, full_id)
-
- # Always register site_id as alias too (with safe check)
- self._register_alias_safe(site_id, full_id)
- self.aliases[full_id] = (
- full_id # full_id can reference itself (no conflict possible)
- )
-
- # Log with alias status
- effective_alias = self._get_effective_alias(alias, full_id)
- self.logger.info(f"Discovered site: {full_id} " f"(path: {effective_alias})")
- except Exception as e:
- self.logger.error(
- f"Failed to load {plugin_type} site '{site_id}': {e}", exc_info=True
- )
-
- def _register_alias_safe(self, alias: str, full_id: str) -> bool:
- """
- Register an alias with duplicate detection.
-
- Args:
- alias: The alias to register
- full_id: The full_id to map the alias to
-
- Returns:
- True if alias was registered, False if it was a duplicate
- """
- if alias in self.aliases:
- existing_full_id = self.aliases[alias]
- if existing_full_id != full_id:
- # Duplicate alias detected
- if alias not in self.alias_conflicts:
- self.alias_conflicts[alias] = [existing_full_id]
- self.alias_conflicts[alias].append(full_id)
- self.logger.info(
- f"Duplicate alias '{alias}': {full_id} conflicts with {existing_full_id}. "
- f"{full_id} will use full_id for endpoint path."
- )
- return False
- else:
- self.aliases[alias] = full_id
- return True
-
- def _get_effective_alias(self, alias: str | None, full_id: str) -> str:
- """
- Get the effective path suffix for a site.
-
- If the alias was taken by another site, returns full_id.
- Otherwise returns the alias (or full_id if no alias).
-
- Args:
- alias: The desired alias
- full_id: The full site ID
-
- Returns:
- The effective path suffix
- """
- if alias:
- # Check if this site owns the alias
- if self.aliases.get(alias) == full_id:
- return alias
- else:
- # Alias was taken, use full_id
- return full_id
- return full_id
-
- def get_alias_conflicts(self) -> dict[str, list[str]]:
- """
- Get all alias conflicts.
-
- Returns:
- Dict mapping conflicted aliases to list of full_ids that wanted them
- """
- return self.alias_conflicts.copy()
-
- def get_effective_path_suffix(self, full_id: str) -> str:
- """
- Get the effective path suffix for a site's endpoint.
-
- Uses alias if available and not conflicted, otherwise full_id.
-
- Args:
- full_id: The full site ID
-
- Returns:
- Path suffix to use in endpoint URL
- """
- site_info = self.sites.get(full_id)
- if not site_info:
- return full_id
-
- alias = site_info.alias
- return self._get_effective_alias(alias, full_id)
-
- def _load_site_config(self, plugin_type: str, site_id: str) -> dict[str, Any] | None:
- """
- Load configuration for a site from environment.
-
- Args:
- plugin_type: Plugin type
- site_id: Site ID
-
- Returns:
- Dict with configuration or None if incomplete
- """
- prefix = f"{plugin_type.upper()}_{site_id.upper()}_"
- config = {}
-
- # Collect all config keys for this site
- for env_key, env_value in os.environ.items():
- if env_key.startswith(prefix):
- # Extract config key (everything after prefix)
- config_key = env_key[len(prefix) :].lower()
- config[config_key] = env_value
-
- if not config:
- return None
-
- self.logger.debug(f"Loaded config for {plugin_type}/{site_id}: {list(config.keys())}")
- return config
-
- def get_site(self, plugin_type: str, site_identifier: str) -> SiteInfo | None:
- """
- Get site information by ID or alias.
-
- Args:
- plugin_type: Plugin type (e.g., 'wordpress')
- site_identifier: Site ID, alias, or full_id
-
- Returns:
- SiteInfo if found, None otherwise
- """
- # Try direct lookup first
- full_id = f"{plugin_type}_{site_identifier}"
- if full_id in self.sites:
- return self.sites[full_id]
-
- # Try alias lookup
- if site_identifier in self.aliases:
- resolved_full_id = self.aliases[site_identifier]
- return self.sites.get(resolved_full_id)
-
- # Try prefixed alias
- prefixed = f"{plugin_type}_{site_identifier}"
- if prefixed in self.aliases:
- resolved_full_id = self.aliases[prefixed]
- return self.sites.get(resolved_full_id)
-
- return None
-
- def get_sites_by_type(self, plugin_type: str) -> list[SiteInfo]:
- """
- Get all sites of a specific plugin type.
-
- Args:
- plugin_type: Plugin type to filter by
-
- Returns:
- List of SiteInfo objects
- """
- return [
- site_info for site_info in self.sites.values() if site_info.plugin_type == plugin_type
- ]
-
- def list_all_sites(self) -> list[dict[str, Any]]:
- """
- List all discovered sites.
-
- Returns:
- List of site info dictionaries
- """
- return [site_info.to_dict() for site_info in self.sites.values()]
-
- def get_site_options(self, plugin_type: str) -> list[str]:
- """
- Get available site options for a plugin type (for schema enum).
-
- Args:
- plugin_type: Plugin type
-
- Returns:
- List of valid site identifiers (IDs and aliases)
- """
- options = set()
- for site_info in self.get_sites_by_type(plugin_type):
- options.add(site_info.site_id)
- if site_info.alias and site_info.alias != site_info.site_id:
- options.add(site_info.alias)
- return sorted(options)
-
-
-# Global site registry instance
-_site_registry: SiteRegistry | None = None
-
-
-def get_site_registry() -> SiteRegistry:
- """Get the global site registry instance."""
- global _site_registry
- if _site_registry is None:
- _site_registry = SiteRegistry()
- return _site_registry
diff --git a/core/templates/dashboard/api-keys/list.html b/core/templates/dashboard/api-keys/list.html
index 2d37010..7f74556 100644
--- a/core/templates/dashboard/api-keys/list.html
+++ b/core/templates/dashboard/api-keys/list.html
@@ -8,7 +8,7 @@
-
+
{% if lang == 'fa' %}مدیریت کلیدهای API برای دسترسی به ابزارها{% else %}Manage API keys for tool access{% endif %}
@@ -24,16 +24,16 @@
-