feat: normalize base_url and strip credentials from logs

New url_util.py (deliberately separate from the vendored files, so a
future upstream re-sync never conflicts with it): normalize_base_url
strips whitespace/trailing slashes and rejects non-http(s) or hostless
URLs; sanitize_url_for_logging masks embedded basic-auth credentials
for Task 7's discovery logging.

Wires normalization into config_flow.validate_input before the client
is built, with a new invalid_base_url form error.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01739Seac5eMAfNwFeiwV5r4
This commit is contained in:
Adam SmithandClaude Sonnet 5 committed 2026-09-17 17:53:10 -04:00
1 parent 564c52159f
commit 914c89dc6b
4 files changed
+86

No files matched your search

@@ -103,6 +103,7 @@ from .const import (
UNSUPPORTED_PRIORITY_SERVICE_TIERS_MODELS, UNSUPPORTED_PRIORITY_SERVICE_TIERS_MODELS,
UNSUPPORTED_WEB_SEARCH_MODELS, UNSUPPORTED_WEB_SEARCH_MODELS,
) )
from .url_util import InvalidBaseURL, normalize_base_url
_LOGGER = logging.getLogger(__name__) _LOGGER = logging.getLogger(__name__)
@@ -119,6 +120,7 @@ async def validate_input(hass: HomeAssistant, data: dict[str, Any]) -> None:
Data has the keys from STEP_USER_DATA_SCHEMA with values provided by the user. Data has the keys from STEP_USER_DATA_SCHEMA with values provided by the user.
""" """
data[CONF_BASE_URL] = normalize_base_url(data[CONF_BASE_URL])
client = openai.AsyncOpenAI( client = openai.AsyncOpenAI(
api_key=data[CONF_API_KEY], api_key=data[CONF_API_KEY],
base_url=data[CONF_BASE_URL], base_url=data[CONF_BASE_URL],
@@ -145,6 +147,8 @@ class OpenAIConfigFlow(ConfigFlow, domain=DOMAIN):
self._async_abort_entries_match(user_input) self._async_abort_entries_match(user_input)
try: try:
await validate_input(self.hass, user_input) await validate_input(self.hass, user_input)
except InvalidBaseURL:
errors["base"] = "invalid_base_url"
except openai.APIConnectionError: except openai.APIConnectionError:
errors["base"] = "cannot_connect" errors["base"] = "cannot_connect"
except openai.AuthenticationError: except openai.AuthenticationError:
@@ -7,6 +7,7 @@
"error": { "error": {
"cannot_connect": "[%key:common::config_flow::error::cannot_connect%]", "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
"invalid_auth": "[%key:common::config_flow::error::invalid_auth%]", "invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
"invalid_base_url": "That does not look like a valid http(s) URL.",
"unknown": "[%key:common::config_flow::error::unknown%]" "unknown": "[%key:common::config_flow::error::unknown%]"
}, },
"step": { "step": {
@@ -0,0 +1,42 @@
"""Base-URL normalization and safe logging.
This module is intentionally separate from the vendored upstream files so that
re-syncing with home-assistant/core never conflicts with it.
"""
from __future__ import annotations
from urllib.parse import urlparse, urlunparse
ALLOWED_SCHEMES = frozenset({"http", "https"})
class InvalidBaseURL(ValueError):
"""Raised when a base URL is unusable."""
def normalize_base_url(url: str) -> str:
"""Strip whitespace and trailing slashes, and validate the scheme."""
cleaned = (url or "").strip()
if not cleaned:
raise InvalidBaseURL("base URL is empty")
parsed = urlparse(cleaned)
if parsed.scheme not in ALLOWED_SCHEMES:
raise InvalidBaseURL(f"unsupported scheme: {parsed.scheme or '(none)'}")
if not parsed.netloc:
raise InvalidBaseURL("base URL has no host")
return urlunparse(parsed._replace(path=parsed.path.rstrip("/")))
def sanitize_url_for_logging(url: str) -> str:
"""Return the URL with any embedded credentials replaced by ***."""
parsed = urlparse(url)
if not parsed.username and not parsed.password:
return url
host = parsed.hostname or ""
if parsed.port:
host = f"{host}:{parsed.port}"
return urlunparse(parsed._replace(netloc=f"***@{host}"))
+39
View File
@@ -0,0 +1,39 @@
"""URL hygiene: normalize input, never log credentials."""
import pytest
from custom_components.openai_compatible.url_util import (
InvalidBaseURL,
normalize_base_url,
sanitize_url_for_logging,
)
@pytest.mark.parametrize(
("raw", "expected"),
[
("http://192.0.2.10:4000/v1", "http://192.0.2.10:4000/v1"),
("http://192.0.2.10:4000/v1/", "http://192.0.2.10:4000/v1"),
(" https://api.openai.com/v1 ", "https://api.openai.com/v1"),
("http://host:4000/v1///", "http://host:4000/v1"),
],
)
def test_normalize(raw, expected):
assert normalize_base_url(raw) == expected
@pytest.mark.parametrize("raw", ["ftp://host/v1", "not-a-url", "", " "])
def test_normalize_rejects(raw):
with pytest.raises(InvalidBaseURL):
normalize_base_url(raw)
def test_sanitize_strips_credentials():
assert (
sanitize_url_for_logging("https://user:secret@host:4000/v1")
== "https://***@host:4000/v1"
)
def test_sanitize_leaves_clean_urls_alone():
assert sanitize_url_for_logging("http://host:4000/v1") == "http://host:4000/v1"