OAuth Manager and OAuth2/OpenID connect Plugin

.gitignore

@@ -19,3 +19,4 @@
 *.pyc
 *.swp
 /venv/
+/.idea

kvmd/apps/__init__.py

@@ -37,6 +37,7 @@
 
 from ..plugins import UnknownPluginError
 from ..plugins.auth import get_auth_service_class
+from ..plugins.auth import get_oauth_service_class
 from ..plugins.hid import get_hid_class
 from ..plugins.atx import get_atx_class
 from ..plugins.msd import get_msd_class
@@ -260,9 +261,17 @@ def _patch_dynamic(  # pylint: disable=too-many-locals
     rebuild = False
 
     if load_auth:
-        scheme["kvmd"]["auth"]["internal"].update(get_auth_service_class(config.kvmd.auth.internal.type).get_plugin_options())
+        scheme["kvmd"]["auth"]["internal"].update(
+            get_auth_service_class(config.kvmd.auth.internal.type).get_plugin_options()
+        )
         if config.kvmd.auth.external.type:
-            scheme["kvmd"]["auth"]["external"].update(get_auth_service_class(config.kvmd.auth.external.type).get_plugin_options())
+            scheme["kvmd"]["auth"]["external"].update(
+                get_auth_service_class(config.kvmd.auth.external.type).get_plugin_options()
+            )
+        if config.kvmd.auth.oauth.enabled:
+            for provider, data in tools.rget(raw_config, "kvmd", "auth", "oauth", "providers").items():
+                scheme["kvmd"]["auth"]["oauth"]["providers"][provider] = get_oauth_service_class(data["type"]).get_plugin_options()
+                scheme["kvmd"]["auth"]["oauth"]["providers"][provider]["type"] = Option(data["type"])
         rebuild = True
 
     for (load, section, get_class) in [
@@ -371,6 +380,13 @@ def _get_config_scheme() -> dict:
                     # Dynamic content
                 },
 
+                "oauth": {
+                    "enabled": Option(False, type=valid_bool),
+                    "providers": {
+                        # Dynamic content
+                    }
+                },
+
                 "totp": {
                     "secret": {
                         "file": Option("/etc/kvmd/totp.secret", type=valid_abs_path, if_empty=""),

kvmd/apps/kvmd/__init__.py

@@ -85,6 +85,9 @@ def main(argv: (list[str] | None)=None) -> None:
             external_kwargs=(config.auth.external._unpack(ignore=["type"]) if config.auth.external.type else {}),
 
             totp_secret_path=config.auth.totp.secret.file,
+
+            oauth_enabled=config.auth.oauth.enabled,
+            oauth_providers=config.auth.oauth.providers._unpack(ignore=["enabled"] if config.auth.oauth.enabled else {}),
         ),
         info_manager=InfoManager(global_config),
         log_reader=(LogReader() if config.log_reader.enabled else None),

kvmd/apps/kvmd/api/auth.py

@@ -24,6 +24,7 @@
 
 from aiohttp.web import Request
 from aiohttp.web import Response
+from aiohttp.web_exceptions import HTTPNotFound, HTTPFound, HTTPUnauthorized
 
 from ....htserver import UnauthorizedError
 from ....htserver import ForbiddenError
@@ -41,6 +42,7 @@
 
 # =====
 _COOKIE_AUTH_TOKEN = "auth_token"
+_COOKIE_OAUTH_SESSION = "oauth-session"
 
 
 async def check_request_auth(auth_manager: AuthManager, exposed: HttpExposed, request: Request) -> None:
@@ -107,3 +109,91 @@ async def __logout_handler(self, request: Request) -> Response:
     @exposed_http("GET", "/auth/check")
     async def __check_handler(self, _: Request) -> Response:
         return make_json_response()
+
+    @exposed_http("GET", "/auth/oauth/providers", auth_required=False)
+    async def __oauth_providers(self, request: Request) -> Response:
+        """
+        Return a json containing the available Providers with short_name and long_name and if oauth is enabled
+        @param request:
+        @return: json with provider infos
+        """
+        response: dict[str, (bool | dict)] = {}
+        if self.__auth_manager.oauth_manager is None:
+            response.update({'enabled': False})
+        else:
+            response.update({'enabled': True, 'providers': self.__auth_manager.oauth_manager.get_providers()})
+        return make_json_response(response)
+
+    @exposed_http("GET", "/auth/oauth/login/{provider}", auth_required=False)
+    async def __oauth(self, request: Request) -> None:
+        """
+        Creates the redirect to the Provider specified in the URL. Checks if the provider is valid.
+        Also sets a cookie containing session information.
+        @param request:
+        @return: redirect to provider
+        """
+        if self.__auth_manager.oauth_manager is None:
+            return
+        provider = format(request.match_info['provider'])
+        if not self.__auth_manager.oauth_manager.valid_provider(provider):
+            raise HTTPNotFound(reason="Unknown provider %s" % provider)
+
+        redirect_url = request.url.with_path(f"/api/auth/oauth/callback/{provider}").with_scheme('https')
+        oauth_cookie = request.cookies.get(_COOKIE_OAUTH_SESSION, "")
+
+        is_valid_session = await self.__auth_manager.oauth_manager.is_valid_session(provider, oauth_cookie)
+        if not is_valid_session:
+            session = await self.__auth_manager.oauth_manager.register_new_session(provider)
+        else:
+            session = oauth_cookie
+
+        response = HTTPFound(
+            await self.__auth_manager.oauth_manager.get_authorize_url(
+                provider=provider, redirect_url=redirect_url, session=session,
+            )
+        )
+        response.set_cookie(name=_COOKIE_OAUTH_SESSION, value=session, secure=True, httponly=True, samesite="Lax")
+
+        # 302 redirect to provider:
+        raise response
+
+    @exposed_http("GET", "/auth/oauth/callback/{provider}", auth_required=False)
+    async def __callback(self, request: Request) -> Response:
+        """
+        After successful login on the side of the provider, the user gets redirected here. If everything is correct,
+        the user gets logged in with the username provided by the Provider.
+        @param request:
+        @return:
+        """
+        if self.__auth_manager.oauth_manager is None:
+            return make_json_response()
+
+        if not request.match_info['provider']:
+            raise HTTPUnauthorized(reason="Provider is missing")
+        provider = format(request.match_info['provider'])
+        if not self.__auth_manager.oauth_manager.valid_provider(provider):
+            raise HTTPNotFound(reason="Unknown provider %s" % provider)
+
+        if _COOKIE_OAUTH_SESSION not in request.cookies.keys():
+            raise HTTPUnauthorized(reason="Cookie is missing")
+        oauth_session = request.cookies[_COOKIE_OAUTH_SESSION]
+
+        if not self.__auth_manager.oauth_manager.is_redirect_from_provider(provider=provider, request_query=dict(request.query)):
+            raise HTTPUnauthorized(reason="Authorization Code is missing")
+
+        redirect_url = request.url.with_query("").with_path(f"/api/auth/oauth/callback/{provider}").with_scheme('https')
+        user = await self.__auth_manager.oauth_manager.get_user_info(
+            provider=provider,
+            oauth_session=oauth_session,
+            request_query=dict(request.query),
+            redirect_url=redirect_url
+        )
+
+        if self.__auth_manager.is_auth_enabled():
+            token = await self.__auth_manager.login_oauth(
+                user=valid_user(user)
+            )
+            if token:
+                return make_json_response(set_cookies={_COOKIE_AUTH_TOKEN: token})
+            raise ForbiddenError()
+        return make_json_response()

kvmd/apps/kvmd/auth.py

@@ -23,6 +23,7 @@
 import secrets
 import pyotp
 
+from .oauth import OAuthManager
 from ...logging import get_logger
 
 from ... import aiotools
@@ -34,8 +35,8 @@
 
 
 # =====
-class AuthManager:
-    def __init__(
+class AuthManager:  # pylint: disable=too-many-instance-attributes
+    def __init__(  # pylint: disable=too-many-arguments
         self,
         enabled: bool,
         unauth_paths: list[str],
@@ -48,6 +49,9 @@ def __init__(
         external_kwargs: dict,
 
         totp_secret_path: str,
+
+        oauth_enabled: bool = False,
+        oauth_providers: (dict | None) = None,
     ) -> None:
 
         self.__enabled = enabled
@@ -70,9 +74,17 @@ def __init__(
             self.__external_service = get_auth_service_class(external_type)(**external_kwargs)
             get_logger().info("Using external auth service %r", self.__external_service.get_plugin_name())
 
+        self.oauth_manager: (OAuthManager | None) = None
+        if enabled and oauth_enabled:
+            if oauth_providers is None:
+                oauth_providers = {}
+            self.oauth_manager = OAuthManager(oauth_providers)
+            get_logger().info("Using OAuth service")
+
         self.__totp_secret_path = totp_secret_path
 
         self.__tokens: dict[str, str] = {}  # {token: user}
+        self.__oauth_tokens: list[str] = []
 
     def is_auth_enabled(self) -> bool:
         return self.__enabled
@@ -124,6 +136,22 @@ async def login(self, user: str, passwd: str) -> (str | None):
         else:
             return None
 
+    async def login_oauth(self, user: str) -> (str | None):
+        """
+        registers the user, who logged in with oauth, with a new token.
+        @param user: the username provided by the oauth provider
+        @return:
+        """
+        assert user == user.strip()
+        assert user
+        assert self.__enabled
+        assert self.oauth_manager
+        token = self.__make_new_token()
+        self.__tokens[token] = user
+
+        get_logger().info("Logged in user with OAuth %r", user)
+        return token
+
     def __make_new_token(self) -> str:
         for _ in range(10):
             token = secrets.token_hex(32)