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())