Added the rest of typing

This commit is contained in:
573dev
2020-10-14 12:59:20 -05:00
parent fb09646ef5
commit 6967dfa4b3
7 changed files with 39 additions and 31 deletions

View File

@@ -1,2 +1 @@
- Implement proper logging (is there some sort of flask logging, or should I just log to a file?)
- Add typing

View File

@@ -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):

View File

@@ -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))

View File

@@ -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()

View File

@@ -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("-", ""))

View File

@@ -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/

View File

@@ -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("/<path:path>", 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",