rewrite: pipenv

cyclonedx_py/_internal/cli.py

@@ -177,7 +177,8 @@ def _validate(self, output: str) -> bool:
             self._logger.warning('Validation skipped.')
             return False
 
-        self._logger.info('Validating result to schema: %s/%s', self._schema_version.to_version(), self._output_format.name)
+        self._logger.info('Validating result to schema: %s/%s',
+                          self._schema_version.to_version(), self._output_format.name)
         from cyclonedx.validation import make_schemabased_validator
 
         validation_error = make_schemabased_validator(
@@ -253,7 +254,6 @@ def run(*, argv: Optional[List[str]] = None, **kwargs: Any) -> int:
     logger.addHandler(lh)
 
     logger.debug('args: %s', args)
-
     try:
         Command(**args, logger=logger)(**args)
     except Exception as error:

cyclonedx_py/_internal/pipenv.py

@@ -16,48 +16,287 @@
 # Copyright (c) OWASP Foundation. All Rights Reserved.
 
 
-from typing import TYPE_CHECKING, Any, BinaryIO
+from typing import TYPE_CHECKING, Any, Dict, Generator, List, Optional, Set, Tuple
 
 from . import BomBuilder
 
 if TYPE_CHECKING:  # pragma: no cover
     from argparse import ArgumentParser
     from logging import Logger
 
+    from cyclonedx.model import ExternalReference
     from cyclonedx.model.bom import Bom
+    from cyclonedx.model.component import Component, ComponentType
+
+    NameDict = Dict[str, Any]
 
 
 # !!! be as lazy loading as possible, as greedy as needed
 # TODO: measure with `/bin/time -v` for max resident size and see if this changes when global imports are used
 
 
 class PipenvBB(BomBuilder):
+    __LOCKFILE_META = '_meta'
 
     @staticmethod
     def make_argument_parser(**kwargs: Any) -> 'ArgumentParser':
-        from argparse import OPTIONAL, ArgumentParser, FileType
+        from argparse import OPTIONAL, ArgumentParser
+        from os import getenv
+
+        from cyclonedx.model.component import ComponentType
+
+        from .utils.args import argparse_type4enum, arpaese_split
 
         p = ArgumentParser(description='Build an SBOM from Pipenv',
                            **kwargs)
-        p.add_argument('lock',
-                       metavar='lock-file',
-                       help='I HELP TODO (default: %(default)s)',
+        # the args shall mimic the ones from Pipenv
+        # see also: https://pipenv.pypa.io/en/latest/configuration.html
+        p.add_argument('--categories',
+                       metavar='CATEGORIES',
+                       dest='categories',
+                       type=arpaese_split((' ', ',')),
+                       default=[])
+        p.add_argument('-d', '--dev',
+                       help='both develop and default packages [env var: PIPENV_DEV]',
+                       action='store_true',
+                       dest='dev',
+                       default=getenv('PIPENV_DEV', '').lower() in ('1', 'true', 'yes'))
+        p.add_argument('--pypi-mirror',
+                       metavar='URL',
+                       help='Specify a PyPI mirror [env var: PIPENV_PYPI_MIRROR]',
+                       dest='pypi_url',
+                       default=getenv('PIPENV_PYPI_MIRROR'))
+        p.add_argument('--pyproject',
+                       metavar='pyproject.toml',
+                       help="Path to the root component's `pyproject.toml` according to PEP621",
+                       dest='pyproject_file',
+                       default=None)
+        _mc_types = [ComponentType.APPLICATION,
+                     ComponentType.FIRMWARE,
+                     ComponentType.LIBRARY]
+        p.add_argument('--mc-type',
+                       metavar='TYPE',
+                       help='Type of the main component'
+                            f' {{choices: {", ".join(t.value for t in _mc_types)}}}'
+                            ' (default: %(default)s)',
+                       dest='mc_type',
+                       choices=_mc_types,
+                       type=argparse_type4enum(ComponentType),
+                       default=ComponentType.APPLICATION.value)
+        p.add_argument('project_directory',
+                       metavar='project-directory',
+                       help='The project directory for Pipenv (default: current working directory)\n'
+                            'Unlike Pipenv tool, there is no auto-detection in this very tool. '  # yet
+                            'Please provide the actual directory that contains `Pipfile` and `Pipfile.lock`',
                        nargs=OPTIONAL,
-                       type=FileType('rb'),
-                       default='Pipfile.lock')
+                       default='.')
         return p
 
     def __init__(self, *,
                  logger: 'Logger',
+                 pypi_url: Optional[str],
                  **__: Any) -> None:
         self._logger = logger
+        self._pypi_url = pypi_url or None  # ignore empty strings
 
     def __call__(self, *,  # type:ignore[override]
-                 lock: BinaryIO,
+                 project_directory: str,
+                 categories: List[str],
+                 dev: bool,
+                 pyproject_file: Optional[str],
+                 mc_type: 'ComponentType',
                  **__: Any) -> 'Bom':
+        from json import loads as json_loads
+        from os.path import join
+
+        # the group-args shall mimic the ones from Pipenv, which uses (comma and/or space)-separated lists
+        # values be like: 'foo bar,bazz' -> ['foo', 'bar', 'bazz']
+        lock_groups: Set[str] = set()
+        if len(categories) == 0:
+            lock_groups.add('default')
+            if dev:
+                lock_groups.add('develop')
+        else:
+            lock_groups.update(categories)
+            lock_groups.discard(self.__LOCKFILE_META)
+            if 'packages' in lock_groups:
+                # replace UI-category with Lock-group
+                lock_groups.remove('packages')
+                lock_groups.add('default')
+            if 'dev-packages' in lock_groups:
+                # replace UI-category with Lock-group
+                lock_groups.remove('dev-packages')
+                lock_groups.add('develop')
+
+        lock_file = join(project_directory, 'Pipfile.lock')
+        try:
+            lock = open(lock_file, 'rt', encoding='utf8', errors='replace')
+        except OSError as err:
+            raise ValueError(f'Could not open lock file: {lock_file}') from err
+        with lock:
+            if pyproject_file is None:
+                rc = None
+            else:
+                from .utils.pep621 import pyproject_file2component
+                rc = pyproject_file2component(pyproject_file, type=mc_type)
+                rc.bom_ref.value = 'root-component'
+
+            return self._make_bom(rc,
+                                  json_loads(lock.read()),
+                                  lock_groups)
+
+    def _make_bom(self, root_c: Optional['Component'],
+                  locker: 'NameDict', use_groups: Set[str]) -> 'Bom':
+        from cyclonedx.model import Property
+        from cyclonedx.model.component import Component, ComponentType
+        from packageurl import PackageURL
+
+        from . import PropertyName
         from .utils.bom import make_bom
 
+        self._logger.debug('use_groups: %r', use_groups)
+
         bom = make_bom()
-        # TODO
+        bom.metadata.component = root_c
+        self._logger.debug('root-component: %r', root_c)
+
+        meta: NameDict = locker[self.__LOCKFILE_META]
+        source_urls: Dict[str, str] = {source['name']: source['url'].rstrip('/') for source in meta.get('sources', ())}
+        if self._pypi_url is not None:
+            source_urls['pypi'] = self._pypi_url.rstrip('/')
+
+        all_components: Dict[str, Component] = {}
+        if root_c:
+            # root for self-installs
+            all_components[root_c.name] = root_c
+        for group_name in use_groups:
+            self._logger.debug('processing group %r ...', group_name)
+            for package_name, package_data in locker.get(group_name, {}).items():
+                if package_name in all_components:
+                    component = all_components[package_name]
+                    self._logger.info('existing component for package %r', package_name)
+                else:
+                    component = all_components[package_name] = Component(
+                        bom_ref=f'{package_name}{package_data.get("version", "")}',
+                        type=ComponentType.LIBRARY,
+                        name=package_name,
+                        version=package_data['version'][2:] if 'version' in package_data else None,
+                        external_references=self.__make_extrefs(package_name, package_data, source_urls),
+                    )
+                    component.purl = PackageURL(type='pypi',
+                                                name=component.name,
+                                                version=component.version,
+                                                qualifiers=self.__purl_qualifiers4lock(package_data, source_urls)
+                                                ) if not self.__is_local(package_data) else None
+                    self._logger.info('add component for package %r', package_name)
+                    self._logger.debug('add component: %r', component)
+                    bom.components.add(component)
+                component.properties.add(Property(
+                    name=PropertyName.PipenvCategory.value,
+                    value=group_name
+                ))
+                component.properties.update(Property(
+                    name=PropertyName.PackageExtra.value,
+                    value=package_extra
+                ) for package_extra in package_data.get('extras', ()))
 
         return bom
+
+    def __is_local(self, data: 'NameDict') -> bool:
+        if 'file' in data:
+            location: str = data['file']
+        elif 'path' in data:
+            location = data['path']
+        else:
+            return False
+        # schema length is expected to be at least 2 chars, to prevent confusion with Windows drive letters `C:\`
+        might_have_schema = location.find(':', 2)
+        if might_have_schema <= 0:
+            return True
+        maybe_schema = location[:might_have_schema]
+        # example data
+        # - file:../MyProject
+        # - file:///home/user/projects/MyProject
+        # - git+file:///home/user/projects/MyProject
+        # - http://acme.org/MyProject/files/foo-bar.tar.gz
+        return maybe_schema == 'file' or maybe_schema.endswith('+file')
+
+    __VCS_TYPES = ('git', 'hg', 'svn', 'bzr')
+    """ VCS types supported by pip.
+        see https://pip.pypa.io/en/latest/topics/vcs-support/#vcs-support
+    """
+
+    def __package_vcs(self, data: 'NameDict') -> Optional[Tuple[str, str]]:
+        for vct in self.__VCS_TYPES:
+            if vct in data:
+                url: str = data[vct]
+                hash_pos = url.find('#')
+                # remove install-annotations, which are behind a `#`
+                return vct, url[:hash_pos] if hash_pos >= 0 else url
+        return None
+
+    def __make_extrefs(self, name: str, data: 'NameDict', source_urls: Dict[str, str]
+                       ) -> Generator['ExternalReference', None, None]:
+        from cyclonedx.exception.model import InvalidUriException, UnknownHashTypeException
+        from cyclonedx.model import ExternalReference, ExternalReferenceType, HashType, XsUri
+
+        hashes = (HashType.from_composite_str(package_hash)
+                  for package_hash
+                  in data.get('hashes', ()))
+        vcs_source = self.__package_vcs(data)
+        try:
+            if vcs_source is not None:
+                vcs_source_url = vcs_source[1]
+                yield ExternalReference(
+                    comment=f'from {vcs_source[0]}',
+                    type=ExternalReferenceType.VCS,
+                    url=XsUri(f'{vcs_source_url}#{data.get("ref", "")}'))
+            elif 'file' in data:
+                yield ExternalReference(
+                    comment='from file',
+                    type=ExternalReferenceType.DISTRIBUTION,
+                    url=XsUri(data['file']),
+                    hashes=hashes)
+            elif 'path' in data:
+                yield ExternalReference(
+                    comment='from path',
+                    type=ExternalReferenceType.DISTRIBUTION,
+                    url=XsUri(data['path']),
+                    hashes=hashes)
+            elif 'index' in data:
+                yield ExternalReference(
+                    comment=f'from explicit index: {data["index"]}',
+                    type=ExternalReferenceType.DISTRIBUTION,
+                    url=XsUri(f'{source_urls[data["index"]]}/{name}/'),
+                    hashes=hashes)
+            else:
+                yield ExternalReference(
+                    comment='from implicit index: pypi',
+                    type=ExternalReferenceType.DISTRIBUTION,
+                    url=XsUri(f'{source_urls["pypi"]}/{name}/'),
+                    hashes=hashes)
+        except (InvalidUriException, UnknownHashTypeException, KeyError) as error:  # pragma: nocover
+            self._logger.debug('skipped dist-extRef for: %r', name, exc_info=error)
+
+    def __purl_qualifiers4lock(self, data: 'NameDict', sourcees: Dict[str, str]) -> 'NameDict':
+        # see https://github.com/package-url/purl-spec/blob/master/PURL-SPECIFICATION.rst
+        qs = {}
+        vcs_source = self.__package_vcs(data)
+        if vcs_source is not None:
+            # see section 3.7.4 in https://github.com/spdx/spdx-spec/blob/cfa1b9d08903/chapters/3-package-information.md
+            # > For version-controlled files, the VCS location syntax is similar to a URL and has the:
+            # > `<vcs_tool>+<transport>://<host_name>[/<path_to_repository>][@<revision_tag_or_branch>][#<sub_path>]`
+            qs['vcs_url'] = f'{vcs_source[1]}@{data["ref"]}'
+        elif 'file' in data:
+            if '://files.pythonhosted.org/' not in data['file']:
+                # skip PURL bloat, do not add implicit information
+                qs['download_url'] = data['file']
+        elif 'index' in data:
+            source_url = sourcees.get(data['index'], 'https://pypi.org/simple')
+            if '://pypi.org/' not in source_url:
+                # skip PURL bloat, do not add implicit information
+                qs['repository_url'] = source_url
+        return qs
+
+    def __make_dependency_graph(self) -> None:
+        pass  # TODO: gather info from `pipenv graph --json-tree` and work with it