Support signed chunk download URLs

legendary/api/egs.py

@@ -22,7 +22,9 @@ class EPCAPI:
     _label = 'Live-EternalKnight'
 
     _oauth_host = 'account-public-service-prod03.ol.epicgames.com'
+    # FIXME: Check whether everything using this can also use `_launcher_host_2`, then replace this with it
     _launcher_host = 'launcher-public-service-prod06.ol.epicgames.com'
+    _launcher_host_2 = 'launcher-public-service-prod.ak.epicgames.com'
     _entitlements_host = 'entitlement-public-service-prod08.ol.epicgames.com'
     _eulatracking_host = 'eulatracking-public-service-prod06.ol.epicgames.com'
     _catalog_host = 'catalog-public-service-prod06.ol.epicgames.com'
@@ -34,6 +36,7 @@ class EPCAPI:
     # _store_gql_host = 'launcher.store.epicgames.com'
     _store_gql_host = 'graphql.epicgames.com'
     _artifact_service_host = 'artifact-public-service-prod.beee.live.use1a.on.epicgames.com'
+    _artifact_delivery_service_host = 'artifact-delivery-service-public-prod.ol.epicgames.com'
 
     def __init__(self, lc='en', cc='US', timeout=10.0):
         self.log = logging.getLogger('EPCAPI')
@@ -233,6 +236,23 @@ def get_game_manifest_by_ticket(self, artifact_id: str, signed_ticket: str, labe
         r.raise_for_status()
         return r.json()
 
+    def get_download_ticket(self, catalog_item_id: str, build_version: str, app_name: str,
+                            namespace: str, label='Live', platform='Windows'):
+        r = self.session.post(f'https://{self._launcher_host_2}/launcher/api/public/assets/v2/ticket',
+                              json=dict(catalogItemId=catalog_item_id, buildVersion=build_version, appName=app_name,
+                                        namespace=namespace, label=label, platform=platform),
+                              timeout=self.request_timeout)
+        r.raise_for_status()
+        return r.json()
+
+    def get_signed_chunk_urls(self, ticket: str, paths: list[str]) -> dict[str, str]:
+        user_id = self.user.get('account_id')
+        r = self.session.post(f'https://{self._artifact_delivery_service_host}/artifact-delivery/api/public/v1/'
+                              f'delivery/account/{user_id}/downloadurls',
+                              json=dict(signedTicket=ticket, chunkIds=paths, deltaId='', clientMetrics={}))
+        r.raise_for_status()
+        return r.json()
+
     def get_library_items(self, include_metadata=True):
         records = []
         r = self.session.get(f'https://{self._library_host}/library/api/public/items',

legendary/cli.py

@@ -5,9 +5,11 @@
 import csv
 import json
 import logging
+import multiprocessing
 import os
 import shlex
 import subprocess
+import threading
 import time
 import webbrowser
 import re
@@ -366,7 +368,7 @@ def list_files(self, args):
             if not game:
                 logger.fatal(f'Could not fetch metadata for "{args.app_name}" (check spelling/account ownership)')
                 exit(1)
-            manifest_data, _ = self.core.get_cdn_manifest(game, platform=args.platform)
+            manifest_data, _, _ = self.core.get_cdn_manifest(game, platform=args.platform)
 
         manifest = self.core.load_manifest(manifest_data)
         files = sorted(manifest.file_manifest_list.elements,
@@ -1012,7 +1014,8 @@ def install_game(self, args):
                                                           override_delta_manifest=args.override_delta_manifest,
                                                           preferred_cdn=args.preferred_cdn,
                                                           disable_https=args.disable_https,
-                                                          bind_ip=args.bind_ip)
+                                                          bind_ip=args.bind_ip,
+                                                          always_use_signed_urls=args.always_use_signed_urls)
 
         # game is either up-to-date or hasn't changed, so we have nothing to do
         if not analysis.dl_size:
@@ -1081,14 +1084,45 @@ def install_game(self, args):
                 print('Aborting...')
                 exit(0)
 
+        ticket_a, ticket_b = multiprocessing.Pipe()
+        sign_a, sign_b = multiprocessing.Pipe()
+
+        def ticket_creator_thread():
+            t = threading.current_thread()
+            while not getattr(t, 'stop', False):
+                if ticket_b.poll(1):
+                    catalog_item_id, build_version, app_name, namespace, label, platform = ticket_b.recv()
+                    ticket_b.send(self.core.egs.get_download_ticket(catalog_item_id, build_version, app_name,
+                                                                    namespace, label, platform))
+
+        def chunk_url_sign_thread():
+            t = threading.current_thread()
+            while not getattr(t, 'stop', False):
+                if sign_b.poll(1):
+                    ticket, chunk_paths = sign_b.recv()
+                    signed_chunk_urls = self.core.egs.get_signed_chunk_urls(ticket, chunk_paths)
+                    if args.disable_https:
+                        for key in signed_chunk_urls:
+                            signed_chunk_urls[key] = signed_chunk_urls[key].replace('https://', 'http://')
+                    sign_b.send(signed_chunk_urls)
+
+
+        ticket_thread = threading.Thread(target=ticket_creator_thread)
+        sign_thread = threading.Thread(target=chunk_url_sign_thread)
+
         start_t = time.time()
 
         try:
             # set up logging stuff (should be moved somewhere else later)
             dlm.logging_queue = self.logging_queue
             dlm.proc_debug = args.dlm_debug
+            dlm.ticket_pipe = ticket_a
+            dlm.sign_pipe = sign_a
 
+            ticket_thread.start()
+            sign_thread.start()
             dlm.start()
+
             dlm.join()
         except Exception as e:
             end_t = time.time()
@@ -1158,6 +1192,11 @@ def install_game(self, args):
                 self.core.install_game(old_igame)
 
             logger.info(f'Finished installation process in {end_t - start_t:.02f} seconds.')
+        finally:
+            ticket_thread.stop = True
+            sign_thread.stop = True
+            ticket_thread.join()
+            sign_thread.join()
 
     def _handle_postinstall(self, postinstall, igame, skip_prereqs=False):
         print('\nThis game lists the following prerequisites to be installed:')
@@ -1265,7 +1304,7 @@ def verify_game(self, args, print_command=True, repair_mode=False, repair_online
 
                 logger.warning('No manifest could be loaded, the file may be missing. Downloading the latest manifest.')
                 game = self.core.get_game(args.app_name, platform=igame.platform)
-                manifest_data, _ = self.core.get_cdn_manifest(game, igame.platform)
+                manifest_data, _, _ = self.core.get_cdn_manifest(game, igame.platform)
             else:
                 logger.critical(f'Manifest appears to be missing! To repair, run "legendary repair '
                                 f'{args.app_name} --repair-and-update", this will however redownload all files '
@@ -1677,6 +1716,7 @@ def info(self, args):
 
         manifest_data = None
         entitlements = None
+        use_signed_url = None
         # load installed manifest or URI
         if args.offline or manifest_uri:
             if app_name and self.core.is_installed(app_name):
@@ -1696,7 +1736,7 @@ def info(self, args):
             game.metadata = egl_meta
             # Get manifest if asset exists for current platform
             if args.platform in game.asset_infos:
-                manifest_data, _ = self.core.get_cdn_manifest(game, args.platform)
+                manifest_data, _, use_signed_url = self.core.get_cdn_manifest(game, args.platform)
 
         if game:
             game_infos = info_items['game']
@@ -1926,6 +1966,11 @@ def info(self, args):
             manifest_info.append(InfoItem('Download size by install tag', 'tag_download_size',
                                           tag_download_size_human or 'N/A', tag_download_size))
 
+        if use_signed_url is not None:
+            info_items["manifest"].append(
+                InfoItem('Uses signed chunk URLs', 'use_signed_urls', use_signed_url, use_signed_url)
+            )
+
         if not args.json:
             def print_info_item(item: InfoItem):
                 if item.value is None:
@@ -2911,6 +2956,8 @@ def main():
                                 help='Do not ask about installing DLCs.')
     install_parser.add_argument('--bind', dest='bind_ip', action='store', metavar='<IPs>', type=str,
                                 help='Comma-separated list of IPs to bind to for downloading')
+    install_parser.add_argument('--always-use-signed-urls', dest='always_use_signed_urls', action='store_true',
+                                help='Always use signed chunk URLs, even if the Epic API indicates not to')
 
     uninstall_parser.add_argument('--keep-files', dest='keep_files', action='store_true',
                                   help='Keep files but remove game from Legendary database')

legendary/core.py

@@ -1262,6 +1262,7 @@ def get_cdn_urls(self, game, platform='Windows'):
             raise ValueError('Manifest response has more than one element!')
 
         manifest_hash = m_api_r['elements'][0]['hash']
+        manifest_use_signed_url: bool = m_api_r['elements'][0]['useSignedUrl']
         base_urls = []
         manifest_urls = []
         for manifest in m_api_r['elements'][0]['manifests']:
@@ -1275,10 +1276,10 @@ def get_cdn_urls(self, game, platform='Windows'):
             else:
                 manifest_urls.append(manifest['uri'])
 
-        return manifest_urls, base_urls, manifest_hash
+        return manifest_urls, base_urls, manifest_hash, manifest_use_signed_url
 
     def get_cdn_manifest(self, game, platform='Windows', disable_https=False):
-        manifest_urls, base_urls, manifest_hash = self.get_cdn_urls(game, platform)
+        manifest_urls, base_urls, manifest_hash, use_signed_url = self.get_cdn_urls(game, platform)
         if not manifest_urls:
             raise ValueError('No manifest URLs returned by API')
 
@@ -1306,7 +1307,7 @@ def get_cdn_manifest(self, game, platform='Windows', disable_https=False):
         if sha1(manifest_bytes).hexdigest() != manifest_hash:
             raise ValueError('Manifest sha hash mismatch!')
 
-        return manifest_bytes, base_urls
+        return manifest_bytes, base_urls, use_signed_url
 
     def get_uri_manifest(self, uri):
         if uri.startswith('http'):
@@ -1341,7 +1342,8 @@ def prepare_download(self, game: Game, base_game: Game = None, base_path: str =
                          repair: bool = False, repair_use_latest: bool = False,
                          disable_delta: bool = False, override_delta_manifest: str = '',
                          egl_guid: str = '', preferred_cdn: str = None,
-                         disable_https: bool = False, bind_ip: str = None) -> (DLManager, AnalysisResult, ManifestMeta):
+                         disable_https: bool = False, bind_ip: str = None, always_use_signed_urls: bool = False
+                         ) -> tuple[DLManager, AnalysisResult, InstalledGame]:
         # load old manifest
         old_manifest = None
 
@@ -1369,11 +1371,13 @@ def prepare_download(self, game: Game, base_game: Game = None, base_path: str =
         if override_manifest:
             self.log.info(f'Overriding manifest with "{override_manifest}"')
             new_manifest_data, _base_urls = self.get_uri_manifest(override_manifest)
+            # FIXME: Populate `use_signed_urls`
+            use_signed_urls = False
             # if override manifest has a base URL use that instead
             if _base_urls:
                 base_urls = _base_urls
         else:
-            new_manifest_data, base_urls = self.get_cdn_manifest(game, platform, disable_https=disable_https)
+            new_manifest_data, base_urls, use_signed_urls = self.get_cdn_manifest(game, platform, disable_https=disable_https)
             # overwrite base urls in metadata with current ones to avoid using old/dead CDNs
             game.base_urls = base_urls
             # save base urls to game metadata
@@ -1509,7 +1513,12 @@ def prepare_download(self, game: Game, base_game: Game = None, base_path: str =
         if not max_workers:
             max_workers = self.lgd.config.getint('Legendary', 'max_workers', fallback=0)
 
-        dlm = DLManager(install_path, base_url, resume_file=resume_file, status_q=status_q,
+        if always_use_signed_urls:
+            use_signed_urls = True
+
+        asset = self.get_asset(game.app_name, platform)
+        dlm = DLManager(install_path, base_url, use_signed_urls, asset,
+                        resume_file=resume_file, status_q=status_q,
                         max_shared_memory=max_shm * 1024 * 1024, max_workers=max_workers,
                         dl_timeout=dl_timeout, bind_ip=bind_ip)
 
@@ -1528,7 +1537,8 @@ def prepare_download(self, game: Game, base_game: Game = None, base_path: str =
             if read_files:
                 raise
             self.log.warning('Memory error encountered, retrying with file read enabled...')
-            dlm = DLManager(install_path, base_url, resume_file=resume_file, status_q=status_q,
+            dlm = DLManager(install_path, base_url, use_signed_urls, asset,
+                        resume_file=resume_file, status_q=status_q,
                         max_shared_memory=max_shm * 1024 * 1024, max_workers=max_workers,
                         dl_timeout=dl_timeout, bind_ip=bind_ip)
             anlres = dlm.run_analysis(manifest=new_manifest, **analysis_kwargs, read_files=True)
@@ -1566,7 +1576,7 @@ def prepare_download(self, game: Game, base_game: Game = None, base_path: str =
                               can_run_offline=offline == 'true', requires_ot=ot == 'true',
                               is_dlc=base_game is not None, install_size=anlres.install_size,
                               egl_guid=egl_guid, install_tags=file_install_tag,
-                              platform=platform, uninstaller=uninstaller)
+                              platform=platform, uninstaller=uninstaller, use_signed_url=use_signed_urls)
 
         return dlm, anlres, igame
 
@@ -1781,9 +1791,11 @@ def import_game(self, game: Game, app_path: str, egl_guid='', platform='Windows'
                 if not needs_verify:
                     self.log.debug(f'No in-progress installation found, assuming complete...')
 
+        # FIXME: Populate `use_signed_url`
+        use_signed_url = False
         if not manifest_data:
             self.log.info(f'Downloading latest manifest for "{game.app_name}"')
-            manifest_data, base_urls = self.get_cdn_manifest(game)
+            manifest_data, base_urls, use_signed_url = self.get_cdn_manifest(game)
             if not game.base_urls:
                 game.base_urls = base_urls
                 self.lgd.set_game_meta(game.app_name, game)
@@ -1809,7 +1821,7 @@ def import_game(self, game: Game, app_path: str, egl_guid='', platform='Windows'
                               executable=new_manifest.meta.launch_exe, can_run_offline=offline == 'true',
                               launch_parameters=new_manifest.meta.launch_command, requires_ot=ot == 'true',
                               needs_verification=needs_verify, install_size=install_size, egl_guid=egl_guid,
-                              platform=platform)
+                              platform=platform, use_signed_url=use_signed_url)
 
         return new_manifest, igame
 
@@ -2060,15 +2072,18 @@ def prepare_overlay_install(self, path=None):
         if not self.logged_in:
             self.egs.start_session(client_credentials=True)
 
-        _manifest, base_urls = self.get_cdn_manifest(EOSOverlayApp)
+        _manifest, base_urls, use_signed_urls = self.get_cdn_manifest(EOSOverlayApp)
         manifest = self.load_manifest(_manifest)
 
         if igame := self.lgd.get_overlay_install_info():
             path = igame.install_path
         else:
             path = path or os.path.join(self.get_default_install_dir(), '.overlay')
 
-        dlm = DLManager(path, base_urls[0])
+        if use_signed_urls:
+            raise ValueError('EOS Overlay requiring signed URLs, not sure what to do here')
+
+        dlm = DLManager(path, base_urls[0], use_signed_urls, GameAsset())
         analysis_result = dlm.run_analysis(manifest=manifest)
 
         install_size = analysis_result.install_size
@@ -2117,7 +2132,7 @@ def prepare_bottle_download(self, bottle_name, manifest_url, base_url=None):
         if os.path.exists(path):
             raise FileExistsError(f'Bottle {bottle_name} already exists')
 
-        dlm = DLManager(path, base_url)
+        dlm = DLManager(path, base_url, False, GameAsset())
         analysis_result = dlm.run_analysis(manifest=manifest)
 
         install_size = analysis_result.install_size