Improve PEK and DN

Author: Schamper

dissect/database/ese/ntds/database.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from functools import lru_cache
+from functools import cached_property, lru_cache
 from io import BytesIO
 from typing import TYPE_CHECKING, BinaryIO
 
@@ -10,12 +10,14 @@
 from dissect.database.ese.ntds.query import Query
 from dissect.database.ese.ntds.schema import Schema
 from dissect.database.ese.ntds.sd import ACL, SecurityDescriptor
-from dissect.database.ese.ntds.util import SearchFlags, encode_value
+from dissect.database.ese.ntds.util import DN, SearchFlags, encode_value
 
 if TYPE_CHECKING:
     from collections.abc import Iterator
 
     from dissect.database.ese.index import Index
+    from dissect.database.ese.ntds.objects import DomainDNS, Top
+    from dissect.database.ese.ntds.pek import PEK
 
 
 class Database:
@@ -48,13 +50,13 @@ def __init__(self, db: Database):
         self.get = lru_cache(4096)(self.get)
         self._make_dn = lru_cache(4096)(self._make_dn)
 
-    def root(self) -> Object:
+    def root(self) -> Top:
         """Return the top-level object in the NTDS database."""
         if (root := next(self.children_of(0), None)) is None:
             raise ValueError("No root object found")
         return root
 
-    def root_domain(self) -> Object:
+    def root_domain(self) -> DomainDNS:
         """Return the root domain object in the NTDS database."""
         obj = self.root()
         while True:
@@ -72,6 +74,11 @@ def root_domain(self) -> Object:
 
         raise ValueError("No root domain object found")
 
+    @cached_property
+    def pek(self) -> PEK:
+        """Return the PEK associated with the root domain."""
+        return self.root_domain().pek
+
     def walk(self) -> Iterator[Object]:
         """Walk through all objects in the NTDS database."""
         stack = [self.root()]
@@ -161,7 +168,7 @@ def children_of(self, dnt: int) -> Iterator[Object]:
             yield Object.from_record(self.db, record)
             record = cursor.next()
 
-    def _make_dn(self, dnt: int) -> str:
+    def _make_dn(self, dnt: int) -> DN:
         """Construct Distinguished Name (DN) from a Directory Number Tag (DNT) value.
 
         This method walks up the parent hierarchy to build the full DN path.
@@ -180,7 +187,9 @@ def _make_dn(self, dnt: int) -> str:
             return ""
 
         parent_dn = self._make_dn(obj.pdnt)
-        return f"{rdn_key}={rdn_value}".upper() + (f",{parent_dn}" if parent_dn else "")
+        dn = f"{rdn_key}={rdn_value}".upper() + (f",{parent_dn}" if parent_dn else "")
+
+        return DN(dn, obj, parent_dn if parent_dn else None)
 
     def _get_index(self, attribute: str) -> Index:
         """Get the index for a given attribute name.

dissect/database/ese/ntds/ntds.py

@@ -39,10 +39,10 @@ def root_domain(self) -> DomainDNS:
         """Return the root domain object of the Active Directory."""
         return self.db.data.root_domain()
 
-    @cached_property
+    @property
     def pek(self) -> PEK:
         """Return the PEK associated with the root domain."""
-        return self.root_domain().pek
+        return self.db.data.pek
 
     def walk(self) -> Iterator[Object]:
         """Walk through all objects in the NTDS database."""

dissect/database/ese/ntds/objects/object.py

@@ -3,7 +3,7 @@
 from functools import cached_property
 from typing import TYPE_CHECKING, Any, ClassVar
 
-from dissect.database.ese.ntds.util import InstanceType, SystemFlags, decode_value
+from dissect.database.ese.ntds.util import DN, InstanceType, SystemFlags, decode_value
 
 if TYPE_CHECKING:
     from collections.abc import Iterator
@@ -181,12 +181,11 @@ def is_head_of_naming_context(self) -> bool:
         return self.instance_type is not None and bool(self.instance_type & InstanceType.HeadOfNamingContext)
 
     @property
-    def distinguished_name(self) -> str | None:
+    def distinguished_name(self) -> DN | None:
         """Return the fully qualified Distinguished Name (DN) for this object."""
-        # return self.db.data._make_dn(self.dnt)
         return self.get("distinguishedName")
 
-    DN = distinguished_name
+    dn = distinguished_name
 
     @cached_property
     def sd(self) -> SecurityDescriptor | None:

dissect/database/ese/ntds/util.py

@@ -12,6 +12,7 @@
     from collections.abc import Callable
 
     from dissect.database.ese.ntds.database import Database
+    from dissect.database.ese.ntds.objects import Object
 
 
 # https://learn.microsoft.com/en-us/openspecs/windows_protocols/ms-adts/7cda533e-d7a4-4aec-a517-91d02ff4a1aa
@@ -144,6 +145,21 @@ class SearchFlags(IntFlag):
     Confidential = 0x00000080
 
 
+def _pek_decrypt(db: Database, value: bytes) -> bytes:
+    """Decrypt a PEK-encrypted blob using the database's PEK, if it's unlocked.
+
+    Args:
+        value: The PEK-encrypted data blob.
+
+    Returns:
+        The decrypted data blob.
+    """
+    if not db.data.pek.unlocked:
+        return value
+
+    return db.data.pek.decrypt(value)
+
+
 ATTRIBUTE_ENCODE_DECODE_MAP: dict[
     str, tuple[Callable[[Database, Any], Any] | None, Callable[[Database, Any], Any] | None]
 ] = {
@@ -162,8 +178,23 @@ class SearchFlags(IntFlag):
         None,
         lambda db, value: float("inf") if int(value) == ((1 << 63) - 1) else wintimestamp(int(value)),
     ),
+    # Protected attributes
+    "unicodePwd": (None, _pek_decrypt),
+    "dBCSPwd": (None, _pek_decrypt),
+    "ntPwdHistory": (None, _pek_decrypt),
+    "lmPwdHistory": (None, _pek_decrypt),
+    "supplementalCredentials": (None, _pek_decrypt),
+    "currentValue": (None, _pek_decrypt),
+    "priorValue": (None, _pek_decrypt),
+    "initialAuthIncoming": (None, _pek_decrypt),
+    "initialAuthOutgoing": (None, _pek_decrypt),
+    "trustAuthIncoming": (None, _pek_decrypt),
+    "trustAuthOutgoing": (None, _pek_decrypt),
+    "msDS-ExecuteScriptPassword": (None, _pek_decrypt),
 }
 
+# TODO add for protected attributes
+
 
 def _ldapDisplayName_to_DNT(db: Database, value: str) -> int | str:
     """Convert an LDAP display name to its corresponding DNT value.
@@ -179,8 +210,10 @@ def _ldapDisplayName_to_DNT(db: Database, value: str) -> int | str:
     return value
 
 
-def _DNT_to_ldapDisplayName(db: Database, value: int) -> str | int:
-    """Convert a DNT value to its corresponding LDAP display name.
+def _DNT_to_ldapDisplayName(db: Database, value: int) -> str | DN | int:
+    """Convert a DNT value to its corresponding LDAP display name or distinguished name.
+
+    For attributes and classes, the LDAP display name is returned. For objects, the distinguished name is returned.
 
     Args:
         value: The Directory Number Tag to look up.
@@ -197,6 +230,18 @@ def _DNT_to_ldapDisplayName(db: Database, value: int) -> str | int:
         return value
 
 
+class DN(str):
+    """A distinguished name (DN) string wrapper. Presents the DN as a string but also retains the underlying object."""
+
+    __slots__ = ("object", "parent")
+
+    def __new__(cls, value: str, object: Object, parent: DN | None = None):
+        instance = super().__new__(cls, value)
+        instance.object = object
+        instance.parent = parent
+        return instance
+
+
 def _oid_to_attrtyp(db: Database, value: str) -> int | str:
     """Convert OID string or LDAP display name to ATTRTYP value.
 

pyproject.toml

@@ -36,6 +36,9 @@ documentation = "https://docs.dissect.tools/en/latest/projects/dissect.database"
 repository = "https://github.com/fox-it/dissect.database"
 
 [project.optional-dependencies]
+full = [
+    "pycryptodome"
+]
 dev = [
     "dissect.cstruct>=4.0.dev,<5.0.dev",
     "dissect.util>=3.24.dev,<4.0.dev",

tests/ese/ntds/test_pek.py

@@ -9,14 +9,22 @@
 def test_pek(goad: NTDS) -> None:
     """Test PEK unlocking and decryption."""
     syskey = bytes.fromhex("079f95655b66f16deb28aa1ab3a81eb0")
-    goad.pek.unlock(syskey)
-    assert goad.pek.unlocked
 
     user = next(goad.users(), None)
     assert user is not None
-    assert user.unicodePwd == bytes.fromhex(
+
+    encrypted = user.unicodePwd
+    # Verify encrypted value
+    assert encrypted == bytes.fromhex(
         "130000000000000029fbdaafb52bf724a51052f668152ac5100000006d06616d95c026064fff245bd256f3d4990f7bffb546f76de566723da4855227"
     )
-    assert goad.pek.decrypt(user.unicodePwd) == bytes.fromhex(
+
+    goad.pek.unlock(syskey)
+    assert goad.pek.unlocked
+
+    # Test decryption of the user's password
+    assert goad.pek.decrypt(encrypted) == bytes.fromhex(
         "06bb564317712dc60761a32914e4048c10101010101010101010101010101010"
     )
+    # Should work transparently now too
+    assert user.unicodePwd == bytes.fromhex("06bb564317712dc60761a32914e4048c10101010101010101010101010101010")