Fix #2122: Standardize User-Agent headers across Importers to prevent blocking

Author: NamanmeetSingh

aboutcode/federated/__init__.py

@@ -26,6 +26,8 @@
 from packageurl import normalize_subpath
 from packageurl import normalize_version
 
+from django.conf import settings
+
 __version__ = "0.1.0"
 
 """
@@ -559,7 +561,7 @@ def from_url(
             federation_name=name,
             config_filename=cls.CONFIG_FILENAME,
         )
-        headers = {"User-Agent": "AboutCode/FederatedCode"}
+        headers = {"User-Agent": settings.VC_USER_AGENT}
         response = requests.get(url=rcf_url, headers=headers)
         if not response.ok:
             raise Exception(f"Failed to fetch Federation config: {rcf_url}")

vulnerabilities/importers/apache_httpd.py

@@ -27,6 +27,7 @@
 from vulnerabilities.utils import create_weaknesses_list
 from vulnerabilities.utils import cwe_regex
 from vulnerabilities.utils import get_item
+from django.conf import settings
 
 logger = logging.getLogger(__name__)
 
@@ -41,7 +42,10 @@ class ApacheHTTPDImporter(Importer):
     def advisory_data(self):
         links = fetch_links(self.base_url)
         for link in links:
-            data = requests.get(link).json()
+            data = requests.get(
+                link,
+                headers={'User-Agent': settings.VC_USER_AGENT}
+            ).json()
             yield self.to_advisory(data)
 
     def to_advisory(self, data):
@@ -150,7 +154,10 @@ def to_version_ranges(self, versions_data, fixed_versions):
 
 def fetch_links(url):
     links = []
-    data = requests.get(url).content
+    data = requests.get(
+        url,
+        headers={'User-Agent': settings.VC_USER_AGENT}
+    ).content
     soup = BeautifulSoup(data, features="lxml")
     for tag in soup.find_all("a"):
         link = tag.get("href")

vulnerabilities/importers/apache_kafka.py

@@ -20,6 +20,7 @@
 from vulnerabilities.importer import AffectedPackage
 from vulnerabilities.importer import Importer
 from vulnerabilities.importer import Reference
+from django.conf import settings
 
 logger = logging.getLogger(__name__)
 
@@ -99,7 +100,10 @@ class ApacheKafkaImporter(Importer):
 
     @staticmethod
     def fetch_advisory_page(self):
-        page = requests.get(self.GH_PAGE_URL)
+        page = requests.get(
+            self.GH_PAGE_URL,
+            headers={'User-Agent': settings.VC_USER_AGENT}
+        )
         return page.content
 
     def advisory_data(self):

vulnerabilities/importers/apache_tomcat.py

@@ -27,6 +27,7 @@
 from vulnerabilities.importer import Reference
 from vulnerabilities.importer import VulnerabilitySeverity
 from vulnerabilities.severity_systems import APACHE_TOMCAT
+from django.conf import settings
 
 LOGGER = logging.getLogger(__name__)
 
@@ -126,15 +127,21 @@ def fetch_advisory_pages(self):
         """
         links = self.fetch_advisory_links("https://tomcat.apache.org/security")
         for page_url in links:
-            yield page_url, requests.get(page_url).content
+            yield page_url, requests.get(
+                page_url,
+                headers={'User-Agent': settings.VC_USER_AGENT}
+            ).content
 
     def fetch_advisory_links(self, url):
         """
         Yield the URLs of each Tomcat version security-related page.
         Each page link is in the form of `https://tomcat.apache.org/security-10.html`,
         for instance, for v10.
         """
-        data = requests.get(url).content
+        data = requests.get(
+            url,
+            headers={'User-Agent': settings.VC_USER_AGENT}
+        ).content
         soup = BeautifulSoup(data, features="lxml")
         for tag in soup.find_all("a"):
             link = tag.get("href")

vulnerabilities/importers/debian.py

@@ -27,6 +27,7 @@
 from vulnerabilities.utils import create_weaknesses_list
 from vulnerabilities.utils import dedupe
 from vulnerabilities.utils import get_item
+from django.conf import settings
 
 logger = logging.getLogger(__name__)
 
@@ -83,7 +84,10 @@ class DebianImporter(Importer):
     importer_name = "Debian Importer"
 
     def get_response(self):
-        response = requests.get(self.api_url)
+        response = requests.get(
+            self.api_url,
+            headers={'User-Agent': settings.VC_USER_AGENT}
+        )
         if response.status_code == 200:
             return response.json()
         raise Exception(

vulnerabilities/importers/debian_oval.py

@@ -14,7 +14,7 @@
 import requests
 
 from vulnerabilities.importer import OvalImporter
-
+from django.conf import settings
 
 class DebianOvalImporter(OvalImporter):
 
@@ -68,7 +68,10 @@ def _fetch(self):
         for release in releases:
             file_url = f"https://www.debian.org/security/oval/oval-definitions-{release}.xml.bz2"
             self.data_url = file_url
-            resp = requests.get(file_url).content
+            resp = requests.get(
+                file_url,
+                headers={'User-Agent': settings.VC_USER_AGENT}
+            ).content
             extracted = bz2.decompress(resp)
             yield (
                 {"type": "deb", "namespace": "debian", "qualifiers": {"distro": release}},

vulnerabilities/importers/gsd.py

@@ -22,6 +22,7 @@
 from vulnerabilities.importer import Reference
 from vulnerabilities.utils import build_description
 from vulnerabilities.utils import dedupe
+from django.conf import settings
 
 logger = logging.getLogger(__name__)
 
@@ -32,7 +33,10 @@ class GSDImporter:  # TODO inherit from Importer
     url = "https://codeload.github.com/cloudsecurityalliance/gsd-database/zip/refs/heads/main"
 
     def advisory_data(self) -> Iterable[AdvisoryData]:
-        response = requests.get(self.url).content
+        response = requests.get(
+            self.url,
+            headers={'User-Agent': settings.VC_USER_AGENT}
+        ).content
         with ZipFile(BytesIO(response)) as zip_file:
             for file_name in zip_file.namelist():
                 if file_name == "gsd-database-main/allowlist.json" or not file_name.endswith(

vulnerabilities/importers/mattermost.py

@@ -25,6 +25,7 @@
 from vulnerabilities.importer import Reference
 from vulnerabilities.importer import VulnerabilitySeverity
 from vulnerabilities.package_managers import GitHubTagsAPI
+from django.conf import settings
 
 SECURITY_UPDATES_URL = "https://mattermost.com/security-updates"
 MM_REPO = {
@@ -36,13 +37,13 @@
 
 class MattermostDataSource(Importer):
     def updated_advisories(self):
-        # FIXME: Change after this https://forum.mattermost.org/t/mattermost-website-returning-403-when-headers-contain-the-word-python/11412
         self.set_api()
         data = requests.get(
-            SECURITY_UPDATES_URL, headers={"user-agent": "aboutcode/vulnerablecode"}
+            SECURITY_UPDATES_URL, 
+            headers={"User-Agent": settings.VC_USER_AGENT},
         ).content
         return self.batch_advisories(self.to_advisories(data))
-
+    
     def set_api(self):
         self.version_api = GitHubTagsAPI()
         asyncio.run(

vulnerabilities/importers/openssl.py

@@ -25,6 +25,7 @@
 from vulnerabilities.importer import Reference
 from vulnerabilities.importer import VulnerabilitySeverity
 from vulnerabilities.severity_systems import SCORING_SYSTEMS
+from django.conf import settings
 
 logger = logging.getLogger(__name__)
 
@@ -36,7 +37,10 @@ class OpensslImporter(Importer):
     importer_name = "OpenSSL Importer"
 
     def fetch(self):
-        response = requests.get(url=self.url)
+        response = requests.get(
+            url=self.url,
+            headers={'User-Agent': settings.VC_USER_AGENT}
+        )
         if not response.status_code == 200:
             logger.error(f"Error while fetching {self.url}: {response.status_code}")
             return

vulnerabilities/importers/postgresql.py

@@ -21,7 +21,7 @@
 from vulnerabilities.importer import Importer
 from vulnerabilities.importer import Reference
 from vulnerabilities.importer import VulnerabilitySeverity
-
+from django.conf import settings
 
 class PostgreSQLImporter(Importer):
 
@@ -37,7 +37,10 @@ def advisory_data(self):
         while True:
             unvisited_urls = known_urls - visited_urls
             for url in unvisited_urls:
-                data = requests.get(url).content
+                data = requests.get(
+                    url,
+                    headers={'User-Agent': settings.VC_USER_AGENT}
+                ).content
                 data_by_url[url] = data
                 visited_urls.add(url)
                 known_urls.update(find_advisory_urls(data))