Infra/lemur
5
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

Remove duplicate code for revoke_certificate

Browse Source
This commit is contained in:
Mathias Petermann 2020-11-10 18:18:45 +01:00
parent fba1fdcc34
commit 31b5f3df86
3 changed files with 26 additions and 44 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
lemur/plugins/lemur_acme/acme_handlers.py
View File

@ -27,7 +27,7 @@ from flask import current_app
from lemur.common.utils import generate_private_key
from lemur.dns_providers import service as dns_provider_service
from lemur.exceptions import InvalidAuthority, UnknownProvider
from lemur.exceptions import InvalidAuthority, UnknownProvider, InvalidConfiguration
from lemur.extensions import metrics, sentry
from lemur.plugins.lemur_acme import cloudflare, dyn, route53, ultradns, powerdns
@ -216,6 +216,27 @@ class AcmeHandler(object):
current_app.logger.debug("Got these domains: {0}".format(domains))
return domains
def revoke_certificate(self, certificate):
if not self.reuse_account(certificate.authority):
raise InvalidConfiguration("There is no ACME account saved, unable to revoke the certificate.")
acme_client, _ = self.acme.setup_acme_client(certificate.authority)
fullchain_com = jose.ComparableX509(
OpenSSL.crypto.load_certificate(
OpenSSL.crypto.FILETYPE_PEM, certificate.body))
try:
acme_client.revoke(fullchain_com, 0) # revocation reason = 0
except (errors.ConflictError, errors.ClientError, errors.Error) as e:
# Certificate already revoked.
current_app.logger.error("Certificate revocation failed with message: " + e.detail)
metrics.send("acme_revoke_certificate_failure", "counter", 1)
return False
current_app.logger.warning("Certificate succesfully revoked: " + certificate.name)
metrics.send("acme_revoke_certificate_success", "counter", 1)
return True
class AcmeDnsHandler(AcmeHandler):

43
lemur/plugins/lemur_acme/plugin.py
View File

@ -12,9 +12,6 @@
.. moduleauthor:: Curtis Castrapel <ccastrapel@netflix.com>
"""
import OpenSSL.crypto
import josepy as jose
from acme import errors
from acme.errors import PollError, WildcardUnsupportedError
from acme.messages import Error as AcmeError
from botocore.exceptions import ClientError
@ -272,25 +269,7 @@ class ACMEIssuerPlugin(IssuerPlugin):
def revoke_certificate(self, certificate, comments):
self.acme = AcmeDnsHandler()
if not self.acme.reuse_account(certificate.authority):
raise InvalidConfiguration("There is no ACME account saved, unable to revoke the certificate.")
acme_client, _ = self.acme.setup_acme_client(certificate.authority)
fullchain_com = jose.ComparableX509(
OpenSSL.crypto.load_certificate(
OpenSSL.crypto.FILETYPE_PEM, certificate.body))
try:
acme_client.revoke(fullchain_com, 0) # revocation reason = 0
except (errors.ConflictError, errors.ClientError, errors.Error) as e:
# Certificate already revoked.
current_app.logger.error("Certificate revocation failed with message: " + e.detail)
metrics.send("acme_revoke_certificate_failure", "counter", 1)
return False
current_app.logger.warning("Certificate succesfully revoked: " + certificate.name)
metrics.send("acme_revoke_certificate_success", "counter", 1)
return True
return self.acme.revoke_certificate(certificate)
class ACMEHttpIssuerPlugin(IssuerPlugin):
@ -391,22 +370,4 @@ class ACMEHttpIssuerPlugin(IssuerPlugin):
def revoke_certificate(self, certificate, comments):
self.acme = AcmeHandler()
if not self.acme.reuse_account(certificate.authority):
raise InvalidConfiguration("There is no ACME account saved, unable to revoke the certificate.")
acme_client, _ = self.acme.setup_acme_client(certificate.authority)
fullchain_com = jose.ComparableX509(
OpenSSL.crypto.load_certificate(
OpenSSL.crypto.FILETYPE_PEM, certificate.body))
try:
acme_client.revoke(fullchain_com, 0) # revocation reason = 0
except (errors.ConflictError, errors.ClientError, errors.Error) as e:
# Certificate already revoked.
current_app.logger.error("Certificate revocation failed with message: " + e.detail)
metrics.send("acme_revoke_certificate_failure", "counter", 1)
return False
current_app.logger.warning("Certificate succesfully revoked: " + certificate.name)
metrics.send("acme_revoke_certificate_success", "counter", 1)
return True
return self.acme.revoke_certificate(certificate)

4
lemur/plugins/lemur_acme/tests/test_acme_dns.py
View File

@ -168,7 +168,7 @@ class TestAcmeDns(unittest.TestCase):
with self.assertRaises(Exception):
self.acme.setup_acme_client(mock_authority)
@patch("lemur.plugins.lemur_acme.plugin.jose.JWK.json_loads")
@patch("lemur.plugins.lemur_acme.acme_handlers.jose.JWK.json_loads")
@patch("lemur.plugins.lemur_acme.acme_handlers.BackwardsCompatibleClientV2")
@patch("lemur.plugins.lemur_acme.acme_handlers.current_app")
def test_setup_acme_client_success_load_account_from_authority(self, mock_current_app, mock_acme, mock_key_json_load):
@ -190,7 +190,7 @@ class TestAcmeDns(unittest.TestCase):
assert result_client
assert not result_registration
@patch("lemur.plugins.lemur_acme.plugin.jose.JWKRSA.fields_to_partial_json")
@patch("lemur.plugins.lemur_acme.acme_handlers.jose.JWKRSA.fields_to_partial_json")
@patch("lemur.plugins.lemur_acme.acme_handlers.authorities_service")
@patch("lemur.plugins.lemur_acme.acme_handlers.BackwardsCompatibleClientV2")
@patch("lemur.plugins.lemur_acme.acme_handlers.current_app")