Auto-update expired tokens if possible

docs/changelog.rst

@@ -2,6 +2,15 @@
 Changelog
 =========
 
+UNRELEASED
+==========
+
+Added
+-----
+
+- Auto-renew tokens when they have expired (if possible) (:issue:`19`)
+
+
 2.0.2 (2023-08-23)
 ==================
 

flask_oidc/__init__.py

@@ -25,25 +25,18 @@
 
 import json
 import logging
-import time
 import warnings
 from functools import wraps
 from urllib.parse import quote_plus
 
+from authlib.common.errors import AuthlibBaseError
 from authlib.integrations.flask_client import OAuth
 from authlib.integrations.flask_oauth2 import ResourceProtector
+from authlib.oauth2.rfc6749 import OAuth2Token
 from authlib.oauth2.rfc7662 import (
     IntrospectTokenValidator as BaseIntrospectTokenValidator,
 )
-from flask import (
-    abort,
-    current_app,
-    g,
-    redirect,
-    request,
-    session,
-    url_for,
-)
+from flask import abort, current_app, g, redirect, request, session, url_for
 
 from .views import auth_routes, legacy_oidc_callback
 
@@ -167,6 +160,7 @@ def init_app(self, app, prefix=None):
                     "OIDC_INTROSPECTION_AUTH_METHOD"
                 ],
             },
+            update_token=self._update_token,
         )
 
         if not app.config["OIDC_RESOURCE_SERVER_ONLY"]:
@@ -195,15 +189,24 @@ def check_token_expiry(self):
                 return
             if request.path == url_for("oidc_auth.logout"):
                 return  # Avoid redirect loop
-            clock_skew = current_app.config["OIDC_CLOCK_SKEW"]
-            if token["expires_at"] - clock_skew < int(time.time()):
+            token = OAuth2Token.from_dict(token)
+            try:
+                self.ensure_active_token(token)
+            except AuthlibBaseError as e:
+                logger.info(f"Could not refresh token {token!r}: {e}")
                 return redirect("{}?reason=expired".format(url_for("oidc_auth.logout")))
         except Exception as e:
-            session.pop("oidc_auth_token", None)
-            session.pop("oidc_auth_profile", None)
             logger.exception("Could not check token expiration")
             abort(500, f"{e.__class__.__name__}: {e}")
 
+    def ensure_active_token(self, token: OAuth2Token):
+        metadata = self.oauth.oidc.load_server_metadata()
+        with self.oauth.oidc._get_oauth_client(**metadata) as session:
+            return session.ensure_active_token(token)
+
+    def _update_token(name, token, refresh_token=None, access_token=None):
+        session["oidc_auth_token"] = g.oidc_id_token = token
+
     @property
     def user_loggedin(self):
         """

tests/test_flask_oidc.py

@@ -7,7 +7,8 @@
 import pytest
 import responses
 from authlib.common.urls import url_decode
-from packaging.version import Version, parse as parse_version
+from packaging.version import Version
+from packaging.version import parse as parse_version
 from werkzeug.exceptions import Unauthorized
 
 from flask_oidc import OpenIDConnect
@@ -81,15 +82,57 @@ def test_ext_logout(test_app, client, dummy_token):
     assert resp.location == expected
 
 
-def test_expired_token(client, dummy_token):
+def test_expired_token(client, dummy_token, mocked_responses):
+    new_token = dummy_token.copy()
+    new_token["access_token"] = "this-is-new"
+    refresh_call = mocked_responses.post("https://test/openidc/Token", json=new_token)
+
     dummy_token["expires_at"] = int(time.time())
     with client.session_transaction() as session:
         session["oidc_auth_token"] = dummy_token
         session["oidc_auth_profile"] = {"nickname": "dummy"}
+
     resp = client.get("/")
+
+    # Make sure we called the token url properly
+    assert refresh_call.call_count == 1
+    call = mocked_responses.calls[1].request
+    body = parse_qs(call.body)
+    assert body == {
+        "grant_type": ["refresh_token"],
+        "refresh_token": ["dummy_refresh_token"],
+        "scope": ["openid profile email"],
+        "client_id": ["MyClient"],
+        "client_secret": ["MySecret"],
+    }
+    # Check that we have the new token in the session
+    assert "oidc_auth_token" in flask.session
+    assert flask.session["oidc_auth_token"]["access_token"] == "this-is-new"
+    # Make sure we went through with the page request
+    assert resp.status_code == 200
+
+
+def test_expired_token_cant_renew(client, dummy_token, mocked_responses):
+    new_token = dummy_token.copy()
+    new_token["access_token"] = "this-is-new"
+    refresh_call = mocked_responses.post(
+        "https://test/openidc/Token", json={"error": "dummy"}, status=401
+    )
+
+    dummy_token["expires_at"] = int(time.time())
+    with client.session_transaction() as session:
+        session["oidc_auth_token"] = dummy_token
+        session["oidc_auth_profile"] = {"nickname": "dummy"}
+
+    resp = client.get("/")
+
+    assert refresh_call.call_count == 1
     assert resp.status_code == 302
     assert resp.location == "/logout?reason=expired"
     resp = client.get(resp.location)
+    assert resp.status_code == 302
+    assert resp.location == "http://localhost/"
+    assert "oidc_auth_token" not in flask.session
 
 
 def test_bad_token(client):
@@ -98,9 +141,7 @@ def test_bad_token(client):
         session["oidc_auth_profile"] = {"nickname": "dummy"}
     resp = client.get("/")
     assert resp.status_code == 500
-    assert "oidc_auth_token" not in flask.session
-    assert "oidc_auth_profile" not in flask.session
-    assert "TypeError: string indices must be integers" in resp.get_data(as_text=True)
+    assert "Internal Server Error" in resp.get_data(as_text=True)
 
 
 def test_user_getinfo(test_app, client, dummy_token):