Converted to plugin & made some extraction generic

Author: B0TAxy

dissect/target/plugins/os/windows/ad/bloodhound.py

@@ -2,18 +2,19 @@
 
 import json
 from functools import cache
+from pathlib import Path
 from typing import TYPE_CHECKING, Any
 
 from dissect.database.ese.ntds import NTDS
 from dissect.database.ese.ntds.c_sd import c_sd
 from dissect.database.ese.ntds.util import UserAccountControl
 
+from dissect.target.plugin import Plugin, UnsupportedPluginError, arg, export
+
 if TYPE_CHECKING:
-    from collections.abc import Callable, Iterator
-    from pathlib import Path
+    from collections.abc import Iterator
 
     from dissect.database.ese.ntds import NTDS
-    from dissect.database.ese.ntds.objects import Object, SecurityObject
     from dissect.database.ese.ntds.sd import SecurityDescriptor
     from flow.record import Record
 
@@ -153,10 +154,10 @@ def extract_sd_data(ntds: NTDS, nt_security_descriptor: int | None) -> tuple[boo
     return c_sd.SECURITY_DESCRIPTOR_CONTROL.SE_DACL_PROTECTED.name in sd.header.Control.name.split("|"), aces
 
 
-class BloodHoundExporter:
-    def __init__(self, ntds: NTDS, output_dir: Path) -> None:
-        self.ntds: NTDS = ntds
-        self.output_dir: Path = output_dir
+class BloodHound(Plugin):
+    def check_compatible(self) -> None:
+        if not self.target.has_function("ad"):
+            raise UnsupportedPluginError("ad plugin is not initialized")
 
     @staticmethod
     def extract_high_value(obj: Record) -> str | None:
@@ -170,8 +171,31 @@ def extract_domain_id(obj: Record) -> str | None:
     def extract_flag_from_enum(obj: Record, flag: UserAccountControl) -> bool:
         return flag.name in obj.user_account_control.split("|")
 
-    def iterate_domains(self, domains: Callable[[], Iterator[Record]]):
-        for domain in domains():
+    def extract_generic_info(self, obj: Record) -> dict[str, Any]:
+        is_acl_protected, aces = extract_sd_data(self.ntds, obj.nt_security_descriptor)
+
+        contained_by = None
+        if obj.parent_guid and obj.parent_type:
+            contained_by = {"ObjectIdentifier": obj.parent_guid, "ObjectType": obj.parent_type}
+
+        return {
+            "ObjectIdentifier": obj.sid,
+            "IsDeleted": obj.is_deleted.value,
+            "IsACLProtected": is_acl_protected,
+            "Aces": aces,
+            "ContainedBy": contained_by,
+        }
+
+    def extract_generic_properties(self, obj: Record) -> dict[str, Any]:
+        return {
+            "domain": obj.domain,  # TODO: Make sure this is robust because it's not from ntds.dit
+            "name": obj.name,
+            "distinguishedname": obj.distinguished_name,
+            "enabled": not self.extract_flag_from_enum(obj, UserAccountControl.ACCOUNTDISABLE),
+        }
+
+    def translate_domains(self) -> Iterator[dict[str, Any]]:
+        for domain in self.target.ad.domains():
             yield {
                 "ObjectIdentifier": domain.sid,
                 "Properties": {
@@ -186,24 +210,17 @@ def iterate_domains(self, domains: Callable[[], Iterator[Record]]):
                 "Links": [],
             }
 
-    def iterate_users(self, users: Callable[[], Iterator[Record]]) -> Iterator[dict[str, Any]]:
+    def translate_users(self) -> Iterator[dict[str, Any]]:
         """Iterate over user records and yield BloodHound-formatted dictionaries."""
-        for user in users():
-            is_acl_protected, aces = extract_sd_data(self.ntds, user.nt_security_descriptor)
+        for user in self.target.ad.users():
             yield {
-                "ObjectIdentifier": user.sid,
-                "IsDeleted": user.is_deleted.value,
-                "IsACLProtected": is_acl_protected,
+                **self.extract_generic_info(user),
                 "HasSIDHistory": user.sid_history,
                 "SPNTargets": user.service_principal_names,  # TODO: Verify this is correct for SPN targeting in BloodHound
                 "PrimaryGroupSID": user.sid.replace(f"-{user.rid}", f"-{user.primary_group_id}"),
                 "AllowedToDelegate": user.allowed_to_delegate,
-                "ContainedBy": {"ObjectIdentifier": user.parent_guid, "ObjectType": user.parent_type},
-                "Aces": aces,
                 "Properties": {
-                    "domain": user.domain,  # TODO: Make sure this is robust because it's not from ntds.dit
-                    "name": user.name,
-                    "distinguishedname": user.distinguished_name,
+                    **self.extract_generic_properties(user),
                     "domainsid": self.extract_domain_id(user),
                     "highvalue": self.extract_high_value(user),
                     "samaccountname": user.sam_name,
@@ -215,7 +232,6 @@ def iterate_users(self, users: Callable[[], Iterator[Record]]) -> Iterator[dict[
                         user, UserAccountControl.TRUSTED_FOR_DELEGATION
                     ),
                     "pwdneverexpires": self.extract_flag_from_enum(user, UserAccountControl.DONT_EXPIRE_PASSWORD),
-                    "enabled": not self.extract_flag_from_enum(user, UserAccountControl.ACCOUNTDISABLE),
                     "trustedtoauth": bool(user.allowed_to_delegate)
                     and self.extract_flag_from_enum(user, UserAccountControl.TRUSTED_TO_AUTHENTICATE_FOR_DELEGATION),
                     "lastlogon": user.logon_last_success_observed.isoformat()
@@ -233,31 +249,27 @@ def iterate_users(self, users: Callable[[], Iterator[Record]]) -> Iterator[dict[
                     "homedirectory": user.home_directory,
                     "userpassword": None,
                     "unixpassword": None,
-                    "unicodepassword": user.nt,  # TODO: figure out lm hash goes here or not
+                    "unicodepassword": user.nt,  # TODO: Figure out lm hash goes here or not
                     "sfupassword": None,
                     "logonscript": user.logon_script,
                     "admincount": user.admin_count.value,
                     "sidhistory": user.sid_history,
                 },
             }
 
-    def iterate_computers(self, computers: Callable[[], Iterator[Record]]):
-        for computer in computers():
+    def translate_computers(self) -> Iterator[dict[str, Any]]:
+        for computer in self.target.ad.computers():
             yield {
-                "ObjectIdentifier": computer.sid,
+                **self.extract_generic_info(computer),
                 "Properties": {
-                    "domain": self.extract_domain_id(computer),
-                    "name": computer.dns_hostname,
-                    "distinguishedname": computer.distinguished_name,
+                    **self.extract_generic_properties(computer),
                     "operatingsystem": computer.operating_system,
-                    "enabled": self.extract_flag_from_enum(computer),
                 },
-                "Aces": extract_sd_data(self.ntds, computer.nt_security_descriptor),
                 "AllowedToDelegate": computer.allowed_to_delegate,
             }
 
-    def iterate_groups(self, groups: Callable[[], Iterator[Record]]):
-        for group in groups():
+    def translate_groups(self) -> Iterator[dict[str, Any]]:
+        for group in self.target.ad.groups():
             yield {
                 "ObjectIdentifier": group.sid,
                 "Properties": {
@@ -269,8 +281,8 @@ def iterate_groups(self, groups: Callable[[], Iterator[Record]]):
                 "Members": group.members,
             }
 
-    def iterate_ous(self, ous: Callable[[], Iterator[Record]]):
-        for ou in ous():
+    def translate_ous(self) -> Iterator[dict[str, Any]]:
+        for ou in self.target.ad.ous():
             yield {
                 "ObjectIdentifier": ou.sid,
                 "Properties": {
@@ -283,8 +295,8 @@ def iterate_ous(self, ous: Callable[[], Iterator[Record]]):
                 "Links": [],
             }
 
-    def iterate_gpos(self, group_policies: Callable[[], Iterator[Record]]):
-        for gpo in group_policies():
+    def translate_gpos(self) -> Iterator[dict[str, Any]]:
+        for gpo in self.target.ad.gpos():
             yield {
                 "ObjectIdentifier": gpo.sid,
                 "Properties": {
@@ -295,32 +307,38 @@ def iterate_gpos(self, group_policies: Callable[[], Iterator[Record]]):
                 "Aces": extract_sd_data(self.ntds, gpo.nt_security_descriptor),
             }
 
-    def write_bloodhound_json(self, records: Callable[[], Iterator[dict[str, Any]]]) -> None:
+    @arg("-o", "--output", dest="output_dir", type=Path, required=True, help="Path to extract BloodHound files to")
+    @export(output="none")
+    def bloodhound(self, output_dir: Path) -> None:
+        """Extract AD objects in BloodHound format and write them iteratively to disk."""
+
         TYPE_TO_FUNCTION_MAPPING = {
-            "users": self.iterate_users,
-            "computers": self.iterate_computers,
-            "domains": self.iterate_domains,
-            "groups": self.iterate_groups,
-            "ous": self.iterate_ous,
-            "gpos": self.iterate_gpos,
+            "users": self.translate_users,
+            "computers": self.translate_computers,
+            "domains": self.translate_domains,
+            "groups": self.translate_groups,
+            "ous": self.translate_ous,
+            "gpos": self.translate_gpos,
         }
 
-        object_type = records.__name__
-        output_path = self.output_dir.joinpath(object_type).with_suffix(".json")
+        output_dir.mkdir(parents=True, exist_ok=True)
+
+        for object_type, translation_function in TYPE_TO_FUNCTION_MAPPING.items():
+            output_path = output_dir.joinpath(object_type).with_suffix(".json")
 
-        metadata = {"methods": 0, "type": object_type, "version": 6, "count": 0}
-        json_start = '{\n\t"data": [\n\t\t'
+            metadata = {"methods": 0, "type": object_type, "version": 6, "count": 0}
+            json_start = '{\n\t"data": [\n\t\t'
 
-        with output_path.open("w", encoding="utf-8") as output_handle:
-            output_handle.write(json_start)
-            first = True
-            for item in TYPE_TO_FUNCTION_MAPPING[object_type](records):
-                if not first:
-                    output_handle.write(",\n\t\t")
+            with output_path.open("w", encoding="utf-8") as output_handle:
+                output_handle.write(json_start)
+                first = True
+                for item in translation_function():
+                    if not first:
+                        output_handle.write(",\n\t\t")
 
-                metadata["count"] += 1
-                output_handle.write(json.dumps(item))
-                first = False
+                    metadata["count"] += 1
+                    output_handle.write(json.dumps(item))
+                    first = False
 
-            json_end = '\n\t\t], \n\t"meta": ' + json.dumps(metadata) + "\n}\n"
-            output_handle.write(json_end)
+                json_end = '\n\t\t], \n\t"meta": ' + json.dumps(metadata) + "\n}\n"
+                output_handle.write(json_end)

dissect/target/plugins/os/windows/ad/ntds.py

@@ -3,16 +3,14 @@
 from datetime import datetime
 from functools import cached_property
 from itertools import zip_longest
-from pathlib import Path
 from typing import TYPE_CHECKING, Any
 
 from dissect.database.ese.ntds import NTDS
 from dissect.database.ese.ntds.util import UserAccountControl
 
 from dissect.target.exceptions import RegistryKeyNotFoundError
 from dissect.target.helpers.record import TargetRecordDescriptor
-from dissect.target.plugin import Plugin, UnsupportedPluginError, arg, export, internal
-from dissect.target.plugins.os.windows.ad.bloodhound import BloodHoundExporter
+from dissect.target.plugin import Plugin, UnsupportedPluginError, export, internal
 from dissect.target.plugins.os.windows.sam import des_decrypt
 
 if TYPE_CHECKING:
@@ -299,21 +297,6 @@ def secretsdump(self) -> Iterator[str]:
                 if "Primary:CLEARTEXT" in supplemental:
                     yield f"{username}:CLEARTEXT:{supplemental['Primary:CLEARTEXT']}"
 
-    @arg("-o", "--output", dest="output_dir", type=Path, required=True, help="Path to extract BloodHound files to")
-    @export(output="none")
-    def bloodhound(self, output_dir: Path) -> None:
-        """Extract AD objects in BloodHound format and write them iteratively to disk."""
-
-        output_dir.mkdir(parents=True, exist_ok=True)
-        bloodhound = BloodHoundExporter(self.ntds, output_dir)
-
-        bloodhound.write_bloodhound_json(self.domains)
-        bloodhound.write_bloodhound_json(self.users)
-        bloodhound.write_bloodhound_json(self.computers)
-        bloodhound.write_bloodhound_json(self.groups)
-        bloodhound.write_bloodhound_json(self.ous)
-        bloodhound.write_bloodhound_json(self.gpos)
-
 
 def extract_object_info(obj: Object) -> dict[str, Any]:
     """Extract generic information from an Object."""