Add nginx tests and other related improvements

pyproject.toml

@@ -6,6 +6,9 @@ build-backend = "setuptools.build_meta"
 [tool.pytest.ini_options]
 DJANGO_SETTINGS_MODULE = "vulnerablecode.settings"
 
+# Fail tests that render templates which make use of invalid template variables.
+FAIL_INVALID_TEMPLATE_VARS = true
+
 markers = [
     "webtest",
 ]
@@ -85,3 +88,4 @@ profile = "black"
 line_length = 100
 force_single_line = true
 skip_gitignore = true
+skip_glob = "*/migrations/*"

setup.cfg

@@ -99,6 +99,8 @@ dev =
 	# misc
     docker-compose
     ipython==8.0.1
+    # used for testing
+    commoncode
 
 [options.entry_points]
 console_scripts =

vulnerabilities/api.py

@@ -147,7 +147,7 @@ def bulk_search(self, request):
             try:
                 purl_string = purl
                 purl = PackageURL.from_string(purl).to_dict()
-            except ValueError as ve:
+            except ValueError:
                 return Response(status=400, data={"Error": f"Invalid Package URL: {purl}"})
             purl_data = Package.objects.filter(
                 **{key: value for key, value in purl.items() if value}

vulnerabilities/helpers.py

@@ -24,6 +24,7 @@
 import dataclasses
 import json
 import logging
+import os
 import re
 from functools import total_ordering
 from typing import List
@@ -38,7 +39,7 @@
 from packageurl import PackageURL
 from univers.version_range import RANGE_CLASS_BY_SCHEMES
 
-LOGGER = logging.getLogger(__name__)
+logger = logging.getLogger(__name__)
 
 cve_regex = re.compile(r"CVE-\d{4}-\d{4,7}", re.IGNORECASE)
 is_cve = cve_regex.match
@@ -75,39 +76,23 @@ def fetch_yaml(url):
 create_etag = MagicMock()
 
 
-def split_markdown_front_matter(lines: str) -> Tuple[str, str]:
+def split_markdown_front_matter(text: str) -> Tuple[str, str]:
     """
-    This function splits lines into markdown front matter and the markdown body
-    and returns list of lines for both
-
-    for example :
-        lines =
-        ---
-        title: ISTIO-SECURITY-2019-001
-        description: Incorrect access control.
-        cves: [CVE-2019-12243]
-        ---
-        # Markdown starts here
-
-    split_markdown_front_matter(lines) would return
-    ['title: ISTIO-SECURITY-2019-001','description: Incorrect access control.'
-    ,'cves: [CVE-2019-12243]'],
-    ["# Markdown starts here"]
+    Return a tuple of (front matter, markdown body) strings split from a
+    ``text`` string. Each can be an empty string. This is used when security
+    advisories are provided in this format.
     """
+    lines = text.splitlines()
+    if not lines:
+        return "", ""
 
-    fmlines = []
-    mdlines = []
-    splitter = mdlines
-
-    for index, line in enumerate(lines.split("\n")):
-        if index == 0 and line.strip().startswith("---"):
-            splitter = fmlines
-        elif line.strip().startswith("---"):
-            splitter = mdlines
-        else:
-            splitter.append(line)
+    if lines[0] == "---":
+        lines = lines[1:]
+        text = "\n".join(lines)
+        frontmatter, _, markdown = text.partition("\n---\n")
+        return frontmatter, markdown
 
-    return "\n".join(fmlines), "\n".join(mdlines)
+    return "", text
 
 
 def contains_alpha(string):
@@ -123,7 +108,7 @@ def requests_with_5xx_retry(max_retries=5, backoff_factor=0.5):
     Returns a requests sessions which retries on 5xx errors with
     a backoff_factor
     """
-    retries = urllib3.util.Retry(
+    retries = urllib3.Retry(
         total=max_retries,
         backoff_factor=backoff_factor,
         raise_on_status=True,
@@ -157,6 +142,30 @@ def __lt__(self, other):
         return self.version < other.version
 
 
+def evolve_purl(purl, **kwargs):
+    """
+    Return a new PackageURL derived from the ``purl`` PackageURL where any of
+    the provided kwarg replaces the corresponding attribute of this PackageURL.
+    Qaulifiers if provided must be a mapping
+    For example::
+    >>> purl = PackageURL.from_string("pkg:generic/this@1.2.3")
+    >>> evolved = PackageURL.from_string("pkg:npm/@baz/that@2.2.3?foo=bar")
+    >>> evolve_purl(purl,
+    ...   type="npm", namespace="@baz", name="that",
+    ...   version="2.2.3", qualifiers={"foo": "bar"}
+    ... ) == evolved
+    True
+
+    """
+    if not kwargs:
+        return PackageURL.from_string(str(purl))
+
+    kwargs = {name: value for name, value in kwargs.items() if hasattr(purl, name)}
+    merged = purl.to_dict()
+    merged.update(kwargs)
+    return PackageURL(**merged)
+
+
 def nearest_patched_package(
     vulnerable_packages: List[PackageURL], resolved_packages: List[PackageURL]
 ) -> List[AffectedPackage]:
@@ -186,32 +195,6 @@ def nearest_patched_package(
     return affected_package_with_patched_package_objects
 
 
-def split_markdown_front_matter(text: str) -> Tuple[str, str]:
-    r"""
-    Return a tuple of (front matter, markdown body) strings split from ``text``.
-    Each can be an empty string.
-
-    >>> text='''---
-    ... title: DUMMY-SECURITY-2019-001
-    ... description: Incorrect access control.
-    ... cves: [CVE-2042-1337]
-    ... ---
-    ... # Markdown starts here
-    ... '''
-    >>> split_markdown_front_matter(text)
-    ('title: DUMMY-SECURITY-2019-001\ndescription: Incorrect access control.\ncves: [CVE-2042-1337]', '# Markdown starts here')
-    """
-    # The doctest contains \n and for the sake of clarity I chose raw strings than escaping those.
-    lines = text.splitlines()
-    if lines[0] == "---":
-        lines = lines[1:]
-        text = "\n".join(lines)
-        frontmatter, _, markdown = text.partition("\n---\n")
-        return frontmatter, markdown
-
-    return "", text
-
-
 # TODO: Replace this with combination of @classmethod and @property after upgrading to python 3.9
 class classproperty(object):
     def __init__(self, fget):
@@ -233,12 +216,54 @@ def get_item(object: dict, *attributes):
     >>> assert(get_item({'a': {'b': {'c': 'd'}}}, 'a', 'b', 'e')) == None
     """
     if not object:
-        LOGGER.error(f"Object is empty: {object}")
         return
     item = object
     for attribute in attributes:
         if attribute not in item:
-            LOGGER.error(f"Missing attribute {attribute} in {item}")
+            logger.error(f"Missing attribute {attribute} in {item}")
             return None
         item = item[attribute]
     return item
+
+
+class GitHubTokenError(Exception):
+    pass
+
+
+class GraphQLError(Exception):
+    pass
+
+
+def fetch_github_graphql_query(graphql_query: dict):
+    """
+    Return results from calling the Github graphql API with the ``graphql_query`` mapping.
+    Raise a GitHubTokenError if the "GH_TOKEN" environment variable is not set.
+    Raise a GraphQLError on query errors.
+    """
+    gh_token = os.environ.get("GH_TOKEN", None)
+    # graphql api cannot work without api token
+    if not gh_token:
+        msg = "Cannot call GitHub API without a token set in the GH_TOKEN environment variable."
+        logger.error(msg)
+        raise GitHubTokenError(msg)
+
+    response = _get_gh_response(gh_token=gh_token, graphql_query=graphql_query)
+
+    message = response.get("message")
+    if message and message == "Bad credentials":
+        raise GitHubTokenError(f"Invalid GitHub token: {message}")
+
+    errors = response.get("errors")
+    if errors:
+        raise GraphQLError(errors)
+
+    return response
+
+
+def _get_gh_response(gh_token, graphql_query):
+    """
+    Convenience function to easy mocking in tests
+    """
+    endpoint = "https://api.github.com/graphql"
+    headers = {"Authorization": f"bearer {gh_token}"}
+    return requests.post(endpoint, headers=headers, json=graphql_query).json()

vulnerabilities/import_runner.py

@@ -21,14 +21,11 @@
 #  VulnerableCode is a free software tool from nexB Inc. and others.
 #  Visit https://github.com/nexB/vulnerablecode/ for support and download.
 
-import dataclasses
 import datetime
-import json
 import logging
 from typing import Iterable
 from typing import List
 
-from vulnerabilities import models
 from vulnerabilities.importer import AdvisoryData
 from vulnerabilities.importer import Importer
 from vulnerabilities.models import Advisory

vulnerabilities/importer.py

@@ -44,6 +44,7 @@
 from univers.versions import Version
 
 from vulnerabilities.helpers import classproperty
+from vulnerabilities.helpers import evolve_purl
 from vulnerabilities.helpers import nearest_patched_package
 from vulnerabilities.oval_parser import OvalParser
 from vulnerabilities.severity_systems import SCORING_SYSTEMS
@@ -58,9 +59,6 @@ class VulnerabilitySeverity:
     value: str
 
     def to_dict(self):
-        """
-        Return a serializable dict that can be converted back using self.from_dict
-        """
         return {
             "system": self.system.identifier,
             "value": self.value,
@@ -69,7 +67,8 @@ def to_dict(self):
     @classmethod
     def from_dict(cls, severity: dict):
         """
-        Return a VulnerabilitySeverity object from dict generated by self.to_dict
+        Return a VulnerabilitySeverity object from a ``severity`` mapping of
+        VulnerabilitySeverity data.
         """
         return cls(system=SCORING_SYSTEMS[severity["system"]], value=severity["value"])
 
@@ -90,9 +89,6 @@ def normalized(self):
         return Reference(reference_id=self.reference_id, url=self.url, severities=severities)
 
     def to_dict(self):
-        """
-        Return a serializable dict that can be converted back using self.from_dict
-        """
         return {
             "reference_id": self.reference_id,
             "url": self.url,
@@ -101,9 +97,6 @@ def to_dict(self):
 
     @classmethod
     def from_dict(cls, ref: dict):
-        """
-        Return a Reference object from dict generated by self.to_dict
-        """
         return cls(
             reference_id=ref["reference_id"],
             url=ref["url"],
@@ -128,9 +121,9 @@ class NoAffectedPackages(Exception):
 @dataclasses.dataclass(order=True, frozen=True)
 class AffectedPackage:
     """
-    Contains a range of affected versions and a fixed version of a given package
-    The PackageURL supplied must *not* have a version
-    It must contain either `affected_version_range` or `fixed_version`
+    Relate a Package URL with a range of affected versions and a fixed version.
+    The Package URL must *not* have a version.
+    AffectedPackage must contain either ``affected_version_range`` or ``fixed_version``.
     """
 
     package: PackageURL
@@ -139,19 +132,21 @@ class AffectedPackage:
 
     def __post_init__(self):
         if self.package.version:
-            raise ValueError("The PackageURL supplied must not have a version")
+            raise ValueError(f"Affected Package URL {self.package!r} cannot have a version.")
+
         if not (self.affected_version_range or self.fixed_version):
             raise ValueError(
-                "Affected Package should at least have either a fixed version or affected version range"
+                f"Affected Package {self.package!r} should have either a fixed version or an "
+                "affected version range."
             )
 
     def get_fixed_purl(self):
         """
-        Return PackageURL corresponding to object's fixed_version
+        Return a Package URL corresponding to object's fixed_version
         """
         if not self.fixed_version:
-            raise ValueError("Affected package should have a fixed version")
-        fixed_purl = self.package._replace(version=str(self.fixed_version))
+            raise ValueError(f"Affected Package {self.package!r} does not have a fixed version")
+        fixed_purl = evolve_purl(purl=self.package, version=str(self.fixed_version))
         return fixed_purl
 
     @classmethod
@@ -256,6 +251,20 @@ def to_dict(self):
             "date_published": self.date_published.isoformat() if self.date_published else None,
         }
 
+    @classmethod
+    def from_dict(cls, advisory_data):
+        date_published = advisory_data["date_published"]
+        transformed = {
+            "aliases": advisory_data["aliases"],
+            "summary": advisory_data["summary"],
+            "affected_packages": [
+                AffectedPackage.from_dict(pkg) for pkg in advisory_data["affected_packages"]
+            ],
+            "references": [Reference.from_dict(ref) for ref in advisory_data["references"]],
+            "date_published": date_published.isoformat() if date_published else None,
+        }
+        return cls(**transformed)
+
 
 class NoLicenseError(Exception):
     pass
@@ -592,7 +601,7 @@ def get_data_from_xml_doc(self, xml_doc: ET.ElementTree, pkg_metadata={}) -> Lis
                     version_class = version_class_by_package_type[pkg_metadata["type"]]
                     version_scheme = version_class.scheme
 
-                    affected_version_range = VersionSpecifier.from_scheme_version_spec_string(
+                    affected_version_range = VersionRange.from_scheme_version_spec_string(
                         version_scheme, affected_version_range
                     )
                     all_versions = self.pkg_manager_api.get(package_name).valid_versions
@@ -623,7 +632,7 @@ def get_data_from_xml_doc(self, xml_doc: ET.ElementTree, pkg_metadata={}) -> Lis
                     )
 
             all_adv.append(
-                Advisory(
+                AdvisoryData(
                     summary=description,
                     affected_packages=affected_packages,
                     vulnerability_id=vuln_id,