Make it possible to specify alternative certificate chains

Twine is amazing in that it actually allows the end user to validate the
SSL when items are being uploaded to a specific pypi repository.

Unfortunately, this validation will only occur for the authorities that
are specified in certifi. This limits two important use cases for
internal and none standard pypi servers:

1. There is no method to use an alternate certificate authority chain,
such as, for instance, is presented from a custom CA root

2. There is no method to use client certificates, which can be an
alternative to username and password.

This commit seeks to redress this, by allowing both of these options to
be specified, with configuration that mirrors that of pip.

tests/test_upload.py

@@ -78,6 +78,7 @@ def test_get_config_old_format(tmpdir):
     try:
         upload.upload(dists=dists, repository="pypi", sign=None, identity=None,
                       username=None, password=None, comment=None,
+                      ca_cert=None, client_cert=None,
                       sign_with=None, config_file=pypirc, skip_existing=False)
     except KeyError as err:
         assert err.args[0] == (

twine/commands/register.py

@@ -23,7 +23,8 @@
 from twine import utils
 
 
-def register(package, repository, username, password, comment, config_file):
+def register(package, repository, username, password, comment, config_file,
+             cert, client_cert):
     config = utils.get_repository_from_config(config_file, repository)
     config["repository"] = utils.normalize_repository_url(
         config["repository"]
@@ -33,8 +34,11 @@ def register(package, repository, username, password, comment, config_file):
 
     username = utils.get_username(username, config)
     password = utils.get_password(password, config)
+    ca_cert = utils.get_cacert(cert, config)
+    client_cert = utils.get_clientcert(client_cert, config)
 
-    repository = Repository(config["repository"], username, password)
+    repository = Repository(config["repository"], username, password,
+                            ca_cert, client_cert)
 
     if not os.path.exists(package):
         raise exc.PackageNotFound(
@@ -78,6 +82,17 @@ def main(args):
         default="~/.pypirc",
         help="The .pypirc config file to use",
     )
+    parser.add_argument(
+        "--cert",
+        metavar="path",
+        help="Path to alternate CA bundle",
+    )
+    parser.add_argument(
+        "--client-cert",
+        metavar="path",
+        help="Path to SSL client certificate, a single file containing the "
+             "private key and the certificate in PEM forma",
+    )
     parser.add_argument(
         "package",
         metavar="package",

twine/commands/upload.py

@@ -62,7 +62,7 @@ def skip_upload(response, skip_existing, package):
 
 
 def upload(dists, repository, sign, identity, username, password, comment,
-           sign_with, config_file, skip_existing):
+           sign_with, config_file, skip_existing, cert, client_cert):
     # Check that a nonsensical option wasn't given
     if not sign and identity:
         raise ValueError("sign must be given along with identity")
@@ -85,8 +85,11 @@ def upload(dists, repository, sign, identity, username, password, comment,
 
     username = utils.get_username(username, config)
     password = utils.get_password(password, config)
+    ca_cert = utils.get_cacert(cert, config)
+    client_cert = utils.get_clientcert(client_cert, config)
 
-    repository = Repository(config["repository"], username, password)
+    repository = Repository(config["repository"], username, password,
+                            ca_cert, client_cert)
 
     for filename in uploads:
         package = PackageFile.from_filename(filename, comment)
@@ -167,6 +170,17 @@ def main(args):
         action="store_true",
         help="Continue uploading files if one already exists",
     )
+    parser.add_argument(
+        "--cert",
+        metavar="path",
+        help="Path to alternate CA bundle",
+    )
+    parser.add_argument(
+        "--client-cert",
+        metavar="path",
+        help="Path to SSL client certificate, a single file containing the "
+             "private key and the certificate in PEM forma",
+    )
     parser.add_argument(
         "dists",
         nargs="+",

twine/repository.py

@@ -21,10 +21,16 @@
 
 
 class Repository(object):
-    def __init__(self, repository_url, username, password):
+    def __init__(self, repository_url, username, password,
+                 ca_root=None, client_cert=None):
         self.url = repository_url
         self.session = requests.session()
         self.session.auth = (username, password)
+        self.ssl_options = {}
+        if ca_root:
+            self.ssl_options['verify'] = ca_root
+        if client_cert:
+            self.ssl_options['cert'] = client_cert
 
     def close(self):
         self.session.close()
@@ -57,6 +63,7 @@ def register(self, package):
             data=encoder,
             allow_redirects=False,
             headers={'Content-Type': encoder.content_type},
+            **self.ssl_options
         )
         # Bug 28. Try to silence a ResourceWarning by releasing the socket.
         resp.close()
@@ -86,6 +93,7 @@ def upload(self, package):
                 data=encoder,
                 allow_redirects=False,
                 headers={'Content-Type': encoder.content_type},
+                **self.ssl_options
             )
 
         return resp

twine/utils.py

@@ -116,14 +116,15 @@ def normalize_repository_url(url):
     return urlunparse(parsed)
 
 
-def get_userpass_value(cli_value, config, key, prompt_strategy):
+def get_userpass_value(cli_value, config, key, prompt_strategy=None):
     """Gets the username / password from config.
 
     Uses the following rules:
 
     1. If it is specified on the cli (`cli_value`), use that.
     2. If `config[key]` is specified, use that.
-    3. Otherwise prompt using `prompt_strategy`.
+    3. If `prompt_strategy`, prompt using `prompt_strategy`.
+    4. Otherwise return None
 
     :param cli_value: The value supplied from the command line or `None`.
     :type cli_value: unicode or `None`
@@ -140,8 +141,10 @@ def get_userpass_value(cli_value, config, key, prompt_strategy):
         return cli_value
     elif config.get(key):
         return config[key]
-    else:
+    elif prompt_strategy:
         return prompt_strategy()
+    else:
+        return None
 
 
 def password_prompt(prompt_text):  # Always expects unicode for our own sanity
@@ -161,3 +164,11 @@ def password_prompt(prompt_text):  # Always expects unicode for our own sanity
     key='password',
     prompt_strategy=password_prompt('Enter your password: '),
 )
+get_cacert = functools.partial(
+    get_userpass_value,
+    key='ca_cert',
+)
+get_clientcert = functools.partial(
+    get_userpass_value,
+    key='client_cert',
+)