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..c1c071a --- /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"] # type:ignore[no-any-return] diff --git a/bumper/mqtt/helper_bot.py b/bumper/mqtt/helper_bot.py index 4ce7060..4f7efae 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("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: diff --git a/bumper/mqtt/proxy.py b/bumper/mqtt/proxy.py new file mode 100644 index 0000000..b1b94b2 --- /dev/null +++ b/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) diff --git a/bumper/mqtt/server.py b/bumper/mqtt/server.py index 3e88255..3e417c2 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,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 - "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, @@ -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) diff --git a/bumper/web/middlewares.py b/bumper/web/middlewares.py new file mode 100644 index 0000000..d85522e --- /dev/null +++ b/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) diff --git a/bumper/web/server.py b/bumper/web/server.py index cc231cb..6144d5d 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,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 ( - HTTPBadRequest, - 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 @@ -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): + def __init__( + self, bindings: list[WebserverBinding] | WebserverBinding, proxy_mode: bool + ): 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, 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.""" @@ -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}) - " + ) + 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 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 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()