Merge branch 'assertion' into dev

Author: rayluo

oauth2cli/assertion.py

@@ -0,0 +1,70 @@
+import time
+import binascii
+import base64
+import uuid
+import logging
+
+import jwt
+
+
+logger = logging.getLogger(__file__)
+
+class Signer(object):
+    def sign_assertion(
+            self, audience, issuer, subject, expires_at,
+            issued_at=None, assertion_id=None, **kwargs):
+        # Names are defined in https://tools.ietf.org/html/rfc7521#section-5
+        raise NotImplementedError("Will be implemented by sub-class")
+
+
+class JwtSigner(Signer):
+    def __init__(self, key, algorithm, sha1_thumbprint=None, headers=None):
+        """Create a signer.
+
+        Args:
+
+            key (str): The key for signing, e.g. a base64 encoded private key.
+            algorithm (str):
+                "RS256", etc.. See https://pyjwt.readthedocs.io/en/latest/algorithms.html
+                RSA and ECDSA algorithms require "pip install cryptography".
+            sha1_thumbprint (str): The x5t aka X.509 certificate SHA-1 thumbprint.
+            headers (dict): Additional headers, e.g. "kid" or "x5c" etc.
+        """
+        self.key = key
+        self.algorithm = algorithm
+        self.headers = headers or {}
+        if sha1_thumbprint:  # https://tools.ietf.org/html/rfc7515#section-4.1.7
+            self.headers["x5t"] = base64.urlsafe_b64encode(
+                binascii.a2b_hex(sha1_thumbprint)).decode()
+
+    def sign_assertion(
+            self, audience, issuer, subject=None, expires_at=None,
+            issued_at=None, assertion_id=None, not_before=None,
+            additional_claims=None, **kwargs):
+        """Sign a JWT Assertion.
+
+        Parameters are defined in https://tools.ietf.org/html/rfc7523#section-3
+        Key-value pairs in additional_claims will be added into payload as-is.
+        """
+        now = time.time()
+        payload = {
+            'aud': audience,
+            'iss': issuer,
+            'sub': subject or issuer,
+            'exp': expires_at or (now + 10*60),  # 10 minutes
+            'iat': issued_at or now,
+            'jti': assertion_id or str(uuid.uuid4()),
+            }
+        if not_before:
+            payload['nbf'] = not_before
+        payload.update(additional_claims or {})
+        try:
+            return jwt.encode(
+                payload, self.key, algorithm=self.algorithm, headers=self.headers)
+        except:
+            if self.algorithm.startswith("RS") or self.algorithm.starswith("ES"):
+                logger.exception(
+                    'Some algorithms requires "pip install cryptography". '
+                    'See https://pyjwt.readthedocs.io/en/latest/installation.html#cryptographic-dependencies-optional')
+            raise
+

oauth2cli/oauth2.py

@@ -22,6 +22,8 @@ def __init__(
             self,
             client_id,
             client_secret=None,  # Triggers HTTP AUTH for Confidential Client
+            client_assertion=None,  # Assertion for Client Authentication
+            client_assertion_type=None,  # The format of the client_assertion
             default_body=None,  # a dict to be sent in each token request,
                 # usually contains Confidential Client authentication parameters
                 # such as {'client_id': 'your_id', 'client_secret': 'secret'}
@@ -31,6 +33,13 @@ def __init__(
         """Initialize a client object to talk all the OAuth2 grants to the server.
 
         Args:
+            client_assertion (str):
+                The client assertion to authenticate this client, per RFC 7521.
+            client_assertion_type (str):
+                If you leave it as the default None, this method will try to make
+                a guess between SAML2 (RFC 7522) and JWT (RFC 7523),
+                the only two profiles defined in RFC 7521.
+                But you can also explicitly provide a value, if needed.
             configuration (dict):
                 It contains the configuration (i.e. metadata) of the auth server.
                 The actual content typically contains keys like
@@ -44,6 +53,13 @@ def __init__(
         self.client_id = client_id
         self.client_secret = client_secret
         self.default_body = default_body or {}
+        if client_assertion is not None:  # See https://tools.ietf.org/html/rfc7521#section-4.2
+            TYPE_JWT = "urn:ietf:params:oauth:client-assertion-type:jwt-bearer"
+            TYPE_SAML2 = "urn:ietf:params:oauth:client-assertion-type:saml2-bearer"
+            if client_assertion_type is None:  # RFC7521 defines only 2 profiles
+                client_assertion_type = TYPE_JWT if "." in client_assertion else TYPE_SAML2
+            self.default_body["client_assertion"] = client_assertion
+            self.default_body["client_assertion_type"] = client_assertion_type
         self.configuration = configuration or {}
         self.logger = logging.getLogger(__name__)
 
@@ -132,6 +148,8 @@ class Client(BaseClient):  # We choose to implement all 4 grants in 1 class
         "DEVICE_CODE": "device_code",
         }
     DEVICE_FLOW_RETRIABLE_ERRORS = ("authorization_pending", "slow_down")
+    GRANT_TYPE_SAML2 = "urn:ietf:params:oauth:grant-type:saml2-bearer"  # RFC7522
+    GRANT_TYPE_JWT = "urn:ietf:params:oauth:grant-type:jwt-bearer"  # RFC7523
 
     def initiate_device_flow(self, scope=None, **kwargs):
         # type: (list, **dict) -> dict
@@ -348,3 +366,23 @@ def obtain_token_with_refresh_token(self, token_item, scope=None,
             return resp
         raise ValueError("token_item should not be a type %s" % type(token_item))
 
+    def obtain_token_with_assertion(
+            self, assertion, grant_type=None, scope=None, **kwargs):
+        # type: (str, Union[str, None], Union[str, list, set, tuple]) -> dict
+        """This method implements Assertion Framework for OAuth2 (RFC 7521).
+        See details at https://tools.ietf.org/html/rfc7521#section-4.1
+
+        :param assertion: The assertion string which will be sent on wire as-is
+        :param grant_type:
+            If you leave it as the default None, this method will try to make
+            a guess between SAML2 (RFC 7522) and JWT (RFC 7523),
+            the only two profiles defined in RFC 7521.
+            But you can also explicitly provide a value, if needed.
+        :param scope: Optional. It must be a subset of previously granted scopes.
+        """
+        if grant_type is None:
+            grant_type = self.GRANT_TYPE_JWT if "." in assertion else self.GRANT_TYPE_SAML2
+        data = kwargs.pop("data", {})
+        data.update(scope=scope, assertion=assertion)
+        return self._obtain_token(grant_type, data=data, **kwargs)
+

setup.py

@@ -31,5 +31,6 @@
     packages=['oauth2cli'],
     install_requires=[
         'requests>=2.0.0',
+        'PyJWT>=1.0.0',
     ]
 )

tests/test_client.py

@@ -11,6 +11,7 @@
 
 from oauth2cli.oauth2 import Client
 from oauth2cli.authcode import obtain_auth_code
+from oauth2cli.assertion import JwtSigner
 from tests import unittest
 
 
@@ -90,10 +91,26 @@ class TestClient(Oauth2TestCase):
 
     @classmethod
     def setUpClass(cls):
-        cls.client = Client(
-            CONFIG['client_id'],
-            client_secret=CONFIG.get('client_secret'),
-            configuration=CONFIG["openid_configuration"])
+        if "client_certificate" in CONFIG:
+            private_key_path = CONFIG["client_certificate"]["private_key_path"]
+            with open(os.path.join(THIS_FOLDER, private_key_path)) as f:
+                private_key = f.read()  # Expecting PEM format
+            cls.client = Client(
+                CONFIG['client_id'],
+                client_assertion=JwtSigner(
+                        private_key,
+                        algorithm="RS256",
+                        sha1_thumbprint=CONFIG["client_certificate"]["thumbprint"]
+                    ).sign_assertion(
+                        audience=CONFIG["openid_configuration"]["token_endpoint"],
+                        issuer=CONFIG["client_id"],
+                    ),
+                configuration=CONFIG["openid_configuration"])
+        else:
+            cls.client = Client(
+                CONFIG['client_id'],
+                client_secret=CONFIG.get('client_secret'),
+                configuration=CONFIG["openid_configuration"])
 
     @unittest.skipUnless("client_secret" in CONFIG, "client_secret missing")
     def test_client_credentials(self):