Added keycloak authentication

docs/Usage/Authentication.md

@@ -41,6 +41,7 @@ Here is the full list:
 |[kerberos-password](../../xpra/server/auth/kerberos_password_auth.py)|Uses kerberos to authenticate a username + password|[#1691](../https://github.com/Xpra-org/xpra/issues/1691)|
 |[kerberos-ticket](../../xpra/server/auth/kerberos_ticket_auth.py)|Uses a kerberos ticket to authenticate a client|[#1691](../https://github.com/Xpra-org/xpra/issues/1691)|
 |[gss_auth](../../xpra/trunk/src/xpra/server/auth/gss_auth.py)|Uses a GSS ticket to authenticate a client|[#1691](../https://github.com/Xpra-org/xpra/issues/1691)|
+|[keycloak](../../xpra/server/auth/keycloak_auth.py)|Uses a keycloak token to authenticate a client|[#3334](../https://github.com/Xpra-org/xpra/issues/3334)|
 |[ldap](../../xpra/server/auth/ldap_auth.py)|Uses ldap via [python-ldap](https://www.python-ldap.org/en/latest/)|[#1791](../https://github.com/Xpra-org/xpra/issues/1791)|
 |[ldap3](../../xpra/server/auth/ldap3_auth.py)|Uses ldap via [python-ldap3](https://github.com/cannatag/ldap3)|[#1791](../https://github.com/Xpra-org/xpra/issues/1791)|
 |[u2f](../../xpra/trunk/src/xpra/server/auth/u2f_auth.py)|[Universal 2nd Factor](https://en.wikipedia.org/wiki/Universal_2nd_Factor)|[#1789](../https://github.com/Xpra-org/xpra/issues/1789)|
@@ -134,7 +135,7 @@ The steps below assume that the client and server have been configured to use au
 * if the server is not configured for authentication, the client connection should be accepted and a warning will be printed
 * if the client is not configured for authentication, a password dialog may show up, and the connection will fail with an authentication error if the correct value is not supplied
 * if multiple authentication modules are specified, the client may bring up multiple authentication dialogs
-* how the client handles the challenges sent by the server can be configured using the `challenge-handlers` option, by default the client will try the following handlers in the specified order: `uri` (whatever password may have been specified in the connection string), `file` (if the `password-file` option was used), `env` (if the environment variable is present), `kerberos`, `gss`, `u2f` and finally `prompt`
+* how the client handles the challenges sent by the server can be configured using the `challenge-handlers` option, by default the client will try the following handlers in the specified order: `uri` (whatever password may have been specified in the connection string), `file` (if the `password-file` option was used), `env` (if the environment variable is present), `kerberos`, `gss`, `keycloak`, `u2f` and finally `prompt`
 </details>
 <details>
   <summary>module and platform specific notes</summary>

xpra/net/digest.py

@@ -67,7 +67,7 @@ def gendigest(digest, password, salt):
         salt = salt.ljust(16, b"\x00")[:16]
         v = generate_response(password, salt)
         return hexstr(v)
-    if digest in ("xor", "kerberos", "gss"):
+    if digest in ("xor", "kerberos", "gss", "keycloak"):
         #kerberos and gss use xor because we need to use the actual token
         #at the other end
         salt = salt.ljust(len(password), b"\x00")[:len(password)]

xpra/server/auth/keycloak_auth.py

@@ -0,0 +1,185 @@
+# This file is part of Xpra.
+# Copyright (C) 2016-2021 Antoine Martin <antoine@xpra.org>
+# Copyright (C) 2022 Nathalie Casati <nat@yuka.ch>
+# Xpra is released under the terms of the GNU GPL v2, or, at your option, any
+# later version. See the file COPYING for details.
+
+import os
+import sys
+import json
+
+from xpra.util import typedict
+from xpra.server.auth.sys_auth_base import SysAuthenticator, log
+
+KEYCLOAK_SERVER_URL = os.environ.get("XPRA_KEYCLOAK_SERVER_URL", "http://localhost:8080/auth/")
+KEYCLOAK_REALM_NAME = os.environ.get("XPRA_KEYCLOAK_REALM_NAME", "example_realm")
+KEYCLOAK_CLIENT_ID  = os.environ.get("XPRA_KEYCLOAK_CLIENT_ID", "example_client")
+KEYCLOAK_CLIENT_SECRET_KEY = os.environ.get("XPRA_KEYCLOAK_CLIENT_SECRET_KEY", "secret")
+KEYCLOAK_REDIRECT_URI = os.environ.get("XPRA_KEYCLOAK_REDIRECT_URI", "http://localhost/login/")
+KEYCLOAK_SCOPE = os.environ.get("XPRA_KEYCLOAK_SCOPE", "openid")
+KEYCLOAK_GRANT_TYPE = os.environ.get("XPRA_KEYCLOAK_GRANT_TYPE", "authorization_code")
+
+
+class Authenticator(SysAuthenticator):
+
+    def __init__(self, **kwargs):
+        self.server_url = kwargs.pop("server_url", KEYCLOAK_SERVER_URL)
+        self.realm_name = kwargs.pop("realm_name", KEYCLOAK_REALM_NAME)
+        self.client_id = kwargs.pop("client_id", KEYCLOAK_CLIENT_ID)
+        self.client_secret_key = kwargs.pop("client_secret_key", KEYCLOAK_CLIENT_SECRET_KEY)
+        self.redirect_uri = kwargs.pop("redirect_uri", KEYCLOAK_REDIRECT_URI)
+        self.scope = kwargs.pop("scope", KEYCLOAK_SCOPE)
+        self.grant_type = kwargs.pop("grant_type", KEYCLOAK_GRANT_TYPE)
+
+        # use keycloak as default prompt
+        kwargs["prompt"] = kwargs.pop("prompt", "keycloak")
+
+        if KEYCLOAK_GRANT_TYPE != "authorization_code":
+          raise(NotImplementedError("Warning: only grant type \"authorization_code\" is currently supported"))
+
+        super().__init__(**kwargs)
+        log("keycloak auth: server_url=%s, client_id=%s, realm_name=%s, redirect_uri=%s, scope=%s, grant_type=%s",
+            self.server_url, self.client_id, self.realm_name, self.redirect_uri, self.scope, self.grant_type)
+
+        try:
+          from oauthlib.oauth2 import WebApplicationClient
+
+          # Get authorization code
+          client = WebApplicationClient(KEYCLOAK_CLIENT_ID)
+          authorization_url = KEYCLOAK_SERVER_URL + 'realms/' + KEYCLOAK_REALM_NAME + '/protocol/openid-connect/auth'
+          self.salt = client.prepare_request_uri(
+                      authorization_url,
+                      redirect_uri = KEYCLOAK_REDIRECT_URI,
+                      scope = [KEYCLOAK_SCOPE],
+          )
+        except ImportError as e:
+          log("check(..)", exc_info=True)
+          log.warn("Warning: cannot use keycloak authentication:")
+          log.warn(" %s", e)
+          # unsure how to fail the auth at this point so we raise the exception
+          raise(e)
+
+    def __repr__(self):
+        return "keycloak"
+
+    def get_challenge(self, digests):
+        assert not self.challenge_sent
+        if "keycloak" not in digests:
+            log.error("Error: client does not support keycloak authentication")
+            return None
+        self.challenge_sent = True
+        return self.salt, "keycloak"
+
+    def check(self, response_json) -> bool:
+        assert self.challenge_sent
+        #log("response_json: %r", response_json)
+
+        if not response_json:
+          log.error("Error: keycloak authentication failed")
+          log.error("Invalid response received from authorization endpoint")
+          return False
+
+        try:
+          response = json.loads(response_json)
+        except json.JSONDecodeError:
+          log.error("Error: keycloak authentication failed")
+          log.error("Invalid response received from authorization endpoint")
+          log("failed to parse json: %r", response_json, exc_info=True)
+          return False 
+
+        if not isinstance(response, dict):
+          log.error("Error: keycloak authentication failed")
+          log.error("Invalid response received from authorization endpoint")
+          log("response is of type %r but dict type is required", type(response), exc_info=True)
+          log("failed to load response %r", response, exc_info=True)
+          return False
+
+        log("check(%r)", response, exc_info=True)
+
+        auth_code = response.get("code")
+        error = response.get("error")
+
+        if error:
+          log.error("Error: keycloak authentication failed")
+          log.error("%s: %s", error, response.get("error_description"))
+          return False
+
+        if not auth_code:
+          log.error("Error: keycloak authentication failed")
+          log.error("Invalid response received from authorization endpoint")
+
+        try:
+          from keycloak import KeycloakOpenID
+          from keycloak.exceptions import KeycloakError
+
+          # Configure client
+          keycloak_openid = KeycloakOpenID(server_url=KEYCLOAK_SERVER_URL,
+                            client_id=KEYCLOAK_CLIENT_ID,
+                            realm_name=KEYCLOAK_REALM_NAME,
+                            client_secret_key=KEYCLOAK_CLIENT_SECRET_KEY)
+
+          # Get well_known
+          config_well_know = keycloak_openid.well_know()
+          log("well_known: %r", config_well_know, exc_info=True)
+
+          # Get token
+          token = keycloak_openid.token(code=auth_code, grant_type=[KEYCLOAK_GRANT_TYPE], redirect_uri=KEYCLOAK_REDIRECT_URI)
+
+          # Verify token
+          access_token = token.get("access_token")
+          if not access_token:
+            log.error("Error: keycloak authentication failed as access token is missing")
+            return False
+
+          token_info = keycloak_openid.introspect(access_token)
+          log("token_info: %r", token_info, exc_info=True)
+
+          token_state = token_info.get("active")
+          if token_state is None:
+            log.error("Error: keycloak authentication failed as token state is missing")
+            return False
+
+          if token_state is False:
+            log.error("Error: keycloak authentication failed as token state not active")
+            return False
+
+          if token_state is True:
+            # Get userinfo
+            user_info = keycloak_openid.userinfo(access_token)
+            log("userinfo_info: %r", user_info, exc_info=True)
+
+            log("keycloak authentication succeeded: token is active")
+            return True
+
+          log.error("Error: keycloak authentication failed as token state is invalid")
+          return False
+        except KeycloakError as e:
+          log.error("Error: keycloak authentication failed")
+          log.error("Error code %s: %s", e.response_code, e.error_message)
+          return False
+        except ImportError as e:
+          log("check(..)", exc_info=True)
+          log.warn("Warning: cannot use keycloak authentication:")
+          log.warn(" %s", e)
+          return False
+
+def main(args):
+    if len(args)!=2:
+        print("invalid number of arguments")
+        print("usage:")
+        print("%s response_json" % (args[0],))
+        return 1
+    response_json = args[1]
+
+    a = Authenticator()
+    a.get_challenge("keycloak")
+
+    if a.check(response_json):
+        print("success")
+        return 0
+    else:
+        print("failed")
+        return -1
+
+if __name__ == "__main__":
+    sys.exit(main(sys.argv))