Hay1tsme/artemis
30
52
Fork 48
Code Issues 5 Pull Requests 1 Packages Projects Releases Wiki Activity

add remaining titles sans CXB

Browse Source
This commit is contained in:
Hay1tsme
2023-10-20 01:09:44 -04:00
parent 179146d676
commit 270703a2ad
6 changed files with 119 additions and 118 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
core/mucha.py
View File

@ -38,8 +38,8 @@ class MuchaServlet:
coloredlogs.install(level=cfg.mucha.loglevel, logger=self.logger, fmt=log_fmt_str)
for _, mod in TitleServlet.title_registry.items():
if hasattr(mod, "index") and hasattr(mod.index, "get_mucha_info"):
enabled, game_cd = mod.index.get_mucha_info(
if hasattr(mod, "get_mucha_info"):
enabled, game_cd = mod.get_mucha_info(
self.config, self.config_dir
)
if enabled:

4
titles/mai2/index.py
View File

@ -157,7 +157,7 @@ class Mai2Servlet(BaseServlet):
def handle_mai2(self, request: Request, game_code: int, matchers: Dict) -> bytes:
endpoint = matchers['endpoint']
version = matchers['version']
version = int(matchers['version'])
if endpoint.lower() == "ping":
return zlib.compress(b'{"returnCode": "1"}')
@ -165,7 +165,7 @@ class Mai2Servlet(BaseServlet):
internal_ver = 0
client_ip = Utils.get_ip_addr(request)
if request.uri.startswith(b"/SDEZ"):
if game_code == "SDEZ":
if version < 105: # 1.0
internal_ver = Mai2Constants.VER_MAIMAI_DX
elif version >= 105 and version < 110: # PLUS

58
titles/ongeki/index.py
View File

@ -12,25 +12,26 @@ from Crypto.Util.Padding import pad
from Crypto.Protocol.KDF import PBKDF2
from Crypto.Hash import SHA1
from os import path
from typing import Tuple
from typing import Tuple, Dict, List
from core.config import CoreConfig
from core.utils import Utils
from titles.ongeki.config import OngekiConfig
from titles.ongeki.const import OngekiConstants
from titles.ongeki.base import OngekiBase
from titles.ongeki.plus import OngekiPlus
from titles.ongeki.summer import OngekiSummer
from titles.ongeki.summerplus import OngekiSummerPlus
from titles.ongeki.red import OngekiRed
from titles.ongeki.redplus import OngekiRedPlus
from titles.ongeki.bright import OngekiBright
from titles.ongeki.brightmemory import OngekiBrightMemory
from core.title import BaseServlet
from .config import OngekiConfig
from .const import OngekiConstants
from .base import OngekiBase
from .plus import OngekiPlus
from .summer import OngekiSummer
from .summerplus import OngekiSummerPlus
from .red import OngekiRed
from .redplus import OngekiRedPlus
from .bright import OngekiBright
from .brightmemory import OngekiBrightMemory
class OngekiServlet:
class OngekiServlet(BaseServlet):
def __init__(self, core_cfg: CoreConfig, cfg_dir: str) -> None:
self.core_cfg = core_cfg
super().__init__(core_cfg, cfg_dir)
self.game_cfg = OngekiConfig()
self.hash_table: Dict[Dict[str, str]] = {}
if path.exists(f"{cfg_dir}/{OngekiConstants.CONFIG_NAME}"):
@ -106,9 +107,7 @@ class OngekiServlet:
)
@classmethod
def get_allnet_info(
cls, game_code: str, core_cfg: CoreConfig, cfg_dir: str
) -> Tuple[bool, str, str]:
def is_game_enabled(cls, game_code: str, core_cfg: CoreConfig, cfg_dir: str) -> bool:
game_cfg = OngekiConfig()
if path.exists(f"{cfg_dir}/{OngekiConstants.CONFIG_NAME}"):
@ -117,30 +116,31 @@ class OngekiServlet:
)
if not game_cfg.server.enable:
return (False, "", "")
if core_cfg.server.is_develop:
return False
return True
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() != 80:
return (
True,
f"http://{core_cfg.title.hostname}:{core_cfg.title.port}/{game_code}/$v/",
f"{core_cfg.title.hostname}:{core_cfg.title.port}/",
f"http://{self.core_cfg.title.hostname}:{self.core_cfg.title.port}/{game_code}/$v/",
f"{self.core_cfg.title.hostname}:{self.core_cfg.title.port}/",
)
return (
True,
f"http://{core_cfg.title.hostname}/{game_code}/$v/",
f"{core_cfg.title.hostname}/",
f"http://{self.core_cfg.title.hostname}/{game_code}/$v/",
f"{self.core_cfg.title.hostname}/",
)
def render_POST(self, request: Request, version: int, url_path: str) -> bytes:
if url_path.lower() == "ping":
def render_POST(self, request: Request, game_code: int, matchers: Dict) -> bytes:
endpoint = matchers['endpoint']
version = matchers['version']
if endpoint.lower() == "ping":
return zlib.compress(b'{"returnCode": 1}')
req_raw = request.content.getvalue()
url_split = url_path.split("/")
encrtped = False
internal_ver = 0
endpoint = url_split[len(url_split) - 1]
client_ip = Utils.get_ip_addr(request)
if version < 105: # 1.0

52
titles/pokken/index.py
View File

@ -1,4 +1,4 @@
from typing import Tuple
from typing import Tuple, List, Dict
from twisted.web.http import Request
from twisted.web import resource
from twisted.internet import reactor
@ -12,6 +12,7 @@ from os import path
from google.protobuf.message import DecodeError
from core import CoreConfig, Utils
from core.title import BaseServlet
from .config import PokkenConfig
from .base import PokkenBase
from .const import PokkenConstants
@ -19,10 +20,9 @@ from .proto import jackal_pb2
from .services import PokkenAdmissionFactory
class PokkenServlet(resource.Resource):
class PokkenServlet(BaseServlet):
def __init__(self, core_cfg: CoreConfig, cfg_dir: str) -> None:
self.isLeaf = True
self.core_cfg = core_cfg
super().__init__(core_cfg, cfg_dir)
self.config_dir = cfg_dir
self.game_cfg = PokkenConfig()
if path.exists(f"{cfg_dir}/pokken.yaml"):
@ -56,9 +56,7 @@ class PokkenServlet(resource.Resource):
self.base = PokkenBase(core_cfg, self.game_cfg)
@classmethod
def get_allnet_info(
cls, game_code: str, core_cfg: CoreConfig, cfg_dir: str
) -> Tuple[bool, str, str]:
def is_game_enabled(cls, game_code: str, core_cfg: CoreConfig, cfg_dir: str) -> bool:
game_cfg = PokkenConfig()
if path.exists(f"{cfg_dir}/{PokkenConstants.CONFIG_NAME}"):
@ -67,18 +65,31 @@ class PokkenServlet(resource.Resource):
)
if not game_cfg.server.enable:
return (False, "", "")
return False
return True
def get_endpoint_matchers(self) -> Tuple[List[Tuple[str, str, Dict]], List[Tuple[str, str, Dict]]]:
return (
True,
f"https://{game_cfg.server.hostname}:{game_cfg.ports.game}/{game_code}/$v/",
f"{game_cfg.server.hostname}/SDAK/$v/",
[],
[
("render_POST", "/pokken/", {}),
("handle_matching", "/pokken/matching", {}),
]
)
def get_allnet_info(self, game_code: str, game_ver: int, keychip: str) -> Tuple[str, str]:
if self.game_cfg.ports.game != 443:
return (
f"https://{self.game_cfg.server.hostname}:{self.game_cfg.ports.game}/pokken/",
f"{self.game_cfg.server.hostname}/SDAK/$v/",
)
return (
f"https://{self.game_cfg.server.hostname}/pokken/",
f"{self.game_cfg.server.hostname}/pokken/",
)
@classmethod
def get_mucha_info(
cls, core_cfg: CoreConfig, cfg_dir: str
) -> Tuple[bool, str, str]:
def get_mucha_info(self, core_cfg: CoreConfig, cfg_dir: str) -> Tuple[bool, str]:
game_cfg = PokkenConfig()
if path.exists(f"{cfg_dir}/{PokkenConstants.CONFIG_NAME}"):
@ -97,12 +108,7 @@ class PokkenServlet(resource.Resource):
self.game_cfg.ports.admission, PokkenAdmissionFactory(self.core_cfg, self.game_cfg)
)
def render_POST(
self, request: Request, version: int = 0, endpoints: str = ""
) -> bytes:
if endpoints == "matching":
return self.handle_matching(request)
def render_POST(self, request: Request, game_code: int, matchers: Dict) -> bytes:
content = request.content.getvalue()
if content == b"":
self.logger.info("Empty request")
@ -131,7 +137,7 @@ class PokkenServlet(resource.Resource):
ret = handler(pokken_request)
return ret
def handle_matching(self, request: Request) -> bytes:
def handle_matching(self, request: Request, game_code: int, matchers: Dict) -> bytes:
if not self.game_cfg.server.enable_matching:
return b""

46
titles/sao/index.py
View File

@ -1,4 +1,4 @@
from typing import Tuple
from typing import Tuple, Dict
from twisted.web.http import Request
from twisted.web import resource
import json, ast
@ -10,16 +10,16 @@ import inflection
from os import path
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(resource.Resource):
class SaoServlet(BaseServlet):
def __init__(self, core_cfg: CoreConfig, cfg_dir: str) -> None:
self.isLeaf = True
self.core_cfg = core_cfg
super().__init__(core_cfg, cfg_dir)
self.config_dir = cfg_dir
self.game_cfg = SaoConfig()
if path.exists(f"{cfg_dir}/sao.yaml"):
@ -53,47 +53,33 @@ class SaoServlet(resource.Resource):
self.base = SaoBase(core_cfg, self.game_cfg)
@classmethod
def get_allnet_info(
cls, game_code: str, core_cfg: CoreConfig, cfg_dir: str
) -> Tuple[bool, str, str]:
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 False
return True
def get_allnet_info(self, game_code: str, game_ver: int, keychip: str) -> Tuple[str, str]:
return (
True,
f"http://{game_cfg.server.hostname}:{game_cfg.server.port}/{game_code}/$v/",
f"{game_cfg.server.hostname}/SDEW/$v/",
f"http://{self.game_cfg.server.hostname}:{self.game_cfg.server.port}/{game_code}/$v/",
f"{self.game_cfg.server.hostname}/SDEW/$v/",
)
@classmethod
def get_mucha_info(
cls, core_cfg: CoreConfig, cfg_dir: str
) -> Tuple[bool, str, str]:
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:
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, "SAO1")
def setup(self) -> None:
pass
def render_POST(
self, request: Request, version: int = 0, endpoints: str = ""
) -> bytes:
def render_POST(self, request: Request, game_code: int, matchers: Dict) -> bytes:
req_url = request.uri.decode()
if req_url == "/matching":
self.logger.info("Matching request")

73
titles/wacca/index.py
View File

@ -5,20 +5,20 @@ import logging
import json
from hashlib import md5
from twisted.web.http import Request
from typing import Dict, Tuple
from typing import Dict, Tuple, List
from os import path
from core import CoreConfig, Utils
from titles.wacca.config import WaccaConfig
from titles.wacca.config import WaccaConfig
from titles.wacca.const import WaccaConstants
from titles.wacca.reverse import WaccaReverse
from titles.wacca.lilyr import WaccaLilyR
from titles.wacca.lily import WaccaLily
from titles.wacca.s import WaccaS
from titles.wacca.base import WaccaBase
from titles.wacca.handlers.base import BaseResponse
from titles.wacca.handlers.helpers import Version
from .config import WaccaConfig
from .config import WaccaConfig
from .const import WaccaConstants
from .reverse import WaccaReverse
from .lilyr import WaccaLilyR
from .lily import WaccaLily
from .s import WaccaS
from .base import WaccaBase
from .handlers.base import BaseResponse
from .handlers.helpers import Version
class WaccaServlet:
@ -61,10 +61,15 @@ class WaccaServlet:
level=self.game_cfg.server.loglevel, logger=self.logger, fmt=log_fmt_str
)
def get_endpoint_matchers(self) -> Tuple[List[Tuple[str, str, Dict]], List[Tuple[str, str, Dict]]]:
return (
[],
[("render_POST", "/WaccaServlet/api/{api}/{endpoint}", {}),
("render_POST", "/WaccaServlet/api/{api}/{branch}/{endpoint}", {}),]
)
@classmethod
def get_allnet_info(
cls, game_code: str, core_cfg: CoreConfig, cfg_dir: str
) -> Tuple[bool, str, str]:
def is_game_enabled(cls, game_code: str, core_cfg: CoreConfig, cfg_dir: str) -> bool:
game_cfg = WaccaConfig()
if path.exists(f"{cfg_dir}/{WaccaConstants.CONFIG_NAME}"):
game_cfg.update(
@ -72,27 +77,43 @@ class WaccaServlet:
)
if not game_cfg.server.enable:
return (False, "", "")
return False
if core_cfg.server.is_develop:
return True
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 (
True,
f"http://{core_cfg.title.hostname}:{core_cfg.title.port}/{game_code}/$v",
f"http://{self.core_cfg.title.hostname}:{self.core_cfg.title.port}/{game_code}/$v",
"",
)
return (True, f"http://{core_cfg.title.hostname}/{game_code}/$v", "")
def render_POST(self, request: Request, version: int, url_path: str) -> bytes:
return (True, f"http://{self.core_cfg.title.hostname}/{game_code}/$v", "")
def render_POST(self, request: Request, game_code: int, matchers: Dict) -> bytes:
def end(resp: Dict) -> bytes:
hash = md5(json.dumps(resp, ensure_ascii=False).encode()).digest()
request.responseHeaders.addRawHeader(b"X-Wacca-Hash", hash.hex().encode())
return json.dumps(resp).encode()
api = matchers['api']
branch = matchers.get('branch', '')
endpoint = matchers['endpoint']
client_ip = Utils.get_ip_addr(request)
if branch:
url_path = f"{api}/{branch}/{endpoint}"
func_to_find = f"handle_{api}_{branch}_{endpoint}_request"
else:
url_path = f"{api}/{endpoint}"
func_to_find = f"handle_{api}_{endpoint}_request"
try:
req_json = json.loads(request.content.getvalue())
version_full = Version(req_json["appVersion"])
except Exception:
self.logger.error(
f"Failed to parse request to {url_path} -> {request.content.getvalue()}"
@ -102,18 +123,6 @@ class WaccaServlet:
resp.message = "不正なリクエスト エラーです"
return end(resp.make())
if "api/" in url_path:
func_to_find = (
"handle_" + url_path.partition("api/")[2].replace("/", "_") + "_request"
)
else:
self.logger.error(f"Malformed url {url_path}")
resp = BaseResponse()
resp.status = 1
resp.message = "Bad URL"
return end(resp.make())
ver_search = int(version_full)
if ver_search < 15000: