refactor(fake-email): replace dns_resolver with email-validator for email domain validation

Signed-off-by: Amine <[email protected]>

Author: AmineRaouane

pyproject.toml

@@ -38,7 +38,7 @@ dependencies = [
     "problog >= 2.2.6,<3.0.0",
     "cryptography >=44.0.0,<45.0.0",
     "semgrep == 1.113.0",
-    "dnspython >=2.7.0,<3.0.0",
+    "email_validator >=2.2.0",
 ]
 keywords = []
 # https://pypi.org/classifiers/

src/macaron/malware_analyzer/README.md

@@ -58,8 +58,8 @@ When a heuristic fails, with `HeuristicResult.FAIL`, then that is an indicator b
     - **Dependency**: None.
 
 11. **Fake Email**
-    - **Description**:  Checks if the package maintainer or author has a suspicious or invalid email .
-    - **Rule**: Return `HeuristicResult.FAIL` if the email format is invalid or the email domain has no MX records ; otherwise, return `HeuristicResult.PASS`.
+    - **Description**:  Checks if the package maintainer or author has a suspicious or invalid email.
+    - **Rule**: Return `HeuristicResult.FAIL` if the email is invalid; otherwise, return `HeuristicResult.PASS`.
     - **Dependency**: None.
 ### Source Code Analysis with Semgrep
 **PyPI Source Code Analyzer**

src/macaron/malware_analyzer/pypi_heuristics/metadata/fake_email.py

@@ -4,12 +4,10 @@
 """The heuristic analyzer to check the email address of the package maintainers."""
 
 import logging
-import re
 
-import dns.resolver as dns_resolver
+from email_validator import EmailNotValidError, ValidatedEmail, validate_email
 
-from macaron.errors import HeuristicAnalyzerValueError
-from macaron.json_tools import JsonType
+from macaron.json_tools import JsonType, json_extract
 from macaron.malware_analyzer.pypi_heuristics.base_analyzer import BaseHeuristicAnalyzer
 from macaron.malware_analyzer.pypi_heuristics.heuristics import HeuristicResult, Heuristics
 from macaron.slsa_analyzer.package_registry.pypi_registry import PyPIPackageJsonAsset
@@ -27,7 +25,7 @@ def __init__(self) -> None:
             depends_on=None,
         )
 
-    def is_valid_email(self, email: str) -> bool:
+    def is_valid_email(self, email: str) -> ValidatedEmail | None:
         """Check if the email format is valid and the domain has MX records.
 
         Parameters
@@ -37,26 +35,21 @@ def is_valid_email(self, email: str) -> bool:
 
         Returns
         -------
-        bool:
-            ``True`` if the email address is valid, ``False`` otherwise.
+        ValidatedEmail | None
+            The validated email object if the email is valid, otherwise None.
 
         Raises
         ------
         HeuristicAnalyzerValueError
             if the failure is due to DNS resolution.
         """
-        if not re.match(r"[^@]+@[^@]+\.[^@]+", email):
-            return False
-
-        domain = email.split("@")[1]
+        emailinfo = None
         try:
-            records = dns_resolver.resolve(domain, "MX")
-            if not records:
-                return False
-        except Exception as err:
-            err_message = f"Failed to resolve domain {domain}: {err}"
-            raise HeuristicAnalyzerValueError(err_message) from err
-        return True
+            emailinfo = validate_email(email, check_deliverability=True)
+        except EmailNotValidError as err:
+            err_message = f"Invalid email address: {email}. Error: {err}"
+            logger.warning(err_message)
+        return emailinfo
 
     def analyze(self, pypi_package_json: PyPIPackageJsonAsset) -> tuple[HeuristicResult, dict[str, JsonType]]:
         """Analyze the package.
@@ -76,16 +69,22 @@ def analyze(self, pypi_package_json: PyPIPackageJsonAsset) -> tuple[HeuristicRes
         HeuristicAnalyzerValueError
             if the analysis fails.
         """
-        data = pypi_package_json.package_json
-        author_email = data.get("info", {}).get("author_email", None)
-        maintainer_email = data.get("info", {}).get("maintainer_email", None)
-        if maintainer_email is None and author_email is None:
-            message = "No maintainers are available"
-            return HeuristicResult.SKIP, {"message": message}
-
-        if author_email is not None and not self.is_valid_email(author_email):
-            return HeuristicResult.FAIL, {"email": author_email}
-        if maintainer_email is not None and not self.is_valid_email(maintainer_email):
-            return HeuristicResult.FAIL, {"email": maintainer_email}
-
-        return HeuristicResult.PASS, {}
+        package_json = pypi_package_json.package_json
+        author_email = json_extract(package_json, ["info", "author_email"], str)
+        maintainer_email = json_extract(package_json, ["info", "maintainer_email"], str)
+
+        if not author_email and not maintainer_email:
+            return HeuristicResult.SKIP, {"message": "No author or maintainer email available."}
+
+        validated_emails: list[JsonType] = []
+        details = ["normalized", "local_part", "domain"]
+
+        for email in [author_email, maintainer_email]:
+            if email:
+                email_info = self.is_valid_email(email)
+                if not email_info:
+                    return HeuristicResult.FAIL, {"email": email}
+
+                validated_emails.append({key: getattr(email_info, key) for key in details})
+
+        return HeuristicResult.PASS, {"validated_emails": validated_emails}

tests/malware_analyzer/pypi/test_fake_email.py

@@ -8,14 +8,15 @@
 from unittest.mock import MagicMock, patch
 
 import pytest
+from email_validator import EmailNotValidError
 
 from macaron.malware_analyzer.pypi_heuristics.heuristics import HeuristicResult
 from macaron.malware_analyzer.pypi_heuristics.metadata.fake_email import FakeEmailAnalyzer
 from macaron.slsa_analyzer.package_registry.pypi_registry import PyPIPackageJsonAsset
 
 
 @pytest.fixture(name="analyzer")
-def analyzer_fixture() -> FakeEmailAnalyzer:
+def analyzer_() -> FakeEmailAnalyzer:
     """Pytest fixture to create a FakeEmailAnalyzer instance."""
     return FakeEmailAnalyzer()
 
@@ -24,132 +25,118 @@ def analyzer_fixture() -> FakeEmailAnalyzer:
 def pypi_package_json_asset_mock_fixture() -> MagicMock:
     """Pytest fixture for a mock PyPIPackageJsonAsset."""
     mock_asset = MagicMock(spec=PyPIPackageJsonAsset)
-    # Default to successful download, tests can override
-    mock_asset.download = MagicMock(return_value=True)
-    # package_json should be set by each test to simulate different PyPI responses
     mock_asset.package_json = {}
     return mock_asset
 
 
-@pytest.fixture(name="mock_dns_resolve")
-def mock_dns_resolve_fixture() -> Generator[MagicMock]:
-    """General purpose mock for dns.resolver.resolve.
+@pytest.fixture(name="mock_validate_email")
+def mock_validate_email_fixture() -> Generator[MagicMock]:
+    """Patch validate_email and mock its behavior."""
+    with patch("macaron.malware_analyzer.pypi_heuristics.metadata.fake_email.validate_email") as mock:
+        yield mock
 
-    Patches where dns_resolver is imported in the module under test.
-    """
-    with patch("macaron.malware_analyzer.pypi_heuristics.metadata.fake_email.dns_resolver.resolve") as mock_resolve:
-        # Default behavior: simulate successful MX record lookup.
-        mock_mx_record = MagicMock()
-        mock_mx_record.exchange = "mail.default-domain.com"
-        mock_resolve.return_value = [mock_mx_record]
-        yield mock_resolve
 
-
-# Tests for the analyze method
 def test_analyze_skip_no_emails_present(analyzer: FakeEmailAnalyzer, pypi_package_json_asset_mock: MagicMock) -> None:
     """Test the analyzer skips if no author_email or maintainer_email is present."""
     pypi_package_json_asset_mock.package_json = {"info": {"author_email": None, "maintainer_email": None}}
     result, info = analyzer.analyze(pypi_package_json_asset_mock)
     assert result == HeuristicResult.SKIP
-    assert info["message"] == "No maintainers are available"
+    assert info["message"] == "No author or maintainer email available."
 
 
 def test_analyze_skip_no_info_key(analyzer: FakeEmailAnalyzer, pypi_package_json_asset_mock: MagicMock) -> None:
     """Test the analyzer skips if 'info' key is missing in PyPI data."""
     pypi_package_json_asset_mock.package_json = {}  # No 'info' key
     result, info = analyzer.analyze(pypi_package_json_asset_mock)
     assert result == HeuristicResult.SKIP
-    assert info["message"] == "No maintainers are available"
+    assert info["message"] == "No author or maintainer email available."
+
 
+def test_analyze_fail_invalid_email(
+    analyzer: FakeEmailAnalyzer, pypi_package_json_asset_mock: MagicMock, mock_validate_email: MagicMock
+) -> None:
+    """Test analyzer fails for an invalid email format."""
+    invalid_email = "invalid-email"
+    pypi_package_json_asset_mock.package_json = {"info": {"author_email": invalid_email, "maintainer_email": None}}
+    mock_validate_email.side_effect = EmailNotValidError("Invalid email.")
 
-def test_analyze_fail_empty_author_email(analyzer: FakeEmailAnalyzer, pypi_package_json_asset_mock: MagicMock) -> None:
-    """Test analyzer fails for empty author_email string (maintainer_email is None)."""
-    pypi_package_json_asset_mock.package_json = {"info": {"author_email": "", "maintainer_email": None}}
     result, info = analyzer.analyze(pypi_package_json_asset_mock)
+
     assert result == HeuristicResult.FAIL
-    assert info["email"] == ""
+    assert info == {"email": invalid_email}
+    mock_validate_email.assert_called_once_with(invalid_email, check_deliverability=True)
 
 
 def test_analyze_pass_only_maintainer_email_valid(
-    analyzer: FakeEmailAnalyzer, pypi_package_json_asset_mock: MagicMock, mock_dns_resolve: MagicMock
+    analyzer: FakeEmailAnalyzer, pypi_package_json_asset_mock: MagicMock, mock_validate_email: MagicMock
 ) -> None:
     """Test analyzer passes when only maintainer_email is present and valid."""
-    mock_mx_record = MagicMock()
-    mock_mx_record.exchange = "mail.example.net"
-    mock_dns_resolve.return_value = [mock_mx_record]
+    email = "[email protected]"
+    pypi_package_json_asset_mock.package_json = {"info": {"author_email": None, "maintainer_email": email}}
+
+    mock_email_info = MagicMock()
+    mock_email_info.normalized = "[email protected]"
+    mock_email_info.local_part = "maintainer"
+    mock_email_info.domain = "example.net"
+    mock_validate_email.return_value = mock_email_info
 
-    pypi_package_json_asset_mock.package_json = {
-        "info": {"author_email": None, "maintainer_email": "[email protected]"}
-    }
     result, info = analyzer.analyze(pypi_package_json_asset_mock)
     assert result == HeuristicResult.PASS
-    assert info == {}
-    mock_dns_resolve.assert_called_once_with("example.net", "MX")
+    assert info["validated_emails"] == [
+        {"normalized": "[email protected]", "local_part": "maintainer", "domain": "example.net"}
+    ]
+    mock_validate_email.assert_called_once_with(email, check_deliverability=True)
 
 
 def test_analyze_pass_both_emails_valid(
-    analyzer: FakeEmailAnalyzer, pypi_package_json_asset_mock: MagicMock, mock_dns_resolve: MagicMock
+    analyzer: FakeEmailAnalyzer, pypi_package_json_asset_mock: MagicMock, mock_validate_email: MagicMock
 ) -> None:
     """Test the analyzer passes when both emails are present and valid."""
 
-    def side_effect_dns_resolve(domain: str, record_type: str = "MX") -> list[MagicMock]:
-        mock_mx = MagicMock()
-        domains = {
-            "MX": {"example.com", "example.net"},
-        }
-        if domain not in domains.get(record_type, set()):
-            pytest.fail(f"Unexpected domain for DNS resolve: {domain}")
-        mock_mx.exchange = f"mail.{domain}"
-        return [mock_mx]
+    def side_effect(email: str, check_deliverability: bool) -> MagicMock:  # pylint: disable=unused-argument
+        local_part, domain = email.split("@")
+        mock_email_info = MagicMock()
+        mock_email_info.normalized = email
+        mock_email_info.local_part = local_part
+        mock_email_info.domain = domain
+        return mock_email_info
 
-    mock_dns_resolve.side_effect = side_effect_dns_resolve
+    mock_validate_email.side_effect = side_effect
 
     pypi_package_json_asset_mock.package_json = {
         "info": {"author_email": "[email protected]", "maintainer_email": "[email protected]"}
     }
     result, info = analyzer.analyze(pypi_package_json_asset_mock)
     assert result == HeuristicResult.PASS
-    assert info == {}
-    assert mock_dns_resolve.call_count == 2
-    mock_dns_resolve.assert_any_call("example.com", "MX")
-    mock_dns_resolve.assert_any_call("example.net", "MX")
-
-
-def test_analyze_fail_author_email_invalid_format(
-    analyzer: FakeEmailAnalyzer, pypi_package_json_asset_mock: MagicMock, mock_dns_resolve: MagicMock
-) -> None:
-    """Test analyzer fails when author_email has an invalid format."""
-    pypi_package_json_asset_mock.package_json = {
-        "info": {"author_email": "bad_email_format", "maintainer_email": "[email protected]"}
-    }
-    result, info = analyzer.analyze(pypi_package_json_asset_mock)
-    assert result == HeuristicResult.FAIL
-    assert info["email"] == "bad_email_format"
-    mock_dns_resolve.assert_not_called()  # Regex check fails before DNS lookup
-
-
-# Tests for the is_valid_email method
-def test_is_valid_email_valid_email_with_mx(analyzer: FakeEmailAnalyzer, mock_dns_resolve: MagicMock) -> None:
-    """Test is_valid_email returns True for a valid email with MX records."""
-    mock_mx_record = MagicMock()
-    mock_mx_record.exchange = "mail.example.com"
-    mock_dns_resolve.return_value = [mock_mx_record]
-    assert analyzer.is_valid_email("[email protected]") is True
-    mock_dns_resolve.assert_called_once_with("example.com", "MX")
-
-
-def test_is_valid_email_invalid_format(analyzer: FakeEmailAnalyzer, mock_dns_resolve: MagicMock) -> None:
-    """Test is_valid_email method with various invalid email formats."""
-    assert not analyzer.is_valid_email("not_an_email")
-    assert not analyzer.is_valid_email("test@")
-    assert not analyzer.is_valid_email("@example.com")
-    assert not analyzer.is_valid_email("test@example")
-    assert not analyzer.is_valid_email("")
-    mock_dns_resolve.assert_not_called()
-
-
-def test_is_valid_email_no_mx_records_returned(analyzer: FakeEmailAnalyzer, mock_dns_resolve: MagicMock) -> None:
-    """Test is_valid_email returns False if DNS resolve returns no MX records."""
-    mock_dns_resolve.return_value = []  # Simulate no MX records found
-    assert analyzer.is_valid_email("[email protected]") is False
-    mock_dns_resolve.assert_called_once_with("no-mx-domain.com", "MX")
+    assert mock_validate_email.call_count == 2
+
+    validated_emails = info.get("validated_emails")
+    assert isinstance(validated_emails, list)
+    assert len(validated_emails) == 2
+    assert {"normalized": "[email protected]", "local_part": "author", "domain": "example.com"} in validated_emails
+    assert {
+        "normalized": "[email protected]",
+        "local_part": "maintainer",
+        "domain": "example.net",
+    } in validated_emails
+
+
+def test_is_valid_email_success(analyzer: FakeEmailAnalyzer, mock_validate_email: MagicMock) -> None:
+    """Test is_valid_email returns the validation object on success."""
+    mock_validated_email = MagicMock()
+    mock_validated_email.normalized = "[email protected]"
+    mock_validated_email.local_part = "test"
+    mock_validated_email.domain = "example.com"
+
+    mock_validate_email.return_value = mock_validated_email
+    result = analyzer.is_valid_email("[email protected]")
+    assert result == mock_validated_email
+    mock_validate_email.assert_called_once_with("[email protected]", check_deliverability=True)
+
+
+def test_is_valid_email_failure(analyzer: FakeEmailAnalyzer, mock_validate_email: MagicMock) -> None:
+    """Test is_valid_email returns None on failure."""
+    mock_validate_email.side_effect = EmailNotValidError("The email address is not valid.")
+    result = analyzer.is_valid_email("invalid-email")
+    assert result is None
+    mock_validate_email.assert_called_once_with("invalid-email", check_deliverability=True)