130 lines
5.2 KiB
Python
130 lines
5.2 KiB
Python
import re
|
|
from impacket.ldap import ldap, ldapasn1
|
|
from impacket.ldap.ldap import LDAPSearchError
|
|
|
|
|
|
class NXCModule:
|
|
"""
|
|
Find PKI Enrollment Services in Active Directory and Certificate Templates Names.
|
|
|
|
Module by Tobias Neitzel (@qtc_de) and Sam Freeside (@snovvcrash)
|
|
"""
|
|
|
|
name = "adcs"
|
|
description = "Find PKI Enrollment Services in Active Directory and Certificate Templates Names"
|
|
supported_protocols = ["ldap"]
|
|
opsec_safe = True
|
|
multiple_hosts = True
|
|
|
|
def __init__(self, context=None, module_options=None):
|
|
self.context = context
|
|
self.module_options = module_options
|
|
self.server = None
|
|
self.regex = None
|
|
|
|
def options(self, context, module_options):
|
|
"""
|
|
SERVER PKI Enrollment Server to enumerate templates for. Default is None, use CN name
|
|
BASE_DN The base domain name for the LDAP query
|
|
"""
|
|
self.context = context
|
|
self.regex = re.compile("(https?://.+)")
|
|
|
|
self.server = None
|
|
self.base_dn = None
|
|
if module_options and "SERVER" in module_options:
|
|
self.server = module_options["SERVER"]
|
|
if module_options and "BASE_DN" in module_options:
|
|
self.base_dn = module_options["BASE_DN"]
|
|
|
|
def on_login(self, context, connection):
|
|
"""On a successful LDAP login we perform a search for all PKI Enrollment Server or Certificate Templates Names."""
|
|
if self.server is None:
|
|
search_filter = "(objectClass=pKIEnrollmentService)"
|
|
else:
|
|
search_filter = f"(distinguishedName=CN={self.server},CN=Enrollment Services,CN=Public Key Services,CN=Services,CN=Configuration,"
|
|
self.context.log.highlight(f"Using PKI CN: {self.server}")
|
|
|
|
context.log.display(f"Starting LDAP search with search filter '{search_filter}'")
|
|
|
|
try:
|
|
sc = ldap.SimplePagedResultsControl()
|
|
base_dn_root = connection.ldapConnection._baseDN if self.base_dn is None else self.base_dn
|
|
|
|
if self.server is None:
|
|
connection.ldapConnection.search(
|
|
searchFilter=search_filter,
|
|
attributes=[],
|
|
sizeLimit=0,
|
|
searchControls=[sc],
|
|
perRecordCallback=self.process_servers,
|
|
searchBase="CN=Configuration," + base_dn_root,
|
|
)
|
|
else:
|
|
connection.ldapConnection.search(
|
|
searchFilter=search_filter + base_dn_root + ")",
|
|
attributes=["certificateTemplates"],
|
|
sizeLimit=0,
|
|
searchControls=[sc],
|
|
perRecordCallback=self.process_templates,
|
|
searchBase="CN=Configuration," + base_dn_root,
|
|
)
|
|
except LDAPSearchError as e:
|
|
context.log.fail(f"Obtained unexpected exception: {e}")
|
|
|
|
def process_servers(self, item):
|
|
"""Function that is called to process the items obtain by the LDAP search when listing PKI Enrollment Servers."""
|
|
if not isinstance(item, ldapasn1.SearchResultEntry):
|
|
return
|
|
|
|
urls = []
|
|
host_name = None
|
|
cn = None
|
|
|
|
try:
|
|
for attribute in item["attributes"]:
|
|
if str(attribute["type"]) == "dNSHostName":
|
|
host_name = attribute["vals"][0].asOctets().decode("utf-8")
|
|
if str(attribute["type"]) == "cn":
|
|
cn = attribute["vals"][0].asOctets().decode("utf-8")
|
|
elif str(attribute["type"]) == "msPKI-Enrollment-Servers":
|
|
values = attribute["vals"]
|
|
|
|
for value in values:
|
|
value = value.asOctets().decode("utf-8")
|
|
match = self.regex.search(value)
|
|
if match:
|
|
urls.append(match.group(1))
|
|
except Exception as e:
|
|
entry = host_name or "item"
|
|
self.context.log.fail(f"Skipping {entry}, cannot process LDAP entry due to error: '{e!s}'")
|
|
|
|
if host_name:
|
|
self.context.log.highlight(f"Found PKI Enrollment Server: {host_name}")
|
|
if cn:
|
|
self.context.log.highlight(f"Found CN: {cn}")
|
|
for url in urls:
|
|
self.context.log.highlight(f"Found PKI Enrollment WebService: {url}")
|
|
|
|
def process_templates(self, item):
|
|
"""Function that is called to process the items obtain by the LDAP search when listing Certificate Templates Names for a specific PKI Enrollment Server."""
|
|
if not isinstance(item, ldapasn1.SearchResultEntry):
|
|
return
|
|
|
|
templates = []
|
|
template_name = None
|
|
|
|
try:
|
|
for attribute in item["attributes"]:
|
|
if str(attribute["type"]) == "certificateTemplates":
|
|
for val in attribute["vals"]:
|
|
template_name = val.asOctets().decode("utf-8")
|
|
templates.append(template_name)
|
|
except Exception as e:
|
|
entry = template_name or "item"
|
|
self.context.log.fail(f"Skipping {entry}, cannot process LDAP entry due to error: '{e}'")
|
|
|
|
if templates:
|
|
for t in templates:
|
|
self.context.log.highlight(f"Found Certificate Template: {t}")
|