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

80.77

/javaobj/v1/marshaller.py

#!/usr/bin/python
# -- Content-Encoding: utf-8 --
"""
Provides functions for writing (writing is WIP currently) Java
objects that will be deserialized by ObjectOutputStream. This form of
object representation is a standard data interchange format in Java world.

javaobj module exposes an API familiar to users of the standard library
marshal, pickle and json modules.

See:
http://download.oracle.com/javase/6/docs/platform/serialization/spec/protocol.html

:authors: Volodymyr Buell, 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

# Standard library
import collections
import logging
import struct

try:
    # Python 2
    from StringIO import StringIO as BytesIO
except ImportError:
    # Python 3+
    from io import BytesIO

# Javaobj modules
from .beans import (
    JavaClass,
    JavaString,
    JavaObject,
    JavaByteArray,
    JavaEnum,
    JavaArray,
)
from ..constants import (
    StreamConstants,
    ClassDescFlags,
    TerminalCode,
    TypeCode,
)
from ..utils import (
    log_debug,
    log_error,
    to_bytes,
    BYTES_TYPE,
    UNICODE_TYPE,
)

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

__all__ = ("JavaObjectMarshaller",)


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

# Documentation strings format
__docformat__ = "restructuredtext en"

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


class JavaObjectMarshaller:
    """
    Serializes objects into Java serialization format
    """

    def __init__(self, stream=None):
        """
        Sets up members

        :param stream: An output stream
        """
        self.object_stream = stream
        self.object_obj = None
        self.object_transformers = []
        self.references = []

    def add_transformer(self, transformer):
        """
        Appends an object transformer to the serialization process

        :param transformer: An object with a transform(obj) method
        """
        self.object_transformers.append(transformer)

    def dump(self, obj):
        """
        Dumps the given object in the Java serialization format
        """
        self.references = []
        self.object_obj = obj
        self.object_stream = BytesIO()
        self._writeStreamHeader()
        self.writeObject(obj)
        return self.object_stream.getvalue()

    def _writeStreamHeader(self):  # pylint:disable=C0103
        """
        Writes the Java serialization magic header in the serialization stream
        """
        self._writeStruct(
            ">HH",
            4,
            (StreamConstants.STREAM_MAGIC, StreamConstants.STREAM_VERSION),
        )

    def writeObject(self, obj):  # pylint:disable=C0103
        """
        Appends an object to the serialization stream

        :param obj: A string or a deserialized Java object
        :raise RuntimeError: Unsupported type
        """
        log_debug("Writing object of type {0}".format(type(obj).__name__))
        if isinstance(obj, JavaArray):
            # Deserialized Java array
            self.write_array(obj)
        elif isinstance(obj, JavaByteArray):
            # Deserialized Java byte array
            self.write_array(obj)
        elif isinstance(obj, JavaEnum):
            # Deserialized Java Enum
            self.write_enum(obj)
        elif isinstance(obj, JavaObject):
            # Deserialized Java object
            self.write_object(obj)
        elif isinstance(obj, JavaString):
            # Deserialized String
            self.write_string(obj)
        elif isinstance(obj, JavaClass):
            # Java class
            self.write_class(obj)
        elif obj is None:
            # Null
            self.write_null()
        elif type(obj) is str:  # pylint:disable=C0123
            # String value
            self.write_blockdata(obj)
        else:
            # Unhandled type
            raise RuntimeError(
                "Object serialization of type {0} is not "
                "supported.".format(type(obj))
            )

    def _writeStruct(self, unpack, length, args):  # pylint:disable=C0103
        """
        Appends data to the serialization stream

        :param unpack: Struct format string
        :param length: Unused
        :param args: Struct arguments
        """
        ba = struct.pack(unpack, *args)
        self.object_stream.write(ba)

    def _writeString(self, obj, use_reference=True):  # pylint:disable=C0103
        """
        Appends a string to the serialization stream

        :param obj: String to serialize
        :param use_reference: If True, allow writing a reference
        """
        # TODO: Convert to "modified UTF-8"
        # http://docs.oracle.com/javase/7/docs/api/java/io/DataInput.html#modified-utf-8
        string = to_bytes(obj, "utf-8")

        if use_reference and isinstance(obj, JavaString):
            try:
                idx = self.references.index(obj)
            except ValueError:
                # First appearance of the string
                self.references.append(obj)
                logging.debug(
                    "*** Adding ref 0x%X for string: %s",
                    len(self.references)
                    - 1
                    + StreamConstants.BASE_REFERENCE_IDX,
                    obj,
                )

                self._writeStruct(">H", 2, (len(string),))
                self.object_stream.write(string)
            else:
                # Write a reference to the previous type
                logging.debug(
                    "*** Reusing ref 0x%X for string: %s",
                    idx + StreamConstants.BASE_REFERENCE_IDX,
                    obj,
                )
                self.write_reference(idx)
        else:
            self._writeStruct(">H", 2, (len(string),))
            self.object_stream.write(string)

    def write_string(self, obj, use_reference=True):
        """
        Writes a Java string with the TC_STRING type marker

        :param obj: The string to print
        :param use_reference: If True, allow writing a reference
        """
        if use_reference and isinstance(obj, JavaString):
            try:
                idx = self.references.index(obj)
            except ValueError:
                # String is not referenced: let _writeString store it
                self._writeStruct(">B", 1, (TerminalCode.TC_STRING,))
                self._writeString(obj, use_reference)
            else:
                # Reuse the referenced string
                logging.debug(
                    "*** Reusing ref 0x%X for String: %s",
                    idx + StreamConstants.BASE_REFERENCE_IDX,
                    obj,
                )
                self.write_reference(idx)
        else:
            # Don't use references
            self._writeStruct(">B", 1, (TerminalCode.TC_STRING,))
            self._writeString(obj, use_reference)

    def write_enum(self, obj):
        """
        Writes an Enum value

        :param obj: A JavaEnum object
        """
        # FIXME: the output doesn't have the same references as the real
        # serializable form
        self._writeStruct(">B", 1, (TerminalCode.TC_ENUM,))

        try:
            idx = self.references.index(obj)
        except ValueError:
            # New reference
            self.references.append(obj)
            logging.debug(
                "*** Adding ref 0x%X for enum: %s",
                len(self.references) - 1 + StreamConstants.BASE_REFERENCE_IDX,
                obj,
            )

            self.write_classdesc(obj.get_class())
        else:
            self.write_reference(idx)

        self.write_string(obj.constant)

    def write_blockdata(self, obj, parent=None):  # pylint:disable=W0613
        """
        Appends a block of data to the serialization stream

        :param obj: String form of the data block
        """
        if isinstance(obj, UNICODE_TYPE):
            # Latin-1: keep bytes as is
            obj = to_bytes(obj, "latin-1")

        length = len(obj)
        if length <= 256:
            # Small block data
            # TC_BLOCKDATA (unsigned byte)<size> (byte)[size]
            self._writeStruct(">B", 1, (TerminalCode.TC_BLOCKDATA,))
            self._writeStruct(">B", 1, (length,))
        else:
            # Large block data
            # TC_BLOCKDATALONG (unsigned int)<size> (byte)[size]
            self._writeStruct(">B", 1, (TerminalCode.TC_BLOCKDATALONG,))
            self._writeStruct(">I", 1, (length,))

        self.object_stream.write(obj)

    def write_null(self):
        """
        Writes a "null" value
        """
        self._writeStruct(">B", 1, (TerminalCode.TC_NULL,))

    def write_object(self, obj, parent=None):
        """
        Writes an object header to the serialization stream

        :param obj: Not yet used
        :param parent: Not yet used
        """
        # Transform object
        for transformer in self.object_transformers:
            tmp_object = transformer.transform(obj)
            if tmp_object is not obj:
                obj = tmp_object
                break

        self._writeStruct(">B", 1, (TerminalCode.TC_OBJECT,))
        cls = obj.get_class()
        self.write_classdesc(cls)

        # Add reference
        self.references.append([])
        logging.debug(
            "*** Adding ref 0x%X for object %s",
            len(self.references) - 1 + StreamConstants.BASE_REFERENCE_IDX,
            obj,
        )

        all_names = collections.deque()
        all_types = collections.deque()
        tmpcls = cls
        while tmpcls:
            all_names.extendleft(reversed(tmpcls.fields_names))
            all_types.extendleft(reversed(tmpcls.fields_types))
            tmpcls = tmpcls.superclass
        del tmpcls

        logging.debug("<=> Field names: %s", all_names)
        logging.debug("<=> Field types: %s", all_types)

        for field_name, field_type in zip(all_names, all_types):
            try:
                logging.debug(
                    "Writing field %s (%s): %s",
                    field_name,
                    field_type,
                    getattr(obj, field_name),
                )
                self._write_value(field_type, getattr(obj, field_name))
            except AttributeError as ex:
                log_error(
                    "No attribute {0} for object {1}\nDir: {2}".format(
                        ex, repr(obj), dir(obj)
                    )
                )
                raise
        del all_names, all_types

        if (
            cls.flags & ClassDescFlags.SC_SERIALIZABLE
            and cls.flags & ClassDescFlags.SC_WRITE_METHOD
            or cls.flags & ClassDescFlags.SC_EXTERNALIZABLE
            and cls.flags & ClassDescFlags.SC_BLOCK_DATA
        ):
            for annotation in obj.annotations:
                log_debug(
                    "Write annotation {0} for {1}".format(
                        repr(annotation), repr(obj)
                    )
                )
                if annotation is None:
                    self.write_null()
                else:
                    self.writeObject(annotation)
            self._writeStruct(">B", 1, (TerminalCode.TC_ENDBLOCKDATA,))

    def write_class(self, obj, parent=None):  # pylint:disable=W0613
        """
        Writes a class to the stream

        :param obj: A JavaClass object
        :param parent:
        """
        self._writeStruct(">B", 1, (TerminalCode.TC_CLASS,))
        self.write_classdesc(obj)

    def write_classdesc(self, obj, parent=None):  # pylint:disable=W0613
        """
        Writes a class description

        :param obj: Class description to write
        :param parent:
        """
        if obj not in self.references:
            # Add reference
            self.references.append(obj)
            logging.debug(
                "*** Adding ref 0x%X for classdesc %s",
                len(self.references) - 1 + StreamConstants.BASE_REFERENCE_IDX,
                obj.name,
            )

            self._writeStruct(">B", 1, (TerminalCode.TC_CLASSDESC,))
            self._writeString(obj.name)
            self._writeStruct(">qB", 1, (obj.serialVersionUID, obj.flags))
            self._writeStruct(">H", 1, (len(obj.fields_names),))

            for field_name, field_type in zip(
                obj.fields_names, obj.fields_types
            ):
                self._writeStruct(
                    ">B", 1, (self._convert_type_to_char(field_type),)
                )
                self._writeString(field_name)
                if ord(field_type[0]) in (
                    TypeCode.TYPE_OBJECT,
                    TypeCode.TYPE_ARRAY,
                ):
                    try:
                        idx = self.references.index(field_type)
                    except ValueError:
                        # First appearance of the type
                        self.references.append(field_type)
                        logging.debug(
                            "*** Adding ref 0x%X for field type %s",
                            len(self.references)
                            - 1
                            + StreamConstants.BASE_REFERENCE_IDX,
                            field_type,
                        )

                        self.write_string(field_type, False)
                    else:
                        # Write a reference to the previous type
                        logging.debug(
                            "*** Reusing ref 0x%X for %s (%s)",
                            idx + StreamConstants.BASE_REFERENCE_IDX,
                            field_type,
                            field_name,
                        )
                        self.write_reference(idx)

            self._writeStruct(">B", 1, (TerminalCode.TC_ENDBLOCKDATA,))
            if obj.superclass:
                self.write_classdesc(obj.superclass)
            else:
                self.write_null()
        else:
            # Use reference
            self.write_reference(self.references.index(obj))

    def write_reference(self, ref_index):
        """
        Writes a reference
        :param ref_index: Local index (0-based) to the reference
        """
        self._writeStruct(
            ">BL",
            1,
            (
                TerminalCode.TC_REFERENCE,
                ref_index + StreamConstants.BASE_REFERENCE_IDX,
            ),
        )

    def write_array(self, obj):
        """
        Writes a JavaArray

        :param obj: A JavaArray object
        """
        classdesc = obj.get_class()
        self._writeStruct(">B", 1, (TerminalCode.TC_ARRAY,))
        self.write_classdesc(classdesc)
        self._writeStruct(">i", 1, (len(obj),))

        # Add reference
        self.references.append(obj)
        logging.debug(
            "*** Adding ref 0x%X for array []",
            len(self.references) - 1 + StreamConstants.BASE_REFERENCE_IDX,
        )

        array_type_code = TypeCode(ord(classdesc.name[0]))
        assert array_type_code == TypeCode.TYPE_ARRAY
        type_code = TypeCode(ord(classdesc.name[1]))

        if type_code == TypeCode.TYPE_OBJECT:
            for o in obj:
                self._write_value(classdesc.name[1:], o)
        elif type_code == TypeCode.TYPE_ARRAY:
            for a in obj:
                self.write_array(a)
        else:
            log_debug("Write array of type {0}".format(chr(type_code.value)))
            for v in obj:
                log_debug("Writing: %s" % v)
                self._write_value(type_code, v)

    def _write_value(self, raw_field_type, value):
        """
        Writes an item of an array

        :param raw_field_type: Value type
        :param value: The value itself
        """
        if isinstance(raw_field_type, (TypeCode, int)):
            field_type = raw_field_type
        else:
            # We don't need details for arrays and objects
            field_type = TypeCode(ord(raw_field_type[0]))

        if field_type == TypeCode.TYPE_BOOLEAN:
            self._writeStruct(">B", 1, (1 if value else 0,))
        elif field_type == TypeCode.TYPE_BYTE:
            self._writeStruct(">b", 1, (value,))
        elif field_type == TypeCode.TYPE_CHAR:
            self._writeStruct(">H", 1, (ord(value),))
        elif field_type == TypeCode.TYPE_SHORT:
            self._writeStruct(">h", 1, (value,))
        elif field_type == TypeCode.TYPE_INTEGER:
            self._writeStruct(">i", 1, (value,))
        elif field_type == TypeCode.TYPE_LONG:
            self._writeStruct(">q", 1, (value,))
        elif field_type == TypeCode.TYPE_FLOAT:
            self._writeStruct(">f", 1, (value,))
        elif field_type == TypeCode.TYPE_DOUBLE:
            self._writeStruct(">d", 1, (value,))
        elif field_type in (TypeCode.TYPE_OBJECT, TypeCode.TYPE_ARRAY):
            if value is None:
                self.write_null()
            elif isinstance(value, JavaEnum):
                self.write_enum(value)
            elif isinstance(value, (JavaArray, JavaByteArray)):
                self.write_array(value)
            elif isinstance(value, JavaObject):
                self.write_object(value)
            elif isinstance(value, JavaString):
                self.write_string(value)
            elif isinstance(value, JavaClass):
                self.write_class(value)
            elif isinstance(value, (BYTES_TYPE, UNICODE_TYPE)):
                self.write_blockdata(value)
            else:
                raise RuntimeError("Unknown typecode: {0}".format(field_type))
        else:
            raise RuntimeError("Unknown typecode: {0}".format(field_type))

    @staticmethod
    def _convert_type_to_char(type_char):
        """
        Converts the given type code to an int

        :param type_char: A type code character
        """
        if isinstance(type_char, TypeCode):
            return type_char.value

        if isinstance(type_char, int):
            return type_char

        if isinstance(type_char, (BYTES_TYPE, UNICODE_TYPE)):
            # Conversion to TypeCode will raise an error if the type
            # is invalid
            return TypeCode(ord(type_char[0])).value

        raise RuntimeError(
            "Typecode {0} ({1}) isn't supported.".format(
                type_char, ord(type_char[0])
            )
        )

1	#!/usr/bin/python
2	# -- Content-Encoding: utf-8 --
3	"""	6✔
4	Provides functions for writing (writing is WIP currently) Java
5	objects that will be deserialized by ObjectOutputStream. This form of
6	object representation is a standard data interchange format in Java world.
7
8	javaobj module exposes an API familiar to users of the standard library
9	marshal, pickle and json modules.
10
11	See:
12	http://download.oracle.com/javase/6/docs/platform/serialization/spec/protocol.html
13
14	:authors: Volodymyr Buell, Thomas Calmant
15	:license: Apache License 2.0
16	:version: 0.4.4
17	:status: Alpha
18
19	..
20
21	Copyright 2024 Thomas Calmant
22
23	Licensed under the Apache License, Version 2.0 (the "License");
24	you may not use this file except in compliance with the License.
25	You may obtain a copy of the License at
26
27	http://www.apache.org/licenses/LICENSE-2.0
28
29	Unless required by applicable law or agreed to in writing, software
30	distributed under the License is distributed on an "AS IS" BASIS,
31	WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
32	See the License for the specific language governing permissions and
33	limitations under the License.
34	"""
35
36	from __future__ import absolute_import	6✔
37
38	# Standard library
39	import collections	6✔
40	import logging	6✔
41	import struct	6✔
42
43	try:	6✔
44	# Python 2
45	from StringIO import StringIO as BytesIO	6✔
46	except ImportError:	6✔
47	# Python 3+
48	from io import BytesIO	6✔
49
50	# Javaobj modules
51	from .beans import (	6✔
52	JavaClass,
53	JavaString,
54	JavaObject,
55	JavaByteArray,
56	JavaEnum,
57	JavaArray,
58	)
59	from ..constants import (	6✔
60	StreamConstants,
61	ClassDescFlags,
62	TerminalCode,
63	TypeCode,
64	)
65	from ..utils import (	6✔
66	log_debug,
67	log_error,
68	to_bytes,
69	BYTES_TYPE,
70	UNICODE_TYPE,
71	)
72
73	# ------------------------------------------------------------------------------
74
75	__all__ = ("JavaObjectMarshaller",)	6✔
76
77
78	# Module version
79	__version_info__ = (0, 4, 4)	6✔
80	__version__ = ".".join(str(x) for x in __version_info__)	6✔
81
82	# Documentation strings format
83	__docformat__ = "restructuredtext en"	6✔
84
85	# ------------------------------------------------------------------------------
86
87
88	class JavaObjectMarshaller:	6✔
89	"""
90	Serializes objects into Java serialization format
91	"""
92
93	def __init__(self, stream=None):	6✔
94	"""
95	Sets up members
96
97	:param stream: An output stream
98	"""
99	self.object_stream = stream	6✔
100	self.object_obj = None	6✔
101	self.object_transformers = []	6✔
102	self.references = []	6✔
103
104	def add_transformer(self, transformer):	6✔
105	"""
106	Appends an object transformer to the serialization process
107
108	:param transformer: An object with a transform(obj) method
109	"""
110	self.object_transformers.append(transformer)	×
111
112	def dump(self, obj):	6✔
113	"""
114	Dumps the given object in the Java serialization format
115	"""
116	self.references = []	6✔
117	self.object_obj = obj	6✔
118	self.object_stream = BytesIO()	6✔
119	self._writeStreamHeader()	6✔
120	self.writeObject(obj)	6✔
121	return self.object_stream.getvalue()	6✔
122
123	def _writeStreamHeader(self): # pylint:disable=C0103	6✔
124	"""
125	Writes the Java serialization magic header in the serialization stream
126	"""
127	self._writeStruct(	6✔
128	">HH",
129	4,
130	(StreamConstants.STREAM_MAGIC, StreamConstants.STREAM_VERSION),
131	)
132
133	def writeObject(self, obj): # pylint:disable=C0103	6✔
134	"""
135	Appends an object to the serialization stream
136
137	:param obj: A string or a deserialized Java object
138	:raise RuntimeError: Unsupported type
139	"""
140	log_debug("Writing object of type {0}".format(type(obj).__name__))	6✔
141	if isinstance(obj, JavaArray):	6✔
142	# Deserialized Java array
143	self.write_array(obj)	6✔
144	elif isinstance(obj, JavaByteArray):	6✔
145	# Deserialized Java byte array
146	self.write_array(obj)	×
147	elif isinstance(obj, JavaEnum):	6✔
148	# Deserialized Java Enum
149	self.write_enum(obj)	×
150	elif isinstance(obj, JavaObject):	6✔
151	# Deserialized Java object
152	self.write_object(obj)	6✔
153	elif isinstance(obj, JavaString):	6✔
154	# Deserialized String
155	self.write_string(obj)	6✔
156	elif isinstance(obj, JavaClass):	6✔
157	# Java class
158	self.write_class(obj)	6✔
159	elif obj is None:	6✔
160	# Null
161	self.write_null()	×
162	elif type(obj) is str: # pylint:disable=C0123	6✔
163	# String value
164	self.write_blockdata(obj)	6✔
165	else:
166	# Unhandled type
167	raise RuntimeError(	×
168	"Object serialization of type {0} is not "
169	"supported.".format(type(obj))
170	)
171
172	def _writeStruct(self, unpack, length, args): # pylint:disable=C0103	6✔
173	"""
174	Appends data to the serialization stream
175
176	:param unpack: Struct format string
177	:param length: Unused
178	:param args: Struct arguments
179	"""
180	ba = struct.pack(unpack, *args)	6✔
181	self.object_stream.write(ba)	6✔
182
183	def _writeString(self, obj, use_reference=True): # pylint:disable=C0103	6✔
184	"""
185	Appends a string to the serialization stream
186
187	:param obj: String to serialize
188	:param use_reference: If True, allow writing a reference
189	"""
190	# TODO: Convert to "modified UTF-8"
191	# http://docs.oracle.com/javase/7/docs/api/java/io/DataInput.html#modified-utf-8
192	string = to_bytes(obj, "utf-8")	6✔
193
194	if use_reference and isinstance(obj, JavaString):	6✔
195	try:	6✔
196	idx = self.references.index(obj)	6✔
197	except ValueError:	6✔
198	# First appearance of the string
199	self.references.append(obj)	6✔
200	logging.debug(	6✔
201	"*** Adding ref 0x%X for string: %s",
202	len(self.references)
203	- 1
204	+ StreamConstants.BASE_REFERENCE_IDX,
205	obj,
206	)
207
208	self._writeStruct(">H", 2, (len(string),))	6✔
209	self.object_stream.write(string)	6✔
210	else:
211	# Write a reference to the previous type
212	logging.debug(	×
213	"*** Reusing ref 0x%X for string: %s",
214	idx + StreamConstants.BASE_REFERENCE_IDX,
215	obj,
216	)
217	self.write_reference(idx)	×
218	else:
219	self._writeStruct(">H", 2, (len(string),))	6✔
220	self.object_stream.write(string)	6✔
221
222	def write_string(self, obj, use_reference=True):	6✔
223	"""
224	Writes a Java string with the TC_STRING type marker
225
226	:param obj: The string to print
227	:param use_reference: If True, allow writing a reference
228	"""
229	if use_reference and isinstance(obj, JavaString):	6✔
230	try:	6✔
231	idx = self.references.index(obj)	6✔
232	except ValueError:	6✔
233	# String is not referenced: let _writeString store it
234	self._writeStruct(">B", 1, (TerminalCode.TC_STRING,))	6✔
235	self._writeString(obj, use_reference)	6✔
236	else:
237	# Reuse the referenced string
238	logging.debug(	6✔
239	"*** Reusing ref 0x%X for String: %s",
240	idx + StreamConstants.BASE_REFERENCE_IDX,
241	obj,
242	)
243	self.write_reference(idx)	6✔
244	else:
245	# Don't use references
246	self._writeStruct(">B", 1, (TerminalCode.TC_STRING,))	6✔
247	self._writeString(obj, use_reference)	6✔
248
249	def write_enum(self, obj):	6✔
250	"""
251	Writes an Enum value
252
253	:param obj: A JavaEnum object
254	"""
255	# FIXME: the output doesn't have the same references as the real
256	# serializable form
257	self._writeStruct(">B", 1, (TerminalCode.TC_ENUM,))	×
258
259	try:	×
260	idx = self.references.index(obj)	×
261	except ValueError:	×
262	# New reference
263	self.references.append(obj)	×
264	logging.debug(	×
265	"*** Adding ref 0x%X for enum: %s",
266	len(self.references) - 1 + StreamConstants.BASE_REFERENCE_IDX,
267	obj,
268	)
269
270	self.write_classdesc(obj.get_class())	×
271	else:
272	self.write_reference(idx)	×
273
274	self.write_string(obj.constant)	×
275
276	def write_blockdata(self, obj, parent=None): # pylint:disable=W0613	6✔
277	"""
278	Appends a block of data to the serialization stream
279
280	:param obj: String form of the data block
281	"""
282	if isinstance(obj, UNICODE_TYPE):	6✔
283	# Latin-1: keep bytes as is
284	obj = to_bytes(obj, "latin-1")	6✔
285
286	length = len(obj)	6✔
287	if length <= 256:	6✔
288	# Small block data
289	# TC_BLOCKDATA (unsigned byte)<size> (byte)[size]
290	self._writeStruct(">B", 1, (TerminalCode.TC_BLOCKDATA,))	6✔
291	self._writeStruct(">B", 1, (length,))	6✔
292	else:
293	# Large block data
294	# TC_BLOCKDATALONG (unsigned int)<size> (byte)[size]
295	self._writeStruct(">B", 1, (TerminalCode.TC_BLOCKDATALONG,))	×
296	self._writeStruct(">I", 1, (length,))	×
297
298	self.object_stream.write(obj)	6✔
299
300	def write_null(self):	6✔
301	"""
302	Writes a "null" value
303	"""
304	self._writeStruct(">B", 1, (TerminalCode.TC_NULL,))	6✔
305
306	def write_object(self, obj, parent=None):	6✔
307	"""
308	Writes an object header to the serialization stream
309
310	:param obj: Not yet used
311	:param parent: Not yet used
312	"""
313	# Transform object
314	for transformer in self.object_transformers:	6✔
315	tmp_object = transformer.transform(obj)	×
316	if tmp_object is not obj:	×
317	obj = tmp_object	×
318	break	×
319
320	self._writeStruct(">B", 1, (TerminalCode.TC_OBJECT,))	6✔
321	cls = obj.get_class()	6✔
322	self.write_classdesc(cls)	6✔
323
324	# Add reference
325	self.references.append([])	6✔
326	logging.debug(	6✔
327	"*** Adding ref 0x%X for object %s",
328	len(self.references) - 1 + StreamConstants.BASE_REFERENCE_IDX,
329	obj,
330	)
331
332	all_names = collections.deque()	6✔
333	all_types = collections.deque()	6✔
334	tmpcls = cls	6✔
335	while tmpcls:	6✔
336	all_names.extendleft(reversed(tmpcls.fields_names))	6✔
337	all_types.extendleft(reversed(tmpcls.fields_types))	6✔
338	tmpcls = tmpcls.superclass	6✔
339	del tmpcls	6✔
340
341	logging.debug("<=> Field names: %s", all_names)	6✔
342	logging.debug("<=> Field types: %s", all_types)	6✔
343
344	for field_name, field_type in zip(all_names, all_types):	6✔
345	try:	6✔
346	logging.debug(	6✔
347	"Writing field %s (%s): %s",
348	field_name,
349	field_type,
350	getattr(obj, field_name),
351	)
352	self._write_value(field_type, getattr(obj, field_name))	6✔
353	except AttributeError as ex:	×
354	log_error(	×
355	"No attribute {0} for object {1}\nDir: {2}".format(
356	ex, repr(obj), dir(obj)
357	)
358	)
359	raise	×
360	del all_names, all_types	6✔
361
362	if (	6✔
363	cls.flags & ClassDescFlags.SC_SERIALIZABLE
364	and cls.flags & ClassDescFlags.SC_WRITE_METHOD
365	or cls.flags & ClassDescFlags.SC_EXTERNALIZABLE
366	and cls.flags & ClassDescFlags.SC_BLOCK_DATA
367	):
368	for annotation in obj.annotations:	6✔
369	log_debug(	×
370	"Write annotation {0} for {1}".format(
371	repr(annotation), repr(obj)
372	)
373	)
374	if annotation is None:	×
375	self.write_null()	×
376	else:
377	self.writeObject(annotation)	×
378	self._writeStruct(">B", 1, (TerminalCode.TC_ENDBLOCKDATA,))	6✔
379
380	def write_class(self, obj, parent=None): # pylint:disable=W0613	6✔
381	"""
382	Writes a class to the stream
383
384	:param obj: A JavaClass object
385	:param parent:
386	"""
387	self._writeStruct(">B", 1, (TerminalCode.TC_CLASS,))	6✔
388	self.write_classdesc(obj)	6✔
389
390	def write_classdesc(self, obj, parent=None): # pylint:disable=W0613	6✔
391	"""
392	Writes a class description
393
394	:param obj: Class description to write
395	:param parent:
396	"""
397	if obj not in self.references:	6✔
398	# Add reference
399	self.references.append(obj)	6✔
400	logging.debug(	6✔
401	"*** Adding ref 0x%X for classdesc %s",
402	len(self.references) - 1 + StreamConstants.BASE_REFERENCE_IDX,
403	obj.name,
404	)
405
406	self._writeStruct(">B", 1, (TerminalCode.TC_CLASSDESC,))	6✔
407	self._writeString(obj.name)	6✔
408	self._writeStruct(">qB", 1, (obj.serialVersionUID, obj.flags))	6✔
409	self._writeStruct(">H", 1, (len(obj.fields_names),))	6✔
410
411	for field_name, field_type in zip(	6✔
412	obj.fields_names, obj.fields_types
413	):
414	self._writeStruct(	6✔
415	">B", 1, (self._convert_type_to_char(field_type),)
416	)
417	self._writeString(field_name)	6✔
418	if ord(field_type[0]) in (	6✔
419	TypeCode.TYPE_OBJECT,
420	TypeCode.TYPE_ARRAY,
421	):
422	try:	6✔
423	idx = self.references.index(field_type)	6✔
424	except ValueError:	6✔
425	# First appearance of the type
426	self.references.append(field_type)	6✔
427	logging.debug(	6✔
428	"*** Adding ref 0x%X for field type %s",
429	len(self.references)
430	- 1
431	+ StreamConstants.BASE_REFERENCE_IDX,
432	field_type,
433	)
434
435	self.write_string(field_type, False)	6✔
436	else:
437	# Write a reference to the previous type
438	logging.debug(	6✔
439	"*** Reusing ref 0x%X for %s (%s)",
440	idx + StreamConstants.BASE_REFERENCE_IDX,
441	field_type,
442	field_name,
443	)
444	self.write_reference(idx)	6✔
445
446	self._writeStruct(">B", 1, (TerminalCode.TC_ENDBLOCKDATA,))	6✔
447	if obj.superclass:	6✔
448	self.write_classdesc(obj.superclass)	6✔
449	else:
450	self.write_null()	6✔
451	else:
452	# Use reference
453	self.write_reference(self.references.index(obj))	6✔
454
455	def write_reference(self, ref_index):	6✔
456	"""
457	Writes a reference
458	:param ref_index: Local index (0-based) to the reference
459	"""
460	self._writeStruct(	6✔
461	">BL",
462	1,
463	(
464	TerminalCode.TC_REFERENCE,
465	ref_index + StreamConstants.BASE_REFERENCE_IDX,
466	),
467	)
468
469	def write_array(self, obj):	6✔
470	"""
471	Writes a JavaArray
472
473	:param obj: A JavaArray object
474	"""
475	classdesc = obj.get_class()	6✔
476	self._writeStruct(">B", 1, (TerminalCode.TC_ARRAY,))	6✔
477	self.write_classdesc(classdesc)	6✔
478	self._writeStruct(">i", 1, (len(obj),))	6✔
479
480	# Add reference
481	self.references.append(obj)	6✔
482	logging.debug(	6✔
483	"*** Adding ref 0x%X for array []",
484	len(self.references) - 1 + StreamConstants.BASE_REFERENCE_IDX,
485	)
486
487	array_type_code = TypeCode(ord(classdesc.name[0]))	6✔
488	assert array_type_code == TypeCode.TYPE_ARRAY	6✔
489	type_code = TypeCode(ord(classdesc.name[1]))	6✔
490
491	if type_code == TypeCode.TYPE_OBJECT:	6✔
492	for o in obj:	6✔
493	self._write_value(classdesc.name[1:], o)	6✔
494	elif type_code == TypeCode.TYPE_ARRAY:	6✔
495	for a in obj:	×
496	self.write_array(a)	×
497	else:
498	log_debug("Write array of type {0}".format(chr(type_code.value)))	6✔
499	for v in obj:	6✔
500	log_debug("Writing: %s" % v)	6✔
501	self._write_value(type_code, v)	6✔
502
503	def _write_value(self, raw_field_type, value):	6✔
504	"""
505	Writes an item of an array
506
507	:param raw_field_type: Value type
508	:param value: The value itself
509	"""
510	if isinstance(raw_field_type, (TypeCode, int)):	6✔
511	field_type = raw_field_type	6✔
512	else:
513	# We don't need details for arrays and objects
514	field_type = TypeCode(ord(raw_field_type[0]))	6✔
515
516	if field_type == TypeCode.TYPE_BOOLEAN:	6✔
517	self._writeStruct(">B", 1, (1 if value else 0,))	6✔
518	elif field_type == TypeCode.TYPE_BYTE:	6✔
519	self._writeStruct(">b", 1, (value,))	6✔
520	elif field_type == TypeCode.TYPE_CHAR:	6✔
521	self._writeStruct(">H", 1, (ord(value),))	6✔
522	elif field_type == TypeCode.TYPE_SHORT:	6✔
523	self._writeStruct(">h", 1, (value,))	×
524	elif field_type == TypeCode.TYPE_INTEGER:	6✔
525	self._writeStruct(">i", 1, (value,))	6✔
526	elif field_type == TypeCode.TYPE_LONG:	6✔
527	self._writeStruct(">q", 1, (value,))	×
528	elif field_type == TypeCode.TYPE_FLOAT:	6✔
529	self._writeStruct(">f", 1, (value,))	×
530	elif field_type == TypeCode.TYPE_DOUBLE:	6✔
531	self._writeStruct(">d", 1, (value,))	×
532	elif field_type in (TypeCode.TYPE_OBJECT, TypeCode.TYPE_ARRAY):	6✔
533	if value is None:	6✔
534	self.write_null()	6✔
535	elif isinstance(value, JavaEnum):	6✔
536	self.write_enum(value)	×
537	elif isinstance(value, (JavaArray, JavaByteArray)):	6✔
538	self.write_array(value)	6✔
539	elif isinstance(value, JavaObject):	6✔
540	self.write_object(value)	6✔
541	elif isinstance(value, JavaString):	6✔
542	self.write_string(value)	6✔
543	elif isinstance(value, JavaClass):	×
544	self.write_class(value)	×
545	elif isinstance(value, (BYTES_TYPE, UNICODE_TYPE)):	×
546	self.write_blockdata(value)	×
547	else:
548	raise RuntimeError("Unknown typecode: {0}".format(field_type))	×
549	else:
550	raise RuntimeError("Unknown typecode: {0}".format(field_type))	×
551
552	@staticmethod	6✔
553	def _convert_type_to_char(type_char):	5✔
554	"""
555	Converts the given type code to an int
556
557	:param type_char: A type code character
558	"""
559	if isinstance(type_char, TypeCode):	6✔
560	return type_char.value	×
561
562	if isinstance(type_char, int):	6✔
563	return type_char	×
564
565	if isinstance(type_char, (BYTES_TYPE, UNICODE_TYPE)):	6✔
566	# Conversion to TypeCode will raise an error if the type
567	# is invalid
568	return TypeCode(ord(type_char[0])).value	6✔
569
570	raise RuntimeError(	×
571	"Typecode {0} ({1}) isn't supported.".format(
572	type_char, ord(type_char[0])
573	)
574	)

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