diff --git a/acme/acme/challenges.py b/acme/acme/challenges.py index 14641af10..96997297b 100644 --- a/acme/acme/challenges.py +++ b/acme/acme/challenges.py @@ -6,13 +6,13 @@ import logging import socket from cryptography.hazmat.primitives import hashes # type: ignore +import josepy as jose import OpenSSL import requests from acme import errors from acme import crypto_util from acme import fields -from acme import jose logger = logging.getLogger(__name__) diff --git a/acme/acme/challenges_test.py b/acme/acme/challenges_test.py index 49e790102..834d569aa 100644 --- a/acme/acme/challenges_test.py +++ b/acme/acme/challenges_test.py @@ -1,6 +1,7 @@ """Tests for acme.challenges.""" import unittest +import josepy as jose import mock import OpenSSL import requests @@ -8,7 +9,6 @@ import requests from six.moves.urllib import parse as urllib_parse # pylint: disable=import-error from acme import errors -from acme import jose from acme import test_util CERT = test_util.load_comparable_cert('cert.pem') diff --git a/acme/acme/client.py b/acme/acme/client.py index 2e07d34d7..dc5efbe86 100644 --- a/acme/acme/client.py +++ b/acme/acme/client.py @@ -10,13 +10,13 @@ import time import six from six.moves import http_client # pylint: disable=import-error +import josepy as jose import OpenSSL import re import requests import sys from acme import errors -from acme import jose from acme import jws from acme import messages @@ -408,7 +408,7 @@ class Client(object): # pylint: disable=too-many-instance-attributes :param str uri: URI of certificate :returns: tuple of the form - (response, :class:`acme.jose.ComparableX509`) + (response, :class:`josepy.util.ComparableX509`) :rtype: tuple """ diff --git a/acme/acme/client_test.py b/acme/acme/client_test.py index 4bd762865..84620fc99 100644 --- a/acme/acme/client_test.py +++ b/acme/acme/client_test.py @@ -5,12 +5,12 @@ import unittest from six.moves import http_client # pylint: disable=import-error +import josepy as jose import mock import requests from acme import challenges from acme import errors -from acme import jose from acme import jws as acme_jws from acme import messages from acme import messages_test diff --git a/acme/acme/crypto_util_test.py b/acme/acme/crypto_util_test.py index da433c5a2..1d7f83ccf 100644 --- a/acme/acme/crypto_util_test.py +++ b/acme/acme/crypto_util_test.py @@ -8,10 +8,10 @@ import unittest import six from six.moves import socketserver #type: ignore # pylint: disable=import-error +import josepy as jose import OpenSSL from acme import errors -from acme import jose from acme import test_util diff --git a/acme/acme/errors.py b/acme/acme/errors.py index 9d991fd75..de5f9d1f4 100644 --- a/acme/acme/errors.py +++ b/acme/acme/errors.py @@ -1,5 +1,5 @@ """ACME errors.""" -from acme.jose import errors as jose_errors +from josepy import errors as jose_errors class Error(Exception): diff --git a/acme/acme/fields.py b/acme/acme/fields.py index 12d09acf4..d7ec78403 100644 --- a/acme/acme/fields.py +++ b/acme/acme/fields.py @@ -1,10 +1,9 @@ """ACME JSON fields.""" import logging +import josepy as jose import pyrfc3339 -from acme import jose - logger = logging.getLogger(__name__) diff --git a/acme/acme/fields_test.py b/acme/acme/fields_test.py index de852b6fa..69dde8b89 100644 --- a/acme/acme/fields_test.py +++ b/acme/acme/fields_test.py @@ -2,10 +2,9 @@ import datetime import unittest +import josepy as jose import pytz -from acme import jose - class FixedTest(unittest.TestCase): """Tests for acme.fields.Fixed.""" diff --git a/acme/acme/jose/__init__.py b/acme/acme/jose/__init__.py deleted file mode 100644 index 9116bc433..000000000 --- a/acme/acme/jose/__init__.py +++ /dev/null @@ -1,82 +0,0 @@ -"""Javascript Object Signing and Encryption (jose). - -This package is a Python implementation of the standards developed by -IETF `Javascript Object Signing and Encryption (Active WG)`_, in -particular the following RFCs: - - - `JSON Web Algorithms (JWA)`_ - - `JSON Web Key (JWK)`_ - - `JSON Web Signature (JWS)`_ - - -.. _`Javascript Object Signing and Encryption (Active WG)`: - https://tools.ietf.org/wg/jose/ - -.. _`JSON Web Algorithms (JWA)`: - https://datatracker.ietf.org/doc/draft-ietf-jose-json-web-algorithms/ - -.. _`JSON Web Key (JWK)`: - https://datatracker.ietf.org/doc/draft-ietf-jose-json-web-key/ - -.. _`JSON Web Signature (JWS)`: - https://datatracker.ietf.org/doc/draft-ietf-jose-json-web-signature/ - -""" -from acme.jose.b64 import ( - b64decode, - b64encode, -) - -from acme.jose.errors import ( - DeserializationError, - SerializationError, - Error, - UnrecognizedTypeError, -) - -from acme.jose.interfaces import JSONDeSerializable - -from acme.jose.json_util import ( - Field, - JSONObjectWithFields, - TypedJSONObjectWithFields, - decode_b64jose, - decode_cert, - decode_csr, - decode_hex16, - encode_b64jose, - encode_cert, - encode_csr, - encode_hex16, -) - -from acme.jose.jwa import ( - HS256, - HS384, - HS512, - JWASignature, - PS256, - PS384, - PS512, - RS256, - RS384, - RS512, -) - -from acme.jose.jwk import ( - JWK, - JWKRSA, -) - -from acme.jose.jws import ( - Header, - JWS, - Signature, -) - -from acme.jose.util import ( - ComparableX509, - ComparableKey, - ComparableRSAKey, - ImmutableMap, -) diff --git a/acme/acme/jose/b64.py b/acme/acme/jose/b64.py deleted file mode 100644 index cf79aa820..000000000 --- a/acme/acme/jose/b64.py +++ /dev/null @@ -1,61 +0,0 @@ -"""JOSE Base64. - -`JOSE Base64`_ is defined as: - - - URL-safe Base64 - - padding stripped - - -.. _`JOSE Base64`: - https://tools.ietf.org/html/draft-ietf-jose-json-web-signature-37#appendix-C - -.. Do NOT try to call this module "base64", as it will "shadow" the - standard library. - -""" -import base64 - -import six - - -def b64encode(data): - """JOSE Base64 encode. - - :param data: Data to be encoded. - :type data: `bytes` - - :returns: JOSE Base64 string. - :rtype: bytes - - :raises TypeError: if `data` is of incorrect type - - """ - if not isinstance(data, six.binary_type): - raise TypeError('argument should be {0}'.format(six.binary_type)) - return base64.urlsafe_b64encode(data).rstrip(b'=') - - -def b64decode(data): - """JOSE Base64 decode. - - :param data: Base64 string to be decoded. If it's unicode, then - only ASCII characters are allowed. - :type data: `bytes` or `unicode` - - :returns: Decoded data. - :rtype: bytes - - :raises TypeError: if input is of incorrect type - :raises ValueError: if input is unicode with non-ASCII characters - - """ - if isinstance(data, six.string_types): - try: - data = data.encode('ascii') - except UnicodeEncodeError: - raise ValueError( - 'unicode argument should contain only ASCII characters') - elif not isinstance(data, six.binary_type): - raise TypeError('argument should be a str or unicode') - - return base64.urlsafe_b64decode(data + b'=' * (4 - (len(data) % 4))) diff --git a/acme/acme/jose/b64_test.py b/acme/acme/jose/b64_test.py deleted file mode 100644 index cbabe2251..000000000 --- a/acme/acme/jose/b64_test.py +++ /dev/null @@ -1,77 +0,0 @@ -"""Tests for acme.jose.b64.""" -import unittest - -import six - - -# https://en.wikipedia.org/wiki/Base64#Examples -B64_PADDING_EXAMPLES = { - b'any carnal pleasure.': (b'YW55IGNhcm5hbCBwbGVhc3VyZS4', b'='), - b'any carnal pleasure': (b'YW55IGNhcm5hbCBwbGVhc3VyZQ', b'=='), - b'any carnal pleasur': (b'YW55IGNhcm5hbCBwbGVhc3Vy', b''), - b'any carnal pleasu': (b'YW55IGNhcm5hbCBwbGVhc3U', b'='), - b'any carnal pleas': (b'YW55IGNhcm5hbCBwbGVhcw', b'=='), -} - - -B64_URL_UNSAFE_EXAMPLES = { - six.int2byte(251) + six.int2byte(239): b'--8', - six.int2byte(255) * 2: b'__8', -} - - -class B64EncodeTest(unittest.TestCase): - """Tests for acme.jose.b64.b64encode.""" - - @classmethod - def _call(cls, data): - from acme.jose.b64 import b64encode - return b64encode(data) - - def test_empty(self): - self.assertEqual(self._call(b''), b'') - - def test_unsafe_url(self): - for text, b64 in six.iteritems(B64_URL_UNSAFE_EXAMPLES): - self.assertEqual(self._call(text), b64) - - def test_different_paddings(self): - for text, (b64, _) in six.iteritems(B64_PADDING_EXAMPLES): - self.assertEqual(self._call(text), b64) - - def test_unicode_fails_with_type_error(self): - self.assertRaises(TypeError, self._call, u'some unicode') - - -class B64DecodeTest(unittest.TestCase): - """Tests for acme.jose.b64.b64decode.""" - - @classmethod - def _call(cls, data): - from acme.jose.b64 import b64decode - return b64decode(data) - - def test_unsafe_url(self): - for text, b64 in six.iteritems(B64_URL_UNSAFE_EXAMPLES): - self.assertEqual(self._call(b64), text) - - def test_input_without_padding(self): - for text, (b64, _) in six.iteritems(B64_PADDING_EXAMPLES): - self.assertEqual(self._call(b64), text) - - def test_input_with_padding(self): - for text, (b64, pad) in six.iteritems(B64_PADDING_EXAMPLES): - self.assertEqual(self._call(b64 + pad), text) - - def test_unicode_with_ascii(self): - self.assertEqual(self._call(u'YQ'), b'a') - - def test_non_ascii_unicode_fails(self): - self.assertRaises(ValueError, self._call, u'\u0105') - - def test_type_error_no_unicode_or_bytes(self): - self.assertRaises(TypeError, self._call, object()) - - -if __name__ == '__main__': - unittest.main() # pragma: no cover diff --git a/acme/acme/jose/errors.py b/acme/acme/jose/errors.py deleted file mode 100644 index 74c9443e1..000000000 --- a/acme/acme/jose/errors.py +++ /dev/null @@ -1,35 +0,0 @@ -"""JOSE errors.""" - - -class Error(Exception): - """Generic JOSE Error.""" - - -class DeserializationError(Error): - """JSON deserialization error.""" - - def __str__(self): - return "Deserialization error: {0}".format( - super(DeserializationError, self).__str__()) - - -class SerializationError(Error): - """JSON serialization error.""" - - -class UnrecognizedTypeError(DeserializationError): - """Unrecognized type error. - - :ivar str typ: The unrecognized type of the JSON object. - :ivar jobj: Full JSON object. - - """ - - def __init__(self, typ, jobj): - self.typ = typ - self.jobj = jobj - super(UnrecognizedTypeError, self).__init__(str(self)) - - def __str__(self): - return '{0} was not recognized, full message: {1}'.format( - self.typ, self.jobj) diff --git a/acme/acme/jose/errors_test.py b/acme/acme/jose/errors_test.py deleted file mode 100644 index 919980920..000000000 --- a/acme/acme/jose/errors_test.py +++ /dev/null @@ -1,17 +0,0 @@ -"""Tests for acme.jose.errors.""" -import unittest - - -class UnrecognizedTypeErrorTest(unittest.TestCase): - def setUp(self): - from acme.jose.errors import UnrecognizedTypeError - self.error = UnrecognizedTypeError('foo', {'type': 'foo'}) - - def test_str(self): - self.assertEqual( - "foo was not recognized, full message: {'type': 'foo'}", - str(self.error)) - - -if __name__ == '__main__': - unittest.main() # pragma: no cover diff --git a/acme/acme/jose/interfaces.py b/acme/acme/jose/interfaces.py deleted file mode 100644 index f841848b3..000000000 --- a/acme/acme/jose/interfaces.py +++ /dev/null @@ -1,216 +0,0 @@ -"""JOSE interfaces.""" -import abc -import collections -import json - -import six - -from acme.jose import errors -from acme.jose import util - -# pylint: disable=no-self-argument,no-method-argument,no-init,inherit-non-class -# pylint: disable=too-few-public-methods - - -@six.add_metaclass(abc.ABCMeta) -class JSONDeSerializable(object): - # pylint: disable=too-few-public-methods - """Interface for (de)serializable JSON objects. - - Please recall, that standard Python library implements - :class:`json.JSONEncoder` and :class:`json.JSONDecoder` that perform - translations based on respective :ref:`conversion tables - ` that look pretty much like the one below (for - complete tables see relevant Python documentation): - - .. _conversion-table: - - ====== ====== - JSON Python - ====== ====== - object dict - ... ... - ====== ====== - - While the above **conversion table** is about translation of JSON - documents to/from the basic Python types only, - :class:`JSONDeSerializable` introduces the following two concepts: - - serialization - Turning an arbitrary Python object into Python object that can - be encoded into a JSON document. **Full serialization** produces - a Python object composed of only basic types as required by the - :ref:`conversion table `. **Partial - serialization** (accomplished by :meth:`to_partial_json`) - produces a Python object that might also be built from other - :class:`JSONDeSerializable` objects. - - deserialization - Turning a decoded Python object (necessarily one of the basic - types as required by the :ref:`conversion table - `) into an arbitrary Python object. - - Serialization produces **serialized object** ("partially serialized - object" or "fully serialized object" for partial and full - serialization respectively) and deserialization produces - **deserialized object**, both usually denoted in the source code as - ``jobj``. - - Wording in the official Python documentation might be confusing - after reading the above, but in the light of those definitions, one - can view :meth:`json.JSONDecoder.decode` as decoder and - deserializer of basic types, :meth:`json.JSONEncoder.default` as - serializer of basic types, :meth:`json.JSONEncoder.encode` as - serializer and encoder of basic types. - - One could extend :mod:`json` to support arbitrary object - (de)serialization either by: - - - overriding :meth:`json.JSONDecoder.decode` and - :meth:`json.JSONEncoder.default` in subclasses - - - or passing ``object_hook`` argument (or ``object_hook_pairs``) - to :func:`json.load`/:func:`json.loads` or ``default`` argument - for :func:`json.dump`/:func:`json.dumps`. - - Interestingly, ``default`` is required to perform only partial - serialization, as :func:`json.dumps` applies ``default`` - recursively. This is the idea behind making :meth:`to_partial_json` - produce only partial serialization, while providing custom - :meth:`json_dumps` that dumps with ``default`` set to - :meth:`json_dump_default`. - - To make further documentation a bit more concrete, please, consider - the following imaginatory implementation example:: - - class Foo(JSONDeSerializable): - def to_partial_json(self): - return 'foo' - - @classmethod - def from_json(cls, jobj): - return Foo() - - class Bar(JSONDeSerializable): - def to_partial_json(self): - return [Foo(), Foo()] - - @classmethod - def from_json(cls, jobj): - return Bar() - - """ - - @abc.abstractmethod - def to_partial_json(self): # pragma: no cover - """Partially serialize. - - Following the example, **partial serialization** means the following:: - - assert isinstance(Bar().to_partial_json()[0], Foo) - assert isinstance(Bar().to_partial_json()[1], Foo) - - # in particular... - assert Bar().to_partial_json() != ['foo', 'foo'] - - :raises acme.jose.errors.SerializationError: - in case of any serialization error. - :returns: Partially serializable object. - - """ - raise NotImplementedError() - - def to_json(self): - """Fully serialize. - - Again, following the example from before, **full serialization** - means the following:: - - assert Bar().to_json() == ['foo', 'foo'] - - :raises acme.jose.errors.SerializationError: - in case of any serialization error. - :returns: Fully serialized object. - - """ - def _serialize(obj): - if isinstance(obj, JSONDeSerializable): - return _serialize(obj.to_partial_json()) - if isinstance(obj, six.string_types): # strings are Sequence - return obj - elif isinstance(obj, list): - return [_serialize(subobj) for subobj in obj] - elif isinstance(obj, collections.Sequence): - # default to tuple, otherwise Mapping could get - # unhashable list - return tuple(_serialize(subobj) for subobj in obj) - elif isinstance(obj, collections.Mapping): - return dict((_serialize(key), _serialize(value)) - for key, value in six.iteritems(obj)) - else: - return obj - - return _serialize(self) - - @util.abstractclassmethod - def from_json(cls, jobj): # pylint: disable=unused-argument - """Deserialize a decoded JSON document. - - :param jobj: Python object, composed of only other basic data - types, as decoded from JSON document. Not necessarily - :class:`dict` (as decoded from "JSON object" document). - - :raises acme.jose.errors.DeserializationError: - if decoding was unsuccessful, e.g. in case of unparseable - X509 certificate, or wrong padding in JOSE base64 encoded - string, etc. - - """ - # TypeError: Can't instantiate abstract class with - # abstract methods from_json, to_partial_json - return cls() # pylint: disable=abstract-class-instantiated - - @classmethod - def json_loads(cls, json_string): - """Deserialize from JSON document string.""" - try: - loads = json.loads(json_string) - except ValueError as error: - raise errors.DeserializationError(error) - return cls.from_json(loads) - - def json_dumps(self, **kwargs): - """Dump to JSON string using proper serializer. - - :returns: JSON document string. - :rtype: str - - """ - return json.dumps(self, default=self.json_dump_default, **kwargs) - - def json_dumps_pretty(self): - """Dump the object to pretty JSON document string. - - :rtype: str - - """ - return self.json_dumps(sort_keys=True, indent=4, separators=(',', ': ')) - - @classmethod - def json_dump_default(cls, python_object): - """Serialize Python object. - - This function is meant to be passed as ``default`` to - :func:`json.dump` or :func:`json.dumps`. They call - ``default(python_object)`` only for non-basic Python types, so - this function necessarily raises :class:`TypeError` if - ``python_object`` is not an instance of - :class:`IJSONSerializable`. - - Please read the class docstring for more information. - - """ - if isinstance(python_object, JSONDeSerializable): - return python_object.to_partial_json() - else: # this branch is necessary, cannot just "return" - raise TypeError(repr(python_object) + ' is not JSON serializable') diff --git a/acme/acme/jose/interfaces_test.py b/acme/acme/jose/interfaces_test.py deleted file mode 100644 index cf98ff371..000000000 --- a/acme/acme/jose/interfaces_test.py +++ /dev/null @@ -1,114 +0,0 @@ -"""Tests for acme.jose.interfaces.""" -import unittest - - -class JSONDeSerializableTest(unittest.TestCase): - # pylint: disable=too-many-instance-attributes - - def setUp(self): - from acme.jose.interfaces import JSONDeSerializable - - # pylint: disable=missing-docstring,invalid-name - - class Basic(JSONDeSerializable): - def __init__(self, v): - self.v = v - - def to_partial_json(self): - return self.v - - @classmethod - def from_json(cls, jobj): - return cls(jobj) - - class Sequence(JSONDeSerializable): - def __init__(self, x, y): - self.x = x - self.y = y - - def to_partial_json(self): - return [self.x, self.y] - - @classmethod - def from_json(cls, jobj): - return cls( - Basic.from_json(jobj[0]), Basic.from_json(jobj[1])) - - class Mapping(JSONDeSerializable): - def __init__(self, x, y): - self.x = x - self.y = y - - def to_partial_json(self): - return {self.x: self.y} - - @classmethod - def from_json(cls, jobj): - pass # pragma: no cover - - self.basic1 = Basic('foo1') - self.basic2 = Basic('foo2') - self.seq = Sequence(self.basic1, self.basic2) - self.mapping = Mapping(self.basic1, self.basic2) - self.nested = Basic([[self.basic1]]) - self.tuple = Basic(('foo',)) - - # pylint: disable=invalid-name - self.Basic = Basic - self.Sequence = Sequence - self.Mapping = Mapping - - def test_to_json_sequence(self): - self.assertEqual(self.seq.to_json(), ['foo1', 'foo2']) - - def test_to_json_mapping(self): - self.assertEqual(self.mapping.to_json(), {'foo1': 'foo2'}) - - def test_to_json_other(self): - mock_value = object() - self.assertTrue(self.Basic(mock_value).to_json() is mock_value) - - def test_to_json_nested(self): - self.assertEqual(self.nested.to_json(), [['foo1']]) - - def test_to_json(self): - self.assertEqual(self.tuple.to_json(), (('foo', ))) - - def test_from_json_not_implemented(self): - from acme.jose.interfaces import JSONDeSerializable - self.assertRaises(TypeError, JSONDeSerializable.from_json, 'xxx') - - def test_json_loads(self): - seq = self.Sequence.json_loads('["foo1", "foo2"]') - self.assertTrue(isinstance(seq, self.Sequence)) - self.assertTrue(isinstance(seq.x, self.Basic)) - self.assertTrue(isinstance(seq.y, self.Basic)) - self.assertEqual(seq.x.v, 'foo1') - self.assertEqual(seq.y.v, 'foo2') - - def test_json_dumps(self): - self.assertEqual('["foo1", "foo2"]', self.seq.json_dumps()) - - def test_json_dumps_pretty(self): - self.assertEqual(self.seq.json_dumps_pretty(), - '[\n "foo1",\n "foo2"\n]') - - def test_json_dump_default(self): - from acme.jose.interfaces import JSONDeSerializable - - self.assertEqual( - 'foo1', JSONDeSerializable.json_dump_default(self.basic1)) - - jobj = JSONDeSerializable.json_dump_default(self.seq) - self.assertEqual(len(jobj), 2) - self.assertTrue(jobj[0] is self.basic1) - self.assertTrue(jobj[1] is self.basic2) - - def test_json_dump_default_type_error(self): - from acme.jose.interfaces import JSONDeSerializable - self.assertRaises( - TypeError, JSONDeSerializable.json_dump_default, object()) - - -if __name__ == '__main__': - unittest.main() # pragma: no cover diff --git a/acme/acme/jose/json_util.py b/acme/acme/jose/json_util.py deleted file mode 100644 index 4baadda5e..000000000 --- a/acme/acme/jose/json_util.py +++ /dev/null @@ -1,485 +0,0 @@ -"""JSON (de)serialization framework. - -The framework presented here is somewhat based on `Go's "json" package`_ -(especially the ``omitempty`` functionality). - -.. _`Go's "json" package`: http://golang.org/pkg/encoding/json/ - -""" -import abc -import binascii -import logging - -import OpenSSL -import six - -from acme.jose import b64 -from acme.jose import errors -from acme.jose import interfaces -from acme.jose import util - - -logger = logging.getLogger(__name__) - - -class Field(object): - """JSON object field. - - :class:`Field` is meant to be used together with - :class:`JSONObjectWithFields`. - - ``encoder`` (``decoder``) is a callable that accepts a single - parameter, i.e. a value to be encoded (decoded), and returns the - serialized (deserialized) value. In case of errors it should raise - :class:`~acme.jose.errors.SerializationError` - (:class:`~acme.jose.errors.DeserializationError`). - - Note, that ``decoder`` should perform partial serialization only. - - :ivar str json_name: Name of the field when encoded to JSON. - :ivar default: Default value (used when not present in JSON object). - :ivar bool omitempty: If ``True`` and the field value is empty, then - it will not be included in the serialized JSON object, and - ``default`` will be used for deserialization. Otherwise, if ``False``, - field is considered as required, value will always be included in the - serialized JSON objected, and it must also be present when - deserializing. - - """ - __slots__ = ('json_name', 'default', 'omitempty', 'fdec', 'fenc') - - def __init__(self, json_name, default=None, omitempty=False, - decoder=None, encoder=None): - # pylint: disable=too-many-arguments - self.json_name = json_name - self.default = default - self.omitempty = omitempty - - self.fdec = self.default_decoder if decoder is None else decoder - self.fenc = self.default_encoder if encoder is None else encoder - - @classmethod - def _empty(cls, value): - """Is the provided value considered "empty" for this field? - - This is useful for subclasses that might want to override the - definition of being empty, e.g. for some more exotic data types. - - """ - return not isinstance(value, bool) and not value - - def omit(self, value): - """Omit the value in output?""" - return self._empty(value) and self.omitempty - - def _update_params(self, **kwargs): - current = dict(json_name=self.json_name, default=self.default, - omitempty=self.omitempty, - decoder=self.fdec, encoder=self.fenc) - current.update(kwargs) - return type(self)(**current) # pylint: disable=star-args - - def decoder(self, fdec): - """Descriptor to change the decoder on JSON object field.""" - return self._update_params(decoder=fdec) - - def encoder(self, fenc): - """Descriptor to change the encoder on JSON object field.""" - return self._update_params(encoder=fenc) - - def decode(self, value): - """Decode a value, optionally with context JSON object.""" - return self.fdec(value) - - def encode(self, value): - """Encode a value, optionally with context JSON object.""" - return self.fenc(value) - - @classmethod - def default_decoder(cls, value): - """Default decoder. - - Recursively deserialize into immutable types ( - :class:`acme.jose.util.frozendict` instead of - :func:`dict`, :func:`tuple` instead of :func:`list`). - - """ - # bases cases for different types returned by json.loads - if isinstance(value, list): - return tuple(cls.default_decoder(subvalue) for subvalue in value) - elif isinstance(value, dict): - return util.frozendict( - dict((cls.default_decoder(key), cls.default_decoder(value)) - for key, value in six.iteritems(value))) - else: # integer or string - return value - - @classmethod - def default_encoder(cls, value): - """Default (passthrough) encoder.""" - # field.to_partial_json() is no good as encoder has to do partial - # serialization only - return value - - -class JSONObjectWithFieldsMeta(abc.ABCMeta): - """Metaclass for :class:`JSONObjectWithFields` and its subclasses. - - It makes sure that, for any class ``cls`` with ``__metaclass__`` - set to ``JSONObjectWithFieldsMeta``: - - 1. All fields (attributes of type :class:`Field`) in the class - definition are moved to the ``cls._fields`` dictionary, where - keys are field attribute names and values are fields themselves. - - 2. ``cls.__slots__`` is extended by all field attribute names - (i.e. not :attr:`Field.json_name`). Original ``cls.__slots__`` - are stored in ``cls._orig_slots``. - - In a consequence, for a field attribute name ``some_field``, - ``cls.some_field`` will be a slot descriptor and not an instance - of :class:`Field`. For example:: - - some_field = Field('someField', default=()) - - class Foo(object): - __metaclass__ = JSONObjectWithFieldsMeta - __slots__ = ('baz',) - some_field = some_field - - assert Foo.__slots__ == ('some_field', 'baz') - assert Foo._orig_slots == () - assert Foo.some_field is not Field - - assert Foo._fields.keys() == ['some_field'] - assert Foo._fields['some_field'] is some_field - - As an implementation note, this metaclass inherits from - :class:`abc.ABCMeta` (and not the usual :class:`type`) to mitigate - the metaclass conflict (:class:`ImmutableMap` and - :class:`JSONDeSerializable`, parents of :class:`JSONObjectWithFields`, - use :class:`abc.ABCMeta` as its metaclass). - - """ - - def __new__(mcs, name, bases, dikt): - fields = {} - - for base in bases: - fields.update(getattr(base, '_fields', {})) - # Do not reorder, this class might override fields from base classes! - for key, value in tuple(six.iteritems(dikt)): - # not six.iterkeys() (in-place edit!) - if isinstance(value, Field): - fields[key] = dikt.pop(key) - - dikt['_orig_slots'] = dikt.get('__slots__', ()) - dikt['__slots__'] = tuple( - list(dikt['_orig_slots']) + list(six.iterkeys(fields))) - dikt['_fields'] = fields - - return abc.ABCMeta.__new__(mcs, name, bases, dikt) - - -@six.add_metaclass(JSONObjectWithFieldsMeta) -class JSONObjectWithFields(util.ImmutableMap, interfaces.JSONDeSerializable): - # pylint: disable=too-few-public-methods - """JSON object with fields. - - Example:: - - class Foo(JSONObjectWithFields): - bar = Field('Bar') - empty = Field('Empty', omitempty=True) - - @bar.encoder - def bar(value): - return value + 'bar' - - @bar.decoder - def bar(value): - if not value.endswith('bar'): - raise errors.DeserializationError('No bar suffix!') - return value[:-3] - - assert Foo(bar='baz').to_partial_json() == {'Bar': 'bazbar'} - assert Foo.from_json({'Bar': 'bazbar'}) == Foo(bar='baz') - assert (Foo.from_json({'Bar': 'bazbar', 'Empty': '!'}) - == Foo(bar='baz', empty='!')) - assert Foo(bar='baz').bar == 'baz' - - """ - - @classmethod - def _defaults(cls): - """Get default fields values.""" - return dict([(slot, field.default) for slot, field - in six.iteritems(cls._fields)]) - - def __init__(self, **kwargs): - # pylint: disable=star-args - super(JSONObjectWithFields, self).__init__( - **(dict(self._defaults(), **kwargs))) - - def encode(self, name): - """Encode a single field. - - :param str name: Name of the field to be encoded. - - :raises errors.SerializationError: if field cannot be serialized - :raises errors.Error: if field could not be found - - """ - try: - field = self._fields[name] - except KeyError: - raise errors.Error("Field not found: {0}".format(name)) - - return field.encode(getattr(self, name)) - - def fields_to_partial_json(self): - """Serialize fields to JSON.""" - jobj = {} - omitted = set() - for slot, field in six.iteritems(self._fields): - value = getattr(self, slot) - - if field.omit(value): - omitted.add((slot, value)) - else: - try: - jobj[field.json_name] = field.encode(value) - except errors.SerializationError as error: - raise errors.SerializationError( - 'Could not encode {0} ({1}): {2}'.format( - slot, value, error)) - return jobj - - def to_partial_json(self): - return self.fields_to_partial_json() - - @classmethod - def _check_required(cls, jobj): - missing = set() - for _, field in six.iteritems(cls._fields): - if not field.omitempty and field.json_name not in jobj: - missing.add(field.json_name) - - if missing: - raise errors.DeserializationError( - 'The following fields are required: {0}'.format( - ','.join(missing))) - - @classmethod - def fields_from_json(cls, jobj): - """Deserialize fields from JSON.""" - cls._check_required(jobj) - fields = {} - for slot, field in six.iteritems(cls._fields): - if field.json_name not in jobj and field.omitempty: - fields[slot] = field.default - else: - value = jobj[field.json_name] - try: - fields[slot] = field.decode(value) - except errors.DeserializationError as error: - raise errors.DeserializationError( - 'Could not decode {0!r} ({1!r}): {2}'.format( - slot, value, error)) - return fields - - @classmethod - def from_json(cls, jobj): - return cls(**cls.fields_from_json(jobj)) - - -def encode_b64jose(data): - """Encode JOSE Base-64 field. - - :param bytes data: - :rtype: `unicode` - - """ - # b64encode produces ASCII characters only - return b64.b64encode(data).decode('ascii') - - -def decode_b64jose(data, size=None, minimum=False): - """Decode JOSE Base-64 field. - - :param unicode data: - :param int size: Required length (after decoding). - :param bool minimum: If ``True``, then `size` will be treated as - minimum required length, as opposed to exact equality. - - :rtype: bytes - - """ - error_cls = TypeError if six.PY2 else binascii.Error - try: - decoded = b64.b64decode(data.encode()) - except error_cls as error: - raise errors.DeserializationError(error) - - if size is not None and ((not minimum and len(decoded) != size) or - (minimum and len(decoded) < size)): - raise errors.DeserializationError( - "Expected at least or exactly {0} bytes".format(size)) - - return decoded - - -def encode_hex16(value): - """Hexlify. - - :param bytes value: - :rtype: unicode - - """ - return binascii.hexlify(value).decode() - - -def decode_hex16(value, size=None, minimum=False): - """Decode hexlified field. - - :param unicode value: - :param int size: Required length (after decoding). - :param bool minimum: If ``True``, then `size` will be treated as - minimum required length, as opposed to exact equality. - - :rtype: bytes - - """ - value = value.encode() - if size is not None and ((not minimum and len(value) != size * 2) or - (minimum and len(value) < size * 2)): - raise errors.DeserializationError() - error_cls = TypeError if six.PY2 else binascii.Error - try: - return binascii.unhexlify(value) - except error_cls as error: - raise errors.DeserializationError(error) - - -def encode_cert(cert): - """Encode certificate as JOSE Base-64 DER. - - :type cert: `OpenSSL.crypto.X509` wrapped in `.ComparableX509` - :rtype: unicode - - """ - return encode_b64jose(OpenSSL.crypto.dump_certificate( - OpenSSL.crypto.FILETYPE_ASN1, cert.wrapped)) - - -def decode_cert(b64der): - """Decode JOSE Base-64 DER-encoded certificate. - - :param unicode b64der: - :rtype: `OpenSSL.crypto.X509` wrapped in `.ComparableX509` - - """ - try: - return util.ComparableX509(OpenSSL.crypto.load_certificate( - OpenSSL.crypto.FILETYPE_ASN1, decode_b64jose(b64der))) - except OpenSSL.crypto.Error as error: - raise errors.DeserializationError(error) - - -def encode_csr(csr): - """Encode CSR as JOSE Base-64 DER. - - :type csr: `OpenSSL.crypto.X509Req` wrapped in `.ComparableX509` - :rtype: unicode - - """ - return encode_b64jose(OpenSSL.crypto.dump_certificate_request( - OpenSSL.crypto.FILETYPE_ASN1, csr.wrapped)) - - -def decode_csr(b64der): - """Decode JOSE Base-64 DER-encoded CSR. - - :param unicode b64der: - :rtype: `OpenSSL.crypto.X509Req` wrapped in `.ComparableX509` - - """ - try: - return util.ComparableX509(OpenSSL.crypto.load_certificate_request( - OpenSSL.crypto.FILETYPE_ASN1, decode_b64jose(b64der))) - except OpenSSL.crypto.Error as error: - raise errors.DeserializationError(error) - - -class TypedJSONObjectWithFields(JSONObjectWithFields): - """JSON object with type.""" - - typ = NotImplemented - """Type of the object. Subclasses must override.""" - - type_field_name = "type" - """Field name used to distinguish different object types. - - Subclasses will probably have to override this. - - """ - - TYPES = NotImplemented - """Types registered for JSON deserialization""" - - @classmethod - def register(cls, type_cls, typ=None): - """Register class for JSON deserialization.""" - typ = type_cls.typ if typ is None else typ - cls.TYPES[typ] = type_cls - return type_cls - - @classmethod - def get_type_cls(cls, jobj): - """Get the registered class for ``jobj``.""" - if cls in six.itervalues(cls.TYPES): - if cls.type_field_name not in jobj: - raise errors.DeserializationError( - "Missing type field ({0})".format(cls.type_field_name)) - # cls is already registered type_cls, force to use it - # so that, e.g Revocation.from_json(jobj) fails if - # jobj["type"] != "revocation". - return cls - - if not isinstance(jobj, dict): - raise errors.DeserializationError( - "{0} is not a dictionary object".format(jobj)) - try: - typ = jobj[cls.type_field_name] - except KeyError: - raise errors.DeserializationError("missing type field") - - try: - return cls.TYPES[typ] - except KeyError: - raise errors.UnrecognizedTypeError(typ, jobj) - - def to_partial_json(self): - """Get JSON serializable object. - - :returns: Serializable JSON object representing ACME typed object. - :meth:`validate` will almost certainly not work, due to reasons - explained in :class:`acme.interfaces.IJSONSerializable`. - :rtype: dict - - """ - jobj = self.fields_to_partial_json() - jobj[self.type_field_name] = self.typ - return jobj - - @classmethod - def from_json(cls, jobj): - """Deserialize ACME object from valid JSON object. - - :raises acme.errors.UnrecognizedTypeError: if type - of the ACME object has not been registered. - - """ - # make sure subclasses don't cause infinite recursive from_json calls - type_cls = cls.get_type_cls(jobj) - return type_cls(**type_cls.fields_from_json(jobj)) diff --git a/acme/acme/jose/json_util_test.py b/acme/acme/jose/json_util_test.py deleted file mode 100644 index 25e36211e..000000000 --- a/acme/acme/jose/json_util_test.py +++ /dev/null @@ -1,381 +0,0 @@ -"""Tests for acme.jose.json_util.""" -import itertools -import unittest - -import mock -import six - -from acme import test_util - -from acme.jose import errors -from acme.jose import interfaces -from acme.jose import util - - -CERT = test_util.load_comparable_cert('cert.pem') -CSR = test_util.load_comparable_csr('csr.pem') - - -class FieldTest(unittest.TestCase): - """Tests for acme.jose.json_util.Field.""" - - def test_no_omit_boolean(self): - from acme.jose.json_util import Field - for default, omitempty, value in itertools.product( - [True, False], [True, False], [True, False]): - self.assertFalse( - Field("foo", default=default, omitempty=omitempty).omit(value)) - - def test_descriptors(self): - mock_value = mock.MagicMock() - - # pylint: disable=missing-docstring - - def decoder(unused_value): - return 'd' - - def encoder(unused_value): - return 'e' - - from acme.jose.json_util import Field - field = Field('foo') - - field = field.encoder(encoder) - self.assertEqual('e', field.encode(mock_value)) - - field = field.decoder(decoder) - self.assertEqual('e', field.encode(mock_value)) - self.assertEqual('d', field.decode(mock_value)) - - def test_default_encoder_is_partial(self): - class MockField(interfaces.JSONDeSerializable): - # pylint: disable=missing-docstring - def to_partial_json(self): - return 'foo' # pragma: no cover - - @classmethod - def from_json(cls, jobj): - pass # pragma: no cover - mock_field = MockField() - - from acme.jose.json_util import Field - self.assertTrue(Field.default_encoder(mock_field) is mock_field) - # in particular... - self.assertNotEqual('foo', Field.default_encoder(mock_field)) - - def test_default_encoder_passthrough(self): - mock_value = mock.MagicMock() - from acme.jose.json_util import Field - self.assertTrue(Field.default_encoder(mock_value) is mock_value) - - def test_default_decoder_list_to_tuple(self): - from acme.jose.json_util import Field - self.assertEqual((1, 2, 3), Field.default_decoder([1, 2, 3])) - - def test_default_decoder_dict_to_frozendict(self): - from acme.jose.json_util import Field - obj = Field.default_decoder({'x': 2}) - self.assertTrue(isinstance(obj, util.frozendict)) - self.assertEqual(obj, util.frozendict(x=2)) - - def test_default_decoder_passthrough(self): - mock_value = mock.MagicMock() - from acme.jose.json_util import Field - self.assertTrue(Field.default_decoder(mock_value) is mock_value) - - -class JSONObjectWithFieldsMetaTest(unittest.TestCase): - """Tests for acme.jose.json_util.JSONObjectWithFieldsMeta.""" - - def setUp(self): - from acme.jose.json_util import Field - from acme.jose.json_util import JSONObjectWithFieldsMeta - self.field = Field('Baz') - self.field2 = Field('Baz2') - # pylint: disable=invalid-name,missing-docstring,too-few-public-methods - # pylint: disable=blacklisted-name - - @six.add_metaclass(JSONObjectWithFieldsMeta) - class A(object): - __slots__ = ('bar',) - baz = self.field - - class B(A): - pass - - class C(A): - baz = self.field2 - - self.a_cls = A - self.b_cls = B - self.c_cls = C - - def test_fields(self): - # pylint: disable=protected-access,no-member - self.assertEqual({'baz': self.field}, self.a_cls._fields) - self.assertEqual({'baz': self.field}, self.b_cls._fields) - - def test_fields_inheritance(self): - # pylint: disable=protected-access,no-member - self.assertEqual({'baz': self.field2}, self.c_cls._fields) - - def test_slots(self): - self.assertEqual(('bar', 'baz'), self.a_cls.__slots__) - self.assertEqual(('baz',), self.b_cls.__slots__) - - def test_orig_slots(self): - # pylint: disable=protected-access,no-member - self.assertEqual(('bar',), self.a_cls._orig_slots) - self.assertEqual((), self.b_cls._orig_slots) - - -class JSONObjectWithFieldsTest(unittest.TestCase): - """Tests for acme.jose.json_util.JSONObjectWithFields.""" - # pylint: disable=protected-access - - def setUp(self): - from acme.jose.json_util import JSONObjectWithFields - from acme.jose.json_util import Field - - class MockJSONObjectWithFields(JSONObjectWithFields): - # pylint: disable=invalid-name,missing-docstring,no-self-argument - # pylint: disable=too-few-public-methods - x = Field('x', omitempty=True, - encoder=(lambda x: x * 2), - decoder=(lambda x: x / 2)) - y = Field('y') - z = Field('Z') # on purpose uppercase - - @y.encoder - def y(value): - if value == 500: - raise errors.SerializationError() - return value - - @y.decoder - def y(value): - if value == 500: - raise errors.DeserializationError() - return value - - # pylint: disable=invalid-name - self.MockJSONObjectWithFields = MockJSONObjectWithFields - self.mock = MockJSONObjectWithFields(x=None, y=2, z=3) - - def test_init_defaults(self): - self.assertEqual(self.mock, self.MockJSONObjectWithFields(y=2, z=3)) - - def test_encode(self): - self.assertEqual(10, self.MockJSONObjectWithFields( - x=5, y=0, z=0).encode("x")) - - def test_encode_wrong_field(self): - self.assertRaises(errors.Error, self.mock.encode, 'foo') - - def test_encode_serialization_error_passthrough(self): - self.assertRaises( - errors.SerializationError, - self.MockJSONObjectWithFields(y=500, z=None).encode, "y") - - def test_fields_to_partial_json_omits_empty(self): - self.assertEqual(self.mock.fields_to_partial_json(), {'y': 2, 'Z': 3}) - - def test_fields_from_json_fills_default_for_empty(self): - self.assertEqual( - {'x': None, 'y': 2, 'z': 3}, - self.MockJSONObjectWithFields.fields_from_json({'y': 2, 'Z': 3})) - - def test_fields_from_json_fails_on_missing(self): - self.assertRaises( - errors.DeserializationError, - self.MockJSONObjectWithFields.fields_from_json, {'y': 0}) - self.assertRaises( - errors.DeserializationError, - self.MockJSONObjectWithFields.fields_from_json, {'Z': 0}) - self.assertRaises( - errors.DeserializationError, - self.MockJSONObjectWithFields.fields_from_json, {'x': 0, 'y': 0}) - self.assertRaises( - errors.DeserializationError, - self.MockJSONObjectWithFields.fields_from_json, {'x': 0, 'Z': 0}) - - def test_fields_to_partial_json_encoder(self): - self.assertEqual( - self.MockJSONObjectWithFields(x=1, y=2, z=3).to_partial_json(), - {'x': 2, 'y': 2, 'Z': 3}) - - def test_fields_from_json_decoder(self): - self.assertEqual( - {'x': 2, 'y': 2, 'z': 3}, - self.MockJSONObjectWithFields.fields_from_json( - {'x': 4, 'y': 2, 'Z': 3})) - - def test_fields_to_partial_json_error_passthrough(self): - self.assertRaises( - errors.SerializationError, self.MockJSONObjectWithFields( - x=1, y=500, z=3).to_partial_json) - - def test_fields_from_json_error_passthrough(self): - self.assertRaises( - errors.DeserializationError, - self.MockJSONObjectWithFields.from_json, - {'x': 4, 'y': 500, 'Z': 3}) - - -class DeEncodersTest(unittest.TestCase): - def setUp(self): - self.b64_cert = ( - u'MIIB3jCCAYigAwIBAgICBTkwDQYJKoZIhvcNAQELBQAwdzELMAkGA1UEBhM' - u'CVVMxETAPBgNVBAgMCE1pY2hpZ2FuMRIwEAYDVQQHDAlBbm4gQXJib3IxKz' - u'ApBgNVBAoMIlVuaXZlcnNpdHkgb2YgTWljaGlnYW4gYW5kIHRoZSBFRkYxF' - u'DASBgNVBAMMC2V4YW1wbGUuY29tMB4XDTE0MTIxMTIyMzQ0NVoXDTE0MTIx' - u'ODIyMzQ0NVowdzELMAkGA1UEBhMCVVMxETAPBgNVBAgMCE1pY2hpZ2FuMRI' - u'wEAYDVQQHDAlBbm4gQXJib3IxKzApBgNVBAoMIlVuaXZlcnNpdHkgb2YgTW' - u'ljaGlnYW4gYW5kIHRoZSBFRkYxFDASBgNVBAMMC2V4YW1wbGUuY29tMFwwD' - u'QYJKoZIhvcNAQEBBQADSwAwSAJBAKx1c7RR7R_drnBSQ_zfx1vQLHUbFLh1' - u'AQQQ5R8DZUXd36efNK79vukFhN9HFoHZiUvOjm0c-pVE6K-EdE_twuUCAwE' - u'AATANBgkqhkiG9w0BAQsFAANBAC24z0IdwIVKSlntksllvr6zJepBH5fMnd' - u'fk3XJp10jT6VE-14KNtjh02a56GoraAvJAT5_H67E8GvJ_ocNnB_o' - ) - self.b64_csr = ( - u'MIIBXTCCAQcCAQAweTELMAkGA1UEBhMCVVMxETAPBgNVBAgMCE1pY2hpZ2F' - u'uMRIwEAYDVQQHDAlBbm4gQXJib3IxDDAKBgNVBAoMA0VGRjEfMB0GA1UECw' - u'wWVW5pdmVyc2l0eSBvZiBNaWNoaWdhbjEUMBIGA1UEAwwLZXhhbXBsZS5jb' - u'20wXDANBgkqhkiG9w0BAQEFAANLADBIAkEArHVztFHtH92ucFJD_N_HW9As' - u'dRsUuHUBBBDlHwNlRd3fp580rv2-6QWE30cWgdmJS86ObRz6lUTor4R0T-3' - u'C5QIDAQABoCkwJwYJKoZIhvcNAQkOMRowGDAWBgNVHREEDzANggtleGFtcG' - u'xlLmNvbTANBgkqhkiG9w0BAQsFAANBAHJH_O6BtC9aGzEVCMGOZ7z9iIRHW' - u'Szr9x_bOzn7hLwsbXPAgO1QxEwL-X-4g20Gn9XBE1N9W6HCIEut2d8wACg' - ) - - def test_encode_b64jose(self): - from acme.jose.json_util import encode_b64jose - encoded = encode_b64jose(b'x') - self.assertTrue(isinstance(encoded, six.string_types)) - self.assertEqual(u'eA', encoded) - - def test_decode_b64jose(self): - from acme.jose.json_util import decode_b64jose - decoded = decode_b64jose(u'eA') - self.assertTrue(isinstance(decoded, six.binary_type)) - self.assertEqual(b'x', decoded) - - def test_decode_b64jose_padding_error(self): - from acme.jose.json_util import decode_b64jose - self.assertRaises(errors.DeserializationError, decode_b64jose, u'x') - - def test_decode_b64jose_size(self): - from acme.jose.json_util import decode_b64jose - self.assertEqual(b'foo', decode_b64jose(u'Zm9v', size=3)) - self.assertRaises( - errors.DeserializationError, decode_b64jose, u'Zm9v', size=2) - self.assertRaises( - errors.DeserializationError, decode_b64jose, u'Zm9v', size=4) - - def test_decode_b64jose_minimum_size(self): - from acme.jose.json_util import decode_b64jose - self.assertEqual(b'foo', decode_b64jose(u'Zm9v', size=3, minimum=True)) - self.assertEqual(b'foo', decode_b64jose(u'Zm9v', size=2, minimum=True)) - self.assertRaises(errors.DeserializationError, decode_b64jose, - u'Zm9v', size=4, minimum=True) - - def test_encode_hex16(self): - from acme.jose.json_util import encode_hex16 - encoded = encode_hex16(b'foo') - self.assertEqual(u'666f6f', encoded) - self.assertTrue(isinstance(encoded, six.string_types)) - - def test_decode_hex16(self): - from acme.jose.json_util import decode_hex16 - decoded = decode_hex16(u'666f6f') - self.assertEqual(b'foo', decoded) - self.assertTrue(isinstance(decoded, six.binary_type)) - - def test_decode_hex16_minimum_size(self): - from acme.jose.json_util import decode_hex16 - self.assertEqual(b'foo', decode_hex16(u'666f6f', size=3, minimum=True)) - self.assertEqual(b'foo', decode_hex16(u'666f6f', size=2, minimum=True)) - self.assertRaises(errors.DeserializationError, decode_hex16, - u'666f6f', size=4, minimum=True) - - def test_decode_hex16_odd_length(self): - from acme.jose.json_util import decode_hex16 - self.assertRaises(errors.DeserializationError, decode_hex16, u'x') - - def test_encode_cert(self): - from acme.jose.json_util import encode_cert - self.assertEqual(self.b64_cert, encode_cert(CERT)) - - def test_decode_cert(self): - from acme.jose.json_util import decode_cert - cert = decode_cert(self.b64_cert) - self.assertTrue(isinstance(cert, util.ComparableX509)) - self.assertEqual(cert, CERT) - self.assertRaises(errors.DeserializationError, decode_cert, u'') - - def test_encode_csr(self): - from acme.jose.json_util import encode_csr - self.assertEqual(self.b64_csr, encode_csr(CSR)) - - def test_decode_csr(self): - from acme.jose.json_util import decode_csr - csr = decode_csr(self.b64_csr) - self.assertTrue(isinstance(csr, util.ComparableX509)) - self.assertEqual(csr, CSR) - self.assertRaises(errors.DeserializationError, decode_csr, u'') - - -class TypedJSONObjectWithFieldsTest(unittest.TestCase): - - def setUp(self): - from acme.jose.json_util import TypedJSONObjectWithFields - - # pylint: disable=missing-docstring,abstract-method - # pylint: disable=too-few-public-methods - - class MockParentTypedJSONObjectWithFields(TypedJSONObjectWithFields): - TYPES = {} - type_field_name = 'type' - - @MockParentTypedJSONObjectWithFields.register - class MockTypedJSONObjectWithFields( - MockParentTypedJSONObjectWithFields): - typ = 'test' - __slots__ = ('foo',) - - @classmethod - def fields_from_json(cls, jobj): - return {'foo': jobj['foo']} - - def fields_to_partial_json(self): - return {'foo': self.foo} - - self.parent_cls = MockParentTypedJSONObjectWithFields - self.msg = MockTypedJSONObjectWithFields(foo='bar') - - def test_to_partial_json(self): - self.assertEqual(self.msg.to_partial_json(), { - 'type': 'test', - 'foo': 'bar', - }) - - def test_from_json_non_dict_fails(self): - for value in [[], (), 5, "asd"]: # all possible input types - self.assertRaises( - errors.DeserializationError, self.parent_cls.from_json, value) - - def test_from_json_dict_no_type_fails(self): - self.assertRaises( - errors.DeserializationError, self.parent_cls.from_json, {}) - - def test_from_json_unknown_type_fails(self): - self.assertRaises(errors.UnrecognizedTypeError, - self.parent_cls.from_json, {'type': 'bar'}) - - def test_from_json_returns_obj(self): - self.assertEqual({'foo': 'bar'}, self.parent_cls.from_json( - {'type': 'test', 'foo': 'bar'})) - - -if __name__ == '__main__': - unittest.main() # pragma: no cover diff --git a/acme/acme/jose/jwa.py b/acme/acme/jose/jwa.py deleted file mode 100644 index 9b682ecab..000000000 --- a/acme/acme/jose/jwa.py +++ /dev/null @@ -1,180 +0,0 @@ -"""JSON Web Algorithm. - -https://tools.ietf.org/html/draft-ietf-jose-json-web-algorithms-40 - -""" -import abc -import collections -import logging - -import cryptography.exceptions -from cryptography.hazmat.backends import default_backend -from cryptography.hazmat.primitives import hashes # type: ignore -from cryptography.hazmat.primitives import hmac # type: ignore -from cryptography.hazmat.primitives.asymmetric import padding # type: ignore - -from acme.jose import errors -from acme.jose import interfaces -from acme.jose import jwk - - -logger = logging.getLogger(__name__) - - -class JWA(interfaces.JSONDeSerializable): # pylint: disable=abstract-method - # pylint: disable=too-few-public-methods - # for some reason disable=abstract-method has to be on the line - # above... - """JSON Web Algorithm.""" - - -class JWASignature(JWA, collections.Hashable): # type: ignore - """JSON Web Signature Algorithm.""" - SIGNATURES = {} # type: dict - - def __init__(self, name): - self.name = name - - def __eq__(self, other): - if not isinstance(other, JWASignature): - return NotImplemented - return self.name == other.name - - def __hash__(self): - return hash((self.__class__, self.name)) - - def __ne__(self, other): - return not self == other - - @classmethod - def register(cls, signature_cls): - """Register class for JSON deserialization.""" - cls.SIGNATURES[signature_cls.name] = signature_cls - return signature_cls - - def to_partial_json(self): - return self.name - - @classmethod - def from_json(cls, jobj): - return cls.SIGNATURES[jobj] - - @abc.abstractmethod - def sign(self, key, msg): # pragma: no cover - """Sign the ``msg`` using ``key``.""" - raise NotImplementedError() - - @abc.abstractmethod - def verify(self, key, msg, sig): # pragma: no cover - """Verify the ``msg` and ``sig`` using ``key``.""" - raise NotImplementedError() - - def __repr__(self): - return self.name - - -class _JWAHS(JWASignature): - - kty = jwk.JWKOct - - def __init__(self, name, hash_): - super(_JWAHS, self).__init__(name) - self.hash = hash_() - - def sign(self, key, msg): - signer = hmac.HMAC(key, self.hash, backend=default_backend()) - signer.update(msg) - return signer.finalize() - - def verify(self, key, msg, sig): - verifier = hmac.HMAC(key, self.hash, backend=default_backend()) - verifier.update(msg) - try: - verifier.verify(sig) - except cryptography.exceptions.InvalidSignature as error: - logger.debug(error, exc_info=True) - return False - else: - return True - - -class _JWARSA(object): - - kty = jwk.JWKRSA - padding = NotImplemented - hash = NotImplemented - - def sign(self, key, msg): - """Sign the ``msg`` using ``key``.""" - try: - signer = key.signer(self.padding, self.hash) - except AttributeError as error: - logger.debug(error, exc_info=True) - raise errors.Error("Public key cannot be used for signing") - except ValueError as error: # digest too large - logger.debug(error, exc_info=True) - raise errors.Error(str(error)) - signer.update(msg) - try: - return signer.finalize() - except ValueError as error: - logger.debug(error, exc_info=True) - raise errors.Error(str(error)) - - def verify(self, key, msg, sig): - """Verify the ``msg` and ``sig`` using ``key``.""" - verifier = key.verifier(sig, self.padding, self.hash) - verifier.update(msg) - try: - verifier.verify() - except cryptography.exceptions.InvalidSignature as error: - logger.debug(error, exc_info=True) - return False - else: - return True - - -class _JWARS(_JWARSA, JWASignature): - - def __init__(self, name, hash_): - super(_JWARS, self).__init__(name) - self.padding = padding.PKCS1v15() - self.hash = hash_() - - -class _JWAPS(_JWARSA, JWASignature): - - def __init__(self, name, hash_): - super(_JWAPS, self).__init__(name) - self.padding = padding.PSS( - mgf=padding.MGF1(hash_()), - salt_length=padding.PSS.MAX_LENGTH) - self.hash = hash_() - - -class _JWAES(JWASignature): # pylint: disable=abstract-class-not-used - - # TODO: implement ES signatures - - def sign(self, key, msg): # pragma: no cover - raise NotImplementedError() - - def verify(self, key, msg, sig): # pragma: no cover - raise NotImplementedError() - - -HS256 = JWASignature.register(_JWAHS('HS256', hashes.SHA256)) -HS384 = JWASignature.register(_JWAHS('HS384', hashes.SHA384)) -HS512 = JWASignature.register(_JWAHS('HS512', hashes.SHA512)) - -RS256 = JWASignature.register(_JWARS('RS256', hashes.SHA256)) -RS384 = JWASignature.register(_JWARS('RS384', hashes.SHA384)) -RS512 = JWASignature.register(_JWARS('RS512', hashes.SHA512)) - -PS256 = JWASignature.register(_JWAPS('PS256', hashes.SHA256)) -PS384 = JWASignature.register(_JWAPS('PS384', hashes.SHA384)) -PS512 = JWASignature.register(_JWAPS('PS512', hashes.SHA512)) - -ES256 = JWASignature.register(_JWAES('ES256')) -ES384 = JWASignature.register(_JWAES('ES384')) -ES512 = JWASignature.register(_JWAES('ES512')) diff --git a/acme/acme/jose/jwa_test.py b/acme/acme/jose/jwa_test.py deleted file mode 100644 index 3328d083a..000000000 --- a/acme/acme/jose/jwa_test.py +++ /dev/null @@ -1,104 +0,0 @@ -"""Tests for acme.jose.jwa.""" -import unittest - -from acme import test_util - -from acme.jose import errors - - -RSA256_KEY = test_util.load_rsa_private_key('rsa256_key.pem') -RSA512_KEY = test_util.load_rsa_private_key('rsa512_key.pem') -RSA1024_KEY = test_util.load_rsa_private_key('rsa1024_key.pem') - - -class JWASignatureTest(unittest.TestCase): - """Tests for acme.jose.jwa.JWASignature.""" - - def setUp(self): - from acme.jose.jwa import JWASignature - - class MockSig(JWASignature): - # pylint: disable=missing-docstring,too-few-public-methods - # pylint: disable=abstract-class-not-used - def sign(self, key, msg): - raise NotImplementedError() # pragma: no cover - - def verify(self, key, msg, sig): - raise NotImplementedError() # pragma: no cover - - # pylint: disable=invalid-name - self.Sig1 = MockSig('Sig1') - self.Sig2 = MockSig('Sig2') - - def test_eq(self): - self.assertEqual(self.Sig1, self.Sig1) - - def test_ne(self): - self.assertNotEqual(self.Sig1, self.Sig2) - - def test_ne_other_type(self): - self.assertNotEqual(self.Sig1, 5) - - def test_repr(self): - self.assertEqual('Sig1', repr(self.Sig1)) - self.assertEqual('Sig2', repr(self.Sig2)) - - def test_to_partial_json(self): - self.assertEqual(self.Sig1.to_partial_json(), 'Sig1') - self.assertEqual(self.Sig2.to_partial_json(), 'Sig2') - - def test_from_json(self): - from acme.jose.jwa import JWASignature - from acme.jose.jwa import RS256 - self.assertTrue(JWASignature.from_json('RS256') is RS256) - - -class JWAHSTest(unittest.TestCase): # pylint: disable=too-few-public-methods - - def test_it(self): - from acme.jose.jwa import HS256 - sig = ( - b"\xceR\xea\xcd\x94\xab\xcf\xfb\xe0\xacA.:\x1a'\x08i\xe2\xc4" - b"\r\x85+\x0e\x85\xaeUZ\xd4\xb3\x97zO" - ) - self.assertEqual(HS256.sign(b'some key', b'foo'), sig) - self.assertTrue(HS256.verify(b'some key', b'foo', sig) is True) - self.assertTrue(HS256.verify(b'some key', b'foo', sig + b'!') is False) - - -class JWARSTest(unittest.TestCase): - - def test_sign_no_private_part(self): - from acme.jose.jwa import RS256 - self.assertRaises( - errors.Error, RS256.sign, RSA512_KEY.public_key(), b'foo') - - def test_sign_key_too_small(self): - from acme.jose.jwa import RS256 - from acme.jose.jwa import PS256 - self.assertRaises(errors.Error, RS256.sign, RSA256_KEY, b'foo') - self.assertRaises(errors.Error, PS256.sign, RSA256_KEY, b'foo') - - def test_rs(self): - from acme.jose.jwa import RS256 - sig = ( - b'|\xc6\xb2\xa4\xab(\x87\x99\xfa*:\xea\xf8\xa0N&}\x9f\x0f\xc0O' - b'\xc6t\xa3\xe6\xfa\xbb"\x15Y\x80Y\xe0\x81\xb8\x88)\xba\x0c\x9c' - b'\xa4\x99\x1e\x19&\xd8\xc7\x99S\x97\xfc\x85\x0cOV\xe6\x07\x99' - b'\xd2\xb9.>}\xfd' - ) - self.assertEqual(RS256.sign(RSA512_KEY, b'foo'), sig) - self.assertTrue(RS256.verify(RSA512_KEY.public_key(), b'foo', sig)) - self.assertFalse(RS256.verify( - RSA512_KEY.public_key(), b'foo', sig + b'!')) - - def test_ps(self): - from acme.jose.jwa import PS256 - sig = PS256.sign(RSA1024_KEY, b'foo') - self.assertTrue(PS256.verify(RSA1024_KEY.public_key(), b'foo', sig)) - self.assertFalse(PS256.verify( - RSA1024_KEY.public_key(), b'foo', sig + b'!')) - - -if __name__ == '__main__': - unittest.main() # pragma: no cover diff --git a/acme/acme/jose/jwk.py b/acme/acme/jose/jwk.py deleted file mode 100644 index 54423f670..000000000 --- a/acme/acme/jose/jwk.py +++ /dev/null @@ -1,281 +0,0 @@ -"""JSON Web Key.""" -import abc -import binascii -import json -import logging - -import cryptography.exceptions -from cryptography.hazmat.backends import default_backend -from cryptography.hazmat.primitives import hashes # type: ignore -from cryptography.hazmat.primitives import serialization -from cryptography.hazmat.primitives.asymmetric import ec # type: ignore -from cryptography.hazmat.primitives.asymmetric import rsa - -import six - -from acme.jose import errors -from acme.jose import json_util -from acme.jose import util - - -logger = logging.getLogger(__name__) - - -class JWK(json_util.TypedJSONObjectWithFields): - # pylint: disable=too-few-public-methods - """JSON Web Key.""" - type_field_name = 'kty' - TYPES = {} # type: dict - cryptography_key_types = () # type: tuple - """Subclasses should override.""" - - required = NotImplemented - """Required members of public key's representation as defined by JWK/JWA.""" - - _thumbprint_json_dumps_params = { - # "no whitespace or line breaks before or after any syntactic - # elements" - 'indent': None, - 'separators': (',', ':'), - # "members ordered lexicographically by the Unicode [UNICODE] - # code points of the member names" - 'sort_keys': True, - } - - def thumbprint(self, hash_function=hashes.SHA256): - """Compute JWK Thumbprint. - - https://tools.ietf.org/html/rfc7638 - - :returns bytes: - - """ - digest = hashes.Hash(hash_function(), backend=default_backend()) - digest.update(json.dumps( - dict((k, v) for k, v in six.iteritems(self.to_json()) - if k in self.required), - **self._thumbprint_json_dumps_params).encode()) - return digest.finalize() - - @abc.abstractmethod - def public_key(self): # pragma: no cover - """Generate JWK with public key. - - For symmetric cryptosystems, this would return ``self``. - - """ - raise NotImplementedError() - - @classmethod - def _load_cryptography_key(cls, data, password=None, backend=None): - backend = default_backend() if backend is None else backend - exceptions = {} - - # private key? - for loader in (serialization.load_pem_private_key, - serialization.load_der_private_key): - try: - return loader(data, password, backend) - except (ValueError, TypeError, - cryptography.exceptions.UnsupportedAlgorithm) as error: - exceptions[loader] = error - - # public key? - for loader in (serialization.load_pem_public_key, - serialization.load_der_public_key): - try: - return loader(data, backend) - except (ValueError, - cryptography.exceptions.UnsupportedAlgorithm) as error: - exceptions[loader] = error - - # no luck - raise errors.Error('Unable to deserialize key: {0}'.format(exceptions)) - - @classmethod - def load(cls, data, password=None, backend=None): - """Load serialized key as JWK. - - :param str data: Public or private key serialized as PEM or DER. - :param str password: Optional password. - :param backend: A `.PEMSerializationBackend` and - `.DERSerializationBackend` provider. - - :raises errors.Error: if unable to deserialize, or unsupported - JWK algorithm - - :returns: JWK of an appropriate type. - :rtype: `JWK` - - """ - try: - key = cls._load_cryptography_key(data, password, backend) - except errors.Error as error: - logger.debug('Loading symmetric key, asymmetric failed: %s', error) - return JWKOct(key=data) - - if cls.typ is not NotImplemented and not isinstance( - key, cls.cryptography_key_types): - raise errors.Error('Unable to deserialize {0} into {1}'.format( - key.__class__, cls.__class__)) - for jwk_cls in six.itervalues(cls.TYPES): - if isinstance(key, jwk_cls.cryptography_key_types): - return jwk_cls(key=key) - raise errors.Error('Unsupported algorithm: {0}'.format(key.__class__)) - - -@JWK.register -class JWKES(JWK): # pragma: no cover - # pylint: disable=abstract-class-not-used - """ES JWK. - - .. warning:: This is not yet implemented! - - """ - typ = 'ES' - cryptography_key_types = ( - ec.EllipticCurvePublicKey, ec.EllipticCurvePrivateKey) - required = ('crv', JWK.type_field_name, 'x', 'y') - - def fields_to_partial_json(self): - raise NotImplementedError() - - @classmethod - def fields_from_json(cls, jobj): - raise NotImplementedError() - - def public_key(self): - raise NotImplementedError() - - -@JWK.register -class JWKOct(JWK): - """Symmetric JWK.""" - typ = 'oct' - __slots__ = ('key',) - required = ('k', JWK.type_field_name) - - def fields_to_partial_json(self): - # TODO: An "alg" member SHOULD also be present to identify the - # algorithm intended to be used with the key, unless the - # application uses another means or convention to determine - # the algorithm used. - return {'k': json_util.encode_b64jose(self.key)} - - @classmethod - def fields_from_json(cls, jobj): - return cls(key=json_util.decode_b64jose(jobj['k'])) - - def public_key(self): - return self - - -@JWK.register -class JWKRSA(JWK): - """RSA JWK. - - :ivar key: `cryptography.hazmat.primitives.rsa.RSAPrivateKey` - or `cryptography.hazmat.primitives.rsa.RSAPublicKey` wrapped - in `.ComparableRSAKey` - - """ - typ = 'RSA' - cryptography_key_types = (rsa.RSAPublicKey, rsa.RSAPrivateKey) - __slots__ = ('key',) - required = ('e', JWK.type_field_name, 'n') - - def __init__(self, *args, **kwargs): - if 'key' in kwargs and not isinstance( - kwargs['key'], util.ComparableRSAKey): - kwargs['key'] = util.ComparableRSAKey(kwargs['key']) - super(JWKRSA, self).__init__(*args, **kwargs) - - @classmethod - def _encode_param(cls, data): - """Encode Base64urlUInt. - - :type data: long - :rtype: unicode - - """ - def _leading_zeros(arg): - if len(arg) % 2: - return '0' + arg - return arg - - return json_util.encode_b64jose(binascii.unhexlify( - _leading_zeros(hex(data)[2:].rstrip('L')))) - - @classmethod - def _decode_param(cls, data): - """Decode Base64urlUInt.""" - try: - return int(binascii.hexlify(json_util.decode_b64jose(data)), 16) - except ValueError: # invalid literal for long() with base 16 - raise errors.DeserializationError() - - def public_key(self): - return type(self)(key=self.key.public_key()) - - @classmethod - def fields_from_json(cls, jobj): - # pylint: disable=invalid-name - n, e = (cls._decode_param(jobj[x]) for x in ('n', 'e')) - public_numbers = rsa.RSAPublicNumbers(e=e, n=n) - if 'd' not in jobj: # public key - key = public_numbers.public_key(default_backend()) - else: # private key - d = cls._decode_param(jobj['d']) - if ('p' in jobj or 'q' in jobj or 'dp' in jobj or - 'dq' in jobj or 'qi' in jobj or 'oth' in jobj): - # "If the producer includes any of the other private - # key parameters, then all of the others MUST be - # present, with the exception of "oth", which MUST - # only be present when more than two prime factors - # were used." - p, q, dp, dq, qi, = all_params = tuple( - jobj.get(x) for x in ('p', 'q', 'dp', 'dq', 'qi')) - if tuple(param for param in all_params if param is None): - raise errors.Error( - 'Some private parameters are missing: {0}'.format( - all_params)) - p, q, dp, dq, qi = tuple( - cls._decode_param(x) for x in all_params) - - # TODO: check for oth - else: - # cryptography>=0.8 - p, q = rsa.rsa_recover_prime_factors(n, e, d) - dp = rsa.rsa_crt_dmp1(d, p) - dq = rsa.rsa_crt_dmq1(d, q) - qi = rsa.rsa_crt_iqmp(p, q) - - key = rsa.RSAPrivateNumbers( - p, q, d, dp, dq, qi, public_numbers).private_key( - default_backend()) - - return cls(key=key) - - def fields_to_partial_json(self): - # pylint: disable=protected-access - if isinstance(self.key._wrapped, rsa.RSAPublicKey): - numbers = self.key.public_numbers() - params = { - 'n': numbers.n, - 'e': numbers.e, - } - else: # rsa.RSAPrivateKey - private = self.key.private_numbers() - public = self.key.public_key().public_numbers() - params = { - 'n': public.n, - 'e': public.e, - 'd': private.d, - 'p': private.p, - 'q': private.q, - 'dp': private.dmp1, - 'dq': private.dmq1, - 'qi': private.iqmp, - } - return dict((key, self._encode_param(value)) - for key, value in six.iteritems(params)) diff --git a/acme/acme/jose/jwk_test.py b/acme/acme/jose/jwk_test.py deleted file mode 100644 index eea5793bf..000000000 --- a/acme/acme/jose/jwk_test.py +++ /dev/null @@ -1,191 +0,0 @@ -"""Tests for acme.jose.jwk.""" -import binascii -import unittest - -from acme import test_util - -from acme.jose import errors -from acme.jose import json_util -from acme.jose import util - - -DSA_PEM = test_util.load_vector('dsa512_key.pem') -RSA256_KEY = test_util.load_rsa_private_key('rsa256_key.pem') -RSA512_KEY = test_util.load_rsa_private_key('rsa512_key.pem') - - -class JWKTest(unittest.TestCase): - """Tests for acme.jose.jwk.JWK.""" - - def test_load(self): - from acme.jose.jwk import JWK - self.assertRaises(errors.Error, JWK.load, DSA_PEM) - - def test_load_subclass_wrong_type(self): - from acme.jose.jwk import JWKRSA - self.assertRaises(errors.Error, JWKRSA.load, DSA_PEM) - - -class JWKTestBaseMixin(object): - """Mixin test for JWK subclass tests.""" - - thumbprint = NotImplemented - - def test_thumbprint_private(self): - self.assertEqual(self.thumbprint, self.jwk.thumbprint()) - - def test_thumbprint_public(self): - self.assertEqual(self.thumbprint, self.jwk.public_key().thumbprint()) - - -class JWKOctTest(unittest.TestCase, JWKTestBaseMixin): - """Tests for acme.jose.jwk.JWKOct.""" - - thumbprint = (b"\xf3\xe7\xbe\xa8`\xd2\xdap\xe9}\x9c\xce>" - b"\xd0\xfcI\xbe\xcd\x92'\xd4o\x0e\xf41\xea" - b"\x8e(\x8a\xb2i\x1c") - - def setUp(self): - from acme.jose.jwk import JWKOct - self.jwk = JWKOct(key=b'foo') - self.jobj = {'kty': 'oct', 'k': json_util.encode_b64jose(b'foo')} - - def test_to_partial_json(self): - self.assertEqual(self.jwk.to_partial_json(), self.jobj) - - def test_from_json(self): - from acme.jose.jwk import JWKOct - self.assertEqual(self.jwk, JWKOct.from_json(self.jobj)) - - def test_from_json_hashable(self): - from acme.jose.jwk import JWKOct - hash(JWKOct.from_json(self.jobj)) - - def test_load(self): - from acme.jose.jwk import JWKOct - self.assertEqual(self.jwk, JWKOct.load(b'foo')) - - def test_public_key(self): - self.assertTrue(self.jwk.public_key() is self.jwk) - - -class JWKRSATest(unittest.TestCase, JWKTestBaseMixin): - """Tests for acme.jose.jwk.JWKRSA.""" - # pylint: disable=too-many-instance-attributes - - thumbprint = (b'\x83K\xdc#3\x98\xca\x98\xed\xcb\x80\x80<\x0c' - b'\xf0\x95\xb9H\xb2*l\xbd$\xe5&|O\x91\xd4 \xb0Y') - - def setUp(self): - from acme.jose.jwk import JWKRSA - self.jwk256 = JWKRSA(key=RSA256_KEY.public_key()) - self.jwk256json = { - 'kty': 'RSA', - 'e': 'AQAB', - 'n': 'm2Fylv-Uz7trgTW8EBHP3FQSMeZs2GNQ6VRo1sIVJEk', - } - # pylint: disable=protected-access - self.jwk256_not_comparable = JWKRSA( - key=RSA256_KEY.public_key()._wrapped) - self.jwk512 = JWKRSA(key=RSA512_KEY.public_key()) - self.jwk512json = { - 'kty': 'RSA', - 'e': 'AQAB', - 'n': 'rHVztFHtH92ucFJD_N_HW9AsdRsUuHUBBBDlHwNlRd3fp5' - '80rv2-6QWE30cWgdmJS86ObRz6lUTor4R0T-3C5Q', - } - self.private = JWKRSA(key=RSA256_KEY) - self.private_json_small = self.jwk256json.copy() - self.private_json_small['d'] = ( - 'lPQED_EPTV0UIBfNI3KP2d9Jlrc2mrMllmf946bu-CE') - self.private_json = self.jwk256json.copy() - self.private_json.update({ - 'd': 'lPQED_EPTV0UIBfNI3KP2d9Jlrc2mrMllmf946bu-CE', - 'p': 'zUVNZn4lLLBD1R6NE8TKNQ', - 'q': 'wcfKfc7kl5jfqXArCRSURQ', - 'dp': 'CWJFq43QvT5Bm5iN8n1okQ', - 'dq': 'bHh2u7etM8LKKCF2pY2UdQ', - 'qi': 'oi45cEkbVoJjAbnQpFY87Q', - }) - self.jwk = self.private - - def test_init_auto_comparable(self): - self.assertTrue(isinstance( - self.jwk256_not_comparable.key, util.ComparableRSAKey)) - self.assertEqual(self.jwk256, self.jwk256_not_comparable) - - def test_encode_param_zero(self): - from acme.jose.jwk import JWKRSA - # pylint: disable=protected-access - # TODO: move encode/decode _param to separate class - self.assertEqual('AA', JWKRSA._encode_param(0)) - - def test_equals(self): - self.assertEqual(self.jwk256, self.jwk256) - self.assertEqual(self.jwk512, self.jwk512) - - def test_not_equals(self): - self.assertNotEqual(self.jwk256, self.jwk512) - self.assertNotEqual(self.jwk512, self.jwk256) - - def test_load(self): - from acme.jose.jwk import JWKRSA - self.assertEqual(self.private, JWKRSA.load( - test_util.load_vector('rsa256_key.pem'))) - - def test_public_key(self): - self.assertEqual(self.jwk256, self.private.public_key()) - - def test_to_partial_json(self): - self.assertEqual(self.jwk256.to_partial_json(), self.jwk256json) - self.assertEqual(self.jwk512.to_partial_json(), self.jwk512json) - self.assertEqual(self.private.to_partial_json(), self.private_json) - - def test_from_json(self): - from acme.jose.jwk import JWK - self.assertEqual( - self.jwk256, JWK.from_json(self.jwk256json)) - self.assertEqual( - self.jwk512, JWK.from_json(self.jwk512json)) - self.assertEqual(self.private, JWK.from_json(self.private_json)) - - def test_from_json_private_small(self): - from acme.jose.jwk import JWK - self.assertEqual(self.private, JWK.from_json(self.private_json_small)) - - def test_from_json_missing_one_additional(self): - from acme.jose.jwk import JWK - del self.private_json['q'] - self.assertRaises(errors.Error, JWK.from_json, self.private_json) - - def test_from_json_hashable(self): - from acme.jose.jwk import JWK - hash(JWK.from_json(self.jwk256json)) - - def test_from_json_non_schema_errors(self): - # valid against schema, but still failing - from acme.jose.jwk import JWK - self.assertRaises(errors.DeserializationError, JWK.from_json, - {'kty': 'RSA', 'e': 'AQAB', 'n': ''}) - self.assertRaises(errors.DeserializationError, JWK.from_json, - {'kty': 'RSA', 'e': 'AQAB', 'n': '1'}) - - def test_thumbprint_go_jose(self): - # https://github.com/square/go-jose/blob/4ddd71883fa547d37fbf598071f04512d8bafee3/jwk.go#L155 - # https://github.com/square/go-jose/blob/4ddd71883fa547d37fbf598071f04512d8bafee3/jwk_test.go#L331-L344 - # https://github.com/square/go-jose/blob/4ddd71883fa547d37fbf598071f04512d8bafee3/jwk_test.go#L384 - from acme.jose.jwk import JWKRSA - key = JWKRSA.json_loads("""{ - "kty": "RSA", - "kid": "bilbo.baggins@hobbiton.example", - "use": "sig", - "n": "n4EPtAOCc9AlkeQHPzHStgAbgs7bTZLwUBZdR8_KuKPEHLd4rHVTeT-O-XV2jRojdNhxJWTDvNd7nqQ0VEiZQHz_AJmSCpMaJMRBSFKrKb2wqVwGU_NsYOYL-QtiWN2lbzcEe6XC0dApr5ydQLrHqkHHig3RBordaZ6Aj-oBHqFEHYpPe7Tpe-OfVfHd1E6cS6M1FZcD1NNLYD5lFHpPI9bTwJlsde3uhGqC0ZCuEHg8lhzwOHrtIQbS0FVbb9k3-tVTU4fg_3L_vniUFAKwuCLqKnS2BYwdq_mzSnbLY7h_qixoR7jig3__kRhuaxwUkRz5iaiQkqgc5gHdrNP5zw", - "e": "AQAB" -}""") - self.assertEqual( - binascii.hexlify(key.thumbprint()), - b"f63838e96077ad1fc01c3f8405774dedc0641f558ebb4b40dccf5f9b6d66a932") - - -if __name__ == '__main__': - unittest.main() # pragma: no cover diff --git a/acme/acme/jose/jws.py b/acme/acme/jose/jws.py deleted file mode 100644 index 5f446e4b1..000000000 --- a/acme/acme/jose/jws.py +++ /dev/null @@ -1,433 +0,0 @@ -"""JOSE Web Signature.""" -import argparse -import base64 -import sys - -import OpenSSL -import six - -from acme.jose import b64 -from acme.jose import errors -from acme.jose import json_util -from acme.jose import jwa -from acme.jose import jwk -from acme.jose import util - - -class MediaType(object): - """MediaType field encoder/decoder.""" - - PREFIX = 'application/' - """MIME Media Type and Content Type prefix.""" - - @classmethod - def decode(cls, value): - """Decoder.""" - # 4.1.10 - if '/' not in value: - if ';' in value: - raise errors.DeserializationError('Unexpected semi-colon') - return cls.PREFIX + value - return value - - @classmethod - def encode(cls, value): - """Encoder.""" - # 4.1.10 - if ';' not in value: - assert value.startswith(cls.PREFIX) - return value[len(cls.PREFIX):] - return value - - -class Header(json_util.JSONObjectWithFields): - """JOSE Header. - - .. warning:: This class supports **only** Registered Header - Parameter Names (as defined in section 4.1 of the - protocol). If you need Public Header Parameter Names (4.2) - or Private Header Parameter Names (4.3), you must subclass - and override :meth:`from_json` and :meth:`to_partial_json` - appropriately. - - .. warning:: This class does not support any extensions through - the "crit" (Critical) Header Parameter (4.1.11) and as a - conforming implementation, :meth:`from_json` treats its - occurrence as an error. Please subclass if you seek for - a different behaviour. - - :ivar x5tS256: "x5t#S256" - :ivar str typ: MIME Media Type, inc. :const:`MediaType.PREFIX`. - :ivar str cty: Content-Type, inc. :const:`MediaType.PREFIX`. - - """ - alg = json_util.Field( - 'alg', decoder=jwa.JWASignature.from_json, omitempty=True) - jku = json_util.Field('jku', omitempty=True) - jwk = json_util.Field('jwk', decoder=jwk.JWK.from_json, omitempty=True) - kid = json_util.Field('kid', omitempty=True) - x5u = json_util.Field('x5u', omitempty=True) - x5c = json_util.Field('x5c', omitempty=True, default=()) - x5t = json_util.Field( - 'x5t', decoder=json_util.decode_b64jose, omitempty=True) - x5tS256 = json_util.Field( - 'x5t#S256', decoder=json_util.decode_b64jose, omitempty=True) - typ = json_util.Field('typ', encoder=MediaType.encode, - decoder=MediaType.decode, omitempty=True) - cty = json_util.Field('cty', encoder=MediaType.encode, - decoder=MediaType.decode, omitempty=True) - crit = json_util.Field('crit', omitempty=True, default=()) - - def not_omitted(self): - """Fields that would not be omitted in the JSON object.""" - return dict((name, getattr(self, name)) - for name, field in six.iteritems(self._fields) - if not field.omit(getattr(self, name))) - - def __add__(self, other): - if not isinstance(other, type(self)): - raise TypeError('Header cannot be added to: {0}'.format( - type(other))) - - not_omitted_self = self.not_omitted() - not_omitted_other = other.not_omitted() - - if set(not_omitted_self).intersection(not_omitted_other): - raise TypeError('Addition of overlapping headers not defined') - - not_omitted_self.update(not_omitted_other) - return type(self)(**not_omitted_self) # pylint: disable=star-args - - def find_key(self): - """Find key based on header. - - .. todo:: Supports only "jwk" header parameter lookup. - - :returns: (Public) key found in the header. - :rtype: .JWK - - :raises acme.jose.errors.Error: if key could not be found - - """ - if self.jwk is None: - raise errors.Error('No key found') - return self.jwk - - @crit.decoder - def crit(unused_value): - # pylint: disable=missing-docstring,no-self-argument,no-self-use - raise errors.DeserializationError( - '"crit" is not supported, please subclass') - - # x5c does NOT use JOSE Base64 (4.1.6) - - @x5c.encoder # type: ignore - def x5c(value): # pylint: disable=missing-docstring,no-self-argument - return [base64.b64encode(OpenSSL.crypto.dump_certificate( - OpenSSL.crypto.FILETYPE_ASN1, cert.wrapped)) for cert in value] - - @x5c.decoder # type: ignore - def x5c(value): # pylint: disable=missing-docstring,no-self-argument - try: - return tuple(util.ComparableX509(OpenSSL.crypto.load_certificate( - OpenSSL.crypto.FILETYPE_ASN1, - base64.b64decode(cert))) for cert in value) - except OpenSSL.crypto.Error as error: - raise errors.DeserializationError(error) - - -class Signature(json_util.JSONObjectWithFields): - """JWS Signature. - - :ivar combined: Combined Header (protected and unprotected, - :class:`Header`). - :ivar unicode protected: JWS protected header (Jose Base-64 decoded). - :ivar header: JWS Unprotected Header (:class:`Header`). - :ivar str signature: The signature. - - """ - header_cls = Header - - __slots__ = ('combined',) - protected = json_util.Field('protected', omitempty=True, default='') - header = json_util.Field( - 'header', omitempty=True, default=header_cls(), - decoder=header_cls.from_json) - signature = json_util.Field( - 'signature', decoder=json_util.decode_b64jose, - encoder=json_util.encode_b64jose) - - @protected.encoder # type: ignore - def protected(value): # pylint: disable=missing-docstring,no-self-argument - # wrong type guess (Signature, not bytes) | pylint: disable=no-member - return json_util.encode_b64jose(value.encode('utf-8')) - - @protected.decoder # type: ignore - def protected(value): # pylint: disable=missing-docstring,no-self-argument - return json_util.decode_b64jose(value).decode('utf-8') - - def __init__(self, **kwargs): - if 'combined' not in kwargs: - kwargs = self._with_combined(kwargs) - super(Signature, self).__init__(**kwargs) - assert self.combined.alg is not None - - @classmethod - def _with_combined(cls, kwargs): - assert 'combined' not in kwargs - header = kwargs.get('header', cls._fields['header'].default) - protected = kwargs.get('protected', cls._fields['protected'].default) - - if protected: - combined = header + cls.header_cls.json_loads(protected) - else: - combined = header - - kwargs['combined'] = combined - return kwargs - - @classmethod - def _msg(cls, protected, payload): - return (b64.b64encode(protected.encode('utf-8')) + b'.' + - b64.b64encode(payload)) - - def verify(self, payload, key=None): - """Verify. - - :param JWK key: Key used for verification. - - """ - key = self.combined.find_key() if key is None else key - return self.combined.alg.verify( - key=key.key, sig=self.signature, - msg=self._msg(self.protected, payload)) - - @classmethod - def sign(cls, payload, key, alg, include_jwk=True, - protect=frozenset(), **kwargs): - """Sign. - - :param JWK key: Key for signature. - - """ - assert isinstance(key, alg.kty) - - header_params = kwargs - header_params['alg'] = alg - if include_jwk: - header_params['jwk'] = key.public_key() - - assert set(header_params).issubset(cls.header_cls._fields) - assert protect.issubset(cls.header_cls._fields) - - protected_params = {} - for header in protect: - if header in header_params: - protected_params[header] = header_params.pop(header) - if protected_params: - # pylint: disable=star-args - protected = cls.header_cls(**protected_params).json_dumps() - else: - protected = '' - - header = cls.header_cls(**header_params) # pylint: disable=star-args - signature = alg.sign(key.key, cls._msg(protected, payload)) - - return cls(protected=protected, header=header, signature=signature) - - def fields_to_partial_json(self): - fields = super(Signature, self).fields_to_partial_json() - if not fields['header'].not_omitted(): - del fields['header'] - return fields - - @classmethod - def fields_from_json(cls, jobj): - fields = super(Signature, cls).fields_from_json(jobj) - fields_with_combined = cls._with_combined(fields) - if 'alg' not in fields_with_combined['combined'].not_omitted(): - raise errors.DeserializationError('alg not present') - return fields_with_combined - - -class JWS(json_util.JSONObjectWithFields): - """JSON Web Signature. - - :ivar str payload: JWS Payload. - :ivar str signature: JWS Signatures. - - """ - __slots__ = ('payload', 'signatures') - - signature_cls = Signature - - def verify(self, key=None): - """Verify.""" - return all(sig.verify(self.payload, key) for sig in self.signatures) - - @classmethod - def sign(cls, payload, **kwargs): - """Sign.""" - return cls(payload=payload, signatures=( - cls.signature_cls.sign(payload=payload, **kwargs),)) - - @property - def signature(self): - """Get a singleton signature. - - :rtype: `signature_cls` - - """ - assert len(self.signatures) == 1 - return self.signatures[0] - - def to_compact(self): - """Compact serialization. - - :rtype: bytes - - """ - assert len(self.signatures) == 1 - - assert 'alg' not in self.signature.header.not_omitted() - # ... it must be in protected - - return ( - b64.b64encode(self.signature.protected.encode('utf-8')) + - b'.' + - b64.b64encode(self.payload) + - b'.' + - b64.b64encode(self.signature.signature)) - - @classmethod - def from_compact(cls, compact): - """Compact deserialization. - - :param bytes compact: - - """ - try: - protected, payload, signature = compact.split(b'.') - except ValueError: - raise errors.DeserializationError( - 'Compact JWS serialization should comprise of exactly' - ' 3 dot-separated components') - - sig = cls.signature_cls( - protected=b64.b64decode(protected).decode('utf-8'), - signature=b64.b64decode(signature)) - return cls(payload=b64.b64decode(payload), signatures=(sig,)) - - def to_partial_json(self, flat=True): # pylint: disable=arguments-differ - assert self.signatures - payload = json_util.encode_b64jose(self.payload) - - if flat and len(self.signatures) == 1: - ret = self.signatures[0].to_partial_json() - ret['payload'] = payload - return ret - else: - return { - 'payload': payload, - 'signatures': self.signatures, - } - - @classmethod - def from_json(cls, jobj): - if 'signature' in jobj and 'signatures' in jobj: - raise errors.DeserializationError('Flat mixed with non-flat') - elif 'signature' in jobj: # flat - return cls(payload=json_util.decode_b64jose(jobj.pop('payload')), - signatures=(cls.signature_cls.from_json(jobj),)) - else: - return cls(payload=json_util.decode_b64jose(jobj['payload']), - signatures=tuple(cls.signature_cls.from_json(sig) - for sig in jobj['signatures'])) - - -class CLI(object): - """JWS CLI.""" - - @classmethod - def sign(cls, args): - """Sign.""" - key = args.alg.kty.load(args.key.read()) - args.key.close() - if args.protect is None: - args.protect = [] - if args.compact: - args.protect.append('alg') - - sig = JWS.sign(payload=sys.stdin.read().encode(), key=key, alg=args.alg, - protect=set(args.protect)) - - if args.compact: - six.print_(sig.to_compact().decode('utf-8')) - else: # JSON - six.print_(sig.json_dumps_pretty()) - - @classmethod - def verify(cls, args): - """Verify.""" - if args.compact: - sig = JWS.from_compact(sys.stdin.read().encode()) - else: # JSON - try: - sig = JWS.json_loads(sys.stdin.read()) - except errors.Error as error: - six.print_(error) - return -1 - - if args.key is not None: - assert args.kty is not None - key = args.kty.load(args.key.read()).public_key() - args.key.close() - else: - key = None - - sys.stdout.write(sig.payload) - return not sig.verify(key=key) - - @classmethod - def _alg_type(cls, arg): - return jwa.JWASignature.from_json(arg) - - @classmethod - def _header_type(cls, arg): - assert arg in Signature.header_cls._fields - return arg - - @classmethod - def _kty_type(cls, arg): - assert arg in jwk.JWK.TYPES - return jwk.JWK.TYPES[arg] - - @classmethod - def run(cls, args=sys.argv[1:]): - """Parse arguments and sign/verify.""" - parser = argparse.ArgumentParser() - parser.add_argument('--compact', action='store_true') - - subparsers = parser.add_subparsers() - parser_sign = subparsers.add_parser('sign') - parser_sign.set_defaults(func=cls.sign) - parser_sign.add_argument( - '-k', '--key', type=argparse.FileType('rb'), required=True) - parser_sign.add_argument( - '-a', '--alg', type=cls._alg_type, default=jwa.RS256) - parser_sign.add_argument( - '-p', '--protect', action='append', type=cls._header_type) - - parser_verify = subparsers.add_parser('verify') - parser_verify.set_defaults(func=cls.verify) - parser_verify.add_argument( - '-k', '--key', type=argparse.FileType('rb'), required=False) - parser_verify.add_argument( - '--kty', type=cls._kty_type, required=False) - - parsed = parser.parse_args(args) - return parsed.func(parsed) - - -if __name__ == '__main__': - exit(CLI.run()) # pragma: no cover diff --git a/acme/acme/jose/jws_test.py b/acme/acme/jose/jws_test.py deleted file mode 100644 index ec91f6a1b..000000000 --- a/acme/acme/jose/jws_test.py +++ /dev/null @@ -1,239 +0,0 @@ -"""Tests for acme.jose.jws.""" -import base64 -import unittest - -import mock -import OpenSSL - -from acme import test_util - -from acme.jose import errors -from acme.jose import json_util -from acme.jose import jwa -from acme.jose import jwk - - -CERT = test_util.load_comparable_cert('cert.pem') -KEY = jwk.JWKRSA.load(test_util.load_vector('rsa512_key.pem')) - - -class MediaTypeTest(unittest.TestCase): - """Tests for acme.jose.jws.MediaType.""" - - def test_decode(self): - from acme.jose.jws import MediaType - self.assertEqual('application/app', MediaType.decode('application/app')) - self.assertEqual('application/app', MediaType.decode('app')) - self.assertRaises( - errors.DeserializationError, MediaType.decode, 'app;foo') - - def test_encode(self): - from acme.jose.jws import MediaType - self.assertEqual('app', MediaType.encode('application/app')) - self.assertEqual('application/app;foo', - MediaType.encode('application/app;foo')) - - -class HeaderTest(unittest.TestCase): - """Tests for acme.jose.jws.Header.""" - - def setUp(self): - from acme.jose.jws import Header - self.header1 = Header(jwk='foo') - self.header2 = Header(jwk='bar') - self.crit = Header(crit=('a', 'b')) - self.empty = Header() - - def test_add_non_empty(self): - from acme.jose.jws import Header - self.assertEqual(Header(jwk='foo', crit=('a', 'b')), - self.header1 + self.crit) - - def test_add_empty(self): - self.assertEqual(self.header1, self.header1 + self.empty) - self.assertEqual(self.header1, self.empty + self.header1) - - def test_add_overlapping_error(self): - self.assertRaises(TypeError, self.header1.__add__, self.header2) - - def test_add_wrong_type_error(self): - self.assertRaises(TypeError, self.header1.__add__, 'xxx') - - def test_crit_decode_always_errors(self): - from acme.jose.jws import Header - self.assertRaises(errors.DeserializationError, Header.from_json, - {'crit': ['a', 'b']}) - - def test_x5c_decoding(self): - from acme.jose.jws import Header - header = Header(x5c=(CERT, CERT)) - jobj = header.to_partial_json() - cert_asn1 = OpenSSL.crypto.dump_certificate( - OpenSSL.crypto.FILETYPE_ASN1, CERT.wrapped) - cert_b64 = base64.b64encode(cert_asn1) - self.assertEqual(jobj, {'x5c': [cert_b64, cert_b64]}) - self.assertEqual(header, Header.from_json(jobj)) - jobj['x5c'][0] = base64.b64encode(b'xxx' + cert_asn1) - self.assertRaises(errors.DeserializationError, Header.from_json, jobj) - - def test_find_key(self): - self.assertEqual('foo', self.header1.find_key()) - self.assertEqual('bar', self.header2.find_key()) - self.assertRaises(errors.Error, self.crit.find_key) - - -class SignatureTest(unittest.TestCase): - """Tests for acme.jose.jws.Signature.""" - - def test_from_json(self): - from acme.jose.jws import Header - from acme.jose.jws import Signature - self.assertEqual( - Signature(signature=b'foo', header=Header(alg=jwa.RS256)), - Signature.from_json( - {'signature': 'Zm9v', 'header': {'alg': 'RS256'}})) - - def test_from_json_no_alg_error(self): - from acme.jose.jws import Signature - self.assertRaises(errors.DeserializationError, - Signature.from_json, {'signature': 'foo'}) - - -class JWSTest(unittest.TestCase): - """Tests for acme.jose.jws.JWS.""" - - def setUp(self): - self.privkey = KEY - self.pubkey = self.privkey.public_key() - - from acme.jose.jws import JWS - self.unprotected = JWS.sign( - payload=b'foo', key=self.privkey, alg=jwa.RS256) - self.protected = JWS.sign( - payload=b'foo', key=self.privkey, alg=jwa.RS256, - protect=frozenset(['jwk', 'alg'])) - self.mixed = JWS.sign( - payload=b'foo', key=self.privkey, alg=jwa.RS256, - protect=frozenset(['alg'])) - - def test_pubkey_jwk(self): - self.assertEqual(self.unprotected.signature.combined.jwk, self.pubkey) - self.assertEqual(self.protected.signature.combined.jwk, self.pubkey) - self.assertEqual(self.mixed.signature.combined.jwk, self.pubkey) - - def test_sign_unprotected(self): - self.assertTrue(self.unprotected.verify()) - - def test_sign_protected(self): - self.assertTrue(self.protected.verify()) - - def test_sign_mixed(self): - self.assertTrue(self.mixed.verify()) - - def test_compact_lost_unprotected(self): - compact = self.mixed.to_compact() - self.assertEqual( - b'eyJhbGciOiAiUlMyNTYifQ.Zm9v.OHdxFVj73l5LpxbFp1AmYX4yJM0Pyb' - b'_893n1zQjpim_eLS5J1F61lkvrCrCDErTEJnBGOGesJ72M7b6Ve1cAJA', - compact) - - from acme.jose.jws import JWS - mixed = JWS.from_compact(compact) - - self.assertNotEqual(self.mixed, mixed) - self.assertEqual( - set(['alg']), set(mixed.signature.combined.not_omitted())) - - def test_from_compact_missing_components(self): - from acme.jose.jws import JWS - self.assertRaises(errors.DeserializationError, JWS.from_compact, b'.') - - def test_json_omitempty(self): - protected_jobj = self.protected.to_partial_json(flat=True) - unprotected_jobj = self.unprotected.to_partial_json(flat=True) - - self.assertTrue('protected' not in unprotected_jobj) - self.assertTrue('header' not in protected_jobj) - - unprotected_jobj['header'] = unprotected_jobj['header'].to_json() - - from acme.jose.jws import JWS - self.assertEqual(JWS.from_json(protected_jobj), self.protected) - self.assertEqual(JWS.from_json(unprotected_jobj), self.unprotected) - - def test_json_flat(self): - jobj_to = { - 'signature': json_util.encode_b64jose( - self.mixed.signature.signature), - 'payload': json_util.encode_b64jose(b'foo'), - 'header': self.mixed.signature.header, - 'protected': json_util.encode_b64jose( - self.mixed.signature.protected.encode('utf-8')), - } - jobj_from = jobj_to.copy() - jobj_from['header'] = jobj_from['header'].to_json() - - self.assertEqual(self.mixed.to_partial_json(flat=True), jobj_to) - from acme.jose.jws import JWS - self.assertEqual(self.mixed, JWS.from_json(jobj_from)) - - def test_json_not_flat(self): - jobj_to = { - 'signatures': (self.mixed.signature,), - 'payload': json_util.encode_b64jose(b'foo'), - } - jobj_from = jobj_to.copy() - jobj_from['signatures'] = [jobj_to['signatures'][0].to_json()] - - self.assertEqual(self.mixed.to_partial_json(flat=False), jobj_to) - from acme.jose.jws import JWS - self.assertEqual(self.mixed, JWS.from_json(jobj_from)) - - def test_from_json_mixed_flat(self): - from acme.jose.jws import JWS - self.assertRaises(errors.DeserializationError, JWS.from_json, - {'signatures': (), 'signature': 'foo'}) - - def test_from_json_hashable(self): - from acme.jose.jws import JWS - hash(JWS.from_json(self.mixed.to_json())) - - -class CLITest(unittest.TestCase): - - def setUp(self): - self.key_path = test_util.vector_path('rsa512_key.pem') - - def test_unverified(self): - from acme.jose.jws import CLI - with mock.patch('sys.stdin') as sin: - sin.read.return_value = '{"payload": "foo", "signature": "xxx"}' - with mock.patch('sys.stdout'): - self.assertEqual(-1, CLI.run(['verify'])) - - def test_json(self): - from acme.jose.jws import CLI - - with mock.patch('sys.stdin') as sin: - sin.read.return_value = 'foo' - with mock.patch('sys.stdout') as sout: - CLI.run(['sign', '-k', self.key_path, '-a', 'RS256', - '-p', 'jwk']) - sin.read.return_value = sout.write.mock_calls[0][1][0] - self.assertEqual(0, CLI.run(['verify'])) - - def test_compact(self): - from acme.jose.jws import CLI - - with mock.patch('sys.stdin') as sin: - sin.read.return_value = 'foo' - with mock.patch('sys.stdout') as sout: - CLI.run(['--compact', 'sign', '-k', self.key_path]) - sin.read.return_value = sout.write.mock_calls[0][1][0] - self.assertEqual(0, CLI.run([ - '--compact', 'verify', '--kty', 'RSA', - '-k', self.key_path])) - - -if __name__ == '__main__': - unittest.main() # pragma: no cover diff --git a/acme/acme/jose/util.py b/acme/acme/jose/util.py deleted file mode 100644 index 26b7e0c5a..000000000 --- a/acme/acme/jose/util.py +++ /dev/null @@ -1,226 +0,0 @@ -"""JOSE utilities.""" -import collections - -from cryptography.hazmat.primitives.asymmetric import rsa -import OpenSSL -import six - - -class abstractclassmethod(classmethod): - # pylint: disable=invalid-name,too-few-public-methods - """Descriptor for an abstract classmethod. - - It augments the :mod:`abc` framework with an abstract - classmethod. This is implemented as :class:`abc.abstractclassmethod` - in the standard Python library starting with version 3.2. - - This particular implementation, allegedly based on Python 3.3 source - code, is stolen from - http://stackoverflow.com/questions/11217878/python-2-7-combine-abc-abstractmethod-and-classmethod. - - """ - __isabstractmethod__ = True - - def __init__(self, target): - target.__isabstractmethod__ = True - super(abstractclassmethod, self).__init__(target) - - -class ComparableX509(object): # pylint: disable=too-few-public-methods - """Wrapper for OpenSSL.crypto.X509** objects that supports __eq__. - - :ivar wrapped: Wrapped certificate or certificate request. - :type wrapped: `OpenSSL.crypto.X509` or `OpenSSL.crypto.X509Req`. - - """ - def __init__(self, wrapped): - assert isinstance(wrapped, OpenSSL.crypto.X509) or isinstance( - wrapped, OpenSSL.crypto.X509Req) - self.wrapped = wrapped - - def __getattr__(self, name): - return getattr(self.wrapped, name) - - def _dump(self, filetype=OpenSSL.crypto.FILETYPE_ASN1): - """Dumps the object into a buffer with the specified encoding. - - :param int filetype: The desired encoding. Should be one of - `OpenSSL.crypto.FILETYPE_ASN1`, - `OpenSSL.crypto.FILETYPE_PEM`, or - `OpenSSL.crypto.FILETYPE_TEXT`. - - :returns: Encoded X509 object. - :rtype: str - - """ - if isinstance(self.wrapped, OpenSSL.crypto.X509): - func = OpenSSL.crypto.dump_certificate - else: # assert in __init__ makes sure this is X509Req - func = OpenSSL.crypto.dump_certificate_request - return func(filetype, self.wrapped) - - def __eq__(self, other): - if not isinstance(other, self.__class__): - return NotImplemented - # pylint: disable=protected-access - return self._dump() == other._dump() - - def __hash__(self): - return hash((self.__class__, self._dump())) - - def __ne__(self, other): - return not self == other - - def __repr__(self): - return '<{0}({1!r})>'.format(self.__class__.__name__, self.wrapped) - - -class ComparableKey(object): # pylint: disable=too-few-public-methods - """Comparable wrapper for `cryptography` keys. - - See https://github.com/pyca/cryptography/issues/2122. - - """ - __hash__ = NotImplemented - - def __init__(self, wrapped): - self._wrapped = wrapped - - def __getattr__(self, name): - return getattr(self._wrapped, name) - - def __eq__(self, other): - # pylint: disable=protected-access - if (not isinstance(other, self.__class__) or - self._wrapped.__class__ is not other._wrapped.__class__): - return NotImplemented - elif hasattr(self._wrapped, 'private_numbers'): - return self.private_numbers() == other.private_numbers() - elif hasattr(self._wrapped, 'public_numbers'): - return self.public_numbers() == other.public_numbers() - else: - return NotImplemented - - def __ne__(self, other): - return not self == other - - def __repr__(self): - return '<{0}({1!r})>'.format(self.__class__.__name__, self._wrapped) - - def public_key(self): - """Get wrapped public key.""" - return self.__class__(self._wrapped.public_key()) - - -class ComparableRSAKey(ComparableKey): # pylint: disable=too-few-public-methods - """Wrapper for `cryptography` RSA keys. - - Wraps around: - - `cryptography.hazmat.primitives.asymmetric.RSAPrivateKey` - - `cryptography.hazmat.primitives.asymmetric.RSAPublicKey` - - """ - - def __hash__(self): - # public_numbers() hasn't got stable hash! - # https://github.com/pyca/cryptography/issues/2143 - if isinstance(self._wrapped, rsa.RSAPrivateKeyWithSerialization): - priv = self.private_numbers() - pub = priv.public_numbers - return hash((self.__class__, priv.p, priv.q, priv.dmp1, - priv.dmq1, priv.iqmp, pub.n, pub.e)) - elif isinstance(self._wrapped, rsa.RSAPublicKeyWithSerialization): - pub = self.public_numbers() - return hash((self.__class__, pub.n, pub.e)) - - -class ImmutableMap(collections.Mapping, collections.Hashable): # type: ignore - # pylint: disable=too-few-public-methods - """Immutable key to value mapping with attribute access.""" - - __slots__ = () - """Must be overridden in subclasses.""" - - def __init__(self, **kwargs): - if set(kwargs) != set(self.__slots__): - raise TypeError( - '__init__() takes exactly the following arguments: {0} ' - '({1} given)'.format(', '.join(self.__slots__), - ', '.join(kwargs) if kwargs else 'none')) - for slot in self.__slots__: - object.__setattr__(self, slot, kwargs.pop(slot)) - - def update(self, **kwargs): - """Return updated map.""" - items = dict(self) - items.update(kwargs) - return type(self)(**items) # pylint: disable=star-args - - def __getitem__(self, key): - try: - return getattr(self, key) - except AttributeError: - raise KeyError(key) - - def __iter__(self): - return iter(self.__slots__) - - def __len__(self): - return len(self.__slots__) - - def __hash__(self): - return hash(tuple(getattr(self, slot) for slot in self.__slots__)) - - def __setattr__(self, name, value): - raise AttributeError("can't set attribute") - - def __repr__(self): - return '{0}({1})'.format(self.__class__.__name__, ', '.join( - '{0}={1!r}'.format(key, value) - for key, value in six.iteritems(self))) - - -class frozendict(collections.Mapping, collections.Hashable): # type: ignore - # pylint: disable=invalid-name,too-few-public-methods - """Frozen dictionary.""" - __slots__ = ('_items', '_keys') - - def __init__(self, *args, **kwargs): - if kwargs and not args: - items = dict(kwargs) - elif len(args) == 1 and isinstance(args[0], collections.Mapping): - items = args[0] - else: - raise TypeError() - # TODO: support generators/iterators - - object.__setattr__(self, '_items', items) - object.__setattr__(self, '_keys', tuple(sorted(six.iterkeys(items)))) - - def __getitem__(self, key): - return self._items[key] - - def __iter__(self): - return iter(self._keys) - - def __len__(self): - return len(self._items) - - def _sorted_items(self): - return tuple((key, self[key]) for key in self._keys) - - def __hash__(self): - return hash(self._sorted_items()) - - def __getattr__(self, name): - try: - return self._items[name] - except KeyError: - raise AttributeError(name) - - def __setattr__(self, name, value): - raise AttributeError("can't set attribute") - - def __repr__(self): - return 'frozendict({0})'.format(', '.join('{0}={1!r}'.format( - key, value) for key, value in self._sorted_items())) diff --git a/acme/acme/jose/util_test.py b/acme/acme/jose/util_test.py deleted file mode 100644 index 0038a6cc1..000000000 --- a/acme/acme/jose/util_test.py +++ /dev/null @@ -1,199 +0,0 @@ -"""Tests for acme.jose.util.""" -import functools -import unittest - -import six - -from acme import test_util - - -class ComparableX509Test(unittest.TestCase): - """Tests for acme.jose.util.ComparableX509.""" - - def setUp(self): - # test_util.load_comparable_{csr,cert} return ComparableX509 - self.req1 = test_util.load_comparable_csr('csr.pem') - self.req2 = test_util.load_comparable_csr('csr.pem') - self.req_other = test_util.load_comparable_csr('csr-san.pem') - - self.cert1 = test_util.load_comparable_cert('cert.pem') - self.cert2 = test_util.load_comparable_cert('cert.pem') - self.cert_other = test_util.load_comparable_cert('cert-san.pem') - - def test_getattr_proxy(self): - self.assertTrue(self.cert1.has_expired()) - - def test_eq(self): - self.assertEqual(self.req1, self.req2) - self.assertEqual(self.cert1, self.cert2) - - def test_ne(self): - self.assertNotEqual(self.req1, self.req_other) - self.assertNotEqual(self.cert1, self.cert_other) - - def test_ne_wrong_types(self): - self.assertNotEqual(self.req1, 5) - self.assertNotEqual(self.cert1, 5) - - def test_hash(self): - self.assertEqual(hash(self.req1), hash(self.req2)) - self.assertNotEqual(hash(self.req1), hash(self.req_other)) - - self.assertEqual(hash(self.cert1), hash(self.cert2)) - self.assertNotEqual(hash(self.cert1), hash(self.cert_other)) - - def test_repr(self): - for x509 in self.req1, self.cert1: - self.assertEqual(repr(x509), - ''.format(x509.wrapped)) - - -class ComparableRSAKeyTest(unittest.TestCase): - """Tests for acme.jose.util.ComparableRSAKey.""" - - def setUp(self): - # test_utl.load_rsa_private_key return ComparableRSAKey - self.key = test_util.load_rsa_private_key('rsa256_key.pem') - self.key_same = test_util.load_rsa_private_key('rsa256_key.pem') - self.key2 = test_util.load_rsa_private_key('rsa512_key.pem') - - def test_getattr_proxy(self): - self.assertEqual(256, self.key.key_size) - - def test_eq(self): - self.assertEqual(self.key, self.key_same) - - def test_ne(self): - self.assertNotEqual(self.key, self.key2) - - def test_ne_different_types(self): - self.assertNotEqual(self.key, 5) - - def test_ne_not_wrapped(self): - # pylint: disable=protected-access - self.assertNotEqual(self.key, self.key_same._wrapped) - - def test_ne_no_serialization(self): - from acme.jose.util import ComparableRSAKey - self.assertNotEqual(ComparableRSAKey(5), ComparableRSAKey(5)) - - def test_hash(self): - self.assertTrue(isinstance(hash(self.key), int)) - self.assertEqual(hash(self.key), hash(self.key_same)) - self.assertNotEqual(hash(self.key), hash(self.key2)) - - def test_repr(self): - self.assertTrue(repr(self.key).startswith( - '=0.6) # rsa_recover_prime_factors (>=0.8) 'cryptography>=0.8', + # formerly known as acme.jose: + 'josepy>=1.0.0', # Connection.set_tlsext_host_name (>=0.13) 'mock', 'PyOpenSSL>=0.13', @@ -74,10 +76,5 @@ setup( 'dev': dev_extras, 'docs': docs_extras, }, - entry_points={ - 'console_scripts': [ - 'jws = acme.jose.jws:CLI.run', - ], - }, test_suite='acme', ) diff --git a/certbot-apache/certbot_apache/tests/util.py b/certbot-apache/certbot_apache/tests/util.py index 2405110c5..ca667465c 100644 --- a/certbot-apache/certbot_apache/tests/util.py +++ b/certbot-apache/certbot_apache/tests/util.py @@ -5,11 +5,10 @@ import sys import unittest import augeas +import josepy as jose import mock import zope.component -from acme import jose - from certbot.display import util as display_util from certbot.plugins import common diff --git a/certbot-compatibility-test/certbot_compatibility_test/util.py b/certbot-compatibility-test/certbot_compatibility_test/util.py index af951aa6a..4155944bd 100644 --- a/certbot-compatibility-test/certbot_compatibility_test/util.py +++ b/certbot-compatibility-test/certbot_compatibility_test/util.py @@ -6,7 +6,8 @@ import re import shutil import tarfile -from acme import jose +import josepy as jose + from acme import test_util from certbot import constants diff --git a/certbot-nginx/certbot_nginx/tests/util.py b/certbot-nginx/certbot_nginx/tests/util.py index 6e1b0d8ff..7b32d8e82 100644 --- a/certbot-nginx/certbot_nginx/tests/util.py +++ b/certbot-nginx/certbot_nginx/tests/util.py @@ -5,11 +5,10 @@ import pkg_resources import tempfile import unittest +import josepy as jose import mock import zope.component -from acme import jose - from certbot import configuration from certbot.tests import util as test_util diff --git a/certbot/account.py b/certbot/account.py index 389f96791..41e980097 100644 --- a/certbot/account.py +++ b/certbot/account.py @@ -7,13 +7,13 @@ import shutil import socket from cryptography.hazmat.primitives import serialization +import josepy as jose import pyrfc3339 import pytz import six import zope.component from acme import fields as acme_fields -from acme import jose from acme import messages from certbot import errors diff --git a/certbot/achallenges.py b/certbot/achallenges.py index f39bb4cec..6535a6b63 100644 --- a/certbot/achallenges.py +++ b/certbot/achallenges.py @@ -19,8 +19,9 @@ Note, that all annotated challenges act as a proxy objects:: """ import logging +import josepy as jose + from acme import challenges -from acme import jose logger = logging.getLogger(__name__) diff --git a/certbot/client.py b/certbot/client.py index ed70fda71..b735421f5 100644 --- a/certbot/client.py +++ b/certbot/client.py @@ -5,13 +5,13 @@ import platform from cryptography.hazmat.backends import default_backend from cryptography.hazmat.primitives.asymmetric import rsa +import josepy as jose import OpenSSL import zope.component from acme import client as acme_client from acme import crypto_util as acme_crypto_util from acme import errors as acme_errors -from acme import jose from acme import messages import certbot diff --git a/certbot/crypto_util.py b/certbot/crypto_util.py index 112ef7c85..3ae16529d 100644 --- a/certbot/crypto_util.py +++ b/certbot/crypto_util.py @@ -14,9 +14,9 @@ import six import zope.component from cryptography.hazmat.backends import default_backend from cryptography import x509 +import josepy as jose from acme import crypto_util as acme_crypto_util -from acme import jose from certbot import errors from certbot import interfaces @@ -368,7 +368,7 @@ def dump_pyopenssl_chain(chain, filetype=OpenSSL.crypto.FILETYPE_PEM): """Dump certificate chain into a bundle. :param list chain: List of `OpenSSL.crypto.X509` (or wrapped in - `acme.jose.ComparableX509`). + :class:`josepy.util.ComparableX509`). """ # XXX: returns empty string when no chain is available, which diff --git a/certbot/main.py b/certbot/main.py index 72af7fbba..1c6432fd9 100644 --- a/certbot/main.py +++ b/certbot/main.py @@ -6,9 +6,9 @@ import os import sys import configobj +import josepy as jose import zope.component -from acme import jose from acme import errors as acme_errors import certbot diff --git a/certbot/plugins/common.py b/certbot/plugins/common.py index 420d15679..002d2f225 100644 --- a/certbot/plugins/common.py +++ b/certbot/plugins/common.py @@ -9,7 +9,7 @@ import OpenSSL import pkg_resources import zope.interface -from acme.jose import util as jose_util +from josepy import util as jose_util from certbot import constants from certbot import crypto_util diff --git a/certbot/plugins/common_test.py b/certbot/plugins/common_test.py index 8ce68bbb5..1a1ca7dcb 100644 --- a/certbot/plugins/common_test.py +++ b/certbot/plugins/common_test.py @@ -5,11 +5,11 @@ import shutil import tempfile import unittest +import josepy as jose import mock import OpenSSL from acme import challenges -from acme import jose from certbot import achallenges from certbot import crypto_util diff --git a/certbot/plugins/dns_test_common.py b/certbot/plugins/dns_test_common.py index d8cd29404..54b656b20 100644 --- a/certbot/plugins/dns_test_common.py +++ b/certbot/plugins/dns_test_common.py @@ -3,10 +3,10 @@ import os import configobj +import josepy as jose import mock import six from acme import challenges -from acme import jose from certbot import achallenges from certbot.tests import acme_util diff --git a/certbot/plugins/dns_test_common_lexicon.py b/certbot/plugins/dns_test_common_lexicon.py index f9c5735e8..a221cf1bf 100644 --- a/certbot/plugins/dns_test_common_lexicon.py +++ b/certbot/plugins/dns_test_common_lexicon.py @@ -1,7 +1,7 @@ """Base test class for DNS authenticators built on Lexicon.""" +import josepy as jose import mock -from acme import jose from requests.exceptions import HTTPError, RequestException from certbot import errors diff --git a/certbot/plugins/standalone_test.py b/certbot/plugins/standalone_test.py index 1ae731e42..5227bc59e 100644 --- a/certbot/plugins/standalone_test.py +++ b/certbot/plugins/standalone_test.py @@ -3,11 +3,11 @@ import argparse import socket import unittest +import josepy as jose import mock import six from acme import challenges -from acme import jose from certbot import achallenges from certbot import errors diff --git a/certbot/plugins/webroot_test.py b/certbot/plugins/webroot_test.py index 92160bdfa..36e2ffba6 100644 --- a/certbot/plugins/webroot_test.py +++ b/certbot/plugins/webroot_test.py @@ -10,11 +10,11 @@ import stat import tempfile import unittest +import josepy as jose import mock import six from acme import challenges -from acme import jose from certbot import achallenges from certbot import errors diff --git a/certbot/tests/account_test.py b/certbot/tests/account_test.py index 7245ad6a1..8ebda56af 100644 --- a/certbot/tests/account_test.py +++ b/certbot/tests/account_test.py @@ -6,10 +6,10 @@ import shutil import stat import unittest +import josepy as jose import mock import pytz -from acme import jose from acme import messages from certbot import errors diff --git a/certbot/tests/acme_util.py b/certbot/tests/acme_util.py index f0549666a..53a2f214a 100644 --- a/certbot/tests/acme_util.py +++ b/certbot/tests/acme_util.py @@ -1,10 +1,10 @@ """ACME utilities for testing.""" import datetime +import josepy as jose import six from acme import challenges -from acme import jose from acme import messages from certbot import auth_handler diff --git a/certbot/tests/client_test.py b/certbot/tests/client_test.py index 09c4a50ca..204f46323 100644 --- a/certbot/tests/client_test.py +++ b/certbot/tests/client_test.py @@ -4,11 +4,11 @@ import shutil import tempfile import unittest +import josepy as jose import OpenSSL import mock from acme import errors as acme_errors -from acme import jose from certbot import account from certbot import errors diff --git a/certbot/tests/display/ops_test.py b/certbot/tests/display/ops_test.py index cb0fb32e3..57d82f839 100644 --- a/certbot/tests/display/ops_test.py +++ b/certbot/tests/display/ops_test.py @@ -4,10 +4,10 @@ import os import sys import unittest +import josepy as jose import mock import zope.component -from acme import jose from acme import messages from certbot import account diff --git a/certbot/tests/main_test.py b/certbot/tests/main_test.py index 1f690df26..04b71dcc7 100644 --- a/certbot/tests/main_test.py +++ b/certbot/tests/main_test.py @@ -11,11 +11,10 @@ import unittest import datetime import pytz +import josepy as jose import six from six.moves import reload_module # pylint: disable=import-error -from acme import jose - from certbot import account from certbot import cli from certbot import constants diff --git a/certbot/tests/util.py b/certbot/tests/util.py index c43b44522..ddd4a1aec 100644 --- a/certbot/tests/util.py +++ b/certbot/tests/util.py @@ -14,11 +14,10 @@ from cryptography.hazmat.backends import default_backend from cryptography.hazmat.primitives import serialization import mock import OpenSSL +import josepy as jose import six from six.moves import reload_module # pylint: disable=import-error -from acme import jose - from certbot import constants from certbot import interfaces from certbot import storage diff --git a/tools/deactivate.py b/tools/deactivate.py index 5facc8436..d43b84552 100644 --- a/tools/deactivate.py +++ b/tools/deactivate.py @@ -18,10 +18,10 @@ import sys from cryptography.hazmat.backends import default_backend from cryptography.hazmat.primitives.asymmetric import rsa from cryptography.hazmat.primitives import serialization +import josepy as jose from acme import client as acme_client from acme import errors as acme_errors -from acme import jose from acme import messages DIRECTORY = os.getenv('DIRECTORY', 'http://localhost:4000/directory')