Invalidate the session when the delegated credential has expired

See gssapi/mod_auth_gssapi#316

Signed-off-by: Aurélien Bompard <[email protected]>

README.md

@@ -5,3 +5,9 @@ A Flask extention to make use of the authentication provided by the
 [mod_auth_gssapi](https://github.com/gssapi/mod_auth_gssapi) extention of
 Apache's HTTPd. See [FASJSON](https://github.com/fedora-infra/fasjson) for a
 usage example.
+
+If you're using sessions from `mod_session` with `mod_auth_gssapi`, set your
+application's `MOD_AUTH_GSSAPI_SESSION_HEADER` configuration variable to the
+value you used in Apache's configuration file for `SessionHeader`. This will
+signal `mod_session` to invalidate the session when the authentication
+credential has expired.

config/httpd.conf

@@ -21,7 +21,7 @@ WSGIScriptReloading Off
   GssapiUseSessions On
   Session On
   SessionCookieName foobar_session path=/foobar;httponly;secure;
-  SessionHeader FOOBAR_SESSION
+  SessionHeader X-Replace-Session
   GssapiSessionKey file:/run/foobar/session.key
 
   GssapiImpersonate On

flask_mod_auth_gssapi/ext.py

@@ -1,7 +1,10 @@
+import logging
 import os
 
 import gssapi
-from flask import abort, g, request
+from flask import abort, current_app, g, redirect, request
+
+_log = logging.getLogger(__name__)
 
 
 class FlaskModAuthGSSAPI:
@@ -12,6 +15,7 @@ def __init__(self, app=None, abort=abort):
 
     def init_app(self, app):
         app.before_request(self._gssapi_check)
+        app.config.setdefault("MOD_AUTH_GSSAPI_SESSION_HEADER", "X-Replace-Session")
 
     def _gssapi_check(self):
         g.gss_name = g.gss_creds = g.principal = g.username = None
@@ -34,6 +38,11 @@ def _gssapi_check(self):
         if not principal:
             return  # Maybe the endpoint is not protected, stop here
 
+        ccache_type, _sep, ccache_location = ccache.partition(":")
+        if ccache_type == "FILE" and not os.path.exists(ccache_location):
+            _log.warning("Delegated credentials not found: %r", ccache_location)
+            return self._clear_session()
+
         gss_name = gssapi.Name(principal, gssapi.NameType.kerberos_principal)
         try:
             creds = gssapi.Credentials(
@@ -49,9 +58,20 @@ def _gssapi_check(self):
         except gssapi.exceptions.ExpiredCredentialsError:
             lifetime = 0
         if lifetime <= 0:
-            self.abort(401, "Credential lifetime has expired")
+            _log.info("Credential lifetime has expired.")
+            return self._clear_session()
 
         g.gss_name = gss_name
         g.gss_creds = creds
         g.principal = gss_name.display_as(gssapi.NameType.kerberos_principal)
         g.username = g.principal.split("@")[0]
+
+    def _clear_session(self):
+        """Unset mod_auth_gssapi's session cookie and redirect to the same URL"""
+        if request.method in ("POST", "PUT", "DELETE"):
+            self.abort(
+                401, "Re-authentication is necessary, please try your request again."
+            )
+        response = redirect(request.url)
+        response.headers[current_app.config["MOD_AUTH_GSSAPI_SESSION_HEADER"]] = "MagBearerToken="
+        return response

tests/test_ext.py

@@ -1,3 +1,4 @@
+import logging
 from types import SimpleNamespace
 
 import pytest
@@ -44,18 +45,29 @@ def test_no_cache(app, wsgi_env):
             assert g.username is None
 
 
-def test_expired(app, wsgi_env, mocker):
+def test_expired(app, wsgi_env, caplog, mocker):
     creds_factory = mocker.patch("gssapi.Credentials")
     creds_factory.return_value = SimpleNamespace(lifetime=0)
-    with app.test_request_context("/", environ_base=wsgi_env):
+    caplog.set_level(logging.INFO)
+    client = app.test_client()
+    response = client.get("/someplace", environ_base=wsgi_env)
+    assert response.status_code == 302
+    assert response.headers["location"] == "http://localhost/someplace"
+    assert caplog.messages == ["Credential lifetime has expired."]
+
+
+def test_expired_unsafe_method(app, wsgi_env, mocker):
+    creds_factory = mocker.patch("gssapi.Credentials")
+    creds_factory.return_value = SimpleNamespace(lifetime=0)
+    with app.test_request_context("/someplace", method="POST", environ_base=wsgi_env):
         with pytest.raises(Unauthorized) as excinfo:
             app.preprocess_request()
             assert g.principal is None
             assert g.username is None
-    assert excinfo.value.description == "Credential lifetime has expired"
+    assert excinfo.value.description == "Re-authentication is necessary, please try your request again."
 
 
-def test_expired_exception(app, wsgi_env, mocker):
+def test_expired_exception(app, wsgi_env, mocker, caplog):
     creds_factory = mocker.patch("gssapi.Credentials")
 
     class MockedCred:
@@ -64,15 +76,15 @@ def lifetime(self):
             raise ExpiredCredentialsError(720896, 100001)
 
     creds_factory.return_value = MockedCred()
-    with app.test_request_context("/", environ_base=wsgi_env):
-        with pytest.raises(Unauthorized) as excinfo:
-            try:
-                app.preprocess_request()
-            except ExpiredCredentialsError:
-                pytest.fail("Did not catch ExpiredCredentialsError on cred.lifetime")
-            assert g.principal is None
-            assert g.username is None
-    assert excinfo.value.description == "Credential lifetime has expired"
+    caplog.set_level(logging.INFO)
+    client = app.test_client()
+    try:
+        response = client.get("/someplace", environ_base=wsgi_env)
+    except ExpiredCredentialsError:
+        pytest.fail("Did not catch ExpiredCredentialsError on cred.lifetime")
+    assert response.status_code == 302
+    assert response.headers["location"] == "http://localhost/someplace"
+    assert caplog.messages == ["Credential lifetime has expired."]
 
 
 def test_nominal(app, wsgi_env, mocker):
@@ -100,3 +112,13 @@ def test_alt_abort(app, wsgi_env, mocker):
     call_args = mock_abort.call_args_list[0][0]
     assert call_args[0] == 403
     assert call_args[1].startswith("Invalid credentials ")
+
+
+def test_ccache_not_found(app, wsgi_env, caplog, mocker):
+    wsgi_env["KRB5CCNAME"] = "FILE:/tmp/does-not-exist"
+    #caplog.set_level(logging.INFO)
+    client = app.test_client()
+    response = client.get("/someplace", environ_base=wsgi_env)
+    assert response.status_code == 302
+    assert response.headers["location"] == "http://localhost/someplace"
+    assert caplog.messages == ["Delegated credentials not found: '/tmp/does-not-exist'"]