feat(server): enable authentication to webhook endpoint (#1015)

* feat(server): add webhook charm options with typed config

* feat(server): update server API with webhook authentication

* tests: add charm unit tests for typed config

* tests: add server tests for webhook authentication

* tests: add missing tests based on copilot review

* add reviewer recommendations

Author: rene-oromtz

server/charm/charmcraft.yaml

@@ -112,6 +112,15 @@ config:
       type: string
       default: "localhost,127.0.0.1,::1"
       description: Resources that we should be able to access bypassing proxy
+    webhook_url:
+      type: string
+      default: "http://test-observer-api.local/"
+      description: The URL for the webhook to send notifications to.
+    webhook_auth:
+      type: string
+      default: ""
+      description: Optional bearer token for authenticating with the webhook endpoint
+
 
 parts:
   charm:

server/charm/pyproject.toml

@@ -9,6 +9,7 @@ dependencies = [
     "pymongo<4.17.1",
     "urllib3>=2.6.3",
     "cosl>=0.0.53",
+    "pydantic>=2.12,<3",
 ]
 
 [dependency-groups]
@@ -63,4 +64,4 @@ ignore = [
 ]
 
 [tool.ruff.lint.pydocstyle]
-convention = "pep257"
+convention = "pep257"

server/charm/src/charm.py

@@ -37,6 +37,8 @@
 from pymongo import MongoClient
 from pymongo.errors import PyMongoError
 
+from config import TestflingerServerConfig
+
 logger = logging.getLogger(__name__)
 
 TESTFLINGER_ADMIN_ID = "testflinger-admin"
@@ -54,6 +56,9 @@ class TestflingerCharm(ops.CharmBase):
     def __init__(self, *args):
         """Initialize the charm."""
         super().__init__(*args)
+        self.typed_config = self.load_config(
+            TestflingerServerConfig, errors="blocked"
+        )
         self.pebble_service_name = "testflinger"
         self.pebble_check_name = "v1_up"
         self.container = self.unit.get_container("testflinger")
@@ -143,7 +148,7 @@ def _require_nginx_route(self):
         # TODO: Remove nginx route when migration to traefik route is completed
         require_nginx_route(
             charm=self,
-            service_hostname=self.config["external_hostname"],
+            service_hostname=self.typed_config.external_hostname,
             service_name=self.app.name,
             service_port=DEFAULT_PORT,
         )
@@ -193,7 +198,7 @@ def _configure_traefik_route(self):
 
         testflinger_base_url = (
             self.traefik_route.external_host
-            or self.config.get("external_hostname", "")
+            or self.typed_config.external_hostname
         )
         if not testflinger_base_url:
             self.unit.status = ops.BlockedStatus(
@@ -345,7 +350,7 @@ def _run_rotation(self, app_env: dict) -> str:
 
     def _on_config_changed(self, _: ops.framework.EventBase) -> None:
         """Handle config changed event."""
-        new_key = self.config["testflinger_secrets_master_key"]
+        new_key = self.typed_config.testflinger_secrets_master_key
         stored_key = self._stored.previous_master_key
 
         # Rotate only if master key was previously defined, has changed,
@@ -389,7 +394,7 @@ def _on_retry_key_rotation_action(self, event: ops.ActionEvent) -> None:
         unit in BlockedStatus. The old key is read from StoredState and the
         new key from the current config.
         """
-        current_key = self.config["testflinger_secrets_master_key"]
+        current_key = self.typed_config.testflinger_secrets_master_key
         stored_key = self._stored.previous_master_key
 
         if not current_key:
@@ -423,7 +428,7 @@ def _on_retry_key_rotation_action(self, event: ops.ActionEvent) -> None:
     @property
     def _pebble_layer(self):
         """Return a dictionary representing a Pebble layer."""
-        keepalive = str(self.config["keepalive"])
+        keepalive = str(self.typed_config.keepalive)
         command = " ".join(
             [
                 "gunicorn",
@@ -482,15 +487,15 @@ def app_environment(self) -> dict:
             "MONGODB_USERNAME": db_data.get("db_username"),
             "MONGODB_PASSWORD": db_data.get("db_password"),
             "MONGODB_DATABASE": db_data.get("db_database"),
-            "MONGODB_MAX_POOL_SIZE": str(self.config["max_pool_size"]),
-            "JWT_SIGNING_KEY": self.config["jwt_signing_key"],
-            "TESTFLINGER_SECRETS_MASTER_KEY": self.config[
-                "testflinger_secrets_master_key"
-            ],
+            "MONGODB_MAX_POOL_SIZE": str(self.typed_config.max_pool_size),
+            "JWT_SIGNING_KEY": self.typed_config.jwt_signing_key,
+            "TESTFLINGER_SECRETS_MASTER_KEY": self.typed_config.testflinger_secrets_master_key,  # noqa: E501
             "TESTFLINGER_KEY_VAULT_URI": self._fetch_keyvault_uri() or "",
-            "HTTP_PROXY": self.config["http_proxy"],
-            "HTTPS_PROXY": self.config["https_proxy"],
-            "NO_PROXY": self.config["no_proxy"],
+            "HTTP_PROXY": self.typed_config.http_proxy,
+            "HTTPS_PROXY": self.typed_config.https_proxy,
+            "NO_PROXY": self.typed_config.no_proxy,
+            "WEBHOOK_URL": self.typed_config.webhook_url,
+            "WEBHOOK_AUTH": self.typed_config.webhook_auth,
         }
         return env
 

server/charm/src/config.py

@@ -0,0 +1,49 @@
+# Copyright 2026 Canonical Ltd.
+# See LICENSE file for licensing details.
+"""Charm configuration."""
+
+from urllib.parse import urlparse
+
+import pydantic
+
+
+class TestflingerServerConfig(pydantic.BaseModel):
+    """Testflinger Server Charm configuration."""
+
+    external_hostname: str = "testflinger.local"
+    keepalive: int = 10
+    max_pool_size: int = 100
+    jwt_signing_key: str = ""
+    testflinger_secrets_master_key: str = ""
+    http_proxy: str = ""
+    https_proxy: str = ""
+    no_proxy: str = "localhost,127.0.0.1,::1"
+    webhook_url: str = "http://test-observer-api.local/"
+    webhook_auth: str = ""
+
+    @pydantic.field_validator("external_hostname")
+    @classmethod
+    def validate_external_hostname(cls, value):
+        """Validate that external_hostname does not include HTTP scheme.
+
+        Protocol to be used is defined by the ingress controller.
+        """
+        if value.startswith(("http://", "https://")):
+            raise ValueError(
+                "external_hostname must not include protocol (http:// or https://)"
+            )
+        return value
+
+    @pydantic.field_validator("webhook_url")
+    @classmethod
+    def validate_webhook_url(cls, value):
+        """Validate that webhook_url includes a protocol and no paths."""
+        parsed_webhook = urlparse(value)
+        if parsed_webhook.scheme not in {"http", "https"}:
+            raise ValueError(
+                "webhook_url must include protocol (http:// or https://)"
+            )
+        if not parsed_webhook.netloc:
+            raise ValueError("webhook_url must include a host")
+
+        return value

server/charm/tests/unit/test_config.py

@@ -0,0 +1,53 @@
+# Copyright 2026 Canonical
+# See LICENSE file for licensing details.
+"""Unit tests for config.py."""
+
+import pytest
+
+from config import TestflingerServerConfig
+
+
+def test_valid_external_hostname():
+    """Test that valid external hostname are accepted."""
+    config = TestflingerServerConfig(external_hostname="testflinger.local")
+    assert config.external_hostname == "testflinger.local"
+
+    config = TestflingerServerConfig(
+        external_hostname="testflinger.canonical.com"
+    )
+    assert config.external_hostname == "testflinger.canonical.com"
+
+
+def test_invalid_external_hostname():
+    """Test that invalid external hostname are rejected."""
+    # Protocol is defined by ingress controller
+    # Reject hostnames that include protocol
+    with pytest.raises(ValueError):
+        TestflingerServerConfig(external_hostname="http://testflinger.local")
+
+    with pytest.raises(ValueError):
+        TestflingerServerConfig(external_hostname="https://testflinger.local")
+
+
+def test_valid_webhook_url():
+    """Test that valid webhook urls are accepted."""
+    config = TestflingerServerConfig(
+        webhook_url="https://test-observer-api.local/"
+    )
+    assert config.webhook_url == "https://test-observer-api.local/"
+
+    config = TestflingerServerConfig(
+        webhook_url="http://test-observer-api.local"
+    )
+    assert config.webhook_url == "http://test-observer-api.local"
+
+
+def test_invalid_webhook_url():
+    """Test that invalid webhook urls are rejected."""
+    # Reject webhook url that do not include protocol
+    with pytest.raises(ValueError):
+        TestflingerServerConfig(webhook_url="test-observer-api.local")
+
+    # Reject webhook url that does not include hostname
+    with pytest.raises(ValueError):
+        TestflingerServerConfig(webhook_url="https:///v1/test-executions/")