diff --git a/example_config/fgoa.yaml b/example_config/fgoa.yaml new file mode 100644 index 0000000..7723ff4 --- /dev/null +++ b/example_config/fgoa.yaml @@ -0,0 +1,3 @@ +server: + enable: True + loglevel: "info" \ No newline at end of file diff --git a/readme.md b/readme.md index 1226784..b63a733 100644 --- a/readme.md +++ b/readme.md @@ -4,6 +4,9 @@ A network service emulator for games running SEGA'S ALL.NET service, and similar # Supported games Games listed below have been tested and confirmed working. Only game versions older then the version currently active in arcades, or games versions that have not recieved a major update in over one year, are supported. ++ Fate/Grand Order Arcade + + 10.80 + + Card Maker + 1.30 + 1.35 diff --git a/titles/fgoa/__init__.py b/titles/fgoa/__init__.py new file mode 100644 index 0000000..f50a6cf --- /dev/null +++ b/titles/fgoa/__init__.py @@ -0,0 +1,5 @@ +from titles.fgoa.index import FGOAServlet +from titles.fgoa.const import FGOAConstants + +index = FGOAServlet +game_codes = [FGOAConstants.GAME_CODE] \ No newline at end of file diff --git a/titles/fgoa/base.py b/titles/fgoa/base.py new file mode 100644 index 0000000..9161cee --- /dev/null +++ b/titles/fgoa/base.py @@ -0,0 +1,30 @@ +from datetime import date, datetime, timedelta +from typing import Any, Dict, List +import json +import logging +from enum import Enum + +from core.config import CoreConfig +from titles.fgoa.config import FGOAConfig +from titles.fgoa.const import FGOAConstants + + +class FGOABase: + def __init__(self, core_cfg: CoreConfig, game_cfg: FGOAConfig) -> None: + self.core_cfg = core_cfg + self.game_config = game_cfg + self.date_time_format = "%Y-%m-%d %H:%M:%S" + self.date_time_format_ext = ( + "%Y-%m-%d %H:%M:%S.%f" # needs to be lopped off at [:-5] + ) + self.date_time_format_short = "%Y-%m-%d" + self.logger = logging.getLogger("fgoa") + self.game = FGOAConstants.GAME_CODE + self.version = FGOAConstants.VER_FGOA_SEASON_1 + + @staticmethod + def _parse_int_ver(version: str) -> str: + return version.replace(".", "")[:3] + + async def handle_game_init_request(self, data: Dict) -> Dict: + return f"" diff --git a/titles/fgoa/config.py b/titles/fgoa/config.py new file mode 100644 index 0000000..a16d54e --- /dev/null +++ b/titles/fgoa/config.py @@ -0,0 +1,24 @@ +from core.config import CoreConfig + + +class FGOAServerConfig: + def __init__(self, parent: "FGOAConfig") -> None: + self.__config = parent + + @property + def enable(self) -> bool: + return CoreConfig.get_config_field( + self.__config, "fgo", "server", "enable", default=True + ) + + @property + def loglevel(self) -> int: + return CoreConfig.str_to_loglevel( + CoreConfig.get_config_field( + self.__config, "fgo", "server", "loglevel", default="info" + ) + ) + +class FGOAConfig(dict): + def __init__(self) -> None: + self.server = FGOAServerConfig(self) diff --git a/titles/fgoa/const.py b/titles/fgoa/const.py new file mode 100644 index 0000000..377a7c5 --- /dev/null +++ b/titles/fgoa/const.py @@ -0,0 +1,14 @@ +class FGOAConstants(): + GAME_CODE = "SDEJ" + + CONFIG_NAME = "fgoa.yaml" + + VER_FGOA_SEASON_1 = 0 + + VERSION_STRING = ( + "Fate/Grand Order Arcade", + ) + + @classmethod + def game_ver_to_string(cls, ver: int): + return cls.VERSION_STRING[ver] diff --git a/titles/fgoa/index.py b/titles/fgoa/index.py new file mode 100644 index 0000000..b1d5a22 --- /dev/null +++ b/titles/fgoa/index.py @@ -0,0 +1,123 @@ +import json +import inflection +import yaml +import string +import logging +import coloredlogs +import zlib +import base64 +import urllib.parse + +from os import path +from typing import Dict, List, Tuple +from logging.handlers import TimedRotatingFileHandler + +from starlette.routing import Route +from starlette.responses import Response +from starlette.requests import Request +from starlette.responses import PlainTextResponse + +from core.config import CoreConfig +from core.title import BaseServlet +from core.utils import Utils + +from titles.fgoa.base import FGOABase +from titles.fgoa.config import FGOAConfig +from titles.fgoa.const import FGOAConstants + +class FGOAServlet(BaseServlet): + def __init__(self, core_cfg: CoreConfig, cfg_dir: str) -> None: + self.core_cfg = core_cfg + self.game_cfg = FGOAConfig() + if path.exists(f"{cfg_dir}/{FGOAConstants.CONFIG_NAME}"): + self.game_cfg.update( + yaml.safe_load(open(f"{cfg_dir}/{FGOAConstants.CONFIG_NAME}")) + ) + + self.versions = [ + FGOABase(core_cfg, self.game_cfg), + ] + + self.logger = logging.getLogger("fgoa") + log_fmt_str = "[%(asctime)s] FGOA | %(levelname)s | %(message)s" + log_fmt = logging.Formatter(log_fmt_str) + fileHandler = TimedRotatingFileHandler( + "{0}/{1}.log".format(self.core_cfg.server.log_dir, "fgoa"), + 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 + ) + + @classmethod + def is_game_enabled(cls, game_code: str, core_cfg: CoreConfig, cfg_dir: str) -> bool: + game_cfg = FGOAConfig() + + if path.exists(f"{cfg_dir}/{FGOAConstants.CONFIG_NAME}"): + game_cfg.update( + yaml.safe_load(open(f"{cfg_dir}/{FGOAConstants.CONFIG_NAME}")) + ) + + if not game_cfg.server.enable: + return False + + return True + + def get_routes(self) -> List[Route]: + return [ + Route("/SDEJ/{version:int}/{endpoint:str}", self.render_POST, 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}", self.core_cfg.server.hostname) + + return (f"http://{self.core_cfg.server.hostname}/{game_code}/{game_ver}", self.core_cfg.server.hostname) + + + async def render_POST(self, request: Request) -> bytes: + version: int = request.path_params.get('version') + endpoint: str = request.path_params.get('endpoint') + req_raw = await request.body() + internal_ver = 0 + client_ip = Utils.get_ip_addr(request) + + 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 + # doing encrypted. The likelyhood of false positives is low but + # technically not 0 + self.logger.error("Encryption not supported at this time") + + self.logger.debug(req_raw) + + self.logger.info(f"v{version} {endpoint} request from {client_ip}") + + func_to_find = "handle_" + inflection.underscore(endpoint) + "_request" + + try: + handler = getattr(self.versions[internal_ver], func_to_find) + resp = await handler(req_raw) + + except Exception as e: + self.logger.error(f"Error handling v{version} method {endpoint} - {e}") + raise + return Response(zlib.compress(b'{"stat": "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"))) \ No newline at end of file