feat: Accept a source for every dependency manager in the config (#340)

Author: sdn4z

README.md

@@ -222,7 +222,8 @@ dependency_file="/my/path/requirements.txt" # it can be either a string or a lis
 selector_method="first_letter"
 logging_level="debug"
 allowlist=["my_package"]
-source="https://mirror-with-trusted-dependencies.com/file.json"
+pypi_source="https://mirror-with-trusted-dependencies.com/file-pypi.json"
+npm_source="https://mirror-with-trusted-dependencies.com/file-npm.json"
 ```
 
 The file format for each reference is as follows:

src/twyn/config/config_handler.py

@@ -35,7 +35,8 @@ class TwynConfiguration:
     dependency_files: set[str]
     selector_method: str
     allowlist: set[str]
-    source: Optional[str]
+    pypi_source: Optional[str]
+    npm_source: Optional[str]
     use_cache: bool
     package_ecosystem: Optional[PackageEcosystems]
     recursive: Optional[bool]
@@ -48,7 +49,8 @@ class ReadTwynConfiguration:
     dependency_files: Optional[set[str]] = field(default_factory=set)
     selector_method: Optional[str] = None
     allowlist: set[str] = field(default_factory=set)
-    source: Optional[str] = None
+    pypi_source: Optional[str] = None
+    npm_source: Optional[str] = None
     use_cache: Optional[bool] = None
     package_ecosystem: Optional[PackageEcosystems] = None
     recursive: Optional[bool] = None
@@ -91,7 +93,6 @@ def resolve_config(
             raise InvalidSelectorMethodError(
                 f"Invalid selector_method '{final_selector_method}'. Must be one of: {valid_methods}"
             )
-
         if use_cache is not None:
             final_use_cache = use_cache
         elif read_config.use_cache is not None:
@@ -110,7 +111,8 @@ def resolve_config(
             dependency_files=dependency_files or read_config.dependency_files or set(),
             selector_method=final_selector_method,
             allowlist=read_config.allowlist,
-            source=read_config.source,
+            pypi_source=read_config.pypi_source,
+            npm_source=read_config.npm_source,
             use_cache=final_use_cache,
             package_ecosystem=package_ecosystem or read_config.package_ecosystem,
             recursive=final_recursive,
@@ -158,7 +160,8 @@ def _get_read_config(self, toml: TOMLDocument) -> ReadTwynConfiguration:
             dependency_files=dependency_file,
             selector_method=twyn_config_data.get("selector_method"),
             allowlist=allowlist,
-            source=twyn_config_data.get("source"),
+            pypi_source=twyn_config_data.get("pypi_source"),
+            npm_source=twyn_config_data.get("npm_source"),
             use_cache=twyn_config_data.get("use_cache"),
             package_ecosystem=twyn_config_data.get("package_ecosystem"),
             recursive=twyn_config_data.get("recursive"),

src/twyn/dependency_managers/dependency_manager.py

@@ -1,44 +1,7 @@
-from dataclasses import dataclass
-from pathlib import Path
-
 from twyn.dependency_managers.exceptions import NoMatchingDependencyManagerError
-from twyn.dependency_parser.parsers.constants import PACKAGE_LOCK_JSON, POETRY_LOCK, REQUIREMENTS_TXT, UV_LOCK
-from twyn.trusted_packages.references import AbstractPackageReference, TopNpmReference, TopPyPiReference
-
-
-@dataclass
-class BaseDependencyManager:
-    """Base class for all `DependencyManagers`.
-
-    It acts as a repository, linking programming languages with trusted packages sources and dependency file names.
-    """
-
-    name: str
-    trusted_packages_source: type[AbstractPackageReference]
-    dependency_files: set[str]
-
-    @classmethod
-    def matches_dependency_file(cls, dependency_file: str) -> bool:
-        return Path(dependency_file).name in cls.dependency_files
-
-    @classmethod
-    def matches_language_name(cls, name: str) -> bool:
-        return cls.name == Path(name).name.lower()
-
-
-@dataclass
-class PypiDependencyManager(BaseDependencyManager):
-    name = "pypi"
-    trusted_packages_source = TopPyPiReference
-    dependency_files = {UV_LOCK, POETRY_LOCK, REQUIREMENTS_TXT}
-
-
-@dataclass
-class NpmDependencyManager(BaseDependencyManager):
-    name = "npm"
-    trusted_packages_source = TopNpmReference
-    dependency_files = {PACKAGE_LOCK_JSON}
-
+from twyn.dependency_managers.managers.base import BaseDependencyManager
+from twyn.dependency_managers.managers.npm_dependency_manager import NpmDependencyManager
+from twyn.dependency_managers.managers.pypi_dependency_manager import PypiDependencyManager
 
 DEPENDENCY_MANAGERS: list[type[BaseDependencyManager]] = [PypiDependencyManager, NpmDependencyManager]
 PACKAGE_ECOSYSTEMS = {x.name for x in DEPENDENCY_MANAGERS}
@@ -53,6 +16,6 @@ def get_dependency_manager_from_file(dependency_file: str) -> type[BaseDependenc
 
 def get_dependency_manager_from_name(name: str) -> type[BaseDependencyManager]:
     for manager in DEPENDENCY_MANAGERS:
-        if manager.matches_language_name(name):
+        if manager.matches_ecosystem_name(name):
             return manager
     raise NoMatchingDependencyManagerError

src/twyn/dependency_managers/exceptions.py

@@ -3,3 +3,7 @@
 
 class NoMatchingDependencyManagerError(TwynError):
     """Error for when a DependencyManger cannot be retrieved based on the provided arguments."""
+
+
+class MultipleSourcesError(TwynError):
+    """Error for when more than one alternative source matches the configuration."""

src/twyn/dependency_managers/managers/base.py

@@ -1,5 +1,6 @@
 from dataclasses import dataclass
 from pathlib import Path
+from typing import Optional
 
 from twyn.trusted_packages.references.base import AbstractPackageReference
 
@@ -20,5 +21,11 @@ def matches_dependency_file(cls, dependency_file: str) -> bool:
         return Path(dependency_file).name in cls.dependency_files
 
     @classmethod
-    def matches_language_name(cls, name: str) -> bool:
+    def matches_ecosystem_name(cls, name: str) -> bool:
         return cls.name == Path(name).name.lower()
+
+    @classmethod
+    def get_alternative_source(cls, sources: dict[str, str]) -> Optional[str]:
+        match = [x for x in sources if x == cls.name]
+
+        return sources[match[0]] if match else None

src/twyn/dependency_managers/utils.py

@@ -13,7 +13,7 @@ def get_dependency_manager_from_file(dependency_file: str) -> type[BaseDependenc
 
 def get_dependency_manager_from_name(name: str) -> type[BaseDependencyManager]:
     for manager in DEPENDENCY_MANAGERS:
-        if manager.matches_language_name(name):
+        if manager.matches_ecosystem_name(name):
             return manager
     raise NoMatchingDependencyManagerError
 

src/twyn/main.py

@@ -79,7 +79,8 @@ def check_dependencies(
     if dependencies:  # Dependencies where input manually, will not read dependency files.
         return _analyze_dependencies_from_input(
             selector_method=selector_method_obj,
-            source=config.source,
+            pypi_source=config.pypi_source,
+            npm_source=config.npm_source,
             maybe_cache_handler=maybe_cache_handler,
             allowlist=config.allowlist,
             show_progress_bar=show_progress_bar,
@@ -100,7 +101,8 @@ def check_dependencies(
 
     return _analyze_packages_from_source(
         selector_method=selector_method_obj,
-        source=config.source,
+        pypi_source=config.pypi_source,
+        npm_source=config.npm_source,
         maybe_cache_handler=maybe_cache_handler,
         allowlist=config.allowlist,
         show_progress_bar=show_progress_bar,
@@ -111,7 +113,8 @@ def check_dependencies(
 def _analyze_dependencies_from_input(
     package_ecosystem: Optional[PackageEcosystems],
     selector_method: SelectorMethod,
-    source: Optional[str],
+    pypi_source: Optional[str],
+    npm_source: Optional[str],
     maybe_cache_handler: Optional[CacheHandler],
     dependencies: set[str],
     allowlist: set[str],
@@ -127,6 +130,7 @@ def _analyze_dependencies_from_input(
         raise InvalidArgumentsError("Not a valid `package_ecosystem`.")
 
     dependency_manager = get_dependency_manager_from_name(package_ecosystem)
+    source = dependency_manager.get_alternative_source({"pypi": pypi_source, "npm": npm_source})
     top_package_reference = dependency_manager.trusted_packages_source(source, maybe_cache_handler)
     trusted_packages = TrustedPackages(
         names=top_package_reference.get_packages(),
@@ -154,7 +158,8 @@ def _analyze_packages_from_source(
     selector_method: SelectorMethod,
     show_progress_bar: bool,
     dependency_files: Optional[set[str]],
-    source: Optional[str],
+    pypi_source: Optional[str],
+    npm_source: Optional[str],
     maybe_cache_handler: Optional[CacheHandler],
 ) -> TyposquatCheckResults:
     """Analyze dependencies from a dependencies file.
@@ -165,6 +170,7 @@ def _analyze_packages_from_source(
 
     dependency_managers = _get_dependency_managers_and_parsers_mapping(dependency_files)
     for dependency_manager, parsers in dependency_managers.items():
+        source = dependency_manager.get_alternative_source({"pypi": pypi_source, "npm": npm_source})
         top_package_reference = dependency_manager.trusted_packages_source(source, maybe_cache_handler)
 
         packages_from_source = top_package_reference.get_packages()

tests/config/test_config_handler.py

@@ -35,7 +35,8 @@ def test_no_enforce_file_on_non_existent_file(self, mock_is_file: Mock) -> None:
             dependency_files=set(),
             selector_method="all",
             allowlist=set(),
-            source=None,
+            pypi_source=None,
+            npm_source=None,
             use_cache=True,
             package_ecosystem=None,
             recursive=False,
@@ -61,7 +62,8 @@ def test_get_twyn_data_from_file(self, pyproject_toml_file: Path) -> None:
             dependency_files={"my_file.txt", "my_other_file.txt"},
             selector_method="all",
             allowlist={"boto4", "boto2"},
-            source=None,
+            pypi_source=None,
+            npm_source=None,
             use_cache=False,
         )
 
@@ -138,7 +140,8 @@ def test_no_load_config_from_cache(self, pyproject_toml_file: Path) -> None:
         assert config.dependency_files == set()
         assert config.use_cache is True
         assert config.selector_method == DEFAULT_SELECTOR_METHOD
-        assert config.source is None
+        assert config.pypi_source is None
+        assert config.npm_source is None
 
     def test_cannot_write_if_file_not_configured(self) -> None:
         with pytest.raises(ConfigFileNotConfiguredError, match="write operation"):

tests/dependency_managers/__init__.py

@@ -0,0 +1,24 @@
+from unittest.mock import Mock
+
+from twyn.dependency_managers.managers.base import BaseDependencyManager
+
+
+class DummyManager(BaseDependencyManager):
+    name = "pypi"
+    trusted_packages_source = Mock()
+    dependency_files = {"requirements.txt", "poetry.lock"}
+
+
+class TestDependencyManager:
+    def test_matches_dependency_file(self) -> None:
+        assert DummyManager.matches_dependency_file("requirements.txt")
+        assert DummyManager.matches_dependency_file("/some/path/poetry.lock")
+        assert not DummyManager.matches_dependency_file("setup.py")
+
+    def test_matches_language_name(self) -> None:
+        assert DummyManager.matches_ecosystem_name("pypi")
+        assert not DummyManager.matches_ecosystem_name("npm")
+
+    def test_get_alternative_source_none(self) -> None:
+        sources = {"npm": "npmjs.com"}
+        assert DummyManager.get_alternative_source(sources) is None