Add no-cache headers to token endpoints

Close #145

src/oic/extension/provider.py

@@ -51,6 +51,7 @@
 from oic.utils.authn.client import UnknownAuthnMethod
 from oic.utils.authn.client import get_client_id
 from oic.utils.authn.client import valid_client_info
+from oic.utils.http_util import OAUTH2_NOCACHE_HEADERS
 from oic.utils.http_util import BadRequest
 from oic.utils.http_util import Forbidden
 from oic.utils.http_util import NoContent
@@ -702,7 +703,7 @@ def code_grant_type(self, areq):
 
         logger.debug("AccessTokenResponse: %s" % atr)
 
-        return Response(atr.to_json(), content="application/json")
+        return Response(atr.to_json(), content="application/json", headers=OAUTH2_NOCACHE_HEADERS)
 
     def client_credentials_grant_type(self, areq):
         _at = self.token_handler.get_access_token(areq['client_id'],

src/oic/oauth2/provider.py

@@ -43,6 +43,7 @@
 from oic.utils.authn.user import NoSuchAuthentication
 from oic.utils.authn.user import TamperAllert
 from oic.utils.authn.user import ToOld
+from oic.utils.http_util import OAUTH2_NOCACHE_HEADERS
 from oic.utils.http_util import BadRequest
 from oic.utils.http_util import CookieDealer
 from oic.utils.http_util import Response
@@ -822,7 +823,7 @@ def token_endpoint(self, authn="", **kwargs):
 
         logger.debug("AccessTokenResponse: %s" % sanitize(atr))
 
-        return Response(atr.to_json(), content="application/json")
+        return Response(atr.to_json(), content="application/json", headers=OAUTH2_NOCACHE_HEADERS)
 
     def verify_endpoint(self, request="", cookie=None, **kwargs):
         _req = parse_qs(request)

src/oic/oic/provider.py

@@ -71,6 +71,7 @@
 from oic.oic.message import RegistrationResponse
 from oic.oic.message import TokenErrorResponse
 from oic.utils import sort_sign_alg
+from oic.utils.http_util import OAUTH2_NOCACHE_HEADERS
 from oic.utils.http_util import BadRequest
 from oic.utils.http_util import Created
 from oic.utils.http_util import Response
@@ -981,7 +982,7 @@ def _access_token_endpoint(self, req, **kwargs):
 
         logger.info("access_token_response: %s" % sanitize(atr.to_dict()))
 
-        return Response(atr.to_json(), content="application/json")
+        return Response(atr.to_json(), content="application/json", headers=OAUTH2_NOCACHE_HEADERS)
 
     def _refresh_access_token_endpoint(self, req, **kwargs):
         _sdb = self.sdb
@@ -1017,7 +1018,7 @@ def _refresh_access_token_endpoint(self, req, **kwargs):
 
         logger.info("access_token_response: %s" % sanitize(atr.to_dict()))
 
-        return Response(atr.to_json(), content="application/json")
+        return Response(atr.to_json(), content="application/json", headers=OAUTH2_NOCACHE_HEADERS)
 
     def token_endpoint(self, request="", authn=None, dtype='urlencoded',
                        **kwargs):

src/oic/utils/http_util.py

@@ -34,6 +34,11 @@
     ("Access-Control-Allow-Headers", "Authorization")
 ]
 
+OAUTH2_NOCACHE_HEADERS = [
+    ('Pragma', 'no-cache'),
+    ('Cache-Control', 'no-store'),
+]
+
 
 class Response(object):
     _template = None

tests/test_oauth2_provider.py

@@ -276,6 +276,34 @@ def test_token_endpoint(self):
                     'refresh_token': '<REDACTED>'}
         assert _eq(eval(logcap.records[6].msg[21:]), expected)
 
+    def test_token_endpoint_no_cache(self):
+        authreq = AuthorizationRequest(state="state",
+                                       redirect_uri="http://example.com/authz",
+                                       client_id="client1")
+
+        _sdb = self.provider.sdb
+        sid = _sdb.access_token.key(user="sub", areq=authreq)
+        access_grant = _sdb.access_token(sid=sid)
+        _sdb[sid] = {
+            "oauth_state": "authz",
+            "sub": "sub",
+            "authzreq": "",
+            "client_id": "client1",
+            "code": access_grant,
+            "code_used": False,
+            "redirect_uri": "http://example.com/authz"
+        }
+
+        # Construct Access token request
+        areq = AccessTokenRequest(code=access_grant,
+                                  redirect_uri="http://example.com/authz",
+                                  client_id="client1",
+                                  client_secret="hemlighet",
+                                  grant_type='authorization_code')
+        resp = self.provider.token_endpoint(request=areq.to_urlencoded())
+        assert resp.headers == [('Pragma', 'no-cache'), ('Cache-Control', 'no-store'),
+                                ('Content-type', 'application/json')]
+
     def test_token_endpoint_unauth(self):
         authreq = AuthorizationRequest(state="state",
                                        redirect_uri="http://example.com/authz",

tests/test_oic_provider.py

@@ -427,6 +427,41 @@ def test_token_endpoint(self):
         assert _eq(atr.keys(),
                    ['token_type', 'id_token', 'access_token', 'scope'])
 
+    def test_token_endpoint_no_cache(self):
+        authreq = AuthorizationRequest(state="state",
+                                       redirect_uri="http://example.com/authz",
+                                       client_id=CLIENT_ID,
+                                       response_type="code",
+                                       scope=["openid"])
+
+        _sdb = self.provider.sdb
+        sid = _sdb.access_token.key(user="sub", areq=authreq)
+        access_grant = _sdb.access_token(sid=sid)
+        ae = AuthnEvent("user", "salt")
+        _sdb[sid] = {
+            "oauth_state": "authz",
+            "authn_event": ae.to_json(),
+            "authzreq": authreq.to_json(),
+            "client_id": CLIENT_ID,
+            "code": access_grant,
+            "code_used": False,
+            "scope": ["openid"],
+            "redirect_uri": "http://example.com/authz",
+        }
+        _sdb.do_sub(sid, "client_salt")
+
+        # Construct Access token request
+        areq = AccessTokenRequest(code=access_grant, client_id=CLIENT_ID,
+                                  redirect_uri="http://example.com/authz",
+                                  client_secret=CLIENT_SECRET,
+                                  grant_type='authorization_code')
+
+        txt = areq.to_urlencoded()
+
+        resp = self.provider.token_endpoint(request=txt)
+        assert resp.headers == [('Pragma', 'no-cache'), ('Cache-Control', 'no-store'),
+                                ('Content-type', 'application/json')]
+
     def test_token_endpoint_refresh(self):
         authreq = AuthorizationRequest(state="state",
                                        redirect_uri="http://example.com/authz",
@@ -1282,3 +1317,48 @@ def test_refresh_access_token_request(self):
         assert atr2['access_token'] != atr['access_token']
         assert atr2['refresh_token'] == atr['refresh_token']
         assert atr2['token_type'] == 'Bearer'
+
+    def test_refresh_access_token_no_cache(self):
+        authreq = AuthorizationRequest(state="state",
+                                       redirect_uri="http://example.com/authz",
+                                       client_id=CLIENT_ID,
+                                       response_type="code",
+                                       scope=["openid", 'offline_access'],
+                                       prompt='consent')
+
+        _sdb = self.provider.sdb
+        sid = _sdb.access_token.key(user="sub", areq=authreq)
+        access_grant = _sdb.access_token(sid=sid)
+        ae = AuthnEvent("user", "salt")
+        _sdb[sid] = {
+            "oauth_state": "authz",
+            "authn_event": ae.to_json(),
+            "authzreq": authreq.to_json(),
+            "client_id": CLIENT_ID,
+            "code": access_grant,
+            "code_used": False,
+            "scope": ["openid", 'offline_access'],
+            "redirect_uri": "http://example.com/authz",
+        }
+        _sdb.do_sub(sid, "client_salt")
+
+        # Construct Access token request
+        areq = AccessTokenRequest(code=access_grant, client_id=CLIENT_ID,
+                                  redirect_uri="http://example.com/authz",
+                                  client_secret=CLIENT_SECRET,
+                                  grant_type='authorization_code')
+
+        txt = areq.to_urlencoded()
+
+        resp = self.provider.token_endpoint(request=txt)
+        atr = AccessTokenResponse().deserialize(resp.message, "json")
+
+        rareq = RefreshAccessTokenRequest(grant_type="refresh_token",
+                                          refresh_token=atr['refresh_token'],
+                                          client_id=CLIENT_ID,
+                                          client_secret=CLIENT_SECRET,
+                                          scope=['openid'])
+
+        resp = self.provider.token_endpoint(request=rareq.to_urlencoded())
+        assert resp.headers == [('Pragma', 'no-cache'), ('Cache-Control', 'no-store'),
+                                ('Content-type', 'application/json')]

tests/test_x_provider.py

@@ -242,6 +242,36 @@ def test_token_endpoint(self):
         atr = AccessTokenResponse().deserialize(resp.message, "json")
         assert _eq(atr.keys(), ['access_token', 'token_type'])
 
+    def test_token_endpoint_no_cache(self):
+        authreq = AuthorizationRequest(state="state",
+                                       redirect_uri="http://example.com/authz",
+                                       client_id="client1")
+
+        _sdb = self.provider.sdb
+        sid = _sdb.access_token.key(user="sub", areq=authreq)
+        access_grant = _sdb.token_factory['code'](sid=sid)
+        _sdb[sid] = {
+            "oauth_state": "authz",
+            "sub": "sub",
+            "authzreq": authreq.to_json(),
+            "client_id": "client1",
+            "code": access_grant,
+            "code_used": False,
+            "redirect_uri": "http://example.com/authz",
+            'response_type': ['code']
+        }
+
+        # Construct Access token request
+        areq = AccessTokenRequest(code=access_grant,
+                                  redirect_uri="http://example.com/authz",
+                                  client_id="client1",
+                                  client_secret="hemlighet",
+                                  grant_type='authorization_code')
+
+        resp = self.provider.token_endpoint(request=areq.to_urlencoded())
+        assert resp.headers == [('Pragma', 'no-cache'), ('Cache-Control', 'no-store'),
+                                ('Content-type', 'application/json')]
+
     def test_token_endpoint_unauth(self):
         authreq = AuthorizationRequest(state="state",
                                        redirect_uri="http://example.com/authz",