Python script to present ShiftLeft NG SAST findings as Bitbucket code insights

shiftleft-bitbucket-insights.py

#!/usr/bin/python
# pip install requests
import os
import sys

import requests

# Collect the required variables
APP_ID = os.getenv("BITBUCKET_REPO_SLUG")
SHIFTLEFT_ORG_ID = os.getenv("SHIFTLEFT_ORG_ID")
SHIFTLEFT_ACCESS_TOKEN = os.getenv("SHIFTLEFT_ACCESS_TOKEN")
BRANCH = os.getenv("BITBUCKET_BRANCH", "")
BITBUCKET_REPO_OWNER = os.getenv("BITBUCKET_REPO_OWNER")
BITBUCKET_REPO_SLUG = os.getenv("BITBUCKET_REPO_SLUG")
BITBUCKET_COMMIT = os.getenv("BITBUCKET_COMMIT")
if not APP_ID or not SHIFTLEFT_ORG_ID:
    print(
        "Run this script from a Bitbucket pipeline or set the required environment variables"
    )
    sys.exit(1)
SL_FINDINGS_URL = (
    f"https://www.shiftleft.io/api/v4/orgs/{SHIFTLEFT_ORG_ID}/apps/{APP_ID}/findings"
)
BITBUCKET_URL = f"http://api.bitbucket.org/2.0/repositories/{BITBUCKET_REPO_OWNER}/{BITBUCKET_REPO_SLUG}/commit/{BITBUCKET_COMMIT}/reports/shiftleft-ngsast"

# Use local bitbucket proxy to avoid the need for app password
proxies = {
    "http": "http://localhost:29418",
    "https": "http://localhost:29418",
}


def convert_severity(severity):
    """Convert ShiftLeft severity to Bitbucket insights"""
    if severity == "critical":
        return "CRITICAL"
    elif severity == "moderate":
        return "MEDIUM"
    return "LOW"


def get_findings():
    """Method to get ShiftLeft findings using v4 api"""
    headers = {
        "Content-Type": "application/json",
        "Authorization": "Bearer " + SHIFTLEFT_ACCESS_TOKEN,
    }
    # Get all the findings
    r = requests.get(SL_FINDINGS_URL, headers=headers)
    if r.status_code == 200:
        raw_response = r.json()
        if raw_response and raw_response.get("response"):
            response = raw_response.get("response")
            total_count = response.get("total_count")
            scan = response.get("scan")
            scan_id = scan.get("id")
            spid = scan.get("internal_id")
            projectSpId = f'sl/{SHIFTLEFT_ORG_ID}/{scan.get("app")}'
            findings = response.get("findings")
            counts = response.get("counts")
            link = f"https://www.shiftleft.io/findingsSummary/{BITBUCKET_REPO_SLUG}?apps={BITBUCKET_REPO_SLUG}&isApp=1"
            data_list = [
                {
                    "title": "Safe to merge?",
                    "type": "BOOLEAN",
                    "value": total_count == 0,
                },
            ]
            # Create a PR report based on the total findings
            rr = requests.put(
                f"{BITBUCKET_URL}{scan_id}",
                proxies=proxies,
                headers={"Content-Type": "application/json",},
                json={
                    "title": "ShiftLeft NG SAST",
                    "details": f"This pull request contains {total_count} issues",
                    "report_type": "SECURITY",
                    "reporter": f"ShiftLeft NextGen Analysis for {BITBUCKET_REPO_SLUG}",
                    "link": link,
                    "remote_link_enabled": True,
                    "logo_url": "https://www.shiftleft.io/static/images/ShiftLeft_logo_white.svg",
                    "result": "FAILED" if total_count else "PASSED",
                    "data": data_list,
                },
            )
            # print(rr.status_code, rr.json())
            # For each finding create an annotation
            # TODO: Enhance the script to pass file and line information based v4 flows api
            for f in findings:
                fid = f.get("id")
                annotation_url = f"{BITBUCKET_URL}{scan_id}/annotations/ngsast{fid}"
                finternal = f.get("internal_id")
                tmpA = finternal.split("/")
                title = tmpA[0]
                occurrenceHash = tmpA[-1]
                annotation = {
                    "title": title,
                    "annotation_type": "VULNERABILITY",
                    "summary": f.get("title"),
                    "details": f.get("description"),
                    "severity": convert_severity(f.get("severity")),
                    "created_on": f.get("created_at"),
                }
                ar = requests.put(
                    annotation_url,
                    proxies=proxies,
                    headers={"Content-Type": "application/json",},
                    json=annotation,
                )
                # print(ar.status_code, ar.json())


if __name__ == "__main__":
    get_findings()