arti
/
deploy-win10-from-linux
1
0
Fork 0
Code Issues 1 Pull Requests Packages Projects Releases Wiki Activity
deploy-win10-from-linux/ntfs_acl.py

469 lines
17 KiB
Python
Raw Permalink Blame History

import sys, os
from io import BytesIO
from construct import *
from pathlib import Path # overrides same name from construct
import ctypes
from textwrap import dedent
def dump_sd(path):
return getxattr(path, "system.ntfs_acl")
def write_sd(path, sd):
if isinstance(sd, SecurityDescriptor): sd = sd.build()
elif isinstance(sd, Container): sd = SecurityDescriptorStruct.build(sd)
os.setxattr(path, "system.ntfs_acl", sd)
_cdll = ctypes.CDLL(None)
#_cdll.libcnotify_verbose_set(1)
def getxattr(path, name):
# We have to use ctypes instead of os.getxattr, because os.getxattr uses a 128B buffer,
# which is too small (ironically, it retries with larger buffer if it gets ERANGE, but
# ntfs-3g returns EIO instead of ERANGE when buffer is too small). It cannot be forced
# to use larger buffer by default.
if isinstance(name, str):
name = name.encode('ascii')
if isinstance(path, os.PathLike):
path = str(path)
if isinstance(path, str):
path = path.encode(sys.getfilesystemencoding())
global _cdll
size = _cdll.getxattr(path, name, None, 0)
buf = ctypes.create_string_buffer(b'', size)
_cdll.getxattr(path, name, buf, size)
return buf.raw
#class ClassAdapter(Adapter):
# def __init__(self, cls, *a, **kw):
# self.adapt_cls = cls
# super().__init__(*a, **kw)
#
# def _decode(self, obj, context, path):
# return self.adapt_cls.from_construct(obj)
#
# def _encode(self, obj, context, path):
# if not isinstance(obj, self.adapt_cls): raise TypeError
# return obj.to_construct()
#
#class ACLAdapter(Adapter):
# def _decode(self, obj, context, path):
# assert len(obj.items) == obj.cnt
# return obj.items
#
# def _encode(self, obj, context, path):
# return dict(revision=2, cnt=len(obj), items=obj)
def _getpath(obj, path):
if isinstance(path, str): path = path.split('.')
for c in path:
obj = obj[c]
return obj
def _setpath(obj, path, val):
if isinstance(path, str): path = path.split('.')
for c in path[:-1]:
obj = obj[c]
obj[path[-1]] = val
class WithSize(Subconstruct):
def __init__(self, size_path, subcon, pattern=b"\x00"):
if not isinstance(pattern, bytes) or len(pattern) != 1:
raise PaddingError("pattern expected to be bytes of length 1")
super().__init__(subcon)
self.size_path = size_path
self.pattern = pattern
def _parse(self, stream, context, path):
position1 = stream.tell()
obj = self.subcon._parsereport(stream, context, path)
#print("obj:", obj)
position2 = stream.tell()
actual_length = (position2 - position1)
formal_length = _getpath(obj, self.size_path)
if formal_length < 0:
raise PaddingError("length cannot be negative", path=path)
pad = formal_length - actual_length
#print("Path:", path)
#print("Subcon:", self.subcon, self is ACEStruct, self is ACLStruct)
#print("Formal length:", formal_length)
#print("Actual length:", actual_length, position1, "..", position2)
if pad < 0:
raise PaddingError("subcon parsed %d bytes but was allowed only %d" % (actual_length, formal_length), path=path)
stream_read(stream, pad, path)
return obj
def _build(self, obj, stream, context, path):
tmpstream = BytesIO()
self.subcon._build(obj, tmpstream, context, path)
size = len(tmpstream.getvalue())
_setpath(obj, self.size_path, size)
#print(size, obj)
return self.subcon._build(obj, stream, context, path)
def _sizeof(self, context, path):
return self.subcon._sizeof(context, path)
# https://github.com/tuxera/ntfs-3g/blob/a4a837025b6ac2b0c44c93e34e22535fe9e95b27/ntfsprogs/ntfssecaudit.c#L1229
SDHeaderStruct = Struct(
"revision" / Const(1, Byte),
Padding(1),
"flags" / FlagsEnum(Int16ul,
# https://docs.microsoft.com/en-us/windows/win32/secauthz/security-descriptor-control
SE_OWNER_DEFAULTED = 0x0001,
SE_GROUP_DEFAULTED = 0x0002,
SE_DACL_PRESENT = 0x0004,
SE_DACL_DEFAULTED = 0x0008,
SE_SACL_DEFAULTED = 0x0008,
SE_SACL_PRESENT = 0x0010,
SE_DACL_AUTO_INHERIT_REQ = 0x0100,
SE_SACL_AUTO_INHERIT_REQ = 0x0200,
SE_DACL_AUTO_INHERITED = 0x0400,
SE_SACL_AUTO_INHERITED = 0x0800,
SE_DACL_PROTECTED = 0x1000,
SE_SACL_PROTECTED = 0x2000,
SE_RM_CONTROL_VALID = 0x4000,
SE_SELF_RELATIVE = 0x8000,
),
"owner_sid_offset" / Int32ul,
"group_sid_offset" / Int32ul,
"sacl_offset" / Int32ul,
"dacl_offset" / Int32ul,
)
# https://docs.microsoft.com/en-us/windows/win32/api/winnt/ns-winnt-sid
# https://docs.microsoft.com/en-us/windows/win32/secauthz/sid-components
# https://github.com/tuxera/ntfs-3g/blob/a4a837025b6ac2b0c44c93e34e22535fe9e95b27/ntfsprogs/ntfssecaudit.c#L957
SIDStruct = Struct(
"revision" / Const(1, Byte),
"cnt" / Byte,
"auth" / BytesInteger(6), # strangely, this is big endian, unlike other fields
"subauth" / Int32ul[this.cnt],
)
# https://docs.microsoft.com/en-us/windows/win32/api/winnt/ns-winnt-acl
# https://github.com/tuxera/ntfs-3g/blob/a4a837025b6ac2b0c44c93e34e22535fe9e95b27/ntfsprogs/ntfssecaudit.c#L1420
ACLHeaderStruct = Struct(
"revision" / Byte, # Const(2, Byte),
Padding(1),
"size" / Int16ul,
"cnt" / Int16ul,
Padding(2),
)
# https://docs.microsoft.com/en-us/windows/win32/secauthz/ace
# https://docs.microsoft.com/en-us/windows/win32/api/winnt/ns-winnt-ace_header
ACCESS_MAX_MS_V2_ACE_TYPE = 0x3
ACCESS_MAX_MS_V3_ACE_TYPE = 0x4
ACCESS_MAX_MS_V4_ACE_TYPE = 0x8
ACCESS_MAX_MS_V5_ACE_TYPE = 0x14
ACEHeaderStruct = Struct(
"type" / Enum(Byte,
# from wine's winnt.h
ACCESS_ALLOWED_ACE_TYPE = 0x0,
ACCESS_DENIED_ACE_TYPE = 0x1,
SYSTEM_AUDIT_ACE_TYPE = 0x2,
SYSTEM_ALARM_ACE_TYPE = 0x3,
ACCESS_ALLOWED_COMPOUND_ACE_TYPE = 0x4,
ACCESS_ALLOWED_OBJECT_ACE_TYPE = 0x5,
ACCESS_DENIED_OBJECT_ACE_TYPE = 0x6,
ACCESS_AUDIT_OBJECT_ACE_TYPE = 0x7,
ACCESS_ALARM_OBJECT_ACE_TYPE = 0x8,
ACCESS_ALLOWED_CALLBACK_ACE_TYPE = 0x9,
ACCESS_DENIED_CALLBACK_ACE_TYPE = 0xa,
ACCESS_ALLOWED_CALLBACK_OBJECT_ACE_TYPE = 0xb,
ACCESS_DENIED_CALLBACK_OBJECT_ACE_TYPE = 0xc,
SYSTEM_AUDIT_CALLBACK_ACE_TYPE = 0xd,
SYSTEM_ALARM_CALLBACK_ACE_TYPE = 0xe,
SYSTEM_AUDIT_CALLBACK_OBJECT_ACE_TYPE = 0xf,
SYSTEM_ALARM_CALLBACK_OBJECT_ACE_TYPE = 0x10,
SYSTEM_MANDATORY_LABEL_ACE_TYPE = 0x11,
SYSTEM_RESOURCE_ATTRIBUTE_ACE_TYPE = 0x12,
SYSTEM_SCOPED_POLICY_ID_ACE_TYPE = 0x13,
SYSTEM_PROCESS_TRUST_LABEL_ACE_TYPE = 0x14,
),
"flags" / FlagsEnum(Byte,
# from wine's winnt.h
# inheritance AceFlags
OBJECT_INHERIT_ACE = 0x01,
CONTAINER_INHERIT_ACE = 0x02,
NO_PROPAGATE_INHERIT_ACE = 0x04,
INHERIT_ONLY_ACE = 0x08,
INHERITED_ACE = 0x10,
# AceFlags mask for what events we (should) audit
SUCCESSFUL_ACCESS_ACE_FLAG = 0x40,
FAILED_ACCESS_ACE_FLAG = 0x80,
),
"size" / Int16ul,
)
# https://docs.microsoft.com/en-us/windows/win32/api/winnt/ns-winnt-access_allowed_ace
AllowDenyBodyStruct = Struct(
# https://docs.microsoft.com/en-us/windows/win32/secauthz/access-mask
"access_mask" / Int32ul,
"sid" / SIDStruct,
)
ACE_BODY_STRUCTS = {
'ACCESS_ALLOWED_ACE_TYPE': AllowDenyBodyStruct,
'ACCESS_DENIED_ACE_TYPE': AllowDenyBodyStruct,
# ...
}
ACEStruct = WithSize('header.size', Struct(
"header" / ACEHeaderStruct,
"body" / Switch(
this.header.type,
ACE_BODY_STRUCTS,
),
#Padding(lambda this: this.header.size - this._subcons.header.sizeof() - this._subcons.body.sizeof()),
))
ACLStruct = WithSize('header.size', Struct(
"header" / ACLHeaderStruct,
"items" / ACEStruct[this.header.cnt],
#Padding(lambda this: this.header.size - this._subcons.header.sizeof() - this._subcons.items.sizeof()),
))
SecurityDescriptorStruct = Struct(
"header" / SDHeaderStruct,
"owner_sid" / Pointer(this.header.owner_sid_offset, SIDStruct),
"group_sid" / Pointer(this.header.group_sid_offset, SIDStruct),
"sacl" / If(lambda this: this.header.flags.get('SE_SACL_PRESENT') and this.header.sacl_offset >= this._subcons.header.sizeof(), Pointer(this.header.sacl_offset, ACLStruct)),
"dacl" / If(lambda this: this.header.flags.get('SE_DACL_PRESENT') and this.header.dacl_offset >= this._subcons.header.sizeof(), Pointer(this.header.dacl_offset, ACLStruct)),
)
class Constructable:
@classmethod
def parse(cls, s):
return cls.from_construct(cls.STRUCT.parse(s))
def build(self):
return self.STRUCT.build(self.to_construct())
class SID(Constructable):
STRUCT = SIDStruct
def __init__(self, revision, auth=None, *subauth):
if isinstance(revision, str):
if not revision.startswith('S-'): raise ValueError
revision, auth, *subauth = [ int(x) for x in revision[2:].split('-') ]
elif isinstance(revision, SID):
sid = revision
revision = sid.revision
auth = sid.auth
subauth = sid.subauth
if auth is None: raise ValueError
self.revision = revision
self.auth = auth
self.subauth = tuple(subauth)
def __eq__(self, other):
if isinstance(other, SID):
return self.revision == other.revision and self.auth == other.auth and self.subauth == other.subauth
elif isinstance(other, str):
return self == SID(other)
else: return NotImplemented
__req__ = __eq__
def __hash__(self):
return hash((self.revision, self.auth, self.subauth))
def __str__(self):
return 'S-' + '-'.join( str(a) for a in (self.revision, self.auth)+self.subauth )
def __repr__(self):
return f'SID({str(self)!r})'
@classmethod
def from_construct(cls, cont):
return cls(cont.revision, cont.auth, *cont.subauth)
def to_construct(self):
return dict(revision=self.revision, cnt=len(self.subauth), auth=self.auth, subauth=self.subauth)
SID_ADMINISTRATORS = SID('S-1-5-32-544') # the built-in Administrators group
SID_SYSTEM = SID('S-1-5-18') # the Local System (NT AUTHORITY\SYSTEM) account
SID_USERS = SID('S-1-5-32-545')
SID_AUTH_USERS = SID('S-1-5-11') # Authenticated Users
SID_EVERYONE = SID('S-1-1-0')
# as empiricaly set by Windows 10
MASK_FULL_CONTROL = 0x1f01ff
MASK_READ_EXECUTE = 0x1200a9
def flags2set(flags):
return frozenset( str(k) for k,v in flags.items() if v if k != '_flagsenum' )
def set2flags(s):
return Container({ f: True for f in set(s) })
class ACE(Constructable):
STRUCT = ACEStruct
ALLOW = 'ACCESS_ALLOWED_ACE_TYPE'
DENY = 'ACCESS_DENIED_ACE_TYPE'
def __init__(self, type, access_mask, sid, flags=None, inheritance=None):
self.type = type
self.access_mask = access_mask
self.sid = sid
if flags is None and inheritance is None:
inheritance = 'all'
flags = set(flags or [])
if inheritance in ('object', 'file', 'all'):
flags |= {'OBJECT_INHERIT_ACE'}
if inheritance in ('container', 'dir', 'all'):
flags |= {'CONTAINER_INHERIT_ACE'}
self.flags = frozenset(flags)
def __repr__(self):
return f'ACE({self.type!r}, {self.access_mask!r}, {self.sid!r}, flags={set(self.flags)!r})'
@classmethod
def from_construct(cls, cont):
if cont.header.type not in (cls.ALLOW, cls.DENY):
raise NotImplementedError
return ACE(str(cont.header.type), cont.body.access_mask, SID.from_construct(cont.body.sid), flags=flags2set(cont.header.flags))
def to_construct(self):
r = Container()
r.header = Container(type=self.type, size=0, flags=set2flags(self.flags))
r.body = Container(access_mask=self.access_mask, sid=self.sid.to_construct(), flags={ f: True for f in self.flags })
return r
def make_inherited(self):
return ACE(self.type, self.access_mask, self.sid, self.flags | {"INHERITED_ACE"} - {"INHERIT_ONLY_ACE"})
class SecurityDescriptor(Constructable):
STRUCT = SecurityDescriptorStruct
def __init__(self, *, owner_sid=SID_ADMINISTRATORS, group_sid=SID_ADMINISTRATORS, dacl=[], sacl=None, sacl_inherit=True, dacl_inherit=True):
self.owner_sid = SID(owner_sid)
self.group_sid = SID(group_sid)
self.sacl = sacl
self.dacl = dacl
self.sacl_inherit = sacl_inherit
self.dacl_inherit = dacl_inherit
def __repr__(self):
r = ['SecurityDescriptor(']
r.append(f' owner_sid={self.owner_sid!r},')
r.append(f' group_sid={self.group_sid!r},')
for name in ('sacl', 'dacl'):
acl = getattr(self, name)
if not acl: continue
r.append(f' {name}=[')
for ace in acl:
r.append(' '*8 + repr(ace))
r.append(' ],')
if not self.sacl_inherit: r.append(' sacl_inherit=False,')
if not self.dacl_inherit: r.append(' dacl_inherit=False,')
r.append(')')
return '\n'.join(r)
@classmethod
def acl_from_construct(cls, cont):
if cont: return [ ACE.from_construct(itm) for itm in cont['items'] ]
else: return None
@classmethod
def acl_to_construct(cls, acl):
# Empty ACL is not the same as no ACL.
# https://docs.microsoft.com/en-us/windows/win32/secauthz/null-dacls-and-empty-dacls
if acl is None:
return None
else:
r = Container()
r.header = Container(revision=2, cnt=len(acl), size=0)
r.items = [ itm.to_construct() for itm in acl ]
return r
@classmethod
def from_construct(cls, cont):
owner_sid = SID.from_construct(cont.owner_sid)
group_sid = SID.from_construct(cont.group_sid)
sacl = cls.acl_from_construct(cont.sacl)
dacl = cls.acl_from_construct(cont.dacl)
flags = flags2set(cont.header.flags)
return SecurityDescriptor(owner_sid=owner_sid, group_sid=group_sid, sacl=sacl, dacl=dacl,
sacl_inherit="SE_SACL_PROTECTED" not in flags,
dacl_inherit="SE_DACL_PROTECTED" not in flags,
)
def to_construct(self):
hdr = Container()
r = Container()
r.header = hdr
r.owner_sid = self.owner_sid.to_construct()
r.group_sid = self.group_sid.to_construct()
r.dacl = self.acl_to_construct(self.dacl)
r.sacl = self.acl_to_construct(self.sacl)
# SE_SELF_RELATIVE specifies that offsets in header are relative to start of SD.
# This should be always set in on-disk SDs.
# https://docs.microsoft.com/en-us/windows/win32/secauthz/absolute-and-self-relative-security-descriptors
flags = {"SE_SELF_RELATIVE"}
# Windows seems to always set SE_DACL_AUTO_INHERITED but not SE_DACL_AUTO_INHERIT_REQ. We mirror that.
if self.sacl is not None: flags |= {"SE_SACL_PRESENT", "SE_SACL_AUTO_INHERITED"}
if self.dacl is not None: flags |= {"SE_DACL_PRESENT", "SE_DACL_AUTO_INHERITED"}
if not self.sacl_inherit:
flags |= {"SE_SACL_PROTECTED"}
if not self.dacl_inherit:
flags |= {"SE_DACL_PROTECTED"}
hdr.flags = set2flags(flags)
sacl_size = len(ACLStruct.build(r.sacl)) if r.sacl else 0
dacl_size = len(ACLStruct.build(r.dacl)) if r.dacl else 0
pos = SDHeaderStruct.sizeof()
def add_element(struct, cont):
nonlocal pos
off = pos
pos += len(struct.build(cont))
return off
hdr.owner_sid_offset = add_element(SIDStruct, r.owner_sid)
hdr.group_sid_offset = add_element(SIDStruct, r.group_sid)
if r.sacl: hdr.sacl_offset = add_element(ACLStruct, r.sacl)
else: hdr.sacl_offset = 0
if r.dacl: hdr.dacl_offset = add_element(ACLStruct, r.dacl)
else: hdr.dacl_offset = 0
return r
def apply_sd_recursively(path, sd=None, *, dacl=None, skip_protected=True, set_owner=False, set_group=False):
path = Path(path)
if sd is None:
sd = SecurityDescriptor.parse(dump_sd(path))
else:
write_sd(path, sd)
def visit(p):
is_dir = p.is_dir()
kind = 'CONTAINER' if is_dir else 'OBJECT'
acl = [ ace.make_inherited() for ace in sd.dacl if f'{kind}_INHERIT_ACE' in ace.flags ]
child_sd = SecurityDescriptor.parse(dump_sd(p))
if (not child_sd.dacl_inherit) and skip_protected: return # do not modify ACL and do not recurse
child_sd.dacl = acl
if set_owner: child_sd.owner_sid = sd.owner_sid
if set_group: child_sd.group_sid = sd.group_sid
print(p, child_sd)
write_sd(p, child_sd)
if is_dir:
for child in p.iterdir(): visit(child)
for p in path.iterdir(): visit(p)
Reference in New Issue View Git Blame Copy Permalink