### # Copyright (c) 2021, Georg Pfuetzenreuter # All rights reserved. # # Redistribution and use in source and binary forms, with or without # modification, are permitted provided that the following conditions are met: # # * Redistributions of source code must retain the above copyright notice, # this list of conditions, and the following disclaimer. # * Redistributions in binary form must reproduce the above copyright notice, # this list of conditions, and the following disclaimer in the # documentation and/or other materials provided with the distribution. # * Neither the name of the author of this software nor the name of # contributors to this software may be used to endorse or promote products # derived from this software without specific prior written consent. # # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" # AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE # IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE # ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE # LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR # CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF # SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS # INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN # CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) # ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE # POSSIBILITY OF SUCH DAMAGE. ### import re import requests import secrets import string from supybot import utils, plugins, ircutils, callbacks, ircmsgs from supybot.commands import * from supybot.ircmsgs import nick try: from supybot.i18n import PluginInternationalization _ = PluginInternationalization('Keycloak') except ImportError: # Placeholder that allows to run the plugin on a bot # without the i18n module _ = lambda x: x class Keycloak(callbacks.Plugin): """Interfaces with Keycloak SSO.""" threaded = True def register(self, irc, msg, args, email): """ registers an account with your username and the specified email address""" server = self.registryValue('backend.server') realm = self.registryValue('backend.realm') tokenurl = self.registryValue('backend.token') usererr = self.registryValue('replies.error') emailverified = self.registryValue('options.emailVerified') firstname = self.registryValue('options.firstName') lastname = self.registryValue('options.lastName') alphabet = string.ascii_letters + string.digits random = ''.join(secrets.choice(alphabet) for i in range(64)) try: tokendl = requests.get(tokenurl) tokendata = tokendl.json() token = tokendata['access_token'] url = server + '/auth/admin/realms/' + realm + '/users' except: print("ERROR: Keycloak token could not be installed.") irc.error(usererr) if re.match(r"[^@]+@[^@]+\.[^@]+", email): pw = random payload = { "firstName": firstname, "lastName": lastname, "email": email, "enabled": "true", "username": msg.nick, "credentials": [{"type": "password", "value": pw, "temporary": emailverified,}], "emailVerified": "false" } response = requests.post( url, headers = {'Content-Type': 'application/json', 'Authorization': 'Bearer ' + token}, json = payload ) print("Keycloak: HTTP Status ", response.status_code) try: print("Keycloak: Response Text: ", response.text) except: print("Keycloak: No or invalid response text. This is not an error.") try: print("Keycloak: Response JSON: ", response.json()) except: print("Keycloak: No or invalid response JSON. This it not an error.") status = response.status_code if status == 201: print(" SSO User " + msg.nick + " created.") irc.queueMsg(msg=ircmsgs.IrcMsg(command='PRIVMSG', args=(msg.nick, f'{pw}'))) irc.reply("OK, I sent you a private message.") if status == 400: print("ERROR: Keycloak indicated that the request is invalid.") irc.error(usererr) if status == 401: print("ERROR: Fix your Keycloak API credentials and/or client roles, doh.") irc.error(usererr) if status == 403: print("ERROR: Keycloak indicated that the authorization provided is not enough to access the resource.") irc.error(usererr) if status == 404: print("ERROR: Keycloak indicated that the requested resource does not exist.") irc.error(usererr) if status == 409: print("ERROR: Keycloak indicated that the resource already exists or \"some other coonflict when processing the request\" occured.") irc.reply("Your username seems to already be registerd.") if status == 415: print("ERROR: Keycloak indicated that the requested media type is not supported.") irc.error(usererr) if status == 500: print("ERROR: Keycloak indicated that the server could not fullfill the request due to \"some unexpected error \".") irc.error(usererr) else: irc.error("Is that a valid email address?") register = wrap(register, ['anything']) def ircprom(self, irc, msg, args, option): """ true/on = enable authentication to your IRC account with an SSO account going by the same username -- false/off = allow authentication to your IRC account ONLY with internal IRC credentials (NickServ) -- Warning: Enabling this without having an SSO account with the same username as your IRC nickname is a security risk.""" user = msg.nick server = self.registryValue('backend.server') realm = self.registryValue('backend.realm') tokenurl = self.registryValue('backend.token') usererr = self.registryValue('replies.error') gid = self.registryValue('options.ircgroup') try: tokendl = requests.get(tokenurl) tokendata = tokendl.json() token = tokendata['access_token'] url = server + '/auth/admin/realms/' + realm + '/users' userdata = requests.get(url, params = {'username': user}, headers = {'Content-Type': 'application/json', 'Authorization': 'Bearer ' + token}) userresp = userdata.json() uid = userresp[0]['id'] print(user, uid) except: print("ERROR: Keycloak token could not be installed.") irc.error(usererr) try: url = server + '/auth/admin/realms/' + realm + '/users/' + uid + '/groups/' + gid if option == 'true' or option == 'on' or option == '1': option = 'enable' response = requests.put( url, headers = {'Content-Type': 'application/json', 'Authorization': 'Bearer ' + token}) if option == 'false' or option == 'off' or option == '0': option == 'disable' response = requests.delete( url, headers = {'Content-Type': 'application/json', 'Authorization': 'Bearer ' + token}) if option != 'true' != 'on' != '1' != 'false' != 'off' != '0': irc.error('Invalid argument.') else: print("Keycloak: HTTP Status ", response.status_code) try: print("Keycloak: Response Text: ", response.text) except: print("Keycloak: No or invalid response text. This is not an error.") try: print("Keycloak: Response JSON: ", response.json()) except: print("Keycloak: No or invalid response JSON. This it not an error.") status = response.status_code if status == 204: print(" SSO user " + user + " is now authorized to authenticate IRC user " + user) irc.queueMsg(msg=ircmsgs.IrcMsg(command='PRIVMSG', args=(msg.nick, f'{pw}'))) irc.reply("OK, I sent you a private message.") if status != 204: print("ERROR: HTTP request did not succeed.") irc.error(usererr) except: print('Operation failed.') ircprom = wrap(ircprom, ['anything']) Class = Keycloak # vim:set shiftwidth=4 softtabstop=4 expandtab textwidth=79: