Compare commits

...

1 Commits

Author SHA1 Message Date
Brad Warren
d435eaf25e Revert "Nginx reversion (#5299)"
This reverts commit c9949411cd.
2017-12-06 16:34:17 -08:00
12 changed files with 437 additions and 69 deletions

View File

@@ -117,6 +117,9 @@ class NginxConfigurator(common.Installer):
# Files to save
self.save_notes = ""
# For creating new vhosts if no names match
self.new_vhost = None
# Add number of outstanding challenges
self._chall_out = 0
@@ -191,9 +194,11 @@ class NginxConfigurator(common.Installer):
"The nginx plugin currently requires --fullchain-path to "
"install a cert.")
vhost = self.choose_vhost(domain)
cert_directives = [['\n', 'ssl_certificate', ' ', fullchain_path],
['\n', 'ssl_certificate_key', ' ', key_path]]
vhost = self.choose_vhost(domain, raise_if_no_match=False)
if vhost is None:
vhost = self._vhost_from_duplicated_default(domain)
cert_directives = [['\n ', 'ssl_certificate', ' ', fullchain_path],
['\n ', 'ssl_certificate_key', ' ', key_path]]
self.parser.add_server_directives(vhost,
cert_directives, replace=True)
@@ -209,7 +214,7 @@ class NginxConfigurator(common.Installer):
#######################
# Vhost parsing methods
#######################
def choose_vhost(self, target_name):
def choose_vhost(self, target_name, raise_if_no_match=True):
"""Chooses a virtual host based on the given domain name.
.. note:: This makes the vhost SSL-enabled if it isn't already. Follows
@@ -223,6 +228,8 @@ class NginxConfigurator(common.Installer):
hostname. Currently we just ignore this.
:param str target_name: domain name
:param bool raise_if_no_match: True iff not finding a match is an error;
otherwise, return None
:returns: ssl vhost associated with name
:rtype: :class:`~certbot_nginx.obj.VirtualHost`
@@ -233,13 +240,16 @@ class NginxConfigurator(common.Installer):
matches = self._get_ranked_matches(target_name)
vhost = self._select_best_name_match(matches)
if not vhost:
# No matches. Raise a misconfiguration error.
raise errors.MisconfigurationError(
("Cannot find a VirtualHost matching domain %s. "
"In order for Certbot to correctly perform the challenge "
"please add a corresponding server_name directive to your "
"nginx configuration: "
"https://nginx.org/en/docs/http/server_names.html") % (target_name))
if raise_if_no_match:
# No matches. Raise a misconfiguration error.
raise errors.MisconfigurationError(
("Cannot find a VirtualHost matching domain %s. "
"In order for Certbot to correctly perform the challenge "
"please add a corresponding server_name directive to your "
"nginx configuration: "
"https://nginx.org/en/docs/http/server_names.html") % (target_name))
else:
return None
else:
# Note: if we are enhancing with ocsp, vhost should already be ssl.
if not vhost.ssl:
@@ -247,6 +257,65 @@ class NginxConfigurator(common.Installer):
return vhost
def ipv6_info(self, port):
"""Returns tuple of booleans (ipv6_active, ipv6only_present)
ipv6_active is true if any server block listens ipv6 address in any port
ipv6only_present is true if ipv6only=on option exists in any server
block ipv6 listen directive for the specified port.
:param str port: Port to check ipv6only=on directive for
:returns: Tuple containing information if IPv6 is enabled in the global
configuration, and existence of ipv6only directive for specified port
:rtype: tuple of type (bool, bool)
"""
vhosts = self.parser.get_vhosts()
ipv6_active = False
ipv6only_present = False
for vh in vhosts:
for addr in vh.addrs:
if addr.ipv6:
ipv6_active = True
if addr.ipv6only and addr.get_port() == port:
ipv6only_present = True
return (ipv6_active, ipv6only_present)
def _vhost_from_duplicated_default(self, domain):
if self.new_vhost is None:
default_vhost = self._get_default_vhost()
self.new_vhost = self.parser.create_new_vhost_from_default(default_vhost)
if not self.new_vhost.ssl:
self._make_server_ssl(self.new_vhost)
self.new_vhost.names = set()
self.new_vhost.names.add(domain)
name_block = [['\n ', 'server_name']]
for name in self.new_vhost.names:
name_block[0].append(' ')
name_block[0].append(name)
self.parser.add_server_directives(self.new_vhost, name_block, replace=True)
return self.new_vhost
def _get_default_vhost(self):
vhost_list = self.parser.get_vhosts()
# if one has default_server set, return that one
default_vhosts = []
for vhost in vhost_list:
for addr in vhost.addrs:
if addr.default:
default_vhosts.append(vhost)
break
if len(default_vhosts) == 1:
return default_vhosts[0]
# TODO: present a list of vhosts for user to choose from
raise errors.MisconfigurationError("Could not automatically find a matching server"
" block. Set the `server_name` directive to use the Nginx installer.")
def _get_ranked_matches(self, target_name):
"""Returns a ranked list of vhosts that match target_name.
The ranking gives preference to SSL vhosts.
@@ -405,9 +474,12 @@ class NginxConfigurator(common.Installer):
all_names.add(host)
elif not common.private_ips_regex.match(host):
# If it isn't a private IP, do a reverse DNS lookup
# TODO: IPv6 support
try:
socket.inet_aton(host)
if addr.ipv6:
host = addr.get_ipv6_exploded()
socket.inet_pton(socket.AF_INET6, host)
else:
socket.inet_pton(socket.AF_INET, host)
all_names.add(socket.gethostbyaddr(host)[0])
except (socket.error, socket.herror, socket.timeout):
continue
@@ -443,16 +515,38 @@ class NginxConfigurator(common.Installer):
:type vhost: :class:`~certbot_nginx.obj.VirtualHost`
"""
ipv6info = self.ipv6_info(self.config.tls_sni_01_port)
ipv6_block = ['']
ipv4_block = ['']
# If the vhost was implicitly listening on the default Nginx port,
# have it continue to do so.
if len(vhost.addrs) == 0:
listen_block = [['\n ', 'listen', ' ', self.DEFAULT_LISTEN_PORT]]
self.parser.add_server_directives(vhost, listen_block, replace=False)
if vhost.ipv6_enabled():
ipv6_block = ['\n ',
'listen',
' ',
'[::]:{0} ssl'.format(self.config.tls_sni_01_port)]
if not ipv6info[1]:
# ipv6only=on is absent in global config
ipv6_block.append(' ')
ipv6_block.append('ipv6only=on')
if vhost.ipv4_enabled():
ipv4_block = ['\n ',
'listen',
' ',
'{0} ssl'.format(self.config.tls_sni_01_port)]
snakeoil_cert, snakeoil_key = self._get_snakeoil_paths()
ssl_block = ([
['\n ', 'listen', ' ', '{0} ssl'.format(self.config.tls_sni_01_port)],
ipv6_block,
ipv4_block,
['\n ', 'ssl_certificate', ' ', snakeoil_cert],
['\n ', 'ssl_certificate_key', ' ', snakeoil_key],
['\n ', 'include', ' ', self.mod_ssl_conf],

View File

@@ -7,6 +7,7 @@ from pyparsing import (
Literal, White, Forward, Group, Optional, OneOrMore, QuotedString, Regex, ZeroOrMore, Combine)
from pyparsing import stringEnd
from pyparsing import restOfLine
import six
logger = logging.getLogger(__name__)
@@ -71,7 +72,7 @@ class RawNginxDumper(object):
"""Iterates the dumped nginx content."""
blocks = blocks or self.blocks
for b0 in blocks:
if isinstance(b0, str):
if isinstance(b0, six.string_types):
yield b0
continue
item = copy.deepcopy(b0)
@@ -88,7 +89,7 @@ class RawNginxDumper(object):
yield '}'
else: # not a block - list of strings
semicolon = ";"
if isinstance(item[0], str) and item[0].strip() == '#': # comment
if isinstance(item[0], six.string_types) and item[0].strip() == '#': # comment
semicolon = ""
yield "".join(item) + semicolon
@@ -145,7 +146,7 @@ def dump(blocks, _file):
return _file.write(dumps(blocks))
spacey = lambda x: (isinstance(x, str) and x.isspace()) or x == ''
spacey = lambda x: (isinstance(x, six.string_types) and x.isspace()) or x == ''
class UnspacedList(list):
"""Wrap a list [of lists], making any whitespace entries magically invisible"""
@@ -189,13 +190,15 @@ class UnspacedList(list):
item, spaced_item = self._coerce(x)
slicepos = self._spaced_position(i) if i < len(self) else len(self.spaced)
self.spaced.insert(slicepos, spaced_item)
list.insert(self, i, item)
if not spacey(item):
list.insert(self, i, item)
self.dirty = True
def append(self, x):
item, spaced_item = self._coerce(x)
self.spaced.append(spaced_item)
list.append(self, item)
if not spacey(item):
list.append(self, item)
self.dirty = True
def extend(self, x):
@@ -226,7 +229,8 @@ class UnspacedList(list):
raise NotImplementedError("Slice operations on UnspacedLists not yet implemented")
item, spaced_item = self._coerce(value)
self.spaced.__setitem__(self._spaced_position(i), spaced_item)
list.__setitem__(self, i, item)
if not spacey(item):
list.__setitem__(self, i, item)
self.dirty = True
def __delitem__(self, i):
@@ -235,8 +239,8 @@ class UnspacedList(list):
self.dirty = True
def __deepcopy__(self, memo):
l = UnspacedList(self[:])
l.spaced = copy.deepcopy(self.spaced, memo=memo)
new_spaced = copy.deepcopy(self.spaced, memo=memo)
l = UnspacedList(new_spaced)
l.dirty = self.dirty
return l

View File

@@ -34,10 +34,13 @@ class Addr(common.Addr):
UNSPECIFIED_IPV4_ADDRESSES = ('', '*', '0.0.0.0')
CANONICAL_UNSPECIFIED_ADDRESS = UNSPECIFIED_IPV4_ADDRESSES[0]
def __init__(self, host, port, ssl, default):
def __init__(self, host, port, ssl, default, ipv6, ipv6only):
# pylint: disable=too-many-arguments
super(Addr, self).__init__((host, port))
self.ssl = ssl
self.default = default
self.ipv6 = ipv6
self.ipv6only = ipv6only
self.unspecified_address = host in self.UNSPECIFIED_IPV4_ADDRESSES
@classmethod
@@ -46,6 +49,8 @@ class Addr(common.Addr):
parts = str_addr.split(' ')
ssl = False
default = False
ipv6 = False
ipv6only = False
host = ''
port = ''
@@ -56,15 +61,25 @@ class Addr(common.Addr):
if addr.startswith('unix:'):
return None
tup = addr.partition(':')
if re.match(r'^\d+$', tup[0]):
# This is a bare port, not a hostname. E.g. listen 80
host = ''
port = tup[0]
# IPv6 check
ipv6_match = re.match(r'\[.*\]', addr)
if ipv6_match:
ipv6 = True
# IPv6 handling
host = ipv6_match.group()
# The rest of the addr string will be the port, if any
port = addr[ipv6_match.end()+1:]
else:
# This is a host-port tuple. E.g. listen 127.0.0.1:*
host = tup[0]
port = tup[2]
# IPv4 handling
tup = addr.partition(':')
if re.match(r'^\d+$', tup[0]):
# This is a bare port, not a hostname. E.g. listen 80
host = ''
port = tup[0]
else:
# This is a host-port tuple. E.g. listen 127.0.0.1:*
host = tup[0]
port = tup[2]
# The rest of the parts are options; we only care about ssl and default
while len(parts) > 0:
@@ -73,8 +88,10 @@ class Addr(common.Addr):
ssl = True
elif nextpart == 'default_server':
default = True
elif nextpart == "ipv6only=on":
ipv6only = True
return cls(host, port, ssl, default)
return cls(host, port, ssl, default, ipv6, ipv6only)
def to_string(self, include_default=True):
"""Return string representation of Addr"""
@@ -114,8 +131,6 @@ class Addr(common.Addr):
self.tup[1]), self.ipv6) == \
common.Addr((other.CANONICAL_UNSPECIFIED_ADDRESS,
other.tup[1]), other.ipv6)
# Nginx plugin currently doesn't support IPv6 but this will
# future-proof it
return super(Addr, self).__eq__(other)
def __eq__(self, other):
@@ -195,10 +210,24 @@ class VirtualHost(object): # pylint: disable=too-few-public-methods
return True
return False
def ipv6_enabled(self):
"""Return true if one or more of the listen directives in vhost supports
IPv6"""
for a in self.addrs:
if a.ipv6:
return True
def ipv4_enabled(self):
"""Return true if one or more of the listen directives in vhost are IPv4
only"""
for a in self.addrs:
if not a.ipv6:
return True
def _find_directive(directives, directive_name):
"""Find a directive of type directive_name in directives
"""
if not directives or isinstance(directives, str) or len(directives) == 0:
if not directives or isinstance(directives, six.string_types) or len(directives) == 0:
return None
if directives[0] == directive_name:

View File

@@ -6,6 +6,8 @@ import os
import pyparsing
import re
import six
from certbot import errors
from certbot_nginx import obj
@@ -312,6 +314,32 @@ class NginxParser(object):
except errors.MisconfigurationError as err:
raise errors.MisconfigurationError("Problem in %s: %s" % (filename, str(err)))
def create_new_vhost_from_default(self, vhost_template):
"""Duplicate the default vhost in the configuration files.
:param :class:`~certbot_nginx.obj.VirtualHost` vhost_template: The vhost
whose information we copy
:returns: A vhost object for the newly created vhost
:rtype: :class:`~certbot_nginx.obj.VirtualHost`
"""
# TODO: https://github.com/certbot/certbot/issues/5185
# put it in the same file as the template, at the same level
enclosing_block = self.parsed[vhost_template.filep]
for index in vhost_template.path[:-1]:
enclosing_block = enclosing_block[index]
new_location = vhost_template.path[-1] + 1
raw_in_parsed = copy.deepcopy(enclosing_block[vhost_template.path[-1]])
enclosing_block.insert(new_location, raw_in_parsed)
new_vhost = copy.deepcopy(vhost_template)
new_vhost.path[-1] = new_location
for addr in new_vhost.addrs:
addr.default = False
for directive in enclosing_block[new_vhost.path[-1]][1]:
if len(directive) > 0 and directive[0] == 'listen' and 'default_server' in directive:
del directive[directive.index('default_server')]
return new_vhost
def _parse_ssl_options(ssl_options):
if ssl_options is not None:
try:
@@ -444,7 +472,7 @@ def _is_include_directive(entry):
"""
return (isinstance(entry, list) and
len(entry) == 2 and entry[0] == 'include' and
isinstance(entry[1], str))
isinstance(entry[1], six.string_types))
def _is_ssl_on_directive(entry):
"""Checks if an nginx parsed entry is an 'ssl on' directive.
@@ -561,7 +589,8 @@ def _add_directive(block, directive, replace):
directive_name = directive[0]
def can_append(loc, dir_name):
""" Can we append this directive to the block? """
return loc is None or (isinstance(dir_name, str) and dir_name in REPEATABLE_DIRECTIVES)
return loc is None or (isinstance(dir_name, six.string_types)
and dir_name in REPEATABLE_DIRECTIVES)
err_fmt = 'tried to insert directive "{0}" but found conflicting "{1}".'

View File

@@ -45,7 +45,7 @@ class NginxConfiguratorTest(util.NginxTest):
def test_prepare(self):
self.assertEqual((1, 6, 2), self.config.version)
self.assertEqual(8, len(self.config.parser.parsed))
self.assertEqual(10, len(self.config.parser.parsed))
@mock.patch("certbot_nginx.configurator.util.exe_exists")
@mock.patch("certbot_nginx.configurator.subprocess.Popen")
@@ -89,7 +89,7 @@ class NginxConfiguratorTest(util.NginxTest):
self.assertEqual(names, set(
["155.225.50.69.nephoscale.net", "www.example.org", "another.alias",
"migration.com", "summer.com", "geese.com", "sslon.com",
"globalssl.com", "globalsslsetssl.com"]))
"globalssl.com", "globalsslsetssl.com", "ipv6.com", "ipv6ssl.com"]))
def test_supported_enhancements(self):
self.assertEqual(['redirect', 'staple-ocsp'],
@@ -131,6 +131,7 @@ class NginxConfiguratorTest(util.NginxTest):
server_conf = set(['somename', 'another.alias', 'alias'])
example_conf = set(['.example.com', 'example.*'])
foo_conf = set(['*.www.foo.com', '*.www.example.com'])
ipv6_conf = set(['ipv6.com'])
results = {'localhost': localhost_conf,
'alias': server_conf,
@@ -139,7 +140,8 @@ class NginxConfiguratorTest(util.NginxTest):
'www.example.com': example_conf,
'test.www.example.com': foo_conf,
'abc.www.foo.com': foo_conf,
'www.bar.co.uk': localhost_conf}
'www.bar.co.uk': localhost_conf,
'ipv6.com': ipv6_conf}
conf_path = {'localhost': "etc_nginx/nginx.conf",
'alias': "etc_nginx/nginx.conf",
@@ -148,7 +150,8 @@ class NginxConfiguratorTest(util.NginxTest):
'www.example.com': "etc_nginx/sites-enabled/example.com",
'test.www.example.com': "etc_nginx/foo.conf",
'abc.www.foo.com': "etc_nginx/foo.conf",
'www.bar.co.uk': "etc_nginx/nginx.conf"}
'www.bar.co.uk': "etc_nginx/nginx.conf",
'ipv6.com': "etc_nginx/sites-enabled/ipv6.com"}
bad_results = ['www.foo.com', 'example', 't.www.bar.co',
'69.255.225.155']
@@ -159,11 +162,24 @@ class NginxConfiguratorTest(util.NginxTest):
self.assertEqual(results[name], vhost.names)
self.assertEqual(conf_path[name], path)
# IPv6 specific checks
if name == "ipv6.com":
self.assertTrue(vhost.ipv6_enabled())
# Make sure that we have SSL enabled also for IPv6 addr
self.assertTrue(
any([True for x in vhost.addrs if x.ssl and x.ipv6]))
for name in bad_results:
self.assertRaises(errors.MisconfigurationError,
self.config.choose_vhost, name)
def test_ipv6only(self):
# ipv6_info: (ipv6_active, ipv6only_present)
self.assertEquals((True, False), self.config.ipv6_info("80"))
# Port 443 has ipv6only=on because of ipv6ssl.com vhost
self.assertEquals((True, True), self.config.ipv6_info("443"))
def test_more_info(self):
self.assertTrue('nginx.conf' in self.config.more_info())
@@ -558,6 +574,145 @@ class NginxConfiguratorTest(util.NginxTest):
self.assertTrue(util.contains_at_depth(
generated_conf, ['ssl_stapling_verify', 'on'], 2))
def test_deploy_no_match_default_set(self):
default_conf = self.config.parser.abs_path('sites-enabled/default')
foo_conf = self.config.parser.abs_path('foo.conf')
del self.config.parser.parsed[foo_conf][2][1][0][1][0] # remove default_server
self.config.version = (1, 3, 1)
self.config.deploy_cert(
"www.nomatch.com",
"example/cert.pem",
"example/key.pem",
"example/chain.pem",
"example/fullchain.pem")
self.config.save()
self.config.parser.load()
parsed_default_conf = util.filter_comments(self.config.parser.parsed[default_conf])
self.assertEqual([[['server'],
[['listen', 'myhost', 'default_server'],
['listen', 'otherhost', 'default_server'],
['server_name', 'www.example.org'],
[['location', '/'],
[['root', 'html'],
['index', 'index.html', 'index.htm']]]]],
[['server'],
[['listen', 'myhost'],
['listen', 'otherhost'],
['server_name', 'www.nomatch.com'],
[['location', '/'],
[['root', 'html'],
['index', 'index.html', 'index.htm']]],
['listen', '5001', 'ssl'],
['ssl_certificate', 'example/fullchain.pem'],
['ssl_certificate_key', 'example/key.pem'],
['include', self.config.mod_ssl_conf],
['ssl_dhparam', self.config.ssl_dhparams]]]],
parsed_default_conf)
self.config.deploy_cert(
"nomatch.com",
"example/cert.pem",
"example/key.pem",
"example/chain.pem",
"example/fullchain.pem")
self.config.save()
self.config.parser.load()
parsed_default_conf = util.filter_comments(self.config.parser.parsed[default_conf])
self.assertTrue(util.contains_at_depth(parsed_default_conf, "nomatch.com", 3))
def test_deploy_no_match_default_set_multi_level_path(self):
default_conf = self.config.parser.abs_path('sites-enabled/default')
foo_conf = self.config.parser.abs_path('foo.conf')
del self.config.parser.parsed[default_conf][0][1][0]
del self.config.parser.parsed[default_conf][0][1][0]
self.config.version = (1, 3, 1)
self.config.deploy_cert(
"www.nomatch.com",
"example/cert.pem",
"example/key.pem",
"example/chain.pem",
"example/fullchain.pem")
self.config.save()
self.config.parser.load()
parsed_foo_conf = util.filter_comments(self.config.parser.parsed[foo_conf])
self.assertEqual([['server'],
[['listen', '*:80', 'ssl'],
['server_name', 'www.nomatch.com'],
['root', '/home/ubuntu/sites/foo/'],
[['location', '/status'], [[['types'], [['image/jpeg', 'jpg']]]]],
[['location', '~', 'case_sensitive\\.php$'], [['index', 'index.php'],
['root', '/var/root']]],
[['location', '~*', 'case_insensitive\\.php$'], []],
[['location', '=', 'exact_match\\.php$'], []],
[['location', '^~', 'ignore_regex\\.php$'], []],
['ssl_certificate', 'example/fullchain.pem'],
['ssl_certificate_key', 'example/key.pem']]],
parsed_foo_conf[1][1][1])
def test_deploy_no_match_no_default_set(self):
default_conf = self.config.parser.abs_path('sites-enabled/default')
foo_conf = self.config.parser.abs_path('foo.conf')
del self.config.parser.parsed[default_conf][0][1][0]
del self.config.parser.parsed[default_conf][0][1][0]
del self.config.parser.parsed[foo_conf][2][1][0][1][0]
self.config.version = (1, 3, 1)
self.assertRaises(errors.MisconfigurationError, self.config.deploy_cert,
"www.nomatch.com", "example/cert.pem", "example/key.pem",
"example/chain.pem", "example/fullchain.pem")
def test_deploy_no_match_fail_multiple_defaults(self):
self.config.version = (1, 3, 1)
self.assertRaises(errors.MisconfigurationError, self.config.deploy_cert,
"www.nomatch.com", "example/cert.pem", "example/key.pem",
"example/chain.pem", "example/fullchain.pem")
def test_deploy_no_match_add_redirect(self):
default_conf = self.config.parser.abs_path('sites-enabled/default')
foo_conf = self.config.parser.abs_path('foo.conf')
del self.config.parser.parsed[foo_conf][2][1][0][1][0] # remove default_server
self.config.version = (1, 3, 1)
self.config.deploy_cert(
"www.nomatch.com",
"example/cert.pem",
"example/key.pem",
"example/chain.pem",
"example/fullchain.pem")
self.config.deploy_cert(
"nomatch.com",
"example/cert.pem",
"example/key.pem",
"example/chain.pem",
"example/fullchain.pem")
self.config.enhance("www.nomatch.com", "redirect")
self.config.save()
self.config.parser.load()
expected = [
['if', '($scheme', '!=', '"https")'],
[['return', '301', 'https://$host$request_uri']]
]
generated_conf = self.config.parser.parsed[default_conf]
self.assertTrue(util.contains_at_depth(generated_conf, expected, 2))
class InstallSslOptionsConfTest(util.NginxTest):
"""Test that the options-ssl-nginx.conf file is installed and updated properly."""

View File

@@ -50,7 +50,9 @@ class NginxParserTest(util.NginxTest): #pylint: disable=too-many-public-methods
'sites-enabled/example.com',
'sites-enabled/migration.com',
'sites-enabled/sslon.com',
'sites-enabled/globalssl.com']]),
'sites-enabled/globalssl.com',
'sites-enabled/ipv6.com',
'sites-enabled/ipv6ssl.com']]),
set(nparser.parsed.keys()))
self.assertEqual([['server_name', 'somename', 'alias', 'another.alias']],
nparser.parsed[nparser.abs_path('server.conf')])
@@ -74,7 +76,7 @@ class NginxParserTest(util.NginxTest): #pylint: disable=too-many-public-methods
parsed = nparser._parse_files(nparser.abs_path(
'sites-enabled/example.com.test'))
self.assertEqual(3, len(glob.glob(nparser.abs_path('*.test'))))
self.assertEqual(5, len(
self.assertEqual(7, len(
glob.glob(nparser.abs_path('sites-enabled/*.test'))))
self.assertEqual([[['server'], [['listen', '69.50.225.155:9000'],
['listen', '127.0.0.1'],
@@ -110,7 +112,8 @@ class NginxParserTest(util.NginxTest): #pylint: disable=too-many-public-methods
vhosts = nparser.get_vhosts()
vhost = obj.VirtualHost(nparser.abs_path('sites-enabled/globalssl.com'),
[obj.Addr('4.8.2.6', '57', True, False)],
[obj.Addr('4.8.2.6', '57', True, False,
False, False)],
True, True, set(['globalssl.com']), [], [0])
globalssl_com = [x for x in vhosts if 'globalssl.com' in x.filep][0]
@@ -121,34 +124,42 @@ class NginxParserTest(util.NginxTest): #pylint: disable=too-many-public-methods
vhosts = nparser.get_vhosts()
vhost1 = obj.VirtualHost(nparser.abs_path('nginx.conf'),
[obj.Addr('', '8080', False, False)],
[obj.Addr('', '8080', False, False,
False, False)],
False, True,
set(['localhost',
r'~^(www\.)?(example|bar)\.']),
[], [10, 1, 9])
vhost2 = obj.VirtualHost(nparser.abs_path('nginx.conf'),
[obj.Addr('somename', '8080', False, False),
obj.Addr('', '8000', False, False)],
[obj.Addr('somename', '8080', False, False,
False, False),
obj.Addr('', '8000', False, False,
False, False)],
False, True,
set(['somename', 'another.alias', 'alias']),
[], [10, 1, 12])
vhost3 = obj.VirtualHost(nparser.abs_path('sites-enabled/example.com'),
[obj.Addr('69.50.225.155', '9000',
False, False),
obj.Addr('127.0.0.1', '', False, False)],
False, False, False, False),
obj.Addr('127.0.0.1', '', False, False,
False, False)],
False, True,
set(['.example.com', 'example.*']), [], [0])
vhost4 = obj.VirtualHost(nparser.abs_path('sites-enabled/default'),
[obj.Addr('myhost', '', False, True)],
[obj.Addr('myhost', '', False, True,
False, False),
obj.Addr('otherhost', '', False, True,
False, False)],
False, True, set(['www.example.org']),
[], [0])
vhost5 = obj.VirtualHost(nparser.abs_path('foo.conf'),
[obj.Addr('*', '80', True, True)],
[obj.Addr('*', '80', True, True,
False, False)],
True, True, set(['*.www.foo.com',
'*.www.example.com']),
[], [2, 1, 0])
self.assertEqual(10, len(vhosts))
self.assertEqual(12, len(vhosts))
example_com = [x for x in vhosts if 'example.com' in x.filep][0]
self.assertEqual(vhost3, example_com)
default = [x for x in vhosts if 'default' in x.filep][0]
@@ -395,6 +406,29 @@ class NginxParserTest(util.NginxTest): #pylint: disable=too-many-public-methods
])
self.assertTrue(server['ssl'])
def test_create_new_vhost_from_default(self):
nparser = parser.NginxParser(self.config_path)
vhosts = nparser.get_vhosts()
default = [x for x in vhosts if 'default' in x.filep][0]
new_vhost = nparser.create_new_vhost_from_default(default)
nparser.filedump(ext='')
# check properties of new vhost
self.assertFalse(next(iter(new_vhost.addrs)).default)
self.assertNotEqual(new_vhost.path, default.path)
# check that things are written to file correctly
new_nparser = parser.NginxParser(self.config_path)
new_vhosts = new_nparser.get_vhosts()
new_defaults = [x for x in new_vhosts if 'default' in x.filep]
self.assertEqual(len(new_defaults), 2)
new_vhost_parsed = new_defaults[1]
self.assertFalse(next(iter(new_vhost_parsed.addrs)).default)
self.assertEqual(next(iter(default.names)), next(iter(new_vhost_parsed.names)))
self.assertEqual(len(default.raw), len(new_vhost_parsed.raw))
self.assertTrue(next(iter(default.addrs)).super_eq(next(iter(new_vhost_parsed.addrs))))
if __name__ == "__main__":
unittest.main() # pragma: no cover

View File

@@ -1,5 +1,6 @@
server {
listen myhost default_server;
listen otherhost default_server;
server_name www.example.org;
location / {

View File

@@ -0,0 +1,5 @@
server {
listen 80;
listen [::]:80;
server_name ipv6.com;
}

View File

@@ -0,0 +1,5 @@
server {
listen 443 ssl;
listen [::]:443 ssl ipv6only=on;
server_name ipv6ssl.com;
}

View File

@@ -66,7 +66,7 @@ class TlsSniPerformTest(util.NginxTest):
self.sni.add_chall(self.achalls[1])
mock_choose.return_value = None
result = self.sni.perform()
self.assertTrue(result is None)
self.assertFalse(result is None)
def test_perform0(self):
responses = self.sni.perform()
@@ -125,10 +125,10 @@ class TlsSniPerformTest(util.NginxTest):
self.sni.add_chall(self.achalls[0])
self.sni.add_chall(self.achalls[2])
v_addr1 = [obj.Addr("69.50.225.155", "9000", True, False),
obj.Addr("127.0.0.1", "", False, False)]
v_addr2 = [obj.Addr("myhost", "", False, True)]
v_addr2_print = [obj.Addr("myhost", "", False, False)]
v_addr1 = [obj.Addr("69.50.225.155", "9000", True, False, False, False),
obj.Addr("127.0.0.1", "", False, False, False, False)]
v_addr2 = [obj.Addr("myhost", "", False, True, False, False)]
v_addr2_print = [obj.Addr("myhost", "", False, False, False, False)]
ll_addr = [v_addr1, v_addr2]
self.sni._mod_config(ll_addr) # pylint: disable=protected-access

View File

@@ -51,19 +51,32 @@ class NginxTlsSni01(common.TLSSNI01):
default_addr = "{0} ssl".format(
self.configurator.config.tls_sni_01_port)
for achall in self.achalls:
vhost = self.configurator.choose_vhost(achall.domain)
if vhost is None:
logger.error(
"No nginx vhost exists with server_name matching: %s. "
"Please specify server_names in the Nginx config.",
achall.domain)
return None
ipv6, ipv6only = self.configurator.ipv6_info(
self.configurator.config.tls_sni_01_port)
if vhost.addrs:
for achall in self.achalls:
vhost = self.configurator.choose_vhost(achall.domain, raise_if_no_match=False)
if vhost is not None and vhost.addrs:
addresses.append(list(vhost.addrs))
else:
addresses.append([obj.Addr.fromstring(default_addr)])
if ipv6:
# If IPv6 is active in Nginx configuration
ipv6_addr = "[::]:{0} ssl".format(
self.configurator.config.tls_sni_01_port)
if not ipv6only:
# If ipv6only=on is not already present in the config
ipv6_addr = ipv6_addr + " ipv6only=on"
addresses.append([obj.Addr.fromstring(default_addr),
obj.Addr.fromstring(ipv6_addr)])
logger.info(("Using default addresses %s and %s for " +
"TLSSNI01 authentication."),
default_addr,
ipv6_addr)
else:
addresses.append([obj.Addr.fromstring(default_addr)])
logger.info("Using default address %s for TLSSNI01 authentication.",
default_addr)
# Create challenge certs
responses = [self._setup_challenge_cert(x) for x in self.achalls]
@@ -117,7 +130,6 @@ class NginxTlsSni01(common.TLSSNI01):
raise errors.MisconfigurationError(
'Certbot could not find an HTTP block to include '
'TLS-SNI-01 challenges in %s.' % root)
config = [self._make_server_block(pair[0], pair[1])
for pair in six.moves.zip(self.achalls, ll_addrs)]
config = nginxparser.UnspacedList(config)

View File

@@ -251,7 +251,7 @@ class Addr(object):
"""Normalized representation of addr/port tuple
"""
if self.ipv6:
return (self._normalize_ipv6(self.tup[0]), self.tup[1])
return (self.get_ipv6_exploded(), self.tup[1])
return self.tup
def __eq__(self, other):