mirror of
https://github.com/certbot/certbot.git
synced 2026-07-31 18:25:24 +02:00
Remove PoP
This commit is contained in:
Brad Warren
@@ -1,99 +0,0 @@
|
|||||||
"""Proof of Possession Identifier Validation Challenge."""
|
|
||||||
import logging
|
|
||||||
import os
|
|
||||||
|
|
||||||
from cryptography import x509
|
|
||||||
from cryptography.hazmat.backends import default_backend
|
|
||||||
import zope.component
|
|
||||||
|
|
||||||
from acme import challenges
|
|
||||||
from acme import jose
|
|
||||||
from acme import other
|
|
||||||
|
|
||||||
from letsencrypt import interfaces
|
|
||||||
from letsencrypt.display import util as display_util
|
|
||||||
|
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
|
||||||
|
|
||||||
|
|
||||||
class ProofOfPossession(object): # pylint: disable=too-few-public-methods
|
|
||||||
"""Proof of Possession Identifier Validation Challenge.
|
|
||||||
|
|
||||||
Based on draft-barnes-acme, section 6.5.
|
|
||||||
|
|
||||||
:ivar installer: Installer object
|
|
||||||
:type installer: :class:`~letsencrypt.interfaces.IInstaller`
|
|
||||||
|
|
||||||
"""
|
|
||||||
def __init__(self, installer):
|
|
||||||
self.installer = installer
|
|
||||||
|
|
||||||
def perform(self, achall):
|
|
||||||
"""Perform the Proof of Possession Challenge.
|
|
||||||
|
|
||||||
:param achall: Proof of Possession Challenge
|
|
||||||
:type achall: :class:`letsencrypt.achallenges.ProofOfPossession`
|
|
||||||
|
|
||||||
:returns: Response or None/False if the challenge cannot be completed
|
|
||||||
:rtype: :class:`acme.challenges.ProofOfPossessionResponse`
|
|
||||||
or False
|
|
||||||
|
|
||||||
"""
|
|
||||||
if (achall.alg in [jose.HS256, jose.HS384, jose.HS512] or
|
|
||||||
not isinstance(achall.hints.jwk, achall.alg.kty)):
|
|
||||||
return None
|
|
||||||
|
|
||||||
for cert, key, _ in self.installer.get_all_certs_keys():
|
|
||||||
with open(cert) as cert_file:
|
|
||||||
cert_data = cert_file.read()
|
|
||||||
try:
|
|
||||||
cert_obj = x509.load_pem_x509_certificate(
|
|
||||||
cert_data, default_backend())
|
|
||||||
except ValueError:
|
|
||||||
try:
|
|
||||||
cert_obj = x509.load_der_x509_certificate(
|
|
||||||
cert_data, default_backend())
|
|
||||||
except ValueError:
|
|
||||||
logger.warn("Certificate is neither PER nor DER: %s", cert)
|
|
||||||
|
|
||||||
cert_key = achall.alg.kty(key=cert_obj.public_key())
|
|
||||||
if cert_key == achall.hints.jwk:
|
|
||||||
return self._gen_response(achall, key)
|
|
||||||
|
|
||||||
# Is there are different prompt we should give the user?
|
|
||||||
code, key = zope.component.getUtility(
|
|
||||||
interfaces.IDisplay).input(
|
|
||||||
"Path to private key for identifier: %s " % achall.domain)
|
|
||||||
if code != display_util.CANCEL:
|
|
||||||
return self._gen_response(achall, key)
|
|
||||||
|
|
||||||
# If we get here, the key wasn't found
|
|
||||||
return False
|
|
||||||
|
|
||||||
def _gen_response(self, achall, key_path): # pylint: disable=no-self-use
|
|
||||||
"""Create the response to the Proof of Possession Challenge.
|
|
||||||
|
|
||||||
:param achall: Proof of Possession Challenge
|
|
||||||
:type achall: :class:`letsencrypt.achallenges.ProofOfPossession`
|
|
||||||
|
|
||||||
:param str key_path: Path to the key corresponding to the hinted to
|
|
||||||
public key.
|
|
||||||
|
|
||||||
:returns: Response or False if the challenge cannot be completed
|
|
||||||
:rtype: :class:`acme.challenges.ProofOfPossessionResponse`
|
|
||||||
or False
|
|
||||||
|
|
||||||
"""
|
|
||||||
if os.path.isfile(key_path):
|
|
||||||
with open(key_path, 'rb') as key:
|
|
||||||
try:
|
|
||||||
# Needs to be changed if JWKES doesn't have a key attribute
|
|
||||||
jwk = achall.alg.kty.load(key.read())
|
|
||||||
sig = other.Signature.from_msg(achall.nonce, jwk.key,
|
|
||||||
alg=achall.alg)
|
|
||||||
except (IndexError, ValueError, TypeError, jose.errors.Error):
|
|
||||||
return False
|
|
||||||
return challenges.ProofOfPossessionResponse(nonce=achall.nonce,
|
|
||||||
signature=sig)
|
|
||||||
return False
|
|
||||||
@@ -1,83 +0,0 @@
|
|||||||
"""Tests for letsencrypt.proof_of_possession."""
|
|
||||||
import os
|
|
||||||
import tempfile
|
|
||||||
import unittest
|
|
||||||
|
|
||||||
import mock
|
|
||||||
|
|
||||||
from acme import challenges
|
|
||||||
from acme import jose
|
|
||||||
from acme import messages
|
|
||||||
|
|
||||||
from letsencrypt import achallenges
|
|
||||||
from letsencrypt import proof_of_possession
|
|
||||||
from letsencrypt.display import util as display_util
|
|
||||||
|
|
||||||
from letsencrypt.tests import test_util
|
|
||||||
|
|
||||||
|
|
||||||
CERT0_PATH = test_util.vector_path("cert.der")
|
|
||||||
CERT2_PATH = test_util.vector_path("dsa_cert.pem")
|
|
||||||
CERT2_KEY_PATH = test_util.vector_path("dsa512_key.pem")
|
|
||||||
CERT3_PATH = test_util.vector_path("matching_cert.pem")
|
|
||||||
CERT3_KEY_PATH = test_util.vector_path("rsa512_key_2.pem")
|
|
||||||
CERT3_KEY = test_util.load_rsa_private_key("rsa512_key_2.pem").public_key()
|
|
||||||
|
|
||||||
|
|
||||||
class ProofOfPossessionTest(unittest.TestCase):
|
|
||||||
def setUp(self):
|
|
||||||
self.installer = mock.MagicMock()
|
|
||||||
self.cert1_path = tempfile.mkstemp()[1]
|
|
||||||
certs = [CERT0_PATH, self.cert1_path, CERT2_PATH, CERT3_PATH]
|
|
||||||
keys = [None, None, CERT2_KEY_PATH, CERT3_KEY_PATH]
|
|
||||||
self.installer.get_all_certs_keys.return_value = zip(
|
|
||||||
certs, keys, 4 * [None])
|
|
||||||
self.proof_of_pos = proof_of_possession.ProofOfPossession(
|
|
||||||
self.installer)
|
|
||||||
|
|
||||||
hints = challenges.ProofOfPossession.Hints(
|
|
||||||
jwk=jose.JWKRSA(key=CERT3_KEY), cert_fingerprints=(),
|
|
||||||
certs=(), serial_numbers=(), subject_key_identifiers=(),
|
|
||||||
issuers=(), authorized_for=())
|
|
||||||
chall = challenges.ProofOfPossession(
|
|
||||||
alg=jose.RS256, nonce='zczv4HMLVe_0kimJ25Juig', hints=hints)
|
|
||||||
challb = messages.ChallengeBody(
|
|
||||||
chall=chall, uri="http://example", status=messages.STATUS_PENDING)
|
|
||||||
self.achall = achallenges.ProofOfPossession(
|
|
||||||
challb=challb, domain="example.com")
|
|
||||||
|
|
||||||
def tearDown(self):
|
|
||||||
os.remove(self.cert1_path)
|
|
||||||
|
|
||||||
def test_perform_bad_challenge(self):
|
|
||||||
hints = challenges.ProofOfPossession.Hints(
|
|
||||||
jwk=jose.jwk.JWKOct(key="foo"), cert_fingerprints=(),
|
|
||||||
certs=(), serial_numbers=(), subject_key_identifiers=(),
|
|
||||||
issuers=(), authorized_for=())
|
|
||||||
chall = challenges.ProofOfPossession(
|
|
||||||
alg=jose.HS512, nonce='zczv4HMLVe_0kimJ25Juig', hints=hints)
|
|
||||||
challb = messages.ChallengeBody(
|
|
||||||
chall=chall, uri="http://example", status=messages.STATUS_PENDING)
|
|
||||||
self.achall = achallenges.ProofOfPossession(
|
|
||||||
challb=challb, domain="example.com")
|
|
||||||
self.assertEqual(self.proof_of_pos.perform(self.achall), None)
|
|
||||||
|
|
||||||
def test_perform_no_input(self):
|
|
||||||
self.assertTrue(self.proof_of_pos.perform(self.achall).verify())
|
|
||||||
|
|
||||||
@mock.patch("letsencrypt.proof_of_possession.zope.component.getUtility")
|
|
||||||
def test_perform_with_input(self, mock_input):
|
|
||||||
# Remove the matching certificate
|
|
||||||
self.installer.get_all_certs_keys.return_value.pop()
|
|
||||||
mock_input().input.side_effect = [(display_util.CANCEL, ""),
|
|
||||||
(display_util.OK, CERT0_PATH),
|
|
||||||
(display_util.OK, "imaginary_file"),
|
|
||||||
(display_util.OK, CERT3_KEY_PATH)]
|
|
||||||
self.assertFalse(self.proof_of_pos.perform(self.achall))
|
|
||||||
self.assertFalse(self.proof_of_pos.perform(self.achall))
|
|
||||||
self.assertFalse(self.proof_of_pos.perform(self.achall))
|
|
||||||
self.assertTrue(self.proof_of_pos.perform(self.achall).verify())
|
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
|
||||||
unittest.main() # pragma: no cover
|
|
||||||
Reference in New Issue
Block a user