importsocketimportthreadingimporttimefrommultiprocessing.connectionimportClient,Listenerdef_test(*,listener:str|None,client:str|None)->tuple[Exception|None,Exception|None,list]:_autkey_listener=listener.encode()iflistenerelseNone_authkey_client=client.encode()ifclientelseNonewithsocket.socket(socket.AF_INET,socket.SOCK_STREAM)ass:s.bind(("localhost",0))address=s.getsockname()thread_exc=Nonethread_got= []client_exc=Nonedeftarget()->None:try:withListener(address,authkey=_autkey_listener)asl:c=l.accept()whileTrue:value=c.recv()ifnotvalue:breakthread_got.append(value)exceptExceptionase:nonlocalthread_excthread_exc=et=threading.Thread(target=target)t.start()try:withClient(address,authkey=_authkey_client)asc:c.send("hello")time.sleep(0.1)c.send(None)t.join()exceptExceptionase:client_exc=ereturnthread_exc,client_exc,thread_gotif__name__=="__main__":print(" no passwords ".center(40,"-"))print(_test(listener=None,client=None))print(" two matching passwords ".center(40,"-"))print(_test(listener="password",client="password"))print(" client has wrong password ".center(40,"-"))print(_test(listener="password",client="wrong"))print(" client has no password ".center(40,"-"))print(_test(listener="password",client=None))# This is the case which deadlocksprint(" listener has no password ".center(40,"-"))print(_test(listener=None,client="password"))

------------- no passwords -------------(None, None, ['hello'])-------- two matching passwords --------(None, None, ['hello'])------ client has wrong password -------(AuthenticationError('digest received was wrong'), AuthenticationError('digest sent was rejected'), [])-------- client has no password --------(AuthenticationError("expected 'md5' of length 16 got 20"), None, [])------- listener has no password -------

ifauthkeyisnotNone:answer_challenge(c,authkey)deliver_challenge(c,authkey)