Source code for univention.s4connector.s4.ntsecurity_descriptor

#!/usr/bin/python3
# -*- coding: utf-8 -*-
#
# Univention S4 Connector
#  nTSecurityDescriptor sync
#
# Copyright 2014-2022 Univention GmbH
#
# https://www.univention.de/
#
# All rights reserved.
#
# The source code of this program is made available
# under the terms of the GNU Affero General Public License version 3
# (GNU AGPL V3) as published by the Free Software Foundation.
#
# Binary versions of this program provided by Univention to you as
# well as other copyrighted, protected or trademarked materials like
# Logos, graphics, fonts, specific documentations and configurations,
# cryptographic keys etc. are subject to a license agreement between
# you and Univention and not subject to the GNU AGPL V3.
#
# In the case you use this program under the terms of the GNU AGPL V3,
# the program is provided in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public
# License with the Debian GNU/Linux or Univention distribution in file
# /usr/share/common-licenses/AGPL-3; if not, see
# <https://www.gnu.org/licenses/>.


import ldap
import univention.debug2 as ud
from samba.dcerpc import security
from samba.ndr import ndr_pack, ndr_unpack
from ldap.controls.readentry import PostReadControl



[docs]def encode_sddl_to_sd_in_ndr(domain_sid, ntsd_sddl):
	ntsd = security.descriptor.from_sddl(ntsd_sddl, domain_sid)
	ntsd_ndr = ndr_pack(ntsd)
	return ntsd_ndr




[docs]def decode_sd_in_ndr_to_sddl(domain_sid, value):
	ntsd = ndr_unpack(security.descriptor, value)
	return ntsd.as_sddl(domain_sid)



# Post-create/modify functions



[docs]def ntsd_to_s4(s4connector, key, object):
	ud.debug(ud.LDAP, ud.INFO, "ntsd_to_s4 object: %s" % object)

	# object dn was already mapped to the s4 DN:
	s4_dn = object['dn']
	modlist = []

	# search the ucs object via
	if 'msNTSecurityDescriptor' not in object['attributes']:
		ud.debug(ud.LDAP, ud.INFO, 'ntsd_to_s4: UCS object does not have a msNTSecurityDescriptor')
		return

	ucs_ntsd_sddl = object['attributes']['msNTSecurityDescriptor'][0].decode('ASCII')
	s4_attributes = s4connector.lo_s4.get(s4_dn, attr=['nTSecurityDescriptor'])
	ntsd_ndr = s4_attributes.get('nTSecurityDescriptor')

	if ntsd_ndr:
		domain_sid = security.dom_sid(s4connector.s4_sid)
		s4_ntsd_sddl = decode_sd_in_ndr_to_sddl(domain_sid, ntsd_ndr[0])
		if s4_ntsd_sddl == ucs_ntsd_sddl:
			ud.debug(ud.LDAP, ud.INFO, 'ntsd_to_s4: nTSecurityDescriptors are equal')
			return

		ud.debug(ud.LDAP, ud.INFO, 'ntsd_to_s4: changing nTSecurityDescriptor from %s to %s' % (s4_ntsd_sddl, ucs_ntsd_sddl))

		ucs_ntsd_ndr = encode_sddl_to_sd_in_ndr(domain_sid, ucs_ntsd_sddl)
		modlist.append((ldap.MOD_REPLACE, 'nTSecurityDescriptor', ucs_ntsd_ndr))

		s4connector.lo_s4.lo.modify_ext_s(s4_dn, modlist)




[docs]def ntsd_to_ucs(s4connector, key, s4_object):
	ud.debug(ud.LDAP, ud.INFO, "ntsd_to_ucs S4 object: %s" % s4_object)
	ud.debug(ud.LDAP, ud.INFO, "ntsd_to_ucs S4 key: %s" % key)

	# modlist
	ml = []

	# search Samba DS expicitly for hidden attribute
	# object dn is already mapped to the UCS DN:
	s4_dn = s4_object.get('dn')
	if not s4_dn:
		return  # ignore

	try:
		s4_attributes = s4connector.lo_s4.get(s4_dn, attr=['nTSecurityDescriptor'], required=True)
	except ldap.NO_SUCH_OBJECT:
		ud.debug(ud.LDAP, ud.WARN, 'ntsd_to_ucs: S4 object (%s) not found' % s4_dn)
		return

	ntsd_ndr = s4_attributes.get('nTSecurityDescriptor')
	if not ntsd_ndr:
		ud.debug(ud.LDAP, ud.INFO, 'ntsd_to_ucs: nTSecurityDescriptor not found in attributes!')
		return

	# search in UCS/OpenLDAP DS to determine modify/add
	ucs_dn = s4_dn
	try:
		ucs_attributes = s4connector.lo.get(ucs_dn, attr=['msNTSecurityDescriptor'])
	except ldap.NO_SUCH_OBJECT:
		ud.debug(ud.LDAP, ud.WARN, 'sid_to_ucs: UCS object (%s) not found' % ucs_dn)
		return

	domain_sid = security.dom_sid(s4connector.s4_sid)
	s4_ntsd_sddl = decode_sd_in_ndr_to_sddl(domain_sid, ntsd_ndr[0]).encode('ASCII')
	ucs_ntsd_sddl = ucs_attributes.get('msNTSecurityDescriptor', [None])[0]
	if not ucs_ntsd_sddl or ucs_ntsd_sddl != s4_ntsd_sddl:
		ml.append(('msNTSecurityDescriptor', ucs_ntsd_sddl, s4_ntsd_sddl))
	if ml:
		ud.debug(ud.LDAP, ud.INFO, 'ntsd_to_ucs: modlist = %s' % ml)
		serverctrls = [PostReadControl(True, ['entryUUID', 'entryCSN'])]
		response = {}
		s4connector.lo.lo.modify(ucs_dn, ml, serverctrls=serverctrls, response=response)
		for c in response.get('ctrls', []):   # If the modify actually did something
			if c.controlType == PostReadControl.controlType:
				entryUUID = c.entry['entryUUID'][0]
				entryCSN = c.entry['entryCSN'][0]
				s4connector._remember_entryCSN_commited_by_connector(entryUUID, entryCSN)