添加互斥锁

添加一些打印

.gitignore

@@ -1 +1,2 @@
 __pycache__/
+*.log

ReadMe.txt

@@ -5,3 +5,7 @@
 2024.5.22
   实现端口转发功能，经测试，ssh、http可以正常使用
 
+2024.5.23
+  其中一方断线后，通知另一方断开连接
+
+

log.py

@@ -0,0 +1,28 @@
+import time
+
+
+
+# 同一个进程中所有调用这个文件的  .py 文件都使用这个变量
+_log_fp=None
+
+def _time():
+    return '['+time.strftime("%Y-%m-%d %H:%M:%S")+']'
+
+def myprint_dec(func):
+    def wrapper(*args, **kwargs):
+        # 在这里添加额外的功能
+        # print(_time(), str(e),file=_log_fp)
+        print(_time(), *args, **kwargs)
+        kwargs["file"]=_log_fp
+        result = func(_time(),*args, **kwargs)
+        _log_fp.flush()
+        return result
+    return wrapper
+
+myprint=myprint_dec(print)
+
+def log_init(file_name:str):
+    global _log_fp
+    if _log_fp is None:
+        _log_fp=open(file_name,mode="w+",encoding="utf-8")
+

server.py

@@ -3,29 +3,33 @@ import threading
 import os
 import sys
 import json
+import time
 
 
 import prot_codec as pc
 import target as tg
+from log import myprint
+from log import log_init
 
 
 
-# 服务器端脚本，用于转发端口数据
+# 代理服务器端脚本，用于转发端口数据
 
 SERVER_PORT = 5345
 
 
 
+
 # 定义个函数,使其专门重复处理客户的请求数据（也就是重复接受一个用户的消息并且重复回答，直到用户选择下线）
-def dispose_client_request(tcp_client_1,tcp_addr):
-    print(f"客户端：{tcp_addr} 已连接")
+def dispose_client_request(tcp_client,tcp_addr):
+    myprint(f"客户端：{tcp_addr} 已连接")
     recv_data=bytearray()
-    target =tg.tcp_target(tcp_client_1)
+    target =tg.tcp_target(tcp_client)
     while True:
         try:
-            recv = tcp_client_1.recv(4096)
+            recv = tcp_client.recv(4096)
         except Exception as e:
-            print(str(e))
+            myprint(str(e))
             break
         if recv:
             recv_data+=recv
@@ -38,8 +42,8 @@ def dispose_client_request(tcp_client_1,tcp_addr):
                 recv_data=recv_data[end+1:]
         else:
             break
-    print(f"客户端：{tcp_addr} 已下线")
-    tcp_client_1.close()
+    myprint(f"客户端：{tcp_addr} 已下线")
+    target.close()
 
 
 
@@ -48,16 +52,17 @@ def start_service():
     tcp_server.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,True)
     tcp_server.bind(("",SERVER_PORT))
 
-    print(f"开始监听 ({SERVER_PORT})")
+    myprint(f"开始监听 ({SERVER_PORT})")
     tcp_server.listen(128)
 
     while True:
-        tcp_client_1 , tcp_client_address = tcp_server.accept()
-        thd = threading.Thread(target = dispose_client_request, args = (tcp_client_1,tcp_client_address))
+        tcp_client , tcp_client_address = tcp_server.accept()
+        thd = threading.Thread(target = dispose_client_request, args = (tcp_client,tcp_client_address))
         # 设置守护主线程  即如果主线程结束了 那子线程中也都销毁了  防止主线程无法退出
         # thd.setDaemon(True)
         thd.start()
 
 
 if __name__ == "__main__":
+    log_init("proxy_service.log")
     start_service()

target.py

@@ -7,6 +7,8 @@ import socket
 
 
 import prot_codec as pc
+from log import myprint
+from log import log_init
 
 
 
@@ -14,44 +16,79 @@ import prot_codec as pc
 # device:
 # server
 # client
+# proxy
 # 定义操作类型
 # option:
 # connect
 # disconnect
 # data
 # login
+# close
 
 
-_tcp_server=None
-_tcp_client=None
+# 这里是公共变量，所有tcp_target对象都可以访问
+# 只支持一对一连接
+_tcp_server:socket.socket=None
+_tcp_client:socket.socket=None
 
 
 
 class tcp_target(object):
-  def __init__(self,tcp_handler:socket):
+  def __init__(self,tcp_handler:socket.socket):
     self.handler=tcp_handler
     self.state="idle"
   def recv_handler(self,recv_data:bytearray):
     global _tcp_client
     global _tcp_server
+    # 解包查看数据类型
     cmd,data=pc.decode(recv_data)
-    print(cmd.decode('utf-8'))
+    myprint(cmd.decode('utf-8'))
     j=json.loads(cmd)
     if(j["device"]=="client"):
       if(j["option"]=="login"):
+        # 登陆时保存客户端句柄，断开之前的句柄
         if(_tcp_client is not None):
           _tcp_client.close()
         _tcp_client=self.handler
       else:
+        # 其他消息原样发送到服务器端
         if(_tcp_server is not None):
           _tcp_server.send(recv_data)
     if(j["device"]=="server"):
       if(j["option"]=="login"):
+        # 登陆时保存客户端句柄，断开之前的句柄
         if(_tcp_server is not None):
           _tcp_server.close()
         _tcp_server=self.handler
+      elif(j["option"]=="keeplive"):
+        # 回复心跳数据
+        cmd={'device':'proxy','option':'keeplive'}
+        data=pc.encode(json.dumps(cmd).encode('utf-8'),b'default')
+        if(_tcp_server is not None):
+          _tcp_server.send(data)
       else:
+        # 其他消息原样发送到客户端
         if(_tcp_client is not None):
           _tcp_client.send(recv_data)
+  # 通知服务器和客户端断开连接
+  def close(self):
+    cmd={'device':'proxy','option':'close'}
+    data=pc.encode(json.dumps(cmd).encode('utf-8'),b'default')
+    # 这里有可能连接已经断开，无法发送
+    if(_tcp_client is not None):
+      try:
+        _tcp_client.send(data)
+        _tcp_client.close()
+      except Exception as e:
+        myprint("target_close:",str(e))
+      _tcp_client=None
+    if(_tcp_server is not None):
+      try:
+        _tcp_server.send(data)
+        _tcp_server.close()
+      except Exception as e:
+        myprint("target_close:",str(e))
+      _tcp_server=None
+
 
 

target_client.py

@@ -10,6 +10,8 @@ import time
 
 
 import prot_codec as pc
+from log import myprint
+from log import log_init
 
 
 
@@ -23,6 +25,13 @@ _LOCAL_PORT = 31234
 
 _remote_client=None
 _local_client=[]
+_local_server=None
+
+
+
+
+_local_client_lock = threading.Lock()
+
 
 
 
@@ -30,28 +39,45 @@ _local_client=[]
 def send_to(ip,port,data:bytearray):
   for item in _local_client:
     if(item[1]==ip and item[2]==port):
-      print(f"recv from remote {ip},{port}")
       item[0].send(data)
       return
-  print(f"can not fond {ip},{port}")
+  myprint(f"can not fond {ip}:{port}")
 
 
 # 关闭指定地址的端口
 def close(ip,port):
+  global _local_client
+  global _local_client_lock
+  _local_client_lock.acquire()
   for item in _local_client:
     if(item[1]==ip and item[2]==port):
+      myprint(f'断开连接 {ip}:{port}')
       item[0].close()
+      _local_client.remove(item)
       break
+  _local_client_lock.release()
 
+# 关闭所有
+def close_all():
+  global _local_client
+  global _local_client_lock
+  _local_client_lock.acquire()
+  for item in _local_client:
+    item[0].close()
+  _local_client=[]
+  _local_client_lock.release()
+  myprint('连接列表已清空')
 
 
 
 
 # 远端数据处理，解包，把负载数据发送到本地服务器
+# 这个线程只会创建一个，用于连接代理服务器
 def remote_client_handler(tcp_remote:socket):
   global _remote_client
   global _local_client
-  print("已连接代理服务器")
+  global _local_server
+  myprint("已连接代理服务器")
   recv_data=bytearray()
   cmd={'device':'client','option':'login'}
   data=pc.encode(json.dumps(cmd).encode('utf-8'),b'default')
@@ -60,7 +86,7 @@ def remote_client_handler(tcp_remote:socket):
     try:
       recv = tcp_remote.recv(4096)
     except Exception as e:
-      print(str(e))
+      myprint("代理服务器连接异常",str(e))
       break
     if recv:
       recv_data+=recv
@@ -70,27 +96,38 @@ def remote_client_handler(tcp_remote:socket):
         if(start==-1 or end==-1):
           break
         cmd,data=pc.decode(recv_data[start:end+1])
-        print(cmd.decode('utf-8'))
+        myprint(cmd.decode('utf-8'))
         try:
           j=json.loads(cmd)
           if(j['device']=='server'):
             if(j['option']=='data'):
+              myprint(f"收到数据 {j['ip']}:{j['port']}")
               send_to(j['ip'],j['port'],data)
             elif(j['option']=='disconnect'):
+              myprint(f"收到服务器的断开通知 {j['ip']}:{j['port']}")
               close(j['ip'],j['port'])
+          elif(j['device']=='proxy'):
+            if(j['option']=='close'):
+              myprint("收到代理服务器的断开通知")
+              close_all()
         except Exception as e:
-          print(str(e))
+          myprint(str(e))
         recv_data=recv_data[end+1:]
     else:
       break
   tcp_remote.close()
+  myprint("代理服务器已断开")
+  close_all()
+  if _local_server is not None:
+    _local_server.close()
 
 
 # 本地数据处理，解包，把负载数据发送到本地服务器
+# 每个本地连接都会创建一个线程
 def local_client_handler(tcp_server:socket,addr):
   global _remote_client
   global _local_client
-  print("addr:",addr)
+  myprint("addr:",addr)
   addr_info=(tcp_server,addr[0],addr[1])
   _local_client.append(addr_info)
   if(_remote_client is not None):
@@ -101,44 +138,53 @@ def local_client_handler(tcp_server:socket,addr):
     try:
       recv = tcp_server.recv(4096)
     except Exception as e:
-      print(str(e))
+      myprint("local:",str(e))
       break
     if recv:
       cmd={'device':'client','option':'data','ip':addr[0],'port':addr[1]}
       data=pc.encode(json.dumps(cmd).encode('utf-8'),recv)
       if _remote_client is not None:
         _remote_client.send(data)
-        print(f"send to remote {addr[0]},{addr[1]}")
+        myprint(f"发送数据 {addr[0]}:{addr[1]}")
     else:
       break
-  tcp_server.close()
-  if(_remote_client is not None):
+  close(addr[0],addr[1])
   cmd={'device':'client','option':'disconnect','ip':addr[0],'port':addr[1]}
   data=pc.encode(json.dumps(cmd).encode('utf-8'),b'default')
+  if(_remote_client is not None):
+    try:
+      myprint(f"发送断开通知到服务器端 {addr[0]}:{addr[1]}")
       _remote_client.send(data)
-  _local_client.remove(addr_info)
+    except Exception as e:
+      myprint("远端连接异常",str(e))
 
 
 
 def main():
   global _remote_client
   global _local_client
+  global _local_server
 
   _remote_client = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
   _remote_client.connect(("1.92.113.30",5345))
   thd = threading.Thread(target = remote_client_handler, args = (_remote_client,))
   thd.start()
-  tcp_server = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
-  tcp_server.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,True)
-  tcp_server.bind(("",_LOCAL_PORT))
-  print(f"开始监听({_LOCAL_PORT})")
-  tcp_server.listen(128)
+  _local_server = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
+  _local_server.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,True)
+  _local_server.bind(("",_LOCAL_PORT))
+  myprint(f"开始监听({_LOCAL_PORT})")
+  _local_server.listen(128)
   while True:
-      temp , temp_address = tcp_server.accept()
+      try:
+        temp , temp_address = _local_server.accept()
+      except Exception as e:
+        myprint("local server:",str(e))
+        break
       thd = threading.Thread(target = local_client_handler, args = (temp,temp_address))
       thd.start()
 
 
 if __name__ == "__main__":
+  log_init("target_client.log")
   main()
 

target_server.py

@@ -5,9 +5,12 @@ import sys
 import json
 import socket
 import threading
+import time
 
 
 import prot_codec as pc
+from log import myprint
+from log import log_init
 
 
 
@@ -20,35 +23,87 @@ import prot_codec as pc
 # 另一个连接本地tcp服务器
 
 
+# 保存连接代理服务器的端口
 _remote_client=None
+# 保存本地连接的tcp客户端端口列表
+# 这个列表的ip和port地址和target_client.py中的同名变量一一对应
 _local_client=[]
 
 
-LOCAL_SERVER_IP = ("192.168.3.166",80)
-# LOCAL_SERVER_IP = ("192.168.3.167",22)
+# LOCAL_SERVER_IP = ("192.168.3.166",80)
+LOCAL_SERVER_IP = ("192.168.1.40",22)
+# LOCAL_SERVER_IP = ("192.168.3.174",22)
 
 
 
+
+
+_local_client_lock = threading.Lock()
+
+
 # 发送数据到指定ip地址和端口
 def send_to(ip,port,data:bytearray):
   for item in _local_client:
     if(item[1]==ip and item[2]==port):
-      print(f"recv from remote {ip},{port}")
       item[0].send(data)
       break
 
 # 关闭指定地址的端口
 def close(ip,port):
+  global _local_client
+  global _local_client_lock
+  _local_client_lock.acquire()
   for item in _local_client:
     if(item[1]==ip and item[2]==port):
+      myprint(f"断开连接 {ip}:{port}")
       item[0].close()
+      # 删除已被关闭的条目
+      _local_client.remove(item)
       break
+  _local_client_lock.release()
+
+# 关闭所有
+def close_all():
+  global _local_client
+  global _local_client_lock
+  _local_client_lock.acquire()
+  for item in _local_client:
+    item[0].close()
+  # 关闭端口之后把列表置空
+  _local_client=[]
+  _local_client_lock.release()
+  myprint('连接列表已清空')
+
+
+_tick_start:float=0
+_tick_end:float=0
+
+# 定时任务
+def remote_keeplive():
+  global _tick_start
+  global _tick_end
+  cmd={'device':'server','option':'keeplive'}
+  data=pc.encode(json.dumps(cmd).encode('utf-8'),b'default')
+  try:
+    _remote_client.send(data)
+  except Exception as e:
+    myprint(str(e))
+  _tick_end=time.perf_counter()
+  # 超过15秒没收到数据则自动断开
+  if(_tick_end-_tick_start>15):
+    myprint("长时间没收到代理服务器的数据回复，主动断开连接")
+    try:
+      _remote_client.close()
+    except Exception as e:
+      myprint(str(e))
+
 
 
 
 
 # 本地数据处理，解包，把负载数据发送到本地服务器
-def local_client_handler(tcp_server:socket,ip,port):
+# 每一个connect都会创建一个线程
+def local_client_handler(tcp_server:socket.socket,ip,port):
   global _remote_client
   global _local_client
   self_info=(tcp_server,ip,port)
@@ -57,43 +112,56 @@ def local_client_handler(tcp_server:socket,ip,port):
     try:
       recv = tcp_server.recv(4096)
     except Exception as e:
-      print(str(e))
+      myprint("本地连接异常",str(e))
       break
     if recv:
       cmd={'device':'server','option':'data','ip':ip,'port':port}
       data=pc.encode(json.dumps(cmd).encode('utf-8'),recv)
       if _remote_client is not None:
         _remote_client.send(data)
-        print(f"send to remote {ip},{port}")
+        myprint(f"发送数据到客户端 {ip}:{port}")
 
     else:
       break
   tcp_server.close()
+  close(ip,port)
+  # 发送连接断开的消息
   cmd={'device':'server','option':'disconnect','ip':ip,'port':port}
   data=pc.encode(json.dumps(cmd).encode('utf-8'),b'default')
   if _remote_client is not None:
+    try:
+      myprint(f"发送断开通知到客户端 {ip}:{port}")
       _remote_client.send(data)
-  _local_client.remove(self_info)
+    except Exception as e:
+      myprint("远端连接异常",str(e))
 
 
 
 
 # 远端数据处理，解包，把负载数据发送到本地服务器
-def remote_client_handler(tcp_client_1:socket):
+def remote_client_handler(tcp_client:socket.socket):
   global _remote_client
   global _local_client
-  print("已连接代理服务器")
+  global _tick_start
+  myprint("已连接代理服务器")
+  timer=threading.Timer(5,remote_keeplive,())
   cmd={'device':'server','option':'login'}
   data=pc.encode(json.dumps(cmd).encode('utf-8'),b'default')
   _remote_client.send(data)
+  timer.start()
   recv_data=bytearray()
+  _tick_start=time.perf_counter()
   while True:
     try:
-      recv = tcp_client_1.recv(4096)
+      recv = tcp_client.recv(4096)
     except Exception as e:
-      print(str(e))
+      myprint("remote:",str(e))
       break
     if recv:
+      timer.cancel()
+      timer=threading.Timer(5,remote_keeplive,())
+      timer.start()
+      _tick_start=time.perf_counter()
       recv_data+=recv
       while True:
         start=recv_data.find(b'\xff')
@@ -101,33 +169,49 @@ def remote_client_handler(tcp_client_1:socket):
         if(start == -1 or end == -1):
           break
         cmd,data=pc.decode(recv_data[start:end+1])
-        print(cmd.decode('utf-8'))
+        myprint(cmd.decode('utf-8'))
         try:
           j=json.loads(cmd)
           if(j['device']=='client'):
             if(j['option']=='connect'):
+              myprint("收到客户端的连接通知")
               temp = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
               temp.connect(LOCAL_SERVER_IP)
               thd = threading.Thread(target = local_client_handler, args = (temp,j['ip'],j['port']))
               thd.start()
             elif(j['option']=='disconnect'):
+              myprint(f"收到客户端的断开通知 {j['ip']}:{j['port']}")
               close(j['ip'],j['port'])
             elif(j['option']=='data'):
+              myprint(f"收到数据 {j['ip']}:{j['port']}")
               send_to(j['ip'],j['port'],data)
+          elif(j['device']=='proxy'):
+            if(j['option']=='close'):
+              myprint("收到代理服务器的断开通知")
+              close_all()
+            elif(j["option"]=='keeplive'):
+              myprint("收到代理服务器的心跳数据")
         except Exception as e:
-          print(str(e))
+          myprint(str(e))
         recv_data=recv_data[end+1:]
     else:
       break
-  tcp_client_1.close()
-  print("socket close.")
-
+  timer.cancel()
+  timer.join()
+  try:
+    tcp_client.close()
+  except Exception as e:
+    myprint(str(e))
+  myprint("与代理服务器的连接已断开")
+  close_all()
 
 
 def main():
   global _remote_client
   global _local_client
 
+  while True:
+    myprint("尝试连接代理服务器")
     _remote_client = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
     _remote_client.connect(("1.92.113.30",5345))
     # thd = threading.Thread(target = remote_client_handler, args = (_remote_client,))
@@ -141,5 +225,6 @@ def main():
 
 
 if __name__ == "__main__":
+  log_init("target_server.log")
   main()