mirror of
https://github.com/vinefeeder/TwinVine.git
synced 2026-07-15 18:10:04 +02:00
450 lines
18 KiB
Python
450 lines
18 KiB
Python
import sys
|
|
from PyQt6.QtWidgets import (
|
|
QApplication, QWidget, QVBoxLayout, QLabel, QLineEdit, QTextEdit, QPushButton, QCheckBox, QFrame, QMessageBox, QHBoxLayout
|
|
)
|
|
from PyQt6.QtCore import Qt, QThread, pyqtSignal
|
|
from PyQt6.QtGui import QPalette, QColor
|
|
import subprocess
|
|
import httpx
|
|
import base64
|
|
import re
|
|
import urllib.parse
|
|
import xml.etree.ElementTree as ET
|
|
from pywidevine.cdm import Cdm
|
|
from pywidevine.device import Device
|
|
from pywidevine.pssh import PSSH
|
|
from base64 import b64encode
|
|
import codecs
|
|
from pathlib import Path
|
|
import os
|
|
import shlex
|
|
|
|
WVD_PATH = "./WVDs/device.wvd"
|
|
WIDEVINE_SYSTEM_ID = 'EDEF8BA9-79D6-4ACE-A3C8-27DCD51D21ED'
|
|
|
|
class DownloadThread(QThread):
|
|
finished = pyqtSignal()
|
|
error = pyqtSignal(str)
|
|
|
|
def __init__(self, command):
|
|
super().__init__()
|
|
self.command = command
|
|
|
|
def run(self):
|
|
try:
|
|
subprocess.run(self.command, check=True)
|
|
self.finished.emit()
|
|
except subprocess.CalledProcessError as e:
|
|
self.error.emit(str(e))
|
|
|
|
class AllHell3App(QWidget):
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.init_ui()
|
|
|
|
def init_ui(self):
|
|
self.setWindowTitle("HellYes!")
|
|
layout = QVBoxLayout()
|
|
|
|
self.mpd_url_label = QLabel("MPD URL")
|
|
layout.addWidget(self.mpd_url_label)
|
|
self.mpd_url_entry = QLineEdit()
|
|
layout.addWidget(self.mpd_url_entry)
|
|
|
|
self.curl_label = QLabel("cURL of License Request")
|
|
layout.addWidget(self.curl_label)
|
|
self.curl_text = QTextEdit()
|
|
layout.addWidget(self.curl_text)
|
|
|
|
highlighted_frame = QFrame()
|
|
highlighted_layout = QVBoxLayout()
|
|
highlighted_frame.setLayout(highlighted_layout)
|
|
highlighted_frame.setStyleSheet("border: 1px solid red;")
|
|
|
|
self.video_name_label = QLabel("Video Name")
|
|
highlighted_layout.addWidget(self.video_name_label)
|
|
self.video_name_entry = QLineEdit()
|
|
highlighted_layout.addWidget(self.video_name_entry)
|
|
|
|
self.fetch_keys_button = QPushButton("Get Keys")
|
|
self.fetch_keys_button.clicked.connect(self.fetch_keys)
|
|
highlighted_layout.addWidget(self.fetch_keys_button)
|
|
|
|
self.download_button = QPushButton("Download Nm~RE")
|
|
self.download_button.clicked.connect(self.download_video)
|
|
highlighted_layout.addWidget(self.download_button)
|
|
|
|
self.download_button2 = QPushButton("Download DASH")
|
|
self.download_button2.clicked.connect(self.download_dash)
|
|
highlighted_layout.addWidget(self.download_button2)
|
|
|
|
layout.addWidget(highlighted_frame)
|
|
|
|
self.keys_output = QTextEdit()
|
|
layout.addWidget(self.keys_output)
|
|
|
|
self.n_m3u8dl_re_label = QLabel("N_m3u8DL-RE command")
|
|
layout.addWidget(self.n_m3u8dl_re_label)
|
|
self.n_m3u8dl_re_output = QTextEdit()
|
|
layout.addWidget(self.n_m3u8dl_re_output)
|
|
|
|
self.dash_mpd_cli_label = QLabel("Dash-MPD-CLI command")
|
|
layout.addWidget(self.dash_mpd_cli_label)
|
|
self.dash_mpd_cli_output = QTextEdit()
|
|
layout.addWidget(self.dash_mpd_cli_output)
|
|
|
|
checkbox_layout = QHBoxLayout()
|
|
self.dark_mode_checkbox = QCheckBox("Dark Mode")
|
|
self.dark_mode_checkbox.setChecked(True)
|
|
self.dark_mode_checkbox.stateChanged.connect(self.toggle_dark_mode)
|
|
checkbox_layout.addWidget(self.dark_mode_checkbox)
|
|
|
|
self.reset_button = QPushButton("Reset")
|
|
self.reset_button.clicked.connect(self.reset_fields)
|
|
checkbox_layout.addWidget(self.reset_button)
|
|
|
|
layout.addLayout(checkbox_layout)
|
|
|
|
self.setLayout(layout)
|
|
self.toggle_dark_mode()
|
|
|
|
def reset_fields(self):
|
|
self.mpd_url_entry.clear()
|
|
self.curl_text.clear()
|
|
self.video_name_entry.clear()
|
|
self.keys_output.clear()
|
|
self.n_m3u8dl_re_output.clear()
|
|
self.dash_mpd_cli_output.clear()
|
|
|
|
|
|
def toggle_dark_mode(self):
|
|
if self.dark_mode_checkbox.isChecked():
|
|
palette = QPalette()
|
|
palette.setColor(QPalette.ColorRole.Window, QColor(53, 53, 53))
|
|
palette.setColor(QPalette.ColorRole.WindowText, Qt.GlobalColor.white)
|
|
palette.setColor(QPalette.ColorRole.Base, QColor(35, 35, 35))
|
|
palette.setColor(QPalette.ColorRole.AlternateBase, QColor(53, 53, 53))
|
|
palette.setColor(QPalette.ColorRole.ToolTipBase, Qt.GlobalColor.black)
|
|
palette.setColor(QPalette.ColorRole.ToolTipText, Qt.GlobalColor.white)
|
|
palette.setColor(QPalette.ColorRole.Text, Qt.GlobalColor.white)
|
|
palette.setColor(QPalette.ColorRole.Button, QColor(53, 53, 53))
|
|
palette.setColor(QPalette.ColorRole.ButtonText, Qt.GlobalColor.white)
|
|
palette.setColor(QPalette.ColorRole.BrightText, Qt.GlobalColor.red)
|
|
palette.setColor(QPalette.ColorRole.Link, QColor(42, 130, 218))
|
|
palette.setColor(QPalette.ColorRole.Highlight, QColor(42, 130, 218))
|
|
palette.setColor(QPalette.ColorRole.HighlightedText, Qt.GlobalColor.black)
|
|
self.setPalette(palette)
|
|
|
|
self.mpd_url_label.setStyleSheet("color: white;")
|
|
self.curl_label.setStyleSheet("color: white;")
|
|
self.video_name_label.setStyleSheet("color: white;")
|
|
self.n_m3u8dl_re_label.setStyleSheet("color: white;")
|
|
self.dash_mpd_cli_label.setStyleSheet("color: white;")
|
|
self.dark_mode_checkbox.setStyleSheet("color: white;")
|
|
self.fetch_keys_button.setStyleSheet("""color: white;
|
|
background-color: #4e4e4e;
|
|
border: 1px solid #6e6e6e;
|
|
padding: 5px;""")
|
|
self.download_button.setStyleSheet("""color: white;
|
|
background-color: #4e4e4e;
|
|
border: 1px solid #6e6e6e;
|
|
padding: 5px;""")
|
|
|
|
self.download_button2.setStyleSheet("""color: white;
|
|
background-color: #4e4e4e;
|
|
border: 1px solid #6e6e6e;
|
|
padding: 5px;""")
|
|
|
|
self.curl_text.setStyleSheet("""color: white; border: 1px solid red;
|
|
background-color: #232323;""")
|
|
self.video_name_label.setStyleSheet("""
|
|
color: white; border: None;""")
|
|
self.mpd_url_entry.setStyleSheet("""color: white; border: 1px solid red;
|
|
background-color: #232323;""")
|
|
|
|
else:
|
|
self.setPalette(QApplication.palette())
|
|
|
|
self.mpd_url_label.setStyleSheet("color: black;")
|
|
self.curl_label.setStyleSheet("color: black;")
|
|
self.video_name_label.setStyleSheet("color: black;")
|
|
self.n_m3u8dl_re_label.setStyleSheet("color: black;")
|
|
self.dash_mpd_cli_label.setStyleSheet("color: black;")
|
|
self.dark_mode_checkbox.setStyleSheet("color: black;")
|
|
self.fetch_keys_button.setStyleSheet("""color: black;
|
|
background-color: #aeaeae;
|
|
border: 1px solid #6e6e6e;
|
|
padding: 5px;""")
|
|
self.download_button.setStyleSheet("""color: black;
|
|
background-color: #aeaeae;
|
|
border: 1px solid #6e6e6e;
|
|
padding: 5px;""")
|
|
self.download_button2.setStyleSheet("""color: black;
|
|
background-color: #aeaeae;
|
|
border: 1px solid #6e6e6e;
|
|
padding: 5px;""")
|
|
|
|
self.curl_text.setStyleSheet("""color: black; border: 1px solid red;
|
|
background-color: white;""")
|
|
self.video_name_label.setStyleSheet("""
|
|
color: black; border: None;""")
|
|
self.mpd_url_entry.setStyleSheet("""color: black; border: 1px solid red;
|
|
background-color: white;""")
|
|
|
|
def fetch_mpd_content(self, mpd_url):
|
|
response = httpx.get(mpd_url)
|
|
response.raise_for_status()
|
|
return response.text
|
|
|
|
def find_default_kid_with_regex(self, mpd_content):
|
|
match = re.search(r'cenc:default_KID="([A-F0-9-]+)"', mpd_content)
|
|
if match:
|
|
return match.group(1)
|
|
return None
|
|
|
|
def extract_or_generate_pssh(self, mpd_content):
|
|
try:
|
|
tree = ET.ElementTree(ET.fromstring(mpd_content))
|
|
root = tree.getroot()
|
|
|
|
namespaces = {
|
|
'cenc': 'urn:mpeg:cenc:2013',
|
|
'': 'urn:mpeg:dash:schema:mpd:2011'
|
|
}
|
|
|
|
default_kid = None
|
|
for elem in root.findall('.//ContentProtection', namespaces):
|
|
scheme_id_uri = elem.attrib.get('schemeIdUri', '').upper()
|
|
if scheme_id_uri == 'URN:MPEG:DASH:MP4PROTECTION:2011':
|
|
default_kid = elem.attrib.get('cenc:default_KID')
|
|
if default_kid:
|
|
break
|
|
|
|
if not default_kid:
|
|
default_kid = self.find_default_kid_with_regex(mpd_content)
|
|
|
|
pssh = None
|
|
for elem in root.findall('.//ContentProtection', namespaces):
|
|
scheme_id_uri = elem.attrib.get('schemeIdUri', '').upper()
|
|
if scheme_id_uri == f'URN:UUID:{WIDEVINE_SYSTEM_ID}':
|
|
pssh_elem = elem.find('cenc:pssh', namespaces)
|
|
if pssh_elem is not None:
|
|
pssh = pssh_elem.text
|
|
break
|
|
|
|
if pssh is not None:
|
|
return pssh
|
|
elif default_kid is not None:
|
|
default_kid = default_kid.replace('-', '')
|
|
s = f'000000387073736800000000edef8ba979d64acea3c827dcd51d21ed000000181210{default_kid}48e3dc959b06'
|
|
return b64encode(bytes.fromhex(s)).decode()
|
|
|
|
else:
|
|
# No pssh or default_KID found
|
|
try:
|
|
pssh = self.get_pssh_from_mpd(self.mpd_url_entry.text()) # init.m4f method
|
|
if pssh:
|
|
return pssh
|
|
else:
|
|
return None
|
|
except:
|
|
return None
|
|
|
|
except ET.ParseError as e:
|
|
try:
|
|
pssh = self.get_pssh_from_mpd(self.mpd_url_entry.text()) # init.m4f method
|
|
return pssh
|
|
except:
|
|
pass
|
|
print(f"Error parsing MPD content: {e}")
|
|
return None
|
|
|
|
|
|
def get_pssh_from_mpd(self, mpd: str):
|
|
print("Extracting PSSH from MPD...")
|
|
|
|
yt_dl = 'yt-dlp'
|
|
init = 'init.m4f'
|
|
|
|
files_to_delete = [init]
|
|
|
|
for file_name in files_to_delete:
|
|
if os.path.exists(file_name):
|
|
os.remove(file_name)
|
|
print(f"{file_name} file successfully deleted.")
|
|
|
|
try:
|
|
subprocess.run([yt_dl, '-q', '--no-warning', '--test', '--allow-u', '-f', 'bestvideo[ext=mp4]/bestaudio[ext=m4a]/best', '-o', init, mpd])
|
|
except FileNotFoundError:
|
|
print("yt-dlp not found. Trying installing it...")
|
|
exit(0)
|
|
|
|
pssh_list = self.extract_pssh_from_file('init.m4f')
|
|
pssh = None
|
|
for target_pssh in pssh_list:
|
|
if 20 < len(target_pssh) < 220:
|
|
pssh = target_pssh
|
|
|
|
print(f'\n{pssh}\n')
|
|
|
|
for file_name in files_to_delete:
|
|
if os.path.exists(file_name):
|
|
os.remove(file_name)
|
|
print(f"{file_name} file successfully deleted.")
|
|
return pssh
|
|
|
|
|
|
|
|
def extract_pssh_from_file(self,file_path: str) -> list:
|
|
print('Extracting PSSHs from init file:', file_path)
|
|
return self.to_pssh(Path(file_path).read_bytes())
|
|
|
|
def find_wv_pssh_offsets(self, raw: bytes) -> list:
|
|
offsets = []
|
|
offset = 0
|
|
while True:
|
|
offset = raw.find(b'pssh', offset)
|
|
if offset == -1:
|
|
break
|
|
size = int.from_bytes(raw[offset-4:offset], byteorder='big')
|
|
pssh_offset = offset - 4
|
|
offsets.append(raw[pssh_offset:pssh_offset+size])
|
|
offset += size
|
|
return offsets
|
|
|
|
def to_pssh(self, content: bytes) -> list:
|
|
wv_offsets = self.find_wv_pssh_offsets(content)
|
|
return [base64.b64encode(wv_offset).decode() for wv_offset in wv_offsets]
|
|
|
|
|
|
|
|
def parse_curl(self, curl_command):
|
|
url_match = re.search(r"curl\s+'(.*?)'", curl_command)
|
|
url = url_match.group(1) if url_match else ""
|
|
|
|
method_match = re.search(r"-X\s+(\w+)", curl_command)
|
|
method = method_match.group(1) if method_match else "UNDEFINED"
|
|
|
|
headers = {}
|
|
headers_matches = re.findall(r"-H\s+'([^:]+):\s*(.*?)'", curl_command)
|
|
for header in headers_matches:
|
|
headers[header[0]] = header[1]
|
|
|
|
data_match = re.search(r"--data(?:-raw)?\s+(?:(\$?')|(\$?{?))(.*?)'", curl_command, re.DOTALL)
|
|
if data_match:
|
|
raw_prefix = data_match.group(1)
|
|
data = data_match.group(3)
|
|
if raw_prefix and raw_prefix.startswith('$'):
|
|
data = None
|
|
else:
|
|
data = data.replace('\\\\', '\\').replace('\\x', '\\\\x')
|
|
try:
|
|
data = codecs.decode(data, 'unicode_escape')
|
|
except Exception as e:
|
|
data = ""
|
|
else:
|
|
data = ""
|
|
return url, method, headers, data
|
|
|
|
def get_key(self, pssh, license_url, headers, data):
|
|
device = Device.load(WVD_PATH)
|
|
cdm = Cdm.from_device(device)
|
|
session_id = cdm.open()
|
|
|
|
challenge = cdm.get_license_challenge(session_id, PSSH(pssh))
|
|
|
|
if data:
|
|
if match := re.search(r'"(CAQ=.*?)"', data):
|
|
challenge = data.replace(match.group(1), base64.b64encode(challenge).decode())
|
|
elif match := re.search(r'"(CAES.*?)"', data):
|
|
challenge = data.replace(match.group(1), base64.b64encode(challenge).decode())
|
|
elif match := re.search(r'=(CAES.*?)(&.*)?$', data):
|
|
b64challenge = base64.b64encode(challenge).decode()
|
|
quoted = urllib.parse.quote_plus(b64challenge)
|
|
challenge = data.replace(match.group(1), quoted)
|
|
|
|
payload = challenge if data is None else challenge
|
|
|
|
license_response = httpx.post(url=license_url, data=payload, headers=headers)
|
|
try:
|
|
license_response.raise_for_status()
|
|
except httpx.HTTPStatusError as e:
|
|
raise e
|
|
|
|
license_content = license_response.content
|
|
try:
|
|
match = re.search(r'"(CAIS.*?)"', license_response.content.decode('utf-8'))
|
|
if match:
|
|
license_content = base64.b64decode(match.group(1))
|
|
except:
|
|
pass
|
|
|
|
if isinstance(license_content, str):
|
|
license_content = base64.b64decode(license_content)
|
|
|
|
cdm.parse_license(session_id, license_content)
|
|
|
|
keys = []
|
|
for key in cdm.get_keys(session_id):
|
|
if key.type == 'CONTENT':
|
|
keys.append(f"--key {key.kid.hex}:{key.key.hex()}")
|
|
|
|
cdm.close(session_id)
|
|
return keys
|
|
|
|
def fetch_keys(self):
|
|
mpd_url = self.mpd_url_entry.text()
|
|
try:
|
|
mpd_content = self.fetch_mpd_content(mpd_url)
|
|
pssh = self.extract_or_generate_pssh(mpd_content)
|
|
if not pssh:
|
|
QMessageBox.critical(self, "Error", "Failed to extract or generate PSSH\nAre you sure the video is Widevine encrypted?.")
|
|
return
|
|
|
|
curl_command = self.curl_text.toPlainText().strip()
|
|
license_url, method, headers, data = self.parse_curl(curl_command)
|
|
keys = self.get_key(pssh, license_url, headers, data)
|
|
self.keys_output.clear()
|
|
self.keys_output.append("\n".join( keys))
|
|
key_str = " ".join(keys)
|
|
|
|
self.n_m3u8dl_re_command = f"N_m3u8DL-RE '{mpd_url}' {key_str} --save-name {self.video_name_entry.text()} -mt -M:format=mkv:muxer=mkvmerge"
|
|
self.n_m3u8dl_re_output.setText(self.n_m3u8dl_re_command)
|
|
|
|
self.dash_mpd_cli_command = f"dash-mpd-cli --quality best --muxer-preference mkv:mkvmerge {key_str} \"{mpd_url}\" --write-subs --output '{self.video_name_entry.text()}.mkv'"
|
|
|
|
self.dash_mpd_cli_output.setText(str(self.dash_mpd_cli_command))
|
|
|
|
except Exception as e:
|
|
QMessageBox.critical(self, "Error", str(e))
|
|
|
|
def download_video(self):
|
|
try:
|
|
self.thread = DownloadThread(shlex.split(self.n_m3u8dl_re_command))
|
|
self.thread.finished.connect(self.on_download_finished)
|
|
self.thread.error.connect(self.on_download_error)
|
|
self.thread.start()
|
|
except Exception as e:
|
|
QMessageBox.critical(self, "Error", str(e))
|
|
|
|
def download_dash(self):
|
|
try:
|
|
self.thread = DownloadThread(shlex.split(self.dash_mpd_cli_command))
|
|
self.thread.finished.connect(self.on_download_finished)
|
|
self.thread.error.connect(self.on_download_error)
|
|
self.thread.start()
|
|
except Exception as e:
|
|
QMessageBox.critical(self, "Error", str(e))
|
|
|
|
def on_download_finished(self):
|
|
QMessageBox.information(self, "Success", "Download finished successfully.")
|
|
|
|
def on_download_error(self, error_message):
|
|
QMessageBox.critical(self, "Error", f"Download failed: {error_message}")
|
|
|
|
if __name__ == "__main__":
|
|
app = QApplication(sys.argv)
|
|
window = AllHell3App()
|
|
window.show()
|
|
sys.exit(app.exec())
|