Add support for custom certificate authority and client certificates

.pre-commit-config.yaml

@@ -3,4 +3,3 @@ repos:
  rev: stable
  hooks:
  - id: black
- language_version: python3.6

docs/docs/repositories.md

@@ -74,6 +74,16 @@ export POETRY_HTTP_BASIC_PYPI_PASSWORD=password
 See [Using environment variables](/configuration#using-environment-variables) for more information
 on how to configure Poetry with environment variables.
 
+#### Custom certificate authority and mutual TLS authentication
+Poetry supports repositories that are secured by a custom certificate authority as well as those that require
+certificate-based client authentication. The following will configure the "foo" repository to validate the repository's
+certificate using a custom certificate authority and use a client certificate (note that these config variables do not
+both need to be set):
+```bash
+ poetry config certificates.foo.cert /path/to/ca.pem
+ poetry config certificates.foo.client-cert /path/to/client.pem
+```
+
 ### Install dependencies from a private repository
 
 Now that you can publish to your private repository, you need to be able to
@@ -105,8 +115,10 @@ From now on, Poetry will also look for packages in your private repository.
 
 If your private repository requires HTTP Basic Auth be sure to add the username and
 password to your `http-basic` configuration using the example above (be sure to use the
-same name that is in the `tool.poetry.source` section). Poetry will use these values
-to authenticate to your private repository when downloading or looking for packages.
+same name that is in the `tool.poetry.source` section). If your repository requires either
+a custom certificate authority or client certificates, similarly refer to the example above to configure the
+`certificates` section. Poetry will use these values to authenticate to your private repository when downloading or
+looking for packages.
 
 
 ### Disabling the PyPI repository

poetry/console/commands/config.py

@@ -228,6 +228,27 @@ def handle(self):
 
  return 0
 
+ # handle certs
+ m = re.match(
+ r"(?:certificates)\.([^.]+)\.(cert|client-cert)", self.argument("key")
+ )
+ if m:
+ if self.option("unset"):
+ config.auth_config_source.remove_property(
+ "certificates.{}.{}".format(m.group(1), m.group(2))
+ )
+
+ return 0
+
+ if len(values) == 1:
+ config.auth_config_source.add_property(
+ "certificates.{}.{}".format(m.group(1), m.group(2)), values[0]
+ )
+ else:
+ raise ValueError("You must pass exactly 1 value")
+
+ return 0
+
  raise ValueError("Setting {} does not exist".format(self.argument("key")))
 
  def _handle_single_value(self, source, key, callbacks, values):

poetry/console/commands/publish.py

@@ -1,5 +1,7 @@
 from cleo import option
 
+from poetry.utils._compat import Path
+
 from .command import Command
 
 
@@ -14,6 +16,15 @@ class PublishCommand(Command):
  ),
  option("username", "u", "The username to access the repository.", flag=False),
  option("password", "p", "The password to access the repository.", flag=False),
+ option(
+ "cert", None, "Certificate authority to access the repository.", flag=False
+ ),
+ option(
+ "client-cert",
+ None,
+ "Client certificate to access the repository.",
+ flag=False,
+ ),
  option("build", None, "Build the package before publishing."),
  ]
 
@@ -57,6 +68,15 @@ def handle(self):
 
  self.line("")
 
+ cert = Path(self.option("cert")) if self.option("cert") else None
+ client_cert = (
+ Path(self.option("client-cert")) if self.option("client-cert") else None
+ )
+
  publisher.publish(
- self.option("repository"), self.option("username"), self.option("password")
+ self.option("repository"),
+ self.option("username"),
+ self.option("password"),
+ cert,
+ client_cert,
  )

poetry/factory.py

@@ -233,7 +233,7 @@ def create_legacy_repository(
  ): # type: (Dict[str, str], Config) -> LegacyRepository
  from .repositories.auth import Auth
  from .repositories.legacy_repository import LegacyRepository
- from .utils.helpers import get_http_basic_auth
+ from .utils.helpers import get_client_cert, get_cert, get_http_basic_auth
 
  if "url" in source:
  # PyPI-like repository
@@ -245,12 +245,18 @@ def create_legacy_repository(
  name = source["name"]
  url = source["url"]
  credentials = get_http_basic_auth(auth_config, name)
- if not credentials:
- return LegacyRepository(name, url)
-
- auth = Auth(url, credentials[0], credentials[1])
-
- return LegacyRepository(name, url, auth=auth)
+ if credentials:
+ auth = Auth(url, credentials[0], credentials[1])
+ else:
+ auth = None
+
+ return LegacyRepository(
+ name,
+ url,
+ auth=auth,
+ cert=get_cert(auth_config, name),
+ client_cert=get_client_cert(auth_config, name),
+ )
 
  @classmethod
  def validate(

poetry/installation/pip_installer.py

@@ -52,6 +52,12 @@ def install(self, package, update=False):
  )
  args += ["--trusted-host", parsed.hostname]
 
+ if repository.cert:
+ args += ["--cert", str(repository.cert)]
+
+ if repository.client_cert:
+ args += ["--client-cert", str(repository.client_cert)]
+
  index_url = repository.authenticated_url
 
  args += ["--index-url", index_url]

poetry/masonry/publishing/publisher.py

@@ -1,6 +1,6 @@
 import logging
 
-from poetry.utils.helpers import get_http_basic_auth
+from poetry.utils.helpers import get_client_cert, get_cert, get_http_basic_auth
 
 from .uploader import Uploader
 
@@ -23,7 +23,7 @@ def __init__(self, poetry, io):
  def files(self):
  return self._uploader.files
 
- def publish(self, repository_name, username, password):
+ def publish(self, repository_name, username, password, cert=None, client_cert=None):
  if repository_name:
  self._io.write_line(
  "Publishing <info>{}</info> (<comment>{}</comment>) "
@@ -74,15 +74,21 @@ def publish(self, repository_name, username, password):
  username = auth[0]
  password = auth[1]
 
- # Requesting missing credentials
- if not username:
- username = self._io.ask("Username:")
+ resolved_client_cert = client_cert or get_client_cert(
+ self._poetry.config, repository_name
+ )
+ # Requesting missing credentials but only if there is not a client cert defined.
+ if not resolved_client_cert:
+ if username is None:
+ username = self._io.ask("Username:")
 
- if password is None:
- password = self._io.ask_hidden("Password:")
-
- # TODO: handle certificates
+ if password is None:
+ password = self._io.ask_hidden("Password:")
 
  self._uploader.auth(username, password)
 
- return self._uploader.upload(url)
+ return self._uploader.upload(
+ url,
+ cert=cert or get_cert(self._poetry.config, repository_name),
+ client_cert=resolved_client_cert,
+ )

poetry/masonry/publishing/uploader.py

@@ -3,7 +3,7 @@
 import math
 import re
 
-from typing import List
+from typing import List, Optional
 
 import requests
 
@@ -14,6 +14,7 @@
 from requests_toolbelt.multipart import MultipartEncoder, MultipartEncoderMonitor
 
 from poetry.__version__ import __version__
+from poetry.utils._compat import Path
 from poetry.utils.helpers import normalize_version
 from poetry.utils.patterns import wheel_file_re
 
@@ -94,9 +95,17 @@ def make_session(self):
  def is_authenticated(self):
  return self._username is not None and self._password is not None
 
- def upload(self, url):
+ def upload(
+ self, url, cert=None, client_cert=None
+ ): # type: (str, Optional[Path], Optional[Path]) -> None
  session = self.make_session()
 
+ if cert:
+ session.verify = str(cert)
+
+ if client_cert:
+ session.cert = str(client_cert)
+
  try:
  self._upload(session, url)
  finally:

poetry/repositories/legacy_repository.py

@@ -155,15 +155,17 @@ def clean_link(self, url):
 
 class LegacyRepository(PyPiRepository):
  def __init__(
- self, name, url, auth=None, disable_cache=False
- ): # type: (str, str, Optional[Auth], bool) -> None
+ self, name, url, auth=None, disable_cache=False, cert=None, client_cert=None
+ ): # type: (str, str, Optional[Auth], bool, Optional[Path], Optional[Path]) -> None
  if name == "pypi":
  raise ValueError("The name [pypi] is reserved for repositories")
 
  self._packages = []
  self._name = name
  self._url = url.rstrip("/")
  self._auth = auth
+ self._client_cert = client_cert
+ self._cert = cert
  self._inspector = Inspector()
  self._cache_dir = Path(CACHE_DIR) / "cache" / "repositories" / name
  self._cache = CacheManager(
@@ -186,8 +188,22 @@ def __init__(
  if not url_parts.username and self._auth:
  self._session.auth = self._auth
 
+ if self._cert:
+ self._session.verify = str(self._cert)
+
+ if self._client_cert:
+ self._session.cert = str(self._client_cert)
+
  self._disable_cache = disable_cache
 
+ @property
+ def cert(self): # type: () -> Optional[Path]
+ return self._cert
+
+ @property
+ def client_cert(self): # type: () -> Optional[Path]
+ return self._client_cert
+
  @property
  def authenticated_url(self): # type: () -> str
  if not self._auth:

poetry/utils/helpers.py

@@ -14,6 +14,7 @@
 
 from poetry.config.config import Config
 from poetry.version import Version
+from poetry.utils._compat import Path
 
 _canonicalize_regex = re.compile("[-_]+")
 
@@ -133,6 +134,22 @@ def get_http_basic_auth(
  return None
 
 
+def get_cert(config, repository_name): # type: (Config, str) -> Optional[Path]
+ cert = config.get("certificates.{}.cert".format(repository_name))
+ if cert:
+ return Path(cert)
+ else:
+ return None
+
+
+def get_client_cert(config, repository_name): # type: (Config, str) -> Optional[Path]
+ client_cert = config.get("certificates.{}.client-cert".format(repository_name))
+ if client_cert:
+ return Path(client_cert)
+ else:
+ return None
+
+
 def _on_rm_error(func, path, exc_info):
  os.chmod(path, stat.S_IWRITE)
  func(path)

tests/console/commands/test_config.py

@@ -99,3 +99,26 @@ def test_set_pypi_token(app, config, config_source, auth_config_source):
  tester.execute("--list")
 
  assert "mytoken" == auth_config_source.config["pypi-token"]["pypi"]
+
+
+def test_set_client_cert(app, config_source, auth_config_source, mocker):
+ init = mocker.spy(ConfigSource, "__init__")
+ command = app.find("config")
+ tester = CommandTester(command)
+
+ tester.execute("certificates.foo.client-cert path/to/cert.pem")
+
+ assert (
+ "path/to/cert.pem"
+ == auth_config_source.config["certificates"]["foo"]["client-cert"]
+ )
+
+
+def test_set_cert(app, config_source, auth_config_source, mocker):
+ init = mocker.spy(ConfigSource, "__init__")
+ command = app.find("config")
+ tester = CommandTester(command)
+
+ tester.execute("certificates.foo.cert path/to/ca.pem")
+
+ assert "path/to/ca.pem" == auth_config_source.config["certificates"]["foo"]["cert"]

tests/console/commands/test_publish.py

@@ -1,3 +1,6 @@
+from poetry.utils._compat import Path
+
+
 def test_publish_returns_non_zero_code_for_upload_errors(app, app_tester, http):
  http.register_uri(
  http.POST, "https://upload.pypi.org/legacy/", status=400, body="Bad Request"
@@ -16,3 +19,22 @@ def test_publish_returns_non_zero_code_for_upload_errors(app, app_tester, http):
 """
 
  assert app_tester.io.fetch_output() == expected
+
+
+def test_publish_with_cert(app_tester, mocker):
+ publisher_publish = mocker.patch("poetry.masonry.publishing.Publisher.publish")
+
+ app_tester.execute("publish --cert path/to/ca.pem")
+
+ assert [
+ (None, None, None, Path("path/to/ca.pem"), None)
+ ] == publisher_publish.call_args
+
+
+def test_publish_with_client_cert(app_tester, mocker):
+ publisher_publish = mocker.patch("poetry.masonry.publishing.Publisher.publish")
+
+ app_tester.execute("publish --client-cert path/to/client.pem")
+ assert [
+ (None, None, None, None, Path("path/to/client.pem"))
+ ] == publisher_publish.call_args