diff --git a/acme/acme/py.typed b/acme/acme/py.typed new file mode 100644 index 000000000..e69de29bb diff --git a/certbot-apache/certbot_apache/_internal/configurator.py b/certbot-apache/certbot_apache/_internal/configurator.py index cf4019ef3..e6ccdbabf 100644 --- a/certbot-apache/certbot_apache/_internal/configurator.py +++ b/certbot-apache/certbot_apache/_internal/configurator.py @@ -2437,10 +2437,9 @@ class ApacheConfigurator(common.Installer, interfaces.Authenticator): except errors.SubprocessError as err: logger.warning("Unable to restart apache using %s", self.options.restart_cmd) - alt_restart = self.options.restart_cmd_alt - if alt_restart: + if self.options.restart_cmd_alt: logger.debug("Trying alternative restart command: %s", - alt_restart) + self.options.restart_cmd_alt) # There is an alternative restart command available # This usually is "restart" verb while original is "graceful" try: diff --git a/certbot-apache/certbot_apache/_internal/override_debian.py b/certbot-apache/certbot_apache/_internal/override_debian.py index 385d3d7a2..78041ceca 100644 --- a/certbot-apache/certbot_apache/_internal/override_debian.py +++ b/certbot-apache/certbot_apache/_internal/override_debian.py @@ -118,7 +118,8 @@ class DebianConfigurator(configurator.ApacheConfigurator): # Generate reversal command. # Try to be safe here... check that we can probably reverse before # applying enmod command - if not util.exe_exists(self.options.dismod): + if (self.options.dismod is None or self.options.enmod is None + or not util.exe_exists(self.options.dismod)): raise errors.MisconfigurationError( "Unable to find a2dismod, please make sure a2enmod and " "a2dismod are configured correctly for certbot.") diff --git a/certbot-apache/certbot_apache/py.typed b/certbot-apache/certbot_apache/py.typed new file mode 100644 index 000000000..e69de29bb diff --git a/certbot-compatibility-test/certbot_compatibility_test/validator.py b/certbot-compatibility-test/certbot_compatibility_test/validator.py index 226b8585b..e2398ffbd 100644 --- a/certbot-compatibility-test/certbot_compatibility_test/validator.py +++ b/certbot-compatibility-test/certbot_compatibility_test/validator.py @@ -1,6 +1,7 @@ """Validators to determine the current webserver configuration""" import logging import socket +from typing import cast import requests @@ -29,7 +30,8 @@ class Validator: logger.exception(str(error)) return False - return presented_cert.digest("sha256") == cert.digest("sha256") + # Despite documentation saying that bytes are expected for digest(), we must provide a str. + return presented_cert.digest(cast(bytes, "sha256")) == cert.digest("sha256") def redirect(self, name, port=80, headers=None): """Test whether webserver redirects to secure connection.""" diff --git a/certbot-dns-cloudflare/certbot_dns_cloudflare/py.typed b/certbot-dns-cloudflare/certbot_dns_cloudflare/py.typed new file mode 100644 index 000000000..e69de29bb diff --git a/certbot-dns-cloudxns/certbot_dns_cloudxns/py.typed b/certbot-dns-cloudxns/certbot_dns_cloudxns/py.typed new file mode 100644 index 000000000..e69de29bb diff --git a/certbot-dns-digitalocean/certbot_dns_digitalocean/py.typed b/certbot-dns-digitalocean/certbot_dns_digitalocean/py.typed new file mode 100644 index 000000000..e69de29bb diff --git a/certbot-dns-dnsimple/certbot_dns_dnsimple/py.typed b/certbot-dns-dnsimple/certbot_dns_dnsimple/py.typed new file mode 100644 index 000000000..e69de29bb diff --git a/certbot-dns-dnsmadeeasy/certbot_dns_dnsmadeeasy/py.typed b/certbot-dns-dnsmadeeasy/certbot_dns_dnsmadeeasy/py.typed new file mode 100644 index 000000000..e69de29bb diff --git a/certbot-dns-gehirn/certbot_dns_gehirn/py.typed b/certbot-dns-gehirn/certbot_dns_gehirn/py.typed new file mode 100644 index 000000000..e69de29bb diff --git a/certbot-dns-google/certbot_dns_google/py.typed b/certbot-dns-google/certbot_dns_google/py.typed new file mode 100644 index 000000000..e69de29bb diff --git a/certbot-dns-linode/certbot_dns_linode/py.typed b/certbot-dns-linode/certbot_dns_linode/py.typed new file mode 100644 index 000000000..e69de29bb diff --git a/certbot-dns-luadns/certbot_dns_luadns/py.typed b/certbot-dns-luadns/certbot_dns_luadns/py.typed new file mode 100644 index 000000000..e69de29bb diff --git a/certbot-dns-nsone/certbot_dns_nsone/py.typed b/certbot-dns-nsone/certbot_dns_nsone/py.typed new file mode 100644 index 000000000..e69de29bb diff --git a/certbot-dns-ovh/certbot_dns_ovh/py.typed b/certbot-dns-ovh/certbot_dns_ovh/py.typed new file mode 100644 index 000000000..e69de29bb diff --git a/certbot-dns-rfc2136/certbot_dns_rfc2136/py.typed b/certbot-dns-rfc2136/certbot_dns_rfc2136/py.typed new file mode 100644 index 000000000..e69de29bb diff --git a/certbot-dns-route53/certbot_dns_route53/py.typed b/certbot-dns-route53/certbot_dns_route53/py.typed new file mode 100644 index 000000000..e69de29bb diff --git a/certbot-dns-sakuracloud/certbot_dns_sakuracloud/py.typed b/certbot-dns-sakuracloud/certbot_dns_sakuracloud/py.typed new file mode 100644 index 000000000..e69de29bb diff --git a/certbot-nginx/certbot_nginx/py.typed b/certbot-nginx/certbot_nginx/py.typed new file mode 100644 index 000000000..e69de29bb diff --git a/certbot/certbot/_internal/client.py b/certbot/certbot/_internal/client.py index 899b051bf..1a6f5ed1b 100644 --- a/certbot/certbot/_internal/client.py +++ b/certbot/certbot/_internal/client.py @@ -2,6 +2,7 @@ import datetime import logging import platform +from typing import cast from typing import Any from typing import Dict from typing import List @@ -224,8 +225,10 @@ def perform_registration(acme, config, tos_cb): raise errors.Error(msg) try: - newreg = messages.NewRegistration.from_data(email=config.email, - external_account_binding=eab) + # TODO: Remove the cast once certbot package is fully typed + newreg = messages.NewRegistration.from_data( + email=config.email, + external_account_binding=cast(Optional[messages.ExternalAccountBinding], eab)) return acme.new_account_and_tos(newreg, tos_cb) except messages.Error as e: if e.code == "invalidEmail" or e.code == "invalidContact": diff --git a/certbot/certbot/_internal/storage.py b/certbot/certbot/_internal/storage.py index 927addb5e..954241c0e 100644 --- a/certbot/certbot/_internal/storage.py +++ b/certbot/certbot/_internal/storage.py @@ -5,6 +5,7 @@ import logging import re import shutil import stat +from typing import cast from typing import Optional import configobj @@ -895,7 +896,8 @@ class RenewableCert(interfaces.RenewableCert): if target is None: raise errors.CertStorageError("could not find the certificate file") with open(target) as f: - return crypto_util.get_names_from_cert(f.read()) + # TODO: Remove the cast once certbot package is fully typed + return crypto_util.get_names_from_cert(cast(bytes, f.read())) def ocsp_revoked(self, version): """Is the specified cert version revoked according to OCSP? diff --git a/certbot/certbot/achallenges.py b/certbot/certbot/achallenges.py index a3ddbdcd3..73666825c 100644 --- a/certbot/certbot/achallenges.py +++ b/certbot/certbot/achallenges.py @@ -18,6 +18,7 @@ Note, that all annotated challenges act as a proxy objects:: """ import logging +from typing import Any from typing import Type import josepy as jose @@ -40,7 +41,7 @@ class AnnotatedChallenge(jose.ImmutableMap): __slots__ = ('challb',) _acme_type: Type[Challenge] = NotImplemented - def __getattr__(self, name): + def __getattr__(self, name: str) -> Any: return getattr(self.challb, name) @@ -48,7 +49,7 @@ class KeyAuthorizationAnnotatedChallenge(AnnotatedChallenge): """Client annotated `KeyAuthorizationChallenge` challenge.""" __slots__ = ('challb', 'domain', 'account_key') - def response_and_validation(self, *args, **kwargs): + def response_and_validation(self, *args: Any, **kwargs: Any) -> Any: """Generate response and validation.""" return self.challb.chall.response_and_validation( self.account_key, *args, **kwargs) diff --git a/certbot/certbot/configuration.py b/certbot/certbot/configuration.py index e39378a6d..4648cbda1 100644 --- a/certbot/certbot/configuration.py +++ b/certbot/certbot/configuration.py @@ -1,5 +1,7 @@ """Certbot user-supplied configuration.""" +import argparse import copy +from typing import Any from typing import List from typing import Optional from urllib import parse @@ -38,7 +40,9 @@ class NamespaceConfig: """ - def __init__(self, namespace): + def __init__(self, namespace: argparse.Namespace) -> None: + self.namespace: argparse.Namespace + # Avoid recursion loop because of the delegation defined in __setattr__ object.__setattr__(self, 'namespace', namespace) self.namespace.config_dir = os.path.abspath(self.namespace.config_dir) @@ -50,10 +54,10 @@ class NamespaceConfig: # Delegate any attribute not explicitly defined to the underlying namespace object. - def __getattr__(self, name): + def __getattr__(self, name: str) -> Any: return getattr(self.namespace, name) - def __setattr__(self, name, value): + def __setattr__(self, name: str, value: Any) -> None: setattr(self.namespace, name, value) @property @@ -61,6 +65,10 @@ class NamespaceConfig: """ACME Directory Resource URI.""" return self.namespace.server + @server.setter + def server(self, server_: str) -> None: + self.namespace.server = server_ + @property def email(self) -> Optional[str]: """Email used for registration and recovery contact. @@ -70,6 +78,10 @@ class NamespaceConfig: """ return self.namespace.email + @email.setter + def email(self, mail: str) -> None: + self.namespace.email = mail + @property def rsa_key_size(self) -> int: """Size of the RSA key.""" @@ -126,32 +138,32 @@ class NamespaceConfig: return self.namespace.work_dir @property - def accounts_dir(self): + def accounts_dir(self) -> str: """Directory where all account information is stored.""" return self.accounts_dir_for_server_path(self.server_path) @property - def backup_dir(self): + def backup_dir(self) -> str: """Configuration backups directory.""" return os.path.join(self.namespace.work_dir, constants.BACKUP_DIR) @property - def csr_dir(self): + def csr_dir(self) -> str: """Directory where new Certificate Signing Requests (CSRs) are saved.""" return os.path.join(self.namespace.config_dir, constants.CSR_DIR) @property - def in_progress_dir(self): + def in_progress_dir(self) -> str: """Directory used before a permanent checkpoint is finalized.""" return os.path.join(self.namespace.work_dir, constants.IN_PROGRESS_DIR) @property - def key_dir(self): + def key_dir(self) -> str: """Keys storage.""" return os.path.join(self.namespace.config_dir, constants.KEY_DIR) @property - def temp_checkpoint_dir(self): + def temp_checkpoint_dir(self) -> str: """Temporary checkpoint directory.""" return os.path.join( self.namespace.work_dir, constants.TEMP_CHECKPOINT_DIR) @@ -233,64 +245,64 @@ class NamespaceConfig: return self.namespace.preferred_chain @property - def server_path(self): + def server_path(self) -> str: """File path based on ``server``.""" parsed = parse.urlparse(self.namespace.server) return (parsed.netloc + parsed.path).replace('/', os.path.sep) - def accounts_dir_for_server_path(self, server_path): + def accounts_dir_for_server_path(self, server_path: str) -> str: """Path to accounts directory based on server_path""" server_path = misc.underscores_for_unsupported_characters_in_path(server_path) return os.path.join( self.namespace.config_dir, constants.ACCOUNTS_DIR, server_path) @property - def default_archive_dir(self): # pylint: disable=missing-function-docstring + def default_archive_dir(self) -> str: # pylint: disable=missing-function-docstring return os.path.join(self.namespace.config_dir, constants.ARCHIVE_DIR) @property - def live_dir(self): # pylint: disable=missing-function-docstring + def live_dir(self) -> str: # pylint: disable=missing-function-docstring return os.path.join(self.namespace.config_dir, constants.LIVE_DIR) @property - def renewal_configs_dir(self): # pylint: disable=missing-function-docstring + def renewal_configs_dir(self) -> str: # pylint: disable=missing-function-docstring return os.path.join( self.namespace.config_dir, constants.RENEWAL_CONFIGS_DIR) @property - def renewal_hooks_dir(self): + def renewal_hooks_dir(self) -> str: """Path to directory with hooks to run with the renew subcommand.""" return os.path.join(self.namespace.config_dir, constants.RENEWAL_HOOKS_DIR) @property - def renewal_pre_hooks_dir(self): + def renewal_pre_hooks_dir(self) -> str: """Path to the pre-hook directory for the renew subcommand.""" return os.path.join(self.renewal_hooks_dir, constants.RENEWAL_PRE_HOOKS_DIR) @property - def renewal_deploy_hooks_dir(self): + def renewal_deploy_hooks_dir(self) -> str: """Path to the deploy-hook directory for the renew subcommand.""" return os.path.join(self.renewal_hooks_dir, constants.RENEWAL_DEPLOY_HOOKS_DIR) @property - def renewal_post_hooks_dir(self): + def renewal_post_hooks_dir(self) -> str: """Path to the post-hook directory for the renew subcommand.""" return os.path.join(self.renewal_hooks_dir, constants.RENEWAL_POST_HOOKS_DIR) # Magic methods - def __deepcopy__(self, _memo): + def __deepcopy__(self, _memo: Any) -> 'NamespaceConfig': # Work around https://bugs.python.org/issue1515 for py26 tests :( :( # https://travis-ci.org/letsencrypt/letsencrypt/jobs/106900743#L3276 new_ns = copy.deepcopy(self.namespace) return type(self)(new_ns) -def _check_config_sanity(config): +def _check_config_sanity(config: NamespaceConfig) -> None: """Validate command line options and display error message if requirements are not met. diff --git a/certbot/certbot/crypto_util.py b/certbot/certbot/crypto_util.py index d277c6856..f7dca341b 100644 --- a/certbot/certbot/crypto_util.py +++ b/certbot/certbot/crypto_util.py @@ -4,18 +4,26 @@ is capable of handling the signatures. """ +import datetime import hashlib import logging import re +from typing import Callable from typing import List +from typing import Optional from typing import Set +from typing import Tuple +from typing import TYPE_CHECKING +from typing import Union import warnings from cryptography import x509 from cryptography.exceptions import InvalidSignature from cryptography.exceptions import UnsupportedAlgorithm from cryptography.hazmat.backends import default_backend +from cryptography.hazmat.primitives import hashes from cryptography.hazmat.primitives.asymmetric import ec +from cryptography.hazmat.primitives.asymmetric.dsa import DSAPublicKey from cryptography.hazmat.primitives.asymmetric.ec import ECDSA from cryptography.hazmat.primitives.asymmetric.ec import EllipticCurvePublicKey from cryptography.hazmat.primitives.asymmetric.padding import PKCS1v15 @@ -23,6 +31,7 @@ from cryptography.hazmat.primitives.asymmetric.rsa import RSAPublicKey from cryptography.hazmat.primitives.serialization import Encoding from cryptography.hazmat.primitives.serialization import NoEncryption from cryptography.hazmat.primitives.serialization import PrivateFormat +import josepy from OpenSSL import crypto from OpenSSL import SSL import pyrfc3339 @@ -34,6 +43,11 @@ from certbot import interfaces from certbot import util from certbot.compat import os +# Cryptography ed448 and ed25519 modules do not exist on oldest tests +if TYPE_CHECKING: + from cryptography.hazmat.primitives.asymmetric.ed448 import Ed448PublicKey + from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey + logger = logging.getLogger(__name__) @@ -87,8 +101,9 @@ def generate_key(key_size: int, key_dir: str, key_type: str = "rsa", # TODO: Remove this call once zope dependencies are removed from Certbot. -def init_save_key(key_size, key_dir, key_type="rsa", elliptic_curve="secp256r1", - keyname="key-certbot.pem"): +def init_save_key(key_size: int, key_dir: str, key_type: str = "rsa", + elliptic_curve: str = "secp256r1", + keyname: str = "key-certbot.pem") -> util.Key: """Initializes and saves a privkey. Inits key and saves it in PEM format on the filesystem. @@ -120,7 +135,7 @@ def init_save_key(key_size, key_dir, key_type="rsa", elliptic_curve="secp256r1", keyname=keyname, strict_permissions=config.strict_permissions) -def generate_csr(privkey: util.Key, names: Set[str], path: str, +def generate_csr(privkey: util.Key, names: Union[List[str], Set[str]], path: str, must_staple: bool = False, strict_permissions: bool = True) -> util.CSR: """Initialize a CSR with the given private key. @@ -151,7 +166,7 @@ def generate_csr(privkey: util.Key, names: Set[str], path: str, # TODO: Remove this call once zope dependencies are removed from Certbot. -def init_save_csr(privkey, names, path): +def init_save_csr(privkey: util.Key, names: Set[str], path: str) -> util.CSR: """Initialize a CSR with the given private key. .. deprecated:: 1.16.0 @@ -182,12 +197,12 @@ def init_save_csr(privkey, names, path): # A. Do more checks to verify that the CSR is trusted/valid # B. Audit the parsing code for vulnerabilities -def valid_csr(csr): +def valid_csr(csr: bytes) -> bool: """Validate CSR. Check if `csr` is a valid CSR for the given domains. - :param str csr: CSR in PEM. + :param bytes csr: CSR in PEM. :returns: Validity of CSR. :rtype: bool @@ -202,11 +217,11 @@ def valid_csr(csr): return False -def csr_matches_pubkey(csr, privkey): +def csr_matches_pubkey(csr: bytes, privkey: bytes) -> bool: """Does private key correspond to the subject public key in the CSR? - :param str csr: CSR in PEM. - :param str privkey: Private key file contents (PEM) + :param bytes csr: CSR in PEM. + :param bytes privkey: Private key file contents (PEM) :returns: Correspondence of private key to CSR subject public key. :rtype: bool @@ -222,11 +237,11 @@ def csr_matches_pubkey(csr, privkey): return False -def import_csr_file(csrfile, data): +def import_csr_file(csrfile: str, data: bytes) -> Tuple[int, util.CSR, List[str]]: """Import a CSR file, which can be either PEM or DER. :param str csrfile: CSR filename - :param str data: contents of the CSR file + :param bytes data: contents of the CSR file :returns: (`crypto.FILETYPE_PEM`, util.CSR object representing the CSR, @@ -251,12 +266,13 @@ def import_csr_file(csrfile, data): return PEM, util.CSR(file=csrfile, data=data_pem, form="pem"), domains -def make_key(bits=1024, key_type="rsa", elliptic_curve=None): +def make_key(bits: int = 1024, key_type: str = "rsa", + elliptic_curve: Optional[str] = None) -> bytes: """Generate PEM encoded RSA|EC key. :param int bits: Number of bits if key_type=rsa. At least 1024 for RSA. - - :param str ec_curve: The elliptic curve to use. + :param str key_type: The type of key to generate, but be rsa or ecdsa + :param str elliptic_curve: The elliptic curve to use. :returns: new RSA or ECDSA key in PEM form with specified number of bits or of type ec_curve when key_type ecdsa is used. @@ -269,6 +285,8 @@ def make_key(bits=1024, key_type="rsa", elliptic_curve=None): key = crypto.PKey() key.generate_key(crypto.TYPE_RSA, bits) elif key_type == 'ecdsa': + if not elliptic_curve: + raise errors.Error("When key_type == ecdsa, elliptic_curve must be set.") try: name = elliptic_curve.upper() if name in ('SECP256R1', 'SECP384R1', 'SECP521R1'): @@ -297,7 +315,7 @@ def make_key(bits=1024, key_type="rsa", elliptic_curve=None): return crypto.dump_privatekey(crypto.FILETYPE_PEM, key) -def valid_privkey(privkey): +def valid_privkey(privkey: str) -> bool: """Is valid RSA private key? :param str privkey: Private key file contents in PEM @@ -313,7 +331,7 @@ def valid_privkey(privkey): return False -def verify_renewable_cert(renewable_cert): +def verify_renewable_cert(renewable_cert: interfaces.RenewableCert) -> None: """For checking that your certs were not corrupted on disk. Several things are checked: @@ -331,7 +349,7 @@ def verify_renewable_cert(renewable_cert): verify_cert_matches_priv_key(renewable_cert.cert_path, renewable_cert.key_path) -def verify_renewable_cert_sig(renewable_cert): +def verify_renewable_cert_sig(renewable_cert: interfaces.RenewableCert) -> None: """Verifies the signature of a RenewableCert object. :param renewable_cert: cert to verify @@ -355,14 +373,16 @@ def verify_renewable_cert_sig(renewable_cert): raise errors.Error(error_str) -def verify_signed_payload(public_key, signature, payload, signature_hash_algorithm): +def verify_signed_payload(public_key: Union[DSAPublicKey, 'Ed25519PublicKey', 'Ed448PublicKey', + EllipticCurvePublicKey, RSAPublicKey], + signature: bytes, payload: bytes, + signature_hash_algorithm: hashes.HashAlgorithm) -> None: """Check the signature of a payload. :param RSAPublicKey/EllipticCurvePublicKey public_key: the public_key to check signature :param bytes signature: the signature bytes :param bytes payload: the payload bytes - :param cryptography.hazmat.primitives.hashes.HashAlgorithm \ - signature_hash_algorithm: algorithm used to hash the payload + :param hashes.HashAlgorithm signature_hash_algorithm: algorithm used to hash the payload :raises InvalidSignature: If signature verification fails. :raises errors.Error: If public key type is not supported @@ -382,10 +402,10 @@ def verify_signed_payload(public_key, signature, payload, signature_hash_algorit verifier.update(payload) verifier.verify() else: - raise errors.Error("Unsupported public key type") + raise errors.Error("Unsupported public key type.") -def verify_cert_matches_priv_key(cert_path, key_path): +def verify_cert_matches_priv_key(cert_path: str, key_path: str) -> None: """ Verifies that the private key and cert match. :param str cert_path: path to a cert in PEM format @@ -407,7 +427,7 @@ def verify_cert_matches_priv_key(cert_path, key_path): raise errors.Error(error_str) -def verify_fullchain(renewable_cert): +def verify_fullchain(renewable_cert: interfaces.RenewableCert) -> None: """ Verifies that fullchain is indeed cert concatenated with chain. :param renewable_cert: cert to verify @@ -434,7 +454,7 @@ def verify_fullchain(renewable_cert): raise e -def pyopenssl_load_certificate(data): +def pyopenssl_load_certificate(data: bytes) -> Tuple[crypto.X509, int]: """Load PEM/DER certificate. :raises errors.Error: @@ -452,8 +472,9 @@ def pyopenssl_load_certificate(data): str(error) for error in openssl_errors))) -def _load_cert_or_req(cert_or_req_str, load_func, - typ=crypto.FILETYPE_PEM): +def _load_cert_or_req(cert_or_req_str: bytes, + load_func: Callable[[int, bytes], Union[crypto.X509, crypto.X509Req]], + typ: int = crypto.FILETYPE_PEM) -> Union[crypto.X509, crypto.X509Req]: try: return load_func(typ, cert_or_req_str) except crypto.Error as err: @@ -462,14 +483,16 @@ def _load_cert_or_req(cert_or_req_str, load_func, raise -def _get_sans_from_cert_or_req(cert_or_req_str, load_func, - typ=crypto.FILETYPE_PEM): +def _get_sans_from_cert_or_req(cert_or_req_str: bytes, + load_func: Callable[[int, bytes], Union[crypto.X509, + crypto.X509Req]], + typ: int = crypto.FILETYPE_PEM) -> List[str]: # pylint: disable=protected-access return acme_crypto_util._pyopenssl_cert_or_req_san(_load_cert_or_req( cert_or_req_str, load_func, typ)) -def get_sans_from_cert(cert, typ=crypto.FILETYPE_PEM): +def get_sans_from_cert(cert: bytes, typ: int = crypto.FILETYPE_PEM) -> List[str]: """Get a list of Subject Alternative Names from a certificate. :param str cert: Certificate (encoded). @@ -483,17 +506,21 @@ def get_sans_from_cert(cert, typ=crypto.FILETYPE_PEM): cert, crypto.load_certificate, typ) -def _get_names_from_cert_or_req(cert_or_req, load_func, typ): +def _get_names_from_cert_or_req(cert_or_req: bytes, + load_func: Callable[[int, bytes], Union[crypto.X509, + crypto.X509Req]], + typ: int) -> List[str]: loaded_cert_or_req = _load_cert_or_req(cert_or_req, load_func, typ) return _get_names_from_loaded_cert_or_req(loaded_cert_or_req) -def _get_names_from_loaded_cert_or_req(loaded_cert_or_req): +def _get_names_from_loaded_cert_or_req(loaded_cert_or_req: Union[crypto.X509, crypto.X509Req] + ) -> List[str]: # pylint: disable=protected-access return acme_crypto_util._pyopenssl_cert_or_req_all_names(loaded_cert_or_req) -def get_names_from_cert(csr, typ=crypto.FILETYPE_PEM): +def get_names_from_cert(cert: bytes, typ: int = crypto.FILETYPE_PEM) -> List[str]: """Get a list of domains from a cert, including the CN if it is set. :param str cert: Certificate (encoded). @@ -504,13 +531,13 @@ def get_names_from_cert(csr, typ=crypto.FILETYPE_PEM): """ return _get_names_from_cert_or_req( - csr, crypto.load_certificate, typ) + cert, crypto.load_certificate, typ) -def get_names_from_req(csr: str, typ: int = crypto.FILETYPE_PEM) -> List[str]: +def get_names_from_req(csr: bytes, typ: int = crypto.FILETYPE_PEM) -> List[str]: """Get a list of domains from a CSR, including the CN if it is set. - :param str cert: CSR (encoded). + :param str csr: CSR (encoded). :param typ: `crypto.FILETYPE_PEM` or `crypto.FILETYPE_ASN1` :returns: A list of domain names. :rtype: list @@ -519,7 +546,8 @@ def get_names_from_req(csr: str, typ: int = crypto.FILETYPE_PEM) -> List[str]: return _get_names_from_cert_or_req(csr, crypto.load_certificate_request, typ) -def dump_pyopenssl_chain(chain, filetype=crypto.FILETYPE_PEM): +def dump_pyopenssl_chain(chain: Union[List[crypto.X509], List[josepy.ComparableX509]], + filetype: int = crypto.FILETYPE_PEM) -> bytes: """Dump certificate chain into a bundle. :param list chain: List of `crypto.X509` (or wrapped in @@ -531,7 +559,7 @@ def dump_pyopenssl_chain(chain, filetype=crypto.FILETYPE_PEM): return acme_crypto_util.dump_pyopenssl_chain(chain, filetype) -def notBefore(cert_path): +def notBefore(cert_path: str) -> datetime.datetime: """When does the cert at cert_path start being valid? :param str cert_path: path to a cert in PEM format @@ -543,7 +571,7 @@ def notBefore(cert_path): return _notAfterBefore(cert_path, crypto.X509.get_notBefore) -def notAfter(cert_path): +def notAfter(cert_path: str) -> datetime.datetime: """When does the cert at cert_path stop being valid? :param str cert_path: path to a cert in PEM format @@ -555,7 +583,8 @@ def notAfter(cert_path): return _notAfterBefore(cert_path, crypto.X509.get_notAfter) -def _notAfterBefore(cert_path, method): +def _notAfterBefore(cert_path: str, + method: Callable[[crypto.X509], Optional[bytes]]) -> datetime.datetime: """Internal helper function for finding notbefore/notafter. :param str cert_path: path to a cert in PEM format @@ -571,6 +600,8 @@ def _notAfterBefore(cert_path, method): x509 = crypto.load_certificate(crypto.FILETYPE_PEM, f.read()) # pyopenssl always returns bytes timestamp = method(x509) + if not timestamp: + raise errors.Error("Error while invoking timestamp method, None has been returned.") reformatted_timestamp = [timestamp[0:4], b"-", timestamp[4:6], b"-", timestamp[6:8], b"T", timestamp[8:10], b":", timestamp[10:12], b":", timestamp[12:]] @@ -580,7 +611,7 @@ def _notAfterBefore(cert_path, method): return pyrfc3339.parse(timestamp_str) -def sha256sum(filename): +def sha256sum(filename: str) -> str: """Compute a sha256sum of a file. NB: In given file, platform specific newlines characters will be converted @@ -607,7 +638,7 @@ CERT_PEM_REGEX = re.compile( ) -def cert_and_chain_from_fullchain(fullchain_pem): +def cert_and_chain_from_fullchain(fullchain_pem: str) -> Tuple[str, str]: """Split fullchain_pem into cert_pem and chain_pem :param str fullchain_pem: concatenated cert + chain @@ -635,7 +666,7 @@ def cert_and_chain_from_fullchain(fullchain_pem): return (certs_normalized[0], "".join(certs_normalized[1:])) -def get_serial_from_cert(cert_path): +def get_serial_from_cert(cert_path: str) -> int: """Retrieve the serial number of a certificate from certificate path :param str cert_path: path to a cert in PEM format @@ -649,7 +680,8 @@ def get_serial_from_cert(cert_path): return x509.get_serial_number() -def find_chain_with_issuer(fullchains, issuer_cn, warn_on_no_match=False): +def find_chain_with_issuer(fullchains: List[str], issuer_cn: str, + warn_on_no_match: bool = False) -> str: """Chooses the first certificate chain from fullchains whose topmost intermediate has an Issuer Common Name matching issuer_cn (in other words the first chain which chains to a root whose name matches issuer_cn). diff --git a/certbot/certbot/errors.py b/certbot/certbot/errors.py index 4c31746be..f43a162fd 100644 --- a/certbot/certbot/errors.py +++ b/certbot/certbot/errors.py @@ -1,4 +1,9 @@ """Certbot client errors.""" +from typing import Set +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from certbot.achallenges import AnnotatedChallenge class Error(Exception): @@ -50,12 +55,12 @@ class FailedChallenges(AuthorizationError): :ivar set failed_achalls: Failed `.AnnotatedChallenge` instances. """ - def __init__(self, failed_achalls): + def __init__(self, failed_achalls: Set['AnnotatedChallenge']) -> None: assert failed_achalls self.failed_achalls = failed_achalls super().__init__() - def __str__(self): + def __str__(self) -> str: return "Failed authorization procedure. {0}".format( ", ".join( "{0} ({1}): {2}".format(achall.domain, achall.typ, achall.error) @@ -94,7 +99,7 @@ class PluginStorageError(PluginError): class StandaloneBindError(Error): """Standalone plugin bind error.""" - def __init__(self, socket_error, port): + def __init__(self, socket_error: OSError, port: int) -> None: super().__init__( "Problem binding to port {0}: {1}".format(port, socket_error)) self.socket_error = socket_error diff --git a/certbot/certbot/interfaces.py b/certbot/certbot/interfaces.py index fa4d1d6ae..16d3124e1 100644 --- a/certbot/certbot/interfaces.py +++ b/certbot/certbot/interfaces.py @@ -4,25 +4,33 @@ from abc import abstractmethod from argparse import ArgumentParser import sys from types import ModuleType +from typing import Any from typing import cast from typing import Iterable from typing import List from typing import Optional +from typing import Type +from typing import TYPE_CHECKING +from typing import Union import warnings import zope.interface from acme.challenges import Challenge from acme.challenges import ChallengeResponse -from certbot.achallenges import AnnotatedChallenge +from acme.client import ClientBase from certbot import configuration +from certbot.achallenges import AnnotatedChallenge + +if TYPE_CHECKING: + from certbot._internal.account import Account class AccountStorage(metaclass=ABCMeta): """Accounts storage interface.""" @abstractmethod - def find_all(self): # pragma: no cover + def find_all(self) -> List['Account']: # pragma: no cover """Find all accounts. :returns: All found accounts. @@ -32,17 +40,20 @@ class AccountStorage(metaclass=ABCMeta): raise NotImplementedError() @abstractmethod - def load(self, account_id): # pragma: no cover + def load(self, account_id: str) -> 'Account': # pragma: no cover """Load an account by its id. :raises .AccountNotFound: if account could not be found :raises .AccountStorageError: if account could not be loaded + :returns: The account loaded + :rtype: .Account + """ raise NotImplementedError() @abstractmethod - def save(self, account, client): # pragma: no cover + def save(self, account: 'Account', client: ClientBase) -> None: # pragma: no cover """Save account. :raises .AccountStorageError: if account could not be saved @@ -96,8 +107,11 @@ class Plugin(metaclass=ABCMeta): description: str = NotImplemented """Short plugin description""" + name: str = NotImplemented + """Unique name of the plugin""" + @abstractmethod - def __init__(self, config: configuration.NamespaceConfig, name: str): + def __init__(self, config: Optional[configuration.NamespaceConfig], name: str) -> None: """Create a new `Plugin`. :param configuration.NamespaceConfig config: Configuration. @@ -167,7 +181,7 @@ class Authenticator(Plugin): """ @abstractmethod - def get_chall_pref(self, domain: str) -> Iterable[Challenge]: + def get_chall_pref(self, domain: str) -> Iterable[Type[Challenge]]: """Return `collections.Iterable` of challenge preferences. :param str domain: Domain for which challenge preferences are sought. @@ -181,7 +195,7 @@ class Authenticator(Plugin): """ @abstractmethod - def perform(self, achalls: List[AnnotatedChallenge]) -> Iterable[ChallengeResponse]: + def perform(self, achalls: List[AnnotatedChallenge]) -> List[ChallengeResponse]: """Perform the given challenge. :param list achalls: Non-empty (guaranteed) list of @@ -189,10 +203,10 @@ class Authenticator(Plugin): instances, such that it contains types found within :func:`get_chall_pref` only. - :returns: `collections.Iterable` of ACME + :returns: list of ACME :class:`~acme.challenges.ChallengeResponse` instances corresponding to each provided :class:`~acme.challenges.Challenge`. - :rtype: :class:`collections.Iterable` of + :rtype: :class:`collections.List` of :class:`acme.challenges.ChallengeResponse`, where responses are required to be returned in the same order as corresponding input challenges @@ -261,7 +275,8 @@ class Installer(Plugin): """ @abstractmethod - def enhance(self, domain: str, enhancement: str, options: Optional[List[str]] = None) -> None: + def enhance(self, domain: str, enhancement: str, + options: Optional[Union[List[str], str]] = None) -> None: """Perform a configuration enhancement. :param str domain: domain for which to provide enhancement @@ -360,7 +375,7 @@ class RenewableCert(metaclass=ABCMeta): @property @abstractmethod - def cert_path(self): + def cert_path(self) -> str: """Path to the certificate file. :rtype: str @@ -369,7 +384,7 @@ class RenewableCert(metaclass=ABCMeta): @property @abstractmethod - def key_path(self): + def key_path(self) -> str: """Path to the private key file. :rtype: str @@ -378,7 +393,7 @@ class RenewableCert(metaclass=ABCMeta): @property @abstractmethod - def chain_path(self): + def chain_path(self) -> str: """Path to the certificate chain file. :rtype: str @@ -387,7 +402,7 @@ class RenewableCert(metaclass=ABCMeta): @property @abstractmethod - def fullchain_path(self): + def fullchain_path(self) -> str: """Path to the full chain file. The full chain is the certificate file plus the chain file. @@ -398,7 +413,7 @@ class RenewableCert(metaclass=ABCMeta): @property @abstractmethod - def lineagename(self): + def lineagename(self) -> str: """Name given to the certificate lineage. :rtype: str @@ -406,7 +421,7 @@ class RenewableCert(metaclass=ABCMeta): """ @abstractmethod - def names(self): + def names(self) -> List[str]: """What are the subject names of this certificate? :returns: the subject names @@ -442,7 +457,7 @@ class GenericUpdater(metaclass=ABCMeta): """ @abstractmethod - def generic_updates(self, lineage, *args, **kwargs): + def generic_updates(self, lineage: RenewableCert, *args: Any, **kwargs: Any) -> None: """Perform any update types defined by the installer. If an installer is a subclass of the class containing this method, this @@ -470,7 +485,7 @@ class RenewDeployer(metaclass=ABCMeta): """ @abstractmethod - def renew_deploy(self, lineage, *args, **kwargs): + def renew_deploy(self, lineage: RenewableCert, *args: Any, **kwargs: Any) -> None: """Perform updates defined by installer when a certificate has been renewed If an installer is a subclass of the class containing this method, this @@ -494,10 +509,10 @@ class _ZopeInterfacesDeprecationModule: Internal class delegating to a module, and displaying warnings when attributes related to Zope interfaces are accessed. """ - def __init__(self, module): + def __init__(self, module: ModuleType) -> None: self.__dict__['_module'] = module - def __getattr__(self, attr): + def __getattr__(self, attr: str) -> None: if attr in ('IConfig', 'IPlugin', 'IPluginFactory', 'IAuthenticator', 'IInstaller', 'IDisplay', 'IReporter'): warnings.warn('{0} attribute in certbot.interfaces module is deprecated ' @@ -505,13 +520,13 @@ class _ZopeInterfacesDeprecationModule: DeprecationWarning, stacklevel=2) return getattr(self._module, attr) - def __setattr__(self, attr, value): # pragma: no cover + def __setattr__(self, attr: str, value: Any) -> None: # pragma: no cover setattr(self._module, attr, value) - def __delattr__(self, attr): # pragma: no cover + def __delattr__(self, attr: str) -> None: # pragma: no cover delattr(self._module, attr) - def __dir__(self): # pragma: no cover + def __dir__(self) -> List[str]: # pragma: no cover return ['_module'] + dir(self._module) diff --git a/certbot/certbot/main.py b/certbot/certbot/main.py index b2fb1dbb7..2120deca8 100644 --- a/certbot/certbot/main.py +++ b/certbot/certbot/main.py @@ -1,8 +1,12 @@ """Certbot main public entry point.""" +from typing import List +from typing import Optional +from typing import Union + from certbot._internal import main as internal_main -def main(cli_args=None): +def main(cli_args: Optional[List[str]] = None) -> Optional[Union[str, int]]: """Run Certbot. :param cli_args: command line to Certbot, defaults to ``sys.argv[1:]`` diff --git a/certbot/certbot/ocsp.py b/certbot/certbot/ocsp.py index c25f2e1b0..84068ebf1 100644 --- a/certbot/certbot/ocsp.py +++ b/certbot/certbot/ocsp.py @@ -38,7 +38,7 @@ logger = logging.getLogger(__name__) class RevocationChecker: """This class figures out OCSP checking on this system, and performs it.""" - def __init__(self, enforce_openssl_binary_usage=False): + def __init__(self, enforce_openssl_binary_usage: bool = False) -> None: self.broken = False self.use_openssl_binary = enforce_openssl_binary_usage or not ocsp @@ -215,7 +215,8 @@ def _check_ocsp_cryptography(cert_path: str, chain_path: str, url: str, timeout: return False -def _check_ocsp_response(response_ocsp, request_ocsp, issuer_cert, cert_path): +def _check_ocsp_response(response_ocsp: 'ocsp.OCSPResponse', request_ocsp: 'ocsp.OCSPRequest', + issuer_cert: x509.Certificate, cert_path: str) -> None: """Verify that the OCSP is valid for several criteria""" # Assert OCSP response corresponds to the certificate we are talking about if response_ocsp.serial_number != request_ocsp.serial_number: @@ -249,13 +250,14 @@ def _check_ocsp_response(response_ocsp, request_ocsp, issuer_cert, cert_path): raise AssertionError('param nextUpdate is in the past.') -def _check_ocsp_response_signature(response_ocsp, issuer_cert, cert_path): +def _check_ocsp_response_signature(response_ocsp: 'ocsp.OCSPResponse', + issuer_cert: x509.Certificate, cert_path: str) -> None: """Verify an OCSP response signature against certificate issuer or responder""" - def _key_hash(cert): + def _key_hash(cert: x509.Certificate) -> bytes: return x509.SubjectKeyIdentifier.from_public_key(cert.public_key()).digest - if response_ocsp.responder_name == issuer_cert.subject or \ - response_ocsp.responder_key_hash == _key_hash(issuer_cert): + if (response_ocsp.responder_name == issuer_cert.subject + or response_ocsp.responder_key_hash == _key_hash(issuer_cert)): # Case where the OCSP responder is also the certificate issuer logger.debug('OCSP response for certificate %s is signed by the certificate\'s issuer.', cert_path) @@ -289,21 +291,23 @@ def _check_ocsp_response_signature(response_ocsp, issuer_cert, cert_path): raise AssertionError('responder is not authorized by issuer to sign OCSP responses') # Following line may raise UnsupportedAlgorithm - chosen_hash = responder_cert.signature_hash_algorithm + chosen_cert_hash = responder_cert.signature_hash_algorithm # For a delegate OCSP responder, we need first check that its certificate is effectively # signed by the certificate issuer. crypto_util.verify_signed_payload(issuer_cert.public_key(), responder_cert.signature, - responder_cert.tbs_certificate_bytes, chosen_hash) + responder_cert.tbs_certificate_bytes, chosen_cert_hash) # Following line may raise UnsupportedAlgorithm - chosen_hash = response_ocsp.signature_hash_algorithm + chosen_response_hash = response_ocsp.signature_hash_algorithm # We check that the OSCP response is effectively signed by the responder # (an authorized delegate one or the certificate issuer itself). + if not chosen_response_hash: + raise AssertionError("no signature hash algorithm defined") crypto_util.verify_signed_payload(responder_cert.public_key(), response_ocsp.signature, - response_ocsp.tbs_response_bytes, chosen_hash) + response_ocsp.tbs_response_bytes, chosen_response_hash) -def _translate_ocsp_query(cert_path, ocsp_output, ocsp_errors): +def _translate_ocsp_query(cert_path: str, ocsp_output: str, ocsp_errors: str) -> bool: """Parse openssl's weird output to work out what it means.""" states = ("good", "revoked", "unknown") diff --git a/certbot/certbot/py.typed b/certbot/certbot/py.typed new file mode 100644 index 000000000..e69de29bb diff --git a/certbot/certbot/reverter.py b/certbot/certbot/reverter.py index edf7de49a..69a8ac83f 100644 --- a/certbot/certbot/reverter.py +++ b/certbot/certbot/reverter.py @@ -5,7 +5,13 @@ import logging import shutil import time import traceback +from typing import Iterable +from typing import List +from typing import Set +from typing import TextIO +from typing import Tuple +from certbot import configuration from certbot import errors from certbot import util from certbot._internal import constants @@ -57,13 +63,13 @@ class Reverter: :type config: :class:`certbot.configuration.NamespaceConfig` """ - def __init__(self, config): + def __init__(self, config: configuration.NamespaceConfig) -> None: self.config = config util.make_or_verify_dir( config.backup_dir, constants.CONFIG_DIRS_MODE, self.config.strict_permissions) - def revert_temporary_config(self): + def revert_temporary_config(self) -> None: """Reload users original configuration files after a temporary save. This function should reinstall the users original configuration files @@ -83,7 +89,7 @@ class Reverter: ) raise errors.ReverterError("Unable to revert temporary config") - def rollback_checkpoints(self, rollback=1): + def rollback_checkpoints(self, rollback: int = 1) -> None: """Revert 'rollback' number of configuration checkpoints. :param int rollback: Number of checkpoints to reverse. A str num will be @@ -125,7 +131,7 @@ class Reverter: "Unable to load checkpoint during rollback") rollback -= 1 - def add_to_temp_checkpoint(self, save_files, save_notes): + def add_to_temp_checkpoint(self, save_files: Set[str], save_notes: str) -> None: """Add files to temporary checkpoint. :param set save_files: set of filepaths to save @@ -135,7 +141,7 @@ class Reverter: self._add_to_checkpoint_dir( self.config.temp_checkpoint_dir, save_files, save_notes) - def add_to_checkpoint(self, save_files, save_notes): + def add_to_checkpoint(self, save_files: Set[str], save_notes: str) -> None: """Add files to a permanent checkpoint. :param set save_files: set of filepaths to save @@ -147,7 +153,7 @@ class Reverter: self._add_to_checkpoint_dir( self.config.in_progress_dir, save_files, save_notes) - def _add_to_checkpoint_dir(self, cp_dir, save_files, save_notes): + def _add_to_checkpoint_dir(self, cp_dir: str, save_files: Set[str], save_notes: str) -> None: """Add save files to checkpoint directory. :param str cp_dir: Checkpoint directory filepath @@ -192,7 +198,7 @@ class Reverter: with open(os.path.join(cp_dir, "CHANGES_SINCE"), "a") as notes_fd: notes_fd.write(save_notes) - def _read_and_append(self, filepath): + def _read_and_append(self, filepath: str) -> Tuple[TextIO, List[str]]: """Reads the file lines and returns a file obj. Read the file returning the lines, and a pointer to the end of the file. @@ -209,7 +215,7 @@ class Reverter: return op_fd, lines - def _recover_checkpoint(self, cp_dir): + def _recover_checkpoint(self, cp_dir: str) -> None: """Recover a specific checkpoint. Recover a specific checkpoint provided by cp_dir @@ -248,7 +254,7 @@ class Reverter: raise errors.ReverterError( "Unable to remove directory: %s" % cp_dir) - def _run_undo_commands(self, filepath): + def _run_undo_commands(self, filepath: str) -> None: """Run all commands in a file.""" # NOTE: csv module uses native strings. That is unicode on Python 3 # It is strongly advised to set newline = '' on Python 3 with CSV, @@ -263,7 +269,7 @@ class Reverter: logger.error( "Unable to run undo command: %s", " ".join(command)) - def _check_tempfile_saves(self, save_files): + def _check_tempfile_saves(self, save_files: Set[str]) -> None: """Verify save isn't overwriting any temporary files. :param set save_files: Set of files about to be saved. @@ -293,7 +299,7 @@ class Reverter: "Attempting to overwrite challenge " "file - %s" % filename) - def register_file_creation(self, temporary, *files): + def register_file_creation(self, temporary: bool, *files: str) -> None: r"""Register the creation of all files during certbot execution. Call this method before writing to the file to make sure that the @@ -332,7 +338,7 @@ class Reverter: if new_fd is not None: new_fd.close() - def register_undo_command(self, temporary, command): + def register_undo_command(self, temporary: bool, command: Iterable[str]) -> None: """Register a command to be run to undo actions taken. .. warning:: This function does not enforce order of operations in terms @@ -362,7 +368,7 @@ class Reverter: raise errors.ReverterError( "Unable to register undo command.") - def _get_cp_dir(self, temporary): + def _get_cp_dir(self, temporary: bool) -> str: """Return the proper reverter directory.""" if temporary: cp_dir = self.config.temp_checkpoint_dir @@ -374,7 +380,7 @@ class Reverter: return cp_dir - def recovery_routine(self): + def recovery_routine(self) -> None: """Revert configuration to most recent finalized checkpoint. Remove all changes (temporary and permanent) that have not been @@ -402,7 +408,7 @@ class Reverter: "Incomplete or failed recovery for IN_PROGRESS checkpoint " "- %s" % self.config.in_progress_dir) - def _remove_contained_files(self, file_list): + def _remove_contained_files(self, file_list: str) -> bool: """Erase all files contained within file_list. :param str file_list: file containing list of file paths to be deleted @@ -440,7 +446,7 @@ class Reverter: return True - def finalize_checkpoint(self, title): + def finalize_checkpoint(self, title: str) -> None: """Finalize the checkpoint. Timestamps and permanently saves all changes made through the use @@ -481,7 +487,7 @@ class Reverter: # rename the directory as a timestamp self._timestamp_progress_dir() - def _checkpoint_timestamp(self): + def _checkpoint_timestamp(self) -> str: "Determine the timestamp of the checkpoint, enforcing monotonicity." timestamp = str(time.time()) others = glob.glob(os.path.join(self.config.backup_dir, "[0-9]*")) @@ -502,7 +508,7 @@ class Reverter: timestamp = timetravel return timestamp - def _timestamp_progress_dir(self): + def _timestamp_progress_dir(self) -> None: """Timestamp the checkpoint.""" # It is possible save checkpoints faster than 1 per second resulting in # collisions in the naming convention. diff --git a/certbot/certbot/tests/acme_util.py b/certbot/certbot/tests/acme_util.py index 49ca88bf5..0c164ecb4 100644 --- a/certbot/certbot/tests/acme_util.py +++ b/certbot/certbot/tests/acme_util.py @@ -1,5 +1,7 @@ """ACME utilities for testing.""" import datetime +from typing import Iterable +from typing import Tuple import josepy as jose @@ -20,13 +22,13 @@ DNS01_2 = challenges.DNS01(token=b"cafecafecafecafecafecafe0feedbac") CHALLENGES = [HTTP01, DNS01] -def gen_combos(challbs): +def gen_combos(challbs: Iterable[messages.ChallengeBody]) -> Tuple[Tuple[int], ...]: """Generate natural combinations for challbs.""" # completing a single DV challenge satisfies the CA return tuple((i,) for i, _ in enumerate(challbs)) -def chall_to_challb(chall, status): +def chall_to_challb(chall: challenges.Challenge, status: messages.Status) -> messages.ChallengeBody: """Return ChallengeBody from Challenge.""" kwargs = { "chall": chall, @@ -56,7 +58,9 @@ DNS01_A_2 = auth_handler.challb_to_achall(DNS01_P_2, JWK, "esimerkki.example.org ACHALLENGES = [HTTP01_A, DNS01_A] -def gen_authzr(authz_status, domain, challs, statuses, combos=True): +def gen_authzr(authz_status: messages.Status, domain: str, challs: Iterable[challenges.Challenge], + statuses: Iterable[messages.Status], + combos: bool = True) -> messages.AuthorizationResource: """Generate an authorization resource. :param authz_status: Status object diff --git a/certbot/certbot/tests/util.py b/certbot/certbot/tests/util.py index bfd3e3fa3..b09a2d2a1 100644 --- a/certbot/certbot/tests/util.py +++ b/certbot/certbot/tests/util.py @@ -2,21 +2,26 @@ from importlib import reload as reload_module import io import logging -from multiprocessing import Event -from multiprocessing import Process +import multiprocessing +from multiprocessing import synchronize import shutil import sys import tempfile +from typing import Any +from typing import Callable +from typing import cast +from typing import IO from typing import Iterable from typing import List from typing import Optional +from typing import Union import unittest import warnings from cryptography.hazmat.backends import default_backend from cryptography.hazmat.primitives import serialization import josepy as jose -import OpenSSL +from OpenSSL import crypto import pkg_resources from certbot import configuration @@ -54,7 +59,8 @@ class DummyInstaller(common.Installer): fullchain_path: str) -> None: pass - def enhance(self, domain: str, enhancement: str, options: Optional[List[str]] = None) -> None: + def enhance(self, domain: str, enhancement: str, + options: Optional[Union[List[str], str]] = None) -> None: pass def supported_enhancements(self) -> List[str]: @@ -70,7 +76,7 @@ class DummyInstaller(common.Installer): pass @classmethod - def add_parser_arguments(cls, add): + def add_parser_arguments(cls, add: Callable[..., None]) -> None: pass def prepare(self) -> None: @@ -80,13 +86,13 @@ class DummyInstaller(common.Installer): pass -def vector_path(*names): +def vector_path(*names: str) -> str: """Path to a test vector.""" return pkg_resources.resource_filename( __name__, os.path.join('testdata', *names)) -def load_vector(*names): +def load_vector(*names: str) -> bytes: """Load contents of a test vector.""" # luckily, resource_string opens file in binary mode data = pkg_resources.resource_string( @@ -100,7 +106,7 @@ def load_vector(*names): return data -def _guess_loader(filename, loader_pem, loader_der): +def _guess_loader(filename: str, loader_pem: int, loader_der: int) -> int: _, ext = os.path.splitext(filename) if ext.lower() == '.pem': return loader_pem @@ -109,41 +115,45 @@ def _guess_loader(filename, loader_pem, loader_der): raise ValueError("Loader could not be recognized based on extension") # pragma: no cover -def load_cert(*names): +def load_cert(*names: str) -> crypto.X509: """Load certificate.""" loader = _guess_loader( - names[-1], OpenSSL.crypto.FILETYPE_PEM, OpenSSL.crypto.FILETYPE_ASN1) - return OpenSSL.crypto.load_certificate(loader, load_vector(*names)) + names[-1], crypto.FILETYPE_PEM, crypto.FILETYPE_ASN1) + return crypto.load_certificate(loader, load_vector(*names)) -def load_csr(*names): +def load_csr(*names: str) -> crypto.X509Req: """Load certificate request.""" loader = _guess_loader( - names[-1], OpenSSL.crypto.FILETYPE_PEM, OpenSSL.crypto.FILETYPE_ASN1) - return OpenSSL.crypto.load_certificate_request(loader, load_vector(*names)) + names[-1], crypto.FILETYPE_PEM, crypto.FILETYPE_ASN1) + return crypto.load_certificate_request(loader, load_vector(*names)) -def load_comparable_csr(*names): +def load_comparable_csr(*names: str) -> jose.ComparableX509: """Load ComparableX509 certificate request.""" return jose.ComparableX509(load_csr(*names)) -def load_rsa_private_key(*names): +def load_rsa_private_key(*names: str) -> jose.ComparableRSAKey: """Load RSA private key.""" - loader = _guess_loader(names[-1], serialization.load_pem_private_key, - serialization.load_der_private_key) - return jose.ComparableRSAKey(loader( + loader = _guess_loader(names[-1], crypto.FILETYPE_PEM, crypto.FILETYPE_ASN1) + loader_fn: Callable[..., Any] + if loader == crypto.FILETYPE_PEM: + loader_fn = serialization.load_pem_private_key + else: + loader_fn = serialization.load_der_private_key + return jose.ComparableRSAKey(loader_fn( load_vector(*names), password=None, backend=default_backend())) -def load_pyopenssl_private_key(*names): +def load_pyopenssl_private_key(*names: str) -> crypto.PKey: """Load pyOpenSSL private key.""" loader = _guess_loader( - names[-1], OpenSSL.crypto.FILETYPE_PEM, OpenSSL.crypto.FILETYPE_ASN1) - return OpenSSL.crypto.load_privatekey(loader, load_vector(*names)) + names[-1], crypto.FILETYPE_PEM, crypto.FILETYPE_ASN1) + return crypto.load_privatekey(loader, load_vector(*names)) -def make_lineage(config_dir, testfile, ec=False): +def make_lineage(config_dir: str, testfile: str, ec: bool = False) -> str: """Creates a lineage defined by testfile. This creates the archive, live, and renewal directories if @@ -151,6 +161,7 @@ def make_lineage(config_dir, testfile, ec=False): :param str config_dir: path to the configuration directory :param str testfile: configuration file to base the lineage on + :param bool ec: True if we generate the lineage with an ECDSA key :returns: path to the renewal conf file for the created lineage :rtype: str @@ -187,7 +198,7 @@ def make_lineage(config_dir, testfile, ec=False): return conf_path -def patch_get_utility(target='zope.component.getUtility'): +def patch_get_utility(target: str = 'zope.component.getUtility') -> mock.MagicMock: """Deprecated, patch certbot.display.util directly or use patch_display_util instead. :param str target: path to patch @@ -199,11 +210,11 @@ def patch_get_utility(target='zope.component.getUtility'): warnings.warn('Decorator certbot.tests.util.patch_get_utility is deprecated. You should now ' 'patch certbot.display.util yourself directly or use ' 'certbot.tests.util.patch_display_util as a temporary workaround.') - return mock.patch(target, new_callable=_create_display_util_mock) + return cast(mock.MagicMock, mock.patch(target, new_callable=_create_display_util_mock)) -def patch_get_utility_with_stdout(target='zope.component.getUtility', - stdout=None): +def patch_get_utility_with_stdout(target: str = 'zope.component.getUtility', + stdout: Optional[IO] = None) -> mock.MagicMock: """Deprecated, patch certbot.display.util directly or use patch_display_util_with_stdout instead. @@ -221,10 +232,10 @@ def patch_get_utility_with_stdout(target='zope.component.getUtility', 'workaround.') stdout = stdout if stdout else io.StringIO() freezable_mock = _create_display_util_mock_with_stdout(stdout) - return mock.patch(target, new=freezable_mock) + return cast(mock.MagicMock, mock.patch(target, new=freezable_mock)) -def patch_display_util(): +def patch_display_util() -> mock.MagicMock: """Patch certbot.display.util to use a special mock display utility. The mock display utility works like a regular mock object, except it also @@ -242,14 +253,15 @@ def patch_display_util(): :returns: patch on the function used internally by certbot.display.util to get a display utility instance - :rtype: unittest.mock._patch + :rtype: mock.MagicMock """ - return mock.patch('certbot._internal.display.obj.get_display', - new_callable=_create_display_util_mock) + return cast(mock.MagicMock, mock.patch('certbot._internal.display.obj.get_display', + new_callable=_create_display_util_mock)) -def patch_display_util_with_stdout(stdout=None): +def patch_display_util_with_stdout( + stdout: Optional[IO] = None) -> mock.MagicMock: """Patch certbot.display.util to use a special mock display utility. The mock display utility works like a regular mock object, except it also @@ -272,13 +284,13 @@ def patch_display_util_with_stdout(stdout=None): expected to have a `write` method :returns: patch on the function used internally by certbot.display.util to get a display utility instance - :rtype: unittest.mock._patch + :rtype: mock.MagicMock """ stdout = stdout if stdout else io.StringIO() - return mock.patch('certbot._internal.display.obj.get_display', - new=_create_display_util_mock_with_stdout(stdout)) + return cast(mock.MagicMock, mock.patch('certbot._internal.display.obj.get_display', + new=_create_display_util_mock_with_stdout(stdout))) class FreezableMock: @@ -294,7 +306,8 @@ class FreezableMock: value of func is ignored. """ - def __init__(self, frozen=False, func=None, return_value=mock.sentinel.DEFAULT): + def __init__(self, frozen: bool = False, func: Callable[..., Any] = None, + return_value: Any = mock.sentinel.DEFAULT) -> None: self._frozen_set = set() if frozen else {'freeze', } self._func = func self._mock = mock.MagicMock() @@ -302,16 +315,16 @@ class FreezableMock: self.return_value = return_value self._frozen = frozen - def freeze(self): + def freeze(self) -> None: """Freeze object preventing further changes.""" self._frozen = True - def __call__(self, *args, **kwargs): + def __call__(self, *args: Any, **kwargs: Any) -> mock.MagicMock: if self._func is not None: self._func(*args, **kwargs) return self._mock(*args, **kwargs) - def __getattribute__(self, name): + def __getattribute__(self, name: str) -> Any: if name == '_frozen': try: return object.__getattribute__(self, name) @@ -324,7 +337,7 @@ class FreezableMock: else: return getattr(object.__getattribute__(self, '_mock'), name) - def __setattr__(self, name, value): + def __setattr__(self, name: str, value: Any) -> None: """ Before it is frozen, attributes are set on the FreezableMock instance and added to the _frozen_set. Attributes in the _frozen_set cannot be changed after the FreezableMock is frozen. In this case, @@ -349,7 +362,7 @@ class FreezableMock: return object.__setattr__(self, name, value) -def _create_display_util_mock(): +def _create_display_util_mock() -> FreezableMock: display = FreezableMock() # Use pylint code for disable to keep on single line under line length limit method_list = [func for func in dir(display_obj.FileDisplay) @@ -363,14 +376,14 @@ def _create_display_util_mock(): return FreezableMock(frozen=True, return_value=display) -def _create_display_util_mock_with_stdout(stdout): - def _write_msg(message, *unused_args, **unused_kwargs): +def _create_display_util_mock_with_stdout(stdout: IO) -> FreezableMock: + def _write_msg(message: str, *unused_args: Any, **unused_kwargs: Any) -> None: """Write to message to stdout. """ if message: stdout.write(message) - def mock_method(*args, **kwargs): + def mock_method(*args: Any, **kwargs: Any) -> None: """ Mock function for display utility methods. """ @@ -394,7 +407,7 @@ def _create_display_util_mock_with_stdout(stdout): return FreezableMock(frozen=True, return_value=display) -def _assert_valid_call(*args, **kwargs): +def _assert_valid_call(*args: Any, **kwargs: Any) -> None: assert_args = [args[0] if args else kwargs['message']] assert_kwargs = {} @@ -408,11 +421,11 @@ def _assert_valid_call(*args, **kwargs): class TempDirTestCase(unittest.TestCase): """Base test class which sets up and tears down a temporary directory""" - def setUp(self): + def setUp(self) -> None: """Execute before test""" self.tempdir = tempfile.mkdtemp() - def tearDown(self): + def tearDown(self) -> None: """Execute after test""" # Cleanup opened resources after a test. This is usually done through atexit handlers in # Certbot, but during tests, atexit will not run registered functions before tearDown is @@ -429,7 +442,7 @@ class TempDirTestCase(unittest.TestCase): class ConfigTestCase(TempDirTestCase): """Test class which sets up a NamespaceConfig object.""" - def setUp(self): + def setUp(self) -> None: super().setUp() self.config = configuration.NamespaceConfig( mock.MagicMock(**constants.CLI_DEFAULTS) @@ -444,7 +457,7 @@ class ConfigTestCase(TempDirTestCase): self.config.namespace.server = "https://example.com" -def _handle_lock(event_in, event_out, path): +def _handle_lock(event_in: synchronize.Event, event_out: synchronize.Event, path: str) -> None: """ Acquire a file lock on given path, then wait to release it. This worker is coordinated using events to signal when the lock should be acquired and released. @@ -463,7 +476,7 @@ def _handle_lock(event_in, event_out, path): my_lock.release() -def lock_and_call(callback, path_to_lock): +def lock_and_call(callback: Callable[[], Any], path_to_lock: str) -> None: """ Grab a lock on path_to_lock from a foreign process then execute the callback. :param callable callback: object to call after acquiring the lock @@ -472,9 +485,10 @@ def lock_and_call(callback, path_to_lock): # Reload certbot.util module to reset internal _LOCKS dictionary. reload_module(util) - emit_event = Event() - receive_event = Event() - process = Process(target=_handle_lock, args=(emit_event, receive_event, path_to_lock)) + emit_event = multiprocessing.Event() + receive_event = multiprocessing.Event() + process = multiprocessing.Process(target=_handle_lock, + args=(emit_event, receive_event, path_to_lock)) process.start() # Wait confirmation that lock is acquired @@ -489,15 +503,15 @@ def lock_and_call(callback, path_to_lock): assert process.exitcode == 0 -def skip_on_windows(reason): +def skip_on_windows(reason: str) -> Callable[[Callable[..., Any]], Callable[..., Any]]: """Decorator to skip permanently a test on Windows. A reason is required.""" - def wrapper(function): + def wrapper(function: Callable[..., Any]) -> Callable[..., Any]: """Wrapped version""" return unittest.skipIf(sys.platform == 'win32', reason)(function) return wrapper -def temp_join(path): +def temp_join(path: str) -> str: """ Return the given path joined to the tempdir path for the current platform Eg.: 'cert' => /tmp/cert (Linux) or 'C:\\Users\\currentuser\\AppData\\Temp\\cert' (Windows) diff --git a/certbot/certbot/util.py b/certbot/certbot/util.py index 2b3e8cae5..50661f183 100644 --- a/certbot/certbot/util.py +++ b/certbot/certbot/util.py @@ -9,11 +9,15 @@ import re import socket import subprocess import sys +from typing import Any +from typing import Callable from typing import Dict from typing import IO from typing import List -from typing import Text +from typing import Optional +from typing import Set from typing import Tuple +from typing import TYPE_CHECKING from typing import Union import warnings @@ -29,6 +33,9 @@ _USE_DISTRO = sys.platform.startswith('linux') if _USE_DISTRO: import distro +if TYPE_CHECKING: + import distutils.version + logger = logging.getLogger(__name__) @@ -61,7 +68,7 @@ _INITIAL_PID = os.getpid() _LOCKS: Dict[str, lock.LockFile] = {} _VERSION_COMPONENT_RE = re.compile(r'(\d+ | [a-z]+ | \.)', re.VERBOSE) -def env_no_snap_for_external_calls(): +def env_no_snap_for_external_calls() -> Dict[str, str]: """ When Certbot is run inside a Snap, certain environment variables are modified. But Certbot sometimes calls out to external programs, @@ -86,7 +93,7 @@ def env_no_snap_for_external_calls(): return env -def run_script(params, log=logger.error): +def run_script(params: List[str], log: Callable[[str], None]=logger.error) -> Tuple[str, str]: """Run the script with the given params. :param list params: List of parameters to pass to subprocess.run @@ -116,7 +123,7 @@ def run_script(params, log=logger.error): return proc.stdout, proc.stderr -def exe_exists(exe): +def exe_exists(exe: Optional[str]) -> bool: """Determine whether path/name refers to an executable. :param str exe: Executable path or name @@ -125,6 +132,9 @@ def exe_exists(exe): :rtype: bool """ + if exe is None: + return False + path, _ = os.path.split(exe) if path: return filesystem.is_executable(exe) @@ -135,7 +145,7 @@ def exe_exists(exe): return False -def lock_dir_until_exit(dir_path): +def lock_dir_until_exit(dir_path: str) -> None: """Lock the directory at dir_path until program exit. :param str dir_path: path to directory @@ -150,7 +160,7 @@ def lock_dir_until_exit(dir_path): _LOCKS[dir_path] = lock.lock_dir(dir_path) -def _release_locks(): +def _release_locks() -> None: for dir_lock in _LOCKS.values(): try: dir_lock.release() @@ -160,7 +170,7 @@ def _release_locks(): _LOCKS.clear() -def set_up_core_dir(directory, mode, strict): +def set_up_core_dir(directory: str, mode: int, strict: bool) -> None: """Ensure directory exists with proper permissions and is locked. :param str directory: Path to a directory. @@ -179,7 +189,7 @@ def set_up_core_dir(directory, mode, strict): raise errors.Error(PERM_ERR_FMT.format(error)) -def make_or_verify_dir(directory, mode=0o755, strict=False): +def make_or_verify_dir(directory: str, mode: int = 0o755, strict: bool = False) -> None: """Make sure directory exists with proper permissions. :param str directory: Path to a directory. @@ -206,7 +216,7 @@ def make_or_verify_dir(directory, mode=0o755, strict=False): raise -def safe_open(path: str, mode: str = "w", chmod=None) -> IO: +def safe_open(path: str, mode: str = "w", chmod: Optional[int] = None) -> IO: """Safely open a file. :param str path: Path to a file. @@ -223,12 +233,12 @@ def safe_open(path: str, mode: str = "w", chmod=None) -> IO: return os.fdopen(fd, mode, *fdopen_args) -def _unique_file(path, filename_pat, count, chmod, mode): +def _unique_file(path: str, filename_pat: Callable[[int], str], count: int, + chmod: int, mode: str) -> Tuple[IO, str]: while True: current_path = os.path.join(path, filename_pat(count)) try: - return safe_open(current_path, chmod=chmod, mode=mode),\ - os.path.abspath(current_path) + return safe_open(current_path, chmod=chmod, mode=mode), os.path.abspath(current_path) except OSError as err: # "File exists," is okay, try a different name. if err.errno != errno.EEXIST: @@ -236,7 +246,7 @@ def _unique_file(path, filename_pat, count, chmod, mode): count += 1 -def unique_file(path, chmod=0o777, mode="w"): +def unique_file(path: str, chmod: int = 0o777, mode: str = "w") -> Tuple[IO, str]: """Safely finds a unique file. :param str path: path/filename.ext @@ -252,7 +262,8 @@ def unique_file(path, chmod=0o777, mode="w"): count=0, chmod=chmod, mode=mode) -def unique_lineage_name(path, filename, chmod=0o644, mode="w"): +def unique_lineage_name(path: str, filename: str, chmod: int = 0o644, + mode: str = "w") -> Tuple[IO, str]: """Safely finds a unique file using lineage convention. :param str path: directory path @@ -279,7 +290,7 @@ def unique_lineage_name(path, filename, chmod=0o644, mode="w"): count=1, chmod=chmod, mode=mode) -def safely_remove(path): +def safely_remove(path: str) -> None: """Remove a file that may not exist.""" try: os.remove(path) @@ -288,7 +299,7 @@ def safely_remove(path): raise -def get_filtered_names(all_names): +def get_filtered_names(all_names: Set[str]) -> Set[str]: """Removes names that aren't considered valid by Let's Encrypt. :param set all_names: all names found in the configuration @@ -305,7 +316,7 @@ def get_filtered_names(all_names): logger.debug('Not suggesting name "%s"', name, exc_info=True) return filtered_names -def get_os_info(): +def get_os_info() -> Tuple[str, str]: """ Get OS name and version @@ -315,7 +326,7 @@ def get_os_info(): return get_python_os_info(pretty=False) -def get_os_info_ua(): +def get_os_info_ua() -> str: """ Get OS name and version string for User Agent @@ -329,7 +340,7 @@ def get_os_info_ua(): return " ".join(get_python_os_info(pretty=True)) return os_info -def get_systemd_os_like(): +def get_systemd_os_like() -> List[str]: """ Get a list of strings that indicate the distribution likeness to other distributions. @@ -342,7 +353,7 @@ def get_systemd_os_like(): return distro.like().split(" ") return [] -def get_var_from_file(varname, filepath="/etc/os-release"): +def get_var_from_file(varname: str, filepath: str = "/etc/os-release") -> str: """ Get single value from a file formatted like systemd /etc/os-release @@ -364,14 +375,14 @@ def get_var_from_file(varname, filepath="/etc/os-release"): return _normalize_string(line.strip()[len(var_string):]) return "" -def _normalize_string(orig): +def _normalize_string(orig: str) -> str: """ Helper function for get_var_from_file() to remove quotes and whitespaces """ return orig.replace('"', '').replace("'", "").strip() -def get_python_os_info(pretty=False): +def get_python_os_info(pretty: bool = False) -> Tuple[str, str]: """ Get Operating System type/distribution and major version using python platform module @@ -430,7 +441,7 @@ def get_python_os_info(pretty=False): EMAIL_REGEX = re.compile("[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+$") -def safe_email(email): +def safe_email(email: str) -> bool: """Scrub email address before using it.""" if EMAIL_REGEX.match(email) is not None: return not email.startswith(".") and ".." not in email @@ -440,11 +451,13 @@ def safe_email(email): class DeprecatedArgumentAction(argparse.Action): """Action to log a warning when an argument is used.""" - def __call__(self, unused1, unused2, unused3, option_string=None): + def __call__(self, unused1: Any, unused2: Any, unused3: Any, + option_string: Optional[str] = None) -> None: warnings.warn("Use of %s is deprecated." % option_string, DeprecationWarning) -def add_deprecated_argument(add_argument, argument_name, nargs): +def add_deprecated_argument(add_argument: Callable[..., None], argument_name: str, + nargs: Union[str, int]) -> None: """Adds a deprecated argument with the name argument_name. Deprecated arguments are not shown in the help. If they are used on @@ -470,11 +483,11 @@ def add_deprecated_argument(add_argument, argument_name, nargs): help=argparse.SUPPRESS, nargs=nargs) -def enforce_le_validity(domain): +def enforce_le_validity(domain: str) -> str: """Checks that Let's Encrypt will consider domain to be valid. :param str domain: FQDN to check - :type domain: `str` or `unicode` + :type domain: `str` :returns: The domain cast to `str`, with ASCII-only contents :rtype: str :raises ConfigurationError: for invalid domains and cases where Let's @@ -503,12 +516,13 @@ def enforce_le_validity(domain): label, domain)) return domain -def enforce_domain_sanity(domain): + +def enforce_domain_sanity(domain: Union[str, bytes]) -> str: """Method which validates domain value and errors out if the requirements are not met. :param domain: Domain to check - :type domain: `str` or `unicode` + :type domain: `str` or `bytes` :raises ConfigurationError: for invalid domains and cases where Let's Encrypt currently will not issue certificates @@ -562,11 +576,11 @@ def enforce_domain_sanity(domain): return domain -def is_ipaddress(address): +def is_ipaddress(address: str) -> bool: """Is given address string form of IP(v4 or v6) address? :param address: address to check - :type address: `str` or `unicode` + :type address: `str` :returns: True if address is valid IP address, otherwise return False. :rtype: bool @@ -585,23 +599,22 @@ def is_ipaddress(address): return False -def is_wildcard_domain(domain): +def is_wildcard_domain(domain: Union[str, bytes]) -> bool: """"Is domain a wildcard domain? :param domain: domain to check - :type domain: `bytes` or `str` or `unicode` + :type domain: `bytes` or `str` :returns: True if domain is a wildcard, otherwise, False :rtype: bool """ - wildcard_marker: Union[Text, bytes] = b"*." if isinstance(domain, str): - wildcard_marker = "*." - return domain.startswith(wildcard_marker) + return domain.startswith("*.") + return domain.startswith(b"*.") -def get_strict_version(normalized): +def get_strict_version(normalized: str) -> "distutils.version.StrictVersion": """Converts a normalized version to a strict version. :param str normalized: normalized version string @@ -619,7 +632,7 @@ def get_strict_version(normalized): return distutils.version.StrictVersion(normalized.replace(".dev", "a")) -def is_staging(srv): +def is_staging(srv: str) -> bool: """ Determine whether a given ACME server is a known test / staging server. @@ -630,7 +643,7 @@ def is_staging(srv): return srv == constants.STAGING_URI or "staging" in srv -def atexit_register(func, *args, **kwargs): +def atexit_register(func: Callable, *args: Any, **kwargs: Any) -> None: """Sets func to be called before the program exits. Special care is taken to ensure func is only called when the process @@ -642,7 +655,7 @@ def atexit_register(func, *args, **kwargs): atexit.register(_atexit_call, func, *args, **kwargs) -def parse_loose_version(version_string): +def parse_loose_version(version_string: str) -> List[Union[int, str]]: """Parses a version string into its components. This code and the returned tuple is based on the now deprecated @@ -668,6 +681,6 @@ def parse_loose_version(version_string): return components -def _atexit_call(func, *args, **kwargs): +def _atexit_call(func: Callable, *args: Any, **kwargs: Any) -> None: if _INITIAL_PID == os.getpid(): func(*args, **kwargs)