diff --git a/TODO.txt b/TODO.txt index dc6ee32..fb7d673 100644 --- a/TODO.txt +++ b/TODO.txt @@ -1,2 +1 @@ - Implement proper logging (is there some sort of flask logging, or should I just log to a file?) -- Add typing diff --git a/v8_server/config.py b/v8_server/config.py index 29dac42..75bf245 100644 --- a/v8_server/config.py +++ b/v8_server/config.py @@ -1,13 +1,13 @@ class Config(object): - DEBUG = False - TESTING = False - DB_SERVER = "localhost" - SECRET_KEY_FILENAME = "v8_server.key" + DEBUG: bool = False + TESTING: bool = False + DB_SERVER: str = "localhost" + SECRET_KEY_FILENAME: str = "v8_server.key" class Development(Config): - DEBUG = True - SECRET_KEY_FILENAME = "dev_v8_server.key" + DEBUG: bool = True + SECRET_KEY_FILENAME: str = "dev_v8_server.key" class Production(Config): diff --git a/v8_server/utils/arc4.py b/v8_server/utils/arc4.py index 798921b..e409bd1 100644 --- a/v8_server/utils/arc4.py +++ b/v8_server/utils/arc4.py @@ -1,16 +1,18 @@ +from typing import Iterable + from Crypto.Cipher import ARC4 from Crypto.Hash import MD5 class EamuseARC4(object): - def __init__(self, eamuse_key): + def __init__(self, eamuse_key) -> None: secret_key = 0x69D74627D985EE2187161570D08D93B12455035B6DF0D8205DF5 key_bytes = bytearray(secret_key.to_bytes(26, "big")) key = MD5.new(eamuse_key + key_bytes).digest() self.arc = ARC4.new(key) - def decrypt(self, data): + def decrypt(self, data: Iterable[int]) -> bytes: return self.arc.decrypt(bytes(data)) - def encrypt(self, data): + def encrypt(self, data: Iterable[int]) -> bytes: return self.arc.encrypt(bytes(data)) diff --git a/v8_server/utils/flask.py b/v8_server/utils/flask.py index 851e155..921d50e 100644 --- a/v8_server/utils/flask.py +++ b/v8_server/utils/flask.py @@ -1,10 +1,12 @@ -import datetime +from datetime import datetime, timedelta from os import urandom from pathlib import Path from tempfile import gettempdir -def generate_secret_key(secret_key_filename, expiry_delta=None): +def generate_secret_key( + secret_key_filename: str, expiry_delta: timedelta = None +) -> bytes: secret_key_file = Path(gettempdir()) / secret_key_filename secret_exists = secret_key_file.exists() diff --git a/v8_server/utils/xml.py b/v8_server/utils/xml.py index 1849049..60b569f 100644 --- a/v8_server/utils/xml.py +++ b/v8_server/utils/xml.py @@ -2,9 +2,11 @@ from binascii import unhexlify from pathlib import Path from random import randint from time import time +from typing import Dict, Tuple import lxml import lzss +from flask import Request from kbinxml import KBinXML from lxml import etree as ET # noqa: N812 @@ -16,7 +18,7 @@ EAMUSE_CONFIG = {"encrypted": False, "compressed": False} REQUESTS_PATH = Path("./requests") -def eamuse_read_xml(request): +def eamuse_read_xml(request: Request) -> Tuple[str, str, str, str, str]: # Get encrypted/compressed data from client headers = request.headers data = request.data @@ -69,7 +71,7 @@ def eamuse_read_xml(request): return xml_text, model, module, method, command -def eamuse_prepare_xml(xml): +def eamuse_prepare_xml(xml: str) -> Tuple[bytes, Dict[str, str]]: x_eamuse_info = f"1-{int(time()):08x}-{randint(0x0000, 0xffff):04x}" key = unhexlify(x_eamuse_info[2:].replace("-", "")) diff --git a/v8_server/version.py b/v8_server/version.py index 63d33c5..7a12095 100644 --- a/v8_server/version.py +++ b/v8_server/version.py @@ -1,7 +1,7 @@ from pathlib import Path -def get_version_number(): +def get_version_number() -> str: # This file must exist in the root of the module, as the following code # tries to find the module root so that it can find the VERSION file. # project_root/ diff --git a/v8_server/view/index.py b/v8_server/view/index.py index 77ed3c5..1e40070 100644 --- a/v8_server/view/index.py +++ b/v8_server/view/index.py @@ -1,4 +1,7 @@ +from typing import Dict, Tuple + from flask import request +from lxml import etree as ET # noqa: N812 from lxml.builder import E from v8_server import app @@ -7,32 +10,32 @@ from v8_server.utils.xml import eamuse_prepare_xml, eamuse_read_xml @app.route("/", defaults={"path": ""}, methods=["GET", "POST"]) @app.route("/", methods=["GET", "POST"]) -def catch_all(path): +def catch_all(path: str) -> str: """ This is currently my catch all route, for whenever a new endpoint pops up that isn't implemented """ d = f""" - {request.args} - {request.form} - {request.files} - {request.values} - {request.json} - {request.data} - {request.headers} + {request.args!r} + {request.form!r} + {request.files!r} + {request.values!r} + {request.json!r} + {request.data!r} + {request.headers!r} """ print(d) return "You want path: %s" % path -def base_response(element, attributes=None): +def base_response(element: str, attributes: Dict[str, str] = None) -> ET: if attributes is None: attributes = {} return E.response(E(element, {**attributes, "expire": "600"})) @app.route("/pcbtracker/service", methods=["POST"]) -def pcbtracker(): +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. @@ -51,7 +54,7 @@ def pcbtracker(): @app.route("/message/service", methods=["POST"]) -def message(): +def message() -> Tuple[bytes, Dict[str, str]]: """ Unknown what this does. Possibly for operator messages? """ @@ -61,7 +64,7 @@ def message(): @app.route("/pcbevent/service", methods=["POST"]) -def pcbevent(): +def pcbevent() -> Tuple[bytes, Dict[str, str]]: """ Handle a PCBEvent request. We do nothing for this aside from logging the event. """ @@ -74,7 +77,7 @@ def pcbevent(): @app.route("/facility/service", methods=["POST"]) -def facility(): +def facility() -> Tuple[bytes, Dict[str, str]]: """ Handle a facility request. The only method of note is the "get" request, which expects to return a bunch of information about the arcade this cabinet is in, @@ -125,7 +128,7 @@ def facility(): @app.route("/package/service", methods=["POST"]) -def package(): +def package() -> Tuple[bytes, Dict[str, str]]: """ This is for supporting downloading of updates. We don't support this. """ @@ -136,13 +139,13 @@ def package(): @app.route("/service/services/services/", methods=["POST"]) -def services(): +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 = [ - "dlstatus", + "cardmng", "eacoin", "facility", "local",