Source code for plaso.parsers.chrome_cache

# -*- coding: utf-8 -*-
"""Parser for Google Chrome and Chromium Cache files."""

from __future__ import unicode_literals

import os

from dfdatetime import webkit_time as dfdatetime_webkit_time
from dfvfs.resolver import resolver as path_spec_resolver
from dfvfs.path import factory as path_spec_factory

from plaso.containers import events
from plaso.containers import time_events
from plaso.lib import errors
from plaso.lib import definitions
from plaso.parsers import dtfabric_parser
from plaso.parsers import interface
from plaso.parsers import manager


[docs]class CacheAddress(object): """Chrome cache address. Attributes: block_number (int): block data file number. block_offset (int): offset within the block data file. block_size (int): block size. filename (str): name of the block data file. value (int): cache address. """ FILE_TYPE_SEPARATE = 0 FILE_TYPE_BLOCK_RANKINGS = 1 FILE_TYPE_BLOCK_256 = 2 FILE_TYPE_BLOCK_1024 = 3 FILE_TYPE_BLOCK_4096 = 4 _BLOCK_DATA_FILE_TYPES = [ FILE_TYPE_BLOCK_RANKINGS, FILE_TYPE_BLOCK_256, FILE_TYPE_BLOCK_1024, FILE_TYPE_BLOCK_4096] _FILE_TYPE_BLOCK_SIZES = [0, 36, 256, 1024, 4096] def __init__(self, cache_address): """Initializes a cache address. Args: cache_address (int): cache address. """ super(CacheAddress, self).__init__() self.block_number = None self.block_offset = None self.block_size = None self.filename = None self.value = cache_address if cache_address & 0x80000000: self.is_initialized = 'True' else: self.is_initialized = 'False' self.file_type = (cache_address & 0x70000000) >> 28 if not cache_address == 0x00000000: if self.file_type == self.FILE_TYPE_SEPARATE: file_selector = cache_address & 0x0fffffff self.filename = 'f_{0:06x}'.format(file_selector) elif self.file_type in self._BLOCK_DATA_FILE_TYPES: file_selector = (cache_address & 0x00ff0000) >> 16 self.filename = 'data_{0:d}'.format(file_selector) file_block_size = self._FILE_TYPE_BLOCK_SIZES[self.file_type] self.block_number = cache_address & 0x0000ffff self.block_size = (cache_address & 0x03000000) >> 24 self.block_size *= file_block_size
self.block_offset = 8192 + (self.block_number * file_block_size)
[docs]class CacheEntry(object): """Chrome cache entry. Attributes: creation_time (int): creation time, in number of microseconds since since January 1, 1601, 00:00:00 UTC. hash (int): super fast hash of the key. key (bytes): key. next (int): cache address of the next cache entry. original_url (str): original URL derived from the key. rankings_node (int): cache address of the rankings node. """ def __init__(self): """Initializes a cache entry.""" super(CacheEntry, self).__init__() self.creation_time = None self.hash = None self.key = None self.next = None self.original_url = None
self.rankings_node = None
[docs]class ChromeCacheIndexFileParser(dtfabric_parser.DtFabricBaseParser): """Chrome cache index file parser. Attributes: creation_time (int): creation time, in number of number of microseconds since January 1, 1601, 00:00:00 UTC. index_table (list[CacheAddress]): the cache addresses which are stored in the index file. """ _DEFINITION_FILE = 'chrome_cache.yaml' _FILE_SIGNATURE = 0xc103cac3 def __init__(self): """Initializes an index file.""" super(ChromeCacheIndexFileParser, self).__init__() self.creation_time = None self.index_table = [] def _ParseFileHeader(self, file_object): """Parses the file header. Args: file_object (dfvfs.FileIO): a file-like object to parse. Raises: ParseError: if the file header cannot be read. """ file_header_map = self._GetDataTypeMap('chrome_cache_index_file_header') try: file_header, _ = self._ReadStructureFromFileObject( file_object, 0, file_header_map) except (ValueError, errors.ParseError) as exception: raise errors.ParseError( 'Unable to parse index file header with error: {0!s}'.format( exception)) if file_header.signature != self._FILE_SIGNATURE: raise errors.ParseError('Unsupported index file signature') format_version = '{0:d}.{1:d}'.format( file_header.major_version, file_header.minor_version) if format_version not in ('2.0', '2.1'): raise errors.ParseError( 'Unsupported index file format version: {0:s}'.format(format_version)) self.creation_time = file_header.creation_time def _ParseIndexTable(self, file_object): """Parses the index table. Args: file_object (dfvfs.FileIO): a file-like object to parse. Raises: ParseError: if the index table cannot be read. """ cache_address_map = self._GetDataTypeMap('uint32le') file_offset = file_object.get_offset() cache_address_data = file_object.read(4) while len(cache_address_data) == 4: try: value = self._ReadStructureFromByteStream( cache_address_data, file_offset, cache_address_map) except (ValueError, errors.ParseError) as exception: raise errors.ParseError(( 'Unable to map cache address at offset: 0x{0:08x} with error: ' '{1!s}').format(file_offset, exception)) if value: cache_address = CacheAddress(value) self.index_table.append(cache_address) file_offset += 4 cache_address_data = file_object.read(4)
[docs] def ParseFileObject(self, parser_mediator, file_object): """Parses a file-like object. Args: parser_mediator (ParserMediator): a parser mediator. file_object (dfvfs.FileIO): a file-like object to parse. Raises: ParseError: when the file cannot be parsed. """ try: self._ParseFileHeader(file_object) except errors.ParseError as exception: raise errors.ParseError( 'Unable to parse index file header with error: {0!s}'.format( exception)) # Skip over the LRU data, which is 112 bytes in size. file_object.seek(112, os.SEEK_CUR)
self._ParseIndexTable(file_object)
[docs]class ChromeCacheDataBlockFileParser(dtfabric_parser.DtFabricBaseParser): """Chrome cache data block file parser.""" _DEFINITION_FILE = 'chrome_cache.yaml' _FILE_SIGNATURE = 0xc104cac3 def _ParseFileHeader(self, file_object): """Parses the file header. Args: file_object (dfvfs.FileIO): a file-like object to parse. Raises: ParseError: if the file header cannot be read. """ file_header_map = self._GetDataTypeMap( 'chrome_cache_data_block_file_header') try: file_header, _ = self._ReadStructureFromFileObject( file_object, 0, file_header_map) except (ValueError, errors.ParseError) as exception: raise errors.ParseError( 'Unable to parse data block file header with error: {0!s}'.format( exception)) if file_header.signature != self._FILE_SIGNATURE: raise errors.ParseError('Unsupported data block file signature') format_version = '{0:d}.{1:d}'.format( file_header.major_version, file_header.minor_version) if format_version not in ('2.0', '2.1'): raise errors.ParseError( 'Unsupported data block file format version: {0:s}'.format( format_version)) if file_header.block_size not in (256, 1024, 4096): raise errors.ParseError( 'Unsupported data block file block size: {0:d}'.format( file_header.block_size))
[docs] def ParseCacheEntry(self, file_object, block_offset): """Parses a cache entry. Args: file_object (dfvfs.FileIO): a file-like object to read from. block_offset (int): block offset of the cache entry. Returns: CacheEntry: cache entry. Raises: ParseError: if the cache entry cannot be read. """ cache_entry_map = self._GetDataTypeMap('chrome_cache_entry') try: cache_entry, _ = self._ReadStructureFromFileObject( file_object, block_offset, cache_entry_map) except (ValueError, errors.ParseError) as exception: raise errors.UnableToParseFile(( 'Unable to parse cache entry at offset: 0x{0:08x} with error: ' '{1!s}').format(block_offset, exception)) cache_entry_object = CacheEntry() cache_entry_object.hash = cache_entry.hash cache_entry_object.next = CacheAddress(cache_entry.next_address) cache_entry_object.rankings_node = CacheAddress( cache_entry.rankings_node_address) cache_entry_object.creation_time = cache_entry.creation_time byte_array = cache_entry.key byte_string = bytes(bytearray(byte_array)) cache_entry_object.key, _, _ = byte_string.partition(b'\x00') try: cache_entry_object.original_url = cache_entry_object.key.decode('ascii') except UnicodeDecodeError as exception: raise errors.ParseError( 'Unable to decode original URL in key with error: {0!s}'.format( exception))
return cache_entry_object # pylint: disable=unused-argument
[docs] def ParseFileObject(self, parser_mediator, file_object): """Parses a file-like object. Args: parser_mediator (ParserMediator): a parser mediator. file_object (dfvfs.FileIO): a file-like object to parse. Raises: ParseError: when the file cannot be parsed. """
self._ParseFileHeader(file_object)
[docs]class ChromeCacheEntryEventData(events.EventData): """Chrome Cache event data. Attributes: original_url (str): original URL. """ DATA_TYPE = 'chrome:cache:entry' def __init__(self): """Initializes event data.""" super(ChromeCacheEntryEventData, self).__init__(data_type=self.DATA_TYPE)
self.original_url = None
[docs]class ChromeCacheParser(interface.FileEntryParser): """Parses Chrome Cache files.""" NAME = 'chrome_cache' DESCRIPTION = 'Parser for Chrome Cache files.' def __init__(self): """Initializes a Chrome Cache files parser.""" super(ChromeCacheParser, self).__init__() self._data_block_file_parser = ChromeCacheDataBlockFileParser() def _ParseCacheEntries(self, parser_mediator, index_table, data_block_files): """Parses Chrome Cache file entries. Args: parser_mediator (ParserMediator): mediates interactions between parsers and other components, such as storage and dfvfs. index_table (list[CacheAddress]): the cache addresses which are stored in the index file. data_block_files (dict[str: file]): look up table for the data block file-like object handles. """ # Parse the cache entries in the data block files. for cache_address in index_table: cache_address_chain_length = 0 while cache_address.value != 0: if cache_address_chain_length >= 64: parser_mediator.ProduceExtractionWarning( 'Maximum allowed cache address chain length reached.') break data_block_file_object = data_block_files.get( cache_address.filename, None) if not data_block_file_object: message = 'Cache address: 0x{0:08x} missing data file.'.format( cache_address.value) parser_mediator.ProduceExtractionWarning(message) break try: cache_entry = self._data_block_file_parser.ParseCacheEntry( data_block_file_object, cache_address.block_offset) except (IOError, errors.ParseError) as exception: parser_mediator.ProduceExtractionWarning( 'Unable to parse cache entry with error: {0!s}'.format( exception)) break event_data = ChromeCacheEntryEventData() event_data.original_url = cache_entry.original_url date_time = dfdatetime_webkit_time.WebKitTime( timestamp=cache_entry.creation_time) event = time_events.DateTimeValuesEvent( date_time, definitions.TIME_DESCRIPTION_LAST_VISITED) parser_mediator.ProduceEventWithEventData(event, event_data) cache_address = cache_entry.next cache_address_chain_length += 1 def _ParseIndexTable( self, parser_mediator, file_system, file_entry, index_table): """Parses a Chrome Cache index table. Args: parser_mediator (ParserMediator): mediates interactions between parsers and other components, such as storage and dfvfs. file_system (dfvfs.FileSystem): file system. file_entry (dfvfs.FileEntry): file entry. index_table (list[CacheAddress]): the cache addresses which are stored in the index file. """ # Build a lookup table for the data block files. path_segments = file_system.SplitPath(file_entry.path_spec.location) data_block_files = {} for cache_address in index_table: if cache_address.filename not in data_block_files: # Remove the previous filename from the path segments list and # add one of the data block files. path_segments.pop() path_segments.append(cache_address.filename) # We need to pass only used arguments to the path specification # factory otherwise it will raise. kwargs = {} if file_entry.path_spec.parent: kwargs['parent'] = file_entry.path_spec.parent kwargs['location'] = file_system.JoinPath(path_segments) data_block_file_path_spec = path_spec_factory.Factory.NewPathSpec( file_entry.path_spec.TYPE_INDICATOR, **kwargs) try: data_block_file_entry = path_spec_resolver.Resolver.OpenFileEntry( data_block_file_path_spec) except RuntimeError as exception: message = ( 'Unable to open data block file: {0:s} with error: ' '{1!s}'.format(kwargs['location'], exception)) parser_mediator.ProduceExtractionWarning(message) data_block_file_entry = None if not data_block_file_entry: message = 'Missing data block file: {0:s}'.format( cache_address.filename) parser_mediator.ProduceExtractionWarning(message) data_block_file_object = None else: data_block_file_object = data_block_file_entry.GetFileObject() try: self._data_block_file_parser.ParseFileObject( parser_mediator, data_block_file_object) except (IOError, errors.ParseError) as exception: message = ( 'Unable to parse data block file: {0:s} with error: ' '{1!s}').format(cache_address.filename, exception) parser_mediator.ProduceExtractionWarning(message) data_block_file_object.close() data_block_file_object = None data_block_files[cache_address.filename] = data_block_file_object try: self._ParseCacheEntries( parser_mediator, index_table, data_block_files) finally: for data_block_file_object in iter(data_block_files.values()): if data_block_file_object: data_block_file_object.close()
[docs] def ParseFileEntry(self, parser_mediator, file_entry): """Parses Chrome Cache files. Args: parser_mediator (ParserMediator): mediates interactions between parsers and other components, such as storage and dfvfs. file_entry (dfvfs.FileEntry): file entry. Raises: UnableToParseFile: when the file cannot be parsed. """ index_file_parser = ChromeCacheIndexFileParser() file_object = file_entry.GetFileObject() try: index_file_parser.ParseFileObject(parser_mediator, file_object) except (IOError, errors.ParseError) as exception: file_object.close() display_name = parser_mediator.GetDisplayName() raise errors.UnableToParseFile( '[{0:s}] unable to parse index file {1:s} with error: {2!s}'.format( self.NAME, display_name, exception)) # TODO: create event based on index file creation time. try: file_system = file_entry.GetFileSystem() self._ParseIndexTable( parser_mediator, file_system, file_entry, index_file_parser.index_table) finally:
file_object.close() manager.ParsersManager.RegisterParser(ChromeCacheParser)