Create abstract Verifier and Signer, remove key_id hack from App Engine and IAM signers (#115)

Author: Jon Wayne Parrott

google/auth/_service_account_info.py

@@ -51,7 +51,7 @@ def from_dict(data, require=None):
             'fields {}.'.format(', '.join(missing)))
 
     # Create a signer.
-    signer = crypt.Signer.from_service_account_info(data)
+    signer = crypt.RSASigner.from_service_account_info(data)
 
     return signer
 

google/auth/app_engine.py

@@ -26,49 +26,33 @@
 
 from google.auth import _helpers
 from google.auth import credentials
+from google.auth import crypt
 
 try:
     from google.appengine.api import app_identity
 except ImportError:
     app_identity = None
 
 
-class Signer(object):
+class Signer(crypt.Signer):
     """Signs messages using the App Engine App Identity service.
 
     This can be used in place of :class:`google.auth.crypt.Signer` when
     running in the App Engine standard environment.
-
-    .. warning::
-        The App Identity service signs bytes using Google-managed keys.
-        Because of this it's possible that the key used to sign bytes will
-        change. In some cases this change can occur between successive calls
-        to :attr:`key_id` and :meth:`sign`. This could result in a signature
-        that was signed with a different key than the one indicated by
-        :attr:`key_id`. It's recommended that if you use this in your code
-        that you account for this behavior by building in retry logic.
     """
 
     @property
     def key_id(self):
         """Optional[str]: The key ID used to identify this private key.
 
-        .. note::
-           This makes a request to the App Identity service.
+        .. warning::
+           This is always ``None``. The key ID used by App Engine can not
+           be reliably determined ahead of time.
         """
-        key_id, _ = app_identity.sign_blob(b'')
-        return key_id
-
-    @staticmethod
-    def sign(message):
-        """Signs a message.
-
-        Args:
-            message (Union[str, bytes]): The message to be signed.
+        return None
 
-        Returns:
-            bytes: The signature of the message.
-        """
+    @_helpers.copy_docstring(crypt.Signer)
+    def sign(self, message):
         message = _helpers.to_bytes(message)
         _, signature = app_identity.sign_blob(message)
         return signature

google/auth/crypt.py

@@ -24,20 +24,21 @@
     valid = crypt.verify_signature(message, signature, cert)
 
 If you're going to verify many messages with the same certificate, you can use
-:class:`Verifier`::
+:class:`RSAVerifier`::
 
     cert = open('certs.pem').read()
-    verifier = crypt.Verifier.from_string(cert)
+    verifier = crypt.RSAVerifier.from_string(cert)
     valid = verifier.verify(message, signature)
 
 
-To sign messages use :class:`Signer` with a private key::
+To sign messages use :class:`RSASigner` with a private key::
 
     private_key = open('private_key.pem').read()
-    signer = crypt.Signer(private_key)
+    signer = crypt.RSASigner(private_key)
     signature = signer.sign(message)
 
 """
+import abc
 import io
 import json
 
@@ -77,23 +78,17 @@ def _bit_list_to_bytes(bit_list):
     byte_vals = bytearray()
     for start in six.moves.xrange(0, num_bits, 8):
         curr_bits = bit_list[start:start + 8]
-        char_val = sum(val * digit
-                       for val, digit in six.moves.zip(_POW2, curr_bits))
+        char_val = sum(
+            val * digit for val, digit in six.moves.zip(_POW2, curr_bits))
         byte_vals.append(char_val)
     return bytes(byte_vals)
 
 
+@six.add_metaclass(abc.ABCMeta)
 class Verifier(object):
-    """This object is used to verify cryptographic signatures.
-
-    Args:
-        public_key (rsa.key.PublicKey): The public key used to verify
-            signatures.
-    """
-
-    def __init__(self, public_key):
-        self._pubkey = public_key
+    """Abstract base class for crytographic signature verifiers."""
 
+    @abc.abstractmethod
     def verify(self, message, signature):
         """Verifies a message against a cryptographic signature.
 
@@ -105,6 +100,24 @@ def verify(self, message, signature):
             bool: True if message was signed by the private key associated
             with the public key that this object was constructed with.
         """
+        # pylint: disable=missing-raises-doc,redundant-returns-doc
+        # (pylint doesn't recognize that this is abstract)
+        raise NotImplementedError('Verify must be implemented')
+
+
+class RSAVerifier(Verifier):
+    """Verifies RSA cryptographic signatures using public keys.
+
+    Args:
+        public_key (rsa.key.PublicKey): The public key used to verify
+            signatures.
+    """
+
+    def __init__(self, public_key):
+        self._pubkey = public_key
+
+    @_helpers.copy_docstring(Verifier)
+    def verify(self, message, signature):
         message = _helpers.to_bytes(message)
         try:
             return rsa.pkcs1.verify(message, signature, self._pubkey)
@@ -145,7 +158,7 @@ def from_string(cls, public_key):
 
 
 def verify_signature(message, signature, certs):
-    """Verify a cryptographic signature.
+    """Verify an RSA cryptographic signature.
 
     Checks that the provided ``signature`` was generated from ``bytes`` using
     the private key associated with the ``cert``.
@@ -163,27 +176,22 @@ def verify_signature(message, signature, certs):
         certs = [certs]
 
     for cert in certs:
-        verifier = Verifier.from_string(cert)
+        verifier = RSAVerifier.from_string(cert)
         if verifier.verify(message, signature):
             return True
     return False
 
 
+@six.add_metaclass(abc.ABCMeta)
 class Signer(object):
-    """Signs messages with a private key.
-
-    Args:
-        private_key (rsa.key.PrivateKey): The private key to sign with.
-        key_id (str): Optional key ID used to identify this private key. This
-            can be useful to associate the private key with its associated
-            public key or certificate.
-    """
+    """Abstract base class for cryptographic signers."""
 
-    def __init__(self, private_key, key_id=None):
-        self._key = private_key
-        self.key_id = key_id
+    @abc.abstractproperty
+    def key_id(self):
         """Optional[str]: The key ID used to identify this private key."""
+        raise NotImplementedError('Key id must be implemented')
 
+    @abc.abstractmethod
     def sign(self, message):
         """Signs a message.
 
@@ -193,6 +201,32 @@ def sign(self, message):
         Returns:
             bytes: The signature of the message.
         """
+        # pylint: disable=missing-raises-doc,redundant-returns-doc
+        # (pylint doesn't recognize that this is abstract)
+        raise NotImplementedError('Sign must be implemented')
+
+
+class RSASigner(Signer):
+    """Signs messages with an RSA private key.
+
+    Args:
+        private_key (rsa.key.PrivateKey): The private key to sign with.
+        key_id (str): Optional key ID used to identify this private key. This
+            can be useful to associate the private key with its associated
+            public key or certificate.
+    """
+
+    def __init__(self, private_key, key_id=None):
+        self._key = private_key
+        self._key_id = key_id
+
+    @property
+    @_helpers.copy_docstring(Signer)
+    def key_id(self):
+        return self._key_id
+
+    @_helpers.copy_docstring(Signer)
+    def sign(self, message):
         message = _helpers.to_bytes(message)
         return rsa.pkcs1.sign(message, self._key, 'SHA-256')
 

google/auth/iam.py

@@ -25,28 +25,20 @@
 from six.moves import http_client
 
 from google.auth import _helpers
+from google.auth import crypt
 from google.auth import exceptions
 
 _IAM_API_ROOT_URI = 'https://iam.googleapis.com/v1'
 _SIGN_BLOB_URI = (
     _IAM_API_ROOT_URI + '/projects/-/serviceAccounts/{}:signBlob?alt=json')
 
 
-class Signer(object):
+class Signer(crypt.Signer):
     """Signs messages using the IAM `signBlob API`_.
 
     This is useful when you need to sign bytes but do not have access to the
     credential's private key file.
 
-    .. warning::
-        The IAM API signs bytes using Google-managed keys. Because of this
-        it's possible that the key used to sign bytes will change. In some
-        cases this change can occur between successive calls to :attr:`key_id`
-        and :meth:`sign`. This could result in a signature that was signed
-        with a different key than the one indicated by :attr:`key_id`. It's
-        recommended that if you use this in your code that you account for
-        this behavior by building in retry logic.
-
     .. _signBlob API:
         https://cloud.google.com/iam/reference/rest/v1/projects.serviceAccounts
         /signBlob
@@ -98,20 +90,13 @@ def _make_signing_request(self, message):
     def key_id(self):
         """Optional[str]: The key ID used to identify this private key.
 
-        .. note::
-            This makes an API request to the IAM API.
+        .. warning::
+           This is always ``None``. The key ID used by IAM can not
+           be reliably determined ahead of time.
         """
-        response = self._make_signing_request('')
-        return response['keyId']
+        return None
 
+    @_helpers.copy_docstring(crypt.Signer)
     def sign(self, message):
-        """Signs a message.
-
-        Args:
-            message (Union[str, bytes]): The message to be signed.
-
-        Returns:
-            bytes: The signature of the message.
-        """
         response = self._make_signing_request(message)
         return base64.b64decode(response['signature'])

tests/oauth2/test_service_account.py

@@ -44,7 +44,7 @@
 
 @pytest.fixture(scope='module')
 def signer():
-    return crypt.Signer.from_string(PRIVATE_KEY_BYTES, '1')
+    return crypt.RSASigner.from_string(PRIVATE_KEY_BYTES, '1')
 
 
 class TestCredentials(object):

tests/test__service_account_info.py

@@ -31,7 +31,7 @@
 
 def test_from_dict():
     signer = _service_account_info.from_dict(SERVICE_ACCOUNT_INFO)
-    assert isinstance(signer, crypt.Signer)
+    assert isinstance(signer, crypt.RSASigner)
     assert signer.key_id == SERVICE_ACCOUNT_INFO['private_key_id']
 
 
@@ -59,5 +59,5 @@ def test_from_filename():
     for key, value in six.iteritems(SERVICE_ACCOUNT_INFO):
         assert info[key] == value
 
-    assert isinstance(signer, crypt.Signer)
+    assert isinstance(signer, crypt.RSASigner)
     assert signer.key_id == SERVICE_ACCOUNT_INFO['private_key_id']

tests/test_app_engine.py

@@ -48,7 +48,7 @@ def test_key_id(self, app_identity_mock):
 
         signer = app_engine.Signer()
 
-        assert signer.key_id == mock.sentinel.key_id
+        assert signer.key_id is None
 
     def test_sign(self, app_identity_mock):
         app_identity_mock.sign_blob.return_value = (