Version 0.2.9

abcache: Add `--dump-master-data` option

.vscode/launch.json

@@ -23,12 +23,30 @@
                 "DEBUG",
                 "abcache", 
                 "--app-version",
-                "3.6.5",
+                "3.8.0",
                 "--app-appHash",
-                "c5e69328-dcfd-fa80-8ec2-b478ed0719d0"
+                "6a32c1dd-bc3e-69f0-c66e-8dd33e3b1ba2"
             ],
             "justMyCode": true
         },
+        {
+            "name": "Python: AbCache (dump Master data)",
+            "type": "python",
+            "request": "launch",
+            "module": "sssekai",
+            "args": [
+                "--log-level",
+                "DEBUG",
+                "abcache", 
+                "--app-version",
+                "3.8.0",
+                "--app-appHash",
+                "6a32c1dd-bc3e-69f0-c66e-8dd33e3b1ba2",
+                "--dump-master-data",
+                "D:\\masterdata"
+            ],
+            "justMyCode": true
+        },        
         {
             "name": "Python: AbCache (no update)",
             "type": "python",

sssekai/__init__.py

@@ -1,5 +1,5 @@
 __VERSION_MAJOR__ = 0
 __VERSION_MINOR__ = 2
-__VERSION_PATCH__ = 8
+__VERSION_PATCH__ = 9
 
 __version__ = '%s.%s.%s' % (__VERSION_MAJOR__,__VERSION_MINOR__,__VERSION_PATCH__)

sssekai/__main__.py

@@ -63,6 +63,8 @@ def write(__s):
     group.add_argument('--download-dir', type=str, help='asset bundle download directory. leave empty if you don\'t want to download anything',default='')
     group.add_argument('--download-ensure-deps',action='store_true',help='ensure dependencies (of the downloaded ones) are downloaded as well')
     group.add_argument('--download-workers', type=int, help='number of download workers (default: %(default)s)',default=4)
+    group = abcache_parser.add_argument_group('extra options', 'NOTE: when *any* of these options are specified, the cache database *won\'t* be updated, and no download will be performed either.')
+    group.add_argument('--dump-master-data', type=str, help='directory to store the dumped master data in JSON', default=None)
     abcache_parser.set_defaults(func=main_abcache)
     # live2dextract
     live2dextract_parser = subparsers.add_parser('live2dextract', help='''Extract Sekai Live2D Models in a AssetBundle''')

sssekai/abcache/__init__.py

@@ -17,7 +17,7 @@ def fromdict(klass : type, d : Union[Mapping, List]):
         return {k:fromdict(klass.__args__[1],v) for k,v in d.items()}
     return d
 
-from requests import Session
+from requests import Session, Response
 from msgpack import unpackb, packb
 
 from sssekai import __version__
@@ -147,6 +147,12 @@ def SEKAI_API_USER(self): return self.SEKAI_API_ENDPOINT + '/api/user'
     @property
     def SEKAI_API_USER_AUTH(self): return f'{self.SEKAI_API_USER}/{self.database.sekai_user_data.userRegistration.userId}/auth?refreshUpdatedResources=False'
     @property
+    def SEKAI_API_USER_SUITE(self): return f'{self.SEKAI_API_ENDPOINT}/api/suite/user/{self.database.sekai_user_data.userRegistration.userId}'
+    @property
+    def SEKAI_API_MASTER_SUITE(self): return f'{self.SEKAI_API_ENDPOINT}/api/suite/master'
+    @property
+    def SEKAI_API_INFORMATION(self): return self.SEKAI_API_ENDPOINT + '/api/information'
+    @property
     def SEKAI_API_GAMEVERSION_ENDPOINT(self): return 'https://game-version.sekai.colorfulpalette.org'
     @property
     def SEKAI_AB_INFO_ENDPOINT(self): return f'https://production-{self.SEKAI_AB_HOST_HASH}-assetbundle-info.sekai.colorfulpalette.org/'
@@ -164,68 +170,84 @@ def SEKAI_ISSUE_SIGNATURE_ENDPOINT(self): return self.SEKAI_ISSUE_ENDPOINT + '/a
     @property
     def abcache_index(self): return self.database.sekai_abcache_index
 
-    def _update_signatures(self):        
-        logger.info('Updating signatures')
-        resp = self.post(self.SEKAI_ISSUE_SIGNATURE_ENDPOINT)
+    def request_packed(self, method : str, url : str, data : dict = None, **kwargs):
+        '''Send a request with packed data. Data will be packed and encrypted before sending.
+
+        Args:
+            method (str): HTTP method
+            url (str): URL
+            data (dict, optional): Payload data. Defaults to None.
+
+        Returns:
+            Response: Response object
+        '''
+        if data is not None:
+            data = packb(data)
+            data = encrypt(data)
+        resp = self.request(method=method, url=url, data=data, **kwargs)
         resp.raise_for_status()
+        return resp
+
+    @staticmethod
+    def response_to_dict(resp : Response):
+        '''Decrypt and unpack a response content to a dictionary.
+
+        Args:
+            resp (Response): Response object
+
+        Returns:
+            dict: Decrypted and unpacked data
+        '''
+        data = decrypt(resp.content)
+        data = unpackb(data)
+        return data
+
+    def _update_signatures(self):        
+        logger.debug('Updating signatures')
+        resp = self.request_packed('POST', self.SEKAI_ISSUE_SIGNATURE_ENDPOINT)        
         self.headers['Cookie'] = resp.headers['Set-Cookie'] 
         # HACK: Per RFC6265, Cookies should not be visible to subdomains since it's not set with Domain attribute (https://github.com/psf/requests/issues/2576)
         # But the other endpoints uses it nontheless. So we have to set it manually.
 
     def _update_user_data(self):
-        logger.info('Updating user data')
+        logger.debug('Updating user data')
         payload = {
             "platform": self.headers['X-Platform'],
             "deviceModel": self.headers['X-DeviceModel'],
             "operatingSystem": self.headers['X-OperatingSystem'],
-        }
-        payload = packb(payload)
-        payload = encrypt(payload)
-        resp = self.post(self.SEKAI_API_USER, data=payload)
-        resp.raise_for_status()
-        data = decrypt(resp.content)
-        data = unpackb(data)
+        }        
+        resp = self.request_packed('POST', self.SEKAI_API_USER, data=payload)
+        data = self.response_to_dict(resp)
         self.database.sekai_user_data = fromdict(SekaiUserData,data)        
 
     def _update_user_auth_data(self):
-        logger.info('Updating user auth data')
+        logger.debug('Updating user auth data')
         payload = {
             "credential": self.database.sekai_user_data.credential,
             "deviceId" : None
-        }
-        payload = packb(payload)
-        payload = encrypt(payload)        
-        resp = self.put(self.SEKAI_API_USER_AUTH, data=payload)
-        resp.raise_for_status()
-        data = decrypt(resp.content)
-        data = unpackb(data)        
+        }    
+        resp = self.request_packed('PUT', self.SEKAI_API_USER_AUTH, data=payload)
+        data = self.response_to_dict(resp)      
         self.database.sekai_user_auth_data = fromdict(SekaiUserAuthData,data)
 
     def _update_system_data(self):
-        logger.info('Updating system data')
-        resp = self.get(self.SEKAI_API_SYSTEM_DATA)
-        resp.raise_for_status()
-        data = decrypt(resp.content)
-        data = unpackb(data)
+        logger.debug('Updating system data')
+        resp = self.request_packed('GET', self.SEKAI_API_SYSTEM_DATA)
+        data = self.response_to_dict(resp)
         self.database.sekai_system_data = fromdict(SekaiSystemData,data)
 
     def _update_gameversion_data(self):
-        logger.info('Updating game version data')
-        resp = self.get(self.SEKAI_API_GAMEVERSION_ENDPOINT + '/' + self.SEKAI_APP_VERSION + '/' + self.SEKAI_APP_HASH)
-        resp.raise_for_status()
-        data = decrypt(resp.content)
-        data = unpackb(data)
+        logger.debug('Updating game version data')
+        resp = self.request_packed('GET', self.SEKAI_API_GAMEVERSION_ENDPOINT + '/' + self.SEKAI_APP_VERSION + '/' + self.SEKAI_APP_HASH)
+        data = self.response_to_dict(resp)
         self.database.sekai_gameversion_data = fromdict(SekaiGameVersionData,data)
 
     def _update_abcache_index(self) -> AbCacheIndex:
-        logger.info('Updating Assetbundle index')
-        resp = self.get(
-            url=self.SEKAI_AB_INFO_ENDPOINT + self.SEKAI_AB_INDEX_PATH
-        )
-        resp.raise_for_status()
-        data = decrypt(resp.content)
-        data = unpackb(data)
-        self.database.sekai_abcache_index = fromdict(AbCacheIndex,data)   
+        logger.debug('Updating Assetbundle index')
+        resp = self.request_packed('GET', self.SEKAI_AB_INFO_ENDPOINT + self.SEKAI_AB_INDEX_PATH)
+        data = self.response_to_dict(resp)
+        self.database.sekai_abcache_index = fromdict(AbCacheIndex,data)
+        return self.database.sekai_abcache_index   
 
     def __init__(self, config : AbCacheConfig) -> None:
         super().__init__()
@@ -245,39 +267,57 @@ def __init__(self, config : AbCacheConfig) -> None:
             'X-App-Hash': self.SEKAI_APP_HASH
         })
 
-    def update(self):
-        logger.info('Updating metadata')        
-        logger.info('Set App version: %s (%s), hash=%s' % (self.config.app_version,self.config.app_platform, self.SEKAI_APP_HASH))
+    def update_download_headers(self):
+        '''Update headers for downloading assetbundles *ONLY*. Functionalities related to user-level data (e.g. Master Data, etc) won't
+        be available. 
+
+        Returns:
+            dict: Updated headers
+        '''
+        self._update_signatures()
+        return self.headers
+
+    def update_client_headers(self):
+        '''Update headers for client-level functionalities. This includes user-level data (e.g. Master Data, etc) and assetbundle data.
+
+        Returns:
+            _type_: _description_
+        '''
+        logger.debug('Updating metadata')        
+        logger.debug('Set App version: %s (%s), hash=%s' % (self.config.app_version,self.config.app_platform, self.SEKAI_APP_HASH))
         self._update_signatures()
         self._update_system_data()
         version_newest = self.database.sekai_system_data.appVersions[-1]
-        logger.info('Newest App version: %s' % version_newest)
+        logger.debug('Newest App version: %s' % version_newest)
         if version_newest.appVersion != self.SEKAI_APP_VERSION:
             logger.warning('App version mismatch. This may cause issues.')            
         self._update_gameversion_data()               
         self._update_user_data()
         self._update_user_auth_data()
+        self.headers.update({
+            'X-Data-Version': self.database.sekai_user_auth_data.dataVersion,
+            'X-Asset-Version': self.database.sekai_user_auth_data.assetVersion,
+            'X-Session-Token': self.database.sekai_user_auth_data.sessionToken
+        })
+        return self.headers
+
+    def update(self):        
+        self.update_client_headers()
         self._update_abcache_index()
         logger.debug('Sekai AssetBundle version: %s' % self.SEKAI_ASSET_VERSION)
         logger.debug('Sekai AssetBundle host hash: %s' % self.SEKAI_AB_HOST_HASH)
 
-
     def save(self, f : BinaryIO):
-        logger.info('Saving cache')
+        logger.debug('Saving cache')
         dump(self.database,f)
 
     def load(self, f : BinaryIO):
-        logger.info('Loading cache')
+        logger.debug('Loading cache')
         self.database = load(f)
 
     def __repr__(self) -> str:
         return f'<AbCache config={self.config} bundles={len(self.abcache_index.bundles)}>'
 
-    def update_download_headers(self):
-        '''Update headers with the latest user auth data. Try to call this before downloading anything.'''
-        self._update_signatures()
-        return self.headers
-
     def get_entry_by_bundle_name(self, bundleName : str) -> AbCacheEntry:
         return self.abcache_index.bundles.get(bundleName, None)
 

sssekai/entrypoint/abcache.py

@@ -1,5 +1,4 @@
-import os, re
-from types import TracebackType
+import os, re, json
 from sssekai.abcache import AbCache, AbCacheConfig, logger
 from sssekai.crypto.AssetBundle import SEKAI_AB_MAGIC, decrypt_headaer_inplace
 from concurrent.futures import ThreadPoolExecutor
@@ -60,6 +59,19 @@ def add_link(self, url, fname, length):
 
 def main_abcache(args):    
     cache = AbCache(AbCacheConfig(args.app_version, args.app_platform, args.app_appHash))
+    if args.dump_master_data:
+        logger.info('Dumping master data to %s', args.dump_master_data)
+        cache.update_client_headers()
+        for split in tqdm(cache.database.sekai_user_auth_data.suiteMasterSplitPath, desc='Pulling'):
+            resp = cache.request_packed('GET', cache.SEKAI_API_ENDPOINT + '/api/' + split)
+            data = cache.response_to_dict(resp)  
+            path = os.path.join(args.dump_master_data, split + '.json')
+            os.makedirs(os.path.dirname(path), exist_ok=True)
+            logger.debug('Saving to %s', path)
+            with open(path, 'w', encoding='utf-8') as f:
+                json.dump(data, f, indent=4, ensure_ascii=False)
+            return  
+
     db_path = os.path.expanduser(args.db)
     os.makedirs(os.path.dirname(db_path), exist_ok=True)
     if not args.no_update: