今日目标

• 理解网络编程的基本概念和协议

• 掌握Socket编程的基础知识

• 学会创建HTTP客户端和服务器

• 了解网络编程的安全思考

网络编程概述

网络编程是计算机之间进行通信的基础，Python提供了丰富的网络编程库：

• socket：底层网络通信

• http.client：HTTP客户端

• urllib：URL处理

• requests：高级HTTP库

• asyncio：异步网络编程

网络协议层次

应用层    HTTP, FTP, SMTP
传输层    TCP, UDP
网络层    IP
链路层    以太网, WiFi
物理层    电缆, 光纤

Socket编程基础

1. 创建Socket

import socket# 创建TCP Sockettcp_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)# 创建UDP Socketudp_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)# Socket类型说明# AF_INET: IPv4地址族# AF_INET6: IPv6地址族# SOCK_STREAM: TCP协议# SOCK_DGRAM: UDP协议

2. TCP客户端示例

import socketimport timedef tcp_client():    """TCP客户端示例"""    # 创建Socket
    client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)    
    try:        # 连接服务器
        server_address = ('localhost', 8888)
        print(f"连接到服务器: {server_address}")
        client.connect(server_address)        
        # 发送数据
        message = "Hello, Server!"
        client.send(message.encode('utf-8'))
        print(f"发送: {message}")        
        # 接收响应
        data = client.recv(1024)
        response = data.decode('utf-8')
        print(f"收到: {response}")        
    except ConnectionRefusedError:
        print("连接被拒绝，请确保服务器正在运行")    except Exception as e:
        print(f"连接错误: {e}")    finally:
        client.close()
        print("连接已关闭")# 运行客户端if __name__ == "__main__":
    tcp_client()

3. TCP服务器示例

import socketimport threadingimport timeclass TCPServer:    def __init__(self, host='localhost', port=8888):
        self.host = host
        self.port = port
        self.server_socket = None
        self.clients = []    
    def start(self):        """启动服务器"""        try:            # 创建服务器Socket
            self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)            
            # 设置端口重用
            self.server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)            
            # 绑定地址
            self.server_socket.bind((self.host, self.port))            
            # 开始监听
            self.server_socket.listen(5)
            print(f"服务器启动成功，监听地址: {self.host}:{self.port}")            
            # 接受客户端连接            while True:
                client_socket, client_address = self.server_socket.accept()
                print(f"新客户端连接: {client_address}")                
                # 为每个客户端创建新线程
                client_thread = threading.Thread(
                    target=self.handle_client,
                    args=(client_socket, client_address)
                )
                client_thread.start()
                self.clients.append(client_socket)                
        except Exception as e:
            print(f"服务器错误: {e}")        finally:
            self.stop()    
    def handle_client(self, client_socket, client_address):        """处理客户端连接"""        try:            while True:                # 接收数据
                data = client_socket.recv(1024)                if not data:                    break                
                # 解码数据
                message = data.decode('utf-8')
                print(f"来自 {client_address}: {message}")                
                # 发送响应
                response = f"服务器收到: {message}"
                client_socket.send(response.encode('utf-8'))                
        except Exception as e:
            print(f"处理客户端 {client_address} 时出错: {e}")        finally:
            client_socket.close()            if client_socket in self.clients:
                self.clients.remove(client_socket)
            print(f"客户端 {client_address} 断开连接")    
    def stop(self):        """停止服务器"""        if self.server_socket:
            self.server_socket.close()        for client in self.clients:
            client.close()
        print("服务器已停止")# 启动服务器if __name__ == "__main__":
    server = TCPServer()    try:
        server.start()    except KeyboardInterrupt:
        print("
正在关闭服务器...")
        server.stop()

HTTP编程

1. 使用urllib

import urllib.requestimport urllib.parseimport urllib.errorimport jsondef http_get_example():    """HTTP GET请求示例"""    try:        # 发送GET请求
        url = "https://jsonplaceholder.typicode.com/posts/1"        with urllib.request.urlopen(url) as response:
            data = response.read()
            result = json.loads(data.decode('utf-8'))
            print("GET请求结果:")
            print(f"标题: {result['title']}")
            print(f"内容: {result['body']}")            
    except urllib.error.URLError as e:
        print(f"URL错误: {e}")    except Exception as e:
        print(f"请求错误: {e}")def http_post_example():    """HTTP POST请求示例"""    try:        # 准备数据
        post_data = {            "title": "Python网络编程",            "body": "学习Python网络编程很有趣",            "userId": 1
        }        
        # 编码数据
        data = json.dumps(post_data).encode('utf-8')        
        # 创建请求
        url = "https://jsonplaceholder.typicode.com/posts"
        req = urllib.request.Request(
            url,
            data=data,
            headers={'Content-Type': 'application/json'}
        )        
        # 发送请求        with urllib.request.urlopen(req) as response:
            result = json.loads(response.read().decode('utf-8'))
            print("POST请求结果:")
            print(f"新帖子ID: {result['id']}")
            print(f"标题: {result['title']}")            
    except urllib.error.URLError as e:
        print(f"URL错误: {e}")    except Exception as e:
        print(f"请求错误: {e}")# 运行示例http_get_example()
print("-" * 50)
http_post_example()

2. 使用requests库

import requestsimport jsonclass HTTPClient:    def __init__(self, base_url="https://jsonplaceholder.typicode.com"):
        self.base_url = base_url
        self.session = requests.Session()
        self.session.headers.update({            'User-Agent': 'Python-HTTPClient/1.0',            'Content-Type': 'application/json'
        })    
    def get(self, endpoint, params=None):        """发送GET请求"""        try:
            url = f"{self.base_url}{endpoint}"
            response = self.session.get(url, params=params, timeout=10)
            response.raise_for_status()            return response.json()        except requests.RequestException as e:
            print(f"GET请求错误: {e}")            return None    
    def post(self, endpoint, data=None):        """发送POST请求"""        try:
            url = f"{self.base_url}{endpoint}"
            response = self.session.post(url, json=data, timeout=10)
            response.raise_for_status()            return response.json()        except requests.RequestException as e:
            print(f"POST请求错误: {e}")            return None    
    def put(self, endpoint, data=None):        """发送PUT请求"""        try:
            url = f"{self.base_url}{endpoint}"
            response = self.session.put(url, json=data, timeout=10)
            response.raise_for_status()            return response.json()        except requests.RequestException as e:
            print(f"PUT请求错误: {e}")            return None    
    def delete(self, endpoint):        """发送DELETE请求"""        try:
            url = f"{self.base_url}{endpoint}"
            response = self.session.delete(url, timeout=10)
            response.raise_for_status()            return response.status_code == 204        except requests.RequestException as e:
            print(f"DELETE请求错误: {e}")            return False# 使用示例client = HTTPClient()# 获取所有帖子posts = client.get("/posts")if posts:
    print(f"获取到 {len(posts)} 个帖子")    for post in posts[:3]:
        print(f"- {post['title']}")# 创建新帖子new_post = {    "title": "Python网络编程教程",    "body": "这是一个关于Python网络编程的详细教程",    "userId": 1}

created_post = client.post("/posts", new_post)if created_post:
    print(f"新帖子创建成功，ID: {created_post['id']}")# 更新帖子update_data = {"title": "更新的标题"}
updated_post = client.put(f"/posts/{created_post['id']}", update_data)if updated_post:
    print(f"帖子更新成功: {updated_post['title']}")# 删除帖子if client.delete(f"/posts/{created_post['id']}"):
    print("帖子删除成功")

网络编程安全

1. 输入验证

import reimport socketdef validate_ip_address(ip):    """验证IP地址格式"""
    pattern = r'^(d{1,3}.){3}d{1,3}$'    if not re.match(pattern, ip):        return False
    
    parts = ip.split('.')    for part in parts:        if not 0 <= int(part) <= 255:            return False    return Truedef validate_port(port):    """验证端口号"""    try:
        port_num = int(port)        return 1 <= port_num <= 65535    except ValueError:        return Falsedef safe_socket_connect(host, port):    """安全的Socket连接"""    if not validate_ip_address(host):        raise ValueError(f"无效的IP地址: {host}")    
    if not validate_port(port):        raise ValueError(f"无效的端口号: {port}")    
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(10)  # 设置超时
        sock.connect((host, port))        return sock    except socket.timeout:        raise ConnectionError("连接超时")    except socket.error as e:        raise ConnectionError(f"连接失败: {e}")

2. 数据加密

import sslimport socketdef create_ssl_client(host, port):    """创建SSL客户端"""    try:        # 创建SSL上下文
        context = ssl.create_default_context()        
        # 创建Socket
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)        
        # 包装为SSL Socket
        ssl_sock = context.wrap_socket(sock, server_hostname=host)        
        # 连接
        ssl_sock.connect((host, port))        return ssl_sock        
    except ssl.SSLError as e:
        print(f"SSL错误: {e}")        return None    except Exception as e:
        print(f"连接错误: {e}")        return Nonedef create_ssl_server(host, port, cert_file, key_file):    """创建SSL服务器"""    try:        # 创建SSL上下文
        context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
        context.load_cert_chain(cert_file, key_file)        
        # 创建Socket
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.bind((host, port))
        sock.listen(5)        
        # 包装为SSL Socket
        ssl_sock = context.wrap_socket(sock, server_side=True)        return ssl_sock        
    except Exception as e:
        print(f"SSL服务器创建失败: {e}")        return None

今日总结

今天我们学习了网络编程的基础知识：

1. Socket编程：TCP和UDP通信的基础

2. HTTP编程：使用urllib和requests进行HTTP请求

3. 网络安全：输入验证、SSL加密等安全思考

4. 实用示例：客户端和服务器开发

网络编程是Python应用开发的重大组成部分，掌握这些知识对于开发Web应用、网络工具、分布式系统等都至关重大。