forked from HomeLab/unraid-mcp
1751bc2984
Addresses issues found by 4 parallel review agents (code-reviewer,
silent-failure-hunter, type-design-analyzer, pr-test-analyzer).
Source fixes:
- core/utils.py: add public safe_display_url() (moved from tools/health.py)
- core/client.py: rename _redact_sensitive → redact_sensitive (public API)
- core/types.py: add SubscriptionData.__post_init__ for tz-aware datetime
enforcement; remove 6 unused type aliases (SystemHealth, APIResponse, etc.)
- subscriptions/manager.py: add exc_info=True to both except-Exception blocks;
add except ValueError break-on-config-error before retry loop; import
redact_sensitive by new public name
- subscriptions/resources.py: re-raise in autostart_subscriptions() so
ensure_subscriptions_started() doesn't permanently set _subscriptions_started
- subscriptions/diagnostics.py: except ToolError: raise before broad except;
use safe_display_url() instead of raw URL slice
- tools/health.py: move _safe_display_url to core/utils; add exc_info=True;
raise ToolError (not return dict) on ImportError
- tools/info.py: use get_args(INFO_ACTIONS) instead of INFO_ACTIONS.__args__
- tools/{array,docker,keys,notifications,rclone,storage,virtualization}.py:
add Literal-vs-ALL_ACTIONS sync check at import time
Test fixes:
- test_health.py: import safe_display_url from core.utils; update
test_diagnose_import_error_internal to expect ToolError (not error dict)
- test_storage.py: add 3 safe_get tests for zero/False/empty-string values
- test_subscription_manager.py: add TestCapLogContentSingleMassiveLine (2 tests)
- test_client.py: rename _redact_sensitive → redact_sensitive; add tests for
new sensitive keys and is_cacheable explicit-keyword form
115 lines
3.8 KiB
Python
115 lines
3.8 KiB
Python
"""Configuration management for Unraid MCP Server.
|
|
|
|
This module handles loading environment variables from multiple .env locations
|
|
and provides all configuration constants used throughout the application.
|
|
"""
|
|
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from dotenv import load_dotenv
|
|
|
|
from ..version import VERSION as APP_VERSION
|
|
|
|
|
|
# Get the script directory (config module location)
|
|
SCRIPT_DIR = Path(__file__).parent # /home/user/code/unraid-mcp/unraid_mcp/config/
|
|
UNRAID_MCP_DIR = SCRIPT_DIR.parent # /home/user/code/unraid-mcp/unraid_mcp/
|
|
PROJECT_ROOT = UNRAID_MCP_DIR.parent # /home/user/code/unraid-mcp/
|
|
|
|
# Load environment variables from .env file
|
|
# In container: First try /app/.env.local (mounted), then project root .env
|
|
dotenv_paths = [
|
|
Path("/app/.env.local"), # Container mount point
|
|
PROJECT_ROOT / ".env.local", # Project root .env.local
|
|
PROJECT_ROOT / ".env", # Project root .env
|
|
UNRAID_MCP_DIR / ".env", # Local .env in unraid_mcp/
|
|
]
|
|
|
|
for dotenv_path in dotenv_paths:
|
|
if dotenv_path.exists():
|
|
load_dotenv(dotenv_path=dotenv_path)
|
|
break
|
|
|
|
# Core API Configuration
|
|
UNRAID_API_URL = os.getenv("UNRAID_API_URL")
|
|
UNRAID_API_KEY = os.getenv("UNRAID_API_KEY")
|
|
|
|
# Server Configuration
|
|
UNRAID_MCP_PORT = int(os.getenv("UNRAID_MCP_PORT", "6970"))
|
|
UNRAID_MCP_HOST = os.getenv("UNRAID_MCP_HOST", "0.0.0.0") # noqa: S104 — intentional for Docker
|
|
UNRAID_MCP_TRANSPORT = os.getenv("UNRAID_MCP_TRANSPORT", "streamable-http").lower()
|
|
|
|
# SSL Configuration
|
|
raw_verify_ssl = os.getenv("UNRAID_VERIFY_SSL", "true").lower()
|
|
if raw_verify_ssl in ["false", "0", "no"]:
|
|
UNRAID_VERIFY_SSL: bool | str = False
|
|
elif raw_verify_ssl in ["true", "1", "yes"]:
|
|
UNRAID_VERIFY_SSL = True
|
|
else: # Path to CA bundle
|
|
UNRAID_VERIFY_SSL = raw_verify_ssl
|
|
|
|
# Logging Configuration
|
|
LOG_LEVEL_STR = os.getenv("UNRAID_MCP_LOG_LEVEL", "INFO").upper()
|
|
LOG_FILE_NAME = os.getenv("UNRAID_MCP_LOG_FILE", "unraid-mcp.log")
|
|
# Use /.dockerenv as the container indicator for robust Docker detection.
|
|
IS_DOCKER = Path("/.dockerenv").exists()
|
|
LOGS_DIR = Path("/app/logs") if IS_DOCKER else PROJECT_ROOT / "logs"
|
|
LOG_FILE_PATH = LOGS_DIR / LOG_FILE_NAME
|
|
|
|
# Ensure logs directory exists; if creation fails, fall back to /tmp.
|
|
try:
|
|
LOGS_DIR.mkdir(parents=True, exist_ok=True)
|
|
except OSError:
|
|
LOGS_DIR = PROJECT_ROOT / ".cache" / "logs"
|
|
LOGS_DIR.mkdir(parents=True, exist_ok=True)
|
|
LOG_FILE_PATH = LOGS_DIR / LOG_FILE_NAME
|
|
|
|
# HTTP Client Configuration
|
|
TIMEOUT_CONFIG = {
|
|
"default": 30,
|
|
"disk_operations": 90, # Longer timeout for SMART data queries
|
|
}
|
|
|
|
|
|
def validate_required_config() -> tuple[bool, list[str]]:
|
|
"""Validate that required configuration is present.
|
|
|
|
Returns:
|
|
bool: True if all required config is present, False otherwise.
|
|
"""
|
|
required_vars = [("UNRAID_API_URL", UNRAID_API_URL), ("UNRAID_API_KEY", UNRAID_API_KEY)]
|
|
|
|
missing = []
|
|
for name, value in required_vars:
|
|
if not value:
|
|
missing.append(name)
|
|
|
|
return len(missing) == 0, missing
|
|
|
|
|
|
def get_config_summary() -> dict[str, Any]:
|
|
"""Get a summary of current configuration (safe for logging).
|
|
|
|
Returns:
|
|
dict: Configuration summary with sensitive data redacted.
|
|
"""
|
|
is_valid, missing = validate_required_config()
|
|
|
|
return {
|
|
"api_url_configured": bool(UNRAID_API_URL),
|
|
"api_url_preview": UNRAID_API_URL[:20] + "..." if UNRAID_API_URL else None,
|
|
"api_key_configured": bool(UNRAID_API_KEY),
|
|
"server_host": UNRAID_MCP_HOST,
|
|
"server_port": UNRAID_MCP_PORT,
|
|
"transport": UNRAID_MCP_TRANSPORT,
|
|
"ssl_verify": UNRAID_VERIFY_SSL,
|
|
"log_level": LOG_LEVEL_STR,
|
|
"log_file": str(LOG_FILE_PATH),
|
|
"config_valid": is_valid,
|
|
"missing_config": missing if not is_valid else None,
|
|
}
|
|
# Re-export application version from a single source of truth.
|
|
VERSION = APP_VERSION
|