Compare commits
1 Commits
3.0.0
...
nginx-reve
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
d435eaf25e |
@@ -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],
|
||||
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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}".'
|
||||
|
||||
|
||||
@@ -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."""
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
server {
|
||||
listen myhost default_server;
|
||||
listen otherhost default_server;
|
||||
server_name www.example.org;
|
||||
|
||||
location / {
|
||||
|
||||
5
certbot-nginx/certbot_nginx/tests/testdata/etc_nginx/sites-enabled/ipv6.com
vendored
Normal file
5
certbot-nginx/certbot_nginx/tests/testdata/etc_nginx/sites-enabled/ipv6.com
vendored
Normal file
@@ -0,0 +1,5 @@
|
||||
server {
|
||||
listen 80;
|
||||
listen [::]:80;
|
||||
server_name ipv6.com;
|
||||
}
|
||||
5
certbot-nginx/certbot_nginx/tests/testdata/etc_nginx/sites-enabled/ipv6ssl.com
vendored
Normal file
5
certbot-nginx/certbot_nginx/tests/testdata/etc_nginx/sites-enabled/ipv6ssl.com
vendored
Normal file
@@ -0,0 +1,5 @@
|
||||
server {
|
||||
listen 443 ssl;
|
||||
listen [::]:443 ssl ipv6only=on;
|
||||
server_name ipv6ssl.com;
|
||||
}
|
||||
@@ -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
|
||||
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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):
|
||||
|
||||
Reference in New Issue
Block a user