learningcircuit--local-deep-research/tests/security/test_notification_validator.py

"""Tests for notification_validator module - notification service URL validation."""

from unittest.mock import patch
import socket

import pytest

from local_deep_research.security.notification_validator import (
    NotificationURLValidationError,
    NotificationURLValidator,
)


class TestNotificationURLValidationError:
    """Tests for NotificationURLValidationError exception."""

    def test_inherits_from_value_error(self):
        """Should inherit from ValueError."""
        assert issubclass(NotificationURLValidationError, ValueError)

    def test_can_be_raised_with_message(self):
        """Should be raisable with a message."""
        with pytest.raises(
            NotificationURLValidationError, match="test message"
        ):
            raise NotificationURLValidationError("test message")


class TestIsPrivateIP:
    """Tests for _is_private_ip static method."""

    def test_localhost_string(self):
        """Should detect 'localhost' as private."""
        assert NotificationURLValidator._is_private_ip("localhost") is True

    def test_localhost_uppercase(self):
        """Should detect 'LOCALHOST' as private (case-insensitive)."""
        assert NotificationURLValidator._is_private_ip("LOCALHOST") is True

    def test_loopback_ipv4(self):
        """Should detect 127.0.0.1 as private."""
        assert NotificationURLValidator._is_private_ip("127.0.0.1") is True

    def test_loopback_ipv6(self):
        """Should detect ::1 as private."""
        assert NotificationURLValidator._is_private_ip("::1") is True

    def test_all_zeros_ipv4(self):
        """Should detect 0.0.0.0 as private."""
        assert NotificationURLValidator._is_private_ip("0.0.0.0") is True

    def test_all_zeros_ipv6(self):
        """Should detect :: as private."""
        assert NotificationURLValidator._is_private_ip("::") is True

    def test_private_10_range(self):
        """Should detect 10.x.x.x as private."""
        assert NotificationURLValidator._is_private_ip("10.0.0.1") is True
        assert NotificationURLValidator._is_private_ip("10.255.255.255") is True

    def test_private_172_range(self):
        """Should detect 172.16-31.x.x as private."""
        assert NotificationURLValidator._is_private_ip("172.16.0.1") is True
        assert NotificationURLValidator._is_private_ip("172.31.255.255") is True

    def test_private_192_range(self):
        """Should detect 192.168.x.x as private."""
        assert NotificationURLValidator._is_private_ip("192.168.0.1") is True
        assert (
            NotificationURLValidator._is_private_ip("192.168.255.255") is True
        )

    def test_link_local_ipv4(self):
        """Should detect link-local 169.254.x.x as private."""
        assert NotificationURLValidator._is_private_ip("169.254.1.1") is True

    def test_public_ipv4(self):
        """Should not detect public IPs as private."""
        assert NotificationURLValidator._is_private_ip("8.8.8.8") is False
        assert NotificationURLValidator._is_private_ip("1.1.1.1") is False
        assert NotificationURLValidator._is_private_ip("93.184.216.34") is False

    def test_hostname_resolving_to_public_ip(self):
        """Should return False for hostnames that resolve to public IPs."""
        fake_result = [
            (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", 0)),
        ]
        with patch("socket.getaddrinfo", return_value=fake_result):
            assert (
                NotificationURLValidator._is_private_ip("example.com") is False
            )

    def test_hostname_resolving_to_private_ip(self):
        """Should return True for hostnames that resolve to private IPs."""
        fake_result = [
            (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("127.0.0.1", 0)),
        ]
        with patch("socket.getaddrinfo", return_value=fake_result):
            assert (
                NotificationURLValidator._is_private_ip("evil.example.com")
                is True
            )

    def test_hostname_dns_failure_returns_false(self):
        """Should return False when DNS resolution fails."""
        with patch(
            "socket.getaddrinfo",
            side_effect=socket.gaierror("Name not resolved"),
        ):
            assert (
                NotificationURLValidator._is_private_ip("nonexistent.invalid")
                is False
            )


class TestValidateServiceUrl:
    """Tests for validate_service_url static method."""

    def test_empty_url_rejected(self):
        """Should reject empty URLs."""
        is_valid, error = NotificationURLValidator.validate_service_url("")
        assert is_valid is False
        assert "non-empty string" in error

    def test_none_url_rejected(self):
        """Should reject None URLs."""
        is_valid, error = NotificationURLValidator.validate_service_url(None)
        assert is_valid is False
        assert "non-empty string" in error

    def test_non_string_url_rejected(self):
        """Should reject non-string URLs."""
        is_valid, error = NotificationURLValidator.validate_service_url(123)
        assert is_valid is False
        assert "non-empty string" in error

    def test_url_without_scheme_rejected(self):
        """Should reject URLs without protocol scheme."""
        is_valid, error = NotificationURLValidator.validate_service_url(
            "example.com/webhook"
        )
        assert is_valid is False
        assert "must have a protocol" in error

    def test_parse_error_does_not_leak_exception_text(self):
        """A urlparse failure must return a generic message, not the
        exception text. The validator error is surfaced to the user by the
        /api/notifications/test-url endpoint, so leaking the exception would
        expose stack-trace fragments (CWE-209, CodeQL py/stack-trace-exposure,
        alert #4775)."""
        secret_marker = "INTERNAL-PARSER-DETAIL-do-not-leak"
        with patch(
            "local_deep_research.security.notification_validator.urlparse",
            side_effect=ValueError(secret_marker),
        ):
            is_valid, error = NotificationURLValidator.validate_service_url(
                "https://example.com/webhook"
            )
        assert is_valid is False
        assert secret_marker not in error
        assert error == "Invalid URL format"

    def test_parse_error_real_input_does_not_leak(self):
        """Real-input companion to the mocked test: an unbalanced IPv6
        bracket makes the stdlib urlparse raise ``ValueError: Invalid IPv6
        URL``, which must surface as the generic message. Guards the reachable
        path against a refactor that stops calling urlparse (or a future
        CPython that stops raising) — something the mocked test cannot catch."""
        is_valid, error = NotificationURLValidator.validate_service_url(
            "http://[::1"
        )
        assert is_valid is False
        assert error == "Invalid URL format"

    def test_file_scheme_blocked(self):
        """Should block file:// scheme."""
        is_valid, error = NotificationURLValidator.validate_service_url(
            "file:///etc/passwd"
        )
        assert is_valid is False
        assert "Blocked unsafe protocol" in error
        assert "file" in error

    def test_ftp_scheme_blocked(self):
        """Should block ftp:// scheme."""
        is_valid, error = NotificationURLValidator.validate_service_url(
            "ftp://ftp.example.com"
        )
        assert is_valid is False
        assert "Blocked unsafe protocol" in error

    def test_javascript_scheme_blocked(self):
        """Should block javascript: scheme."""
        is_valid, error = NotificationURLValidator.validate_service_url(
            "javascript:alert(1)"
        )
        assert is_valid is False
        assert "Blocked unsafe protocol" in error

    def test_data_scheme_blocked(self):
        """Should block data: scheme."""
        is_valid, error = NotificationURLValidator.validate_service_url(
            "data:text/html,<script>alert(1)</script>"
        )
        assert is_valid is False
        assert "Blocked unsafe protocol" in error

    def test_unknown_scheme_rejected(self):
        """Should reject unknown/unsupported schemes."""
        is_valid, error = NotificationURLValidator.validate_service_url(
            "custom://example.com"
        )
        assert is_valid is False
        assert "Unsupported protocol" in error

    def test_https_valid(self):
        """Should accept https:// URLs to public hosts."""
        is_valid, error = NotificationURLValidator.validate_service_url(
            "https://webhook.example.com/notify"
        )
        assert is_valid is True
        assert error is None

    def test_http_valid(self):
        """Should accept http:// URLs to public hosts."""
        is_valid, error = NotificationURLValidator.validate_service_url(
            "http://webhook.example.com/notify"
        )
        assert is_valid is True
        assert error is None

    def test_discord_scheme_valid(self):
        """Should accept discord:// URLs."""
        is_valid, error = NotificationURLValidator.validate_service_url(
            "discord://webhook_id/webhook_token"
        )
        assert is_valid is True
        assert error is None

    def test_slack_scheme_valid(self):
        """Should accept slack:// URLs."""
        is_valid, error = NotificationURLValidator.validate_service_url(
            "slack://token_a/token_b/token_c"
        )
        assert is_valid is True
        assert error is None

    def test_telegram_scheme_valid(self):
        """Should accept telegram:// URLs."""
        is_valid, error = NotificationURLValidator.validate_service_url(
            "telegram://bot_token/chat_id"
        )
        assert is_valid is True
        assert error is None

    def test_mailto_scheme_valid(self):
        """Should accept mailto: URLs."""
        is_valid, error = NotificationURLValidator.validate_service_url(
            "mailto://user@example.com"
        )
        assert is_valid is True
        assert error is None

    def test_ntfy_scheme_valid(self):
        """Should accept ntfy:// URLs."""
        is_valid, error = NotificationURLValidator.validate_service_url(
            "ntfy://topic"
        )
        assert is_valid is True
        assert error is None

    def test_ntfys_scheme_valid(self):
        """Should accept ntfys:// URLs (HTTPS variant of ntfy)."""
        is_valid, error = NotificationURLValidator.validate_service_url(
            "ntfys://topic"
        )
        assert is_valid is True
        assert error is None

    def test_http_localhost_blocked(self):
        """Should block http://localhost by default."""
        is_valid, error = NotificationURLValidator.validate_service_url(
            "http://localhost:5000/webhook"
        )
        assert is_valid is False
        assert "Blocked private/internal IP" in error

    def test_http_127_blocked(self):
        """Should block http://127.0.0.1 by default."""
        is_valid, error = NotificationURLValidator.validate_service_url(
            "http://127.0.0.1/webhook"
        )
        assert is_valid is False
        assert "Blocked private/internal IP" in error

    def test_http_private_ip_blocked(self):
        """Should block http to private IPs by default."""
        is_valid, error = NotificationURLValidator.validate_service_url(
            "http://192.168.1.100/webhook"
        )
        assert is_valid is False
        assert "Blocked private/internal IP" in error

    def test_http_localhost_allowed_with_flag(self):
        """Should allow localhost when allow_private_ips=True."""
        is_valid, error = NotificationURLValidator.validate_service_url(
            "http://localhost:5000/webhook", allow_private_ips=True
        )
        assert is_valid is True
        assert error is None

    def test_http_private_ip_allowed_with_flag(self):
        """Should allow private IPs when allow_private_ips=True."""
        is_valid, error = NotificationURLValidator.validate_service_url(
            "http://192.168.1.100/webhook", allow_private_ips=True
        )
        assert is_valid is True
        assert error is None

    def test_whitespace_stripped(self):
        """Should strip whitespace from URL."""
        is_valid, error = NotificationURLValidator.validate_service_url(
            "  https://example.com/webhook  "
        )
        assert is_valid is True
        assert error is None


class TestParserDifferentialBypass:
    """
    Tests for the parser-differential SSRF bypass (GHSA-g23j-2vwm-5c25)
    in the notification flow.  The same bypass that affected
    ``ssrf_validator.validate_url`` also affected
    ``NotificationURLValidator.validate_service_url`` because both used
    ``urlparse(url).hostname`` for the SSRF check.
    """

    def test_advisory_canonical_payload_blocked(self):
        is_valid, error = NotificationURLValidator.validate_service_url(
            "http://127.0.0.1:6666\\@1.1.1.1"
        )
        assert is_valid is False
        assert error is not None

    def test_post_prepare_canonicalised_form_blocked(self):
        """Layer-2 verification on the notification flow."""
        is_valid, error = NotificationURLValidator.validate_service_url(
            "http://127.0.0.1:6666/%5C@1.1.1.1"
        )
        assert is_valid is False
        assert error is not None
        assert "127.0.0.1" in error  # Layer 2 reports the actual host

    def test_backslash_no_port(self):
        is_valid, _ = NotificationURLValidator.validate_service_url(
            "http://127.0.0.1\\@1.1.1.1"
        )
        assert is_valid is False

    def test_tab_in_url_blocked(self):
        is_valid, _ = NotificationURLValidator.validate_service_url(
            "https://example.com/path\there"
        )
        assert is_valid is False

    def test_null_byte_blocked(self):
        is_valid, _ = NotificationURLValidator.validate_service_url(
            "http://127.0.0.1\x00@1.1.1.1"
        )
        assert is_valid is False

    def test_apprise_discord_still_works(self):
        is_valid, error = NotificationURLValidator.validate_service_url(
            "discord://webhook_id/token"
        )
        assert is_valid is True
        assert error is None

    def test_apprise_slack_still_works(self):
        is_valid, error = NotificationURLValidator.validate_service_url(
            "slack://TestApp@TokenA/TokenB/TokenC"
        )
        assert is_valid is True
        assert error is None

    def test_apprise_mailto_with_credentials(self):
        is_valid, error = NotificationURLValidator.validate_service_url(
            "mailto://user:pass@smtp.gmail.com"
        )
        assert is_valid is True
        assert error is None

    def test_apprise_signal_url_accepted(self):
        """signal:// (Apprise's Signal-API-REST transport) is allowed.

        Regression test for #4006: the validator previously rejected the
        Signal scheme with "Unsupported protocol".  Apprise handles its
        own host validation for non-http schemes, so private-IP hosts
        like signal-api-rest containers on the LAN must round-trip.
        """
        is_valid, error = NotificationURLValidator.validate_service_url(
            "signal://192.168.50.20:8739/+15551234567/+15557654321"
        )
        assert is_valid is True
        assert error is None

    def test_ipv6_unspecified_blocked(self):
        """``::`` (and equivalent forms) routes to local host on Linux."""
        is_valid, _ = NotificationURLValidator.validate_service_url(
            "http://[::]/"
        )
        assert is_valid is False

    def test_ipv6_unspecified_zero_form_blocked(self):
        """``0::`` bypasses the literal-string ``::`` allow-list at
        ``_is_private_ip`` — must be caught via the ip_address normalisation
        path against ``::/128`` in BLOCKED_IP_RANGES."""
        is_valid, _ = NotificationURLValidator.validate_service_url(
            "http://[0::]/"
        )
        assert is_valid is False

    def test_ipv6_unspecified_full_form_blocked(self):
        is_valid, _ = NotificationURLValidator.validate_service_url(
            "http://[0:0:0:0:0:0:0:0]/"
        )
        assert is_valid is False


class TestCloudMetadataBlockedForPluginSchemes:
    """Plugin-scheme IMDS guard.

    Apprise translates schemes like signal://host/... into HTTP requests
    against the URL host (e.g. POST http://host/v2/send), so cloud-
    metadata IPs reached through a plugin scheme would otherwise bypass
    the IMDS protection enforced for http/https. ``validate_service_url``
    must reject them under every flag combination.
    """

    METADATA_IPS = (
        "169.254.169.254",  # AWS IMDSv1/v2, Azure, OCI, DigitalOcean
        "169.254.170.2",  # AWS ECS task metadata v3
        "169.254.170.23",  # AWS ECS task metadata v4
        "169.254.0.23",  # Tencent Cloud
        "100.100.100.200",  # AlibabaCloud
    )

    # Plugin schemes that resolve a user-supplied host into an outbound
    # HTTP request under Apprise. (Schemes whose "host" slot holds an
    # opaque token — discord, slack, telegram, pushover, teams — are
    # covered by the existing positive tests; an IP-shaped token would
    # still trip this guard, which is fine.)
    HOST_BEARING_SCHEMES = (
        "signal",
        "gotify",
        "ntfy",
        "ntfys",
        "mattermost",
        "rocketchat",
        "matrix",
        "json",
        "xml",
        "form",
    )

    @pytest.mark.parametrize("ip", METADATA_IPS)
    @pytest.mark.parametrize("scheme", HOST_BEARING_SCHEMES)
    def test_metadata_ip_blocked_by_default(self, scheme, ip):
        url = f"{scheme}://{ip}/path"
        is_valid, error = NotificationURLValidator.validate_service_url(url)
        assert is_valid is False
        assert "cloud-metadata" in error.lower()

    @pytest.mark.parametrize("ip", METADATA_IPS)
    @pytest.mark.parametrize("scheme", HOST_BEARING_SCHEMES)
    def test_metadata_ip_blocked_even_with_allow_private_ips(self, scheme, ip):
        """allow_private_ips=True unlocks LAN reach, NOT IMDS."""
        url = f"{scheme}://{ip}/path"
        is_valid, error = NotificationURLValidator.validate_service_url(
            url, allow_private_ips=True
        )
        assert is_valid is False
        assert "cloud-metadata" in error.lower()

    def test_mailto_metadata_host_blocked(self):
        """mailto://user@169.254.169.254/... must not reach IMDS."""
        is_valid, error = NotificationURLValidator.validate_service_url(
            "mailto://user:pass@169.254.169.254/recipient"
        )
        assert is_valid is False
        assert "cloud-metadata" in error.lower()

    def test_signal_lan_host_still_allowed(self):
        """LAN signal-api-rest container (#4006 use case) keeps working."""
        is_valid, error = NotificationURLValidator.validate_service_url(
            "signal://192.168.50.20:8739/+15551234567/+15557654321"
        )
        assert is_valid is True
        assert error is None

    def test_gotify_lan_host_still_allowed(self):
        is_valid, error = NotificationURLValidator.validate_service_url(
            "gotify://10.0.0.5:8080/AbCdEf123"
        )
        assert is_valid is True
        assert error is None

    def test_signal_loopback_still_allowed(self):
        """Plugin schemes pointing at localhost (same-host self-hosted
        container) round-trip without the operator opt-in — only the
        absolute IMDS block fires for plugin schemes."""
        is_valid, error = NotificationURLValidator.validate_service_url(
            "signal://127.0.0.1:8739/+15551234567/+15557654321"
        )
        assert is_valid is True
        assert error is None

    def test_plugin_scheme_token_host_unaffected(self):
        """Schemes whose 'host' slot is an opaque token (discord, slack,
        telegram, pushover, teams) keep working — the IMDS check is a
        no-op against non-IP strings."""
        for url in (
            "discord://webhook_id/token",
            "slack://TestApp@TokenA/TokenB/TokenC",
            "telegram://bottoken/ChatID",
            "pushover://user@token",
            "teams://group@token1/token2/token3",
        ):
            is_valid, error = NotificationURLValidator.validate_service_url(url)
            assert is_valid is True, f"{url} should be valid, got: {error}"

    def test_signal_metadata_hostname_via_dns_blocked(self):
        """DNS-resolved hostname pointing at IMDS is rejected — closes
        the easy ``signal://imds.attacker.example/...`` variant of the
        bypass. (The full DNS-rebinding TOCTOU window is a separately
        documented residual risk; this test only covers single-resolve
        attackers.)"""
        with patch("socket.getaddrinfo") as mock_dns:
            mock_dns.return_value = [
                (
                    socket.AF_INET,
                    socket.SOCK_STREAM,
                    0,
                    "",
                    ("169.254.169.254", 0),
                )
            ]
            is_valid, error = NotificationURLValidator.validate_service_url(
                "signal://imds.attacker.example/+15551234567/+15557654321"
            )
            assert is_valid is False
            assert "cloud-metadata" in error.lower()


class TestValidateServiceUrlStrict:
    """Tests for validate_service_url_strict static method."""

    def test_valid_url_returns_true(self):
        """Should return True for valid URLs."""
        result = NotificationURLValidator.validate_service_url_strict(
            "https://example.com/webhook"
        )
        assert result is True

    def test_invalid_url_raises_exception(self):
        """Should raise NotificationURLValidationError for invalid URLs."""
        with pytest.raises(NotificationURLValidationError) as exc_info:
            NotificationURLValidator.validate_service_url_strict(
                "file:///etc/passwd"
            )
        assert "validation failed" in str(exc_info.value)

    def test_private_ip_raises_exception(self):
        """Should raise exception for private IPs by default."""
        with pytest.raises(NotificationURLValidationError) as exc_info:
            NotificationURLValidator.validate_service_url_strict(
                "http://localhost/webhook"
            )
        assert "Blocked private/internal IP" in str(exc_info.value)

    def test_private_ip_allowed_with_flag(self):
        """Should not raise when allow_private_ips=True."""
        result = NotificationURLValidator.validate_service_url_strict(
            "http://localhost/webhook", allow_private_ips=True
        )
        assert result is True


class TestValidateMultipleUrls:
    """Tests for validate_multiple_urls static method."""

    def test_empty_urls_rejected(self):
        """Should reject empty URL string."""
        is_valid, error = NotificationURLValidator.validate_multiple_urls("")
        assert is_valid is False
        assert "non-empty string" in error

    def test_none_urls_rejected(self):
        """Should reject None."""
        is_valid, error = NotificationURLValidator.validate_multiple_urls(None)
        assert is_valid is False
        assert "non-empty string" in error

    def test_only_separators_rejected(self):
        """Should reject string with only separators."""
        is_valid, error = NotificationURLValidator.validate_multiple_urls(",,,")
        assert is_valid is False
        assert "No valid URLs found" in error

    def test_single_valid_url(self):
        """Should accept single valid URL."""
        is_valid, error = NotificationURLValidator.validate_multiple_urls(
            "https://example.com/webhook"
        )
        assert is_valid is True
        assert error is None

    def test_multiple_valid_urls(self):
        """Should accept multiple valid URLs."""
        urls = "https://example.com/webhook,discord://id/token,slack://token"
        is_valid, error = NotificationURLValidator.validate_multiple_urls(urls)
        assert is_valid is True
        assert error is None

    def test_one_invalid_url_fails_all(self):
        """Should fail if any URL is invalid."""
        urls = "https://example.com/webhook,file:///etc/passwd"
        is_valid, error = NotificationURLValidator.validate_multiple_urls(urls)
        assert is_valid is False
        assert "file" in error.lower()

    def test_whitespace_in_urls_stripped(self):
        """Should handle whitespace around URLs."""
        urls = "  https://example.com/webhook  ,  discord://id/token  "
        is_valid, error = NotificationURLValidator.validate_multiple_urls(urls)
        assert is_valid is True
        assert error is None

    def test_custom_separator(self):
        """Should support custom separator."""
        urls = "https://example.com/webhook|discord://id/token"
        is_valid, error = NotificationURLValidator.validate_multiple_urls(
            urls, separator="|"
        )
        assert is_valid is True
        assert error is None

    def test_private_ip_in_multiple_blocked(self):
        """Should block private IPs in multiple URLs."""
        urls = "https://example.com/webhook,http://localhost/webhook"
        is_valid, error = NotificationURLValidator.validate_multiple_urls(urls)
        assert is_valid is False
        assert "Blocked private/internal IP" in error

    def test_private_ip_allowed_with_flag(self):
        """Should allow private IPs when flag is set."""
        urls = "https://example.com/webhook,http://localhost/webhook"
        is_valid, error = NotificationURLValidator.validate_multiple_urls(
            urls, allow_private_ips=True
        )
        assert is_valid is True
        assert error is None


class TestClassConstants:
    """Tests for class constants."""

    def test_blocked_schemes_contains_dangerous_protocols(self):
        """BLOCKED_SCHEMES should contain dangerous protocols."""
        blocked = NotificationURLValidator.BLOCKED_SCHEMES
        assert "file" in blocked
        assert "ftp" in blocked
        assert "javascript" in blocked
        assert "data" in blocked

    def test_allowed_schemes_contains_common_services(self):
        """ALLOWED_SCHEMES should contain common notification services."""
        allowed = NotificationURLValidator.ALLOWED_SCHEMES
        assert "http" in allowed
        assert "https" in allowed
        assert "discord" in allowed
        assert "slack" in allowed
        assert "telegram" in allowed
        assert "mailto" in allowed
        assert "ntfys" in allowed
        assert "signal" in allowed

    def test_private_ip_ranges_exist(self):
        """PRIVATE_IP_RANGES should contain RFC1918 and other private ranges."""
        ranges = NotificationURLValidator.PRIVATE_IP_RANGES
        assert len(ranges) > 0
        # Check some expected ranges are present
        range_strings = [str(r) for r in ranges]
        assert "127.0.0.0/8" in range_strings
        assert "10.0.0.0/8" in range_strings
        assert "192.168.0.0/16" in range_strings


class TestNat64EnvOptOutInNotificationValidator:
    """Mirror of ssrf_validator's TestNat64EnvOptOut for the notification
    path. The notification validator must honor the same operator
    opt-in semantics AND keep the cloud-metadata block absolute."""

    def test_nat64_wkp_blocked_when_env_unset(self, monkeypatch):
        monkeypatch.delenv("LDR_SECURITY_ALLOW_NAT64", raising=False)
        # 64:ff9b::a00:1 is the NAT64 wrap of 10.0.0.1.
        assert NotificationURLValidator._is_private_ip("64:ff9b::a00:1") is True

    def test_nat64_wkp_allowed_when_env_true(self, monkeypatch):
        monkeypatch.setenv("LDR_SECURITY_ALLOW_NAT64", "true")
        # NAT64 wrap of 8.8.8.8 — canonical IPv6-only-deployment use case.
        assert (
            NotificationURLValidator._is_private_ip("64:ff9b::808:808") is False
        )

    def test_nat64_local_use_allowed_when_env_true(self, monkeypatch):
        monkeypatch.setenv("LDR_SECURITY_ALLOW_NAT64", "true")
        assert (
            NotificationURLValidator._is_private_ip("64:ff9b:1::808:808")
            is False
        )

    def test_imds_via_nat64_wkp_wrap_blocked_under_env_true(self, monkeypatch):
        """[64:ff9b::a9fe:a9fe] — NAT64 WKP wrap of 169.254.169.254.
        Must remain blocked even with the operator opt-in. Mirrors the
        ssrf_validator embedded-IPv4 IMDS check."""
        monkeypatch.setenv("LDR_SECURITY_ALLOW_NAT64", "true")
        assert (
            NotificationURLValidator._is_private_ip("64:ff9b::a9fe:a9fe")
            is True
        )

    def test_imds_via_nat64_local_use_wrap_blocked_under_env_true(
        self, monkeypatch
    ):
        """Same lock-in for the RFC 8215 local-use prefix wrap."""
        monkeypatch.setenv("LDR_SECURITY_ALLOW_NAT64", "true")
        assert (
            NotificationURLValidator._is_private_ip("64:ff9b:1::a9fe:a9fe")
            is True
        )

    def test_ecs_metadata_via_nat64_wrap_blocked_under_env_true(
        self, monkeypatch
    ):
        """169.254.170.2 = 0xa9feaa02 — AWS ECS task metadata v3."""
        monkeypatch.setenv("LDR_SECURITY_ALLOW_NAT64", "true")
        assert (
            NotificationURLValidator._is_private_ip("64:ff9b::a9fe:aa02")
            is True
        )

    def test_alibaba_metadata_via_nat64_wrap_blocked_under_env_true(
        self, monkeypatch
    ):
        """100.100.100.200 = 0x646464c8 — AlibabaCloud metadata."""
        monkeypatch.setenv("LDR_SECURITY_ALLOW_NAT64", "true")
        assert (
            NotificationURLValidator._is_private_ip("64:ff9b::6464:64c8")
            is True
        )

    def test_env_does_not_unblock_6to4_in_notification_path(self, monkeypatch):
        monkeypatch.setenv("LDR_SECURITY_ALLOW_NAT64", "true")
        assert (
            NotificationURLValidator._is_private_ip("2002:c0a8:101::") is True
        )

    def test_env_does_not_unblock_teredo_in_notification_path(
        self, monkeypatch
    ):
        monkeypatch.setenv("LDR_SECURITY_ALLOW_NAT64", "true")
        assert NotificationURLValidator._is_private_ip("2001::1") is True

    def test_imds_via_nat64_wrap_blocked_when_env_unset(self, monkeypatch):
        """Sanity: the IMDS embedded-IPv4 check fires regardless of env
        state — when env is unset, the NAT64 prefix entry already blocks
        directly, but the embedded-IPv4 path is still well-formed."""
        monkeypatch.delenv("LDR_SECURITY_ALLOW_NAT64", raising=False)
        assert (
            NotificationURLValidator._is_private_ip("64:ff9b::a9fe:a9fe")
            is True
        )

    def test_ipv4_mapped_imds_blocked(self, monkeypatch):
        """Cross-validator parity: ssrf_validator unwraps IPv4-mapped
        IPv6 (``::ffff:169.254.169.254``) before the IMDS literal check.
        notification_validator must do the same — otherwise an attacker
        who can configure a webhook URL can reach IMDS via the IPv4-
        mapped form. Pre-PR this was a real gap; locked in here so it
        cannot regress."""
        monkeypatch.delenv("LDR_SECURITY_ALLOW_NAT64", raising=False)
        assert (
            NotificationURLValidator._is_private_ip("::ffff:169.254.169.254")
            is True
        )

    def test_ipv4_mapped_loopback_blocked(self, monkeypatch):
        """Same parity check for the loopback IPv4-mapped form."""
        monkeypatch.delenv("LDR_SECURITY_ALLOW_NAT64", raising=False)
        assert (
            NotificationURLValidator._is_private_ip("::ffff:127.0.0.1") is True
        )

    def test_ipv4_mapped_public_ip_passes(self, monkeypatch):
        """Anti-collision: the unwrap must not over-block public IPv4."""
        monkeypatch.delenv("LDR_SECURITY_ALLOW_NAT64", raising=False)
        assert (
            NotificationURLValidator._is_private_ip("::ffff:8.8.8.8") is False
        )

    def test_validate_service_url_imds_blocked_under_allow_private_ips(self):
        """Round-3 audit regression: validate_service_url with
        allow_private_ips=True previously short-circuited the entire
        host check, allowing http://169.254.169.254/ through. The opt-in
        is for self-hosted webhooks on internal networks, not for IMDS
        exfiltration. ALWAYS_BLOCKED_METADATA_IPS must remain absolute."""
        is_valid, error = NotificationURLValidator.validate_service_url(
            "http://169.254.169.254/latest/meta-data/",
            allow_private_ips=True,
        )
        assert is_valid is False
        assert error is not None

    def test_validate_service_url_imds_v6_mapped_blocked_under_allow_private_ips(
        self,
    ):
        is_valid, _ = NotificationURLValidator.validate_service_url(
            "http://[::ffff:169.254.169.254]/", allow_private_ips=True
        )
        assert is_valid is False

    def test_validate_service_url_imds_via_nat64_wkp_blocked_under_allow_private_ips(
        self,
    ):
        is_valid, _ = NotificationURLValidator.validate_service_url(
            "http://[64:ff9b::a9fe:a9fe]/", allow_private_ips=True
        )
        assert is_valid is False

    def test_validate_service_url_imds_via_nat64_local_use_blocked_under_allow_private_ips(
        self,
    ):
        is_valid, _ = NotificationURLValidator.validate_service_url(
            "http://[64:ff9b:1::a9fe:a9fe]/", allow_private_ips=True
        )
        assert is_valid is False

    def test_validate_service_url_alibaba_metadata_blocked_under_allow_private_ips(
        self,
    ):
        """100.100.100.200 is in ALWAYS_BLOCKED_METADATA_IPS and ALSO in
        the CGNAT range (100.64.0.0/10) — pre-fix the carve-out for
        CGNAT under allow_private_ips=True would have leaked it."""
        is_valid, _ = NotificationURLValidator.validate_service_url(
            "http://100.100.100.200/", allow_private_ips=True
        )
        assert is_valid is False

    def test_validate_service_url_rfc1918_allowed_under_allow_private_ips(self):
        """Anti-collision: the fix must not over-block legitimate
        self-hosted webhook destinations. allow_private_ips=True is
        designed for exactly this case."""
        is_valid, _ = NotificationURLValidator.validate_service_url(
            "http://192.168.1.100/webhook", allow_private_ips=True
        )
        assert is_valid is True

    def test_validate_service_url_localhost_allowed_under_allow_private_ips(
        self,
    ):
        is_valid, _ = NotificationURLValidator.validate_service_url(
            "http://localhost:5000/webhook", allow_private_ips=True
        )
        assert is_valid is True

    def test_dns_resolved_imds_via_nat64_blocked_under_env_true(
        self, monkeypatch
    ):
        """Hostname-resolution branch: a hostname that resolves to a
        NAT64-wrapped IMDS IPv4 must still be blocked under env opt-in.
        This exercises the second call site of _ip_matches_blocked_range."""
        monkeypatch.setenv("LDR_SECURITY_ALLOW_NAT64", "true")
        # AF_INET6 result tuple: (family, type, proto, canonname, sockaddr)
        # sockaddr for IPv6 is (host, port, flowinfo, scopeid)
        with patch(
            "socket.getaddrinfo",
            return_value=[
                (
                    socket.AF_INET6,
                    socket.SOCK_STREAM,
                    6,
                    "",
                    ("64:ff9b::a9fe:a9fe", 0, 0, 0),
                )
            ],
        ):
            assert (
                NotificationURLValidator._is_private_ip("imds.attacker.example")
                is True
            )