Restrict notification action links to application paths

src/govoplan_notifications/backend/schemas.py

@@ -3,12 +3,28 @@ from __future__ import annotations
 from datetime import datetime
 from typing import Any, Literal
 
-from pydantic import BaseModel, ConfigDict, Field
+from pydantic import BaseModel, ConfigDict, Field, field_validator
 
 
 NotificationStatus = Literal["pending", "queued", "sending", "sent", "failed", "skipped", "cancelled"]
 
 
+def normalize_notification_action_url(value: str | None) -> str | None:
+    if value is None:
+        return None
+    candidate = value.strip()
+    if not candidate:
+        return None
+    if (
+        not candidate.startswith("/")
+        or candidate.startswith("//")
+        or "\\" in candidate
+        or any(ord(character) < 32 or ord(character) == 127 for character in candidate)
+    ):
+        raise ValueError("Notification action URL must be an application-relative path")
+    return candidate
+
+
 class NotificationCreateRequest(BaseModel):
     model_config = ConfigDict(extra="forbid")
 
@@ -31,6 +47,11 @@ class NotificationCreateRequest(BaseModel):
     payload: dict[str, Any] = Field(default_factory=dict)
     metadata: dict[str, Any] = Field(default_factory=dict)
 
+    @field_validator("action_url")
+    @classmethod
+    def validate_action_url(cls, value: str | None) -> str | None:
+        return normalize_notification_action_url(value)
+
 
 class NotificationUpdateRequest(BaseModel):
     model_config = ConfigDict(extra="forbid")

src/govoplan_notifications/backend/service.py

@@ -12,7 +12,12 @@ from sqlalchemy.orm import Session
 from govoplan_core.core.notifications import NotificationDispatchRequest
 from govoplan_core.db.base import utcnow
 from govoplan_notifications.backend.db.models import NotificationDeliveryAttempt, NotificationMessage, NotificationPreference
-from govoplan_notifications.backend.schemas import NotificationCreateRequest, NotificationPreferencesUpdateRequest, NotificationUpdateRequest
+from govoplan_notifications.backend.schemas import (
+    NotificationCreateRequest,
+    NotificationPreferencesUpdateRequest,
+    NotificationUpdateRequest,
+    normalize_notification_action_url,
+)
 
 
 class NotificationError(ValueError):
@@ -81,6 +86,15 @@ def _clean_source_modules(values: list[str]) -> list[str]:
     return cleaned
 
 
+def _safe_notification_action_url(value: str | None) -> str | None:
+    try:
+        return normalize_notification_action_url(value)
+    except ValueError:
+        # Legacy rows predate action URL validation. Never project an unsafe
+        # stored value back into a clickable browser link.
+        return None
+
+
 def create_notification(
     session: Session,
     *,
@@ -101,7 +115,7 @@ def create_notification(
         subject=payload.subject,
         body_text=payload.body_text,
         body_html=payload.body_html,
-        action_url=payload.action_url,
+        action_url=normalize_notification_action_url(payload.action_url),
         priority=payload.priority,
         not_before_at=payload.not_before_at,
         status="queued" if payload.enqueue_delivery else "pending",
@@ -485,7 +499,7 @@ def notification_response(notification: NotificationMessage) -> dict[str, Any]:
         "subject": notification.subject,
         "body_text": notification.body_text,
         "body_html": notification.body_html,
-        "action_url": notification.action_url,
+        "action_url": _safe_notification_action_url(notification.action_url),
         "priority": notification.priority,
         "status": notification.status,
         "not_before_at": response_datetime(notification.not_before_at),