Added proxy and tests for proxy

README.rst

@@ -44,8 +44,7 @@ Usage
 -----
 
 This module is implemented as a primarily asynchronous library
-(asyncio), mostly through the ``Session`` class, which is exported as
-default. Because the library is asynchronous, you must wait for it to be
+(asyncio), mostly through the `Session <https://pylibmeshctrl.readthedocs.io/en/latest/api/meshctrl.html#meshctrl.session.Session>`__ class. Because the library is asynchronous, you must wait for it to be
 initialized before interacting with the server. The preferred way to do
 this is to use the async context manager pattern:
 
@@ -53,20 +52,20 @@ this is to use the async context manager pattern:
 
    import meshctrl
 
-   async with meshctrl.session.Session(url, **options):
+   async with meshctrl.Session(url, **options):
        print(await session.list_users())
        ...
 
 However, if you prefer to instantiate the object yourself, you can
-simply use the ``initialized`` property:
+simply use the `initialized <https://pylibmeshctrl.readthedocs.io/en/latest/api/meshctrl.html#meshctrl.session.Session.initialized>`__ property:
 
 .. code:: python
 
-   session = meshctrl.session.Session(url, **options)
+   session = meshctrl.Session(url, **options)
    await session.initialized.wait()
 
 Note that, in this case, you will be rquired to clean up tho session
-using its ``close`` method.
+using its `close <https://pylibmeshctrl.readthedocs.io/en/latest/api/meshctrl.html#meshctrl.session.Session.close>`__ method.
 
 Session Parameters
 ------------------

setup.cfg

@@ -46,6 +46,7 @@ install_requires =
     importlib-metadata
     cryptography>=43.0.3
     websockets>=13.1
+    python-socks[asyncio]
 
 
 [options.packages.find]

src/meshctrl/exceptions.py

@@ -22,9 +22,6 @@ class FileTransferError(MeshCtrlError):
 
     Attributes:
         stats (dict): {"result" (str): Human readable result, "size" (int): number of bytes successfully transferred}
-        initialized (asyncio.Event): Event marking if the Session initialization has finished. Wait on this to wait for a connection.
-        alive (bool): Whether the session connection is currently alive
-        closed (asyncio.Event): Event that occurs when the session closes permanently
     """
     def __init__(self, message, stats):
         self.stats = stats

src/meshctrl/session.py

@@ -8,6 +8,8 @@ import json
 import datetime
 import io
 import ssl
+import urllib
+from python_socks.async_.asyncio import Proxy
 from . import constants
 from . import exceptions
 from . import util
@@ -31,6 +33,7 @@ class Session(object):
         proxy (str): "url:port" to use for proxy server NOTE: This is currently not implemented due to a limitation of the undersying websocket library. Upvote the issue if you find this important.
         token (str): Login token. This appears to be superfluous
         ignore_ssl (bool): Ignore SSL errors
+        auto_reconnect (bool): In case of server failure, attempt to auto reconnect. All outstanding requests will be killed.
 
     Returns:
         :py:class:`Session`: Session connected to url
@@ -92,6 +95,7 @@ class Session(object):
         self._inflight = set()
         self._file_tunnels = {}
         self._ignore_ssl = ignore_ssl
+        self.auto_reconnect = auto_reconnect
 
         self._eventer = util.Eventer()
 
@@ -120,16 +124,17 @@ class Session(object):
                 ssl_context.verify_mode = ssl.CERT_NONE
                 options = { "ssl": ssl_context }
 
-            # Setup the HTTP proxy if needed
-            # if (self._proxy != None):
-            #     options.agent = new https_proxy_agent(urllib.parse(self._proxy))
-
             headers = websockets.datastructures.Headers()
 
             if (self._password):
                 token = self._token if self._token else b""
                 headers['x-meshauth'] = (base64.b64encode(self._user.encode()) + b',' + base64.b64encode(self._password.encode()) + token).decode()
 
+            if self._proxy:
+                proxy = Proxy.from_url(self._proxy)
+                parsed = urllib.parse.urlparse(self.url)
+                options["sock"] = await proxy.connect(dest_host=parsed.hostname, dest_port=parsed.port)
+
             options["additional_headers"] = headers
             async for websocket in websockets.asyncio.client.connect(self.url, process_exception=util._process_websocket_exception, **options):
                 self.alive = True
@@ -159,13 +164,20 @@ class Session(object):
     async def _send_data_task(self, websocket):
         while True:
             message = await self._message_queue.get()
-            print(f"{self._user} send: {message}\n")
             await websocket.send(message)
 
     async def _listen_data_task(self, websocket):
         async for message in websocket:
-            print(f"{self._user} recv: {message}\n")
-            data = json.loads(message)
+            await self._eventer.emit("raw", message)
+            # Meshcentral does pong wrong and breaks our parsing, so fix it here.
+            if message == '{action:"pong"}':
+                message = '{"action":"pong"}'
+
+            # Can't process non-json data, don't even try
+            try:
+                data = json.loads(message)
+            except SyntaxError:
+                continue
             action = data.get("action", None)
             await self._eventer.emit("server_event", data)
             if action == "close":
@@ -234,14 +246,14 @@ class Session(object):
         return response
 
     @util._check_socket
-    async def _send_command_no_response_id(self, data, timeout=None):
+    async def _send_command_no_response_id(self, data, action_override=None, timeout=None):
         responded = asyncio.Event()
         response = None
         async def _(data):
             nonlocal response
             response = data
             responded.set()
-        self._eventer.once(data["action"], _)
+        self._eventer.once(action_override if action_override is not None else data["action"], _)
         await self._message_queue.put(json.dumps(data))
         await asyncio.wait_for(responded.wait(), timeout=timeout)
         if isinstance(response, Exception):
@@ -268,6 +280,23 @@ class Session(object):
         """
         return self._user_info
     
+    async def ping(self, timeout=None):
+        '''
+        Ping the server. WARNING: Non namespaced call. Calling this function again before it returns may cause unintended consequences.
+
+        Args:
+            timeout (int): duration in seconds to wait for a response before throwing an error
+
+        Returns:
+            dict: {"action": "pong"}
+
+        Raises:
+            :py:class:`~meshctrl.exceptions.ServerError`: Error from server
+            :py:class:`~meshctrl.exceptions.SocketError`: Info about socket closure
+            asyncio.TimeoutError: Command timed out
+        '''
+        data = await self._send_command_no_response_id({"action": "ping"}, action_override="pong", timeout=timeout)
+        return data
 
     async def list_device_groups(self, timeout=None):
         '''
@@ -284,7 +313,7 @@ class Session(object):
             :py:class:`~meshctrl.exceptions.SocketError`: Info about socket closure
             asyncio.TimeoutError: Command timed out
         '''
-        data = await self._send_command({"action": "meshes"}, "list_device_groups", timeout)
+        data = await self._send_command({"action": "meshes"}, "list_device_groups", timeout=timeout)
         return [mesh.Mesh(m["_id"], self, **m) for m in data["meshes"]]
 
 
@@ -323,7 +352,7 @@ class Session(object):
             op["name"] = name
         if message:
             op["msg"] = message
-        data = await self._send_command(op, "send_invite_email", timeout)
+        data = await self._send_command(op, "send_invite_email", timeout=timeout)
         if ("result" in data and data["result"].lower() != "ok"):
             raise exceptions.ServerError(data["result"])
         return True
@@ -359,7 +388,7 @@ class Session(object):
             op["meshname"] = group
         if flags != None:
             op["flags"] = flags
-        data = await self._send_command(op, "generate_invite_link", timeout)
+        data = await self._send_command(op, "generate_invite_link", timeout=timeout)
         if ("result" in data and data["result"].lower() != "ok"):
             raise exceptions.ServerError(data["result"])
         del data["tag"]
@@ -382,7 +411,7 @@ class Session(object):
             :py:class:`~meshctrl.exceptions.SocketError`: Info about socket closure
             asyncio.TimeoutError: Command timed out
         '''
-        data = await self._send_command({"action": "users"}, "list_users", timeout)
+        data = await self._send_command({"action": "users"}, "list_users", timeout=timeout)
         if ("result" in data and data["result"].lower() != "ok"):
             raise exceptions.ServerError(data["result"])
         return data["users"]
@@ -401,7 +430,7 @@ class Session(object):
             :py:class:`~meshctrl.exceptions.SocketError`: Info about socket closure
             asyncio.TimeoutError: Command timed out
         '''
-        return (await self._send_command({"action": "wssessioncount"}, "list_user_sessions", timeout))["wssessions"]
+        return (await self._send_command({"action": "wssessioncount"}, "list_user_sessions", timeout=timeout))["wssessions"]
 
     
     async def list_devices(self, details=False, group=None, meshid=None, timeout=None):
@@ -426,14 +455,14 @@ class Session(object):
         tasks = []
         async with asyncio.TaskGroup() as tg:
             if details:
-                tasks.append(tg.create_task(self._send_command_no_response_id({"action": "getDeviceDetails", "type":"json"}, timeout)))
+                tasks.append(tg.create_task(self._send_command_no_response_id({"action": "getDeviceDetails", "type":"json"}, timeout=timeout)))
             elif group:
-                tasks.append(tg.create_task(self._send_command({ "action": 'nodes', "meshname": group}, "list_devices", timeout)))
+                tasks.append(tg.create_task(self._send_command({ "action": 'nodes', "meshname": group}, "list_devices", timeout=timeout)))
             elif meshid:
-                tasks.append(tg.create_task(self._send_command({ "action": 'nodes', "meshid": meshid}, "list_devices", timeout)))
+                tasks.append(tg.create_task(self._send_command({ "action": 'nodes', "meshid": meshid}, "list_devices", timeout=timeout)))
             else:
-                tasks.append(tg.create_task(self._send_command({ "action": 'meshes' }, "list_devices", timeout)))
-                tasks.append(tg.create_task(self._send_command({ "action": 'nodes' }, "list_devices", timeout)))
+                tasks.append(tg.create_task(self._send_command({ "action": 'meshes' }, "list_devices", timeout=timeout)))
+                tasks.append(tg.create_task(self._send_command({ "action": 'nodes' }, "list_devices", timeout=timeout)))
 
         res0 = tasks[0].result()
         if "result" in res0:
@@ -478,6 +507,24 @@ class Session(object):
                 node["mesh"] = mesh.Mesh(node.get("meshid"), self)
         return [device.Device(n["_id"], self, **n) for n in nodes]
 
+    async def raw_messages(self):
+        '''
+        Listen to raw messages from the server. These will be strings that have not been parsed at all. Consider this an emergency fallback if meshcentral sends something odd. You will get every message from the websocket.
+
+        Returns:
+            generator(data): A generator which will generate every message the server sends
+        '''
+        event_queue = asyncio.Queue()
+        async def _(data):
+            await event_queue.put(data)
+        self._eventer.on("raw", _)
+        try:
+            while True:
+                data = await event_queue.get()
+                yield data
+        finally:
+            self._eventer.off("server_event", _)
+
     async def events(self, filter=None):
         '''
         Listen to events from the server
@@ -535,7 +582,7 @@ class Session(object):
         if limit:
             cmd["limit"] = limit
 
-        data = await self._send_command(cmd, "list_events", timeout)
+        data = await self._send_command(cmd, "list_events", timeout=timeout)
         return data["events"]
 
     async def list_login_tokens(self, timeout=None):
@@ -552,7 +599,7 @@ class Session(object):
             :py:class:`~meshctrl.exceptions.SocketError`: Info about socket closure
             asyncio.TimeoutError: Command timed out
          '''
-        return (await self._send_command_no_response_id({"action": "loginTokens"}, timeout))["loginTokens"]
+        return (await self._send_command_no_response_id({"action": "loginTokens"}, timeout=timeout))["loginTokens"]
 
     async def add_login_token(self, name, expire=None, timeout=None):
         '''
@@ -571,7 +618,7 @@ class Session(object):
             asyncio.TimeoutError: Command timed out
          '''
         cmd = { "action": 'createLoginToken', "name": name, "expire": 0 if not expire else expire }
-        data = await self._send_command_no_response_id(cmd, timeout)
+        data = await self._send_command_no_response_id(cmd, timeout=timeout)
         del data["action"]
         return data
 
@@ -603,7 +650,7 @@ class Session(object):
                         name = token["tokenUser"]
                         break
             realnames.append(name)
-        return (await self._send_command_no_response_id({ "action": 'loginTokens', "remove": realnames }, timeout))["loginTokens"]
+        return (await self._send_command_no_response_id({ "action": 'loginTokens', "remove": realnames }, timeout=timeout))["loginTokens"]
 
     async def add_user(self, name, password=None, randompass=False, domain=None, email=None, emailverified=False, resetpass=False, realname=None, phone=None, rights=None, timeout=None):
         '''
@@ -651,7 +698,7 @@ class Session(object):
         if isinstance(realname, str):
             op["realname"] = realname
 
-        data = await self._send_command(op, "add_user", timeout)
+        data = await self._send_command(op, "add_user", timeout=timeout)
         if data.get("result", "ok").lower() != "ok":
             raise exceptions.ServerError(data["result"])
         return True
@@ -706,7 +753,7 @@ class Session(object):
             op["realname"] = realname
         if realname is True:
             op["realname"] = ''
-        data = await self._send_command(op, "edit_user", timeout) 
+        data = await self._send_command(op, "edit_user", timeout=timeout) 
         if data.get("result", "ok").lower() != "ok":
             raise exceptions.ServerError(data["result"])
         return True
@@ -732,7 +779,7 @@ class Session(object):
         elif (self._domain is not None) and ("/" not in userid):
             userid = f"user/{self._domain}/{userid}"
 
-        data = await self._send_command({ "action": 'deleteuser', "userid": userid }, "remove_user", timeout)
+        data = await self._send_command({ "action": 'deleteuser', "userid": userid }, "remove_user", timeout=timeout)
         if data.get("result", "ok").lower() != "ok":
             raise exceptions.ServerError(data["result"])
         return True
@@ -760,7 +807,7 @@ class Session(object):
             op["domain"] = self._domain
         elif self._domain is not None:
             op["domain"] = self._domain
-        data = await self._send_command(op, "add_user_group", timeout)
+        data = await self._send_command(op, "add_user_group", timeout=timeout)
         if data.get("result", "ok").lower() != "ok":
             raise exceptions.ServerError(data["result"])
 
@@ -795,7 +842,7 @@ class Session(object):
 
         if (not groupid.startswith("ugrp/")):
             groupid = f"ugrp//{groupid}"
-        data = await self._send_command({ "action": 'deleteusergroup', "ugrpid": groupid }, "remove_user_group", timeout)
+        data = await self._send_command({ "action": 'deleteusergroup', "ugrpid": groupid }, "remove_user_group", timeout=timeout)
         if data.get("result", "ok").lower() != "ok":
             raise exceptions.ServerError(data["result"])
         return True
@@ -814,7 +861,7 @@ class Session(object):
             :py:class:`~meshctrl.exceptions.SocketError`: Info about socket closure
             asyncio.TimeoutError: Command timed out
         '''
-        r = await self._send_command({"action": "usergroups"}, "list_user_groups", timeout)
+        r = await self._send_command({"action": "usergroups"}, "list_user_groups", timeout=timeout)
         groups = []
         for key, val in r["ugroups"].items():
             val["_id"] = key
@@ -888,7 +935,7 @@ class Session(object):
         async with asyncio.TaskGroup() as tg:
             tasks.append(tg.create_task(asyncio.wait_for(_(tg), timeout=timeout)))
             tasks.append(tg.create_task(asyncio.wait_for(__(tg), timeout=timeout)))
-            tasks.append(tg.create_task(self._send_command({ "action": 'addusertousergroup', "ugrpid": groupid, "usernames": usernames}, "add_users_to_user_group", timeout)))
+            tasks.append(tg.create_task(self._send_command({ "action": 'addusertousergroup', "ugrpid": groupid, "usernames": usernames}, "add_users_to_user_group", timeout=timeout)))
 
 
         res = tasks[2].result()
@@ -922,7 +969,7 @@ class Session(object):
         if (not groupid.startswith("ugrp/")):
             groupid = f"ugrp//{groupid}"
 
-        data = await self._send_command({ "action": 'removeuserfromusergroup', "ugrpid": groupid, "userid": userid }, "remove_from_user_group", timeout)
+        data = await self._send_command({ "action": 'removeuserfromusergroup', "ugrpid": groupid, "userid": userid }, "remove_from_user_group", timeout=timeout)
 
         if data.get("result", "ok").lower() != "ok":
             raise exceptions.ServerError(data["result"])
@@ -953,7 +1000,7 @@ class Session(object):
         if rights is None:
             rights = 0
 
-        data = await self._send_command({ "action": 'adddeviceuser', "nodeid": nodeid, "userids": userids, "rights": rights}, "add_users_to_device", timeout)
+        data = await self._send_command({ "action": 'adddeviceuser', "nodeid": nodeid, "userids": userids, "rights": rights}, "add_users_to_device", timeout=timeout)
 
         if data.get("result", "ok").lower() != "ok":
             raise exceptions.ServerError(data["result"])
@@ -981,7 +1028,7 @@ class Session(object):
 
         userids = [f"user//{u}" if  not u.startswith("user//") else u for u in userids]
 
-        data = await self._send_command({ "action": 'adddeviceuser', "nodeid": nodeid, "usernames": userids, "rights": 0, "remove": True }, "remove_users_from_device", timeout)
+        data = await self._send_command({ "action": 'adddeviceuser', "nodeid": nodeid, "usernames": userids, "rights": 0, "remove": True }, "remove_users_from_device", timeout=timeout)
 
         if data.get("result", "ok").lower() != "ok":
             raise exceptions.ServerError(data["result"])
@@ -1018,7 +1065,7 @@ class Session(object):
         if consent:
             op["consent"] = consent
 
-        data = await self._send_command(op, "add_device_group", timeout)
+        data = await self._send_command(op, "add_device_group", timeout=timeout)
         if data.get("result", "ok").lower() != "ok":
             raise exceptions.ServerError(data["result"])
 
@@ -1053,7 +1100,7 @@ class Session(object):
             op["meshname"] = meshid
             del op["meshid"]
 
-        data = await self._send_command(op, "remove_device_group", timeout)
+        data = await self._send_command(op, "remove_device_group", timeout=timeout)
 
         if data.get("result", "ok").lower() != "ok":
             raise exceptions.ServerError(data["result"])
@@ -1107,7 +1154,7 @@ class Session(object):
         if consent is not None:
             op["consent"] = consent
 
-        data = await self._send_command(op, "edit_device_group", timeout)
+        data = await self._send_command(op, "edit_device_group", timeout=timeout)
 
         if data.get("result", "ok").lower() != "ok":
             raise exceptions.ServerError(data["result"])
@@ -1138,7 +1185,7 @@ class Session(object):
             op["meshname"] = meshid
             del op["meshid"]
 
-        data = await self._send_command(op, "move_to_device_group", timeout)
+        data = await self._send_command(op, "move_to_device_group", timeout=timeout)
 
         if data.get("result", "ok").lower() != "ok":
             raise exceptions.ServerError(data["result"])
@@ -1171,7 +1218,7 @@ class Session(object):
             op["meshname"] = meshid
             del op["meshid"]
 
-        data = await self._send_command(op, "add_user_to_device_group", timeout)
+        data = await self._send_command(op, "add_user_to_device_group", timeout=timeout)
         results = data["result"].split(",")
         out = {}
         for i, result in enumerate(results):
@@ -1213,7 +1260,7 @@ class Session(object):
         tasks = []
         async with asyncio.TaskGroup() as tg:
             for userid in userids:
-                tasks.append(tg.create_task(self._send_command({ "action": 'removemeshuser', "userid": userid } | id_obj, "remove_users_from_device_group", timeout)))
+                tasks.append(tg.create_task(self._send_command({ "action": 'removemeshuser', "userid": userid } | id_obj, "remove_users_from_device_group", timeout=timeout)))
 
         out = {}
         for i, task in enumerate(tasks):
@@ -1247,7 +1294,7 @@ class Session(object):
         if userid:
             op["userid"] = userid
 
-        data = await self._send_command(op, "broadcast", timeout)
+        data = await self._send_command(op, "broadcast", timeout=timeout)
 
         if data.get("result", "ok").lower() != "ok":
             raise exceptions.ServerError(data["result"])
@@ -1271,10 +1318,10 @@ class Session(object):
         '''
         tasks = []
         async with asyncio.TaskGroup() as tg:
-            tasks.append(tg.create_task(self._send_command({ "action": 'nodes' }, "device_info", timeout)))
-            tasks.append(tg.create_task(self._send_command_no_response_id({ "action": 'getnetworkinfo', "nodeid": nodeid }, timeout)))
-            tasks.append(tg.create_task(self._send_command_no_response_id({ "action": 'lastconnect', "nodeid": nodeid }, timeout)))
-            tasks.append(tg.create_task(self._send_command({ "action": 'getsysinfo', "nodeid": nodeid, "nodeinfo": True }, "device_info", timeout)))
+            tasks.append(tg.create_task(self._send_command({ "action": 'nodes' }, "device_info", timeout=timeout)))
+            tasks.append(tg.create_task(self._send_command_no_response_id({ "action": 'getnetworkinfo', "nodeid": nodeid }, timeout=timeout)))
+            tasks.append(tg.create_task(self._send_command_no_response_id({ "action": 'lastconnect', "nodeid": nodeid }, timeout=timeout)))
+            tasks.append(tg.create_task(self._send_command({ "action": 'getsysinfo', "nodeid": nodeid, "nodeinfo": True }, "device_info", timeout=timeout)))
             tasks.append(tg.create_task(self.list_device_groups(timeout=timeout)))
 
         nodes, network, lastconnect, sysinfo, meshes = (_.result() for _ in tasks)
@@ -1344,7 +1391,7 @@ class Session(object):
         if consent is not None:
             op["consent"] = consent
 
-        data = await self._send_command(op, "edit_device", timeout)
+        data = await self._send_command(op, "edit_device", timeout=timeout)
 
         if data.get("result", "ok").lower() != "ok":
             raise exceptions.ServerError(data["result"])
@@ -1402,7 +1449,7 @@ class Session(object):
                         continue
                     result[node]["result"].append(event["value"])
         async def __(command):
-            data = await self._send_command(command, "run_command", timeout)
+            data = await self._send_command(command, "run_command", timeout=timeout)
 
             if data.get("result", "ok").lower() != "ok":
                 raise exceptions.ServerError(data["result"])
@@ -1479,7 +1526,7 @@ class Session(object):
         if isinstance(nodeids, str):
             nodeids = [nodeids]
 
-        return await self._send_command({ "action": 'wakedevices', "nodeids": nodeids }, "wake_devices", timeout)
+        return await self._send_command({ "action": 'wakedevices', "nodeids": nodeids }, "wake_devices", timeout=timeout)
 
     async def reset_devices(self, nodeids, timeout=None):
         '''
@@ -1500,7 +1547,7 @@ class Session(object):
         if isinstance(nodeids, str):
             nodeids = [nodeids]
 
-        return await self._send_command({ "action": 'poweraction', "nodeids": nodeids, "actiontype": 3 }, "reset_devices", timeout)
+        return await self._send_command({ "action": 'poweraction', "nodeids": nodeids, "actiontype": 3 }, "reset_devices", timeout=timeout)
 
     async def sleep_devices(self, nodeids, timeout=None):
         '''
@@ -1521,7 +1568,7 @@ class Session(object):
         if isinstance(nodeids, str):
             nodeids = [nodeids]
 
-        return await self._send_command({ "action": 'poweraction', "nodeids": nodeids, "actiontype": 4 }, "sleep_devices", timeout)
+        return await self._send_command({ "action": 'poweraction', "nodeids": nodeids, "actiontype": 4 }, "sleep_devices", timeout=timeout)
 
     async def power_off_devices(self, nodeids, timeout=None):
         ''' 
@@ -1542,7 +1589,7 @@ class Session(object):
         if isinstance(nodeids, str):
             nodeids = [nodeids]
 
-        return await self._send_command({ "action": 'poweraction', "nodeids": nodeids, "actiontype": 2 }, "power_off_devices", timeout)
+        return await self._send_command({ "action": 'poweraction', "nodeids": nodeids, "actiontype": 2 }, "power_off_devices", timeout=timeout)
 
     async def list_device_shares(self, nodeid, timeout=None):
         '''
@@ -1559,7 +1606,7 @@ class Session(object):
             :py:class:`~meshctrl.exceptions.SocketError`: Info about socket closure
             asyncio.TimeoutError: Command timed out
         '''
-        data = await self._send_command_no_response_id({ "action": 'deviceShares', "nodeid": nodeid }, timeout)
+        data = await self._send_command_no_response_id({ "action": 'deviceShares', "nodeid": nodeid }, timeout=timeout)
         if data.get("result", "ok").lower() != "ok":
             raise exceptions.ServerError(data["result"])
 
@@ -1604,7 +1651,7 @@ class Session(object):
             end = int(start.timestamp())
         if end <= start:
             raise ValueError("End time must be ahead of start time")
-        data = await self._send_command({ "action": 'createDeviceShareLink', "nodeid": nodeid, "guestname": name, "p": constants.SharingTypeEnum[type], "consent": consent, "start": start, "end": end }, "add_device_share", timeout)
+        data = await self._send_command({ "action": 'createDeviceShareLink', "nodeid": nodeid, "guestname": name, "p": constants.SharingTypeEnum[type], "consent": consent, "start": start, "end": end }, "add_device_share", timeout=timeout)
         
         if data.get("result", "ok").lower() != "ok":
             raise exceptions.ServerError(data["result"])
@@ -1633,7 +1680,7 @@ class Session(object):
             :py:class:`~meshctrl.exceptions.SocketError`: Info about socket closure
             asyncio.TimeoutError: Command timed out
         '''
-        data = await self._send_command({ "action": 'removeDeviceShare', "nodeid": nodeid, "publicid": shareid }, "remove_device_share", timeout)
+        data = await self._send_command({ "action": 'removeDeviceShare', "nodeid": nodeid, "publicid": shareid }, "remove_device_share", timeout=timeout)
         
         if data.get("result", "ok").lower() != "ok":
             raise exceptions.ServerError(data["result"])
@@ -1669,7 +1716,7 @@ class Session(object):
         tasks = []
         async with asyncio.TaskGroup() as tg:
             tasks.append(tg.create_task(asyncio.wait_for(_(), timeout=timeout)))
-            tasks.append({ "action": 'msg', "type": 'openUrl', "nodeid": nodeid, "url": url }, "device_open_url", timeout)
+            tasks.append({ "action": 'msg', "type": 'openUrl', "nodeid": nodeid, "url": url }, "device_open_url", timeout=timeout)
 
         res = tasks[1].result()
         success = tasks[2].result()
@@ -1701,7 +1748,7 @@ class Session(object):
             :py:class:`~meshctrl.exceptions.SocketError`: Info about socket closure
             asyncio.TimeoutError: Command timed out
          '''
-        data = await self._send_command({ "action": 'msg', "type": 'messagebox', "nodeid": nodeid, "title": title, "msg": message }, "device_message", timeout)
+        data = await self._send_command({ "action": 'msg', "type": 'messagebox', "nodeid": nodeid, "title": title, "msg": message }, "device_message", timeout=timeout)
 
         if data.get("result", "ok").lower() != "ok":
             raise exceptions.ServerError(data["result"])
@@ -1731,7 +1778,7 @@ class Session(object):
         if isinstance(nodeids, str):
             nodeids = [nodeids]
 
-        data = self._send_command({ "action": 'toast', "nodeids": nodeids, "title": "MeshCentral", "msg": message }, "device_toast", timeout)
+        data = self._send_command({ "action": 'toast', "nodeids": nodeids, "title": "MeshCentral", "msg": message }, "device_toast", timeout=timeout)
 
         if data.get("result", "ok").lower() != "ok":
             raise exceptions.ServerError(data["result"])
@@ -1817,17 +1864,20 @@ class Session(object):
             :py:class:`~meshctrl.exceptions.FileTransferCancelled`: File transfer cancelled. Info available on the `stats` property
 
         Returns:
-            io.IOBase: The stream which has been downloaded into. Cursor will be at the end of the stream.
+            io.IOBase: The stream which has been downloaded into. Cursor will be at the beginning of where the file is downloaded.
         '''
         if target is None:
             target = io.BytesIO()
+        start = target.tell()
         if unique_file_tunnel:
             async with self.file_explorer(nodeid) as files:
                 await files.download(source, target)
+                target.seek(start)
                 return target
         else:
             files = await self._cached_file_explorer(nodeid, nodeid)
             await files.download(source, target, timeout=timeout)
+            target.seek(start)
             return target
 
     async def download_file(self, nodeid, source, filepath, unique_file_tunnel=False, timeout=None):
@@ -1845,10 +1895,10 @@ class Session(object):
             :py:class:`~meshctrl.exceptions.FileTransferCancelled`: File transfer cancelled. Info available on the `stats` property
 
         Returns:
-            io.IOBase: The stream which has been downloaded into. Cursor will be at the end of the stream.
+            None
          '''
         with open(filepath, "wb") as f:
-            return await self.download(nodeid, source, f, unique_file_tunnel, timeout=timeout)
+            await self.download(nodeid, source, f, unique_file_tunnel, timeout=timeout)
 
     async def _cached_file_explorer(self, nodeid, _id):
         if (_id not in self._file_tunnels or not self._file_tunnels[_id].alive):

src/meshctrl/tunnel.py

@@ -4,6 +4,8 @@ import websockets.asyncio
 import websockets.asyncio.client
 import asyncio
 import ssl
+from python_socks.async_.asyncio import Proxy
+import urllib
 from . import exceptions
 from . import util
 from . import constants
@@ -52,10 +54,6 @@ class Tunnel(object):
                 ssl_context.verify_mode = ssl.CERT_NONE
                 options = { "ssl": ssl_context }
 
-            # Setup the HTTP proxy if needed
-            # if (self._session._proxy != None):
-            #     options.agent = new https_proxy_agent(urllib.parse(this._proxy))
-
             if (self.node_id.split('/') != 3) and (self._session._currentDomain is not None):
                 self.node_id = f"node/{self._session._currentDomain}/{self.node_id}"
 
@@ -72,13 +70,11 @@ class Tunnel(object):
 
             self.url = self._session.url.replace('/control.ashx', '/meshrelay.ashx?browser=1&p=' + str(self._protocol) + '&nodeid=' + self.node_id + '&id=' + self._tunnel_id + '&auth=' + authcookie["cookie"])
 
-            # headers = websockets.datastructures.Headers()
+            if self._session._proxy:
+                proxy = Proxy.from_url(self._session._proxy)
+                parsed = urllib.parse.urlparse(self.url)
+                options["sock"] = await proxy.connect(dest_host=parsed.hostname, dest_port=parsed.port)
 
-            # if (self._password):
-            #     token = self._token if self._token else b""
-            #     headers['x-meshauth'] = (base64.b64encode(self._user.encode()) + b',' + base64.b64encode(self._password.encode()) + token).decode()
-
-            # options["additional_headers"] = headers
             async for websocket in websockets.asyncio.client.connect(self.url, process_exception=util._process_websocket_exception, **options):
                 self.alive = True
                 self._socket_open.set()

tests/.gitignore

@@ -1 +1,2 @@
-/data
+data
+environment/scripts/meshcentral/users.json

tests/environment/__init__.py

@@ -54,13 +54,16 @@ class TestEnvironment(object):
         self._subp = None
         self.mcurl = "wss://localhost:8086"
         self.clienturl = "http://localhost:5000"
-        self._dockerurl = "host.docker.internal:8086"
+        self.dockerurl = "host.docker.internal:8086"
+        self.proxyurl = "http://localhost:3128"
 
     def __enter__(self):
         global _docker_process
         if _docker_process is not None:
             self._subp = _docker_process
             return self
+        # Destroy the env in case it wasn't killed correctly last time.
+        subprocess.check_call(["docker", "compose", "down"], stdout=subprocess.DEVNULL, cwd=thisdir)
         self._subp = _docker_process = subprocess.Popen(["docker", "compose", "up", "--build", "--force-recreate", "--no-deps"], stdout=subprocess.DEVNULL, cwd=thisdir)
         timeout = 30
         start = time.time()
@@ -88,7 +91,7 @@ class TestEnvironment(object):
         pass
 
     def create_agent(self, meshid):
-        return Agent(meshid, self.mcurl, self.clienturl, self._dockerurl)
+        return Agent(meshid, self.mcurl, self.clienturl, self.dockerurl)
 
 def _kill_docker_process():
     if _docker_process is not None:

tests/environment/compose.yaml

@@ -19,9 +19,9 @@ services:
     #   # mongodb data-directory - A must for data persistence
     #   - ./meshcentral/mongodb_data:/data/db
     networks:
-     - meshctrl
+      - meshctrl
     extra_hosts:
-    - "host.docker.internal:host-gateway"
+      - "host.docker.internal:host-gateway"
 
   meshcentral:
     restart: always
@@ -49,4 +49,21 @@ services:
     healthcheck:
       test: curl -k --fail https://localhost:443/ || exit 1
       interval: 5s
-      timeout: 120s
+      timeout: 120s
+
+  squid:
+    image: ubuntu/squid:latest
+    restart: unless-stopped
+    container_name: meshctrl-squid
+    ports:
+      - 3128:3128
+
+    networks:
+      - meshctrl
+
+    extra_hosts:
+      - "host.docker.internal:host-gateway"
+
+    volumes:
+      - ./config/squid/conf.d:/etc/squid/conf.d
+      - ./config/squid/squid.conf:/etc/squid/squid.conf

tests/environment/config/squid/conf.d/meshctrl.conf

@@ -0,0 +1,11 @@
+# Logs are managed by logrotate on Debian
+logfile_rotate 0
+
+acl all src all
+acl Safe_ports port 8086
+acl SSS_ports port 8086
+http_access allow all
+debug_options ALL,0 85,2 88,2
+
+# Set max_filedescriptors to avoid using system's RLIMIT_NOFILE. See LP: #1978272
+max_filedescriptors 1024

tests/environment/scripts/meshcentral/users.json

@@ -1 +0,0 @@
-{"admin": "3U6zP4iIes5ISH15XxjYLjJcCdw9jU0m", "privileged": "aiIO0zLMGsU7++FYVDNxhlpYlZ1andRB", "unprivileged": "Cz9OMV1wkVd9pXdWi4lkBAAu6TMt43MA"}

tests/test_files.py

@@ -7,7 +7,7 @@ import io
 import random
 
 async def test_commands(env):
-    async with meshctrl.session.Session(env.mcurl, user="admin", password=env.users["admin"], ignore_ssl=True) as admin_session:
+    async with meshctrl.Session("wss://" + env.dockerurl, user="admin", password=env.users["admin"], ignore_ssl=True, proxy=env.proxyurl) as admin_session:
         mesh = await admin_session.add_device_group("test", description="This is a test group", amtonly=False, features=0, consent=0, timeout=10)
         try:
             with env.create_agent(mesh.short_meshid) as agent:
@@ -53,7 +53,7 @@ async def test_commands(env):
             assert (await admin_session.remove_device_group(mesh.meshid, timeout=10)), "Failed to remove device group"
 
 async def test_upload_download(env):
-    async with meshctrl.session.Session(env.mcurl, user="admin", password=env.users["admin"], ignore_ssl=True) as admin_session:
+    async with meshctrl.Session("wss://" + env.dockerurl, user="admin", password=env.users["admin"], ignore_ssl=True, proxy=env.proxyurl) as admin_session:
         mesh = await admin_session.add_device_group("test", description="This is a test group", amtonly=False, features=0, consent=0, timeout=10)
         try:
             with env.create_agent(mesh.short_meshid) as agent:

tests/test_sanity.py

@@ -8,14 +8,30 @@ import ssl
 import requests
 
 async def test_sanity(env):
-    async with meshctrl.session.Session(env.mcurl, user="unprivileged", password=env.users["unprivileged"], ignore_ssl=True) as s:
+    async with meshctrl.Session(env.mcurl, user="unprivileged", password=env.users["unprivileged"], ignore_ssl=True) as s:
+        got_pong = asyncio.Event()
+        async def _():
+            async for raw in s.raw_messages():
+                if raw == '{action:"pong"}':
+                    got_pong.set()
+                    break
+        ping_task = None
+        async with asyncio.TaskGroup() as tg:
+            tg.create_task(asyncio.wait_for(_(), timeout=5))
+            tg.create_task(asyncio.wait_for(got_pong.wait(), timeout=5))
+            ping_task = tg.create_task(s.ping(timeout=10))
+        print("\ninfo ping: {}\n".format(ping_task.result()))
         print("\ninfo user_info: {}\n".format(await s.user_info()))
         print("\ninfo server_info: {}\n".format(await s.server_info()))
         pass
 
+async def test_proxy(env):
+    async with meshctrl.Session("wss://" + env.dockerurl, user="unprivileged", password=env.users["unprivileged"], ignore_ssl=True, proxy=env.proxyurl) as s:
+        pass
+
 async def test_ssl(env):
     try:
-        async with meshctrl.session.Session(env.mcurl, user="unprivileged", password=env.users["unprivileged"], ignore_ssl=False) as s:
+        async with meshctrl.Session(env.mcurl, user="unprivileged", password=env.users["unprivileged"], ignore_ssl=False) as s:
             pass
     except* ssl.SSLCertVerificationError:
         pass

tests/test_session.py

@@ -8,8 +8,8 @@ import io
 thisdir = os.path.dirname(os.path.realpath(__file__))
 
 async def test_admin(env):
-    async with meshctrl.session.Session(env.mcurl, user="admin", password=env.users["admin"], ignore_ssl=True) as admin_session,\
-               meshctrl.session.Session(env.mcurl, user="privileged", password=env.users["privileged"], ignore_ssl=True) as privileged_session:
+    async with meshctrl.Session(env.mcurl, user="admin", password=env.users["admin"], ignore_ssl=True) as admin_session,\
+               meshctrl.Session(env.mcurl, user="privileged", password=env.users["privileged"], ignore_ssl=True) as privileged_session:
         admin_users = await admin_session.list_users(timeout=10)
         print("\ninfo list_users: {}\n".format(admin_users))
         try:
@@ -34,22 +34,22 @@ async def test_admin(env):
 
 async def test_users(env):
     try:
-        async with meshctrl.session.Session(env.mcurl[3:], user="admin", password=env.users["admin"], ignore_ssl=True) as admin_session:
+        async with meshctrl.Session(env.mcurl[3:], user="admin", password=env.users["admin"], ignore_ssl=True) as admin_session:
             pass
     except* ValueError:
         pass
     else:
         raise Exception("Connected with bad URL")
     try:
-        async with meshctrl.session.Session(env.mcurl, user="admin", ignore_ssl=True) as admin_session:
+        async with meshctrl.Session(env.mcurl, user="admin", ignore_ssl=True) as admin_session:
             pass
     except* meshctrl.exceptions.MeshCtrlError:
         pass
     else:
         raise Exception("Connected with no password")
-    async with meshctrl.session.Session(env.mcurl+"/", user="admin", password=env.users["admin"], ignore_ssl=True) as admin_session,\
-               meshctrl.session.Session(env.mcurl, user="privileged", password=env.users["privileged"], ignore_ssl=True) as privileged_session,\
-               meshctrl.session.Session(env.mcurl, user="unprivileged", password=env.users["unprivileged"], ignore_ssl=True) as unprivileged_session:
+    async with meshctrl.Session(env.mcurl+"/", user="admin", password=env.users["admin"], ignore_ssl=True) as admin_session,\
+               meshctrl.Session(env.mcurl, user="privileged", password=env.users["privileged"], ignore_ssl=True) as privileged_session,\
+               meshctrl.Session(env.mcurl, user="unprivileged", password=env.users["unprivileged"], ignore_ssl=True) as unprivileged_session:
 
         assert len(await admin_session.list_users(timeout=10)) == 3, "Wrong number of users"
 
@@ -74,17 +74,17 @@ async def test_users(env):
         assert len(await admin_session.list_users(timeout=10)) == 3, "Failed to remove user"
 
 async def test_login_token(env):
-    async with meshctrl.session.Session(env.mcurl, user="unprivileged", password=env.users["unprivileged"], ignore_ssl=True) as s:
+    async with meshctrl.Session(env.mcurl, user="unprivileged", password=env.users["unprivileged"], ignore_ssl=True) as s:
         token = await s.add_login_token("test", expire=1, timeout=10)
         print("\ninfo add_login_token: {}\n".format(token))
 
-        async with meshctrl.session.Session(env.mcurl, user=token["tokenUser"], password=token["tokenPass"], ignore_ssl=True) as s2:
+        async with meshctrl.Session(env.mcurl, user=token["tokenUser"], password=token["tokenPass"], ignore_ssl=True) as s2:
             assert (await s2.user_info())["_id"] == (await s.user_info())["_id"], "Login token logged into wrong account"
         # Wait for the login token to expire
         await asyncio.sleep(65)
 
         try:
-            async with meshctrl.session.Session(env.mcurl, user=token["tokenUser"], password=token["tokenPass"], ignore_ssl=True) as s2:
+            async with meshctrl.Session(env.mcurl, user=token["tokenUser"], password=token["tokenPass"], ignore_ssl=True) as s2:
                 pass
         except:
             pass
@@ -94,7 +94,7 @@ async def test_login_token(env):
         token = await s.add_login_token("test2", timeout=10)
         token2 = await s.add_login_token("test3", timeout=10)
         print("\ninfo add_login_token_no_expire: {}\n".format(token))
-        async with meshctrl.session.Session(env.mcurl, user=token["tokenUser"], password=token["tokenPass"], ignore_ssl=True) as s2:
+        async with meshctrl.Session(env.mcurl, user=token["tokenUser"], password=token["tokenPass"], ignore_ssl=True) as s2:
             assert (await s2.user_info())["_id"] == (await s.user_info())["_id"], "Login token logged into wrong account"
 
         r = await s.list_login_tokens(timeout=10)
@@ -107,9 +107,9 @@ async def test_login_token(env):
         assert len(await s.remove_login_token([token2["name"]], timeout=10)) == 0, "Residual login tokens"
 
 async def test_mesh_device(env):
-    async with meshctrl.session.Session(env.mcurl, user="admin", password=env.users["admin"], ignore_ssl=True) as admin_session,\
-               meshctrl.session.Session(env.mcurl, user="privileged", password=env.users["privileged"], ignore_ssl=True) as privileged_session,\
-               meshctrl.session.Session(env.mcurl, user="unprivileged", password=env.users["unprivileged"], ignore_ssl=True) as unprivileged_session:
+    async with meshctrl.Session(env.mcurl, user="admin", password=env.users["admin"], ignore_ssl=True) as admin_session,\
+               meshctrl.Session(env.mcurl, user="privileged", password=env.users["privileged"], ignore_ssl=True) as privileged_session,\
+               meshctrl.Session(env.mcurl, user="unprivileged", password=env.users["unprivileged"], ignore_ssl=True) as unprivileged_session:
         # Test creating a mesh
         mesh = await admin_session.add_device_group("test", description="This is a test group", amtonly=False, features=0, consent=0, timeout=10)
         print("\ninfo add_device_group: {}\n".format(mesh))
@@ -266,9 +266,9 @@ async def test_mesh_device(env):
         assert not (await admin_session.add_users_to_device_group((await privileged_session.user_info())["_id"], mesh.meshid, rights=meshctrl.constants.MeshRights.fullrights, timeout=5))[(await privileged_session.user_info())["_id"]]["success"], "Added user to device group which doesn't exist?"
 
 async def test_user_groups(env):
-    async with meshctrl.session.Session(env.mcurl, user="admin", password=env.users["admin"], ignore_ssl=True) as admin_session,\
-               meshctrl.session.Session(env.mcurl, user="privileged", password=env.users["privileged"], ignore_ssl=True) as privileged_session,\
-               meshctrl.session.Session(env.mcurl, user="unprivileged", password=env.users["unprivileged"], ignore_ssl=True) as unprivileged_session:
+    async with meshctrl.Session(env.mcurl, user="admin", password=env.users["admin"], ignore_ssl=True) as admin_session,\
+               meshctrl.Session(env.mcurl, user="privileged", password=env.users["privileged"], ignore_ssl=True) as privileged_session,\
+               meshctrl.Session(env.mcurl, user="unprivileged", password=env.users["unprivileged"], ignore_ssl=True) as unprivileged_session:
         
         user_group = await admin_session.add_user_group("test", description="aoeu")
         print("\ninfo add_user_group: {}\n".format(user_group))
@@ -294,7 +294,7 @@ async def test_user_groups(env):
         assert await admin_session.remove_user_group(user_group2.id.split("/")[-1])
 
 async def test_events(env):
-    async with meshctrl.session.Session(env.mcurl, user="admin", password=env.users["admin"], ignore_ssl=True) as admin_session:
+    async with meshctrl.Session(env.mcurl, user="admin", password=env.users["admin"], ignore_ssl=True) as admin_session:
         await admin_session.list_events()
         mesh = await admin_session.add_device_group("test", description="This is a test group", amtonly=False, features=0, consent=0, timeout=10)
         try:
@@ -310,7 +310,7 @@ async def test_events(env):
                         await asyncio.sleep(1)
                     else:
                         break
-                async with meshctrl.session.Session(env.mcurl, user="privileged", password=env.users["privileged"], ignore_ssl=True) as privileged_session:
+                async with meshctrl.Session(env.mcurl, user="privileged", password=env.users["privileged"], ignore_ssl=True) as privileged_session:
                     
                     # assert len(await privileged_session.list_events()) == 0, "non-admin user has access to admin events"
 
@@ -337,8 +337,8 @@ async def test_events(env):
             assert (await admin_session.remove_device_group(mesh.meshid, timeout=10)), "Failed to remove device group"
             
 async def test_interuser(env):
-    async with meshctrl.session.Session(env.mcurl, user="admin", password=env.users["admin"], ignore_ssl=True) as admin_session,\
-               meshctrl.session.Session(env.mcurl, user="privileged", password=env.users["privileged"], ignore_ssl=True) as privileged_session:
+    async with meshctrl.Session(env.mcurl, user="admin", password=env.users["admin"], ignore_ssl=True) as admin_session,\
+               meshctrl.Session(env.mcurl, user="privileged", password=env.users["privileged"], ignore_ssl=True) as privileged_session:
         got_message = asyncio.Event()
         async def _():
             async for message in admin_session.events({"action": "interuser"}):
@@ -361,7 +361,7 @@ async def test_interuser(env):
             tg.create_task(asyncio.wait_for(got_message.wait(), 5))
 
 async def test_session_files(env):
-    async with meshctrl.session.Session(env.mcurl, user="admin", password=env.users["admin"], ignore_ssl=True) as admin_session:
+    async with meshctrl.Session(env.mcurl, user="admin", password=env.users["admin"], ignore_ssl=True) as admin_session:
         mesh = await admin_session.add_device_group("test", description="This is a test group", amtonly=False, features=0, consent=0, timeout=10)
         try:
             with env.create_agent(mesh.short_meshid) as agent:
@@ -396,11 +396,9 @@ async def test_session_files(env):
                 assert r["size"] == len(randdata), "Uploaded wrong number of bytes"
 
                 s = await admin_session.download(agent.nodeid, f"{pwd}/test", timeout=5)
-                s.seek(0)
                 assert s.read() == randdata, "Downloaded bad data"
 
                 await admin_session.download(agent.nodeid, f"{pwd}/test", downfilestream, timeout=5)
-                downfilestream.seek(0)
                 assert downfilestream.read() == randdata, "Downloaded bad data"
 
                 await admin_session.download_file(agent.nodeid, f"{pwd}/test2", os.path.join(thisdir, "data", "test"), timeout=5)

tests/test_shell.py

@@ -5,7 +5,7 @@ import meshctrl
 import requests
 
 async def test_shell(env):
-    async with meshctrl.session.Session(env.mcurl, user="admin", password=env.users["admin"], ignore_ssl=True) as admin_session:
+    async with meshctrl.Session(env.mcurl, user="admin", password=env.users["admin"], ignore_ssl=True) as admin_session:
         mesh = await admin_session.add_device_group("test", description="This is a test group", amtonly=False, features=0, consent=0, timeout=10)
         try:
             with env.create_agent(mesh.short_meshid) as agent:
@@ -40,7 +40,7 @@ async def test_shell(env):
 
 
 async def test_smart_shell(env):
-    async with meshctrl.session.Session(env.mcurl, user="admin", password=env.users["admin"], ignore_ssl=True) as admin_session:
+    async with meshctrl.Session(env.mcurl, user="admin", password=env.users["admin"], ignore_ssl=True) as admin_session:
         mesh = await admin_session.add_device_group("test", description="This is a test group", amtonly=False, features=0, consent=0, timeout=10)
         try:
             with env.create_agent(mesh.short_meshid) as agent: