scan

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Multi-language static analysis scanner
"""
import argparse
import logging
import os
import sys
import tempfile
import uuid

import lib.analysis as analysis
import lib.config as config
import lib.convert as convertLib
import lib.context as context
import lib.utils as utils
from lib.executor import exec_tool, execute_default_cmd
from lib.telemetry import track

logging.basicConfig(
    level=logging.INFO, format="%(levelname)s [%(asctime)s] %(message)s"
)
LOG = logging.getLogger(__name__)

at_logo = """
  ___            _____ _                    _
 / _ \          |_   _| |                  | |
/ /_\ \_ __  _ __ | | | |__  _ __ ___  __ _| |_
|  _  | '_ \| '_ \| | | '_ \| '__/ _ \/ _` | __|
| | | | |_) | |_) | | | | | | | |  __/ (_| | |_
\_| |_/ .__/| .__/\_/ |_| |_|_|  \___|\__,_|\__|
      | |   | |
      |_|   |_|
"""


def build_args():
    """
    Constructs command line arguments for the scanner
    """
    parser = argparse.ArgumentParser(
        description="Wrapper for various static analysis tools"
    )
    parser.add_argument("--src", dest="src_dir", help="Source directory")
    parser.add_argument("--out_dir", dest="reports_dir", help="Reports directory")
    parser.add_argument(
        "--type",
        dest="scan_type",
        help="Override project type if auto-detection is incorrect. Comma separated values for multiple types. Eg: python,bash,credscan",
    )
    parser.add_argument(
        "--convert",
        action="store_true",
        default=True,
        dest="convert",
        help="Convert results to sarif json format",
    )
    parser.add_argument(
        "--no-error",
        action="store_true",
        default=False,
        dest="noerror",
        help="Continue on error to prevent build from breaking",
    )
    return parser.parse_args()


def scan(type_list, src, reports_dir, convert):
    """
    Method to initiate scan of the codebase

    Args:
      type_list List of project type
      src Project dir
      reports_dir Directory for output reports
      convert Boolean to enable normalisation of reports json
    """
    for type_str in type_list:
        if config.get("scan_tools_args_map").get(type_str):
            cmd_map_list = config.get("scan_tools_args_map").get(type_str)
            # Default command list can be in the form of a list or dict
            if isinstance(cmd_map_list, list):
                execute_default_cmd(
                    cmd_map_list, type_str, type_str, src, reports_dir, convert,
                )
            elif isinstance(cmd_map_list, dict):
                for cmd_key, cmd_val in cmd_map_list.items():
                    execute_default_cmd(
                        cmd_val, type_str, cmd_key, src, reports_dir, convert,
                    )
        else:
            # Look for any _scan function in this module for execution
            try:
                getattr(sys.modules[__name__], "%s_scan" % type_str)(
                    src, reports_dir, convert
                )
            except Exception as e:
                LOG.error("Received exception: {}".format(e))
                # LOG.warning("Project type is not supported: {}".format(type_str))


def python_scan(src, reports_dir, convert):
    """
    Method to initiate scan of the python codebase

    Args:
      src Project dir
      reports_dir Directory for output reports
      convert Boolean to enable normalisation of reports json
    """
    bandit_scan(src, reports_dir, convert)


def bandit_scan(src, reports_dir, convert):
    """
    Method to initiate bandit scan of the python codebase

    Args:
      src Project dir
      reports_dir Directory for output reports
      convert Boolean to enable normalisation of reports json
    """
    convert_args = []
    report_fname = utils.get_report_file(
        "bandit", reports_dir, convert, ext_name="json"
    )
    if reports_dir or convert:
        convert_args = [
            "-o",
            report_fname,
            "-f",
            "json",
        ]
    bandit_cmd = "bandit"
    bandit_args = [
        bandit_cmd,
        "-r",
        "-a",
        "vuln",
        "-ii",
        "-ll",
        *convert_args,
        "-x",
        ",".join(config.get("ignore_directories")),
        src,
    ]
    exec_tool(bandit_args)
    if convert:
        crep_fname = utils.get_report_file(
            bandit_cmd, reports_dir, convert, ext_name="sarif"
        )
        convertLib.convert_file(
            bandit_cmd, bandit_args[1:], src, report_fname, crep_fname,
        )


def java_scan(src, reports_dir, convert):
    """
    Method to initiate scan of the java codebase

    Args:
      src Project dir
      reports_dir Directory for output reports
      convert Boolean to enable normalisation of reports json
    """
    findsecbugs_scan(src, reports_dir, convert)


def findsecbugs_scan(src, reports_dir, convert):
    """
    Method to initiate findsecbugs scan of the java codebase

    Args:
      src Project dir
      reports_dir Directory for output reports
      convert Boolean to enable normalisation of reports json
    """
    report_fname = utils.get_report_file(
        "findsecbugs", reports_dir, convert, ext_name="xml"
    )
    findsec_cmd = [
        "java",
        "-jar",
        os.environ["SPOTBUGS_HOME"] + "/lib/spotbugs.jar",
    ]
    jar_files = utils.find_jar_files()
    with tempfile.NamedTemporaryFile(mode="w") as fp:
        fp.writelines([str(x) + "\n" for x in jar_files])
        jars_list = fp.name
        findsec_args = [
            *findsec_cmd,
            "-textui",
            "-include",
            os.environ["APP_SRC_DIR"] + "/spotbugs/include.xml",
            "-exclude",
            os.environ["APP_SRC_DIR"] + "/spotbugs/exclude.xml",
            "-noClassOk",
            "-auxclasspathFromFile",
            jars_list,
            "-sourcepath",
            src,
            "-quiet",
            "-medium",
            "-xml:withMessages",
            "-effort:max",
            "-nested:false",
            "-output",
            report_fname,
            src,
        ]
        exec_tool(findsec_args, src)
        if convert:
            # We need the filelist to fix the file location paths
            j_files = utils.find_files(src, ".java")
            crep_fname = utils.get_report_file(
                "findsecbugs", reports_dir, convert, ext_name="sarif"
            )
            convertLib.convert_file(
                "findsecbugs", findsec_args[1:], src, report_fname, crep_fname, j_files,
            )


def nodejs_scan(src, reports_dir, convert):
    """
    Method to initiate scan of the node.js codebase

    Args:
      src Project dir
      reports_dir Directory for output reports
      convert Boolean to enable normalisation of reports json
    """
    sec_scan(src, reports_dir, convert)


def sec_scan(src, reports_dir, convert):
    """
    Method to initiate Nodejs sec scan of the node.js codebase

    Args:
      src Project dir
      reports_dir Directory for output reports
      convert Boolean to enable normalisation of reports json
    """
    convert_args = []
    report_fname = utils.get_report_file("nodejsscan", reports_dir, convert)
    if reports_dir or convert:
        convert_args = [
            "--output",
            report_fname,
        ]
    sec_cmd = "nodejsscan"
    sec_args = [sec_cmd, *convert_args, "-d", src]
    exec_tool(sec_args, src)
    if convert:
        crep_fname = utils.get_report_file(
            "nodejsscan", reports_dir, convert, ext_name="sarif"
        )
        convertLib.convert_file(
            "nodejsscan", sec_args[1:], src, report_fname, crep_fname,
        )


def bomgen(src, reports_dir, convert):
    """
    Method to generate cyclonedx bom file using cdxgen

    Args:
      src Project dir
      reports_dir Directory for output reports
      convert Boolean to enable normalisation of reports json
    """
    report_fname = utils.get_report_file("bom", reports_dir, convert, ext_name="xml")
    bom_args = ["cdxgen", "-o", report_fname, src]
    exec_tool(bom_args, src)


def main():
    args = build_args()
    src_dir = args.src_dir
    if not args.src_dir:
        src_dir = os.getcwd()
    type = args.scan_type
    # Get or construct the run uuid
    run_uuid = os.environ.get("SCAN_ID", str(uuid.uuid4()))
    config.set("run_uuid", run_uuid)
    # Set the source directory as an environment variable
    config.set("SAST_SCAN_SRC_DIR", src_dir)
    repo_context = context.find_repo_details(src_dir)
    workspace = os.environ.get("WORKSPACE", None)
    if not workspace:
        workspace = utils.get_workspace(repo_context)
        if workspace:
            config.set("WORKSPACE", workspace)
    config.reload()

    # Identify project type
    if not type:
        # Check the local config first. If not try auto detection
        type = config.get("scan_type")
        if type:
            type = type.split(",")
        else:
            type = utils.detect_project_type(src_dir)
    else:
        type = type.split(",")
    print(at_logo)
    LOG.info("Scanning {} using scan plugins {}".format(src_dir, type))
    reports_dir = args.reports_dir
    if not reports_dir:
        reports_dir = os.path.join(src_dir, "reports")
    scan(type, src_dir, reports_dir, args.convert)
    if reports_dir:
        sarif_files = utils.find_files(reports_dir, ".sarif")
        agg_fname = utils.get_report_file(
            "all-" + run_uuid, reports_dir, False, ext_name="json"
        )
        report_summary, build_status = analysis.summary(sarif_files, agg_fname)
        if report_summary:
            analysis.print_summary(report_summary)
            track(
                {
                    "id": run_uuid,
                    "repo_context": repo_context,
                    "report_summary": report_summary,
                }
            )
            if not args.noerror:
                sys.exit(1 if build_status == "fail" else 0)


if __name__ == "__main__":
    main()