feat: strip authentication secrets from download/registry urls (#647)

strip secret authentication information from registries and download urls - in accordance with PEP 610 
> `url` MUST be stripped of any sensitive authentication information, for security reasons.
> [...]
> Additionally, the user:password section of the URL MAY be a well-known, non security sensitive string.


---------

Signed-off-by: Jan Kowalleck <jan.kowalleck@gmail.com>

Author: jkowalleck

cyclonedx_py/_internal/pipenv.py

@@ -33,6 +33,7 @@
 from .utils.args import arparse_split
 from .utils.cdx import make_bom
 from .utils.pyproject import pyproject_file2component
+from .utils.secret import redact_auth_from_url
 
 if TYPE_CHECKING:  # pragma: no cover
     from logging import Logger
@@ -141,9 +142,12 @@ def _make_bom(self, root_c: Optional['Component'],
         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', ())}
+        source_urls: Dict[str, str] = {
+            source['name']: redact_auth_from_url(source['url']).rstrip('/')
+            for source in meta.get('sources', ())
+        }
         if self._pypi_url is not None:
-            source_urls['pypi'] = self._pypi_url.rstrip('/')
+            source_urls['pypi'] = redact_auth_from_url(self._pypi_url).rstrip('/')
 
         all_components: Dict[str, Component] = {}
         if root_c:
@@ -223,7 +227,7 @@ def __make_extrefs(self, name: str, data: 'NameDict', source_urls: Dict[str, str
         vcs_source = self.__package_vcs(data)
         try:
             if vcs_source is not None:
-                vcs_source_url = vcs_source[1]
+                vcs_source_url = redact_auth_from_url(vcs_source[1])
                 yield ExternalReference(
                     comment=f'from {vcs_source[0]}',
                     type=ExternalReferenceType.VCS,
@@ -232,7 +236,7 @@ def __make_extrefs(self, name: str, data: 'NameDict', source_urls: Dict[str, str
                 yield ExternalReference(
                     comment='from file',
                     type=ExternalReferenceType.DISTRIBUTION,
-                    url=XsUri(data['file']),
+                    url=XsUri(redact_auth_from_url(data['file'])),
                     hashes=hashes)
             elif 'path' in data:
                 yield ExternalReference(
@@ -263,16 +267,16 @@ def __purl_qualifiers4lock(self, data: 'NameDict', sourcees: Dict[str, str]) ->
             # 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"]}'
+            qs['vcs_url'] = f'{redact_auth_from_url(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']
+                qs['download_url'] = redact_auth_from_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
+                qs['repository_url'] = redact_auth_from_url(source_url.rstrip('/'))
         return qs
 
     def __make_dependency_graph(self) -> None:

cyclonedx_py/_internal/poetry.py

@@ -33,6 +33,7 @@
 from .cli_common import add_argument_mc_type
 from .utils.cdx import make_bom
 from .utils.poetry import poetry2component
+from .utils.secret import redact_auth_from_url
 from .utils.toml import toml_loads
 
 if TYPE_CHECKING:  # pragma: no cover
@@ -387,17 +388,17 @@ def __purl_qualifiers4lock(self, package: 'NameDict') -> 'NameDict':
             # 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'{source["type"]}+{source["url"]}@' + \
+            qs['vcs_url'] = f'{source["type"]}+{redact_auth_from_url(source["url"])}@' + \
                 source.get('resolved_reference', source.get('reference', ''))
         elif source_type == 'url':
             if '://files.pythonhosted.org/' not in source['url']:
                 # skip PURL bloat, do not add implicit information
-                qs['download_url'] = source['url']
+                qs['download_url'] = redact_auth_from_url(source['url'])
         elif source_type == 'legacy':
             source_url = package['source'].get('url', 'https://pypi.org/simple')
             if '://pypi.org/' not in source_url:
                 # skip PURL bloat, do not add implicit information
-                qs['repository_url'] = source_url
+                qs['repository_url'] = redact_auth_from_url(source_url)
 
         return qs
 
@@ -415,7 +416,7 @@ def __extrefs4lock(self, package: 'NameDict') -> Generator['ExternalReference',
             yield from self.__extrefs4lock_vcs(package)
 
     def __extrefs4lock_legacy(self, package: 'NameDict') -> Generator['ExternalReference', None, None]:
-        source_url = package['source'].get('url', 'https://pypi.org/simple')
+        source_url = redact_auth_from_url(package['source'].get('url', 'https://pypi.org/simple'))
         for file in package['files']:
             try:
                 yield ExternalReference(
@@ -433,7 +434,7 @@ def __extrefs4lock_url(self, package: 'NameDict') -> Generator['ExternalReferenc
             yield ExternalReference(
                 comment='from url',
                 type=ExternalReferenceType.DISTRIBUTION,
-                url=XsUri(package['source']['url']),
+                url=XsUri(redact_auth_from_url(package['source']['url'])),
                 hashes=[HashType.from_composite_str(package['files'][0]['hash'])] if len(package['files']) else None
             )
         except (InvalidUriException, UnknownHashTypeException) as error:  # pragma: nocover
@@ -444,7 +445,7 @@ def __extrefs4lock_file(self, package: 'NameDict') -> Generator['ExternalReferen
             yield ExternalReference(
                 comment='from file',
                 type=ExternalReferenceType.DISTRIBUTION,
-                url=XsUri(package['source']['url']),
+                url=XsUri(redact_auth_from_url(package['source']['url'])),
                 hashes=[HashType.from_composite_str(package['files'][0]['hash'])] if len(package['files']) else None
             )
         except (InvalidUriException, UnknownHashTypeException) as error:  # pragma: nocover
@@ -455,7 +456,7 @@ def __extrefs4lock_directory(self, package: 'NameDict') -> Generator['ExternalRe
             yield ExternalReference(
                 comment='from directory',
                 type=ExternalReferenceType.DISTRIBUTION,
-                url=XsUri(package['source']['url'])
+                url=XsUri(redact_auth_from_url(package['source']['url']))
                 # no hash for a source-directory
             )
         except InvalidUriException as error:  # pragma: nocover
@@ -468,7 +469,7 @@ def __extrefs4lock_vcs(self, package: 'NameDict') -> Generator['ExternalReferenc
             yield ExternalReference(
                 comment='from VCS',
                 type=ExternalReferenceType.VCS,
-                url=XsUri(f'{source["type"]}+{source["url"]}#{vcs_ref}')
+                url=XsUri(f'{source["type"]}+{redact_auth_from_url(source["url"])}#{vcs_ref}')
                 # no hashes, has source.resolved_reference instead, which is a property
             )
         except InvalidUriException as error:  # pragma: nocover

cyclonedx_py/_internal/requirements.py

@@ -18,9 +18,10 @@
 
 from argparse import OPTIONAL, ArgumentParser
 from functools import reduce
+from itertools import chain
 from os import unlink
 from textwrap import dedent
-from typing import TYPE_CHECKING, Any, Generator, List, Optional, Set
+from typing import TYPE_CHECKING, Any, Generator, Iterable, Optional, Set
 
 from cyclonedx.exception.model import InvalidUriException, UnknownHashTypeException
 from cyclonedx.model import ExternalReference, ExternalReferenceType, HashType, Property, XsUri
@@ -33,6 +34,7 @@
 from .utils.cdx import make_bom
 from .utils.io import io2file
 from .utils.pyproject import pyproject_file2component
+from .utils.secret import redact_auth_from_url
 
 if TYPE_CHECKING:  # pragma: no cover
     from logging import Logger
@@ -97,11 +99,11 @@ def make_argument_parser(**kwargs: Any) -> 'ArgumentParser':
     def __init__(self, *,
                  logger: 'Logger',
                  index_url: str,
-                 extra_index_urls: List[str],
+                 extra_index_urls: Iterable[str],
                  **__: Any) -> None:
         self._logger = logger
         self._index_url = index_url
-        self._extra_index_urls = set(extra_index_urls)
+        self._extra_index_urls = tuple(extra_index_urls)
 
     def __call__(self, *,  # type:ignore[override]
                  requirements_file: str,
@@ -136,9 +138,15 @@ def _make_bom(self, root_c: Optional['Component'], rf: 'RequirementsFile') -> 'B
         return bom
 
     def _add_components(self, bom: 'Bom', rf: 'RequirementsFile') -> None:
-        index_url = reduce(lambda c, i: i.options.get('index_url') or c, rf.options, self._index_url)
-        extra_index_urls = self._extra_index_urls.union(*(
-            i.options['extra_index_urls'] for i in rf.options if 'extra_index_urls' in i.options))
+        index_url = redact_auth_from_url(reduce(
+            lambda c, i: i.options.get('index_url') or c, rf.options, self._index_url
+        ).rstrip('/'))
+        extra_index_urls = set(map(
+            lambda u: redact_auth_from_url(u.rstrip('/')),
+            chain(self._extra_index_urls, chain.from_iterable(
+                i.options['extra_index_urls'] for i in rf.options if 'extra_index_urls' in i.options
+            ))
+        ))
         self._logger.debug('index_url = %r', index_url)
         self._logger.debug('extra_index_urls = %r', extra_index_urls)
 
@@ -178,11 +186,12 @@ def _make_component(self, req: 'InstallRequirement',
             elif req.is_url:
                 if '://files.pythonhosted.org/' not in req.link.url:
                     # skip PURL bloat, do not add implicit information
-                    purl_qualifiers['vcs_url' if req.is_vcs_url else 'download_url'] = req.link.url
+                    purl_qualifiers['vcs_url' if req.is_vcs_url else 'download_url'] = redact_auth_from_url(
+                        req.link.url)
                 external_references.append(ExternalReference(
                     comment='explicit dist url',
                     type=ExternalReferenceType.VCS if req.is_vcs_url else ExternalReferenceType.DISTRIBUTION,
-                    url=XsUri(req.link.url),
+                    url=XsUri(redact_auth_from_url(req.link.url)),
                     hashes=hashes))
             else:
                 # url based on https://warehouse.pypa.io/api-reference/legacy.html
@@ -203,7 +212,7 @@ def _make_component(self, req: 'InstallRequirement',
 
         return Component(
             bom_ref=f'requirements-L{req.line_number}',
-            description=f'requirements line {req.line_number}: {req.line}',
+            description=f'requirements line {req.line_number}: {redact_auth_from_url(req.line)}',
             type=ComponentType.LIBRARY,
             name=name or 'unknown',
             version=version,

cyclonedx_py/_internal/utils/secret.py

@@ -0,0 +1,27 @@
+# This file is part of CycloneDX Python Lib
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+# SPDX-License-Identifier: Apache-2.0
+# Copyright (c) OWASP Foundation. All Rights Reserved.
+
+from re import compile as re_compile
+
+_URL_AUTH_MATCHER = re_compile(r'(?<=://)[^/@:]+:[^/@]+@')
+_URL_AUTH_REPLACE = ''  # drop auth - in accordance with PEP 610
+
+
+def redact_auth_from_url(s: str) -> str:
+    # is intended to work on any string that contains an url.
+    return _URL_AUTH_MATCHER.sub(_URL_AUTH_REPLACE, s) \
+        if '@' in s else s

tests/__init__.py

@@ -16,11 +16,12 @@
 # Copyright (c) OWASP Foundation. All Rights Reserved.
 
 
-import re
 from json import dumps as json_dumps
 from os import getenv
 from os.path import dirname, join
 from pathlib import Path
+from re import sub as re_sub
+from sys import stderr
 from typing import Union
 from unittest import TestCase
 from xml.sax.saxutils import escape as xml_escape, quoteattr as xml_quoteattr  # nosec:B406
@@ -31,11 +32,11 @@
 
 RECREATE_SNAPSHOTS = '1' == getenv('CDX_TEST_RECREATE_SNAPSHOTS')
 if RECREATE_SNAPSHOTS:
-    print('!!! WILL RECREATE ALL SNAPSHOTS !!!')
+    print('!!! WILL RECREATE ALL SNAPSHOTS !!!', file=stderr)
 
 INIT_TESTBEDS = '1' != getenv('CDX_TEST_SKIP_INIT_TESTBEDS')
 if INIT_TESTBEDS:
-    print('!!! WILL INIT TESTBEDS !!!')
+    print('!!! WILL INIT TESTBEDS !!!', file=stderr)
 
 _TESTDATA_DIRECTORY = join(dirname(__file__), '_data')
 
@@ -102,15 +103,15 @@ def make_xml_comparable(bom: str) -> str:
         '        <vendor>CycloneDX</vendor>\n'
         '        <name>cyclonedx-bom</name>\n'
         '        <version>thisVersion-testing</version>')
-    bom = re.sub(  # replace metadata.tools.version
+    bom = re_sub(  # replace metadata.tools.version
         '        <vendor>CycloneDX</vendor>\n'
         '        <name>cyclonedx-python-lib</name>\n'
         '        <version>.*?</version>',
         '        <vendor>CycloneDX</vendor>\n'
         '        <name>cyclonedx-python-lib</name>\n'
         '        <version>libVersion-testing</version>',
         bom)
-    bom = re.sub(  # replace metadata.tools.externalReferences
+    bom = re_sub(  # replace metadata.tools.externalReferences
         '        <vendor>CycloneDX</vendor>\n'
         '        <name>cyclonedx-python-lib</name>\n'
         r'        <version>(.*?)</version>\n'
@@ -132,15 +133,15 @@ def make_json_comparable(bom: str) -> str:
         '        "name": "cyclonedx-bom",\n'
         '        "vendor": "CycloneDX",\n'
         '        "version": "thisVersion-testing"')
-    bom = re.sub(  # replace metadata.tools.version
+    bom = re_sub(  # replace metadata.tools.version
         '        "name": "cyclonedx-python-lib",\n'
         '        "vendor": "CycloneDX",\n'
         '        "version": ".*?"',
         '        "name": "cyclonedx-python-lib",\n'
         '        "vendor": "CycloneDX",\n'
         '        "version": "libVersion-testing"',
         bom)
-    bom = re.sub(  # replace metadata.tools.externalReferences
+    bom = re_sub(  # replace metadata.tools.externalReferences
         r'        "externalReferences": \[[\s\S]*?\],\n'
         '        "name": "cyclonedx-python-lib",\n'
         '        "vendor": "CycloneDX"',

tests/_data/infiles/_helpers/pypi-proxy.py

@@ -22,17 +22,17 @@
 This might be needed to play certain setups.
 """
 
-import sys
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from os import unlink
+from sys import argv, stderr
 from urllib.request import urlretrieve
 
 
 class PypiProxyReqHandler(BaseHTTPRequestHandler):
     def do_GET(self) -> None:  # noqa:N802
-        print('> ', self.path, file=sys.stderr)
+        print('PyPI-PROXY > ', self.path, file=stderr)
         p, m = urlretrieve(f'https://pypi.org{self.path}')  # nosec B310
-        print('< ', p, file=sys.stderr)
+        print('PyPI-PROXY < ', p, file=stderr)
         self.send_response(200)
         for k, v in m.items():
             self.send_header(k, v)
@@ -42,11 +42,15 @@ def do_GET(self) -> None:  # noqa:N802
         unlink(p)
 
 
+def make_proxy(port: int) -> ThreadingHTTPServer:
+    server_address = ('', port)
+    return ThreadingHTTPServer(server_address, PypiProxyReqHandler)
+
+
 if __name__ == '__main__':
-    server_address = ('', int(sys.argv[1]) if len(sys.argv) >= 2 else 8080)
-    httpd = ThreadingHTTPServer(server_address, PypiProxyReqHandler)
-    print(f'running PyPI proxy at: {server_address!r}', file=sys.stderr)
+    proxy = make_proxy(int(argv[1]) if len(argv) >= 2 else 8080)
+    print(f'running PyPI proxy at: {proxy.server_address!r}', file=stderr)
     try:
-        httpd.serve_forever()
+        proxy.serve_forever()
     except KeyboardInterrupt:
-        httpd.server_close()
+        proxy.server_close()