diff --git a/py/tests/test_czzy.py b/py/tests/test_czzy.py index 513f33f..d0318bc 100644 --- a/py/tests/test_czzy.py +++ b/py/tests/test_czzy.py @@ -1,8 +1,12 @@ import unittest +from base64 import b64encode from importlib.machinery import SourceFileLoader from pathlib import Path from unittest.mock import patch +from Cryptodome.Cipher import AES +from Cryptodome.Util.Padding import pad + ROOT = Path(__file__).resolve().parents[1] MODULE = SourceFileLoader("czzy_spider", str(ROOT / "厂长资源.py")).load_module() @@ -147,6 +151,55 @@ class TestCZZYSpider(unittest.TestCase): self.assertEqual(result["list"][0]["vod_id"], "/movie/example.html") self.assertEqual(result["list"][0]["vod_name"], "示例影片") + def test_extract_iframe_src(self): + html = '' + self.assertEqual( + self.spider._extract_iframe_src(html, "https://www.czzy89.com"), + "https://www.czzy89.com/player-v2/test", + ) + + def test_extract_player_url_prefers_mysvg_then_art_url(self): + self.assertEqual( + self.spider._extract_player_url_from_iframe("var mysvg='https://video.example/a.m3u8';"), + "https://video.example/a.m3u8", + ) + self.assertEqual( + self.spider._extract_player_url_from_iframe("art.url='https://video.example/b.m3u8';"), + "https://video.example/b.m3u8", + ) + + def test_extract_player_url_decodes_data_payload(self): + original = "https://video.example/data.m3u8" + middle = len(original) // 2 + obfuscated = original[:middle] + "ABCDEFG" + original[middle:] + encoded = "".join("{:02x}".format(ord(ch)) for ch in obfuscated)[::-1] + html = 'var config = {"data":"%s"};' % encoded + self.assertEqual(self.spider._extract_player_url_from_iframe(html), original) + + def test_extract_player_url_decrypts_player_payload(self): + payload = '{"url":"https://video.example/encrypted.m3u8"}' + key = b"VFBTzdujpR9FWBhe" + iv = b"1234567890abcdef" + cipher = AES.new(key, AES.MODE_CBC, iv) + encrypted = b64encode(cipher.encrypt(pad(payload.encode("utf-8"), AES.block_size))).decode("utf-8") + html = 'var player="%s";var rand="%s";' % (encrypted, iv.decode("utf-8")) + self.assertEqual( + self.spider._extract_player_url_from_iframe(html), + "https://video.example/encrypted.m3u8", + ) + + def test_extract_player_url_supports_wp_nonce_fallback(self): + html = """ + + """ + self.assertEqual( + self.spider._extract_player_url_from_iframe(html), + "https://video.example/wp.m3u8", + ) + if __name__ == "__main__": unittest.main() diff --git a/py/厂长资源.py b/py/厂长资源.py index e1924bf..48da7b7 100644 --- a/py/厂长资源.py +++ b/py/厂长资源.py @@ -1,7 +1,13 @@ # coding=utf-8 +import base64 +import json +import re import sys from urllib.parse import quote, urljoin +from Cryptodome.Cipher import AES +from Cryptodome.Util.Padding import unpad + from base.spider import Spider as BaseSpider sys.path.append("..") @@ -209,6 +215,66 @@ class Spider(BaseSpider): } return {"list": [vod]} + def _extract_iframe_src(self, html, host): + match = re.search(r"]+src=['\"]([^'\"]+)['\"]", html, re.I) + if not match: + return "" + return self._normalize_url(match.group(1), host) + + def _decode_data_url(self, value): + try: + encrypted = value[::-1] + temp = "" + for idx in range(0, len(encrypted), 2): + pair = encrypted[idx:idx + 2] + if len(pair) == 2: + temp += chr(int(pair, 16)) + middle = (len(temp) - 7) // 2 + return temp[:middle] + temp[middle + 7:] + except Exception: + return "" + + def _decrypt_player_payload(self, cipher_text, iv): + try: + cipher = AES.new(b"VFBTzdujpR9FWBhe", AES.MODE_CBC, iv.encode("utf-8")) + raw = base64.b64decode(cipher_text) + value = unpad(cipher.decrypt(raw), AES.block_size).decode("utf-8") + return json.loads(value).get("url", "") + except Exception: + return "" + + def _extract_player_url_from_iframe(self, html): + match = re.search( + r"var\s+player\s*=\s*[\"']([^\"']+)[\"'].*?var\s+rand\s*=\s*[\"']([^\"']+)[\"']", + html, + re.S, + ) + if match: + value = self._decrypt_player_payload(match.group(1), match.group(2)) + if value: + return value + + match = re.search(r"[\"']data[\"']\s*:\s*[\"']([^\"']+)[\"']", html) + if match: + value = self._decode_data_url(match.group(1)) + if value: + return value + + match = re.search(r"\bmysvg\b\s*=\s*[\"']([^\"']+)[\"']", html, re.I) + if match: + return match.group(1) + + match = re.search(r"art\.url\s*=\s*[\"']([^\"']+)[\"']", html, re.I) + if match: + return match.group(1) + + if "window.wp_nonce" in html: + match = re.search(r"url\s*:\s*[\"']([^\"']+)[\"']", html, re.I) + if match: + return match.group(1) + + return "" + def categoryContent(self, tid, pg, filter, extend): path = self.category_paths.get(tid, self.category_paths["movie"]).format(pg=pg) html, host = self._request_html(path, expect_xpath="//a[@href]")