From a0f22d21cec0b8ab29eadd9285bd876205df24c7 Mon Sep 17 00:00:00 2001 From: Adrien Ferrand Date: Mon, 25 Oct 2021 00:43:21 +0200 Subject: [PATCH] Add type annotations to the `acme` project (#9036) * Start more types * Second run * Work in progress * Types in all acme module * Various fixes * Various fixes * Final fixes * Disallow untyped defs for acme project * Fix coverage * Remote unecessary type ignore * Use Mapping instead of Dict as input whenever it is possible * Update acme/acme/client.py Co-authored-by: Brad Warren * Update acme/acme/client.py Co-authored-by: alexzorin * Various fixes * Fix code * Fix code * Update acme/acme/client.py Co-authored-by: Brad Warren * Update acme/acme/challenges.py Co-authored-by: Brad Warren * Update acme/acme/client.py Co-authored-by: Brad Warren * Fix deactivate_registration and simplify signature of update_registration * Do not leak personal data during account deactivation * Clean more Dicts * New fix to not leak contact field in the account deactivation payload. * Add ignore for python 3.6 type check * Revert "Add ignore for python 3.6 type check" This reverts commit da7338137b798e3ace34de15ed12f76ec3cf3888. * Let's find a smarter way than "type: ignore" * Update certbot/certbot/_internal/account.py Co-authored-by: alexzorin * Fix an annotation Co-authored-by: Brad Warren Co-authored-by: alexzorin --- acme/acme/challenges.py | 96 ++++---- acme/acme/client.py | 211 +++++++++++------- acme/acme/crypto_util.py | 85 ++++--- acme/acme/errors.py | 35 ++- acme/acme/fields.py | 17 +- acme/acme/jws.py | 13 +- acme/acme/magic_typing.py | 5 +- acme/acme/messages.py | 108 +++++---- acme/acme/mixins.py | 21 +- acme/acme/standalone.py | 67 +++--- acme/acme/util.py | 7 +- acme/tests/challenges_test.py | 4 +- .../certbot_apache/_internal/parser.py | 2 +- certbot/certbot/_internal/account.py | 8 +- certbot/certbot/_internal/client.py | 7 +- certbot/certbot/compat/_path.py | 2 +- certbot/certbot/compat/os.py | 2 +- certbot/certbot/plugins/dns_common_lexicon.py | 4 +- certbot/certbot/plugins/dns_test_common.py | 2 +- .../plugins/dns_test_common_lexicon.py | 2 +- mypy.ini | 5 +- tests/lock_test.py | 5 +- tox.ini | 8 +- 23 files changed, 431 insertions(+), 285 deletions(-) diff --git a/acme/acme/challenges.py b/acme/acme/challenges.py index 2737f9f22..de842a261 100644 --- a/acme/acme/challenges.py +++ b/acme/acme/challenges.py @@ -5,6 +5,12 @@ import functools import hashlib import logging import socket +from typing import cast +from typing import Any +from typing import Dict +from typing import Mapping +from typing import Optional +from typing import Tuple from typing import Type from cryptography.hazmat.primitives import hashes @@ -25,10 +31,10 @@ logger = logging.getLogger(__name__) class Challenge(jose.TypedJSONObjectWithFields): # _fields_to_partial_json """ACME challenge.""" - TYPES: dict = {} + TYPES: Dict[str, Type['Challenge']] = {} @classmethod - def from_json(cls, jobj): + def from_json(cls, jobj: Mapping[str, Any]) -> 'Challenge': try: return super().from_json(jobj) except jose.UnrecognizedTypeError as error: @@ -39,7 +45,7 @@ class Challenge(jose.TypedJSONObjectWithFields): class ChallengeResponse(ResourceMixin, TypeMixin, jose.TypedJSONObjectWithFields): # _fields_to_partial_json """ACME challenge response.""" - TYPES: dict = {} + TYPES: Dict[str, Type['ChallengeResponse']] = {} resource_type = 'challenge' resource = fields.Resource(resource_type) @@ -57,15 +63,15 @@ class UnrecognizedChallenge(Challenge): """ - def __init__(self, jobj): + def __init__(self, jobj: Mapping[str, Any]) -> None: super().__init__() object.__setattr__(self, "jobj", jobj) - def to_partial_json(self): + def to_partial_json(self) -> Dict[str, Any]: return self.jobj # pylint: disable=no-member @classmethod - def from_json(cls, jobj): + def from_json(cls, jobj: Mapping[str, Any]) -> 'UnrecognizedChallenge': return cls(jobj) @@ -79,13 +85,13 @@ class _TokenChallenge(Challenge): """Minimum size of the :attr:`token` in bytes.""" # TODO: acme-spec doesn't specify token as base64-encoded value - token = jose.Field( + token: bytes = jose.Field( "token", encoder=jose.encode_b64jose, decoder=functools.partial( jose.decode_b64jose, size=TOKEN_SIZE, minimum=True)) # XXX: rename to ~token_good_for_url @property - def good_token(self): # XXX: @token.decoder + def good_token(self) -> bool: # XXX: @token.decoder """Is `token` good? .. todo:: acme-spec wants "It MUST NOT contain any non-ASCII @@ -108,7 +114,7 @@ class KeyAuthorizationChallengeResponse(ChallengeResponse): key_authorization = jose.Field("keyAuthorization") thumbprint_hash_function = hashes.SHA256 - def verify(self, chall, account_public_key): + def verify(self, chall: 'KeyAuthorizationChallenge', account_public_key: jose.JWK) -> bool: """Verify the key authorization. :param KeyAuthorization chall: Challenge that corresponds to @@ -140,7 +146,7 @@ class KeyAuthorizationChallengeResponse(ChallengeResponse): return True - def to_partial_json(self): + def to_partial_json(self) -> Dict[str, Any]: jobj = super().to_partial_json() jobj.pop('keyAuthorization', None) return jobj @@ -158,7 +164,7 @@ class KeyAuthorizationChallenge(_TokenChallenge, metaclass=abc.ABCMeta): thumbprint_hash_function = ( KeyAuthorizationChallengeResponse.thumbprint_hash_function) - def key_authorization(self, account_key): + def key_authorization(self, account_key: jose.JWK) -> str: """Generate Key Authorization. :param JWK account_key: @@ -169,7 +175,7 @@ class KeyAuthorizationChallenge(_TokenChallenge, metaclass=abc.ABCMeta): account_key.thumbprint( hash_function=self.thumbprint_hash_function)).decode() - def response(self, account_key): + def response(self, account_key: jose.JWK) -> KeyAuthorizationChallengeResponse: """Generate response to the challenge. :param JWK account_key: @@ -182,7 +188,7 @@ class KeyAuthorizationChallenge(_TokenChallenge, metaclass=abc.ABCMeta): key_authorization=self.key_authorization(account_key)) @abc.abstractmethod - def validation(self, account_key, **kwargs): + def validation(self, account_key: jose.JWK, **kwargs: Any) -> Any: """Generate validation for the challenge. Subclasses must implement this method, but they are likely to @@ -196,7 +202,8 @@ class KeyAuthorizationChallenge(_TokenChallenge, metaclass=abc.ABCMeta): """ raise NotImplementedError() # pragma: no cover - def response_and_validation(self, account_key, *args, **kwargs): + def response_and_validation(self, account_key: jose.JWK, *args: Any, **kwargs: Any + ) -> Tuple[KeyAuthorizationChallengeResponse, Any]: """Generate response and validation. Convenience function that return results of `response` and @@ -215,7 +222,7 @@ class DNS01Response(KeyAuthorizationChallengeResponse): """ACME dns-01 challenge response.""" typ = "dns-01" - def simple_verify(self, chall, domain, account_public_key): # pylint: disable=unused-argument + def simple_verify(self, chall: 'DNS01', domain: str, account_public_key: jose.JWK) -> bool: # pylint: disable=unused-argument """Simple verify. This method no longer checks DNS records and is a simple wrapper @@ -246,7 +253,7 @@ class DNS01(KeyAuthorizationChallenge): LABEL = "_acme-challenge" """Label clients prepend to the domain name being validated.""" - def validation(self, account_key, **unused_kwargs): + def validation(self, account_key: jose.JWK, **unused_kwargs: Any) -> str: """Generate validation. :param JWK account_key: @@ -256,7 +263,7 @@ class DNS01(KeyAuthorizationChallenge): return jose.b64encode(hashlib.sha256(self.key_authorization( account_key).encode("utf-8")).digest()).decode() - def validation_domain_name(self, name): + def validation_domain_name(self, name: str) -> str: """Domain name for TXT validation record. :param unicode name: Domain name being validated. @@ -281,7 +288,8 @@ class HTTP01Response(KeyAuthorizationChallengeResponse): WHITESPACE_CUTSET = "\n\r\t " """Whitespace characters which should be ignored at the end of the body.""" - def simple_verify(self, chall, domain, account_public_key, port=None): + def simple_verify(self, chall: 'HTTP01', domain: str, account_public_key: jose.JWK, + port: Optional[int] = None) -> bool: """Simple verify. :param challenges.SimpleHTTP chall: Corresponding challenge. @@ -346,7 +354,7 @@ class HTTP01(KeyAuthorizationChallenge): """URI root path for the server provisioned resource.""" @property - def path(self): + def path(self) -> str: """Path (starting with '/') for provisioned resource. :rtype: string @@ -354,7 +362,7 @@ class HTTP01(KeyAuthorizationChallenge): """ return '/' + self.URI_ROOT_PATH + '/' + self.encode('token') - def uri(self, domain): + def uri(self, domain: str) -> str: """Create an URI to the provisioned resource. Forms an URI to the HTTPS server provisioned resource @@ -366,7 +374,7 @@ class HTTP01(KeyAuthorizationChallenge): """ return "http://" + domain + self.path - def validation(self, account_key, **unused_kwargs): + def validation(self, account_key: jose.JWK, **unused_kwargs: Any) -> str: """Generate validation. :param JWK account_key: @@ -393,11 +401,12 @@ class TLSALPN01Response(KeyAuthorizationChallengeResponse): ACME_TLS_1_PROTOCOL = "acme-tls/1" @property - def h(self): + def h(self) -> bytes: """Hash value stored in challenge certificate""" return hashlib.sha256(self.key_authorization.encode('utf-8')).digest() - def gen_cert(self, domain, key=None, bits=2048): + def gen_cert(self, domain: str, key: Optional[crypto.PKey] = None, bits: int = 2048 + ) -> Tuple[crypto.X509, crypto.PKey]: """Generate tls-alpn-01 certificate. :param unicode domain: Domain verified by the challenge. @@ -413,15 +422,15 @@ class TLSALPN01Response(KeyAuthorizationChallengeResponse): key = crypto.PKey() key.generate_key(crypto.TYPE_RSA, bits) - der_value = b"DER:" + codecs.encode(self.h, 'hex') acme_extension = crypto.X509Extension(self.ID_PE_ACME_IDENTIFIER_V1, - critical=True, value=der_value) + critical=True, value=der_value) return crypto_util.gen_ss_cert(key, [domain], force_san=True, - extensions=[acme_extension]), key + extensions=[acme_extension]), key - def probe_cert(self, domain, host=None, port=None): + def probe_cert(self, domain: str, host: Optional[str] = None, + port: Optional[int] = None) -> crypto.X509: """Probe tls-alpn-01 challenge certificate. :param unicode domain: domain being validated, required. @@ -435,10 +444,10 @@ class TLSALPN01Response(KeyAuthorizationChallengeResponse): if port is None: port = self.PORT - return crypto_util.probe_sni(host=host, port=port, name=domain, - alpn_protocols=[self.ACME_TLS_1_PROTOCOL]) + return crypto_util.probe_sni(host=host.encode(), port=port, name=domain.encode(), + alpn_protocols=[self.ACME_TLS_1_PROTOCOL]) - def verify_cert(self, domain, cert): + def verify_cert(self, domain: str, cert: crypto.X509) -> bool: """Verify tls-alpn-01 challenge certificate. :param unicode domain: Domain name being validated. @@ -450,7 +459,10 @@ class TLSALPN01Response(KeyAuthorizationChallengeResponse): """ # pylint: disable=protected-access names = crypto_util._pyopenssl_cert_or_req_all_names(cert) - logger.debug('Certificate %s. SANs: %s', cert.digest('sha256'), names) + # Type ignore needed due to + # https://github.com/pyca/pyopenssl/issues/730. + logger.debug('Certificate %s. SANs: %s', + cert.digest('sha256'), names) # type: ignore[arg-type] if len(names) != 1 or names[0].lower() != domain.lower(): return False @@ -465,8 +477,9 @@ class TLSALPN01Response(KeyAuthorizationChallengeResponse): return False # pylint: disable=too-many-arguments - def simple_verify(self, chall, domain, account_public_key, - cert=None, host=None, port=None): + def simple_verify(self, chall: 'TLSALPN01', domain: str, account_public_key: jose.JWK, + cert: Optional[crypto.X509] = None, host: Optional[str] = None, + port: Optional[int] = None) -> bool: """Simple verify. Verify ``validation`` using ``account_public_key``, optionally @@ -506,7 +519,7 @@ class TLSALPN01(KeyAuthorizationChallenge): response_cls = TLSALPN01Response typ = response_cls.typ - def validation(self, account_key, **kwargs): + def validation(self, account_key: jose.JWK, **kwargs: Any) -> Tuple[crypto.X509, crypto.PKey]: """Generate validation. :param JWK account_key: @@ -523,7 +536,7 @@ class TLSALPN01(KeyAuthorizationChallenge): domain=kwargs.get('domain')) @staticmethod - def is_supported(): + def is_supported() -> bool: """ Check if TLS-ALPN-01 challenge is supported on this machine. This implies that a recent version of OpenSSL is installed (>= 1.0.2), @@ -545,7 +558,8 @@ class DNS(_TokenChallenge): LABEL = "_acme-challenge" """Label clients prepend to the domain name being validated.""" - def gen_validation(self, account_key, alg=jose.RS256, **kwargs): + def gen_validation(self, account_key: jose.JWK, alg: jose.JWASignature = jose.RS256, + **kwargs: Any) -> jose.JWS: """Generate validation. :param .JWK account_key: Private account key. @@ -559,7 +573,7 @@ class DNS(_TokenChallenge): payload=self.json_dumps(sort_keys=True).encode('utf-8'), key=account_key, alg=alg, **kwargs) - def check_validation(self, validation, account_public_key): + def check_validation(self, validation: jose.JWS, account_public_key: jose.JWK) -> bool: """Check validation. :param JWS validation: @@ -576,7 +590,7 @@ class DNS(_TokenChallenge): logger.debug("Checking validation for DNS failed: %s", error) return False - def gen_response(self, account_key, **kwargs): + def gen_response(self, account_key: jose.JWK, **kwargs: Any) -> 'DNSResponse': """Generate response. :param .JWK account_key: Private account key. @@ -588,7 +602,7 @@ class DNS(_TokenChallenge): return DNSResponse(validation=self.gen_validation( account_key, **kwargs)) - def validation_domain_name(self, name): + def validation_domain_name(self, name: str) -> str: """Domain name for TXT validation record. :param unicode name: Domain name being validated. @@ -608,7 +622,7 @@ class DNSResponse(ChallengeResponse): validation = jose.Field("validation", decoder=jose.JWS.from_json) - def check_validation(self, chall, account_public_key): + def check_validation(self, chall: 'DNS', account_public_key: jose.JWK) -> bool: """Check validation. :param challenges.DNS chall: @@ -617,4 +631,4 @@ class DNSResponse(ChallengeResponse): :rtype: bool """ - return chall.check_validation(self.validation, account_public_key) + return chall.check_validation(cast(jose.JWS, self.validation), account_public_key) diff --git a/acme/acme/client.py b/acme/acme/client.py index 7ab1eb657..ae2f261aa 100644 --- a/acme/acme/client.py +++ b/acme/acme/client.py @@ -13,11 +13,16 @@ import re import sys import time from types import ModuleType +from typing import Any +from typing import Callable from typing import cast from typing import Dict +from typing import Iterable from typing import List +from typing import Optional from typing import Set from typing import Text +from typing import Tuple from typing import Union import warnings @@ -48,7 +53,8 @@ class ClientBase: :ivar .ClientNetwork net: Client network. :ivar int acme_version: ACME protocol version. 1 or 2. """ - def __init__(self, directory, net, acme_version): + def __init__(self, directory: messages.Directory, net: 'ClientNetwork', + acme_version: int) -> None: """Initialize. :param .messages.Directory directory: Directory Resource @@ -60,7 +66,9 @@ class ClientBase: self.acme_version = acme_version @classmethod - def _regr_from_response(cls, response, uri=None, terms_of_service=None): + def _regr_from_response(cls, response: requests.Response, uri: Optional[str] = None, + terms_of_service: Optional[str] = None + ) -> messages.RegistrationResource: if 'terms-of-service' in response.links: terms_of_service = response.links['terms-of-service']['url'] @@ -69,7 +77,8 @@ class ClientBase: uri=response.headers.get('Location', uri), terms_of_service=terms_of_service) - def _send_recv_regr(self, regr, body): + def _send_recv_regr(self, regr: messages.RegistrationResource, + body: messages.Registration) -> messages.RegistrationResource: response = self._post(regr.uri, body) # TODO: Boulder returns httplib.ACCEPTED @@ -82,7 +91,7 @@ class ClientBase: response, uri=regr.uri, terms_of_service=regr.terms_of_service) - def _post(self, *args, **kwargs): + def _post(self, *args: Any, **kwargs: Any) -> requests.Response: """Wrapper around self.net.post that adds the acme_version. """ @@ -91,7 +100,9 @@ class ClientBase: kwargs.setdefault('new_nonce_url', getattr(self.directory, 'newNonce')) return self.net.post(*args, **kwargs) - def update_registration(self, regr, update=None): + def update_registration(self, regr: messages.RegistrationResource, + update: Optional[messages.Registration] = None + ) -> messages.RegistrationResource: """Update registration. :param messages.RegistrationResource regr: Registration Resource. @@ -108,7 +119,8 @@ class ClientBase: self.net.account = updated_regr return updated_regr - def deactivate_registration(self, regr): + def deactivate_registration(self, regr: messages.RegistrationResource + ) -> messages.RegistrationResource: """Deactivate registration. :param messages.RegistrationResource regr: The Registration Resource @@ -118,7 +130,8 @@ class ClientBase: :rtype: `.RegistrationResource` """ - return self.update_registration(regr, update={'status': 'deactivated'}) + return self.update_registration(regr, messages.Registration.from_json( + {"status": "deactivated", "contact": None})) def deactivate_authorization(self, authzr: messages.AuthorizationResource @@ -137,7 +150,9 @@ class ClientBase: return self._authzr_from_response(response, authzr.body.identifier, authzr.uri) - def _authzr_from_response(self, response, identifier=None, uri=None): + def _authzr_from_response(self, response: requests.Response, + identifier: Optional[messages.Identifier] = None, + uri: Optional[str] = None) -> messages.AuthorizationResource: authzr = messages.AuthorizationResource( body=messages.Authorization.from_json(response.json()), uri=response.headers.get('Location', uri)) @@ -145,7 +160,8 @@ class ClientBase: raise errors.UnexpectedUpdate(authzr) return authzr - def answer_challenge(self, challb, response): + def answer_challenge(self, challb: messages.ChallengeBody, response: requests.Response + ) -> messages.ChallengeResource: """Answer challenge. :param challb: Challenge Resource body. @@ -174,7 +190,7 @@ class ClientBase: return challr @classmethod - def retry_after(cls, response, default): + def retry_after(cls, response: requests.Response, default: int) -> datetime.datetime: """Compute next `poll` time based on response ``Retry-After`` header. Handles integers and various datestring formats per @@ -205,7 +221,7 @@ class ClientBase: return datetime.datetime.now() + datetime.timedelta(seconds=seconds) - def _revoke(self, cert, rsn, url): + def _revoke(self, cert: jose.ComparableX509, rsn: int, url: str) -> None: """Revoke certificate. :param .ComparableX509 cert: `OpenSSL.crypto.X509` wrapped in @@ -247,8 +263,9 @@ class Client(ClientBase): """ - def __init__(self, directory, key, alg=jose.RS256, verify_ssl=True, - net=None): + def __init__(self, directory: messages.Directory, key: jose.JWK, + alg: jose.JWASignature=jose.RS256, verify_ssl: bool = True, + net: Optional['ClientNetwork'] = None) -> None: """Initialize. :param directory: Directory Resource (`.messages.Directory`) or @@ -263,9 +280,10 @@ class Client(ClientBase): directory = messages.Directory.from_json( net.get(directory).json()) super().__init__(directory=directory, - net=net, acme_version=1) + net=net, acme_version=1) - def register(self, new_reg=None): + def register(self, new_reg: Optional[messages.NewRegistration] = None + ) -> messages.RegistrationResource: """Register. :param .NewRegistration new_reg: @@ -282,16 +300,18 @@ class Client(ClientBase): # "Instance of 'Field' has no key/contact member" bug: return self._regr_from_response(response) - def query_registration(self, regr): + def query_registration(self, regr: messages.RegistrationResource + ) -> messages.RegistrationResource: """Query server about registration. - :param messages.RegistrationResource: Existing Registration + :param messages.RegistrationResource regr: Existing Registration Resource. """ return self._send_recv_regr(regr, messages.UpdateRegistration()) - def agree_to_tos(self, regr): + def agree_to_tos(self, regr: messages.RegistrationResource + ) -> messages.RegistrationResource: """Agree to the terms-of-service. Agree to the terms-of-service in a Registration Resource. @@ -306,7 +326,8 @@ class Client(ClientBase): return self.update_registration( regr.update(body=regr.body.update(agreement=regr.terms_of_service))) - def request_challenges(self, identifier, new_authzr_uri=None): + def request_challenges(self, identifier: messages.Identifier, + new_authzr_uri: Optional[str] = None) -> messages.AuthorizationResource: """Request challenges. :param .messages.Identifier identifier: Identifier to be challenged. @@ -332,7 +353,8 @@ class Client(ClientBase): assert response.status_code == http_client.CREATED return self._authzr_from_response(response, identifier) - def request_domain_challenges(self, domain, new_authzr_uri=None): + def request_domain_challenges(self, domain: str,new_authzr_uri: Optional[str] = None + ) -> messages.AuthorizationResource: """Request challenges for domain names. This is simply a convenience function that wraps around @@ -352,7 +374,9 @@ class Client(ClientBase): return self.request_challenges(messages.Identifier( typ=messages.IDENTIFIER_FQDN, value=domain), new_authzr_uri) - def request_issuance(self, csr, authzrs): + def request_issuance(self, csr: jose.ComparableX509, + authzrs: Iterable[messages.AuthorizationResource] + ) -> messages.CertificateResource: """Request issuance. :param csr: CSR @@ -389,7 +413,8 @@ class Client(ClientBase): body=jose.ComparableX509(OpenSSL.crypto.load_certificate( OpenSSL.crypto.FILETYPE_ASN1, response.content))) - def poll(self, authzr): + def poll(self, authzr: messages.AuthorizationResource + ) -> Tuple[messages.AuthorizationResource, requests.Response]: """Poll Authorization Resource for status. :param authzr: Authorization Resource @@ -405,8 +430,11 @@ class Client(ClientBase): response, authzr.body.identifier, authzr.uri) return updated_authzr, response - def poll_and_request_issuance( - self, csr, authzrs, mintime=5, max_attempts=10): + def poll_and_request_issuance(self, csr: jose.ComparableX509, + authzrs: Iterable[messages.AuthorizationResource], + mintime: int = 5, max_attempts: int = 10 + ) -> Tuple[messages.CertificateResource, + Tuple[messages.AuthorizationResource, ...]]: """Poll and request issuance. This function polls all provided Authorization Resource URIs @@ -480,7 +508,7 @@ class Client(ClientBase): updated_authzrs = tuple(updated[authzr] for authzr in authzrs) return self.request_issuance(csr, updated_authzrs), updated_authzrs - def _get_cert(self, uri): + def _get_cert(self, uri: str) -> Tuple[requests.Response, jose.ComparableX509]: """Returns certificate from URI. :param str uri: URI of certificate @@ -496,7 +524,7 @@ class Client(ClientBase): return response, jose.ComparableX509(OpenSSL.crypto.load_certificate( OpenSSL.crypto.FILETYPE_ASN1, response.content)) - def check_cert(self, certr): + def check_cert(self, certr: messages.CertificateResource) -> messages.CertificateResource: """Check for new cert. :param certr: Certificate Resource @@ -515,7 +543,7 @@ class Client(ClientBase): raise errors.UnexpectedUpdate(response.text) return certr.update(body=cert) - def refresh(self, certr): + def refresh(self, certr: messages.CertificateResource) -> messages.CertificateResource: """Refresh certificate. :param certr: Certificate Resource @@ -530,7 +558,8 @@ class Client(ClientBase): # respond with status code 403 (Forbidden) return self.check_cert(certr) - def fetch_chain(self, certr, max_length=10): + def fetch_chain(self, certr: messages.CertificateResource, + max_length: int = 10) -> List[jose.ComparableX509]: """Fetch chain for certificate. :param .CertificateResource certr: Certificate Resource @@ -559,7 +588,7 @@ class Client(ClientBase): "Recursion limit reached. Didn't get {0}".format(uri)) return chain - def revoke(self, cert, rsn): + def revoke(self, cert: jose.ComparableX509, rsn: int) -> None: """Revoke certificate. :param .ComparableX509 cert: `OpenSSL.crypto.X509` wrapped in @@ -570,7 +599,7 @@ class Client(ClientBase): :raises .ClientError: If revocation is unsuccessful. """ - return self._revoke(cert, rsn, self.directory[messages.Revocation]) + self._revoke(cert, rsn, self.directory[cast(str, messages.Revocation)]) class ClientV2(ClientBase): @@ -580,16 +609,15 @@ class ClientV2(ClientBase): :ivar .ClientNetwork net: Client network. """ - def __init__(self, directory, net): + def __init__(self, directory: messages.Directory, net: 'ClientNetwork') -> None: """Initialize. :param .messages.Directory directory: Directory Resource :param .ClientNetwork net: Client network. """ - super().__init__(directory=directory, - net=net, acme_version=2) + super().__init__(directory=directory, net=net, acme_version=2) - def new_account(self, new_account): + def new_account(self, new_account: messages.NewRegistration) -> messages.RegistrationResource: """Register. :param .NewRegistration new_account: @@ -602,16 +630,17 @@ class ClientV2(ClientBase): response = self._post(self.directory['newAccount'], new_account) # if account already exists if response.status_code == 200 and 'Location' in response.headers: - raise errors.ConflictError(response.headers.get('Location')) + raise errors.ConflictError(response.headers['Location']) # "Instance of 'Field' has no key/contact member" bug: regr = self._regr_from_response(response) self.net.account = regr return regr - def query_registration(self, regr): + def query_registration(self, regr: messages.RegistrationResource + ) -> messages.RegistrationResource: """Query server about registration. - :param messages.RegistrationResource: Existing Registration + :param messages.RegistrationResource regr: Existing Registration Resource. """ @@ -623,7 +652,9 @@ class ClientV2(ClientBase): terms_of_service=regr.terms_of_service) return self.net.account - def update_registration(self, regr, update=None): + def update_registration(self, regr: messages.RegistrationResource, + update: Optional[messages.Registration] = None + ) -> messages.RegistrationResource: """Update registration. :param messages.RegistrationResource regr: Registration Resource. @@ -638,7 +669,7 @@ class ClientV2(ClientBase): new_regr = self._get_v2_account(regr) return super().update_registration(new_regr, update) - def _get_v2_account(self, regr): + def _get_v2_account(self, regr: messages.RegistrationResource) -> messages.RegistrationResource: self.net.account = None only_existing_reg = regr.body.update(only_return_existing=True) response = self._post(self.directory['newAccount'], only_existing_reg) @@ -647,10 +678,10 @@ class ClientV2(ClientBase): self.net.account = new_regr return new_regr - def new_order(self, csr_pem): + def new_order(self, csr_pem: bytes) -> messages.OrderResource: """Request a new Order object from the server. - :param str csr_pem: A CSR in PEM format. + :param bytes csr_pem: A CSR in PEM format. :returns: The newly created order. :rtype: OrderResource @@ -682,7 +713,8 @@ class ClientV2(ClientBase): authorizations=authorizations, csr_pem=csr_pem) - def poll(self, authzr): + def poll(self, authzr: messages.AuthorizationResource + ) -> Tuple[messages.AuthorizationResource, requests.Response]: """Poll Authorization Resource for status. :param authzr: Authorization Resource @@ -698,7 +730,8 @@ class ClientV2(ClientBase): response, authzr.body.identifier, authzr.uri) return updated_authzr, response - def poll_and_finalize(self, orderr, deadline=None): + def poll_and_finalize(self, orderr: messages.OrderResource, + deadline: Optional[datetime.datetime] = None) -> messages.OrderResource: """Poll authorizations and finalize the order. If no deadline is provided, this method will timeout after 90 @@ -716,7 +749,8 @@ class ClientV2(ClientBase): orderr = self.poll_authorizations(orderr, deadline) return self.finalize_order(orderr, deadline) - def poll_authorizations(self, orderr, deadline): + def poll_authorizations(self, orderr: messages.OrderResource, deadline: datetime.datetime + ) -> messages.OrderResource: """Poll Order Resource for status.""" responses = [] for url in orderr.body.authorizations: @@ -740,7 +774,8 @@ class ClientV2(ClientBase): raise errors.ValidationError(failed) return orderr.update(authorizations=responses) - def finalize_order(self, orderr, deadline, fetch_alternative_chains=False): + def finalize_order(self, orderr: messages.OrderResource, deadline: datetime.datetime, + fetch_alternative_chains: bool = False) -> messages.OrderResource: """Finalize an order and obtain a certificate. :param messages.OrderResource orderr: order to finalize @@ -772,7 +807,7 @@ class ClientV2(ClientBase): return orderr raise errors.TimeoutError() - def revoke(self, cert, rsn): + def revoke(self, cert: jose.ComparableX509, rsn: int) -> None: """Revoke certificate. :param .ComparableX509 cert: `OpenSSL.crypto.X509` wrapped in @@ -783,13 +818,13 @@ class ClientV2(ClientBase): :raises .ClientError: If revocation is unsuccessful. """ - return self._revoke(cert, rsn, self.directory['revokeCert']) + self._revoke(cert, rsn, self.directory['revokeCert']) - def external_account_required(self): + def external_account_required(self) -> bool: """Checks if ACME server requires External Account Binding authentication.""" return hasattr(self.directory, 'meta') and self.directory.meta.external_account_required - def _post_as_get(self, *args, **kwargs): + def _post_as_get(self, *args: Any, **kwargs: Any) -> requests.Response: """ Send GET request using the POST-as-GET protocol. :param args: @@ -799,7 +834,7 @@ class ClientV2(ClientBase): new_args = args[:1] + (None,) + args[1:] return self._post(*new_args, **kwargs) - def _get_links(self, response, relation_type): + def _get_links(self, response: requests.Response, relation_type: str) -> List[str]: """ Retrieves all Link URIs of relation_type from the response. :param requests.Response response: The requests HTTP response. @@ -836,7 +871,7 @@ class BackwardsCompatibleClientV2: :ivar .ClientBase client: either Client or ClientV2 """ - def __init__(self, net, key, server): + def __init__(self, net: 'ClientNetwork', key: jose.JWK, server: str) -> None: directory = messages.Directory.from_json(net.get(server).json()) self.acme_version = self._acme_version_from_directory(directory) self.client: Union[Client, ClientV2] @@ -845,17 +880,19 @@ class BackwardsCompatibleClientV2: else: self.client = ClientV2(directory, net=net) - def __getattr__(self, name): + def __getattr__(self, name: str) -> Any: return getattr(self.client, name) - def new_account_and_tos(self, regr, check_tos_cb=None): + def new_account_and_tos(self, regr: messages.NewRegistration, + check_tos_cb: Optional[Callable[[str], None]] = None + ) -> messages.RegistrationResource: """Combined register and agree_tos for V1, new_account for V2 :param .NewRegistration regr: :param callable check_tos_cb: callback that raises an error if the check does not work """ - def _assess_tos(tos): + def _assess_tos(tos: str) -> None: if check_tos_cb is not None: check_tos_cb(tos) if self.acme_version == 1: @@ -872,13 +909,13 @@ class BackwardsCompatibleClientV2: regr = regr.update(terms_of_service_agreed=True) return client_v2.new_account(regr) - def new_order(self, csr_pem): + def new_order(self, csr_pem: bytes) -> messages.OrderResource: """Request a new Order object from the server. If using ACMEv1, returns a dummy OrderResource with only the authorizations field filled in. - :param str csr_pem: A CSR in PEM format. + :param bytes csr_pem: A CSR in PEM format. :returns: The newly created order. :rtype: OrderResource @@ -898,7 +935,8 @@ class BackwardsCompatibleClientV2: return messages.OrderResource(authorizations=authorizations, csr_pem=csr_pem) return cast(ClientV2, self.client).new_order(csr_pem) - def finalize_order(self, orderr, deadline, fetch_alternative_chains=False): + def finalize_order(self, orderr: messages.OrderResource, deadline: datetime.datetime, + fetch_alternative_chains: bool = False) -> messages.OrderResource: """Finalize an order and obtain a certificate. :param messages.OrderResource orderr: order to finalize @@ -933,13 +971,13 @@ class BackwardsCompatibleClientV2: cert = OpenSSL.crypto.dump_certificate( OpenSSL.crypto.FILETYPE_PEM, certr.body.wrapped).decode() - chain = crypto_util.dump_pyopenssl_chain(chain).decode() + chain_str = crypto_util.dump_pyopenssl_chain(chain).decode() - return orderr.update(fullchain_pem=(cert + chain)) + return orderr.update(fullchain_pem=(cert + chain_str)) return cast(ClientV2, self.client).finalize_order( orderr, deadline, fetch_alternative_chains) - def revoke(self, cert, rsn): + def revoke(self, cert: jose.ComparableX509, rsn: int) -> None: """Revoke certificate. :param .ComparableX509 cert: `OpenSSL.crypto.X509` wrapped in @@ -950,14 +988,14 @@ class BackwardsCompatibleClientV2: :raises .ClientError: If revocation is unsuccessful. """ - return self.client.revoke(cert, rsn) + self.client.revoke(cert, rsn) - def _acme_version_from_directory(self, directory): + def _acme_version_from_directory(self, directory: messages.Directory) -> int: if hasattr(directory, 'newNonce'): return 2 return 1 - def external_account_required(self): + def external_account_required(self) -> bool: """Checks if the server requires an external account for ACMEv2 servers. Always return False for ACMEv1 servers, as it doesn't use External Account Binding.""" @@ -989,9 +1027,10 @@ class ClientNetwork: :param source_address: Optional source address to bind to when making requests. :type source_address: str or tuple(str, int) """ - def __init__(self, key, account=None, alg=jose.RS256, verify_ssl=True, - user_agent='acme-python', timeout=DEFAULT_NETWORK_TIMEOUT, - source_address=None): + def __init__(self, key: jose.JWK, account: Optional[messages.RegistrationResource] = None, + alg: jose.JWASignature = jose.RS256, verify_ssl: bool = True, + user_agent: str = 'acme-python', timeout: int = DEFAULT_NETWORK_TIMEOUT, + source_address: Optional[Union[str, Tuple[str, int]]] = None) -> None: self.key = key self.account = account self.alg = alg @@ -1008,7 +1047,7 @@ class ClientNetwork: self.session.mount("http://", adapter) self.session.mount("https://", adapter) - def __del__(self): + def __del__(self) -> None: # Try to close the session, but don't show exceptions to the # user if the call to close() fails. See #4840. try: @@ -1016,14 +1055,15 @@ class ClientNetwork: except Exception: # pylint: disable=broad-except pass - def _wrap_in_jws(self, obj, nonce, url, acme_version): + def _wrap_in_jws(self, obj: jose.JSONDeSerializable, nonce: str, url: str, + acme_version: int) -> jose.JWS: """Wrap `JSONDeSerializable` object in JWS. .. todo:: Implement ``acmePath``. :param josepy.JSONDeSerializable obj: :param str url: The URL to which this object will be POSTed - :param bytes nonce: + :param str nonce: :rtype: `josepy.JWS` """ @@ -1045,7 +1085,8 @@ class ClientNetwork: return jws.JWS.sign(jobj, **kwargs).json_dumps(indent=2) @classmethod - def _check_response(cls, response, content_type=None): + def _check_response(cls, response: requests.Response, + content_type: Optional[str] = None) -> requests.Response: """Check response content and its type. .. note:: @@ -1075,7 +1116,7 @@ class ClientNetwork: jobj = None if response.status_code == 409: - raise errors.ConflictError(response.headers.get('Location')) + raise errors.ConflictError(response.headers.get('Location', 'UNKNOWN-LOCATION')) if not response.ok: if jobj is not None: @@ -1103,7 +1144,7 @@ class ClientNetwork: return response - def _send_request(self, method, url, *args, **kwargs): + def _send_request(self, method: str, url: str, *args: Any, **kwargs: Any) -> requests.Response: """Send HTTP request. Makes sure that `verify_ssl` is respected. Logs request and @@ -1178,7 +1219,7 @@ class ClientNetwork: debug_content) return response - def head(self, *args, **kwargs): + def head(self, *args: Any, **kwargs: Any) -> requests.Response: """Send HEAD request without checking the response. Note, that `_check_response` is not called, as it is expected @@ -1188,12 +1229,13 @@ class ClientNetwork: """ return self._send_request('HEAD', *args, **kwargs) - def get(self, url, content_type=JSON_CONTENT_TYPE, **kwargs): + def get(self, url: str, content_type: str = JSON_CONTENT_TYPE, + **kwargs: Any) -> requests.Response: """Send GET request and check response.""" return self._check_response( self._send_request('GET', url, **kwargs), content_type=content_type) - def _add_nonce(self, response): + def _add_nonce(self, response: requests.Response) -> None: if self.REPLAY_NONCE_HEADER in response.headers: nonce = response.headers[self.REPLAY_NONCE_HEADER] try: @@ -1205,7 +1247,7 @@ class ClientNetwork: else: raise errors.MissingNonce(response) - def _get_nonce(self, url, new_nonce_url): + def _get_nonce(self, url: str, new_nonce_url: str) -> str: if not self._nonces: logger.debug('Requesting fresh nonce') if new_nonce_url is None: @@ -1216,7 +1258,7 @@ class ClientNetwork: self._add_nonce(response) return self._nonces.pop() - def post(self, *args, **kwargs): + def post(self, *args: Any, **kwargs: Any) -> requests.Response: """POST object wrapped in `.JWS` and check response. If the server responded with a badNonce error, the request will @@ -1231,8 +1273,9 @@ class ClientNetwork: return self._post_once(*args, **kwargs) raise - def _post_once(self, url, obj, content_type=JOSE_CONTENT_TYPE, - acme_version=1, **kwargs): + def _post_once(self, url: str, obj: jose.JSONDeSerializable, + content_type: str = JOSE_CONTENT_TYPE, acme_version: int = 1, + **kwargs: Any) -> requests.Response: new_nonce_url = kwargs.pop('new_nonce_url', None) data = self._wrap_in_jws(obj, self._get_nonce(url, new_nonce_url), url, acme_version) kwargs.setdefault('headers', {'Content-Type': content_type}) @@ -1250,23 +1293,23 @@ class _ClientDeprecationModule: Internal class delegating to a module, and displaying warnings when attributes related to deprecated attributes in the acme.client module. """ - def __init__(self, module): + def __init__(self, module: ModuleType) -> None: self.__dict__['_module'] = module - def __getattr__(self, attr): + def __getattr__(self, attr: str) -> Any: if attr in ('Client', 'BackwardsCompatibleClientV2'): warnings.warn('The {0} attribute in acme.client is deprecated ' 'and will be removed soon.'.format(attr), 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/acme/acme/crypto_util.py b/acme/acme/crypto_util.py index 2f3395486..681dd18b1 100644 --- a/acme/acme/crypto_util.py +++ b/acme/acme/crypto_util.py @@ -1,11 +1,17 @@ """Crypto utilities.""" import binascii import contextlib +import ipaddress import logging import os import re import socket +from typing import Any from typing import Callable +from typing import List +from typing import Mapping +from typing import Optional +from typing import Set from typing import Tuple from typing import Union @@ -28,10 +34,10 @@ _DEFAULT_SSL_METHOD = SSL.SSLv23_METHOD class _DefaultCertSelection: - def __init__(self, certs): + def __init__(self, certs: Mapping[bytes, Tuple[crypto.PKey, crypto.X509]]): self.certs = certs - def __call__(self, connection): + def __call__(self, connection: SSL.Connection) -> Optional[Tuple[crypto.PKey, crypto.X509]]: server_name = connection.get_servername() return self.certs.get(server_name, None) @@ -49,9 +55,13 @@ class SSLSocket: # pylint: disable=too-few-public-methods `certs` parameter would be ignored, and therefore must be empty. """ - def __init__(self, sock, certs=None, - method=_DEFAULT_SSL_METHOD, alpn_selection=None, - cert_selection=None): + def __init__(self, sock: socket.socket, + certs: Optional[Mapping[bytes, Tuple[crypto.PKey, crypto.X509]]] = None, + method: int = _DEFAULT_SSL_METHOD, + alpn_selection: Optional[Callable[[SSL.Connection, List[bytes]], bytes]] = None, + cert_selection: Optional[Callable[[SSL.Connection], + Tuple[crypto.PKey, crypto.X509]]] = None + ) -> None: self.sock = sock self.alpn_selection = alpn_selection self.method = method @@ -59,14 +69,18 @@ class SSLSocket: # pylint: disable=too-few-public-methods raise ValueError("Neither cert_selection or certs specified.") if cert_selection and certs: raise ValueError("Both cert_selection and certs specified.") - if cert_selection is None: - cert_selection = _DefaultCertSelection(certs) - self.cert_selection = cert_selection + actual_cert_selection: Union[_DefaultCertSelection, + Optional[Callable[[SSL.Connection], + Tuple[crypto.PKey, + crypto.X509]]]] = cert_selection + if actual_cert_selection is None: + actual_cert_selection = _DefaultCertSelection(certs if certs else {}) + self.cert_selection = actual_cert_selection - def __getattr__(self, name): + def __getattr__(self, name: str) -> Any: return getattr(self.sock, name) - def _pick_certificate_cb(self, connection): + def _pick_certificate_cb(self, connection: SSL.Connection) -> None: """SNI certificate callback. This method will set a new OpenSSL context object for this @@ -98,17 +112,17 @@ class SSLSocket: # pylint: disable=too-few-public-methods # pylint: disable=missing-function-docstring - def __init__(self, connection): + def __init__(self, connection: SSL.Connection) -> None: self._wrapped = connection - def __getattr__(self, name): + def __getattr__(self, name: str) -> Any: return getattr(self._wrapped, name) - def shutdown(self, *unused_args): + def shutdown(self, *unused_args: Any) -> bool: # OpenSSL.SSL.Connection.shutdown doesn't accept any args return self._wrapped.shutdown() - def accept(self): # pylint: disable=missing-function-docstring + def accept(self) -> Tuple[FakeConnection, Any]: # pylint: disable=missing-function-docstring sock, addr = self.sock.accept() context = SSL.Context(self.method) @@ -132,9 +146,9 @@ class SSLSocket: # pylint: disable=too-few-public-methods return ssl_sock, addr -def probe_sni(name, host, port=443, timeout=300, # pylint: disable=too-many-arguments - method=_DEFAULT_SSL_METHOD, source_address=('', 0), - alpn_protocols=None): +def probe_sni(name: bytes, host: bytes, port: int = 443, timeout: int = 300, # pylint: disable=too-many-arguments + method: int = _DEFAULT_SSL_METHOD, source_address: Tuple[str, int] = ('', 0), + alpn_protocols: Optional[List[str]] = None) -> crypto.X509: """Probe SNI server for SSL certificate. :param bytes name: Byte string to send as the server name in the @@ -147,7 +161,7 @@ def probe_sni(name, host, port=443, timeout=300, # pylint: disable=too-many-argu of source interface). See `socket.creation_connection` for more info. Available only in Python 2.7+. :param alpn_protocols: Protocols to request using ALPN. - :type alpn_protocols: `list` of `bytes` + :type alpn_protocols: `list` of `str` :raises acme.errors.Error: In case of any problems. @@ -168,8 +182,8 @@ def probe_sni(name, host, port=443, timeout=300, # pylint: disable=too-many-argu source_address[1] ) if any(source_address) else "" ) - socket_tuple: Tuple[str, int] = (host, port) - sock = socket.create_connection(socket_tuple, **socket_kwargs) + socket_tuple: Tuple[bytes, int] = (host, port) + sock = socket.create_connection(socket_tuple, **socket_kwargs) # type: ignore[arg-type] except socket.error as error: raise errors.Error(error) @@ -187,7 +201,10 @@ def probe_sni(name, host, port=443, timeout=300, # pylint: disable=too-many-argu return client_ssl.get_peer_certificate() -def make_csr(private_key_pem, domains=None, must_staple=False, ipaddrs=None): +def make_csr(private_key_pem: bytes, domains: Optional[Union[Set[str], List[str]]] = None, + must_staple: bool = False, + ipaddrs: Optional[List[Union[ipaddress.IPv4Address, ipaddress.IPv6Address]]] = None + ) -> bytes: """Generate a CSR containing domains or IPs as subjectAltNames. :param buffer private_key_pem: Private key, in PEM PKCS#8 format. @@ -238,7 +255,8 @@ def make_csr(private_key_pem, domains=None, must_staple=False, ipaddrs=None): crypto.FILETYPE_PEM, csr) -def _pyopenssl_cert_or_req_all_names(loaded_cert_or_req): +def _pyopenssl_cert_or_req_all_names(loaded_cert_or_req: Union[crypto.X509, crypto.X509Req] + ) -> List[str]: # unlike its name this only outputs DNS names, other type of idents will ignored common_name = loaded_cert_or_req.get_subject().CN sans = _pyopenssl_cert_or_req_san(loaded_cert_or_req) @@ -248,7 +266,7 @@ def _pyopenssl_cert_or_req_all_names(loaded_cert_or_req): return [common_name] + [d for d in sans if d != common_name] -def _pyopenssl_cert_or_req_san(cert_or_req): +def _pyopenssl_cert_or_req_san(cert_or_req: Union[crypto.X509, crypto.X509Req]) -> List[str]: """Get Subject Alternative Names from certificate or CSR using pyOpenSSL. .. todo:: Implement directly in PyOpenSSL! @@ -275,7 +293,7 @@ def _pyopenssl_cert_or_req_san(cert_or_req): for part in sans_parts if part.startswith(prefix)] -def _pyopenssl_cert_or_req_san_ip(cert_or_req): +def _pyopenssl_cert_or_req_san_ip(cert_or_req: Union[crypto.X509, crypto.X509Req]) -> List[str]: """Get Subject Alternative Names IPs from certificate or CSR using pyOpenSSL. :param cert_or_req: Certificate or CSR. @@ -295,7 +313,7 @@ def _pyopenssl_cert_or_req_san_ip(cert_or_req): return [part[len(prefix):] for part in sans_parts if part.startswith(prefix)] -def _pyopenssl_extract_san_list_raw(cert_or_req): +def _pyopenssl_extract_san_list_raw(cert_or_req: Union[crypto.X509, crypto.X509Req]) -> List[str]: """Get raw SAN string from cert or csr, parse it as UTF-8 and return. :param cert_or_req: Certificate or CSR. @@ -312,10 +330,9 @@ def _pyopenssl_extract_san_list_raw(cert_or_req): if isinstance(cert_or_req, crypto.X509): # pylint: disable=line-too-long - func: Union[Callable[[int, crypto.X509Req], bytes], Callable[[int, crypto.X509], bytes]] = crypto.dump_certificate + text = crypto.dump_certificate(crypto.FILETYPE_TEXT, cert_or_req).decode('utf-8') else: - func = crypto.dump_certificate_request - text = func(crypto.FILETYPE_TEXT, cert_or_req).decode("utf-8") + text = crypto.dump_certificate_request(crypto.FILETYPE_TEXT, cert_or_req).decode('utf-8') # WARNING: this function does not support multiple SANs extensions. # Multiple X509v3 extensions of the same type is disallowed by RFC 5280. raw_san = re.search(r"X509v3 Subject Alternative Name:(?: critical)?\s*(.*)", text) @@ -327,8 +344,12 @@ def _pyopenssl_extract_san_list_raw(cert_or_req): return sans_parts -def gen_ss_cert(key, domains=None, not_before=None, - validity=(7 * 24 * 60 * 60), force_san=True, extensions=None, ips=None): +def gen_ss_cert(key: crypto.PKey, domains: Optional[List[str]] = None, + not_before: Optional[int] = None, + validity: int = (7 * 24 * 60 * 60), force_san: bool = True, + extensions: Optional[List[crypto.X509Extension]] = None, + ips: Optional[List[Union[ipaddress.IPv4Address, ipaddress.IPv4Address]]] = None + ) -> crypto.X509: """Generate new self-signed certificate. :type domains: `list` of `unicode` @@ -389,7 +410,7 @@ def gen_ss_cert(key, domains=None, not_before=None, return cert -def dump_pyopenssl_chain(chain, filetype=crypto.FILETYPE_PEM): +def dump_pyopenssl_chain(chain: List[crypto.X509], filetype: int = crypto.FILETYPE_PEM) -> bytes: """Dump certificate chain into a bundle. :param list chain: List of `OpenSSL.crypto.X509` (or wrapped in @@ -402,7 +423,7 @@ def dump_pyopenssl_chain(chain, filetype=crypto.FILETYPE_PEM): # XXX: returns empty string when no chain is available, which # shuts up RenewableCert, but might not be the best solution... - def _dump_cert(cert): + def _dump_cert(cert: Union[jose.ComparableX509, crypto.X509]) -> bytes: if isinstance(cert, jose.ComparableX509): cert = cert.wrapped return crypto.dump_certificate(filetype, cert) diff --git a/acme/acme/errors.py b/acme/acme/errors.py index b47ed88da..b8407da7e 100644 --- a/acme/acme/errors.py +++ b/acme/acme/errors.py @@ -1,5 +1,17 @@ """ACME errors.""" +import typing +from typing import Any +from typing import List +from typing import Mapping +from typing import Set + from josepy import errors as jose_errors +import requests + +# We import acme.messages only during type check to avoid circular dependencies. Type references +# to acme.message.* must be quoted to be lazily initialized and avoid compilation errors. +if typing.TYPE_CHECKING: + from acme import messages # pragma: no cover class Error(Exception): @@ -28,12 +40,12 @@ class NonceError(ClientError): class BadNonce(NonceError): """Bad nonce error.""" - def __init__(self, nonce, error, *args): + def __init__(self, nonce: str, error: Exception, *args: Any) -> None: super().__init__(*args) self.nonce = nonce self.error = error - def __str__(self): + def __str__(self) -> str: return 'Invalid nonce ({0!r}): {1}'.format(self.nonce, self.error) @@ -47,11 +59,11 @@ class MissingNonce(NonceError): :ivar requests.Response ~.response: HTTP Response """ - def __init__(self, response, *args): + def __init__(self, response: requests.Response, *args: Any) -> None: super().__init__(*args) self.response = response - def __str__(self): + def __str__(self) -> str: return ('Server {0} response did not include a replay ' 'nonce, headers: {1} (This may be a service outage)'.format( self.response.request.method, self.response.headers)) @@ -69,17 +81,20 @@ class PollError(ClientError): to the most recently updated one """ - def __init__(self, exhausted, updated): + def __init__(self, exhausted: Set['messages.AuthorizationResource'], + updated: Mapping['messages.AuthorizationResource', + 'messages.AuthorizationResource'] + ) -> None: self.exhausted = exhausted self.updated = updated super().__init__() @property - def timeout(self): + def timeout(self) -> bool: """Was the error caused by timeout?""" return bool(self.exhausted) - def __repr__(self): + def __repr__(self) -> str: return '{0}(exhausted={1!r}, updated={2!r})'.format( self.__class__.__name__, self.exhausted, self.updated) @@ -88,7 +103,7 @@ class ValidationError(Error): """Error for authorization failures. Contains a list of authorization resources, each of which is invalid and should have an error field. """ - def __init__(self, failed_authzrs): + def __init__(self, failed_authzrs: List['messages.AuthorizationResource']) -> None: self.failed_authzrs = failed_authzrs super().__init__() @@ -100,7 +115,7 @@ class TimeoutError(Error): # pylint: disable=redefined-builtin class IssuanceError(Error): """Error sent by the server after requesting issuance of a certificate.""" - def __init__(self, error): + def __init__(self, error: 'messages.Error') -> None: """Initialize. :param messages.Error error: The error provided by the server. @@ -117,7 +132,7 @@ class ConflictError(ClientError): Also used in V2 of the ACME client for the same purpose. """ - def __init__(self, location): + def __init__(self, location: str) -> None: self.location = location super().__init__() diff --git a/acme/acme/fields.py b/acme/acme/fields.py index bd915e47d..b15489020 100644 --- a/acme/acme/fields.py +++ b/acme/acme/fields.py @@ -1,4 +1,7 @@ """ACME JSON fields.""" +import datetime +from typing import Any + import logging import josepy as jose @@ -10,17 +13,17 @@ logger = logging.getLogger(__name__) class Fixed(jose.Field): """Fixed field.""" - def __init__(self, json_name, value): + def __init__(self, json_name: str, value: Any) -> None: self.value = value super().__init__( json_name=json_name, default=value, omitempty=False) - def decode(self, value): + def decode(self, value: Any) -> Any: if value != self.value: raise jose.DeserializationError('Expected {0!r}'.format(self.value)) return self.value - def encode(self, value): + def encode(self, value: Any) -> Any: if value != self.value: logger.warning( 'Overriding fixed field (%s) with %r', self.json_name, value) @@ -37,11 +40,11 @@ class RFC3339Field(jose.Field): """ @classmethod - def default_encoder(cls, value): + def default_encoder(cls, value: datetime.datetime) -> str: return pyrfc3339.generate(value) @classmethod - def default_decoder(cls, value): + def default_decoder(cls, value: str) -> datetime.datetime: try: return pyrfc3339.parse(value) except ValueError as error: @@ -51,12 +54,12 @@ class RFC3339Field(jose.Field): class Resource(jose.Field): """Resource MITM field.""" - def __init__(self, resource_type, *args, **kwargs): + def __init__(self, resource_type: str, *args: Any, **kwargs: Any) -> None: self.resource_type = resource_type super().__init__( 'resource', default=resource_type, *args, **kwargs) - def decode(self, value): + def decode(self, value: Any) -> Any: if value != self.resource_type: raise jose.DeserializationError( 'Wrong resource type: {0} instead of {1}'.format( diff --git a/acme/acme/jws.py b/acme/acme/jws.py index d9d3e12c3..bde1bb590 100644 --- a/acme/acme/jws.py +++ b/acme/acme/jws.py @@ -4,6 +4,8 @@ The JWS implementation in josepy only implements the base JOSE standard. In order to support the new header fields defined in ACME, this module defines some ACME-specific classes that layer on top of josepy. """ +from typing import Optional + import josepy as jose @@ -17,7 +19,7 @@ class Header(jose.Header): # Mypy does not understand the josepy magic happening here, and falsely claims # that nonce is redefined. Let's ignore the type check here. @nonce.decoder # type: ignore - def nonce(value): # pylint: disable=no-self-argument,missing-function-docstring + def nonce(value: str) -> bytes: # pylint: disable=no-self-argument,missing-function-docstring try: return jose.decode_b64jose(value) except jose.DeserializationError as error: @@ -46,11 +48,12 @@ class JWS(jose.JWS): @classmethod # pylint: disable=arguments-differ - def sign(cls, payload, key, alg, nonce, url=None, kid=None): + def sign(cls, payload: bytes, key: jose.JWK, alg: jose.JWASignature, nonce: Optional[bytes], + url: Optional[str] = None, kid: Optional[str] = None) -> jose.JWS: # Per ACME spec, jwk and kid are mutually exclusive, so only include a # jwk field if kid is not provided. include_jwk = kid is None return super().sign(payload, key=key, alg=alg, - protect=frozenset(['nonce', 'url', 'kid', 'jwk', 'alg']), - nonce=nonce, url=url, kid=kid, - include_jwk=include_jwk) + protect=frozenset(['nonce', 'url', 'kid', 'jwk', 'alg']), + nonce=nonce, url=url, kid=kid, + include_jwk=include_jwk) diff --git a/acme/acme/magic_typing.py b/acme/acme/magic_typing.py index 796987415..b05d2c4bc 100644 --- a/acme/acme/magic_typing.py +++ b/acme/acme/magic_typing.py @@ -6,12 +6,13 @@ available. This code is being kept for now for backwards compatibility. """ import warnings from typing import * # pylint: disable=wildcard-import, unused-wildcard-import -from typing import Collection, IO +from typing import Any warnings.warn("acme.magic_typing is deprecated and will be removed in a future release.", DeprecationWarning) + class TypingClass: """Ignore import errors by getting anything""" - def __getattr__(self, name): + def __getattr__(self, name: str) -> Any: return None # pragma: no cover diff --git a/acme/acme/messages.py b/acme/acme/messages.py index 9c519a009..0dd579a98 100644 --- a/acme/acme/messages.py +++ b/acme/acme/messages.py @@ -3,7 +3,13 @@ from collections.abc import Hashable import json from typing import Any from typing import Dict +from typing import Iterator +from typing import List +from typing import Mapping +from typing import MutableMapping +from typing import Tuple from typing import Type +from typing import Optional import josepy as jose @@ -57,7 +63,7 @@ ERROR_TYPE_DESCRIPTIONS.update(dict( # add errors with old prefix, deprecate me (OLD_ERROR_PREFIX + name, desc) for name, desc in ERROR_CODES.items())) -def is_acme_error(err): +def is_acme_error(err: BaseException) -> bool: """Check if argument is an ACME error.""" if isinstance(err, Error) and (err.typ is not None): return (ERROR_PREFIX in err.typ) or (OLD_ERROR_PREFIX in err.typ) @@ -79,7 +85,7 @@ class Error(jose.JSONObjectWithFields, errors.Error): detail = jose.Field('detail', omitempty=True) @classmethod - def with_code(cls, code, **kwargs): + def with_code(cls, code: str, **kwargs: Any) -> 'Error': """Create an Error instance with an ACME Error code. :unicode code: An ACME error code, like 'dnssec'. @@ -95,7 +101,7 @@ class Error(jose.JSONObjectWithFields, errors.Error): return cls(typ=typ, **kwargs) # type: ignore @property - def description(self): + def description(self) -> Optional[str]: """Hardcoded error description based on its type. :returns: Description if standard ACME error or ``None``. @@ -105,7 +111,7 @@ class Error(jose.JSONObjectWithFields, errors.Error): return ERROR_TYPE_DESCRIPTIONS.get(self.typ) @property - def code(self): + def code(self) -> Optional[str]: """ACME error code. Basically self.typ without the ERROR_PREFIX. @@ -119,7 +125,7 @@ class Error(jose.JSONObjectWithFields, errors.Error): return code return None - def __str__(self): + def __str__(self) -> str: return b' :: '.join( part.encode('ascii', 'backslashreplace') for part in (self.typ, self.description, self.detail, self.title) @@ -131,34 +137,36 @@ class _Constant(jose.JSONDeSerializable, Hashable): __slots__ = ('name',) POSSIBLE_NAMES: Dict[str, '_Constant'] = NotImplemented - def __init__(self, name): + def __init__(self, name: str) -> None: super().__init__() self.POSSIBLE_NAMES[name] = self # pylint: disable=unsupported-assignment-operation self.name = name - def to_partial_json(self): + def to_partial_json(self) -> str: return self.name @classmethod - def from_json(cls, jobj): + def from_json(cls, jobj: str) -> '_Constant': if jobj not in cls.POSSIBLE_NAMES: # pylint: disable=unsupported-membership-test raise jose.DeserializationError( '{0} not recognized'.format(cls.__name__)) return cls.POSSIBLE_NAMES[jobj] - def __repr__(self): + def __repr__(self) -> str: return '{0}({1})'.format(self.__class__.__name__, self.name) - def __eq__(self, other): + def __eq__(self, other: Any) -> bool: return isinstance(other, type(self)) and other.name == self.name - def __hash__(self): + def __hash__(self) -> int: return hash((self.__class__, self.name)) class Status(_Constant): """ACME "status" field.""" - POSSIBLE_NAMES: dict = {} + POSSIBLE_NAMES: Dict[str, 'Status'] = {} + + STATUS_UNKNOWN = Status('unknown') STATUS_PENDING = Status('pending') STATUS_PROCESSING = Status('processing') @@ -172,7 +180,8 @@ STATUS_DEACTIVATED = Status('deactivated') class IdentifierType(_Constant): """ACME identifier type.""" POSSIBLE_NAMES: Dict[str, 'IdentifierType'] = {} - # class def ends here + + IDENTIFIER_FQDN = IdentifierType('dns') # IdentifierDNS in Boulder IDENTIFIER_IP = IdentifierType('ip') # IdentifierIP in pebble - not in Boulder yet @@ -191,7 +200,7 @@ class Identifier(jose.JSONObjectWithFields): class Directory(jose.JSONDeSerializable): """Directory.""" - _REGISTERED_TYPES: Dict[str, Type[Any]] = {} + _REGISTERED_TYPES: Dict[str, Type['Directory']] = {} class Meta(jose.JSONObjectWithFields): """Directory Meta.""" @@ -201,60 +210,59 @@ class Directory(jose.JSONDeSerializable): caa_identities = jose.Field('caaIdentities', omitempty=True) external_account_required = jose.Field('externalAccountRequired', omitempty=True) - def __init__(self, **kwargs): + def __init__(self, **kwargs: Any) -> None: kwargs = {self._internal_name(k): v for k, v in kwargs.items()} super().__init__(**kwargs) @property - def terms_of_service(self): + def terms_of_service(self) -> str: """URL for the CA TOS""" return self._terms_of_service or self._terms_of_service_v2 - def __iter__(self): + def __iter__(self) -> Iterator[str]: # When iterating over fields, use the external name 'terms_of_service' instead of # the internal '_terms_of_service'. for name in super().__iter__(): yield name[1:] if name == '_terms_of_service' else name - def _internal_name(self, name): + def _internal_name(self, name: str) -> str: return '_' + name if name == 'terms_of_service' else name - @classmethod - def _canon_key(cls, key): + def _canon_key(cls, key: str) -> str: return getattr(key, 'resource_type', key) @classmethod - def register(cls, resource_body_cls: Type[Any]) -> Type[Any]: + def register(cls, resource_body_cls: Type['Directory']) -> Type['Directory']: """Register resource.""" resource_type = resource_body_cls.resource_type assert resource_type not in cls._REGISTERED_TYPES cls._REGISTERED_TYPES[resource_type] = resource_body_cls return resource_body_cls - def __init__(self, jobj): + def __init__(self, jobj: Mapping[str, Any]) -> None: canon_jobj = util.map_keys(jobj, self._canon_key) # TODO: check that everything is an absolute URL; acme-spec is # not clear on that self._jobj = canon_jobj - def __getattr__(self, name): + def __getattr__(self, name: str) -> Any: try: return self[name.replace('_', '-')] except KeyError as error: raise AttributeError(str(error)) - def __getitem__(self, name): + def __getitem__(self, name: str) -> Any: try: return self._jobj[self._canon_key(name)] except KeyError: raise KeyError('Directory field "' + self._canon_key(name) + '" not found') - def to_partial_json(self): + def to_partial_json(self) -> Dict[str, Any]: return self._jobj @classmethod - def from_json(cls, jobj): + def from_json(cls, jobj: MutableMapping[str, Any]) -> 'Directory': jobj['meta'] = cls.Meta.from_json(jobj.pop('meta', {})) return cls(jobj) @@ -285,7 +293,8 @@ class ExternalAccountBinding: """ACME External Account Binding""" @classmethod - def from_data(cls, account_public_key, kid, hmac_key, directory): + def from_data(cls, account_public_key: jose.JWK, kid: str, hmac_key: str, + directory: Directory) -> Dict[str, Any]: """Create External Account Binding Resource from contact details, kid and hmac.""" key_json = json.dumps(account_public_key.to_partial_json()).encode() @@ -325,7 +334,9 @@ class Registration(ResourceBody): email_prefix = 'mailto:' @classmethod - def from_data(cls, phone=None, email=None, external_account_binding=None, **kwargs): + def from_data(cls, phone: Optional[str] = None, email: Optional[str] = None, + external_account_binding: Optional[ExternalAccountBinding] = None, + **kwargs: Any) -> 'Registration': """ Create registration resource from contact details. @@ -354,19 +365,19 @@ class Registration(ResourceBody): return cls(**kwargs) - def __init__(self, **kwargs): + def __init__(self, **kwargs: Any) -> None: """Note if the user provides a value for the `contact` member.""" - if 'contact' in kwargs: + if 'contact' in kwargs and kwargs['contact'] is not None: # Avoid the __setattr__ used by jose.TypedJSONObjectWithFields object.__setattr__(self, '_add_contact', True) super().__init__(**kwargs) - def _filter_contact(self, prefix): + def _filter_contact(self, prefix: str) -> Tuple[str, ...]: return tuple( detail[len(prefix):] for detail in self.contact # pylint: disable=not-an-iterable if detail.startswith(prefix)) - def _add_contact_if_appropriate(self, jobj): + def _add_contact_if_appropriate(self, jobj: Dict[str, Any]) -> Dict[str, Any]: """ The `contact` member of Registration objects should not be required when de-serializing (as it would be if the Fields' `omitempty` flag were `False`), but @@ -383,23 +394,23 @@ class Registration(ResourceBody): return jobj - def to_partial_json(self): + def to_partial_json(self) -> Dict[str, Any]: """Modify josepy.JSONDeserializable.to_partial_json()""" jobj = super().to_partial_json() return self._add_contact_if_appropriate(jobj) - def fields_to_partial_json(self): + def fields_to_partial_json(self) -> Dict[str, Any]: """Modify josepy.JSONObjectWithFields.fields_to_partial_json()""" jobj = super().fields_to_partial_json() return self._add_contact_if_appropriate(jobj) @property - def phones(self): + def phones(self) -> Tuple[str, ...]: """All phones found in the ``contact`` field.""" return self._filter_contact(self.phone_prefix) @property - def emails(self): + def emails(self) -> Tuple[str, ...]: """All emails found in the ``contact`` field.""" return self._filter_contact(self.email_prefix) @@ -460,39 +471,39 @@ class ChallengeBody(ResourceBody): error = jose.Field('error', decoder=Error.from_json, omitempty=True, default=None) - def __init__(self, **kwargs): + def __init__(self, **kwargs: Any) -> None: kwargs = {self._internal_name(k): v for k, v in kwargs.items()} super().__init__(**kwargs) - def encode(self, name): + def encode(self, name: str) -> Any: return super().encode(self._internal_name(name)) - def to_partial_json(self): + def to_partial_json(self) -> Dict[str, Any]: jobj = super().to_partial_json() jobj.update(self.chall.to_partial_json()) return jobj @classmethod - def fields_from_json(cls, jobj): + def fields_from_json(cls, jobj: Mapping[str, Any]) -> Dict[str, Any]: jobj_fields = super().fields_from_json(jobj) jobj_fields['chall'] = challenges.Challenge.from_json(jobj) return jobj_fields @property - def uri(self): + def uri(self) -> str: """The URL of this challenge.""" return self._url or self._uri - def __getattr__(self, name): + def __getattr__(self, name: str) -> Any: return getattr(self.chall, name) - def __iter__(self): + def __iter__(self) -> Iterator[str]: # When iterating over fields, use the external name 'uri' instead of # the internal '_uri'. for name in super().__iter__(): yield name[1:] if name == '_uri' else name - def _internal_name(self, name): + def _internal_name(self, name: str) -> str: return '_' + name if name == 'uri' else name @@ -507,7 +518,7 @@ class ChallengeResource(Resource): authzr_uri = jose.Field('authzr_uri') @property - def uri(self): + def uri(self) -> str: """The URL of the challenge body.""" return self.body.uri @@ -538,11 +549,11 @@ class Authorization(ResourceBody): # Mypy does not understand the josepy magic happening here, and falsely claims # that challenge is redefined. Let's ignore the type check here. @challenges.decoder # type: ignore - def challenges(value): # pylint: disable=no-self-argument,missing-function-docstring + def challenges(value: List[Mapping[str, Any]]) -> Tuple[ChallengeBody, ...]: # pylint: disable=no-self-argument,missing-function-docstring return tuple(ChallengeBody.from_json(chall) for chall in value) @property - def resolved_combinations(self): + def resolved_combinations(self) -> Tuple[Tuple[Dict[str, Any], ...], ...]: """Combinations with challenges instead of indices.""" return tuple(tuple(self.challenges[idx] for idx in combo) for combo in self.combinations) # pylint: disable=not-an-iterable @@ -639,9 +650,10 @@ class Order(ResourceBody): # Mypy does not understand the josepy magic happening here, and falsely claims # that identifiers is redefined. Let's ignore the type check here. @identifiers.decoder # type: ignore - def identifiers(value): # pylint: disable=no-self-argument,missing-function-docstring + def identifiers(value: List[Mapping[str, Any]]) -> Tuple[Identifier, ...]: # pylint: disable=no-self-argument,missing-function-docstring return tuple(Identifier.from_json(identifier) for identifier in value) + class OrderResource(ResourceWithURI): """Order Resource. diff --git a/acme/acme/mixins.py b/acme/acme/mixins.py index 6d58e0889..38551d550 100644 --- a/acme/acme/mixins.py +++ b/acme/acme/mixins.py @@ -1,21 +1,23 @@ """Useful mixins for Challenge and Resource objects""" +from typing import Any +from typing import Dict class VersionedLEACMEMixin: """This mixin stores the version of Let's Encrypt's endpoint being used.""" @property - def le_acme_version(self): + def le_acme_version(self) -> int: """Define the version of ACME protocol to use""" return getattr(self, '_le_acme_version', 1) @le_acme_version.setter - def le_acme_version(self, version): + def le_acme_version(self, version: int) -> None: # We need to use object.__setattr__ to not depend on the specific implementation of # __setattr__ in current class (eg. jose.TypedJSONObjectWithFields raises AttributeError # for any attempt to set an attribute to make objects immutable). object.__setattr__(self, '_le_acme_version', version) - def __setattr__(self, key, value): + def __setattr__(self, key: str, value: Any) -> None: if key == 'le_acme_version': # Required for @property to operate properly. See comment above. object.__setattr__(self, key, value) @@ -28,12 +30,12 @@ class ResourceMixin(VersionedLEACMEMixin): This mixin generates a RFC8555 compliant JWS payload by removing the `resource` field if needed (eg. ACME v2 protocol). """ - def to_partial_json(self): + def to_partial_json(self) -> Dict[str, Any]: """See josepy.JSONDeserializable.to_partial_json()""" return _safe_jobj_compliance(super(), 'to_partial_json', 'resource') - def fields_to_partial_json(self): + def fields_to_partial_json(self) -> Dict[str, Any]: """See josepy.JSONObjectWithFields.fields_to_partial_json()""" return _safe_jobj_compliance(super(), 'fields_to_partial_json', 'resource') @@ -44,20 +46,21 @@ class TypeMixin(VersionedLEACMEMixin): This mixin allows generation of a RFC8555 compliant JWS payload by removing the `type` field if needed (eg. ACME v2 protocol). """ - def to_partial_json(self): + def to_partial_json(self) -> Dict[str, Any]: """See josepy.JSONDeserializable.to_partial_json()""" return _safe_jobj_compliance(super(), 'to_partial_json', 'type') - def fields_to_partial_json(self): + def fields_to_partial_json(self) -> Dict[str, Any]: """See josepy.JSONObjectWithFields.fields_to_partial_json()""" return _safe_jobj_compliance(super(), 'fields_to_partial_json', 'type') -def _safe_jobj_compliance(instance, jobj_method, uncompliant_field): +def _safe_jobj_compliance(instance: Any, jobj_method: str, + uncompliant_field: str) -> Dict[str, Any]: if hasattr(instance, jobj_method): - jobj = getattr(instance, jobj_method)() + jobj: Dict[str, Any] = getattr(instance, jobj_method)() if instance.le_acme_version == 2: jobj.pop(uncompliant_field, None) return jobj diff --git a/acme/acme/standalone.py b/acme/acme/standalone.py index e2672fb94..cd2caa221 100644 --- a/acme/acme/standalone.py +++ b/acme/acme/standalone.py @@ -7,8 +7,16 @@ import logging import socket import socketserver import threading +from typing import Any from typing import List +from typing import Mapping from typing import Optional +from typing import Set +from typing import Tuple +from typing import Type + +from OpenSSL import crypto +from OpenSSL import SSL from acme import challenges from acme import crypto_util @@ -19,7 +27,7 @@ logger = logging.getLogger(__name__) class TLSServer(socketserver.TCPServer): """Generic TLS Server.""" - def __init__(self, *args, **kwargs): + def __init__(self, *args: Any, **kwargs: Any) -> None: self.ipv6 = kwargs.pop("ipv6", False) if self.ipv6: self.address_family = socket.AF_INET6 @@ -31,18 +39,19 @@ class TLSServer(socketserver.TCPServer): self.allow_reuse_address = kwargs.pop("allow_reuse_address", True) socketserver.TCPServer.__init__(self, *args, **kwargs) - def _wrap_sock(self): + def _wrap_sock(self) -> None: self.socket = crypto_util.SSLSocket( self.socket, cert_selection=self._cert_selection, alpn_selection=getattr(self, '_alpn_selection', None), method=self.method) - def _cert_selection(self, connection): # pragma: no cover + def _cert_selection(self, connection: SSL.Connection + ) -> Tuple[crypto.PKey, crypto.X509]: # pragma: no cover """Callback selecting certificate for connection.""" server_name = connection.get_servername() return self.certs.get(server_name, None) - def server_bind(self): + def server_bind(self) -> None: self._wrap_sock() return socketserver.TCPServer.server_bind(self) @@ -62,7 +71,8 @@ class BaseDualNetworkedServers: If two servers are instantiated, they will serve on the same port. """ - def __init__(self, ServerClass, server_address, *remaining_args, **kwargs): + def __init__(self, ServerClass: Type[socketserver.TCPServer], server_address: Tuple[str, int], + *remaining_args: Any, **kwargs: Any) -> None: port = server_address[1] self.threads: List[threading.Thread] = [] self.servers: List[socketserver.BaseServer] = [] @@ -111,7 +121,7 @@ class BaseDualNetworkedServers: else: # pragma: no cover raise socket.error("Could not bind to IPv4 or IPv6.") - def serve_forever(self): + def serve_forever(self) -> None: """Wraps socketserver.TCPServer.serve_forever""" for server in self.servers: thread = threading.Thread( @@ -119,11 +129,11 @@ class BaseDualNetworkedServers: thread.start() self.threads.append(thread) - def getsocknames(self): + def getsocknames(self) -> List[Tuple[str, int]]: """Wraps socketserver.TCPServer.socket.getsockname""" return [server.socket.getsockname() for server in self.servers] - def shutdown_and_server_close(self): + def shutdown_and_server_close(self) -> None: """Wraps socketserver.TCPServer.shutdown, socketserver.TCPServer.server_close, and threading.Thread.join""" for server in self.servers: @@ -139,13 +149,16 @@ class TLSALPN01Server(TLSServer, ACMEServerMixin): ACME_TLS_1_PROTOCOL = b"acme-tls/1" - def __init__(self, server_address, certs, challenge_certs, ipv6=False): + def __init__(self, server_address: Tuple[str, int], + certs: List[Tuple[crypto.PKey, crypto.X509]], + challenge_certs: Mapping[str, Tuple[crypto.PKey, crypto.X509]], + ipv6: bool = False) -> None: TLSServer.__init__( self, server_address, _BaseRequestHandlerWithLogging, certs=certs, ipv6=ipv6) self.challenge_certs = challenge_certs - def _cert_selection(self, connection): + def _cert_selection(self, connection: SSL.Connection) -> Tuple[crypto.PKey, crypto.X509]: # TODO: We would like to serve challenge cert only if asked for it via # ALPN. To do this, we need to retrieve the list of protos from client # hello, but this is currently impossible with openssl [0], and ALPN @@ -155,9 +168,9 @@ class TLSALPN01Server(TLSServer, ACMEServerMixin): # [0] https://github.com/openssl/openssl/issues/4952 server_name = connection.get_servername() logger.debug("Serving challenge cert for server name %s", server_name) - return self.challenge_certs.get(server_name, None) + return self.challenge_certs[server_name] - def _alpn_selection(self, _connection, alpn_protos): + def _alpn_selection(self, _connection: SSL.Connection, alpn_protos: List[bytes]) -> bytes: """Callback to select alpn protocol.""" if len(alpn_protos) == 1 and alpn_protos[0] == self.ACME_TLS_1_PROTOCOL: logger.debug("Agreed on %s ALPN", self.ACME_TLS_1_PROTOCOL) @@ -171,7 +184,7 @@ class TLSALPN01Server(TLSServer, ACMEServerMixin): class HTTPServer(BaseHTTPServer.HTTPServer): """Generic HTTP Server.""" - def __init__(self, *args, **kwargs): + def __init__(self, *args: Any, **kwargs: Any) -> None: self.ipv6 = kwargs.pop("ipv6", False) if self.ipv6: self.address_family = socket.AF_INET6 @@ -183,7 +196,8 @@ class HTTPServer(BaseHTTPServer.HTTPServer): class HTTP01Server(HTTPServer, ACMEServerMixin): """HTTP01 Server.""" - def __init__(self, server_address, resources, ipv6=False, timeout=30): + def __init__(self, server_address: Tuple[str, int], resources: Set[challenges.HTTP01], + ipv6: bool = False, timeout: int = 30) -> None: HTTPServer.__init__( self, server_address, HTTP01RequestHandler.partial_init( simple_http_resources=resources, timeout=timeout), ipv6=ipv6) @@ -193,7 +207,7 @@ class HTTP01DualNetworkedServers(BaseDualNetworkedServers): """HTTP01Server Wrapper. Tries everything for both. Failures for one don't affect the other.""" - def __init__(self, *args, **kwargs): + def __init__(self, *args: Any, **kwargs: Any) -> None: BaseDualNetworkedServers.__init__(self, HTTP01Server, *args, **kwargs) @@ -209,7 +223,7 @@ class HTTP01RequestHandler(BaseHTTPServer.BaseHTTPRequestHandler): HTTP01Resource = collections.namedtuple( "HTTP01Resource", "chall response validation") - def __init__(self, *args, **kwargs): + def __init__(self, *args: Any, **kwargs: Any) -> None: self.simple_http_resources = kwargs.pop("simple_http_resources", set()) self._timeout = kwargs.pop('timeout', 30) BaseHTTPServer.BaseHTTPRequestHandler.__init__(self, *args, **kwargs) @@ -222,7 +236,7 @@ class HTTP01RequestHandler(BaseHTTPServer.BaseHTTPRequestHandler): # everyone happy, we statically redefine 'timeout' as a method property, and set the # timeout value in a new internal instance-level property _timeout. @property - def timeout(self): + def timeout(self) -> int: # type: ignore[override] """ The default timeout this server should apply to requests. :return: timeout to apply @@ -230,16 +244,16 @@ class HTTP01RequestHandler(BaseHTTPServer.BaseHTTPRequestHandler): """ return self._timeout - def log_message(self, format, *args): # pylint: disable=redefined-builtin + def log_message(self, format: str, *args: Any) -> None: # pylint: disable=redefined-builtin """Log arbitrary message.""" logger.debug("%s - - %s", self.client_address[0], format % args) - def handle(self): + def handle(self) -> None: """Handle request.""" self.log_message("Incoming request") BaseHTTPServer.BaseHTTPRequestHandler.handle(self) - def do_GET(self): # pylint: disable=invalid-name,missing-function-docstring + def do_GET(self) -> None: # pylint: disable=invalid-name,missing-function-docstring if self.path == "/": self.handle_index() elif self.path.startswith("/" + challenges.HTTP01.URI_ROOT_PATH): @@ -247,21 +261,21 @@ class HTTP01RequestHandler(BaseHTTPServer.BaseHTTPRequestHandler): else: self.handle_404() - def handle_index(self): + def handle_index(self) -> None: """Handle index page.""" self.send_response(200) self.send_header("Content-Type", "text/html") self.end_headers() self.wfile.write(self.server.server_version.encode()) - def handle_404(self): + def handle_404(self) -> None: """Handler 404 Not Found errors.""" self.send_response(http_client.NOT_FOUND, message="Not Found") self.send_header("Content-type", "text/html") self.end_headers() self.wfile.write(b"404") - def handle_simple_http_resource(self): + def handle_simple_http_resource(self) -> None: """Handle HTTP01 provisioned resources.""" for resource in self.simple_http_resources: if resource.chall.path == self.path: @@ -277,7 +291,8 @@ class HTTP01RequestHandler(BaseHTTPServer.BaseHTTPRequestHandler): self.path) @classmethod - def partial_init(cls, simple_http_resources, timeout): + def partial_init(cls, simple_http_resources: Set[challenges.HTTP01], + timeout: int) -> 'functools.partial[HTTP01RequestHandler]': """Partially initialize this handler. This is useful because `socketserver.BaseServer` takes @@ -293,11 +308,11 @@ class HTTP01RequestHandler(BaseHTTPServer.BaseHTTPRequestHandler): class _BaseRequestHandlerWithLogging(socketserver.BaseRequestHandler): """BaseRequestHandler with logging.""" - def log_message(self, format, *args): # pylint: disable=redefined-builtin + def log_message(self, format: str, *args: Any) -> None: # pylint: disable=redefined-builtin """Log arbitrary message.""" logger.debug("%s - - %s", self.client_address[0], format % args) - def handle(self): + def handle(self) -> None: """Handle request.""" self.log_message("Incoming request") socketserver.BaseRequestHandler.handle(self) diff --git a/acme/acme/util.py b/acme/acme/util.py index f9c11713e..f027ddcba 100644 --- a/acme/acme/util.py +++ b/acme/acme/util.py @@ -1,5 +1,10 @@ """ACME utilities.""" +from typing import Any +from typing import Callable +from typing import Dict +from typing import Mapping -def map_keys(dikt, func): + +def map_keys(dikt: Mapping[Any, Any], func: Callable[[Any], Any]) -> Dict[Any, Any]: """Map dictionary keys.""" return {func(key): value for key, value in dikt.items()} diff --git a/acme/tests/challenges_test.py b/acme/tests/challenges_test.py index 71bf61976..ed8470fa6 100644 --- a/acme/tests/challenges_test.py +++ b/acme/tests/challenges_test.py @@ -326,12 +326,12 @@ class TLSALPN01ResponseTest(unittest.TestCase): self.response.probe_cert('foo.com') mock_gethostbyname.assert_called_once_with('foo.com') mock_probe_sni.assert_called_once_with( - host='127.0.0.1', port=self.response.PORT, name='foo.com', + host=b'127.0.0.1', port=self.response.PORT, name=b'foo.com', alpn_protocols=['acme-tls/1']) self.response.probe_cert('foo.com', host='8.8.8.8') mock_probe_sni.assert_called_with( - host='8.8.8.8', port=mock.ANY, name='foo.com', + host=b'8.8.8.8', port=mock.ANY, name=b'foo.com', alpn_protocols=['acme-tls/1']) @mock.patch('acme.challenges.TLSALPN01Response.probe_cert') diff --git a/certbot-apache/certbot_apache/_internal/parser.py b/certbot-apache/certbot_apache/_internal/parser.py index 3c705e066..773f5f1e4 100644 --- a/certbot-apache/certbot_apache/_internal/parser.py +++ b/certbot-apache/certbot_apache/_internal/parser.py @@ -15,7 +15,7 @@ from certbot_apache._internal import constants try: from augeas import Augeas except ImportError: # pragma: no cover - Augeas = None # type: ignore + Augeas = None logger = logging.getLogger(__name__) diff --git a/certbot/certbot/_internal/account.py b/certbot/certbot/_internal/account.py index 5469b8396..3d45b8ed6 100644 --- a/certbot/certbot/_internal/account.py +++ b/certbot/certbot/_internal/account.py @@ -5,6 +5,9 @@ import hashlib import logging import shutil import socket +from typing import cast +from typing import Any +from typing import Mapping from cryptography.hazmat.primitives import serialization import josepy as jose @@ -63,7 +66,10 @@ class Account: try: hasher = hashlib.md5() except ValueError: - hasher = hashlib.new('md5', usedforsecurity=False) # type: ignore + # This cast + dictionary expansion is made to make mypy happy without the need of a + # "type: ignore" directive that will also require to disable the check on useless + # "type: ignore" directives when mypy is run on Python 3.9+. + hasher = hashlib.new('md5', **cast(Mapping[str, Any], {"usedforsecurity": False})) hasher.update(self.key.key.public_key().public_bytes( encoding=serialization.Encoding.PEM, diff --git a/certbot/certbot/_internal/client.py b/certbot/certbot/_internal/client.py index a217e889d..899b051bf 100644 --- a/certbot/certbot/_internal/client.py +++ b/certbot/certbot/_internal/client.py @@ -2,7 +2,11 @@ import datetime import logging import platform -from typing import List, Optional, Union +from typing import Any +from typing import Dict +from typing import List +from typing import Optional +from typing import Union import warnings from cryptography.hazmat.backends import default_backend @@ -203,6 +207,7 @@ def perform_registration(acme, config, tos_cb): """ eab_credentials_supplied = config.eab_kid and config.eab_hmac_key + eab: Optional[Dict[str, Any]] if eab_credentials_supplied: account_public_key = acme.client.net.key.public_key() eab = messages.ExternalAccountBinding.from_data(account_public_key=account_public_key, diff --git a/certbot/certbot/compat/_path.py b/certbot/certbot/compat/_path.py index a5db3a575..c36421b7e 100644 --- a/certbot/certbot/compat/_path.py +++ b/certbot/certbot/compat/_path.py @@ -13,7 +13,7 @@ from __future__ import absolute_import # First round of wrapping: we import statically all public attributes exposed by the os.path # module. This allows in particular to have pylint, mypy, IDEs be aware that most of os.path # members are available in certbot.compat.path. -from os.path import * # type: ignore # pylint: disable=wildcard-import,unused-wildcard-import,os-module-forbidden +from os.path import * # pylint: disable=wildcard-import,unused-wildcard-import,os-module-forbidden # Second round of wrapping: we import dynamically all attributes from the os.path module that have # not yet been imported by the first round (static star import). diff --git a/certbot/certbot/compat/os.py b/certbot/certbot/compat/os.py index 653877b50..cdfc6fbb1 100644 --- a/certbot/certbot/compat/os.py +++ b/certbot/certbot/compat/os.py @@ -21,7 +21,7 @@ from __future__ import absolute_import # First round of wrapping: we import statically all public attributes exposed by the os module # This allows in particular to have pylint, mypy, IDEs be aware that most of os members are # available in certbot.compat.os. -from os import * # type: ignore # pylint: disable=wildcard-import,unused-wildcard-import,redefined-builtin,os-module-forbidden +from os import * # pylint: disable=wildcard-import,unused-wildcard-import,redefined-builtin,os-module-forbidden # Second round of wrapping: we import dynamically all attributes from the os module that have not # yet been imported by the first round (static import). This covers in particular the case of diff --git a/certbot/certbot/plugins/dns_common_lexicon.py b/certbot/certbot/plugins/dns_common_lexicon.py index 32213999d..cede6a4fd 100644 --- a/certbot/certbot/plugins/dns_common_lexicon.py +++ b/certbot/certbot/plugins/dns_common_lexicon.py @@ -19,8 +19,8 @@ try: from lexicon.config import ConfigResolver from lexicon.providers.base import Provider except ImportError: - ConfigResolver = None # type: ignore - Provider = None # type: ignore + ConfigResolver = None + Provider = None logger = logging.getLogger(__name__) diff --git a/certbot/certbot/plugins/dns_test_common.py b/certbot/certbot/plugins/dns_test_common.py index 7a8df9329..64e07d063 100644 --- a/certbot/certbot/plugins/dns_test_common.py +++ b/certbot/certbot/plugins/dns_test_common.py @@ -14,7 +14,7 @@ from certbot.tests import util as test_util if typing.TYPE_CHECKING: from typing_extensions import Protocol else: - Protocol = object # type: ignore + Protocol = object diff --git a/certbot/certbot/plugins/dns_test_common_lexicon.py b/certbot/certbot/plugins/dns_test_common_lexicon.py index 3292eaf80..f4beee64e 100644 --- a/certbot/certbot/plugins/dns_test_common_lexicon.py +++ b/certbot/certbot/plugins/dns_test_common_lexicon.py @@ -20,7 +20,7 @@ except ImportError: # pragma: no cover if typing.TYPE_CHECKING: from typing_extensions import Protocol else: - Protocol = object # type: ignore + Protocol = object diff --git a/mypy.ini b/mypy.ini index cf7a3cc2b..47d31a827 100644 --- a/mypy.ini +++ b/mypy.ini @@ -1,6 +1,5 @@ [mypy] check_untyped_defs = True ignore_missing_imports = True - -[mypy-acme.magic_typing_test] -ignore_errors = True +warn_unused_ignores = True +show_error_codes = True diff --git a/tests/lock_test.py b/tests/lock_test.py index d0da8e369..4ddbd3aef 100644 --- a/tests/lock_test.py +++ b/tests/lock_test.py @@ -144,8 +144,7 @@ def setup_certificate(workspace): :rtype: `tuple` """ # Generate key - # See comment on cryptography import about type: ignore - private_key = rsa.generate_private_key( # type: ignore + private_key = rsa.generate_private_key( public_exponent=65537, key_size=2048, backend=default_backend() @@ -169,7 +168,7 @@ def setup_certificate(workspace): key_path = os.path.join(workspace, 'cert.key') with open(key_path, 'wb') as file_handle: - file_handle.write(private_key.private_bytes( # type: ignore + file_handle.write(private_key.private_bytes( encoding=serialization.Encoding.PEM, format=serialization.PrivateFormat.TraditionalOpenSSL, encryption_algorithm=serialization.NoEncryption() diff --git a/tox.ini b/tox.ini index 467a00400..554ff5b68 100644 --- a/tox.ini +++ b/tox.ini @@ -20,7 +20,8 @@ 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 -source_paths = acme/acme certbot/certbot certbot-ci/certbot_integration_tests certbot-apache/certbot_apache 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 +fully_typed_source_paths = acme/acme +partially_typed_source_paths = certbot/certbot certbot-ci/certbot_integration_tests certbot-apache/certbot_apache 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 = @@ -125,7 +126,7 @@ 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]source_paths} + python -m pylint --reports=n --rcfile=.pylintrc {[base]fully_typed_source_paths} {[base]partially_typed_source_paths} // TODO: Re-enable strict checks for optionals with appropriate type corrections or code redesign. [testenv:mypy{,-win,-posix}] @@ -133,7 +134,8 @@ basepython = python3 commands = win: {[base]pip_install} {[base]win_all_packages} !win: {[base]pip_install} {[base]all_packages} certbot-ci - mypy {[base]source_paths} + mypy --disallow-untyped-defs {[base]fully_typed_source_paths} + mypy {[base]partially_typed_source_paths} [testenv:apacheconftest] commands =