脱敏说明：目标站点域名、AES 密钥/IV 与签名值已脱敏，仅保留分析方法与流程。
搜索
import requests
url = "https://api-target-01.example.com/api.php/api/search/mv"
payload='timestamp=1724380094&data=c%2BrZEZylJXvSpVbGzxhjZnE6YRwVVRfSKKIyDsElnt%2FVPhZZ7DX2PFo1GY9sR6DHF2ZYZ%2B5cTStJ%2F3VjfBa43lu0EMj5oYygww1Afw8n4DW%2B05bdxMC9gQCGj7GDbMXNYwAfFGks1WbYZVSPBgjfitnhlrKs74ETXwd5xUL5%2FD6O4nAZXnBWjTbdXSRPoXzcn7Kh4tJ1opwIjDZ6nUTBaRs6K0g30rQk3DlHy%2BsacUBTKQ5HDsCg9gy9oj4%2F4WRF&sign=REDACTED'
headers = {
'Host': 'api-target-01.example.com',
'user-agent': 'okhttp/3.14.9',
'content-type': 'application/x-www-form-urlencoded'
}
response = requests.request("POST", url, headers=headers, data=payload)
print(response.text)
请求参数
timestamp=1724380094
data=c%2BrZEZylJXvSpVbGzxhjZnE6YRwVVRfSKKIyDsElnt%2FVPhZZ7DX2PFo1GY9sR6DHF2ZYZ%2B5cTStJ%2F3VjfBa43lu0EMj5oYygww1Afw8n4DW%2B05bdxMC9gQCGj7GDbMXNYwAfFGks1WbYZVSPBgjfitnhlrKs74ETXwd5xUL5%2FD6O4nAZXnBWjTbdXSRPoXzcn7Kh4tJ1opwIjDZ6nUTBaRs6K0g30rQk3DlHy%2BsacUBTKQ5HDsCg9gy9oj4%2F4WRF
sign=REDACTED
timestamp
秒级时间戳
data的组成
拼接json
{"oauth_id":"REDACTED","bundle_id":"com.target.app","version":"2.0.5","oauth_type":"web","language":"zh","via":"pwa","kwy":"搜索关键词","page":1,"limit":30,"style":"0"}
-
运算模式：
AES-CBC (密码块链)
-
填充模式：
NonePKCS7
-
密钥长度：
128 bits
密钥： REDACTED
偏移： REDACTED
sign的组成
- 拼接字符串data={上面字符串加密的data}×tamp={上面的时间戳后拼接一个固定字符串}REDACTED
- 然后对他进行sha-256处理
- 对上文进行md5处理
- 得到sign，发包
注意！！！
此时获取到的视频只有2分钟，如果想获取完整视频需要将域名前的120play或者10play子域名替换成m3u8然后请求得到密文，再次解密才可获取视频
成品
'''
Descripttion:
version: 39
Author: sikuai
Date: 2024-08-23 10:32:53
LastEditors: sikuai
LastEditTime: 2024-11-24 12:45:59
'''
import requests
import hashlib
import json
from Crypto.Cipher import AES
import binascii
import base64
import time
import os
import subprocess
import shutil
# AES加密配置
AESKEY = 'REDACTED' # 密钥
AESIV = 'REDACTED' # 偏移量
DECRYPTION_KEY = 'REDACTED3ebace31' # 解密密钥
class AESTool:
def __init__(self):
self.key = AESKEY.encode('utf-8')
self.iv = AESIV.encode('utf-8')
def pkcs7padding(self, text):
padding = 16 - len(text) % 16
return text + chr(padding) * padding
def pkcs7unpadding(self, text):
padding = ord(text[-1])
return text[:-padding]
def aes_encrypt(self, content):
cipher = AES.new(self.key, AES.MODE_CBC, self.iv)
content_padding = self.pkcs7padding(content)
encrypt_bytes = cipher.encrypt(content_padding.encode('utf-8'))
return base64.b64encode(encrypt_bytes).decode('utf-8')
def aes_decrypt(self, content):
cipher = AES.new(self.key, AES.MODE_CBC, self.iv)
decrypted_bytes = cipher.decrypt(base64.b64decode(content))
return self.pkcs7unpadding(decrypted_bytes.decode('utf-8'))
# 解密m3u8加密后的密文
def decrypt_m3u8_cipher(self, cipher_text):
# 提取CFB模式的偏移量
iv = binascii.unhexlify(cipher_text[:32])
# 提取密文
encrypted_data = binascii.unhexlify(cipher_text[32:])
# 解密密钥
decryption_key = binascii.unhexlify(DECRYPTION_KEY)
# 创建AES解密器
cipher = AES.new(decryption_key, AES.MODE_CFB, iv, segment_size=128)
# 解密数据
decrypted_data = cipher.decrypt(encrypted_data)
# 确保解密后的数据不为空
if decrypted_data:
# 如果使用了PKCS7填充，需要去除填充
# 这里假设解密后的数据使用了PKCS7填充
padding_len = decrypted_data[-1]
if isinstance(padding_len, int):
unpaded_data = decrypted_data[:-padding_len]
else:
unpaded_data = decrypted_data
return unpaded_data.decode('utf-8')
else:
raise ValueError("Decrypted data is empty.")
def search():
# 准备要加密的JSON数据
data_json = {
"oauth_id": "REDACTED",
"bundle_id": "com.target.app",
"version": "2.0.5",
"oauth_type": "web",
"language": "zh",
"via": "pwa",
"kwy": "正太",
"page": 1,
"limit": 30,
"style": "0"
}
# 生成时间戳
timestamp = str(int(time.time()))
aes_tool = AESTool()
# 将JSON数据转换为字符串并加密
encrypted_data = aes_tool.aes_encrypt(json.dumps(data_json))
# 生成sign
sign_str = f"data={encrypted_data}×tamp={timestamp}REDACTED"
sign_md5 = hashlib.md5(hashlib.sha256(sign_str.encode('utf-8')).hexdigest().encode('utf-8')).hexdigest()
# 发送HTTP请求
url = "https://api-target-01.example.com/api.php/api/search/mv"
headers = {
'Host': 'api-target-01.example.com',
'user-agent': 'okhttp/3.14.9',
'content-type': 'application/x-www-form-urlencoded'
}
payload = f"timestamp={timestamp}&data={encrypted_data}&sign={sign_md5}"
return url, headers, payload
def list_tab(page):
#https://api-target-02.example.com/api.phpapi/tabnew/list_tab_mv
#{"oauth_id":"REDACTED","bundle_id":"com.target.app","version":"2.0.5","oauth_type":"web","language":"zh","via":"pwa","tab_id":"93","limit":30,"page":2,"sort":"hot"}
data_json = {
"oauth_id": "REDACTED",
"bundle_id": "com.target.app",
"version": "2.0.5",
"oauth_type": "web",
"language": "zh",
"via": "pwa",
"tab_id": "93",
"limit": 30,
"page": page,
"sort": "hot"
}
# 生成时间戳
timestamp = str(int(time.time()))
aes_tool = AESTool()
encrypted_data = aes_tool.aes_encrypt(json.dumps(data_json))
sign_str = f"data={encrypted_data}×tamp={timestamp}REDACTED"
sign_md5 = hashlib.md5(hashlib.sha256(sign_str.encode('utf-8')).hexdigest().encode('utf-8')).hexdigest()
url = "https://api-target-02.example.com/api.php/api/tabnew/list_tab_mv"
headers = {
'Host': 'api-target-02.example.com',
'user-agent': 'okhttp/3.14.9',
'content-type': 'application/x-www-form-urlencoded'
}
payload = f"timestamp={timestamp}&data={encrypted_data}&sign={sign_md5}"
return url, headers, payload
def download(url, headers, payload):
aes_tool = AESTool()
try:
response = requests.post(url, headers=headers, data=payload)
response.raise_for_status() # 检查请求是否成功
response_json = response.json()
# 检查errcode
if response_json.get('errcode') == 0:
# 解密data字段
decrypted_data = aes_tool.aes_decrypt(response_json.get('data'))
# 处理解密后的json数据
data_list = json.loads(decrypted_data).get('data', {}).get('list', [])
# 读取现有标题并去掉换行符
with open('mv_title.txt', 'r') as f:
mv_id_title_list = set(line.strip() for line in f.readlines())
for item in data_list:
title = item['title']
if title in mv_id_title_list:
print(f"已存在: {title}")
continue
else:
with open('mv_title.txt', 'a') as f:
f.write(f"{title}\n")
mv_id_title_list.add(title) # 更新集合
print(item['id'], title, item['play_url'])
# 对item['play_url']进行处理，子域名替换成m3u8，通过split(.)实现
play_url = item['play_url']
play_url_list = play_url.split('.')
play_url_list[0] ='m3u8'
play_url = '.'.join(play_url_list)
play_url = "https://" + play_url
print(f"处理后的播放地址: {play_url}")
# 访问play_url并获取响应内容
res_text = requests.get(play_url).text
# 内容是m3u8加密后的密文，先进行切分，前32位为cfb偏移量，DECRYPTION_KEY为解密密钥，后面的内容为密文，aes、pkcs7，nopadding解密
# 注意密文，密钥，偏移量均为hex字符串，解密时记得处理
aes_tool = AESTool()
decrypted_content = aes_tool.decrypt_m3u8_cipher(res_text)
# print(decrypted_content)
# 保存解密后的内容到文件
with open(f"{item['id']}{item['title']}.m3u8", 'w', encoding='utf-8') as f:
f.write(decrypted_content)
# 调用ffmpeg命令行工具下载 命令行执行： 1.exe "f"{item['id']}{item['title']}.m3u8"" --workDir "C:\Users\sikuai\Desktop\代码开发\python\暗网" --saveName "{item['title']}"
# 构建命令行参数列表
current_dir = os.getcwd()
# 构建命令行参数列表，使用相对路径
command = [
'N_m3u8DL-RE.exe',
os.path.join(current_dir, f"{item['id']}{item['title']}.m3u8"),
'--tmp-dir',
os.path.join(current_dir, item["title"]),
'--save-name',
f"{item['title']}",
'--save-dir',
current_dir,
'--no-log',
'--log-level=WARN'
]
subprocess.run(command, shell=True)
# 删除临时文件和{item['title']}工作文件夹
os.remove(f"{item['id']}{item['title']}.m3u8")
# shutil.rmtree(f"{item['title']}")
else:
print("Error: errcode is not 0.")
except requests.RequestException as e:
print(f"请求异常: {e}")
except Exception as e:
print(f"发生了错误: {e}")
if __name__ == '__main__':
aes_tool = AESTool()
# 下载for循环
for i in range(1, 100):
print(f"正在下载第{i}页...")
list_tab_url, list_tab_headers, list_tab_payload = list_tab(i)
download(list_tab_url, list_tab_headers, list_tab_payload)