# Mattermost Bot. # Copyright (c) 2016-2022 by Someone (aka. Jan Vales ) # published under MIT-License import atexit import json import logging import os import pprint import signal import sys import threading import traceback import urllib from inspect import cleandoc from http.server import BaseHTTPRequestHandler, HTTPServer from urllib.request import Request, urlopen from socketserver import ThreadingMixIn import mattermost import mattermost.ws logger = logging.getLogger(__name__) class MMBot(): def __init__(self, local_websrv_hostname="localhost", local_websrv_port=18065, local_public_websrv_hostname="localhost", local_public_websrv_port=18066, api_user=None, api_user_pw=None, api_bearer=None, mm_api_url="http://localhost:8065/api", mm_ws_url="ws://localhost:8065/api/v4/websocket", debug_chan_id=None): self.local_websrv_hostname = local_websrv_hostname self.local_websrv_port = local_websrv_port self.local_public_websrv_hostname = local_public_websrv_hostname self.local_public_websrv_port = local_public_websrv_port self.api_user = api_user self.mm_api_url = mm_api_url self.mm_ws_url = mm_ws_url self.modules = {} self.wsmodules = {} self.modules_public = {} self.api = None self.debug_chan_id = debug_chan_id self.command_stats = {} self.admin_ids = [] self.mmws = None self.sigusr1_cnt = 0 atexit.register(self.on_shutdown) signal.signal(signal.SIGUSR1, self.on_SIGUSR1) signal.signal(signal.SIGTERM, self.shutdown) signal.signal(signal.SIGINT, self.shutdown) # monkey-patch thread naming try: import pyprctl def _rename_current_thread(self, name, thread): if thread == threading.current_thread(): thread.setName(name) pyprctl.set_name(name) def _bootstrap_named_thread(self): self.rename_current_thread(self._name, threading.current_thread()) self.original_bootstrap() threading.Thread.original_bootstrap = threading.Thread._bootstrap threading.Thread._bootstrap = _bootstrap_named_thread except ImportError: logger.error('pyprctl module is not installed. You will not be able to see thread names') def _rename_current_thread(self, name, thread): if thread == threading.current_thread(): thread.setName(name) threading.Thread.rename_current_thread = _rename_current_thread # Core-Command: /use-data self.USETOPICS = {"bot":cleandoc("""##### Here I am, brain the size of a planet, and they ask me to fill in some missing MM-features and be fun ... It gives me a headache. Written by **``@someone``** in python3. Big thanks to contributors: **``@ju``**, **``@gittenburg``** Inspiring ideas by: ``@bearza``, ``@frunobulax``, **``@x5468656f``** Feel like contributing too? Talk to **``@someone``**. :) The repository is here: https://git.somenet.org/pub/jan/mattermost-bot.git """)} if api_user is not None and api_user_pw is not None: logger.info("User credentials given. Trying to login.") self.api = mattermost.MMApi(self.mm_api_url) self.api.login(api_user, api_user_pw) elif api_user is not None: self.api = mattermost.MMApi(self.mm_api_url) self.api.login(bearer=api_bearer) # Register a module with the bot. def register(self, module): if module.TEAM_ID not in self.modules: self.modules[module.TEAM_ID] = {} if module.TRIGGER not in self.modules[module.TEAM_ID]: self.modules[module.TEAM_ID][module.TRIGGER] = module module._on_register(self) else: raise Exception("Multiple registration attempts for module: "+module.TRIGGER+" and team: "+module.TEAM_ID) # Register a websocket handling module with the bot. # There is no useful way to discriminate WS-events by originating teams atm. :( def register_ws(self, module, eventlist): for evtype in eventlist: if evtype not in self.wsmodules: self.wsmodules[evtype] = {} if module.NAME not in self.wsmodules[evtype]: self.wsmodules[evtype][module.NAME] = module module._on_register_ws_evtype(self, evtype) else: raise Exception("Multiple registration attempts for module: "+module.NAME+" and evtype: "+evtype) # Register a public facing module with the bot. def register_public(self, module): self.modules_public = {} if module.TRIGGER not in self.modules_public: self.modules_public[module.TRIGGER] = module module._on_register_public(self) else: raise Exception("Multiple registration attempts for module: "+module.TRIGGER+" and team: "+module.TEAM_ID) def start(self): logger.info("Starting: Almost there.") logger.info(pprint.pformat(self.modules)) logger.info(pprint.pformat(self.wsmodules)) logger.info(pprint.pformat(self.modules_public)) if self.mm_ws_url is not None: self.mmws = mattermost.ws.MMws(self.websocket_handler, self.api, self.mm_ws_url) if self.local_public_websrv_hostname is not None and self.local_public_websrv_port is not None: th = threading.Thread(target=self.start_webserver_public, daemon=True) th.setName("http public") th.start() if self.local_websrv_hostname is not None and self.local_websrv_port is not None: self.start_webserver_mattermost() def on_shutdown(self): logger.info("Shutting down ...") if self.mmws: self.mmws.close_websocket() for team_modules in self.modules: for module in self.modules[team_modules]: self.modules[team_modules][module]._on_shutdown() for evtype in self.wsmodules: for module in self.wsmodules[evtype]: self.wsmodules[evtype][module]._on_shutdown() self.api.logout() self.command_stats_dump() logger.info("BYE.") ######## # misc # ######## def shutdown(self, unk1=None, unk2=None, exit_code=0): sys.exit(exit_code) def on_SIGUSR1(self, unk1=None, unk2=None): logger.info("on_SIGUSR1()") self.sigusr1_cnt += 1 self.command_stats_inc("internal::SIGUSR1") self.command_stats_dump() for team_modules in self.modules: for module in self.modules[team_modules]: self.modules[team_modules][module]._on_SIGUSR1(self.sigusr1_cnt) for evtype in self.wsmodules: for module in self.wsmodules[evtype]: self.wsmodules[evtype][module]._on_SIGUSR1(self.sigusr1_cnt) def debug_chan(self, message, root_id=None): if self.debug_chan_id is None: logger.error("debug_chan() called, but debug_chan_id is unspecified.") return None return self.api.create_post(self.debug_chan_id, "``AUTODELETE-DAY``\n"+message, root_id=root_id) def command_stats_inc(self, command, amount=1): if command in self.command_stats: self.command_stats[command] += amount else: self.command_stats[command] = amount def command_stats_dump(self): self.dump_stats_json(self.command_stats, "/tmp/somebot_command_stats.json", "#command_usage #mmstats") self.command_stats = {} def dump_stats_json(self, stats_data, file_path, header="", footer="", no_data_text=""): do_write = False if stats_data: do_write = True try: with open(file_path, "r") as file: old_stats = json.load(file) for item, cnt in old_stats["data"].items(): if item in stats_data: stats_data[item] += cnt else: stats_data[item] = cnt except (FileNotFoundError, json.JSONDecodeError, KeyError): do_write = True # if no data, but file exists: skip write if not do_write and "header" in old_stats and "footer" in old_stats and "no_data_text" in old_stats and old_stats["header"] == header and old_stats["footer"] == footer and old_stats["no_data_text"] == no_data_text: return logger.info("dump_stats_json(): writing file: %s", file_path) self.command_stats_inc("internal::dump_stats_json:"+file_path) with open(file_path, "w", encoding="utf-8") as file: json.dump({"header":header, "footer":footer, "no_data_text":no_data_text, "data":dict(sorted(stats_data.items()))}, file, ensure_ascii=False, indent=2) ########################## # Bot's websocket client # ########################## def websocket_handler(self, mmws, event_data): for evtype in self.wsmodules: if evtype == event_data["event"]: for module_name in self.wsmodules[evtype]: try: if self.wsmodules[evtype][module_name].on_WS_EVENT(event_data): self.command_stats_inc("ws::"+evtype+"::"+module_name) except Exception as exc: self.debug_chan("##### Exception in ``"+evtype+"::"+module_name+"``: ``"+repr(exc)+"``\n# :boom::boom::boom::boom::boom:") logger.exception("websocket_handler(): Exception in: %s\nEvent-data:%s", evtype+"::"+module_name, pprint.pformat(event_data)) ############################## # Bot's mattermost webserver # ############################## def start_webserver_mattermost(self): logger.info("Starting mattermost facing webserver.") class HTTPRequestHandler(BaseHTTPRequestHandler): bot = None handled = False responseURL = None def version_string(self): return "SomeBot" def do_POST(self): self.handled = False if self.headers["Content-Type"] == "application/x-www-form-urlencoded": data = urllib.parse.parse_qs(self.rfile.read(int(self.headers["Content-Length"])).decode("utf-8"), keep_blank_values=True) # only accept first occurence data = {k: v[0] for k, v in data.items()} elif self.headers["Content-Type"] == "application/json": data = json.loads(self.rfile.read(int(self.headers["Content-Length"])).decode("utf-8")) else: self.respond(415) return # store responseURL if "response_url" in data: self.responseURL = data["response_url"] # handle call logger.debug("do_POST(): request incomming.") try: module = 'not yet known' splitpath = self.path.strip("/").split("/") if splitpath[0] in self.bot.modules and splitpath[1] in self.bot.modules[splitpath[0]]: module = self.bot.modules[splitpath[0]][splitpath[1]] # /command if len(splitpath) > 2 and splitpath[2] == "command": if "token" in data and module.mm_secret_token == data["token"]: self.bot.command_stats_inc("/"+splitpath[1]+"/"+splitpath[2]) module._on_POST(self, data) else: logger.error("do_POST(): Auth problem: Shutting down: mm_secret_token mismatch expected/got -%s-%s-", module.mm_secret_token, data["token"]) traceback.print_stack() self.respond(403) self.bot.on_shutdown() os._exit(1) # interactive button-handler. TODO auth! elif len(splitpath) > 2 and splitpath[2] == "interactive": self.bot.command_stats_inc("/"+splitpath[1]+"/"+splitpath[2]) module._on_POST_interactive(self, data) # dialog-handler: TODO: auth! elif len(splitpath) > 2 and splitpath[2] == "dialog": self.bot.command_stats_inc("/"+splitpath[1]+"/"+splitpath[2]) module._on_POST_dialog(self, data) else: # Invalid command action logger.error("do_POST(): Invalid command action.") self.bot.debug_chan("do_POST(): Invalid command action.\n# :boom::boom::boom::boom::boom:\n```\n"+self.path+"\n```") self.respond_cmd_err("Invalid command action") else: # Invalid command/unknown command logger.error("do_POST(): Invalid command/unknown command.") self.bot.debug_chan("do_POST(): Invalid command/unknown command.\n# :boom::boom::boom::boom::boom:\n```\n"+self.path+"\n```") self.respond_cmd_err("Invalid command/unknown command") # always try to fail to retain userinput. If previously responded to, nothing will happen. self.respond(400, if_nonzero_secondary='ignore') except Exception as exc: self.bot.debug_chan("##### Exception in ``"+self.path.strip("/")+"``: ``"+repr(exc)+"``\n# :boom::boom::boom::boom::boom:") logger.exception("do_POST(): Exception in: %s\nRequest-data:%s", self.path.strip("/"), pprint.pformat(data)) self.respond_cmd_err("A serverside error occured. @someone should have been contacted.", http_code=500, if_nonzero_secondary='ignore') # Send a response to the channel. def respond_cmd_chan(self, message, props=None, att=None, http_code=200): data = {"skip_slack_parsing":True, "response_type":"in_channel", "text":message} if props: data.update({"props": props}) else: props={} if att: props.update({"attachments": att}) data.update({"props": props}) self.respond(http_code, data) # Send a ephemeral response to the user. def respond_cmd_temp(self, message, props=None, att=None, http_code=200): data = {"skip_slack_parsing":True, "response_type":"ephemeral", "text":message} if props: data.update({"props": props}) else: props={} if att: props.update({"attachments": att}) data.update({"props": props}) self.respond(http_code, data) def respond_interactive_temp(self, message): # cant be secondary, because no response url (interactive messages only?) self.respond(200, {"skip_slack_parsing":True, "ephemeral_text": message}) # Use to send a failure to the user. Use only the first time during a request. Should retain input on clientside. def respond_cmd_err(self, message, props=None, http_code=400, if_nonzero_secondary='exc'): data = {"skip_slack_parsing":True, "response_type":"ephemeral", "text": "## :x: Failure! :(\n#### "+message} if props: data.update({"props": props}) # must be 2 separatecalls, as the message is ignored in a non-200. self.respond(http_code=http_code, if_nonzero_secondary=if_nonzero_secondary) self.respond(000, data) def respond_interactive_err(self, message): # cant be secondary, because no response url (interactive messages only?) self.respond(200, {"skip_slack_parsing":True, "ephemeral_text":"## :x: Failure! :(\n#### "+message}) def respond(self, http_code=200, data=None, if_nonzero_secondary='exc'): """ First response call must have a valid http code. Secondary responses should have http_code = 0. use if_nonzero_secondary = 'ignore' to ignore response with http_code != 0. use if_nonzero_secondary = 'force' to send secondary with http_code != 0. """ if data is None: data = {} # First response if not self.handled: if http_code >= 600 or http_code < 100: raise Exception("respond(): Primary response must have a valid http code.") self.handled = True self.send_response(http_code) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", len(bytes(json.dumps(data), "utf8"))) self.end_headers() self.wfile.write(bytes(json.dumps(data), "utf8")) logger.debug("respond(): Primary response send.") # Secondary responses else: if http_code != 0 and if_nonzero_secondary == "ignore": logger.info("respond(): Secondary responses must have a zero http code, but if_nonzero_secondary='ignore'. Doing nothing.") return elif http_code != 0 and if_nonzero_secondary == "force": logger.warning("respond(): Secondary responses must have a zero http code, but if_nonzero_secondary='force'. Sending anyway.") traceback.print_stack() elif http_code != 0: raise Exception("respond(): Secondary responses must have a zero http code.") if not self.responseURL: raise Exception("respond(): Secondary response attempt without response url.") logger.debug("respond(): Secondary response. Using responseURL: %s", self.responseURL) req = Request(self.responseURL, data=bytes(json.dumps(data), "utf8"), method='POST') req.add_header("Content-Type", "application/json") conn = urlopen(req, timeout=3) logger.debug("respond(): Secondary response send. Status: %s", conn.status) class MyHTTPServer(ThreadingMixIn, HTTPServer): def serve_forever(self, bot): self.RequestHandlerClass.bot = bot HTTPServer.serve_forever(self) self.httpd = MyHTTPServer((self.local_websrv_hostname, self.local_websrv_port), HTTPRequestHandler) self.httpd.serve_forever(self) ################################# # Bot's public facing webserver # ################################# def start_webserver_public(self): logger.info("Starting public facing webserver.") class HTTPPublicRequestHandler(BaseHTTPRequestHandler): bot = None handled = False def version_string(self): return "SomeBot" def do_POST(self): self.handled = False data = None if self.headers["Content-Type"] == "application/x-www-form-urlencoded": data = urllib.parse.parse_qs(self.rfile.read(int(self.headers["Content-Length"])).decode("utf-8"), keep_blank_values=True) # only accept first occurence data = {k: v[0] for k, v in data.items()} elif self.headers["Content-Type"] == "application/json": data = json.loads(self.rfile.read(int(self.headers["Content-Length"])).decode("utf-8")) # else: # self.respond_public(415) # return logger.debug("do_public_POST(): request incomming.") try: module = 'not yet known' splitpath = self.path.strip("/").split("/") if splitpath[0] in self.bot.modules_public: module = self.bot.modules_public[splitpath[0]] self.bot.command_stats_inc("public-post::"+splitpath[0]) module._on_public_POST(self, data) else: # Invalid command/unknown command logger.error("do_public_POST(): Invalid command/unknown command") #self.bot.debug_chan("do_public_POST(): Invalid command/unknown command.\n# :boom::boom::boom::boom::boom:\n```\n"+self.path+"\n```") self.respond_public(400, {"error":"Invalid command/unknown command"} ,if_nonzero_secondary='ignore') # always try to fail to retain userinput. If previously responded to, nothing will happen. self.respond_public(400, if_nonzero_secondary='ignore') except Exception as exc: self.bot.debug_chan("##### Exception in ``"+self.path.strip("/")+"``: ``"+repr(exc)+"``\n# :boom::boom::boom::boom::boom:") logger.exception("do_public_POST(): Exception in: %s\nRequest-data:%s", self.path.strip("/"), pprint.pformat(data)) self.respond_public(500, {"error":"A serverside error occured. @someone should have been contacted."}, if_nonzero_secondary='ignore') def do_GET(self): self.handled = False data = None logger.debug("do_public_GET(): request incomming.") try: module = 'not yet known' splitpath = self.path.strip("/").split("/") if splitpath[0] in self.bot.modules_public: module = self.bot.modules_public[splitpath[0]] self.bot.command_stats_inc("public-get::"+splitpath[0]) module._on_public_GET(self, data) else: # Invalid command/unknown command logger.error("do_public_GET(): Invalid command/unknown command") #self.bot.debug_chan("do_public_GET(): Invalid command/unknown command.\n# :boom::boom::boom::boom::boom:\n```\n"+self.path+"\n```") self.respond_public(400, {"error":"Invalid command/unknown command"} ,if_nonzero_secondary='ignore') # always try to fail to retain userinput. If previously responded to, nothing will happen. self.respond_public(400, if_nonzero_secondary='ignore') except Exception as exc: self.bot.debug_chan("##### Exception in ``"+self.path.strip("/")+"``: ``"+repr(exc)+"``\n# :boom::boom::boom::boom::boom:") logger.exception("do_public_GET(): Exception in: %s\nRequest-data:%s", self.path.strip("/"), pprint.pformat(data)) self.respond_public(500, {"error":"A serverside error occured. @someone should have been contacted."}, if_nonzero_secondary='ignore') def respond_public(self, http_code=200, data=None, if_nonzero_secondary='exc'): """ First response call must have a valid http code. """ if data is None: data = {} # First response if not self.handled: if http_code >= 600 or http_code < 100: raise Exception("respond_public(): Primary response must have a valid http code.") self.handled = True self.send_response(http_code) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", len(bytes(json.dumps(data), "utf8"))) self.end_headers() self.wfile.write(bytes(json.dumps(data), "utf8")) logger.debug("respond_public(): Primary response send.") # Secondary responses else: if http_code != 0 and if_nonzero_secondary == "ignore": logger.info("respond_public(): Secondary responses must have a zero http code, but if_nonzero_secondary='ignore'. Doing nothing.") return elif if_nonzero_secondary != "ignore": raise Exception("respond_public(): Secondary responses must be ignored.") class MyHTTPServer(ThreadingMixIn, HTTPServer): def serve_forever(self, bot): self.RequestHandlerClass.bot = bot HTTPServer.serve_forever(self) self.httpd = MyHTTPServer((self.local_public_websrv_hostname, self.local_public_websrv_port), HTTPPublicRequestHandler) self.httpd.serve_forever(self)