From 420561cbfcbdbddaa49dab98e9b2eae134c25693 Mon Sep 17 00:00:00 2001 From: mavis Date: Sun, 5 Jul 2026 21:18:59 +0000 Subject: [PATCH] feat: extract _validate_notification_target to routes_email_validation --- backend/secuscan/routes.py | 42 ++---- backend/secuscan/routes_email_validation.py | 105 +++++++++++++ .../unit/test_routes_email_validation.py | 141 ++++++++++++++++++ 3 files changed, 256 insertions(+), 32 deletions(-) create mode 100644 backend/secuscan/routes_email_validation.py create mode 100644 testing/backend/unit/test_routes_email_validation.py diff --git a/backend/secuscan/routes.py b/backend/secuscan/routes.py index 14252b25..faa209c8 100644 --- a/backend/secuscan/routes.py +++ b/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]: diff --git a/backend/secuscan/routes_email_validation.py b/backend/secuscan/routes_email_validation.py new file mode 100644 index 00000000..ab7519e1 --- /dev/null +++ b/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) diff --git a/testing/backend/unit/test_routes_email_validation.py b/testing/backend/unit/test_routes_email_validation.py new file mode 100644 index 00000000..e57b5233 --- /dev/null +++ b/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, " ")