artemis/core/mucha.py

342 lines
12 KiB
Python
Raw Normal View History

2024-01-09 20:59:58 +00:00
from typing import Dict, Any, Optional
import logging, coloredlogs
from logging.handlers import TimedRotatingFileHandler
2024-01-09 08:07:04 +00:00
from starlette.requests import Request
from datetime import datetime
2023-07-12 04:41:53 +00:00
from Crypto.Cipher import Blowfish
import pytz
from .config import CoreConfig
from .utils import Utils
from .title import TitleServlet
2023-03-09 16:38:58 +00:00
class MuchaServlet:
2024-01-09 08:07:04 +00:00
mucha_registry: Dict[str, str] = {}
2023-03-09 16:38:58 +00:00
def __init__(self, cfg: CoreConfig, cfg_dir: str) -> None:
self.config = cfg
2023-03-05 03:46:26 +00:00
self.config_dir = cfg_dir
2023-03-09 16:38:58 +00:00
self.logger = logging.getLogger("mucha")
log_fmt_str = "[%(asctime)s] Mucha | %(levelname)s | %(message)s"
log_fmt = logging.Formatter(log_fmt_str)
2023-03-09 16:38:58 +00:00
fileHandler = TimedRotatingFileHandler(
"{0}/{1}.log".format(self.config.server.log_dir, "mucha"),
when="d",
backupCount=10,
)
fileHandler.setFormatter(log_fmt)
2023-03-09 16:38:58 +00:00
consoleHandler = logging.StreamHandler()
consoleHandler.setFormatter(log_fmt)
self.logger.addHandler(fileHandler)
self.logger.addHandler(consoleHandler)
2023-03-09 16:38:58 +00:00
self.logger.setLevel(cfg.mucha.loglevel)
coloredlogs.install(level=cfg.mucha.loglevel, logger=self.logger, fmt=log_fmt_str)
for _, mod in TitleServlet.title_registry.items():
if hasattr(mod, "get_mucha_info"):
2024-01-09 08:07:04 +00:00
enabled, game_cds, netid_prefixes = mod.get_mucha_info(
2023-03-09 16:38:58 +00:00
self.config, self.config_dir
)
2023-03-05 03:46:26 +00:00
if enabled:
2024-01-09 08:07:04 +00:00
for x in range(len(game_cds)):
self.mucha_registry[game_cds[x]] = netid_prefixes[x]
2023-03-09 16:38:58 +00:00
self.logger.info(f"Serving {len(self.mucha_registry)} games")
2023-03-05 03:46:26 +00:00
2024-01-22 19:59:53 +00:00
def handle_boardauth(self, request: Request) -> bytes:
req_dict = self.mucha_preprocess(request.content.getvalue())
client_ip = Utils.get_ip_addr(request)
if req_dict is None:
2023-03-09 16:38:58 +00:00
self.logger.error(
f"Error processing mucha request {request.content.getvalue()}"
)
2023-07-12 04:41:53 +00:00
return b"RESULTS=000"
req = MuchaAuthRequest(req_dict)
2023-07-12 04:41:53 +00:00
self.logger.info(f"Boardauth request from {client_ip} for {req.gameVer}")
2023-02-21 21:46:43 +00:00
self.logger.debug(f"Mucha request {vars(req)}")
2023-03-05 03:46:26 +00:00
if req.gameCd not in self.mucha_registry:
2023-08-08 14:17:56 +00:00
self.logger.warning(f"Unknown gameCd {req.gameCd}")
2023-07-12 04:41:53 +00:00
return b"RESULTS=000"
2023-03-05 03:46:26 +00:00
2023-03-05 16:42:03 +00:00
# TODO: Decrypt S/N
2023-07-16 20:58:34 +00:00
b_key = b""
for x in range(8):
b_key += req.sendDate[(x - 1) & 7].encode()
2023-03-09 16:38:58 +00:00
2023-07-16 20:58:34 +00:00
cipher = Blowfish.new(b_key, Blowfish.MODE_ECB)
sn_decrypt = cipher.decrypt(bytes.fromhex(req.serialNum))
self.logger.debug(f"Decrypt SN to {sn_decrypt.hex()}")
2023-07-12 04:41:53 +00:00
2023-03-09 16:38:58 +00:00
resp = MuchaAuthResponse(
2024-01-09 08:07:04 +00:00
f"{self.config.server.hostname}{':' + str(self.config.server.port) if self.config.server.is_develop else ''}"
2023-03-09 16:38:58 +00:00
)
2023-02-21 21:46:43 +00:00
self.logger.debug(f"Mucha response {vars(resp)}")
return self.mucha_postprocess(vars(resp))
2023-03-09 16:38:58 +00:00
2024-01-22 19:59:53 +00:00
def handle_updatecheck(self, request: Request) -> bytes:
req_dict = self.mucha_preprocess(request.content.getvalue())
client_ip = Utils.get_ip_addr(request)
if req_dict is None:
2023-03-09 16:38:58 +00:00
self.logger.error(
f"Error processing mucha request {request.content.getvalue()}"
)
2023-07-12 04:41:53 +00:00
return b"RESULTS=000"
req = MuchaUpdateRequest(req_dict)
2023-07-12 04:41:53 +00:00
self.logger.info(f"Updatecheck request from {client_ip} for {req.gameVer}")
2023-02-21 21:46:43 +00:00
self.logger.debug(f"Mucha request {vars(req)}")
2023-03-05 03:46:26 +00:00
if req.gameCd not in self.mucha_registry:
2023-08-08 14:17:56 +00:00
self.logger.warning(f"Unknown gameCd {req.gameCd}")
2023-07-12 04:41:53 +00:00
return b"RESULTS=000"
2023-02-21 21:46:43 +00:00
2024-01-09 08:07:04 +00:00
resp = MuchaUpdateResponse(req.gameVer, f"{self.config.server.hostname}{':' + str(self.config.server.port) if self.config.server.is_develop else ''}")
2023-02-21 21:46:43 +00:00
self.logger.debug(f"Mucha response {vars(resp)}")
return self.mucha_postprocess(vars(resp))
2024-01-22 19:59:53 +00:00
def handle_dlstate(self, request: Request) -> bytes:
2023-07-12 04:41:53 +00:00
req_dict = self.mucha_preprocess(request.content.getvalue())
client_ip = Utils.get_ip_addr(request)
if req_dict is None:
self.logger.error(
f"Error processing mucha request {request.content.getvalue()}"
)
return b""
req = MuchaDownloadStateRequest(req_dict)
self.logger.info(f"DownloadState request from {client_ip} for {req.gameCd} -> {req.updateVer}")
self.logger.debug(f"request {vars(req)}")
return b"RESULTS=001"
def mucha_preprocess(self, data: bytes) -> Optional[Dict]:
try:
ret: Dict[str, Any] = {}
2023-03-09 16:38:58 +00:00
for x in data.decode().split("&"):
kvp = x.split("=")
if len(kvp) == 2:
ret[kvp[0]] = kvp[1]
return ret
2023-03-09 16:38:58 +00:00
2023-07-16 20:58:34 +00:00
except Exception:
self.logger.error(f"Error processing mucha request {data}")
return None
def mucha_postprocess(self, data: dict) -> Optional[bytes]:
try:
urlencode = "&".join(f"{k}={v}" for k, v in data.items())
return urlencode.encode()
2023-07-16 20:58:34 +00:00
except Exception:
self.logger.error("Error processing mucha response")
return None
2023-03-09 16:38:58 +00:00
class MuchaAuthRequest:
def __init__(self, request: Dict) -> None:
2023-03-09 17:17:10 +00:00
# gameCd + boardType + countryCd + version
self.gameVer = request.get("gameVer", "")
self.sendDate = request.get("sendDate", "") # %Y%m%d
self.serialNum = request.get("serialNum", "")
self.gameCd = request.get("gameCd", "")
self.boardType = request.get("boardType", "")
self.boardId = request.get("boardId", "")
self.mac = request.get("mac", "")
self.placeId = request.get("placeId", "")
self.storeRouterIp = request.get("storeRouterIp", "")
self.countryCd = request.get("countryCd", "")
self.useToken = request.get("useToken", "")
self.allToken = request.get("allToken", "")
2023-03-09 16:38:58 +00:00
class MuchaAuthResponse:
def __init__(self, mucha_url: str) -> None:
2023-03-09 16:38:58 +00:00
self.RESULTS = "001"
self.AUTH_INTERVAL = "86400"
self.SERVER_TIME = datetime.strftime(datetime.now(), "%Y%m%d%H%M")
self.UTC_SERVER_TIME = datetime.strftime(datetime.now(pytz.UTC), "%Y%m%d%H%M")
2023-03-09 16:38:58 +00:00
self.CHARGE_URL = f"https://{mucha_url}/charge/"
self.FILE_URL = f"https://{mucha_url}/file/"
self.URL_1 = f"https://{mucha_url}/url1/"
self.URL_2 = f"https://{mucha_url}/url2/"
self.URL_3 = f"https://{mucha_url}/url3/"
2023-03-09 16:38:58 +00:00
self.PLACE_ID = "JPN123"
self.COUNTRY_CD = "JPN"
self.SHOP_NAME = "TestShop!"
self.SHOP_NICKNAME = "TestShop"
self.AREA_0 = "008"
self.AREA_1 = "009"
self.AREA_2 = "010"
self.AREA_3 = "011"
self.AREA_FULL_0 = ""
self.AREA_FULL_1 = ""
self.AREA_FULL_2 = ""
self.AREA_FULL_3 = ""
2023-03-09 16:38:58 +00:00
self.SHOP_NAME_EN = "TestShop!"
self.SHOP_NICKNAME_EN = "TestShop"
self.AREA_0_EN = "008"
self.AREA_1_EN = "009"
self.AREA_2_EN = "010"
self.AREA_3_EN = "011"
self.AREA_FULL_0_EN = ""
self.AREA_FULL_1_EN = ""
self.AREA_FULL_2_EN = ""
self.AREA_FULL_3_EN = ""
2023-03-09 16:38:58 +00:00
self.PREFECTURE_ID = "1"
self.EXPIRATION_DATE = "null"
self.USE_TOKEN = "0"
self.CONSUME_TOKEN = "0"
self.DONGLE_FLG = "1"
self.FORCE_BOOT = "0"
2023-03-09 16:38:58 +00:00
class MuchaUpdateRequest:
def __init__(self, request: Dict) -> None:
2023-03-09 17:17:10 +00:00
self.gameVer = request.get("gameVer", "")
self.gameCd = request.get("gameCd", "")
self.serialNum = request.get("serialNum", "")
self.countryCd = request.get("countryCd", "")
self.placeId = request.get("placeId", "")
self.storeRouterIp = request.get("storeRouterIp", "")
2023-03-09 16:38:58 +00:00
class MuchaUpdateResponse:
def __init__(self, game_ver: str, mucha_url: str) -> None:
2023-07-12 04:41:53 +00:00
self.RESULTS = "001"
self.EXE_VER = game_ver
self.UPDATE_VER_1 = game_ver
2023-07-12 04:41:53 +00:00
self.UPDATE_URL_1 = f"http://{mucha_url}/updUrl1/"
self.UPDATE_SIZE_1 = "20"
self.CHECK_CRC_1 = "0000000000000000"
self.CHECK_URL_1 = f"http://{mucha_url}/checkUrl/"
self.CHECK_SIZE_1 = "20"
self.INFO_SIZE_1 = "0"
self.COM_SIZE_1 = "0"
self.COM_TIME_1 = "0"
self.LAN_INFO_SIZE_1 = "0"
2023-07-12 04:41:53 +00:00
self.USER_ID = ""
self.PASSWORD = ""
2023-03-05 14:43:05 +00:00
2023-07-12 04:41:53 +00:00
"""
RESULTS
EXE_VER
UPDATE_VER_%d
UPDATE_URL_%d
UPDATE_SIZE_%d
CHECK_CRC_%d
CHECK_URL_%d
CHECK_SIZE_%d
2023-03-09 16:38:58 +00:00
2023-07-12 04:41:53 +00:00
INFO_SIZE_1
COM_SIZE_1
COM_TIME_1
LAN_INFO_SIZE_1
USER_ID
PASSWORD
"""
2023-03-09 16:38:58 +00:00
class MuchaUpdateResponseStub:
2023-03-05 14:43:05 +00:00
def __init__(self, game_ver: str) -> None:
self.RESULTS = "001"
self.UPDATE_VER_1 = game_ver
2023-07-12 04:41:53 +00:00
class MuchaDownloadStateRequest:
def __init__(self, request: Dict) -> None:
self.gameCd = request.get("gameCd", "")
self.updateVer = request.get("updateVer", "")
self.serialNum = request.get("serialNum", "")
self.fileSize = request.get("fileSize", "")
self.compFileSize = request.get("compFileSize", "")
self.boardId = request.get("boardId", "")
self.placeId = request.get("placeId", "")
self.storeRouterIp = request.get("storeRouterIp", "")
class MuchaDownloadErrorRequest:
def __init__(self, request: Dict) -> None:
self.gameCd = request.get("gameCd", "")
self.updateVer = request.get("updateVer", "")
self.serialNum = request.get("serialNum", "")
self.downloadUrl = request.get("downloadUrl", "")
self.errCd = request.get("errCd", "")
self.errMessage = request.get("errMessage", "")
self.boardId = request.get("boardId", "")
self.placeId = request.get("placeId", "")
self.storeRouterIp = request.get("storeRouterIp", "")
class MuchaRegiAuthRequest:
def __init__(self, request: Dict) -> None:
self.gameCd = request.get("gameCd", "")
self.serialNum = request.get("serialNum", "") # Encrypted
self.countryCd = request.get("countryCd", "")
self.registrationCd = request.get("registrationCd", "")
self.sendDate = request.get("sendDate", "")
self.useToken = request.get("useToken", "")
self.allToken = request.get("allToken", "")
self.placeId = request.get("placeId", "")
self.storeRouterIp = request.get("storeRouterIp", "")
class MuchaRegiAuthResponse:
def __init__(self) -> None:
self.RESULTS = "001" # 001 = success, 099, 098, 097 = fail, others = fail
self.ALL_TOKEN = "0" # Encrypted
self.ADD_TOKEN = "0" # Encrypted
class MuchaTokenStateRequest:
def __init__(self, request: Dict) -> None:
self.gameCd = request.get("gameCd", "")
self.serialNum = request.get("serialNum", "")
self.countryCd = request.get("countryCd", "")
self.useToken = request.get("useToken", "")
self.allToken = request.get("allToken", "")
self.placeId = request.get("placeId", "")
self.storeRouterIp = request.get("storeRouterIp", "")
class MuchaTokenStateResponse:
def __init__(self) -> None:
self.RESULTS = "001"
class MuchaTokenMarginStateRequest:
def __init__(self, request: Dict) -> None:
self.gameCd = request.get("gameCd", "")
self.serialNum = request.get("serialNum", "")
self.countryCd = request.get("countryCd", "")
self.placeId = request.get("placeId", "")
self.limitLowerToken = request.get("limitLowerToken", 0)
self.limitUpperToken = request.get("limitUpperToken", 0)
self.settlementMonth = request.get("settlementMonth", 0)
class MuchaTokenMarginStateResponse:
def __init__(self) -> None:
self.RESULTS = "001"
self.LIMIT_LOWER_TOKEN = 0
self.LIMIT_UPPER_TOKEN = 0
self.LAST_SETTLEMENT_MONTH = 0
self.LAST_LIMIT_LOWER_TOKEN = 0
self.LAST_LIMIT_UPPER_TOKEN = 0
self.SETTLEMENT_MONTH = 0