HLS AES-128 加密视频解密实战:Fiddler 抓包提取密钥与 Python 3.12 自动化下载

流媒体技术已经成为现代互联网视频传输的主流方案,其中 HLS(HTTP Live Streaming)协议因其良好的兼容性和适应性被广泛采用。然而,许多平台为了保护版权内容,会对 HLS 视频流进行 AES-128 加密处理。本文将深入探讨如何通过 Fiddler 抓包工具提取加密密钥,并利用 Python 3.12 实现自动化下载和解密流程。

1. HLS 加密原理与关键技术解析

HLS 协议的核心是将视频分割成一系列小的 TS(Transport Stream)文件,并通过 M3U8 索引文件进行管理。当视频内容被加密时,M3U8 文件中会包含 #EXT-X-KEY 标签,指明加密方法和密钥获取方式。

AES-128 加密是 HLS 最常用的加密方案,其工作流程如下:

  1. 密钥生成 :内容提供商使用 16 字节随机数生成加密密钥
  2. 分片加密 :每个 TS 文件使用相同的密钥进行 AES-128 CBC 模式加密
  3. 密钥分发 :密钥通过 HTTPS 单独传输,M3U8 文件仅包含密钥 URI
  4. 初始向量(IV) :可为固定值或随机生成,用于 CBC 模式的初始块

典型的加密 M3U8 文件示例如下:

#EXTM3U
#EXT-X-VERSION:3
#EXT-X-TARGETDURATION:10
#EXT-X-MEDIA-SEQUENCE:0
#EXT-X-KEY:METHOD=AES-128,URI="https://example.com/key.key",IV=0x00000000000000000000000000000000
#EXTINF:10.000000,
segment0.ts
#EXTINF:10.000000,
segment1.ts

2. Fiddler 抓包与密钥提取实战

Fiddler 是一款强大的 HTTP 调试代理工具,在分析加密视频流时尤为有用。以下是使用 Fiddler 捕获密钥的具体步骤:

2.1 Fiddler 基础配置

  1. 安装 HTTPS 证书

    • 打开 Fiddler → Tools → Options → HTTPS
    • 勾选 "Decrypt HTTPS traffic"
    • 点击 "Actions" → "Trust Root Certificate"
  2. 过滤目标流量

    • 在右侧 Filters 面板中设置目标主机名
    • 或使用 Ctrl+F 搜索 ".m3u8" 或 "key"

2.2 密钥捕获与提取

当播放加密视频时,Fiddler 会捕获到两类关键请求:

  1. M3U8 文件请求 :包含 #EXT-X-KEY 标签
  2. 密钥请求 :返回 16 字节二进制密钥数据

获取密钥的两种方法:

方法一:直接查看响应体

  1. 找到密钥请求(通常为 key/key.key/encryptkey 等路径)
  2. 在 Inspectors → Headers 查看 Content-Length 应为 16
  3. 切换到 HexView 标签页,复制 16 进制数据

方法二:Base64 编码导出

  1. 右键点击密钥请求 → Save → Response → Entire Response
  2. 用文本编辑器打开保存的文件,删除 HTTP 头部分
  3. 对剩余二进制数据进行 Base64 编码

注意:某些平台会动态生成密钥,需要确保在同一个会话中完成抓包和下载操作。

3. Python 自动化下载解密实现

下面是一个完整的 Python 3.12 脚本,实现从 M3U8 解析到 TS 下载解密的自动化流程:

import os
import re
import requests
from Crypto.Cipher import AES
from concurrent.futures import ThreadPoolExecutor
from urllib.parse import urljoin

class HLSDownloader:
    def __init__(self, m3u8_url, output_file='output.mp4'):
        self.m3u8_url = m3u8_url
        self.output_file = output_file
        self.session = requests.Session()
        self.session.headers.update({
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)'
        })

    def fetch_key(self, key_uri, iv=None):
        """获取并解析解密密钥"""
        response = self.session.get(key_uri)
        if response.status_code != 200:
            raise ValueError(f"密钥获取失败: {response.status_code}")
        
        key = response.content
        if len(key) != 16:
            raise ValueError("密钥长度必须为16字节")
        
        # 处理初始向量
        if iv is None:
            iv = bytes([0]*16)
        elif isinstance(iv, str):
            if iv.startswith('0x'):
                iv = bytes.fromhex(iv[2:])
            else:
                iv = iv.encode('utf-8')
        
        return key, iv

    def parse_m3u8(self):
        """解析M3U8文件获取TS片段和加密信息"""
        response = self.session.get(self.m3u8_url)
        if response.status_code != 200:
            raise ValueError("M3U8文件获取失败")
        
        content = response.text.splitlines()
        ts_list = []
        key_uri = None
        iv = None
        
        for line in content:
            line = line.strip()
            if line.startswith('#EXT-X-KEY'):
                # 解析加密信息
                match = re.search(r'URI="([^"]+)"', line)
                if match:
                    key_uri = urljoin(self.m3u8_url, match.group(1))
                
                iv_match = re.search(r'IV=([^,]+)', line)
                if iv_match:
                    iv = iv_match.group(1)
            
            elif line and not line.startswith('#'):
                # 添加TS片段
                ts_url = urljoin(self.m3u8_url, line)
                ts_list.append(ts_url)
        
        return ts_list, key_uri, iv

    def download_ts(self, ts_url, cipher=None):
        """下载并解密单个TS片段"""
        response = self.session.get(ts_url, stream=True)
        if response.status_code != 200:
            print(f"下载失败: {ts_url}")
            return None
        
        data = response.content
        if cipher:
            data = cipher.decrypt(data)
        
        return data

    def run(self):
        """执行完整下载流程"""
        # 1. 解析M3U8
        ts_list, key_uri, iv = self.parse_m3u8()
        if not ts_list:
            raise ValueError("未找到有效的TS片段")
        
        # 2. 获取解密密钥
        cipher = None
        if key_uri:
            key, iv = self.fetch_key(key_uri, iv)
            cipher = AES.new(key, AES.MODE_CBC, iv=iv)
            print(f"成功获取解密密钥: {key.hex()}")
        
        # 3. 多线程下载TS片段
        print(f"开始下载 {len(ts_list)} 个TS片段...")
        with ThreadPoolExecutor(max_workers=8) as executor:
            futures = [executor.submit(self.download_ts, ts, cipher) for ts in ts_list]
            ts_data = [f.result() for f in futures if f.result()]
        
        # 4. 合并TS文件
        print("合并TS片段...")
        with open(self.output_file, 'wb') as f:
            for data in ts_data:
                f.write(data)
        
        print(f"视频下载完成: {self.output_file}")

if __name__ == '__main__':
    # 使用示例
    m3u8_url = "https://example.com/playlist.m3u8"
    downloader = HLSDownloader(m3u8_url)
    downloader.run()

3.1 关键代码解析

  1. 密钥处理

    • fetch_key 方法处理密钥URI和初始向量
    • 支持十六进制( 0x... )和字符串格式的IV
    • 自动验证密钥长度是否为16字节
  2. M3U8解析

    • 使用正则表达式提取 #EXT-X-KEY 标签
    • 自动补全相对路径的TS片段URL
  3. 并发下载

    • 使用 ThreadPoolExecutor 实现多线程下载
    • 每个线程独立处理下载和解密
  4. 解密处理

    • 使用 PyCryptodome 库的 AES CBC 模式
    • 解密在下载流中实时完成,减少内存占用

3.2 依赖安装

执行脚本前需安装以下依赖:

pip install requests pycryptodome

4. 高级技巧与异常处理

在实际应用中,可能会遇到各种特殊情况,以下是几个常见问题的解决方案:

4.1 动态密钥与鉴权

许多商业平台会采用动态密钥和访问控制:

# 处理带鉴权参数的密钥请求示例
def fetch_key_with_auth(self, key_uri, auth_token):
    params = {'token': auth_token}
    response = self.session.get(key_uri, params=params)
    return response.content

4.2 TS 片段校验

为确保下载完整性,可添加校验逻辑:

def download_ts(self, ts_url, cipher=None, retry=3):
    for attempt in range(retry):
        try:
            response = self.session.get(ts_url, timeout=10)
            response.raise_for_status()
            data = response.content
            if len(data) == 0:
                raise ValueError("空内容")
            
            if cipher:
                data = cipher.decrypt(data)
            
            # 简单校验TS文件头
            if data[0] != 0x47:  # TS同步字节
                raise ValueError("无效TS格式")
            
            return data
        except Exception as e:
            if attempt == retry - 1:
                print(f"下载失败 {ts_url}: {str(e)}")
                return None

4.3 性能优化

对于大型视频,可采用以下优化策略:

  1. 分段合并 :每下载N个TS文件就合并一次,减少内存压力
  2. 进度显示 :添加tqdm进度条显示下载进度
  3. 断点续传 :记录已下载的TS片段,支持中断后继续
from tqdm import tqdm

def run_with_progress(self):
    ts_list, key_uri, iv = self.parse_m3u8()
    cipher = self.prepare_cipher(key_uri, iv)
    
    with tqdm(total=len(ts_list), desc="下载进度") as pbar:
        with ThreadPoolExecutor(max_workers=8) as executor:
            futures = {executor.submit(self.download_ts, ts, cipher): ts for ts in ts_list}
            with open(self.output_file, 'wb') as f:
                for future in as_completed(futures):
                    data = future.result()
                    if data:
                        f.write(data)
                    pbar.update(1)

5. 安全与法律注意事项

在实施HLS视频下载方案时,必须注意以下关键点:

  1. 版权合规

    • 仅下载拥有合法权限的内容
    • 不得规避技术保护措施用于侵权用途
  2. 网络安全

    • 使用HTTPS确保密钥传输安全
    • 避免在代码中硬编码敏感信息
  3. 资源节制

    • 添加合理的请求间隔(如 time.sleep(0.1) )
    • 设置User-Agent标识合规爬虫
  4. 数据保护

    • 下载的内容不应在未授权情况下二次分发
    • 及时删除临时解密文件

在实际开发中,建议将下载速度控制在合理范围内,并尊重网站的robots.txt协议。对于商业级应用,应考虑使用官方API而非逆向工程方案。

Logo

Agent 垂直技术社区,欢迎活跃、内容共建。

更多推荐