Kayori/artemis
1
0
Fork 0
forked from Hay1tsme/artemis
Code Pull Requests Activity
319aea098f
artemis/titles/mai2/index.py
zaphkito 08891d0851 mai2: some improve for DX earlier version and return game code in uri (#125) 
Attention: There are all talking about maimai DX and newer version, not Pre-DX

dx and newer version request these but no used, they are just exist in game code, only found `oldServerUrl` used in SDEZ 1.00, this should also fix SDGA and SDGB try to visit `ServerUrl + movieServerUrl` although that just because of SEGA shit code
tested work

![image](/attachments/f2c79134-4651-4976-8278-bbcf268f424a)

Reviewed-on: Hay1tsme/artemis#125
Co-authored-by: zaphkito <zaphkito@noreply.gitea.tendokyu.moe>
Co-committed-by: zaphkito <zaphkito@noreply.gitea.tendokyu.moe>
2024-05-05 05:41:14 +00:00

441 lines
17 KiB
Python
Raw Blame History

from starlette.requests import Request
from starlette.responses import Response, JSONResponse
from starlette.routing import Route
import json
import inflection
import yaml
import logging, coloredlogs
import zlib
import string
from logging.handlers import TimedRotatingFileHandler
from os import path, mkdir
from typing import Tuple, List, Dict
from Crypto.Hash import MD5
from Crypto.Cipher import AES
from Crypto.Util.Padding import pad
from core.config import CoreConfig
from core.utils import Utils
from core.title import BaseServlet
from .config import Mai2Config
from .const import Mai2Constants
from .base import Mai2Base
from .finale import Mai2Finale
from .dx import Mai2DX
from .dxplus import Mai2DXPlus
from .splash import Mai2Splash
from .splashplus import Mai2SplashPlus
from .universe import Mai2Universe
from .universeplus import Mai2UniversePlus
from .festival import Mai2Festival
from .festivalplus import Mai2FestivalPlus
from .buddies import Mai2Buddies
class Mai2Servlet(BaseServlet):
def __init__(self, core_cfg: CoreConfig, cfg_dir: str) -> None:
super().__init__(core_cfg, cfg_dir)
self.game_cfg = Mai2Config()
self.hash_table: Dict[int, Dict[str, str]] = {}
if path.exists(f"{cfg_dir}/{Mai2Constants.CONFIG_NAME}"):
self.game_cfg.update(
yaml.safe_load(open(f"{cfg_dir}/{Mai2Constants.CONFIG_NAME}"))
)
self.versions = [
Mai2Base,
None,
None,
None,
None,
None,
None,
None,
None,
None,
None,
None,
Mai2Finale,
Mai2DX,
Mai2DXPlus,
Mai2Splash,
Mai2SplashPlus,
Mai2Universe,
Mai2UniversePlus,
Mai2Festival,
Mai2FestivalPlus,
Mai2Buddies
]
self.logger = logging.getLogger("mai2")
if not hasattr(self.logger, "initted"):
log_fmt_str = "[%(asctime)s] Mai2 | %(levelname)s | %(message)s"
log_fmt = logging.Formatter(log_fmt_str)
fileHandler = TimedRotatingFileHandler(
"{0}/{1}.log".format(self.core_cfg.server.log_dir, "mai2"),
encoding="utf8",
when="d",
backupCount=10,
)
fileHandler.setFormatter(log_fmt)
consoleHandler = logging.StreamHandler()
consoleHandler.setFormatter(log_fmt)
self.logger.addHandler(fileHandler)
self.logger.addHandler(consoleHandler)
self.logger.setLevel(self.game_cfg.server.loglevel)
coloredlogs.install(
level=self.game_cfg.server.loglevel, logger=self.logger, fmt=log_fmt_str
)
self.logger.initted = True
for version, keys in self.game_cfg.crypto.keys.items():
if version < Mai2Constants.VER_MAIMAI_DX:
continue
if len(keys) < 3:
continue
self.hash_table[version] = {}
method_list = [
method
for method in dir(self.versions[version])
if not method.startswith("__")
]
for method in method_list:
# handle_method_api_request -> HandleMethodApiRequest
# remove the first 6 chars and the final 7 chars to get the canonical
# endpoint name.
method_fixed = inflection.camelize(method)[6:-7]
hash = MD5.new((method_fixed + keys[2]).encode())
# truncate unused bytes like the game does
hashed_name = hash.hexdigest()
self.hash_table[version][hashed_name] = method_fixed
self.logger.debug(
"Hashed v%s method %s with %s to get %s",
version, method_fixed, keys[2], hashed_name
)
@classmethod
def is_game_enabled(
cls, game_code: str, core_cfg: CoreConfig, cfg_dir: str
) -> bool:
game_cfg = Mai2Config()
if path.exists(f"{cfg_dir}/{Mai2Constants.CONFIG_NAME}"):
game_cfg.update(
yaml.safe_load(open(f"{cfg_dir}/{Mai2Constants.CONFIG_NAME}"))
)
if not game_cfg.server.enable:
return False
return True
def get_routes(self) -> List[Route]:
return [
Route("/{game:str}/{version:int}/MaimaiServlet/api/movie/{endpoint:str}", self.handle_movie, methods=['GET', 'POST']),
Route("/{game:str}/{version:int}/MaimaiServlet/old/{endpoint:str}", self.handle_old_srv),
Route("/{game:str}/{version:int}/MaimaiServlet/old/{endpoint:str}/{placeid:str}/{keychip:str}/{userid:int}", self.handle_old_srv_userdata),
Route("/{game:str}/{version:int}/MaimaiServlet/old/{endpoint:str}/{userid:int}", self.handle_old_srv_userdata),
Route("/{game:str}/{version:int}/MaimaiServlet/old/{endpoint:str}/{userid:int}", self.handle_old_srv_userdata),
Route("/{game:str}/{version:int}/MaimaiServlet/usbdl/{endpoint:str}", self.handle_usbdl),
Route("/{game:str}/{version:int}/MaimaiServlet/deliver/{endpoint:str}", self.handle_deliver),
Route("/{game:str}/{version:int}/MaimaiServlet/{endpoint:str}", self.handle_mai, methods=['POST']),
Route("/{game:str}/{version:int}/Maimai2Servlet/{endpoint:str}", self.handle_mai2, methods=['POST']),
]
def get_allnet_info(self, game_code: str, game_ver: int, keychip: str) -> Tuple[str, str]:
if not self.core_cfg.server.is_using_proxy and Utils.get_title_port(self.core_cfg) != 80:
return (
f"http://{self.core_cfg.server.hostname}:{Utils.get_title_port(self.core_cfg)}/{game_code}/{game_ver}/",
f"{self.core_cfg.server.hostname}",
)
return (
f"http://{self.core_cfg.server.hostname}/{game_code}/{game_ver}/",
f"{self.core_cfg.server.hostname}",
)
def setup(self):
if (
self.game_cfg.uploads.photos
and self.game_cfg.uploads.photos_dir
and not path.exists(self.game_cfg.uploads.photos_dir)
):
try:
mkdir(self.game_cfg.uploads.photos_dir)
except Exception:
self.logger.error(
f"Failed to make photo upload directory at {self.game_cfg.uploads.photos_dir}"
)
if (
self.game_cfg.uploads.movies
and self.game_cfg.uploads.movies_dir
and not path.exists(self.game_cfg.uploads.movies_dir)
):
try:
mkdir(self.game_cfg.uploads.movies_dir)
except Exception:
self.logger.error(
f"Failed to make movie upload directory at {self.game_cfg.uploads.movies_dir}"
)
async def handle_movie(self, request: Request):
return JSONResponse()
async def handle_usbdl(self, request: Request):
return Response("OK")
async def handle_deliver(self, request: Request):
return Response(status_code=404)
async def handle_mai(self, request: Request) -> bytes:
endpoint: str = request.path_params.get('endpoint')
version: int = request.path_params.get('version')
if endpoint.lower() == "ping":
return Response(zlib.compress(b'{"returnCode": "1"}'))
req_raw = await request.body()
internal_ver = 0
client_ip = Utils.get_ip_addr(request)
if version < 110: # 1.0
internal_ver = Mai2Constants.VER_MAIMAI
elif version >= 110 and version < 120: # Plus
internal_ver = Mai2Constants.VER_MAIMAI_PLUS
elif version >= 120 and version < 130: # Green
internal_ver = Mai2Constants.VER_MAIMAI_GREEN
elif version >= 130 and version < 140: # Green Plus
internal_ver = Mai2Constants.VER_MAIMAI_GREEN_PLUS
elif version >= 140 and version < 150: # Orange
internal_ver = Mai2Constants.VER_MAIMAI_ORANGE
elif version >= 150 and version < 160: # Orange Plus
internal_ver = Mai2Constants.VER_MAIMAI_ORANGE_PLUS
elif version >= 160 and version < 170: # Pink
internal_ver = Mai2Constants.VER_MAIMAI_PINK
elif version >= 170 and version < 180: # Pink Plus
internal_ver = Mai2Constants.VER_MAIMAI_PINK_PLUS
elif version >= 180 and version < 185: # Murasaki
internal_ver = Mai2Constants.VER_MAIMAI_MURASAKI
elif version >= 185 and version < 190: # Murasaki Plus
internal_ver = Mai2Constants.VER_MAIMAI_MURASAKI_PLUS
elif version >= 190 and version < 195: # Milk
internal_ver = Mai2Constants.VER_MAIMAI_MILK
elif version >= 195 and version < 197: # Milk Plus
internal_ver = Mai2Constants.VER_MAIMAI_MILK_PLUS
elif version >= 197: # Finale
internal_ver = Mai2Constants.VER_MAIMAI_FINALE
try:
unzip = zlib.decompress(req_raw)
except zlib.error as e:
self.logger.error(
f"Failed to decompress v{version} {endpoint} request -> {e}"
)
return Response(zlib.compress(b'{"stat": "0"}'))
req_data = json.loads(unzip)
self.logger.info(f"v{version} {endpoint} request from {client_ip}")
self.logger.debug(req_data)
func_to_find = "handle_" + inflection.underscore(endpoint) + "_request"
handler_cls = self.versions[internal_ver](self.core_cfg, self.game_cfg)
if not hasattr(handler_cls, func_to_find):
self.logger.warning(f"Unhandled v{version} request {endpoint}")
resp = {"returnCode": 1}
else:
try:
handler = getattr(handler_cls, func_to_find)
resp = await handler(req_data)
except Exception as e:
self.logger.error(f"Error handling v{version} method {endpoint} - {e}")
return Response(zlib.compress(b'{"returnCode": "0"}'))
if resp is None:
resp = {"returnCode": 1}
self.logger.debug(f"Response {resp}")
return Response(zlib.compress(json.dumps(resp, ensure_ascii=False).encode("utf-8")))
async def handle_mai2(self, request: Request) -> bytes:
endpoint: str = request.path_params.get('endpoint')
version: int = request.path_params.get('version')
game_code = request.path_params.get('game')
if endpoint.lower() == "ping":
return Response(zlib.compress(b'{"returnCode": "1"}'))
req_raw = await request.body()
internal_ver = 0
client_ip = Utils.get_ip_addr(request)
encrypted = False
if game_code == "SDEZ": # JP
if version < 110: # 1.0
internal_ver = Mai2Constants.VER_MAIMAI_DX
elif version >= 110 and version < 114: # PLUS
internal_ver = Mai2Constants.VER_MAIMAI_DX_PLUS
elif version >= 114 and version < 117: # Splash
internal_ver = Mai2Constants.VER_MAIMAI_DX_SPLASH
elif version >= 117 and version < 120: # Splash PLUS
internal_ver = Mai2Constants.VER_MAIMAI_DX_SPLASH_PLUS
elif version >= 120 and version < 125: # UNiVERSE
internal_ver = Mai2Constants.VER_MAIMAI_DX_UNIVERSE
elif version >= 125 and version < 130: # UNiVERSE PLUS
internal_ver = Mai2Constants.VER_MAIMAI_DX_UNIVERSE_PLUS
elif version >= 130 and version < 135: # FESTiVAL
internal_ver = Mai2Constants.VER_MAIMAI_DX_FESTIVAL
elif version >= 135 and version < 140: # FESTiVAL PLUS
internal_ver = Mai2Constants.VER_MAIMAI_DX_FESTIVAL_PLUS
elif version >= 140: # BUDDiES
internal_ver = Mai2Constants.VER_MAIMAI_DX_BUDDIES
elif game_code == "SDGA": # Int
if version < 105: # 1.0
internal_ver = Mai2Constants.VER_MAIMAI_DX
elif version >= 105 and version < 110: # PLUS
internal_ver = Mai2Constants.VER_MAIMAI_DX_PLUS
elif version >= 110 and version < 115: # Splash
internal_ver = Mai2Constants.VER_MAIMAI_DX_SPLASH
elif version >= 115 and version < 120: # Splash PLUS
internal_ver = Mai2Constants.VER_MAIMAI_DX_SPLASH_PLUS
elif version >= 120 and version < 125: # UNiVERSE
internal_ver = Mai2Constants.VER_MAIMAI_DX_UNIVERSE
elif version >= 125 and version < 130: # UNiVERSE PLUS
internal_ver = Mai2Constants.VER_MAIMAI_DX_UNIVERSE_PLUS
elif version >= 130 and version < 135: # FESTiVAL
internal_ver = Mai2Constants.VER_MAIMAI_DX_FESTIVAL
elif version >= 135 and version < 140: # FESTiVAL PLUS
internal_ver = Mai2Constants.VER_MAIMAI_DX_FESTIVAL_PLUS
if all(c in string.hexdigits for c in endpoint) and len(endpoint) == 32:
# If we get a 32 character long hex string, it's a hash and we're
# dealing with an encrypted request. False positives shouldn't happen
# as long as requests are suffixed with `Api`.
if internal_ver not in self.hash_table:
self.logger.error(
"v%s does not support encryption or no keys entered",
version,
)
return Response(zlib.compress(b'{"stat": "0"}'))
elif endpoint.lower() not in self.hash_table[internal_ver]:
self.logger.error(
"No hash found for v%s endpoint %s",
version, endpoint
)
return Response(zlib.compress(b'{"stat": "0"}'))
endpoint = self.hash_table[internal_ver][endpoint.lower()]
try:
crypt = AES.new(
bytes.fromhex(self.game_cfg.crypto.keys[internal_ver][0]),
AES.MODE_CBC,
bytes.fromhex(self.game_cfg.crypto.keys[internal_ver][1]),
)
req_raw = crypt.decrypt(req_raw)
except Exception as e:
self.logger.error(
"Failed to decrypt v%s request to %s",
version, endpoint,
exc_info=e,
)
return Response(zlib.compress(b'{"stat": "0"}'))
encrypted = True
if (
not encrypted
and self.game_cfg.crypto.encrypted_only
and version >= 110
):
self.logger.error(
"Unencrypted v%s %s request, but config is set to encrypted only: %r",
version, endpoint, req_raw
)
return Response(zlib.compress(b'{"stat": "0"}'))
try:
unzip = zlib.decompress(req_raw)
except zlib.error as e:
self.logger.error(
f"Failed to decompress v{version} {endpoint} request -> {e}"
)
return Response(zlib.compress(b'{"stat": "0"}'))
req_data = json.loads(unzip)
self.logger.info(f"v{version} {endpoint} request from {client_ip}")
self.logger.debug(req_data)
endpoint = (
endpoint.replace("MaimaiExp", "")
if game_code == Mai2Constants.GAME_CODE_DX_INT
else endpoint
)
func_to_find = "handle_" + inflection.underscore(endpoint) + "_request"
handler_cls = self.versions[internal_ver](self.core_cfg, self.game_cfg)
if not hasattr(handler_cls, func_to_find):
self.logger.warning(f"Unhandled v{version} request {endpoint}")
resp = {"returnCode": 1}
else:
try:
handler = getattr(handler_cls, func_to_find)
resp = await handler(req_data)
except Exception as e:
self.logger.error(f"Error handling v{version} method {endpoint} - {e}")
return Response(zlib.compress(b'{"stat": "0"}'))
if resp is None:
resp = {"returnCode": 1}
self.logger.debug(f"Response {resp}")
zipped = zlib.compress(json.dumps(resp, ensure_ascii=False).encode("utf-8"))
if not encrypted or version < 110:
return Response(zipped)
padded = pad(zipped, 16)
crypt = AES.new(
bytes.fromhex(self.game_cfg.crypto.keys[internal_ver][0]),
AES.MODE_CBC,
bytes.fromhex(self.game_cfg.crypto.keys[internal_ver][1]),
)
return Response(crypt.encrypt(padded))
async def handle_old_srv(self, request: Request) -> bytes:
endpoint = request.path_params.get('endpoint')
version = request.path_params.get('version')
self.logger.info(f"v{version} old server {endpoint}")
return Response(zlib.compress(b"ok"))
async def handle_old_srv_userdata(self, request: Request) -> bytes:
endpoint = request.path_params.get('endpoint')
version = request.path_params.get('version')
self.logger.info(f"v{version} old server {endpoint}")
return Response(zlib.compress(b"{}"))
View Git Blame Copy Permalink