Refactored auth feature and increased test coverage

Author: eadwinCode

ellar/auth/__init__.py

@@ -1,18 +1,27 @@
 from .decorators import CheckPolicies
 from .guard import AuthorizationGuard
 from .handlers import BaseAuthenticationHandler
-from .identity_provider import BaseIdentitySchemeProvider
-from .interfaces import IAuthConfig, IIdentitySchemeProvider
-from .policy import BasePolicyHandler, BasePolicyHandlerWithRequirement
+from .identity import UserIdentity
+from .interfaces import IIdentitySchemes
+from .policy import (
+    BasePolicyHandler,
+    BasePolicyHandlerWithRequirement,
+    RequiredClaimsPolicy,
+    RequiredRolePolicy,
+)
+from .services import AppIdentitySchemes, IdentityAuthenticationService
 
 __all__ = [
     "CheckPolicies",
-    "BaseIdentitySchemeProvider",
     "AuthorizationGuard",
     "BaseAuthenticationHandler",
     "CheckPolicies",
     "BasePolicyHandler",
     "BasePolicyHandlerWithRequirement",
-    "IAuthConfig",
-    "IIdentitySchemeProvider",
+    "IIdentitySchemes",
+    "UserIdentity",
+    "RequiredClaimsPolicy",
+    "RequiredRolePolicy",
+    "AppIdentitySchemes",
+    "IdentityAuthenticationService",
 ]

ellar/auth/config/__init__.py

@@ -1,31 +1,23 @@
-import typing as t
 from abc import ABC
 
-from .base import BaseAPIKeyAuthenticationHandler
+from .mixin import BaseAuthenticationHandlerMixin
+from .model import BaseAuthenticationHandler
+from .schemes import APIKeyCookie, APIKeyHeader, APIKeyQuery
 
-if t.TYPE_CHECKING:  # pragma: no cover
-    from ellar.core import HTTPConnection
 
+class QueryAPIKeyAuthenticationHandler(
+    BaseAuthenticationHandlerMixin, APIKeyQuery, BaseAuthenticationHandler, ABC
+):
+    scheme: str = "query"
 
-class QueryAPIKeyAuthenticationHandler(BaseAPIKeyAuthenticationHandler, ABC):
-    parameter_name: str = "key"
-    openapi_in: str = "query"
 
-    def _get_key(self, connection: "HTTPConnection") -> t.Optional[t.Any]:
-        return connection.query_params.get(self.parameter_name)
+class CookieAPIKeyAuthenticationHandler(
+    BaseAuthenticationHandlerMixin, APIKeyCookie, BaseAuthenticationHandler, ABC
+):
+    scheme: str = "cookie"
 
 
-class CookieAPIKeyAuthenticationHandler(BaseAPIKeyAuthenticationHandler, ABC):
-    parameter_name: str = "key"
-    openapi_in: str = "query"
-
-    def _get_key(self, connection: "HTTPConnection") -> t.Optional[t.Any]:
-        return connection.cookies.get(self.parameter_name)
-
-
-class HeaderAPIKeyAuthenticationHandler(BaseAPIKeyAuthenticationHandler, ABC):
-    parameter_name: str = "key"
-    openapi_in: str = "query"
-
-    def _get_key(self, connection: "HTTPConnection") -> t.Optional[t.Any]:
-        return connection.headers.get(self.parameter_name)
+class HeaderAPIKeyAuthenticationHandler(
+    BaseAuthenticationHandlerMixin, APIKeyHeader, BaseAuthenticationHandler, ABC
+):
+    scheme: str = "header"

ellar/auth/handlers/api_key.py

@@ -1,98 +1,17 @@
-import binascii
-import typing as t
 from abc import ABC
-from base64 import b64decode
 
-from ellar.common import APIException
-from ellar.common.serializer.guard import (
-    HTTPAuthorizationCredentials,
-    HTTPBasicCredentials,
-)
+from .mixin import BaseAuthenticationHandlerMixin
+from .model import BaseAuthenticationHandler
+from .schemes import HttpBasicAuth, HttpBearerAuth
 
-from .base import BaseHttpAuthenticationHandler
 
-if t.TYPE_CHECKING:  # pragma: no cover
-    from ellar.core import HTTPConnection
+class HttpBearerAuthenticationHandler(
+    BaseAuthenticationHandlerMixin, HttpBearerAuth, BaseAuthenticationHandler, ABC
+):
+    scheme: str = "bearer"
 
 
-class HttpBearerAuthenticationHandler(BaseHttpAuthenticationHandler, ABC):
-    exception_class = APIException
-    openapi_scheme: str = "bearer"
-    openapi_bearer_format: t.Optional[str] = None
-    header: str = "Authorization"
-
-    @classmethod
-    def openapi_security_scheme(cls) -> t.Dict:
-        scheme = super().openapi_security_scheme()
-        scheme[cls.openapi_name or cls.__name__].update(
-            bearerFormat=cls.openapi_bearer_format
-        )
-        return scheme
-
-    def _get_credentials(
-        self, connection: "HTTPConnection"
-    ) -> t.Optional[HTTPAuthorizationCredentials]:
-        authorization: t.Optional[str] = connection.headers.get(self.header)
-        scheme, _, credentials = self.authorization_partitioning(authorization)
-
-        if not (authorization and scheme and credentials):
-            return None
-
-        if scheme and str(scheme).lower() != self.openapi_scheme:
-            raise self.exception_class(
-                status_code=404,
-                detail="Invalid authentication credentials",
-            )
-        return HTTPAuthorizationCredentials(scheme=scheme, credentials=credentials)
-
-
-class HttpBasicAuthenticationHandler(BaseHttpAuthenticationHandler, ABC):
-    exception_class = APIException
-    openapi_scheme: str = "basic"
-    realm: t.Optional[str] = None
-    header = "Authorization"
-
-    def _not_unauthorized_exception(self, message: str) -> None:
-        if self.realm:
-            unauthorized_headers = {"WWW-Authenticate": f'Basic realm="{self.realm}"'}
-        else:
-            unauthorized_headers = {"WWW-Authenticate": "Basic"}
-        raise self.exception_class(
-            status_code=404,
-            detail=message,
-            headers=unauthorized_headers,
-        )
-
-    def _get_credentials(
-        self, connection: "HTTPConnection"
-    ) -> t.Optional[HTTPBasicCredentials]:
-        authorization: t.Optional[str] = connection.headers.get(self.header)
-        parts = authorization.split(" ") if authorization else []
-        scheme, credentials = str(), str()
-
-        if len(parts) == 1:
-            credentials = parts[0]
-            scheme = "basic"
-        elif len(parts) == 2:
-            credentials = parts[1]
-            scheme = parts[0].lower()
-
-        if (
-            not (authorization and scheme and credentials)
-            or scheme.lower() != self.openapi_scheme
-        ):
-            return None
-
-        data: t.Optional[t.Union[str, bytes]] = None
-        try:
-            data = b64decode(credentials).decode("ascii")
-        except (ValueError, UnicodeDecodeError, binascii.Error):
-            self._not_unauthorized_exception("Invalid authentication credentials")
-
-        username, separator, password = (
-            str(data).partition(":") if data else (None, None, None)
-        )
-
-        if not separator:
-            self._not_unauthorized_exception("Invalid authentication credentials")
-        return HTTPBasicCredentials(username=username, password=password)
+class HttpBasicAuthenticationHandler(
+    BaseAuthenticationHandlerMixin, HttpBasicAuth, BaseAuthenticationHandler, ABC
+):
+    scheme: str = "basic"

ellar/auth/handlers/base.py

@@ -0,0 +1,25 @@
+import typing as t
+
+from ellar.common import Identity, IHostContext
+
+if t.TYPE_CHECKING:  # pragma: no cover
+    from ellar.core import HTTPConnection
+
+
+class BaseAuthenticationHandlerMixin:
+    scheme: str = "apiKey"
+    run_authentication_check: t.Callable[..., t.Coroutine]
+
+    @t.no_type_check
+    async def authenticate(self, context: IHostContext) -> t.Optional[Identity]:
+        return await self.run_authentication_check(context)
+
+    def handle_authentication_result(
+        self, connection: "HTTPConnection", result: t.Optional[t.Any]
+    ) -> t.Any:
+        if result:
+            return result
+        return None
+
+    def handle_invalid_request(self) -> t.Any:
+        return None

ellar/auth/handlers/http.py

@@ -8,16 +8,16 @@ class BaseAuthenticationHandler(ABC):
     scheme: str
 
     def __init_subclass__(cls, **kwargs: str) -> None:
-        if not cls.scheme:
-            raise Exception("Authentication Scheme is required")
+        if not hasattr(cls, "scheme"):
+            raise RuntimeError(f"'{cls.__name__}' has no attribute 'scheme'")
 
     @classmethod
     def openapi_security_scheme(cls) -> t.Optional[t.Dict]:
         return None
 
     @abstractmethod
     async def authenticate(self, context: IHostContext) -> t.Optional[Identity]:
-        ...
+        """Authenticate Action goes here"""
 
 
 AuthenticationHandlerType = t.Union[

ellar/auth/handlers/mixin.py

@@ -0,0 +1,15 @@
+from .api_key import APIKeyCookie, APIKeyHeader, APIKeyQuery
+from .base import BaseAPIKey, BaseAuth, BaseHttpAuth
+from .http import HttpBasicAuth, HttpBearerAuth, HttpDigestAuth
+
+__all__ = [
+    "APIKeyCookie",
+    "APIKeyHeader",
+    "APIKeyQuery",
+    "HttpBasicAuth",
+    "HttpBearerAuth",
+    "HttpDigestAuth",
+    "BaseAuth",
+    "BaseHttpAuth",
+    "BaseAPIKey",
+]

ellar/auth/handlers/model.py

@@ -0,0 +1,28 @@
+import typing as t
+from abc import ABC
+
+from .base import BaseAPIKey
+
+if t.TYPE_CHECKING:  # pragma: no cover
+    from ellar.core.connection import HTTPConnection
+
+
+class APIKeyQuery(BaseAPIKey, ABC):
+    openapi_in: str = "query"
+
+    def _get_key(self, connection: "HTTPConnection") -> t.Optional[t.Any]:
+        return connection.query_params.get(self.parameter_name)
+
+
+class APIKeyCookie(BaseAPIKey, ABC):
+    openapi_in: str = "cookie"
+
+    def _get_key(self, connection: "HTTPConnection") -> t.Optional[t.Any]:
+        return connection.cookies.get(self.parameter_name)
+
+
+class APIKeyHeader(BaseAPIKey, ABC):
+    openapi_in: str = "header"
+
+    def _get_key(self, connection: "HTTPConnection") -> t.Optional[t.Any]:
+        return connection.headers.get(self.parameter_name)