Merge pull request #1860 from letsencrypt/wip-1546

Implement <Reinstall> option in addition to <Renew> or <Cancel>

letsencrypt/cli.py

@@ -205,76 +205,131 @@ def _find_duplicative_certs(config, domains):
         if candidate_names == set(domains):
             identical_names_cert = candidate_lineage
         elif candidate_names.issubset(set(domains)):
-            subset_names_cert = candidate_lineage
+            # This logic finds and returns the largest subset-names cert
+            # in the case where there are several available.
+            if subset_names_cert is None:
+                subset_names_cert = candidate_lineage
+            elif len(candidate_names) > len(subset_names_cert.names()):
+                subset_names_cert = candidate_lineage
 
     return identical_names_cert, subset_names_cert
 
 
 def _treat_as_renewal(config, domains):
-    """Determine whether or not the call should be treated as a renewal.
+    """Determine whether there are duplicated names and how to handle them.
 
-    :returns: RenewableCert or None if renewal shouldn't occur.
-    :rtype: :class:`.storage.RenewableCert`
+    :returns: Two-element tuple containing desired new-certificate behavior as
+              a string token ("reinstall", "renew", or "newcert"), plus either
+              a RenewableCert instance or None if renewal shouldn't occur.
 
     :raises .Error: If the user would like to rerun the client again.
 
     """
-    renewal = False
-
     # Considering the possibility that the requested certificate is
     # related to an existing certificate.  (config.duplicate, which
     # is set with --duplicate, skips all of this logic and forces any
     # kind of certificate to be obtained with renewal = False.)
-    if not config.duplicate:
-        ident_names_cert, subset_names_cert = _find_duplicative_certs(
-            config, domains)
-        # I am not sure whether that correctly reads the systemwide
-        # configuration file.
-        question = None
-        if ident_names_cert is not None:
-            question = (
-                "You have an existing certificate that contains exactly the "
-                "same domains you requested (ref: {0}){br}{br}Do you want to "
-                "renew and replace this certificate with a newly-issued one?"
-            ).format(ident_names_cert.configfile.filename, br=os.linesep)
-        elif subset_names_cert is not None:
-            question = (
-                "You have an existing certificate that contains a portion of "
-                "the domains you requested (ref: {0}){br}{br}It contains these "
-                "names: {1}{br}{br}You requested these names for the new "
-                "certificate: {2}.{br}{br}Do you want to replace this existing "
-                "certificate with the new certificate?"
-            ).format(subset_names_cert.configfile.filename,
-                     ", ".join(subset_names_cert.names()),
-                     ", ".join(domains),
-                     br=os.linesep)
-        if question is None:
-            # We aren't in a duplicative-names situation at all, so we don't
-            # have to tell or ask the user anything about this.
-            pass
-        elif config.renew_by_default or zope.component.getUtility(
-                interfaces.IDisplay).yesno(question, "Replace", "Cancel"):
-            renewal = True
-        else:
-            reporter_util = zope.component.getUtility(interfaces.IReporter)
-            reporter_util.add_message(
-                "To obtain a new certificate that {0} an existing certificate "
-                "in its domain-name coverage, you must use the --duplicate "
-                "option.{br}{br}For example:{br}{br}{1} --duplicate {2}".format(
-                    "duplicates" if ident_names_cert is not None else
-                    "overlaps with",
-                    sys.argv[0], " ".join(sys.argv[1:]),
-                    br=os.linesep
-                ),
-                reporter_util.HIGH_PRIORITY)
-            raise errors.Error(
-                "User did not use proper CLI and would like "
-                "to reinvoke the client.")
+    if config.duplicate:
+        return "newcert", None
+    # TODO: Also address superset case
+    ident_names_cert, subset_names_cert = _find_duplicative_certs(config, domains)
+    # XXX ^ schoen is not sure whether that correctly reads the systemwide
+    # configuration file.
+    if ident_names_cert is None and subset_names_cert is None:
+        return "newcert", None
 
-        if renewal:
-            return ident_names_cert if ident_names_cert is not None else subset_names_cert
+    if ident_names_cert is not None:
+        return _handle_identical_cert_request(config, ident_names_cert)
+    elif subset_names_cert is not None:
+        return _handle_subset_cert_request(config, domains, subset_names_cert)
 
-    return None
+def _handle_identical_cert_request(config, cert):
+    """Figure out what to do if a cert has the same names as a perviously obtained one
+
+    :param storage.RenewableCert cert:
+
+    :returns: Tuple of (string, cert_or_None) as per _treat_as_renewal
+    :rtype: tuple
+
+    """
+    if config.renew_by_default:
+        logger.info("Auto-renewal forced with --renew-by-default...")
+        return "renew", cert
+    if cert.should_autorenew(interactive=True):
+        logger.info("Cert is due for renewal, auto-renewing...")
+        return "renew", cert
+    if config.reinstall:
+        # Set with --reinstall, force an identical certificate to be
+        # reinstalled without further prompting.
+        return "reinstall", cert
+
+    question = (
+        "You have an existing certificate that contains exactly the same "
+        "domains you requested and isn't close to expiry."
+        "{br}(ref: {0}){br}{br}What would you like to do?"
+    ).format(cert.configfile.filename, br=os.linesep)
+
+    if config.verb == "run":
+        keep_opt = "Attempt to reinstall this existing certificate"
+    elif config.verb == "certonly":
+        keep_opt = "Keep the existing certificate for now"
+    choices = [keep_opt,
+               "Renew & replace the cert (limit ~5 per 7 days)",
+               "Cancel this operation and do nothing"]
+
+    display = zope.component.getUtility(interfaces.IDisplay)
+    response = display.menu(question, choices, "OK", "Cancel")
+    if response[0] == "cancel" or response[1] == 2:
+        # TODO: Add notification related to command-line options for
+        #       skipping the menu for this case.
+        raise errors.Error(
+            "User chose to cancel the operation and may "
+            "reinvoke the client.")
+    elif response[1] == 0:
+        return "reinstall", cert
+    elif response[1] == 1:
+        return "renew", cert
+    else:
+        assert False, "This is impossible"
+
+def _handle_subset_cert_request(config, domains, cert):
+    """Figure out what to do if a previous cert had a subset of the names now requested
+
+    :param storage.RenewableCert cert:
+
+    :returns: Tuple of (string, cert_or_None) as per _treat_as_renewal
+    :rtype: tuple
+
+    """
+    existing = ", ".join(cert.names())
+    question = (
+        "You have an existing certificate that contains a portion of "
+        "the domains you requested (ref: {0}){br}{br}It contains these "
+        "names: {1}{br}{br}You requested these names for the new "
+        "certificate: {2}.{br}{br}Do you want to expand and replace this existing "
+        "certificate with the new certificate?"
+    ).format(cert.configfile.filename,
+             existing,
+             ", ".join(domains),
+             br=os.linesep)
+    if config.expand or config.renew_by_default or zope.component.getUtility(
+            interfaces.IDisplay).yesno(question, "Expand", "Cancel"):
+        return "renew", cert
+    else:
+        reporter_util = zope.component.getUtility(interfaces.IReporter)
+        reporter_util.add_message(
+            "To obtain a new certificate that contains these names without "
+            "replacing your existing certificate for {0}, you must use the "
+            "--duplicate option.{br}{br}"
+            "For example:{br}{br}{1} --duplicate {2}".format(
+                existing,
+                sys.argv[0], " ".join(sys.argv[1:]),
+                br=os.linesep
+            ),
+            reporter_util.HIGH_PRIORITY)
+        raise errors.Error(
+            "User chose to cancel the operation and may "
+            "reinvoke the client.")
 
 
 def _report_new_cert(cert_path, fullchain_path):
@@ -314,10 +369,21 @@ def _suggest_donate():
 
 def _auth_from_domains(le_client, config, domains):
     """Authenticate and enroll certificate."""
-    # Note: This can raise errors... caught above us though.
-    lineage = _treat_as_renewal(config, domains)
+    # Note: This can raise errors... caught above us though. This is now
+    # a three-way case: reinstall (which results in a no-op here because
+    # although there is a relevant lineage, we don't do anything to it
+    # inside this function -- we don't obtain a new certificate), renew
+    # (which results in treating the request as a renewal), or newcert
+    # (which results in treating the request as a new certificate request).
 
-    if lineage is not None:
+    action, lineage = _treat_as_renewal(config, domains)
+    if action == "reinstall":
+        # The lineage already exists; allow the caller to try installing
+        # it without getting a new certificate at all.
+        return lineage
+    elif action == "renew":
+        original_server = lineage.configuration["renewalparams"]["server"]
+        _avoid_invalidating_lineage(config, lineage, original_server)
         # TODO: schoen wishes to reuse key - discussion
         # https://github.com/letsencrypt/letsencrypt/pull/777/files#r40498574
         new_certr, new_chain, new_key, _ = le_client.obtain_certificate(domains)
@@ -331,7 +397,7 @@ def _auth_from_domains(le_client, config, domains):
         # TODO: Check return value of save_successor
         # TODO: Also update lineage renewal config with any relevant
         #       configuration values from this attempt? <- Absolutely (jdkasten)
-    else:
+    elif action == "newcert":
         # TREAT AS NEW REQUEST
         lineage = le_client.obtain_and_enroll_certificate(domains)
         if not lineage:
@@ -341,6 +407,27 @@ def _auth_from_domains(le_client, config, domains):
 
     return lineage
 
+def _avoid_invalidating_lineage(config, lineage, original_server):
+    "Do not renew a valid cert with one from a staging server!"
+    def _is_staging(srv):
+        return srv == constants.STAGING_URI or "staging" in srv
+
+    # Some lineages may have begun with --staging, but then had production certs
+    # added to them
+    latest_cert = OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM,
+                                                  open(lineage.cert).read())
+    # all our test certs are from happy hacker fake CA, though maybe one day
+    # we should test more methodically
+    now_valid = not "fake" in repr(latest_cert.get_issuer()).lower()
+
+    if _is_staging(config.server):
+        if not _is_staging(original_server) or now_valid:
+            if not config.break_my_certs:
+                names = ", ".join(lineage.names())
+                raise errors.Error(
+                    "You've asked to renew/replace a seemingly valid certificate with "
+                    "a test certificate (domains: {0}). We will not do that "
+                    "unless you use the --break-my-certs flag!".format(names))
 
 def set_configurator(previously, now):
     """
@@ -485,7 +572,6 @@ def run(args, config, plugins):  # pylint: disable=too-many-branches,too-many-lo
 
 def obtain_cert(args, config, plugins):
     """Authenticate & obtain cert, but do not install it."""
-
     if args.domains and args.csr is not None:
         # TODO: --csr could have a priority, when --domains is
         # supplied, check if CSR matches given domains?
@@ -695,6 +781,7 @@ class HelpfulArgumentParser(object):
         """
         parsed_args = self.parser.parse_args(self.args)
         parsed_args.func = self.VERBS[self.verb]
+        parsed_args.verb = self.verb
 
         # Do any post-parsing homework here
 
@@ -877,13 +964,19 @@ def prepare_and_parse_args(plugins, args):
                 help="Domain names to apply. For multiple domains you can use "
                 "multiple -d flags or enter a comma separated list of domains "
                 "as a parameter.")
-    helpful.add(
-        None, "--duplicate", dest="duplicate", action="store_true",
-        help="Allow getting a certificate that duplicates an existing one")
-
     helpful.add_group(
         "automation",
         description="Arguments for automating execution & other tweaks")
+    helpful.add(
+        "automation", "--keep-until-expiring", "--keep", "--reinstall",
+        dest="reinstall", action="store_true",
+        help="If the requested cert matches an existing cert, always keep the "
+             "existing one until it is due for renewal (for the "
+             "'run' subcommand this means reinstall the existing cert)")
+    helpful.add(
+        "automation", "--expand", action="store_true",
+        help="If an existing cert covers some subset of the requested names, "
+             "always expand and replace it with the additional names.")
     helpful.add(
         "automation", "--version", action="version",
         version="%(prog)s {0}".format(letsencrypt.__version__),
@@ -891,13 +984,18 @@ def prepare_and_parse_args(plugins, args):
     helpful.add(
         "automation", "--renew-by-default", action="store_true",
         help="Select renewal by default when domains are a superset of a "
-             "previously attained cert")
+             "previously attained cert (often --keep-until-expiring is "
+             "more appropriate). Implies --expand.")
     helpful.add(
         "automation", "--agree-tos", dest="tos", action="store_true",
         help="Agree to the Let's Encrypt Subscriber Agreement")
     helpful.add(
         "automation", "--account", metavar="ACCOUNT_ID",
         help="Account ID to use")
+    helpful.add(
+        "automation", "--duplicate", dest="duplicate", action="store_true",
+        help="Allow making a certificate lineage that duplicates an existing one "
+             "(both can be renewed in parallel)")
 
     helpful.add_group(
         "testing", description="The following flags are meant for "
@@ -918,7 +1016,10 @@ def prepare_and_parse_args(plugins, args):
     helpful.add(
         "testing", "--http-01-port", type=int, dest="http01_port",
         default=flag_default("http01_port"), help=config_help("http01_port"))
-
+    helpful.add(
+        "testing", "--break-my-certs", action="store_true",
+        help="Be willing to replace or renew valid certs with invalid "
+             "(testing/staging) certs")
     helpful.add_group(
         "security", description="Security parameters & server settings")
     helpful.add(

letsencrypt/storage.py

@@ -471,7 +471,7 @@ class RenewableCert(object):  # pylint: disable=too-many-instance-attributes
         return ("autodeploy" not in self.configuration or
                 self.configuration.as_bool("autodeploy"))
 
-    def should_autodeploy(self):
+    def should_autodeploy(self, interactive=False):
         """Should this lineage now automatically deploy a newer version?
 
         This is a policy question and does not only depend on whether
@@ -480,12 +480,16 @@ class RenewableCert(object):  # pylint: disable=too-many-instance-attributes
         exists, and whether the time interval for autodeployment has
         been reached.)
 
+        :param bool interactive: set to True to examine the question
+            regardless of whether the renewal configuration allows
+            automated deployment (for interactive use). Default False.
+
         :returns: whether the lineage now ought to autodeploy an
             existing newer cert version
         :rtype: bool
 
         """
-        if self.autodeployment_is_enabled():
+        if interactive or self.autodeployment_is_enabled():
             if self.has_pending_deployment():
                 interval = self.configuration.get("deploy_before_expiry",
                                                   "5 days")
@@ -529,7 +533,7 @@ class RenewableCert(object):  # pylint: disable=too-many-instance-attributes
         return ("autorenew" not in self.configuration or
                 self.configuration.as_bool("autorenew"))
 
-    def should_autorenew(self):
+    def should_autorenew(self, interactive=False):
         """Should we now try to autorenew the most recent cert version?
 
         This is a policy question and does not only depend on whether
@@ -540,12 +544,16 @@ class RenewableCert(object):  # pylint: disable=too-many-instance-attributes
         Note that this examines the numerically most recent cert version,
         not the currently deployed version.
 
+        :param bool interactive: set to True to examine the question
+            regardless of whether the renewal configuration allows
+            automated renewal (for interactive use). Default False.
+
         :returns: whether an attempt should now be made to autorenew the
             most current cert version in this lineage
         :rtype: bool
 
         """
-        if self.autorenewal_is_enabled():
+        if interactive or self.autorenewal_is_enabled():
             # Consider whether to attempt to autorenew this cert now
 
             # Renewals on the basis of revocation
@@ -559,8 +567,8 @@ class RenewableCert(object):  # pylint: disable=too-many-instance-attributes
                 "cert", self.latest_common_version()))
             now = pytz.UTC.fromutc(datetime.datetime.utcnow())
             if expiry < add_time_interval(now, interval):
-                logger.debug("Should renew, certificate "
-                             "has been expired since %s.",
+                logger.debug("Should renew, less than %s before certificate "
+                             "expiry %s.", interval,
                              expiry.strftime("%Y-%m-%d %H:%M:%S %Z"))
                 return True
         return False

letsencrypt/tests/cli_test.py

@@ -403,7 +403,7 @@ class CLITest(unittest.TestCase):  # pylint: disable=too-many-public-methods
 
     def _certonly_new_request_common(self, mock_client):
         with mock.patch('letsencrypt.cli._treat_as_renewal') as mock_renewal:
-            mock_renewal.return_value = None
+            mock_renewal.return_value = ("newcert", None)
             with mock.patch('letsencrypt.cli._init_le_client') as mock_init:
                 mock_init.return_value = mock_client
                 self._call(['-d', 'foo.bar', '-a', 'standalone', 'certonly'])
@@ -413,13 +413,13 @@ class CLITest(unittest.TestCase):  # pylint: disable=too-many-public-methods
     @mock.patch('letsencrypt.cli._treat_as_renewal')
     @mock.patch('letsencrypt.cli._init_le_client')
     def test_certonly_renewal(self, mock_init, mock_renewal, mock_get_utility, _suggest):
-        cert_path = '/etc/letsencrypt/live/foo.bar/cert.pem'
+        cert_path = 'letsencrypt/tests/testdata/cert.pem'
         chain_path = '/etc/letsencrypt/live/foo.bar/fullchain.pem'
 
         mock_lineage = mock.MagicMock(cert=cert_path, fullchain=chain_path)
         mock_cert = mock.MagicMock(body='body')
         mock_key = mock.MagicMock(pem='pem_key')
-        mock_renewal.return_value = mock_lineage
+        mock_renewal.return_value = ("renew", mock_lineage)
         mock_client = mock.MagicMock()
         mock_client.obtain_certificate.return_value = (mock_cert, 'chain',
                                                        mock_key, 'csr')