python/samba/sd_utils.py

   1 # Utility methods for security descriptor manipulation
   2 #
   3 # Copyright Nadezhda Ivanova 2010 <nivanova@samba.org>
   4 #
   5 # This program is free software; you can redistribute it and/or modify
   6 # it under the terms of the GNU General Public License as published by
   7 # the Free Software Foundation; either version 3 of the License, or
   8 # (at your option) any later version.
   9 #
  10 # This program is distributed in the hope that it will be useful,
  11 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  12 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13 # GNU General Public License for more details.
  14 #
  15 # You should have received a copy of the GNU General Public License
  16 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
  17 #
  18
  19 """Utility methods for security descriptor manipulation."""
  20
  21 from ldb import FLAG_MOD_REPLACE, SCOPE_BASE, Dn, Message, MessageElement
  22
  23 import samba
  24 from samba.dcerpc import security
  25 from samba.ndr import ndr_deepcopy, ndr_pack, ndr_unpack
  26 from samba.ntstatus import NT_STATUS_OBJECT_NAME_NOT_FOUND
  27
  28
  29 def escaped_claim_id(claim_id):
  30     """Encode claim attribute names according to [MS-DTYP] 2.5.1 ("attr-char2")
  31
  32     Some characters must be encoded as %hhhh, while others must not be.
  33     Of the optional ones, we encode some control characters.
  34
  35     The \x00 byte is also encoded, which is useful for tests, but it
  36     is forbidden in either form.
  37     """
  38     escapes = '\x00\t\n\x0b\x0c\r !"%&()<=>|'
  39     return ''.join(c
  40                    if c not in escapes
  41                    else f'%{ord(c):04x}'
  42                    for c in claim_id)
  43
  44
  45 class SDUtils(object):
  46     """Some utilities for manipulation of security descriptors on objects."""
  47
  48     def __init__(self, samdb):
  49         self.ldb = samdb
  50         self.domain_sid = security.dom_sid(self.ldb.get_domain_sid())
  51
  52     def modify_sd_on_dn(self, object_dn, sd, controls=None):
  53         """Modify security descriptor using either SDDL string
  54             or security.descriptor object
  55         """
  56         m = Message()
  57         if isinstance(object_dn, Dn):
  58             m.dn = object_dn
  59         else:
  60             m.dn = Dn(self.ldb, object_dn)
  61
  62         assert (isinstance(sd, str) or isinstance(sd, security.descriptor))
  63         if isinstance(sd, str):
  64             tmp_desc = security.descriptor.from_sddl(sd, self.domain_sid)
  65         elif isinstance(sd, security.descriptor):
  66             tmp_desc = sd
  67
  68         m["nTSecurityDescriptor"] = MessageElement(ndr_pack(tmp_desc),
  69                                                    FLAG_MOD_REPLACE,
  70                                                    "nTSecurityDescriptor")
  71         self.ldb.modify(m, controls)
  72
  73     def read_sd_on_dn(self, object_dn, controls=None):
  74         res = self.ldb.search(object_dn, SCOPE_BASE, None,
  75                               ["nTSecurityDescriptor"], controls=controls)
  76         desc = res[0]["nTSecurityDescriptor"][0]
  77         return ndr_unpack(security.descriptor, desc)
  78
  79     def get_object_sid(self, object_dn):
  80         res = self.ldb.search(object_dn)
  81         return ndr_unpack(security.dom_sid, res[0]["objectSid"][0])
  82
  83     def update_aces_in_dacl(self, dn, del_aces=None, add_aces=None,
  84                             sddl_attr=None, controls=None):
  85         if del_aces is None:
  86             del_aces = []
  87         if add_aces is None:
  88             add_aces = []
  89
  90         def ace_from_sddl(ace_sddl):
  91             ace_sd = security.descriptor.from_sddl("D:" + ace_sddl, self.domain_sid)
  92             assert len(ace_sd.dacl.aces) == 1
  93             return ace_sd.dacl.aces[0]
  94
  95         if sddl_attr is None:
  96             if controls is None:
  97                 controls = ["sd_flags:1:%d" % security.SECINFO_DACL]
  98             sd = self.read_sd_on_dn(dn, controls=controls)
  99             if not sd.type & security.SEC_DESC_DACL_PROTECTED:
 100                 # if the DACL is not protected remove all
 101                 # inherited aces, as they will be re-inherited
 102                 # on the server, we need a ndr_deepcopy in order
 103                 # to avoid reference problems while deleting
 104                 # the aces while looping over them
 105                 dacl_copy = ndr_deepcopy(sd.dacl)
 106                 for ace in dacl_copy.aces:
 107                     if ace.flags & security.SEC_ACE_FLAG_INHERITED_ACE:
 108                         try:
 109                             sd.dacl_del_ace(ace)
 110                         except samba.NTSTATUSError as err:
 111                             if err.args[0] != NT_STATUS_OBJECT_NAME_NOT_FOUND:
 112                                 raise err
 113                             # dacl_del_ace may remove more than
 114                             # one ace, so we may not find it anymore
 115                             pass
 116         else:
 117             if controls is None:
 118                 controls = []
 119             res = self.ldb.search(dn, SCOPE_BASE, None,
 120                                   [sddl_attr], controls=controls)
 121             old_sddl = str(res[0][sddl_attr][0])
 122             sd = security.descriptor.from_sddl(old_sddl, self.domain_sid)
 123
 124         num_changes = 0
 125         del_ignored = []
 126         add_ignored = []
 127         inherited_ignored = []
 128
 129         for ace in del_aces:
 130             if isinstance(ace, str):
 131                 ace = ace_from_sddl(ace)
 132             assert isinstance(ace, security.ace)
 133
 134             if ace.flags & security.SEC_ACE_FLAG_INHERITED_ACE:
 135                 inherited_ignored.append(ace)
 136                 continue
 137
 138             if ace not in sd.dacl.aces:
 139                 del_ignored.append(ace)
 140                 continue
 141
 142             sd.dacl_del_ace(ace)
 143             num_changes += 1
 144
 145         for ace in add_aces:
 146             add_idx = -1
 147             if isinstance(ace, dict):
 148                 if "idx" in ace:
 149                     add_idx = ace["idx"]
 150                 ace = ace["ace"]
 151             if isinstance(ace, str):
 152                 ace = ace_from_sddl(ace)
 153             assert isinstance(ace, security.ace)
 154
 155             if ace.flags & security.SEC_ACE_FLAG_INHERITED_ACE:
 156                 inherited_ignored.append(ace)
 157                 continue
 158
 159             if ace in sd.dacl.aces:
 160                 add_ignored.append(ace)
 161                 continue
 162
 163             sd.dacl_add(ace, add_idx)
 164             num_changes += 1
 165
 166         if num_changes == 0:
 167             return del_ignored, add_ignored, inherited_ignored
 168
 169         if sddl_attr is None:
 170             self.modify_sd_on_dn(dn, sd, controls=controls)
 171         else:
 172             new_sddl = sd.as_sddl(self.domain_sid)
 173             m = Message()
 174             m.dn = dn
 175             m[sddl_attr] = MessageElement(new_sddl.encode('ascii'),
 176                                           FLAG_MOD_REPLACE,
 177                                           sddl_attr)
 178             self.ldb.modify(m, controls=controls)
 179
 180         return del_ignored, add_ignored, inherited_ignored
 181
 182     def dacl_prepend_aces(self, object_dn, aces, controls=None):
 183         """Prepend an ACE (or more) to an objects security descriptor
 184         """
 185         ace_sd = security.descriptor.from_sddl("D:" + aces, self.domain_sid)
 186         add_aces = []
 187         add_idx = 0
 188         for ace in ace_sd.dacl.aces:
 189             add_aces.append({"idx": add_idx, "ace": ace})
 190             add_idx += 1
 191         _, ai, ii = self.update_aces_in_dacl(object_dn, add_aces=add_aces,
 192                                              controls=controls)
 193         return ai, ii
 194
 195     def dacl_add_ace(self, object_dn, ace):
 196         """Add an ACE (or more) to an objects security descriptor
 197         """
 198         _, _ = self.dacl_prepend_aces(object_dn, ace,
 199                                       controls=["show_deleted:1"])
 200
 201     def dacl_append_aces(self, object_dn, aces, controls=None):
 202         """Append an ACE (or more) to an objects security descriptor
 203         """
 204         ace_sd = security.descriptor.from_sddl("D:" + aces, self.domain_sid)
 205         add_aces = []
 206         for ace in ace_sd.dacl.aces:
 207             add_aces.append(ace)
 208         _, ai, ii = self.update_aces_in_dacl(object_dn, add_aces=add_aces,
 209                                              controls=controls)
 210         return ai, ii
 211
 212     def dacl_delete_aces(self, object_dn, aces, controls=None):
 213         """Delete an ACE (or more) to an objects security descriptor
 214         """
 215         del_sd = security.descriptor.from_sddl("D:" + aces, self.domain_sid)
 216         del_aces = []
 217         for ace in del_sd.dacl.aces:
 218             del_aces.append(ace)
 219         di, _, ii = self.update_aces_in_dacl(object_dn, del_aces=del_aces,
 220                                              controls=controls)
 221         return di, ii
 222
 223     def get_sd_as_sddl(self, object_dn, controls=None):
 224         """Return object nTSecurityDescriptor in SDDL format
 225         """
 226         if controls is None:
 227             controls = []
 228         desc = self.read_sd_on_dn(object_dn, controls + ["show_deleted:1"])
 229         return desc.as_sddl(self.domain_sid)