Add initial attestation PoC code.

omega/analyzer/worker/run-analysis-assertion.ps1

@@ -0,0 +1,58 @@
+param (
+    [Parameter(Mandatory=$true)]
+    [string]
+    $PackageUrl,
+
+    [Parameter(Mandatory=$false)]
+    [string]
+    $PreviousVersion,
+
+    [Parameter(Mandatory=$false)]
+    [string]
+    $OutputDirectoryName = "results",
+
+    [Parameter(Mandatory=$false)]
+    [string]
+    $LibrariesIOAPIKey
+)
+
+$IMAGE_TAG = "latest"
+
+# Create directory, get absolute path for Docker
+New-Item -ItemType Directory -Force -Path $OutputDirectoryName | Out-Null
+$OutputDirectoryName = (Resolve-Path $OutputDirectoryName).Path
+
+# Ensure we have a version
+if (!$PackageUrl.Contains("@")) {
+    $parts = $PackageUrl -split "/"
+    $Name = $parts[1]
+    $Type = ($parts[0] -split ":")[1]
+    $res = Invoke-WebRequest -UseBasicParsing -Uri "http://deps.dev/_/s/$Type/p/$Name"
+    $data = $res.Content | ConvertFrom-Json
+    $version = $data.version.version
+    $PackageUrl = "pkg:$Type/$Name@$version"
+    Write-Host "Normalized PackageUrl to $PackageUrl"
+}
+
+
+if ($LibrariesIOAPIKey -ne $null)
+{
+    docker run --rm -it -e "LIBRARIES_IO_API_KEY=$LibrariesIOAPIKey" --mount type=bind,source=$OutputDirectoryName,target=/opt/export openssf/omega-toolshed:$IMAGE_TAG $PackageUrl $PreviousVersion
+}
+else
+{
+    docker run --rm -it --mount type=bind,source=$OutputDirectoryName,target=/opt/export openssf/omega-toolshed:$IMAGE_TAG $PackageUrl $PreviousVersion
+}
+Write-Output "Package successfully analyzed, results in $OutputDirectoryName"
+
+# Create security review (if needed)
+Write-Output "Running assertions..."
+
+cd tools
+Get-ChildItem -Path $OutputDirectoryName -Filter '*.sarif' -Recurse | %{
+    python .\create-assertion.py --assertion NoCriticalSecurityFindingsByTool --package-url $PackageUrl --private-key private-key.pem --input_file $_.FullName > $OutputDirectoryName\assertion-no-critical-security-findings-by-tool__{$_.Name}.json
+}
+python .\create-assertion.py --assertion NoPubliclyKnownVulnerabilities --package-url $PackageUrl --private-key private-key.pem > $OutputDirectoryName\assertion-no-publicly-known-vulnerabilities.json
+cd ..
+
+Write-Output "Operation Complete"

omega/analyzer/worker/tools/assertions/__init__.py

@@ -0,0 +1,15 @@
+# From https://github.com/python/cpython/blob/main/Lib/distutils/util.py
+# This will be removed in Python 3.12, so we'll keep a copy of it.
+def strtobool(val):
+    """Convert a string representation of truth to true (1) or false (0).
+    True values are 'y', 'yes', 't', 'true', 'on', and '1'; false values
+    are 'n', 'no', 'f', 'false', 'off', and '0'.  Raises ValueError if
+    'val' is anything else.
+    """
+    val = val.lower()
+    if val in ("y", "yes", "t", "true", "on", "1"):
+        return 1
+    elif val in ("n", "no", "f", "false", "off", "0"):
+        return 0
+    else:
+        raise ValueError("invalid truth value %r" % (val,))

omega/analyzer/worker/tools/assertions/attestations.md

@@ -0,0 +1,52 @@
+# Attestations
+
+Step 1: Create a key pair
+
+```
+openssl ecparam -name prime256v1 -genkey -noout -out private-key.pem
+openssl ec -in private-key.pem -pubout -out public-key.pem
+```
+
+Step 2: Create an assertion
+
+```
+python .\create-assertion.py
+    --assertion ManualReviewAssertion
+    --package-url pkg:npm/left-pad@1.3.0
+    --assertion_pass true
+    --review_text 'This was a lot of fun.'
+    --private-key .\private-key.pem
+    --input_file ..\results\npm\left-pad\1.3.0\reference-binaries\npm-left-pad@1.3.0.tgz
+
+{
+  "_type": "https://github.com/ossf/alpha-omega/omega-analysis-toolchain/Statement/v0.1",
+  "_comment": "Generated by the Omega Analysis Toolchain",
+  "subject": [
+    {
+      "type": "https://github.com/ossf/alpha-omega/omega-analysis-toolchain/Types/PackageURL/v0.1",
+      "purl": "pkg:npm/left-pad@1.3.0",
+      "digest": {
+        "alg": "sha256",
+        "value": "hwwP4QliI6WNT4gy0Ip+ZR6i/K245od7L9wmtmLUgd0="
+      },
+      "filename": "npm-left-pad@1.3.0.tgz"
+    }
+  ],
+  "predicateType": "https://github.com/ossf/alpha-omega/omega-analysis-toolchain/Predicate/v0.1",
+  "predicate": {
+    "review_text": "This was a lot of fun."
+  },
+  "status": "pass",
+  "signature": "MEQCIG3RfjZb/LMjpwSDvapI6TJDzLS/5moghpvWLFyHwZTkAiBJNdCrNrr+rDfaJqk3uMnRCCnKkFRegjbS2sjKyAQSYg=="
+}
+```
+
+The other assertion types available are in the assertions directory (subtypes of BaseAssertion)
+
+* ManualAssertion
+* ManualReviewAssertion
+* NoPubliclyKnownVulnerabilities
+* PackageIsReproducible
+* NoCriticalSecurityFindingsByTool
+
+Please provide feedback in [#28](https://github.com/ossf/alpha-omega/issues/28).

omega/analyzer/worker/tools/assertions/base.py

@@ -0,0 +1,106 @@
+import base64
+import datetime
+import hashlib
+import json
+import logging
+import os
+
+from . import strtobool
+from .sarif_processor import SarifProcessor
+
+
+class BaseAssertion:
+    required_args = ["package_url", "input_binaries"]
+
+    def __init__(self, kwargs):
+        self.args = kwargs
+        self.assertion_timestamp = (
+            datetime.datetime.now().astimezone().replace(microsecond=0).isoformat()
+        )
+
+        for arg in self.get_all_required_args():
+            if arg not in self.args or self.args.get(arg) is None:
+                raise ValueError(f"Missing required argument {arg}")
+
+    @classmethod
+    def get_all_required_args(cls) -> list[str]:
+        """
+        Calculates required arguments from current and all parent classes.
+        """
+        return list(
+            set(
+                [getattr(base, "required_args") for base in cls.__bases__][0]
+                + cls.required_args
+            )
+        )
+
+    def base_assertion(self):
+        if self.__class__ == BaseAssertion:
+            raise NotImplementedError("base_assertion must be called on subclasses")
+
+        assertion = {
+            "_type": "https://github.com/ossf/alpha-omega/omega-analysis-toolchain/Statement/v0.1",
+            "_comment": "Generated by the Omega Analysis Toolchain",
+            "subject": [],
+            "predicateType": "https://github.com/ossf/alpha-omega/omega-analysis-toolchain/Predicate/v0.1",
+            "predicate": {},
+        }
+
+        if "input_binaries" in self.args:
+            subject = {
+                "type": "https://github.com/ossf/alpha-omega/omega-analysis-toolchain/Types/PackageURL/v0.1",
+                "purl": str(self.args["package_url"]),
+            }
+            for binary in self.args["input_binaries"].split(","):
+                try:
+                    with open(binary, "rb") as f:
+                        subject["digest"] = {
+                            "alg": "sha256",
+                            "value": base64.b64encode(
+                                hashlib.sha256(f.read()).digest()
+                            ).decode("ascii"),
+                        }
+                    subject["filename"] = os.path.basename(binary)
+                except Exception as msg:
+                    logging.warning("Error calculating digest for %s: %s", binary, msg)
+            assertion["subject"].append(subject)
+
+        else:
+            assertion["subject"].append(
+                {
+                    "type": "https://github.com/ossf/alpha-omega/omega-analysis-toolchain/Types/PackageURL/v0.1",
+                    "purl": str(self.args["package_url"]),
+                }
+            )
+        return assertion
+
+
+class BaseSARIFAssertion(BaseAssertion):
+    required_args = ["input_file"]
+
+    def __init__(self, kwargs):
+        super().__init__(kwargs)
+        self.sarif_file = kwargs["input_file"]
+
+    def enumerate_findings(self):
+        with open(self.sarif_file, "r", encoding="utf-8") as f:
+            sarif = json.load(f)
+
+        processor = SarifProcessor(sarif)
+        findings = filter(self.filter_lambda, processor.findings)
+
+        return findings
+
+    def base_assertion(self):
+        if self.__class__ == BaseSARIFAssertion:
+            raise NotImplementedError("base_assertion must be called on subclasses")
+
+        return {
+            "type": self.__class__.__name__,
+            "assertion_timestamp": self.assertion_timestamp,
+            "version": self.version,
+            "subject": {
+                "purl": str(self.args["package_url"]),
+            },
+            "predicate": {},
+        }

omega/analyzer/worker/tools/assertions/manual_assertion.py

@@ -0,0 +1,22 @@
+import json
+import logging
+
+import requests
+from packageurl import PackageURL
+
+from . import strtobool
+from .base import BaseAssertion
+
+
+class ManualAssertion(BaseAssertion):
+    required_args = ["assertion_pass", "predicate_json"]
+    version = "0.1.0"
+
+    def emit(self):
+        data = json.loads(self.args["predicate_json"])
+        is_pass = strtobool(self.args["assertion_pass"])
+
+        assertion = self.base_assertion()
+        assertion["status"] = "pass" if is_pass else "fail"
+        assertion["predicate"] = data
+        return assertion

omega/analyzer/worker/tools/assertions/manual_review_assertion.py

@@ -0,0 +1,21 @@
+import json
+import logging
+
+import requests
+from packageurl import PackageURL
+
+from .base import BaseAssertion
+
+
+class ManualReviewAssertion(BaseAssertion):
+    required_args = ["review_text", "assertion_pass"]
+    version = "0.1.0"
+
+    def emit(self):
+        review_text = self.args["review_text"]
+        is_pass = bool(self.args["assertion_pass"])
+
+        assertion = self.base_assertion()
+        assertion["status"] = "pass" if is_pass else "fail"
+        assertion["predicate"] = {"review_text": review_text}
+        return assertion

omega/analyzer/worker/tools/assertions/public_vulnerabilities.py

@@ -0,0 +1,36 @@
+import logging
+
+import requests
+from packageurl import PackageURL
+
+from .base import BaseAssertion
+
+
+class NoPubliclyKnownVulnerabilities(BaseAssertion):
+    version = "0.1.0"
+
+    def emit(self):
+        logging.info("Checking deps.dev for public vulnerabilities...")
+        package_url = PackageURL.from_string(self.args.get("package_url"))
+
+        if package_url.namespace:
+            res = requests.get(
+                f"https://deps.dev/_/s/{package_url.type}/p/{package_url.namespace}/{package_url.name}/v/{package_url.version}"
+            )
+        else:
+            res = requests.get(
+                f"https://deps.dev/_/s/{package_url.type}/p/{package_url.name}/v/{package_url.version}"
+            )
+
+        if res.status_code == 200:
+            deps_metadata = res.json()
+            vulnerabilities = deps_metadata.get("version", {}).get("advisories", [])
+
+            assertion = self.base_assertion()
+            assertion["status"] = "pass" if not len(vulnerabilities) else "fail"
+            assertion["predicate"]["public_vulnerabilities"] = len(vulnerabilities)
+            return assertion
+        else:
+            logging.warning(
+                "deps.dev returned a non-200 status code. Skipping public vulnerability check."
+            )

omega/analyzer/worker/tools/assertions/reproducible_package.py

@@ -0,0 +1,56 @@
+import json
+import logging
+import os
+import subprocess
+import tempfile
+import uuid
+
+import requests
+from packageurl import PackageURL
+
+from .base import BaseAssertion
+
+
+class PackageIsReproducible(BaseAssertion):
+    required_args = ["package_url"]
+    version = "0.1.0"
+
+    def emit(self):
+        """Checks if the package is reproducible (via oss-reproducible)."""
+        logging.info("Checking for reproducibility using oss-reproducible...")
+        result = False
+        is_error = False
+
+        output_filename = os.path.join(
+            tempfile.gettempdir(), str(uuid.uuid4()) + ".json"
+        )
+        try:
+            res = subprocess.run(
+                [
+                    "oss-reproducible",
+                    "-o",
+                    output_filename,
+                    self.args.get("package_url"),
+                ],
+                timeout=900,
+            )
+            if res.returncode == 0:
+                with open(output_filename, "r") as f:
+                    data = json.load(f)
+                    if len(data) > 0 and data[0].get("IsReproducible") == True:
+                        result = True
+        except Exception as msg:
+            logging.warning("Error running oss-reproducible: %s", msg, exc_info=True)
+            is_error = True
+
+        try:
+            os.remove(output_filename)
+        except:
+            pass
+
+        if not is_error:
+            assertion = self.base_assertion()
+            assertion["status"] = "pass" if result == True else "fail"
+            return assertion
+        else:
+            logging.warning("oss-reproducible was not run successfully.")