Updated to new oauth format. Removed pypasser dependency.

Reviewed-on: #1

README.md

@@ -2,8 +2,10 @@
 Efficient KF3/DMM game launcher.
 
 ## Requirements
-Python 3
+- Python 3 and pip.
-KF3 installed from DMM. DMM itself does not need to be running.
+- DMM account that owns KF3
+
+KF3 installed from DMM is recommended but not required.
 
 ## Instructions
 Download the files from this repo. Unzip them to a safe place.
@@ -16,5 +18,4 @@ Input your DMM email and then password when prompted, and `yes` if you want to u
 
 If you make a mistake or you want to change login details, edit the `kfp2g.cfg` config file, or delete it to run the configuration again.
 
-## Known issues
+If KF3 needs to update, use `update_and_run_KF3.bat` and it will install any new updates.
-Retrieving launch arguments - error 308: Every once in a while, DMM games may have updated terms of service that need to be accepted. This can only be done via the official DMM app.

dmmBypass.py

@@ -2,40 +2,57 @@ import os
 import sys
 import json
 import hashlib
+import pathlib
 import argparse
-import requests
 import subprocess
 import urllib.parse
 from uuid import getnode
+from datetime import datetime, timedelta
+
+# Set to False to allow packet sniffing/capture.
+# Set separately for dmmUpdater
+SSL_VERIFY = True
 
 parser = argparse.ArgumentParser(description='DMM bypass script')
 parser.add_argument('-g', '--game', type=str, help="DMM code name of the game", default="kfp2g")
+parser.add_argument('-t', '--type', help="DMM game type (ACL/GCL)", default="GCL")
 parser.add_argument('-u', '--update', help="Check for game update before launching", action='store_true')
 args = parser.parse_args()
 
-config_name = args.game + '.cfg'
+def load_config(game_id):
+    config_name = game_id + ".cfg"
+    if not os.path.exists(config_name):
+        return None
+    else:
+        with open(config_name, "rt", encoding="utf-8") as f:
+            return json.load(f)
 
-if not os.path.exists(config_name):
+def save_config(game_id, config):
-    subprocess.check_call([sys.executable, "-m", "pip", "install", "requests", "beautifulsoup4", "PyPasser"])
+    with open(game_id + '.cfg', 'wt', encoding="utf-8") as f:
+        json.dump(config, f, indent=1, ensure_ascii=False)
+
+config = load_config(args.game)
+
+if config is None:
+    subprocess.check_call([sys.executable, "-m", "pip", "install", "requests", "beautifulsoup4"])
     config = {
         "game_id" : args.game,
         "file_path" : input('Enter full file path to KF3 .exe: ').strip('\"'),
         "dmm_login" : input('DMM Login (email): '),
         "dmm_password" : input('DMM Password: '),
-        "use_proxy" : input('Your login tokens will be sent through my VPN machine.\nIs that okay? (yes/no): ').lower() == "yes"
+        "use_proxy" : input('Your login data will be sent through my VPN.\nIs that okay? (yes/no): ').lower() == "yes"
     }
-    with open(config_name, 'wt', encoding="utf-8") as f:
+
-        json.dump(config, f, indent=1, ensure_ascii=False)
+    save_config(args.game, config)
 else:
-    with open(config_name, "rt", encoding="utf-8") as f:
+    print(f'Loaded settings from {args.game}.cfg')
-        config = json.load(f)
-    print(f'Loaded settings from {config_name}')
 
 config["update_game"] = args.update
+config["game_type"] = args.type
 
+import requests
 import dmmUpdater
 from bs4 import BeautifulSoup
-from pypasser import reCaptchaV3
 
 def get_hash(data):
     sha_obj = hashlib.sha256()
@@ -47,122 +64,191 @@ def get_mac():
     mac = getnode()
     return ':'.join(("%012X" % mac)[i:i+2] for i in range(0, 12, 2)).lower()
 
-def retrieve_login_token(session : requests.Session):
+def get_game_root_path(user_path, game_dir, exe_name):
+    """Return valid game install directory"""
+    path = pathlib.Path(user_path)
+    parts = list(path.parts)
+    if path.name == exe_name:
+        return str(path.parent.resolve())
+    elif game_dir in parts:
+        #get index of last occurence of game_dir in user path (+1)
+        dir_idx = len(parts) - parts[::-1].index(game_dir)
+        parts = parts[:dir_idx]
+        return str(pathlib.Path(*parts).resolve())
+    else:
+        raise Exception(f"Path to game files is incorrect! Ensure it points to {exe_name} or contains {game_dir}!")
+
+def retrieve_oauth_code(login, password, session : requests.Session):
     try:
-        print("Retrieving login form")
+        print("Retrieving login url")
-        url = "https://accounts.dmm.com/service/login/password"
+        url = "https://apidgp-gameplayer.games.dmm.com/v5/auth/login/url"
-        result = session.get(url)
+        data = {"prompt":"choose"}
+        result = session.post(url, json=data, verify=SSL_VERIFY)
+        result.raise_for_status()
+        url = result.json()["data"]["url"]
+        encoded_url = urllib.parse.quote_plus(url.replace("https://accounts.dmm.com/service/oauth/select/=/path=", "").replace("oauth/select/", "oauth/").replace("accounts?prompt=choose", "accounts"))
+
+        print("Retrieving login form token")
+        result = session.get(url, verify=SSL_VERIFY)
         result.raise_for_status()
         page = BeautifulSoup(result.content, 'html.parser')
         token = page.find('input', attrs={"name":"token"}).get("value")
-        return token
+
+        print("Retrieving oauth link")
+        url = "https://accounts.dmm.com/service/oauth/authenticate"
+        headers = {"Content-Type": "application/x-www-form-urlencoded",
+                   "check_done_login":"true"}
+        data = f"token={token}&login_id={login}&password={password}&use_auto_login=1&path={encoded_url}&recaptchaToken="
+        result = session.post(url, data, headers=headers, verify=SSL_VERIFY)
+        result.raise_for_status()
+        page = BeautifulSoup(result.content, 'html.parser')
+        final_url = page.find('input', attrs={"id":"ga-param-service-url"}).get("value")
+
+        print("Retrieving oauth token")
+        result = session.get(final_url, allow_redirects=False, verify=SSL_VERIFY)
+        redirect_url = result.next.url
+        oauth_token = redirect_url.split("=")[-1]
+
+        return oauth_token
+    
     except Exception as e:
         print("Failed to retrieve login form:", e)
 
-def retrieve_captcha_token():
+def retrieve_access_token(oauth_token, session: requests.Session):
-    try:
+    url = "https://apidgp-gameplayer.games.dmm.com/v5/auth/accesstoken/issue"
-        captcha = reCaptchaV3("https://www.google.com/recaptcha/enterprise/anchor?ar=1&k=6LfZLQEVAAAAAC-8pKwFNuzVoJW4tfUCghBX_7ZE&co=aHR0cHM6Ly9hY2NvdW50cy5kbW0uY29tOjQ0Mw..&hl=ja&v=1Bq_oiMBd4XPUhKDwr0YL1Js&size=invisible")
+    result = session.post(url, json={"code":oauth_token}, verify=SSL_VERIFY)
-        return captcha
+    result.raise_for_status()
-    except Exception as e:
+    data = result.json()["data"]
-        print("Failed to solve captcha:", e)
+    return data["access_token"], data["expires_in_seconds"]
 
-def retrieve_auth_keys(login, password, token, captcha, session : requests.Session):
+def agree_to_game_terms(game_id, use_proxy):
     try:
-        print("Logging in")
+        print("Accepting updated game terms of use")
-        url = "https://accounts.dmm.com/service/login/password/authenticate"
-        data = f"token={token}&login_id={login}&password={password}&prompt=&device=games-player&recaptchaToken={captcha}"
-        headers = {"Content-Type": "application/x-www-form-urlencoded"}
-        result = session.post(url, data, headers=headers)
-        result.raise_for_status()
-        cookies = result.cookies.get_dict()
-        return cookies["login_secure_id"], cookies["login_session_id"]
-    except Exception as e:
-        print("Failed to log in:", e)
-        return None, None
-    
-def retrieve_update_params(game_id, login_secure, login_session, use_proxy):
-    try:
-        print("Retrieving update file list")
-        data = {"product_id":game_id,"game_type":"GCL","game_os":"win"}
         headers = {"User-Agent": "DMMGamePlayer5-Win/5.3.12 Electron/32.1.0", "Client-App": "DMMGamePlayer5", "Client-version": "5.3.12", "Content-Type": "application/json"}
-        cookies = {"login_secure_id":login_secure, "login_session_id":login_session}
+        url = "https://katworks.sytes.net/proxy/agreement" if use_proxy else "https://apidgp-gameplayer.games.dmm.com/v5/agreement/confirm/client" 
-        url = "https://katworks.sytes.net/KF/Api/DMM/filelist" if use_proxy else "https://apidgp-gameplayer.games.dmm.com/v5/r2/filelist/cl"
+        data = {"product_id":game_id,"is_notification":False,"is_myapp":False}
-        result = requests.post(url, cookies=cookies, headers=headers, json=data)
+        result = requests.post(url, headers=headers, json=data)
         result.raise_for_status()
         result_json = result.json()
         if result_json["result_code"] != 100:
             raise Exception(f'{result_json["result_code"]}: {result_json["error"]}')
-        data = result_json["data"]
+        return True
-        game_version = data["latest_version"]
-        print("Latest version:", game_version)
-        file_list_url = data["file_list_url"]
-        file_list_params = data["sign"]
-        file_list_params = "?" + file_list_params.replace(";", "&").replace("CloudFront-", "")
-        return file_list_url, file_list_params
     except Exception as e:
-        print("Failed to retrieve update file list:", e)
+        print("Failed to accept terms of use:", e)
-        return None, None
+        return False
-    
+
-def retrieve_launch_params(game_id, mac_addr, hdd_serial, motherboard, login_secure, login_session, use_proxy):
+def retrieve_launch_params(game_id, game_type, mac_addr, hdd_serial, motherboard, access_token, session: requests.Session, use_proxy = False, update_game = False):
     try:
         print("Retrieving launch arguments")
-        data = {"product_id":game_id,"game_type":"GCL","game_os":"win","launch_type":"LIB","mac_address":mac_addr,"hdd_serial":hdd_serial,"motherboard":motherboard,"user_os":"win"}
+        data = {"product_id":game_id,"game_type":game_type,"game_os":"win","launch_type":"LIB","mac_address":mac_addr,"hdd_serial":hdd_serial,"motherboard":motherboard,"user_os":"win"}
-        headers = {"User-Agent": "DMMGamePlayer5-Win/5.3.12 Electron/32.1.0", "Client-App": "DMMGamePlayer5", "Client-version": "5.3.12", "Content-Type": "application/json"}
+        headers = {"User-Agent": "DMMGamePlayer5-Win/5.3.12 Electron/32.1.0",
-        cookies = {"login_secure_id":login_secure, "login_session_id":login_session}
+                   "Client-App": "DMMGamePlayer5",
-        url = "https://katworks.sytes.net/KF/Api/DMM/launch" if use_proxy else "https://apidgp-gameplayer.games.dmm.com/v5/r2/launch/cl"
+                   "Client-version": "5.3.12",
-        result = requests.post(url, cookies=cookies, headers=headers, json=data)
+                   "Content-Type": "application/json",
+                   "actauth":access_token}
+        if update_game:
+            url = "https://katworks.sytes.net/proxy/launchAndUpdate" if use_proxy else "https://apidgp-gameplayer.games.dmm.com/v5/r2/launch/cl" 
+        else:
+            url = "https://katworks.sytes.net/proxy/launch" if use_proxy else "https://apidgp-gameplayer.games.dmm.com/v5/launch/cl" 
+        result = session.post(url, headers=headers, json=data, verify=SSL_VERIFY)
         result.raise_for_status()
         result_json = result.json()
-        if result_json["result_code"] != 100:
+        if result_json["result_code"] == 308:
+            if agree_to_game_terms(game_id, use_proxy):
+                result = session.post(url, headers=headers, json=data, verify=SSL_VERIFY)
+                result.raise_for_status()
+                result_json = result.json()
+                if result_json["result_code"] != 100:
+                    raise Exception(f'{result_json["result_code"]}: {result_json["error"]}')
+            else:
+                raise Exception("Failed to agree to updated game terms of use. Use the DMM app to confirm.")
+        elif result_json["result_code"] != 100:
             raise Exception(f'{result_json["result_code"]}: {result_json["error"]}')
         data = result_json["data"]
-        return data["execute_args"]
+        return data
     except Exception as e:
-        print("Failed to retrieve launch arguments:", e)
+        if str(e).startswith("203:"):
+            config = load_config(game_id)
+            config["saved_login"] = None
+            save_config(game_id, config)
+            print("DMM session has expired. Saved login data cleared. Please run the program again.")
+        else:
+            print("Failed to retrieve launch arguments:", e)
 
 def main(config):
-    game_id = config["game_id"]
+    #required arguments
+    game_id     = config["game_id"]
+    game_type   = config["game_type"]
     exe_location = config["file_path"]
-    login = urllib.parse.quote_plus(config["dmm_login"])
+    login       = urllib.parse.quote_plus(config["dmm_login"])
-    password = urllib.parse.quote_plus(config["dmm_password"])
+    password    = urllib.parse.quote_plus(config["dmm_password"])
-    update_game = config["update_game"] if "update_game" in config else False
+    #optional arguments
-    use_proxy = config["use_proxy"] if "use_proxy" in config else False
+    update_game = config.get("update_game", False)
+    use_proxy   = config.get("use_proxy", False)
+    saved_login = config.get("saved_login", None)
+    #dmm requires these values
     mac_addr = get_mac()
-    hdd_serial = get_hash('')
+    hdd_serial = get_hash('') #DMM sends an empty hash as well
-    #actual moterboard serial is unknown, but this works
+    motherboard =  get_hash(getnode()) #actual moterboard serial is unknown, but this works
-    motherboard =  get_hash(getnode())
+
+    current_time = datetime.now()
 
     with requests.Session() as session:
-        token = retrieve_login_token(session)
+        if saved_login is not None:
-        captcha = retrieve_captcha_token()
+            login_expiration = datetime.fromtimestamp(saved_login["expiration"])
+            if login_expiration < current_time:
+                print("Login data has expired")
+                saved_login = None
 
-        if token == None or captcha == None:
+        if saved_login is not None:
-            return
+            access_token = saved_login["access_token"]
+            print("Loaded token from previous session")
+        else:
+            oauth_code = retrieve_oauth_code(login, password, session)
+            if oauth_code == None:
+                return
+            
+            access_token, expiration_seconds = retrieve_access_token(oauth_code, session)
 
-        login_secure, login_session = retrieve_auth_keys(login, password, token, captcha, session)
+            token_expiration = int((current_time + timedelta(seconds=expiration_seconds)).timestamp())
-        if login_secure == None or login_session == None:
+            config["saved_login"] = saved_login = {"access_token" : access_token, "expiration": token_expiration}
-            return
+            del(config["update_game"])
+            save_config(game_id, config)
         
         if not use_proxy: input("Enable VPN now and press Enter")
 
-        if update_game:
+        #execute_args, file_list_url, file_access_params
-            file_list_url, file_access_params = retrieve_update_params(game_id, login_secure, login_session, use_proxy)
+        launch_data : dict = retrieve_launch_params(game_id, game_type, mac_addr, hdd_serial, motherboard, access_token, session, use_proxy, update_game)
-            if file_list_url == None or file_access_params == None:
+                
-                return
+        if launch_data == None:
+            return
+        
+        execute_args : str = launch_data.get("execute_args", None)
 
-        execute_args = retrieve_launch_params(game_id, mac_addr, hdd_serial, motherboard, login_secure, login_session, use_proxy)
         if execute_args == None:
             return
+        
+        if update_game:
+            file_list_url = launch_data.get("file_list_url", None)
+            file_list_params : str = launch_data.get("sign", None)
+            if file_list_url == None or file_list_params == None:
+                return
+
+            print("Latest version:", launch_data["latest_version"])
+            file_list_params = "?" + file_list_params.replace(";", "&").replace("CloudFront-", "")
+
+        exec_name = launch_data["exec_file_name"]
+        install_dir = launch_data["install_dir"]
+        game_root_path = get_game_root_path(exe_location, install_dir, exec_name)
 
         if not use_proxy: input("Disable VPN now and press Enter")
 
         if update_game:
-            dmmUpdater.update_game(os.path.dirname(exe_location), file_list_url, file_access_params)
+            dmmUpdater.update_game(game_root_path, file_list_url, file_list_params)
 
         print("Starting game")
-        args = [exe_location] + execute_args.split()
+        args = [os.path.join(game_root_path, exec_name)] + execute_args.split()
-        print(args)
         subprocess.Popen(args, start_new_session=True) 
-        input("Done. Press enter to exit (this will close the game)")
+        input("Done. Press enter to exit")
 
 main(config)

dmmUpdater.py

@@ -4,11 +4,16 @@ import hashlib
 import requests
 import urllib.parse
 from urllib.request import urlretrieve
+from pathlib import Path
+
+# Set to False to allow packet sniffing/capture.
+# Set separately for dmmBypass
+SSL_VERIFY = True
 
 def get_file_list(url):
     url = "https://apidgp-gameplayer.games.dmm.com" + url
     print("Retrieving file list from " + url)
-    result = requests.get(url)
+    result = requests.get(url, verify=SSL_VERIFY)
     result.raise_for_status()
     data = result.json()["data"]
     return data["domain"], data["file_list"]
@@ -27,33 +32,28 @@ def get_file_hash(file_path):
 def update_game(game_path, files_url, files_param):
     print("Updating game")
     server_url, server_files = get_file_list(files_url)
-    server_file_dict = {file["local_path"]: file for file in server_files}
+    server_file_dict = {str(Path(game_path, file["local_path"].lstrip('/')).resolve()): file for file in server_files}
-
+    local_file_dict = {str(Path(dp, f).resolve()): "" for dp, dn, filenames in os.walk(game_path) for f in filenames}
-    local_files = [os.path.join(dp, f).replace("\\", "/") for dp, dn, filenames in os.walk(game_path) for f in filenames]
-    local_file_dict = {"/" + os.path.relpath(r, game_path).replace("\\", "/"): {"abs_path":r, "hash":""} for r in local_files}
     
     files_to_download = []
     files_to_delete = []
-    for server_file_key in server_file_dict.keys():
+    for abs_file_path in server_file_dict.keys():
-        server_file = server_file_dict[server_file_key]
+        server_file = server_file_dict[abs_file_path]
         
-        if server_file_key in local_file_dict:
+        if abs_file_path in local_file_dict:
-            local_file = local_file_dict[server_file_key]
             if server_file["force_delete_flg"]:
-                files_to_delete.append(local_file["abs_path"])
+                files_to_delete.append(abs_file_path)
             else:
-                local_file["hash"] = get_file_hash(local_file["abs_path"])
+                local_file_hash = get_file_hash(abs_file_path)
                 
-                if server_file["check_hash_flg"] and local_file["hash"] == server_file["hash"]:
+                if server_file["check_hash_flg"] and local_file_hash == server_file["hash"]:
                     continue
 
                 download_url = urllib.parse.urljoin(server_url, server_file["path"]) + files_param
-                download_path = game_path.replace("\\", "/") + server_file_key
+                files_to_download.append({"url":download_url, "path":abs_file_path})
-                files_to_download.append({"url":download_url, "path":download_path})
         else:
             download_url = urllib.parse.urljoin(server_url, server_file["path"]) + files_param
-            download_path = game_path.replace("\\", "/") + server_file_key
+            files_to_download.append({"url":download_url, "path":abs_file_path})
-            files_to_download.append({"url":download_url, "path":download_path})
         
     print("Files to download:", len(files_to_download))
 
@@ -75,7 +75,7 @@ def update_game(game_path, files_url, files_param):
                 file = files_to_download[index]
                 url, path = file["url"], file["path"]
                 
-                response = requests.get(url, timeout=10, stream=True)
+                response = requests.get(url, timeout=10, stream=True, verify=SSL_VERIFY)
                 total_size = int(response.headers.get("content-length", 0))
                 block_size = 1024 * 1024
                 downloaded = 0