cli.py

# -*- coding: utf-8 -*-
from __future__ import absolute_import

import json
import logging
import os
import sys
import tempfile

import click

from safety import safety
from safety.alerts import alert
from safety.constants import EXIT_CODE_VULNERABILITIES_FOUND, EXIT_CODE_OK, EXIT_CODE_FAILURE
from safety.errors import SafetyException, SafetyError
from safety.formatter import SafetyFormatter
from safety.output_utils import should_add_nl
from safety.safety import get_packages, read_vulnerabilities, fetch_policy, post_results
from safety.util import get_proxy_dict, get_packages_licenses, output_exception, \
    MutuallyExclusiveOption, DependentOption, transform_ignore, SafetyPolicyFile, active_color_if_needed, \
    get_processed_options, get_safety_version, json_alias, bare_alias, SafetyContext, is_a_remote_mirror, \
    filter_announcements

LOG = logging.getLogger(__name__)


@click.group()
@click.option('--debug/--no-debug', default=False)
@click.option('--telemetry/--disable-telemetry', default=True, hidden=True)
@click.option('--disable-optional-telemetry-data', default=False, cls=MutuallyExclusiveOption,
              mutually_exclusive=["telemetry", "disable-telemetry"], is_flag=True, show_default=True)
@click.version_option(version=get_safety_version())
@click.pass_context
def cli(ctx, debug, telemetry, disable_optional_telemetry_data):
    """
    Safety checks Python dependencies for known security vulnerabilities and suggests the proper
    remediations for vulnerabilities detected. Safety can be run on developer machines, in CI/CD pipelines and
    on production systems.
    """
    SafetyContext().safety_source = 'cli'
    ctx.telemetry = telemetry and not disable_optional_telemetry_data
    level = logging.CRITICAL
    if debug:
        level = logging.DEBUG

    logging.basicConfig(format='%(asctime)s %(name)s => %(message)s', level=level)

    LOG.info(f'Telemetry enabled: {ctx.telemetry}')

    @ctx.call_on_close
    def clean_up_on_close():
        LOG.debug('Calling clean up on close function.')
        safety.close_session()


@cli.command()
@click.option("--key", default="", envvar="SAFETY_API_KEY",
              help="API Key for pyup.io's vulnerability database. Can be set as SAFETY_API_KEY "
                   "environment variable. Default: empty")
@click.option("--db", default="",
              help="Path to a local or remote vulnerability database. Default: empty")
@click.option("--full-report/--short-report", default=False, cls=MutuallyExclusiveOption,
              mutually_exclusive=["output", "json", "bare"],
              with_values={"output": ['json', 'bare'], "json": [True, False], "bare": [True, False]},
              help='Full reports include a security advisory (if available). Default: --short-report')
@click.option("--cache", is_flag=False, flag_value=60, default=0,
              help="Cache requests to the vulnerability database locally. Default: 0 seconds",
              hidden=True)
@click.option("--stdin", default=False, cls=MutuallyExclusiveOption, mutually_exclusive=["files"],
              help="Read input from stdin.", is_flag=True, show_default=True)
@click.option("files", "--file", "-r", multiple=True, type=click.File(), cls=MutuallyExclusiveOption,
              mutually_exclusive=["stdin"],
              help="Read input from one (or multiple) requirement files. Default: empty")
@click.option("--ignore", "-i", multiple=True, type=str, default=[], callback=transform_ignore,
              help="Ignore one (or multiple) vulnerabilities by ID. Default: empty")
@click.option('--json', default=False, cls=MutuallyExclusiveOption, mutually_exclusive=["output", "bare"],
              with_values={"output": ['screen', 'text', 'bare', 'json'], "bare": [True, False]}, callback=json_alias,
              hidden=True, is_flag=True, show_default=True)
@click.option('--bare', default=False, cls=MutuallyExclusiveOption, mutually_exclusive=["output", "json"],
              with_values={"output": ['screen', 'text', 'bare', 'json'], "json": [True, False]}, callback=bare_alias,
              hidden=True, is_flag=True, show_default=True)
@click.option('--output', "-o", type=click.Choice(['screen', 'text', 'json', 'bare'], case_sensitive=False),
              default='screen', callback=active_color_if_needed, envvar='SAFETY_OUTPUT')
@click.option("--proxy-protocol", "-pr", type=click.Choice(['http', 'https']), default='https', cls=DependentOption, required_options=['proxy_host'],
              help="Proxy protocol (https or http) --proxy-protocol")
@click.option("--proxy-host", "-ph", multiple=False, type=str, default=None,
              help="Proxy host IP or DNS --proxy-host")
@click.option("--proxy-port", "-pp", multiple=False, type=int, default=80, cls=DependentOption, required_options=['proxy_host'],
              help="Proxy port number --proxy-port")
@click.option("--exit-code/--continue-on-error", default=True,
              help="Output standard exit codes. Default: --exit-code")
@click.option("--policy-file", type=SafetyPolicyFile(), default='.safety-policy.yml',
              help="Define the policy file to be used")
@click.option("--audit-and-monitor/--disable-audit-and-monitor", default=True,
              help="Send results back to pyup.io for viewing on your dashboard. Requires an API key.")
@click.option("--project", default=None,
              help="Project to associate this scan with on pyup.io. Defaults to a canonicalized github style name if available, otherwise unknown")

@click.option("--save-json", default="", help="Path to where output file will be placed, if the path is a directory, "
                                              "Safety will use safety-report.json as filename. Default: empty")
@click.pass_context
def check(ctx, key, db, full_report, stdin, files, cache, ignore, output, json, bare, proxy_protocol, proxy_host, proxy_port,
          exit_code, policy_file, save_json, audit_and_monitor, project):
    """
    Find vulnerabilities in Python dependencies at the target provided.

    """
    LOG.info('Running check command')

    try:
        packages = get_packages(files, stdin)
        proxy_dictionary = get_proxy_dict(proxy_protocol, proxy_host, proxy_port)

        if key:
            server_policies = fetch_policy(key=key, proxy=proxy_dictionary)
            server_audit_and_monitor = server_policies["audit_and_monitor"]
            server_safety_policy = server_policies["safety_policy"]
        else:
            server_audit_and_monitor = False
            server_safety_policy = ""

        if server_safety_policy and policy_file:
            click.secho(
                "Warning: both a local policy file '{policy_filename}' and a server sent policy are present. "
                "Continuing with the local policy file.".format(policy_filename=policy_file['filename']),
                fg="yellow",
                file=sys.stderr
            )
        elif server_safety_policy:
            with tempfile.NamedTemporaryFile(prefix='server-safety-policy-') as tmp:
                tmp.write(server_safety_policy.encode('utf-8'))
                tmp.seek(0)

                policy_file = SafetyPolicyFile().convert(tmp.name, param=None, ctx=None)
                LOG.info('Using server side policy file')

        ignore_severity_rules = None
        ignore, ignore_severity_rules, exit_code = get_processed_options(policy_file, ignore,
                                                                         ignore_severity_rules, exit_code)

        is_env_scan = not stdin and not files
        params = {'stdin': stdin, 'files': files, 'policy_file': policy_file, 'continue_on_error': not exit_code,
                  'ignore_severity_rules': ignore_severity_rules, 'project': project, 'audit_and_monitor': server_audit_and_monitor and audit_and_monitor}
        LOG.info('Calling the check function')
        vulns, db_full = safety.check(packages=packages, key=key, db_mirror=db, cached=cache, ignore_vulns=ignore,
                                      ignore_severity_rules=ignore_severity_rules, proxy=proxy_dictionary,
                                      include_ignored=True, is_env_scan=is_env_scan, telemetry=ctx.parent.telemetry,
                                      params=params)
        LOG.debug('Vulnerabilities returned: %s', vulns)
        LOG.debug('full database returned is None: %s', db_full is None)

        LOG.info('Safety is going to calculate remediations')
        remediations = safety.calculate_remediations(vulns, db_full)

        announcements = []
        if not db or is_a_remote_mirror(db):
            LOG.info('Not local DB used, Getting announcements')
            announcements = safety.get_announcements(key=key, proxy=proxy_dictionary, telemetry=ctx.parent.telemetry)

        json_report = None
        if save_json or (server_audit_and_monitor and audit_and_monitor):
            default_name = 'safety-report.json'
            json_report = SafetyFormatter(output='json').render_vulnerabilities(announcements, vulns, remediations,
                                                                                full_report, packages)

            if server_audit_and_monitor and audit_and_monitor:
                policy_contents = ''
                if policy_file:
                    policy_contents = policy_file.get('raw', '')

                r = post_results(key=key, proxy=proxy_dictionary, safety_json=json_report, policy_file=policy_contents)
                SafetyContext().params['audit_and_monitor_url'] = r.get('url')

            if save_json:
                if os.path.isdir(save_json):
                    save_json = os.path.join(save_json, default_name)

                with open(save_json, 'w+') as output_json_file:
                    output_json_file.write(json_report)

        LOG.info('Safety is going to render the vulnerabilities report using %s output', output)
        if json_report and output == 'json':
            output_report = json_report
        else:
            output_report = SafetyFormatter(output=output).render_vulnerabilities(announcements, vulns, remediations,
                                                                                  full_report, packages)

        # Announcements are send to stderr if not terminal, it doesn't depend on "exit_code" value
        stderr_announcements = filter_announcements(announcements=announcements, by_type='error')
        if stderr_announcements and (not sys.stdout.isatty() and os.environ.get("SAFETY_OS_DESCRIPTION", None) != 'run'):
            LOG.info('sys.stdout is not a tty, error announcements are going to be send to stderr')
            click.secho(SafetyFormatter(output='text').render_announcements(stderr_announcements), fg="red",
                        file=sys.stderr)

        found_vulns = list(filter(lambda v: not v.ignored, vulns))
        LOG.info('Vulnerabilities found (Not ignored): %s', len(found_vulns))
        LOG.info('All vulnerabilities found (ignored and Not ignored): %s', len(vulns))

        click.secho(output_report, nl=should_add_nl(output, found_vulns), file=sys.stdout)

        if exit_code and found_vulns:
            LOG.info('Exiting with default code for vulnerabilities found')
            sys.exit(EXIT_CODE_VULNERABILITIES_FOUND)

        sys.exit(EXIT_CODE_OK)

    except SafetyError as e:
        LOG.exception('Expected SafetyError happened: %s', e)
        output_exception(e, exit_code_output=exit_code)
    except Exception as e:
        LOG.exception('Unexpected Exception happened: %s', e)
        exception = e if isinstance(e, SafetyException) else SafetyException(info=e)
        output_exception(exception, exit_code_output=exit_code)


@cli.command()
@click.option("--full-report/--short-report", default=False, cls=MutuallyExclusiveOption, mutually_exclusive=["output"], with_values={"output": ['json', 'bare']},
              help='Full reports include a security advisory (if available). Default: '
                   '--short-report')
@click.option('--output', "-o", type=click.Choice(['screen', 'text', 'json', 'bare'], case_sensitive=False),
              default='screen', callback=active_color_if_needed)
@click.option("file", "--file", "-f", type=click.File(), required=True,
              help="Read input from an insecure report file. Default: empty")
@click.pass_context
def review(ctx, full_report, output, file):
    """
    Show an output from a previous exported JSON report.
    """
    LOG.info('Running check command')
    report = {}

    try:
        report = read_vulnerabilities(file)
    except SafetyError as e:
        LOG.exception('Expected SafetyError happened: %s', e)
        output_exception(e, exit_code_output=True)
    except Exception as e:
        LOG.exception('Unexpected Exception happened: %s', e)
        exception = e if isinstance(e, SafetyException) else SafetyException(info=e)
        output_exception(exception, exit_code_output=True)

    params = {'file': file}
    vulns, remediations, packages = safety.review(report, params=params)

    announcements = safety.get_announcements(key=None, proxy=None, telemetry=ctx.parent.telemetry)
    output_report = SafetyFormatter(output=output).render_vulnerabilities(announcements, vulns, remediations,
                                                                          full_report, packages)

    found_vulns = list(filter(lambda v: not v.ignored, vulns))
    click.secho(output_report, nl=should_add_nl(output, found_vulns), file=sys.stdout)
    sys.exit(EXIT_CODE_OK)


@cli.command()
@click.option("--key", envvar="SAFETY_API_KEY",
              help="API Key for pyup.io's vulnerability database. Can be set as SAFETY_API_KEY "
                   "environment variable. Default: empty")
@click.option("--db", default="",
              help="Path to a local license database. Default: empty")
@click.option('--output', "-o", type=click.Choice(['screen', 'text', 'json', 'bare'], case_sensitive=False),
              default='screen')
@click.option("--cache", default=0,
              help='Whether license database file should be cached.'
                   'Default: 0 seconds')
@click.option("files", "--file", "-r", multiple=True, type=click.File(),
              help="Read input from one (or multiple) requirement files. Default: empty")
@click.option("proxyhost", "--proxy-host", "-ph", multiple=False, type=str, default=None,
              help="Proxy host IP or DNS --proxy-host")
@click.option("proxyport", "--proxy-port", "-pp", multiple=False, type=int, default=80,
              help="Proxy port number --proxy-port")
@click.option("proxyprotocol", "--proxy-protocol", "-pr", multiple=False, type=str, default='http',
              help="Proxy protocol (https or http) --proxy-protocol")
@click.pass_context
def license(ctx, key, db, output, cache, files, proxyprotocol, proxyhost, proxyport):
    """
    Find the open source licenses used by your Python dependencies.
    """
    LOG.info('Running license command')
    packages = get_packages(files, False)

    proxy_dictionary = get_proxy_dict(proxyprotocol, proxyhost, proxyport)
    licenses_db = {}

    try:
        licenses_db = safety.get_licenses(key=key, db_mirror=db, cached=cache, proxy=proxy_dictionary,
                                          telemetry=ctx.parent.telemetry)
    except SafetyError as e:
        LOG.exception('Expected SafetyError happened: %s', e)
        output_exception(e, exit_code_output=False)
    except Exception as e:
        LOG.exception('Unexpected Exception happened: %s', e)
        exception = e if isinstance(e, SafetyException) else SafetyException(info=e)
        output_exception(exception, exit_code_output=False)

    filtered_packages_licenses = get_packages_licenses(packages=packages, licenses_db=licenses_db)

    announcements = []
    if not db:
        announcements = safety.get_announcements(key=key, proxy=proxy_dictionary, telemetry=ctx.parent.telemetry)

    output_report = SafetyFormatter(output=output).render_licenses(announcements, filtered_packages_licenses)

    click.secho(output_report, nl=True)


@cli.command()
@click.option("--path", default=".", help="Path where the generated file will be saved. Default: current directory")
@click.argument('name')
@click.pass_context
def generate(ctx, name, path):
    """Create a boilerplate supported file type.

    NAME is the name of the file type to generate. Valid values are: policy_file
    """
    if name != 'policy_file':
        click.secho(f'This Safety version only supports "policy_file" generation. "{name}" is not supported.', fg='red',
                    file=sys.stderr)
        sys.exit(EXIT_CODE_FAILURE)

    LOG.info('Running generate %s', name)

    if not os.path.exists(path):
        click.secho(f'The path "{path}" does not exist.', fg='red',
                    file=sys.stderr)
        sys.exit(EXIT_CODE_FAILURE)

    policy = os.path.join(path, '.safety-policy.yml')
    ROOT = os.path.dirname(os.path.abspath(__file__))

    try:
        with open(policy, "w") as f:
            f.write(open(os.path.join(ROOT, 'safety-policy-template.yml')).read())
            LOG.debug('Safety created the policy file.')
            msg = f'A default Safety policy file has been generated! Review the file contents in the path {path} in the ' \
                  'file: .safety-policy.yml'
            click.secho(msg, fg='green')
    except Exception as exc:
        if isinstance(exc, OSError):
            LOG.debug('Unable to generate %s because: %s', name, exc.errno)

        click.secho(f'Unable to generate {name}, because: {str(exc)} error.', fg='red',
                    file=sys.stderr)
        sys.exit(EXIT_CODE_FAILURE)


@cli.command()
@click.option("--path", default=".safety-policy.yml", help="Path where the generated file will be saved. Default: current directory")
@click.argument('name')
@click.pass_context
def validate(ctx, name, path):
    """Verify the validity of a supported file type.

    NAME is the name of the file type to validate. Valid values are: policy_file
    """
    if name != 'policy_file':
        click.secho(f'This Safety version only supports "policy_file" validation. "{name}" is not supported.', fg='red',
                    file=sys.stderr)
        sys.exit(EXIT_CODE_FAILURE)

    LOG.info('Running validate %s', name)

    if not os.path.exists(path):
        click.secho(f'The path "{path}" does not exist.', fg='red', file=sys.stderr)
        sys.exit(EXIT_CODE_FAILURE)

    try:
        values = SafetyPolicyFile().convert(path, None, None)
    except Exception as e:
        click.secho(str(e).lstrip(), fg='red', file=sys.stderr)
        sys.exit(EXIT_CODE_FAILURE)

    del values['raw']

    click.secho(f'The Safety policy file was successfully parsed with the following values:', fg='green')
    click.secho(json.dumps(values, indent=4, default=str))

cli.add_command(alert)


if __name__ == "__main__":
    cli()