mirror of
https://github.com/sploutchy/impacket
synced 2026-06-08 17:32:01 +00:00
asolino
0afacbea5a
This script will gather data about the domain's users and their corresponding email addresses. It will also include some extra information about last logon and last password set attributes. You can enable or disable the the attributes shown in the final table by changing the values in line 184 and headers in line 190. If no entries are returned that means users don't have email addresses specified.
263 lines
10 KiB
Python
Executable File
263 lines
10 KiB
Python
Executable File
#!/usr/bin/python
|
|
# Copyright (c) 2016 CORE Security Technologies
|
|
#
|
|
# This software is provided under under a slightly modified version
|
|
# of the Apache Software License. See the accompanying LICENSE file
|
|
# for more information.
|
|
#
|
|
# Author:
|
|
# Alberto Solino (@agsolino)
|
|
#
|
|
# Description:
|
|
# This script will gather data about the domain's users and their corresponding email addresses. It will also
|
|
# include some extra information about last logon and last password set attributes.
|
|
# You can enable or disable the the attributes shown in the final table by changing the values in line 184 and
|
|
# headers in line 190.
|
|
# If no entries are returned that means users don't have email addresses specified.
|
|
#
|
|
# Reference for:
|
|
# LDAP
|
|
#
|
|
|
|
|
|
import argparse
|
|
import logging
|
|
import os
|
|
import sys
|
|
from datetime import datetime
|
|
from binascii import hexlify, unhexlify
|
|
|
|
from pyasn1.codec.der import decoder
|
|
from impacket import version
|
|
from impacket.dcerpc.v5.samr import UF_ACCOUNTDISABLE, UF_NORMAL_ACCOUNT
|
|
from impacket.examples import logger
|
|
from impacket.krb5 import constants
|
|
from impacket.krb5.asn1 import TGS_REP
|
|
from impacket.krb5.ccache import CCache
|
|
from impacket.krb5.kerberosv5 import getKerberosTGT, getKerberosTGS
|
|
from impacket.krb5.types import Principal
|
|
from impacket.ldap import ldap, ldapasn1
|
|
from impacket.smbconnection import SMBConnection
|
|
|
|
|
|
class GetADUsers:
|
|
@staticmethod
|
|
def printTable(items, header):
|
|
colLen = []
|
|
for i, col in enumerate(header):
|
|
rowMaxLen = max([len(row[i]) for row in items])
|
|
colLen.append(max(rowMaxLen, len(col)))
|
|
|
|
outputFormat = ' '.join(['{%d:%ds} ' % (num, width) for num, width in enumerate(colLen)])
|
|
|
|
# Print header
|
|
print outputFormat.format(*header)
|
|
print ' '.join(['-' * itemLen for itemLen in colLen])
|
|
|
|
# And now the rows
|
|
for row in items:
|
|
print outputFormat.format(*row)
|
|
|
|
def __init__(self, username, password, domain, cmdLineOptions):
|
|
self.options = cmdLineOptions
|
|
self.__username = username
|
|
self.__password = password
|
|
self.__domain = domain
|
|
self.__lmhash = ''
|
|
self.__nthash = ''
|
|
self.__aesKey = cmdLineOptions.aesKey
|
|
self.__doKerberos = cmdLineOptions.k
|
|
self.__target = None
|
|
self.__kdcHost = cmdLineOptions.dc_ip
|
|
self.__requestUser = cmdLineOptions.user
|
|
if cmdLineOptions.hashes is not None:
|
|
self.__lmhash, self.__nthash = cmdLineOptions.hashes.split(':')
|
|
|
|
# Create the baseDN
|
|
domainParts = self.__domain.split('.')
|
|
self.baseDN = ''
|
|
for i in domainParts:
|
|
self.baseDN += 'dc=%s,' % i
|
|
# Remove last ','
|
|
self.baseDN = self.baseDN[:-1]
|
|
|
|
def getMachineName(self):
|
|
if self.__kdcHost is not None:
|
|
s = SMBConnection(self.__kdcHost, self.__kdcHost)
|
|
else:
|
|
s = SMBConnection(self.__domain, self.__domain)
|
|
try:
|
|
s.login('', '')
|
|
except Exception:
|
|
logging.debug('Error while anonymous logging into %s' % self.__domain)
|
|
|
|
s.logoff()
|
|
return s.getServerName()
|
|
|
|
@staticmethod
|
|
def getUnixTime(t):
|
|
t -= 116444736000000000
|
|
t /= 10000000
|
|
return t
|
|
|
|
def run(self):
|
|
if self.__doKerberos:
|
|
self.__target = self.getMachineName()
|
|
else:
|
|
if self.__kdcHost is not None:
|
|
self.__target = self.__kdcHost
|
|
else:
|
|
self.__target = self.__domain
|
|
|
|
# Connect to LDAP
|
|
try:
|
|
ldapConnection = ldap.LDAPConnection('ldap://%s'%self.__target, self.baseDN, self.__kdcHost)
|
|
if self.__doKerberos is not True:
|
|
ldapConnection.login(self.__username, self.__password, self.__domain, self.__lmhash, self.__nthash)
|
|
else:
|
|
ldapConnection.kerberosLogin(self.__username, self.__password, self.__domain, self.__lmhash, self.__nthash,
|
|
self.__aesKey, kdcHost=self.__kdcHost)
|
|
except ldap.LDAPSessionError, e:
|
|
if str(e).find('strongerAuthRequired') >= 0:
|
|
# We need to try SSL
|
|
ldapConnection = ldap.LDAPConnection('ldaps://%s' % self.__target, self.baseDN, self.__kdcHost)
|
|
if self.__doKerberos is not True:
|
|
ldapConnection.login(self.__username, self.__password, self.__domain, self.__lmhash, self.__nthash)
|
|
else:
|
|
ldapConnection.kerberosLogin(self.__username, self.__password, self.__domain, self.__lmhash, self.__nthash,
|
|
self.__aesKey, kdcHost=self.__kdcHost)
|
|
else:
|
|
raise
|
|
|
|
# Building the search filter
|
|
searchFilter = "(&(sAMAccountName=*)(mail=*)"
|
|
|
|
if self.__requestUser is not None:
|
|
searchFilter += '(sAMAccountName:=%s))' % self.__requestUser
|
|
else:
|
|
searchFilter += ')'
|
|
|
|
try:
|
|
logging.info('Querying %s for information about domain. Be patient...' % self.__target)
|
|
resp = ldapConnection.search(searchFilter=searchFilter,
|
|
attributes=['sAMAccountName', 'pwdLastSet', 'mail', 'lastLogon'],
|
|
sizeLimit=999)
|
|
except ldap.LDAPSearchError, e:
|
|
if e.getErrorString().find('sizeLimitExceeded') >= 0:
|
|
logging.debug('sizeLimitExceeded exception caught, giving up and processing the data received')
|
|
# We reached the sizeLimit, process the answers we have already and that's it. Until we implement
|
|
# paged queries
|
|
resp = e.getAnswers()
|
|
pass
|
|
else:
|
|
raise
|
|
|
|
answers = []
|
|
logging.debug('Total of records returned %d' % len(resp))
|
|
|
|
for item in resp:
|
|
if isinstance(item, ldapasn1.SearchResultEntry) is not True:
|
|
continue
|
|
sAMAccountName = ''
|
|
pwdLastSet = ''
|
|
mail = ''
|
|
lastLogon = 'N/A'
|
|
try:
|
|
for attribute in item['attributes']:
|
|
if attribute['type'] == 'sAMAccountName':
|
|
if str(attribute['vals'][0]).endswith('$') is False:
|
|
# User Account
|
|
sAMAccountName = str(attribute['vals'][0])
|
|
elif attribute['type'] == 'pwdLastSet':
|
|
if str(attribute['vals'][0]) == '0':
|
|
pwdLastSet = '<never>'
|
|
else:
|
|
pwdLastSet = str(datetime.fromtimestamp(self.getUnixTime(int(str(attribute['vals'][0])))))
|
|
elif attribute['type'] == 'lastLogon':
|
|
if str(attribute['vals'][0]) == '0':
|
|
lastLogon = '<never>'
|
|
else:
|
|
lastLogon = str(datetime.fromtimestamp(self.getUnixTime(int(str(attribute['vals'][0])))))
|
|
elif attribute['type'] == 'mail':
|
|
mail = str(attribute['vals'][0])
|
|
|
|
answers.append([sAMAccountName, mail, pwdLastSet, lastLogon])
|
|
except Exception, e:
|
|
logging.error('Skipping item, cannot process due to error %s' % str(e))
|
|
pass
|
|
|
|
if len(answers)>0:
|
|
self.printTable(answers, header=[ "Name", "Email", "PasswordLastSet", "LastLogon"])
|
|
print '\n\n'
|
|
|
|
else:
|
|
print "No entries found!"
|
|
|
|
|
|
# Process command-line arguments.
|
|
if __name__ == '__main__':
|
|
# Init the example's logger theme
|
|
logger.init()
|
|
print version.BANNER
|
|
|
|
parser = argparse.ArgumentParser(add_help = True, description = "Queries target domain for users data")
|
|
|
|
parser.add_argument('target', action='store', help='domain/username[:password]')
|
|
parser.add_argument('-user', action='store', metavar='username', help='Requests data for specific user ')
|
|
parser.add_argument('-debug', action='store_true', help='Turn DEBUG output ON')
|
|
|
|
group = parser.add_argument_group('authentication')
|
|
|
|
group.add_argument('-hashes', action="store", metavar = "LMHASH:NTHASH", help='NTLM hashes, format is LMHASH:NTHASH')
|
|
group.add_argument('-no-pass', action="store_true", help='don\'t ask for password (useful for -k)')
|
|
group.add_argument('-k', action="store_true", help='Use Kerberos authentication. Grabs credentials from ccache file '
|
|
'(KRB5CCNAME) based on target parameters. If valid credentials '
|
|
'cannot be found, it will use the ones specified in the command '
|
|
'line')
|
|
group.add_argument('-aesKey', action="store", metavar = "hex key", help='AES key to use for Kerberos Authentication '
|
|
'(128 or 256 bits)')
|
|
group.add_argument('-dc-ip', action='store',metavar = "ip address", help='IP Address of the domain controller. If '
|
|
'ommited it use the domain part (FQDN) '
|
|
'specified in the target parameter')
|
|
|
|
if len(sys.argv)==1:
|
|
parser.print_help()
|
|
sys.exit(1)
|
|
|
|
options = parser.parse_args()
|
|
|
|
if options.debug is True:
|
|
logging.getLogger().setLevel(logging.DEBUG)
|
|
else:
|
|
logging.getLogger().setLevel(logging.INFO)
|
|
|
|
import re
|
|
# This is because I'm lazy with regex
|
|
# ToDo: We need to change the regex to fullfil domain/username[:password]
|
|
targetParam = options.target+'@'
|
|
domain, username, password, address = re.compile('(?:(?:([^/@:]*)/)?([^@:]*)(?::([^@]*))?@)?(.*)').match(targetParam).groups('')
|
|
|
|
#In case the password contains '@'
|
|
if '@' in address:
|
|
password = password + '@' + address.rpartition('@')[0]
|
|
address = address.rpartition('@')[2]
|
|
|
|
if domain is '':
|
|
logging.critical('Domain should be specified!')
|
|
sys.exit(1)
|
|
|
|
if password == '' and username != '' and options.hashes is None and options.no_pass is False and options.aesKey is None:
|
|
from getpass import getpass
|
|
password = getpass("Password:")
|
|
|
|
if options.aesKey is not None:
|
|
options.k = True
|
|
|
|
try:
|
|
executer = GetADUsers(username, password, domain, options)
|
|
executer.run()
|
|
except Exception, e:
|
|
#import traceback
|
|
#print traceback.print_exc()
|
|
print str(e)
|