Merge pull request #26 from IdentityPython/iat_exp

Iat exp

Author: rohe

Makefile

@@ -40,10 +40,10 @@ test:
 .PHONY: test
 
 isort:
-	@pipenv run isort --recursive $(OICDIR) $(TESTDIR)
+	@pipenv run isort $(OICDIR) $(TESTDIR)
 
 check-isort:
-	@pipenv run isort --recursive --diff --check-only $(OICDIR) $(TESTDIR)
+	@pipenv run isort --diff --check-only $(OICDIR) $(TESTDIR)
 .PHONY: isort check-isort
 
 check-pylama:

src/oidcmsg/__init__.py

@@ -1,6 +1,8 @@
 __author__ = 'Roland Hedberg'
 __version__ = '1.1.3'
 
+import os
+
 VERIFIED_CLAIM_PREFIX = '__verified'
 
 

src/oidcmsg/oidc/__init__.py

@@ -824,6 +824,11 @@ def verify(self, **kwargs):
         else:
             if (_iat + _storage_time) < (_now - _skew):
                 raise IATError('Issued too long ago')
+            elif _iat > _now + _skew:
+                raise IATError('Issued sometime in the future')
+
+        if _exp < _iat:
+            raise IATError('Expiration time can not be earlier the issued at')
 
         if 'nonce' in kwargs and 'nonce' in self:
             if kwargs['nonce'] != self['nonce']:

src/oidcmsg/storage/abfile.py

@@ -87,14 +87,16 @@ def __getitem__(self, item):
         :return:
         """
         item = self.key_conv.serialize(item)
-
-        if self.is_changed(item):
-            logger.info("File content change in {}".format(item))
-            fname = os.path.join(self.fdir, item)
-            self.storage[item] = self._read_info(fname)
-
-        logger.debug('Read from "%s"', item)
-        return self.storage[item]
+        if self._is_file(item):
+            if self.is_changed(item):
+                logger.info("File content change in {}".format(item))
+                fname = os.path.join(self.fdir, item)
+                self.storage[item] = self._read_info(fname)
+
+            logger.debug('Read from "%s"', item)
+            return self.storage[item]
+        else:
+            raise KeyError(item)
 
     def __setitem__(self, key, value):
         """
@@ -163,31 +165,32 @@ def get_mtime(fname):
 
         return mtime
 
+    def _is_file(self, item):
+        fname = os.path.join(self.fdir, item)
+        return os.path.isfile(fname)
+
     def is_changed(self, item):
         """
-        Find out if this item has been modified since last
+        Find out if this item has been modified since last.
+        When I get here I know that item points to an existing file.
 
         :param item: A key
         :return: True/False
         """
         fname = os.path.join(self.fdir, item)
-        if os.path.isfile(fname):
-            mtime = self.get_mtime(fname)
+        mtime = self.get_mtime(fname)
 
-            try:
-                _ftime = self.fmtime[item]
-            except KeyError:  # Never been seen before
-                self.fmtime[item] = mtime
-                return True
-
-            if mtime > _ftime:  # has changed
-                self.fmtime[item] = mtime
-                return True
-            else:
-                return False
+        try:
+            _ftime = self.fmtime[item]
+        except KeyError:  # Never been seen before
+            self.fmtime[item] = mtime
+            return True
+
+        if mtime > _ftime:  # has changed
+            self.fmtime[item] = mtime
+            return True
         else:
-            logger.error('Could not access {}'.format(fname))
-            raise KeyError(item)
+            return False
 
     def _read_info(self, fname):
         if os.path.isfile(fname):

tests/test_06_oidc.py

@@ -35,6 +35,8 @@
 from oidcmsg.oidc import CHashError
 from oidcmsg.oidc import Claims
 from oidcmsg.oidc import DiscoveryRequest
+from oidcmsg.oidc import EXPError
+from oidcmsg.oidc import IATError
 from oidcmsg.oidc import IdToken
 from oidcmsg.oidc import Link
 from oidcmsg.oidc import OpenIDSchema
@@ -929,6 +931,68 @@ def test_id_token():
     idt.verify()
 
 
+def test_id_token_expired():
+    _now = time_util.utc_time_sans_frac()
+
+    idt = IdToken(**{
+        "sub": "553df2bcf909104751cfd8b2",
+        "aud": [
+            "5542958437706128204e0000",
+            "554295ce3770612820620000"
+        ],
+        "auth_time": 1441364872,
+        "azp": "554295ce3770612820620000",
+        "at_hash": "L4Ign7TCAD_EppRbHAuCyw",
+        "iat": _now - 200,
+        "exp": _now - 100,
+        "iss": "https://sso.qa.7pass.ctf.prosiebensat1.com"
+    })
+
+    with pytest.raises(EXPError):
+        idt.verify()
+
+
+def test_id_token_iat_in_the_future():
+    _now = time_util.utc_time_sans_frac()
+
+    idt = IdToken(**{
+        "sub": "553df2bcf909104751cfd8b2",
+        "aud": [
+            "5542958437706128204e0000",
+            "554295ce3770612820620000"
+        ],
+        "auth_time": 1441364872,
+        "azp": "554295ce3770612820620000",
+        "at_hash": "L4Ign7TCAD_EppRbHAuCyw",
+        "iat": _now + 600,
+        "exp": _now + 1200,
+        "iss": "https://sso.qa.7pass.ctf.prosiebensat1.com"
+    })
+
+    with pytest.raises(IATError):
+        idt.verify()
+
+
+def test_id_token_exp_before_iat():
+    _now = time_util.utc_time_sans_frac()
+
+    idt = IdToken(**{
+        "sub": "553df2bcf909104751cfd8b2",
+        "aud": [
+            "5542958437706128204e0000",
+            "554295ce3770612820620000"
+        ],
+        "auth_time": 1441364872,
+        "azp": "554295ce3770612820620000",
+        "at_hash": "L4Ign7TCAD_EppRbHAuCyw",
+        "iat": _now + 50,
+        "exp": _now,
+        "iss": "https://sso.qa.7pass.ctf.prosiebensat1.com"
+    })
+
+    with pytest.raises(IATError):
+        idt.verify(skew=100)
+
 class TestAccessTokenRequest(object):
     def test_example(self):
         _txt = 'grant_type=authorization_code&code=SplxlOBeZQQYbYS6WxSbIA' \

tox.ini

@@ -4,7 +4,7 @@ envlist = py{36,37,38},docs,quality
 [testenv]
 passenv = CI TRAVIS TRAVIS_*
 commands =
-    py.test --cov-report= --cov=oicmsg tests/ -m "not network" {posargs}
+    py.test --cov-report= --cov=oidcmsg tests/ -m "not network" {posargs}
     codecov
 extras = testing
 deps =
@@ -20,7 +20,7 @@ commands = sphinx-build -b html doc/ doc/_build/html -W
 ignore_errors = True
 extras = quality
 commands =
-    isort --recursive --diff --check-only src/ tests/
+    isort --diff --check-only src/ tests/
     pylama src/ tests/
 
 [pep8]