Python--网络编程
3. 网络编程与Socket
3.1 Socket基础
创建Socket
import socket
# TCP Socket
tcp_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# UDP Socket
udp_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
服务器端函数
函数 | 描述 |
---|---|
bind((host, port)) | 绑定地址和端口 |
listen(backlog) | 开始监听传入连接 |
accept() | 接受连接,返回新Socket和客户端地址 |
客户端函数
函数 | 描述 |
---|---|
connect((host, port)) | 连接到服务器 |
send(data) | 发送数据(TCP) |
sendto(data, address) | 发送数据(UDP) |
3.2 高级Socket功能
超时与异常处理
import socket
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.settimeout(5) # 设置超时为5秒
try:
s.connect(("example.com", 80))
except socket.timeout:
print("连接超时!")
except socket.error as e:
print(f"连接错误: {e}")
finally:
s.close()
多线程Socket服务器
import socket
import threading
def handle_client(conn, addr):
print(f"客户端 {addr} 已连接")
data = conn.recv(1024)
conn.send(b"Server: Received")
conn.close()
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.bind(("0.0.0.0", 8080))
server.listen(5)
while True:
conn, addr = server.accept()
thread = threading.Thread(target=handle_client, args=(conn, addr))
thread.start()
实现简单HTTP请求
# 发送HTTP GET请求
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.connect(("example.com", 80))
s.send(b"GET / HTTP/1.1\r\nHost: example.com\r\n\r\n")
response = s.recv(4096)
print(response.decode())
s.close()
3.3 TCP通信示例
服务器端:
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.bind(("0.0.0.0", 8080))
server_socket.listen(5)
conn, addr = server_socket.accept()
data = conn.recv(1024)
conn.send(b"Received")
conn.close()
客户端:
client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client_socket.connect(("127.0.0.1", 8080))
client_socket.send(b"Hello Server")
response = client_socket.recv(1024)
client_socket.close()
3.4 UDP通信示例
服务器端:
udp_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
udp_socket.bind(("0.0.0.0", 9090))
data, addr = udp_socket.recvfrom(1024)
udp_socket.sendto(b"ACK", addr)
客户端:
udp_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
udp_socket.sendto(b"Hello", ("127.0.0.1", 9090))
response, addr = udp_socket.recvfrom(1024)
3.5 端口扫描与主机探测
使用 python-nmap 库进行端口扫描:
import nmap
scanner = nmap.PortScanner()
scanner.scan("192.168.1.1", "1-1024", "-sS") # SYN扫描
print(scanner.all_hosts())
print(scanner["192.168.1.1"].all_tcp())
4. 工具与资源
-
正则表达式在线测试工具:https://regex101.com
-
网络调试工具:Wireshark、Postman、tcpdump
-
Python库:
-
python-nmap:pip install python-nmap
-
requests:HTTP请求库
-
scapy:构造和解析网络数据包。
pip install scapy
-
asyncio:实现异步网络编程。
import asyncio async def tcp_echo_client(message): reader, writer = await asyncio.open_connection("127.0.0.1", 8888) writer.write(message.encode()) data = await reader.read(100) print(f"Received: {data.decode()}") writer.close() asyncio.run(tcp_echo_client("Hello Server"))
-