AwinSimpleP2P/server1.py

import threading
import uuid

from utils import *

services = {
    'ssh': 22,
    'alist': 5244,
    'minecraft': 25565
}
server_name = 'awin_server'
co_server = 'www.awin-x.top'
co_port = 5000

server_token = f"server_{uuid.uuid4().hex[:8]}"
my_charactor = 'server'

co_server_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
co_server_sock.connect((co_server, co_port))

udp_sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
udp_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
udp_sock.bind(('0.0.0.0', 0))
listen_port = udp_sock.getsockname()[1]

tcp_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
tcp_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
tcp_sock.bind(('0.0.0.0', listen_port))
tcp_sock.listen(5)
print(f"监听端口: {listen_port}")

conns = {}

clients = {}


def heart_beat_thread():
    heart_beat_pack = pack_data('ping')
    while True:
        print('ping 到 co_server')
        send_pack_tcp(co_server_sock, heart_beat_pack)
        time.sleep(10)


def handle_punch_to(message):
    client_name = message['client_name']
    client_token = message['client_token']
    target_addr = tuple(message['target_addr'])
    clients[client_name] = {
        'client_token': client_token,
        'addr': target_addr,
    }
    print(f'打洞到{target_addr}')
    for i in range(100):
        send_action_udp(udp_sock, target_addr, 'punch')
        time.sleep(0.01)


def handle_punch_request(message):
    co_server_punch_port = message['co_server_punch_port']
    print(f'暴露端口：{listen_port}->{(co_server, co_server_punch_port)}')
    for i in range(5):
        send_action_udp(udp_sock, (co_server, co_server_punch_port), 'server_punch_port')
        time.sleep(0.1)


def handle_bye(sock):
    print(f"收到来自 {sock.getpeername()} 的断开连接请求")
    try:
        send_action_tcp(sock, 'bye', 'bye from server')
        sock.close()
    except Exception as e:
        print(e)


def handle_data(conn_id, data):
    conns[conn_id]['conn_sock'].sendall(data)


def forward_data_thread(conn_id):
    conn = conns[conn_id]
    while conn_id in conns:
        data = conn['conn_sock'].recv(4096)
        try:
            send_data_tcp(conn['client_sock'], data, {'conn_id': conn_id})
        except Exception as e:
            print(e[:10]+"->service connection breaks")
            break


def handle_connect_service(message):
    service = message['service']
    service_port = services[service]
    client_name = message['client_name']
    conn_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    conn_sock.connect(('127.0.0.1', service_port))
    conn_id = message['conn_id']
    conns[conn_id] = {
        'conn_sock': conn_sock,
        'client_name': client_name,
        'service': service
    }
    threading.Thread(target=forward_data_thread, args=conn_id,  daemon=True).start()


def handle_disconnect_service(message):
    conn_id = message['conn_id']
    if conn_id in conns:
        conn_sock = conns[conn_id]['conn_sock']
        conn_sock.close()
        del conns[conn_id]


def client_thread(client_sock, client_addr):
    while True:
        action, message, data = recv_tcp(client_sock)
        if action == 'data':
            handle_data(message['conn_id'], data)
        elif action == 'bye':
            handle_bye(client_sock)
            break
        elif action == 'connect_service':
            handle_connect_service(message)
        elif action == 'disconnect_service':
            handle_disconnect_service(message)
        else:
            send_action_tcp(client_sock, 'error', f"未知操作: {action}")


def tcp_accept_thread():
    while True:
        sock, addr = tcp_sock.accept()
        print(f"收到来自 {addr} 的连接")
        action, message, data = recv_tcp(sock)
        if action == 'client_hello':
            client_name = message['client_name']
            client_token = message['client_token']
            if client_name in clients:
                if clients[client_name]['client_token'] == client_token:
                    if 'sock' in clients[client_name]:
                        send_action_tcp(sock, 'error', f"{client_name} 已存在")
                        print(f"{client_name} 已存在")
                    else:
                        clients[client_name]['sock'] = sock
                        clients[client_name]['addr'] = addr
                        send_action_tcp(sock, 'success', f"{client_name} 已连接")
                        threading.Thread(target=client_thread, args=(sock, addr)).start()
                        print(f"{client_name} 已连接")
                else:
                    print(f'{addr},token 错误')
            else:
                print(f'{addr},未注册')
        else:
            print(f'{addr},未知操作')
            sock.close()


def main():
    send_action_tcp(co_server_sock, 'server_hello', {
        'name': server_name,
        'token': server_token,
        'charactor': my_charactor
    })
    action, message, data = recv_tcp(co_server_sock)
    if action == 'co_server_hello':
        co_server_token = message['token']
    else:
        print(f"连接到协服务器失败{message['message']}")
        return

    threading.Thread(target=heart_beat_thread, daemon=True).start()

    while True:
        action, message, data = recv_tcp(co_server_sock)
        if action == 'punch_to':
            handle_punch_to(message)
        elif action == 'punch_request':
            handle_punch_request(message)
        elif action == 'pong':
            print('pong 来自 co_server')
        # elif action == 'data':
        #     handle_data(co_server_sock, co_server_addr, message, data)
        elif action == 'bye':
            handle_bye(co_server_sock)
        else:
            print(f'收到来自co_server的未知消息{action}')


if __name__ == '__main__':
    main()