Update CA for Python 3.13+

Author: YeisonCardona

chaski/node.py

@@ -610,11 +610,13 @@ def __init__(
 
         # Request an SSL certificate for secure communication if specified
         if request_ssl_certificate:
-            asyncio.sleep(1)
-            asyncio.create_task(self.request_ssl_certificate(request_ssl_certificate))
-
-    #             if os.path.exists(self.ssl_context_client,
-    #                               self.ssl_context_server)
+            loop = asyncio.get_event_loop()
+            loop.call_later(
+                1,
+                lambda: asyncio.create_task(
+                    self.request_ssl_certificate(request_ssl_certificate)
+                ),
+            )
 
     # ----------------------------------------------------------------------
     def __repr__(self) -> str:

chaski/scripts/remote_proxy.py

@@ -6,23 +6,30 @@
 
 logging.basicConfig(level=logging.DEBUG)
 
-parser = argparse.ArgumentParser(description='Chaski Remote Server')
+parser = argparse.ArgumentParser(description="Chaski Remote Server")
 parser.add_argument(
-    '-p',
-    '--port',
+    "-i",
+    "--ip",
     type=str,
-    default='65434',
-    help='Port number to run the server on',
+    default="127.0.0.1",
+    help="IP address to run the server on",
 )
 parser.add_argument(
-    '-n',
-    '--name',
+    "-p",
+    "--port",
     type=str,
-    default='ChaskiRemote',
-    help='Name of the server',
+    default="65434",
+    help="Port number to run the server on",
 )
 parser.add_argument(
-    'modules', type=str, help='Comma-separated list of available modules'
+    "-n",
+    "--name",
+    type=str,
+    default="ChaskiRemote",
+    help="Name of the server",
+)
+parser.add_argument(
+    "modules", type=str, help="Comma-separated list of available modules"
 )
 
 args = parser.parse_args()
@@ -32,9 +39,10 @@
 async def run():
     """"""
     server = ChaskiRemote(
+        ip=args.ip,
         port=args.port,
         name=args.name,
-        available=args.modules.split(','),
+        available=args.modules.split(","),
         run=False,
     )
 
@@ -48,5 +56,5 @@ def main():
     asyncio.run(run())
 
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     main()

chaski/scripts/terminate_connections.py

@@ -6,7 +6,6 @@
 # ----------------------------------------------------------------------
 def close_connections(port):
     """"""
-    print(f"Cleaning connections on port {port}...")
 
     try:
         # Run the lsof command to get the PIDs of active connections on the specified port
@@ -17,9 +16,7 @@ def close_connections(port):
         )
         pids = result.stdout.strip().splitlines()
 
-        if not pids:
-            print(f"No active connections on port {port}.")
-        else:
+        if pids:
             # Close all active connections
             for pid in pids:
                 print(f"Closing connection with PID {pid}...")
@@ -37,18 +34,19 @@ def main():
 
     try:
         port_range = sys.argv[1]
-        START_PORT, END_PORT = map(int, port_range.split('-'))
+        START_PORT, END_PORT = map(int, port_range.split("-"))
         if START_PORT > END_PORT:
             raise ValueError(
                 "The start port must be less than or equal to the end port."
             )
     except ValueError as e:
-        print(f"Error in port range: {e}")
-        sys.exit(1)
+        START_PORT = 65440
+        END_PORT = 65500
 
     # Iterate over the range of ports and close active connections
     for port in range(START_PORT, END_PORT + 1):
         close_connections(port)
 
+
 if __name__ == "__main__":
-    main()
+    main()

chaski/streamer.py

@@ -39,7 +39,7 @@ class ChaskiStreamer(ChaskiNode):
     # ----------------------------------------------------------------------
     def __init__(
         self,
-        destination_folder: str = '.',
+        destination_folder: str = ".",
         chunk_size: int = 8192,
         file_handling_callback: callable = None,
         allow_incoming_files: bool = False,
@@ -48,6 +48,8 @@ def __init__(
         *args: tuple,
         **kwargs: dict,
     ):
+        # Initialize lock for file transfer to prevent concurrent transfers
+        self._file_transfer_lock = asyncio.Lock()
         """
         Initialize a new instance of ChaskiStreamer.
 
@@ -86,8 +88,8 @@ def __init__(
         self.terminate_stream_flag = False
 
         self.enable_message_propagation()
-        self.add_propagation_command('ChaskiMessage')
-        self.add_propagation_command('ChaskiStorageRequest')
+        self.add_propagation_command("ChaskiMessage")
+        self.add_propagation_command("ChaskiStorageRequest")
 
         if persistent_storage:
             self.persistent_storage = PersistentStorage()
@@ -101,7 +103,7 @@ def __repr__(self):
         such as the IP address and port. If the instance is a root node, it prepends an
         asterisk (*) to the string.
         """
-        h = '*' if self.paired else ''
+        h = "*" if self.paired else ""
         return h + self.address
 
     # ----------------------------------------------------------------------
@@ -122,7 +124,7 @@ def address(self) -> str:
 
     # ----------------------------------------------------------------------
     @classmethod
-    def get_hash(cls, file: str, algorithm: str = 'sha256') -> str:
+    def get_hash(cls, file: str, algorithm: str = "sha256") -> str:
         """
         Compute the hash of a file using the specified algorithm.
 
@@ -143,7 +145,7 @@ def get_hash(cls, file: str, algorithm: str = 'sha256') -> str:
             The hexadecimal hash digest of the file.
         """
         hash_func = hashlib.new(algorithm)
-        with open(file, 'rb') as f:
+        with open(file, "rb") as f:
             while chunk := f.read(8192):  # Read the file in 8192 byte blocks
                 hash_func.update(chunk)
         return hash_func.hexdigest()
@@ -188,7 +190,7 @@ def _get_status(self, **kwargs) -> dict:
         }
 
     # ----------------------------------------------------------------------
-    async def __aenter__(self) -> Generator['Message', None, None]:
+    async def __aenter__(self) -> Generator["Message", None, None]:
         """
         Enter the asynchronous context for streaming messages.
 
@@ -208,7 +210,7 @@ async def __aexit__(
         self,
         exception_type: type,
         exception_value: BaseException,
-        exception_traceback: 'TracebackType',
+        exception_traceback: "TracebackType",
     ) -> None:
         """
         Exit the runtime context related to this object and stop the streamer.
@@ -250,10 +252,10 @@ async def push(self, topic: str, data: bytes = None) -> None:
         data : bytes, optional
             The byte-encoded data to be sent with the message. This could be any binary payload that subscribers are expected to process.
         """
-        await self._write('ChaskiMessage', data=data, topic=topic)
+        await self._write("ChaskiMessage", data=data, topic=topic)
 
     # ----------------------------------------------------------------------
-    async def _process_ChaskiMessage(self, message: 'Message', edge: 'Edge') -> None:
+    async def _process_ChaskiMessage(self, message: "Message", edge: "Edge") -> None:
         """
         Process an incoming Chaski message and place it onto the message queue.
 
@@ -307,7 +309,7 @@ def deactivate_file_transfer(self) -> None:
         self.allow_incoming_files = False
 
     # ----------------------------------------------------------------------
-    async def message_stream(self) -> Generator['Message', None, None]:
+    async def message_stream(self) -> Generator["Message", None, None]:
         """
         Asynchronously generate messages from the message queue.
 
@@ -358,7 +360,7 @@ def terminate_stream(self) -> None:
     async def push_file(
         self,
         topic: str,
-        file: 'IOBase',
+        file: "IOBase",
         filename: str = None,
         data: dict = {},
     ):
@@ -386,37 +388,44 @@ async def push_file(
         This method uses asynchronous I/O to read the file in chunks and send each chunk
         without blocking the event loop. It ensures that the entire file is processed and sent
         even if the process involves multiple chunks.
+
+        This method uses a lock to ensure only one file transfer happens at a time.
         """
-        size = 0
-        # Initialize a SHA-256 hash function for computing the hash digest of the file chunks
-        hash_func = hashlib.new('sha256')
-        while True:
-            # Read the next chunk of data from the file up to the specified chunk size
-            chunk = file.read(self.chunk_size)
-            # Increment the size by the length of the current chunk
-            size += len(chunk)
-            # Update the hash function with the current chunk of data.
-            hash_func.update(chunk)
-            # Package the chunked file data along with metadata such as filename, hash, and chunk size
-            package_data = {
-                'filename': (filename if filename else os.path.split(file.name)[-1]),
-                'chunk': chunk,
-                'hash': hash_func.hexdigest(),
-                'data': data,
-                'chunk_size': self.chunk_size,
-                'size': size,
-            }
-
-            # Send the chunked file data as a message to the specified topic and yield control to the event loop
-            await self._write('ChaskiFile', data=package_data, topic=topic)
-            await asyncio.sleep(0)  # very important sleep
-
-            # If no more chunks are available to read, the file transfer is complete
-            if not chunk:
-                break
+
+        # Acquire the lock to ensure only one file transfer at a time
+        async with self._file_transfer_lock:
+            size = 0
+            # Initialize a SHA-256 hash function for computing the hash digest of the file chunks
+            hash_func = hashlib.new("sha256")
+
+            while True:
+                # Read the next chunk of data from the file up to the specified chunk size
+                chunk = file.read(self.chunk_size)
+                # If no more chunks are available to read, the file transfer is complete
+                if not chunk:
+                    break
+                # Increment the size by the length of the current chunk
+                size += len(chunk)
+                # Update the hash function with the current chunk of data.
+                hash_func.update(chunk)
+                # Package the chunked file data along with metadata such as filename, hash, and chunk size
+                package_data = {
+                    "filename": (
+                        filename if filename else os.path.split(file.name)[-1]
+                    ),
+                    "chunk": chunk,
+                    "hash": hash_func.hexdigest(),
+                    "data": data,
+                    "chunk_size": self.chunk_size,
+                    "size": size,
+                }
+
+                # Send the chunked file data as a message to the specified topic and yield control to the event loop
+                await self._write("ChaskiFile", data=package_data, topic=topic)
+                await asyncio.sleep(0)  # very important sleep
 
     # ----------------------------------------------------------------------
-    async def _process_ChaskiFile(self, message: 'Message', edge: 'Edge') -> None:
+    async def _process_ChaskiFile(self, message: "Message", edge: "Edge") -> None:
         """
         Process an incoming ChaskiFile message and append each chunk of data to the target file.
 
@@ -444,10 +453,10 @@ async def _process_ChaskiFile(self, message: 'Message', edge: 'Edge') -> None:
             return
 
         # Append incoming file chunk data to the target file in append-binary mode
-        if chunk := message.data.pop('chunk'):
+        if chunk := message.data.pop("chunk"):
             with open(
-                os.path.join(self.destination_folder, message.data['filename']),
-                'ab',
+                os.path.join(self.destination_folder, message.data["filename"]),
+                "ab",
             ) as file:
                 # Write the current chunk to the target file in append-binary mode
                 file.write(chunk)
@@ -459,7 +468,7 @@ async def _process_ChaskiFile(self, message: 'Message', edge: 'Edge') -> None:
                 self.file_handling_callback(
                     **{
                         **message.data,
-                        'destiny_folder': self.destination_folder,
+                        "destiny_folder": self.destination_folder,
                     }
                 )
 

test/test_connection.py

@@ -27,6 +27,7 @@
 import asyncio
 from chaski.utils.auto import create_nodes
 from typing import Optional
+from chaski.scripts import terminate_connections
 
 
 ########################################################################
@@ -40,8 +41,11 @@ class _TestConnections:
     communication protocols, such as IPv4 and IPv6.
     """
 
+    def tearDown(self):
+        terminate_connections.main()
+
     # ----------------------------------------------------------------------
-    async def _close_nodes(self, nodes: list['ChaskiNode']):
+    async def _close_nodes(self, nodes: list["ChaskiNode"]):
         """
         Close all ChaskiNode instances in the provided list.
 
@@ -54,11 +58,12 @@ async def _close_nodes(self, nodes: list['ChaskiNode']):
             A list containing instances of ChaskiNode that need to be stopped.
         """
         for node in nodes:
+            await asyncio.sleep(0.3)
             await node.stop()
 
     # ----------------------------------------------------------------------
     def assertConnection(
-        self, node1: 'ChaskiNode', node2: 'ChaskiNode', msg: Optional[str] = None
+        self, node1: "ChaskiNode", node2: "ChaskiNode", msg: Optional[str] = None
     ):
         """
         Assert that two ChaskiNodes are connected to each other.
@@ -358,7 +363,7 @@ async def test_response_udp(self):
         }
         response_data = await nodes[0]._test_generic_request_udp(dummy_data)
         self.assertEqual(
-            dummy_data, response_data, 'Mismatch between sent and received data'
+            dummy_data, response_data, "Mismatch between sent and received data"
         )
 
         await self._close_nodes(nodes)
@@ -414,7 +419,7 @@ async def asyncSetUp(self) -> None:
         -----
         This method is automatically invoked by the testing framework and typically does not need to be called explicitly.
         """
-        self.ip = '127.0.0.1'
+        self.ip = "127.0.0.1"
         await asyncio.sleep(0)
 
     async def test_single_connections(self):
@@ -489,7 +494,7 @@ async def asyncSetUp(self) -> None:
         -----
         This method is automatically invoked by the testing framework and typically does not need to be called explicitly.
         """
-        self.ip = '::1'
+        self.ip = "::1"
         await asyncio.sleep(0)
 
     async def test_single_connections(self):
@@ -514,5 +519,5 @@ async def test_response_udp(self):
         return await super().test_response_udp()
 
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     unittest.main()