847 lines
36 KiB
Python
847 lines
36 KiB
Python
"""Legacy OAuth 2.1 authorization server for the HA-MCP Server webhook.
|
|
|
|
Ported from the webhook-proxy add-on's ``mcp_proxy/oauth.py`` ``OAuthProvider``
|
|
+ ``AuthorizeView`` + ``TokenView`` (the proven ``legacy`` mode). Self-hosted,
|
|
single-tenant authorization server with a static client_id/client_secret pair,
|
|
for MCP clients that need a credential to paste rather than HA core's native
|
|
OAuth (``ha_auth``) — currently just Google Gemini Spark, whose custom
|
|
connected apps use the Client ID Metadata Document pattern that HA core's
|
|
``/auth/authorize`` does not yet support for cross-origin redirect_uris
|
|
(home-assistant/core#176282).
|
|
|
|
Unlike the add-on, this module holds no reference to ``hass``, the webhook id,
|
|
or a public base URL: the discovery-document layer (RFC 8414 / RFC 9728) and
|
|
base-URL resolution already live in ``mcp_webhook.py``, shared with the
|
|
``ha_auth`` mode, and are extended there to be mode-aware. This module owns
|
|
only the OAuth-specific state (tokens, PKCE codes, client auth) and the two
|
|
root views the add-on's ``ha_auth`` mode never needed (``ha_auth`` mode has HA
|
|
core serve its own ``/auth/authorize`` + ``/auth/token``).
|
|
|
|
Tokens are opaque, HMAC-SHA256 signed (``body.sig``), and carry enough state
|
|
(``{kind, iat, exp, jti, cid}``) to validate without a server-side store, so
|
|
the integration survives HA restarts. ``cid`` (the client_id at issuance time)
|
|
means rotating the client_id revokes every outstanding token at the restart
|
|
that rebinds the root views with the new identity — until then the bound
|
|
provider keeps the old client_id and old tokens keep validating (see
|
|
:func:`bind_legacy_views`).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import binascii
|
|
import hashlib
|
|
import hmac
|
|
import ipaddress
|
|
import json
|
|
import logging
|
|
import re
|
|
import secrets
|
|
import time
|
|
from collections.abc import Callable
|
|
from html import escape
|
|
from typing import TYPE_CHECKING, TypedDict
|
|
from urllib.parse import unquote_plus, urlparse
|
|
|
|
from aiohttp import web
|
|
from homeassistant.components.http import HomeAssistantView
|
|
|
|
from .const import WEBHOOK_AUTH_LEGACY
|
|
|
|
if TYPE_CHECKING:
|
|
from homeassistant.core import HomeAssistant
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
# Registered at the HA ROOT rather than under a component-namespaced path, to
|
|
# match the add-on's legacy mode (see the ownership guard below). This was
|
|
# motivated by an MCP client that builds ``<host>/authorize`` from the resource
|
|
# host root instead of reading the ``authorization_endpoint`` metadata field —
|
|
# observed with Google Gemini Spark, which is what legacy mode exists to serve.
|
|
# claude.ai, by contrast, DOES honor the advertised ``authorization_endpoint``:
|
|
# issue #1969's none-mode auto-approve server advertises a component-scoped
|
|
# ``/authorize`` (see :mod:`oauth_autoapprove`) and claude.ai calls exactly that,
|
|
# proven live. So the root registration is for the metadata-ignoring clients, not
|
|
# a hard requirement for claude.ai.
|
|
AUTHORIZE_PATH = "/authorize"
|
|
TOKEN_PATH = "/token"
|
|
|
|
ACCESS_TOKEN_TTL = 60 * 60 # 1 hour
|
|
REFRESH_TOKEN_TTL = 30 * 24 * 60 * 60 # 30 days
|
|
AUTH_CODE_TTL = 5 * 60 # 5 minutes
|
|
TOKEN_KIND_ACCESS = "access"
|
|
TOKEN_KIND_REFRESH = "refresh"
|
|
|
|
# RFC 6749 §5.1: a /token response body carries the access/refresh credentials,
|
|
# so it MUST NOT be cached by any intermediary (reverse proxy, Nabu Casa, etc.).
|
|
_TOKEN_RESPONSE_HEADERS = {"Cache-Control": "no-store", "Pragma": "no-cache"}
|
|
|
|
# RFC 8252 §7.3: native/CLI OAuth clients (e.g. GitHub Copilot CLI) receive the
|
|
# authorization code on a loopback redirect, for which the spec explicitly
|
|
# permits a plain http scheme. Every non-loopback redirect must still be https.
|
|
_LOOPBACK_HOSTNAMES = frozenset({"localhost"})
|
|
|
|
# RFC 7636 §4.1: code_verifier is 43-128 chars from the unreserved URL set.
|
|
PKCE_VERIFIER_MIN = 43
|
|
PKCE_VERIFIER_MAX = 128
|
|
# SHA-256 → 32 bytes → 43 base64url chars (no padding).
|
|
PKCE_S256_CHALLENGE_LEN = 43
|
|
_PKCE_VERIFIER_RE = re.compile(r"^[A-Za-z0-9._~-]+$")
|
|
_PKCE_CHALLENGE_RE = re.compile(r"^[A-Za-z0-9_-]{43}$")
|
|
|
|
# Pending-code dict cap. An attacker spamming /authorize with valid params
|
|
# could grow the dict between the prune passes that run on each issuance.
|
|
# 1000 codes is well past anything legitimate (5-min TTL, single-tenant).
|
|
MAX_PENDING_CODES = 1000
|
|
|
|
_RESTART_HINT = (
|
|
"If this persists, fully restart Home Assistant (Settings -> System -> "
|
|
"Restart) -- Home Assistant cannot rebind the /authorize and /token "
|
|
"endpoints to new credentials without a full restart."
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Root-route ownership guard
|
|
# ---------------------------------------------------------------------------
|
|
#
|
|
# The webhook-proxy add-on's legacy mode (both the stable and dev flavors)
|
|
# ALSO claims the root /authorize + /token routes in the same HA instance —
|
|
# aiohttp lets the first-registered path win and silently shadows a later
|
|
# duplicate, and HA cannot unregister a bound view until it restarts. These
|
|
# two literals are therefore the SAME strings as
|
|
# ``mcp_proxy.OAUTH_ROUTE_OWNER_KEY`` / ``mcp_proxy.OAUTH_ROUTE_KEY_FINGERPRINT``
|
|
# (deliberately domain-neutral, not namespaced under this component's DOMAIN)
|
|
# so whichever integration binds first can be recognized by the other and the
|
|
# second registrant fails loud instead of being silently shadowed.
|
|
OAUTH_ROUTE_OWNER_KEY = "webhook_proxy_oauth_route_owner"
|
|
OAUTH_ROUTE_KEY_FINGERPRINT = "webhook_proxy_oauth_route_key_fingerprint"
|
|
|
|
# TOP-LEVEL hass.data key (NOT under DOMAIN, so it survives
|
|
# async_unload_entry's hass.data.pop(DOMAIN)) holding the LegacyOAuthProvider
|
|
# instance bound to the root views for this HA session. A reload reuses it
|
|
# when the credentials match; see bind_legacy_views.
|
|
_LEGACY_PROVIDER_KEY = "ha_mcp_tools_oauth_legacy_provider"
|
|
|
|
# TOP-LEVEL hass.data flag recording whether the currently-bound root views were
|
|
# registered MID-SESSION (hass already running → the route is not actually live
|
|
# until a full HA restart) rather than at boot (live immediately). It cannot be
|
|
# cleared without a real process restart, which wipes hass.data — so it stays
|
|
# True for the life of a session that late-bound, and is absent/False for a
|
|
# session that bound cleanly at boot. Read on every reuse so an unrelated reload
|
|
# before that restart does not falsely report "no restart needed" and clear the
|
|
# repair (the views are still not live). See bind_legacy_views.
|
|
_LEGACY_PENDING_RESTART_KEY = "ha_mcp_tools_oauth_legacy_pending_restart"
|
|
|
|
# This component's DOMAIN, duplicated here (rather than imported) to avoid a
|
|
# module-level dependency on const.DOMAIN for the ownership-marker value —
|
|
# the value written IS "ha_mcp_tools", checked against by name below.
|
|
_DOMAIN = "ha_mcp_tools"
|
|
|
|
|
|
class LegacyOAuthRouteConflict(RuntimeError):
|
|
"""Raised when another integration already owns the root OAuth routes."""
|
|
|
|
|
|
def _oauth_route_fingerprint(
|
|
client_id: str, client_secret: str, signing_key: bytes
|
|
) -> str:
|
|
"""Stable fingerprint of the OAuth identity bound to the root views."""
|
|
h = hashlib.sha256()
|
|
h.update(client_id.encode())
|
|
h.update(b"\0")
|
|
h.update(client_secret.encode())
|
|
h.update(b"\0")
|
|
h.update(signing_key)
|
|
return h.hexdigest()
|
|
|
|
|
|
def _normalize_signing_key(signing_key: bytes | str) -> bytes:
|
|
"""Accept either raw bytes or a hex string (how entry.data stores it,
|
|
since entry.data must be JSON-serializable)."""
|
|
return bytes.fromhex(signing_key) if isinstance(signing_key, str) else signing_key
|
|
|
|
|
|
def bind_legacy_views(
|
|
hass: HomeAssistant,
|
|
client_id: str,
|
|
client_secret: str,
|
|
signing_key: bytes | str,
|
|
) -> tuple[LegacyOAuthProvider, bool]:
|
|
"""Bind the root ``/authorize`` + ``/token`` views at most once per HA session.
|
|
|
|
Returns ``(provider, restart_needed)``. ``provider`` is the identity now
|
|
authoritative for the webhook's bearer gate — on a reload with unchanged
|
|
credentials this REUSES the already-bound provider (aiohttp cannot rebind
|
|
a view, so a fresh provider object would mint tokens the bound views never
|
|
issued and vice versa). ``restart_needed`` is True when:
|
|
|
|
* the currently-bound views were registered after HA finished starting (a
|
|
route registered while ``hass.is_running`` is not actually live until a
|
|
restart — this mirrors the add-on's ``oauth_restart_needed =
|
|
hass.is_running``) and that restart has not happened yet — this pending
|
|
state persists across config-entry reloads until a real HA restart wipes
|
|
``hass.data``, or
|
|
* the credentials changed since the currently-bound views were registered
|
|
(the bound views keep serving the OLD identity until a restart rebinds
|
|
them with the new one).
|
|
|
|
Raises :class:`LegacyOAuthRouteConflict` when the webhook-proxy add-on (or
|
|
its dev flavor) already owns the root routes in this HA instance.
|
|
"""
|
|
key_bytes = _normalize_signing_key(signing_key)
|
|
fingerprint = _oauth_route_fingerprint(client_id, client_secret, key_bytes)
|
|
|
|
owner = hass.data.get(OAUTH_ROUTE_OWNER_KEY)
|
|
if owner is not None and owner != _DOMAIN:
|
|
_LOGGER.error(
|
|
"HA-MCP: cannot enable legacy OAuth mode -- the Webhook Proxy "
|
|
"add-on ('%s') already owns the root /authorize and /token routes "
|
|
"in this Home Assistant instance, and Home Assistant cannot "
|
|
"release them until it restarts. Stop that add-on and restart "
|
|
"Home Assistant, then enable legacy mode again.",
|
|
owner,
|
|
)
|
|
raise LegacyOAuthRouteConflict(owner)
|
|
|
|
bound_provider = hass.data.get(_LEGACY_PROVIDER_KEY)
|
|
if owner == _DOMAIN and isinstance(bound_provider, LegacyOAuthProvider):
|
|
bound_fingerprint = hass.data.get(OAUTH_ROUTE_KEY_FINGERPRINT)
|
|
creds_changed = bound_fingerprint != fingerprint
|
|
if creds_changed:
|
|
_LOGGER.warning(
|
|
"HA-MCP: legacy OAuth credentials changed but the bound root "
|
|
"views still use the previous ones -- a Home Assistant "
|
|
"restart is required to activate the new credentials."
|
|
)
|
|
# Still restart-pending if the views were late-bound this session (flag
|
|
# persisted below) OR the credentials just changed. Reading the flag
|
|
# rather than recomputing keeps an unrelated reload from clearing a
|
|
# still-pending restart repair while the routes remain not-live.
|
|
pending = bool(hass.data.get(_LEGACY_PENDING_RESTART_KEY))
|
|
return bound_provider, pending or creds_changed
|
|
|
|
# First registration this HA session.
|
|
provider = LegacyOAuthProvider(
|
|
client_id=client_id,
|
|
client_secret=client_secret,
|
|
signing_key=key_bytes,
|
|
active_mode_getter=lambda: _live_auth_mode(hass),
|
|
)
|
|
hass.http.register_view(AuthorizeView(provider))
|
|
hass.http.register_view(TokenView(provider))
|
|
hass.data[OAUTH_ROUTE_OWNER_KEY] = _DOMAIN
|
|
hass.data[OAUTH_ROUTE_KEY_FINGERPRINT] = fingerprint
|
|
hass.data[_LEGACY_PROVIDER_KEY] = provider
|
|
# A first registration happening mid-session isn't live until a full HA
|
|
# restart; flag it. At HA boot (hass.is_running is still False while
|
|
# integrations are being set up) it binds cleanly. Persist the pending
|
|
# state so a later reload before that restart reuses it (see the reuse
|
|
# branch above) rather than recomputing "no restart needed".
|
|
pending_restart = hass.is_running
|
|
hass.data[_LEGACY_PENDING_RESTART_KEY] = pending_restart
|
|
return provider, pending_restart
|
|
|
|
|
|
def legacy_credentials_active(
|
|
hass: HomeAssistant,
|
|
client_id: str,
|
|
client_secret: str,
|
|
signing_key: bytes | str,
|
|
) -> bool:
|
|
"""Whether the bound root views currently serve exactly these credentials.
|
|
|
|
False while a credential rotation is pending a restart (the bound provider
|
|
keeps the previous identity until then — see :func:`bind_legacy_views`),
|
|
when another integration owns the routes, or when legacy OAuth was never
|
|
bound this session. Callers use this to withhold rotated credentials from
|
|
surfaces a still-valid old-identity token can read — the admin startup log
|
|
in particular, which is reachable through the server's own log tools
|
|
(review finding on #1880).
|
|
"""
|
|
if hass.data.get(OAUTH_ROUTE_OWNER_KEY) != _DOMAIN:
|
|
return False
|
|
bound = hass.data.get(OAUTH_ROUTE_KEY_FINGERPRINT)
|
|
if not isinstance(bound, str):
|
|
return False
|
|
current = _oauth_route_fingerprint(
|
|
client_id, client_secret, _normalize_signing_key(signing_key)
|
|
)
|
|
return hmac.compare_digest(bound, current)
|
|
|
|
|
|
def legacy_restart_pending(hass: HomeAssistant) -> bool:
|
|
"""Whether the root views were bound mid-session and are not live until a
|
|
restart (see :func:`bind_legacy_views`). Distinct from
|
|
:func:`legacy_credentials_active`: at a mid-session FIRST enable the bound
|
|
views serve exactly the current credentials (active is True) yet
|
|
``/authorize`` is not live until the pending restart — the admin surfaces
|
|
(options hint, startup log) use this to caveat credentials that are
|
|
correct but not yet serving."""
|
|
return bool(hass.data.get(_LEGACY_PENDING_RESTART_KEY))
|
|
|
|
|
|
def _live_auth_mode(hass: HomeAssistant) -> str | None:
|
|
"""Read the CURRENTLY configured webhook auth mode from hass.data.
|
|
|
|
Deferred import to avoid a module cycle: mcp_webhook imports this module
|
|
for the views/provider it registers, so this module cannot import
|
|
mcp_webhook at load time.
|
|
"""
|
|
from .mcp_webhook import active_auth_mode
|
|
|
|
return active_auth_mode(hass)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Small helpers (ported from the add-on's oauth.py)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _b64url_encode(raw: bytes) -> str:
|
|
return base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii")
|
|
|
|
|
|
def _b64url_decode(s: str) -> bytes:
|
|
pad = "=" * (-len(s) % 4)
|
|
return base64.urlsafe_b64decode(s + pad)
|
|
|
|
|
|
def _is_loopback_host(hostname: str) -> bool:
|
|
"""True for the loopback hosts RFC 8252 §7.3/§8.3 allows over plain http."""
|
|
if hostname in _LOOPBACK_HOSTNAMES:
|
|
return True
|
|
try:
|
|
# Covers all of 127.0.0.0/8 and ::1, not just the literal 127.0.0.1.
|
|
return ipaddress.ip_address(hostname).is_loopback
|
|
except ValueError:
|
|
return False
|
|
|
|
|
|
def _is_valid_redirect_uri(redirect_uri: str) -> bool:
|
|
"""Spec-floor validation for OAuth redirect_uri: an https:// URL — or an
|
|
http:// loopback URL (RFC 8252 §7.3, for native/CLI clients) — with a
|
|
non-empty host, a valid port, and no fragment. Single-tenant — no per-client
|
|
allowlist, but reject the obvious bad shapes that would let an attacker
|
|
direct the flow to an empty/malformed URL."""
|
|
if not redirect_uri:
|
|
return False
|
|
try:
|
|
parsed = urlparse(redirect_uri)
|
|
# Accessing .port validates it: urlparse defers the range/format check
|
|
# until access, so a crafted ':999999' or ':abc' port raises ValueError
|
|
# HERE (→ clean 400) instead of later in yarl inside _redirect_with,
|
|
# where it would escape as an uncaught 500 on an unauthenticated view.
|
|
_ = parsed.port
|
|
except ValueError:
|
|
return False
|
|
if not parsed.hostname:
|
|
return False
|
|
if parsed.scheme == "http":
|
|
# Plain http only for loopback callbacks (native-client flow).
|
|
if not _is_loopback_host(parsed.hostname):
|
|
return False
|
|
elif parsed.scheme != "https":
|
|
return False
|
|
# Fragments are not allowed in OAuth redirect URIs (RFC 6749 §3.1.2).
|
|
return not parsed.fragment
|
|
|
|
|
|
def _text_error(
|
|
status: int, message: str, *, restart_hint: bool = False
|
|
) -> web.Response:
|
|
"""Plain-text error response. ``restart_hint`` appends ``_RESTART_HINT`` —
|
|
set it only for the stale-registration cases a full HA restart actually
|
|
unsticks (invalid client_id), not for client-side request mistakes."""
|
|
text = f"{message}. {_RESTART_HINT}" if restart_hint else message
|
|
return web.Response(status=status, text=text)
|
|
|
|
|
|
def _json_error(
|
|
error: str,
|
|
status: int,
|
|
headers: dict[str, str] | None = None,
|
|
*,
|
|
restart_hint: bool = False,
|
|
) -> web.Response:
|
|
"""OAuth JSON error response. ``restart_hint`` carries ``_RESTART_HINT`` in
|
|
``error_description`` — set it only for the stale-registration case
|
|
(``invalid_client``), not client-side protocol errors."""
|
|
body = {"error": error}
|
|
if restart_hint:
|
|
body["error_description"] = _RESTART_HINT
|
|
return web.json_response(body, status=status, headers=headers)
|
|
|
|
|
|
def _json_not_found() -> web.Response:
|
|
"""404 for a root view whose route is disabled in the active mode, or
|
|
whose bound provider is stale (see LegacyOAuthProvider.is_active)."""
|
|
return web.json_response({"error": "not_found"}, status=404)
|
|
|
|
|
|
class _PendingCode(TypedDict):
|
|
"""Shape of an entry in PKCECodeStore._codes. TypedDict so a typo on
|
|
one of these keys fails type-check rather than silently treating it as
|
|
missing."""
|
|
|
|
redirect_uri: str
|
|
code_challenge: str
|
|
expires: float
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# PKCECodeStore (shared PKCE S256 authorization-code lifecycle)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class PKCECodeStore:
|
|
"""In-memory PKCE (S256) authorization-code store.
|
|
|
|
Shared by :class:`LegacyOAuthProvider` and the none-mode auto-approve
|
|
server (:mod:`oauth_autoapprove`) so the one-shot code lifecycle — issue at
|
|
``/authorize``, verify + consume at ``/token`` — has a single
|
|
implementation instead of two copies. Codes are short-lived
|
|
(:data:`AUTH_CODE_TTL`), one-shot, bound to the ``redirect_uri`` +
|
|
``code_challenge`` presented at issuance, and capped
|
|
(:data:`MAX_PENDING_CODES`) with an expiry prune on each issue. A restart
|
|
wipes the store, which only forces in-flight authorize/token round-trips to
|
|
retry.
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
self._codes: dict[str, _PendingCode] = {}
|
|
|
|
def issue_code(self, redirect_uri: str, code_challenge: str) -> str | None:
|
|
"""Issue a one-shot authorization code, or None if the pending-code
|
|
store is at capacity (signals an abuse attempt — see MAX_PENDING_CODES)."""
|
|
now = time.time()
|
|
self._codes = {k: v for k, v in self._codes.items() if v["expires"] > now}
|
|
if len(self._codes) >= MAX_PENDING_CODES:
|
|
_LOGGER.warning(
|
|
"HA-MCP OAuth: pending-code store at cap (%d); refusing "
|
|
"new issuance until existing codes expire or are consumed.",
|
|
MAX_PENDING_CODES,
|
|
)
|
|
return None
|
|
code = secrets.token_urlsafe(32)
|
|
self._codes[code] = {
|
|
"redirect_uri": redirect_uri,
|
|
"code_challenge": code_challenge,
|
|
"expires": now + AUTH_CODE_TTL,
|
|
}
|
|
return code
|
|
|
|
def consume_code(self, code: str, redirect_uri: str, code_verifier: str) -> bool:
|
|
"""One-shot consume ``code``, verifying its PKCE S256 challenge.
|
|
|
|
Returns True only for a live, unexpired code whose stored
|
|
``redirect_uri`` matches and whose ``code_challenge`` equals
|
|
``base64url(SHA-256(code_verifier))``. The code is popped (one-shot)
|
|
before any check that can fail, so a failed attempt still burns it.
|
|
"""
|
|
# Validate the verifier shape per RFC 7636 §4.1 before doing any
|
|
# crypto. A confused client passing an empty/short verifier should be
|
|
# rejected explicitly rather than silently hashing junk.
|
|
if not (PKCE_VERIFIER_MIN <= len(code_verifier) <= PKCE_VERIFIER_MAX):
|
|
return False
|
|
if not _PKCE_VERIFIER_RE.match(code_verifier):
|
|
return False
|
|
entry = self._codes.pop(code, None)
|
|
if entry is None:
|
|
return False
|
|
if entry["expires"] < time.time():
|
|
return False
|
|
if entry["redirect_uri"] != redirect_uri:
|
|
return False
|
|
# PKCE S256 verification: SHA-256(verifier) base64url(no pad) == challenge
|
|
derived = _b64url_encode(hashlib.sha256(code_verifier.encode()).digest())
|
|
return hmac.compare_digest(
|
|
derived.encode("ascii"), entry["code_challenge"].encode("ascii")
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# LegacyOAuthProvider
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class LegacyOAuthProvider:
|
|
"""Holds legacy-OAuth state: token issue/validate, PKCE codes, client auth.
|
|
|
|
Constructed once per bind (see :func:`bind_legacy_views`), not once per
|
|
config-entry reload — see that function's docstring for why. Holds no
|
|
reference to ``hass``; ``active_mode_getter`` is how the bound root views
|
|
learn whether legacy is STILL the live mode on each request (a reload that
|
|
switches away leaves this same instance bound but inactive — see
|
|
:meth:`is_active`).
|
|
"""
|
|
|
|
def __init__(
|
|
self,
|
|
client_id: str,
|
|
client_secret: str,
|
|
signing_key: bytes | str,
|
|
active_mode_getter: Callable[[], str | None],
|
|
) -> None:
|
|
if not client_id:
|
|
raise ValueError("client_id must be a non-empty string")
|
|
if not client_secret:
|
|
raise ValueError("client_secret must be a non-empty string")
|
|
key_bytes = _normalize_signing_key(signing_key)
|
|
if len(key_bytes) < 32:
|
|
raise ValueError("signing_key must be at least 32 bytes")
|
|
self._client_id = client_id
|
|
self._client_secret = client_secret
|
|
self._signing_key = key_bytes
|
|
self._active_mode_getter = active_mode_getter
|
|
# PKCE authorization codes live in the shared store (also used by the
|
|
# none-mode auto-approve server) — see :class:`PKCECodeStore`.
|
|
self._code_store = PKCECodeStore()
|
|
|
|
@property
|
|
def client_id(self) -> str:
|
|
return self._client_id
|
|
|
|
def is_active(self) -> bool:
|
|
"""True iff legacy is the CURRENTLY configured + live webhook auth mode."""
|
|
return self._active_mode_getter() == WEBHOOK_AUTH_LEGACY
|
|
|
|
# -----------------------------------------------------------------
|
|
# Token issuance / validation
|
|
# -----------------------------------------------------------------
|
|
|
|
def _issue_token(self, kind: str, ttl: int) -> str:
|
|
now = int(time.time())
|
|
payload = {
|
|
"kind": kind,
|
|
"iat": now,
|
|
"exp": now + ttl,
|
|
"jti": secrets.token_urlsafe(12),
|
|
"cid": self._client_id,
|
|
}
|
|
body = _b64url_encode(json.dumps(payload, separators=(",", ":")).encode())
|
|
sig = hmac.new(self._signing_key, body.encode("ascii"), hashlib.sha256).digest()
|
|
return f"{body}.{_b64url_encode(sig)}"
|
|
|
|
def _validate_token(self, token: str, expected_kind: str) -> bool:
|
|
try:
|
|
body, sig_part = token.rsplit(".", 1)
|
|
except ValueError:
|
|
return False
|
|
try:
|
|
actual_sig = _b64url_decode(sig_part)
|
|
# body.encode("ascii") is inside the try: a bearer whose
|
|
# pre-signature segment carries a non-ASCII char raises
|
|
# UnicodeEncodeError, which must be caught here (return False)
|
|
# rather than escaping the webhook gate.
|
|
expected_sig = hmac.new(
|
|
self._signing_key, body.encode("ascii"), hashlib.sha256
|
|
).digest()
|
|
except (ValueError, binascii.Error, UnicodeEncodeError):
|
|
return False
|
|
if not hmac.compare_digest(actual_sig, expected_sig):
|
|
return False
|
|
try:
|
|
payload = json.loads(_b64url_decode(body))
|
|
except (ValueError, json.JSONDecodeError):
|
|
return False
|
|
if not isinstance(payload, dict):
|
|
return False
|
|
if payload.get("kind") != expected_kind:
|
|
return False
|
|
if payload.get("cid") != self._client_id:
|
|
# Token was issued for a previous client_id config — reject so a
|
|
# client_id rotation revokes outstanding tokens once the restart
|
|
# binds a provider carrying the new identity. Pre-restart the
|
|
# bound provider still holds the OLD client_id, so old tokens
|
|
# keep validating until then (see bind_legacy_views).
|
|
return False
|
|
# Valid up to but not including `exp` (RFC 7519 §4.1.4 convention).
|
|
return bool(payload.get("exp", 0) > int(time.time()))
|
|
|
|
def issue_access_token(self) -> str:
|
|
return self._issue_token(TOKEN_KIND_ACCESS, ACCESS_TOKEN_TTL)
|
|
|
|
def issue_refresh_token(self) -> str:
|
|
return self._issue_token(TOKEN_KIND_REFRESH, REFRESH_TOKEN_TTL)
|
|
|
|
def validate_access_token(self, token: str) -> bool:
|
|
return self._validate_token(token, TOKEN_KIND_ACCESS)
|
|
|
|
def validate_refresh_token(self, token: str) -> bool:
|
|
return self._validate_token(token, TOKEN_KIND_REFRESH)
|
|
|
|
def validate_bearer(self, request: web.Request) -> bool:
|
|
header = request.headers.get("Authorization", "")
|
|
if not header.lower().startswith("bearer "):
|
|
return False
|
|
token = header[7:].strip()
|
|
return self.validate_access_token(token)
|
|
|
|
# -----------------------------------------------------------------
|
|
# Authorization codes (PKCE)
|
|
# -----------------------------------------------------------------
|
|
|
|
def issue_code(self, redirect_uri: str, code_challenge: str) -> str | None:
|
|
"""Issue a one-shot PKCE-bound authorization code (see PKCECodeStore)."""
|
|
return self._code_store.issue_code(redirect_uri, code_challenge)
|
|
|
|
def consume_code(self, code: str, redirect_uri: str, code_verifier: str) -> bool:
|
|
"""Verify PKCE S256 + one-shot consume a code (see PKCECodeStore)."""
|
|
return self._code_store.consume_code(code, redirect_uri, code_verifier)
|
|
|
|
# -----------------------------------------------------------------
|
|
# Client authentication
|
|
# -----------------------------------------------------------------
|
|
|
|
def authenticate_client(
|
|
self, client_id: str | None, client_secret: str | None
|
|
) -> bool:
|
|
if not client_id or not client_secret:
|
|
return False
|
|
return hmac.compare_digest(
|
|
client_id.encode(), self._client_id.encode()
|
|
) and hmac.compare_digest(client_secret.encode(), self._client_secret.encode())
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Views (root /authorize + /token)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class AuthorizeView(HomeAssistantView):
|
|
"""OAuth /authorize endpoint with a minimal consent page."""
|
|
|
|
requires_auth = False
|
|
url = AUTHORIZE_PATH
|
|
name = "ha_mcp_tools:oauth:authorize"
|
|
|
|
def __init__(self, provider: LegacyOAuthProvider) -> None:
|
|
self._provider = provider
|
|
|
|
@staticmethod
|
|
def _redirect_with(redirect_uri: str, **params: str) -> web.Response:
|
|
# yarl ships with aiohttp and handles existing-query-string merging
|
|
# plus parameter encoding correctly — safer than hand-rolling.
|
|
import yarl
|
|
|
|
url = yarl.URL(redirect_uri).update_query(params)
|
|
return web.Response(status=302, headers={"Location": str(url)})
|
|
|
|
async def get(self, request: web.Request) -> web.Response:
|
|
if not self._provider.is_active():
|
|
# Serve ONLY while legacy is the live mode. Both ha_auth/none (HA
|
|
# core or the secret URL is the authority) and a not-yet-live
|
|
# entry mean this root view must not serve — HA can't rebind or
|
|
# drop it without a restart, so a mode switch away leaves it
|
|
# bound. Refuse it.
|
|
return _text_error(404, "not found")
|
|
params = request.query
|
|
client_id = params.get("client_id", "")
|
|
redirect_uri = params.get("redirect_uri", "")
|
|
state = params.get("state", "")
|
|
code_challenge = params.get("code_challenge", "")
|
|
code_challenge_method = params.get("code_challenge_method", "")
|
|
response_type = params.get("response_type", "")
|
|
|
|
err = self._validate_authorize_params(
|
|
response_type=response_type,
|
|
client_id=client_id,
|
|
redirect_uri=redirect_uri,
|
|
code_challenge=code_challenge,
|
|
code_challenge_method=code_challenge_method,
|
|
)
|
|
if err is not None:
|
|
return err
|
|
|
|
# Render minimal consent page. Showing the redirect_uri lets the user
|
|
# verify the flow goes back to a domain they recognize.
|
|
html = f"""<!DOCTYPE html>
|
|
<html>
|
|
<head>
|
|
<meta charset="utf-8">
|
|
<title>Authorize MCP Connector</title>
|
|
<style>
|
|
body {{ font-family: system-ui, sans-serif; max-width: 36rem; margin: 4rem auto; padding: 0 1rem; }}
|
|
code {{ background: #f4f4f4; padding: 2px 6px; border-radius: 3px; word-break: break-all; }}
|
|
button {{ padding: 0.5rem 1rem; font-size: 1rem; margin-right: 0.5rem; }}
|
|
.approve {{ background: #2563eb; color: white; border: none; }}
|
|
.deny {{ background: #e5e7eb; color: #111; border: none; }}
|
|
</style>
|
|
</head>
|
|
<body>
|
|
<h1>Authorize MCP Connector</h1>
|
|
<p>An MCP client is requesting access to your Home Assistant MCP server.</p>
|
|
<p>It will redirect to:<br><code>{escape(redirect_uri)}</code></p>
|
|
<p>Only allow this if you started this connection yourself.</p>
|
|
<form method="POST" action="{AUTHORIZE_PATH}">
|
|
<input type="hidden" name="client_id" value="{escape(client_id)}">
|
|
<input type="hidden" name="redirect_uri" value="{escape(redirect_uri)}">
|
|
<input type="hidden" name="state" value="{escape(state)}">
|
|
<input type="hidden" name="code_challenge" value="{escape(code_challenge)}">
|
|
<button class="approve" type="submit" name="action" value="approve">Allow</button>
|
|
<button class="deny" type="submit" name="action" value="deny">Deny</button>
|
|
</form>
|
|
</body>
|
|
</html>"""
|
|
return web.Response(text=html, content_type="text/html")
|
|
|
|
async def post(self, request: web.Request) -> web.Response:
|
|
if not self._provider.is_active():
|
|
return _text_error(404, "not found")
|
|
data = await request.post()
|
|
action = str(data.get("action", ""))
|
|
client_id = str(data.get("client_id", ""))
|
|
redirect_uri = str(data.get("redirect_uri", ""))
|
|
state = str(data.get("state", ""))
|
|
code_challenge = str(data.get("code_challenge", ""))
|
|
|
|
# Re-validate everything from the form — never trust hidden fields.
|
|
# response_type/method aren't carried on the POST so we hard-code the
|
|
# spec values here; the validator still applies all the same rules to
|
|
# the user-influenceable fields.
|
|
err = self._validate_authorize_params(
|
|
response_type="code",
|
|
client_id=client_id,
|
|
redirect_uri=redirect_uri,
|
|
code_challenge=code_challenge,
|
|
code_challenge_method="S256",
|
|
)
|
|
if err is not None:
|
|
return err
|
|
|
|
if action == "deny":
|
|
return self._redirect_with(redirect_uri, error="access_denied", state=state)
|
|
if action != "approve":
|
|
return _text_error(400, "invalid action")
|
|
|
|
code = self._provider.issue_code(redirect_uri, code_challenge)
|
|
if code is None:
|
|
# Pending-code store at cap → signal back per RFC 6749 §4.1.2.1
|
|
# instead of silently failing.
|
|
return self._redirect_with(
|
|
redirect_uri, error="temporarily_unavailable", state=state
|
|
)
|
|
return self._redirect_with(redirect_uri, code=code, state=state)
|
|
|
|
def _validate_authorize_params(
|
|
self,
|
|
*,
|
|
response_type: str,
|
|
client_id: str,
|
|
redirect_uri: str,
|
|
code_challenge: str,
|
|
code_challenge_method: str,
|
|
) -> web.Response | None:
|
|
"""Return a 400 web.Response if any /authorize param is invalid, or
|
|
None if all checks pass. Centralized so GET and POST share identical
|
|
validation — the POST path explicitly re-validates the hidden form
|
|
fields rather than trusting them."""
|
|
if response_type != "code":
|
|
return _text_error(400, "unsupported_response_type")
|
|
if code_challenge_method != "S256":
|
|
return _text_error(400, "invalid code_challenge_method (S256 required)")
|
|
if not _PKCE_CHALLENGE_RE.match(code_challenge):
|
|
return _text_error(
|
|
400, "invalid code_challenge (must be 43-char base64url)"
|
|
)
|
|
if client_id != self._provider.client_id:
|
|
return _text_error(400, "invalid client_id", restart_hint=True)
|
|
if not _is_valid_redirect_uri(redirect_uri):
|
|
return _text_error(
|
|
400,
|
|
"redirect_uri must be an https:// URL (or an http:// loopback "
|
|
"URL) with a valid host and port",
|
|
)
|
|
return None
|
|
|
|
|
|
class TokenView(HomeAssistantView):
|
|
"""OAuth /token endpoint: authorization_code + refresh_token grants."""
|
|
|
|
requires_auth = False
|
|
cors_allowed = True
|
|
url = TOKEN_PATH
|
|
name = "ha_mcp_tools:oauth:token"
|
|
|
|
def __init__(self, provider: LegacyOAuthProvider) -> None:
|
|
self._provider = provider
|
|
|
|
@staticmethod
|
|
def _extract_client_creds(
|
|
request: web.Request, form: dict
|
|
) -> tuple[str | None, str | None]:
|
|
"""Pull client_id/secret from Basic auth header OR form body."""
|
|
header = request.headers.get("Authorization", "")
|
|
if header.lower().startswith("basic "):
|
|
try:
|
|
decoded = base64.b64decode(header[6:].strip(), validate=True).decode(
|
|
"utf-8"
|
|
)
|
|
except (ValueError, UnicodeDecodeError, binascii.Error):
|
|
return None, None
|
|
if ":" in decoded:
|
|
cid, _, sec = decoded.partition(":")
|
|
# RFC 6749 §2.3.1: client_secret_basic values are
|
|
# application/x-www-form-urlencoded before base64, so decode
|
|
# them back with unquote_PLUS ("+" means space in that
|
|
# encoding — plain unquote would leave it literal). A no-op for
|
|
# the generated credentials (URL-safe alphabets, nothing to
|
|
# decode) but required for custom overrides containing reserved
|
|
# characters — matches the form-body path below, which aiohttp
|
|
# also form-decodes ("+" → space).
|
|
return unquote_plus(cid), unquote_plus(sec)
|
|
return None, None
|
|
return form.get("client_id"), form.get("client_secret")
|
|
|
|
async def post(self, request: web.Request) -> web.Response:
|
|
if not self._provider.is_active():
|
|
return _json_not_found()
|
|
form = dict(await request.post())
|
|
client_id, client_secret = self._extract_client_creds(request, form)
|
|
if not self._provider.authenticate_client(client_id, client_secret):
|
|
return _json_error(
|
|
"invalid_client",
|
|
401,
|
|
headers={"WWW-Authenticate": 'Basic realm="HA-MCP OAuth"'},
|
|
restart_hint=True,
|
|
)
|
|
|
|
grant_type = form.get("grant_type", "")
|
|
if grant_type == "authorization_code":
|
|
return await self._handle_authorization_code(form)
|
|
if grant_type == "refresh_token":
|
|
return await self._handle_refresh(form)
|
|
return _json_error("unsupported_grant_type", 400)
|
|
|
|
async def _handle_authorization_code(self, form: dict) -> web.Response:
|
|
code = str(form.get("code", ""))
|
|
redirect_uri = str(form.get("redirect_uri", ""))
|
|
code_verifier = str(form.get("code_verifier", ""))
|
|
if not (code and redirect_uri and code_verifier):
|
|
return _json_error("invalid_request", 400)
|
|
if not self._provider.consume_code(code, redirect_uri, code_verifier):
|
|
return _json_error("invalid_grant", 400)
|
|
return web.json_response(
|
|
{
|
|
"access_token": self._provider.issue_access_token(),
|
|
"token_type": "Bearer",
|
|
"expires_in": ACCESS_TOKEN_TTL,
|
|
"refresh_token": self._provider.issue_refresh_token(),
|
|
},
|
|
headers=_TOKEN_RESPONSE_HEADERS,
|
|
)
|
|
|
|
async def _handle_refresh(self, form: dict) -> web.Response:
|
|
refresh = str(form.get("refresh_token", ""))
|
|
if not refresh or not self._provider.validate_refresh_token(refresh):
|
|
return _json_error("invalid_grant", 400)
|
|
return web.json_response(
|
|
{
|
|
"access_token": self._provider.issue_access_token(),
|
|
"token_type": "Bearer",
|
|
"expires_in": ACCESS_TOKEN_TTL,
|
|
"refresh_token": self._provider.issue_refresh_token(),
|
|
},
|
|
headers=_TOKEN_RESPONSE_HEADERS,
|
|
)
|