diff --git a/config/dpkg/control b/config/dpkg/control index 4b3621c2..c43e586b 100644 --- a/config/dpkg/control +++ b/config/dpkg/control @@ -9,7 +9,7 @@ Homepage: https://github.com/log2timeline/dfvfs Package: python3-dfvfs Architecture: all -Depends: libbde-python3 (>= 20220121), libcaes-python3 (>= 20240114), libewf-python3 (>= 20131210), libfcrypto-python3 (>= 20240114), libfsapfs-python3 (>= 20220709), libfsext-python3 (>= 20220829), libfsfat-python3 (>= 20220925), libfshfs-python3 (>= 20220831), libfsntfs-python3 (>= 20211229), libfsxfs-python3 (>= 20260702), libfvde-python3 (>= 20220121), libfwnt-python3 (>= 20210717), libluksde-python3 (>= 20220121), libmodi-python3 (>= 20210405), libphdi-python3 (>= 20220228), libqcow-python3 (>= 20201213), libsigscan-python3 (>= 20230109), libsmdev-python3 (>= 20140529), libsmraw-python3 (>= 20140612), libvhdi-python3 (>= 20201014), libvmdk-python3 (>= 20140421), libvsapm-python3 (>= 20260713), libvsgpt-python3 (>= 20211115), libvshadow-python3 (>= 20160109), libvslvm-python3 (>= 20160109), python3-dfdatetime (>= 20221112), python3-dtfabric (>= 20230518), python3-pytsk3 (>= 20210419), python3-xattr (>= 0.7.2), python3-yaml (>= 3.10), ${misc:Depends} +Depends: libbde-python3 (>= 20220121), libcaes-python3 (>= 20240114), libewf-python3 (>= 20131210), libfcrypto-python3 (>= 20240114), libfsapfs-python3 (>= 20220709), libfsext-python3 (>= 20220829), libfsfat-python3 (>= 20220925), libfshfs-python3 (>= 20220831), libfsntfs-python3 (>= 20211229), libfsxfs-python3 (>= 20260702), libfvde-python3 (>= 20220121), libfwnt-python3 (>= 20210717), libluksde-python3 (>= 20220121), libmodi-python3 (>= 20210405), libphdi-python3 (>= 20220228), libqcow-python3 (>= 20201213), libsigscan-python3 (>= 20230109), libsmdev-python3 (>= 20140529), libsmraw-python3 (>= 20140612), libvhdi-python3 (>= 20201014), libvmdk-python3 (>= 20140421), libvsapm-python3 (>= 20260713), libvsgpt-python3 (>= 20211115), libvshadow-python3 (>= 20160109), libvslvm-python3 (>= 20160109), python3-dfdatetime (>= 20221112), python3-dtfabric (>= 20230518), python3-pytsk3 (>= 20260715), python3-xattr (>= 0.7.2), python3-yaml (>= 3.10), ${misc:Depends} Description: Python 3 module of dfVFS dfVFS, or Digital Forensics Virtual File System, provides read-only access to file-system objects from various storage media types and file formats. The goal diff --git a/dependencies.ini b/dependencies.ini index 61c7c5fc..2a2490df 100644 --- a/dependencies.ini +++ b/dependencies.ini @@ -164,7 +164,7 @@ version_property: get_version() [pytsk3] dpkg_name: python3-pytsk3 -minimum_version: 20210419 +minimum_version: 20260715 rpm_name: python3-pytsk3 version_property: get_version() diff --git a/dfvfs/analyzer/__init__.py b/dfvfs/analyzer/__init__.py index b518517f..ccc2459d 100644 --- a/dfvfs/analyzer/__init__.py +++ b/dfvfs/analyzer/__init__.py @@ -26,11 +26,6 @@ except ImportError: pass -try: - from dfvfs.analyzer import tsk_partition_analyzer_helper -except ImportError: - pass - from dfvfs.analyzer import vhdi_analyzer_helper from dfvfs.analyzer import vmdk_analyzer_helper from dfvfs.analyzer import vshadow_analyzer_helper diff --git a/dfvfs/analyzer/analyzer.py b/dfvfs/analyzer/analyzer.py index e2d60ac6..91ea2735 100644 --- a/dfvfs/analyzer/analyzer.py +++ b/dfvfs/analyzer/analyzer.py @@ -138,9 +138,9 @@ def _GetSpecificationStore(cls, format_category): format_category (str): format category. Returns: - tuple[FormatSpecificationStore, list[AnalyzerHelper]]: a format - specification store and remaining analyzer helpers that do not have - a format specification. + tuple[FormatSpecificationStore, list[AnalyzerHelper]]: a format specification + store and remaining analyzer helpers that do not have a format + specification. """ specification_store = specification.FormatSpecificationStore() remainder_list = [] @@ -173,11 +173,11 @@ def _GetTypeIndicators( Args: signature_scanner (pysigscan.scanner): signature scanner. specification_store (FormatSpecificationStore): specification store. - remainder_list (list[AnalyzerHelper]): remaining analyzer helpers that - do not have a format specification. + remainder_list (list[AnalyzerHelper]): remaining analyzer helpers that do not + have a format specification. path_spec (PathSpec): path specification. - resolver_context (Optional[Context]): resolver context, where None - represents the built-in context which is not multi process safe. + resolver_context (Optional[Context]): resolver context, where None represents + the built-in context which is not multi process safe. Returns: list[str]: supported format type indicators. @@ -195,7 +195,6 @@ def _GetTypeIndicators( format_specification = specification_store.GetSpecificationBySignature( scan_result.identifier ) - if format_specification.identifier not in type_indicator_list: type_indicator_list.append(format_specification.identifier) @@ -220,10 +219,8 @@ def DeregisterHelper(cls, analyzer_helper): """ if analyzer_helper.type_indicator not in cls._analyzer_helpers: raise KeyError( - ( - f"Analyzer helper object not set for type indicator: " - f"{analyzer_helper.type_indicator:s}." - ) + f"Analyzer helper object not set for type indicator: " + f"{analyzer_helper.type_indicator:s}." ) analyzer_helper = cls._analyzer_helpers[analyzer_helper.type_indicator] @@ -238,8 +235,8 @@ def GetArchiveTypeIndicators(cls, path_spec, resolver_context=None): Args: path_spec (PathSpec): path specification. - resolver_context (Optional[Context]): resolver context, where None - represents the built-in context which is not multi process safe. + resolver_context (Optional[Context]): resolver context, where None represents + the built-in context which is not multi process safe. Returns: list[str]: supported format type indicators. @@ -396,7 +393,6 @@ def GetVolumeSystemTypeIndicators(cls, path_spec, resolver_context=None): path_spec, resolver_context=resolver_context, ) - if ( len(type_indicators) > 1 and definitions.TYPE_INDICATOR_TSK_PARTITION in type_indicators @@ -420,10 +416,8 @@ def RegisterHelper(cls, analyzer_helper): """ if analyzer_helper.type_indicator in cls._analyzer_helpers: raise KeyError( - ( - f"Analyzer helper object already set for type indicator: " - f"{analyzer_helper.type_indicator:s}." - ) + f"Analyzer helper object already set for type indicator: " + f"{analyzer_helper.type_indicator:s}." ) cls._FlushCache(analyzer_helper.format_categories) diff --git a/dfvfs/analyzer/tsk_partition_analyzer_helper.py b/dfvfs/analyzer/tsk_partition_analyzer_helper.py deleted file mode 100644 index 4054c5b8..00000000 --- a/dfvfs/analyzer/tsk_partition_analyzer_helper.py +++ /dev/null @@ -1,38 +0,0 @@ -"""The SleuthKit (TSK) partition format analyzer helper implementation.""" - -import pytsk3 - -from dfvfs.analyzer import analyzer -from dfvfs.analyzer import analyzer_helper -from dfvfs.lib import definitions -from dfvfs.lib import tsk_image - - -class TSKPartitionAnalyzerHelper(analyzer_helper.AnalyzerHelper): - """TSK partition analyzer helper.""" - - FORMAT_CATEGORIES = frozenset([definitions.FORMAT_CATEGORY_VOLUME_SYSTEM]) - - TYPE_INDICATOR = definitions.TYPE_INDICATOR_TSK_PARTITION - - def AnalyzeFileObject(self, file_object): - """Retrieves the format specification. - - Args: - file_object (FileIO): file-like object. - - Returns: - str: type indicator if the file-like object contains a supported format - or None otherwise. - """ - tsk_image_object = tsk_image.TSKFileSystemImage(file_object) - - try: - pytsk3.Volume_Info(tsk_image_object) - except OSError: - return None - - return self.type_indicator - - -analyzer.Analyzer.RegisterHelper(TSKPartitionAnalyzerHelper()) diff --git a/dfvfs/helpers/source_scanner.py b/dfvfs/helpers/source_scanner.py index 02446acf..c0d16f1f 100644 --- a/dfvfs/helpers/source_scanner.py +++ b/dfvfs/helpers/source_scanner.py @@ -18,11 +18,14 @@ * which VSS stores to default to. """ +import pytsk3 + from dfvfs.analyzer import analyzer from dfvfs.lib import apfs_helper from dfvfs.lib import definitions from dfvfs.lib import errors from dfvfs.lib import raw_helper +from dfvfs.lib import tsk_image from dfvfs.path import factory as path_spec_factory from dfvfs.resolver import resolver @@ -31,8 +34,7 @@ class SourceScanNode: """Source scan node. Attributes: - credential (tuple[str, str]): credential used to unlock the source scan - node. + credential (tuple[str, str]): credential used to unlock the source scan node. path_spec (PathSpec): path specification. parent_node (SourceScanNode): source scan parent node. scanned (bool): True if the source scan node has been fully scanned. @@ -105,8 +107,8 @@ def IsFileSystem(self): def IsSystemLevel(self): """Determines if the scan node has a path specification at system-level. - System-level is an indication used if the path specification is - handled by the operating system and should not have a parent. + System-level is an indication used if the path specification is handled by the + operating system and should not have a parent. Returns: bool: True if the scan node has a path specification at system-level. @@ -252,8 +254,8 @@ def HasFileSystemScanNodes(self): def HasLockedScanNodes(self): """Determines if a locked scan node was detected during the scan. - A locked scan node is e.g. an encrypted volume for which a credential, - e.g. password, to unlock the volume is not available. + A locked scan node is e.g. an encrypted volume for which a credential, e.g. + password, to unlock the volume is not available. Returns: bool: True if a locked scan node was detected during the scan. @@ -274,8 +276,8 @@ def HasScanNode(self, path_spec): def IsLockedScanNode(self, path_spec): """Determines if a scan node is locked. - A locked scan node is e.g. an encrypted volume for which a credential, - e.g. password, to unlock the volume is not available. + A locked scan node is e.g. an encrypted volume for which a credential, e.g. + password, to unlock the volume is not available. Args: path_spec (PathSpec): path specification. @@ -289,8 +291,7 @@ def IsSourceTypeDirectory(self): """Determines if the source type is a directory. Returns: - bool: True if the source type is a directory, False if not or - None if not set. + bool: True if the source type is a directory, False if not or None if not set. """ if not self.source_type: return None @@ -332,7 +333,6 @@ def OpenSourcePath(self, source_path): source_path_spec = path_spec_factory.Factory.NewPathSpec( definitions.TYPE_INDICATOR_OS, location=source_path ) - self.AddScanNode(source_path_spec, None) def RemoveScanNode(self, path_spec): @@ -381,8 +381,8 @@ def UnlockScanNode(self, path_spec, credential_identifier, credential_data): Args: path_spec (PathSpec): path specification. - credential_identifier (str): credential identifier used to unlock - the scan node. + credential_identifier (str): credential identifier used to unlock the scan + node. credential_data (bytes): credential data used to unlock the scan node. Raises: @@ -405,19 +405,20 @@ def UnlockScanNode(self, path_spec, credential_identifier, credential_data): class SourceScanner: """Searcher to find volumes within a volume system.""" - def __init__(self, resolver_context=None): + def __init__(self, resolver_context=None, sector_size=None): """Initializes a source scanner. Args: - resolver_context (Optional[Context]): resolver context, where None - indicates to use the built-in context which is not multi process - safe. + resolver_context (Optional[Context]): resolver context, where None indicates + to use the built-in context which is not multi process safe. + sector_size (Optional[int]): number of bytes per sector. """ super().__init__() self._resolver_context = resolver_context + self._sector_size = sector_size or 512 - # TODO: add functions to check if path spec type is a storage media image - # type, file system type, etc. + # TODO: add functions to check if path spec type is a storage media image type, file + # system type, etc. def _ScanNode(self, scan_context, scan_node, auto_recurse=True): """Scans a node for supported formats. @@ -425,8 +426,8 @@ def _ScanNode(self, scan_context, scan_node, auto_recurse=True): Args: scan_context (SourceScannerContext): source scanner context. scan_node (SourceScanNode): source scan node. - auto_recurse (Optional[bool]): True if the scan should automatically - recurse as far as possible. + auto_recurse (Optional[bool]): True if the scan should automatically recurse + as far as possible. Raises: BackEndError: if the source cannot be scanned. @@ -445,7 +446,6 @@ def _ScanNode(self, scan_context, scan_node, auto_recurse=True): system_level_file_entry = resolver.Resolver.OpenFileEntry( scan_node.path_spec, resolver_context=self._resolver_context ) - if system_level_file_entry is None: raise errors.BackEndError("Unable to open file entry.") @@ -468,9 +468,9 @@ def _ScanNode(self, scan_context, scan_node, auto_recurse=True): if not auto_recurse: return - # In case we did not find a storage media image type we keep looking - # since not all RAW storage media image naming schemas are known and - # its type can only detected by its content. + # In case we did not find a storage media image type we keep looking since + # not all RAW storage media image naming schemas are known and its type can + # only detected by its content. source_path_spec = None while True: @@ -482,8 +482,8 @@ def _ScanNode(self, scan_context, scan_node, auto_recurse=True): self._ScanEncryptedVolumeNode(scan_context, scan_node) if scan_context.IsLockedScanNode(scan_node.path_spec): - # Scan node is locked, such as an encrypted volume, and we cannot - # scan it for a volume system. + # Scan node is locked, such as an encrypted volume, and we cannot scan + # it for a volume system. break source_path_spec = self.ScanForVolumeSystem(scan_node.path_spec) @@ -506,7 +506,6 @@ def _ScanNode(self, scan_context, scan_node, auto_recurse=True): self._ScanVolumeSystemRootNode( scan_context, scan_node, auto_recurse=auto_recurse ) - # Check for an empty GPT with MBR partitions. if ( scan_node.type_indicator == definitions.TYPE_INDICATOR_GPT @@ -535,17 +534,16 @@ def _ScanNode(self, scan_context, scan_node, auto_recurse=True): if not scan_context.updated: break - # In case we did not find a volume system type we keep looking - # since we could be dealing with a storage media image that contains - # a single volume. + # In case we did not find a volume system type we keep looking since we could be + # dealing with a storage media image that contains a single volume. # No need to scan the root of a volume system for a file system. if scan_node.IsVolumeSystemRoot(): pass elif scan_context.IsLockedScanNode(scan_node.path_spec): - # Scan node is locked, such as an encrypted volume, and we cannot - # scan it for a file system. + # Scan node is locked, such as an encrypted volume, and we cannot scan it + # for a file system. pass elif ( @@ -553,8 +551,8 @@ def _ScanNode(self, scan_context, scan_node, auto_recurse=True): and auto_recurse and scan_node.path_spec != scan_path_spec ): - # Since scanning for file systems in VSS snapshot volumes can - # be expensive we only do this when explicitly asked for. + # Since scanning for file systems in VSS snapshot volumes can be expensive + # we only do this when explicitly asked for. pass elif not scan_node.IsFileSystem(): @@ -582,8 +580,7 @@ def _ScanNode(self, scan_context, scan_node, auto_recurse=True): scan_context.SetSourceType(source_type) - # If all scans failed mark the scan node as scanned so we do not scan it - # again. + # If all scans failed mark the scan node as scanned so we do not scan it again. if not scan_node.scanned: scan_node.scanned = True @@ -607,8 +604,8 @@ def _ScanEncryptedVolumeNode(self, scan_context, scan_node): if not file_entry.Unlock(): scan_context.LockScanNode(scan_node.path_spec) - # For BitLocker To Go add a scan node for the unencrypted part of - # the volume. + # For BitLocker To Go add a scan node for the unencrypted part of the + # volume. if scan_node.type_indicator == definitions.TYPE_INDICATOR_BDE: path_spec = self.ScanForFileSystem(scan_node.path_spec.parent) if path_spec: @@ -620,8 +617,8 @@ def _ScanVolumeSystemRootNode(self, scan_context, scan_node, auto_recurse=True): Args: scan_context (SourceScannerContext): source scanner context. scan_node (SourceScanNode): source scan node. - auto_recurse (Optional[bool]): True if the scan should automatically - recurse as far as possible. + auto_recurse (Optional[bool]): True if the scan should automatically recurse + as far as possible. Raises: ValueError: if the scan context or scan node is invalid. @@ -647,7 +644,6 @@ def _ScanVolumeSystemRootNode(self, scan_context, scan_node, auto_recurse=True): sub_scan_node = scan_context.AddScanNode( sub_file_entry.path_spec, scan_node ) - if scan_node.type_indicator == definitions.TYPE_INDICATOR_VSHADOW: # Since scanning for file systems in VSS snapshot volumes can # be expensive we only do this when explicitly asked for. @@ -678,11 +674,11 @@ def Scan(self, scan_context, auto_recurse=True, scan_path_spec=None): Args: scan_context (SourceScannerContext): source scanner context. - auto_recurse (Optional[bool]): True if the scan should automatically - recurse as far as possible. - scan_path_spec (Optional[PathSpec]): path specification to indicate - where the source scanner should continue scanning, where None - indicates the scanner will start with the sources. + auto_recurse (Optional[bool]): True if the scan should automatically recurse + as far as possible. + scan_path_spec (Optional[PathSpec]): path specification to indicate where the + source scanner should continue scanning, where None indicates the scanner + will start with the sources. Raises: ValueError: if the scan context is invalid. @@ -708,12 +704,12 @@ def ScanForFileSystem(self, source_path_spec): source_path_spec (PathSpec): source path specification. Returns: - PathSpec: file system path specification or None if no supported file - system type was found. + PathSpec: file system path specification or None if no supported file system + type was found. Raises: - BackEndError: if the source cannot be scanned or more than one file - system type is found. + BackEndError: if the source cannot be scanned or more than one file system + type is found. """ if source_path_spec.type_indicator == ( definitions.TYPE_INDICATOR_APFS_CONTAINER @@ -730,10 +726,7 @@ def ScanForFileSystem(self, source_path_spec): ) except RuntimeError as exception: raise errors.BackEndError( - ( - f"Unable to process source path specification with error: " - f"{exception!s}" - ) + f"Unable to process source path specification with error: {exception!s}" ) if not type_indicators: @@ -767,10 +760,9 @@ def ScanForFileSystem(self, source_path_spec): file_system_path_spec = path_spec_factory.Factory.NewPathSpec( type_indicator, location=root_location, parent=source_path_spec ) - if type_indicator == definitions.TYPE_INDICATOR_TSK: - # Check if the file system can be opened since the file system by - # signature detection results in false positives. + # Check if the file system can be opened since the file system by signature + # detection results in false positives. try: resolver.Resolver.OpenFileSystem( file_system_path_spec, resolver_context=self._resolver_context @@ -800,22 +792,18 @@ def ScanForStorageMediaImage(self, source_path_spec): ) except RuntimeError as exception: raise errors.BackEndError( - ( - f"Unable to process source path specification with error: " - f"{exception!s}" - ) + f"Unable to process source path specification with error: {exception!s}" ) if not type_indicators: - # The RAW storage media image type cannot be detected based on - # a signature so we try to detect it based on common file naming schemas. + # The RAW storage media image type cannot be detected based on a signature + # so we try to detect it based on common file naming schemas. file_system = resolver.Resolver.OpenFileSystem( source_path_spec, resolver_context=self._resolver_context ) raw_path_spec = path_spec_factory.Factory.NewPathSpec( definitions.TYPE_INDICATOR_RAW, parent=source_path_spec ) - try: # The RAW glob function will raise a PathSpecError if the path # specification is unsuitable for globbing. @@ -851,17 +839,17 @@ def ScanForVolumeSystem(self, source_path_spec): BackEndError: if the source cannot be scanned or more than one volume system type is found. """ - if source_path_spec.type_indicator == definitions.TYPE_INDICATOR_VSHADOW: - # It is technically possible to scan for VSS-in-VSS but makes no sense - # to do so. + source_type_indicator = source_path_spec.type_indicator + + if source_type_indicator == definitions.TYPE_INDICATOR_VSHADOW: + # It is technically possible to scan for VSS-in-VSS but makes no sense to + # do so. return None if source_path_spec.IsVolumeSystemRoot(): return source_path_spec - if source_path_spec.type_indicator == ( - definitions.TYPE_INDICATOR_APFS_CONTAINER - ): + if source_type_indicator == definitions.TYPE_INDICATOR_APFS_CONTAINER: # Currently pyfsapfs does not support reading from a volume as a device. # Also see: https://github.com/log2timeline/dfvfs/issues/332 return None @@ -872,30 +860,54 @@ def ScanForVolumeSystem(self, source_path_spec): ) except (OSError, RuntimeError) as exception: raise errors.BackEndError( - ( - f"Unable to process source path specification with error: " - f"{exception!s}" - ) + f"Unable to process source path specification with error: {exception!s}" ) + scan_for_sector_size = False if not type_indicators: - return None + scan_for_sector_size = True + else: + if len(type_indicators) > 1: + raise errors.BackEndError( + "Unsupported source found more than one volume system types." + ) - if len(type_indicators) > 1: - raise errors.BackEndError( - "Unsupported source found more than one volume system types." + type_indicator = type_indicators[0] + + # pysigscan found a volume system sigature however for the TSK partition + # table type we also need the sector size. + scan_for_sector_size = ( + type_indicator == definitions.TYPE_INDICATOR_TSK_PARTITION ) - # Ignore the TSK partition table type when detected within other partition - # table types. - type_indicator = type_indicators[0] + if scan_for_sector_size: + # Ignore the TSK partition table type when detected within other partition + # table types. + if source_type_indicator in definitions.PARTITION_TABLE_TYPE_INDICATORS: + return None - if type_indicator == definitions.TYPE_INDICATOR_TSK_PARTITION: - if source_path_spec.type_indicator in ( - definitions.PARTITION_TABLE_TYPE_INDICATORS - ): + file_object = resolver.Resolver.OpenFileObject( + source_path_spec, resolver_context=self._resolver_context + ) + try: + tsk_image_object = tsk_image.TSKFileSystemImage( + file_object, + sector_size=self._sector_size, + ) + volume_info = pytsk3.Volume_Info(tsk_image_object) + except OSError: + volume_info = None + + if not volume_info: return None + return path_spec_factory.Factory.NewPathSpec( + definitions.TYPE_INDICATOR_TSK_PARTITION, + location="/", + parent=source_path_spec, + sector_size=self._sector_size, + ) + if type_indicator in definitions.VOLUME_SYSTEM_TYPE_INDICATORS: return path_spec_factory.Factory.NewPathSpec( type_indicator, location="/", parent=source_path_spec @@ -911,8 +923,8 @@ def Unlock(self, scan_context, path_spec, credential_identifier, credential_data Args: scan_context (SourceScannerContext): source scanner context. path_spec (PathSpec): path specification of the locked scan node. - credential_identifier (str): credential identifier used to unlock - the scan node. + credential_identifier (str): credential identifier used to unlock the scan + node. credential_data (bytes): credential data used to unlock the scan node. Returns: @@ -931,7 +943,6 @@ def Unlock(self, scan_context, path_spec, credential_identifier, credential_data resolver.Resolver.key_chain.SetCredential( path_spec, credential_identifier, credential_data ) - if path_spec.type_indicator == definitions.TYPE_INDICATOR_APFS_CONTAINER: # Currently pyfsapfs does not support reading from a volume as a device. # Also see: https://github.com/log2timeline/dfvfs/issues/332 diff --git a/dfvfs/helpers/volume_scanner.py b/dfvfs/helpers/volume_scanner.py index 684e1909..44deda3a 100644 --- a/dfvfs/helpers/volume_scanner.py +++ b/dfvfs/helpers/volume_scanner.py @@ -17,14 +17,13 @@ class VolumeScannerOptions: """Volume scanner options. Attributes: - credentials (list[tuple[str, str]]): credentials, per type, to unlock - volumes. + credentials (list[tuple[str, str]]): credentials, per type, to unlock volumes. partitions (list[str]): partition identifiers. - scan_mode (str): mode that defines how the VolumeScanner should scan - for volumes and snapshots. + scan_mode (str): mode that defines how the VolumeScanner should scan for volumes + and snapshots. snapshots (list[str]): snapshot identifiers. - volumes (list[str]): volume identifiers, e.g. those of an APFS or LVM - volume system. + volumes (list[str]): volume identifiers, e.g. those of an APFS or LVM volume + system. """ # Scan all volumes and snapshots for available file systems. @@ -171,16 +170,17 @@ def UnlockEncryptedVolume( class VolumeScanner: """Volume scanner.""" - def __init__(self, mediator=None): + def __init__(self, mediator=None, sector_size=None): """Initializes a volume scanner. Args: mediator (Optional[VolumeScannerMediator]): a volume scanner mediator. + sector_size (Optional[int]): number of bytes per sector. """ super().__init__() self._mediator = mediator self._source_path = None - self._source_scanner = source_scanner.SourceScanner() + self._source_scanner = source_scanner.SourceScanner(sector_size=sector_size) self._source_type = None def _GetBasePathSpecs(self, scan_context, options): @@ -229,9 +229,9 @@ def _GetBasePathSpecs(self, scan_context, options): def _GetPartitionIdentifiers(self, scan_node, options): """Determines the partition identifiers. - This function determines which partition identifiers need to be scanned - based on the volume scanner options. If no options are provided and there - is more than a single partition the mediator is used to ask the user. + This function determines which partition identifiers need to be scanned based + on the volume scanner options. If no options are provided and there is more + than a single partition the mediator is used to ask the user. Args: scan_node (SourceScanNode): scan node. @@ -420,16 +420,16 @@ def _NormalizedVolumeIdentifiers( Args: volume_system (VolumeSystem): volume system. - volume_identifiers (list[int|str]): allowed volume identifiers, formatted - as an integer or string with prefix. + volume_identifiers (list[int|str]): allowed volume identifiers, formatted as + an integer or string with prefix. prefix (Optional[str]): volume identifier prefix. Returns: list[str]: volume identifiers with prefix. Raises: - ScannerError: if the volume identifier is not supported or no volume - could be found that corresponds with the identifier. + ScannerError: if the volume identifier is not supported or no volume could be + found that corresponds with the identifier. """ normalized_volume_identifiers = [] for volume_identifier in volume_identifiers: @@ -466,10 +466,10 @@ def _ScanEncryptedVolume(self, scan_context, scan_node, options): options (VolumeScannerOptions): volume scanner options. Raises: - ScannerError: if the format of or within the source is not supported, - the scan node is invalid, there are no credentials defined for - the format or no mediator is provided and a locked scan node was - found, e.g. an encrypted volume, + ScannerError: if the format of or within the source is not supported, the scan + node is invalid, there are no credentials defined for the format or no + mediator is provided and a locked scan node was found, e.g. an encrypted + volume. """ if not scan_node or not scan_node.path_spec: raise errors.ScannerError("Invalid or missing scan node.") @@ -550,9 +550,9 @@ def _ScanSource(self, source_path): SourceScannerContext: source scanner context. Raises: - ScannerError: if the source path does not exists, or if the source path - is not a file or directory, or if the format of or within the source - file is not supported. + ScannerError: if the source path does not exists, or if the source path is not + a file or directory, or if the format of or within the source file is not + supported. """ if not source_path: raise errors.ScannerError("Invalid source path.") @@ -578,9 +578,9 @@ def _ScanSourcePathSpec(self, source_path_spec): SourceScannerContext: source scanner context. Raises: - ScannerError: if the source path does not exists, or if the source path - is not a file or directory, or if the format of or within the source - file is not supported. + ScannerError: if the source path does not exists, or if the source path is not + a file or directory, or if the format of or within the source file is not + supported. """ scan_context = source_scanner.SourceScannerContext() scan_context.AddScanNode(source_path_spec, None) @@ -604,8 +604,8 @@ def _ScanVolume(self, scan_context, scan_node, options, base_path_specs): base_path_specs (list[PathSpec]): file system base path specifications. Raises: - ScannerError: if the format of or within the source - is not supported or the scan node is invalid. + ScannerError: if the format of or within the source is not supported or the + scan node is invalid. """ if not scan_node or not scan_node.path_spec: raise errors.ScannerError("Invalid or missing scan node.") @@ -662,8 +662,8 @@ def _ScanVolumeSystemRoot(self, scan_context, scan_node, options, base_path_spec base_path_specs (list[PathSpec]): file system base path specifications. Raises: - ScannerError: if the scan node is invalid, the scan node type is not - supported or if a sub scan node cannot be retrieved. + ScannerError: if the scan node is invalid, the scan node type is not supported + or if a sub scan node cannot be retrieved. """ if not scan_node or not scan_node.path_spec: raise errors.ScannerError("Invalid scan node.") @@ -721,17 +721,17 @@ def GetBasePathSpecs(self, source_path, options=None): Args: source_path (str): source path. - options (Optional[VolumeScannerOptions]): volume scanner options. If None - the default volume scanner options are used, which are defined in the + options (Optional[VolumeScannerOptions]): volume scanner options. If None the + default volume scanner options are used, which are defined in the VolumeScannerOptions class. Returns: list[PathSpec]: path specifications. Raises: - ScannerError: if the source path does not exists, or if the source path - is not a file or directory, or if the format of or within the source - file is not supported. + ScannerError: if the source path does not exists, or if the source path is not + a file or directory, or if the format of or within the source file is not + supported. """ if not options: options = VolumeScannerOptions() @@ -830,17 +830,17 @@ def ScanForWindowsVolume(self, source_path, options=None): Args: source_path (str): source path. - options (Optional[VolumeScannerOptions]): volume scanner options. If None - the default volume scanner options are used, which are defined in the + options (Optional[VolumeScannerOptions]): volume scanner options. If None the + default volume scanner options are used, which are defined in the VolumeScannerOptions class. Returns: bool: True if a Windows volume was found. Raises: - ScannerError: if the source path does not exists, or if the source path - is not a file or directory, or if the format of or within the source - file is not supported. + ScannerError: if the source path does not exists, or if the source path is not + a file or directory, or if the format of or within the source file is not + supported. """ if not options: options = VolumeScannerOptions() diff --git a/dfvfs/lib/tsk_image.py b/dfvfs/lib/tsk_image.py index d0e5547e..f08190cf 100644 --- a/dfvfs/lib/tsk_image.py +++ b/dfvfs/lib/tsk_image.py @@ -7,11 +7,12 @@ class TSKFileSystemImage(pytsk3.Img_Info): """Pytsk3 image object using a file-like object.""" - def __init__(self, file_object): + def __init__(self, file_object, sector_size=None): """Initializes an image object. Args: file_object (FileIO): file-like object. + sector_size (Optional[int]): number of bytes per sector. Raises: ValueError: if the file-like object is invalid. @@ -21,16 +22,20 @@ def __init__(self, file_object): # pytsk3.Img_Info does not let you set attributes after initialization. self._file_object = file_object - # Using the old parent class invocation style otherwise some versions - # of pylint complain also setting type to RAW or EXTERNAL to make sure - # Img_Info does not do detection. + + # Using the old parent class invocation style otherwise some versions of pylint + # complain also setting type to RAW or EXTERNAL to make sure Img_Info does not + # do detection. tsk_img_type = getattr(pytsk3, "TSK_IMG_TYPE_EXTERNAL", pytsk3.TSK_IMG_TYPE_RAW) - # Note that we want url to be a binary string in Python 2 and a Unicode - # string in Python 3. Hence the string is not prefixed. - pytsk3.Img_Info.__init__(self, url="", type=tsk_img_type) - # Note: that the following functions do not follow the style guide - # because they are part of the pytsk3.Img_Info object interface. + # Note that we want url to be a binary string in Python 2 and a Unicode string + # in Python 3. Hence the string is not prefixed. + pytsk3.Img_Info.__init__( + self, url="", type=tsk_img_type, sector_size=sector_size or 512 + ) + + # Note: that the following functions do not follow the style guide because they are + # part of the pytsk3.Img_Info object interface. # pylint: disable=invalid-name def close(self): diff --git a/dfvfs/lib/tsk_partition.py b/dfvfs/lib/tsk_partition.py index ae33d20e..dd562659 100644 --- a/dfvfs/lib/tsk_partition.py +++ b/dfvfs/lib/tsk_partition.py @@ -19,6 +19,7 @@ def GetTSKVsPartByPathSpec(tsk_volume, path_spec): """ location = getattr(path_spec, "location", None) part_index = getattr(path_spec, "part_index", None) + sector_size = getattr(path_spec, "sector_size", None) start_offset = getattr(path_spec, "start_offset", None) partition_index = None @@ -36,7 +37,7 @@ def GetTSKVsPartByPathSpec(tsk_volume, path_spec): if location is None and start_offset is None: return None, None - bytes_per_sector = TSKVolumeGetBytesPerSector(tsk_volume) + bytes_per_sector = sector_size or TSKVolumeGetBytesPerSector(tsk_volume) current_part_index = 0 current_partition_index = 0 tsk_vs_part = None @@ -73,8 +74,8 @@ def GetTSKVsPartByPathSpec(tsk_volume, path_spec): current_part_index += 1 - # Note that here we cannot solely rely on testing if tsk_vs_part is set - # since the for loop will exit with tsk_vs_part set. + # Note that here we cannot solely rely on testing if tsk_vs_part is set since the + # for loop will exit with tsk_vs_part set. if tsk_vs_part is None or current_part_index >= number_of_tsk_vs_parts: return None, None @@ -92,13 +93,14 @@ def TSKVolumeGetBytesPerSector(tsk_volume): Returns: int: number of bytes per sector or 512 by default. """ - # Note that because pytsk3.Volume_Info does not explicitly defines info - # we need to check if the attribute exists and has a value other - # than None. Default to 512 otherwise. - if hasattr(tsk_volume, "info") and tsk_volume.info is not None: - block_size = getattr(tsk_volume.info, "block_size", 512) - else: + # Note that because pytsk3.Volume_Info does not explicitly defines info we need to + # check if the attribute exists and has a value other than None. Default to 512 + # otherwise. + tsk_volume_info = getattr(tsk_volume, "info", None) + if tsk_volume_info is None: block_size = 512 + else: + block_size = getattr(tsk_volume_info, "block_size", 512) return block_size @@ -112,8 +114,8 @@ def TSKVsPartGetNumberOfSectors(tsk_vs_part): Returns: int: number of sectors or None. """ - # Note that because pytsk3.TSK_VS_PART_INFO does not explicitly defines - # len we need to check if the attribute exists. + # Note that because pytsk3.TSK_VS_PART_INFO does not explicitly defines len we need + # to check if the attribute exists. return getattr(tsk_vs_part, "len", None) @@ -126,8 +128,8 @@ def TSKVsPartGetStartSector(tsk_vs_part): Returns: int: start sector or None. """ - # Note that because pytsk3.TSK_VS_PART_INFO does not explicitly defines - # start we need to check if the attribute exists. + # Note that because pytsk3.TSK_VS_PART_INFO does not explicitly defines start we + # need to check if the attribute exists. return getattr(tsk_vs_part, "start", None) @@ -140,8 +142,8 @@ def TSKVsPartIsAllocated(tsk_vs_part): Returns: bool: True if the volume system part is allocated, False otherwise. """ - # Note that because pytsk3.TSK_VS_PART_INFO does not explicitly defines - # flags need to check if the attribute exists. + # Note that because pytsk3.TSK_VS_PART_INFO does not explicitly defines flags need + # to check if the attribute exists. # The flags are an instance of TSK_VS_PART_FLAG_ENUM. tsk_vs_part_flags = getattr(tsk_vs_part, "flags", None) @@ -149,18 +151,17 @@ def TSKVsPartIsAllocated(tsk_vs_part): tsk_vs_part_flags is not None and tsk_vs_part_flags == pytsk3.TSK_VS_PART_FLAG_ALLOC ) - tsk_vs_vstype = getattr(tsk_vs_part.vs, "vstype", pytsk3.TSK_VS_TYPE_UNSUPP) - # For BSD disklabel consider c and d partitions (slot 2 and 3) as metadata - # part of the disklabel volume itself and not a partition of the volume. + # For BSD disklabel consider c and d partitions (slot 2 and 3) as metadata part of + # the disklabel volume itself and not a partition of the volume. if tsk_vs_vstype == pytsk3.TSK_VS_TYPE_BSD: tsk_vs_part_slot_num = getattr(tsk_vs_part, "slot_num", None) if tsk_vs_part_slot_num in (None, 2, 3): is_allocated = False - # For APM partition tables the description needs to be checked to determine - # the usage of the part. + # For APM partition tables the description needs to be checked to determine the + # usage of the part. elif tsk_vs_vstype == pytsk3.TSK_VS_TYPE_MAC: tsk_vs_part_desc = getattr(tsk_vs_part, "desc", None) diff --git a/dfvfs/path/tsk_partition_path_spec.py b/dfvfs/path/tsk_partition_path_spec.py index 65735b68..013a1317 100644 --- a/dfvfs/path/tsk_partition_path_spec.py +++ b/dfvfs/path/tsk_partition_path_spec.py @@ -17,7 +17,13 @@ class TSKPartitionPathSpec(path_spec.PathSpec): TYPE_INDICATOR = definitions.TYPE_INDICATOR_TSK_PARTITION def __init__( - self, location=None, parent=None, part_index=None, start_offset=None, **kwargs + self, + location=None, + parent=None, + part_index=None, + sector_size=None, + start_offset=None, + **kwargs, ): """Initializes a path specification. @@ -27,6 +33,7 @@ def __init__( location (Optional[str]): location. parent (Optional[PathSpec]): parent path specification. part_index (Optional[int]): part index. + sector_size (Optional[int]): number of bytes per sector. start_offset (Optional[int]): start offset. Raises: @@ -38,6 +45,7 @@ def __init__( super().__init__(parent=parent, **kwargs) self.location = location self.part_index = part_index + self.sector_size = sector_size self.start_offset = start_offset @property @@ -49,6 +57,8 @@ def comparable(self): string_parts.append(f"location: {self.location:s}") if self.part_index is not None: string_parts.append(f"part index: {self.part_index:d}") + if self.sector_size is not None: + string_parts.append(f"sector size: {self.sector_size:d}") if self.start_offset is not None: string_parts.append(f"start offset: 0x{self.start_offset:08x}") diff --git a/dfvfs/vfs/tsk_partition_directory.py b/dfvfs/vfs/tsk_partition_directory.py index fd43f4fa..96d51412 100644 --- a/dfvfs/vfs/tsk_partition_directory.py +++ b/dfvfs/vfs/tsk_partition_directory.py @@ -13,14 +13,15 @@ class TSKPartitionDirectory(directory.Directory): def _EntriesGenerator(self): """Retrieves directory entries. - Since a directory can contain a vast number of entries using - a generator is more memory efficient. + Since a directory can contain a vast number of entries using a generator is more + memory efficient. Yields: TSKPartitionPathSpec: a path specification. """ location = getattr(self.path_spec, "location", None) part_index = getattr(self.path_spec, "part_index", None) + sector_size = getattr(self.path_spec, "sector_size", None) start_offset = getattr(self.path_spec, "start_offset", None) # Only the virtual root file has directory entries. @@ -31,7 +32,9 @@ def _EntriesGenerator(self): and location == self._file_system.LOCATION_ROOT ): tsk_volume = self._file_system.GetTSKVolume() - bytes_per_sector = tsk_partition.TSKVolumeGetBytesPerSector(tsk_volume) + bytes_per_sector = sector_size or tsk_partition.TSKVolumeGetBytesPerSector( + tsk_volume + ) part_index = 0 partition_index = 0 @@ -63,6 +66,8 @@ def _EntriesGenerator(self): start_sector = tsk_partition.TSKVsPartGetStartSector(tsk_vs_part) + if sector_size is not None: + kwargs["sector_size"] = sector_size if start_sector is not None: kwargs["start_offset"] = start_sector * bytes_per_sector diff --git a/dfvfs/vfs/tsk_partition_file_system.py b/dfvfs/vfs/tsk_partition_file_system.py index 2dd6f55a..4ca742eb 100644 --- a/dfvfs/vfs/tsk_partition_file_system.py +++ b/dfvfs/vfs/tsk_partition_file_system.py @@ -41,8 +41,8 @@ def _Open(self, mode="rb"): """Opens the file system object defined by path specification. Args: - mode (Optional[str]): file access mode. The default is 'rb' which - represents read-only binary. + mode (Optional[str]): file access mode. The default is 'rb' which represents + read-only binary. Raises: AccessError: if the access to open the file was denied. @@ -56,8 +56,9 @@ def _Open(self, mode="rb"): file_object = resolver.Resolver.OpenFileObject( self._path_spec.parent, resolver_context=self._resolver_context ) - - tsk_image_object = tsk_image.TSKFileSystemImage(file_object) + tsk_image_object = tsk_image.TSKFileSystemImage( + file_object, sector_size=self._path_spec.sector_size + ) tsk_volume = pytsk3.Volume_Info(tsk_image_object) self._file_object = file_object @@ -75,9 +76,8 @@ def FileEntryExistsByPathSpec(self, path_spec): tsk_vs_part, _ = tsk_partition.GetTSKVsPartByPathSpec( self._tsk_volume, path_spec ) - - # The virtual root file has no corresponding TSK volume system part object - # but should have a location. + # The virtual root file has no corresponding TSK volume system part object but + # should have a location. if tsk_vs_part is None: location = getattr(path_spec, "location", None) return location is not None and location == self.LOCATION_ROOT @@ -96,11 +96,10 @@ def GetFileEntryByPathSpec(self, path_spec): tsk_vs_part, partition_index = tsk_partition.GetTSKVsPartByPathSpec( self._tsk_volume, path_spec ) - location = getattr(path_spec, "location", None) - # The virtual root file has no corresponding TSK volume system part object - # but should have a location. + # The virtual root file has no corresponding TSK volume system part object but + # should have a location. if tsk_vs_part is None: if location is None or location != self.LOCATION_ROOT: return None diff --git a/docs/sources/Path-specifications.md b/docs/sources/Path-specifications.md index 9aa5a4ca..ea135ae7 100644 --- a/docs/sources/Path-specifications.md +++ b/docs/sources/Path-specifications.md @@ -340,6 +340,7 @@ systems. location | The location of the volume within the volume system parent | The parent path specification part_index | The SleuthKit part index that indicates the volume within the volume system +sector_size | The number of bytes per sector start_offset | The start offset, in bytes, of the volume within the volume system ### The VHD storage media image type diff --git a/pyproject.toml b/pyproject.toml index be19a98b..dea0573e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "dfvfs" -version = "20260714" +version = "20260715" description = "Digital Forensics Virtual File System (dfVFS)" maintainers = [ { name = "Log2Timeline maintainers", email = "log2timeline-maintainers@googlegroups.com" }, @@ -46,7 +46,7 @@ dependencies = [ "libvsgpt-python >= 20211115", "libvshadow-python >= 20160109", "libvslvm-python >= 20160109", - "pytsk3 >= 20210419", + "pytsk3 >= 20260715", "xattr >= 0.7.2 ; platform_system != \"Windows\"", ] diff --git a/tests/analyzer/analyzer.py b/tests/analyzer/analyzer.py index 3b3da6f6..ce1179d6 100644 --- a/tests/analyzer/analyzer.py +++ b/tests/analyzer/analyzer.py @@ -75,7 +75,6 @@ def testHelperRegistration(self): self.assertEqual( len(analyzer.Analyzer._analyzer_helpers), number_of_helpers + 1 ) - with self.assertRaises(KeyError): analyzer.Analyzer.RegisterHelper(test_helper) @@ -384,7 +383,6 @@ def testGetVolumeSystemTypeIndicatorsCS(self): path_spec = tsk_partition_path_spec.TSKPartitionPathSpec( location="/p1", parent=path_spec ) - expected_type_indicators = [definitions.TYPE_INDICATOR_CS] type_indicators = analyzer.Analyzer.GetVolumeSystemTypeIndicators(path_spec) self.assertEqual(type_indicators, expected_type_indicators) @@ -422,17 +420,6 @@ def testGetVolumeSystemTypeIndicatorsLVM(self): type_indicators = analyzer.Analyzer.GetVolumeSystemTypeIndicators(path_spec) self.assertEqual(type_indicators, expected_type_indicators) - def testGetVolumeSystemTypeIndicatorsMBR(self): - """Tests the GetVolumeSystemTypeIndicators function on MBR partitions.""" - test_file = self._GetTestFilePath(["mbr.raw"]) - self._SkipIfPathNotExists(test_file) - - path_spec = os_path_spec.OSPathSpec(location=test_file) - - expected_type_indicators = [definitions.TYPE_INDICATOR_TSK_PARTITION] - type_indicators = analyzer.Analyzer.GetVolumeSystemTypeIndicators(path_spec) - self.assertEqual(type_indicators, expected_type_indicators) - def testGetVolumeSystemTypeIndicatorsVSS(self): """Tests the GetVolumeSystemTypeIndicators function on a VSS volume.""" test_file = self._GetTestFilePath(["vss.raw"]) diff --git a/tests/helpers/source_scanner.py b/tests/helpers/source_scanner.py index 86377583..b5620ec0 100644 --- a/tests/helpers/source_scanner.py +++ b/tests/helpers/source_scanner.py @@ -311,7 +311,6 @@ def testGetVolumeIdentifiers(self): test_tsk_partition_path_spec = path_spec_factory.Factory.NewPathSpec( definitions.TYPE_INDICATOR_TSK_PARTITION, parent=test_raw_path_spec ) - volume_system = tsk_volume_system.TSKVolumeSystem() volume_system.Open(test_tsk_partition_path_spec) @@ -330,7 +329,6 @@ def testScanOnAPFS(self): self.assertEqual( scan_context.source_type, definitions.SOURCE_TYPE_STORAGE_MEDIA_IMAGE ) - scan_node = scan_context.GetRootScanNode() self.assertIsNotNone(scan_node) @@ -364,7 +362,6 @@ def testScanOnAPM(self): self.assertEqual( scan_context.source_type, definitions.SOURCE_TYPE_STORAGE_MEDIA_IMAGE ) - scan_node = scan_context.GetRootScanNode() self.assertIsNotNone(scan_node) self.assertEqual(scan_node.type_indicator, definitions.TYPE_INDICATOR_OS) @@ -404,7 +401,6 @@ def testScanOnEncryptedAPFS(self): self.assertEqual( scan_context.source_type, definitions.SOURCE_TYPE_STORAGE_MEDIA_IMAGE ) - scan_node = scan_context.GetRootScanNode() self.assertIsNotNone(scan_node) self.assertEqual(scan_node.type_indicator, definitions.TYPE_INDICATOR_OS) @@ -451,7 +447,6 @@ def testScanOnEncryptedAPFS(self): self._source_scanner.Unlock( scan_context, scan_node.path_spec, "password", self._APFS_PASSWORD ) - self.assertFalse(scan_context.IsLockedScanNode(scan_node.path_spec)) self._source_scanner.Scan(scan_context, scan_path_spec=scan_node.path_spec) @@ -473,7 +468,6 @@ def testScanOnGPT(self): self.assertEqual( scan_context.source_type, definitions.SOURCE_TYPE_STORAGE_MEDIA_IMAGE ) - scan_node = self._GetTestScanNode(scan_context) self.assertIsNotNone(scan_node) self.assertEqual(scan_node.type_indicator, definitions.PREFERRED_GPT_BACK_END) @@ -506,14 +500,12 @@ def testScanOnEmtpyGPTWithMBR(self): self.assertEqual( scan_node.sub_nodes[0].type_indicator, definitions.PREFERRED_GPT_BACK_END ) - scan_node = scan_node.sub_nodes[1] self.assertIsNotNone(scan_node) self.assertEqual( scan_node.type_indicator, definitions.TYPE_INDICATOR_TSK_PARTITION ) - scan_node = scan_node.sub_nodes[2].GetSubNodeByLocation("/") self.assertIsNotNone(scan_node) self.assertEqual(scan_node.type_indicator, definitions.PREFERRED_EXT_BACK_END) @@ -530,7 +522,6 @@ def testScanOnLVM(self): self.assertEqual( scan_context.source_type, definitions.SOURCE_TYPE_STORAGE_MEDIA_IMAGE ) - scan_node = self._GetTestScanNode(scan_context) self.assertIsNotNone(scan_node) self.assertEqual(scan_node.type_indicator, definitions.TYPE_INDICATOR_LVM) @@ -553,7 +544,6 @@ def testScanOnLUKSDE(self): self.assertEqual( scan_context.source_type, definitions.SOURCE_TYPE_STORAGE_MEDIA_IMAGE ) - scan_node = self._GetTestScanNode(scan_context) self.assertIsNotNone(scan_node) self.assertEqual(scan_node.type_indicator, definitions.TYPE_INDICATOR_LUKSDE) @@ -563,7 +553,6 @@ def testScanOnLUKSDE(self): self._source_scanner.Unlock( scan_context, scan_node.path_spec, "password", self._LUKSDE_PASSWORD ) - self.assertFalse(scan_context.IsLockedScanNode(scan_node.path_spec)) self._source_scanner.Scan(scan_context, scan_path_spec=scan_node.path_spec) @@ -586,13 +575,11 @@ def testScanOnMBR(self): self.assertEqual( scan_context.source_type, definitions.SOURCE_TYPE_STORAGE_MEDIA_IMAGE ) - scan_node = self._GetTestScanNode(scan_context) self.assertIsNotNone(scan_node) self.assertEqual( scan_node.type_indicator, definitions.TYPE_INDICATOR_TSK_PARTITION ) - self.assertEqual(len(scan_node.sub_nodes), 8) scan_node = scan_node.sub_nodes[6].GetSubNodeByLocation("/") @@ -611,7 +598,6 @@ def testScanOnVSS(self): self.assertEqual( scan_context.source_type, definitions.SOURCE_TYPE_STORAGE_MEDIA_IMAGE ) - scan_node = self._GetTestScanNode(scan_context) self.assertIsNotNone(scan_node) self.assertEqual(scan_node.type_indicator, definitions.TYPE_INDICATOR_RAW) @@ -655,7 +641,6 @@ def testScanOnBDE(self): self.assertEqual( scan_context.source_type, definitions.SOURCE_TYPE_STORAGE_MEDIA_IMAGE ) - scan_node = self._GetTestScanNode(scan_context) self.assertIsNotNone(scan_node) self.assertEqual(scan_node.type_indicator, definitions.TYPE_INDICATOR_RAW) @@ -670,7 +655,6 @@ def testScanOnBDE(self): self._source_scanner.Unlock( scan_context, scan_node.path_spec, "password", self._BDE_PASSWORD ) - self.assertFalse(scan_context.IsLockedScanNode(scan_node.path_spec)) self._source_scanner.Scan(scan_context, scan_path_spec=scan_node.path_spec) @@ -695,7 +679,6 @@ def testScanOnFVDE(self): self.assertEqual( scan_context.source_type, definitions.SOURCE_TYPE_STORAGE_MEDIA_IMAGE ) - scan_node = scan_context.GetRootScanNode() self.assertIsNotNone(scan_node) self.assertEqual(scan_node.type_indicator, definitions.TYPE_INDICATOR_OS) @@ -738,7 +721,6 @@ def testScanOnFVDE(self): self._source_scanner.Unlock( scan_context, scan_node.path_spec, "password", self._FVDE_PASSWORD ) - self.assertFalse(scan_context.IsLockedScanNode(scan_node.path_spec)) self._source_scanner.Scan(scan_context, scan_path_spec=scan_node.path_spec) @@ -793,7 +775,6 @@ def testScanOnRAW(self): self.assertEqual( scan_context.source_type, definitions.SOURCE_TYPE_STORAGE_MEDIA_IMAGE ) - scan_node = self._GetTestScanNode(scan_context) self.assertIsNotNone(scan_node) self.assertIsNotNone(scan_node.path_spec) @@ -824,7 +805,6 @@ def testScanForFileSystemOnVSS(self): test_vss_path_spec = path_spec_factory.Factory.NewPathSpec( definitions.TYPE_INDICATOR_VSHADOW, store_index=1, parent=test_raw_path_spec ) - path_spec = self._source_scanner.ScanForFileSystem(test_vss_path_spec) self.assertIsNotNone(path_spec) @@ -839,7 +819,6 @@ def testScanForFileSystemOnBodyFile(self): test_os_path_spec = path_spec_factory.Factory.NewPathSpec( definitions.TYPE_INDICATOR_OS, location=test_path ) - path_spec = self._source_scanner.ScanForFileSystem(test_os_path_spec) self.assertIsNone(path_spec) @@ -851,7 +830,6 @@ def testScanForStorageMediaImageOnRAW(self): test_os_path_spec = path_spec_factory.Factory.NewPathSpec( definitions.TYPE_INDICATOR_OS, location=test_path ) - path_spec = self._source_scanner.ScanForStorageMediaImage(test_os_path_spec) self.assertIsNotNone(path_spec) self.assertEqual(path_spec.type_indicator, definitions.TYPE_INDICATOR_RAW) @@ -864,7 +842,6 @@ def testScanForStorageMediaImageOnSplitRAW(self): test_os_path_spec = path_spec_factory.Factory.NewPathSpec( definitions.TYPE_INDICATOR_OS, location=test_path ) - path_spec = self._source_scanner.ScanForStorageMediaImage(test_os_path_spec) self.assertIsNotNone(path_spec) self.assertEqual(path_spec.type_indicator, definitions.TYPE_INDICATOR_RAW) @@ -877,7 +854,6 @@ def testScanForStorageMediaImageOnEWF(self): test_os_path_spec = path_spec_factory.Factory.NewPathSpec( definitions.TYPE_INDICATOR_OS, location=test_path ) - path_spec = self._source_scanner.ScanForStorageMediaImage(test_os_path_spec) self.assertIsNotNone(path_spec) self.assertEqual(path_spec.type_indicator, definitions.TYPE_INDICATOR_EWF) @@ -890,7 +866,6 @@ def testScanForStorageMediaImageOnQCOW(self): test_os_path_spec = path_spec_factory.Factory.NewPathSpec( definitions.TYPE_INDICATOR_OS, location=test_path ) - path_spec = self._source_scanner.ScanForStorageMediaImage(test_os_path_spec) self.assertIsNotNone(path_spec) self.assertEqual(path_spec.type_indicator, definitions.TYPE_INDICATOR_QCOW) @@ -903,7 +878,6 @@ def testScanForStorageMediaImageOnVHDI(self): test_os_path_spec = path_spec_factory.Factory.NewPathSpec( definitions.TYPE_INDICATOR_OS, location=test_path ) - path_spec = self._source_scanner.ScanForStorageMediaImage(test_os_path_spec) self.assertIsNotNone(path_spec) self.assertEqual(path_spec.type_indicator, definitions.TYPE_INDICATOR_VHDI) @@ -914,7 +888,6 @@ def testScanForStorageMediaImageOnVHDI(self): test_os_path_spec = path_spec_factory.Factory.NewPathSpec( definitions.TYPE_INDICATOR_OS, location=test_path ) - path_spec = self._source_scanner.ScanForStorageMediaImage(test_os_path_spec) self.assertIsNotNone(path_spec) self.assertEqual(path_spec.type_indicator, definitions.TYPE_INDICATOR_VHDI) @@ -927,7 +900,6 @@ def testScanForStorageMediaImageOnVMDK(self): test_os_path_spec = path_spec_factory.Factory.NewPathSpec( definitions.TYPE_INDICATOR_OS, location=test_path ) - path_spec = self._source_scanner.ScanForStorageMediaImage(test_os_path_spec) self.assertIsNotNone(path_spec) self.assertEqual(path_spec.type_indicator, definitions.TYPE_INDICATOR_VMDK) @@ -940,7 +912,6 @@ def testScanForStorageMediaImageOnBodyFile(self): test_os_path_spec = path_spec_factory.Factory.NewPathSpec( definitions.TYPE_INDICATOR_OS, location=test_path ) - path_spec = self._source_scanner.ScanForStorageMediaImage(test_os_path_spec) self.assertIsNone(path_spec) @@ -952,7 +923,6 @@ def testScanForVolumeSystemOnPartitionedImage(self): test_os_path_spec = path_spec_factory.Factory.NewPathSpec( definitions.TYPE_INDICATOR_OS, location=test_path ) - path_spec = self._source_scanner.ScanForVolumeSystem(test_os_path_spec) self.assertIsNotNone(path_spec) self.assertEqual( @@ -970,7 +940,6 @@ def testScanForVolumeSystemOnVSS(self): test_raw_path_spec = path_spec_factory.Factory.NewPathSpec( definitions.TYPE_INDICATOR_RAW, parent=test_os_path_spec ) - path_spec = self._source_scanner.ScanForVolumeSystem(test_raw_path_spec) self.assertIsNotNone(path_spec) self.assertEqual(path_spec.type_indicator, definitions.TYPE_INDICATOR_VSHADOW) @@ -985,7 +954,6 @@ def testScanForVolumeSystemOnBDE(self): test_os_path_spec = path_spec_factory.Factory.NewPathSpec( definitions.TYPE_INDICATOR_OS, location=test_path ) - path_spec = self._source_scanner.ScanForVolumeSystem(test_os_path_spec) self.assertIsNotNone(path_spec) self.assertEqual(path_spec.type_indicator, definitions.TYPE_INDICATOR_BDE) @@ -998,7 +966,6 @@ def testScanForVolumeSystemOnBodyFile(self): test_os_path_spec = path_spec_factory.Factory.NewPathSpec( definitions.TYPE_INDICATOR_OS, location=test_path ) - path_spec = self._source_scanner.ScanForVolumeSystem(test_os_path_spec) self.assertIsNone(path_spec) diff --git a/tests/helpers/volume_scanner.py b/tests/helpers/volume_scanner.py index 50b0b2f5..dee07d7b 100644 --- a/tests/helpers/volume_scanner.py +++ b/tests/helpers/volume_scanner.py @@ -248,7 +248,6 @@ def testGetVolumeIdentifiers(self): test_lvm_path_spec = path_spec_factory.Factory.NewPathSpec( definitions.TYPE_INDICATOR_LVM, parent=test_raw_path_spec ) - volume_system = lvm_volume_system.LVMVolumeSystem() volume_system.Open(test_lvm_path_spec) @@ -294,7 +293,6 @@ def testGetVolumeSnapshotIdentifiers(self): test_vss_path_spec = path_spec_factory.Factory.NewPathSpec( definitions.TYPE_INDICATOR_VSHADOW, parent=test_raw_path_spec ) - volume_system = vshadow_volume_system.VShadowVolumeSystem() volume_system.Open(test_vss_path_spec) @@ -344,7 +342,6 @@ def testNormalizedVolumeIdentifiersMBR(self): test_tsk_partition_path_spec = path_spec_factory.Factory.NewPathSpec( definitions.TYPE_INDICATOR_TSK_PARTITION, parent=test_raw_path_spec ) - volume_system = tsk_volume_system.TSKVolumeSystem() volume_system.Open(test_tsk_partition_path_spec) @@ -384,7 +381,6 @@ def testNormalizedVolumeIdentifiersVSS(self): test_vss_path_spec = path_spec_factory.Factory.NewPathSpec( definitions.TYPE_INDICATOR_VSHADOW, parent=test_raw_path_spec ) - volume_system = vshadow_volume_system.VShadowVolumeSystem() volume_system.Open(test_vss_path_spec) @@ -451,7 +447,6 @@ def testScanEncryptedVolumeOnBDE(self): self.assertEqual( fat_scan_node.type_indicator, definitions.PREFERRED_FAT_BACK_END ) - test_scanner._ScanEncryptedVolume(scan_context, bde_scan_node, test_options) self.assertEqual(len(bde_scan_node.sub_nodes), 1) @@ -459,7 +454,6 @@ def testScanEncryptedVolumeOnBDE(self): self.assertEqual( fat_scan_node.type_indicator, definitions.PREFERRED_FAT_BACK_END ) - # Test without mediator. resolver.Resolver.key_chain.Empty() @@ -482,7 +476,6 @@ def testScanEncryptedVolumeOnBDE(self): self.assertEqual( fat_scan_node.type_indicator, definitions.PREFERRED_FAT_BACK_END ) - with self.assertRaises(errors.ScannerError): test_scanner._ScanEncryptedVolume(scan_context, bde_scan_node, test_options) @@ -913,7 +906,6 @@ def testGetBasePathSpecsOnRAW(self): test_ext_path_spec = path_spec_factory.Factory.NewPathSpec( definitions.PREFERRED_EXT_BACK_END, location="/", parent=test_raw_path_spec ) - test_mediator = TestVolumeScannerMediator() test_scanner = volume_scanner.VolumeScanner(mediator=test_mediator) @@ -923,7 +915,6 @@ def testGetBasePathSpecsOnRAW(self): base_path_specs = [ base_path_spec.comparable for base_path_spec in base_path_specs ] - self.assertEqual(base_path_specs, expected_base_path_specs) # Test error conditions. @@ -944,11 +935,11 @@ def testGetBasePathSpecsOnMBR(self): test_raw_path_spec = path_spec_factory.Factory.NewPathSpec( definitions.TYPE_INDICATOR_RAW, parent=test_os_path_spec ) - test_tsk_partition_path_spec = path_spec_factory.Factory.NewPathSpec( definitions.TYPE_INDICATOR_TSK_PARTITION, location="/p1", part_index=2, + sector_size=512, start_offset=0x00000200, parent=test_raw_path_spec, ) @@ -957,10 +948,10 @@ def testGetBasePathSpecsOnMBR(self): location="/", parent=test_tsk_partition_path_spec, ) - test_tsk_partition_path_spec = path_spec_factory.Factory.NewPathSpec( definitions.TYPE_INDICATOR_TSK_PARTITION, location="/p5", + sector_size=512, part_index=6, start_offset=0x00010600, parent=test_raw_path_spec, @@ -970,7 +961,6 @@ def testGetBasePathSpecsOnMBR(self): location="/", parent=test_tsk_partition_path_spec, ) - test_mediator = TestVolumeScannerMediator() test_scanner = volume_scanner.VolumeScanner(mediator=test_mediator) @@ -978,12 +968,10 @@ def testGetBasePathSpecsOnMBR(self): test_ext_path_spec1.comparable, test_ext_path_spec2.comparable, ] - base_path_specs = test_scanner.GetBasePathSpecs(test_path) base_path_specs = [ base_path_spec.comparable for base_path_spec in base_path_specs ] - self.assertEqual(base_path_specs, expected_base_path_specs) def testGetBasePathSpecsOnDirectory(self): @@ -994,7 +982,6 @@ def testGetBasePathSpecsOnDirectory(self): test_os_path_spec = path_spec_factory.Factory.NewPathSpec( definitions.TYPE_INDICATOR_OS, location=test_path ) - test_mediator = TestVolumeScannerMediator() test_scanner = volume_scanner.VolumeScanner(mediator=test_mediator) @@ -1004,7 +991,6 @@ def testGetBasePathSpecsOnDirectory(self): base_path_specs = [ base_path_spec.comparable for base_path_spec in base_path_specs ] - self.assertEqual(base_path_specs, expected_base_path_specs) @@ -1027,7 +1013,6 @@ def testScanFileSystem(self): test_tsk_path_spec = path_spec_factory.Factory.NewPathSpec( definitions.TYPE_INDICATOR_TSK, location="/", parent=test_qcow_path_spec ) - test_mediator = TestVolumeScannerMediator() test_scanner = volume_scanner.WindowsVolumeScanner(mediator=test_mediator) diff --git a/tests/path/tsk_partition_path_spec.py b/tests/path/tsk_partition_path_spec.py index f97853b8..878a6906 100644 --- a/tests/path/tsk_partition_path_spec.py +++ b/tests/path/tsk_partition_path_spec.py @@ -20,25 +20,21 @@ def testInitialize(self): path_spec = tsk_partition_path_spec.TSKPartitionPathSpec( location="/p2", parent=self._path_spec ) - self.assertIsNotNone(path_spec) path_spec = tsk_partition_path_spec.TSKPartitionPathSpec( part_index=1, parent=self._path_spec ) - self.assertIsNotNone(path_spec) path_spec = tsk_partition_path_spec.TSKPartitionPathSpec( start_offset=0x2000, parent=self._path_spec ) - self.assertIsNotNone(path_spec) path_spec = tsk_partition_path_spec.TSKPartitionPathSpec( location="/p2", part_index=1, parent=self._path_spec ) - self.assertIsNotNone(path_spec) with self.assertRaises(ValueError): @@ -62,49 +58,51 @@ def testComparable(self): path_spec = tsk_partition_path_spec.TSKPartitionPathSpec( location="/p2", parent=self._path_spec ) - self.assertIsNotNone(path_spec) expected_comparable = "\n".join( ["type: TEST", "type: TSK_PARTITION, location: /p2", ""] ) - self.assertEqual(path_spec.comparable, expected_comparable) path_spec = tsk_partition_path_spec.TSKPartitionPathSpec( part_index=1, parent=self._path_spec ) - self.assertIsNotNone(path_spec) expected_comparable = "\n".join( ["type: TEST", "type: TSK_PARTITION, part index: 1", ""] ) + self.assertEqual(path_spec.comparable, expected_comparable) + + path_spec = tsk_partition_path_spec.TSKPartitionPathSpec( + sector_size=2048, parent=self._path_spec + ) + self.assertIsNotNone(path_spec) + expected_comparable = "\n".join( + ["type: TEST", "type: TSK_PARTITION, sector size: 2048", ""] + ) self.assertEqual(path_spec.comparable, expected_comparable) path_spec = tsk_partition_path_spec.TSKPartitionPathSpec( start_offset=0x2000, parent=self._path_spec ) - self.assertIsNotNone(path_spec) expected_comparable = "\n".join( ["type: TEST", "type: TSK_PARTITION, start offset: 0x00002000", ""] ) - self.assertEqual(path_spec.comparable, expected_comparable) path_spec = tsk_partition_path_spec.TSKPartitionPathSpec( location="/p2", part_index=1, parent=self._path_spec ) - self.assertIsNotNone(path_spec) expected_comparable = "\n".join( ["type: TEST", "type: TSK_PARTITION, location: /p2, part index: 1", ""] ) - self.assertEqual(path_spec.comparable, expected_comparable)