Move decoders/encoders to top-level acme.util

letsencrypt/acme/challenges.py

@@ -117,8 +117,8 @@ class DVSNI(DVChallenge):
 
     @classmethod
     def from_valid_json(cls, jobj):
-        return cls(r=cls._decode_b64jose(jobj["r"], cls.R_SIZE),
+        return cls(r=util.decode_b64jose(jobj["r"], cls.R_SIZE),
-                   nonce=cls._decode_hex16(jobj["nonce"], cls.NONCE_SIZE))
+                   nonce=util.decode_hex16(jobj["nonce"], cls.NONCE_SIZE))
 
 
 @ChallengeResponse.register
@@ -162,7 +162,7 @@ class DVSNIResponse(ChallengeResponse):
 
     @classmethod
     def from_valid_json(cls, jobj):
-        return cls(s=cls._decode_b64jose(jobj["s"], cls.S_SIZE))
+        return cls(s=util.decode_b64jose(jobj["s"], cls.S_SIZE))
 
 
 @Challenge.register
@@ -271,7 +271,7 @@ class ProofOfPossession(ClientChallenge):
             fields = {"jwk": self.jwk}
             add = functools.partial(_extend_if_not_empty, fields)
             add(self.cert_fingerprints, "certFingerprints")
-            add([self._encode_cert(cert) for cert in self.certs], "certs")
+            add([util.encode_cert(cert) for cert in self.certs], "certs")
             add(self.subject_key_identifiers, "subjectKeyIdentifiers")
             add(self.serial_numbers, "serialNumbers")
             add(self.issuers, "issuers")
@@ -283,7 +283,7 @@ class ProofOfPossession(ClientChallenge):
             return cls(
                 jwk=other.JWK.from_valid_json(jobj["jwk"]),
                 cert_fingerprints=jobj.get("certFingerprints", []),
-                certs=[cls._decode_cert(cert)
+                certs=[util.decode_cert(cert)
                        for cert in jobj.get("certs", [])],
                 subject_key_identifiers=jobj.get("subjectKeyIdentifiers", []),
                 serial_numbers=jobj.get("serialNumbers", []),
@@ -300,7 +300,7 @@ class ProofOfPossession(ClientChallenge):
     @classmethod
     def from_valid_json(cls, jobj):
         return cls(alg=jobj["alg"],
-                   nonce=cls._decode_b64jose(jobj["nonce"], cls.NONCE_SIZE),
+                   nonce=util.decode_b64jose(jobj["nonce"], cls.NONCE_SIZE),
                    hints=cls.Hints.from_valid_json(jobj["hints"]))
 
 
@@ -329,7 +329,7 @@ class ProofOfPossessionResponse(ChallengeResponse):
 
     @classmethod
     def from_valid_json(cls, jobj):
-        return cls(nonce=cls._decode_b64jose(jobj["nonce"], cls.NONCE_SIZE),
+        return cls(nonce=util.decode_b64jose(jobj["nonce"], cls.NONCE_SIZE),
                    signature=other.Signature.from_valid_json(jobj["signature"]))
 
 

letsencrypt/acme/messages.py

@@ -125,7 +125,7 @@ class Challenge(Message):
         # TODO: turn "combinations" elements into sets?
         # TODO: turn "combinations" into set?
         return cls(session_id=jobj["sessionID"],
-                   nonce=cls._decode_b64jose(jobj["nonce"]),
+                   nonce=util.decode_b64jose(jobj["nonce"]),
                    challenges=[challenges.Challenge.from_valid_json(chall)
                                for chall in jobj["challenges"]],
                    combinations=jobj.get("combinations", []))
@@ -245,7 +245,7 @@ class AuthorizationRequest(Message):
     def from_valid_json(cls, jobj):
         return cls(
             session_id=jobj["sessionID"],
-            nonce=cls._decode_b64jose(jobj["nonce"]),
+            nonce=util.decode_b64jose(jobj["nonce"]),
             responses=[challenges.ChallengeResponse.from_valid_json(chall)
                        for chall in jobj["responses"]],
             signature=other.Signature.from_valid_json(jobj["signature"]),
@@ -268,17 +268,17 @@ class Certificate(Message):
     __slots__ = ("certificate", "chain", "refresh")
 
     def _fields_to_json(self):
-        fields = {"certificate": self._encode_cert(self.certificate)}
+        fields = {"certificate": util.encode_cert(self.certificate)}
         if self.chain:
-            fields["chain"] = [self._encode_cert(cert) for cert in self.chain]
+            fields["chain"] = [util.encode_cert(cert) for cert in self.chain]
         if self.refresh is not None:
             fields["refresh"] = self.refresh
         return fields
 
     @classmethod
     def from_valid_json(cls, jobj):
-        return cls(certificate=cls._decode_cert(jobj["certificate"]),
+        return cls(certificate=util.decode_cert(jobj["certificate"]),
-                   chain=[cls._decode_cert(cert) for cert in
+                   chain=[util.decode_cert(cert) for cert in
                           jobj.get("chain", [])],
                    refresh=jobj.get("refresh"))
 
@@ -328,13 +328,13 @@ class CertificateRequest(Message):
 
     def _fields_to_json(self):
         return {
-            "csr": self._encode_csr(self.csr),
+            "csr": util.encode_csr(self.csr),
             "signature": self.signature,
         }
 
     @classmethod
     def from_valid_json(cls, jobj):
-        return cls(csr=cls._decode_csr(jobj["csr"]),
+        return cls(csr=util.decode_csr(jobj["csr"]),
                    signature=other.Signature.from_valid_json(jobj["signature"]))
 
 
@@ -449,13 +449,13 @@ class RevocationRequest(Message):
 
     def _fields_to_json(self):
         return {
-            "certificate": self._encode_cert(self.certificate),
+            "certificate": util.encode_cert(self.certificate),
             "signature": self.signature,
         }
 
     @classmethod
     def from_valid_json(cls, jobj):
-        return cls(certificate=cls._decode_cert(jobj["certificate"]),
+        return cls(certificate=util.decode_cert(jobj["certificate"]),
                    signature=other.Signature.from_valid_json(jobj["signature"]))
 
 

letsencrypt/acme/other.py

@@ -34,7 +34,7 @@ class JWK(util.ACMEObject):
     @classmethod
     def _decode_param(cls, data):
         try:
-            return long(binascii.hexlify(cls._decode_b64jose(data)), 16)
+            return long(binascii.hexlify(util.decode_b64jose(data)), 16)
         except ValueError:  # invalid literal for long() with base 16
             raise errors.ValidationError(data)
 
@@ -123,7 +123,7 @@ class Signature(util.ACMEObject):
     @classmethod
     def from_valid_json(cls, jobj):
         assert jobj['alg'] == 'RS256'  # TODO: support other algorithms
-        return cls(alg=jobj['alg'], sig=cls._decode_b64jose(jobj['sig']),
+        return cls(alg=jobj['alg'], sig=util.decode_b64jose(jobj['sig']),
-                   nonce=cls._decode_b64jose(
+                   nonce=util.decode_b64jose(
                        jobj['nonce'], cls.NONCE_SIZE, minimum=True),
                    jwk=JWK.from_valid_json(jobj['jwk']))

letsencrypt/acme/util.py

@@ -99,49 +99,99 @@ class ACMEObject(ImmutableMap):  # pylint: disable=too-few-public-methods
         """Deserialize from valid JSON object."""
         raise NotImplementedError()
 
-    @classmethod
-    def _decode_b64jose(cls, data, size=None, minimum=False):
-        try:
-            decoded = jose.b64decode(data)
-        except TypeError:
-            raise errors.ValidationError()
 
-        if size is not None and ((not minimum and len(decoded) != size)
+def decode_b64jose(value, size=None, minimum=False):
-                                 or (minimum and len(decoded) < size)):
+    """Decode ACME object JOSE Base64 encoded field.
-            raise errors.ValidationError()
 
-        return decoded
+    :param str value: Encoded field value.
+    :param int size: If specified, this function will check if data size
+        (after decoding) matches.
+    :param bool minimum: If ``True``, then ``size`` is the minimum required
+        size, otherwise ``size`` must be exact.
 
-    @classmethod
+    :raises letsencrypt.acme.errors.ValidationError: if anything goes wrong
-    def _decode_hex16(cls, data, size=None, minimum=False):
+    :returns: Decoded value.
-        if size is not None and ((not minimum and len(data) != size * 2)
-                                 or (minimum and len(data) < size * 2)):
-            raise errors.ValidationError()
-        return binascii.unhexlify(data)
 
-    @classmethod
+    """
-    def _encode_cert(cls, cert):
+    try:
-        return jose.b64encode(cert.as_der())
+        decoded = jose.b64decode(value)
+    except TypeError:
+        raise errors.ValidationError()
 
-    @classmethod
+    if size is not None and ((not minimum and len(decoded) != size)
-    def _decode_cert(cls, b64der):
+                             or (minimum and len(decoded) < size)):
-        try:
+        raise errors.ValidationError()
-            return ComparableX509(M2Crypto.X509.load_cert_der_string(
-                cls._decode_b64jose(b64der)))
-        except M2Crypto.X509.X509Error:
-            raise errors.ValidationError()
 
-    @classmethod
+    return decoded
-    def _encode_csr(cls, csr):
-        return cls._encode_cert(csr)
 
-    @classmethod
+
-    def _decode_csr(cls, b64der):
+def decode_hex16(value, size=None, minimum=False):
-        try:
+    """Decode ACME object hex16-encoded field.
-            return ComparableX509(M2Crypto.X509.load_request_der_string(
+
-                cls._decode_b64jose(b64der)))
+    :param str value: Encoded field value.
-        except M2Crypto.X509.X509Error:
+    :param int size: If specified, this function will check if data size
-            raise errors.ValidationError()
+        (after decoding) matches.
+    :param bool minimum: If ``True``, then ``size`` is the minimum required
+        size, otherwise ``size`` must be exact.
+
+    """
+    # binascii.hexlify.__doc__: "The resulting string is therefore twice
+    # as long as the length of data."
+    if size is not None and ((not minimum and len(value) != size * 2)
+                             or (minimum and len(value) < size * 2)):
+        raise errors.ValidationError()
+    try:
+        return binascii.unhexlify(value)
+    except TypeError as error:  # odd-length string (binascci.unhexlify.__doc__)
+        raise errors.ValidationError(error)
+
+
+def encode_cert(cert):
+    """Encode ACME object X509 certificate field."""
+    return jose.b64encode(cert.as_der())
+
+
+def decode_cert(b64der):
+    """Decode ACME object X509 certificate field.
+
+    :param str b64der: Input data that's meant to be valid base64
+        DER-encoded certificate.
+
+    :raises letsencrypt.acme.errors.ValidationError: if anything goes wrong
+
+    :returns: Decoded certificate.
+    :rtype: :class:`M2Crypto.X509.X509` wrapped in :class:`ComparableX509`.
+
+    """
+    try:
+        return ComparableX509(M2Crypto.X509.load_cert_der_string(
+            decode_b64jose(b64der)))
+    except M2Crypto.X509.X509Error:
+        raise errors.ValidationError()
+
+
+def encode_csr(csr):
+    """Encode ACME object CSR field."""
+    return encode_cert(csr)
+
+
+def decode_csr(b64der):
+    """Decode ACME object CSR field.
+
+    :param str b64der: Input data that's meant to be valid base64
+        DER-encoded CSR.
+
+    :raises letsencrypt.acme.errors.ValidationError: if anything goes wrong
+
+    :returns: Decoded certificate.
+    :rtype: :class:`M2Crypto.X509.X509` wrapped in :class:`ComparableX509`.
+
+    """
+    try:
+        return ComparableX509(M2Crypto.X509.load_request_der_string(
+            decode_b64jose(b64der)))
+    except M2Crypto.X509.X509Error:
+        raise errors.ValidationError()
 
 
 class TypedACMEObject(ACMEObject):

letsencrypt/acme/util_test.py

@@ -109,8 +109,8 @@ class ImmutableMapTest(unittest.TestCase):
         self.assertEqual("B(x='foo', y='bar')", repr(self.B(x='foo', y='bar')))
 
 
-class ACMEObjectTest(unittest.TestCase):
+class EncodersAndDecodersTest(unittest.TestCase):
-    """Tests for letsencrypt.acme.util.ACMEObject."""
+    """Tests for encoders and decoders from letsencrypt.acme.util"""
     # pylint: disable=protected-access
 
     def setUp(self):
@@ -139,57 +139,62 @@ class ACMEObjectTest(unittest.TestCase):
         )
 
     def test_decode_b64_jose_padding_error(self):
-        from letsencrypt.acme.util import ACMEObject
+        from letsencrypt.acme.util import decode_b64jose
-        self.assertRaises(
+        self.assertRaises(errors.ValidationError, decode_b64jose, 'x')
-            errors.ValidationError, ACMEObject._decode_b64jose, 'x')
 
     def test_decode_b64_jose_size(self):
-        from letsencrypt.acme.util import ACMEObject
+        from letsencrypt.acme.util import decode_b64jose
-        self.assertEqual('foo', ACMEObject._decode_b64jose('Zm9v', size=3))
+        self.assertEqual('foo', decode_b64jose('Zm9v', size=3))
         self.assertRaises(
-            errors.ValidationError, ACMEObject._decode_b64jose, 'Zm9v', size=2)
+            errors.ValidationError, decode_b64jose, 'Zm9v', size=2)
         self.assertRaises(
-            errors.ValidationError, ACMEObject._decode_b64jose, 'Zm9v', size=4)
+            errors.ValidationError, decode_b64jose, 'Zm9v', size=4)
 
     def test_decode_b64_jose_minimum_size(self):
-        from letsencrypt.acme.util import ACMEObject
+        from letsencrypt.acme.util import decode_b64jose
-        self.assertEqual(
+        self.assertEqual('foo', decode_b64jose('Zm9v', size=3, minimum=True))
-            'foo', ACMEObject._decode_b64jose('Zm9v', size=3, minimum=True))
+        self.assertEqual('foo', decode_b64jose('Zm9v', size=2, minimum=True))
-        self.assertEqual(
+        self.assertRaises(errors.ValidationError, decode_b64jose,
-            'foo', ACMEObject._decode_b64jose('Zm9v', size=2, minimum=True))
-        self.assertRaises(errors.ValidationError, ACMEObject._decode_b64jose,
                           'Zm9v', size=4, minimum=True)
 
     def test_decode_hex16(self):
-        from letsencrypt.acme.util import ACMEObject
+        from letsencrypt.acme.util import decode_hex16
-        self.assertEqual('foo', ACMEObject._decode_hex16('666f6f'))
+        self.assertEqual('foo', decode_hex16('666f6f'))
 
     def test_decode_hex16_minimum_size(self):
-        from letsencrypt.acme.util import ACMEObject
+        from letsencrypt.acme.util import decode_hex16
-        self.assertEqual(
+        self.assertEqual('foo', decode_hex16('666f6f', size=3, minimum=True))
-            'foo', ACMEObject._decode_hex16('666f6f', size=3, minimum=True))
+        self.assertEqual('foo', decode_hex16('666f6f', size=2, minimum=True))
-        self.assertEqual(
+        self.assertRaises(errors.ValidationError, decode_hex16,
-            'foo', ACMEObject._decode_hex16('666f6f', size=2, minimum=True))
-        self.assertRaises(errors.ValidationError, ACMEObject._decode_hex16,
                           '666f6f', size=4, minimum=True)
 
+    def test_decode_hex16_odd_length(self):
+        from letsencrypt.acme.util import decode_hex16
+        self.assertRaises(errors.ValidationError, decode_hex16, 'x')
+
     def test_encode_cert(self):
-        from letsencrypt.acme.util import ACMEObject
+        from letsencrypt.acme.util import encode_cert
-        self.assertEqual(self.b64_cert, ACMEObject._encode_cert(CERT))
+        self.assertEqual(self.b64_cert, encode_cert(CERT))
 
     def test_decode_cert(self):
-        from letsencrypt.acme.util import ACMEObject
+        from letsencrypt.acme.util import ComparableX509
-        self.assertEqual(CERT, ACMEObject._decode_cert(self.b64_cert))
+        from letsencrypt.acme.util import decode_cert
-        self.assertRaises(errors.ValidationError, ACMEObject._decode_cert, '')
+        cert = decode_cert(self.b64_cert)
+        self.assertTrue(isinstance(cert, ComparableX509))
+        self.assertEqual(cert, CERT)
+        self.assertRaises(errors.ValidationError, decode_cert, '')
 
     def test_encode_csr(self):
-        from letsencrypt.acme.util import ACMEObject
+        from letsencrypt.acme.util import encode_csr
-        self.assertEqual(self.b64_csr, ACMEObject._encode_csr(CSR))
+        self.assertEqual(self.b64_csr, encode_csr(CSR))
 
     def test_decode_csr(self):
-        from letsencrypt.acme.util import ACMEObject
+        from letsencrypt.acme.util import ComparableX509
-        self.assertEqual(CSR, ACMEObject._decode_csr(self.b64_csr))
+        from letsencrypt.acme.util import decode_csr
-        self.assertRaises(errors.ValidationError, ACMEObject._decode_csr, '')
+        csr = decode_csr(self.b64_csr)
+        self.assertTrue(isinstance(csr, ComparableX509))
+        self.assertEqual(csr, CSR)
+        self.assertRaises(errors.ValidationError, decode_csr, '')
 
 
 class TypedACMEObjectTest(unittest.TestCase):