mirror of
https://github.com/certbot/certbot.git
synced 2026-08-03 12:02:03 +02:00
Enable checking of type annotation in Nginx plugin (#5997)
* Adds type checking for certbot-nginx * First pass at type annotation in certbot-nginx * Ensure linting is disabled for timing imports * Makes container types specific per PR comments * Removes unnecessary lint option
This commit is contained in:
committed by
Brad Warren
parent
802fcc99ee
commit
307f45f88f
@@ -28,6 +28,9 @@ from certbot_nginx import nginxparser
|
|||||||
from certbot_nginx import parser
|
from certbot_nginx import parser
|
||||||
from certbot_nginx import tls_sni_01
|
from certbot_nginx import tls_sni_01
|
||||||
from certbot_nginx import http_01
|
from certbot_nginx import http_01
|
||||||
|
from certbot_nginx import obj # pylint: disable=unused-import
|
||||||
|
from acme.magic_typing import List, Dict, Set # pylint: disable=unused-import, no-name-in-module
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
@@ -98,8 +101,8 @@ class NginxConfigurator(common.Installer):
|
|||||||
|
|
||||||
# List of vhosts configured per wildcard domain on this run.
|
# List of vhosts configured per wildcard domain on this run.
|
||||||
# used by deploy_cert() and enhance()
|
# used by deploy_cert() and enhance()
|
||||||
self._wildcard_vhosts = {}
|
self._wildcard_vhosts = {} # type: Dict[str, List[obj.VirtualHost]]
|
||||||
self._wildcard_redirect_vhosts = {}
|
self._wildcard_redirect_vhosts = {} # type: Dict[str, List[obj.VirtualHost]]
|
||||||
|
|
||||||
# Add number of outstanding challenges
|
# Add number of outstanding challenges
|
||||||
self._chall_out = 0
|
self._chall_out = 0
|
||||||
@@ -528,7 +531,7 @@ class NginxConfigurator(common.Installer):
|
|||||||
:rtype: set
|
:rtype: set
|
||||||
|
|
||||||
"""
|
"""
|
||||||
all_names = set()
|
all_names = set() # type: Set[str]
|
||||||
|
|
||||||
for vhost in self.parser.get_vhosts():
|
for vhost in self.parser.get_vhosts():
|
||||||
all_names.update(vhost.names)
|
all_names.update(vhost.names)
|
||||||
@@ -824,7 +827,7 @@ class NginxConfigurator(common.Installer):
|
|||||||
self.parser.add_server_directives(vhost,
|
self.parser.add_server_directives(vhost,
|
||||||
stapling_directives)
|
stapling_directives)
|
||||||
except errors.MisconfigurationError as error:
|
except errors.MisconfigurationError as error:
|
||||||
logger.debug(error)
|
logger.debug(str(error))
|
||||||
raise errors.PluginError("An error occurred while enabling OCSP "
|
raise errors.PluginError("An error occurred while enabling OCSP "
|
||||||
"stapling for {0}.".format(vhost.names))
|
"stapling for {0}.".format(vhost.names))
|
||||||
|
|
||||||
@@ -892,7 +895,7 @@ class NginxConfigurator(common.Installer):
|
|||||||
universal_newlines=True)
|
universal_newlines=True)
|
||||||
text = proc.communicate()[1] # nginx prints output to stderr
|
text = proc.communicate()[1] # nginx prints output to stderr
|
||||||
except (OSError, ValueError) as error:
|
except (OSError, ValueError) as error:
|
||||||
logger.debug(error, exc_info=True)
|
logger.debug(str(error), exc_info=True)
|
||||||
raise errors.PluginError(
|
raise errors.PluginError(
|
||||||
"Unable to run %s -V" % self.conf('ctl'))
|
"Unable to run %s -V" % self.conf('ctl'))
|
||||||
|
|
||||||
|
|||||||
@@ -10,6 +10,7 @@ from certbot.plugins import common
|
|||||||
|
|
||||||
from certbot_nginx import obj
|
from certbot_nginx import obj
|
||||||
from certbot_nginx import nginxparser
|
from certbot_nginx import nginxparser
|
||||||
|
from acme.magic_typing import List # pylint: disable=unused-import, no-name-in-module
|
||||||
|
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
@@ -113,7 +114,7 @@ class NginxHttp01(common.ChallengePerformer):
|
|||||||
:returns: list of :class:`certbot_nginx.obj.Addr` to apply
|
:returns: list of :class:`certbot_nginx.obj.Addr` to apply
|
||||||
:rtype: list
|
:rtype: list
|
||||||
"""
|
"""
|
||||||
addresses = []
|
addresses = [] # type: List[obj.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)
|
||||||
|
|||||||
@@ -248,7 +248,7 @@ class UnspacedList(list):
|
|||||||
"""Recurse through the parse tree to figure out if any sublists are dirty"""
|
"""Recurse through the parse tree to figure out if any sublists are dirty"""
|
||||||
if self.dirty:
|
if self.dirty:
|
||||||
return True
|
return True
|
||||||
return any((isinstance(x, list) and x.is_dirty() for x in self))
|
return any((isinstance(x, UnspacedList) and x.is_dirty() for x in self))
|
||||||
|
|
||||||
def _spaced_position(self, idx):
|
def _spaced_position(self, idx):
|
||||||
"Convert from indexes in the unspaced list to positions in the spaced one"
|
"Convert from indexes in the unspaced list to positions in the spaced one"
|
||||||
|
|||||||
@@ -13,7 +13,7 @@ from certbot import errors
|
|||||||
|
|
||||||
from certbot_nginx import obj
|
from certbot_nginx import obj
|
||||||
from certbot_nginx import nginxparser
|
from certbot_nginx import nginxparser
|
||||||
|
from acme.magic_typing import Union, Dict, Set, Any, List, Tuple # pylint: disable=unused-import, no-name-in-module
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
@@ -28,7 +28,7 @@ class NginxParser(object):
|
|||||||
"""
|
"""
|
||||||
|
|
||||||
def __init__(self, root):
|
def __init__(self, root):
|
||||||
self.parsed = {}
|
self.parsed = {} # type: Dict[str, Union[List, nginxparser.UnspacedList]]
|
||||||
self.root = os.path.abspath(root)
|
self.root = os.path.abspath(root)
|
||||||
self.config_root = self._find_config_root()
|
self.config_root = self._find_config_root()
|
||||||
|
|
||||||
@@ -90,7 +90,7 @@ class NginxParser(object):
|
|||||||
"""
|
"""
|
||||||
servers = self._get_raw_servers()
|
servers = self._get_raw_servers()
|
||||||
|
|
||||||
addr_to_ssl = {}
|
addr_to_ssl = {} # type: Dict[Tuple[str, str], bool]
|
||||||
for filename in servers:
|
for filename in servers:
|
||||||
for server, _ in servers[filename]:
|
for server, _ in servers[filename]:
|
||||||
# Parse the server block to save addr info
|
# Parse the server block to save addr info
|
||||||
@@ -104,9 +104,10 @@ class NginxParser(object):
|
|||||||
|
|
||||||
def _get_raw_servers(self):
|
def _get_raw_servers(self):
|
||||||
# pylint: disable=cell-var-from-loop
|
# pylint: disable=cell-var-from-loop
|
||||||
|
# type: () -> Dict
|
||||||
"""Get a map of unparsed all server blocks
|
"""Get a map of unparsed all server blocks
|
||||||
"""
|
"""
|
||||||
servers = {}
|
servers = {} # type: Dict[str, Union[List, nginxparser.UnspacedList]]
|
||||||
for filename in self.parsed:
|
for filename in self.parsed:
|
||||||
tree = self.parsed[filename]
|
tree = self.parsed[filename]
|
||||||
servers[filename] = []
|
servers[filename] = []
|
||||||
@@ -727,9 +728,9 @@ def _parse_server_raw(server):
|
|||||||
:rtype: dict
|
:rtype: dict
|
||||||
|
|
||||||
"""
|
"""
|
||||||
parsed_server = {'addrs': set(),
|
addrs = set() # type: Set[obj.Addr]
|
||||||
'ssl': False,
|
ssl = False # type: bool
|
||||||
'names': set()}
|
names = set() # type: Set[str]
|
||||||
|
|
||||||
apply_ssl_to_all_addrs = False
|
apply_ssl_to_all_addrs = False
|
||||||
|
|
||||||
@@ -739,17 +740,21 @@ def _parse_server_raw(server):
|
|||||||
if directive[0] == 'listen':
|
if directive[0] == 'listen':
|
||||||
addr = obj.Addr.fromstring(" ".join(directive[1:]))
|
addr = obj.Addr.fromstring(" ".join(directive[1:]))
|
||||||
if addr:
|
if addr:
|
||||||
parsed_server['addrs'].add(addr)
|
addrs.add(addr)
|
||||||
if addr.ssl:
|
if addr.ssl:
|
||||||
parsed_server['ssl'] = True
|
ssl = True
|
||||||
elif directive[0] == 'server_name':
|
elif directive[0] == 'server_name':
|
||||||
parsed_server['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):
|
||||||
parsed_server['ssl'] = True
|
ssl = True
|
||||||
apply_ssl_to_all_addrs = True
|
apply_ssl_to_all_addrs = True
|
||||||
|
|
||||||
if apply_ssl_to_all_addrs:
|
if apply_ssl_to_all_addrs:
|
||||||
for addr in parsed_server['addrs']:
|
for addr in addrs:
|
||||||
addr.ssl = True
|
addr.ssl = True
|
||||||
|
|
||||||
return parsed_server
|
return {
|
||||||
|
'addrs': addrs,
|
||||||
|
'ssl': ssl,
|
||||||
|
'names': names
|
||||||
|
}
|
||||||
|
|||||||
@@ -11,6 +11,7 @@ from certbot_nginx import nginxparser
|
|||||||
from certbot_nginx import obj
|
from certbot_nginx import obj
|
||||||
from certbot_nginx import parser
|
from certbot_nginx import parser
|
||||||
from certbot_nginx.tests import util
|
from certbot_nginx.tests import util
|
||||||
|
from acme.magic_typing import List # pylint: disable=unused-import, no-name-in-module
|
||||||
|
|
||||||
|
|
||||||
class NginxParserTest(util.NginxTest): #pylint: disable=too-many-public-methods
|
class NginxParserTest(util.NginxTest): #pylint: disable=too-many-public-methods
|
||||||
@@ -99,7 +100,7 @@ class NginxParserTest(util.NginxTest): #pylint: disable=too-many-public-methods
|
|||||||
([[[0], [3], [4]], [[5], [3], [0]]], [])]
|
([[[0], [3], [4]], [[5], [3], [0]]], [])]
|
||||||
|
|
||||||
for mylist, result in mylists:
|
for mylist, result in mylists:
|
||||||
paths = []
|
paths = [] # type: List[List[int]]
|
||||||
parser._do_for_subarray(mylist,
|
parser._do_for_subarray(mylist,
|
||||||
lambda x: isinstance(x, list) and
|
lambda x: isinstance(x, list) and
|
||||||
len(x) >= 1 and
|
len(x) >= 1 and
|
||||||
|
|||||||
Reference in New Issue
Block a user