Improve parsing of complex ACLs (#23)

Co-authored-by: Schamper <[email protected]>

dissect/ntfs/c_ntfs.py

@@ -400,6 +400,8 @@
     SYSTEM_MANDATORY_LABEL          = 0x11,
     SYSTEM_RESOURCE_ATTRIBUTE       = 0x12,
     SYSTEM_SCOPED_POLICY_ID         = 0x13,
+    SYSTEM_PROCESS_TRUST_LABEL      = 0x14,
+    SYSTEM_ACCESS_FILTER            = 0x15,
 };
 
 flag ACE_FLAGS : BYTE {
@@ -412,6 +414,15 @@
     FAILED_ACCESS_ACE_FLAG      = 0x80,
 };
 
+flag ACE_OBJECT_FLAGS : DWORD {
+    ACE_OBJECT_TYPE_PRESENT           = 0x01,
+    ACE_INHERITED_OBJECT_TYPE_PRESENT = 0x02,
+};
+
+enum COMPOUND_ACE_TYPE : USHORT {
+    COMPOUND_ACE_IMPERSONATION  = 0x01,
+};
+
 typedef struct _ACL {
     BYTE        AclRevision;
     BYTE        Sbz1;
@@ -549,6 +560,7 @@
 IO_REPARSE_TAG = c_ntfs.IO_REPARSE_TAG
 ACCESS_MASK = c_ntfs.ACCESS_MASK
 ACE_TYPE = c_ntfs.ACE_TYPE
+ACE_OBJECT_FLAGS = c_ntfs.ACE_OBJECT_FLAGS
 COLLATION = c_ntfs.COLLATION
 
 # Some useful magic numbers and constants

dissect/ntfs/secure.py

@@ -7,7 +7,7 @@
 from dissect.cstruct import Instance
 from dissect.util.sid import read_sid
 
-from dissect.ntfs.c_ntfs import ACE_TYPE, c_ntfs
+from dissect.ntfs.c_ntfs import ACE_OBJECT_FLAGS, ACE_TYPE, c_ntfs
 from dissect.ntfs.mft import MftRecord
 
 
@@ -158,23 +158,38 @@ def __init__(self, fh: BinaryIO):
         self.object_type = None
         self.inherited_object_type = None
         self.sid = None
+        self.compound_type = None
+        self.server_sid = None
 
         buf = io.BytesIO(self.data)
         if self.is_standard_ace:
             self.mask = c_ntfs.DWORD(buf)
             self.sid = read_sid(buf)
+        elif self.is_compound_ace:
+            self.mask = c_ntfs.DWORD(buf)
+            self.compound_type = c_ntfs.COMPOUND_ACE_TYPE(buf)
+            c_ntfs.USHORT(buf)
+            self.server_sid = read_sid(buf)
+            self.sid = read_sid(buf)
         elif self.is_object_ace:
             self.mask = c_ntfs.DWORD(buf)
             self.flags = c_ntfs.DWORD(buf)
-            self.object_type = UUID(bytes_le=buf.read(16))
-            self.inherited_object_type = UUID(bytes_le=buf.read(16))
+            if self.flags & ACE_OBJECT_FLAGS.ACE_OBJECT_TYPE_PRESENT:
+                self.object_type = UUID(bytes_le=buf.read(16))
+            if self.flags & ACE_OBJECT_FLAGS.ACE_INHERITED_OBJECT_TYPE_PRESENT:
+                self.inherited_object_type = UUID(bytes_le=buf.read(16))
             self.sid = read_sid(buf)
 
         self.application_data = buf.read() or None
 
     def __repr__(self) -> str:
         if self.is_standard_ace:
             return f"<{self.header.AceType.name} mask=0x{self.mask:x} sid={self.sid}>"
+        elif self.is_compound_ace:
+            return (
+                f"<{self.header.AceType.name} mask=0x{self.mask:x} type={self.compound_type.name}"
+                f" server_sid={self.server_sid} client_sid={self.sid}>"
+            )
         elif self.is_object_ace:
             return (
                 f"<{self.header.AceType.name} mask=0x{self.mask:x} flags={self.flags} object_type={self.object_type}"
@@ -196,16 +211,22 @@ def is_standard_ace(self) -> bool:
             ACE_TYPE.ACCESS_DENIED,
             ACE_TYPE.SYSTEM_AUDIT,
             ACE_TYPE.SYSTEM_ALARM,
-            ACE_TYPE.ACCESS_ALLOWED_COMPOUND,
             ACE_TYPE.ACCESS_ALLOWED_CALLBACK,
             ACE_TYPE.ACCESS_DENIED_CALLBACK,
             ACE_TYPE.SYSTEM_AUDIT_CALLBACK,
             ACE_TYPE.SYSTEM_ALARM_CALLBACK,
             ACE_TYPE.SYSTEM_MANDATORY_LABEL,
             ACE_TYPE.SYSTEM_RESOURCE_ATTRIBUTE,
             ACE_TYPE.SYSTEM_SCOPED_POLICY_ID,
+            ACE_TYPE.SYSTEM_PROCESS_TRUST_LABEL,
+            ACE_TYPE.SYSTEM_ACCESS_FILTER,
         )
 
+    @property
+    def is_compound_ace(self) -> bool:
+        """Return whether this ACE is a compound ACE."""
+        return self.header.AceType in (ACE_TYPE.ACCESS_ALLOWED_COMPOUND,)
+
     @property
     def is_object_ace(self) -> bool:
         """Return whether this ACE is an object ACE."""

tests/conftest.py

@@ -2,42 +2,48 @@
 import gzip
 import io
 import os
+from typing import BinaryIO, Iterator
 
 import pytest
 from dissect.util.stream import MappingStream
 
 
-def absolute_path(filename):
+def absolute_path(filename: str) -> str:
     return os.path.join(os.path.dirname(__file__), filename)
 
 
-def open_file_gz(name, mode="rb"):
+def open_file_gz(name: str, mode: str = "rb") -> Iterator[BinaryIO]:
     with gzip.GzipFile(absolute_path(name), mode) as f:
         yield f
 
 
 @pytest.fixture
-def ntfs_bin():
+def ntfs_bin() -> Iterator[BinaryIO]:
     yield from open_file_gz("data/ntfs.bin.gz")
 
 
 @pytest.fixture
-def mft_bin():
+def mft_bin() -> Iterator[BinaryIO]:
     yield from open_file_gz("data/mft.bin.gz")
 
 
 @pytest.fixture
-def sds_bin():
+def sds_bin() -> Iterator[BinaryIO]:
     yield from open_file_gz("data/sds.bin.gz")
 
 
 @pytest.fixture
-def boot_2m_bin():
+def sds_complex_bin() -> Iterator[BinaryIO]:
+    yield from open_file_gz("data/sds_complex.bin.gz")
+
+
+@pytest.fixture
+def boot_2m_bin() -> Iterator[BinaryIO]:
     yield from open_file_gz("data/boot_2m.bin.gz")
 
 
 @pytest.fixture
-def ntfs_fragmented_mft_fh():
+def ntfs_fragmented_mft_fh() -> Iterator[BinaryIO]:
     # Test data from https://github.com/msuhanov/ntfs-samples
     # This is from the file ntfs_extremely_fragmented_mft.raw which has, as the name implies, a heavily fragmented MFT
     # The entire file is way too large, so only take just enough data that we actually need to make dissect.ntfs happy

tests/test_attr.py

@@ -7,7 +7,7 @@
 from dissect.ntfs.exceptions import VolumeNotAvailableError
 
 
-def test_attributes():
+def test_attributes() -> None:
     # Single $STANDARD_INFORMATION attribute
     data = bytes.fromhex(
         "100000006000000000001800000000004800000018000000d2145d665666d801"
@@ -64,7 +64,7 @@ def test_attributes():
         attr.open()
 
 
-def test_reparse_point_moint_point():
+def test_reparse_point_moint_point() -> None:
     data = bytes.fromhex(
         "c00000005800000000000000000004004000000018000000030000a038000000"
         "00001a001c0012005c003f003f005c0043003a005c0054006100720067006500"
@@ -80,7 +80,7 @@ def test_reparse_point_moint_point():
     assert not attr.relative
 
 
-def test_reparse_point_symlink():
+def test_reparse_point_symlink() -> None:
     data = bytes.fromhex(
         "c000000058000000000000000000040040000000180000000c0000a038000000"
         "12001a00000012000000000043003a005c005400610072006700650074005c00"

tests/test_index.py

@@ -1,13 +1,14 @@
 import io
 import struct
+from typing import BinaryIO
 from unittest.mock import Mock
 
 from dissect.ntfs.c_ntfs import ATTRIBUTE_TYPE_CODE, c_ntfs
 from dissect.ntfs.index import IndexEntry, Match, _cmp_filename, _cmp_ulong
 from dissect.ntfs.ntfs import NTFS
 
 
-def mock_filename_entry(filename):
+def mock_filename_entry(filename: str) -> IndexEntry:
     attribute = c_ntfs._FILE_NAME(
         FileNameLength=len(filename),
         FileName=filename,
@@ -23,7 +24,7 @@ def mock_filename_entry(filename):
     return IndexEntry(mock_index, io.BytesIO(data), 0)
 
 
-def mock_ulong_entry(value):
+def mock_ulong_entry(value: int) -> IndexEntry:
     header = c_ntfs._INDEX_ENTRY(
         Length=len(c_ntfs._INDEX_ENTRY) + 4,
         KeyLength=4,
@@ -35,7 +36,7 @@ def mock_ulong_entry(value):
     return IndexEntry(mock_index, io.BytesIO(data), 0)
 
 
-def test_cmp_filename():
+def test_cmp_filename() -> None:
     entry = mock_filename_entry("bbbb")
 
     assert _cmp_filename(entry, "CCCC") == Match.Greater
@@ -48,15 +49,15 @@ def test_cmp_filename():
     assert _cmp_filename(entry, "CONFIG") == Match.Less
 
 
-def test_cmp_ulong():
+def test_cmp_ulong() -> None:
     entry = mock_ulong_entry(100)
 
     assert _cmp_ulong(entry, 99) == Match.Less
     assert _cmp_ulong(entry, 100) == Match.Equal
     assert _cmp_ulong(entry, 101) == Match.Greater
 
 
-def test_index_lookup(ntfs_bin):
+def test_index_lookup(ntfs_bin: BinaryIO) -> None:
     fs = NTFS(ntfs_bin)
 
     root = fs.mft.get("Large Directory")

tests/test_mft.py

@@ -1,4 +1,5 @@
 import io
+from typing import BinaryIO
 
 import pytest
 
@@ -8,7 +9,7 @@
 from dissect.ntfs.ntfs import NTFS
 
 
-def test_mft(mft_bin):
+def test_mft(mft_bin: BinaryIO) -> None:
     fs = NTFS(mft=mft_bin)
 
     assert fs.mft
@@ -18,7 +19,7 @@ def test_mft(mft_bin):
         assert fs.mft.get(FILE_NUMBER_MFT).open()
 
 
-def test_mft_record_get_no_mft(mft_bin):
+def test_mft_record_get_no_mft(mft_bin: BinaryIO) -> None:
     fs = NTFS(mft=mft_bin)
 
     root = fs.mft.root
@@ -28,7 +29,7 @@ def test_mft_record_get_no_mft(mft_bin):
         root.get("$MFT")
 
 
-def test_mft_record():
+def test_mft_record() -> None:
     # Single MFT record of the $MFT file itself.
     data = bytes.fromhex(
         "46494c453000030051511000000000000100010038000100a001000000040000"

tests/test_ntfs.py

@@ -1,12 +1,13 @@
 from io import BytesIO
+from typing import BinaryIO
 
 import pytest
 
 from dissect.ntfs.exceptions import FileNotFoundError, NotADirectoryError
 from dissect.ntfs.ntfs import NTFS
 
 
-def test_ntfs(ntfs_bin):
+def test_ntfs(ntfs_bin: BinaryIO) -> None:
     fs = NTFS(ntfs_bin)
 
     assert fs.sector_size == 512
@@ -49,12 +50,12 @@ def test_ntfs(ntfs_bin):
     assert fs.mft.get("Directory/File 1.txt").full_path() == "Directory\\File 1.txt"
 
 
-def test_ntfs_large_sector(boot_2m_bin):
+def test_ntfs_large_sector(boot_2m_bin: BinaryIO) -> None:
     fs = NTFS(boot=boot_2m_bin)
     assert fs.cluster_size == 0x200000
 
 
-def test_ntfs_64k_sector():
+def test_ntfs_64k_sector() -> None:
     boot_sector = """
     eb52904e5446532020202000028000000000000000f800003f00ff0000080400
     0000000080008000ffef3b060000000000c00000000000000100000000000000
@@ -78,6 +79,6 @@ def test_ntfs_64k_sector():
     assert fs.cluster_size == 0x10000
 
 
-def test_fragmented_mft(ntfs_fragmented_mft_fh):
+def test_fragmented_mft(ntfs_fragmented_mft_fh: BinaryIO) -> None:
     fs = NTFS(ntfs_fragmented_mft_fh)
     assert len(fs.mft.fh.runlist) == 238

tests/test_secure.py

@@ -1,10 +1,12 @@
+from typing import BinaryIO
+
 import pytest
 
 from dissect.ntfs.ntfs import NTFS
 from dissect.ntfs.secure import Secure
 
 
-def test_secure(ntfs_bin):
+def test_secure(ntfs_bin: BinaryIO) -> None:
     fs = NTFS(ntfs_bin)
 
     assert fs.secure
@@ -28,14 +30,33 @@ def test_secure(ntfs_bin):
     assert len(list(fs.secure.descriptors())) == 24
 
 
-def test_secure_file(sds_bin):
+def test_secure_file(sds_bin: BinaryIO) -> None:
     secure = Secure(sds=sds_bin)
 
     sd = secure.lookup(256)
     assert sd.owner == "S-1-5-18"
     assert sd.group == "S-1-5-32-544"
 
 
-def test_secure_fail():
+def test_secure_complex_acl(sds_complex_bin: BinaryIO) -> None:
+    secure = Secure(sds=sds_complex_bin)
+
+    sd = secure.lookup(259)
+    assert sd.owner == "S-1-5-21-3090333131-159632407-777084872-1001"
+    assert sd.group == "S-1-5-21-3090333131-159632407-777084872-513"
+    assert len(sd.sacl.ace) == 2
+    assert list(map(repr, sd.sacl.ace)) == [
+        "<SYSTEM_MANDATORY_LABEL mask=0x7 sid=S-1-16-4096>",
+        "<SYSTEM_ACCESS_FILTER mask=0x1200a9 sid=S-1-1-0>",
+    ]
+    assert len(sd.dacl.ace) == 3
+    assert list(map(repr, sd.dacl.ace)) == [
+        "<ACCESS_ALLOWED_COMPOUND mask=0x1f01ff type=COMPOUND_ACE_IMPERSONATION server_sid=S-1-5-11 client_sid=S-1-5-32-545>",  # noqa: E501
+        "<ACCESS_ALLOWED_OBJECT mask=0x1f01ff flags=0 object_type=None inherited_object_type=None sid=S-1-1-0>",
+        "<ACCESS_ALLOWED_CALLBACK_OBJECT mask=0x1f01ff flags=3 object_type=01234567-89ab-cdef-1111-111111111111 inherited_object_type=22222222-2222-2222-0123-456789abcdef sid=S-1-5-32-545>",  # noqa: E501
+    ]
+
+
+def test_secure_fail() -> None:
     with pytest.raises(ValueError):
         Secure()

tests/test_usnjrnl.py

@@ -3,7 +3,7 @@
 from dissect.ntfs.usnjrnl import UsnRecord
 
 
-def test_usnjrnl_record_v4():
+def test_usnjrnl_record_v4() -> None:
     data = bytes.fromhex(
         "5000000004000000c1000000000001000000000000000000bf00000000000100"
         "0000000000000000d00201000000000003810080000000000000000001001000"
@@ -15,7 +15,7 @@ def test_usnjrnl_record_v4():
     assert record.extents[0].Length == 0x284000
 
 
-def test_usnjrnl_record_v2():
+def test_usnjrnl_record_v2() -> None:
     data = bytes.fromhex(
         "5800000002000000c100000000000100bf000000000001002003010000000000"
         "6252641a86a4d7010381008000000000000000002000000018003c0069007300"

tests/test_util.py

@@ -3,7 +3,7 @@
 from dissect.ntfs.util import AttributeMap, apply_fixup
 
 
-def test_fixup():
+def test_fixup() -> None:
     buf = bytearray(
         b"FILE\x30\x00"
         + (b"\x00" * 42)
@@ -39,7 +39,7 @@ def test_fixup():
     assert fixed[2046:2048] == b"\xFC\x00"
 
 
-def test_attribute_map():
+def test_attribute_map() -> None:
     attr_map = AttributeMap()
     assert len(attr_map) == 0
     assert attr_map.STANDARD_INFORMATION == []