snyk_code_test.py

# You'll need to install the library if you haven't already
!pip install scikit-learn requests

import json, subprocess, requests
import numpy as np
from sklearn.ensemble import IsolationForest
import os

# You might need to adjust your PATH to include the npm global bin directory
# This command tries to find the npm global bin directory and add it to the PATH
# in case it's not already included.
npm_bin_path = os.popen('npm bin -g').read().strip()
if npm_bin_path and npm_bin_path not in os.environ['PATH']:
    os.environ['PATH'] += os.pathsep + npm_bin_path
    print(f"Added {npm_bin_path} to PATH for this session.")

# 1.1: Install prerequisites
#    pip install scikit-learn requests


# 1. Run Snyk CLI and capture JSON output
# Remove check=True so that the CalledProcessError is not raised immediately
proc = subprocess.run(
    ["snyk", "test", "--json"],
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE, # Capture stderr for potential error messages
    check=False # Do not raise CalledProcessError for non-zero exit codes
)

# Check the return code to see if Snyk found vulnerabilities
# A return code of 0 means no vulnerabilities were found.
# A non-zero return code (like 1 or 3) means vulnerabilities were found.
# Refer to Snyk CLI documentation for specific exit codes if needed.
if proc.returncode != 0:
    print(f"Snyk test completed with exit code {proc.returncode}. Vulnerabilities likely found.")
    # You can optionally print stderr for more details from Snyk
    # print("Snyk stderr:", proc.stderr.decode())
    # Decide whether to fail the build here based on Snyk's exit code
    # If you want to fail if Snyk finds *any* vulnerabilities, you can exit here.
    # exit(1)
    # If you want to continue and process based on your own logic (critical/anomaly), proceed.
else:
    print("Snyk test completed with exit code 0. No vulnerabilities found by Snyk.")
    # If Snyk found no vulnerabilities, there's no JSON output to parse,
    # so we can exit successfully here.
    print("✅ Security checks passed (no vulnerabilities reported by Snyk)")
    exit(0)


# Proceed with vulnerability processing only if Snyk returned non-zero (found vulns)
try:
    vulns = json.loads(proc.stdout)

    # 2. Extract features per vulnerability
    #    e.g., severity (0=low,1=medium,2=high,3=critical),
    #          package age (in days),
    #          num of vulnerable versions
    def extract_features(v):
        sev_map = {"low":0,"medium":1,"high":2,"critical":3}
        severity = sev_map.get(v["severity"], 0) # Use .get with a default in case of unexpected severity
        # Attempt to get package age, handle potential errors
        try:
            pkg_age_response = requests.get(f"https://libraries.io/{v['packageManager']}/{v['name']}/{v['version']}/latest_release")
            pkg_age_response.raise_for_status() # Raise an HTTPError for bad responses (4xx or 5xx)
            pkg_age_data = pkg_age_response.json()
            pkg_age = pkg_age_data.get("days_since", 0) # Default to 0 if days_since is missing
        except requests.exceptions.RequestException as e:
            print(f"Warning: Could not retrieve package age for {v['name']}@{v['version']}: {e}")
            pkg_age = 0 # Default package age if request fails
        except Exception as e:
             print(f"Warning: Error processing package age data for {v['name']}@{v['version']}: {e}")
             pkg_age = 0

        versions = len(v.get("versions", [])) # Use .get with empty list default
        return [severity, pkg_age, versions]

    # Check if 'vulnerabilities' key exists and is not empty
    if "vulnerabilities" in vulns and vulns["vulnerabilities"]:
        X = np.array([extract_features(v) for v in vulns["vulnerabilities"]])

        # 3. Fit or load a pretrained Isolation Forest
        #    In production, you’d persist this model and only call .predict()
        iso = IsolationForest(contamination=0.05, random_state=0)
        iso.fit(X)            # Train on historical data in a real setup
        labels = iso.predict(X)  # -1 = anomaly, 1 = normal

        # 4. Decision logic
        has_critical = any(v.get("severity") == "critical" for v in vulns["vulnerabilities"]) # Use .get
        has_anomaly  = any(l == -1 for l in labels)

        if has_critical or has_anomaly:
            print("🚨 Security build failed")
            exit(1)
        else:
            print("✅ Security checks passed")
            exit(0)
    else:
        print("Snyk test completed, but no vulnerabilities found in the output.")
        print("✅ Security checks passed (no vulnerabilities found in Snyk output)")
        exit(0)

except json.JSONDecodeError as e:
    print(f"Error decoding JSON from Snyk output: {e}")
    print("Snyk stdout:", proc.stdout.decode())
    print("Snyk stderr:", proc.stderr.decode())
    print("🚨 Security build failed due to invalid Snyk output.")
    exit(1)
except Exception as e:
    print(f"An unexpected error occurred during vulnerability processing: {e}")
    print("🚨 Security build failed due to processing error.")
    exit(1)