Python 网络编程指南

import socket

# 创建 TCP Socket
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

# 创建 UDP Socket
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

import socket

def tcp_client(host: str, port: int, message: str) -> str:
    """TCP 客户端：连接服务器，发送消息，接收响应"""
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.settimeout(10)  # 设置超时，防止永久阻塞
        s.connect((host, port))
        s.sendall(message.encode('utf-8'))  # 使用 sendall 确保完整发送

        # 接收响应
        chunks = []
        while True:
            chunk = s.recv(4096)
            if not chunk:
                break
            chunks.append(chunk)

    return b''.join(chunks).decode('utf-8')


# 使用示例
response = tcp_client('127.0.0.1', 9999, 'Hello, Server!')
print(response)

import socket
import threading


def handle_client(conn: socket.socket, addr: tuple[str, int]) -> None:
    """处理单个客户端连接"""
    print(f'新连接: {addr}')
    try:
        conn.settimeout(30)  # 客户端超时
        while True:
            data = conn.recv(1024)
            if not data:
                break
            print(f'收到 [{addr}]: {data.decode("utf-8").strip()}')
            conn.sendall(f'Echo: {data.decode("utf-8")}'.encode('utf-8'))
    except socket.timeout:
        print(f'客户端超时: {addr}')
    except ConnectionResetError:
        print(f'客户端断开: {addr}')
    finally:
        conn.close()


def tcp_server(host: str = '127.0.0.1', port: int = 9999) -> None:
    """多线程 TCP Echo 服务器"""
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)  # 端口复用
    server.bind((host, port))
    server.listen(5)  # 最大等待连接数
    print(f'服务端启动: {host}:{port}')

    try:
        while True:
            conn, addr = server.accept()
            t = threading.Thread(target=handle_client, args=(conn, addr))
            t.daemon = True
            t.start()
    except KeyboardInterrupt:
        print('\n服务端关闭')
    finally:
        server.close()


if __name__ == '__main__':
    tcp_server()

import socket


def udp_server(host: str = '127.0.0.1', port: int = 9999) -> None:
    """UDP Echo 服务器"""
    server = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    server.bind((host, port))
    print(f'UDP 服务端启动: {host}:{port}')

    while True:
        data, addr = server.recvfrom(1024)
        print(f'收到 [{addr}]: {data.decode("utf-8")}')
        server.sendto(f'Echo: {data.decode("utf-8")}'.encode('utf-8'), addr)


def udp_client(host: str = '127.0.0.1', port: int = 9999) -> None:
    """UDP 客户端"""
    client = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    client.settimeout(5)

    for msg in ['Hello', 'UDP', 'World']:
        client.sendto(msg.encode('utf-8'), (host, port))
        try:
            data, addr = client.recvfrom(1024)
            print(f'服务端响应: {data.decode("utf-8")}')
        except socket.timeout:
            print('请求超时')

    client.close()

import socketserver


class EchoHandler(socketserver.BaseRequestHandler):
    """TCP Echo 处理器"""
    def handle(self) -> None:
        print(f'新连接: {self.client_address}')
        while True:
            data = self.request.recv(1024)
            if not data:
                break
            self.request.sendall(f'Echo: {data.decode()}'.encode())


class UDPHandler(socketserver.BaseRequestHandler):
    """UDP Echo 处理器"""
    def handle(self) -> None:
        data = self.request[0]  # UDP 数据在 request[0]
        socket = self.request[1]  # UDP socket 在 request[1]
        print(f'收到 [{self.client_address}]: {data.decode()}')
        socket.sendto(f'Echo: {data.decode()}'.encode(), self.client_address)


# 使用
if __name__ == '__main__':
    # 多线程 TCP 服务器
    with socketserver.ThreadingTCPServer(('127.0.0.1', 9999), EchoHandler) as server:
        print('TCP 服务器启动')
        server.serve_forever()

import selectors
import socket
import types


def accept_connection(sel: selectors.DefaultSelector, sock: socket.socket) -> None:
    """接受新连接"""
    conn, addr = sock.accept()
    print(f'新连接: {addr}')
    conn.setblocking(False)

    data = types.SimpleNamespace(addr=addr, inb=b'', outb=b'')
    events = selectors.EVENT_READ | selectors.EVENT_WRITE
    sel.register(conn, events, data=data)


def service_connection(sel: selectors.DefaultSelector, key: selectors.SelectorKey, mask: int) -> None:
    """处理已有连接的读写"""
    sock = key.fileobj
    data = key.data

    if mask & selectors.EVENT_READ:
        recv_data = sock.recv(1024)
        if recv_data:
            data.outb += recv_data  # 收到的数据加入发送缓冲
        else:
            print(f'关闭连接: {data.addr}')
            sel.unregister(sock)
            sock.close()

    if mask & selectors.EVENT_WRITE and data.outb:
        sent = sock.send(data.outb)  # Echo 回去
        data.outb = data.outb[sent:]


def run_selector_server(host: str = '127.0.0.1', port: int = 9999) -> None:
    """基于 selectors 的单线程高并发 Echo 服务器"""
    sel = selectors.DefaultSelector()

    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    sock.bind((host, port))
    sock.listen()
    sock.setblocking(False)

    sel.register(sock, selectors.EVENT_READ, data=None)
    print(f'Selector 服务器启动: {host}:{port}')

    try:
        while True:
            events = sel.select(timeout=None)
            for key, mask in events:
                if key.data is None:
                    accept_connection(sel, key.fileobj)
                else:
                    service_connection(sel, key, mask)
    except KeyboardInterrupt:
        print('\n服务器关闭')
    finally:
        sel.close()
        sock.close()

import socket
import ssl


def tls_server(host: str = '127.0.0.1', port: int = 8443) -> None:
    """TLS 加密 Echo 服务器"""
    context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
    context.load_cert_chain(certfile="server.crt", keyfile="server.key")

    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server.bind((host, port))
    server.listen(5)

    with context.wrap_socket(server, server_side=True) as tls_sock:
        print(f'TLS 服务器启动: {host}:{port}')
        while True:
            conn, addr = tls_sock.accept()
            try:
                data = conn.recv(1024)
                if data:
                    print(f'收到 [{addr}]: {data.decode()}')
                    conn.sendall(f'Secure Echo: {data.decode()}'.encode())
            finally:
                conn.close()

import socket
import ssl


def tls_client(host: str = '127.0.0.1', port: int = 8443) -> None:
    """TLS 加密客户端"""
    context = ssl.create_default_context(ssl.Purpose.SERVER_AUTH)
    # 自签名证书需要加载 CA 或禁用验证（仅测试用）
    context.check_hostname = False
    context.verify_mode = ssl.CERT_NONE

    with socket.create_connection((host, port)) as sock:
        with context.wrap_socket(sock, server_hostname=host) as tls_sock:
            tls_sock.sendall(b'Hello, TLS Server!')
            data = tls_sock.recv(1024)
            print(f'响应: {data.decode()}')

import socket
import struct
import json


def send_file(sock: socket.socket, file_path: str) -> None:
    """发送文件：先发元数据，再发内容"""
    import os
    file_size = os.path.getsize(file_path)
    file_name = os.path.basename(file_path)

    # 发送元数据（JSON 头 + 4字节长度前缀）
    metadata = json.dumps({"name": file_name, "size": file_size}).encode()
    sock.sendall(struct.pack('!I', len(metadata)))  # 元数据长度
    sock.sendall(metadata)  # 元数据内容

    # 发送文件内容
    with open(file_path, 'rb') as f:
        while True:
            chunk = f.read(65536)  # 64KB 块
            if not chunk:
                break
            sock.sendall(chunk)

    print(f'已发送: {file_name} ({file_size} bytes)')


def receive_file(sock: socket.socket, save_dir: str = '.') -> str:
    """接收文件：先收元数据，再收内容"""
    # 接收元数据长度
    raw_len = _recv_exactly(sock, 4)
    meta_len = struct.unpack('!I', raw_len)[0]

    # 接收元数据
    metadata = json.loads(_recv_exactly(sock, meta_len))
    file_name = metadata['name']
    file_size = metadata['size']

    # 接收文件内容
    save_path = os.path.join(save_dir, file_name)
    received = 0
    with open(save_path, 'wb') as f:
        while received < file_size:
            chunk = sock.recv(min(65536, file_size - received))
            if not chunk:
                break
            f.write(chunk)
            received += len(chunk)

    print(f'已接收: {file_name} ({received} bytes)')
    return save_path


def _recv_exactly(sock: socket.socket, n: int) -> bytes:
    """精确接收 n 字节"""
    data = b''
    while len(data) < n:
        chunk = sock.recv(n - len(data))
        if not chunk:
            raise ConnectionError('连接断开')
        data += chunk
    return data

import socket
import threading
from collections.abc import Callable


class ChatServer:
    """简单的多人聊天服务器"""

    def __init__(self, host: str = '127.0.0.1', port: int = 8888) -> None:
        self.host = host
        self.port = port
        self.clients: dict[socket.socket, str] = {}
        self.lock = threading.Lock()

    def broadcast(self, message: str, exclude: socket.socket | None = None) -> None:
        """向所有客户端广播消息"""
        with self.lock:
            for client in list(self.clients.keys()):
                if client != exclude:
                    try:
                        client.sendall(message.encode('utf-8'))
                    except ConnectionError:
                        self.remove_client(client)

    def remove_client(self, client: socket.socket) -> None:
        """移除客户端"""
        with self.lock:
            name = self.clients.pop(client, 'Unknown')
        try:
            client.close()
        except OSError:
            pass
        self.broadcast(f'[系统] {name} 离开了聊天室\n')

    def handle_client(self, conn: socket.socket, addr: tuple[str, int]) -> None:
        """处理客户端消息"""
        try:
            # 第一个消息是用户名
            name = conn.recv(1024).decode('utf-8').strip()
            with self.lock:
                self.clients[conn] = name
            self.broadcast(f'[系统] {name} 加入了聊天室\n', exclude=conn)

            while True:
                data = conn.recv(1024)
                if not data:
                    break
                msg = data.decode('utf-8').strip()
                if msg:
                    self.broadcast(f'[{name}] {msg}\n', exclude=conn)
        except ConnectionResetError:
            pass
        finally:
            self.remove_client(conn)

    def start(self) -> None:
        """启动聊天服务器"""
        server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        server.bind((self.host, self.port))
        server.listen(10)
        print(f'聊天服务器启动: {self.host}:{self.port}')

        try:
            while True:
                conn, addr = server.accept()
                t = threading.Thread(target=self.handle_client, args=(conn, addr))
                t.daemon = True
                t.start()
        except KeyboardInterrupt:
            print('\n服务器关闭')
        finally:
            server.close()


if __name__ == '__main__':
    ChatServer().start()

import socket
import concurrent.futures
from dataclasses import dataclass


@dataclass
class ScanResult:
    port: int
    is_open: bool
    service: str


# 常见端口与服务映射
COMMON_PORTS = {
    21: "FTP", 22: "SSH", 23: "Telnet", 25: "SMTP",
    53: "DNS", 80: "HTTP", 110: "POP3", 143: "IMAP",
    443: "HTTPS", 993: "IMAPS", 995: "POP3S",
    3306: "MySQL", 5432: "PostgreSQL", 6379: "Redis",
    8080: "HTTP-Alt", 8443: "HTTPS-Alt", 27017: "MongoDB",
}


def scan_port(host: str, port: int, timeout: float = 1.0) -> ScanResult:
    """扫描单个端口"""
    try:
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
            s.settimeout(timeout)
            result = s.connect_ex((host, port))
            is_open = result == 0
            service = COMMON_PORTS.get(port, "Unknown")
            return ScanResult(port=port, is_open=is_open, service=service)
    except socket.error:
        return ScanResult(port=port, is_open=False, service="")


def port_scan(host: str, ports: range | list[int], timeout: float = 1.0, workers: int = 100) -> list[ScanResult]:
    """并发端口扫描"""
    open_ports = []
    with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as executor:
        futures = {executor.submit(scan_port, host, p, timeout): p for p in ports}
        for future in concurrent.futures.as_completed(futures):
            result = future.result()
            if result.is_open:
                open_ports.append(result)

    return sorted(open_ports, key=lambda r: r.port)


# 使用示例
if __name__ == '__main__':
    target = '127.0.0.1'
    print(f'扫描 {target} ...')
    results = port_scan(target, range(1, 1024), timeout=0.5)
    for r in results:
        print(f'  端口 {r.port:>5} 开放  [{r.service}]')
    print(f'共发现 {len(results)} 个开放端口')

# ❌ 问题：连续发送两条消息，接收方可能一次收到 "HelloWorld"
conn.sendall(b'Hello')
conn.sendall(b'World')
# 接收方: data = conn.recv(1024) → b'HelloWorld'  # 粘在一起了！

# ✅ 解决方案1：长度前缀
import struct

def send_message(sock, data: bytes) -> None:
    """先发4字节长度，再发数据"""
    sock.sendall(struct.pack('!I', len(data)))
    sock.sendall(data)

def recv_message(sock) -> bytes:
    """先收4字节长度，再收完整数据"""
    raw_len = _recv_exactly(sock, 4)
    msg_len = struct.unpack('!I', raw_len)[0]
    return _recv_exactly(sock, msg_len)

# ✅ 解决方案2：换行分隔符（适合文本协议）
def send_line(sock, text: str) -> None:
    sock.sendall((text + '\n').encode('utf-8'))