avoid running Process instances beacuse of a bug in python3.13 python/cpython#134381

Author: anadrianmanrique

impacket/smbserver.py

@@ -4887,9 +4887,10 @@ class SimpleSMBServer:
     :param string configFile: a file with all the servers' configuration. If no file specified, this class will create the basic parameters needed to run. You will need to add your shares manually tho. See addShare() method
     """
 
-    def __init__(self, listenAddress='0.0.0.0', listenPort=445, configFile=''):
+    def __init__(self, listenAddress='0.0.0.0', listenPort=445, configFile='', smbserverclass=SMBSERVER):
         if configFile != '':
-            self.__server = SMBSERVER((listenAddress, listenPort))
+            #self.__server = SMBSERVER((listenAddress, listenPort))
+            self.__server = smbserverclass((listenAddress, listenPort))
             self.__server.processConfigFile(configFile)
             self.__smbConfig = None
         else:
@@ -4914,7 +4915,7 @@ def __init__(self, listenAddress='0.0.0.0', listenPort=445, configFile=''):
             self.__smbConfig.set('IPC$', 'read only', 'yes')
             self.__smbConfig.set('IPC$', 'share type', '3')
             self.__smbConfig.set('IPC$', 'path', '')
-            self.__server = SMBSERVER((listenAddress, listenPort), config_parser=self.__smbConfig)
+            self.__server = smbserverclass((listenAddress, listenPort), config_parser=self.__smbConfig)
             self.__server.processConfigFile()
 
         # Now we have to register the MS-SRVS server. This specially important for
@@ -4926,6 +4927,9 @@ def __init__(self, listenAddress='0.0.0.0', listenPort=445, configFile=''):
         self.__server.registerNamedPipe('srvsvc', ('127.0.0.1', self.__srvsServer.getListenPort()))
         self.__server.registerNamedPipe('wkssvc', ('127.0.0.1', self.__wkstServer.getListenPort()))
 
+    def getServer(self):
+        return self.__server
+
     def start(self):
         self.__srvsServer.start()
         self.__wkstServer.start()

tests/SMB_RPC/test_smbserver.py

@@ -83,9 +83,36 @@
 from six import PY2, StringIO, BytesIO, b, assertRaisesRegex, assertCountEqual
 
 from impacket.smb import SMB_DIALECT
-from impacket.smbserver import normalize_path, isInFileJail, SimpleSMBServer
+from impacket.smbserver import normalize_path, isInFileJail, SimpleSMBServer, SMBSERVER
 from impacket.smbconnection import SMBConnection, SessionError, compute_lmhash, compute_nthash
+from threading import Thread
 
+import select
+import socket
+
+class StoppableMixin():
+    def serve_forever(self):
+        self.must_serve = True
+        self.timeout = 0.1
+
+        while self.must_serve:
+            self.handle_request()
+
+    def close_request(self,request):
+        if self.must_serve:
+            request.close()
+
+    def get_request(self):
+        timeout = 0.1
+        while self.must_serve:
+            _read,_,_ = select.select([self.socket],[],[],timeout)
+            
+            if _read and self.must_serve:
+                return self.socket.accept()
+        raise socket.error
+
+class SMBSERVERForTests(StoppableMixin,SMBSERVER):
+    pass
 
 class SMBServerUnitTests(unittest.TestCase):
     """Unit tests for the SMBServer
@@ -214,7 +241,10 @@ def tearDown(self):
         self.stop_smbserver()
 
     def get_smbserver(self, add_credential=True, add_share=True):
-        smbserver = SimpleSMBServer(listenAddress=self.address, listenPort=int(self.port))
+        #smbserver = SimpleSMBServerForTests(listenAddress=self.address, listenPort=int(self.port))
+        # smbserver should be run in a host thread and also be able to be terminated in order to run several times
+        # different configurations.
+        smbserver = SimpleSMBServer(listenAddress=self.address, listenPort=int(self.port),smbserverclass=SMBSERVERForTests)
         if add_credential:
             smbserver.addCredential(self.username, 0, self.lmhash, self.nthash)
         if add_share:
@@ -232,18 +262,25 @@ def start_smbserver(self, server):
         """Starts the SimpleSMBServer process.
         """
         self.server = server
-        self.server_process = Process(target=server.start)
+        #self.server_process = Process(target=server.start)
+        # avoid using Process beacuse of bug in python3.13 https://github.com/python/cpython/issues/134381
+        # TODO: remove these changes once a bugfix gets backported.
+        self.server_process = Thread(target=server.start)
+        self.server_process.daemon = True
         self.server_process.start()
 
     def stop_smbserver(self):
         """Stops the SimpleSMBServer process and wait for insider threads to join.
         """
         if self.server:
             self.server.stop()
+            self.server.getServer().must_serve=False
             self.server = None
         if self.server_process:
-            self.server_process.terminate()
+            #self.server_process.terminate()
+            #self.server_process._stop()
             sleep(0.1)
+            self.server_process.join()
             self.server_process = None
 
     def test_smbserver_login_valid(self):