Add test coverage for _validate_notification_target

backend/secuscan/routes.py

@@ -23,6 +23,8 @@
     iter_raw_output_chunks,
     parse_json_fields,
 )
+from .routes_email_validation import validate_notification_target  # noqa: E402
+
 
 # Re-exported for backward compatibility with integration tests
 SSE_RAW_OUTPUT_CHUNK_SIZE = 64 * 1024
@@ -132,39 +134,15 @@ def _json_payload(value: Any, fallback: str) -> str:
 
 router = APIRouter(prefix="/api/v1", dependencies=[Depends(require_api_key)])
 
-_EMAIL_PATTERN = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$")
-
-
+# _validate_notification_target moved to routes_email_validation
 def _validate_notification_target(channel_type: NotificationChannelType, target: str) -> str:
-    cleaned = target.strip()
-    if not cleaned:
-        raise HTTPException(status_code=400, detail="Notification target is required")
-
-    if channel_type == NotificationChannelType.WEBHOOK:
-        is_valid, error = validate_url(cleaned)
-        if not is_valid:
-            raise HTTPException(status_code=400, detail=error or "Invalid webhook URL")
-
-        if settings.notification_ssrf_enabled:
-            from .validation import resolve_and_validate_target, validate_webhook_target
-            ssrf_ok, ssrf_err = resolve_and_validate_target(cleaned)
-            if not ssrf_ok:
-                raise HTTPException(
-                    status_code=400,
-                    detail=f"Webhook target blocked by SSRF protection: {ssrf_err}"
-                )
-            # Additional independent check against notification_blocked_ip_ranges
-            target_ok, target_err = validate_webhook_target(cleaned)
-            if not target_ok:
-                raise HTTPException(
-                    status_code=400,
-                    detail=f"Webhook target blocked by SSRF protection: {target_err}"
-                )
-        return cleaned
-
-    if not _EMAIL_PATTERN.match(cleaned):
-        raise HTTPException(status_code=400, detail="Invalid email address")
-    return cleaned
+    """Re-export via NotificationChannelType string sentinel to the validation helper."""
+    from .routes_email_validation import validate_notification_target as _vt
+    _channel_str = channel_type.value if hasattr(channel_type, "value") else str(channel_type)
+    try:
+        return _vt(_channel_str, target, notification_ssrf_enabled=settings.notification_ssrf_enabled)
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))
 
 
 def _serialize_notification_rule(row: Dict[str, Any]) -> Dict[str, Any]:

backend/secuscan/routes_email_validation.py

@@ -0,0 +1,105 @@
+"""
+Email-format and notification-target validation helpers.
+
+Extracted from routes.py so they can be unit-tested without pulling in the
+FastAPI / xhtml2pdf / reportlab import chain.
+
+Public API
+----------
+validate_email_format(email: str) -> str
+    Strips whitespace and validates against a simple RFC-5321-ish pattern.
+    Raises ValueError if the address is blank or malformed.
+
+validate_notification_target(channel_type: str, target: str, *,
+                             notification_ssrf_enabled: bool = True) -> str
+    Validates the target for the given channel type.
+    For "webhook": validates the URL and optionally runs SSRF checks.
+    For "email": validates the email format.
+    Returns the stripped target on success.
+    Raises ValueError on failure (HTTPException is raised by the caller).
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Optional, Tuple
+
+_EMAIL_PATTERN = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$")
+
+# Sentinel strings that map to NotificationChannelType members.
+WEBHOOK_TYPE = "webhook"
+EMAIL_TYPE = "email"
+
+
+def validate_email_format(email: str) -> str:
+    """Strip and validate an email address.
+
+    Args:
+        email: Raw address string.
+
+    Returns:
+        The stripped address.
+
+    Raises:
+        ValueError: If the address is blank or malformed.
+    """
+    cleaned = email.strip()
+    if not cleaned:
+        raise ValueError("Email address is required")
+    if not _EMAIL_PATTERN.match(cleaned):
+        raise ValueError("Invalid email address")
+    return cleaned
+
+
+def validate_webhook_url(url: str) -> Tuple[bool, Optional[str]]:
+    """
+    Basic URL scheme check for webhook targets.
+
+    Returns (True, None) for http/https URLs; (False, reason) otherwise.
+    """
+    from .validation import validate_url
+    return validate_url(url)
+
+
+def validate_notification_target(
+    channel_type: str,
+    target: str,
+    *,
+    notification_ssrf_enabled: bool = True,
+) -> str:
+    """Validate and return the cleaned notification target.
+
+    Args:
+        channel_type: ``"webhook"`` or ``"email"`` (string sentinel, not enum).
+        target:        The target URL or email address.
+        notification_ssrf_enabled:
+            Pass ``False`` to skip SSRF checks for webhook targets.
+            Defaults to ``True`` (mirrors the application default).
+
+    Returns:
+        The stripped target string on success.
+
+    Raises:
+        ValueError: When the target is blank or fails validation.
+    """
+    cleaned = target.strip()
+    if not cleaned:
+        raise ValueError("Notification target is required")
+
+    if channel_type == WEBHOOK_TYPE:
+        is_valid, error = validate_webhook_url(cleaned)
+        if not is_valid:
+            raise ValueError(error or "Invalid webhook URL")
+
+        if notification_ssrf_enabled:
+            from .validation import resolve_and_validate_target, validate_webhook_target
+            ssrf_ok, ssrf_err = resolve_and_validate_target(cleaned)
+            if not ssrf_ok:
+                raise ValueError(f"Webhook target blocked by SSRF protection: {ssrf_err}")
+            target_ok, target_err = validate_webhook_target(cleaned)
+            if not target_ok:
+                raise ValueError(f"Webhook target blocked by SSRF protection: {target_err}")
+        return cleaned
+
+    # email channel
+    return validate_email_format(cleaned)

testing/backend/unit/test_routes_email_validation.py

@@ -0,0 +1,141 @@
+"""
+Unit tests for validate_notification_target in backend.secuscan.routes_email_validation.
+
+Tests the notification-target validation path for both webhook and email channels.
+"""
+
+import pytest
+from backend.secuscan.routes_email_validation import (
+    validate_notification_target,
+    validate_email_format,
+    WEBHOOK_TYPE,
+    EMAIL_TYPE,
+)
+
+
+# ---------------------------------------------------------------------------
+# validate_email_format
+# ---------------------------------------------------------------------------
+
+class TestValidateEmailFormat:
+    def test_valid_address_stripped(self):
+        result = validate_email_format("  Test@Example.COM  ")
+        assert result == "Test@Example.COM"
+
+    def test_valid_simple_address(self):
+        assert validate_email_format("alice@example.org") == "alice@example.org"
+
+    def test_missing_at_symbol_raises(self):
+        with pytest.raises(ValueError, match="Invalid email address"):
+            validate_email_format("notanemail")
+
+    def test_missing_domain_raises(self):
+        with pytest.raises(ValueError, match="Invalid email address"):
+            validate_email_format("user@")
+
+    def test_empty_string_raises(self):
+        with pytest.raises(ValueError, match="Email address is required"):
+            validate_email_format("")
+
+    def test_whitespace_only_raises(self):
+        with pytest.raises(ValueError, match="Email address is required"):
+            validate_email_format("   ")
+
+    def test_leading_trailing_spaces_stripped(self):
+        assert validate_email_format("  bob@example.com  ") == "bob@example.com"
+
+
+# ---------------------------------------------------------------------------
+# validate_notification_target (webhook channel)
+# ---------------------------------------------------------------------------
+
+class TestValidateNotificationTargetWebhook:
+    def test_valid_http_url(self, monkeypatch):
+        monkeypatch.setattr(
+            "backend.secuscan.routes_email_validation.validate_webhook_url",
+            lambda url: (True, None),
+        )
+        result = validate_notification_target(WEBHOOK_TYPE, "http://example.com/hook")
+        assert result == "http://example.com/hook"
+
+    def test_valid_https_url(self, monkeypatch):
+        monkeypatch.setattr(
+            "backend.secuscan.routes_email_validation.validate_webhook_url",
+            lambda url: (True, None),
+        )
+        result = validate_notification_target(WEBHOOK_TYPE, "  https://discord.com/api/webhooks/1/abc  ")
+        assert result == "https://discord.com/api/webhooks/1/abc"
+
+    def test_invalid_url_raises(self, monkeypatch):
+        monkeypatch.setattr(
+            "backend.secuscan.routes_email_validation.validate_webhook_url",
+            lambda url: (False, "URL must start with http or https"),
+        )
+        with pytest.raises(ValueError, match="URL must start with http or https"):
+            validate_notification_target(WEBHOOK_TYPE, "ftp://evil.com")
+
+    def test_empty_target_raises(self, monkeypatch):
+        with pytest.raises(ValueError, match="Notification target is required"):
+            validate_notification_target(WEBHOOK_TYPE, "")
+
+    def test_whitespace_target_raises(self, monkeypatch):
+        with pytest.raises(ValueError, match="Notification target is required"):
+            validate_notification_target(WEBHOOK_TYPE, "   ")
+
+    def test_ssrf_blocked_raises(self, monkeypatch):
+        monkeypatch.setattr(
+            "backend.secuscan.routes_email_validation.validate_webhook_url",
+            lambda url: (True, None),
+        )
+        monkeypatch.setattr(
+            "backend.secuscan.validation.resolve_and_validate_target",
+            lambda url: (False, "Resolved to private IP 10.0.0.1"),
+        )
+        monkeypatch.setattr(
+            "backend.secuscan.validation.validate_webhook_target",
+            lambda url: (True, None),
+        )
+        with pytest.raises(ValueError, match="SSRF protection"):
+            validate_notification_target(WEBHOOK_TYPE, "https://evil.com/hook")
+
+    def test_ssrf_passes_when_disabled(self, monkeypatch):
+        monkeypatch.setattr(
+            "backend.secuscan.routes_email_validation.validate_webhook_url",
+            lambda url: (True, None),
+        )
+        # Even if SSRF would block, it should be skipped
+        monkeypatch.setattr(
+            "backend.secuscan.validation.resolve_and_validate_target",
+            lambda url: (False, "Blocked"),
+        )
+        result = validate_notification_target(
+            WEBHOOK_TYPE, "https://public.example.com/hook",
+            notification_ssrf_enabled=False,
+        )
+        assert result == "https://public.example.com/hook"
+
+
+# ---------------------------------------------------------------------------
+# validate_notification_target (email channel)
+# ---------------------------------------------------------------------------
+
+class TestValidateNotificationTargetEmail:
+    def test_valid_email(self):
+        result = validate_notification_target(EMAIL_TYPE, "alice@example.org")
+        assert result == "alice@example.org"
+
+    def test_email_stripped(self):
+        result = validate_notification_target(EMAIL_TYPE, "  bob@example.org  ")
+        assert result == "bob@example.org"
+
+    def test_invalid_email_raises(self):
+        with pytest.raises(ValueError, match="Invalid email address"):
+            validate_notification_target(EMAIL_TYPE, "not-an-email")
+
+    def test_empty_email_raises(self):
+        with pytest.raises(ValueError, match="Notification target is required"):
+            validate_notification_target(EMAIL_TYPE, "")
+
+    def test_whitespace_email_raises(self):
+        with pytest.raises(ValueError, match="Notification target is required"):
+            validate_notification_target(EMAIL_TYPE, "  ")