Merge pull request #854 from gbip/853_session_state

Use session_state to backup consumer state if available

CHANGELOG.md

@@ -6,6 +6,9 @@ The format is based on the [KeepAChangeLog] project.
 [KeepAChangeLog]: https://keepachangelog.com/
 
 ## Unreleased
+
+- [#854] Improve OIDC Session Management support by using the `session_state` parameter from an *Authentication Response* (if available) as a key to store `Consumer` data.
+
 ### Changed
 - [#847] Using pydantic for settings instead of custom class
 - [#851], [#852] Add `authn_method` to `Consumer.complete`

src/oic/oic/consumer.py

@@ -461,10 +461,19 @@ def parse_authz(
             self.verify_id_token(idt, self.authz_req.get(_state or atr["state"]))
         return aresp, atr, idt
 
-    def complete(self, state, authn_method: str = "client_secret_basic"):
+    def complete(
+        self,
+        state,
+        authn_method: str = "client_secret_basic",
+        session_state: Optional[str] = None,
+    ):
         """
         Do the access token request, the last step in a code flow.
 
+        'session_state' is an optional parameter that can be provided if the Authorization Server support OIDC Session
+        Management.
+        If provided, it is used as the key in the session database to store the consumer data.
+
         If Implicit flow was used then this method is never used.
         """
         args = {"redirect_uri": self.redirect_uris[0]}
@@ -496,7 +505,12 @@ def complete(self, state, authn_method: str = "client_secret_basic"):
         if resp.type() == "ErrorResponse":
             raise TokenError(resp.error, resp)
 
-        self._backup(state)
+        if session_state is not None:
+            # Use session_state from Authorization server, as per §2
+            # from https://openid.net/specs/openid-connect-session-1_0.html
+            self._backup(session_state)
+        else:
+            self._backup(state)
 
         return resp
 

tests/test_oic_consumer.py

@@ -306,8 +306,59 @@ def test_complete(self):
             resp = self.consumer.complete(_state)
         assert isinstance(resp, AccessTokenResponse)
         assert _eq(resp.keys(), ["token_type", "state", "access_token", "scope"])
+        assert _state in self.consumer.sdb  # Consumer has been saved using 'state'
+        assert resp["state"] == _state
 
+    def test_complete_with_session_management(self):
+        _state = "state0"
+        _session_state = "o75rvk4#5#JDYD`w"
+        args = {
+            "client_id": self.consumer.client_id,
+            "response_type": "code",
+            "scope": ["openid"],
+        }
+
+        location = "https://example.com/cb?code=code&state=state0"
+        with responses.RequestsMock() as rsps:
+            rsps.add(
+                responses.GET,
+                "https://example.com/authorization",
+                status=302,
+                headers={"location": location},
+            )
+            rsps.add(
+                responses.POST,
+                "https://example.com/token",
+                content_type="application/json",
+                json={
+                    "access_token": "some_token",
+                    "token_type": "bearer",
+                    "state": "state0",
+                    "scope": "openid",
+                    "session_state": _session_state,
+                },
+            )
+            result = self.consumer.do_authorization_request(
+                state=_state, request_args=args
+            )
+            parsed = urlparse(result.headers["location"])
+
+            self.consumer.parse_response(
+                AuthorizationResponse, info=parsed.query, sformat="urlencoded"
+            )
+
+            resp = self.consumer.complete(_state, session_state=_session_state)
+
+        assert isinstance(resp, AccessTokenResponse)
+        assert _eq(
+            resp.keys(),
+            ["token_type", "state", "access_token", "scope", "session_state"],
+        )
+        assert (
+            _session_state in self.consumer.sdb
+        )  # Consumer has been saved using 'session_state'
         assert resp["state"] == _state
+        assert resp["session_state"] == _session_state
 
     def test_parse_authz(self):
         _state = "state0"