3
2
forked from Dniel97/artemis
artemis/titles/sao/index.py
2024-01-22 15:49:06 -05:00

159 lines
5.8 KiB
Python

from typing import Tuple, Dict, List
from starlette.requests import Request
from starlette.responses import Response
from starlette.routing import Route
import yaml
import logging, coloredlogs
from logging.handlers import TimedRotatingFileHandler
from os import path
from Crypto.Cipher import Blowfish
from hashlib import md5
import secrets
from core import CoreConfig, Utils
from core.title import BaseServlet
from titles.sao.config import SaoConfig
from titles.sao.const import SaoConstants
from titles.sao.base import SaoBase
from titles.sao.handlers.base import *
class SaoServlet(BaseServlet):
def __init__(self, core_cfg: CoreConfig, cfg_dir: str) -> None:
super().__init__(core_cfg, cfg_dir)
self.config_dir = cfg_dir
self.game_cfg = SaoConfig()
if path.exists(f"{cfg_dir}/sao.yaml"):
self.game_cfg.update(yaml.safe_load(open(f"{cfg_dir}/sao.yaml")))
self.logger = logging.getLogger("sao")
if not hasattr(self.logger, "inited"):
log_fmt_str = "[%(asctime)s] SAO | %(levelname)s | %(message)s"
log_fmt = logging.Formatter(log_fmt_str)
fileHandler = TimedRotatingFileHandler(
"{0}/{1}.log".format(self.core_cfg.server.log_dir, "sao"),
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.inited = True
self.base = SaoBase(core_cfg, self.game_cfg)
self.static_hash = None
if self.game_cfg.hash.verify_hash:
self.static_hash = md5(self.game_cfg.hash.hash_base.encode()).digest() # Greate hashing guys, really validates the data
def get_routes(self) -> List[Route]:
return [
Route("/{datecode:int}/proto/if/{category:str}/{endpoint:str}", self.render_POST, methods=['POST'])
]
@classmethod
def is_game_enabled(cls, game_code: str, core_cfg: CoreConfig, cfg_dir: str) -> bool:
game_cfg = SaoConfig()
if path.exists(f"{cfg_dir}/{SaoConstants.CONFIG_NAME}"):
game_cfg.update(
yaml.safe_load(open(f"{cfg_dir}/{SaoConstants.CONFIG_NAME}"))
)
if not game_cfg.server.enable:
return False
return True
def get_allnet_info(self, game_code: str, game_ver: int, keychip: str) -> Tuple[str, str]:
port_ssl = Utils.get_title_port_ssl(self.core_cfg)
port_normal = Utils.get_title_port(self.core_cfg)
proto = "http"
port = f":{port_normal}" if port_normal != 80 else ""
if self.game_cfg.server.use_https:
proto = "https"
port = f":{port_ssl}" if port_ssl != 443 else ""
return (f"{proto}://{self.core_cfg.server.hostname}{port}/", "")
def get_mucha_info(self, core_cfg: CoreConfig, cfg_dir: str) -> Tuple[bool, str]:
if not self.game_cfg.server.enable:
return (False, [], [])
return (True, SaoConstants.GAME_CDS, SaoConstants.NETID_PREFIX)
async def render_POST(self, request: Request) -> bytes:
endpoint = request.path_params.get('endpoint', '')
iv = b""
req_raw = await request.body()
if len(req_raw) < 40:
self.logger.warn(f"Malformed request to {endpoint} - {req_raw.hex()}")
return Response()
req_header = SaoRequestHeader(req_raw)
cmd_str = f"{req_header.cmd:04x}"
if self.game_cfg.hash.verify_hash and self.static_hash != req_header.hash:
self.logger.error(f"Hash mismatch! Expecting {self.static_hash} but recieved {req_header.hash}")
return Response()
if self.game_cfg.crypt.enable:
iv = req_raw[40:48]
cipher = Blowfish.new(self.game_cfg.crypt.key.encode(), Blowfish.MODE_CBC, iv)
crypt_data = req_raw[48:]
req_data = cipher.decrypt(crypt_data)
self.logger.debug(f"Decrypted {req_data.hex()} with IV {iv.hex()}")
else:
req_data = req_raw[40:]
handler = getattr(self.base, f"handle_{cmd_str}", self.base.handle_noop)
self.logger.info(f"{endpoint} - {cmd_str} request")
self.logger.debug(f"Request: {req_raw.hex()}")
resp = await handler(req_header, req_data)
if resp is None:
resp = SaoNoopResponse(req_header.cmd + 1).make()
if type(resp) == bytes:
pass
elif issubclass(resp, SaoBaseResponse):
resp = resp.make()
else:
self.logger.error(f"Unknown response type {type(resp)}")
return Response()
self.logger.debug(f"Response: {resp.hex()}")
if self.game_cfg.crypt.enable:
iv = secrets.token_bytes(8)
data_to_crypt = resp[24:]
while len(data_to_crypt) % 8 != 0:
data_to_crypt += b"\x00"
cipher = Blowfish.new(self.game_cfg.crypt.key.encode(), Blowfish.MODE_CBC, iv)
data_crypt = cipher.encrypt(data_to_crypt)
crypt_data_len = len(data_crypt) + len(iv)
tmp = struct.pack("!I", crypt_data_len) # does it want the length of the encrypted response??
resp = resp[:20] + tmp + iv + data_crypt
self.logger.debug(f"Encrypted Response: {resp.hex()}")
return Response(resp, media_type="text/html; charset=utf-8")