Improve typosquatting top-packages sources for npm and PyPI

guarddog/analyzer/metadata/npm/typosquatting.py

@@ -1,48 +1,81 @@
+import json
+import os
+import time
 from typing import Optional
 
+import requests
+
 from guarddog.analyzer.metadata.typosquatting import TyposquatDetector
+from guarddog.utils.config import TOP_PACKAGES_CACHE_LOCATION
+
+_NPMS_URL = "https://api.npms.io/v2/search?q=not:unstable&size=250&from={offset}"
+_TOP_N = 10000
+_PAGE_SIZE = 250
+_CACHE_FILE = "top_npm_packages.json"
+_REFRESH_DAYS = 30
 
 
 class NPMTyposquatDetector(TyposquatDetector):
-    """Detector for typosquatting attacks. Detects if a package name is a typosquat of one of the top 5000 packages.
+    """Detector for typosquatting attacks. Detects if a package name is a typosquat of one of the top 10000 packages.
     Checks for distance one Levenshtein, one-off character swaps, permutations
     around hyphens, and substrings.
 
     Attributes:
-        popular_packages (set): set of top 5000 most popular packages from npm
+        popular_packages (set): set of top 10k most popular packages from npm
     """
 
     def _get_top_packages(self) -> set:
-        """
-        Gets the top 8000 most popular NPM packages.
-        Uses the base class implementation with NPM-specific parameters.
-        """
-        return self._get_top_packages_with_refresh(
-            packages_filename="top_npm_packages.json",
-            popular_packages_url="https://github.com/LeoDog896/npm-rank/releases/download/latest/raw.json",
-            refresh_days=30,
+        resources_dir = TOP_PACKAGES_CACHE_LOCATION or os.path.abspath(
+            os.path.join(os.path.dirname(__file__), "../resources")
         )
+        cache_path = os.path.join(resources_dir, _CACHE_FILE)
+        cache = self._load_cache_file(cache_path)
+
+        if not self._cache_is_expired(cache, days=_REFRESH_DAYS):
+            packages = (cache or {}).get("packages") or []
+            return set(packages)
+
+        packages = self._fetch_from_npms()
+        if packages:
+            with open(cache_path, "w+") as f:
+                json.dump(
+                    {"downloaded_timestamp": int(time.time()), "packages": packages},
+                    f,
+                    ensure_ascii=False,
+                    indent=4,
+                )
+            return set(packages)
+
+        # Fall back to stale cache rather than returning empty
+        packages = (cache or {}).get("packages") or []
+        return set(packages)
+
+    def _fetch_from_npms(self) -> list:
+        packages = []
+        for offset in range(0, _TOP_N, _PAGE_SIZE):
+            try:
+                resp = requests.get(
+                    _NPMS_URL.format(offset=offset),
+                    headers={"User-Agent": "guarddog/1.0"},
+                    timeout=30,
+                )
+                resp.raise_for_status()
+                data = resp.json()
+            except Exception:
+                break
+            batch = [obj["package"]["name"] for obj in data.get("results", [])]
+            packages.extend(batch)
+            if len(batch) < _PAGE_SIZE:
+                break
+        return packages[:_TOP_N]
 
     def _extract_package_names(self, data: dict | list | None) -> list | None:
-        """
-        Extract package names from NPM data structure.
-
-        Network response format: [{"name": "package-name", ...}, ...]
-        Local file format: ["package-name", "package-name", ...]
-
-        This method handles both formats and limits to top 8000 packages.
-        """
         if data is None:
             return None
 
-        # If data is already a list of strings (local file format)
-        if isinstance(data, list) and len(data) > 0:
-            if isinstance(data[0], str):
-                return data
-
-            # If data is list of dicts (network response format)
-            if isinstance(data[0], dict) and "name" in data[0]:
-                return [item["name"] for item in data[0:8000]]
+        # Local cache format: list of strings
+        if isinstance(data, list) and len(data) > 0 and isinstance(data[0], str):
+            return data
 
         return None
 

guarddog/analyzer/metadata/pypi/typosquatting.py

@@ -1,47 +1,90 @@
+import json
 import logging
+import os
+import time
+from datetime import datetime, timedelta
 from typing import Optional
 
 import packaging.utils
+import requests
 
 from guarddog.analyzer.metadata.typosquatting import TyposquatDetector
+from guarddog.utils.config import TOP_PACKAGES_CACHE_LOCATION
 
 log = logging.getLogger("guarddog")
 
+_CLICKHOUSE_URL = "https://sql-clickhouse.clickhouse.com"
+_TOP_N = 10000
+_CACHE_FILE = "top_pypi_packages.json"
+_REFRESH_DAYS = 30
+
 
 class PypiTyposquatDetector(TyposquatDetector):
     """
-    Detector for typosquatting attacks. Detects if a package name is a typosquat of one of the top 1000 packages.
+    Detector for typosquatting attacks. Detects if a package name is a typosquat of one of the top 10000 packages.
     Checks for distance one Levenshtein, one-off character swaps, permutations
     around hyphens, and substrings.
 
     Attributes:
-        popular_packages (list): list of top 5000 downloaded packages from PyPI
+        popular_packages (list): list of top 10k downloaded packages from PyPI
     """
 
     def _get_top_packages(self) -> set:
-        """
-        Gets the package information of the top 5000 most downloaded PyPI packages.
-        Uses the base class implementation with PyPI-specific parameters.
-        """
-        packages = self._get_top_packages_with_refresh(
-            packages_filename="top_pypi_packages.json",
-            popular_packages_url="https://hugovk.github.io/top-pypi-packages/top-pypi-packages.min.json",
-            refresh_days=30,
+        resources_dir = TOP_PACKAGES_CACHE_LOCATION or os.path.abspath(
+            os.path.join(os.path.dirname(__file__), "../resources")
         )
-
-        # Apply canonicalization to PyPI package names
+        cache_path = os.path.join(resources_dir, _CACHE_FILE)
+        cache = self._load_cache_file(cache_path)
+
+        if not self._cache_is_expired(cache, days=_REFRESH_DAYS):
+            packages = (cache or {}).get("packages") or []
+            return set(map(self._canonicalize_name, packages))
+
+        packages = self._fetch_from_clickhouse()
+        if packages:
+            with open(cache_path, "w+") as f:
+                json.dump(
+                    {"downloaded_timestamp": int(time.time()), "packages": packages},
+                    f,
+                    ensure_ascii=False,
+                    indent=4,
+                )
+            return set(map(self._canonicalize_name, packages))
+
+        # Fall back to stale cache rather than returning empty
+        packages = (cache or {}).get("packages") or []
         return set(map(self._canonicalize_name, packages))
 
+    def _fetch_from_clickhouse(self) -> list:
+        last_month = (datetime.now() - timedelta(days=32)).strftime("%Y-%m-01")
+        sql = (
+            f"SELECT SUM(count) AS download_count, project "
+            f"FROM pypi.pypi_downloads_per_month "
+            f"WHERE month = '{last_month}' "
+            f"GROUP BY project ORDER BY download_count DESC "
+            f"LIMIT {_TOP_N}"
+        )
+        try:
+            resp = requests.post(
+                _CLICKHOUSE_URL,
+                params={"user": "demo", "default_format": "JSON"},
+                data=sql.encode("utf-8"),
+                headers={"User-Agent": "guarddog/1.0"},
+                timeout=30,
+            )
+            resp.raise_for_status()
+            return [row["project"] for row in resp.json().get("data", [])]
+        except Exception as e:
+            log.warning(f"Failed to fetch PyPI top packages from ClickHouse: {e}")
+            return []
+
     def _extract_package_names(self, data: dict | list | None) -> list | None:
-        """
-        Extract package names from PyPI data structure.
-        PyPI data has format: {"rows": [{"project": "name", "download_count": ...}, ...]}
-        """
         if data is None:
             return None
 
-        if isinstance(data, dict) and "rows" in data:
-            return [row["project"] for row in data["rows"]]
+        # Local cache format: list of strings
+        if isinstance(data, list) and len(data) > 0 and isinstance(data[0], str):
+            return data
 
         return None