8591112367

Committed 07 Apr 2024 07:18PM UTC coverage: 78.701%. First build

Build # 8591112367

Build Type

push

github

Committed by

tcalmant

Commit Message

Added 3.11 & 3.12 to GitHub actions

Run Details

1611 of 2047 relevant lines covered (78.7%)

4.71 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

69.36

/javaobj/v2/core.py

#!/usr/bin/env python3
"""
Second parsing approach for javaobj, using the same approach as jdeserialize
See: https://github.com/frohoff/jdeserialize

:authors: Thomas Calmant
:license: Apache License 2.0
:version: 0.4.4
:status: Alpha

..

    Copyright 2024 Thomas Calmant

    Licensed under the Apache License, Version 2.0 (the "License");
    you may not use this file except in compliance with the License.
    You may obtain a copy of the License at

        http://www.apache.org/licenses/LICENSE-2.0

    Unless required by applicable law or agreed to in writing, software
    distributed under the License is distributed on an "AS IS" BASIS,
    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    See the License for the specific language governing permissions and
    limitations under the License.
"""

from __future__ import absolute_import

import logging
import os
from typing import (  # pylint:disable=W0611
    IO,
    Any,
    Callable,
    Dict,
    List,
    Optional,
)

from ..constants import (
    PRIMITIVE_TYPES,
    StreamConstants,
    TerminalCode,
    TypeCode,
)
from ..modifiedutf8 import (  # pylint:disable=W0611  # noqa: F401
    decode_modified_utf8,
)
from . import api  # pylint:disable=W0611
from .beans import (
    BlockData,
    ClassDataType,
    ClassDescType,
    ExceptionRead,
    ExceptionState,
    FieldType,
    JavaArray,
    JavaClass,
    JavaClassDesc,
    JavaEnum,
    JavaField,
    JavaInstance,
    JavaString,
    ParsedJavaContent,
)
from .stream import DataStreamReader
from .transformers import DefaultObjectTransformer

# ------------------------------------------------------------------------------

# Module version
__version_info__ = (0, 4, 4)
__version__ = ".".join(str(x) for x in __version_info__)

# Documentation strings format
__docformat__ = "restructuredtext en"

# ------------------------------------------------------------------------------


class JavaStreamParser(api.IJavaStreamParser):
    """
    Parses a Java stream
    """

    def __init__(self, fd, transformers):
        # type: (IO[bytes], List[api.ObjectTransformer]) -> None
        """
        :param fd: File-object to read from
        :param transformers: Custom object transformers
        """
        # Input stream
        self.__fd = fd
        self.__reader = DataStreamReader(fd)

        # Object transformers
        self.__transformers = list(transformers)

        # Logger
        self._log = logging.getLogger("javaobj.parser")

        # Handles
        self.__handle_maps = []  # type: List[Dict[int, ParsedJavaContent]]
        self.__handles = {}  # type: Dict[int, ParsedJavaContent]

        # Initial handle value
        self.__current_handle = StreamConstants.BASE_REFERENCE_IDX.value

        # Definition of the type code handlers
        # Each takes the type code as argument
        self.__type_code_handlers = {
            TerminalCode.TC_OBJECT: self._do_object,
            TerminalCode.TC_CLASS: self._do_class,
            TerminalCode.TC_ARRAY: self._do_array,
            TerminalCode.TC_STRING: self._read_new_string,
            TerminalCode.TC_LONGSTRING: self._read_new_string,
            TerminalCode.TC_ENUM: self._do_enum,
            TerminalCode.TC_CLASSDESC: self._do_classdesc,
            TerminalCode.TC_PROXYCLASSDESC: self._do_classdesc,
            TerminalCode.TC_REFERENCE: self._do_reference,
            TerminalCode.TC_NULL: self._do_null,
            TerminalCode.TC_EXCEPTION: self._do_exception,
            TerminalCode.TC_BLOCKDATA: self._do_block_data,
            TerminalCode.TC_BLOCKDATALONG: self._do_block_data,
        }  # type: Dict[int, Callable[[int], ParsedJavaContent]]

    def run(self):
        # type: () -> List[ParsedJavaContent]
        """
        Parses the input stream
        """
        # Check the magic byte
        magic = self.__reader.read_ushort()
        if magic != StreamConstants.STREAM_MAGIC:
            raise ValueError("Invalid file magic: 0x{0:x}".format(magic))

        # Check the stream version
        version = self.__reader.read_ushort()
        if version != StreamConstants.STREAM_VERSION:
            raise ValueError("Invalid file version: 0x{0:x}".format(version))

        # Reset internal state
        self._reset()

        # Read content
        contents = []  # type: List[ParsedJavaContent]
        while True:
            self._log.info("Reading next content")
            start = self.__fd.tell()
            try:
                type_code = self.__reader.read_byte()
            except EOFError:
                # End of file
                break

            if type_code == TerminalCode.TC_RESET:
                # Explicit reset
                self._reset()
                continue

            parsed_content = self._read_content(type_code, True)
            self._log.debug("Read: %s", parsed_content)
            if parsed_content is not None and parsed_content.is_exception:
                # Get the raw data between the start of the object and our
                # current position
                end = self.__fd.tell()
                self.__fd.seek(start, os.SEEK_SET)
                stream_data = self.__fd.read(end - start)

                # Prepare an exception object
                parsed_content = ExceptionState(parsed_content, stream_data)

            contents.append(parsed_content)

        for content in self.__handles.values():
            content.validate()

        # TODO: connect member classes ? (see jdeserialize @ 864)

        if self.__handles:
            self.__handle_maps.append(self.__handles.copy())

        return contents

    def dump(self, content):
        # type: (List[ParsedJavaContent]) -> str
        """
        Dumps to a string the given objects
        """
        lines = []  # type: List[str]

        # Stream content
        lines.append("//// BEGIN stream content output")
        lines.extend(str(c) for c in content)
        lines.append("//// END stream content output")
        lines.append("")

        lines.append("//// BEGIN instance dump")
        for c in self.__handles.values():
            if isinstance(c, JavaInstance):
                instance = c  # type: JavaInstance
                lines.extend(self._dump_instance(instance))
        lines.append("//// END instance dump")
        lines.append("")
        return "\n".join(lines)

    @staticmethod
    def _dump_instance(instance):
        # type: (JavaInstance) -> List[str]
        """
        Dumps an instance to a set of lines
        """
        lines = []  # type: List[str]
        lines.append(
            "[instance 0x{0:x}: 0x{1:x} / {2}".format(
                instance.handle,
                instance.classdesc.handle,
                instance.classdesc.name,
            )
        )

        if instance.annotations:
            lines.append("\tobject annotations:")
            for cd, annotation in instance.annotations.items():
                lines.append("\t" + (cd.name or "null"))
                for c in annotation:
                    lines.append("\t\t" + str(c))

        if instance.field_data:
            lines.append("\tfield data:")
            for field, obj in instance.field_data.items():
                line = "\t\t" + (field.name or "null") + ": "
                if isinstance(obj, ParsedJavaContent):
                    content = obj  # type: ParsedJavaContent
                    h = content.handle
                    if h == instance.handle:
                        line += "this"
                    else:
                        line += "r0x{0:x}".format(h)

                    line += ": " + str(content)
                else:
                    line += str(obj)

                lines.append(line)

        lines.append("]")
        return lines

    def _reset(self):
        """
        Resets the internal state of the parser
        """
        if self.__handles:
            self.__handle_maps.append(self.__handles.copy())

        self.__handles.clear()

        # Reset handle index
        self.__current_handle = StreamConstants.BASE_REFERENCE_IDX

    def _new_handle(self):
        # type: () -> int
        """
        Returns a new handle value
        """
        handle = self.__current_handle
        self.__current_handle += 1
        return handle

    def _set_handle(self, handle, content):
        # type: (int, ParsedJavaContent) -> None
        """
        Stores the reference to an object
        """
        if handle in self.__handles:
            raise ValueError("Trying to reset handle {0:x}".format(handle))

        self.__handles[handle] = content

    @staticmethod
    def _do_null(_):
        """
        The easiest one
        """
        return None

    def _read_content(self, type_code, block_data, class_desc=None):
        # type: (int, bool, Optional[JavaClassDesc]) -> ParsedJavaContent
        """
        Parses the next content
        """
        if not block_data and type_code in (
            TerminalCode.TC_BLOCKDATA,
            TerminalCode.TC_BLOCKDATALONG,
        ):
            raise ValueError("Got a block data, but not allowed here.")

        try:
            # Look for a handler for that type code
            handler = self.__type_code_handlers[type_code]
        except KeyError:
            # Look for an external reader
            if (
                class_desc
                and class_desc.name
                and class_desc.data_type == ClassDataType.WRCLASS
            ):
                # Return its result immediately
                return self._custom_readObject(class_desc.name)

            # No valid custom reader: abandon
            raise ValueError("Unknown type code: 0x{0:x}".format(type_code))
        else:
            try:
                # Parse the object
                return handler(type_code)
            except ExceptionRead as ex:
                # We found an exception object: return it (raise later)
                return ex.exception_object

    def _read_new_string(self, type_code):
        # type: (int) -> JavaString
        """
        Reads a Java String
        """
        if type_code == TerminalCode.TC_REFERENCE:
            # Got a reference
            previous = self._do_reference()
            if not isinstance(previous, JavaString):
                raise ValueError("Invalid reference to a Java string")
            return previous

        # Assign a new handle
        handle = self._new_handle()

        # Read the length
        if type_code == TerminalCode.TC_STRING:
            length = self.__reader.read_ushort()
        elif type_code == TerminalCode.TC_LONGSTRING:
            length = self.__reader.read_long()
            if length < 0 or length > 2147483647:
                raise ValueError("Invalid string length: {0}".format(length))

            if length < 65536:
                self._log.warning("Small string stored as a long one")

        # Parse the content
        data = self.__fd.read(length)
        java_str = JavaString(handle, data)

        # Store the reference to the string
        self._set_handle(handle, java_str)
        return java_str

    def _read_classdesc(self):
        # type: () -> JavaClassDesc
        """
        Reads a class description with its type code
        """
        type_code = self.__reader.read_byte()
        return self._do_classdesc(type_code)

    def _do_classdesc(self, type_code):
        # type: (int) -> JavaClassDesc
        """
        Parses a class description
        """
        if type_code == TerminalCode.TC_CLASSDESC:
            # Do the real job
            name = self.__reader.read_UTF()
            serial_version_uid = self.__reader.read_long()
            handle = self._new_handle()
            desc_flags = self.__reader.read_byte()
            nb_fields = self.__reader.read_short()

            if nb_fields < 0:
                raise ValueError("Invalid field count: {0}".format(nb_fields))

            fields = []  # type: List[JavaField]
            for _ in range(nb_fields):
                field_type = self.__reader.read_byte()
                field_name = self.__reader.read_UTF()
                class_name = None

                if field_type in (TypeCode.TYPE_OBJECT, TypeCode.TYPE_ARRAY):
                    # String type code
                    str_type_code = self.__reader.read_byte()
                    class_name = self._read_new_string(str_type_code)
                elif field_type not in PRIMITIVE_TYPES:
                    raise ValueError(
                        "Invalid field type char: 0x{0:x}".format(field_type)
                    )

                fields.append(
                    JavaField(FieldType(field_type), field_name, class_name)
                )

            # Setup the class description bean
            class_desc = JavaClassDesc(ClassDescType.NORMALCLASS)
            class_desc.name = name
            class_desc.serial_version_uid = serial_version_uid
            class_desc.handle = handle
            class_desc.desc_flags = desc_flags
            class_desc.fields = fields
            class_desc.annotations = self._read_class_annotations(class_desc)
            class_desc.super_class = self._read_classdesc()

            if class_desc.super_class:
                class_desc.super_class.is_super_class = True

            # Store the reference to the parsed bean
            self._set_handle(handle, class_desc)
            return class_desc
        elif type_code == TerminalCode.TC_NULL:
            # Null reference
            return None
        elif type_code == TerminalCode.TC_REFERENCE:
            # Reference to an already loading class description
            previous = self._do_reference()
            if not isinstance(previous, JavaClassDesc):
                raise ValueError(
                    "Referenced object is not a class description"
                )
            return previous
        elif type_code == TerminalCode.TC_PROXYCLASSDESC:
            # Proxy class description
            handle = self._new_handle()
            nb_interfaces = self.__reader.read_int()
            interfaces = [
                self.__reader.read_UTF() for _ in range(nb_interfaces)
            ]

            class_desc = JavaClassDesc(ClassDescType.PROXYCLASS)
            class_desc.handle = handle
            class_desc.interfaces = interfaces
            class_desc.annotations = self._read_class_annotations()
            class_desc.super_class = self._read_classdesc()

            if class_desc.super_class:
                class_desc.super_class.is_super_class = True

            # Store the reference to the parsed bean
            self._set_handle(handle, class_desc)
            return class_desc

        raise ValueError("Expected a valid class description starter")

    def _custom_readObject(self, class_name):
        # type: (str) -> ParsedJavaContent
        """
        Reads an object with a custom serialization process

        :param class_name: Name of the class to load
        :return: The parsed object
        :raise ValueError: Unknown kind of class
        """
        self.__fd.seek(-1, os.SEEK_CUR)
        for transformer in self.__transformers:
            class_data = transformer.load_custom_writeObject(
                self, self.__reader, class_name
            )
            if class_data:
                return class_data

        raise ValueError("Custom readObject can not be processed")

    def _read_class_annotations(self, class_desc=None):
        # type: (Optional[JavaClassDesc]) -> List[ParsedJavaContent]
        """
        Reads the annotations associated to a class
        """
        contents = []  # type: List[ParsedJavaContent]
        while True:
            type_code = self.__reader.read_byte()
            if type_code == TerminalCode.TC_ENDBLOCKDATA:
                # We're done here
                return contents
            elif type_code == TerminalCode.TC_RESET:
                # Reset references
                self._reset()
                continue

            java_object = self._read_content(type_code, True, class_desc)

            if java_object is not None and java_object.is_exception:
                # Found an exception: raise it
                raise ExceptionRead(java_object)

            contents.append(java_object)

        raise Exception("Class annotation reading stopped before end")

    def _create_instance(self, class_desc):
        # type: (JavaClassDesc) -> JavaInstance
        """
        Creates a JavaInstance object, by a transformer if possible
        """
        # Try to create the transformed object
        for transformer in self.__transformers:
            instance = transformer.create_instance(class_desc)
            if instance is not None:
                if class_desc.name:
                    instance.is_external_instance = not self._is_default_supported(
                        class_desc.name
                    )
                return instance

        return JavaInstance()

    def _do_object(self, type_code=0):
        # type: (int) -> JavaInstance
        """
        Parses an object
        """
        # Parse the object class description
        class_desc = self._read_classdesc()

        # Assign a new handle
        handle = self._new_handle()
        self._log.debug(
            "Reading new object: handle %x, classdesc %s", handle, class_desc
        )

        # Prepare the instance object
        instance = self._create_instance(class_desc)
        instance.classdesc = class_desc
        instance.handle = handle

        # Store the instance
        self._set_handle(handle, instance)

        # Read the instance content
        self._read_class_data(instance)
        self._log.debug("Done reading object handle %x", handle)
        return instance

    def _is_default_supported(self, class_name):
        # type: (str) -> bool
        """
        Checks if this class is supported by the default object transformer
        """
        default_transf = [
            x
            for x in self.__transformers
            if isinstance(x, DefaultObjectTransformer)
        ]
        return (
            bool(default_transf)
            and class_name in default_transf[0]._type_mapper
        )

    def _read_class_data(self, instance):
        # type: (JavaInstance) -> None
        """
        Reads the content of an instance
        """
        # Read the class hierarchy
        classes = []  # type: List[JavaClassDesc]
        instance.classdesc.get_hierarchy(classes)

        all_data = {}  # type: Dict[JavaClassDesc, Dict[JavaField, Any]]
        annotations = {}  # type: Dict[JavaClassDesc, List[ParsedJavaContent]]

        for cd in classes:
            values = {}  # type: Dict[JavaField, Any]
            cd.validate()
            if (
                cd.data_type == ClassDataType.NOWRCLASS
                or cd.data_type == ClassDataType.WRCLASS
            ):
                if (
                    cd.data_type == ClassDataType.WRCLASS
                    and instance.is_external_instance
                ):
                    annotations[cd] = self._read_class_annotations(cd)
                else:
                    for field in cd.fields:
                        values[field] = self._read_field_value(field.type)
                    all_data[cd] = values

                    if cd.data_type == ClassDataType.WRCLASS:
                        annotations[cd] = self._read_class_annotations(cd)
            else:
                if cd.data_type == ClassDataType.OBJECT_ANNOTATION:
                    # Call the transformer if possible
                    if not instance.load_from_blockdata(self, self.__reader):
                        # Can't read :/
                        raise ValueError(
                            "hit externalizable with nonzero SC_BLOCK_DATA; "
                            "can't interpret data"
                        )
                annotations[cd] = self._read_class_annotations(cd)

        # Fill the instance object
        instance.annotations = annotations
        instance.field_data = all_data

        # Load transformation from the fields and annotations
        instance.load_from_instance()

    def _read_field_value(self, field_type):
        # type: (FieldType) -> Any
        """
        Reads the value of an instance field
        """
        if field_type == FieldType.BYTE:
            return self.__reader.read_byte()
        if field_type == FieldType.CHAR:
            return self.__reader.read_char()
        if field_type == FieldType.DOUBLE:
            return self.__reader.read_double()
        if field_type == FieldType.FLOAT:
            return self.__reader.read_float()
        if field_type == FieldType.INTEGER:
            return self.__reader.read_int()
        if field_type == FieldType.LONG:
            return self.__reader.read_long()
        if field_type == FieldType.SHORT:
            return self.__reader.read_short()
        if field_type == FieldType.BOOLEAN:
            return self.__reader.read_bool()
        if field_type in (FieldType.OBJECT, FieldType.ARRAY):
            sub_type_code = self.__reader.read_byte()
            if field_type == FieldType.ARRAY:
                if sub_type_code == TerminalCode.TC_NULL:
                    # Seems required, according to issue #46
                    return None
                if sub_type_code == TerminalCode.TC_REFERENCE:
                    return self._do_classdesc(sub_type_code)
                if sub_type_code != TerminalCode.TC_ARRAY:
                    raise ValueError(
                        "Array type listed, but type code != TC_ARRAY"
                    )

            content = self._read_content(sub_type_code, False)
            if content is not None and content.is_exception:
                raise ExceptionRead(content)

            return content

        raise ValueError("Can't process type: {0}".format(field_type))

    def _do_reference(self, type_code=0):
        # type: (int) -> ParsedJavaContent
        """
        Returns an object already parsed
        """
        handle = self.__reader.read_int()
        try:
            return self.__handles[handle]
        except KeyError:
            raise ValueError("Invalid reference handle: {0:x}".format(handle))

    def _do_enum(self, type_code):
        # type: (int) -> JavaEnum
        """
        Parses an enumeration
        """
        cd = self._read_classdesc()
        if cd is None:
            raise ValueError("Enum description can't be null")

        handle = self._new_handle()

        # Read the enum string
        sub_type_code = self.__reader.read_byte()
        enum_str = self._read_new_string(sub_type_code)
        cd.enum_constants.add(enum_str.value)

        # Store the object
        enum_obj = JavaEnum(handle, cd, enum_str)
        self._set_handle(handle, enum_obj)
        return enum_obj

    def _do_class(self, type_code):
        # type: (int) -> JavaClass
        """
        Parses a class
        """
        cd = self._read_classdesc()
        handle = self._new_handle()
        class_obj = JavaClass(handle, cd)

        # Store the class object
        self._set_handle(handle, class_obj)
        return class_obj

    def _do_array(self, type_code):
        # type: (int) -> JavaArray
        """
        Parses an array
        """
        cd = self._read_classdesc()
        handle = self._new_handle()
        if not cd.name or len(cd.name) < 2:
            raise ValueError("Invalid name in array class description")

        # ParsedJavaContent type
        content_type_byte = ord(cd.name[1].encode("latin1"))
        field_type = FieldType(content_type_byte)

        # Array size
        size = self.__reader.read_int()
        if size < 0:
            raise ValueError("Invalid array size")

        # Array content
        for transformer in self.__transformers:
            content = transformer.load_array(
                self.__reader, field_type.type_code(), size
            )
            if content is not None:
                break
        else:
            content = [self._read_field_value(field_type) for _ in range(size)]

        return JavaArray(handle, cd, field_type, content)

    def _do_exception(self, type_code):
        # type: (int) -> ParsedJavaContent
        """
        Read the content of a thrown exception
        """
        # Start by resetting current state
        self._reset()

        type_code = self.__reader.read_byte()
        if type_code == TerminalCode.TC_RESET:
            raise ValueError("TC_RESET read while reading exception")

        content = self._read_content(type_code, False)
        if content is None:
            raise ValueError("Null exception object")

        if not isinstance(content, JavaInstance):
            raise ValueError("Exception object is not an instance")

        if content.is_exception:
            raise ExceptionRead(content)

        # Strange object ?
        content.is_exception = True
        self._reset()
        return content

    def _do_block_data(self, type_code):
        # type: (int) -> BlockData
        """
        Reads a block data
        """
        # Parse the size
        if type_code == TerminalCode.TC_BLOCKDATA:
            size = self.__reader.read_ubyte()
        elif type_code == TerminalCode.TC_BLOCKDATALONG:
            size = self.__reader.read_int()
        else:
            raise ValueError("Invalid type code for blockdata")

        if size < 0:
            raise ValueError("Invalid value for block data size")

        # Read the block
        data = self.__fd.read(size)
        return BlockData(data)

tcalmant / python-javaobj / 8591112367

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous