refactor(config): move check_deliverability setting to defaults.ini

Signed-off-by: Amine <amine.raouane@enim.ac.ma>

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",
-    "email_validator >=2.2.0,<3.0.0",
+    "email-validator >=2.2.0,<3.0.0",
 ]
 keywords = []
 # https://pypi.org/classifiers/

src/macaron/config/defaults.ini

@@ -612,6 +612,9 @@ cost = 1.0
 # The path to the file that contains the list of popular packages.
 popular_packages_path =
 
+# A boolean value that determines whether to check the deliverability of the email address.
+check_deliverability = True
+
 # ==== The following sections are for source code analysis using Semgrep ====
 # rulesets: a reference to a 'ruleset' in this section refers to a Semgrep .yaml file containing one or more rules.
 # rules: a reference to a 'rule' in this section refers to an individual rule ID, specified by the '- id:' field in

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

@@ -4,9 +4,12 @@
 """The heuristic analyzer to check the email address of the package maintainers."""
 
 import logging
+import re
 
 from email_validator import EmailNotValidError, ValidatedEmail, validate_email
 
+from macaron.config.defaults import defaults
+from macaron.errors import HeuristicAnalyzerValueError
 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
@@ -18,12 +21,49 @@
 class FakeEmailAnalyzer(BaseHeuristicAnalyzer):
     """Analyze the email address of the package maintainers."""
 
+    PATTERN = re.compile(
+        r"""\b            # word‑boundary
+        [A-Za-z0-9]+      # first alpha‑numeric segment
+        (?:\.[A-Za-z0-9]+)*   # optional “.segment” repeats
+        @
+        [A-Za-z0-9]+      # domain name segment
+        (?:\.[A-Za-z0-9]+)*   # optional sub‑domains
+        \.[A-Za-z]{2,}    # top‑level domain (at least 2 letters)
+        \b""",
+        re.VERBOSE,
+    )
+
     def __init__(self) -> None:
         super().__init__(
             name="fake_email_analyzer",
             heuristic=Heuristics.FAKE_EMAIL,
             depends_on=None,
         )
+        self.check_deliverability: bool = self._load_defaults()
+
+    def _load_defaults(self) -> bool:
+        """Load the default values from defaults.ini."""
+        section_name = "heuristic.pypi"
+        if defaults.has_section(section_name):
+            section = defaults[section_name]
+            return section.getboolean("check_deliverability", fallback=True)
+        return True
+
+    def get_emails(self, email_field: str) -> list[str]:
+        """Extract emails from the given email field.
+
+        Parameters
+        ----------
+        email_field: str
+            The email field from which to extract emails.
+
+        Returns
+        -------
+        list[str]
+            A list of emails extracted from the email field.
+        """
+        emails = self.PATTERN.findall(email_field)
+        return [email.strip() for email in emails if email.strip()]
 
     def is_valid_email(self, email: str) -> ValidatedEmail | None:
         """Check if the email format is valid and the domain has MX records.
@@ -37,15 +77,10 @@ def is_valid_email(self, email: str) -> ValidatedEmail | None:
         -------
         ValidatedEmail | None
             The validated email object if the email is valid, otherwise None.
-
-        Raises
-        ------
-        HeuristicAnalyzerValueError
-            if the failure is due to DNS resolution.
         """
         emailinfo = None
         try:
-            emailinfo = validate_email(email, check_deliverability=True)
+            emailinfo = validate_email(email, check_deliverability=self.check_deliverability)
         except EmailNotValidError as err:
             err_message = f"Invalid email address: {email}. Error: {err}"
             logger.warning(err_message)
@@ -63,15 +98,10 @@ def analyze(self, pypi_package_json: PyPIPackageJsonAsset) -> tuple[HeuristicRes
         -------
         tuple[HeuristicResult, dict[str, JsonType]]:
             The result and related information collected during the analysis.
-
-        Raises
-        ------
-        HeuristicAnalyzerValueError
-            if the analysis fails.
         """
         package_json = pypi_package_json.package_json
         if not package_json.get("info", {}):
-            return HeuristicResult.SKIP, {"message": "No package info available."}
+            raise HeuristicAnalyzerValueError("No package info available.")
 
         author_email = json_extract(package_json, ["info", "author_email"], str)
         maintainer_email = json_extract(package_json, ["info", "maintainer_email"], str)
@@ -82,12 +112,17 @@ def analyze(self, pypi_package_json: PyPIPackageJsonAsset) -> tuple[HeuristicRes
         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}
+        for email_field in [author_email, maintainer_email]:
+            if email_field:
+                emails = self.get_emails(email_field)
+                if not emails:
+                    return HeuristicResult.FAIL, {"message": "no emails found in the email field"}
+
+                for email in emails:
+                    email_info = self.is_valid_email(email)
+                    if not email_info:
+                        return HeuristicResult.FAIL, {"invalid_email": email}
 
-                validated_emails.append({key: getattr(email_info, key) for key in details})
+                    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

@@ -4,12 +4,11 @@
 """Tests for the FakeEmailAnalyzer heuristic."""
 
 
-from collections.abc import Generator
-from unittest.mock import MagicMock, patch
+from unittest.mock import MagicMock
 
 import pytest
-from email_validator import EmailNotValidError
 
+from macaron.errors import HeuristicAnalyzerValueError
 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
@@ -22,20 +21,13 @@ def analyzer_() -> FakeEmailAnalyzer:
 
 
 @pytest.fixture(name="pypi_package_json_asset_mock")
-def pypi_package_json_asset_mock_fixture() -> MagicMock:
+def pypi_package_json_asset_mock_() -> MagicMock:
     """Pytest fixture for a mock PyPIPackageJsonAsset."""
     mock_asset = MagicMock(spec=PyPIPackageJsonAsset)
     mock_asset.package_json = {}
     return mock_asset
 
 
-@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
-
-
 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}}
@@ -44,99 +36,98 @@ def test_analyze_skip_no_emails_present(analyzer: FakeEmailAnalyzer, pypi_packag
     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."""
+def test_analyze_raises_error_for_missing_info_key(
+    analyzer: FakeEmailAnalyzer, pypi_package_json_asset_mock: MagicMock
+) -> None:
+    """Test the analyzer raises an error if the 'info' key is missing in the 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 package info available."
+    with pytest.raises(HeuristicAnalyzerValueError) as exc_info:
+        analyzer.analyze(pypi_package_json_asset_mock)
+    assert "No package info available." in str(exc_info.value)
 
 
-def test_analyze_fail_invalid_email(
-    analyzer: FakeEmailAnalyzer, pypi_package_json_asset_mock: MagicMock, mock_validate_email: MagicMock
+def test_analyze_fail_no_email_found_in_field(
+    analyzer: FakeEmailAnalyzer, pypi_package_json_asset_mock: MagicMock
 ) -> None:
-    """Test analyzer fails for an invalid email format."""
-    invalid_email = "invalid-email"
+    """Test the analyzer fails if an email field does not contain a parsable email address."""
+    pypi_package_json_asset_mock.package_json = {"info": {"author_email": "not an email", "maintainer_email": None}}
+    result, info = analyzer.analyze(pypi_package_json_asset_mock)
+    assert result == HeuristicResult.FAIL
+    assert info == {"message": "no emails found in the email field"}
+
+
+def test_analyze_fail_invalid_email(analyzer: FakeEmailAnalyzer, pypi_package_json_asset_mock: MagicMock) -> None:
+    """Test analyzer fails if the email field contains an invalid email format."""
+    invalid_email = "user@example"
     pypi_package_json_asset_mock.package_json = {"info": {"author_email": invalid_email, "maintainer_email": None}}
-    mock_validate_email.side_effect = EmailNotValidError("Invalid email.")
 
     result, info = analyzer.analyze(pypi_package_json_asset_mock)
-
     assert result == HeuristicResult.FAIL
-    assert info == {"email": invalid_email}
-    mock_validate_email.assert_called_once_with(invalid_email, check_deliverability=True)
+    assert info == {"message": "no emails found in the email field"}
 
 
 def test_analyze_pass_only_maintainer_email_valid(
-    analyzer: FakeEmailAnalyzer, pypi_package_json_asset_mock: MagicMock, mock_validate_email: MagicMock
+    analyzer: FakeEmailAnalyzer, pypi_package_json_asset_mock: MagicMock
 ) -> None:
-    """Test analyzer passes when only maintainer_email is present and valid."""
+    """Test the analyzer passes if only a valid maintainer_email is present and deliverability is not checked."""
     email = "maintainer@example.net"
     pypi_package_json_asset_mock.package_json = {"info": {"author_email": None, "maintainer_email": email}}
+    result, info = analyzer.analyze(pypi_package_json_asset_mock)
 
-    mock_email_info = MagicMock()
-    mock_email_info.normalized = "maintainer@example.net"
-    mock_email_info.local_part = "maintainer"
-    mock_email_info.domain = "example.net"
-    mock_validate_email.return_value = mock_email_info
+    if analyzer.check_deliverability:
+        assert result == HeuristicResult.FAIL
+        assert info == {"invalid_email": email}
+        return
 
-    result, info = analyzer.analyze(pypi_package_json_asset_mock)
     assert result == HeuristicResult.PASS
     assert info["validated_emails"] == [
         {"normalized": "maintainer@example.net", "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_validate_email: MagicMock
-) -> None:
-    """Test the analyzer passes when both emails are present and valid."""
-
-    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_validate_email.side_effect = side_effect
+def test_analyze_pass_both_emails_valid(analyzer: FakeEmailAnalyzer, pypi_package_json_asset_mock: MagicMock) -> None:
+    """Test the analyzer passes if both emails are valid and deliverability is not checked."""
+    author_email = "example@gmail.com"
+    author_local_part, author_domain = author_email.split("@")
+    maintainer_email = "maintainer@example.net"
+    maintainer_local_part, maintainer_domain = maintainer_email.split("@")
 
     pypi_package_json_asset_mock.package_json = {
-        "info": {"author_email": "author@example.com", "maintainer_email": "maintainer@example.net"}
+        "info": {"author_email": author_email, "maintainer_email": maintainer_email}
     }
     result, info = analyzer.analyze(pypi_package_json_asset_mock)
+    if analyzer.check_deliverability:
+        assert result == HeuristicResult.FAIL
+        assert info == {"invalid_email": maintainer_email}
+        return
+
     assert result == HeuristicResult.PASS
-    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": "author@example.com", "local_part": "author", "domain": "example.com"} in validated_emails
+    assert {"normalized": author_email, "local_part": author_local_part, "domain": author_domain} in validated_emails
     assert {
-        "normalized": "maintainer@example.net",
-        "local_part": "maintainer",
-        "domain": "example.net",
+        "normalized": maintainer_email,
+        "local_part": maintainer_local_part,
+        "domain": maintainer_domain,
     } 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 = "test@example.com"
-    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("test@example.com")
-    assert result == mock_validated_email
-    mock_validate_email.assert_called_once_with("test@example.com", check_deliverability=True)
-
-
-def test_is_valid_email_failure(analyzer: FakeEmailAnalyzer, mock_validate_email: MagicMock) -> None:
+def test_is_valid_email_failure(analyzer: FakeEmailAnalyzer) -> 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)
+
+
+def test_get_emails(analyzer: FakeEmailAnalyzer) -> None:
+    """Test the get_emails method."""
+    email_field = "test@example.com, another test <another@example.org>"
+    expected = ["test@example.com", "another@example.org"]
+    assert analyzer.get_emails(email_field) == expected
+
+    email_field_no_email = "this is not an email"
+    assert analyzer.get_emails(email_field_no_email) == []
+
+    email_field_empty = ""
+    assert analyzer.get_emails(email_field_empty) == []