From 72f3b5d049c05f125e45bf79fcd3bc50f9c969c3 Mon Sep 17 00:00:00 2001 From: Brian Martin Date: Thu, 6 Jun 2019 22:46:40 -0400 Subject: [PATCH] change execution bumper.start/shutdown - allow envionrment variables (prep for docker) - move bumper.db to new "data" directory start_bumper reworked - use argparse --- bumper/__init__.py | 157 +++++++++++++++++++++++++++++++++++++------ bumper/confserver.py | 58 +++++++++------- bumper/mqttserver.py | 6 +- bumper/xmppserver.py | 8 +-- start_bumper.py | 111 ++++++------------------------ tests/test_init.py | 14 ---- 6 files changed, 197 insertions(+), 157 deletions(-) diff --git a/bumper/__init__.py b/bumper/__init__.py index 4280b1b..2d9edf1 100644 --- a/bumper/__init__.py +++ b/bumper/__init__.py @@ -6,6 +6,7 @@ from bumper.xmppserver import XMPPServer import asyncio import json import time +import distutils from datetime import datetime, timedelta import platform import os, sys @@ -13,19 +14,64 @@ import logging from logging.handlers import RotatingFileHandler from base64 import b64decode, b64encode from tinydb import TinyDB, Query -import json from tinydb.storages import MemoryStorage +import socket -ca_cert = "./certs/CA/cacert.pem" -server_cert = "./certs/cert.pem" -server_key = "./certs/key.pem" +def strtobool(strbool): + if str(strbool).lower() in ['true', '1', 't', 'y', 'on','yes']: + return True + else: + return False +# os.environ['PYTHONASYNCIODEBUG'] = '1' # Uncomment to enable ASYNCIODEBUG + +bumper_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), os.pardir)) +# Set defaults from environment variables first +# Certs +ca_cert = os.environ.get("BUMPER_CA") or os.path.join(bumper_dir, "certs/ca.crt") +server_cert = os.environ.get("BUMPER_CERT") or os.path.join( + bumper_dir, "certs/bumper.crt" +) +server_key = os.environ.get("BUMPER_KEY") or os.path.join( + bumper_dir, "certs/bumper.key" +) + +if not ( + os.path.exists(ca_cert) or os.path.exists(server_cert) or os.path.exists(server_key) +): + logging.log(logging.FATAL, "Certificate(s) don't exist at paths specified") + os._exit(1) + +# Folders +logs_dir = os.environ.get("BUMPER_LOGS") or os.path.join(bumper_dir, "logs") +os.makedirs(logs_dir, exist_ok=True) # Ensure logs directory exists or create +data_dir = os.environ.get("BUMPER_DATA") or os.path.join(bumper_dir, "data") +os.makedirs(data_dir, exist_ok=True) # Ensure data directory exists or create + +# Listeners +bumper_listen = os.environ.get("BUMPER_LISTEN") or socket.gethostbyname( + socket.gethostname() +) +if not bumper_listen: + logging.log(logging.FATAL, "No listen address configured!") + os._exit(1) + +bumper_announce_ip = os.environ.get("BUMPER_ANNOUNCE_IP") or bumper_listen + +# Other +bumper_debug = strtobool(os.environ.get("BUMPER_DEBUG")) or False use_auth = False token_validity_seconds = 3600 # 1 hour db = None -# Logs -os.makedirs("logs", exist_ok=True) # Ensure logs directory exists or create +mqtt_server = None +mqtt_helperbot = None +conf_server = None +conf_server_2 = None +xmpp_server = None + +shutting_down = False + # Set format for all logs logformat = logging.Formatter( "[%(asctime)s] :: %(levelname)s :: %(name)s :: %(module)s :: %(funcName)s :: %(lineno)d :: %(message)s" @@ -74,9 +120,90 @@ xmppserverlog.addHandler(xmpp_rotate) # Override the logging level # xmppserverlog.setLevel(logging.INFO) - logging.getLogger("asyncio").setLevel(logging.CRITICAL + 1) # Ignore this logger +async def start(): + + try: + loop = asyncio.get_event_loop() + except: + loop = asyncio.new_event_loop() + + if bumper_debug: + logging.basicConfig( + level=logging.DEBUG, + format="[%(asctime)s] :: %(levelname)s :: %(name)s :: %(module)s :: %(funcName)s :: %(lineno)d :: %(message)s", + ) + loop.set_debug(True) # Set asyncio loop to debug + # logging.getLogger("asyncio").setLevel(logging.DEBUG) # Show debug asyncio logs (disabled in init, uncomment for debugging asyncio) + else: + logging.basicConfig( + level=logging.INFO, + format="[%(asctime)s] :: %(levelname)s :: %(name)s :: %(message)s", + ) + + bumperlog.info("Starting Bumper") + global mqtt_server + mqtt_server = MQTTServer((bumper_listen, 8883)) + global mqtt_helperbot + mqtt_helperbot = MQTTHelperBot((bumper_listen, 8883)) + global conf_server + conf_server = ConfServer( + (bumper_listen, 443), usessl=True, helperbot=mqtt_helperbot + ) + global conf_server_2 + conf_server_2 = ConfServer( + (bumper_listen, 8007), usessl=False, helperbot=mqtt_helperbot + ) + global xmpp_server + xmpp_server = XMPPServer((bumper_listen, 5223)) + + # Start web servers + conf_server.confserver_app() + asyncio.create_task(conf_server.start_server()) + + conf_server_2.confserver_app() + asyncio.create_task(conf_server_2.start_server()) + + # Start MQTT Server + asyncio.create_task(mqtt_server.broker_coro()) + + # Start MQTT Helperbot + asyncio.create_task(mqtt_helperbot.start_helper_bot()) + + # Start XMPP Server + asyncio.create_task(xmpp_server.start_async_server()) + + # Start maintenance + while not shutting_down: + asyncio.create_task(maintenance()) + await asyncio.sleep(30) + + +async def maintenance(): + revoke_expired_tokens() + + +async def shutdown(): + try: + bumperlog.info("Shutting down") + await mqtt_server.broker.shutdown() + xmpp_server.server.close() + await xmpp_server.server.wait_closed() + await conf_server.stop_server() + await conf_server_2.stop_server() + global shutting_down + shutting_down = True + + except asyncio.CancelledError: + bumperlog.info("Coroutine canceled") + + except Exception as e: + bumperlog.info("Exception: {}".format(e)) + + finally: + bumperlog.info("Shutdown complete") + def get_milli_time(timetoconvert): return int(round(timetoconvert * 1000)) @@ -89,20 +216,8 @@ def db_file(): return os_db_path() -def os_db_path(createdir=True): - # createdir - Set to False during tests to not create paths - if platform.system() == "Windows": - if createdir: - os.makedirs( - os.getenv("APPDATA"), exist_ok=True - ) # Ensure db_path directory exists or create - return os.path.join(os.getenv("APPDATA"), "bumper.db") - else: - if createdir: - os.makedirs( - os.path.expanduser("~/.config"), exist_ok=True - ) # Ensure db_path directory exists or create - return os.path.expanduser("~/.config/bumper.db") +def os_db_path(): # createdir=True): + return os.path.join(data_dir, "bumper.db") def db_get(): diff --git a/bumper/confserver.py b/bumper/confserver.py index 5180769..ae5d51d 100644 --- a/bumper/confserver.py +++ b/bumper/confserver.py @@ -63,6 +63,7 @@ class ConfServer: self.confthread = None self.run_async = False self.app = None + self.site = None def confserver_app(self): self.app = web.Application(loop=asyncio.get_event_loop()) @@ -173,7 +174,7 @@ class ConfServer: if self.usessl: ssl_ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH) ssl_ctx.load_cert_chain(bumper.server_cert, bumper.server_key) - site = web.TCPSite( + self.site = web.TCPSite( runner, host=self.address[0], port=self.address[1], @@ -181,9 +182,9 @@ class ConfServer: ) else: - site = web.TCPSite(runner, host=self.address[0], port=self.address[1]) + self.site = web.TCPSite(runner, host=self.address[0], port=self.address[1]) - await site.start() + await self.site.start() except PermissionError as e: if "bind" in e.strerror: @@ -198,6 +199,13 @@ class ConfServer: confserverlog.exception("{}".format(e)) exit(1) + async def stop_server(self): + try: + await self.site.stop() + + except Exception as e: + confserverlog.exception("{}".format(e)) + async def handle_base(self, request): try: # TODO - API Options here for viewing clients, tokens, restarting the server, etc. @@ -877,19 +885,26 @@ class ConfServer: if todo == "FindBest": service = postbody["service"] if service == "EcoMsgNew": - srvip = socket.gethostbyname(socket.gethostname()) + srvip = bumper.bumper_announce_ip srvport = 5223 confserverlog.info( - "Reporting FindBest-EcoMsgNew Server to Bot as: {}:{}".format( + "Announcing EcoMsgNew Server to bot as: {}:{}".format( srvip, srvport ) ) - body = {"result": "ok", "ip": srvip, "port": srvport} + msgserver = {"ip": srvip, "port": srvport, "result": "ok"} + msgserver = json.dumps(msgserver) + msgserver = msgserver.replace( + " ", "" + ) # bot seems to be very picky about having no spaces, only way was with text + + return web.json_response(text=msgserver) + elif service == "EcoUpdate": srvip = "47.88.66.164" # EcoVacs Server srvport = 8005 confserverlog.info( - "Reporting FindBest-EcoUpdate Server to Bot as: {}:{}".format( + "Announcing EcoUpdate Server to bot as: {}:{}".format( srvip, srvport ) ) @@ -937,9 +952,7 @@ class ConfServer: bumper.bot_remove(postbody["did"]) body = {"result": "ok", "todo": "result"} - confserverlog.debug( - "\r\n POST: {} \r\n Response: {}".format(postbody, body) - ) + confserverlog.debug("POST: {} - Response: {}".format(postbody, body)) return web.json_response(body) @@ -982,9 +995,7 @@ class ConfServer: "todo": "result", } - confserverlog.debug( - "\r\n POST: {} \r\n Response: {}".format(postbody, body) - ) + confserverlog.debug("POST: {} - Response: {}".format(postbody, body)) return web.json_response(body) @@ -1012,10 +1023,10 @@ class ConfServer: if todo == "FindBest": service = postbody["service"] if service == "EcoMsgNew": - srvip = socket.gethostbyname(socket.gethostname()) + srvip = bumper.bumper_announce_ip srvport = 5223 confserverlog.info( - "Reporting FindBest-EcoMsgNew Server to Bot as: {}:{}".format( + "Announcing EcoMsgNew Server to bot as: {}:{}".format( srvip, srvport ) ) @@ -1025,17 +1036,18 @@ class ConfServer: " ", "" ) # bot seems to be very picky about having no spaces, only way was with text - confserverlog.debug( - "\r\n POST: {} \r\n Response: {}".format(postbody, msgserver) - ) return web.json_response(text=msgserver) elif service == "EcoUpdate": - body = {"result": "ok", "ip": "47.88.66.164", "port": 8005} + srvip = "47.88.66.164" # EcoVacs Server + srvport = 8005 + confserverlog.info( + "Announcing EcoUpdate Server to bot as: {}:{}".format( + srvip, srvport + ) + ) + body = {"result": "ok", "ip": srvip, "port": srvport} - confserverlog.debug( - "\r\n POST: {} \r\n Response: {}".format(postbody, body) - ) return web.json_response(body) except Exception as e: @@ -1097,7 +1109,7 @@ class ConfServer: body = {"ret": "ok", "logs": []} confserverlog.debug( - "\r\n POST: {} \r\n Response: {}".format(json_body, body) + "POST: {} - Response: {}".format(json_body, body) ) return web.json_response(body) diff --git a/bumper/mqttserver.py b/bumper/mqttserver.py index 4edaca7..fa6d45e 100644 --- a/bumper/mqttserver.py +++ b/bumper/mqttserver.py @@ -300,8 +300,8 @@ class BumperMQTTServer_Plugin: "eco-ng", ) - mqttserverlog.debug( - "new bot authenticated SN: {} DID: {}".format( + mqttserverlog.info( + "bot authenticated SN: {} DID: {}".format( username, didsplit[0] ) ) @@ -324,7 +324,7 @@ class BumperMQTTServer_Plugin: if auth: bumper.client_add(userid, realm, resource) - mqttserverlog.debug( + mqttserverlog.info( "client authenticated {}".format(userid) ) authenticated = True diff --git a/bumper/xmppserver.py b/bumper/xmppserver.py index 453de22..199eb12 100644 --- a/bumper/xmppserver.py +++ b/bumper/xmppserver.py @@ -51,10 +51,6 @@ class XMPPServer: class XMPPServer_Protocol(asyncio.Protocol): client_id = None exit_flag = False - server_cert = "./certs/cert.pem" - server_key = "./certs/key.pem" - ssl_ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH) - ssl_ctx.load_cert_chain(server_cert, server_key) aclient = None def connection_made(self, transport): @@ -506,7 +502,7 @@ class XMPPAsyncClient: if self.devclass: # if there is a devclass it is a bot bumper.bot_add(self.uid, self.uid, self.devclass, "atom", "eco-legacy") self.type = self.BOT - xmppserverlog.debug("bot authenticated {}".format(self.uid)) + xmppserverlog.info("bot authenticated SN: {}".format(self.uid)) # Send response self.send( '' @@ -525,7 +521,7 @@ class XMPPAsyncClient: if auth: self.type = self.CONTROLLER bumper.client_add(self.uid, "bumper", self.clientresource) - xmppserverlog.debug("client authenticated {}".format(self.uid)) + xmppserverlog.info("client authenticated {}".format(self.uid)) # Client authenticated, move to next state self._set_state("INIT") diff --git a/start_bumper.py b/start_bumper.py index 677e219..5bc8356 100644 --- a/start_bumper.py +++ b/start_bumper.py @@ -1,102 +1,33 @@ #!/usr/bin/env python3 -import logging import bumper -import sys, socket -import time -import platform -import os - -# os.environ['PYTHONASYNCIODEBUG'] = '1' # Uncomment to enable ASYNCIODEBUG +import argparse import asyncio - -async def main(): - try: - loop = asyncio.get_event_loop() - except: - loop = asyncio.new_event_loop() - - args = sys.argv - listen_host = "" - - if len(args) > 0: - if "--debug" in args: - logging.basicConfig( - level=logging.DEBUG, - format="[%(asctime)s] :: %(levelname)s :: %(name)s :: %(module)s :: %(funcName)s :: %(lineno)d :: %(message)s", - ) - loop.set_debug(True) # Set asyncio loop to debug - # logging.getLogger("asyncio").setLevel(logging.DEBUG) # Show debug asyncio logs (disabled in init, uncomment for debugging asyncio) - else: - logging.basicConfig( - level=logging.INFO, - format="[%(asctime)s] :: %(levelname)s :: %(name)s :: %(message)s", - ) - - if "--listen" in args: - listen_host = args[args.index("--listen") + 1] - - if listen_host == "": - if platform.system() == "Darwin": # If a Mac, use 0.0.0.0 for listening - listen_host = "0.0.0.0" - else: - listen_host = socket.gethostbyname(socket.gethostname()) - - conf_address_443 = (listen_host, 443) - conf_address_8007 = (listen_host, 8007) - xmpp_address = (listen_host, 5223) - mqtt_address = (listen_host, 8883) - - mqtt_server = bumper.MQTTServer(mqtt_address) - mqtt_helperbot = bumper.MQTTHelperBot(mqtt_address) - conf_server = bumper.ConfServer( - conf_address_443, usessl=True, helperbot=mqtt_helperbot - ) - conf_server_2 = bumper.ConfServer( - conf_address_8007, usessl=False, helperbot=mqtt_helperbot - ) - xmpp_server = bumper.XMPPServer(xmpp_address) - - try: - # Start web servers - conf_server.confserver_app() - asyncio.create_task(conf_server.start_server()) - - conf_server_2.confserver_app() - asyncio.create_task(conf_server_2.start_server()) - - # Start MQTT Server - asyncio.create_task(mqtt_server.broker_coro()) - - # Start MQTT Helperbot - asyncio.create_task(mqtt_helperbot.start_helper_bot()) - - # Start XMPP Server - asyncio.create_task(xmpp_server.start_async_server()) - - maintain = asyncio.create_task(maintenance_tasks()) - await maintain # Keeps the loop running until this exits - - finally: - # Cleanup and close tasks/loop - for task in asyncio.Task.all_tasks(): - task.cancel() - loop.close() - - -async def maintenance_tasks(): - while True: - await asyncio.sleep(30) # Sleep 30 seconds - bumper.revoke_expired_tokens() - - if __name__ == "__main__": try: - asyncio.run(main()) + parser = argparse.ArgumentParser() + parser.add_argument("--listen", type=str, default=None, help="listen address") + parser.add_argument( + "--announce", type=str, default=None, help="announce address (for bot)" + ) + parser.add_argument("--debug", action="store_true") + args = parser.parse_args() + + if args.debug: + bumper.bumper_debug = True + + if args.listen: + bumper.bumper_listen = args.listen + + if args.announce: + bumper.bumper_announce_ip = args.announce + + asyncio.run(bumper.start()) except KeyboardInterrupt: bumper.bumperlog.info("Keyboard Interrupt!") pass + finally: - bumper.bumperlog.info("Bumper Exiting!") + asyncio.run(bumper.shutdown()) diff --git a/tests/test_init.py b/tests/test_init.py index d1e9060..3a38fae 100644 --- a/tests/test_init.py +++ b/tests/test_init.py @@ -23,20 +23,6 @@ def test_user_db(): if os.path.exists("tests/tmp.db"): os.remove("tests/tmp.db") # Remove existing db - # Test os_db_path - platform.system = mock.MagicMock(return_value="Windows") - p = platform.system() - os.getenv = mock.MagicMock(return_value="C:\\AppData") - o = os.getenv("APPDATA") - assert bumper.os_db_path(createdir=False) == os.path.join( - os.getenv("APPDATA"), "bumper.db" - ) - - platform.system = mock.MagicMock(return_value="Linux") - assert bumper.os_db_path(createdir=False) == os.path.expanduser( - "~/.config/bumper.db" - ) - bumper.db = "tests/tmp.db" # Set db location for testing bumper.user_add("testuser") # Add testuser