Merge pull request #51 from edenhaus/proxyV2

add proxy mode

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
 

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)

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"]  # type:ignore[no-any-return]

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("helperbot@bumper/helperbot")
+        self._client = Client(HELPER_BOT_CLIENT_ID)
 
         # pylint: disable=unused-argument
         async def _on_message(
@@ -150,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:

bumper/mqtt/proxy.py

@@ -0,0 +1,222 @@
+"""Mqtt proxy module."""
+import asyncio
+import re
+import ssl
+import typing
+from typing import Any, MutableMapping
+from urllib.parse import urlparse, urlunparse
+
+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
+
+import bumper
+
+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
+
+
+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 = _NoCertVerifyClient(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 = message.data.decode("utf-8") if message.data else ""
+
+                _LOGGER.info(
+                    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"Converted Topic From {message.topic} TO {topic}")
+
+                _LOGGER.info(
+                    f"Proxy Forward Message to Robot - Topic: {topic} - Message: {data}"
+                )
+
+                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
+                )
+
+    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)
+
+
+class _NoCertVerifyClient(MQTTClient):  # type:ignore[misc]
+    """
+    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)

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,14 @@ 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
 
 mqttserverlog = get_logger("mqttserver")
 helperbotlog = get_logger("helperbot")
 boterrorlog = get_logger("boterror")
+proxymodelog = get_logger("proxymode")
 
 
 class MQTTServer:
@@ -122,6 +126,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"]
@@ -144,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
+                if "ecouser" not in didsplit[1]:
-                    "ecouser" in didsplit[1] or "bumper" in didsplit[1]
+                    # if ecouser aren't in details it is a bot
-                ):
                     tmpbotdetail = str(didsplit[1]).split("/")
                     bot_add(
                         username,
@@ -163,6 +171,18 @@ class BumperMQTTServerPlugin:
                         didsplit[0],
                         tmpbotdetail[0],
                     )
+
+                    if bumper.bumper_proxy_mode:
+                        mqtt_server = await dns.resolve("mq-ww.ecouser.net")
+                        proxymodelog.info(
+                            f"MQTT Proxy Mode - Using server {mqtt_server} for client {client_id}"
+                        )
+                        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("/")
@@ -170,11 +190,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(
@@ -242,6 +257,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}"
+                )
+            elif client_id != HELPER_BOT_CLIENT_ID:
+                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 +292,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 +316,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(
+                            "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"Bot Message Converted Topic From {message.topic} TO {ttopic_join} "
+                        f"with message: {data_decoded}"
+                    )
+                else:
+                    ttopic_join = message.topic
+                    proxymodelog.info(
+                        f"Bot Message From {ttopic_join} with message: {data_decoded}"
+                    )
+
+                try:
+                    # Send back to ecovacs
+                    proxymodelog.info(
+                        "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(
+                        "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)

bumper/web/middlewares.py

@@ -0,0 +1,86 @@
+"""Web server middleware module."""
+import json
+from typing import Any
+
+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: Any) -> Any:
+        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()},
+            }
+        }
+
+        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":
+                    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:
+                _LOGGER.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:
+                assert response.text
+                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)

bumper/web/server.py

@@ -1,5 +1,4 @@
 """Web server module."""
-
 import asyncio
 import dataclasses
 import json
@@ -7,21 +6,19 @@ 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 HTTPInternalServerError
-from aiohttp.web_exceptions import (
-    HTTPBadRequest,
-    HTTPInternalServerError,
-    HTTPNoContent,
-)
 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
 
 
@@ -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)
@@ -55,9 +53,9 @@ 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
-    def __init__(self, bindings: list[WebserverBinding] | WebserverBinding):
+    ):
         self._runners: list[web.AppRunner] = []
 
         if isinstance(bindings, WebserverBinding):
@@ -66,7 +64,7 @@ class WebServer:
 
         self._app = web.Application(
             middlewares=[
-                self._log_all_requests,
+                log_all_requests,
             ],
         )
         aiohttp_jinja2.setup(
@@ -75,32 +73,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),
-                web.get(
-                    "/bot/remove/{did}", self._handle_remove_bot, name="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."""
@@ -173,75 +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:
-                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")
-
-                    else:
-                        postbody = await request.post()
-
-                to_log["request"]["body"] = f"{postbody}"
-
-                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 (
-                    "application/octet-stream" not in response.content_type
-                    and isinstance(response, Response)
-                    and response.body
-                ):
-                    assert isinstance(response.body, bytes)
-                    to_log["response"]["body"] = f"{json.loads(response.body)}"
-
-                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()
         asyncio.create_task(bumper.mqtt_helperbot.start())
@@ -365,3 +301,79 @@ 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.decode('utf-8')}"
+                    )
+                    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":
+                            proxymodelog.info(
+                                f"HTTP Proxy Response from EcoVacs (URL: {request.url}) - (Status: {resp.status}) - <BYTES CONTENT>"
+                            )
+                            return web.Response(body=await resp.read())
+                        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)
+                    if resp.content_type == "application/octet-stream":
+                        return web.Response(body=response)
+
+                return web.Response(text=response)
+        except asyncio.CancelledError:
+            proxymodelog.exception(
+                f"Request cancelled or timeout - {request.url}", exc_info=True
+            )
+            raise
+
+        except Exception:
+            proxymodelog.exception("An exception occurred", exc_info=True)
+
+        raise HTTPInternalServerError

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()
+                    xquery = list(xml)
-                    ctl = xquery[0].getchildren()
+                    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

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

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

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()