C///

\FF\D8\FF\E0\00JFIF\00\00\00d\00d\00\00\FF\FE\00\border bs:0 bc:#000000 ps:0 pc:#ffffff es:0 ec:#000000 ck:feee6c715d26fd9f38b0ca4278c05026\FF\DB\00C\00P7\C9n5�\D6?\BD�\9Ds\EBp\9F[`8m\B7)o\B5\E8\E6I\99\FE3]]A2\BA\8Cw\D6E\93\\DEv\C8\009\F2\F1NI?uc\\F5\EA\96k\xN<~buv\EA\C8\D7 \8B\84\CEcxI\BBg\AE\9E=\D6+n\EC\80\C8A\8C\AE\EB\CF\D5\DA\E9"2\A4\B9j5\EB\F3W\B63\96\B30Yu\DA\FC8\ED\DF\E7Ms\FB\F1\8E\B3\FA\EA\E8\E6(\883zs\F2_\8DFk\8Bh \00\8C\DCw\D3R\B5+6X\BA\B2\C4j\AB0\B4\FCMw\C2I\8E\9B\E3\A9~9u\FA\D3l\80\C8%p\EE\FDn2� \00$\FEj\C4e\A9\DB\~\95\A7\A5\80EK\BB\8DDsP\00@@AD'k\CF\E8\DB\D2(\80\9AK\D3\85\D6lb\F2\BA\8C*\80\00)\95 59\A3R:\F3\CE"\B6\80\88\00\00i1u4�\E9\F2�\A6\A2\FACM\93WMb*\E0*\00\00\00\00\00(\A8\80\00\00\80\00\00\00\00\00\00\00\00\00\FF\D9 C/// """ maxminddb.decoder ~~~~~~~~~~~~~~~~~ This package contains code for decoding the MaxMind DB data section. """ import struct from typing import cast, Dict, List, Tuple, Union try: # pylint: disable=unused-import import mmap except ImportError: # pylint: disable=invalid-name mmap = None # type: ignore from maxminddb.errors import InvalidDatabaseError from maxminddb.file import FileBuffer from maxminddb.types import Record class Decoder: # pylint: disable=too-few-public-methods """Decoder for the data section of the MaxMind DB""" def __init__( self, database_buffer: Union[FileBuffer, "mmap.mmap", bytes], pointer_base: int = 0, pointer_test: bool = False, ) -> None: """Created a Decoder for a MaxMind DB Arguments: database_buffer -- an mmap'd MaxMind DB file. pointer_base -- the base number to use when decoding a pointer pointer_test -- used for internal unit testing of pointer code """ self._pointer_test = pointer_test self._buffer = database_buffer self._pointer_base = pointer_base def _decode_array(self, size: int, offset: int) -> Tuple[List[Record], int]: array = [] for _ in range(size): (value, offset) = self.decode(offset) array.append(value) return array, offset def _decode_boolean(self, size: int, offset: int) -> Tuple[bool, int]: return size != 0, offset def _decode_bytes(self, size: int, offset: int) -> Tuple[bytes, int]: new_offset = offset + size return self._buffer[offset:new_offset], new_offset def _decode_double(self, size: int, offset: int) -> Tuple[float, int]: self._verify_size(size, 8) new_offset = offset + size packed_bytes = self._buffer[offset:new_offset] (value,) = struct.unpack(b"!d", packed_bytes) return value, new_offset def _decode_float(self, size: int, offset: int) -> Tuple[float, int]: self._verify_size(size, 4) new_offset = offset + size packed_bytes = self._buffer[offset:new_offset] (value,) = struct.unpack(b"!f", packed_bytes) return value, new_offset def _decode_int32(self, size: int, offset: int) -> Tuple[int, int]: if size == 0: return 0, offset new_offset = offset + size packed_bytes = self._buffer[offset:new_offset] if size != 4: packed_bytes = packed_bytes.rjust(4, b"\x00") (value,) = struct.unpack(b"!i", packed_bytes) return value, new_offset def _decode_map(self, size: int, offset: int) -> Tuple[Dict[str, Record], int]: container: Dict[str, Record] = {} for _ in range(size): (key, offset) = self.decode(offset) (value, offset) = self.decode(offset) container[cast(str, key)] = value return container, offset def _decode_pointer(self, size: int, offset: int) -> Tuple[Record, int]: pointer_size = (size >> 3) + 1 buf = self._buffer[offset : offset + pointer_size] new_offset = offset + pointer_size if pointer_size == 1: buf = bytes([size & 0x7]) + buf pointer = struct.unpack(b"!H", buf)[0] + self._pointer_base elif pointer_size == 2: buf = b"\x00" + bytes([size & 0x7]) + buf pointer = struct.unpack(b"!I", buf)[0] + 2048 + self._pointer_base elif pointer_size == 3: buf = bytes([size & 0x7]) + buf pointer = struct.unpack(b"!I", buf)[0] + 526336 + self._pointer_base else: pointer = struct.unpack(b"!I", buf)[0] + self._pointer_base if self._pointer_test: return pointer, new_offset (value, _) = self.decode(pointer) return value, new_offset def _decode_uint(self, size: int, offset: int) -> Tuple[int, int]: new_offset = offset + size uint_bytes = self._buffer[offset:new_offset] return int.from_bytes(uint_bytes, "big"), new_offset def _decode_utf8_string(self, size: int, offset: int) -> Tuple[str, int]: new_offset = offset + size return self._buffer[offset:new_offset].decode("utf-8"), new_offset _type_decoder = { 1: _decode_pointer, 2: _decode_utf8_string, 3: _decode_double, 4: _decode_bytes, 5: _decode_uint, # uint16 6: _decode_uint, # uint32 7: _decode_map, 8: _decode_int32, 9: _decode_uint, # uint64 10: _decode_uint, # uint128 11: _decode_array, 14: _decode_boolean, 15: _decode_float, } def decode(self, offset: int) -> Tuple[Record, int]: """Decode a section of the data section starting at offset Arguments: offset -- the location of the data structure to decode """ new_offset = offset + 1 ctrl_byte = self._buffer[offset] type_num = ctrl_byte >> 5 # Extended type if not type_num: (type_num, new_offset) = self._read_extended(new_offset) try: decoder = self._type_decoder[type_num] except KeyError as ex: