mirror of
https://github.com/certbot/certbot.git
synced 2026-07-30 17:54:25 +02:00
fixup fromstring return types (#10162)
in https://github.com/certbot/certbot/pull/9124 we had the problem of certbot-nginx's `Addr.fromstring` method possibly returning None which is not possible in the `Addr` method in the certbot base class or in certbot-apache. we fixed this by telling mypy the common `Addr.fromstring` method returns an `Optional[Addr]` (despite it actually always returning an `Addr`) and then unnecessarily complicating certbot-apache's code a bit. the need for extra complexity with this approach is going even further in https://github.com/certbot/certbot/pull/10151 where we have to use `cast` to assure mypy that the type isn't actually `Optional`. i personally don't like all this
This commit is contained in:
@@ -994,9 +994,7 @@ class ApacheConfigurator(common.Configurator):
|
|||||||
for arg in args:
|
for arg in args:
|
||||||
arg_value = self.parser.get_arg(arg)
|
arg_value = self.parser.get_arg(arg)
|
||||||
if arg_value is not None:
|
if arg_value is not None:
|
||||||
addr = obj.Addr.fromstring(arg_value)
|
addrs.add(obj.Addr.fromstring(arg_value))
|
||||||
if addr is not None:
|
|
||||||
addrs.add(addr)
|
|
||||||
is_ssl = False
|
is_ssl = False
|
||||||
|
|
||||||
if self.parser.find_dir("SSLEngine", "on", start=path, exclude=False):
|
if self.parser.find_dir("SSLEngine", "on", start=path, exclude=False):
|
||||||
@@ -1126,9 +1124,7 @@ class ApacheConfigurator(common.Configurator):
|
|||||||
"""
|
"""
|
||||||
addrs = set()
|
addrs = set()
|
||||||
for param in node.parameters:
|
for param in node.parameters:
|
||||||
addr = obj.Addr.fromstring(param)
|
addrs.add(obj.Addr.fromstring(param))
|
||||||
if addr:
|
|
||||||
addrs.add(addr)
|
|
||||||
|
|
||||||
is_ssl = False
|
is_ssl = False
|
||||||
# Exclusion to match the behavior in get_virtual_hosts_v2
|
# Exclusion to match the behavior in get_virtual_hosts_v2
|
||||||
@@ -1646,10 +1642,9 @@ class ApacheConfigurator(common.Configurator):
|
|||||||
for addr in ssl_addr_p:
|
for addr in ssl_addr_p:
|
||||||
old_addr = obj.Addr.fromstring(
|
old_addr = obj.Addr.fromstring(
|
||||||
str(self.parser.get_arg(addr)))
|
str(self.parser.get_arg(addr)))
|
||||||
if old_addr:
|
ssl_addr = old_addr.get_addr_obj("443")
|
||||||
ssl_addr = old_addr.get_addr_obj("443")
|
self.parser.aug.set(addr, str(ssl_addr))
|
||||||
self.parser.aug.set(addr, str(ssl_addr))
|
ssl_addrs.add(ssl_addr)
|
||||||
ssl_addrs.add(ssl_addr)
|
|
||||||
|
|
||||||
return ssl_addrs
|
return ssl_addrs
|
||||||
|
|
||||||
|
|||||||
@@ -146,7 +146,7 @@ class NginxHttp01(common.ChallengePerformer):
|
|||||||
:returns: list of :class:`certbot_nginx._internal.obj.Addr` to apply
|
:returns: list of :class:`certbot_nginx._internal.obj.Addr` to apply
|
||||||
:rtype: list
|
:rtype: list
|
||||||
"""
|
"""
|
||||||
addresses: List[Optional[Addr]] = []
|
addresses: List[Addr] = []
|
||||||
default_addr = "%s" % self.configurator.config.http01_port
|
default_addr = "%s" % self.configurator.config.http01_port
|
||||||
ipv6_addr = "[::]:{0}".format(
|
ipv6_addr = "[::]:{0}".format(
|
||||||
self.configurator.config.http01_port)
|
self.configurator.config.http01_port)
|
||||||
@@ -169,7 +169,7 @@ class NginxHttp01(common.ChallengePerformer):
|
|||||||
logger.debug("Using default address %s for authentication.",
|
logger.debug("Using default address %s for authentication.",
|
||||||
default_addr)
|
default_addr)
|
||||||
|
|
||||||
return [address for address in addresses if address]
|
return addresses
|
||||||
|
|
||||||
def _get_validation_path(self, achall: KeyAuthorizationAnnotatedChallenge) -> str:
|
def _get_validation_path(self, achall: KeyAuthorizationAnnotatedChallenge) -> str:
|
||||||
return os.sep + os.path.join(challenges.HTTP01.URI_ROOT_PATH, achall.chall.encode("token"))
|
return os.sep + os.path.join(challenges.HTTP01.URI_ROOT_PATH, achall.chall.encode("token"))
|
||||||
|
|||||||
@@ -12,6 +12,10 @@ from certbot.plugins import common
|
|||||||
ADD_HEADER_DIRECTIVE = 'add_header'
|
ADD_HEADER_DIRECTIVE = 'add_header'
|
||||||
|
|
||||||
|
|
||||||
|
class SocketAddrError(Exception):
|
||||||
|
"""Raised when a UNIX-domain socket address is encountered."""
|
||||||
|
|
||||||
|
|
||||||
class Addr(common.Addr):
|
class Addr(common.Addr):
|
||||||
r"""Represents an Nginx address, i.e. what comes after the 'listen'
|
r"""Represents an Nginx address, i.e. what comes after the 'listen'
|
||||||
directive.
|
directive.
|
||||||
@@ -51,8 +55,15 @@ class Addr(common.Addr):
|
|||||||
self.unspecified_address = host in self.UNSPECIFIED_IPV4_ADDRESSES
|
self.unspecified_address = host in self.UNSPECIFIED_IPV4_ADDRESSES
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def fromstring(cls, str_addr: str) -> Optional["Addr"]:
|
def fromstring(cls, str_addr: str) -> "Addr":
|
||||||
"""Initialize Addr from string."""
|
"""Initialize Addr from string.
|
||||||
|
|
||||||
|
:param str str_addr: nginx address string
|
||||||
|
:returns: parsed nginx address
|
||||||
|
:rtype: Addr
|
||||||
|
:raises SocketAddrError: if a UNIX-domain socket address is given
|
||||||
|
|
||||||
|
"""
|
||||||
parts = str_addr.split(' ')
|
parts = str_addr.split(' ')
|
||||||
ssl = False
|
ssl = False
|
||||||
default = False
|
default = False
|
||||||
@@ -64,9 +75,9 @@ class Addr(common.Addr):
|
|||||||
# The first part must be the address
|
# The first part must be the address
|
||||||
addr = parts.pop(0)
|
addr = parts.pop(0)
|
||||||
|
|
||||||
# Ignore UNIX-domain sockets
|
# Raise for UNIX-domain sockets
|
||||||
if addr.startswith('unix:'):
|
if addr.startswith('unix:'):
|
||||||
return None
|
raise SocketAddrError(f'encountered UNIX-domain socket address {str_addr}')
|
||||||
|
|
||||||
# IPv6 check
|
# IPv6 check
|
||||||
ipv6_match = re.match(r'\[.*\]', addr)
|
ipv6_match = re.match(r'\[.*\]', addr)
|
||||||
|
|||||||
@@ -794,11 +794,14 @@ def _parse_server_raw(server: UnspacedList) -> Dict[str, Any]:
|
|||||||
if not directive:
|
if not directive:
|
||||||
continue
|
continue
|
||||||
if directive[0] == 'listen':
|
if directive[0] == 'listen':
|
||||||
addr = obj.Addr.fromstring(" ".join(directive[1:]))
|
try:
|
||||||
if addr:
|
addr = obj.Addr.fromstring(" ".join(directive[1:]))
|
||||||
addrs.add(addr)
|
except obj.SocketAddrError:
|
||||||
if addr.ssl:
|
# Ignore UNIX-domain socket addresses
|
||||||
ssl = True
|
continue
|
||||||
|
addrs.add(addr)
|
||||||
|
if addr.ssl:
|
||||||
|
ssl = True
|
||||||
elif directive[0] == 'server_name':
|
elif directive[0] == 'server_name':
|
||||||
names.update(x.strip('"\'') for x in directive[1:])
|
names.update(x.strip('"\'') for x in directive[1:])
|
||||||
elif _is_ssl_on_directive(directive):
|
elif _is_ssl_on_directive(directive):
|
||||||
|
|||||||
@@ -16,8 +16,7 @@ class AddrTest(unittest.TestCase):
|
|||||||
self.addr4 = Addr.fromstring("*:80 default_server ssl")
|
self.addr4 = Addr.fromstring("*:80 default_server ssl")
|
||||||
self.addr5 = Addr.fromstring("myhost")
|
self.addr5 = Addr.fromstring("myhost")
|
||||||
self.addr6 = Addr.fromstring("80 default_server spdy")
|
self.addr6 = Addr.fromstring("80 default_server spdy")
|
||||||
self.addr7 = Addr.fromstring("unix:/var/run/nginx.sock")
|
self.addr7 = Addr.fromstring("*:80 default ssl")
|
||||||
self.addr8 = Addr.fromstring("*:80 default ssl")
|
|
||||||
|
|
||||||
def test_fromstring(self):
|
def test_fromstring(self):
|
||||||
assert self.addr1.get_addr() == "192.168.1.1"
|
assert self.addr1.get_addr() == "192.168.1.1"
|
||||||
@@ -50,9 +49,13 @@ class AddrTest(unittest.TestCase):
|
|||||||
assert self.addr6.ssl is False
|
assert self.addr6.ssl is False
|
||||||
assert self.addr6.default is True
|
assert self.addr6.default is True
|
||||||
|
|
||||||
assert self.addr8.default is True
|
assert self.addr7.default is True
|
||||||
|
|
||||||
assert self.addr7 is None
|
def test_fromstring_socket(self):
|
||||||
|
from certbot_nginx._internal.obj import Addr, SocketAddrError
|
||||||
|
socket_string = r"unix:/var/run/nginx.sock"
|
||||||
|
with pytest.raises(SocketAddrError, match=socket_string):
|
||||||
|
Addr.fromstring(socket_string)
|
||||||
|
|
||||||
def test_str(self):
|
def test_str(self):
|
||||||
assert str(self.addr1) == "192.168.1.1"
|
assert str(self.addr1) == "192.168.1.1"
|
||||||
@@ -61,7 +64,7 @@ class AddrTest(unittest.TestCase):
|
|||||||
assert str(self.addr4) == "*:80 default_server ssl"
|
assert str(self.addr4) == "*:80 default_server ssl"
|
||||||
assert str(self.addr5) == "myhost"
|
assert str(self.addr5) == "myhost"
|
||||||
assert str(self.addr6) == "80 default_server"
|
assert str(self.addr6) == "80 default_server"
|
||||||
assert str(self.addr8) == "*:80 default_server ssl"
|
assert str(self.addr7) == "*:80 default_server ssl"
|
||||||
|
|
||||||
def test_to_string(self):
|
def test_to_string(self):
|
||||||
assert self.addr1.to_string() == "192.168.1.1"
|
assert self.addr1.to_string() == "192.168.1.1"
|
||||||
|
|||||||
@@ -261,7 +261,7 @@ class Addr:
|
|||||||
self.ipv6 = ipv6
|
self.ipv6 = ipv6
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def fromstring(cls: Type[GenericAddr], str_addr: str) -> Optional[GenericAddr]:
|
def fromstring(cls: Type[GenericAddr], str_addr: str) -> GenericAddr:
|
||||||
"""Initialize Addr from string."""
|
"""Initialize Addr from string."""
|
||||||
if str_addr.startswith('['):
|
if str_addr.startswith('['):
|
||||||
# ipv6 addresses starts with [
|
# ipv6 addresses starts with [
|
||||||
|
|||||||
Reference in New Issue
Block a user