4. TLS记录层封装与解析

任务目标

实现一个最小可运行的TLS 1.2记录层封装与解析模块,支持将应用数据按最大分片长度(2^14字节)分段,对每个分片构造TLS记录头(5字节),并模拟加密与MAC计算。解析端能够从字节流中识别记录边界、校验类型与版本、解密并重组应用数据。

核心实现

import os
import hashlib
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.primitives import padding

# TLS 1.2 记录层常量
RECORD_HEADER_LEN = 5
MAX_PLAINTEXT_LEN = 2**14  # 16384 字节
CONTENT_TYPE_APPLICATION_DATA = 23
TLS_VERSION_1_2 = b'\x03\x03'

class TLSRecordLayer:
    def __init__(self, key: bytes, iv: bytes, mac_key: bytes):
        """key: AES-128-CBC 密钥(16B), iv: 初始IV(16B), mac_key: HMAC-SHA256密钥"""
        self.key = key
        self.iv = iv
        self.mac_key = mac_key
        self.sequence_number = 0  # 发送端序列号

    def _encrypt(self, plaintext: bytes) -> bytes:
        """AES-128-CBC加密 + PKCS7填充"""
        padder = padding.PKCS7(128).padder()
        padded_data = padder.update(plaintext) + padder.finalize()
        cipher = Cipher(algorithms.AES(self.key), modes.CBC(self.iv))
        encryptor = cipher.encryptor()
        return encryptor.update(padded_data) + encryptor.finalize()

    def _decrypt(self, ciphertext: bytes) -> bytes:
        """AES-128-CBC解密 + 去除PKCS7填充"""
        cipher = Cipher(algorithms.AES(self.key), modes.CBC(self.iv))
        decryptor = cipher.decryptor()
        padded_data = decryptor.update(ciphertext) + decryptor.finalize()
        unpadder = padding.PKCS7(128).unpadder()
        return unpadder.update(padded_data) + unpadder.finalize()

    def _compute_mac(self, content_type: int, version: bytes, plaintext: bytes) -> bytes:
        """HMAC-SHA256,覆盖序列号、类型、版本、长度和明文"""
        seq = self.sequence_number.to_bytes(8, 'big')
        length = len(plaintext).to_bytes(2, 'big')
        data = seq + bytes([content_type]) + version + length + plaintext
        return hashlib.sha256(data + self.mac_key).digest()[:16]  # 简化MAC

    def seal(self, app_data: bytes) -> list:
        """封装:分段 + 加密 + MAC,返回记录列表"""
        records = []
        for i in range(0, len(app_data), MAX_PLAINTEXT_LEN):
            chunk = app_data[i:i + MAX_PLAINTEXT_LEN]
            mac = self._compute_mac(CONTENT_TYPE_APPLICATION_DATA, TLS_VERSION_1_2, chunk)
            plaintext_with_mac = chunk + mac
            ciphertext = self._encrypt(plaintext_with_mac)
            header = bytes([CONTENT_TYPE_APPLICATION_DATA]) + TLS_VERSION_1_2 + len(ciphertext).to_bytes(2, 'big')
            records.append(header + ciphertext)
            self.sequence_number += 1
        return records

    def open(self, records: list) -> bytes:
        """解析:解密 + 验证MAC + 重组"""
        plaintext_parts = []
        for record in records:
            if len(record) < RECORD_HEADER_LEN:
                raise ValueError("记录长度不足")
            content_type = record[0]
            version = record[1:3]
            length = int.from_bytes(record[3:5], 'big')
            if content_type != CONTENT_TYPE_APPLICATION_DATA:
                raise ValueError(f"不支持的内容类型: {content_type}")
            if version != TLS_VERSION_1_2:
                raise ValueError("TLS版本不匹配")
            if len(record) != RECORD_HEADER_LEN + length:
                raise ValueError("记录长度字段与实际不符")
            ciphertext = record[RECORD_HEADER_LEN:]
            plaintext_with_mac = self._decrypt(ciphertext)
            # 分离明文和MAC(MAC固定16字节)
            plaintext = plaintext_with_mac[:-16]
            received_mac = plaintext_with_mac[-16:]
            # 验证MAC(使用接收端序列号)
            expected_mac = self._compute_mac(content_type, version, plaintext)
            if received_mac != expected_mac:
                raise ValueError("MAC验证失败")
            plaintext_parts.append(plaintext)
        return b''.join(plaintext_parts)

# 使用示例
if __name__ == '__main__':
    # 初始化参数(实际TLS握手后协商生成)
    key = os.urandom(16)
    iv = os.urandom(16)
    mac_key = os.urandom(32)

    sender = TLSRecordLayer(key, iv, mac_key)
    receiver = TLSRecordLayer(key, iv, mac_key)

    # 测试数据:超过一个分片长度,触发分段
    app_data = b'A' * 20000 + b'B' * 1000
    records = sender.seal(app_data)
    print(f"分段数量: {len(records)}")
    for i, rec in enumerate(records):
        print(f"记录{i+1}: 总长度={len(rec)}B, 密文长度={len(rec)-5}B")

    # 接收端解析
    recovered = receiver.open(records)
    print(f"重组数据长度: {len(recovered)}B")
    print(f"数据一致性: {recovered == app_data}")

运行与验证

依赖安装:

pip install cryptography

运行方式:

python tls_record_layer.py

通过条件:程序输出显示分段数量为2(20000字节数据被拆分为16384+3616),每条记录长度为密文长度+5字节头,重组数据与原始数据完全一致(True)。若修改MAC或密文任意字节,open()应抛出ValueError。

关键说明

  • 本实现聚焦记录层核心机制:分段、加密、MAC、重组。密钥交换、证书验证等握手流程不在范围内。
  • MAC计算采用简化方案(SHA256截断16字节),实际TLS使用HMAC且密钥扩展方式不同,但验证了记录层完整性保护逻辑。
  • 接收端序列号需与发送端同步递增,本示例通过相同初始化顺序保证一致性。
  • 加密使用AES-128-CBC,实际TLS 1.2还支持GCM等模式,但CBC足以演示记录层封装流程。