tcp_port_tran/target_server.py

import os
import sys
import json
import socket
import threading
import time


import prot_codec as pc
from log import myprint
from log import log_init







# tcp服务端脚本
# 一个连接远端转发服务器
# 另一个连接本地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.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):
      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))





# 本地数据处理，解包，把负载数据发送到本地服务器
# 每一个connect都会创建一个线程
def local_client_handler(tcp_server:socket.socket,ip,port):
  global _remote_client
  global _local_client
  self_info=(tcp_server,ip,port)
  _local_client.append(self_info)
  while True:
    try:
      recv = tcp_server.recv(4096)
    except Exception as 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)
        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)
    except Exception as e:
      myprint("远端连接异常",str(e))




# 远端数据处理，解包，把负载数据发送到本地服务器
def remote_client_handler(tcp_client:socket.socket):
  global _remote_client
  global _local_client
  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.recv(4096)
    except Exception as 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')
        end=recv_data.find(b'\xfe')
        if(start == -1 or end == -1):
          break
        cmd,data=pc.decode(recv_data[start:end+1])
        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:
          myprint(str(e))
        recv_data=recv_data[end+1:]
    else:
      break
  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,))
    # thd.start()
    # thd.join()

    remote_client_handler(_remote_client)





if __name__ == "__main__":
  log_init("target_server.log")
  main()