bemaniutils/bemani/protocol/protocol.py

import binascii
import hashlib
from typing import Optional
from typing_extensions import Final

from bemani.protocol.lz77 import Lz77
from bemani.protocol.binary import BinaryEncoding
from bemani.protocol.xml import XmlEncoding
from bemani.protocol.node import Node


class EAmuseException(Exception):
    """
    An exception thrown when we encounter an error with E-Amusement encapsulation.
    """


class EAmuseProtocol:
    """
    A wrapper object that encapsulates encoding/decoding the E-Amusement protocol by Konami.
    """

    SHARED_SECRET: Final[
        bytes
    ] = b"\x69\xD7\x46\x27\xD9\x85\xEE\x21\x87\x16\x15\x70\xD0\x8D\x93\xB1\x24\x55\x03\x5B\x6D\xF0\xD8\x20\x5D\xF5"

    XML: Final[int] = 1
    BINARY: Final[int] = 2
    BINARY_DECOMPRESSED: Final[int] = 3

    SHIFT_JIS_LEGACY: Final[str] = "shift-jis-legacy"
    SHIFT_JIS: Final[str] = "shift-jis"
    EUC_JP: Final[str] = "euc-jp"
    UTF_8: Final[str] = "utf-8"
    ASCII: Final[str] = "ascii"

    def __init__(self) -> None:
        """
        Initialize the object.
        """
        self.last_text_encoding: Optional[str] = None
        self.last_packet_encoding: Optional[int] = None

    def _rc4_crypt(self, data: bytes, key: bytes) -> bytes:
        """
        Given a data blob and a key blob, perform RC4 encryption/decryption.

        Parameters:
            data - Binary string representing data to be encrypted/decrypted
            key - Binary string representing the key to use

        Returns:
            binary string representing the encrypted/decrypted data
        """
        S = list(range(256))
        j = 0
        out = []

        # KSA Phase
        for i in range(256):
            j = (j + S[i] + key[i % len(key)]) & 0xFF
            S[i], S[j] = S[j], S[i]

        # PRGA Phase
        i = j = 0
        for char in data:
            i = (i + 1) & 0xFF
            j = (j + S[i]) & 0xFF
            S[i], S[j] = S[j], S[i]
            out.append(char ^ S[(S[i] + S[j]) & 0xFF])

        return bytes(out)

    def __decrypt(self, encryption_key: Optional[str], data: bytes) -> bytes:
        """
        Given data and an optional encryption key, decrypt the data.

        Parameters:
            encryption_key - A string encryption key as returned from a HTTP request.
                             Should be in the form 1-xxyyzzww-aabb. If it is None, this
                             performs a null decrypt.
            data - Binary string representing data to transform.

        Returns:
            binary string representing transformed data
        """
        key: Optional[bytes] = None
        if encryption_key:
            # Key is concatenated with the shared secret above
            version, first, second = encryption_key.split("-")
            key = (
                binascii.unhexlify((first + second).encode("ascii"))
                + EAmuseProtocol.SHARED_SECRET
            )

            # Next, key is sent through MD5 to derive the real key
            m = hashlib.md5()
            m.update(key)
            key = m.digest()

        if key:
            # This is an encrypted old-style packet
            return self._rc4_crypt(data, key)

        # No encryption
        return data

    def __encrypt(self, encryption_key: Optional[str], data: bytes) -> bytes:
        """
        Given data and an optional encryption key, encrypt the data.

        Parameters:
            encryption_key - A string encryption key as returned from a HTTP request.
                             Should be in the form 1-xxyyzzww-aabb. If it is None, this
                             performs a null decrypt.
            data - Binary string representing data to transform.

        Returns:
            binary string representing transformed data
        """
        # RC4 is symmetric
        return self.__decrypt(encryption_key, data)

    def __decompress(self, compression: Optional[str], data: bytes) -> bytes:
        """
        Given data and an optional compression scheme, decompress the data.

        Parameters:
            compression - A string specifying the compression used. Should
                          be of the form 'l7zz' or 'none'. You can also pass in
                          the python value None.
            data - Binary string representing data to transform.

        Returns:
            binary string representing transformed data
        """
        if compression is None or compression == "none":
            # This isn't compressed
            return data
        elif compression == "lz77":
            # This is a compressed new-style packet
            lz = Lz77()
            return lz.decompress(data)
        else:
            raise EAmuseException(f"Unknown compression {compression}")

    def __compress(self, compression: Optional[str], data: bytes) -> bytes:
        """
        Given data and an optional compression scheme, compress the data.

        Parameters:
            compression - A string specifying the compression used. Should
                          be of the form 'l7zz' or 'none'. The python value
                          None will also be recognized as 'none'.
            data - Binary string representing data to transform.

        Returns:
            binary string representing transformed data
        """
        if compression is None or compression == "none":
            # This isn't compressed
            return data
        elif compression == "lz77":
            # This is a compressed new-style packet
            lz = Lz77()
            return lz.compress(data)
        else:
            raise EAmuseException(f"Unknown compression {compression}")

    def __decode(self, data: bytes) -> Node:
        """
        Given data, decode the data into a Node tree.

        Parameters:
            data - Binary string representing data to decode.

        Returns:
            Node tree on success or None on failure.
        """
        # Assume it's a binary page
        binary = BinaryEncoding()
        ret = binary.decode(data, skip_on_exceptions=True)

        if ret is not None:
            # We got a result, it was binary
            self.last_text_encoding = binary.encoding
            self.last_packet_encoding = EAmuseProtocol.BINARY

            return ret

        # Assume its XML
        xml = XmlEncoding()
        ret = xml.decode(data, skip_on_exceptions=True)

        if ret is not None:
            # We got a result, it was XML
            self.last_text_encoding = xml.encoding
            self.last_packet_encoding = EAmuseProtocol.XML

            return ret

        # Couldn't decode
        raise EAmuseException("Unknown packet encoding")

    def __encode(self, tree: Node, text_encoding: str, packet_encoding: int) -> bytes:
        """
        Given a Node tree, encode the data into the given packet encoding.

        Parameters:
            tree - A Node object representing the root of the tree to encode.
            text_encoding - The text encoding for any strings that will be encoded.
                            Should be EAmuseProtocol.SHIFT_JIS, EAmuseProtocol.EUC_JP or
                            EAmuseProtocol.UTF8.
            packet_encoding - The encoding used for the packet. Should be EAmuseProtocol.XML
                              or EAmuseProtocol.BINARY.

        Returns:
            A string blob representing the encoded packet.
        """
        if packet_encoding == EAmuseProtocol.BINARY:
            # It's binary, encode it
            binary = BinaryEncoding()
            return binary.encode(tree, encoding=text_encoding)
        elif packet_encoding == EAmuseProtocol.BINARY_DECOMPRESSED:
            # It's binary, encode it
            binary = BinaryEncoding()
            return binary.encode(tree, encoding=text_encoding, compressed=False)
        elif packet_encoding == EAmuseProtocol.XML:
            # It's XML, encode it
            xml = XmlEncoding()
            return xml.encode(tree, encoding=text_encoding)
        else:
            raise EAmuseException(f"Invalid packet encoding {packet_encoding}")

    def decode(
        self, compression: Optional[str], encryption: Optional[str], data: bytes
    ) -> Node:
        """
        Given a request with optional compression and encryption set, decrypt,
        decompress and decode the data, returning a parsed tree.

        Parameters:
            compression - A string specifying the compression type, should be 'lz77' or 'none'.
                          The python value None can also be passed in.
            encryption - A string specifying the encryption key, or None if no encryption.
            data - A binary string of data to parse.

        Returns:
            A Node tree structure representing the parsed request, or None on failure.
        """
        data = self.__decrypt(encryption, data)
        data = self.__decompress(compression, data)
        return self.__decode(data)

    def encode(
        self,
        compression: Optional[str],
        encryption: Optional[str],
        tree: Node,
        text_encoding: Optional[str] = None,
        packet_encoding: Optional[int] = None,
    ) -> bytes:
        """
        Given a response with optional compression and encryption set, encode, compress
        and encrypt the data, returning a binary blob suitable for forwarding on a network.

        Parameters:
            compression - A string specifying the compression type, should be 'lz77' or 'none'.
                          The python value None can also be passed in.
            encryption - A string specifying the encryption key, or None if no encryption.
            data - A binary string of data to parse.
            text_encoding - A text encoding to use. If not provided, uses the text encoding of the
                            last decoded packet. See __encode for values.
            packet_encpding - A packet encoding to use. If not provided, uses the packet encoding
                              of the last decoded packet. See __encode for values.

        Returns:
            A blob of data representing the encoded packet.
        """
        # Either auto-set response based on request, or explicitly override in parameters
        if text_encoding is None:
            text_encoding = self.last_text_encoding
        if text_encoding is None:
            raise EAmuseException("Unknown text encoding")

        if packet_encoding is None:
            packet_encoding = self.last_packet_encoding
        if packet_encoding is None:
            raise EAmuseException("Unknown packet encoding")

        # Clear last packet since we sent a response
        self.last_text_encoding = None
        self.last_packet_encoding = None

        data = self.__encode(tree, text_encoding, packet_encoding)
        data = self.__compress(compression, data)
        return self.__encrypt(encryption, data)