Handle unexpected key type migration.

certbot-ci/certbot_integration_tests/certbot_tests/test_main.py

@@ -499,6 +499,16 @@ def test_renew_with_ec_keys(context):
     assert_elliptic_key(key2, SECP384R1)
     assert 280 < os.stat(key2).st_size < 320  # ec keys of 384 bits are ~310 bytes
 
+    # We expect here that the command will fail because without --key-type specified,
+    # Certbot must errored out to prevent changing an existing certificate key type,
+    # without explicit user consent (by specifying both --cert-name and --key-type).
+    with pytest.raises(subprocess.CalledProcessError):
+        context.certbot([
+            'certonly',
+            '--force-renewal',
+            '-d', certname
+        ])
+
 
 def test_ocsp_must_staple(context):
     """Test that OCSP Must-Staple is correctly set in the generated certificate."""

certbot/certbot/_internal/main.py

@@ -11,7 +11,7 @@ import josepy as jose
 import zope.component
 
 from acme import errors as acme_errors
-from acme.magic_typing import Union, Iterable, Optional  # pylint: disable=unused-import
+from acme.magic_typing import Union, Iterable, Optional, List, Tuple  # pylint: disable=unused-import
 import certbot
 from certbot import crypto_util
 from certbot import errors
@@ -130,7 +130,37 @@ def _get_and_save_cert(le_client, config, domains=None, certname=None, lineage=N
     return lineage
 
 
-def _handle_subset_cert_request(config, domains, cert):
+def _handle_unexpected_key_type_migration(config, cert):
+    # type: (configuration.NamespaceConfig, storage.RenewableCert) -> None
+    """
+    This function ensures that the user will not implicitly migrate an existing key
+    from one type to another in the situation where a certificate for that lineage
+    already exist and they have not provided explicitly --key-type and --cert-name.
+    :param config: Current configuration provided by the client
+    :param cert: Matching certificate that could be renewed
+    """
+    if (config.verb in ["certonly", "run"]
+            and (not cli.set_by_cli("key_type") or not cli.set_by_cli("certname"))):
+
+        new_key_type = config.key_type.upper()
+        cur_key_type = cert.private_key_type.upper()
+
+        if new_key_type != cur_key_type:
+            logger.error("A certificate already exists for that name, the list of provided "
+                         "domains or a subset of this list. This certificate uses a key of %s "
+                         "type, and you have not provided the --key-type flag. By default in this "
+                         "case Certbot would generate a new certificate with a key of %s type. "
+                         "Please confirm that you really want to change the type of the key by "
+                         "setting both --cert-name and --key-type CLI flags.",
+                         cur_key_type, new_key_type)
+            raise errors.Error("Command canceled.")
+
+
+def _handle_subset_cert_request(config,  # type: configuration.NamespaceConfig
+                                domains,  # type: List[str]
+                                cert  # type: storage.RenewableCert
+                                ):
+    # type: (...) -> Tuple[str, Optional[storage.RenewableCert]]
     """Figure out what to do if a previous cert had a subset of the names now requested
 
     :param config: Configuration object
@@ -147,6 +177,8 @@ def _handle_subset_cert_request(config, domains, cert):
     :rtype: `tuple` of `str`
 
     """
+    _handle_unexpected_key_type_migration(config, cert)
+
     existing = ", ".join(cert.names())
     question = (
         "You have an existing certificate that contains a portion of "
@@ -177,7 +209,10 @@ def _handle_subset_cert_request(config, domains, cert):
     raise errors.Error(USER_CANCELLED)
 
 
-def _handle_identical_cert_request(config, lineage):
+def _handle_identical_cert_request(config,  # type: configuration.NamespaceConfig
+                                   lineage,  # type: storage.RenewableCert
+                                   ):
+    # type: (...) -> Tuple[str, Optional[storage.RenewableCert]]
     """Figure out what to do if a lineage has the same names as a previously obtained one
 
     :param config: Configuration object
@@ -191,6 +226,8 @@ def _handle_identical_cert_request(config, lineage):
     :rtype: `tuple` of `str`
 
     """
+    _handle_unexpected_key_type_migration(config, lineage)
+
     if not lineage.ensure_deployed():
         return "reinstall", lineage
     if renewal.should_renew(config, lineage):
@@ -266,6 +303,7 @@ def _find_lineage_for_domains(config, domains):
         return _handle_subset_cert_request(config, domains, subset_names_cert)
     return None, None
 
+
 def _find_cert(config, domains, certname):
     """Finds an existing certificate object given domains and/or a certificate name.
 
@@ -289,7 +327,12 @@ def _find_cert(config, domains, certname):
         logger.info("Keeping the existing certificate")
     return (action != "reinstall"), lineage
 
-def _find_lineage_for_domains_and_certname(config, domains, certname):
+
+def _find_lineage_for_domains_and_certname(config,  # type: configuration.NamespaceConfig
+                                           domains,  # type: List[str]
+                                           certname  # type: str
+                                           ):
+    # type: (...) -> Tuple[str, Optional[storage.RenewableCert]]
     """Find appropriate lineage based on given domains and/or certname.
 
     :param config: Configuration object
@@ -316,8 +359,9 @@ def _find_lineage_for_domains_and_certname(config, domains, certname):
     if lineage:
         if domains:
             if set(cert_manager.domains_for_certname(config, certname)) != set(domains):
+                _handle_unexpected_key_type_migration(config, lineage)
                 _ask_user_to_confirm_new_names(config, domains, certname,
-                    lineage.names()) # raises if no
+                                               lineage.names())  # raises if no
                 return "renew", lineage
         # unnecessarily specified domains or no domains specified
         return _handle_identical_cert_request(config, lineage)
@@ -386,6 +430,7 @@ def _ask_user_to_confirm_new_names(config, new_domains, certname, old_domains):
     if not obj.yesno(msg, "Update cert", "Cancel", default=True):
         raise errors.ConfigurationError("Specified mismatched cert name and domains.")
 
+
 def _find_domains_or_certname(config, installer, question=None):
     """Retrieve domains and certname from config or user input.
 

certbot/tests/main_test.py

@@ -49,14 +49,47 @@ RSA2048_KEY_PATH = test_util.vector_path('rsa2048_key.pem')
 SS_CERT_PATH = test_util.vector_path('cert_2048.pem')
 
 
-class TestHandleIdenticalCerts(unittest.TestCase):
-    """Test for certbot._internal.main._handle_identical_cert_request"""
-    def test_handle_identical_cert_request_pending(self):
+class TestHandleCerts(unittest.TestCase):
+    """Test for certbot._internal.main._handle_* methods"""
+    @mock.patch("certbot._internal.main._handle_unexpected_key_type_migration")
+    def test_handle_identical_cert_request_pending(self, mock_handle_migration):
         mock_lineage = mock.Mock()
         mock_lineage.ensure_deployed.return_value = False
         # pylint: disable=protected-access
         ret = main._handle_identical_cert_request(mock.Mock(), mock_lineage)
         self.assertEqual(ret, ("reinstall", mock_lineage))
+        self.assertTrue(mock_handle_migration.called)
+
+    @mock.patch("certbot._internal.main._handle_unexpected_key_type_migration")
+    def test_handle_subset_cert_request(self, mock_handle_migration):
+        mock_config = mock.Mock()
+        mock_config.expand = True
+        mock_lineage = mock.Mock()
+        mock_lineage.names.return_value = ["dummy1", "dummy2"]
+        ret = main._handle_subset_cert_request(mock_config, ["dummy1"], mock_lineage)
+        self.assertEqual(ret, ("renew", mock_lineage))
+        self.assertTrue(mock_handle_migration.called)
+
+    @mock.patch("certbot._internal.main.cli.set_by_cli")
+    def test_handle_unexpected_key_type_migration(self, mock_set):
+        config = mock.Mock()
+        config.verb = "certonly"
+        config.key_type = "rsa"
+        cert = mock.Mock()
+        cert.private_key_type = "ecdsa"
+
+        mock_set.return_value = True
+        main._handle_unexpected_key_type_migration(config, cert)
+
+        mock_set.return_value = False
+        with self.assertRaises(errors.Error) as raised:
+            main._handle_unexpected_key_type_migration(config, cert)
+        self.assertTrue("Command canceled." in str(raised.exception))
+
+        mock_set.side_effect = lambda var: var != "key_type"
+        with self.assertRaises(errors.Error) as raised:
+            main._handle_unexpected_key_type_migration(config, cert)
+        self.assertTrue("Command canceled." in str(raised.exception))
 
 
 class RunTest(test_util.ConfigTestCase):
@@ -163,9 +196,10 @@ class CertonlyTest(unittest.TestCase):
     @mock.patch('certbot._internal.cert_manager.lineage_for_certname')
     @mock.patch('certbot._internal.cert_manager.domains_for_certname')
     @mock.patch('certbot._internal.renewal.renew_cert')
+    @mock.patch('certbot._internal.main._handle_unexpected_key_type_migration')
     @mock.patch('certbot._internal.main._report_new_cert')
     def test_find_lineage_for_domains_and_certname(self, mock_report_cert,
-        mock_renew_cert, mock_domains, mock_lineage):
+        mock_handle_type, mock_renew_cert, mock_domains, mock_lineage):
         domains = ['example.com', 'test.org']
         mock_domains.return_value = domains
         mock_lineage.names.return_value = domains
@@ -175,6 +209,7 @@ class CertonlyTest(unittest.TestCase):
         self.assertTrue(mock_domains.call_count == 1)
         self.assertTrue(mock_renew_cert.call_count == 1)
         self.assertTrue(mock_report_cert.call_count == 1)
+        self.assertTrue(mock_handle_type.call_count == 1)
 
         # user confirms updating lineage with new domains
         self._call(('certonly --webroot -d example.com -d test.com '
@@ -183,11 +218,12 @@ class CertonlyTest(unittest.TestCase):
         self.assertTrue(mock_domains.call_count == 2)
         self.assertTrue(mock_renew_cert.call_count == 2)
         self.assertTrue(mock_report_cert.call_count == 2)
+        self.assertTrue(mock_handle_type.call_count == 2)
 
         # error in _ask_user_to_confirm_new_names
         self.mock_get_utility().yesno.return_value = False
         self.assertRaises(errors.ConfigurationError, self._call,
-            ('certonly --webroot -d example.com -d test.com --cert-name example.com').split())
+            'certonly --webroot -d example.com -d test.com --cert-name example.com'.split())
 
     @mock.patch('certbot._internal.cert_manager.domains_for_certname')
     @mock.patch('certbot.display.ops.choose_names')
@@ -982,6 +1018,7 @@ class MainTest(test_util.ConfigTestCase):
         mock_lineage.should_autorenew.return_value = due_for_renewal
         mock_lineage.has_pending_deployment.return_value = False
         mock_lineage.names.return_value = ['isnot.org']
+        mock_lineage.private_key_type = 'RSA'
         mock_certr = mock.MagicMock()
         mock_key = mock.MagicMock(pem='pem_key')
         mock_client = mock.MagicMock()