path: root/python/samba/dbchecker.py

# Samba4 AD database checker
#
# Copyright (C) Andrew Tridgell 2011
# Copyright (C) Matthieu Patou <mat@matws.net> 2011
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed 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 General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
#

import ldb
import samba
import time
from base64 import b64decode
from samba import dsdb
from samba import common
from samba.dcerpc import misc
from samba.dcerpc import drsuapi
from samba.ndr import ndr_unpack, ndr_pack
from samba.dcerpc import drsblobs
from samba.common import dsdb_Dn
from samba.dcerpc import security
from samba.descriptor import get_wellknown_sds, get_diff_sds
from samba.auth import system_session, admin_session


class dbcheck(object):
    """check a SAM database for errors"""

    def __init__(self, samdb, samdb_schema=None, verbose=False, fix=False,
                 yes=False, quiet=False, in_transaction=False,
                 reset_well_known_acls=False):
        self.samdb = samdb
        self.dict_oid_name = None
        self.samdb_schema = (samdb_schema or samdb)
        self.verbose = verbose
        self.fix = fix
        self.yes = yes
        self.quiet = quiet
        self.remove_all_unknown_attributes = False
        self.remove_all_empty_attributes = False
        self.fix_all_normalisation = False
        self.fix_all_DN_GUIDs = False
        self.fix_all_binary_dn = False
        self.remove_all_deleted_DN_links = False
        self.fix_all_target_mismatch = False
        self.fix_all_metadata = False
        self.fix_time_metadata = False
        self.fix_all_missing_backlinks = False
        self.fix_all_orphaned_backlinks = False
        self.fix_rmd_flags = False
        self.fix_ntsecuritydescriptor = False
        self.fix_ntsecuritydescriptor_owner_group = False
        self.seize_fsmo_role = False
        self.move_to_lost_and_found = False
        self.fix_instancetype = False
        self.fix_replmetadata_zero_invocationid = False
        self.fix_replmetadata_unsorted_attid = False
        self.fix_deleted_deleted_objects = False
        self.fix_dn = False
        self.fix_base64_userparameters = False
        self.fix_utf8_userparameters = False
        self.fix_doubled_userparameters = False
        self.reset_well_known_acls = reset_well_known_acls
        self.reset_all_well_known_acls = False
        self.in_transaction = in_transaction
        self.infrastructure_dn = ldb.Dn(samdb, "CN=Infrastructure," + samdb.domain_dn())
        self.naming_dn = ldb.Dn(samdb, "CN=Partitions,%s" % samdb.get_config_basedn())
        self.schema_dn = samdb.get_schema_basedn()
        self.rid_dn = ldb.Dn(samdb, "CN=RID Manager$,CN=System," + samdb.domain_dn())
        self.ntds_dsa = ldb.Dn(samdb, samdb.get_dsServiceName())
        self.class_schemaIDGUID = {}
        self.wellknown_sds = get_wellknown_sds(self.samdb)
        self.fix_all_missing_objectclass = False

        self.dn_set = set()

        self.name_map = {}
        try:
            res = samdb.search(base="CN=DnsAdmins,CN=Users,%s" % samdb.domain_dn(), scope=ldb.SCOPE_BASE,
                           attrs=["objectSid"])
            dnsadmins_sid = ndr_unpack(security.dom_sid, res[0]["objectSid"][0])
            self.name_map['DnsAdmins'] = str(dnsadmins_sid)
        except ldb.LdbError, (enum, estr):
            if enum != ldb.ERR_NO_SUCH_OBJECT:
                raise
            pass

        self.system_session_info = system_session()
        self.admin_session_info = admin_session(None, samdb.get_domain_sid())

        res = self.samdb.search(base=self.ntds_dsa, scope=ldb.SCOPE_BASE, attrs=['msDS-hasMasterNCs', 'hasMasterNCs'])
        if "msDS-hasMasterNCs" in res[0]:
            self.write_ncs = res[0]["msDS-hasMasterNCs"]
        else:
            # If the Forest Level is less than 2003 then there is no
            # msDS-hasMasterNCs, so we fall back to hasMasterNCs
            # no need to merge as all the NCs that are in hasMasterNCs must
            # also be in msDS-hasMasterNCs (but not the opposite)
            if "hasMasterNCs" in res[0]:
                self.write_ncs = res[0]["hasMasterNCs"]
            else:
                self.write_ncs = None

        res = self.samdb.search(base="", scope=ldb.SCOPE_BASE, attrs=['namingContexts'])
        try:
            ncs = res[0]["namingContexts"]
            self.deleted_objects_containers = []
            for nc in ncs:
                try:
                    dn = self.samdb.get_wellknown_dn(ldb.Dn(self.samdb, nc),
                                                     dsdb.DS_GUID_DELETED_OBJECTS_CONTAINER)
                    self.deleted_objects_containers.append(dn)
                except KeyError:
                    pass
        except KeyError:
            pass
        except IndexError:
            pass

    def check_database(self, DN=None, scope=ldb.SCOPE_SUBTREE, controls=[], attrs=['*']):
        '''perform a database check, returning the number of errors found'''
        res = self.samdb.search(base=DN, scope=scope, attrs=['dn'], controls=controls)
        self.report('Checking %u objects' % len(res))
        error_count = 0

        for object in res:
            self.dn_set.add(str(object.dn))
            error_count += self.check_object(object.dn, attrs=attrs)

        if DN is None:
            error_count += self.check_rootdse()

        if error_count != 0 and not self.fix:
            self.report("Please use --fix to fix these errors")

        self.report('Checked %u objects (%u errors)' % (len(res), error_count))
        return error_count

    def report(self, msg):
        '''print a message unless quiet is set'''
        if not self.quiet:
            print(msg)

    def confirm(self, msg, allow_all=False, forced=False):
        '''confirm a change'''
        if not self.fix:
            return False
        if self.quiet:
            return self.yes
        if self.yes:
            forced = True
        return common.confirm(msg, forced=forced, allow_all=allow_all)

    ################################################################
    # a local confirm function with support for 'all'
    def confirm_all(self, msg, all_attr):
        '''confirm a change with support for "all" '''
        if not self.fix:
            return False
        if self.quiet:
            return self.yes
        if getattr(self, all_attr) == 'NONE':
            return False
        if getattr(self, all_attr) == 'ALL':
            forced = True
        else:
            forced = self.yes
        c = common.confirm(msg, forced=forced, allow