mirror of
https://github.com/certbot/certbot.git
synced 2026-06-03 13:59:02 -04:00
Add dvsni tests
This commit is contained in:
yan
parent
23d27f4659
commit
6f4af62f61
5 changed files with 224 additions and 49 deletions
|
|
@ -1,7 +1,11 @@
|
|||
"""NginxDVSNI"""
|
||||
import logging
|
||||
import os
|
||||
|
||||
from letsencrypt.client import errors
|
||||
from letsencrypt.client.plugins.apache.dvsni import ApacheDvsni
|
||||
from letsencrypt.client.plugins.nginx import obj
|
||||
from letsencrypt.client.plugins.nginx.nginxparser import dump
|
||||
|
||||
|
||||
class NginxDvsni(ApacheDvsni):
|
||||
|
|
@ -29,35 +33,110 @@ class NginxDvsni(ApacheDvsni):
|
|||
"""
|
||||
|
||||
def perform(self):
|
||||
"""Perform a DVSNI challenge on Nginx."""
|
||||
"""Perform a DVSNI challenge on Nginx.
|
||||
|
||||
:returns: list of :class:`letsencrypt.acme.challenges.DVSNIResponse`
|
||||
:rtype: list
|
||||
|
||||
"""
|
||||
if not self.achalls:
|
||||
return []
|
||||
|
||||
self.configurator.save()
|
||||
|
||||
addresses = []
|
||||
default_addr = "443 default_server ssl"
|
||||
|
||||
for achall in self.achalls:
|
||||
vhost = self.configurator.choose_vhost(achall.domain)
|
||||
if vhost is None:
|
||||
logging.error(
|
||||
"No nginx vhost exists with servername or alias of: %s",
|
||||
"No nginx vhost exists with server_name or alias of: %s",
|
||||
achall.domain)
|
||||
logging.error("No default 443 nginx vhost exists")
|
||||
logging.error("Please specify servernames in the Nginx config")
|
||||
logging.error("Please specify server_names in the Nginx config")
|
||||
return None
|
||||
|
||||
for addr in vhost.addrs:
|
||||
if addr.default:
|
||||
addresses.append([obj.Addr.fromstring(default_addr)])
|
||||
break
|
||||
else:
|
||||
addresses.append(list(vhost.addrs))
|
||||
|
||||
responses = []
|
||||
# Create challenge certs
|
||||
responses = [self._setup_challenge_cert(x) for x in self.achalls]
|
||||
|
||||
# Create all of the challenge certs
|
||||
# for achall in self.achalls:
|
||||
# responses.append(self._setup_challenge_cert(achall))
|
||||
|
||||
# Setup the configuration
|
||||
# self._mod_config(addresses)
|
||||
# Set up the configuration
|
||||
self._mod_config(addresses)
|
||||
|
||||
# Save reversible changes
|
||||
self.configurator.save("SNI Challenge", True)
|
||||
|
||||
return responses
|
||||
|
||||
def _mod_config(self, ll_addrs):
|
||||
"""Modifies Nginx config to include challenge server blocks.
|
||||
|
||||
:param list ll_addrs: list of lists of
|
||||
:class:`letsencrypt.client.plugins.apache.obj.Addr` to apply
|
||||
|
||||
:raises errors.LetsEncryptMisconfigurationError:
|
||||
Unable to find a suitable HTTP block to include DVSNI hosts.
|
||||
|
||||
"""
|
||||
# Add the 'include' statement for the challenges if it doesn't exist
|
||||
# already in the main config
|
||||
included = False
|
||||
directive = ['include', self.challenge_conf]
|
||||
root = self.configurator.parser.loc["root"]
|
||||
main = self.configurator.parser.parsed[root]
|
||||
for entry in main:
|
||||
if entry[0] == ['http']:
|
||||
body = entry[1]
|
||||
if directive not in body:
|
||||
body.append(directive)
|
||||
included = True
|
||||
break
|
||||
if not included:
|
||||
raise errors.LetsEncryptMisconfigurationError(
|
||||
'LetsEncrypt could not find an HTTP block to include DVSNI '
|
||||
'challenges in %s.' % root)
|
||||
|
||||
config = []
|
||||
for idx, addrs in enumerate(ll_addrs):
|
||||
config.append(self._make_server_block(self.achalls[idx], addrs))
|
||||
|
||||
self.configurator.reverter.register_file_creation(
|
||||
True, self.challenge_conf)
|
||||
|
||||
with open(self.challenge_conf, "w") as new_conf:
|
||||
dump(config, new_conf)
|
||||
|
||||
def _make_server_block(self, achall, addrs):
|
||||
"""Creates a server block for a DVSNI challenge.
|
||||
|
||||
:param achall: Annotated DVSNI challenge.
|
||||
:type achall: :class:`letsencrypt.client.achallenges.DVSNI`
|
||||
|
||||
:param list addrs: addresses of challenged domain
|
||||
:class:`list` of type :class:`~nginx.obj.Addr`
|
||||
|
||||
:returns: server block for the challenge host
|
||||
:rtype: list
|
||||
|
||||
"""
|
||||
block = []
|
||||
for addr in addrs:
|
||||
block.append(['listen', str(addr)])
|
||||
|
||||
block.append(['server_name', achall.nonce_domain])
|
||||
block.append(['include', self.configurator.parser.loc["ssl_options"]])
|
||||
block.append(['ssl_certificate', self.get_cert_file(achall)])
|
||||
block.append(['ssl_certificate_key', achall.key.file])
|
||||
|
||||
document_root = os.path.join(
|
||||
self.configurator.config.config_dir, "dvsni_page")
|
||||
block.append([['location', '/'], [['root', document_root]]])
|
||||
|
||||
return [['server'], block]
|
||||
|
|
|
|||
|
|
@ -67,12 +67,20 @@ class Addr(ApacheAddr):
|
|||
return cls(host, port, ssl, default)
|
||||
|
||||
def __str__(self):
|
||||
parts = ''
|
||||
if self.tup[0] and self.tup[1]:
|
||||
return "%s:%s" % self.tup
|
||||
parts = "%s:%s" % self.tup
|
||||
elif self.tup[0]:
|
||||
return self.tup[0]
|
||||
parts = self.tup[0]
|
||||
else:
|
||||
return self.tup[1]
|
||||
parts = self.tup[1]
|
||||
|
||||
if self.default:
|
||||
parts += ' default_server'
|
||||
if self.ssl:
|
||||
parts += ' ssl'
|
||||
|
||||
return parts
|
||||
|
||||
def __eq__(self, other):
|
||||
if isinstance(other, self.__class__):
|
||||
|
|
|
|||
|
|
@ -33,6 +33,7 @@ class NginxParser(object):
|
|||
"""Loads Nginx files into a parsed tree.
|
||||
|
||||
"""
|
||||
self.parsed = {}
|
||||
self._parse_recursively(self.loc["root"])
|
||||
|
||||
def _parse_recursively(self, filepath):
|
||||
|
|
@ -252,8 +253,9 @@ class NginxParser(object):
|
|||
|
||||
def add_server_directives(self, filename, names, directives,
|
||||
replace=False):
|
||||
"""Add or replace directives in server blocks whose server_name set
|
||||
is 'names'. If replace is True, this raises a misconfiguration error
|
||||
"""Add or replace directives in server blocks identified by server_name.
|
||||
|
||||
..note :: If replace is True, this raises a misconfiguration error
|
||||
if the directive does not already exist.
|
||||
|
||||
..todo :: Doesn't match server blocks whose server_name directives are
|
||||
|
|
|
|||
|
|
@ -6,13 +6,16 @@ import shutil
|
|||
import mock
|
||||
|
||||
from letsencrypt.acme import challenges
|
||||
from letsencrypt.acme import messages2
|
||||
|
||||
from letsencrypt.client import achallenges
|
||||
from letsencrypt.client import errors
|
||||
from letsencrypt.client import le_util
|
||||
|
||||
from letsencrypt.client.plugins.nginx.obj import Addr
|
||||
from letsencrypt.client.plugins.nginx.tests import util
|
||||
|
||||
from letsencrypt.client.tests import acme_util
|
||||
|
||||
|
||||
class DvsniPerformTest(util.NginxTest):
|
||||
"""Test the NginxDVSNI challenge."""
|
||||
|
|
@ -36,25 +39,29 @@ class DvsniPerformTest(util.NginxTest):
|
|||
|
||||
self.achalls = [
|
||||
achallenges.DVSNI(
|
||||
challb=messages2.ChallengeBody(
|
||||
chall=challenges.DVSNI(
|
||||
challb=acme_util.chall_to_challb(
|
||||
challenges.DVSNI(
|
||||
r="foo",
|
||||
nonce="bar",
|
||||
),
|
||||
uri="https://letsencrypt-ca.org/chall0_uri",
|
||||
status=messages2.Status("pending"),
|
||||
), domain="www.example.com", key=auth_key),
|
||||
nonce="bar"
|
||||
), "pending"),
|
||||
domain="www.example.com", key=auth_key),
|
||||
achallenges.DVSNI(
|
||||
challb=messages2.ChallengeBody(
|
||||
chall=challenges.DVSNI(
|
||||
challb=acme_util.chall_to_challb(
|
||||
challenges.DVSNI(
|
||||
r="\xba\xa9\xda?<m\xaewmx\xea\xad\xadv\xf4\x02\xc9y\x80"
|
||||
"\xe2_X\t\xe7\xc7\xa4\t\xca\xf7&\x945",
|
||||
nonce="Y\xed\x01L\xac\x95\xf7pW\xb1\xd7"
|
||||
"\xa1\xb2\xc5\x96\xba",
|
||||
),
|
||||
uri="https://letsencrypt-ca.org/chall1_uri",
|
||||
status=messages2.Status("pending"),
|
||||
), domain="blah", key=auth_key),
|
||||
"\xa1\xb2\xc5\x96\xba"
|
||||
), "pending"),
|
||||
domain="blah", key=auth_key),
|
||||
achallenges.DVSNI(
|
||||
challb=acme_util.chall_to_challb(
|
||||
challenges.DVSNI(
|
||||
r="\x8c\x8a\xbf_-f\\cw\xee\xd6\xf8/\xa5\xe3\xfd\xeb9"
|
||||
"\xf1\xf5\xb9\xefVM\xc9w\xa4u\x9c\xe1\x87\xb4",
|
||||
nonce="7\xbc^\xb7]>\x00\xa1\x9bOcU\x84^Z\x18"
|
||||
), "pending"),
|
||||
domain="www.example.org", key=auth_key)
|
||||
]
|
||||
|
||||
def tearDown(self):
|
||||
|
|
@ -69,26 +76,105 @@ class DvsniPerformTest(util.NginxTest):
|
|||
|
||||
@mock.patch("letsencrypt.client.plugins.nginx.configurator."
|
||||
"NginxConfigurator.save")
|
||||
def test_perform0(self, mock_save):
|
||||
self.sni.add_chall(self.achalls[0])
|
||||
responses = self.sni.perform()
|
||||
self.assertEqual([], responses)
|
||||
self.assertEqual(mock_save.call_count, 2)
|
||||
|
||||
def test_setup_challenge_cert(self):
|
||||
# This is a helper function that can be used for handling
|
||||
# open context managers more elegantly. It avoids dealing with
|
||||
# __enter__ and __exit__ calls.
|
||||
# http://www.voidspace.org.uk/python/mock/helpers.html#mock.mock_open
|
||||
pass
|
||||
|
||||
@mock.patch("letsencrypt.client.plugins.nginx.configurator."
|
||||
"NginxConfigurator.save")
|
||||
def test_perform1(self, mock_save):
|
||||
def test_perform(self, mock_save):
|
||||
self.sni.add_chall(self.achalls[1])
|
||||
responses = self.sni.perform()
|
||||
self.assertEqual(None, responses)
|
||||
self.assertEqual(mock_save.call_count, 1)
|
||||
|
||||
def test_perform0(self):
|
||||
responses = self.sni.perform()
|
||||
self.assertEqual([], responses)
|
||||
|
||||
@mock.patch("letsencrypt.client.plugins.nginx.configurator."
|
||||
"NginxConfigurator.save")
|
||||
def test_perform1(self, mock_save):
|
||||
self.sni.add_chall(self.achalls[0])
|
||||
mock_setup_cert = mock.MagicMock(
|
||||
return_value=challenges.DVSNIResponse(s="nginxS1"))
|
||||
|
||||
# pylint: disable=protected-access
|
||||
self.sni._setup_challenge_cert = mock_setup_cert
|
||||
|
||||
responses = self.sni.perform()
|
||||
|
||||
mock_setup_cert.assert_called_once_with(self.achalls[0])
|
||||
self.assertEqual([challenges.DVSNIResponse(s="nginxS1")], responses)
|
||||
self.assertEqual(mock_save.call_count, 2)
|
||||
|
||||
# Make sure challenge config is included in main config
|
||||
http = self.sni.configurator.parser.parsed[
|
||||
self.sni.configurator.parser.loc["root"]][-1]
|
||||
self.assertTrue(['include', self.sni.challenge_conf] in http[1])
|
||||
|
||||
def test_perform2(self):
|
||||
self.sni.add_chall(self.achalls[0])
|
||||
self.sni.add_chall(self.achalls[2])
|
||||
|
||||
mock_setup_cert = mock.MagicMock(side_effect=[
|
||||
challenges.DVSNIResponse(s="nginxS0"),
|
||||
challenges.DVSNIResponse(s="nginxS1")])
|
||||
# pylint: disable=protected-access
|
||||
self.sni._setup_challenge_cert = mock_setup_cert
|
||||
|
||||
responses = self.sni.perform()
|
||||
|
||||
self.assertEqual(mock_setup_cert.call_count, 2)
|
||||
|
||||
self.assertEqual(
|
||||
mock_setup_cert.call_args_list[0], mock.call(self.achalls[0]))
|
||||
self.assertEqual(
|
||||
mock_setup_cert.call_args_list[1], mock.call(self.achalls[2]))
|
||||
|
||||
http = self.sni.configurator.parser.parsed[
|
||||
self.sni.configurator.parser.loc["root"]][-1]
|
||||
self.assertTrue(['include', self.sni.challenge_conf] in http[1])
|
||||
|
||||
self.assertEqual(len(responses), 2)
|
||||
for i in xrange(2):
|
||||
self.assertEqual(responses[i].s, "nginxS%d" % i)
|
||||
|
||||
def test_mod_config(self):
|
||||
self.sni.add_chall(self.achalls[0])
|
||||
self.sni.add_chall(self.achalls[2])
|
||||
|
||||
v_addr1 = [Addr("69.50.225.155", "9000", True, False),
|
||||
Addr("127.0.0.1", "", False, False)]
|
||||
v_addr2 = [Addr("myhost", "", False, True)]
|
||||
ll_addr = [v_addr1, v_addr2]
|
||||
self.sni._mod_config(ll_addr) # pylint: disable=protected-access
|
||||
|
||||
self.sni.configurator.save()
|
||||
|
||||
self.sni.configurator.parser.load()
|
||||
|
||||
http = self.sni.configurator.parser.parsed[
|
||||
self.sni.configurator.parser.loc["root"]][-1]
|
||||
self.assertTrue(['include', self.sni.challenge_conf] in http[1])
|
||||
|
||||
vhosts = self.sni.configurator.parser.get_vhosts()
|
||||
vhs = []
|
||||
for vhost in vhosts:
|
||||
if vhost.filep == self.sni.challenge_conf:
|
||||
vhs.append(vhost)
|
||||
|
||||
for vhost in vhs:
|
||||
if vhost.addrs == set(v_addr1):
|
||||
self.assertEqual(
|
||||
vhost.names, set([self.achalls[0].nonce_domain]))
|
||||
else:
|
||||
self.assertEqual(vhost.addrs, set(v_addr2))
|
||||
self.assertEqual(
|
||||
vhost.names, set([self.achalls[2].nonce_domain]))
|
||||
|
||||
self.assertEqual(len(vhs), 2)
|
||||
|
||||
def test_mod_config_fail(self):
|
||||
root = self.sni.configurator.parser.loc["root"]
|
||||
self.sni.configurator.parser.parsed[root] = [['include', 'foo.conf']]
|
||||
# pylint: disable=protected-access
|
||||
self.assertRaises(errors.LetsEncryptMisconfigurationError,
|
||||
self.sni._mod_config, [])
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
|
|
|||
|
|
@ -49,11 +49,11 @@ class AddrTest(unittest.TestCase):
|
|||
|
||||
def test_str(self):
|
||||
self.assertEqual(str(self.addr1), "192.168.1.1")
|
||||
self.assertEqual(str(self.addr2), "192.168.1.1:*")
|
||||
self.assertEqual(str(self.addr2), "192.168.1.1:* ssl")
|
||||
self.assertEqual(str(self.addr3), "192.168.1.1:80")
|
||||
self.assertEqual(str(self.addr4), "*:80")
|
||||
self.assertEqual(str(self.addr4), "*:80 default_server ssl")
|
||||
self.assertEqual(str(self.addr5), "myhost")
|
||||
self.assertEqual(str(self.addr6), "80")
|
||||
self.assertEqual(str(self.addr6), "80 default_server")
|
||||
|
||||
def test_eq(self):
|
||||
from letsencrypt.client.plugins.nginx.obj import Addr
|
||||
|
|
|
|||
Loading…
Reference in a new issue