[v1.3.0] Only trigger vuln threshold on fixable vulns (#122)

* Add --threshold-fixable-only to CLI

* implemented business logic

* changed 'threshold_fixable_only' from str to bool

* Added more test coverage and CLI refinements

* debugging failing unit test

* test threshold-fixable-only in workflow

* test threshold-fixable-only in workflow

* debugging CI/CD

* debugging CI/CD

* debugging

* debugging

* debugging

* debugging

* removed debug log showing CLI arguments

* add missing argument, fixed_vuln_counts

* simplify get_fixed_vuln_counts() return values

* refactor return types in get_scan_result()

* refactor

* refine get_fixed_vuln_counts()

* update test_get_fixed_vuln_counts()

* testing case sensitivity

* revert 'TRUE' to 'true'

* use debug log when vuln doesnt have rating

* integrate --show-only-fixable-vulns (part 1)

* integrate only show fixable vulns

* test example workflows

* fix CLI input arguments

* remove leading '-' character for conditional inclusion

* add a no-op CLI arg (workaround)

* enable new arguments in workflows

* fix failing test

* update workflows for prod

---------

Co-authored-by: Michael Long <mlongii@amazon.com>

Author: bluesentinelsec

.github/workflows/test_vuln_thresholds.yml

@@ -45,8 +45,8 @@ jobs:
           low_threshold: 1
           other_threshold: 1
           sbomgen_version: "latest"
+          threshold_fixable_only: true
+          show_only_fixable_vulns: true
 
       - name: Fail if vulnerability threshold is exceeded
         run: if [[  ${{ steps.inspector.outputs.vulnerability_threshold_exceeded }} != "1" ]]; then echo "test failed"; else echo "test passed"; fi
-
-        # TODO: handle failure case

action.yml

@@ -110,6 +110,18 @@ inputs:
     description: "Specifies the OS and CPU arch of the container image you wish to scan. Valid inputs are of the form 'os/cpu/variant' for example, 'linux/amd64', 'linux/arm64/v8', etc. If no platform is specified, the system will use the same platform as the host that is performing the scan. This argument only affects container image scans. Requires inspector-sbomgen 1.5.1 or later."
     required: False
 
+  threshold_fixable_only:
+    description: 'If set to true, only count vulnerabilities with a fix towards threshold exceeded condition.'
+    required: False
+    default: false
+    type: boolean
+
+  show_only_fixable_vulns:
+    description: "If set to true, this action will show only fixed vulnerabilities in the GitHub Actions step summary page. All vulnerability metadata is still retained in the raw Inspector scan files."
+    required: False
+    default: false
+    type: boolean
+
 outputs:
   artifact_sbom:
     description: "The filepath to the artifact's software bill of materials."
@@ -148,6 +160,8 @@ runs:
     - --out-dockerfile-scan-md=${{ inputs.output_inspector_dockerfile_scan_path_markdown }}
     - --sbomgen-version=${{ inputs.sbomgen_version }}
     - --thresholds
+    - ${{ inputs.threshold_fixable_only == 'true' && '--threshold-fixable-only' || '--no-op' }}
+    - ${{ inputs.show_only_fixable_vulns == 'true' && '--show-only-fixable-vulns'|| '--no-op' }}
     - --critical=${{ inputs.critical_threshold }}
     - --high=${{ inputs.high_threshold }}
     - --medium=${{ inputs.medium_threshold }}

entrypoint/entrypoint/cli.py

@@ -50,14 +50,19 @@ def init(sys_argv=None) -> argparse.Namespace:
                         help="Specifies one or more files and/or directories that should NOT be inventoried.")
     parser.add_argument("--timeout", type=str, default="600",
                         help="The amount of time in seconds that inspector-sbomgne will run. When this timeout is exceeded, sbomgen will gracefully conclude and present any findings discovered up to that point.")
-    parser.add_argument("--show-only-fixed-vulnerabilities", action="store_true", help="If set, this program will only show fixed vulnerabilities in the GitHub Actions job summary page.")
+    parser.add_argument("--show-only-fixable-vulns", action="store_true", default=False,
+                        help="Only show fixed vulnerabilities in the GitHub Actions job summary page.")
+    parser.add_argument("--threshold-fixable-only", action="store_true", default=False,
+                        help="Only count vulnerabilities with a fix towards threshold exceeded condition.")
 
     parser.add_argument("--platform", type=str,
                         help="Specifies the OS and CPU arch of the container image you wish to scan. Valid inputs are "
                              "of the form 'os/cpu/variant' for example, 'linux/amd64', 'linux/arm64/v8', etc. If no platform is "
                              "specified, the system will use the same platform as the host that is performing the "
                              "scan. This argument only affects container image scans. Requires inspector-sbomgen "
                              "1.5.1 or later.")
+    parser.add_argument("--no-op", action="store_true", default=False,
+                        help="A no operation argument, used as the default from the GitHub Actions caller when boolean arguments are not set. This is a workaround because GitHub Actions doesn't have a clean way to invoke or not invoke action='store_true' arguments")
 
     args = ""
     if sys_argv:

entrypoint/entrypoint/fixed_vulns.py

@@ -0,0 +1,10 @@
+from dataclasses import dataclass
+
+
+@dataclass
+class FixedVulns:
+    criticals: int
+    highs: int
+    mediums: int
+    lows: int
+    others: int

entrypoint/entrypoint/orchestrator.py

@@ -3,12 +3,13 @@
 import logging
 import os
 import platform
+import re
 import shutil
 import sys
 import tempfile
-import re
+import typing
 
-from entrypoint import dockerfile, executor, exporter, installer, pkg_vuln
+from entrypoint import dockerfile, executor, exporter, installer, pkg_vuln, fixed_vulns
 
 
 def execute(args) -> int:
@@ -26,12 +27,13 @@ def execute(args) -> int:
     set_github_actions_output('inspector_scan_results', args.out_scan)
 
     logging.info("tallying vulnerabilities")
-    succeeded, scan_result = get_scan_result(args)
+    succeeded, scan_result, fixed_vuln_counts = get_scan_result(args)
     require_true(succeeded, "unable to tally vulnerabilities")
 
     print_vuln_count_summary(scan_result)
 
-    set_flag_if_vuln_threshold_exceeded(args, scan_result)
+    vuln_counts = fixed_vuln_counts if args.threshold_fixable_only else scan_result
+    set_env_var_if_vuln_threshold_exceeded(args, vuln_counts)
 
     write_pkg_vuln_report_csv(args.out_scan_csv, scan_result)
     set_github_actions_output('inspector_scan_results_csv', args.out_scan_csv)
@@ -166,7 +168,7 @@ def invoke_sbomgen(args) -> int:
 
     elif "archive" in args.artifact_type.lower():
         args.artifact_type = "archive"
-        path_arg = "--path"        
+        path_arg = "--path"
 
     else:
         logging.error(
@@ -201,7 +203,8 @@ def invoke_sbomgen(args) -> int:
         if args.platform:
             platform_arg = args.platform.lower()
             if not is_valid_container_platform(platform_arg):
-                logging.fatal(f"received invalid container image platform: '{args.platform}'. Platform should be of the form 'os/cpu/variant' such as 'linux/amd64' or 'linux/arm64/v8'")       
+                logging.fatal(
+                    f"received invalid container image platform: '{args.platform}'. Platform should be of the form 'os/cpu/variant' such as 'linux/amd64' or 'linux/arm64/v8'")
             sbomgen_args.append("--platform")
             sbomgen_args.append(platform_arg)
 
@@ -232,40 +235,45 @@ def invoke_inspector_scan(src_sbom, dst_scan):
     return ret
 
 
-def get_scan_result(args) -> tuple[bool, exporter.InspectorScanResult]:
-    if args.show_only_fixed_vulnerabilities == True:
-        succeeded, criticals, highs, mediums, lows, others = get_fixed_vuln_counts(args.out_scan)
-        if succeeded is False:
-            return False, None
-    else:
-        succeeded, criticals, highs, mediums, lows, others = get_vuln_counts(args.out_scan)
-        if succeeded is False:
-            return False, None
+def get_scan_result(args) -> tuple[bool, exporter.InspectorScanResult, fixed_vulns.FixedVulns]:
+    scan_result = exporter.InspectorScanResult(vulnerabilities=[pkg_vuln.Vulnerability()])
+    fixed_vulns_counts = fixed_vulns.FixedVulns(criticals=0, highs=0, mediums=0, lows=0, others=0)
+
+    succeeded, fixed_vulns_counts = get_fixed_vuln_counts(
+        args.out_scan)
+    if succeeded is False:
+        return False, scan_result, fixed_vulns_counts
+
+    succeeded, criticals, highs, mediums, lows, others = get_vuln_counts(args.out_scan)
+    if succeeded is False:
+        return False, scan_result, fixed_vulns_counts
 
     try:
         with open(args.out_scan, "r") as f:
             inspector_scan = json.load(f)
             vulns = pkg_vuln.parse_inspector_scan_result(inspector_scan)
-            if args.show_only_fixed_vulnerabilities:
-                for vuln in vulns:
-                    if vuln.fixed_ver == "null":
-                        vulns.remove(vuln)
+
     except Exception as e:
         logging.error(e)
-        return False, None
-    
+        return False, scan_result, fixed_vulns_counts
+
+    if args.show_only_fixable_vulns:
+        for vuln in vulns:
+            if vuln.fixed_ver == "null":
+                vulns.remove(vuln)
+
     scan_result = exporter.InspectorScanResult(
         vulnerabilities=vulns,
         artifact_name=args.artifact_path,
         artifact_type=args.artifact_type,
-        criticals=criticals,
-        highs=highs,
-        mediums=mediums,
-        lows=lows,
-        others=others
+        criticals=str(criticals),
+        highs=str(highs),
+        mediums=str(mediums),
+        lows=str(lows),
+        others=str(others)
     )
 
-    return succeeded, scan_result
+    return succeeded, scan_result, fixed_vulns_counts
 
 
 def set_github_actions_output(key, value):
@@ -276,6 +284,10 @@ def set_github_actions_output(key, value):
         logging.info(f"setting github actions output: {key}:{value}")
         os.system(f'echo "{key}={value}" >> "$GITHUB_OUTPUT"')
 
+    # set an ENV VAR so that outputs
+    # look the same locally or on GitHub Actions
+    os.environ[key] = str(value)
+
     return
 
 
@@ -340,78 +352,83 @@ def get_vuln_counts(inspector_scan_path: str) -> tuple[bool, int, int, int, int,
 
     return True, criticals, highs, mediums, lows, others
 
-def get_fixed_vuln_counts(inspector_scan_path: str) -> tuple[bool, int, int, int, int, int]:
-    criticals_fixed = 0
-    highs_fixed = 0
-    mediums_fixed = 0
-    lows_fixed = 0
-    others_fixed = 0
+
+def get_fixed_vuln_counts(inspector_scan_path: str) -> tuple[bool, fixed_vulns.FixedVulns]:
+    fixed_vulns_counts = fixed_vulns.FixedVulns(criticals=0, highs=0, mediums=0, lows=0, others=0)
 
     scan_contents = ""
     try:
         with open(inspector_scan_path, 'r') as f:
             scan_contents = json.load(f)
     except Exception as e:
         logging.error(e)
-        return False, criticals, highs, mediums, lows, others
+        return False, fixed_vulns_counts
 
     # find the sbom->metadata->properties object
     scan_contents = scan_contents.get("sbom")
     if scan_contents is None:
         logging.error(
             f"expected Inspector scan results with 'sbom' as root object, but it was not found in file {inspector_scan_path}")
-        return False, criticals, highs, mediums, lows, others
+        return False, fixed_vulns_counts
 
     vulnerabilities = scan_contents.get("vulnerabilities")
 
     if vulnerabilities is None:
         # no vulnerabilities found
-        return True, criticals_fixed, highs_fixed, mediums_fixed, lows_fixed, others_fixed
-    
+        return True, fixed_vulns_counts
+
+    is_fix_available_key = "amazon:inspector:sbom_scanner:fixed_version:comp-"
     for vuln in vulnerabilities:
         props = vuln.get("properties")
         if props is None:
-            logging.error(f"expected vulnerability with 'properties' key but none was found in file {inspector_scan_path}")
+            logging.debug(
+                f"expected vulnerability with 'properties' key but none was found in file {inspector_scan_path}")
             continue
 
         for prop in props:
             name = prop.get("name")
             if name is None:
-                logging.error(f"expected vulnerability with 'name' key but none was found in file {inspector_scan_path}")
                 continue
-            if "amazon:inspector:sbom_scanner:fixed_version:comp-" in name:
-                rating = vuln.get("ratings")
-                if rating is None:
-                    logging.error(f"expected vulnerability with 'rating' key but none was found in file {inspector_scan_path}")
+            if is_fix_available_key not in name:
+                continue
+
+            # vuln has an available fix
+            rating = vuln.get("ratings")
+            if rating is None:
+                logging.error(
+                    f"expected vulnerability with 'rating' key but none was found in file {inspector_scan_path}")
+                continue
+
+            previous_score = 0
+            previous_severity = ""
+            for each_rating in rating:
+                severity = each_rating.get("severity")
+                if severity is None:
+                    logging.error(
+                        f"expected vulnerability with 'severity' key but none was found in file {inspector_scan_path}")
+                    continue
+                score = each_rating.get("score")
+                if score is None:
+                    logging.debug(
+                        f"expected vulnerability with 'score' key but none was found in file {inspector_scan_path}")
                     continue
+                if previous_score < score:
+                    previous_score = score
+                    previous_severity = severity
+
+            if previous_severity == "critical":
+                fixed_vulns_counts.criticals += 1
+            elif previous_severity == "high":
+                fixed_vulns_counts.highs += 1
+            elif previous_severity == "medium":
+                fixed_vulns_counts.mediums += 1
+            elif previous_severity == "low":
+                fixed_vulns_counts.lows += 1
+            else:
+                fixed_vulns_counts.others += 1
+
+    return True, fixed_vulns_counts
 
-                previous_score = 0
-                previous_severity = ""
-                for each_rating in rating:
-                    severity = each_rating.get("severity")
-                    if severity is None:
-                        logging.error(f"expected vulnerability with 'severity' key but none was found in file {inspector_scan_path}")
-                        continue
-                    score = each_rating.get("score")
-                    if score is None:
-                        logging.error(f"expected vulnerability with 'score' key but none was found in file {inspector_scan_path}")
-                        continue
-                    if previous_score < score:
-                        previous_score = score
-                        previous_severity = severity
-                
-                if previous_severity == "none":
-                    others_fixed += 1
-                elif previous_severity == "critical":
-                    criticals_fixed += 1
-                elif previous_severity == "high":
-                    highs_fixed += 1
-                elif previous_severity == "medium":
-                    mediums_fixed += 1
-                elif previous_severity == "low":
-                    lows_fixed += 1
-    
-    return True, criticals_fixed, highs_fixed, mediums_fixed, lows_fixed, others_fixed
 
 def install_sbomgen(args):
     os_name = platform.system()
@@ -452,12 +469,14 @@ def write_pkg_vuln_report_markdown(out_scan_markdown, scan_result: exporter.Insp
     return markdown
 
 
-def set_flag_if_vuln_threshold_exceeded(args, scan_result: exporter.InspectorScanResult):
-    is_exceeded = exceeds_threshold(scan_result.criticals, args.critical,
-                                    scan_result.highs, args.high,
-                                    scan_result.mediums, args.medium,
-                                    scan_result.lows, args.low,
-                                    scan_result.others, args.other)
+def set_env_var_if_vuln_threshold_exceeded(args,
+                                           vuln_counts: typing.Union[
+                                               exporter.InspectorScanResult, fixed_vulns.FixedVulns]):
+    is_exceeded = exceeds_threshold(vuln_counts.criticals, args.critical,
+                                    vuln_counts.highs, args.high,
+                                    vuln_counts.mediums, args.medium,
+                                    vuln_counts.lows, args.low,
+                                    vuln_counts.others, args.other)
 
     if is_exceeded and args.thresholds:
         set_github_actions_output('vulnerability_threshold_exceeded', 1)
@@ -471,19 +490,19 @@ def exceeds_threshold(criticals, critical_threshold,
                       lows, low_threshold,
                       others, other_threshold) -> bool:
     is_threshold_exceed = False
-    if 0 < critical_threshold <= criticals:
+    if 0 < critical_threshold <= int(criticals):
         is_threshold_exceed = True
 
-    if 0 < high_threshold <= highs:
+    if 0 < high_threshold <= int(highs):
         is_threshold_exceed = True
 
-    if 0 < medium_threshold <= mediums:
+    if 0 < medium_threshold <= int(mediums):
         is_threshold_exceed = True
 
-    if 0 < low_threshold <= lows:
+    if 0 < low_threshold <= int(lows):
         is_threshold_exceed = True
 
-    if 0 < other_threshold <= others:
+    if 0 < other_threshold <= int(others):
         is_threshold_exceed = True
 
     return is_threshold_exceed
@@ -533,6 +552,7 @@ def require_true(expr: bool, msg: str):
         logging.error(msg)
         exit(1)
 
+
 def is_valid_container_platform(img_platform):
     # regex for detecting 'os/cpu/variant'
     # os/cpu are required whereas variant is optional