Enforce http(s) schema on service base_url fields

Add a shared ServiceBaseUrl type (BeforeValidator + Field description) in
integrations/base.py and apply it to base_url across all six service configs
(grafana, prometheus, alertmanager, jellyfin, jellyseerr, nextcloud). Missing
http:// or https:// schema now fails fast with a clear 422 instead of breaking
HTTP clients silently. Tests cover reject/accept cases; REQUIREMENTS updated.
This commit is contained in:
Developer
2026-06-26 09:52:20 +00:00
parent 56b919ea1f
commit eebc86a52b
9 changed files with 71 additions and 9 deletions
@@ -6,6 +6,7 @@ from typing import Any
from media_library_viewer_api.integrations.base import (
SecretField,
ServiceBaseUrl,
ServiceConfigBase,
ServiceDefinition,
WidgetConfigBase,
@@ -16,7 +17,7 @@ from media_library_viewer_api.integrations.base import (
class AlertmanagerConfig(ServiceConfigBase):
"""Non-secret Alertmanager connection config."""
base_url: str
base_url: ServiceBaseUrl
timeout_seconds: int = 5
@@ -16,9 +16,37 @@ map. There is no runtime plugin loading.
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any
from typing import Annotated, Any
from pydantic import BaseModel
from pydantic import BaseModel, BeforeValidator, Field
def _validate_service_base_url(value: Any) -> str:
"""Require an absolute http(s) URL for service ``base_url`` fields.
Relative hosts (e.g. ``grafana.example.com``) break downstream HTTP clients
because ``requests`` treats them as relative paths, so we fail fast with a
clear error instead of letting the call silently malfunction.
"""
if not isinstance(value, str):
raise ValueError("base_url must be a string starting with http:// or https://")
text = value.strip()
if not text:
raise ValueError("base_url must not be empty")
lowered = text.lower()
if not (lowered.startswith("http://") or lowered.startswith("https://")):
raise ValueError("base_url must start with http:// or https:// (include the schema)")
return text
#: Shared annotated type for service ``base_url`` fields. applying the validator
#: uniformly across every integration so missing schemas are rejected at the
#: config boundary with a helpful message.
ServiceBaseUrl = Annotated[
str,
Field(description="Absolute URL including the http:// or https:// schema."),
BeforeValidator(_validate_service_base_url),
]
class ServiceConfigBase(BaseModel):
@@ -26,6 +54,9 @@ class ServiceConfigBase(BaseModel):
Subclass this in each integration module and declare the connection fields.
The JSON schema is derived via ``model_json_schema()`` and exposed to the UI.
Connection URLs should use the :data:`ServiceBaseUrl` type so the
``http(s)://`` schema is enforced consistently across integrations.
"""
@@ -4,6 +4,7 @@ from __future__ import annotations
from media_library_viewer_api.integrations.base import (
SecretField,
ServiceBaseUrl,
ServiceConfigBase,
ServiceDefinition,
WidgetConfigBase,
@@ -14,7 +15,7 @@ from media_library_viewer_api.integrations.base import (
class GrafanaConfig(ServiceConfigBase):
"""Non-secret Grafana connection config."""
base_url: str
base_url: ServiceBaseUrl
timeout_seconds: int = 5
@@ -4,6 +4,7 @@ from __future__ import annotations
from media_library_viewer_api.integrations.base import (
SecretField,
ServiceBaseUrl,
ServiceConfigBase,
ServiceDefinition,
WidgetConfigBase,
@@ -14,7 +15,7 @@ from media_library_viewer_api.integrations.base import (
class JellyfinConfig(ServiceConfigBase):
"""Non-secret Jellyfin connection config."""
base_url: str
base_url: ServiceBaseUrl
user_id: str = ""
timeout_seconds: int = 10
@@ -9,6 +9,7 @@ from __future__ import annotations
from media_library_viewer_api.integrations.base import (
SecretField,
ServiceBaseUrl,
ServiceConfigBase,
ServiceDefinition,
)
@@ -17,7 +18,7 @@ from media_library_viewer_api.integrations.base import (
class JellyseerrConfig(ServiceConfigBase):
"""Non-secret Jellyseerr connection config."""
base_url: str
base_url: ServiceBaseUrl
DEFINITION = ServiceDefinition(
@@ -8,6 +8,7 @@ from __future__ import annotations
from media_library_viewer_api.integrations.base import (
SecretField,
ServiceBaseUrl,
ServiceConfigBase,
ServiceDefinition,
)
@@ -16,7 +17,7 @@ from media_library_viewer_api.integrations.base import (
class NextcloudConfig(ServiceConfigBase):
"""Non-secret Nextcloud connection config."""
base_url: str
base_url: ServiceBaseUrl
username: str = ""
@@ -4,6 +4,7 @@ from __future__ import annotations
from media_library_viewer_api.integrations.base import (
SecretField,
ServiceBaseUrl,
ServiceConfigBase,
ServiceDefinition,
WidgetConfigBase,
@@ -14,7 +15,7 @@ from media_library_viewer_api.integrations.base import (
class PrometheusConfig(ServiceConfigBase):
"""Non-secret Prometheus connection config."""
base_url: str
base_url: ServiceBaseUrl
timeout_seconds: int = 10
+22 -1
View File
@@ -8,6 +8,7 @@ from unittest.mock import patch
import pytest
from cryptography.fernet import Fernet
from fastapi.testclient import TestClient
from pydantic import ValidationError
from media_library_viewer_api.dependencies import get_settings_store
from media_library_viewer_api.integrations.registry import (
@@ -244,7 +245,8 @@ def test_invalid_config_rejected(client):
"/api/services/instances",
json={"service_type": "grafana", "name": "x", "config": {"base_url": ""}},
)
# Pydantic accepts empty string; force a real validation error via bad type.
assert response.status_code == 422
# Force a real validation error via bad type.
response = client.post(
"/api/services/instances",
json={"service_type": "grafana", "name": "x", "config": {"timeout_seconds": "fast"}},
@@ -252,6 +254,25 @@ def test_invalid_config_rejected(client):
assert response.status_code == 422
@pytest.mark.parametrize(
"bad_url", ["grafana.example.com", "localhost:3000", "//grafana.example.com", "ftp://grafana.example.com"]
)
def test_service_base_url_requires_http_schema(bad_url):
"""Every service base_url must include an http:// or https:// schema."""
model = get_service_definition("grafana").config_model
with pytest.raises(ValidationError):
model.model_validate({"base_url": bad_url, "timeout_seconds": 5})
@pytest.mark.parametrize(
"service_type", ["grafana", "prometheus", "alertmanager", "jellyfin", "jellyseerr", "nextcloud"]
)
def test_service_base_url_accepts_absolute_urls(service_type):
model = get_service_definition(service_type).config_model
instance = model.model_validate({"base_url": "https://example.com"})
assert instance.base_url == "https://example.com"
def test_unknown_secret_field_rejected(client):
response = client.post(
"/api/services/instances",
+4
View File
@@ -271,6 +271,10 @@ Service definitions live as Pydantic modules in the backend
(`integrations/`); they declare the service config schema, secret fields, and
the widget kinds the service provides. There is no runtime plugin loading.
Every service `base_url` uses the shared `ServiceBaseUrl` type, which rejects
values missing an `http://` or `https://` schema with a clear validation error
(relative hosts break downstream HTTP clients).
### Services
- **Grafana** — base URL + optional API key; provides a dashboard-link widget.