feat: mitm download

config.toml

@@ -23,3 +23,7 @@ afterDownloaded = ""
 embedMetadata = ["title", "artist", "album", "album_artist", "composer",
     "genre", "created", "track", "tracknum", "disk", "lyrics", "cover", "copyright",
     "record_company", "upc", "isrc"]
+
+[mitm]
+host = "127.0.0.1"
+port = "11080"

pyproject.toml

@@ -20,6 +20,7 @@ pure-python-adb = "^0.3.0.dev0"
 frida = "^16.2.1"
 tenacity = "^8.2.3"
 prompt-toolkit = "^3.0.43"
+mitmproxy = "^10.3.0"
 
 [build-system]
 requires = ["poetry-core"]

src/cmd.py

@@ -13,9 +13,9 @@ from src.api import get_token
 from src.config import Config
 from src.rip import rip_song, rip_album
 from src.types import GlobalAuthParams
-from src.url import AppleMusicURL, URLType
 from src.url import AppleMusicURL, URLType, Song
 from src.utils import get_song_id_from_m3u8
+from src.mitm import start_proxy
 
 
 class NewInteractiveShell:
@@ -47,6 +47,11 @@ class NewInteractiveShell:
                                  default="alac")
         m3u8_parser.add_argument("-f", "--force", type=bool, default=False)
         subparser.add_parser("exit")
+        mitm_parser = subparser.add_parser("mitm")
+        mitm_parser.add_argument("-c", "--codec",
+                                 choices=["alac", "ec3", "aac", "aac-binaural", "aac-downmix", "ac3"],
+                                 default="alac")
+        mitm_parser.add_argument("-f", "--force", type=bool, default=False)
 
         logger.remove()
         logger.add(lambda msg: print_formatted_text(ANSI(msg), end=""), colorize=True, level="INFO")
@@ -74,20 +79,17 @@ class NewInteractiveShell:
         match cmds[0]:
             case "download":
                 await self.do_download(args.url, args.codec, args.force)
+            case "m3u8":
+                await self.do_m3u8(args.url, args.codec, args.force)
+            case "mitm":
+                await self.do_mitm(args.codec, args.force)
             case "exit":
                 self.loop.stop()
                 sys.exit()
 
     async def do_download(self, raw_url: str, codec: str, force_download: bool):
         url = AppleMusicURL.parse_url(raw_url)
-        devices = self.storefront_device_mapping.get(url.storefront)
-        if not devices:
-            logger.error(f"No device is available to decrypt the specified region: {url.storefront}")
-        available_devices = [device for device in devices if not device.decryptLock.locked()]
-        if not available_devices:
-            available_device: Device = random.choice(devices)
-        else:
-            available_device: Device = random.choice(available_devices)
+        available_device = await self._get_available_device(url.storefront)
         global_auth_param = GlobalAuthParams.from_auth_params_and_token(available_device.get_auth_params(),
                                                                         self.anonymous_access_token)
         match url.type:
@@ -97,6 +99,50 @@ class NewInteractiveShell:
             case URLType.Album:
                 task = self.loop.create_task(rip_album(url, global_auth_param, codec, self.config, available_device))
 
+    async def do_m3u8(self, m3u8_url: str, codec: str, force_download: bool):
+        song_id = get_song_id_from_m3u8(m3u8_url)
+        song = Song(id=song_id, storefront=self.config.region.defaultStorefront, url="", type=URLType.Song)
+        available_device = await self._get_available_device(self.config.region.defaultStorefront)
+        global_auth_param = GlobalAuthParams.from_auth_params_and_token(available_device.get_auth_params(),
+                                                                        self.anonymous_access_token)
+        self.loop.create_task(
+            rip_song(song, global_auth_param, codec, self.config, available_device, force_save=force_download,
+                     specified_m3u8=m3u8_url)
+        )
+
+    async def do_mitm(self, codec: str, force_download: bool):
+        available_device = await self._get_available_device(self.config.region.defaultStorefront)
+        global_auth_param = GlobalAuthParams.from_auth_params_and_token(available_device.get_auth_params(),
+                                                                        self.anonymous_access_token)
+        m3u8_urls = set()
+        tasks = set()
+
+        def callback(m3u8_url):
+            if m3u8_url in m3u8_urls:
+                return
+            song_id = get_song_id_from_m3u8(m3u8_url)
+            song = Song(id=song_id, storefront=self.config.region.defaultStorefront, url="", type=URLType.Song)
+            task = self.loop.create_task(
+                rip_song(song, global_auth_param, codec, self.config, available_device, force_save=force_download,
+                         specified_m3u8=m3u8_url)
+            )
+            tasks.update(task)
+            task.add_done_callback(tasks.remove)
+            m3u8_urls.update(m3u8_url)
+
+        self.loop.create_task(start_proxy(self.config.mitm.host, self.config.mitm.port, callback))
+
+    async def _get_available_device(self, storefront: str):
+        devices = self.storefront_device_mapping.get(storefront)
+        if not devices:
+            logger.error(f"No device is available to decrypt the specified region: {storefront}")
+        available_devices = [device for device in devices if not device.decryptLock.locked()]
+        if not available_devices:
+            available_device: Device = random.choice(devices)
+        else:
+            available_device: Device = random.choice(available_devices)
+        return available_device
+
     async def handle_command(self):
         session = PromptSession("> ")
 

src/config.py

@@ -31,11 +31,17 @@ class Metadata(BaseModel):
     embedMetadata: list[str]
 
 
+class Mitm(BaseModel):
+    host: str
+    port: int
+
+
 class Config(BaseModel):
-    language: Language
+    region: Region
     devices: list[Device]
     download: Download
     metadata: Metadata
+    mitm: Mitm
 
     @classmethod
     def load_from_config(cls, config_file: str = "config.toml"):

src/mitm.py

@@ -0,0 +1,33 @@
+import plistlib
+
+import mitmproxy.http
+from mitmproxy import options
+from mitmproxy.tools import dump
+from loguru import logger
+
+
+class RequestHandler:
+    def __init__(self, callback):
+        self.callback = callback
+
+    def response(self, flow: mitmproxy.http.HTTPFlow):
+        if flow.request.host == "play.itunes.apple.com" and flow.request.path == "/WebObjects/MZPlay.woa/wa/subPlaybackDispatch":
+            data = plistlib.loads(flow.response.content)
+            m3u8 = data["songList"][0]["hls-playlist-url"]
+            flow.response.status_code = 500
+            self.callback(m3u8)
+
+
+async def start_proxy(host, port, callback):
+    opts = options.Options(listen_host=host, listen_port=port, mode=["socks5"])
+
+    master = dump.DumpMaster(
+        opts,
+        with_termlog=False,
+        with_dumper=False,
+    )
+    master.addons.add(RequestHandler(callback))
+
+    logger.info(f"Mitmproxy started at socks5://{host}:{port}")
+
+    await master.run()