Python实战手把手教你解密并下载AES-128加密的M3U8视频流附完整代码最近在帮朋友处理一个在线教育平台的视频下载需求时遇到了AES-128加密的M3U8视频流。这种加密方式在各大视频平台都很常见但完整实现解密下载的教程却不多。本文将带你从零开始用Python实现整个解密下载流程包括分析网页结构、获取密钥、解密TS片段和最终合并视频。1. 理解M3U8视频流与AES加密原理M3U8是一种基于HTTP Live StreamingHLS协议的视频播放列表格式它将整个视频分割成多个小的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更活跃维护如果你遇到安装问题可以尝试以下替代方案库名替代方案备注pycryptodomecryptography功能更全面但API不同requestsurllib3更底层使用稍复杂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(fFailed 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:METHODAES-128,URI([^]), m3u8_content) iv_info re.search(rIV0x([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(fFailed 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(fFailed 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, fsegment_{index:04d}.ts) with open(output_path, wb) as f: f.write(decrypted_data) return output_path4. 完整工作流程与异常处理现在我们把所有部分组合起来形成完整的下载流程def download_video(self, m3u8_url, output_diroutput): # 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(fDownloaded {i1}/{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(fMerged video saved to {output_path})如果需要转换为MP4格式可以使用FFmpegffmpeg -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_workers5) 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(stopstop_after_attempt(3)) def download_with_retry(self, url): return self.session.get(url)在实际使用中我发现最常遇到的问题是不正确的IV处理。有些平台会使用全零的IV而有些则会动态生成。如果解密后的视频开头几秒是乱码但后面正常很可能就是IV值设置有问题。