Harness for recording responses for each finding made by Yelps detect secrets tool.

detect_secrets_assessment.py

import argparse
import json
import subprocess
import linecache
import os
import sys
from pprint import pprint
import tempfile
import csv


EDITOR = os.environ.get('EDITOR', 'vim')
CONTEXT = 10
ANSWERS = {
    "1": "FP: No hardcoded credentials present.",
    "2": "FP: Example or dummy credential.",
    "3": "TP: Valid credential, but not for production use.",
    "4": "TP: VALID PRODUCTION CREDENTIAL DETECTED!",
    "5": "OTHER: Follow up.",
}

def iter_findings(report, skip_to=0):
    data = json.loads(report.read())
    pos = 0
    for filename, secrets in data['results'].items():
        for secret in secrets:
            if pos >= skip_to: 
                yield dict({'filename': filename}, **secret)
            else:
                print(f"skipping record: {pos}")
            pos += 1


def show_editor(finding, secret):
    """
    Allow custom responses for each secret detected.
    """
    with tempfile.NamedTemporaryFile('w+') as tf: 
        tf.write("\n");
        tf.write("#####################################################\n")
        tf.write("#\n")
        tf.write(f"#   FILE: {finding['filename']}\n")
        tf.write(f"#   TYPE: {finding['type']}\n")
        tf.write(f'# SECRET:\n')
        tf.write(secret)
        tf.write("#\n")
        tf.write("#####################################################\n")
        for num, ans in ANSWERS.items():
            tf.write(f"# {num}) {ans}\n")
        tf.flush()

        subprocess.call([EDITOR, tf.name])

        tf.seek(0)
        content = tf.read()
        comment = ""
        for line in content.split('\n'):
            if line.strip() and not line.startswith('#'):
                comment += line

        return comment.strip()


def show_prompt(finding, secret):
    """
    Use fixed responses (for churning through lots of results).
    """
    print("")
    print("#####################################################\n")
    print("#")
    print(f"#   FILE: {finding['filename']}")
    print(f"#   TYPE: {finding['type']}")
    print(f'# SECRET:')
    print(secret)
    print("#")
    print("#####################################################\n")
    for num, ans in ANSWERS.items():
        print(f"# {num}) {ans}")

    return input("Is this a secret?:")

def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--infile", required=True, type=argparse.FileType('r'))
    parser.add_argument("--outfile", required=True, type=argparse.FileType('a', encoding='UTF-8'))
    parser.add_argument("--root", default=".")
    parser.add_argument("--offset", type=int, default=0)
    args = parser.parse_args()

    csvwriter = csv.writer(args.outfile)
    for finding in iter_findings(args.infile, args.offset):

        lines = []
        filename = os.path.join(args.root, finding['filename'])
        line = finding['line_number']
        credential_type = finding['type']
        credential_hash = finding['hashed_secret']

        content = []
        for context in range(-CONTEXT, CONTEXT):
            filecontent = linecache.getline(filename, line + context)
            if not filecontent:
                  filecontent = '\n'
            if context == 0:
                content.append('# >>> |' + filecontent)
            else:
                content.append('#     |' + filecontent)
            
        response = show_prompt(finding, ''.join(content))
        while response.strip() not in ANSWERS.keys():
            print(f"Invalid selection: {response}")
            response = show_prompt(finding, ''.join(content))

        comment = ANSWERS[response] 
        csvwriter.writerow([filename, line, credential_type, credential_hash, comment])
        args.outfile.flush()


if __name__ == "__main__":
    main()