Massive refactor

v8_server/eamuse/services/pcbtracker.py

@@ -0,0 +1,25 @@
+from lxml.builder import E
+
+
+class PCBTracker(object):
+    """
+    Handle the PCBTracker request.
+
+    The only method of note is the "alive" method which returns whether PASELI should be
+    active or not for this session.
+
+    Example:
+        <call model="K32:J:B:A:2011033000" srcid="00010203040506070809">
+            <pcbtracker hardid="010074D435AAD895" method="alive" softid=""/>
+        </call>
+    """
+
+    # We don't support PASELI on GFDM: V8
+    PASELI_ACTIVE = False
+
+    # Methods
+    ALIVE = "alive"
+
+    @classmethod
+    def alive(cls):
+        return E.response(E.pcbtracker({"ecenable": "1" if cls.PASELI_ACTIVE else "0"}))

v8_server/eamuse/services/services.py

@@ -0,0 +1,255 @@
+from __future__ import annotations
+
+import logging
+from binascii import unhexlify
+from datetime import datetime
+from enum import IntEnum
+from random import randint
+from time import time
+from typing import Any, Callable, Dict, Optional, Tuple, Union
+
+from flask import Flask, Request
+from kbinxml import KBinXML
+from lxml import etree
+from lxml.builder import E
+from lxml.etree import _Element as eElement
+
+from v8_server import LOG_PATH
+from v8_server.eamuse.utils.arc4 import EAmuseARC4
+from v8_server.eamuse.utils.eamuse import Model
+from v8_server.eamuse.utils.lz77 import Lz77
+from v8_server.eamuse.utils.xml import get_xml_attrib, get_xml_tag
+
+
+# We want a general logger, and a special logger to log requests separately
+logger = logging.getLogger(__name__)
+rlogger = logging.getLogger("requests")
+
+
+# Use an Enum for the different services for minimal xml
+class ServiceType(IntEnum):
+    PCBTRACKER = 1
+    MESSAGE = 2
+    PCBEVENT = 3
+    FACILITY = 4
+    PACKAGE = 5
+    CARDMNG = 6
+    LOCAL = 7
+
+
+class Services(object):
+    """
+    Handles a service request from the eAmuse Server
+
+    Example:
+        <call model="K32:J:B:A:2011033000" srcid="00010203040506070809">
+            <services method="get"/>
+        </call>
+    """
+
+    # Default service url that GFDM uses. You will need to set up your network so that
+    # this URL points to this server.
+    SERVICE_URL = "https://eamuse.konami.fun"
+
+    # The base route that GFDM uses to query the eAmuse server to get the list of
+    # offered services
+    SERVICES_ROUTE = "/service/services/services/"
+
+    # Default NTP url
+    # XXX: Maybe needs to be configurable? What if the machine this is running on
+    # doesn't have internet access?
+    NTP_URL = "ntp//pool.ntp.org"
+
+    def __init__(
+        self,
+        expire: int = 600,
+        method: str = "get",
+        mode: str = "operation",
+        status: int = 0,
+    ) -> None:
+        self.expire = expire
+        self.method = method
+        self.mode = mode
+        self.status = status
+        self.services = self.generate_services()
+
+    def get_services(self) -> etree:
+        return E.response(
+            E.services(
+                expire=str(self.expire),
+                method=self.method,
+                mode=self.mode,
+                status=str(self.status),
+                *[E.item({"name": k, "url": self.services[k]}) for k in self.services],
+            )
+        )
+
+    def generate_services(self) -> Dict[str, str]:
+        # The Localhost IP Address
+        ip = "127.0.0.1"
+        services = {
+            "ntp": self.NTP_URL,
+            "keepalive": (
+                f"{self.SERVICE_URL}/keepalive?"
+                f"pa={ip}&ia={ip}&ga={ip}&ma={ip}&t1=2&t2=10"
+            ),
+            **{
+                n.lower(): f"{self.SERVICE_URL}/{m.value}"
+                for n, m in ServiceType.__members__.items()
+            },
+        }
+        return services
+
+    @classmethod
+    def route(cls, app: Flask, _type: ServiceType) -> Callable[[Any], Any]:
+        def decorator(f):
+            func = app.route(f"/{_type.value}", methods=["POST"])(f)
+            return func
+
+        return decorator
+
+    def __repr__(self) -> str:
+        return (
+            f"Services<expire: {self.expire}, "
+            f'method: "{self.method}", mode: "{self.mode}", status: {self.status}, '
+            f"services: {list(self.services.keys())}>"
+        )
+
+
+class ServiceRequest(object):
+    # eAmuse Header tags we care about
+    X_EAMUSE_INFO = "x-eamuse-info"
+    X_COMPRESS = "x-compress"
+
+    # The encoding that we will use for this data
+    ENCODING = "UTF-8"
+
+    # Log dir for the requests
+    LOG_DIR = LOG_PATH / "requests"
+
+    def __init__(self, request: Request) -> None:
+        # Save the request so we can refer back to it
+        self._request = request
+
+        self.model: Optional[Model] = None
+        self.module: Optional[str] = None
+        self.method: Optional[str] = None
+        self.command: Optional[str] = None
+        self.encrypted = self.X_EAMUSE_INFO in request.headers
+        self.compression = (
+            request.headers[self.X_COMPRESS]
+            if self.X_COMPRESS in request.headers
+            else "none"
+        )
+        self.compressed = self.compression != "none"
+
+        # Parse the request data
+        self.xml = self.read()
+
+    def read(self) -> etree:
+        # Lets grab the raw data from the request
+        xml_bin = self._request.data
+
+        # Decrypt the data if necessary
+        x_eamuse_info: Optional[str] = None
+        if self.encrypted:
+            x_eamuse_info, key = self._get_encryption_data()
+            xml_bin = EAmuseARC4(key).decrypt(xml_bin)
+
+        # De-Compress the data if necessary
+        # Right now we only support `lz77`
+        if self.compressed and self.compression == "lz77":
+            xml_bin = Lz77().decompress(xml_bin)
+
+        # Convert the binary xml data to text bytes, and save a copy
+        xml_bytes = KBinXML(xml_bin).to_text().encode(self.ENCODING)
+        self._save_xml(xml_bytes, "req", x_eamuse_info)
+
+        # Convert the XML text to an eTree
+        xml_root = etree.fromstring(xml_bytes)
+
+        # Grab the common xml information that we need
+        # <call model="_MODEL_" srcid="_SRCID_">
+        #     <_MODULE_ method="_METHOD_" command="_COMMAND_">
+        # </call>
+
+        # First grab the model
+        self.model = Model.from_modelstring(get_xml_attrib(xml_root, "model"))
+
+        # The module is the first child of the root
+        module = xml_root[0]
+        self.module = get_xml_tag(module)
+        self.method = get_xml_attrib(module, "method")
+        self.command = get_xml_attrib(module, "command")
+
+        rlogger.info(self.__repr__())
+        rlogger.debug(f"Request:\n {xml_bytes.decode(self.ENCODING)}")
+        return xml_root
+
+    def response(self, xml_bytes: Union[bytes, eElement]):
+        # Firstly, let's make sure xml_bytes is a bytes object
+        if type(xml_bytes) == eElement:
+            xml_bytes = etree.tostring(xml_bytes, pretty_print=True)
+
+        # Generate our own encryption key
+        x_eamuse_info, key = self._make_encryption_data()
+
+        # Save our xml response
+        self._save_xml(xml_bytes, "resp", x_eamuse_info if self.encrypted else None)
+
+        # Convert our xml to binary
+        xml_bin = KBinXML(xml_bytes).to_binary()
+
+        # Common Headers
+        headers = {
+            "Content-Type": "application/octet_stream",
+            "Server": "Microsoft-HTTPAPI/2.0",
+        }
+
+        # Compress the data if necessary
+        # Right now we only support `lz77`
+        if self.compressed and self.compression == "lz77":
+            xml_bin = Lz77().compress(xml_bin)
+            headers[self.X_COMPRESS] = "lz77"
+
+        # Encrypt the data if necessary
+        if self.encrypted:
+            xml_bin = EAmuseARC4(key).encrypt(xml_bin)
+            headers[self.X_EAMUSE_INFO] = x_eamuse_info
+
+        rlogger.debug(f"Response:\n{xml_bytes.decode(self.ENCODING)}")
+
+        return xml_bin, headers
+
+    def _get_encryption_data(self) -> Tuple[str, bytes]:
+        x_eamuse_info = self._request.headers[self.X_EAMUSE_INFO]
+        key = unhexlify(x_eamuse_info[2:].replace("-", ""))
+        return x_eamuse_info, key
+
+    def _make_encryption_data(self) -> Tuple[str, bytes]:
+        info = f"1-{int(time()):08x}-{randint(0x0000, 0xffff):04x}"
+        key = unhexlify(info[2:].replace("-", ""))
+        return info, key
+
+    def _save_xml(self, data: bytes, kind: str, _id: Optional[str]) -> None:
+        # Always make sure the dir exists
+        self.LOG_DIR.mkdir(parents=True, exist_ok=True)
+
+        # We want a unique identifier to match requests and responses, so let's use the
+        # x-eamuse-info header if it exists, else just a hash of the data
+        uid = _id if _id is not None else str(abs(hash(self._request.data)))[0:8]
+
+        # Write out the data
+        date = datetime.now().strftime("%Y_%m_%d_%H_%M_%S")
+        filepath = self.LOG_DIR / f"eamuse_{date}_{uid}_{kind}.xml"
+        with filepath.open("wb") as f:
+            logging.debug(f"Writing File: {filepath}")
+            f.write(data)
+
+    def __repr__(self) -> str:
+        return (
+            f'ServiceRequest<model: "{self.model}", module: "{self.module}", '
+            f'method: "{self.method}", command: "{self.command}", '
+            f"encrypted: {self.encrypted}, compressed: {self.compressed}, "
+            f'compression: "{self.compression}">'
+        )

v8_server/eamuse/utils/arc4.py

@@ -4,7 +4,7 @@ from Crypto.Cipher import ARC4
 from Crypto.Hash import MD5
 
 
-class EamuseARC4(object):
+class EAmuseARC4(object):
     def __init__(self, eamuse_key) -> None:
         secret_key = 0x69D74627D985EE2187161570D08D93B12455035B6DF0D8205DF5
         key_bytes = bytearray(secret_key.to_bytes(26, "big"))

v8_server/eamuse/utils/eamuse.py

@@ -1,6 +1,21 @@
+from __future__ import annotations
+
 from typing import Dict, Optional
 
 
+class XMLBinTypes(object):
+    s8 = "s8"
+    u8 = "u8"
+    s16 = "s16"
+    u16 = "u16"
+    s32 = "s32"
+    u32 = "u32"
+    s64 = "s64"
+    u64 = "u64"
+    ip4 = "ip4"
+    time = "time"
+
+
 def e_type(_type, count: Optional[int] = None) -> Dict[str, str]:
     result = {"__type": _type}
     if count is not None:
@@ -34,7 +49,7 @@ class Model:
         self.version = version
 
     @staticmethod
-    def from_modelstring(model: str) -> "Model":
+    def from_modelstring(model: Optional[str]) -> Optional[Model]:
         """
         Parse a modelstring and return a Model
         Parameters:
@@ -43,6 +58,9 @@ class Model:
         Returns:
             A Model object.
         """
+        if model is None:
+            return None
+
         parts = model.split(":")
         if len(parts) == 5:
             game, dest, spec, rev, version = parts
@@ -50,10 +68,15 @@ class Model:
         elif len(parts) == 4:
             game, dest, spec, rev = parts
             return Model(game, dest, spec, rev, None)
-        raise Exception("Couldn't parse model {}".format(model))
+        raise Exception(f"Couldn't parse model: {model}")
+
+    def __repr__(self) -> str:
+        version = f", version: {self.version}" if self.version is not None else ""
+        return (
+            f'Model<game: "{self.game}", destination: "{self.dest}", '
+            f'spec: "{self.spec}", revision: "{self.rev}"{version}>'
+        )
 
     def __str__(self) -> str:
-        if self.version is None:
-            return f"{self.game}:{self.dest}:{self.spec}:{self.rev}"
-        else:
-            return f"{self.game}:{self.dest}:{self.spec}:{self.rev}:{self.version}"
+        version = f":{self.version}" if self.version is not None else ""
+        return f"{self.game}:{self.dest}:{self.spec}:{self.rev}{version}"

v8_server/eamuse/utils/xml.py

@@ -0,0 +1,20 @@
+from datetime import datetime
+from typing import Optional
+
+from lxml import etree
+
+
+def get_xml_tag(xml: etree) -> str:
+    return str(xml.tag)
+
+
+def get_xml_attrib(xml: etree, name: str) -> str:
+    return str(xml.attrib[name]) if name in xml.attrib else "None"
+
+
+def format_date(timestamp: Optional[int]) -> str:
+    if timestamp is None:
+        return "None"
+
+    dt = datetime.fromtimestamp(timestamp)
+    return dt.strftime("%Y-%m-%d %H:%M:%S")

v8_server/utils/xml.py

@@ -1,167 +0,0 @@
-import logging
-from binascii import unhexlify
-from datetime import datetime
-from random import randint
-from time import time
-from typing import Dict, Optional, Tuple, Union
-
-import lxml
-from flask import Request
-from kbinxml import KBinXML
-from lxml import etree as ET  # noqa: N812
-
-from v8_server import LOG_PATH
-from v8_server.utils.arc4 import EamuseARC4
-from v8_server.utils.lz77 import Lz77
-
-
-# We want a general logger, and a special logger to log requests separately
-logger = logging.getLogger(__name__)
-rlogger = logging.getLogger("requests")
-
-# eAmuse Header Tags
-X_EAMUSE_INFO = "x-eamuse-info"
-X_COMPRESS = "x-compress"
-
-
-def is_encrypted(request: Request) -> bool:
-    return X_EAMUSE_INFO in request.headers
-
-
-def get_encryption_key(request: Request) -> Tuple[str, bytes]:
-    info = request.headers[X_EAMUSE_INFO]
-    key = unhexlify(info[2:].replace("-", ""))
-    return info, key
-
-
-def make_encryption_key() -> Tuple[str, bytes]:
-    info = f"1-{int(time()):08x}-{randint(0x0000, 0xffff):04x}"
-    key = unhexlify(info[2:].replace("-", ""))
-    return info, key
-
-
-def is_compressed(request: Request) -> bool:
-    return X_COMPRESS in request.headers and request.headers[X_COMPRESS] != "none"
-
-
-def get_compression_type(request: Request) -> str:
-    return request.headers[X_COMPRESS]
-
-
-def get_xml_tag(xml: lxml.etree._Element) -> str:
-    return str(xml.tag)
-
-
-def get_xml_attrib(xml: lxml.etree._Element, name: str) -> str:
-    return str(xml.attrib[name]) if name in xml.attrib else "None"
-
-
-def format_date(timestamp: Optional[int]) -> str:
-    if timestamp is None:
-        return "None"
-
-    dt = datetime.fromtimestamp(timestamp)
-    return dt.strftime("%Y-%m-%d %H:%M:%S")
-
-
-def save_xml(data: bytes, request: Request, kind: str, _type: str = "xml") -> None:
-    # Always make sure the dir exists
-    dirpath = LOG_PATH / "requests"
-    dirpath.mkdir(parents=True, exist_ok=True)
-
-    # We want a unique identifier to match requests and responses, so lets use the
-    # x-eamuse-info header if it exists, else just a hash of the data
-    info = str(abs(hash(request.data)))[0:8]
-    if is_encrypted(request):
-        x_eamuse_info = get_encryption_key(request)[0]
-        info = x_eamuse_info.replace("-", "_")[2:]
-
-    # Write out the data
-    date = datetime.now().strftime("%Y_%m_%d_%H_%M_%S")
-    filepath = dirpath / f"eamuse_{date}_{info}_{kind}.{_type}"
-    with filepath.open("wb") as f:
-        logging.debug(f"Writing File: {filepath}")
-        f.write(data)
-
-
-def eamuse_read_xml(request: Request) -> Tuple[lxml.etree._Element, str, str, str, str]:
-    # Get the raw xml data from the request
-    xml_bin = request.data
-
-    # Decrypt the data if necessary
-    if is_encrypted(request):
-        _, key = get_encryption_key(request)
-        xml_bin = EamuseARC4(key).decrypt(xml_bin)
-
-    # Decompress the data if necessary
-    # Right now we only de-compress lz77
-    if is_compressed(request) and get_compression_type(request) == "lz77":
-        xml_bin = Lz77().decompress(xml_bin)
-
-    # Convert the binary xml data to text bytes and save a copy
-    try:
-        xml_bytes = KBinXML(xml_bin).to_text().encode("UTF-8")
-    except Exception:
-        print(xml_bin)
-        raise
-    save_xml(xml_bytes, request, "req")
-
-    # Convert the xml text to an eTree
-    root = ET.fromstring(xml_bytes)
-
-    # Grab the xml information we care about
-    model = get_xml_attrib(root, "mode")
-    module = get_xml_tag(root[0])
-    method = get_xml_attrib(root[0], "method")
-    command = get_xml_attrib(root[0], "command")
-
-    rlogger.debug(
-        "---- Request ----\n"
-        f"[ {'Model':^20} | {'Module':^15} | {'Method':^15} | {'Command':^20} ]\n"
-        f"[ {model:^20} | {module:^15} | {method:^15} | {command:^20} ]\n"
-        f"{xml_bytes.decode('UTF-8')}\n"
-    )
-
-    # Return XML and important fields
-    return root, model, module, method, command
-
-
-def eamuse_prepare_xml(
-    xml_bytes: Union[bytes, lxml.etree._Element], request: Request
-) -> Tuple[bytes, Dict[str, str]]:
-    # Make sure xml_bytes is a bytes object
-    if type(xml_bytes) == lxml.etree._Element:
-        xml_bytes = ET.tostring(xml_bytes, pretty_print=True)
-
-    # Lets save our response
-    save_xml(xml_bytes, request, "resp")
-
-    # Lets make our own encryption key
-    x_eamuse_info, key = make_encryption_key()
-
-    # Common headers
-    headers = {
-        "Content-Type": "applicaton/octet_stream",
-        "Server": "Microsoft-HTTPAPI/2.0",
-    }
-
-    # Convert XML to binary and save
-    xml_bin = KBinXML(xml_bytes).to_binary()
-
-    # TODO: I don't know that we really need to save the binary xml
-    # save_xml(xml_bin, request, "resp", _type="bin")
-
-    # Compress if necessary
-    # Right now we only compress lz77
-    if is_compressed(request) and get_compression_type(request) == "lz77":
-        headers[X_COMPRESS] = "lz77"
-        xml_bin = Lz77().compress(xml_bin)
-
-    # Encrypt if necessary
-    if is_encrypted(request):
-        headers[X_EAMUSE_INFO] = x_eamuse_info
-        xml_bin = EamuseARC4(key).encrypt(xml_bin)
-
-    rlogger.debug(f"---- Response ----\n{xml_bytes.decode('UTF-8')}\n")
-
-    return xml_bin, headers

v8_server/view/index.py

@@ -7,15 +7,20 @@ from lxml.builder import E
 from sqlalchemy.orm.exc import MultipleResultsFound
 
 from v8_server import app
+from v8_server.eamuse.services.pcbtracker import PCBTracker
+from v8_server.eamuse.services.services import ServiceRequest, Services, ServiceType
+from v8_server.eamuse.utils.eamuse import e_type
+from v8_server.eamuse.utils.xml import get_xml_attrib
 from v8_server.model.connection import Database
 from v8_server.model.user import Card, ExtID, Profile, RefID, User
-from v8_server.utils.eamuse import e_type
-from v8_server.utils.xml import eamuse_prepare_xml, eamuse_read_xml, get_xml_attrib
 
 
-@app.route("/", defaults={"path": ""}, methods=["GET", "POST"])
-@app.route("/<path:path>", methods=["GET", "POST"])
-def catch_all(path: str) -> str:
+FlaskResponse = Tuple[bytes, Dict[str, str]]
+
+
+@app.route("/", defaults={"u_path": ""}, methods=["GET", "POST"])
+@app.route("/<path:u_path>", methods=["GET", "POST"])
+def catch_all(u_path: str) -> str:
     """
     This is currently my catch all route, for whenever a new endpoint pops up that isn't
     implemented
@@ -35,7 +40,7 @@ def catch_all(path: str) -> str:
         f"{header_str[:-1]}\n"
     )
     app.logger.debug(d)
-    return "You want path: %s" % path
+    return "You want path: %s" % u_path
 
 
 def base_response(element: str, attributes: Dict[str, str] = None) -> ET:
@@ -44,25 +49,19 @@ def base_response(element: str, attributes: Dict[str, str] = None) -> ET:
     return E.response(E(element, {**attributes, "expire": "600"}))
 
 
-@app.route("/pcbtracker/service", methods=["POST"])
-def pcbtracker() -> Tuple[bytes, Dict[str, str]]:
-    """
-    Handle a PCBTracker.alive request. The only method of note is the "alive" method
-    which returns whether PASELI should be active or not for this session.
+@Services.route(app, ServiceType.PCBTRACKER)
+def pcbtracker_service() -> FlaskResponse:
+    req = ServiceRequest(request)
 
-    For V8 it should not be active.
-    """
-    method = eamuse_read_xml(request)[3]
-
-    if method == "alive":
-        response = base_response("pcbtracker", {"ecenable": "0"})
+    if req.method == PCBTracker.ALIVE:
+        response = PCBTracker.alive()
     else:
-        # There shoulnd't really even be any other methods
-        raise Exception("Not sure how to handle this PCBTracker Request")
+        raise Exception(f"Not sure how to handle this PCBTracker Request: {req}")
 
-    return eamuse_prepare_xml(response, request)
+    return req.response(response)
 
 
+'''
 @app.route("/message/service", methods=["POST"])
 def message() -> Tuple[bytes, Dict[str, str]]:
     """
@@ -363,46 +362,11 @@ def local() -> Tuple[bytes, Dict[str, str]]:
 
     return eamuse_prepare_xml(response, request)
 
+'''
 
-@app.route("/service/services/services/", methods=["POST"])
-def services() -> Tuple[bytes, Dict[str, str]]:
-    # We don't need to actually read the data here, but let's do it anyway as it saves a
-    # copy
-    _ = eamuse_read_xml(request)
 
-    service_names = [
-        "cardmng",
-        "eacoin",
-        "facility",
-        "local",
-        "message",
-        "netlog",
-        "package",
-        "pcbevent",
-        "pcbtracker",
-        "sidmgr",
-        "userdata",
-        "userid",
-        "eemall",
-    ]
-
-    services = {
-        "ntp": "ntp://pool.ntp.org",
-        "keepalive": (
-            "http://eamuse.konami.fun/"
-            "keepalive?pa=127.0.0.1&ia=127.0.0.1&ga=127.0.0.1&ma=127.0.0.1&t1=2&t2=10"
-        ),
-        **{k: f"http://eamuse.konami.fun/{k}/service" for k in service_names},
-    }
-
-    response = E.response(
-        E.services(
-            expire="600",
-            method="get",
-            mode="operation",
-            status="0",
-            *[E.item({"name": k, "url": services[k]}) for k in services],
-        )
-    )
-
-    return eamuse_prepare_xml(response, request)
+@app.route(Services.SERVICES_ROUTE, methods=["POST"])
+def services_service() -> Tuple[bytes, Dict[str, str]]:
+    s_req = ServiceRequest(request)
+    services = Services().get_services()
+    return s_req.response(services)