emrg: a size or an offset in the Win32 ABI is measured from its struct

emrg/sandbox/win32/abi.py

@@ -95,8 +95,35 @@
 SID_MAX_SUB_AUTHORITIES = 15
 #: ACE flag marking inherited entries.
 INHERITED_ACE = 0x10
-#: Maximum SID allocation size in bytes.
-SECURITY_MAX_SID_SIZE = 68
+class Sid(ctypes.Structure):
+    """``SID`` — the security identifier, sized by its own declaration.
+
+    Declared here, beside the constant that describes it, so that constant can be
+    *measured*: a two-byte header, a six-byte identifier authority and the
+    sub-authority array.  The struct holds no pointer, so its layout is the same
+    on a 32-bit and a 64-bit interpreter — which is exactly why a literal ``68``
+    reads as correct and stays correct by accident, with nothing saying so.
+
+    The offsets the ACL walk reads are taken from here rather than written down:
+    :attr:`SubAuthority` starts the variable-length tail an inline ACE SID is
+    read from, and :attr:`SubAuthorityCount` is the byte that sizes it.
+    """
+
+    _fields_ = [
+        ("Revision", ctypes.c_ubyte),
+        ("SubAuthorityCount", ctypes.c_ubyte),
+        ("IdentifierAuthority", ctypes.c_ubyte * 6),
+        ("SubAuthority", ctypes.c_uint32 * SID_MAX_SUB_AUTHORITIES),
+    ]
+
+
+#: The most bytes a ``SID`` may occupy, and the buffer size a SID is copied into.
+#:
+#: The SDK defines this as ``sizeof(SID)`` with the sub-authority array at its
+#: maximum, so it is taken from the struct above and never written down.  A
+#: literal here would be a number a reader cannot check: the declaration is what
+#: says how large a SID is, and it costs nothing to ask it (issue #2517).
+SECURITY_MAX_SID_SIZE = ctypes.sizeof(Sid)
 
 # --- registry keys (issue #2492 Stage 2) -----------------------------------
 

emrg/sandbox/win32/acl.py

@@ -26,6 +26,8 @@
 from emrg.sandbox.win32 import abi
 from emrg.sandbox.win32.ffi import (
     ACLStructure,
+    AccessAllowedAce,
+    AceHeader,
     ExplicitAccessW,
     Overlapped,
     Win32Bindings,
@@ -303,24 +305,29 @@ def has_exact_grant(
     header = ACLStructure.from_address(old_acl)
     acl_size = int(header.AclSize)
     ace_count = int(header.AceCount)
-    if acl_size < 8 or acl_size > 1_048_576:
+    header_size = ctypes.sizeof(ACLStructure)
+    if acl_size < header_size or acl_size > 1_048_576:
         return False  # implausible: fall back to the merge path
     target = read_sid_bytes(api, sid_ptr)
-    offset = 8  # the first ACE follows the 8-byte ACL header
+    offset = header_size  # the first ACE follows the ACL header
     for _ in range(ace_count):
-        ace_type = ctypes.c_ubyte.from_address(old_acl + offset).value
-        ace_flags = ctypes.c_ubyte.from_address(old_acl + offset + 1).value
-        ace_size = int(ctypes.c_uint16.from_address(old_acl + offset + 2).value)
-        if ace_size < 8 or offset + ace_size > acl_size:
+        ace = old_acl + offset
+        ace_type = ctypes.c_ubyte.from_address(ace).value
+        ace_flags = ctypes.c_ubyte.from_address(ace + AceHeader.AceFlags.offset).value
+        ace_size = int(ctypes.c_uint16.from_address(ace + AceHeader.AceSize.offset).value)
+        if ace_size < AccessAllowedAce.SidStart.offset or offset + ace_size > acl_size:
             return False  # implausible: fall back to the merge path
-        ace_mask = int(ctypes.c_uint32.from_address(old_acl + offset + 4).value)
+        ace_mask = int(ctypes.c_uint32.from_address(ace + AccessAllowedAce.Mask.offset).value)
         exact = (
             ace_type == abi.ACCESS_ALLOWED_ACE_TYPE
             and ace_flags == abi.SUB_CONTAINERS_AND_OBJECTS_INHERIT
             and ace_mask == mask
         )
         if exact:
-            inline = _inline_sid_bytes(old_acl + offset + 8, ace_size - 8)
+            inline = _inline_sid_bytes(
+                ace + AccessAllowedAce.SidStart.offset,
+                ace_size - AccessAllowedAce.SidStart.offset,
+            )
             if inline is not None and inline == target:
                 return True
         offset += ace_size
@@ -334,12 +341,14 @@ def _inline_sid_bytes(address: int, available: int) -> bytes | None:
     :param available: how many bytes of the ACE remain at that address.
     :returns: the SID's bytes, or ``None`` when the SID does not fit.
     """
-    if available < 8:
+    if available < abi.Sid.SubAuthority.offset:
         return None
-    sub_authority_count = int(ctypes.c_ubyte.from_address(address + 1).value)
+    sub_authority_count = int(
+        ctypes.c_ubyte.from_address(address + abi.Sid.SubAuthorityCount.offset).value
+    )
     if sub_authority_count > abi.SID_MAX_SUB_AUTHORITIES:
         return None
-    length = 8 + 4 * sub_authority_count
+    length = abi.Sid.SubAuthority.offset + ctypes.sizeof(ctypes.c_uint32) * sub_authority_count
     if length > available:
         return None
     return ctypes.string_at(address, length)

emrg/sandbox/win32/ffi.py

@@ -106,6 +106,42 @@ class ACLStructure(ctypes.Structure):
     ]
 
 
+class AceHeader(ctypes.Structure):
+    """``ACE_HEADER`` — the three fields every ACE begins with.
+
+    The ACL walk reads these three by offset, and a number read byte-wise is a
+    number nobody can check against anything.  Declared, the walk names the field
+    it is reading and the layout is the ruler (issue #2517).
+
+    Fixed-width field types, deliberately: ``ctypes.wintypes.DWORD`` is
+    ``c_ulong``, which is four bytes on Windows and **eight** on a 64-bit Linux or
+    macOS interpreter, so a ``DWORD`` in a Win32 declaration describes a different
+    struct on the very legs that run the guard.  Measured here: the same bytes
+    put ``Mask`` at offset 4 with these types and at 8 with ``wintypes.DWORD``.
+    """
+
+    _fields_ = [
+        ("AceType", ctypes.c_ubyte),
+        ("AceFlags", ctypes.c_ubyte),
+        ("AceSize", ctypes.c_uint16),
+    ]
+
+
+class AccessAllowedAce(ctypes.Structure):
+    """``ACCESS_ALLOWED_ACE`` — the header, the access mask, and the inline SID.
+
+    ``SidStart`` is the first four bytes of the SID the ACE carries **inside
+    itself**, not a pointer to one, so its offset is where a bounded reader starts
+    and the ACE's own size is what bounds it.
+    """
+
+    _fields_ = [
+        ("Header", AceHeader),
+        ("Mask", ctypes.c_uint32),
+        ("SidStart", ctypes.c_uint32),
+    ]
+
+
 class StartupInfoW(ctypes.Structure):
     """``STARTUPINFOW``."""
 

tests/test_bash_v2_win32.py

@@ -15,6 +15,7 @@
 re-invented.
 """
 
+import ast
 import ctypes
 import gc
 import json
@@ -23,6 +24,7 @@
 import subprocess
 import sys
 import tempfile
+from pathlib import Path
 
 import pytest
 
@@ -149,6 +151,115 @@ def test_the_descriptor_buffer_holds_the_whole_struct_the_api_fills():
     ]
 
 
+def _abi_measurement_faults(tree):
+    """The places a size or an offset is written down instead of measured (issue #2517).
+
+    :param tree: a parsed module.
+    :returns: one line per fault, empty when the module measures what it needs.
+    """
+    faults = []
+    for node in tree.body:
+        if isinstance(node, ast.Assign):
+            targets, value = node.targets, node.value
+        elif isinstance(node, ast.AnnAssign):
+            targets, value = [node.target], node.value or ast.Constant(None)
+        else:
+            continue
+        if not isinstance(value, ast.Constant) or type(value.value) is not int:
+            continue
+        for target in targets:
+            name = getattr(target, "id", "")
+            if name.upper().endswith(("_SIZE", "_LENGTH")):
+                faults.append(f"{name} = {value.value} written down")
+    for node in ast.walk(tree):
+        if not isinstance(node, ast.Call):
+            continue
+        func = node.func
+        called = getattr(func, "attr", getattr(func, "id", ""))
+        if called not in {"from_address", "string_at"}:
+            continue
+        for arg in node.args:
+            for inner in ast.walk(arg):
+                if isinstance(inner, ast.Constant) and type(inner.value) is int:
+                    faults.append(f"{called}(…) offset written as {inner.value}")
+    return faults
+
+
+def test_a_size_or_an_offset_in_the_win32_abi_is_measured_from_its_struct():
+    """A size or an offset beside a Win32 struct is measured from the declaration.
+
+    Issue #2517, the class #2509 was one instance of. ``SECURITY_MAX_SID_SIZE``
+    was written down as ``68``: correct, and correct by the accident of the
+    struct's shape, with nothing saying so — and the same number was copied again
+    into two test buffers, so a changed constant would have left both sites
+    passing against a number they no longer matched. ``acl.py`` walked an ACE by
+    reading at ``+1``, ``+2``, ``+4`` and ``+8``, which are the
+    ``ACE_HEADER``/``ACCESS_ALLOWED_ACE`` layout and were checkable against
+    nothing.
+
+    This is an AST question, not a Windows one — a declaration lays out on any
+    interpreter — which is why the guard lives in the file's cross-platform half
+    and catches on the macOS and Linux legs, the legs that stay green.
+    """
+    package = Path(__file__).resolve().parent.parent / "emrg" / "sandbox" / "win32"
+    modules = sorted(package.glob("*.py"))
+    assert len(modules) >= 5, f"the ABI package was not read: {package}"
+
+    faults = [
+        f"{path.name}: {fault}"
+        for path in modules
+        for fault in _abi_measurement_faults(ast.parse(path.read_text(encoding="utf-8")))
+    ]
+    assert not faults, "sizes and offsets come from the declaration:\n" + "\n".join(faults)
+
+    # The matcher discriminates, proven on snippets rather than on today's tree:
+    # the constant #2509 shipped, and #2517's hand-computed offset.
+    assert _abi_measurement_faults(ast.parse("SECURITY_DESCRIPTOR_MIN_LENGTH = 20\n"))
+    assert _abi_measurement_faults(ast.parse("ace = ctypes.c_ubyte.from_address(base + 1)\n"))
+    # …and it stays quiet on the shape it asks for.
+    assert not _abi_measurement_faults(
+        ast.parse("SECURITY_DESCRIPTOR_MIN_LENGTH = ctypes.sizeof(SecurityDescriptor)\n")
+    )
+    assert not _abi_measurement_faults(
+        ast.parse("ace = ctypes.c_ubyte.from_address(base + AceHeader.AceFlags.offset)\n")
+    )
+
+
+def test_the_declared_win32_structs_measure_the_offsets_the_acl_walk_reads():
+    """``Sid``, ``AceHeader`` and ``AccessAllowedAce`` are the ruler the walk reads by.
+
+    Both directions, because either alone is satisfiable by a wrong declaration:
+    the value must equal the struct's own size, and the struct must have the shape
+    the platform defines. A truncated declaration shrinks both sides together, so
+    the SDK's numbers are pinned as well.
+    """
+    sid = abi.Sid
+    assert [name for name, _ in sid._fields_] == [
+        "Revision",
+        "SubAuthorityCount",
+        "IdentifierAuthority",
+        "SubAuthority",
+    ]
+    # The sub-authority array is as long as the constant allows, and its bytes are
+    # read through the field itself: a _ctypes field is a descriptor, not a list.
+    assert sid.SubAuthority.size == abi.SID_MAX_SUB_AUTHORITIES * ctypes.sizeof(ctypes.c_uint32)
+    # A SID holds no pointer, so this layout is the same at either interpreter
+    # width — and the SDK's own answer for it is 68.
+    assert ctypes.sizeof(sid) == abi.SECURITY_MAX_SID_SIZE
+    assert abi.SECURITY_MAX_SID_SIZE == 68
+    assert sid.SubAuthorityCount.offset == 1
+    assert sid.SubAuthority.offset == 8
+
+    assert ctypes.sizeof(ffi.ACLStructure) == 8
+    assert (
+        ffi.AceHeader.AceType.offset,
+        ffi.AceHeader.AceFlags.offset,
+        ffi.AceHeader.AceSize.offset,
+    ) == (0, 1, 2)
+    assert ffi.AccessAllowedAce.Mask.offset == 4
+    assert ffi.AccessAllowedAce.SidStart.offset == 8
+
+
 # ── capability SIDs ───────────────────────────────────────────────────────
 
 
@@ -1683,7 +1794,7 @@ def test_grant_caller_grants_a_sid_that_its_own_buffer_still_owns(tmp_path, monk
     def fake_caller_ace(bindings, path, sid_ptr):
         for _ in range(256):
             ctypes.create_string_buffer(_FAKE_TOKEN_USER_SIZE)  # this block's own class
-            ctypes.create_string_buffer(68)  # SECURITY_MAX_SID_SIZE
+            ctypes.create_string_buffer(abi.SECURITY_MAX_SID_SIZE)  # the SID buffer's own class
         gc.collect()
         granted.append((path, sid_ptr, ctypes.string_at(sid_ptr, 8)))
 
@@ -2152,7 +2263,7 @@ def test_the_sid_memory_survives_the_allocations_that_used_to_land_on_it():
     before = ctypes.string_at(handle.address, 8)
     for _ in range(256):
         ctypes.create_string_buffer(16)  # the size class of a SID_AND_ATTRIBUTES array
-        ctypes.create_string_buffer(68)  # SECURITY_MAX_SID_SIZE, the well-known SID's own
+        ctypes.create_string_buffer(abi.SECURITY_MAX_SID_SIZE)  # the well-known SID's own class
     gc.collect()
     assert ctypes.string_at(handle.address, 8) == before