Merge pull request #17 from IdentityPython/develop

Better handling on error in authentication

Author: rohe

pyproject.toml

@@ -7,7 +7,7 @@ build-backend = "setuptools.build_meta"
 
 [metadata]
 name = "idpyoidc"
-version = "1.0.11"
+version = "1.0.12"
 author = "Roland Hedberg"
 author_email = "[email protected]"
 description = "Everything OAuth2 and OIDC"

src/idpyoidc/__init__.py

@@ -1,5 +1,5 @@
 __author__ = "Roland Hedberg"
-__version__ = "1.0.11"
+__version__ = "1.0.12"
 
 import os
 from typing import Dict

src/idpyoidc/server/__init__.py

@@ -7,7 +7,6 @@
 from cryptojwt import KeyJar
 
 from idpyoidc.impexp import ImpExp
-from idpyoidc.message.oidc import RegistrationRequest
 from idpyoidc.server import authz
 from idpyoidc.server.client_authn import client_auth_setup
 from idpyoidc.server.configure import ASConfiguration

src/idpyoidc/server/cookie_handler.py

@@ -15,7 +15,9 @@
 from cryptojwt.jws.hmac import HMACSigner
 from cryptojwt.jwt import utc_time_sans_frac
 from cryptojwt.key_jar import init_key_jar
+from cryptojwt.utils import as_bytes
 
+from idpyoidc.encrypter import init_encrypter
 from idpyoidc.server.util import lv_pack
 from idpyoidc.server.util import lv_unpack
 from idpyoidc.time_util import epoch_in_a_while
@@ -37,17 +39,25 @@ def __init__(
         keys: Optional[dict] = None,
         sign_alg: [str] = "SHA256",
         name: Optional[dict] = None,
+        crypt_config: Optional[dict] = None,
         **kwargs,
     ):
+        self.sign_key = None
+        self.enc_key = None
+        self.crypt = None
 
         if keys:
             key_jar = init_key_jar(**keys)
-            _keys = key_jar.get_signing_key(key_type="oct", kid="sig")
+            _keys = key_jar.get_signing_key(key_type="oct")
             if _keys:
                 self.sign_key = _keys[0]
-            _keys = key_jar.get_encrypt_key(key_type="oct", kid="enc")
+            _keys = key_jar.get_encrypt_key(key_type="oct")
             if _keys:
                 self.enc_key = _keys[0]
+        elif crypt_config:
+            _crypt = init_encrypter(crypt_config)
+            self.crypt = _crypt["encrypter"]
+            self.crypt_config = _crypt["conf"]
         else:
             if sign_key:
                 if isinstance(sign_key, SYMKey):
@@ -132,6 +142,11 @@ def _sign_enc_payload(self, payload: str, timestamp: Optional[Union[int, str]] =
                 base64.b64encode(ctx),
                 base64.b64encode(tag),
             ]
+        elif self.crypt:
+            msg = lv_pack(timestamp, payload)
+            cookie_payload = [
+                bytes_timestamp,
+                base64.b64encode(self.crypt.encrypt(msg.encode()))]
         else:
             cookie_payload = [bytes_timestamp, bytes_load, base64.b64encode(mac)]
 
@@ -147,6 +162,15 @@ def _ver_dec_content(self, parts):
 
         if parts is None:
             return None
+        elif len(parts) == 2:
+            t0, enc_payload = parts
+            if not self.crypt:
+                raise VerificationError("Can not decrypt")
+            msg = self.crypt.decrypt(base64.b64decode(as_bytes(enc_payload)))
+            t1, payload = lv_unpack(msg.decode("utf-8"))
+            if t0 != t1:
+                raise VerificationError('Suspicious timestamp')
+            return payload, t1
         elif len(parts) == 3:
             # verify the cookie signature
             timestamp, payload, b64_mac = parts
@@ -255,15 +279,15 @@ def parse_cookie(self, name: str, cookies: List[dict]) -> Optional[List[dict]]:
         LOGGER.debug("Looking for '{}' cookies".format(name))
         res = []
         for _cookie in cookies:
-            LOGGER.debug("Cookie: {}".format(_cookie))
+            LOGGER.debug(f"Cookie: {_cookie}")
             if "name" in _cookie and _cookie["name"] == name:
                 _content = self._ver_dec_content(_cookie["value"].split("|"))
                 if _content:
-                    payload, timestamp = self._ver_dec_content(_cookie["value"].split("|"))
+                    payload, timestamp = _content
                     value, typ = payload.split("::")
                     res.append({"value": value, "type": typ, "timestamp": timestamp})
                 else:
-                    LOGGER.debug(f"Could not verify {name} cookie")
+                    LOGGER.debug(f"Could not verify '{name}' cookie")
         return res
 
 

src/idpyoidc/server/oauth2/authorization.py

@@ -603,21 +603,22 @@ def setup_auth(
                         if _csi.is_active() is False:
                             identity = None
 
-        authn_args = authn_args_gather(request, authn_class_ref, cinfo, **kwargs)
         _mngr = _context.session_manager
         _session_id = ""
 
         # To authenticate or Not
         if not identity:  # No!
             logger.info("No active authentication")
-            logger.debug("Known clients: {}".format(list(_context.cdb.keys())))
+            # logger.debug("Known clients: {}".format(list(_context.cdb.keys())))
 
             if "prompt" in request and "none" in request["prompt"]:
                 # Need to authenticate but not allowed
                 return self._login_required_error(redirect_uri, request)
             else:
+                authn_args = authn_args_gather(request, authn_class_ref, cinfo, **kwargs)
                 return {"function": authn, "args": authn_args}
         else:
+            authn_args = authn_args_gather(request, authn_class_ref, cinfo, **kwargs)
             logger.info(f"Active authentication: {identity}")
             if re_authenticate(request, authn):
                 # demand re-authentication
@@ -994,7 +995,7 @@ def process_request(
         cinfo = _context.cdb[_cid]
         # logger.debug("client {}: {}".format(_cid, cinfo))
 
-        # this apply the default optionally deny_unknown_scopes policy
+        # this applies the default optionally deny_unknown_scopes policy
         check_unknown_scopes_policy(request, _cid, _context)
 
         if http_info is None:
@@ -1004,7 +1005,11 @@ def process_request(
         if _cookies:
             logger.debug("parse_cookie@process_request")
             _session_cookie_name = _context.cookie_handler.name["session"]
-            _my_cookies = _context.cookie_handler.parse_cookie(_session_cookie_name, _cookies)
+            try:
+                _my_cookies = _context.cookie_handler.parse_cookie(_session_cookie_name, _cookies)
+            except Exception as err:
+                logger.info(f"Parse cookie failed due to: {err}")
+                _my_cookies = {}
         else:
             _my_cookies = {}
 

src/idpyoidc/server/user_authn/user.py

@@ -13,6 +13,9 @@
 
 from idpyoidc.server.exception import FailedAuthentication
 from idpyoidc.server.exception import ImproperlyConfigured
+from idpyoidc.server.exception import InconsistentDatabase
+from idpyoidc.server.exception import NoSuchClientSession
+from idpyoidc.server.exception import NoSuchGrant
 from idpyoidc.server.exception import OnlyForTestingWarning
 from idpyoidc.time_util import utc_time_sans_frac
 from idpyoidc.util import instantiate
@@ -64,8 +67,8 @@ def authenticated_as(self, client_id, cookie=None, **kwargs):
             return None, 0
         else:
             _info = self.cookie_info(cookie, client_id)
-            logger.debug("authenticated_as: cookie info={}".format(_info))
             if _info:
+                logger.debug("authenticated_as: cookie info={}".format(_info))
                 if "max_age" in kwargs and kwargs["max_age"] != 0:
                     _max_age = kwargs["max_age"]
                     _now = utc_time_sans_frac()
@@ -74,6 +77,9 @@ def authenticated_as(self, client_id, cookie=None, **kwargs):
                             "Too old by {} seconds".format(_now - (_info["timestamp"] + _max_age))
                         )
                         return None, 0
+            else:
+                logger.info("Failed to find session based on cookie")
+
             return _info, utc_time_sans_frac()
 
     def verify(self, *args, **kwargs):
@@ -109,9 +115,18 @@ def cookie_info(self, cookie: List[dict], client_id: str) -> dict:
             for val in cookie:
                 _info = json.loads(val["value"])
                 _info["timestamp"] = int(val["timestamp"])
+
+                # verify session ID
+                try:
+                    _context.session_manager[_info["sid"]]
+                except (KeyError, ValueError, InconsistentDatabase,
+                        NoSuchClientSession, NoSuchGrant) as err:
+                    logger.info(f"Verifying session ID fail due to {err}")
+                    return {}
+
                 session_id = _context.session_manager.decrypt_session_id(_info["sid"])
                 logger.debug("cookie_info: session id={}".format(session_id))
-                # _, cid, _ = _context.session_manager.decrypt_session_id(_info["sid"])
+
                 if session_id[1] != client_id:
                     continue
                 else:
@@ -138,13 +153,13 @@ class UserPassJinja2(UserAuthnMethod):
     url_endpoint = "/verify/user_pass_jinja"
 
     def __init__(
-        self,
-        db,
-        template_handler,
-        template="user_pass.jinja2",
-        server_get=None,
-        verify_endpoint="",
-        **kwargs,
+            self,
+            db,
+            template_handler,
+            template="user_pass.jinja2",
+            server_get=None,
+            verify_endpoint="",
+            **kwargs,
     ):
 
         super(UserPassJinja2, self).__init__(server_get=server_get)

tests/private/cookie_jwks.json

@@ -1 +1 @@
-{"keys": [{"kty": "oct", "use": "enc", "kid": "enc", "k": "XlZGhRrHFOQwgIJZSV8g23cxVoL27xyv"}, {"kty": "oct", "use": "sig", "kid": "sig", "k": "TD6UZ7GmkUeC1oejdCGTf2YAp7FHeGfd"}]}
+{"keys": [{"kty": "oct", "use": "enc", "kid": "enc", "k": "G6JN24md0JKKBD16Aq02wPpvW6QrlgzI"}, {"kty": "oct", "use": "sig", "kid": "sig", "k": "8YmUB_TM3eW3-uZHUqgsuiXSkMpgboUk"}]}

tests/test_server_13_user_authn.py

@@ -4,6 +4,7 @@
 import pytest
 
 from idpyoidc.server import Server
+from idpyoidc.server.authn_event import create_authn_event
 from idpyoidc.server.configure import OPConfiguration
 from idpyoidc.server.user_authn.authn_context import INTERNETPROTOCOLPASSWORD
 from idpyoidc.server.user_authn.authn_context import UNSPECIFIED
@@ -80,6 +81,20 @@ def create_endpoint_context(self):
         }
         self.server = Server(OPConfiguration(conf=conf, base_path=BASEDIR), cwd=BASEDIR)
         self.endpoint_context = self.server.endpoint_context
+        self.session_manager = self.endpoint_context.session_manager
+        self.user_id = "diana"
+
+    def _create_session(self, auth_req, sub_type="public", sector_identifier=""):
+        if sector_identifier:
+            authz_req = auth_req.copy()
+            authz_req["sector_identifier_uri"] = sector_identifier
+        else:
+            authz_req = auth_req
+        client_id = authz_req["client_id"]
+        ae = create_authn_event(self.user_id)
+        return self.session_manager.create_session(
+            ae, authz_req, self.user_id, client_id=client_id, sub_type=sub_type
+        )
 
     def test_authenticated_as_without_cookie(self):
         authn_item = self.endpoint_context.authn_broker.pick(INTERNETPROTOCOLPASSWORD)
@@ -93,12 +108,11 @@ def test_authenticated_as_with_cookie(self):
         method = authn_item[0]["method"]
 
         authn_req = {"state": "state_identifier", "client_id": "client 12345"}
+        _sid = self._create_session(authn_req)
         _cookie = self.endpoint_context.new_cookie(
             name=self.endpoint_context.cookie_handler.name["session"],
             sub="diana",
-            sid=self.endpoint_context.session_manager.encrypted_session_id(
-                "diana", "client 12345", "abcdefgh"
-            ),
+            sid=_sid,
             state=authn_req["state"],
             client_id=authn_req["client_id"],
         )

tests/test_server_20g_cookie_handler.py

@@ -3,6 +3,7 @@
 
 from idpyoidc.server.cookie_handler import CookieHandler
 from idpyoidc.server.cookie_handler import compute_session_state
+from tests import CRYPT_CONFIG
 
 KEYDEFS = [
     {"type": "OCT", "kid": "sig", "use": ["sig"]},
@@ -233,3 +234,56 @@ def test_mult_cookie(self):
 def test_compute_session_state():
     hv = compute_session_state("state", "salt", "client_id", "https://example.com/redirect")
     assert hv == "d21113fbe4b54661ae45f3a3233b0f865ccc646af248274b6fa5664267540e29.salt"
+
+
+class TestCookieHandlerFernetEnc(object):
+    @pytest.fixture(autouse=True)
+    def make_cookie_content_handler(self):
+        cookie_conf = {
+            "crypt_config": CRYPT_CONFIG,
+        }
+
+        self.cookie_handler = CookieHandler(**cookie_conf)
+
+    def test_make_cookie_content(self):
+        _cookie_info = self.cookie_handler.make_cookie_content("idpyoidc.server", "value", "sso")
+        assert _cookie_info
+        assert set(_cookie_info.keys()) == {"name", "value", "samesite", "httponly", "secure"}
+        assert len(_cookie_info["value"].split("|")) == 2
+
+    def test_make_cookie_content_max_age(self):
+        _cookie_info = self.cookie_handler.make_cookie_content(
+            "idpyoidc.server", "value", "sso", max_age=3600
+        )
+        assert _cookie_info
+        assert set(_cookie_info.keys()) == {
+            "name",
+            "value",
+            "max-age",
+            "samesite",
+            "httponly",
+            "secure",
+        }
+        assert len(_cookie_info["value"].split("|")) == 2
+
+    def test_read_cookie_info(self):
+        _cookie_info = [self.cookie_handler.make_cookie_content("idpyoidc.server", "value", "sso")]
+        returned = [{"name": c["name"], "value": c["value"]} for c in _cookie_info]
+        _info = self.cookie_handler.parse_cookie("idpyoidc.server", returned)
+        assert len(_info) == 1
+        assert set(_info[0].keys()) == {"value", "type", "timestamp"}
+        assert _info[0]["value"] == "value"
+        assert _info[0]["type"] == "sso"
+
+    def test_mult_cookie(self):
+        _cookie = [
+            self.cookie_handler.make_cookie_content("idpyoidc.server", "value", "sso"),
+            self.cookie_handler.make_cookie_content("idpyoidc.server", "session_state", "session"),
+        ]
+        assert len(_cookie) == 2
+        _c_info = self.cookie_handler.parse_cookie("idpyoidc.server", _cookie)
+        assert len(_c_info) == 2
+        assert _c_info[0]["value"] == "value"
+        assert _c_info[0]["type"] == "sso"
+        assert _c_info[1]["value"] == "session_state"
+        assert _c_info[1]["type"] == "session"