chore: coverage of user_authn.user increased

Author: peppelinux

src/oidcop/user_authn/user.py

@@ -169,12 +169,14 @@ def __call__(self, **kwargs):
             ),
             OnlyForTestingWarning,
         )
-
+        if not self.server_get:
+            raise Exception(
+                f"{self.__class__.__name__} doesn't have a working server_get"
+            )
         _context = self.server_get("endpoint_context")
         # Stores information need afterwards in a signed JWT that then
         # appears as a hidden input in the form
         jws = create_signed_jwt(_context.issuer, _context.keyjar, **kwargs)
-
         _kwargs = self.kwargs.copy()
         for attr in ["policy", "tos", "logo"]:
             _uri = "{}_uri".format(attr)
@@ -220,8 +222,8 @@ def authenticated_as(self, client_id, cookie=None, authorization="", **kwargs):
             authorization = authorization[6:]
 
         (user, pwd) = base64.b64decode(authorization).split(b":")
-        user = unquote(user)
-        self.verify_password(user, pwd)
+        user = unquote(user.decode())
+        self.verify_password(user, pwd.decode())
         res = {"uid": user}
         if cookie:
             res.update(self.cookie_info(cookie, client_id))
@@ -237,7 +239,7 @@ def __init__(self, ttl, symkey, server_get=None):
         if symkey is not None and symkey == "":
             msg = "SymKeyAuthn.symkey cannot be an empty value"
             raise ImproperlyConfigured(msg)
-        self.symkey = symkey
+        self.symkey = symkey.encode() if isinstance(symkey, str) else symkey
         self.ttl = ttl
 
     def authenticated_as(self, client_id, cookie=None, authorization="", **kwargs):
@@ -252,7 +254,7 @@ def authenticated_as(self, client_id, cookie=None, authorization="", **kwargs):
         try:
             aesgcm = AESGCM(self.symkey)
             user = aesgcm.decrypt(iv, encmsg, None)
-        except (AssertionError, KeyError):
+        except (AssertionError, KeyError): # pragma: no-cover
             raise FailedAuthentication("Decryption failed")
 
         res = {"uid": user}
@@ -278,8 +280,7 @@ def authenticated_as(self, client_id="", cookie=None, authorization="", **kwargs
         :param kwargs: extra key word arguments
         :return:
         """
-
-        if self.fail:
+        if self.fail: # pragma: no-cover
             raise self.fail()
 
         res = {"uid": self.user}

tests/test_12_user_authn.py

@@ -1,3 +1,4 @@
+import base64
 import os
 
 import pytest
@@ -7,8 +8,12 @@
 from oidcop.user_authn.authn_context import UNSPECIFIED
 from oidcop.user_authn.user import NoAuthn
 from oidcop.user_authn.user import UserPassJinja2
+from oidcop.user_authn.user import BasicAuthn
+from oidcop.user_authn.user import NoAuthn
+from oidcop.user_authn.user import SymKeyAuthn
 from oidcop.util import JSONDictDB
 
+
 KEYDEFS = [
     {"type": "RSA", "key": "", "use": ["sig"]},
     {"type": "EC", "crv": "P-256", "use": ["sig"]},
@@ -66,7 +71,7 @@ def create_endpoint_context(self):
                     },
                 },
             },
-            "template_dir": "template",
+            "template_dir": "tests/templates",
         }
         server = Server(conf)
         self.endpoint_context = server.endpoint_context
@@ -96,3 +101,32 @@ def test_authenticated_as_with_cookie(self):
         _info, _time_stamp = method.authenticated_as("client 12345", [_cookie])
         assert set(_info.keys()) == {"sub", "sid", "state", "client_id"}
         assert _info["sub"] == "diana"
+
+    def test_userpassjinja2(self):
+        db = {
+                "class": JSONDictDB,
+                "kwargs": {"filename": full_path("passwd.json")},
+        }
+        template_handler = self.endpoint_context.template_handler
+        sg = self.endpoint_context.session_manager.token_handler.handler['access_token'].kwargs['server_get']
+        res = UserPassJinja2(db, template_handler, server_get=sg)
+        res()
+        assert 'page_header' in res.kwargs
+
+
+    def test_basic_auth(self):
+        sg = self.endpoint_context.session_manager.token_handler.handler['access_token'].kwargs['server_get']
+        basic_auth = base64.b64encode(b'diana:krall').decode()
+        ba = BasicAuthn(pwd={'diana': 'krall'},
+                        server_get=sg)
+        ba.authenticated_as(
+            client_id='', authorization=f"Basic {basic_auth}"
+        )
+
+    def test_no_auth(self):
+        sg = self.endpoint_context.session_manager.token_handler.handler['access_token'].kwargs['server_get']
+        basic_auth = base64.b64encode(
+            b'D\xfd\x8a\x85\xa6\xd1\x16\xe4\\6\x1e\x9ds~\xc3\t\x95\x99\x83\x91\x1f\xfb:iviviviv'
+        )
+        ba = SymKeyAuthn(symkey=b'0'*32, ttl=600, server_get=sg)
+        ba.authenticated_as(client_id='', authorization=basic_auth)