How to use josepy - 10 common examples

def uninitialized_client(key=None):
    if key is None:
        key = josepy.JWKRSA(key=rsa.generate_private_key(65537, 2048, default_backend()))
    net = acme_client.ClientNetwork(key, user_agent="Boulder integration tester")
    directory = messages.Directory.from_json(net.get(DIRECTORY_V2).json())
    return acme_client.ClientV2(directory, net)

def make_client(email=None):
    """Build an acme.Client and register a new account with a random key."""
    key = josepy.JWKRSA(key=rsa.generate_private_key(65537, 2048, default_backend()))

    net = acme_client.ClientNetwork(key, user_agent="Boulder integration tester")

    client = acme_client.Client(DIRECTORY, key=key, net=net)
    account = client.register(messages.NewRegistration.from_data(email=email))
    client.agree_to_tos(account)
    client.account = account
    return client

def load_comparable_csr(*names):
    """Load ComparableX509 certificate request."""
    return jose.ComparableX509(load_csr(*names))

cleanup = chisel2.do_http_challenges(client, order.authorizations)
    try:
        order = client.poll_and_finalize(order)
    finally:
        cleanup()

    # Create a new client with the JWK as the cert private key
    jwk = josepy.JWKRSA(key=key)
    net = acme_client.ClientNetwork(key, user_agent="Boulder integration tester")

    directory = Directory.from_json(net.get(chisel2.DIRECTORY_V2).json())
    new_client = acme_client.ClientV2(directory, net)

    cert = OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM, order.fullchain_pem)
    reset_akamai_purges()
    client.revoke(josepy.ComparableX509(cert), 0)

    cert_file = tempfile.NamedTemporaryFile(
        dir=tempdir, suffix='.test_revoke_by_privkey.pem',
        mode='w+', delete=False)
    cert_file.write(OpenSSL.crypto.dump_certificate(
        OpenSSL.crypto.FILETYPE_PEM, cert).decode())
    cert_file.close()
    verify_ocsp(cert_file.name, "/tmp/intermediate-cert-rsa-a.pem", "http://localhost:4002", "revoked")
    verify_akamai_purge()

def _test_create_client(self):
        with start_action(action_type=u'integration:create_client').context():
            self.key = JWKRSA(key=generate_private_key('rsa'))
            return (
                DeferredContext(self._create_client(self.key))
                .addActionFinish())

def load_rsa_private_key(*names):
    """Load RSA private key."""
    loader = _guess_loader(names[-1], serialization.load_pem_private_key,
                           serialization.load_der_private_key)
    return jose.ComparableRSAKey(loader(
        load_vector(*names), password=None, backend=default_backend()))

"""Base test class for DNS authenticators."""

import configobj
import josepy as jose
import mock
import six

from acme import challenges

from certbot import achallenges
from certbot.compat import security
from certbot.tests import acme_util
from certbot.tests import util as test_util

DOMAIN = 'example.com'
KEY = jose.JWKRSA.load(test_util.load_vector("rsa512_key.pem"))


class BaseAuthenticatorTest(object):
    """
    A base test class to reduce duplication between test code for DNS Authenticator Plugins.

    Assumes:
     * That subclasses also subclass unittest.TestCase
     * That the authenticator is stored as self.auth
    """

    achall = achallenges.KeyAuthorizationAnnotatedChallenge(
        challb=acme_util.DNS01, domain=DOMAIN, account_key=KEY)

    def test_more_info(self):
        # pylint: disable=no-member

"""Base test class for DNS authenticators built on Lexicon."""

import josepy as jose
import mock
from requests.exceptions import HTTPError
from requests.exceptions import RequestException

from certbot import errors
from certbot.plugins import dns_test_common
from certbot.tests import util as test_util

DOMAIN = 'example.com'
KEY = jose.JWKRSA.load(test_util.load_vector("rsa512_key.pem"))

# These classes are intended to be subclassed/mixed in, so not all members are defined.
# pylint: disable=no-member

class BaseLexiconAuthenticatorTest(dns_test_common.BaseAuthenticatorTest):

    def test_perform(self):
        self.auth.perform([self.achall])

        expected = [mock.call.add_txt_record(DOMAIN, '_acme-challenge.'+DOMAIN, mock.ANY)]
        self.assertEqual(expected, self.mock_client.mock_calls)

    def test_cleanup(self):
        self.auth._attempt_cleanup = True  # _attempt_cleanup | pylint: disable=protected-access
        self.auth.cleanup([self.achall])

class Registration(ResourceBody):
    """Registration Resource Body.

    :ivar josepy.jwk.JWK key: Public key.
    :ivar tuple contact: Contact information following ACME spec,
        `tuple` of `unicode`.
    :ivar unicode agreement:

    """
    # on new-reg key server ignores 'key' and populates it based on
    # JWS.signature.combined.jwk
    key = jose.Field('key', omitempty=True, decoder=jose.JWK.from_json)
    contact = jose.Field('contact', omitempty=True, default=())
    agreement = jose.Field('agreement', omitempty=True)
    status = jose.Field('status', omitempty=True)
    terms_of_service_agreed = jose.Field('termsOfServiceAgreed', omitempty=True)
    only_return_existing = jose.Field('onlyReturnExisting', omitempty=True)
    external_account_binding = jose.Field('externalAccountBinding', omitempty=True)

    phone_prefix = 'tel:'
    email_prefix = 'mailto:'

    @classmethod
    def from_data(cls, phone=None, email=None, external_account_binding=None, **kwargs):
        """Create registration resource from contact details."""
        details = list(kwargs.pop('contact', ()))
        if phone is not None:
            details.append(cls.phone_prefix + phone)
        if email is not None:
            details.extend([cls.email_prefix + mail for mail in email.split(',')])
        kwargs['contact'] = tuple(details)

def get_or_gen_key(ctx, account_key_path, new_account_key_size):
    account_key_path = os.path.expanduser(account_key_path)
    if os.path.exists(account_key_path):
        logger.debug('opening existing account key %s', account_key_path)
        with open(account_key_path, 'rb') as key_file:
            key_contents = key_file.read()
            try:
                try:
                    account_key = jose.JWKRSA(key=serialization.load_pem_private_key(key_contents, None,
                                              default_backend()))
                except TypeError:  # password required
                    password = click.prompt('Password for %s' % account_key_path, hide_input=True, default=None)
                    key = serialization.load_pem_private_key(key_contents, password.encode('utf-8'), default_backend())
                    account_key = jose.JWKRSA(key=key)
            except ValueError as e:
                logger.error('could not open key %s: %s', account_key_path, e)
                ctx.exit(1)
    else:
        logger.warning('no account key found; creating a new %d bit key in %s', new_account_key_size, account_key_path)
        account_key = jose.JWKRSA(key=rsa.generate_private_key(
            public_exponent=65537,
            key_size=new_account_key_size,
            backend=default_backend()))
        try:
            os.makedirs(os.path.dirname(account_key_path), 0o750)