Python实战：手把手教你解密并下载AES-128加密的M3U8视频流（附完整代码）
Python实战：手把手教你解密并下载AES-128加密的M3U8视频流（附完整代码）
最近在帮朋友处理一个在线教育平台的视频下载需求时，遇到了AES-128加密的M3U8视频流。这种加密方式在各大视频平台都很常见，但完整实现解密下载的教程却不多。本文将带你从零开始，用Python实现整个解密下载流程，包括分析网页结构、获取密钥、解密TS片段和最终合并视频。
1. 理解M3U8视频流与AES加密原理
M3U8是一种基于HTTP Live Streaming（HLS）协议的视频播放列表格式，它将整个视频分割成多个小的TS文件片段。当视频内容被AES-128加密时，每个TS片段都需要使用密钥解密后才能正常播放。
AES-128加密需要三个关键要素：
- 密钥(Key)：16字节的加密密钥
- 初始化向量(IV)：16字节的随机数，用于增强加密强度
- 加密模式：通常使用CBC模式
在实际应用中，密钥可能通过以下几种方式提供：
- 直接内嵌在M3U8文件中
- 通过单独URL获取
- 需要计算或转换得到
注意：处理加密视频时，请确保你有权下载和使用这些内容，遵守相关网站的使用条款。
2. 环境准备与工具安装
开始之前，我们需要准备以下Python库：
pip install requests pycryptodome关键库说明：
requests：用于HTTP请求获取M3U8文件和TS片段pycryptodome：提供AES解密功能（比pycrypto更活跃维护）
如果你遇到安装问题，可以尝试以下替代方案：
| 库名 | 替代方案 | 备注 |
|---|---|---|
| pycryptodome | cryptography | 功能更全面但API不同 |
| requests | urllib3 | 更底层，使用稍复杂 |
3. 完整代码实现与分步解析
下面是我们实现的核心代码，我将逐部分解释其工作原理：
import requests import re import os from Crypto.Cipher import AES from urllib.parse import urljoin class M3U8Downloader: def __init__(self, base_url): self.base_url = base_url self.session = requests.Session() self.session.headers.update({ 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)' }) def download_m3u8(self, m3u8_url): response = self.session.get(m3u8_url) if response.status_code != 200: raise Exception(f"Failed to fetch M3U8: {response.status_code}") return response.text3.1 解析M3U8文件获取关键信息
M3U8文件中包含TS片段列表和加密信息，我们需要从中提取：
- 密钥URL（如果有加密）
- 所有TS片段的URL
- 可能的IV值
def parse_m3u8(self, m3u8_content): # 提取加密信息 key_info = re.search(r'#EXT-X-KEY:METHOD=AES-128,URI="([^"]+)"', m3u8_content) iv_info = re.search(r'IV=0x([0-9a-fA-F]+)', m3u8_content) # 提取所有TS片段 ts_urls = re.findall(r'^[^#].*\.ts$', m3u8_content, re.MULTILINE) return { 'key_url': key_info.group(1) if key_info else None, 'iv': iv_info.group(1) if iv_info else '0'*32, 'ts_urls': ts_urls }3.2 下载并处理密钥
获取密钥后，可能需要根据实际情况进行转换：
def get_key(self, key_url): if not key_url.startswith('http'): key_url = urljoin(self.base_url, key_url) response = self.session.get(key_url) if response.status_code != 200: raise Exception(f"Failed to fetch key: {response.status_code}") return response.content3.3 下载并解密TS片段
这是最核心的部分，我们使用AES-CBC模式解密每个TS片段：
def download_and_decrypt_ts(self, ts_url, key, iv, output_dir, index): if not os.path.exists(output_dir): os.makedirs(output_dir) full_url = urljoin(self.base_url, ts_url) response = self.session.get(full_url) if response.status_code != 200: print(f"Failed to download {ts_url}") return None cipher = AES.new(key, AES.MODE_CBC, iv) decrypted_data = cipher.decrypt(response.content) output_path = os.path.join(output_dir, f"segment_{index:04d}.ts") with open(output_path, 'wb') as f: f.write(decrypted_data) return output_path4. 完整工作流程与异常处理
现在我们把所有部分组合起来，形成完整的下载流程：
def download_video(self, m3u8_url, output_dir='output'): # 1. 下载M3U8文件 m3u8_content = self.download_m3u8(m3u8_url) # 2. 解析M3U8内容 m3u8_info = self.parse_m3u8(m3u8_content) # 3. 获取密钥 key = self.get_key(m3u8_info['key_url']) if m3u8_info['key_url'] else None iv = bytes.fromhex(m3u8_info['iv']) # 4. 下载并解密所有TS片段 downloaded_files = [] for i, ts_url in enumerate(m3u8_info['ts_urls']): file_path = self.download_and_decrypt_ts( ts_url, key, iv, output_dir, i ) if file_path: downloaded_files.append(file_path) print(f"Downloaded {i+1}/{len(m3u8_info['ts_urls'])}") return downloaded_files常见错误及解决方案：
密钥获取失败：
- 检查密钥URL是否正确
- 可能需要添加特定的请求头
解密后视频无法播放：
- 确认IV值是否正确
- 检查密钥是否需要额外处理（如Base64解码）
TS片段下载失败：
- 可能是反爬机制触发
- 尝试添加Referer等请求头
5. 合并TS文件与格式转换
下载完所有TS片段后，我们可以将它们合并为一个完整的视频文件：
def merge_ts_files(ts_files, output_path): with open(output_path, 'wb') as merged: for ts_file in sorted(ts_files): with open(ts_file, 'rb') as f: merged.write(f.read()) print(f"Merged video saved to {output_path}")如果需要转换为MP4格式，可以使用FFmpeg：
ffmpeg -i merged.ts -c copy output.mp46. 实战案例：处理特殊情况的技巧
在实际项目中，你可能会遇到以下特殊情况：
动态密钥：每次请求密钥URL返回不同的密钥
- 解决方案：在下载每个TS片段前重新获取密钥
密钥转换：密钥可能需要Base64解码或十六进制转换
import base64 key = base64.b64decode(key_content)分片M3U8：主M3U8指向多个子M3U8
- 需要递归处理所有子M3U8文件
反爬机制：
- 添加合理的请求间隔
- 模拟浏览器行为（User-Agent、Cookies等）
7. 性能优化与代码改进
对于大量TS片段，我们可以进行以下优化：
多线程下载：
from concurrent.futures import ThreadPoolExecutor with ThreadPoolExecutor(max_workers=5) as executor: futures = [ executor.submit( self.download_and_decrypt_ts, ts_url, key, iv, output_dir, i ) for i, ts_url in enumerate(ts_urls) ]断点续传：
- 记录已下载的片段
- 跳过已存在的文件
进度显示：
from tqdm import tqdm for i, ts_url in enumerate(tqdm(ts_urls)): self.download_and_decrypt_ts(ts_url, key, iv, output_dir, i)错误重试机制：
from tenacity import retry, stop_after_attempt @retry(stop=stop_after_attempt(3)) def download_with_retry(self, url): return self.session.get(url)
在实际使用中，我发现最常遇到的问题是不正确的IV处理。有些平台会使用全零的IV，而有些则会动态生成。如果解密后的视频开头几秒是乱码，但后面正常，很可能就是IV值设置有问题。