From 6fb5e99e0623cc42e67975cf12ced5be7a5d598b Mon Sep 17 00:00:00 2001 From: Robert Resch Date: Sat, 20 Aug 2022 12:22:31 +0200 Subject: [PATCH 01/10] add proxy mode --- bumper/__init__.py | 27 ++++++- bumper/db.py | 14 ++++ bumper/dns.py | 11 +++ bumper/mqtt/helper_bot.py | 2 +- bumper/mqtt/proxy.py | 89 +++++++++++++++++++++++ bumper/mqtt/server.py | 94 +++++++++++++++++++++++- bumper/web/server.py | 149 +++++++++++++++++++++++++++++--------- requirements.txt | 2 + tests/conftest.py | 2 +- tests/web/test_server.py | 6 +- 10 files changed, 351 insertions(+), 45 deletions(-) create mode 100644 bumper/dns.py create mode 100644 bumper/mqtt/proxy.py diff --git a/bumper/__init__.py b/bumper/__init__.py index ad3e3be..34ffc95 100644 --- a/bumper/__init__.py +++ b/bumper/__init__.py @@ -3,9 +3,13 @@ import logging import os import socket import sys -from typing import Any -from bumper.db import revoke_expired_oauths, revoke_expired_tokens +from bumper.db import ( + bot_reset_connectionStatus, + client_reset_connectionStatus, + revoke_expired_oauths, + revoke_expired_tokens, +) from bumper.mqtt.helper_bot import HelperBot from bumper.mqtt.server import MQTTServer from bumper.util import get_logger, log_to_stdout @@ -50,6 +54,7 @@ bumper_debug = strtobool(os.environ.get("BUMPER_DEBUG")) or False use_auth = False token_validity_seconds = 3600 # 1 hour oauth_validity_days = 15 +bumper_proxy_mode = strtobool(os.environ.get("BUMPER_PROXY_MODE")) or False mqtt_server: MQTTServer mqtt_helperbot: HelperBot @@ -60,6 +65,7 @@ shutting_down = False bumperlog = get_logger("bumper") logging.getLogger("asyncio").setLevel(logging.CRITICAL + 1) # Ignore this logger +proxymodelog = get_logger("proxymode") web_server_https_port = os.environ.get("WEB_SERVER_HTTPS_PORT") or 443 mqtt_listen_port = 8883 @@ -71,6 +77,10 @@ web_server_bindings = [ async def start() -> None: + # Reset xmpp/mqtt to false in database for bots and clients + bot_reset_connectionStatus() + client_reset_connectionStatus() + try: loop = asyncio.get_event_loop() except: @@ -102,12 +112,16 @@ async def start() -> None: return bumperlog.info("Starting Bumper") + + if bumper_proxy_mode: + bumperlog.info("Proxy Mode Enabled") + global mqtt_server mqtt_server = MQTTServer(bumper_listen, mqtt_listen_port) global mqtt_helperbot mqtt_helperbot = HelperBot(bumper_listen, mqtt_listen_port) global web_server - web_server = WebServer(web_server_bindings) + web_server = WebServer(web_server_bindings, bumper_proxy_mode) global xmpp_server xmpp_server = XMPPServer(bumper_listen, xmpp_listen_port) @@ -194,12 +208,19 @@ def main(argv: None | list[str] = None) -> None: help="announce address to bots on checkin", ) parser.add_argument("--debug", action="store_true", help="enable debug logs") + parser.add_argument( + "--proxy-mode", action="store_true", help="enable proxy mode" + ) args = parser.parse_args(args=argv) if args.debug: bumper_debug = True + if args.proxy_mode: + global bumper_proxy_mode + bumper_proxy_mode = True + if args.listen: bumper_listen = args.listen diff --git a/bumper/db.py b/bumper/db.py index b73e89f..cfbd006 100644 --- a/bumper/db.py +++ b/bumper/db.py @@ -415,3 +415,17 @@ def client_set_xmpp(resource: str, xmpp: bool) -> None: clients = db_get().table("clients") Client = Query() clients.upsert({"xmpp_connection": xmpp}, Client.resource == resource) + + +def bot_reset_connectionStatus() -> None: + bots = db_get().table("bots") + for bot in bots: + bot_set_mqtt(bot["did"], False) + bot_set_xmpp(bot["did"], False) + + +def client_reset_connectionStatus() -> None: + clients = db_get().table("clients") + for client in clients: + client_set_mqtt(client["resource"], False) + client_set_xmpp(client["resource"], False) diff --git a/bumper/dns.py b/bumper/dns.py new file mode 100644 index 0000000..e480de3 --- /dev/null +++ b/bumper/dns.py @@ -0,0 +1,11 @@ +from aiohttp import AsyncResolver + + +def get_resolver_with_public_nameserver() -> AsyncResolver: + # requires aiodns + return AsyncResolver(nameservers=["1.1.1.1", "8.8.8.8"]) + + +async def resolve(host: str) -> str: + hosts = await get_resolver_with_public_nameserver().resolve(host) + return hosts[0]["host"] diff --git a/bumper/mqtt/helper_bot.py b/bumper/mqtt/helper_bot.py index 4ce7060..4dd6ccb 100644 --- a/bumper/mqtt/helper_bot.py +++ b/bumper/mqtt/helper_bot.py @@ -46,7 +46,7 @@ class HelperBot: self._port = port self._client_id = "helperbot@bumper/helperbot" self._timeout = timeout - self._client = Client("helperbot@bumper/helperbot") + self._client = Client(self._client_id) # pylint: disable=unused-argument async def _on_message( diff --git a/bumper/mqtt/proxy.py b/bumper/mqtt/proxy.py new file mode 100644 index 0000000..8b15711 --- /dev/null +++ b/bumper/mqtt/proxy.py @@ -0,0 +1,89 @@ +"""Mqtt proxy module.""" +import asyncio +import re +from typing import Any, MutableMapping + +from amqtt.client import MQTTClient +from amqtt.mqtt.constants import QOS_0, QOS_1, QOS_2 +from cachetools import TTLCache + +from ..util import get_logger + +_LOGGER = get_logger("proxymode") + +# iot/p2p/[command]]/[sender did]/[sender class]]/[sender resource] +# /[receiver did]/[receiver class]]/[receiver resource]/[q|p/[request id/j +# [q|p] q-> request p-> response + +TOPIC_P2P = re.compile( + "iot/p2p/(?P[^/]+)/(?P[^/]+)/(?P[^/]+)/(?P[^/]+)/" + "/(?P[^/]+)/(?P[^/]+)/(?P[^/]+)/(?P[^/]+)/" + "(?P[^/]+)/(?P[^/]+)" +) + + +class ProxyClient: + """Mqtt client, which proxies all messages to the ecovacs servers.""" + + def __init__( + self, + client_id: str, + host: str, + port: int = 443, + config: dict[str, Any] | None = None, + timeout: float = 180, + ): + self.request_mapper: MutableMapping[str, str] = TTLCache( + maxsize=timeout * 60, ttl=timeout * 1.1 + ) + self._client = MQTTClient(client_id=client_id, config=config) + self._host = host + self._port = port + + async def connect(self, username: str, password: str) -> None: + try: + await self._client.connect( + f"mqtts://{username}:{password}@{self._host}:{self._port}" + ) + except Exception: + _LOGGER.exception("An exception occurred during startup", exc_info=True) + raise + + asyncio.create_task(self._handle_messages()) + + async def _handle_messages(self) -> None: + while self._client.session.transitions.is_connected(): + try: + message = await self._client.deliver_message() + data = str(message.data.decode("utf-8")) + + _LOGGER.info( + f"MQTT Proxy Client - Message Received From Ecovacs - Topic: {message.topic} - Message: {data}" + ) + topic = message.topic + ttopic = topic.split("/") + if ttopic[1] == "p2p": + self.request_mapper[ttopic[10]] = ttopic[3] + ttopic[3] = "proxyhelper" + topic = "/".join(ttopic) + _LOGGER.info( + f"MQTT Proxy Client - Converted Topic From {message.topic} TO {topic}" + ) + + _LOGGER.info( + f"MQTT Proxy Client - Proxy Forward Message to Robot - Topic: {topic} - Message: {data}" + ) + await self._client.publish(topic, data.encode(), QOS_0) + except Exception: # pylint: disable=broad-except + _LOGGER.error( + "An error occurred during handling a message", exc_info=True + ) + + async def subscribe(self, topic: str, qos: QOS_0 | QOS_1 | QOS_2 = QOS_0) -> None: + await self._client.subscribe([(topic, qos)]) + + async def disconnect(self) -> None: + await self._client.disconnect() + + async def publish(self, topic: str, message: bytes, qos: int | None = None) -> None: + await self._client.publish(topic, message, qos) diff --git a/bumper/mqtt/server.py b/bumper/mqtt/server.py index 3e88255..a2614f0 100644 --- a/bumper/mqtt/server.py +++ b/bumper/mqtt/server.py @@ -1,15 +1,16 @@ """Server module.""" - import os from typing import Any import amqtt import pkg_resources from amqtt.broker import Broker, BrokerContext +from amqtt.mqtt.constants import QOS_0, QOS_1, QOS_2 from amqtt.session import IncomingApplicationMessage, Session from passlib.apps import custom_app_context as pwd_context import bumper +from bumper import dns from bumper.db import ( bot_add, bot_get, @@ -19,11 +20,13 @@ from bumper.db import ( client_get, client_set_mqtt, ) +from bumper.mqtt.proxy import ProxyClient from bumper.util import get_logger mqttserverlog = get_logger("mqttserver") helperbotlog = get_logger("helperbot") boterrorlog = get_logger("boterror") +proxymodelog = get_logger("proxymode") class MQTTServer: @@ -122,6 +125,7 @@ class BumperMQTTServerPlugin: """MQTT Server plugin which handles the authentication.""" def __init__(self, context: BrokerContext) -> None: + self._proxy_clients: dict[str, ProxyClient] = {} self.context = context try: self.auth_config = self.context.config["auth"] @@ -163,6 +167,32 @@ class BumperMQTTServerPlugin: didsplit[0], tmpbotdetail[0], ) + + if bumper.bumper_proxy_mode: + mqtt_server = await dns.resolve("mq-ww.ecouser.net") + if mqtt_server: + proxymodelog.info( + f"MQTT Proxy Mode - Using server {mqtt_server}" + ) + else: + proxymodelog.error( + "MQTT Proxy Mode - No server found! Load defaults or " + "set mqtt_server in config_proxymode table!" + ) + proxymodelog.exception( + f"MQTT Proxy Mode - Exiting due to no MQTT Server configured!" + ) + exit(1) + + proxymodelog.info( + f"MQTT Proxy Mode - Proxy Bot to MQTT - Client_id: {client_id} - Username: {username}" + ) + proxy = ProxyClient( + client_id, mqtt_server, config={"check_hostname": False} + ) + self._proxy_clients[client_id] = proxy + await proxy.connect(username, password) + return True tmpclientdetail = str(didsplit[1]).split("/") @@ -242,6 +272,21 @@ class BumperMQTTServerPlugin: return users + async def on_broker_client_subscribed( + self, client_id: str, topic: str, qos: QOS_0 | QOS_1 | QOS_2 + ) -> None: + if bumper.bumper_proxy_mode: + # if proxy mode, also subscribe on ecovacs server + if client_id in self._proxy_clients: + await self._proxy_clients[client_id].subscribe(topic, qos) + proxymodelog.info( + f"MQTT Proxy Mode - New MQTT Topic Subscription - Client: {client_id} - Topic: {topic}" + ) + else: + proxymodelog.warning( + f"MQTT Proxy Mode - No proxy client found! - Client: {client_id} - Topic: {topic}" + ) + async def on_broker_client_connected(self, client_id: str) -> None: """On client connected.""" self._set_client_connected(client_id, True) @@ -262,12 +307,13 @@ class BumperMQTTServerPlugin: client_set_mqtt(client["resource"], connected) async def on_broker_message_received( # pylint: disable=no-self-use - self, message: IncomingApplicationMessage, **_: dict[str, Any] + self, message: IncomingApplicationMessage, client_id: str ) -> None: """On message received.""" topic = message.topic topic_split = str(topic).split("/") data_decoded = str(message.data.decode("utf-8")) + if topic_split[6] == "helperbot": # Response to command _log__helperbot_message("Received Response", topic, data_decoded) @@ -285,6 +331,50 @@ class BumperMQTTServerPlugin: else: _log__helperbot_message("Received Message", topic, data_decoded) + if bumper.bumper_proxy_mode and client_id in self._proxy_clients: + if not topic_split[3] == "proxyhelper": + # if from proxyhelper, don't send back to ecovacs...yet + if topic_split[6] == "proxyhelper": + ttopic = message.topic.split("/") + ttopic[6] = self._proxy_clients[client_id].request_mapper.pop( + ttopic[10], "" + ) + if ttopic[6] == "": + proxymodelog.warning( + "MQTT Proxy Client - Request mapper is missing entry, " + f"probably request took to long... Client_id: {client_id}" + f" - Request_id: {ttopic[10]}" + ) + return + + ttopic_join = "/".join(ttopic) + proxymodelog.info( + f"MQTT Proxy Client - Bot Message Converted Topic From {message.topic} TO {ttopic_join} " + f"with message: {data_decoded}" + ) + else: + ttopic_join = message.topic + proxymodelog.info( + f"MQTT Proxy Client - Bot Message From {ttopic_join} with message: {data_decoded}" + ) + + try: + # Send back to ecovacs + proxymodelog.info( + "MQTT Proxy Client - Proxy Forward Message to Ecovacs - Topic:" + f" {ttopic_join} - Message: {data_decoded}" + ) + await self._proxy_clients[client_id].publish( + ttopic_join, data_decoded.encode(), message.qos + ) + except Exception: # pylint: disable=broad-except + proxymodelog.error( + "MQTT Proxy Client - Forwarding to Ecovacs - Exception", + exc_info=True, + ) + async def on_broker_client_disconnected(self, client_id: str) -> None: """On client disconnect.""" + if bumper.bumper_proxy_mode and client_id in self._proxy_clients: + await self._proxy_clients.pop(client_id).disconnect() self._set_client_connected(client_id, False) diff --git a/bumper/web/server.py b/bumper/web/server.py index cc231cb..c506d44 100644 --- a/bumper/web/server.py +++ b/bumper/web/server.py @@ -1,5 +1,4 @@ """Web server module.""" - import asyncio import dataclasses import json @@ -7,20 +6,18 @@ import logging import os import ssl +import aiohttp import aiohttp_jinja2 import jinja2 from aiohttp import web from aiohttp.typedefs import Handler -from aiohttp.web_exceptions import ( - HTTPBadRequest, - HTTPInternalServerError, - HTTPNoContent, -) +from aiohttp.web_exceptions import HTTPInternalServerError, HTTPNoContent from aiohttp.web_request import Request from aiohttp.web_response import Response, StreamResponse import bumper from bumper.db import bot_get, bot_remove, client_get, client_remove, db_get +from bumper.dns import get_resolver_with_public_nameserver from bumper.util import get_logger from bumper.web.plugins import add_plugins @@ -41,6 +38,7 @@ class _aiohttp_filter(logging.Filter): confserverlog = get_logger("confserver") # Add logging filter above to aiohttp.access logging.getLogger("aiohttp.access").addFilter(_aiohttp_filter()) +proxymodelog = logging.getLogger("proxymode") @dataclasses.dataclass(frozen=True) @@ -57,7 +55,9 @@ class WebServer: _EXCLUDE_FROM_LOGGING = ["base", "remove-bot", "remove-client", "restart-service"] - def __init__(self, bindings: list[WebserverBinding] | WebserverBinding): + def __init__( + self, bindings: list[WebserverBinding] | WebserverBinding, proxy_mode: bool + ): self._runners: list[web.AppRunner] = [] if isinstance(bindings, WebserverBinding): @@ -75,32 +75,39 @@ class WebServer: os.path.join(bumper.bumper_dir, "bumper", "web", "templates") ), ) - self._add_routes() + self._add_routes(proxy_mode) self._app.freeze() # no modification allowed anymore - def _add_routes(self) -> None: + def _add_routes(self, proxy_mode: bool) -> None: self._app.add_routes( [ - web.get("", self._handle_base, name="base"), - web.get( - "/bot/remove/{did}", self._handle_remove_bot, name="remove-bot" - ), + web.get("/bot/remove/{did}", self._handle_remove_bot), web.get( "/client/remove/{resource}", self._handle_remove_client, - name="remove-client", ), web.get( "/restart_{service}", self._handle_restart_service, - name="restart-service", ), - web.post("/lookup.do", self._handle_lookup), - web.post("/newauth.do", self._handle_newauth), ] ) - add_plugins(self._app) + if proxy_mode: + self._app.add_routes( + [ + web.route("*", "/{path:.*}", self._handle_proxy), + ] + ) + else: + self._app.add_routes( + [ + web.get("", self._handle_base), + web.post("/lookup.do", self._handle_lookup), + web.post("/newauth.do", self._handle_newauth), + ] + ) + add_plugins(self._app) async def start(self) -> None: """Start server.""" @@ -189,22 +196,11 @@ class WebServer: } } try: - postbody = None if request.content_length: - if request.content_type == "application/x-www-form-urlencoded": - postbody = await request.post() - - elif request.content_type == "application/json": - try: - postbody = json.loads(await request.text()) - except Exception as e: - confserverlog.error(f"Request body not json: {e}") - raise HTTPBadRequest(reason="Body was not json") - + if request.content_type == "application/json": + to_log["request"]["body"] = await request.text() else: - postbody = await request.post() - - to_log["request"]["body"] = f"{postbody}" + to_log["request"]["body"] = f"{await request.post()}" response = await handler(request) if response is None: @@ -217,13 +213,17 @@ class WebServer: to_log["response"] = { "status": f"{response.status}", } + if ( - "application/octet-stream" not in response.content_type - and isinstance(response, Response) + isinstance(response, Response) and response.body + and ( + response.content_type.startswith("text") + or response.content_type == "application/json" + ) ): assert isinstance(response.body, bytes) - to_log["response"]["body"] = f"{json.loads(response.body)}" + to_log["response"]["body"] = response.body.decode("utf-8") confserverlog.debug(json.dumps(to_log)) @@ -365,3 +365,82 @@ class WebServer: confserverlog.exception(f"{e}") raise HTTPInternalServerError + + async def _handle_proxy(self, request: Request) -> Response: + try: + if request.raw_path == "/": + return await self._handle_base(request) + if request.raw_path == "/lookup.do": + return await self._handle_lookup(request) + # use bumper to handle lookup so bot gets Bumper IP and not Ecovacs + + async with aiohttp.ClientSession( + headers=request.headers, + connector=aiohttp.TCPConnector( + verify_ssl=False, resolver=get_resolver_with_public_nameserver() + ), + ) as session: + if request.content.total_bytes > 0: + read_body = await request.read() + proxymodelog.info( + f"HTTP Proxy Request to EcoVacs (body=true) (URL:{request.url}) - {read_body}" + ) + if request.content_type == "application/x-www-form-urlencoded": + # android apps use form + fdata = await request.post() + async with session.request( + request.method, request.url, data=fdata + ) as resp: + response = await resp.text() + proxymodelog.info( + f"HTTP Proxy Response from EcoVacs (URL: {request.url}) - (Status: {resp.status}) - {response}" + ) + else: + # handle json + jdata = read_body.decode("utf8") + jdata = json.loads(jdata) + async with session.request( + request.method, request.url, json=jdata + ) as resp: + response = await resp.text() + proxymodelog.info( + f"HTTP Proxy Response from EcoVacs (URL: {request.url}) - (Status: {resp.status}) - {response}" + ) + + else: + proxymodelog.info( + f"HTTP Proxy Request to EcoVacs (body=false) (URL:{request.url})" + ) + async with session.request(request.method, request.url) as resp: + if resp.content_type == "application/octet-stream": + response = await resp.read() + proxymodelog.info( + f"HTTP Proxy Response from EcoVacs (URL: {request.url}) - (Status: {resp.status}) - " + ) + else: + response = await resp.text() + proxymodelog.info( + f"HTTP Proxy Response from EcoVacs (URL: {request.url}) - (Status: {resp.status}) - {response}" + ) + + if resp.status == 200: + if resp.content_type == "application/json": + response = json.loads(response) + return web.json_response(response) + elif resp.content_type == "application/octet-stream": + return web.Response(body=response) + else: + return web.Response(text=response) + + else: + return web.Response(text=response) + + except asyncio.CancelledError: + proxymodelog.exception( + f"Request cancelled or timeout - {request.url} - {jdata}", exc_info=True + ) + + except Exception: + proxymodelog.exception("An exception occurred", exc_info=True) + + raise HTTPInternalServerError diff --git a/requirements.txt b/requirements.txt index 9c78c3a..e9b4061 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,3 +1,4 @@ +aiodns==3.0.0 aiohttp==3.8.1 aiohttp-jinja2==1.5 cachetools==5.2.0 @@ -5,3 +6,4 @@ git+https://github.com/Yakifo/amqtt@master#amqtt==11.0.0 gmqtt==0.6.11 Jinja2==3.1.2 tinydb==4.7.0 +websockets==10.3 diff --git a/tests/conftest.py b/tests/conftest.py index 30176be..0e6b9d9 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -54,7 +54,7 @@ async def helper_bot(mqtt_server: MQTTServer): @pytest.fixture async def webserver_client(aiohttp_client): - webserver = bumper.WebServer(WebserverBinding(HOST, WEBSERVER_PORT, False)) + webserver = bumper.WebServer(WebserverBinding(HOST, WEBSERVER_PORT, False), False) client = await aiohttp_client(webserver._app) yield client diff --git a/tests/web/test_server.py b/tests/web/test_server.py index 943b056..47464c5 100644 --- a/tests/web/test_server.py +++ b/tests/web/test_server.py @@ -12,7 +12,7 @@ from tests import HOST, MQTT_PORT, WEBSERVER_PORT def create_webserver(): - return WebServer(WebserverBinding(HOST, WEBSERVER_PORT, False)) + return WebServer(WebserverBinding(HOST, WEBSERVER_PORT, False), False) def async_return(result): @@ -27,12 +27,12 @@ def remove_existing_db(): async def test_webserver_ssl(): - webserver = WebServer(WebserverBinding(HOST, WEBSERVER_PORT, True)) + webserver = WebServer(WebserverBinding(HOST, WEBSERVER_PORT, True), False) await webserver.start() async def test_webserver_no_ssl(): - webserver = WebServer(WebserverBinding(HOST, 11112, False)) + webserver = WebServer(WebserverBinding(HOST, 11112, False), False) await webserver.start() From 308a590e8e6dac67fc743f6a17ceef48a68f59e5 Mon Sep 17 00:00:00 2001 From: Robert Resch Date: Sat, 20 Aug 2022 13:19:38 +0200 Subject: [PATCH 02/10] improve request logging --- bumper/web/middlewares.py | 81 +++++++++++++++++++++++++++++++++++++++ bumper/web/server.py | 71 ++-------------------------------- 2 files changed, 85 insertions(+), 67 deletions(-) create mode 100644 bumper/web/middlewares.py diff --git a/bumper/web/middlewares.py b/bumper/web/middlewares.py new file mode 100644 index 0000000..f5e3b00 --- /dev/null +++ b/bumper/web/middlewares.py @@ -0,0 +1,81 @@ +"""Web server middleware module.""" +import json + +from aiohttp import web +from aiohttp.typedefs import Handler +from aiohttp.web_exceptions import HTTPNoContent +from aiohttp.web_request import Request +from aiohttp.web_response import Response, StreamResponse + +from bumper.util import get_logger + +_LOGGER = get_logger("webserver_requests") + + +class CustomEncoder(json.JSONEncoder): + def default(self, obj): + if isinstance(obj, set): + return list(obj) + return json.JSONEncoder.default(self, obj) + +_EXCLUDE_FROM_LOGGING = ["base", "remove-bot", "remove-client", "restart-service"] + + +@web.middleware +async def log_all_requests( + request: Request, handler: Handler +) -> StreamResponse: + if request.match_info.route.name not in _EXCLUDE_FROM_LOGGING: + to_log = { + "request": { + "method": request.method, + "url": str(request.url), + "path": request.path, + "query_string": request.query_string, + "headers": {h for h in request.headers.items()}, + "route_resource": request.match_info.route.resource.canonical, + } + } + try: + if request.content_length: + if request.content_type == "application/json": + to_log["request"]["body"] = await request.json() + else: + to_log["request"]["body"] = {h for h in await request.post()} + + response = await handler(request) + if response is None: + confserverlog.warning( # type:ignore[unreachable] + "Response was null!" + ) + _LOGGER.warning(json.dumps(to_log, cls=CustomEncoder)) + raise HTTPNoContent + + to_log["response"] = { + "status": f"{response.status}", + } + + if ( + isinstance(response, Response) + and response.body + ): + if response.content_type == "application/json": + to_log["response"]["body"] = json.loads(response.text) + elif response.content_type.startswith("text"): + to_log["response"]["body"] = response.text + + return response + + except web.HTTPNotFound: + _LOGGER.debug(f"Request path {request.raw_path} not found") + raise + + except Exception: + _LOGGER.exception("An exception occurred in the logging middleware.", exc_info=True) + raise + + finally: + _LOGGER.debug(json.dumps(to_log, cls=CustomEncoder)) + + else: + return await handler(request) diff --git a/bumper/web/server.py b/bumper/web/server.py index c506d44..9ba1f9b 100644 --- a/bumper/web/server.py +++ b/bumper/web/server.py @@ -10,15 +10,15 @@ import aiohttp import aiohttp_jinja2 import jinja2 from aiohttp import web -from aiohttp.typedefs import Handler -from aiohttp.web_exceptions import HTTPInternalServerError, HTTPNoContent +from aiohttp.web_exceptions import HTTPInternalServerError from aiohttp.web_request import Request -from aiohttp.web_response import Response, StreamResponse +from aiohttp.web_response import Response import bumper from bumper.db import bot_get, bot_remove, client_get, client_remove, db_get from bumper.dns import get_resolver_with_public_nameserver from bumper.util import get_logger +from bumper.web.middlewares import log_all_requests from bumper.web.plugins import add_plugins @@ -53,8 +53,6 @@ class WebserverBinding: class WebServer: """Web server.""" - _EXCLUDE_FROM_LOGGING = ["base", "remove-bot", "remove-client", "restart-service"] - def __init__( self, bindings: list[WebserverBinding] | WebserverBinding, proxy_mode: bool ): @@ -66,7 +64,7 @@ class WebServer: self._app = web.Application( middlewares=[ - self._log_all_requests, + log_all_requests, ], ) aiohttp_jinja2.setup( @@ -180,67 +178,6 @@ class WebServer: raise HTTPInternalServerError - @web.middleware - async def _log_all_requests( - self, request: Request, handler: Handler - ) -> StreamResponse: - if request.match_info.route.name not in self._EXCLUDE_FROM_LOGGING: - to_log = { - "request": { - "route_name": f"{request.match_info.route.name}", - "method": f"{request.method}", - "path": f"{request.path}", - "query_string": f"{request.query_string}", - "raw_path": f"{request.raw_path}", - "raw_headers": f'{",".join(map("{}".format, request.raw_headers))}', - } - } - try: - if request.content_length: - if request.content_type == "application/json": - to_log["request"]["body"] = await request.text() - else: - to_log["request"]["body"] = f"{await request.post()}" - - response = await handler(request) - if response is None: - confserverlog.warning( # type:ignore[unreachable] - "Response was null!" - ) - confserverlog.warning(json.dumps(to_log)) - raise HTTPNoContent - - to_log["response"] = { - "status": f"{response.status}", - } - - if ( - isinstance(response, Response) - and response.body - and ( - response.content_type.startswith("text") - or response.content_type == "application/json" - ) - ): - assert isinstance(response.body, bytes) - to_log["response"]["body"] = response.body.decode("utf-8") - - confserverlog.debug(json.dumps(to_log)) - - return response - - except web.HTTPNotFound as notfound: - confserverlog.debug(f"Request path {request.raw_path} not found") - confserverlog.debug(json.dumps(to_log)) - return notfound - - except Exception as e: - confserverlog.exception(f"{e}") - confserverlog.error(json.dumps(to_log)) - raise e - - else: - return await handler(request) async def _restart_helper_bot(self) -> None: await bumper.mqtt_helperbot.disconnect() From 574e91bb38d7b8e9480eaf4e92f592f5e2341112 Mon Sep 17 00:00:00 2001 From: Robert Resch Date: Sat, 20 Aug 2022 13:58:20 +0200 Subject: [PATCH 03/10] fix "UnboundLocalError" --- bumper/web/middlewares.py | 14 ++++++-------- bumper/web/server.py | 3 +-- 2 files changed, 7 insertions(+), 10 deletions(-) diff --git a/bumper/web/middlewares.py b/bumper/web/middlewares.py index f5e3b00..3731447 100644 --- a/bumper/web/middlewares.py +++ b/bumper/web/middlewares.py @@ -18,13 +18,12 @@ class CustomEncoder(json.JSONEncoder): return list(obj) return json.JSONEncoder.default(self, obj) + _EXCLUDE_FROM_LOGGING = ["base", "remove-bot", "remove-client", "restart-service"] @web.middleware -async def log_all_requests( - request: Request, handler: Handler -) -> StreamResponse: +async def log_all_requests(request: Request, handler: Handler) -> StreamResponse: if request.match_info.route.name not in _EXCLUDE_FROM_LOGGING: to_log = { "request": { @@ -55,10 +54,7 @@ async def log_all_requests( "status": f"{response.status}", } - if ( - isinstance(response, Response) - and response.body - ): + if isinstance(response, Response) and response.body: if response.content_type == "application/json": to_log["response"]["body"] = json.loads(response.text) elif response.content_type.startswith("text"): @@ -71,7 +67,9 @@ async def log_all_requests( raise except Exception: - _LOGGER.exception("An exception occurred in the logging middleware.", exc_info=True) + _LOGGER.exception( + "An exception occurred in the logging middleware.", exc_info=True + ) raise finally: diff --git a/bumper/web/server.py b/bumper/web/server.py index 9ba1f9b..d1c633c 100644 --- a/bumper/web/server.py +++ b/bumper/web/server.py @@ -178,7 +178,6 @@ class WebServer: raise HTTPInternalServerError - async def _restart_helper_bot(self) -> None: await bumper.mqtt_helperbot.disconnect() asyncio.create_task(bumper.mqtt_helperbot.start()) @@ -374,7 +373,7 @@ class WebServer: except asyncio.CancelledError: proxymodelog.exception( - f"Request cancelled or timeout - {request.url} - {jdata}", exc_info=True + f"Request cancelled or timeout - {request.url}", exc_info=True ) except Exception: From 8d7fd6c7590ae91c17269771dc978b16d0667c86 Mon Sep 17 00:00:00 2001 From: Robert Resch Date: Sat, 20 Aug 2022 14:03:20 +0200 Subject: [PATCH 04/10] raise CancelledError --- bumper/web/server.py | 1 + 1 file changed, 1 insertion(+) diff --git a/bumper/web/server.py b/bumper/web/server.py index d1c633c..8e8d544 100644 --- a/bumper/web/server.py +++ b/bumper/web/server.py @@ -375,6 +375,7 @@ class WebServer: proxymodelog.exception( f"Request cancelled or timeout - {request.url}", exc_info=True ) + raise except Exception: proxymodelog.exception("An exception occurred", exc_info=True) From 8abf7e0f02cdd4938e97185f1d1549baa33ecffa Mon Sep 17 00:00:00 2001 From: Robert Resch Date: Sat, 20 Aug 2022 14:10:42 +0200 Subject: [PATCH 05/10] ignore helperbot --- bumper/mqtt/helper_bot.py | 6 ++++-- bumper/mqtt/server.py | 3 ++- 2 files changed, 6 insertions(+), 3 deletions(-) diff --git a/bumper/mqtt/helper_bot.py b/bumper/mqtt/helper_bot.py index 4dd6ccb..bcb62de 100644 --- a/bumper/mqtt/helper_bot.py +++ b/bumper/mqtt/helper_bot.py @@ -35,6 +35,9 @@ class CommandDto: self._event.set() +HELPER_BOT_CLIENT_ID = "helperbot@bumper/helperbot" + + class HelperBot: """Helper bot, which converts commands from the rest api to mqtt ones.""" @@ -44,9 +47,8 @@ class HelperBot: ) self._host = host self._port = port - self._client_id = "helperbot@bumper/helperbot" self._timeout = timeout - self._client = Client(self._client_id) + self._client = Client(HELPER_BOT_CLIENT_ID) # pylint: disable=unused-argument async def _on_message( diff --git a/bumper/mqtt/server.py b/bumper/mqtt/server.py index a2614f0..83733ae 100644 --- a/bumper/mqtt/server.py +++ b/bumper/mqtt/server.py @@ -20,6 +20,7 @@ from bumper.db import ( client_get, client_set_mqtt, ) +from bumper.mqtt.helper_bot import HELPER_BOT_CLIENT_ID from bumper.mqtt.proxy import ProxyClient from bumper.util import get_logger @@ -282,7 +283,7 @@ class BumperMQTTServerPlugin: proxymodelog.info( f"MQTT Proxy Mode - New MQTT Topic Subscription - Client: {client_id} - Topic: {topic}" ) - else: + elif client_id != HELPER_BOT_CLIENT_ID: proxymodelog.warning( f"MQTT Proxy Mode - No proxy client found! - Client: {client_id} - Topic: {topic}" ) From d07ba472be4ff0af072c26f8ff2f6c5e0e5faa66 Mon Sep 17 00:00:00 2001 From: Robert Resch Date: Sat, 20 Aug 2022 14:53:52 +0200 Subject: [PATCH 06/10] allow self signed certs --- bumper/mqtt/proxy.py | 138 ++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 136 insertions(+), 2 deletions(-) diff --git a/bumper/mqtt/proxy.py b/bumper/mqtt/proxy.py index 8b15711..ee786b7 100644 --- a/bumper/mqtt/proxy.py +++ b/bumper/mqtt/proxy.py @@ -1,11 +1,27 @@ """Mqtt proxy module.""" import asyncio import re +import ssl +import typing from typing import Any, MutableMapping +from urllib.parse import urlparse, urlunparse -from amqtt.client import MQTTClient +import websockets +from amqtt.adapters import ( + StreamReaderAdapter, + StreamWriterAdapter, + WebSocketsReader, + WebSocketsWriter, +) +from amqtt.client import ConnectException, MQTTClient +from amqtt.mqtt.connack import CONNECTION_ACCEPTED from amqtt.mqtt.constants import QOS_0, QOS_1, QOS_2 +from amqtt.mqtt.protocol.client_handler import ClientProtocolHandler +from amqtt.mqtt.protocol.handler import ProtocolHandlerException from cachetools import TTLCache +from websockets.exceptions import InvalidHandshake, InvalidURI +from websockets.legacy.client import connect +from websockets.typing import Subprotocol from ..util import get_logger @@ -36,7 +52,7 @@ class ProxyClient: self.request_mapper: MutableMapping[str, str] = TTLCache( maxsize=timeout * 60, ttl=timeout * 1.1 ) - self._client = MQTTClient(client_id=client_id, config=config) + self._client = _NoCertVerifyClient(client_id=client_id, config=config) self._host = host self._port = port @@ -87,3 +103,121 @@ class ProxyClient: async def publish(self, topic: str, message: bytes, qos: int | None = None) -> None: await self._client.publish(topic, message, qos) + + +class _NoCertVerifyClient(MQTTClient): + """ + Mqtt client, which is not verify the certificate. + + Purpose is only to add "sc.verify_mode = ssl.CERT_NONE # Ignore verify of cert" + """ + + @typing.no_type_check + async def _connect_coro(self): + kwargs = dict() + + # Decode URI attributes + uri_attributes = urlparse(self.session.broker_uri) + scheme = uri_attributes.scheme + secure = True if scheme in ("mqtts", "wss") else False + self.session.username = ( + self.session.username if self.session.username else uri_attributes.username + ) + self.session.password = ( + self.session.password if self.session.password else uri_attributes.password + ) + self.session.remote_address = uri_attributes.hostname + self.session.remote_port = uri_attributes.port + if scheme in ("mqtt", "mqtts") and not self.session.remote_port: + self.session.remote_port = 8883 if scheme == "mqtts" else 1883 + if scheme in ("ws", "wss") and not self.session.remote_port: + self.session.remote_port = 443 if scheme == "wss" else 80 + if scheme in ("ws", "wss"): + # Rewrite URI to conform to https://tools.ietf.org/html/rfc6455#section-3 + uri = ( + scheme, + self.session.remote_address + ":" + str(self.session.remote_port), + uri_attributes[2], + uri_attributes[3], + uri_attributes[4], + uri_attributes[5], + ) + self.session.broker_uri = urlunparse(uri) + # Init protocol handler + # if not self._handler: + self._handler = ClientProtocolHandler(self.plugins_manager) + + if secure: + sc = ssl.create_default_context( + ssl.Purpose.SERVER_AUTH, + cafile=self.session.cafile, + capath=self.session.capath, + cadata=self.session.cadata, + ) + if "certfile" in self.config and "keyfile" in self.config: + sc.load_cert_chain(self.config["certfile"], self.config["keyfile"]) + if "check_hostname" in self.config and isinstance( + self.config["check_hostname"], bool + ): + sc.check_hostname = self.config["check_hostname"] + + sc.verify_mode = ssl.CERT_NONE # Ignore verify of cert + kwargs["ssl"] = sc + + try: + reader = None + writer = None + self._connected_state.clear() + # Open connection + if scheme in ("mqtt", "mqtts"): + conn_reader, conn_writer = await asyncio.open_connection( + self.session.remote_address, self.session.remote_port, **kwargs + ) + reader = StreamReaderAdapter(conn_reader) + writer = StreamWriterAdapter(conn_writer) + elif scheme in ("ws", "wss"): + websocket = await websockets.connect( + self.session.broker_uri, + subprotocols=["mqtt"], + extra_headers=self.extra_headers, + **kwargs, + ) + reader = WebSocketsReader(websocket) + writer = WebSocketsWriter(websocket) + # Start MQTT protocol + self._handler.attach(self.session, reader, writer) + return_code = await self._handler.mqtt_connect() + if return_code is not CONNECTION_ACCEPTED: + self.session.transitions.disconnect() + self.logger.warning("Connection rejected with code '%s'" % return_code) + exc = ConnectException("Connection rejected by broker") + exc.return_code = return_code + raise exc + else: + # Handle MQTT protocol + await self._handler.start() + self.session.transitions.connect() + self._connected_state.set() + self.logger.debug( + "connected to %s:%s" + % (self.session.remote_address, self.session.remote_port) + ) + return return_code + except InvalidURI as iuri: + self.logger.warning( + "connection failed: invalid URI '%s'" % self.session.broker_uri + ) + self.session.transitions.disconnect() + raise ConnectException( + "connection failed: invalid URI '%s'" % self.session.broker_uri, iuri + ) + except InvalidHandshake as ihs: + self.logger.warning("connection failed: invalid websocket handshake") + self.session.transitions.disconnect() + raise ConnectException( + "connection failed: invalid websocket handshake", ihs + ) + except (ProtocolHandlerException, ConnectionError, OSError) as e: + self.logger.warning("MQTT connection failed: %r" % e) + self.session.transitions.disconnect() + raise ConnectException(e) From e7f5d0846fdbedada1be1aa2b6f9419bdbdd1447 Mon Sep 17 00:00:00 2001 From: Robert Resch Date: Sat, 20 Aug 2022 16:41:36 +0200 Subject: [PATCH 07/10] fix proxy --- bumper/mqtt/helper_bot.py | 3 +++ bumper/mqtt/proxy.py | 23 ++++++++++++++--------- bumper/mqtt/server.py | 24 +++++++++++------------- 3 files changed, 28 insertions(+), 22 deletions(-) diff --git a/bumper/mqtt/helper_bot.py b/bumper/mqtt/helper_bot.py index bcb62de..4f7efae 100644 --- a/bumper/mqtt/helper_bot.py +++ b/bumper/mqtt/helper_bot.py @@ -152,6 +152,9 @@ class HelperBot: finally: self._commands.pop(request_id, None) + def publish(self, topic: str, data: bytes) -> None: + self._client.publish(topic, data) + async def disconnect(self) -> None: """Disconnect client.""" if self.is_connected: diff --git a/bumper/mqtt/proxy.py b/bumper/mqtt/proxy.py index ee786b7..981ec39 100644 --- a/bumper/mqtt/proxy.py +++ b/bumper/mqtt/proxy.py @@ -20,8 +20,8 @@ from amqtt.mqtt.protocol.client_handler import ClientProtocolHandler from amqtt.mqtt.protocol.handler import ProtocolHandlerException from cachetools import TTLCache from websockets.exceptions import InvalidHandshake, InvalidURI -from websockets.legacy.client import connect -from websockets.typing import Subprotocol + +import bumper from ..util import get_logger @@ -74,22 +74,27 @@ class ProxyClient: data = str(message.data.decode("utf-8")) _LOGGER.info( - f"MQTT Proxy Client - Message Received From Ecovacs - Topic: {message.topic} - Message: {data}" + f"Message Received From Ecovacs - Topic: {message.topic} - Message: {data}" ) topic = message.topic ttopic = topic.split("/") if ttopic[1] == "p2p": + if ttopic[3] == "proxyhelper": + _LOGGER.error( + f'"proxyhelper" was sender - INVALID!! Topic: {topic}' + ) + continue + self.request_mapper[ttopic[10]] = ttopic[3] ttopic[3] = "proxyhelper" topic = "/".join(ttopic) - _LOGGER.info( - f"MQTT Proxy Client - Converted Topic From {message.topic} TO {topic}" - ) + _LOGGER.info(f"Converted Topic From {message.topic} TO {topic}") _LOGGER.info( - f"MQTT Proxy Client - Proxy Forward Message to Robot - Topic: {topic} - Message: {data}" + f"Proxy Forward Message to Robot - Topic: {topic} - Message: {data}" ) - await self._client.publish(topic, data.encode(), QOS_0) + + bumper.mqtt_helperbot.publish(topic, message.data) except Exception: # pylint: disable=broad-except _LOGGER.error( "An error occurred during handling a message", exc_info=True @@ -105,7 +110,7 @@ class ProxyClient: await self._client.publish(topic, message, qos) -class _NoCertVerifyClient(MQTTClient): +class _NoCertVerifyClient(MQTTClient): # type:ignore[misc] """ Mqtt client, which is not verify the certificate. diff --git a/bumper/mqtt/server.py b/bumper/mqtt/server.py index 83733ae..36551dc 100644 --- a/bumper/mqtt/server.py +++ b/bumper/mqtt/server.py @@ -149,11 +149,14 @@ class BumperMQTTServerPlugin: client_id = session.client_id try: + if client_id == HELPER_BOT_CLIENT_ID: + mqttserverlog.info("Bumper Authentication Success - Helperbot") + return True + if "@" in client_id: didsplit = str(client_id).split("@") - if not ( # if ecouser or bumper aren't in details it is a bot - "ecouser" in didsplit[1] or "bumper" in didsplit[1] - ): + if "ecouser" not in didsplit[1]: + # if ecouser aren't in details it is a bot tmpbotdetail = str(didsplit[1]).split("/") bot_add( username, @@ -201,11 +204,6 @@ class BumperMQTTServerPlugin: realm = tmpclientdetail[0] resource = tmpclientdetail[1] - if userid == "helperbot": - mqttserverlog.info( - "Bumper Authentication Success - Helperbot: %s", client_id - ) - return True if check_authcode(didsplit[0], password) or not bumper.use_auth: client_add(userid, realm, resource) mqttserverlog.info( @@ -342,7 +340,7 @@ class BumperMQTTServerPlugin: ) if ttopic[6] == "": proxymodelog.warning( - "MQTT Proxy Client - Request mapper is missing entry, " + "Request mapper is missing entry, " f"probably request took to long... Client_id: {client_id}" f" - Request_id: {ttopic[10]}" ) @@ -350,19 +348,19 @@ class BumperMQTTServerPlugin: ttopic_join = "/".join(ttopic) proxymodelog.info( - f"MQTT Proxy Client - Bot Message Converted Topic From {message.topic} TO {ttopic_join} " + f"Bot Message Converted Topic From {message.topic} TO {ttopic_join} " f"with message: {data_decoded}" ) else: ttopic_join = message.topic proxymodelog.info( - f"MQTT Proxy Client - Bot Message From {ttopic_join} with message: {data_decoded}" + f"Bot Message From {ttopic_join} with message: {data_decoded}" ) try: # Send back to ecovacs proxymodelog.info( - "MQTT Proxy Client - Proxy Forward Message to Ecovacs - Topic:" + "Proxy Forward Message to Ecovacs - Topic:" f" {ttopic_join} - Message: {data_decoded}" ) await self._proxy_clients[client_id].publish( @@ -370,7 +368,7 @@ class BumperMQTTServerPlugin: ) except Exception: # pylint: disable=broad-except proxymodelog.error( - "MQTT Proxy Client - Forwarding to Ecovacs - Exception", + "Forwarding to Ecovacs - Exception", exc_info=True, ) From e7c03f8cf029cfe88c86352b943ed850223b3743 Mon Sep 17 00:00:00 2001 From: Robert Resch Date: Sat, 20 Aug 2022 16:43:17 +0200 Subject: [PATCH 08/10] fix mypy --- bumper/dns.py | 2 +- bumper/web/middlewares.py | 13 ++++++++++--- bumper/web/server.py | 12 ++++-------- bumper/xmppserver.py | 9 +++++---- 4 files changed, 20 insertions(+), 16 deletions(-) diff --git a/bumper/dns.py b/bumper/dns.py index e480de3..c1c071a 100644 --- a/bumper/dns.py +++ b/bumper/dns.py @@ -8,4 +8,4 @@ def get_resolver_with_public_nameserver() -> AsyncResolver: async def resolve(host: str) -> str: hosts = await get_resolver_with_public_nameserver().resolve(host) - return hosts[0]["host"] + return hosts[0]["host"] # type:ignore[no-any-return] diff --git a/bumper/web/middlewares.py b/bumper/web/middlewares.py index 3731447..d85522e 100644 --- a/bumper/web/middlewares.py +++ b/bumper/web/middlewares.py @@ -1,5 +1,6 @@ """Web server middleware module.""" import json +from typing import Any from aiohttp import web from aiohttp.typedefs import Handler @@ -13,7 +14,7 @@ _LOGGER = get_logger("webserver_requests") class CustomEncoder(json.JSONEncoder): - def default(self, obj): + def default(self, obj: Any) -> Any: if isinstance(obj, set): return list(obj) return json.JSONEncoder.default(self, obj) @@ -32,9 +33,14 @@ async def log_all_requests(request: Request, handler: Handler) -> StreamResponse "path": request.path, "query_string": request.query_string, "headers": {h for h in request.headers.items()}, - "route_resource": request.match_info.route.resource.canonical, } } + + if request.match_info.route.resource: + to_log["request"][ + "route_resource" + ] = request.match_info.route.resource.canonical + try: if request.content_length: if request.content_type == "application/json": @@ -44,7 +50,7 @@ async def log_all_requests(request: Request, handler: Handler) -> StreamResponse response = await handler(request) if response is None: - confserverlog.warning( # type:ignore[unreachable] + _LOGGER.warning( # type:ignore[unreachable] "Response was null!" ) _LOGGER.warning(json.dumps(to_log, cls=CustomEncoder)) @@ -55,6 +61,7 @@ async def log_all_requests(request: Request, handler: Handler) -> StreamResponse } if isinstance(response, Response) and response.body: + assert response.text if response.content_type == "application/json": to_log["response"]["body"] = json.loads(response.text) elif response.content_type.startswith("text"): diff --git a/bumper/web/server.py b/bumper/web/server.py index 8e8d544..6144d5d 100644 --- a/bumper/web/server.py +++ b/bumper/web/server.py @@ -319,7 +319,7 @@ class WebServer: if request.content.total_bytes > 0: read_body = await request.read() proxymodelog.info( - f"HTTP Proxy Request to EcoVacs (body=true) (URL:{request.url}) - {read_body}" + f"HTTP Proxy Request to EcoVacs (body=true) (URL:{request.url}) - {read_body.decode('utf-8')}" ) if request.content_type == "application/x-www-form-urlencoded": # android apps use form @@ -349,10 +349,10 @@ class WebServer: ) async with session.request(request.method, request.url) as resp: if resp.content_type == "application/octet-stream": - response = await resp.read() proxymodelog.info( f"HTTP Proxy Response from EcoVacs (URL: {request.url}) - (Status: {resp.status}) - " ) + return web.Response(body=await resp.read()) else: response = await resp.text() proxymodelog.info( @@ -363,14 +363,10 @@ class WebServer: if resp.content_type == "application/json": response = json.loads(response) return web.json_response(response) - elif resp.content_type == "application/octet-stream": + if resp.content_type == "application/octet-stream": return web.Response(body=response) - else: - return web.Response(text=response) - - else: - return web.Response(text=response) + return web.Response(text=response) except asyncio.CancelledError: proxymodelog.exception( f"Request cancelled or timeout - {request.url}", exc_info=True diff --git a/bumper/xmppserver.py b/bumper/xmppserver.py index af16cde..f07c3d5 100644 --- a/bumper/xmppserver.py +++ b/bumper/xmppserver.py @@ -230,7 +230,7 @@ class XMPPAsyncClient: xmppserverlog.debug( "admin username received from bot: {}".format(ctl.get("admin")) ) - #XMPPServer.client_id = ctl.get("admin") + # XMPPServer.client_id = ctl.get("admin") return # forward @@ -314,8 +314,8 @@ class XMPPAsyncClient: xmppserverlog.info( "Bot reported user has no permissions, Bumper will attempt to add user to bot. This is typical if bot was last on Ecovacs Network." ) - xquery = xml.getchildren() - ctl = xquery[0].getchildren() + xquery = list(xml) + ctl = list(xquery[0]) if "error" in ctl[0].attrib: ctlerr = ctl[0].attrib["error"] adminuser = ctlerr.replace( @@ -328,7 +328,8 @@ class XMPPAsyncClient: adminuser.startswith("fuid_") or adminuser.startswith("fusername_") or bumper.use_auth - ): # if not fuid_ then its ecovacs OR ignore bumper auth + ): + # if not fuid_ then its ecovacs OR ignore bumper auth # TODO: Implement auth later, should this user have access to bot? # Add user jid to bot From 80ffc65620305a5ae378b4e65378d91042810e41 Mon Sep 17 00:00:00 2001 From: Robert Resch Date: Sat, 20 Aug 2022 19:37:01 +0200 Subject: [PATCH 09/10] data can be None --- bumper/mqtt/proxy.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/bumper/mqtt/proxy.py b/bumper/mqtt/proxy.py index 981ec39..24261ca 100644 --- a/bumper/mqtt/proxy.py +++ b/bumper/mqtt/proxy.py @@ -71,7 +71,7 @@ class ProxyClient: while self._client.session.transitions.is_connected(): try: message = await self._client.deliver_message() - data = str(message.data.decode("utf-8")) + data = message.data.decode("utf-8") if message.data else "" _LOGGER.info( f"Message Received From Ecovacs - Topic: {message.topic} - Message: {data}" From b92cead7d5d0caa3bf936a447a0a5b5cb2f12f46 Mon Sep 17 00:00:00 2001 From: Robert Resch Date: Sat, 20 Aug 2022 19:40:04 +0200 Subject: [PATCH 10/10] remove not required code --- bumper/mqtt/proxy.py | 6 ------ bumper/mqtt/server.py | 16 +--------------- 2 files changed, 1 insertion(+), 21 deletions(-) diff --git a/bumper/mqtt/proxy.py b/bumper/mqtt/proxy.py index 24261ca..b1b94b2 100644 --- a/bumper/mqtt/proxy.py +++ b/bumper/mqtt/proxy.py @@ -31,12 +31,6 @@ _LOGGER = get_logger("proxymode") # /[receiver did]/[receiver class]]/[receiver resource]/[q|p/[request id/j # [q|p] q-> request p-> response -TOPIC_P2P = re.compile( - "iot/p2p/(?P[^/]+)/(?P[^/]+)/(?P[^/]+)/(?P[^/]+)/" - "/(?P[^/]+)/(?P[^/]+)/(?P[^/]+)/(?P[^/]+)/" - "(?P[^/]+)/(?P[^/]+)" -) - class ProxyClient: """Mqtt client, which proxies all messages to the ecovacs servers.""" diff --git a/bumper/mqtt/server.py b/bumper/mqtt/server.py index 36551dc..3e417c2 100644 --- a/bumper/mqtt/server.py +++ b/bumper/mqtt/server.py @@ -174,22 +174,8 @@ class BumperMQTTServerPlugin: if bumper.bumper_proxy_mode: mqtt_server = await dns.resolve("mq-ww.ecouser.net") - if mqtt_server: - proxymodelog.info( - f"MQTT Proxy Mode - Using server {mqtt_server}" - ) - else: - proxymodelog.error( - "MQTT Proxy Mode - No server found! Load defaults or " - "set mqtt_server in config_proxymode table!" - ) - proxymodelog.exception( - f"MQTT Proxy Mode - Exiting due to no MQTT Server configured!" - ) - exit(1) - proxymodelog.info( - f"MQTT Proxy Mode - Proxy Bot to MQTT - Client_id: {client_id} - Username: {username}" + f"MQTT Proxy Mode - Using server {mqtt_server} for client {client_id}" ) proxy = ProxyClient( client_id, mqtt_server, config={"check_hostname": False}