starred/certbot-certbot
1
0
Fork 0
mirror of https://github.com/certbot/certbot.git synced 2026-07-27 16:30:31 +02:00
Code Issues Packages Projects Releases Wiki Activity

Fully type certbot apache (#9177)

Browse Source 
* Work in progress

* Work in progress

* Work in progress

* Work in progress

* Fix issues around nullability of VirtualHost.path, may discuss that during review

* Work in progress

* Fix remaining types

* Various lint fixes

* Reconfigure tox and mypy to disallow untyped defs globally

* Cleanup compatibility tests

* Use cast for unused v2 logic

* Improve types

* Remove unused comment

* Fix coverage

* Better types

* Fix another type

* Update certbot-apache/certbot_apache/_internal/apacheparser.py

Co-authored-by: alexzorin <alex@zor.io>

* Update certbot-apache/certbot_apache/_internal/assertions.py

Co-authored-by: alexzorin <alex@zor.io>

* Fix type

* Various fixes

* Refactor imports

* Keep naming convention consistent on TypeVars

* Improve types

* Improve types

* Remove remaining Sequence[str] in the project

Co-authored-by: alexzorin <alex@zor.io>
This commit is contained in:
Adrien Ferrand
2022-01-31 19:17:40 +11:00
committed by GitHub
co-authored by alexzorin
parent fb1b105ba2
commit 0181a0b07f
29 changed files with 322 additions and 278 deletions
Download Patch File Download Diff File Expand all files Collapse all files
certbot-apache/certbot_apache/_internal/apache_util.py
+2 -5
View File
@@ -4,12 +4,10 @@ import fnmatch
import logging
import re
import subprocess
from typing import Any
from typing import Dict
from typing import Iterable
from typing import List
from typing import Optional
from typing import Sequence
from typing import Tuple
import pkg_resources
@@ -21,7 +19,7 @@ from certbot.compat import os
logger = logging.getLogger(__name__)
def get_mod_deps(mod_name: str) -> Any:
def get_mod_deps(mod_name: str) -> List[str]:
"""Get known module dependencies.
.. note:: This does not need to be accurate in order for the client to
@@ -135,7 +133,6 @@ def included_in_paths(filepath: str, paths: Iterable[str]) -> bool:
:returns: True if included
:rtype: bool
"""
return any(fnmatch.fnmatch(filepath, path) for path in paths)
@@ -212,7 +209,7 @@ def parse_from_subprocess(command: List[str], regexp: str) -> List[str]:
return re.compile(regexp).findall(stdout)
def _get_runtime_cfg(command: Sequence[str]) -> str:
def _get_runtime_cfg(command: List[str]) -> str:
"""
Get runtime configuration info.
certbot-apache/certbot_apache/_internal/apacheparser.py
+25 -28
View File
@@ -1,5 +1,6 @@
""" apacheconfig implementation of the ParserNode interfaces """
from typing import Any
from typing import Iterable
from typing import List
from typing import Optional
from typing import Tuple
@@ -7,6 +8,7 @@ from typing import Tuple
from certbot_apache._internal import assertions
from certbot_apache._internal import interfaces
from certbot_apache._internal import parsernode_util as util
from certbot_apache._internal.interfaces import ParserNode
class ApacheParserNode(interfaces.ParserNode):
@@ -16,21 +18,20 @@ class ApacheParserNode(interfaces.ParserNode):
by parsing the equivalent configuration text using the apacheconfig library.
"""
def __init__(self, **kwargs: Any):
def __init__(self, **kwargs: Any) -> None:
# pylint: disable=unused-variable
ancestor, dirty, filepath, metadata = util.parsernode_kwargs(kwargs)
super().__init__(**kwargs)
self.ancestor: str = ancestor
self.filepath: str = filepath
self.dirty: bool = dirty
self.metadata: Any = metadata
self.ancestor = ancestor
self.filepath = filepath
self.dirty = dirty
self.metadata = metadata
self._raw: Any = self.metadata["ac_ast"]
def save(self, msg: str) -> None:
pass # pragma: no cover
# pylint: disable=unused-variable
def find_ancestors(self, name: str) -> List["ApacheBlockNode"]:
def find_ancestors(self, name: str) -> List["ApacheParserNode"]: # pylint: disable=unused-variable
"""Find ancestor BlockNodes with a given name"""
return [ApacheBlockNode(name=assertions.PASS,
parameters=assertions.PASS,
@@ -42,12 +43,12 @@ class ApacheParserNode(interfaces.ParserNode):
class ApacheCommentNode(ApacheParserNode):
""" apacheconfig implementation of CommentNode interface """
def __init__(self, **kwargs: Any):
comment, kwargs = util.commentnode_kwargs(kwargs) # pylint: disable=unused-variable
def __init__(self, **kwargs: Any) -> None:
comment, kwargs = util.commentnode_kwargs(kwargs)
super().__init__(**kwargs)
self.comment = comment
def __eq__(self, other: Any):
def __eq__(self, other: Any) -> bool:
if isinstance(other, self.__class__):
return (self.comment == other.comment and
self.dirty == other.dirty and
@@ -60,12 +61,12 @@ class ApacheCommentNode(ApacheParserNode):
class ApacheDirectiveNode(ApacheParserNode):
""" apacheconfig implementation of DirectiveNode interface """
def __init__(self, **kwargs: Any):
def __init__(self, **kwargs: Any) -> None:
name, parameters, enabled, kwargs = util.directivenode_kwargs(kwargs)
super().__init__(**kwargs)
self.name: str = name
self.parameters: List[str] = parameters
self.enabled: bool = enabled
self.name = name
self.parameters = parameters
self.enabled = enabled
self.include: Optional[str] = None
def __eq__(self, other: Any) -> bool:
@@ -79,7 +80,7 @@ class ApacheDirectiveNode(ApacheParserNode):
self.metadata == other.metadata)
return False # pragma: no cover
def set_parameters(self, _parameters):
def set_parameters(self, _parameters: Iterable[str]) -> None:
"""Sets the parameters for DirectiveNode"""
return # pragma: no cover
@@ -87,11 +88,11 @@ class ApacheDirectiveNode(ApacheParserNode):
class ApacheBlockNode(ApacheDirectiveNode):
""" apacheconfig implementation of BlockNode interface """
def __init__(self, **kwargs: Any):
def __init__(self, **kwargs: Any) -> None:
super().__init__(**kwargs)
self.children: Tuple[ApacheParserNode, ...] = ()
def __eq__(self, other):
def __eq__(self, other: Any) -> bool:
if isinstance(other, self.__class__):
return (self.name == other.name and
self.filepath == other.filepath and
@@ -104,9 +105,8 @@ class ApacheBlockNode(ApacheDirectiveNode):
return False # pragma: no cover
# pylint: disable=unused-argument
def add_child_block(
self, name: str, parameters: Optional[str] = None, position: Optional[int] = None
) -> "ApacheBlockNode": # pragma: no cover
def add_child_block(self, name: str, parameters: Optional[List[str]] = None,
position: Optional[int] = None) -> "ApacheBlockNode": # pragma: no cover
"""Adds a new BlockNode to the sequence of children"""
new_block = ApacheBlockNode(name=assertions.PASS,
parameters=assertions.PASS,
@@ -117,9 +117,8 @@ class ApacheBlockNode(ApacheDirectiveNode):
return new_block
# pylint: disable=unused-argument
def add_child_directive(
self, name: str, parameters: Optional[str] = None, position: Optional[int] = None
) -> ApacheDirectiveNode: # pragma: no cover
def add_child_directive(self, name: str, parameters: Optional[List[str]] = None,
position: int = None) -> ApacheDirectiveNode: # pragma: no cover
"""Adds a new DirectiveNode to the sequence of children"""
new_dir = ApacheDirectiveNode(name=assertions.PASS,
parameters=assertions.PASS,
@@ -142,8 +141,7 @@ class ApacheBlockNode(ApacheDirectiveNode):
self.children += (new_comment,)
return new_comment
# pylint: disable=unused-argument
def find_blocks(self, name, exclude: bool = True) -> List["ApacheBlockNode"]:
def find_blocks(self, name: str, exclude: bool = True) -> List["ApacheBlockNode"]: # pylint: disable=unused-argument
"""Recursive search of BlockNodes from the sequence of children"""
return [ApacheBlockNode(name=assertions.PASS,
parameters=assertions.PASS,
@@ -151,8 +149,7 @@ class ApacheBlockNode(ApacheDirectiveNode):
filepath=assertions.PASS,
metadata=self.metadata)]
# pylint: disable=unused-argument
def find_directives(self, name: str, exclude: bool = True) -> List[ApacheDirectiveNode]:
def find_directives(self, name: str, exclude: bool = True) -> List[ApacheDirectiveNode]: # pylint: disable=unused-argument
"""Recursive search of DirectiveNodes from the sequence of children"""
return [ApacheDirectiveNode(name=assertions.PASS,
parameters=assertions.PASS,
@@ -168,7 +165,7 @@ class ApacheBlockNode(ApacheDirectiveNode):
filepath=assertions.PASS,
metadata=self.metadata)]
def delete_child(self, child: "ApacheBlockNode") -> None:
def delete_child(self, child: ParserNode) -> None:
"""Deletes a ParserNode from the sequence of children"""
return # pragma: no cover
certbot-apache/certbot_apache/_internal/assertions.py
+28 -16
View File
@@ -1,13 +1,21 @@
"""Dual parser node assertions"""
import fnmatch
from typing import Any
from typing import Iterable
from typing import List
from typing import Optional
from typing import Union
from certbot_apache._internal import interfaces
from certbot_apache._internal.interfaces import CommentNode
from certbot_apache._internal.interfaces import DirectiveNode
from certbot_apache._internal.interfaces import ParserNode
from certbot_apache._internal.obj import VirtualHost
PASS = "CERTBOT_PASS_ASSERT"
def assertEqual(first: Any, second: Any) -> None:
def assertEqual(first: ParserNode, second: ParserNode) -> None:
""" Equality assertion """
if isinstance(first, interfaces.CommentNode):
@@ -31,23 +39,26 @@ def assertEqual(first: Any, second: Any) -> None:
assert first.filepath == second.filepath
# pragma: no cover
def assertEqualComment(first: Any, second: Any) -> None:
def assertEqualComment(first: ParserNode, second: ParserNode) -> None: # pragma: no cover
""" Equality assertion for CommentNode """
assert isinstance(first, interfaces.CommentNode)
assert isinstance(second, interfaces.CommentNode)
if not isPass(first.comment) and not isPass(second.comment): # type: ignore
assert first.comment == second.comment # type: ignore
if not isPass(first.comment) and not isPass(second.comment):
assert first.comment == second.comment
def _assertEqualDirectiveComponents(first: Any, second: Any) -> None: # pragma: no cover
def _assertEqualDirectiveComponents(first: ParserNode, # pragma: no cover
second: ParserNode) -> None:
""" Handles assertion for instance variables for DirectiveNode and BlockNode"""
# Enabled value cannot be asserted, because Augeas implementation
# is unable to figure that out.
# assert first.enabled == second.enabled
assert isinstance(first, DirectiveNode)
assert isinstance(second, DirectiveNode)
if not isPass(first.name) and not isPass(second.name):
assert first.name == second.name
@@ -55,7 +66,7 @@ def _assertEqualDirectiveComponents(first: Any, second: Any) -> None: # pragma:
assert first.parameters == second.parameters
def assertEqualDirective(first: Any, second: Any) -> None:
def assertEqualDirective(first: ParserNode, second: ParserNode) -> None:
""" Equality assertion for DirectiveNode """
assert isinstance(first, interfaces.DirectiveNode)
@@ -63,14 +74,14 @@ def assertEqualDirective(first: Any, second: Any) -> None:
_assertEqualDirectiveComponents(first, second)
def isPass(value) -> bool: # pragma: no cover
def isPass(value: Any) -> bool: # pragma: no cover
"""Checks if the value is set to PASS"""
if isinstance(value, bool):
return True
return PASS in value
def isPassDirective(block):
def isPassDirective(block: DirectiveNode) -> bool:
""" Checks if BlockNode or DirectiveNode should pass the assertion """
if isPass(block.name):
@@ -82,28 +93,29 @@ def isPassDirective(block):
return False
def isPassComment(comment):
def isPassComment(comment: CommentNode) -> bool:
""" Checks if CommentNode should pass the assertion """
if isPass(comment.comment):
return True
if isPass(comment.filepath): # pragma: no cover
if isPass(comment.filepath): # pragma: no cover
return True
return False
def isPassNodeList(nodelist): # pragma: no cover
def isPassNodeList(nodelist: List[Union[DirectiveNode, CommentNode]]) -> bool: # pragma: no cover
""" Checks if a ParserNode in the nodelist should pass the assertion,
this function is used for results of find_* methods. Unimplemented find_*
methods should return a sequence containing a single ParserNode instance
with assertion pass string."""
node: Optional[Union[DirectiveNode, CommentNode]]
try:
node = nodelist[0]
except IndexError:
node = None
if not node: # pragma: no cover
if not node: # pragma: no cover
return False
if isinstance(node, interfaces.DirectiveNode):
@@ -111,13 +123,13 @@ def isPassNodeList(nodelist): # pragma: no cover
return isPassComment(node)
def assertEqualSimple(first, second):
def assertEqualSimple(first: Any, second: Any) -> None:
""" Simple assertion """
if not isPass(first) and not isPass(second):
assert first == second
def isEqualVirtualHost(first, second) -> bool:
def isEqualVirtualHost(first: VirtualHost, second: VirtualHost) -> bool:
"""
Checks that two VirtualHost objects are similar. There are some built
in differences with the implementations: VirtualHost created by ParserNode
@@ -138,7 +150,7 @@ def isEqualVirtualHost(first, second) -> bool:
)
def assertEqualPathsList(first, second): # pragma: no cover
def assertEqualPathsList(first: Iterable[str], second: Iterable[str]) -> None: # pragma: no cover
"""
Checks that the two lists of file paths match. This assertion allows for wildcard
paths.
certbot-apache/certbot_apache/_internal/augeasparser.py
+31 -29
View File
@@ -74,27 +74,28 @@ from typing import Set
from typing import Tuple
from typing import Union
from certbot import errors
from certbot.compat import os
from certbot_apache._internal import apache_util
from certbot_apache._internal import assertions
from certbot_apache._internal import interfaces
from certbot_apache._internal import parser
from certbot_apache._internal import parsernode_util as util
from certbot import errors
from certbot.compat import os
class AugeasParserNode(interfaces.ParserNode):
""" Augeas implementation of ParserNode interface """
def __init__(self, **kwargs: Any):
def __init__(self, **kwargs: Any) -> None:
# pylint: disable=unused-variable
ancestor, dirty, filepath, metadata = util.parsernode_kwargs(kwargs)
super().__init__(**kwargs)
self.ancestor: str = ancestor
self.filepath: str = filepath
self.dirty: bool = dirty
self.metadata: Dict[str, Any] = metadata
self.parser: parser.ApacheParser = cast(parser.ApacheParser,
self.ancestor = ancestor
self.filepath = filepath
self.dirty = dirty
self.metadata = metadata
self.parser = cast(parser.ApacheParser,
self.metadata.get("augeasparser"))
try:
if self.metadata["augeaspath"].endswith("/"):
@@ -109,17 +110,17 @@ class AugeasParserNode(interfaces.ParserNode):
def save(self, msg: Iterable[str]) -> None:
self.parser.save(msg)
def find_ancestors(self, name: str) -> List["AugeasBlockNode"]:
def find_ancestors(self, name: str) -> List["AugeasParserNode"]:
"""
Searches for ancestor BlockNodes with a given name.
:param str name: Name of the BlockNode parent to search for
:returns: List of matching ancestor nodes.
:rtype: list of AugeasBlockNode
:rtype: list of AugeasParserNode
"""
ancestors: List[AugeasBlockNode] = []
ancestors: List["AugeasParserNode"] = []
parent = self.metadata["augeaspath"]
while True:
@@ -129,7 +130,7 @@ class AugeasParserNode(interfaces.ParserNode):
if not parent or parent == "/files":
break
anc = self._create_blocknode(parent)
if anc.name.lower() == name.lower():
if anc.name is not None and anc.name.lower() == name.lower():
ancestors.append(anc)
return ancestors
@@ -180,7 +181,7 @@ class AugeasParserNode(interfaces.ParserNode):
class AugeasCommentNode(AugeasParserNode):
""" Augeas implementation of CommentNode interface """
def __init__(self, **kwargs: Any):
def __init__(self, **kwargs: Any) -> None:
comment, kwargs = util.commentnode_kwargs(kwargs) # pylint: disable=unused-variable
super().__init__(**kwargs)
self.comment = comment
@@ -198,11 +199,11 @@ class AugeasCommentNode(AugeasParserNode):
class AugeasDirectiveNode(AugeasParserNode):
""" Augeas implementation of DirectiveNode interface """
def __init__(self, **kwargs: Any):
def __init__(self, **kwargs: Any) -> None:
name, parameters, enabled, kwargs = util.directivenode_kwargs(kwargs)
super().__init__(**kwargs)
self.name: str = name
self.enabled: bool = enabled
self.name = name
self.enabled = enabled
if parameters:
self.set_parameters(parameters)
@@ -217,7 +218,7 @@ class AugeasDirectiveNode(AugeasParserNode):
self.metadata == other.metadata)
return False
def set_parameters(self, parameters: List[str]):
def set_parameters(self, parameters: Iterable[str]) -> None:
"""
Sets parameters of a DirectiveNode or BlockNode object.
@@ -236,7 +237,7 @@ class AugeasDirectiveNode(AugeasParserNode):
self.parser.aug.set(param_path, param)
@property
def parameters(self) -> Tuple[Optional[str], ...]:
def parameters(self) -> Tuple[str, ...]:
"""
Fetches the parameters from Augeas tree, ensuring that the sequence always
represents the current state
@@ -246,17 +247,18 @@ class AugeasDirectiveNode(AugeasParserNode):
"""
return tuple(self._aug_get_params(self.metadata["augeaspath"]))
def _aug_get_params(self, path: str) -> List[Optional[str]]:
def _aug_get_params(self, path: str) -> List[str]:
"""Helper function to get parameters for DirectiveNodes and BlockNodes"""
arg_paths = self.parser.aug.match(path + "/arg")
return [self.parser.get_arg(apath) for apath in arg_paths]
args = [self.parser.get_arg(apath) for apath in arg_paths]
return [arg for arg in args if arg is not None]
class AugeasBlockNode(AugeasDirectiveNode):
""" Augeas implementation of BlockNode interface """
def __init__(self, **kwargs: Any):
def __init__(self, **kwargs: Any) -> None:
super().__init__(**kwargs)
self.children: Tuple["AugeasBlockNode", ...] = ()
@@ -273,9 +275,9 @@ class AugeasBlockNode(AugeasDirectiveNode):
return False
# pylint: disable=unused-argument
def add_child_block(
self, name: str, parameters: Optional[str] = None, position: Optional[int] = None
) -> "AugeasBlockNode": # pragma: no cover
def add_child_block(self, name: str, # pragma: no cover
parameters: Optional[List[str]] = None,
position: Optional[int] = None) -> "AugeasBlockNode":
"""Adds a new BlockNode to the sequence of children"""
insertpath, realpath, before = self._aug_resolve_child_position(
@@ -289,7 +291,7 @@ class AugeasBlockNode(AugeasDirectiveNode):
# Check if the file was included from the root config or initial state
file_path = apache_util.get_file_path(realpath)
if file_path is None:
raise errors.Error(f"No file path found for vhost: {realpath}")
raise errors.Error(f"No file path found for vhost: {realpath}") # pragma: no cover
enabled = self.parser.parsed_in_original(file_path)
# Parameters will be set at the initialization of the new object
@@ -303,9 +305,9 @@ class AugeasBlockNode(AugeasDirectiveNode):
)
# pylint: disable=unused-argument
def add_child_directive(
self, name: str, parameters=None, position=None
) -> "AugeasDirectiveNode": # pragma: no cover
def add_child_directive(self, name: str, # pragma: no cover
parameters: Optional[List[str]] = None,
position: Optional[int] = None) -> AugeasDirectiveNode:
"""Adds a new DirectiveNode to the sequence of children"""
if not parameters:
@@ -324,7 +326,7 @@ class AugeasBlockNode(AugeasDirectiveNode):
# Check if the file was included from the root config or initial state
file_path = apache_util.get_file_path(realpath)
if file_path is None:
raise errors.Error(f"No file path found for vhost: {realpath}")
raise errors.Error(f"No file path found for vhost: {realpath}") # pragma: no cover
enabled = self.parser.parsed_in_original(file_path)
return AugeasDirectiveNode(
certbot-apache/certbot_apache/_internal/configurator.py
+40 -32
View File
@@ -1,12 +1,12 @@
"""Apache Configurator."""
# pylint: disable=too-many-lines
from collections import defaultdict
import copy
import fnmatch
import logging
import re
import socket
import time
# pylint: disable=too-many-lines
from collections import defaultdict
from typing import Any
from typing import Callable
from typing import cast
@@ -21,6 +21,16 @@ from typing import Tuple
from typing import Type
from typing import Union
from certbot_apache._internal import apache_util
from certbot_apache._internal import assertions
from certbot_apache._internal import constants
from certbot_apache._internal import display_ops
from certbot_apache._internal import dualparser
from certbot_apache._internal import http_01
from certbot_apache._internal import obj
from certbot_apache._internal import parser
from certbot_apache._internal.apacheparser import ApacheBlockNode
from acme import challenges
from certbot import achallenges
from certbot import errors
@@ -32,14 +42,6 @@ from certbot.interfaces import RenewableCert
from certbot.plugins import common
from certbot.plugins.enhancements import AutoHSTSEnhancement
from certbot.plugins.util import path_surgery
from certbot_apache._internal import apache_util
from certbot_apache._internal import assertions
from certbot_apache._internal import constants
from certbot_apache._internal import display_ops
from certbot_apache._internal import dualparser
from certbot_apache._internal import http_01
from certbot_apache._internal import obj
from certbot_apache._internal import parser
try:
import apacheconfig
@@ -73,7 +75,7 @@ class OsOptions:
handle_sites: bool = False,
challenge_location: str = "/etc/apache2",
apache_bin: Optional[str] = None,
):
) -> None:
self.server_root = server_root
self.vhost_root = vhost_root
self.vhost_files = vhost_files
@@ -185,7 +187,7 @@ class ApacheConfigurator(common.Configurator):
self.options.conftest_cmd[0] = self.options.ctl
@classmethod
def add_parser_arguments(cls, add: Callable):
def add_parser_arguments(cls, add: Callable[..., None]) -> None:
# When adding, modifying or deleting command line arguments, be sure to
# include the changes in the list used in method _prepare_options() to
# ensure consistent behavior.
@@ -223,7 +225,7 @@ class ApacheConfigurator(common.Configurator):
add("bin", default=DEFAULTS.bin,
help="Full path to apache2/httpd binary")
def __init__(self, *args: Any, **kwargs: Any):
def __init__(self, *args: Any, **kwargs: Any) -> None:
"""Initialize an Apache Configurator.
:param tup version: version of Apache as a tuple (2, 4, 7)
@@ -398,7 +400,7 @@ class ApacheConfigurator(common.Configurator):
" Apache configuration?".format(self.options.server_root))
self._prepared = True
def save(self, title: Optional[str] = None, temporary: bool = False):
def save(self, title: Optional[str] = None, temporary: bool = False) -> None:
"""Saves all changes to the configuration files.
This function first checks for save errors, if none are found,
@@ -447,7 +449,7 @@ class ApacheConfigurator(common.Configurator):
self.revert_temporary_config()
self.parser.aug.load()
def rollback_checkpoints(self, rollback: int = 1):
def rollback_checkpoints(self, rollback: int = 1) -> None:
"""Rollback saved checkpoints.
:param int rollback: Number of checkpoints to revert
@@ -560,10 +562,11 @@ class ApacheConfigurator(common.Configurator):
def _generate_no_suitable_vhost_error(self, target_name: str) -> errors.PluginError:
"""
Notifies the user that Certbot could not find a vhost to secure
and raises an error.
Prepare an error to notify the user that Certbot could not find a vhost
to secure.
:param str target_name: The server name that could not be mapped
:raises errors.PluginError: Raised unconditionally
:return: An instance of PluginError
:rtype: errors.PluginError
"""
return errors.PluginError(
"Certbot could not find a VirtualHost for {0} in the Apache "
@@ -743,7 +746,8 @@ class ApacheConfigurator(common.Configurator):
if not vhost.ssl:
addrs = self._get_proposed_addrs(vhost, "443")
# TODO: Conflicts is too conservative
if not any(vhost.enabled and vhost.conflicts(addrs) for vhost in self.vhosts):
if not any(one_vhost.enabled and one_vhost.conflicts(addrs) for
one_vhost in self.vhosts):
vhost = self.make_vhost_ssl(vhost)
else:
logger.error(
@@ -1102,10 +1106,10 @@ class ApacheConfigurator(common.Configurator):
vhs = []
vhosts = self.parser_root.find_blocks("VirtualHost", exclude=False)
for vhblock in vhosts:
vhs.append(self._create_vhost_v2(vhblock))
vhs.append(self._create_vhost_v2(cast(ApacheBlockNode, vhblock)))
return vhs
def _create_vhost_v2(self, node) -> obj.VirtualHost:
def _create_vhost_v2(self, node: ApacheBlockNode) -> obj.VirtualHost:
"""Used by get_virtual_hosts_v2 to create vhost objects using ParserNode
interfaces.
:param ApacheBlockNode node: The BlockNode object of VirtualHost block
@@ -1133,6 +1137,9 @@ class ApacheConfigurator(common.Configurator):
if addr.get_port() == "443":
is_ssl = True
if node.filepath is None:
raise errors.Error("Node filepath cannot be None.") # pragma: no cover
enabled = apache_util.included_in_paths(node.filepath, self.parsed_paths)
macro = False
@@ -1148,11 +1155,13 @@ class ApacheConfigurator(common.Configurator):
self._populate_vhost_names_v2(vhost)
return vhost
def _populate_vhost_names_v2(self, vhost: obj.VirtualHost):
def _populate_vhost_names_v2(self, vhost: obj.VirtualHost) -> None:
"""Helper function that populates the VirtualHost names.
:param host: In progress vhost whose names will be added
:type host: :class:`~certbot_apache.obj.VirtualHost`
"""
if not vhost.node:
raise errors.PluginError("Current VirtualHost has no node.") # pragma: no cover
servername_match = vhost.node.find_directives("ServerName", exclude=False)
serveralias_match = vhost.node.find_directives("ServerAlias", exclude=False)
@@ -1167,7 +1176,7 @@ class ApacheConfigurator(common.Configurator):
vhost.aliases.add(serveralias)
vhost.name = servername
def is_name_vhost(self, target_addr: obj.Addr):
def is_name_vhost(self, target_addr: obj.Addr) -> bool:
"""Returns if vhost is a name based vhost
NameVirtualHost was deprecated in Apache 2.4 as all VirtualHosts are
@@ -1186,9 +1195,9 @@ class ApacheConfigurator(common.Configurator):
# search for NameVirtualHost directive for ip_addr
# note ip_addr can be FQDN although Apache does not recommend it
return (self.version >= (2, 4) or
self.parser.find_dir("NameVirtualHost", str(target_addr)))
bool(self.parser.find_dir("NameVirtualHost", str(target_addr))))
def add_name_vhost(self, addr: obj.Addr):
def add_name_vhost(self, addr: obj.Addr) -> None:
"""Adds NameVirtualHost directive for given address.
:param addr: Address that will be added as NameVirtualHost directive
@@ -1295,7 +1304,7 @@ class ApacheConfigurator(common.Configurator):
self.save_notes += (f"Added Listen {listen} directive to "
f"{self.parser.loc['listen']}\n")
def _add_listens_https(self, listens: Set[str], listens_orig: List[str], port: str):
def _add_listens_https(self, listens: Set[str], listens_orig: List[str], port: str) -> None:
"""Helper method for ensure_listen to figure out which new
listen statements need adding for listening HTTPS on port
@@ -1681,7 +1690,7 @@ class ApacheConfigurator(common.Configurator):
return ssl_addrs
def _clean_vhost(self, vhost: obj.VirtualHost):
def _clean_vhost(self, vhost: obj.VirtualHost) -> None:
# remove duplicated or conflicting ssl directives
self._deduplicate_directives(vhost.path,
["SSLCertificateFile",
@@ -1689,7 +1698,7 @@ class ApacheConfigurator(common.Configurator):
# remove all problematic directives
self._remove_directives(vhost.path, ["SSLCertificateChainFile"])
def _deduplicate_directives(self, vh_path: Optional[str], directives: List[str]):
def _deduplicate_directives(self, vh_path: Optional[str], directives: List[str]) -> None:
for directive in directives:
while len(self.parser.find_dir(directive, None,
vh_path, False)) > 1:
@@ -1697,7 +1706,7 @@ class ApacheConfigurator(common.Configurator):
vh_path, False)
self.parser.aug.remove(re.sub(r"/\w*$", "", directive_path[0]))
def _remove_directives(self, vh_path: Optional[str], directives: List[str]):
def _remove_directives(self, vh_path: Optional[str], directives: List[str]) -> None:
for directive in directives:
while self.parser.find_dir(directive, None, vh_path, False):
directive_path = self.parser.find_dir(directive, None,
@@ -1815,8 +1824,7 @@ class ApacheConfigurator(common.Configurator):
if id_comment:
# Use the first value, multiple ones shouldn't exist
comment = self.parser.get_arg(id_comment[0])
if comment is not None:
return comment.split(" ")[-1]
return comment.split(" ")[-1] if comment else None
return None
def add_vhost_id(self, vhost: obj.VirtualHost) -> Optional[str]:
@@ -2031,7 +2039,7 @@ class ApacheConfigurator(common.Configurator):
self.save()
logger.info(msg)
def _set_http_header(self, ssl_vhost: obj.VirtualHost, header_substring: str):
def _set_http_header(self, ssl_vhost: obj.VirtualHost, header_substring: str) -> None:
"""Enables header that is identified by header_substring on ssl_vhost.
If the header identified by header_substring is not already set,
certbot-apache/certbot_apache/_internal/constants.py
+2 -1
View File
@@ -1,5 +1,6 @@
"""Apache plugin constants."""
from typing import List, Dict
from typing import Dict
from typing import List
import pkg_resources
certbot-apache/certbot_apache/_internal/display_ops.py
+12 -9
View File
@@ -3,24 +3,26 @@ import logging
from typing import Iterable
from typing import List
from typing import Optional
from typing import Sequence
from typing import Tuple
from certbot_apache._internal.obj import VirtualHost
from certbot import errors
from certbot.compat import os
from certbot.display import util as display_util
from certbot_apache._internal import obj
logger = logging.getLogger(__name__)
def select_vhost_multiple(vhosts: Optional[List[obj.VirtualHost]]) -> List[obj.VirtualHost]:
def select_vhost_multiple(vhosts: Optional[List[VirtualHost]]) -> List[VirtualHost]:
"""Select multiple Vhosts to install the certificate for
:param vhosts: Available Apache VirtualHosts
:type vhosts: :class:`list` of type `~obj.VirtualHost`
:type vhosts: :class:`list` of type `~VirtualHost`
:returns: List of VirtualHosts
:rtype: :class:`list`of type `~obj.Vhost`
:rtype: :class:`list`of type `~VirtualHost`
"""
if not vhosts:
return []
@@ -37,7 +39,7 @@ def select_vhost_multiple(vhosts: Optional[List[obj.VirtualHost]]) -> List[obj.V
return []
def _reversemap_vhosts(names: Iterable[str], vhosts: List[obj.VirtualHost]):
def _reversemap_vhosts(names: Iterable[str], vhosts: Iterable[VirtualHost]) -> List[VirtualHost]:
"""Helper function for select_vhost_multiple for mapping string
representations back to actual vhost objects"""
return_vhosts = []
@@ -49,12 +51,13 @@ def _reversemap_vhosts(names: Iterable[str], vhosts: List[obj.VirtualHost]):
return return_vhosts
def select_vhost(domain: str, vhosts: List[obj.VirtualHost]) -> Optional[obj.VirtualHost]:
def select_vhost(domain: str, vhosts: Sequence[VirtualHost]) -> Optional[VirtualHost]:
"""Select an appropriate Apache Vhost.
:param domain: Domain to select
:param str domain: Domain for vhost selection
:param vhosts: Available Apache VirtualHosts
:type vhosts: :class:`list` of type `~obj.Vhost`
:type vhosts: :class:`list` of type `~VirtualHost`
:returns: VirtualHost or `None`
:rtype: `~obj.Vhost` or `None`
@@ -68,7 +71,7 @@ def select_vhost(domain: str, vhosts: List[obj.VirtualHost]) -> Optional[obj.Vir
return None
def _vhost_menu(domain: str, vhosts: List[obj.VirtualHost]) -> Tuple[str, int]:
def _vhost_menu(domain: str, vhosts: Iterable[VirtualHost]) -> Tuple[str, int]:
"""Select an appropriate Apache Vhost.
:param vhosts: Available Apache Virtual Hosts
certbot-apache/certbot_apache/_internal/dualparser.py
+57 -50
View File
@@ -1,24 +1,40 @@
""" Dual ParserNode implementation """
from typing import Any
from typing import Callable
from typing import Generic
from typing import Iterable
from typing import List
from typing import Optional
from typing import Sequence
from typing import Set
from typing import Tuple
from typing import Type
from typing import TYPE_CHECKING
from typing import TypeVar
from certbot_apache._internal import apacheparser
from certbot_apache._internal import assertions
from certbot_apache._internal import augeasparser
from certbot_apache._internal import interfaces
if TYPE_CHECKING:
from certbot_apache._internal.apacheparser import ApacheParserNode # pragma: no cover
from certbot_apache._internal.augeasparser import AugeasParserNode # pragma: no cover
class DualNodeBase:
GenericAugeasParserNode = TypeVar("GenericAugeasParserNode", bound="AugeasParserNode")
GenericApacheParserNode = TypeVar("GenericApacheParserNode", bound="ApacheParserNode")
GenericDualNode = TypeVar("GenericDualNode", bound="DualNodeBase")
class DualNodeBase(Generic[GenericAugeasParserNode, GenericApacheParserNode]):
""" Dual parser interface for in development testing. This is used as the
base class for dual parser interface classes. This class handles runtime
attribute value assertions."""
def save(self, msg: str): # pragma: no cover
def __init__(self, primary: GenericAugeasParserNode,
secondary: GenericApacheParserNode) -> None:
self.primary = primary
self.secondary = secondary
def save(self, msg: str) -> None: # pragma: no cover
""" Call save for both parsers """
self.primary.save(msg)
self.secondary.save(msg)
@@ -37,13 +53,12 @@ class DualNodeBase:
assertions.assertEqualSimple(firstval, secondval)
return firstval
def find_ancestors(self, name: str) -> Sequence[interfaces.ParserNode]:
def find_ancestors(self, name: str) -> List["DualNodeBase"]:
""" Traverses the ancestor tree and returns ancestors matching name """
return self._find_helper(DualBlockNode, "find_ancestors", name)
def _find_helper(
self, nodeclass: Callable, findfunc: str, search: str, **kwargs: Any
) -> List[apacheparser.ApacheBlockNode]:
def _find_helper(self, nodeclass: Type[GenericDualNode], findfunc: str, search: str,
**kwargs: Any) -> List[GenericDualNode]:
"""A helper for find_* functions. The function specific attributes should
be passed as keyword arguments.
@@ -83,10 +98,11 @@ class DualNodeBase:
return new_nodes
class DualCommentNode(DualNodeBase):
class DualCommentNode(DualNodeBase[augeasparser.AugeasCommentNode,
apacheparser.ApacheCommentNode]):
""" Dual parser implementation of CommentNode interface """
def __init__(self, **kwargs: Any):
def __init__(self, **kwargs: Any) -> None:
""" This initialization implementation allows ordinary initialization
of CommentNode objects as well as creating a DualCommentNode object
using precreated or fetched CommentNode objects if provided as optional
@@ -106,19 +122,21 @@ class DualCommentNode(DualNodeBase):
if primary or secondary:
assert primary and secondary
self.primary = primary
self.secondary = secondary
super().__init__(primary, secondary)
else:
self.primary = augeasparser.AugeasCommentNode(**kwargs)
self.secondary = apacheparser.ApacheCommentNode(**kwargs)
super().__init__(augeasparser.AugeasCommentNode(**kwargs),
apacheparser.ApacheCommentNode(**kwargs))
assertions.assertEqual(self.primary, self.secondary)
class DualDirectiveNode(DualNodeBase):
class DualDirectiveNode(DualNodeBase[augeasparser.AugeasDirectiveNode,
apacheparser.ApacheDirectiveNode]):
""" Dual parser implementation of DirectiveNode interface """
def __init__(self, **kwargs: Any):
parameters: str
def __init__(self, **kwargs: Any) -> None:
""" This initialization implementation allows ordinary initialization
of DirectiveNode objects as well as creating a DualDirectiveNode object
using precreated or fetched DirectiveNode objects if provided as optional
@@ -138,19 +156,14 @@ class DualDirectiveNode(DualNodeBase):
if primary or secondary:
assert primary and secondary
self.primary = primary
self.secondary = secondary
super().__init__(primary, secondary)
else:
self.primary = augeasparser.AugeasDirectiveNode(
**kwargs
)
self.secondary = apacheparser.ApacheDirectiveNode(
**kwargs
)
super().__init__(augeasparser.AugeasDirectiveNode(**kwargs),
apacheparser.ApacheDirectiveNode(**kwargs))
assertions.assertEqual(self.primary, self.secondary)
def set_parameters(self, parameters):
def set_parameters(self, parameters: Iterable[str]) -> None:
""" Sets parameters and asserts that both implementation successfully
set the parameter sequence """
@@ -159,10 +172,11 @@ class DualDirectiveNode(DualNodeBase):
assertions.assertEqual(self.primary, self.secondary)
class DualBlockNode(DualNodeBase):
class DualBlockNode(DualNodeBase[augeasparser.AugeasBlockNode,
apacheparser.ApacheBlockNode]):
""" Dual parser implementation of BlockNode interface """
def __init__(self, **kwargs: Any):
def __init__(self, **kwargs: Any) -> None:
""" This initialization implementation allows ordinary initialization
of BlockNode objects as well as creating a DualBlockNode object
using precreated or fetched BlockNode objects if provided as optional
@@ -182,17 +196,15 @@ class DualBlockNode(DualNodeBase):
if primary or secondary:
assert primary and secondary
self.primary = primary
self.secondary = secondary
super().__init__(primary, secondary)
else:
self.primary = augeasparser.AugeasBlockNode(**kwargs)
self.secondary = apacheparser.ApacheBlockNode(**kwargs)
super().__init__(augeasparser.AugeasBlockNode(**kwargs),
apacheparser.ApacheBlockNode(**kwargs))
assertions.assertEqual(self.primary, self.secondary)
def add_child_block(
self, name: str, parameters: Optional[str] = None, position: Optional[int] = None
) -> "DualBlockNode":
def add_child_block(self, name: str, parameters: Optional[List[str]] = None,
position: Optional[int] = None) -> "DualBlockNode":
""" Creates a new child BlockNode, asserts that both implementations
did it in a similar way, and returns a newly created DualBlockNode object
encapsulating both of the newly created objects """
@@ -202,9 +214,8 @@ class DualBlockNode(DualNodeBase):
assertions.assertEqual(primary_new, secondary_new)
return DualBlockNode(primary=primary_new, secondary=secondary_new)
def add_child_directive(
self, name: str, parameters: Optional[str] = None, position: Optional[int] = None
) -> DualDirectiveNode:
def add_child_directive(self, name: str, parameters: Optional[List[str]] = None,
position: Optional[int] = None) -> DualDirectiveNode:
""" Creates a new child DirectiveNode, asserts that both implementations
did it in a similar way, and returns a newly created DualDirectiveNode
object encapsulating both of the newly created objects """
@@ -214,9 +225,8 @@ class DualBlockNode(DualNodeBase):
assertions.assertEqual(primary_new, secondary_new)
return DualDirectiveNode(primary=primary_new, secondary=secondary_new)
def add_child_comment(
self, comment: str = "", position: Optional[int] = None
) -> DualCommentNode:
def add_child_comment(self, comment: str = "",
position: Optional[int] = None) -> DualCommentNode:
""" Creates a new child CommentNode, asserts that both implementations
did it in a similar way, and returns a newly created DualCommentNode
object encapsulating both of the newly created objects """
@@ -226,11 +236,9 @@ class DualBlockNode(DualNodeBase):
assertions.assertEqual(primary_new, secondary_new)
return DualCommentNode(primary=primary_new, secondary=secondary_new)
def _create_matching_list(
self,
primary_list: List[interfaces.ParserNode],
secondary_list: List[interfaces.ParserNode],
) -> List[Tuple[interfaces.ParserNode, interfaces.ParserNode]]:
def _create_matching_list(self, primary_list: Iterable[interfaces.ParserNode],
secondary_list: Iterable[interfaces.ParserNode]
) -> List[Tuple[interfaces.ParserNode, interfaces.ParserNode]]:
""" Matches the list of primary_list to a list of secondary_list and
returns a list of tuples. This is used to create results for find_
methods.
@@ -257,7 +265,7 @@ class DualBlockNode(DualNodeBase):
raise AssertionError("Could not find a matching node.")
return matched
def find_blocks(self, name: str, exclude: bool = True) -> List[apacheparser.ApacheBlockNode]:
def find_blocks(self, name: str, exclude: bool = True) -> List["DualBlockNode"]:
"""
Performs a search for BlockNodes using both implementations and does simple
checks for results. This is built upon the assumption that unimplemented
@@ -269,8 +277,7 @@ class DualBlockNode(DualNodeBase):
return self._find_helper(DualBlockNode, "find_blocks", name,
exclude=exclude)
def find_directives(self, name: str, exclude: bool = True
) -> Sequence[apacheparser.ApacheDirectiveNode]:
def find_directives(self, name: str, exclude: bool = True) -> List[DualDirectiveNode]:
"""
Performs a search for DirectiveNodes using both implementations and
checks the results. This is built upon the assumption that unimplemented
@@ -282,7 +289,7 @@ class DualBlockNode(DualNodeBase):
return self._find_helper(DualDirectiveNode, "find_directives", name,
exclude=exclude)
def find_comments(self, comment: str) -> Sequence[apacheparser.ApacheParserNode]:
def find_comments(self, comment: str) -> List[DualCommentNode]:
"""
Performs a search for CommentNodes using both implementations and
checks the results. This is built upon the assumption that unimplemented
@@ -293,7 +300,7 @@ class DualBlockNode(DualNodeBase):
return self._find_helper(DualCommentNode, "find_comments", comment)
def delete_child(self, child: "DualBlockNode"):
def delete_child(self, child: "DualBlockNode") -> None:
"""Deletes a child from the ParserNode implementations. The actual
ParserNode implementations are used here directly in order to be able
to match a child to the list of children."""
certbot-apache/certbot_apache/_internal/entrypoint.py
+5 -4
View File
@@ -1,8 +1,7 @@
""" Entry point for Apache Plugin """
from typing import Callable
from typing import Dict
from typing import Type
from certbot import util
from certbot_apache._internal import configurator
from certbot_apache._internal import override_arch
from certbot_apache._internal import override_centos
@@ -13,7 +12,9 @@ from certbot_apache._internal import override_gentoo
from certbot_apache._internal import override_suse
from certbot_apache._internal import override_void
OVERRIDE_CLASSES: Dict[str, Callable] = {
from certbot import util
OVERRIDE_CLASSES: Dict[str, Type[configurator.ApacheConfigurator]] = {
"arch": override_arch.ArchConfigurator,
"cloudlinux": override_centos.CentOSConfigurator,
"darwin": override_darwin.DarwinConfigurator,
@@ -41,7 +42,7 @@ OVERRIDE_CLASSES: Dict[str, Callable] = {
}
def get_configurator():
def get_configurator() -> Type[configurator.ApacheConfigurator]:
""" Get correct configurator class based on the OS fingerprint """
os_name, os_version = util.get_os_info()
os_name = os_name.lower()
certbot-apache/certbot_apache/_internal/http_01.py
+3 -2
View File
@@ -5,14 +5,15 @@ from typing import List
from typing import Set
from typing import TYPE_CHECKING
from certbot_apache._internal.obj import VirtualHost
from certbot_apache._internal.parser import get_aug_path
from acme.challenges import KeyAuthorizationChallengeResponse
from certbot import errors
from certbot.achallenges import KeyAuthorizationAnnotatedChallenge
from certbot.compat import filesystem
from certbot.compat import os
from certbot.plugins import common
from certbot_apache._internal.obj import VirtualHost
from certbot_apache._internal.parser import get_aug_path
if TYPE_CHECKING:
from certbot_apache._internal.configurator import ApacheConfigurator # pragma: no cover
certbot-apache/certbot_apache/_internal/interfaces.py
+40 -29
View File
@@ -98,11 +98,15 @@ names and parameters in case insensitive manner. This does not apply to everythi
however, for example the parameters of a conditional statement may be case sensitive.
For this reason the internal representation of data should not ignore the case.
"""
import abc
from typing import Any
from typing import Dict
from typing import List
from typing import Optional
from typing import Tuple
from typing import TypeVar
GenericParserNode = TypeVar("GenericParserNode", bound="ParserNode")
class ParserNode(metaclass=abc.ABCMeta):
@@ -147,9 +151,13 @@ class ParserNode(metaclass=abc.ABCMeta):
# for the ParserNode instance.
metadata: Dict[str, Any]
"""
ancestor: Optional["ParserNode"]
dirty: bool
filepath: Optional[str]
metadata: Dict[str, Any]
@abc.abstractmethod
def __init__(self, **kwargs: Any):
def __init__(self, **kwargs: Any) -> None:
"""
Initializes the ParserNode instance, and sets the ParserNode specific
instance variables. This is not meant to be used directly, but through
@@ -192,7 +200,7 @@ class ParserNode(metaclass=abc.ABCMeta):
"""
@abc.abstractmethod
def find_ancestors(self, name: str):
def find_ancestors(self: GenericParserNode, name: str) -> List[GenericParserNode]:
"""
Traverses the ancestor tree up, searching for BlockNodes with a specific
name.
@@ -221,9 +229,10 @@ class CommentNode(ParserNode, metaclass=abc.ABCMeta):
comment: str
"""
comment: str
@abc.abstractmethod
def __init__(self, **kwargs: Any):
def __init__(self, **kwargs: Any) -> None:
"""
Initializes the CommentNode instance and sets its instance variables.
@@ -275,6 +284,9 @@ class DirectiveNode(ParserNode, metaclass=abc.ABCMeta):
parameters: Tuple[str, ...]
"""
enabled: bool
name: Optional[str]
parameters: Tuple[str, ...]
@abc.abstractmethod
def __init__(self, **kwargs: Any) -> None:
@@ -366,11 +378,11 @@ class BlockNode(DirectiveNode, metaclass=abc.ABCMeta):
children: Tuple[ParserNode, ...]
"""
children: Tuple[ParserNode, ...]
@abc.abstractmethod
def add_child_block(
self, name: str, parameters: List[str] = None, position: int = None
) -> "BlockNode":
def add_child_block(self, name: str, parameters: Optional[List[str]] = None,
position: Optional[int] = None) -> "BlockNode":
"""
Adds a new BlockNode child node with provided values and marks the callee
BlockNode dirty. This is used to add new children to the AST. The preceding
@@ -390,9 +402,8 @@ class BlockNode(DirectiveNode, metaclass=abc.ABCMeta):
"""
@abc.abstractmethod
def add_child_directive(
self, name: str, parameters: Optional[List[str]] = None, position: Optional[int] = None
) -> "DirectiveNode":
def add_child_directive(self, name: str, parameters: Optional[List[str]] = None,
position: Optional[int] = None) -> DirectiveNode:
"""
Adds a new DirectiveNode child node with provided values and marks the
callee BlockNode dirty. This is used to add new children to the AST. The
@@ -413,7 +424,7 @@ class BlockNode(DirectiveNode, metaclass=abc.ABCMeta):
"""
@abc.abstractmethod
def add_child_comment(self, comment: str = "", position: Optional[int] = None) -> "CommentNode":
def add_child_comment(self, comment: str = "", position: Optional[int] = None) -> CommentNode:
"""
Adds a new CommentNode child node with provided value and marks the
callee BlockNode dirty. This is used to add new children to the AST. The
@@ -450,23 +461,7 @@ class BlockNode(DirectiveNode, metaclass=abc.ABCMeta):
"""
@abc.abstractmethod
def find_comments(self, comment: str) -> List["CommentNode"]:
"""
Find comments with value containing the search term.
This method walks the child tree of ParserNodes under the instance it was
called from. This way it is possible to search for the whole configuration
tree, when starting from root node, or to do a partial search when starting
from a specified branch. The lookup should be case sensitive.
:param str comment: The content of comment to search for
:returns: A list of found CommentNode objects.
"""
@abc.abstractmethod
def find_directives(self, name: str, exclude: bool = True):
def find_directives(self, name: str, exclude: bool = True) -> List[DirectiveNode]:
"""
Find a directive by name. This method walks the child tree of ParserNodes
under the instance it was called from. This way it is possible to search
@@ -484,7 +479,23 @@ class BlockNode(DirectiveNode, metaclass=abc.ABCMeta):
"""
@abc.abstractmethod
def delete_child(self, child: "ParserNode") -> None:
def find_comments(self, comment: str) -> List[CommentNode]:
"""
Find comments with value containing the search term.
This method walks the child tree of ParserNodes under the instance it was
called from. This way it is possible to search for the whole configuration
tree, when starting from root node, or to do a partial search when starting
from a specified branch. The lookup should be case sensitive.
:param str comment: The content of comment to search for
:returns: A list of found CommentNode objects.
"""
@abc.abstractmethod
def delete_child(self, child: ParserNode) -> None:
"""
Remove a specified child node from the list of children of the called
BlockNode object.
certbot-apache/certbot_apache/_internal/obj.py
+16 -12
View File
@@ -5,15 +5,19 @@ from typing import Iterable
from typing import Optional
from typing import Pattern
from typing import Set
from typing import Union
from certbot_apache._internal.apacheparser import ApacheBlockNode
from certbot_apache._internal.augeasparser import AugeasBlockNode
from certbot_apache._internal.dualparser import DualBlockNode
from certbot.plugins import common
from certbot_apache._internal import interfaces
class Addr(common.Addr):
"""Represents an Apache address."""
def __eq__(self, other: Any):
def __eq__(self, other: Any) -> bool:
"""This is defined as equivalent within Apache.
ip_addr:* == ip_addr
@@ -25,10 +29,10 @@ class Addr(common.Addr):
self.is_wildcard() and other.is_wildcard()))
return False
def __repr__(self):
def __repr__(self) -> str:
return f"certbot_apache._internal.obj.Addr({repr(self.tup)})"
def __hash__(self): # pylint: disable=useless-super-delegation
def __hash__(self) -> int: # pylint: disable=useless-super-delegation
# Python 3 requires explicit overridden for __hash__ if __eq__ or
# __cmp__ is overridden. See https://bugs.python.org/issue2235
return super().__hash__()
@@ -125,11 +129,11 @@ class VirtualHost:
# ?: is used for not returning enclosed characters
strip_name: Pattern = re.compile(r"^(?:.+://)?([^ :$]*)")
def __init__(
self, filepath: str, path: str, addrs: Set["Addr"],
ssl: bool, enabled: bool, name: Optional[str] = None,
aliases: Optional[Set[str]] = None, modmacro: bool = False,
ancestor: Optional["VirtualHost"] = None, node = None):
def __init__(self, filepath: str, path: str, addrs: Set["Addr"], ssl: bool,
enabled: bool, name: Optional[str] = None, aliases: Optional[Set[str]] = None,
modmacro: bool = False, ancestor: Optional["VirtualHost"] = None,
node: Optional[Union[ApacheBlockNode, AugeasBlockNode, DualBlockNode]] = None
) -> None:
"""Initialize a VH."""
self.filep = filepath
@@ -141,7 +145,7 @@ class VirtualHost:
self.enabled = enabled
self.modmacro = modmacro
self.ancestor = ancestor
self.node: interfaces.BlockNode = node
self.node = node
def get_names(self) -> Set[str]:
"""Return a set of all names."""
@@ -153,7 +157,7 @@ class VirtualHost:
return all_names
def __str__(self):
def __str__(self) -> str:
return (
f"File: {self.filep}\n"
f"Vhost path: {self.path}\n"
@@ -185,7 +189,7 @@ class VirtualHost:
return False
def __hash__(self):
def __hash__(self) -> int:
return hash((self.filep, self.path,
tuple(self.addrs), tuple(self.get_names()),
self.ssl, self.enabled, self.modmacro))
certbot-apache/certbot_apache/_internal/override_centos.py
+5 -4
View File
@@ -4,14 +4,15 @@ from typing import Any
from typing import cast
from typing import List
from certbot import errors
from certbot import util
from certbot.errors import MisconfigurationError
from certbot_apache._internal import apache_util
from certbot_apache._internal import configurator
from certbot_apache._internal import parser
from certbot_apache._internal.configurator import OsOptions
from certbot import errors
from certbot import util
from certbot.errors import MisconfigurationError
logger = logging.getLogger(__name__)
@@ -78,7 +79,7 @@ class CentOSConfigurator(configurator.ApacheConfigurator):
return CentOSParser(
self.options.server_root, self, self.options.vhost_root, self.version)
def _deploy_cert(self, *args: Any, **kwargs: Any): # pylint: disable=arguments-differ
def _deploy_cert(self, *args: Any, **kwargs: Any) -> None: # pylint: disable=arguments-differ
"""
Override _deploy_cert in order to ensure that the Apache configuration
has "LoadModule ssl_module..." before parsing the VirtualHost configuration
certbot-apache/certbot_apache/_internal/override_debian.py
+5 -4
View File
@@ -1,14 +1,15 @@
""" Distribution specific override class for Debian family (Ubuntu/Debian) """
import logging
from certbot_apache._internal import apache_util
from certbot_apache._internal import configurator
from certbot_apache._internal.configurator import OsOptions
from certbot_apache._internal.obj import VirtualHost
from certbot import errors
from certbot import util
from certbot.compat import filesystem
from certbot.compat import os
from certbot_apache._internal import apache_util
from certbot_apache._internal import configurator
from certbot_apache._internal.configurator import OsOptions
from certbot_apache._internal.obj import VirtualHost
logger = logging.getLogger(__name__)
certbot-apache/certbot_apache/_internal/override_fedora.py
+6 -3
View File
@@ -1,11 +1,14 @@
""" Distribution specific override class for Fedora 29+ """
from certbot import errors
from certbot import util
from typing import Any
from certbot_apache._internal import apache_util
from certbot_apache._internal import configurator
from certbot_apache._internal import parser
from certbot_apache._internal.configurator import OsOptions
from certbot import errors
from certbot import util
class FedoraConfigurator(configurator.ApacheConfigurator):
"""Fedora 29+ specific ApacheConfigurator override class"""
@@ -68,7 +71,7 @@ class FedoraConfigurator(configurator.ApacheConfigurator):
class FedoraParser(parser.ApacheParser):
"""Fedora 29+ specific ApacheParser override class"""
def __init__(self, *args, **kwargs) -> None:
def __init__(self, *args: Any, **kwargs: Any) -> None:
# Fedora 29+ specific configuration file for Apache
self.sysconfig_filep = "/etc/sysconfig/httpd"
super().__init__(*args, **kwargs)
certbot-apache/certbot_apache/_internal/parser.py
+12 -9
View File
@@ -11,14 +11,15 @@ from typing import Mapping
from typing import Optional
from typing import Pattern
from typing import Set
from typing import TYPE_CHECKING
from typing import Tuple
from typing import TYPE_CHECKING
from typing import Union
from certbot_apache._internal import apache_util
from certbot_apache._internal import constants
from certbot import errors
from certbot.compat import os
from certbot_apache._internal import apache_util
from certbot_apache._internal import constants
if TYPE_CHECKING:
from certbot_apache._internal.configurator import ApacheConfigurator # pragma: no cover
@@ -125,7 +126,7 @@ class ApacheParser:
self.aug.get(path + "/message")))
raise errors.PluginError(msg)
def check_aug_version(self) -> bool:
def check_aug_version(self) -> Union[bool, List[str]]:
""" Checks that we have recent enough version of libaugeas.
If augeas version is recent enough, it will support case insensitive
regexp matching"""
@@ -214,7 +215,7 @@ class ApacheParser:
self.aug.remove("/files/"+sf)
self.aug.load()
def _log_save_errors(self, ex_errs: List[str]) -> None:
def _log_save_errors(self, ex_errs: Iterable[str]) -> None:
"""Log errors due to bad Augeas save.
:param list ex_errs: Existing errors before save
@@ -364,7 +365,7 @@ class ApacheParser:
:param str aug_conf_path: Desired Augeas config path to add directive
:param str directive: Directive you would like to add, e.g. Listen
:param args: Values of the directive; str "443" or list of str
:param args: Values of the directive; list of str (eg. ["443"])
:type args: list
"""
@@ -431,7 +432,7 @@ class ApacheParser:
return retpath
def add_dir(
self, aug_conf_path: str, directive: Optional[str], args: Union[List[str], str]
self, aug_conf_path: Optional[str], directive: Optional[str], args: Union[List[str], str]
) -> None:
"""Appends directive to the end fo the file given by aug_conf_path.
@@ -444,6 +445,7 @@ class ApacheParser:
:type args: list or str
"""
aug_conf_path = aug_conf_path if aug_conf_path else ""
self.aug.set(aug_conf_path + "/directive[last() + 1]", directive)
if isinstance(args, list):
for i, value in enumerate(args, 1):
@@ -452,7 +454,7 @@ class ApacheParser:
else:
self.aug.set(aug_conf_path + "/directive[last()]/arg", args)
def add_dir_beginning(self, aug_conf_path: str, dirname: str,
def add_dir_beginning(self, aug_conf_path: Optional[str], dirname: str,
args: Union[List[str], str]) -> None:
"""Adds the directive to the beginning of defined aug_conf_path.
@@ -461,6 +463,7 @@ class ApacheParser:
:param args: Value of the directive. ie. Listen 443, 443 is arg
:type args: list or str
"""
aug_conf_path = aug_conf_path if aug_conf_path else ""
first_dir = aug_conf_path + "/directive[1]"
if self.aug.get(first_dir):
self.aug.insert(first_dir, "directive", True)
@@ -598,7 +601,7 @@ class ApacheParser:
allargs = self.aug.match(match + '*')
return [self.get_arg(arg) for arg in allargs]
def get_arg(self, match: Optional[str]) -> Optional[str]:
def get_arg(self, match: str) -> Optional[str]:
"""Uses augeas.get to get argument value and interprets result.
This also converts all variables and parameters appropriately.
certbot-apache/certbot_apache/_internal/parsernode_util.py
+9 -5
View File
@@ -1,12 +1,14 @@
"""ParserNode utils"""
from typing import Dict
from typing import Any
from typing import List
from typing import Dict
from typing import Iterable
from typing import Optional
from typing import Tuple
from certbot_apache._internal.interfaces import ParserNode
def validate_kwargs(kwargs: Dict[str, Any], required_names: List[str]) -> Dict[str, Any]:
def validate_kwargs(kwargs: Dict[str, Any], required_names: Iterable[str]) -> Dict[str, Any]:
"""
Ensures that the kwargs dict has all the expected values. This function modifies
the kwargs dictionary, and hence the returned dictionary should be used instead
@@ -30,7 +32,8 @@ def validate_kwargs(kwargs: Dict[str, Any], required_names: List[str]) -> Dict[s
return validated_kwargs
def parsernode_kwargs(kwargs: Dict[str, Any]) -> Tuple[Any, ...]:
def parsernode_kwargs(kwargs: Dict[str, Any]
) -> Tuple[Optional[ParserNode], bool, Optional[str], Dict[str, Any]]:
"""
Validates keyword arguments for ParserNode. This function modifies the kwargs
dictionary, and hence the returned dictionary should be used instead in the
@@ -95,7 +98,8 @@ def commentnode_kwargs(kwargs: Dict[str, Any]) -> Tuple[Optional[str], Dict[str,
return comment, kwargs
def directivenode_kwargs(kwargs: Dict[str, Any]) -> Tuple[Any, Any, Any, Dict]:
def directivenode_kwargs(kwargs: Dict[str, Any]
) -> Tuple[Optional[str], Tuple[str, ...], bool, Dict[str, Any]]:
"""
Validates keyword arguments for DirectiveNode and BlockNode and sets the
default values for optional kwargs. This function modifies the kwargs
certbot-apache/tests/configurator_test.py
+1 -1
View File
@@ -447,7 +447,7 @@ class MultipleVhostsTest(util.ApacheTest):
addr = obj.Addr.fromstring("*:80")
self.assertIs(self.config.is_name_vhost(addr), True)
self.config.version = (2, 2)
self.assertEqual(self.config.is_name_vhost(addr), [])
self.assertIs(self.config.is_name_vhost(addr), False)
def test_add_name_vhost(self):
self.config.add_name_vhost(obj.Addr.fromstring("*:443"))
certbot-ci/certbot_integration_tests/utils/acme_server.py
+3 -4
View File
@@ -18,7 +18,6 @@ from typing import Dict
from typing import List
from typing import Mapping
from typing import Optional
from typing import Sequence
from typing import Type
import requests
@@ -42,7 +41,7 @@ class ACMEServer:
ACMEServer is also a context manager, and so can be used to ensure ACME server is
started/stopped upon context enter/exit.
"""
def __init__(self, acme_server: str, nodes: Sequence[str], http_proxy: bool = True,
def __init__(self, acme_server: str, nodes: List[str], http_proxy: bool = True,
stdout: bool = False, dns_server: Optional[str] = None,
http_01_port: int = DEFAULT_HTTP_01_PORT) -> None:
"""
@@ -121,7 +120,7 @@ class ACMEServer:
traceback: Optional[TracebackType]) -> None:
self.stop()
def _construct_acme_xdist(self, acme_server: str, nodes: Sequence[str]) -> None:
def _construct_acme_xdist(self, acme_server: str, nodes: List[str]) -> None:
"""Generate and return the acme_xdist dict"""
acme_xdist = {'acme_server': acme_server, 'challtestsrv_port': CHALLTESTSRV_PORT}
@@ -246,7 +245,7 @@ class ACMEServer:
self._launch_process(command)
print('=> Finished configuring the HTTP proxy.')
def _launch_process(self, command: Sequence[str], cwd: str = os.getcwd(),
def _launch_process(self, command: List[str], cwd: str = os.getcwd(),
env: Optional[Mapping[str, str]] = None,
force_stderr: bool = False) -> subprocess.Popen:
"""Launch silently a subprocess OS command"""
certbot-ci/certbot_integration_tests/utils/certbot_call.py
+2 -3
View File
@@ -8,7 +8,6 @@ import sys
from typing import Dict
from typing import List
from typing import Mapping
from typing import Sequence
from typing import Tuple
import certbot_integration_tests
@@ -16,7 +15,7 @@ import certbot_integration_tests
from certbot_integration_tests.utils.constants import *
def certbot_test(certbot_args: Sequence[str], directory_url: str, http_01_port: int,
def certbot_test(certbot_args: List[str], directory_url: str, http_01_port: int,
tls_alpn_01_port: int, config_dir: str, workspace: str,
force_renew: bool = True) -> Tuple[str, str]:
"""
@@ -101,7 +100,7 @@ def _compute_additional_args(workspace: str, environ: Mapping[str, str],
return additional_args
def _prepare_args_env(certbot_args: Sequence[str], directory_url: str, http_01_port: int,
def _prepare_args_env(certbot_args: List[str], directory_url: str, http_01_port: int,
tls_alpn_01_port: int, config_dir: str, workspace: str,
force_renew: bool) -> Tuple[List[str], Dict[str, str]]:
certbot-ci/certbot_integration_tests/utils/dns_server.py
+3 -2
View File
@@ -9,8 +9,9 @@ import sys
import tempfile
import time
from types import TracebackType
from typing import Any, Sequence
from typing import Any
from typing import Dict
from typing import List
from typing import Optional
from typing import Type
@@ -34,7 +35,7 @@ class DNSServer:
future to support parallelization (https://github.com/certbot/certbot/issues/8455).
"""
def __init__(self, unused_nodes: Sequence[str], show_output: bool = False) -> None:
def __init__(self, unused_nodes: List[str], show_output: bool = False) -> None:
"""
Create an DNSServer instance.
:param list nodes: list of node names that will be setup by pytest xdist
certbot-compatibility-test/certbot_compatibility_test/configurators/common.py
+2 -1
View File
@@ -21,6 +21,7 @@ from certbot_compatibility_test import util
from acme import challenges
from acme.challenges import Challenge
from certbot._internal import constants
from certbot.plugins import common
from certbot.achallenges import AnnotatedChallenge
logger = logging.getLogger(__name__)
@@ -49,7 +50,7 @@ class Proxy(interfaces.ConfiguratorProxy):
self.args = args
self.http_port = 80
self.https_port = 443
self._configurator: interfaces.Configurator
self._configurator: common.Configurator
self._all_names: Optional[Set[str]] = None
self._test_names: Optional[Set[str]] = None
certbot-compatibility-test/certbot_compatibility_test/configurators/nginx/common.py
+1 -4
View File
@@ -2,12 +2,10 @@
import os
import shutil
import subprocess
from typing import cast
from typing import Set
from typing import Tuple
from certbot_compatibility_test import errors
from certbot_compatibility_test import interfaces
from certbot_compatibility_test import util
from certbot_compatibility_test.configurators import common as configurators_common
from certbot_nginx._internal import configurator
@@ -48,8 +46,7 @@ class Proxy(configurators_common.Proxy):
setattr(self.le_config, "nginx_" + k, constants.os_constant(k))
conf = configuration.NamespaceConfig(self.le_config)
self._configurator = cast(interfaces.Configurator, configurator.NginxConfigurator(
config=conf, name="nginx"))
self._configurator = configurator.NginxConfigurator(config=conf, name="nginx")
self._configurator.prepare()
def cleanup_from_tests(self) -> None:
certbot-compatibility-test/certbot_compatibility_test/interfaces.py
-4
View File
@@ -63,7 +63,3 @@ class InstallerProxy(PluginProxy, interfaces.Installer, metaclass=ABCMeta):
class ConfiguratorProxy(AuthenticatorProxy, InstallerProxy, metaclass=ABCMeta):
"""Wraps a Certbot configurator"""
class Configurator(interfaces.Installer, interfaces.Authenticator, metaclass=ABCMeta):
"""Represents a plugin that has both Installer and Authenticator capabilities"""
certbot/certbot/display/ops.py
+3 -4
View File
@@ -6,7 +6,6 @@ from typing import Callable
from typing import Iterable
from typing import List
from typing import Optional
from typing import Sequence
from typing import Tuple
from certbot import errors
@@ -242,7 +241,7 @@ def _choose_names_manually(prompt_prefix: str = "") -> List[str]:
return []
def success_installation(domains: Sequence[str]) -> None:
def success_installation(domains: List[str]) -> None:
"""Display a box confirming the installation of HTTPS.
:param list domains: domain names which were enabled
@@ -254,7 +253,7 @@ def success_installation(domains: Sequence[str]) -> None:
)
def success_renewal(unused_domains: Sequence[str]) -> None:
def success_renewal(unused_domains: List[str]) -> None:
"""Display a box confirming the renewal of an existing certificate.
:param list domains: domain names which were renewed
@@ -296,7 +295,7 @@ def report_executed_command(command_name: str, returncode: int, stdout: str, std
logger.warning("%s ran with error output:\n%s", command_name, indent(err_s, ' '))
def _gen_https_names(domains: Sequence[str]) -> str:
def _gen_https_names(domains: List[str]) -> str:
"""Returns a string of the https domains.
Domains are formatted nicely with ``https://`` prepended to each.
certbot/certbot/plugins/common.py
+1 -1
View File
@@ -12,9 +12,9 @@ from typing import Iterable
from typing import List
from typing import Optional
from typing import Set
from typing import Tuple
from typing import Type
from typing import TypeVar
from typing import Tuple
import pkg_resources
mypy.ini
+1 -1
View File
@@ -1,5 +1,5 @@
[mypy]
check_untyped_defs = True
ignore_missing_imports = True
warn_unused_ignores = True
show_error_codes = True
disallow_untyped_defs = True
tests/lock_test.py
+4 -5
View File
@@ -11,7 +11,6 @@ import sys
import tempfile
from typing import Iterable
from typing import List
from typing import Sequence
from typing import Tuple
from cryptography import x509
@@ -185,7 +184,7 @@ def setup_certificate(workspace: str) -> Tuple[str, str]:
return key_path, cert_path
def test_command(command: Sequence[str], directories: Iterable[str]) -> None:
def test_command(command: List[str], directories: Iterable[str]) -> None:
"""Assert Certbot acquires locks in a specific order.
command is run repeatedly testing that Certbot acquires locks on
@@ -202,7 +201,7 @@ def test_command(command: Sequence[str], directories: Iterable[str]) -> None:
dir_lock.release()
def check_error(command: Sequence[str], dir_path: str) -> None:
def check_error(command: List[str], dir_path: str) -> None:
"""Run command and verify it fails to acquire the lock for dir_path.
:param str command: certbot command to run
@@ -227,7 +226,7 @@ def check_error(command: Sequence[str], dir_path: str) -> None:
report_failure(err_msg, out, err)
def check_call(args: Sequence[str]) -> str:
def check_call(args: List[str]) -> str:
"""Simple imitation of subprocess.check_call.
This function is only available in subprocess in Python 2.7+.
@@ -257,7 +256,7 @@ def report_failure(err_msg: str, out: str, err: str) -> None:
sys.exit(err_msg)
def subprocess_call(args: Sequence[str]) -> Tuple[int, str, str]:
def subprocess_call(args: List[str]) -> Tuple[int, str, str]:
"""Run a command with subprocess and return the result.
:param list args: program and it's arguments to be run
tox.ini
+3 -6
View File
@@ -20,8 +20,7 @@ install_and_test = python {toxinidir}/tools/install_and_test.py
dns_packages = certbot-dns-cloudflare certbot-dns-cloudxns certbot-dns-digitalocean certbot-dns-dnsimple certbot-dns-dnsmadeeasy certbot-dns-gehirn certbot-dns-google certbot-dns-linode certbot-dns-luadns certbot-dns-nsone certbot-dns-ovh certbot-dns-rfc2136 certbot-dns-route53 certbot-dns-sakuracloud
win_all_packages = acme[test] certbot[test] {[base]dns_packages} certbot-nginx
all_packages = {[base]win_all_packages} certbot-apache
fully_typed_source_paths = acme/acme certbot/certbot certbot-ci/certbot_integration_tests certbot-ci/snap_integration_tests certbot-ci/windows_installer_integration_tests certbot-compatibility-test/certbot_compatibility_test certbot-dns-cloudflare/certbot_dns_cloudflare certbot-dns-cloudxns/certbot_dns_cloudxns certbot-dns-digitalocean/certbot_dns_digitalocean certbot-dns-dnsimple/certbot_dns_dnsimple certbot-dns-dnsmadeeasy/certbot_dns_dnsmadeeasy certbot-dns-gehirn/certbot_dns_gehirn certbot-dns-google/certbot_dns_google certbot-dns-linode/certbot_dns_linode certbot-dns-luadns/certbot_dns_luadns certbot-dns-nsone/certbot_dns_nsone certbot-dns-ovh/certbot_dns_ovh certbot-dns-rfc2136/certbot_dns_rfc2136 certbot-dns-route53/certbot_dns_route53 certbot-dns-sakuracloud/certbot_dns_sakuracloud certbot-nginx/certbot_nginx tests/lock_test.py
partially_typed_source_paths = certbot-apache/certbot_apache
source_paths = acme/acme certbot/certbot certbot-apache/certbot_apache certbot-ci/certbot_integration_tests certbot-ci/snap_integration_tests certbot-ci/windows_installer_integration_tests certbot-compatibility-test/certbot_compatibility_test certbot-dns-cloudflare/certbot_dns_cloudflare certbot-dns-cloudxns/certbot_dns_cloudxns certbot-dns-digitalocean/certbot_dns_digitalocean certbot-dns-dnsimple/certbot_dns_dnsimple certbot-dns-dnsmadeeasy/certbot_dns_dnsmadeeasy certbot-dns-gehirn/certbot_dns_gehirn certbot-dns-google/certbot_dns_google certbot-dns-linode/certbot_dns_linode certbot-dns-luadns/certbot_dns_luadns certbot-dns-nsone/certbot_dns_nsone certbot-dns-ovh/certbot_dns_ovh certbot-dns-rfc2136/certbot_dns_rfc2136 certbot-dns-route53/certbot_dns_route53 certbot-dns-sakuracloud/certbot_dns_sakuracloud certbot-nginx/certbot_nginx tests/lock_test.py
[testenv]
passenv =
@@ -126,16 +125,14 @@ basepython = python3
commands =
win: {[base]pip_install} {[base]win_all_packages}
!win: {[base]pip_install} {[base]all_packages}
python -m pylint --reports=n --rcfile=.pylintrc {[base]fully_typed_source_paths} {[base]partially_typed_source_paths}
python -m pylint --reports=n --rcfile=.pylintrc {[base]source_paths}
// TODO: Re-enable strict checks for optionals with appropriate type corrections or code redesign.
[testenv:mypy{,-win,-posix}]
basepython = python3
commands =
win: {[base]pip_install} {[base]win_all_packages}
!win: {[base]pip_install} {[base]all_packages} certbot-ci
mypy --disallow-untyped-defs {[base]fully_typed_source_paths}
mypy {[base]partially_typed_source_paths}
mypy {[base]source_paths}
[testenv:apacheconftest]
commands =