"""ACME client API."""
# pylint: disable=too-many-lines
# This pylint disable can be deleted once the deprecated ACMEv1 code is
# removed.
import base64
import collections
import datetime
from email.utils import parsedate_tz
import heapq
import http.client as http_client
import logging
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
import josepy as jose
import OpenSSL
import requests
from requests.adapters import HTTPAdapter
from requests.utils import parse_header_links
from requests_toolbelt.adapters.source import SourceAddressAdapter
from acme import crypto_util
from acme import errors
from acme import jws
from acme import messages
from acme.mixins import VersionedLEACMEMixin
logger = logging.getLogger(__name__)
DEFAULT_NETWORK_TIMEOUT = 45
DER_CONTENT_TYPE = 'application/pkix-cert'
[docs]class ClientBase:
"""ACME client base object.
:ivar messages.Directory directory:
:ivar .ClientNetwork net: Client network.
:ivar int acme_version: ACME protocol version. 1 or 2.
"""
def __init__(self, directory: messages.Directory, net: 'ClientNetwork',
acme_version: int) -> None:
"""Initialize.
:param .messages.Directory directory: Directory Resource
:param .ClientNetwork net: Client network.
:param int acme_version: ACME protocol version. 1 or 2.
"""
self.directory = directory
self.net = net
self.acme_version = acme_version
@classmethod
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']
return messages.RegistrationResource(
body=messages.Registration.from_json(response.json()),
uri=response.headers.get('Location', uri),
terms_of_service=terms_of_service)
def _send_recv_regr(self, regr: messages.RegistrationResource,
body: messages.Registration) -> messages.RegistrationResource:
response = self._post(regr.uri, body)
# TODO: Boulder returns httplib.ACCEPTED
#assert response.status_code == httplib.OK
# TODO: Boulder does not set Location or Link on update
# (c.f. acme-spec #94)
return self._regr_from_response(
response, uri=regr.uri,
terms_of_service=regr.terms_of_service)
def _post(self, *args: Any, **kwargs: Any) -> requests.Response:
"""Wrapper around self.net.post that adds the acme_version.
"""
kwargs.setdefault('acme_version', self.acme_version)
if hasattr(self.directory, 'newNonce'):
kwargs.setdefault('new_nonce_url', getattr(self.directory, 'newNonce'))
return self.net.post(*args, **kwargs)
[docs] def update_registration(self, regr: messages.RegistrationResource,
update: Optional[messages.Registration] = None
) -> messages.RegistrationResource:
"""Update registration.
:param messages.RegistrationResource regr: Registration Resource.
:param messages.Registration update: Updated body of the
resource. If not provided, body will be taken from `regr`.
:returns: Updated Registration Resource.
:rtype: `.RegistrationResource`
"""
update = regr.body if update is None else update
body = messages.UpdateRegistration(**dict(update))
updated_regr = self._send_recv_regr(regr, body=body)
self.net.account = updated_regr
return updated_regr
[docs] def deactivate_registration(self, regr: messages.RegistrationResource
) -> messages.RegistrationResource:
"""Deactivate registration.
:param messages.RegistrationResource regr: The Registration Resource
to be deactivated.
:returns: The Registration resource that was deactivated.
:rtype: `.RegistrationResource`
"""
return self.update_registration(regr, messages.Registration.from_json(
{"status": "deactivated", "contact": None}))
[docs] def deactivate_authorization(self,
authzr: messages.AuthorizationResource
) -> messages.AuthorizationResource:
"""Deactivate authorization.
:param messages.AuthorizationResource authzr: The Authorization resource
to be deactivated.
:returns: The Authorization resource that was deactivated.
:rtype: `.AuthorizationResource`
"""
body = messages.UpdateAuthorization(status='deactivated')
response = self._post(authzr.uri, body)
return self._authzr_from_response(response,
authzr.body.identifier, authzr.uri)
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))
if identifier is not None and authzr.body.identifier != identifier:
raise errors.UnexpectedUpdate(authzr)
return authzr
[docs] def answer_challenge(self, challb: messages.ChallengeBody, response: requests.Response
) -> messages.ChallengeResource:
"""Answer challenge.
:param challb: Challenge Resource body.
:type challb: `.ChallengeBody`
:param response: Corresponding Challenge response
:type response: `.challenges.ChallengeResponse`
:returns: Challenge Resource with updated body.
:rtype: `.ChallengeResource`
:raises .UnexpectedUpdate:
"""
response = self._post(challb.uri, response)
try:
authzr_uri = response.links['up']['url']
except KeyError:
raise errors.ClientError('"up" Link header missing')
challr = messages.ChallengeResource(
authzr_uri=authzr_uri,
body=messages.ChallengeBody.from_json(response.json