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

Add type annotations to the acme project (#9036)

Browse Source 
* 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 <bmw@users.noreply.github.com>

* Update acme/acme/client.py

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

* Various fixes

* Fix code

* Fix code

* Update acme/acme/client.py

Co-authored-by: Brad Warren <bmw@users.noreply.github.com>

* Update acme/acme/challenges.py

Co-authored-by: Brad Warren <bmw@users.noreply.github.com>

* Update acme/acme/client.py

Co-authored-by: Brad Warren <bmw@users.noreply.github.com>

* 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 da7338137b.

* Let's find a smarter way than "type: ignore"

* Update certbot/certbot/_internal/account.py

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

* Fix an annotation

Co-authored-by: Brad Warren <bmw@users.noreply.github.com>
Co-authored-by: alexzorin <alex@zor.io>
This commit is contained in:
Adrien Ferrand
2021-10-25 09:43:21 +11:00
committed by GitHub
co-authored by Brad Warren alexzorin
parent 94af235713
commit a0f22d21ce
23 changed files with 431 additions and 285 deletions
Download Patch File Download Diff File Expand all files Collapse all files
acme/acme/challenges.py
+55 -41
View File
@@ -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)
acme/acme/client.py
+127 -84
View File
@@ -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)
acme/acme/crypto_util.py
+53 -32
View File
@@ -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)
acme/acme/errors.py
+25 -10
View File
@@ -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__()
acme/acme/fields.py
+10 -7
View File
@@ -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(
acme/acme/jws.py
+8 -5
View File
@@ -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)
acme/acme/magic_typing.py
+3 -2
View File
@@ -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
acme/acme/messages.py
+60 -48
View File
@@ -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.
acme/acme/mixins.py
+12 -9
View File
@@ -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
acme/acme/standalone.py
+41 -26
View File
@@ -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)
acme/acme/util.py
+6 -1
View File
@@ -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()}
acme/tests/challenges_test.py
+2 -2
View File
@@ -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')
certbot-apache/certbot_apache/_internal/parser.py
+1 -1
View File
@@ -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__)
certbot/certbot/_internal/account.py
+7 -1
View File
@@ -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,
certbot/certbot/_internal/client.py
+6 -1
View File
@@ -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,
certbot/certbot/compat/_path.py
+1 -1
View File
@@ -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).
certbot/certbot/compat/os.py
+1 -1
View File
@@ -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
certbot/certbot/plugins/dns_common_lexicon.py
+2 -2
View File
@@ -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__)
certbot/certbot/plugins/dns_test_common.py
+1 -1
View File
@@ -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
certbot/certbot/plugins/dns_test_common_lexicon.py
+1 -1
View File
@@ -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
mypy.ini
+2 -3
View File
@@ -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
tests/lock_test.py
+2 -3
View File
@@ -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()
tox.ini
+5 -3
View File
@@ -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 =