AI-written code trusted a user-supplied `ha_url` in OAuth flows without validation, letting unauthenticated attackers probe internal networks via server-side requests.

+for Home Assistant MCP Server. Users authenticate via a consent form where they+provide their Home Assistant URL and Long-Lived Access Token (LLAT).+"""++import logging+import secrets+import time+from typing import Any+from urllib.parse import urlencode++import httpx+from mcp.server.auth.provider import (+    AccessToken,+    AuthorizationCode,+    AuthorizationParams,+    AuthorizeError,+    RefreshToken,+    TokenError,+    construct_redirect_uri,+)+from mcp.shared.auth import OAuthClientInformationFull, OAuthToken+from pydantic import AnyHttpUrl+from starlette.requests import Request+from starlette.responses import HTMLResponse, RedirectResponse, Response+from starlette.routing import Route++from fastmcp.server.auth.auth import (+    ClientRegistrationOptions,+    OAuthProvider,+    RevocationOptions,+)++from .consent_form import create_consent_html, create_error_html++logger = logging.getLogger(__name__)++# Token expiration times+AUTH_CODE_EXPIRY_SECONDS = 5 * 60  # 5 minutes+ACCESS_TOKEN_EXPIRY_SECONDS = 60 * 60  # 1 hour+REFRESH_TOKEN_EXPIRY_SECONDS = 7 * 24 * 60 * 60  # 7 days+++class HomeAssistantCredentials:+    """Stores Home Assistant credentials for a client."""++    def __init__(self, ha_url: str, ha_token: str):+        self.ha_url = ha_url.rstrip("/")+        self.ha_token = ha_token+        self.validated_at = time.time()++    def to_dict(self) -> dict[str, Any]:+        """Convert to dictionary for storage."""+        return {+            "ha_url": self.ha_url,+            "ha_token": self.ha_token,+            "validated_at": self.validated_at,+        }+++class HomeAssistantOAuthProvider(OAuthProvider):+    """+    OAuth 2.1 provider for Home Assistant MCP Server.++    This provider implements the full OAuth 2.1 flow with:+    - Dynamic Client Registration (DCR)+    - PKCE support+    - Custom consent form for collecting HA credentials+    - Token management with refresh token support++    The consent form collects the user's Home Assistant URL and+    Long-Lived Access Token, validates them, and stores them+    securely for subsequent API calls.+    """++    def __init__(+        self,+        base_url: AnyHttpUrl | str,+        issuer_url: AnyHttpUrl | str | None = None,+        service_documentation_url: AnyHttpUrl | str | None = None,+        client_registration_options: ClientRegistrationOptions | None = None,+        revocation_options: RevocationOptions | None = None,+        required_scopes: list[str] | None = None,+    ):+        """+        Initialize the Home Assistant OAuth provider.

+                _CONFIG_ERROR_MESSAGE.format(missing_vars="\n".join(missing_vars)),+                file=sys.stderr,+            )             sys.exit(1)      # For other validation errors, show the original error with guidance-    print(f"""+    print(+        f""" ==============================================================================                     Home Assistant MCP Server - Configuration Error ==============================================================================@@ -82,18 +84,22 @@ def _handle_config_error(error: Exception) -> None:   https://github.com/homeassistant-ai/ha-mcp#-installation  ==============================================================================-""", file=sys.stderr)+""",+        file=sys.stderr,+    )     sys.exit(1)   def _create_server():     """Create server instance (deferred to avoid import during smoke test)."""     try:         from ha_mcp.server import HomeAssistantSmartMCPServer  # type: ignore[import-not-found]+         return HomeAssistantSmartMCPServer()     except Exception as e:         # Check if this is a pydantic validation error (missing env vars)         from pydantic import ValidationError+         if isinstance(e, ValidationError):             _handle_config_error(e)         raise@@ -123,6 +129,7 @@ def _get_server(): # This is accessed when the module is imported, so we need deferred creation class _DeferredMCP:     """Wrapper that defers MCP creation until actually accessed."""+     def __getattr__(self, name: str) -> Any:         return getattr(_get_mcp(), name) @@ -142,6 +149,7 @@ async def _cleanup_resources() -> None:     # Close WebSocket listener service if running     try:         from ha_mcp.client.websocket_listener import stop_websocket_listener+         await stop_websocket_listener()         logger.debug("WebSocket listener stopped")     except Exception as e:@@ -150,6 +158,7 @@ async def _cleanup_resources() -> None:     # Close WebSocket manager connections     try:         from ha_mcp.client.websocket_client import websocket_manager+         await websocket_manager.disconnect()         logger.debug("WebSocket manager disconnected")     except Exception as e:@@ -241,8 +250,7 @@ async def _run_with_graceful_shutdown() -> None:         # Clean up resources with timeout         try:             await asyncio.wait_for(-                _cleanup_resources(),-                timeout=SHUTDOWN_TIMEOUT_SECONDS+                _cleanup_resources(), timeout=SHUTDOWN_TIMEOUT_SECONDS             )         except TimeoutError:             logger.warning("Resource cleanup timed out")@@ -269,6 +277,7 @@ def main() -> None:     # Check for smoke test flag     if "--smoke-test" in sys.argv:         from ha_mcp.smoke_test import main as smoke_test_main+         sys.exit(smoke_test_main())      # Configure logging before server creation@@ -363,8 +372,7 @@ async def _run_http_with_graceful_shutdown(         # Clean up resources with timeout         try:             await asyncio.wait_for(

+Home Assistant URL and Long-Lived Access Token (LLAT).+"""+++def create_consent_html(+    client_id: str,+    client_name: str | None,+    redirect_uri: str,+    state: str,+    scopes: list[str],+    error_message: str | None = None,+) -> str:+    """+    Generate HTML consent form for Home Assistant authentication.++    Args:+        client_id: OAuth client ID+        client_name: Human-readable client name+        redirect_uri: OAuth redirect URI+        state: OAuth state parameter+        scopes: Requested OAuth scopes+        error_message: Optional error message to display++    Returns:+        HTML string for the consent form+    """+    display_name = client_name or client_id+    scopes_display = ", ".join(scopes) if scopes else "full access"++    error_html = ""+    if error_message:+        error_html = f"""+        <div class="error-message">+            <strong>Error:</strong> {error_message}+        </div>+        """++    return f"""+<!DOCTYPE html>+<html lang="en">+<head>+    <meta charset="UTF-8">+    <meta name="viewport" content="width=device-width, initial-scale=1.0">+    <title>Connect to Home Assistant</title>+    <style>+        :root {{+            --primary-color: #03a9f4;+            --primary-hover: #0288d1;+            --error-color: #f44336;+            --error-bg: #ffebee;+            --success-color: #4caf50;+            --text-color: #212121;+            --text-secondary: #757575;+            --border-color: #e0e0e0;+            --bg-color: #fafafa;+            --card-bg: #ffffff;+        }}++        @media (prefers-color-scheme: dark) {{+            :root {{+                --primary-color: #29b6f6;+                --primary-hover: #4fc3f7;+                --error-color: #ef5350;+                --error-bg: #3e2723;+                --success-color: #66bb6a;+                --text-color: #e0e0e0;+                --text-secondary: #9e9e9e;+                --border-color: #424242;+                --bg-color: #121212;+                --card-bg: #1e1e1e;+            }}+        }}++        * {{+            box-sizing: border-box;+            margin: 0;+            padding: 0;+        }}++        body {{+            font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, Oxygen, Ubuntu, Cantarell, sans-serif;+            background-color: var(--bg-color);+            color: var(--text-color);+            min-height: 100vh;+            display: flex;

@@ -32,10 +31,13 @@ class OAuthProxyClient:      This class is necessary because tools capture a reference to the client at registration time.     The proxy allows us to inject different credentials per-request based on OAuth token claims.++    The Home Assistant URL is fixed server-side (HOMEASSISTANT_URL env var).+    Only the access token varies per-user (from OAuth consent form).     """ -    def __init__(self, auth_provider: "HomeAssistantOAuthProvider") -> None:-        self._auth_provider = auth_provider+    def __init__(self, ha_url: str) -> None:+        self._ha_url = ha_url.rstrip("/")         self._oauth_clients: dict[str, HomeAssistantClient] = {}         self._lock = threading.Lock() @@ -52,26 +54,25 @@ def _get_oauth_client(self) -> "HomeAssistantClient":             logger.warning("No access token in context")             raise RuntimeError("No OAuth token in request context") -        # Extract HA credentials from token claims+        # Extract HA token from claims (URL is server-side config)         claims = token.claims -        if not claims or "ha_url" not in claims or "ha_token" not in claims:+        if not claims or "ha_token" not in claims:             logger.error(f"OAuth token missing HA credentials. Keys present: {list(claims.keys()) if claims else []}")             raise RuntimeError("No Home Assistant credentials in OAuth token claims") -        ha_url = claims["ha_url"]         ha_token = claims["ha_token"] -        # Hash credentials for cache key to avoid raw tokens appearing in dict keys-        client_key = hashlib.sha256(f"{ha_url}:{ha_token}".encode()).hexdigest()+        # Hash token for cache key to avoid raw tokens appearing in dict keys+        client_key = hashlib.sha256(ha_token.encode()).hexdigest()          with self._lock:             if client_key not in self._oauth_clients:                 self._oauth_clients[client_key] = HomeAssistantClient(-                    base_url=ha_url,+                    base_url=self._ha_url,                     token=ha_token,                 )-                logger.info(f"Created OAuth client for {ha_url}")+                logger.info(f"Created OAuth client for {self._ha_url}")              return self._oauth_clients[client_key] @@ -610,18 +611,19 @@ def main_sse() -> None: def main_oauth() -> None:     """Run server with OAuth 2.1 authentication over HTTP. -    This mode enables zero-config authentication for MCP clients like Claude.ai.-    Users authenticate via a consent form where they enter their Home Assistant-    URL and Long-Lived Access Token.+    This mode enables per-user authentication for MCP clients like Claude.ai.+    Users authenticate via a consent form where they provide their+    Long-Lived Access Token.      Environment:+    - HOMEASSISTANT_URL (required): URL of the Home Assistant instance     - MCP_BASE_URL (required): Public URL where this server is accessible (e.g., https://your-tunnel.com)     - MCP_PORT (optional, default: 8086)     - MCP_SECRET_PATH (optional, default: "/mcp")     - LOG_LEVEL (optional, default: INFO) -    Note: HOMEASSISTANT_URL and HOMEASSISTANT_TOKEN are NOT required in this mode.-    They are collected via the OAuth consent form.+    Note: HOMEASSISTANT_TOKEN is NOT required in this mode.+    Per-user tokens are collected via the OAuth consent form.     """     # Configure logging for OAuth mode     log_level = os.getenv("LOG_LEVEL", "INFO").upper()@@ -633,21 +635,45 @@ def main_oauth() -> None:      port, path = _get_http_runtime(default_port=8086)     base_url = os.getenv("MCP_BASE_URL")+    ha_url = os.getenv("HOMEASSISTANT_URL") +    missing = []     if not base_url:

 import binascii@@ -15,7 +15,6 @@ from typing import Any from urllib.parse import urlencode -import httpx from fastmcp.server.auth.auth import (     AccessToken,  # FastMCP version has claims field     ClientRegistrationOptions,@@ -49,15 +48,13 @@ class HomeAssistantCredentials:     """Stores Home Assistant credentials for a client.""" -    def __init__(self, ha_url: str, ha_token: str):-        self.ha_url = ha_url.rstrip("/")+    def __init__(self, ha_token: str):         self.ha_token = ha_token         self.validated_at = time.time()      def to_dict(self) -> dict[str, Any]:         """Convert to dictionary for storage."""         return {-            "ha_url": self.ha_url,             "ha_token": self.ha_token,             "validated_at": self.validated_at,         }@@ -73,11 +70,11 @@ class HomeAssistantOAuthProvider(OAuthProvider):     - Custom consent form for collecting HA credentials     - Stateless access tokens (base64-encoded JSON) -    The consent form collects the user's Home Assistant URL and-    Long-Lived Access Token, validates them, and encodes them into-    stateless access tokens for subsequent API calls.+    The consent form collects the user's Long-Lived Access Token,+    which is encoded into stateless access tokens for subsequent API calls.+    The Home Assistant URL is configured server-side via HOMEASSISTANT_URL. -    Access tokens are base64-encoded JSON containing HA credentials.+    Access tokens are base64-encoded JSON containing the HA token.     No encryption or signing - security comes from HTTPS transport     and the LLAT itself being the authorization boundary.     """@@ -140,28 +137,27 @@ def __init__(          logger.info(f"HomeAssistantOAuthProvider initialized with base_url={base_url}") -    def _encode_credentials(self, ha_url: str, ha_token: str) -> str:+    def _encode_credentials(self, ha_token: str) -> str:         """-        Encode HA credentials into a stateless access token.+        Encode HA token into a stateless access token. -        Tokens are base64-encoded JSON containing HA credentials.+        Tokens are base64-encoded JSON containing the HA token.         No encryption or signing - credentials are readable but transmitted over HTTPS.         The LLAT itself provides the security boundary.         """         payload = {-            "ha_url": ha_url,             "ha_token": ha_token,             "iat": int(time.time()),         }         json_str = json.dumps(payload)         encoded = urlsafe_b64encode(json_str.encode()).decode().rstrip("=")         return encoded -    def _decode_credentials(self, token: str) -> tuple[str, str] | None:+    def _decode_credentials(self, token: str) -> str | None:         """-        Decode access token to extract HA credentials.+        Decode access token to extract HA token. -        Returns (ha_url, ha_token) or None if invalid.+        Returns ha_token or None if invalid.         """         try:             # Add padding if needed@@ -172,11 +168,10 @@ def _decode_credentials(self, token: str) -> tuple[str, str] | None:             decoded = urlsafe_b64decode(token.encode()).decode()             payload = json.loads(decoded) -            ha_url = payload.get("ha_url")

+import html+from urllib.parse import urlparse+++def _extract_domain(redirect_uri: str) -> str:+    """Extract display domain from redirect URI."""+    try:+        parsed = urlparse(redirect_uri)+        return parsed.netloc or redirect_uri+    except (AttributeError, TypeError, ValueError):+        return redirect_uri+  def create_consent_html(     client_id: str,-    client_name: str | None,     redirect_uri: str,     state: str,-    scopes: list[str],+    txn_id: str,     error_message: str | None = None, ) -> str:     """     Generate HTML consent form for Home Assistant authentication.      Args:         client_id: OAuth client ID-        client_name: Human-readable client name-        redirect_uri: OAuth redirect URI+        redirect_uri: OAuth redirect URI (used to derive the display domain)         state: OAuth state parameter-        scopes: Requested OAuth scopes+        txn_id: Transaction ID for this authorization request         error_message: Optional error message to display      Returns:         HTML string for the consent form     """-    display_name = client_name or client_id-    scopes_display = ", ".join(scopes) if scopes else "full access"+    domain = _extract_domain(redirect_uri)+    safe_domain = html.escape(domain)+    safe_client_id = html.escape(client_id)+    safe_redirect_uri = html.escape(redirect_uri)+    safe_state = html.escape(state)+    safe_txn_id = html.escape(txn_id)      error_html = ""     if error_message:+        safe_error = html.escape(error_message)         error_html = f"""         <div class="error-message">-            <strong>Error:</strong> {error_message}+            <strong>Error:</strong> {safe_error}         </div>         """ @@ -52,7 +67,8 @@ def create_consent_html(             --primary-hover: #0288d1;             --error-color: #f44336;             --error-bg: #ffebee;-            --success-color: #4caf50;+            --warning-color: #ff9800;+            --warning-bg: #fff3e0;             --text-color: #212121;             --text-secondary: #757575;             --border-color: #e0e0e0;@@ -66,7 +82,8 @@ def create_consent_html(                 --primary-hover: #4fc3f7;                 --error-color: #ef5350;                 --error-bg: #3e2723;-                --success-color: #66bb6a;+                --warning-color: #ffb74d;+                --warning-bg: #2a1f0a;                 --text-color: #e0e0e0;                 --text-secondary: #9e9e9e;                 --border-color: #424242;@@ -123,28 +140,6 @@ def create_consent_html(             font-size: 14px;         }} -        .client-info {{