Merge branch 'develop'

README.md

@@ -45,7 +45,7 @@ $ ntfsdump /Windows/System32/winevt/Logs -o ./dump ./path/to/your/imagefile.raw
 extracting from E01 image (included splited-E01).
 
 ```.bash
-$ls
+$ ls
 imagefile.E01
 imagefile.E02
 imagefile.E03
@@ -71,6 +71,12 @@ $ ntfsfind '.*\.evtx' ./path/to/your/imagefile.raw | ntfsdump ./path/to/your/ima
 --version, -v:
     show program's version number and exit.
 
+--quiet, -q:
+    flat to suppress standard output.
+
+--nolog:
+    flag to no logs are output.
+
 --volume-num, -n:
     NTFS volume number (default: autodetect).
 
@@ -96,6 +102,20 @@ The image file to be processed must meet the following conditions.
 Additional file formats will be added in the future.  
 If you have any questions, please submit an issue.  
 
+
+## LogFormat
+ntfsdump outputs logs in the following format.  
+By default, it outputs the files to the current directory, but if you do not need them, please use the `--nolog` option.
+
+```
+- ntfsdump v{{version}} - 
+2022-01-01T00:00:00.000000: [{{EventName}}] {{Description}}
+2022-01-01T00:00:00.000000: [{{EventName}}] {{Description}}
+2022-01-01T00:00:00.000000: [{{EventName}}] {{Description}}
+...
+```
+
+
 ## Installation
 
 ### via PyPI

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "ntfsdump"
-version = "2.2.1"
+version = "2.3.0"
 description = "A tool for extract any files from an NTFS volume on an image file."
 authors = ["sumeshi <sum3sh1@protonmail.com>"]
 license = "MIT"

src/ntfsdump/__init__.py

@@ -1,14 +1,29 @@
 # coding: utf-8
-from typing import List, Literal, Optional
+from typing import Literal, Optional
 
 from ntfsdump.presenters.NtfsDumpPresenter import NtfsDumpPresenter
 
-def ntfsdump(imagefile_path: str, output_path: str, target_queries: List[str], volume_num: Optional[int] = None, file_type: Literal['raw', 'e01'] = 'raw'):
 
+def ntfsdump(
+    imagefile_path: str,
+    output_path: str,
+    target_queries: list[str],
+    volume_num: Optional[int] = None,
+    file_type: Literal['raw', 'e01'] = 'raw'
+):
+    """A tool for extract any files from an NTFS volume on an image file.
+
+    Args:
+        imagefile_path (str): target image file path.
+        output_path (str): output target file path, or output target directory path.
+        target_queries (list[str]): query for extracted file paths.
+        volume_num (Optional[int], optional): system volume number. Defaults to None.
+        file_type (Literal['raw', 'e01'], optional): target image file format. Defaults to 'raw'.
+    """
     NtfsDumpPresenter().ntfsdump(
-        imagefile_path=imagefile_path,
-        output_path=output_path,
-        target_queries=target_queries,
-        volume_num=volume_num,
-        file_type=file_type,
+        imagefile_path,
+        output_path,
+        target_queries,
+        volume_num,
+        file_type,
     )

src/ntfsdump/models/ImageFile.py

@@ -3,15 +3,16 @@
 from typing import List, Literal, Optional
 
 from ntfsdump.models.NtfsVolume import NtfsVolume
+from ntfsdump.models.Log import Log
 
 import pytsk3
 import pyewf
 
+
 class ewf_Img_Info(pytsk3.Img_Info):
   def __init__(self, ewf_handle):
     self._ewf_handle = ewf_handle
-    super(ewf_Img_Info, self).__init__(
-        url="", type=pytsk3.TSK_IMG_TYPE_EXTERNAL)
+    super(ewf_Img_Info, self).__init__(url="", type=pytsk3.TSK_IMG_TYPE_EXTERNAL)
 
   def close(self):
     self._ewf_handle.close()
@@ -27,18 +28,26 @@ def get_size(self):
 class ImageFile(object):
     def __init__(self, path: Path, volume_num: Optional[int], file_type: Literal['raw', 'e01'] = 'raw'):
         self.path: Path = path
+        self.logger: Log = Log()
         self.block_size: int = 512
         self.file_type: Literal['raw', 'e01'] = file_type
         self.volumes: List[NtfsVolume] = self.__analyze_partitions()
         self.main_volume: NtfsVolume = self.__auto_detect_main_partition(volume_num)
 
+        self.logger.log(f"[analyze] {len(self.volumes)} volumes were detected as NTFS volumes.", 'system')
+        for index, volume in enumerate(self.volumes):
+            self.logger.log(f"[analyze] NTFS Volume {index}: {volume.description}", 'system')
+        self.logger.log(f"[analyze] Volume {self.volumes.index(self.main_volume)} was automatically detected as the main partition.", 'system')
+
     def __analyze_partitions(self) -> List[NtfsVolume]:
         if self.file_type == 'e01':
+            self.logger.log(f"[analyze] E01 Format Image", 'system')
             filenames = pyewf.glob(str(self.path))
             ewf_handle = pyewf.handle()
             ewf_handle.open(filenames)
             img_info = ewf_Img_Info(ewf_handle)
         else:
+            self.logger.log(f"[analyze] Raw Format Image", 'system')
             img_info = pytsk3.Img_Info(str(self.path))
 
         volumes = pytsk3.Volume_Info(img_info)

src/ntfsdump/models/Log.py

@@ -0,0 +1,77 @@
+# coding: utf-8
+from pathlib import Path
+from typing import Literal
+from traceback import format_exc
+from datetime import datetime
+
+from ntfsdump.models.MetaData import MetaData
+
+
+def get_datetime() -> datetime:
+    return datetime.utcnow()
+
+def get_logfile_time():
+    if not MetaData.run_time:
+        MetaData.run_time = get_datetime()
+    return MetaData.run_time.strftime('%Y%m%d_%H%M%S_%f')
+
+
+class Log(object):
+    def __init__(
+        self,
+        path: Path = Path('.', f"{MetaData.name}_{get_logfile_time()}.log"),
+    ):
+        """Logging class
+
+        Args:
+            path (Path, optional): path of log file. Defaults to Path('.', f"{MetaData.name}_{get_logfile_time()}.log").
+            is_quiet (bool, optional): flag to supress standard output. Defaults to False.
+        """
+        self.path = path
+        self.is_quiet = MetaData.quiet
+
+        if not MetaData.nolog:
+            self.__create_logfile()
+
+    def __create_logfile(self):
+        if not self.path.exists():
+            self.path.write_text(f"- {MetaData.name} v{MetaData.version} - \n")
+
+    def __write_to_log(self, message: str):
+        try:
+            with self.path.open('a') as f:
+                f.write(f"{get_datetime().isoformat()}: {message}\n")
+        except Exception as e:
+            self.print_danger(format_exc())
+
+    def print_info(self, message: str):
+        """print with cyan color
+
+        Args:
+            message (str): a message to be printed.
+        """
+        print(f"\033[36m{message}\033[0m")
+
+    def print_danger(self, message: str):
+        """print with red color
+
+        Args:
+            message (str): a message to be printed.
+        """
+        print(f"\033[31m{message}\033[0m")
+
+    def log(self, message: str, type: Literal['system', 'info', 'danger'] = 'system'):
+        """print and write message to logfile
+
+        Args:
+            message (str): a message to be logged.
+            type (Literal['system', 'info', 'danger']): 'system ' is used only for logging.
+        """
+        if not MetaData.nolog:
+            self.__write_to_log(message)
+
+        if not self.is_quiet:
+            if type == 'info':
+                self.print_info(message) 
+            elif type == 'danger':
+                self.print_danger(message) 

src/ntfsdump/models/MetaData.py

@@ -0,0 +1,12 @@
+# coding: utf-8
+from typing import Optional, Final
+from datetime import datetime
+from importlib.metadata import version
+
+
+class MetaData(object):
+    name: Final[str] = 'ntfsdump'
+    version: Final[str] = version(name)
+    run_time: Optional[datetime] = None
+    quiet: bool = True
+    nolog: bool = True

src/ntfsdump/models/NtfsVolume.py

@@ -2,6 +2,8 @@
 from pathlib import Path
 from typing import List, Optional
 
+from ntfsdump.models.Log import Log
+
 import pytsk3
 
 
@@ -14,6 +16,10 @@ def __init__(self, path: Path, addr: int, description: str, start_byte: int, end
         self.start_byte = start_byte
         self.end_byte = end_byte
         self.fs_info = fs_info
+        self.logger = Log()
+
+    def __clean_query(self, query: str) -> str:
+        return query.replace('\\', '/')
 
     def __is_dir(self, query: str) -> bool:
         f = self.fs_info.open(query)
@@ -31,7 +37,6 @@ def __list_artifacts(self, query: str) -> List[str]:
         ]
 
     def __read_file(self, query: str) -> bytes:
-
         f = self.fs_info.open(query)
 
         offset = 0
@@ -51,10 +56,9 @@ def __read_file(self, query: str) -> bytes:
         return result
 
     def __read_alternate_data_stream(self, query: str, ads: str) -> Optional[bytes]:
-        query = query.replace('\\', '/')
         f = self.fs_info.open(query)
 
-        OFFSET = 0
+        offset = 0
         BUFF_SIZE = 1024 * 1024
 
         ads_attribute = None
@@ -67,11 +71,11 @@ def __read_alternate_data_stream(self, query: str, ads: str) -> Optional[bytes]:
             result = bytes()
             ADS_SIZE = ads_attribute.info.size
 
-            while OFFSET < ADS_SIZE:
-                available_to_read = min(BUFF_SIZE, ADS_SIZE - OFFSET)
-                data = f.read_random(OFFSET, available_to_read, ads_attribute.info.type, ads_attribute.info.id)
+            while offset < ADS_SIZE:
+                available_to_read = min(BUFF_SIZE, ADS_SIZE - offset)
+                data = f.read_random(offset , available_to_read, ads_attribute.info.type, ads_attribute.info.id)
                 if not data: break
-                OFFSET += len(data)
+                offset += len(data)
                 result += data
             return result
 
@@ -81,15 +85,16 @@ def __write_file(self, destination_path: Path, content: Optional[bytes], filenam
         # destination path is a file
         try:
             destination_path.write_bytes(content)
-            print(f"dumped: {destination_path}")
+            self.logger.log(f"[dumped] {destination_path}", 'info')
 
         # destination path is a directory
         except IsADirectoryError:
             Path(destination_path / filename).write_bytes(content)
-            print(f"dumped: {Path(destination_path / filename)}")
+            self.logger.log(f"[dumped] {Path(destination_path / filename)}", 'info')
 
     def dump_files(self, query: str, destination_path: Path) -> None:
-        query = query.replace('\\', '/')
+        query = self.__clean_query(query)
+        self.logger.log(f"[query] {query}", 'system')
 
         if self.__is_dir(query):
             for artifact in self.__list_artifacts(query):
@@ -135,5 +140,5 @@ def dump_files(self, query: str, destination_path: Path) -> None:
                 content = self.__read_file(query)
                 self.__write_file(destination_path, content, filename)
             except Exception as e:
-                print(e)
-                print(f"dump error: {query}")
+                self.logger.log(f"[error] {query}", 'danger')
+                self.logger.log(e, 'danger')

src/ntfsdump/presenters/NtfsDumpPresenter.py

@@ -5,7 +5,14 @@
 from ntfsdump.models.ImageFile import ImageFile
 
 class NtfsDumpPresenter(object):
-    def ntfsdump(self, imagefile_path: str, output_path: str, target_queries: List[str], volume_num: Optional[int] = None, file_type: Literal['raw', 'e01'] = 'raw'):
+    def ntfsdump(
+        self,
+        imagefile_path: str,
+        output_path: str,
+        target_queries: List[str],
+        volume_num: Optional[int] = None,
+        file_type: Literal['raw', 'e01'] = 'raw'
+    ):
         # dump files
         image = ImageFile(Path(imagefile_path), volume_num, file_type)
         for target_query in target_queries:

src/ntfsdump/views/BaseView.py

@@ -1,7 +1,8 @@
 # coding: utf-8
 import argparse
 from abc import ABCMeta, abstractmethod
-from importlib.metadata import version
+
+from ntfsdump.models.MetaData import MetaData
 
 
 class BaseView(metaclass=ABCMeta):
@@ -11,13 +12,10 @@ def __init__(self):
         self.__define_common_options()
 
     def __define_common_options(self):
-        self.parser.add_argument("--version", "-v", action="version", version=version('ntfsdump'))
+        self.parser.add_argument("--version", "-v", action="version", version=MetaData.version)
         self.parser.add_argument("--quiet", "-q", action='store_true', help="flag to suppress standard output.")
+        self.parser.add_argument("--nolog", action='store_true', help="flag to no logs are output.")
 
     @abstractmethod
     def define_options(self):
         pass
-
-    def log(self, message: str, is_quiet: bool):
-        if not is_quiet:
-            print(message)