Improve plugin arguments

dissect/target/plugins/apps/other/env.py

@@ -29,8 +29,8 @@ def check_compatible(self) -> None:
         pass
 
     @export(record=EnvironmentFileRecord)
-    @arg("--env-path", help="path to scan environment files in", required=True)
-    @arg("--extension", help="extension of files to scan", default="env")
+    @arg("--env-path", required=True, help="path to scan environment files in")
+    @arg("--extension", default="env", help="extension of files to scan")
     def envfile(self, env_path: str, extension: str = "env") -> Iterator[EnvironmentFileRecord]:
         """Yield environment variables found in ``.env`` files at the provided path."""
 

dissect/target/plugins/apps/vpn/openvpn.py

@@ -162,7 +162,7 @@ def _load_config(self, parser: OpenVPNParser, config_path: fsutil.TargetPath) ->
         return parser.parsed_data
 
     @export(record=[OpenVPNServer, OpenVPNClient])
-    @arg("--export-key", action="store_true")
+    @arg("--export-key", action="store_true", help="export private keys to records")
     def config(self, export_key: bool = False) -> Iterator[OpenVPNServer | OpenVPNClient]:
         """Parses config files from openvpn interfaces."""
         # We define the parser here so we can reuse it

dissect/target/plugins/apps/webserver/caddy.py

@@ -7,9 +7,9 @@
 
 from dissect.util.ts import from_unix
 
-from dissect.target import plugin
 from dissect.target.exceptions import FileNotFoundError, UnsupportedPluginError
 from dissect.target.helpers.fsutil import basename, open_decompress
+from dissect.target.plugin import export
 from dissect.target.plugins.apps.webserver.webserver import (
     WebserverAccessLogRecord,
     WebserverPlugin,
@@ -95,7 +95,7 @@ def get_log_paths(self) -> list[Path]:
 
         return log_paths
 
-    @plugin.export(record=WebserverAccessLogRecord)
+    @export(record=WebserverAccessLogRecord)
     def access(self) -> Iterator[WebserverAccessLogRecord]:
         """Parses Caddy V1 CRF and Caddy V2 JSON access logs.
 

dissect/target/plugins/apps/webserver/iis.py

@@ -8,11 +8,11 @@
 from defusedxml import ElementTree
 from flow.record.base import RE_VALID_FIELD_NAME
 
-from dissect.target import plugin
 from dissect.target.exceptions import FileNotFoundError as DissectFileNotFoundError
 from dissect.target.exceptions import PluginError, UnsupportedPluginError
 from dissect.target.helpers.fsutil import has_glob_magic
 from dissect.target.helpers.record import TargetRecordDescriptor
+from dissect.target.plugin import export
 from dissect.target.plugins.apps.webserver.webserver import (
     WebserverAccessLogRecord,
     WebserverPlugin,
@@ -302,7 +302,7 @@ def parse_w3c_format_log(self, path: Path) -> Iterator[TargetRecordDescriptor]:
                 **{normalise_field_name(field): raw.get(field) for field in extra_fields},
             )
 
-    @plugin.export(record=BasicRecordDescriptor)
+    @export(record=BasicRecordDescriptor)
     def logs(self) -> Iterator[TargetRecordDescriptor]:
         """Return contents of IIS (v7 and above) log files.
 
@@ -323,7 +323,7 @@ def logs(self) -> Iterator[TargetRecordDescriptor]:
             self.target.log.info("Parsing IIS log file %s in %s format", log_file, log_format)
             yield from parse_func(log_file)
 
-    @plugin.export(record=WebserverAccessLogRecord)
+    @export(record=WebserverAccessLogRecord)
     def access(self) -> Iterator[WebserverAccessLogRecord]:
         """Return contents of IIS (v7 and above) log files in unified WebserverAccessLogRecord format.
 

dissect/target/plugins/filesystem/icat.py

@@ -21,13 +21,13 @@ def check_compatible(self) -> None:
         if not any(fs.__type__ in self.FS_SUPPORTED for fs in filesystems):
             raise UnsupportedPluginError("No supported filesystems found")
 
-    @arg("--segment", "--inode", "-i", dest="inum", required=True, type=int, help="MFT segment or inode number")
+    @arg("-i", "--inode", "--segment", dest="inum", type=int, required=True, help="MFT segment or inode number")
     @arg(
         "--fs",
         type=int,
         help="optional filesystem index, zero indexed. Defaults to the 'sysvol' or '/' filesystem otherwise",
     )
-    @arg("--ads", type=str, default="", help="Alternate Data Stream name")
+    @arg("--ads", default="", help="Alternate Data Stream name")
     @export(output="none")
     def icat(self, inum: int, fs: int | None, ads: str) -> None:
         """Output the contents of a file based on its MFT segment or inode number. Supports Alternate Data Streams

dissect/target/plugins/filesystem/ntfs/mft.py

@@ -158,7 +158,7 @@ def check_compatible(self) -> None:
         action="store_true",
         help="compacts the MFT entry timestamps into a single record",
     )
-    @arg("--fs", type=int, default=None, help="optional filesystem index, zero indexed")
+    @arg("--fs", type=int, help="optional filesystem index, zero indexed")
     @arg("--start", type=int, default=0, help="the first MFT segment number")
     @arg("--end", type=int, default=-1, help="the last MFT segment number")
     @arg(

dissect/target/plugins/filesystem/yara.py

@@ -50,7 +50,7 @@ def check_compatible(self) -> None:
     @arg("-r", "--rules", required=True, nargs="*", help="path(s) to YARA rule file(s) or folder(s)")
     @arg("-p", "--path", default="/", help="path on target(s) to recursively scan")
     @arg("-m", "--max-size", type=int, default=DEFAULT_MAX_SCAN_SIZE, help="maximum file size in bytes to scan")
-    @arg("-c", "--check", default=False, action="store_true", help="check if every YARA rule is valid")
+    @arg("-c", "--check", action="store_true", help="check if every YARA rule is valid")
     @export(record=YaraMatchRecord)
     def yara(
         self,

dissect/target/plugins/general/loaders.py

@@ -18,7 +18,7 @@ def check_compatible(self) -> None:
     # For now we use --as-json, but in the future this should be changed to inherit --json from target-query.
     # https://github.com/fox-it/dissect.target/pull/841
     # https://github.com/fox-it/dissect.target/issues/889
-    @arg("--as-json", dest="as_json", action="store_true")
+    @arg("--as-json", dest="as_json", action="store_true", help="output in JSON format")
     def loaders(self, as_json: bool = False) -> None:
         """List the available loaders."""
 

dissect/target/plugins/general/osinfo.py

@@ -5,9 +5,9 @@
 
 from flow.record import GroupedRecord
 
-from dissect.target import plugin
 from dissect.target.exceptions import UnsupportedPluginError
 from dissect.target.helpers.record import TargetRecordDescriptor
+from dissect.target.plugin import Plugin, export
 
 OSInfoRecord = TargetRecordDescriptor(
     "generic/osinfo",
@@ -18,14 +18,14 @@
 )
 
 
-class OSInfoPlugin(plugin.Plugin):
+class OSInfoPlugin(Plugin):
     """Convenience plugin that wraps _os.* functions in records."""
 
     def check_compatible(self) -> None:
         if not self.target._os_plugin:
             raise UnsupportedPluginError("No operating system detected on target")
 
-    @plugin.export(record=OSInfoRecord)
+    @export(record=OSInfoRecord)
     def osinfo(self) -> Iterator[OSInfoRecord | GroupedRecord]:
         """Yield grouped records with target OS info."""
         for os_func in self.target._os.__functions__:

dissect/target/plugins/general/plugins.py

@@ -59,16 +59,24 @@ def generate_functions_json(functions: list[plugin.FunctionDescriptor] | None =
 
     for desc in functions or _get_default_functions():
         docstring = desc.func.__doc__.split("\n\n", 1)[0].strip() if desc.func.__doc__ else None
-        arguments = [
-            {
+        arguments = []
+
+        for name, _arg in desc.args:
+            is_bool_action = _arg.get("action", "") in ("store_true", "store_false")
+            arg_desc = {
                 "name": name[0],
-                "type": getattr(arg.get("type"), "__name__", None),
-                "help": arg.get("help"),
-                "default": arg.get("default"),
-                "required": arg.get("required", False),
+                # infer the type either by store_*, type argument and fallback to default str.
+                # See: https://docs.python.org/3/library/argparse.html#type
+                "type": "bool" if is_bool_action else getattr(_arg.get("type"), "__name__", "str"),
+                "help": _arg.get("help"),
+                "default": _arg.get("action") == "store_false"
+                if is_bool_action
+                else (_arg.get("default") or _arg.get("const")),
+                # required can either be set explicitly or is implied with '--' style arguments.
+                "required": _arg.get("required", False),
             }
-            for name, arg in desc.args
-        ]
+
+            arguments.append(arg_desc)
 
         loaded.append(
             {
@@ -151,12 +159,12 @@ def check_compatible(self) -> None:
         pass
 
     @export(output="none", cache=False)
-    @arg("--docs", dest="print_docs", action="store_true")
+    @arg("--docs", dest="print_docs", action="store_true", help="output docstrings")
     # NOTE: We would prefer to re-use arguments across plugins from argparse in query.py, but that is not possible yet.
     # For now we use --as-json, but in the future this should be changed to inherit --json from target-query.
     # https://github.com/fox-it/dissect.target/pull/841
     # https://github.com/fox-it/dissect.target/issues/889
-    @arg("--as-json", dest="as_json", action="store_true")
+    @arg("--as-json", dest="as_json", action="store_true", help="output in JSON format")
     def plugins(self, print_docs: bool = False, as_json: bool = False) -> None:
         """Print all available plugins."""
         if as_json:

dissect/target/plugins/os/unix/etc/etc.py

@@ -70,8 +70,8 @@ def _sub(
                 yield UnixConfigTreeRecord(**data)
 
     @export(record=UnixConfigTreeRecord)
-    @arg("--glob", dest="pattern", required=False, default="*", type=str, help="Glob-style pattern to search for")
-    @arg("--root", dest="root", required=False, default="/", type=str, help="Path to use as root for search")
+    @arg("--glob", dest="pattern", default="*", help="Glob-style pattern to search for")
+    @arg("--root", dest="root", default="/", help="Path to use as root for search")
     def etc(self, pattern: str, root: str) -> Iterator[UnixConfigTreeRecord]:
         """Yield etc configuration records."""
 

dissect/target/plugins/os/windows/defender/_plugin.py

@@ -5,9 +5,9 @@
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, TextIO
 
-from dissect.target import plugin
 from dissect.target.exceptions import UnsupportedPluginError
 from dissect.target.helpers.record import TargetRecordDescriptor
+from dissect.target.plugin import Plugin, arg, export
 from dissect.target.plugins.os.windows.defender.mplog import (
     DEFENDER_MPLOG_BLOCK_PATTERNS,
     DEFENDER_MPLOG_LINE,
@@ -137,7 +137,7 @@ def filter_func(record: Record) -> bool:
     return filter(filter_func, records)
 
 
-class MicrosoftDefenderPlugin(plugin.Plugin):
+class MicrosoftDefenderPlugin(Plugin):
     """Plugin that parses artifacts created by Microsoft Defender.
 
     This includes the EVTX logs, as well as recovery of artefacts from the quarantine folder.
@@ -161,7 +161,7 @@ def check_compatible(self) -> None:
         ):
             raise UnsupportedPluginError("No Defender objects found")
 
-    @plugin.export(record=DefenderLogRecord)
+    @export(record=DefenderLogRecord)
     def evtx(self) -> Iterator[DefenderLogRecord]:
         """Parse Microsoft Defender evtx log files."""
 
@@ -185,7 +185,7 @@ def evtx(self) -> Iterator[DefenderLogRecord]:
 
             yield DefenderLogRecord(**record_fields, _target=self.target)
 
-    @plugin.export(record=[DefenderQuarantineRecord, DefenderFileQuarantineRecord])
+    @export(record=[DefenderQuarantineRecord, DefenderFileQuarantineRecord])
     def quarantine(self) -> Iterator[DefenderQuarantineRecord | DefenderFileQuarantineRecord]:
         """Parse the quarantine folder of Microsoft Defender for quarantine entry resources.
 
@@ -225,7 +225,7 @@ def quarantine(self) -> Iterator[DefenderQuarantineRecord | DefenderFileQuaranti
                     # For anything other than a file, we yield a generic DefenderQuarantineRecord.
                     yield DefenderQuarantineRecord(**fields, _target=self.target)
 
-    @plugin.export(record=DefenderExclusionRecord)
+    @export(record=DefenderExclusionRecord)
     def exclusions(self) -> Iterator[DefenderExclusionRecord]:
         """Yield Microsoft Defender exclusions from the Registry."""
 
@@ -413,7 +413,7 @@ def _mplog(
             yield from self._mplog_line(mplog_line, source, tzinfo=tzinfo)
             yield from self._mplog_block(mplog_line, mplog, source, tzinfo=tzinfo)
 
-    @plugin.export(
+    @export(
         record=[
             DefenderMPLogProcessImageRecord,
             DefenderMPLogMinFilUSSRecord,
@@ -472,15 +472,15 @@ def mplog(
                 except UnicodeError:
                     continue
 
-    @plugin.arg(
-        "--output",
+    @arg(
         "-o",
+        "--output",
         dest="output_dir",
         type=Path,
         required=True,
         help="Path to recover quarantined file to.",
     )
-    @plugin.export(output="none")
+    @export(output="none")
     def recover(self, output_dir: Path) -> None:
         """Recover files that have been placed into quarantine by Microsoft Defender.
 

dissect/target/plugins/os/windows/lnk.py

@@ -136,7 +136,7 @@ def check_compatible(self) -> None:
                 return
         raise UnsupportedPluginError("No folders containing link files found")
 
-    @arg("--path", "-p", dest="path", default=None, help="Path to directory or .lnk file in target")
+    @arg("-p", "--path", help="path to directory or .lnk file in target")
     @export(record=LnkRecord)
     def lnk(self, path: str | None = None) -> Iterator[LnkRecord]:
         """Parse all .lnk files in /ProgramData, /Users, and /Windows or from a specified path in record format.

dissect/target/plugins/os/windows/log/evtx.py

@@ -9,9 +9,9 @@
 from dissect.eventlog.exceptions import MalformedElfChnkException
 from flow.record import Record, utils
 
-from dissect.target import plugin
 from dissect.target.exceptions import FilesystemError
 from dissect.target.helpers.record import DynamicDescriptor, TargetRecordDescriptor
+from dissect.target.plugin import Plugin, arg, export
 from dissect.target.plugins.os.windows.log.evt import WindowsEventlogsMixin
 
 if TYPE_CHECKING:
@@ -26,7 +26,7 @@
 EVTX_GLOB = "*.evtx"
 
 
-class EvtxPlugin(WindowsEventlogsMixin, plugin.Plugin):
+class EvtxPlugin(WindowsEventlogsMixin, Plugin):
     """Plugin for fetching and parsing Windows Eventlog Files (``*.evtx``)."""
 
     RECORD_NAME = "filesystem/windows/evtx"
@@ -39,9 +39,9 @@ def __init__(self, target: Target):
         super().__init__(target)
         self._create_event_descriptor = lru_cache(4096)(self._create_event_descriptor)
 
-    @plugin.arg("--logs-dir", help="logs directory to scan")
-    @plugin.arg("--log-file-glob", default=EVTX_GLOB, help="glob pattern to match a log file name")
-    @plugin.export(record=DynamicDescriptor(["datetime"]))
+    @arg("--logs-dir", help="logs directory to scan")
+    @arg("--log-file-glob", default=EVTX_GLOB, help="glob pattern to match a log file name")
+    @export(record=DynamicDescriptor(["datetime"]))
     def evtx(self, log_file_glob: str = EVTX_GLOB, logs_dir: str | None = None) -> Iterator[DynamicDescriptor]:
         """Return entries from Windows Event log files (``*.evtx``).
 
@@ -86,7 +86,7 @@ def evtx(self, log_file_glob: str = EVTX_GLOB, logs_dir: str | None = None) -> I
             for event in evtx.Evtx(entry_data):
                 yield self._build_record(event, entry)
 
-    @plugin.export(record=DynamicDescriptor(["datetime"]))
+    @export(record=DynamicDescriptor(["datetime"]))
     def scraped_evtx(self) -> Iterator[DynamicDescriptor]:
         """Return EVTX log file records scraped from target disks."""
         yield from self.target.scrape.scrape_chunks_from_disks(

dissect/target/plugins/os/windows/rdpcache.py

@@ -381,34 +381,33 @@ def paths(self) -> Iterator[RDPCacheRecord]:
                 _target=self.target,
             )
 
-    @arg("--output", "-o", dest="output_dir", type=Path, required=True, help="path to recover bitmap files to")
+    @arg("-o", "--output", dest="output_dir", type=Path, required=True, help="path to recover bitmap files to")
     @arg(
-        "--grid",
         "-g",
+        "--grid",
         dest="as_grid",
         action="store_true",
         help="assemble all recovered tiles of a cache into a single grid image with borders around each tile",
     )
     @arg(
-        "--collage",
         "-c",
+        "--collage",
         dest="as_collage",
         action="store_true",
         help="assemble all recovered tiles of a cache into one image",
     )
     @arg(
-        "--no-tiles",
         "-n",
+        "--no-tiles",
         dest="no_individual_tiles",
         action="store_true",
         help="do not save each recovered tile as a separate bitmap",
     )
     @arg(
-        "--remnants",
         "-r",
+        "--remnants",
         choices=["only", "include", "exclude"],
         default="exclude",
-        type=str,
         help="include old leftover colordata in between tiles as separate, 'remnant' tiles",
     )
     @export(output="none")

dissect/target/plugins/os/windows/thumbcache.py

@@ -99,13 +99,13 @@ def _parse_thumbcache(
                 self.target.log.error("Error parsing thumbcache: %s", e)  # noqa: TRY400
                 self.target.log.debug(e, exc_info=e)
 
-    @arg("--output", "-o", dest="output_dir", type=Path, help="Path to extract thumbcache thumbnails to.")
+    @arg("-o", "--output", dest="output_dir", type=Path, help="path to extract thumbcache thumbnails to")
     @export(record=[ThumbcacheRecord, IndexRecord])
     def thumbcache(self, output_dir: Path | None = None) -> Iterator[ThumbcacheRecord | IndexRecord]:
         """Yield thumbcache thumbnails."""
         yield from self._parse_thumbcache(ThumbcacheRecord, "thumbcache", output_dir)
 
-    @arg("--output", "-o", dest="output_dir", type=Path, help="Path to extract iconcache thumbnails to.")
+    @arg("-o", "--output", dest="output_dir", type=Path, help="path to extract iconcache thumbnails to")
     @export(record=[IconcacheRecord, IndexRecord])
     def iconcache(self, output_dir: Path | None = None) -> Iterator[IconcacheRecord | IndexRecord]:
         """Yield iconcache thumbnails."""

dissect/target/plugins/scrape/qfind.py

@@ -24,9 +24,9 @@ class QFindPlugin(Plugin):
     def check_compatible(self) -> None:
         pass
 
-    @arg("-n", "--needles", type=str, nargs="*", metavar="NEEDLES", help="needles to search for")
+    @arg("-n", "--needles", nargs="*", metavar="NEEDLES", help="needles to search for")
     @arg("-nf", "--needle-file", type=Path, help="file containing the needles to search for")
-    @arg("-e", "--encoding", type=str, help="encode text needles with these comma separated encodings")
+    @arg("-e", "--encoding", help="encode text needles with these comma separated encodings")
     @arg("--no-hex-decode", action="store_true", help="do not automatically add decoded hex needles (only in raw mode)")
     @arg("-R", "--raw", action="store_true", help="show raw hex dumps instead of post-processed string output")
     @arg("-i", "--ignore-case", action="store_true", help="case insensitive search")