Using dparse to read requirements and fixes for custom integrations

safety/cli.py

@@ -17,7 +17,7 @@
 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
+    get_processed_options, get_safety_version, json_alias, bare_alias, SafetyContext, is_a_remote_mirror
 
 LOG = logging.getLogger(__name__)
 
@@ -51,7 +51,7 @@ def cli(ctx, debug, telemetry, disable_optional_telemetry_data):
               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 vulnerability database. Default: empty")
+              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]},
@@ -105,7 +105,7 @@ def check(ctx, key, db, full_report, stdin, files, cache, ignore, output, json,
         proxy_dictionary = get_proxy_dict(proxy_protocol, proxy_host, proxy_port)
 
         announcements = []
-        if not db:
+        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)
 

safety/output_utils.py

@@ -1,12 +1,14 @@
 import json
 import logging
+import os
 import textwrap
 from datetime import datetime
 
 import click
 
 from safety.constants import RED, YELLOW
-from safety.util import get_safety_version, Package, get_terminal_size, SafetyContext, build_telemetry_data, build_git_data
+from safety.util import get_safety_version, Package, get_terminal_size, \
+    SafetyContext, build_telemetry_data, build_git_data, is_a_remote_mirror
 
 LOG = logging.getLogger(__name__)
 
@@ -485,16 +487,24 @@ def build_report_for_review_vuln_report(as_dict=False):
 
 def build_using_sentence(key, db):
     key_sentence = []
+    custom_integration = os.environ.get('SAFETY_CUSTOM_INTEGRATION',
+                                        'false').lower() == 'true'
 
     if key:
         key_sentence = [{'style': True, 'value': 'an API KEY'},
                         {'style': False, 'value': ' and the '}]
         db_name = 'PyUp Commercial'
+    elif db and custom_integration and is_a_remote_mirror(db):
+        return []
     else:
         db_name = 'non-commercial'
 
     if db:
-        db_name = "local file {0}".format(db)
+        db_type = 'local file'
+        if is_a_remote_mirror(db):
+            db_type = 'remote URL'
+
+        db_name = f"{db_type} {db}"
 
     database_sentence = [{'style': True, 'value': db_name + ' database'}]
 
@@ -629,6 +639,7 @@ def get_report_brief_info(as_dict=False, report_type=1, **kwargs):
     brief_data['json_version'] = 1
 
     using_sentence = build_using_sentence(key, db)
+    using_sentence_section = [nl] if not using_sentence else [nl] + [build_using_sentence(key, db)]
     scanned_count_sentence = build_scanned_count_sentence(packages)
 
     timestamp = [{'style': False, 'value': 'Timestamp '}, {'style': True, 'value': current_time}]
@@ -638,7 +649,7 @@ def get_report_brief_info(as_dict=False, report_type=1, **kwargs):
      {'style': False, 'value': ' is scanning for '},
      {'style': True, 'value': scanning_types.get(context.command, {}).get('name', '')},
      {'style': True, 'value': '...'}] + safety_policy_used + audit_and_monitor, action_executed
-     ] + [nl] + scanned_items + [nl] + [using_sentence] + [scanned_count_sentence] + [timestamp]
+     ] + [nl] + scanned_items + using_sentence_section + [scanned_count_sentence] + [timestamp]
 
     brief_info.extend(additional_data)
 

safety/safety.py

@@ -19,7 +19,7 @@
                      RequestTimeoutError, ServerError, MalformedDatabase)
 from .models import Vulnerability, CVE, Severity
 from .util import RequirementFile, read_requirements, Package, build_telemetry_data, sync_safety_context, SafetyContext, \
-    validate_expiration_date
+    validate_expiration_date, is_a_remote_mirror
 
 session = requests.session()
 
@@ -216,7 +216,7 @@ def fetch_database(full=False, key=False, db=False, cached=0, proxy=None, teleme
     db_name = "insecure_full.json" if full else "insecure.json"
     for mirror in mirrors:
         # mirror can either be a local path or a URL
-        if mirror.startswith("http://") or mirror.startswith("https://"):
+        if is_a_remote_mirror(mirror):
             data = fetch_database_url(mirror, db_name=db_name, key=key, cached=cached, proxy=proxy, telemetry=telemetry)
         else:
             data = fetch_database_file(mirror, db_name=db_name)
@@ -509,7 +509,7 @@ def get_licenses(key=False, db_mirror=False, cached=0, proxy=None, telemetry=Tru
 
     for mirror in mirrors:
         # mirror can either be a local path or a URL
-        if mirror.startswith("http://") or mirror.startswith("https://"):
+        if is_a_remote_mirror(mirror):
             licenses = fetch_database_url(mirror, db_name=db_name, key=key, cached=cached, proxy=proxy,
                                           telemetry=telemetry)
         else:
@@ -522,20 +522,36 @@ def get_licenses(key=False, db_mirror=False, cached=0, proxy=None, telemetry=Tru
 def get_announcements(key, proxy, telemetry=True):
     LOG.info('Getting announcements')
 
-    body = build_telemetry_data(telemetry=telemetry)
-
     announcements = []
     headers = {}
 
     if key:
         headers["X-Api-Key"] = key
 
     url = f"{API_BASE_URL}announcements/"
+    method = 'post'
+    data = build_telemetry_data(telemetry=telemetry)
+    request_kwargs = {'headers': headers, 'proxies': proxy, 'timeout': 3}
+    data_keyword = 'json'
+
+    source = os.environ.get('SAFETY_ANNOUNCEMENTS_URL', None)
+
+    if source:
+        LOG.debug(f'Getting the announcement from a different source: {source}')
+        url = source
+        method = 'get'
+        data = {
+            'telemetry': json.dumps(build_telemetry_data(telemetry=telemetry))}
+        data_keyword = 'params'
+
+    request_kwargs[data_keyword] = data
+    request_kwargs['url'] = url
 
-    LOG.debug(f'Telemetry body sent: {body}')
+    LOG.debug(f'Telemetry data sent: {data}')
 
     try:
-        r = session.post(url=url, json=body, headers=headers, timeout=2, proxies=proxy)
+        request_func = getattr(session, method)
+        r = request_func(**request_kwargs)
         LOG.debug(r.text)
     except Exception as e:
         LOG.info('Unexpected but HANDLED Exception happened getting the announcements: %s', e)

safety/util.py

@@ -1,3 +1,4 @@
+import json
 import logging
 import os
 import platform
@@ -9,8 +10,10 @@
 from typing import List
 
 import click
+import dparse.parser
 from click import BadParameter
 from dparse.parser import setuptools_parse_requirements_backport as _parse_requirements
+from dparse import parse, filetypes
 from packaging.utils import canonicalize_name
 from packaging.version import parse as parse_version
 from ruamel.yaml import YAML
@@ -21,102 +24,56 @@
 
 LOG = logging.getLogger(__name__)
 
-def iter_lines(fh, lineno=0):
-    for line in fh.readlines()[lineno:]:
-        yield line
 
+def is_a_remote_mirror(mirror):
+    return mirror.startswith("http://") or mirror.startswith("https://")
 
-def parse_line(line):
-    if line.startswith('-e') or line.startswith('http://') or line.startswith('https://'):
-        if "#egg=" in line:
-            line = line.split("#egg=")[-1]
-    if ' --hash' in line:
-        line = line.split(" --hash")[0]
-    return _parse_requirements(line)
 
+def is_supported_by_parser(path):
+    supported_types = (".txt", ".in", ".yml", ".ini", "Pipfile",
+                       "Pipfile.lock", "setup.cfg", "poetry.lock")
+    return path.endswith(supported_types)
 
-def read_requirements(fh, resolve=False):
+
+def read_requirements(fh, resolve=True):
     """
     Reads requirements from a file like object and (optionally) from referenced files.
     :param fh: file like object to read from
     :param resolve: boolean. resolves referenced files.
     :return: generator
     """
     is_temp_file = not hasattr(fh, 'name')
-    for num, line in enumerate(iter_lines(fh)):
-        line = line.strip()
-        if not line:
-            # skip empty lines
-            continue
-        if line.startswith('#') or \
-            line.startswith('-i') or \
-            line.startswith('--index-url') or \
-            line.startswith('--extra-index-url') or \
-            line.startswith('-f') or line.startswith('--find-links') or \
-            line.startswith('--no-index') or line.startswith('--allow-external') or \
-            line.startswith('--allow-unverified') or line.startswith('-Z') or \
-            line.startswith('--always-unzip'):
-            # skip unsupported lines
-            continue
-        elif line.startswith('-r') or line.startswith('--requirement'):
-            # got a referenced file here, try to resolve the path
-            # if this is a tempfile, skip
-            if is_temp_file:
-                continue
-
-            # strip away the recursive flag
-            prefixes = ["-r", "--requirement"]
-            filename = line.strip()
-            for prefix in prefixes:
-                if filename.startswith(prefix):
-                    filename = filename[len(prefix):].strip()
-
-            # if there is a comment, remove it
-            if " #" in filename:
-                filename = filename.split(" #")[0].strip()
-            req_file_path = os.path.join(os.path.dirname(fh.name), filename)
-            if resolve:
-                # recursively yield the resolved requirements
-                if os.path.exists(req_file_path):
-                    with open(req_file_path) as _fh:
-                        for req in read_requirements(_fh, resolve=True):
-                            yield req
-            else:
-                yield RequirementFile(path=req_file_path)
-        else:
-            try:
-                parseable_line = line
-                # multiline requirements are not parseable
-                if "\\" in line:
-                    parseable_line = line.replace("\\", "")
-                    for next_line in iter_lines(fh, num + 1):
-                        parseable_line += next_line.strip().replace("\\", "")
-                        line += "\n" + next_line
-                        if "\\" in next_line:
-                            continue
-                        break
-                req, = parse_line(parseable_line)
-                if len(req.specifier._specs) == 1 and \
-                        next(iter(req.specifier._specs))._spec[0] == "==":
-                    yield Package(name=req.name, version=next(iter(req.specifier._specs))._spec[1],
-                                  found='temp_file' if is_temp_file else fh.name, insecure_versions=[],
-                                  secure_versions=[], latest_version=None,
-                                  latest_version_without_known_vulnerabilities=None, more_info_url=None)
-                else:
-                    try:
-                        fname = fh.name
-                    except AttributeError:
-                        fname = line
-
-                    click.secho(
-                        "Warning: unpinned requirement '{req}' found in {fname}, "
-                        "unable to check.".format(req=req.name,
-                                                  fname=fname),
-                        fg="yellow",
-                        file=sys.stderr
-                    )
-            except ValueError:
-                continue
+    path = None
+    found = 'temp_file'
+    file_type = filetypes.requirements_txt
+
+    if not is_temp_file and is_supported_by_parser(fh.name):
+        path = fh.name
+        found = path
+        file_type = None
+
+    dependency_file = parse(fh.read(), path=path, resolve=resolve,
+                            file_type=file_type)
+    for dep in dependency_file.resolved_dependencies:
+        try:
+            spec = next(iter(dep.specs))._spec
+        except StopIteration:
+            click.secho(
+                f"Warning: unpinned requirement '{dep.name}' found in {path}, "
+                "unable to check.",
+                fg="yellow",
+                file=sys.stderr
+            )
+            return
+
+        version = spec[1]
+        if spec[0] == '==':
+            yield Package(name=dep.name, version=version,
+                          found=found,
+                          insecure_versions=[],
+                          secure_versions=[], latest_version=None,
+                          latest_version_without_known_vulnerabilities=None,
+                          more_info_url=None)
 
 
 def get_proxy_dict(proxy_protocol, proxy_host, proxy_port):