pikvm · intelfx · Jun 17, 2025 · Jun 17, 2025 · Jun 17, 2025
diff --git a/.gitignore b/.gitignore
@@ -19,3 +19,4 @@
 *.pyc
 *.swp
 /venv/
+/.idea
diff --git a/kvmd/apps/__init__.py b/kvmd/apps/__init__.py
@@ -35,6 +35,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
@@ -268,9 +269,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 [
@@ -385,6 +394,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=""),

diff --git a/kvmd/apps/kvmd/__init__.py b/kvmd/apps/kvmd/__init__.py
@@ -89,6 +89,9 @@ def main(argv: (list[str] | None)=None) -> None:
             ext_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),

diff --git a/kvmd/apps/kvmd/api/auth.py b/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, HTTPBadRequest
 
 from ....htserver import UnauthorizedError
 from ....htserver import ForbiddenError
@@ -40,9 +41,12 @@
 
 from ..auth import AuthManager
 
+from ....logging import get_logger
+
 
 # =====
 _COOKIE_AUTH_TOKEN = "auth_token"
+_COOKIE_OAUTH_SESSION = "oauth-session"
 
 
 async def _check_xhdr(auth_manager: AuthManager, _: HttpExposed, req: Request) -> bool:
@@ -136,3 +140,98 @@ async def __logout_handler(self, req: Request) -> Response:
     @exposed_http("GET", "/auth/check", allow_usc=False)
     async def __check_handler(self, _: Request) -> Response:
         return make_json_response()
+
+    @exposed_http("GET", "/auth/oauth/providers", auth_required=False, allow_usc=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, allow_usc=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:
+            raise HTTPBadRequest(reason="Auth disabled")
+
+        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, allow_usc=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:
+            raise HTTPBadRequest(reason="Auth disabled")
+
+        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_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 not user:
+            raise ForbiddenError()
+
+        token = await self.__auth_manager.login_oauth(
+            user=valid_user(user)
+        )
+        get_logger().info(f"OAUTH CALLBACK: {token=}")
+        if not token:
+            raise ForbiddenError()
+
+        response = HTTPFound(
+            request.url.with_path("").with_scheme("https")
+        )
+        response.set_cookie(name=_COOKIE_AUTH_TOKEN, value=token, samesite="Lax", httponly=True)
+        return response
diff --git a/kvmd/apps/kvmd/auth.py b/kvmd/apps/kvmd/auth.py
@@ -29,6 +29,7 @@
 import secrets
 import pyotp
 
+from .oauth import OAuthManager
 from ...logging import get_logger
 
 from ... import aiotools
@@ -69,6 +70,9 @@ def __init__(
         ext_kwargs: dict,
 
         totp_secret_path: str,
+
+        oauth_enabled: bool = False,
+        oauth_providers: (dict | None) = None,
     ) -> None:
 
         logger = get_logger(0)
@@ -107,10 +111,20 @@ def __init__(
             logger.info("Using external auth service %r",
                         self.__ext_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.__sessions: dict[str, _Session] = {}  # {token: session}
 
+        self.__tokens: dict[str, str] = {}  # {token: user}
+        self.__oauth_tokens: list[str] = []
+
     def is_auth_enabled(self) -> bool:
         return self.__enabled
 
@@ -172,6 +186,28 @@ async def login(self, user: str, passwd: str, expire: int) -> (str | None):
 
         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()
+        session = _Session(
+            user=user,
+            expire_ts=self.__make_expire_ts(0),
+        )
+        self.__sessions[token] = session
+        get_logger(0).info("Logged in via OAuth (user %r); expire=%s, sessions_now=%d",
+                           session.user,
+                           self.__format_expire_ts(session.expire_ts),
+                           self.__get_sessions_number(session.user))
+        return token
+
     def __make_new_token(self) -> str:
         for _ in range(10):
             token = secrets.token_hex(32)
-Original file line number
+Diff line change
@@ Expand Up / @@ -19,3 +19,4 @@ @@
     *.pyc
     *.swp
     /venv/
+    /.idea