Merge branch 'expires-in' into 'main'

Ensures the token `expires_in` claim and the `access_token` `exp` claim have the same value.

See merge request yaal/canaille!83

CHANGES.rst

@@ -3,6 +3,13 @@ All notable changes to this project will be documented in this file.
 The format is based on `Keep a Changelog <https://keepachangelog.com/en/1.0.0/>`_,
 and this project adheres to `Semantic Versioning <https://semver.org/spec/v2.0.0.html>`_.
 
+Fixed
+*****
+
+- Ensures the token `expires_in` claim and the `access_token` `exp` claim
+  have the same value. :pr:`83`
+
+
 [0.0.18] - 2022-12-28
 =====================
 

canaille/oidc/oauth.py

@@ -39,6 +39,7 @@ from .models import Token
 DEFAULT_JWT_KTY = "RSA"
 DEFAULT_JWT_ALG = "RS256"
 DEFAULT_JWT_EXP = 3600
+AUTHORIZATION_CODE_LIFETIME = 84400
 
 
 def exists_nonce(nonce, req):
@@ -135,7 +136,7 @@ def save_authorization_code(code, request):
         scope=scope,
         nonce=nonce,
         issue_date=now,
-        lifetime=str(84000),
+        lifetime=str(AUTHORIZATION_CODE_LIFETIME),
         challenge=request.data.get("code_challenge"),
         challenge_method=request.data.get("code_challenge_method"),
     )
@@ -247,7 +248,7 @@ def save_token(token, request):
         type=token["token_type"],
         access_token=token["access_token"],
         issue_date=now,
-        lifetime=str(token["expires_in"]),
+        lifetime=token["expires_in"],
         scope=token["scope"],
         client=request.client.dn,
         refresh_token=token.get("refresh_token"),
@@ -391,17 +392,23 @@ class CodeChallenge(_CodeChallenge):
         return authorization_code.challenge_method
 
 
-def generate_access_token(client, grant_type, user, scope):
-    audience = [Client.get(dn).client_id for dn in client.audience]
-    return generate_id_token(
-        {}, generate_user_info(user, scope), aud=audience, **get_jwt_config(grant_type)
-    )
-
-
 authorization = AuthorizationServer()
 require_oauth = ResourceProtector()
 
 
+def generate_access_token(client, grant_type, user, scope):
+    audience = [Client.get(dn).client_id for dn in client.audience]
+    bearer_token_generator = authorization._token_generators["default"]
+    kwargs = {
+        "token": {},
+        "user_info": generate_user_info(user, scope),
+        "aud": audience,
+        **get_jwt_config(grant_type),
+    }
+    kwargs["exp"] = bearer_token_generator._get_expires_in(client, grant_type)
+    return generate_id_token(**kwargs)
+
+
 def setup_oauth(app):
     authorization.init_app(app, query_client=query_client, save_token=save_token)
 

tests/oidc/test_authorization_code_flow.py

@@ -8,6 +8,7 @@ from canaille.models import User
 from canaille.oidc.models import AuthorizationCode
 from canaille.oidc.models import Consent
 from canaille.oidc.models import Token
+from canaille.oidc.oauth import setup_oauth
 from werkzeug.security import gen_salt
 
 from . import client_credentials
@@ -79,12 +80,16 @@ def test_authorization_code_flow(
         "address",
         "phone",
     }
+    claims = jwt.decode(access_token, keypair[1])
+    assert claims["sub"] == logged_user.uid[0]
+    assert claims["name"] == logged_user.cn[0]
+    assert claims["aud"] == [client.client_id, other_client.client_id]
 
     id_token = res.json["id_token"]
     claims = jwt.decode(id_token, keypair[1])
-    assert logged_user.uid[0] == claims["sub"]
+    assert claims["sub"] == logged_user.uid[0]
-    assert logged_user.cn[0] == claims["name"]
+    assert claims["name"] == logged_user.cn[0]
-    assert [client.client_id, other_client.client_id] == claims["aud"]
+    assert claims["aud"] == [client.client_id, other_client.client_id]
 
     res = testclient.get(
         "/oauth/userinfo",
@@ -912,3 +917,109 @@ def test_refresh_token_with_invalid_user(testclient, client):
         "error_description": 'There is no "user" for this token.',
     }
     Token.get(access_token=access_token).delete()
+
+
+def test_token_default_expiration_date(testclient, logged_user, client, keypair):
+    res = testclient.get(
+        "/oauth/authorize",
+        params=dict(
+            response_type="code",
+            client_id=client.client_id,
+            scope="openid profile email groups address phone",
+            nonce="somenonce",
+        ),
+        status=200,
+    )
+
+    res = res.form.submit(name="answer", value="accept", status=302)
+    params = parse_qs(urlsplit(res.location).query)
+    code = params["code"][0]
+    authcode = AuthorizationCode.get(code=code)
+    assert authcode.lifetime == 84400
+
+    res = testclient.post(
+        "/oauth/token",
+        params=dict(
+            grant_type="authorization_code",
+            code=code,
+            scope="openid profile email groups address phone",
+            redirect_uri=client.redirect_uris[0],
+        ),
+        headers={"Authorization": f"Basic {client_credentials(client)}"},
+        status=200,
+    )
+
+    assert res.json["expires_in"] == 864000
+
+    access_token = res.json["access_token"]
+    token = Token.get(access_token=access_token)
+    assert token.lifetime == 864000
+
+    claims = jwt.decode(access_token, keypair[1])
+    assert claims["exp"] - claims["iat"] == 864000
+
+    id_token = res.json["id_token"]
+    claims = jwt.decode(id_token, keypair[1])
+    assert claims["exp"] - claims["iat"] == 3600
+
+    consents = Consent.filter(client=client.dn, subject=logged_user.dn)
+    for consent in consents:
+        consent.delete()
+
+
+def test_token_custom_expiration_date(testclient, logged_user, client, keypair):
+    testclient.app.config["OAUTH2_TOKEN_EXPIRES_IN"] = {
+        "authorization_code": 1000,
+        "implicit": 2000,
+        "password": 3000,
+        "client_credentials": 4000,
+        "urn:ietf:params:oauth:grant-type:jwt-bearer": 5000,
+    }
+    testclient.app.config["JWT"]["EXP"] = 6000
+    setup_oauth(testclient.app)
+
+    res = testclient.get(
+        "/oauth/authorize",
+        params=dict(
+            response_type="code",
+            client_id=client.client_id,
+            scope="openid profile email groups address phone",
+            nonce="somenonce",
+        ),
+        status=200,
+    )
+
+    res = res.form.submit(name="answer", value="accept", status=302)
+    params = parse_qs(urlsplit(res.location).query)
+    code = params["code"][0]
+    authcode = AuthorizationCode.get(code=code)
+    assert authcode.lifetime == 84400
+
+    res = testclient.post(
+        "/oauth/token",
+        params=dict(
+            grant_type="authorization_code",
+            code=code,
+            scope="openid profile email groups address phone",
+            redirect_uri=client.redirect_uris[0],
+        ),
+        headers={"Authorization": f"Basic {client_credentials(client)}"},
+        status=200,
+    )
+
+    assert res.json["expires_in"] == 1000
+
+    access_token = res.json["access_token"]
+    token = Token.get(access_token=access_token)
+    assert token.lifetime == 1000
+
+    claims = jwt.decode(access_token, keypair[1])
+    assert claims["exp"] - claims["iat"] == 1000
+
+    id_token = res.json["id_token"]
+    claims = jwt.decode(id_token, keypair[1])
+    assert claims["exp"] - claims["iat"] == 6000
+
+    consents = Consent.filter(client=client.dn, subject=logged_user.dn)
+    for consent in consents:
+        consent.delete()