pgw00g/artemis
1
0
Fork 0
forked from Dniel97/artemis
Code Pull Requests Activity

copy allnet functionality over

Browse Source
This commit is contained in:
Hay1tsme 2023-02-18 00:00:30 -05:00
parent abb25aa328
commit 8616c6d064
5 changed files with 190 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

193
core/allnet.py
View File

@ -1,4 +1,4 @@
from typing import Dict, List, Any, Optional
from typing import Dict, List, Any, Optional, Tuple
import logging, coloredlogs
from logging.handlers import TimedRotatingFileHandler
from twisted.web.http import Request
@ -6,6 +6,10 @@ from datetime import datetime
import pytz
import base64
import zlib
from Crypto.PublicKey import RSA
from Crypto.Hash import SHA
from Crypto.Signature import PKCS1_v1_5
from time import strptime
from core.config import CoreConfig
from core.data import Data
@ -17,6 +21,7 @@ class AllnetServlet:
self.config = core_cfg
self.config_folder = cfg_folder
self.data = Data(core_cfg)
self.uri_registry: Dict[str, Tuple[str, str]] = {}
self.logger = logging.getLogger("allnet")
if not hasattr(self.logger, "initialized"):
@ -36,28 +41,194 @@ class AllnetServlet:
coloredlogs.install(level=core_cfg.allnet.loglevel, logger=self.logger, fmt=log_fmt_str)
self.logger.initialized = True
if "game_registry" not in globals():
globals()["game_registry"] = Utils.get_all_titles()
plugins = Utils.get_all_titles()
if len(globals()["game_registry"]) == 0:
if len(plugins) == 0:
self.logger.error("No games detected!")
for _, mod in plugins.items():
for code in mod.game_codes:
if hasattr(mod, "use_default_title") and mod.use_default_title:
if hasattr(mod, "include_protocol") and mod.include_protocol:
if hasattr(mod, "title_secure") and mod.title_secure:
uri = "https://"
else:
uri = "http://"
else:
uri = ""
if core_cfg.server.is_develop:
uri += f"{core_cfg.title.hostname}:{core_cfg.title.port}"
else:
uri += f"{core_cfg.title.hostname}"
uri += f"/{code}/$v"
if hasattr(mod, "trailing_slash") and mod.trailing_slash:
uri += "/"
else:
if hasattr(mod, "uri"):
uri = mod.uri
else:
uri = ""
if hasattr(mod, "host"):
host = mod.host
elif hasattr(mod, "use_default_host") and mod.use_default_host:
if core_cfg.server.is_develop:
host = f"{core_cfg.title.hostname}:{core_cfg.title.port}"
else:
host = f"{core_cfg.title.hostname}"
else:
host = ""
self.uri_registry[code] = (uri, host)
self.logger.info(f"Allnet serving {len(self.uri_registry)} games on port {core_cfg.allnet.port}")
def handle_poweron(self, request: Request):
request_ip = request.getClientAddress().host
try:
req = AllnetPowerOnRequest(self.allnet_req_to_dict(request.content.getvalue()))
# Validate the request. Currently we only validate the fields we plan on using
if not req.game_id or not req.ver or not req.token or not req.serial or not req.ip:
raise AllnetRequestException(f"Bad request params {vars(req)}")
raise AllnetRequestException(f"Bad auth request params from {request_ip} - {vars(req)}")
except AllnetRequestException as e:
self.logger.error(e)
return b""
if req.format_ver == 3:
resp = AllnetPowerOnResponse3(req.token)
else:
resp = AllnetPowerOnResponse2()
if req.game_id not in self.uri_registry:
msg = f"Unrecognised game {req.game_id} attempted allnet auth from {request_ip}."
self.data.base.log_event("allnet", "ALLNET_AUTH_UNKNOWN_GAME", logging.WARN, msg)
self.logger.warn(msg)
resp.stat = 0
return self.dict_to_http_form_string([vars(resp)])
resp.uri, resp.host = self.uri_registry[req.game_id]
machine = self.data.arcade.get_machine(req.serial)
if machine is None and not self.config.server.allow_unregistered_serials:
msg = f"Unrecognised serial {req.serial} attempted allnet auth from {request_ip}."
self.data.base.log_event("allnet", "ALLNET_AUTH_UNKNOWN_SERIAL", logging.WARN, msg)
self.logger.warn(msg)
resp.stat = 0
return self.dict_to_http_form_string([vars(resp)])
if machine is not None:
arcade = self.data.arcade.get_arcade(machine["arcade"])
req.country = arcade["country"] if machine["country"] is None else machine["country"]
req.place_id = arcade["id"]
req.allnet_id = machine["id"]
req.name = arcade["name"]
req.nickname = arcade["nickname"]
req.region0 = arcade["region_id"]
req.region_name0 = arcade["country"]
req.region_name1 = arcade["state"]
req.region_name2 = arcade["city"]
req.client_timezone = arcade["timezone"] if arcade["timezone"] is not None else "+0900"
msg = f"{req.serial} authenticated from {request_ip}: {req.game_id} v{req.ver}"
self.data.base.log_event("allnet", "ALLNET_AUTH_SUCCESS", logging.INFO, msg)
self.logger.info(msg)
return self.dict_to_http_form_string([vars(resp)])
def handle_dlorder(self, request: Request):
request_ip = request.getClientAddress().host
try:
req = AllnetDownloadOrderRequest(self.allnet_req_to_dict(request.content.getvalue()))
# Validate the request. Currently we only validate the fields we plan on using
if not req.game_id or not req.ver or not req.token or not req.serial or not req.ip:
raise AllnetRequestException(f"Bad auth request params from {request_ip} - {vars(req)}")
except AllnetRequestException as e:
self.logger.error(e)
return b""
def handle_dlorder(self, request: Request):
pass
resp = AllnetDownloadOrderResponse()
if not self.config.allnet.allow_online_updates:
return self.dict_to_http_form_string(vars(resp))
else: # TODO: Actual dlorder response
return self.dict_to_http_form_string(vars(resp))
def handle_billing_request(self, request: Request):
pass
req_dict = self.billing_req_to_dict(request.content.getvalue())
request_ip = request.getClientAddress()
if req_dict is None:
self.logger.error(f"Failed to parse request {request.content.getvalue()}")
return b""
self.logger.debug(f"request {req_dict}")
rsa = RSA.import_key(open(self.config.billing.sign_key, 'rb').read())
signer = PKCS1_v1_5.new(rsa)
digest = SHA.new()
kc_playlimit = int(req_dict[0]["playlimit"])
kc_nearfull = int(req_dict[0]["nearfull"])
kc_billigtype = int(req_dict[0]["billingtype"])
kc_playcount = int(req_dict[0]["playcnt"])
kc_serial: str = req_dict[0]["keychipid"]
kc_game: str = req_dict[0]["gameid"]
kc_date = strptime(req_dict[0]["date"], "%Y%m%d%H%M%S")
kc_serial_bytes = kc_serial.encode()
machine = self.data.arcade.get_machine(kc_serial)
if machine is None and not self.config.server.allow_unregistered_serials:
msg = f"Unrecognised serial {kc_serial} attempted billing checkin from {request_ip} for game {kc_game}."
self.data.base.log_event("allnet", "BILLING_CHECKIN_NG_SERIAL", logging.WARN, msg)
self.logger.warn(msg)
resp = BillingResponse("", "", "", "")
resp.result = "1"
return self.dict_to_http_form_string([vars(resp)])
msg = f"Billing checkin from {request.getClientIP()}: game {kc_game} keychip {kc_serial} playcount " \
f"{kc_playcount} billing_type {kc_billigtype} nearfull {kc_nearfull} playlimit {kc_playlimit}"
self.logger.info(msg)
self.data.base.log_event('billing', 'BILLING_CHECKIN_OK', logging.INFO, msg)
while kc_playcount > kc_playlimit:
kc_playlimit += 1024
kc_nearfull += 1024
playlimit = kc_playlimit
nearfull = kc_nearfull + (kc_billigtype * 0x00010000)
digest.update(playlimit.to_bytes(4, 'little') + kc_serial_bytes)
playlimit_sig = signer.sign(digest).hex()
digest = SHA.new()
digest.update(nearfull.to_bytes(4, 'little') + kc_serial_bytes)
nearfull_sig = signer.sign(digest).hex()
# TODO: playhistory
resp = BillingResponse(playlimit, playlimit_sig, nearfull, nearfull_sig)
resp_str = self.dict_to_http_form_string([vars(resp)])
if resp_str is None:
self.logger.error(f"Failed to parse response {vars(resp)}")
self.logger.debug(f"response {vars(resp)}")
return resp_str.encode("utf-8")
def kvp_to_dict(self, *kvp: str) -> List[Dict[str, Any]]:
ret: List[Dict[str, Any]] = []
@ -207,7 +378,7 @@ class AllnetDownloadOrderResponse():
self.uri = uri
class BillingResponse():
def __init__(self, playlimit: str, playlimit_sig: str, nearfull: str, nearfull_sig: str,
def __init__(self, playlimit: str = "", playlimit_sig: str = "", nearfull: str = "", nearfull_sig: str = "",
playhistory: str = "000000/0:000000/0:000000/0") -> None:
self.result = "0"
@ -226,4 +397,6 @@ class BillingResponse():
# YYYY -> 4 digit year, MM -> 2 digit month, C -> Playcount during that period
class AllnetRequestException(Exception):
pass
def __init__(self, message="Allnet Request Error") -> None:
self.message = message
super().__init__(self.message)

4
core/config.py
View File

@ -14,8 +14,8 @@ class ServerConfig:
return CoreConfig.get_config_field(self.__config, 'core', 'server', 'allow_user_registration', default=True)
@property
def allow_unregistered_games(self) -> bool:
return CoreConfig.get_config_field(self.__config, 'core', 'server', 'allow_unregistered_games', default=True)
def allow_unregistered_serials(self) -> bool:
return CoreConfig.get_config_field(self.__config, 'core', 'server', 'allow_unregistered_serials', default=True)
@property
def name(self) -> str:

7
core/title.py
View File

@ -33,10 +33,9 @@ class TitleServlet():
coloredlogs.install(level=core_cfg.title.loglevel, logger=self.logger, fmt=log_fmt_str)
self.logger.initialized = True
if "game_registry" not in globals():
globals()["game_registry"] = Utils.get_all_titles()
plugins = Utils.get_all_titles()
for folder, mod in globals()["game_registry"].items():
for folder, mod in plugins.items():
if hasattr(mod, "game_codes") and hasattr(mod, "index"):
handler_cls = mod.index(self.config, self.config_folder)
if hasattr(handler_cls, "setup"):
@ -48,7 +47,7 @@ class TitleServlet():
else:
self.logger.error(f"{folder} missing game_code or index in __init__.py")
self.logger.info(f"Serving {len(globals()['game_registry'])} game codes")
self.logger.info(f"Serving {len(self.title_registry)} game codes on port {core_cfg.title.port}")
def render_GET(self, request: Request, endpoints: dict) -> bytes:
code = endpoints["game"]

2
docs/config.md
View File

@ -2,7 +2,7 @@
## Server
- `listen_address`: IP Address or hostname that the server will listen for connections on. Set to 127.0.0.1 for local only, or 0.0.0.0 for all interfaces. Default `127.0.0.1`
- `allow_user_registration`: Allows users to register in-game via the AimeDB `register` function. Disable to be able to control who can use cards on your server. Default `True`
- `allow_unregistered_games`: Allows games that do not have registered keychips to connect and authenticate. Disable to restrict who can connect to your server. Recomended to disable for production setups. Default `True`
- `allow_unregistered_serials`: Allows games that do not have registered keychips to connect and authenticate. Disable to restrict who can connect to your server. Recomended to disable for production setups. Default `True`
- `name`: Name for the server, used by some games in their default MOTDs. Default `ARTEMiS`
- `is_develop`: Flags that the server is a development instance without a proxy standing in front of it. Setting to `False` tells the server not to listen for SSL, because the proxy should be handling all SSL-related things, among other things. Default `True`
- `log_dir`: Directory to store logs. Server MUST have read and write permissions to this directory or you will have issues. Default `logs`

2
example_config/core.yaml
View File

@ -1,7 +1,7 @@
server:
listen_address: "127.0.0.1"
allow_user_registration: True
allow_unregistered_games: True
allow_unregistered_serials: True
name: "ARTEMiS"
is_develop: True
log_dir: "logs"