admin/cisagov-snafflepy
1
0
Fork 0
mirror of https://github.com/cisagov/snafflepy synced 2026-09-24 18:22:23 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #22 from robert-todora/dev

Browse Source 
added file download capability
This commit is contained in:
Robert Todora
2023-07-25 12:57:14 -05:00
committed by GitHub
parent 312e62b421 0602a8b4dc
commit cae6ba8406
6 changed files with 91 additions and 87 deletions
Download Patch File Download Diff File Expand all files Collapse all files
snaffcore/DefaultRules/ShareRules/Keep/KeepDollarShares.toml
+2 -2
View File
@@ -2,9 +2,9 @@
EnumerationScope = "ShareEnumeration" EnumerationScope = "ShareEnumeration"
RuleName = "KeepDollarShares" RuleName = "KeepDollarShares"
MatchAction = "Snaffle" MatchAction = "Snaffle"
Description = "Notifies the user that they can read C$ or ADMIN$ or something fun/noisy, but doesn't actually scan inside it." Description = "Notifies the user that C$ or ADMIN$ is visible on file share, but doesn't actually scan inside it."
MatchLocation = "ShareName" MatchLocation = "ShareName"
WordListType = "EndsWith" WordListType = "Exact"
MatchLength = 0 MatchLength = 0
WordList = ["\\\\C\\$", WordList = ["\\\\C\\$",
"\\\\ADMIN\\$", "ADMIN\\$", "C\\$"] "\\\\ADMIN\\$", "ADMIN\\$", "C\\$"]
snaffcore/classifier.py
+32 -26
View File
@@ -2,10 +2,11 @@ import re
import toml import toml
import os import os
import logging import logging
import pprint # import pprint
import termcolor
from impacket.smbconnection import SessionError, SMBConnection from impacket.smbconnection import SessionError, SMBConnection
from .file import *
log = logging.getLogger('snafflepy.classifier') log = logging.getLogger('snafflepy.classifier')
@@ -45,47 +46,42 @@ class Rules:
log.warning( log.warning(
f"{dict_rule['RuleName']} is invalid, please check your syntax!") f"{dict_rule['RuleName']} is invalid, please check your syntax!")
#pprint.pprint(self.share_classifiers)
# pprint.pprint(self.directory_classifiers)
# pprint.pprint(self.file_classifiers)
# pprint.pprint(self.contents_classifiers)
# pprint.pprint(self.postmatch_classifiers)
# TODO # TODO
def is_interest_file(file:RemoteFile, rules, smb_client) -> bool:
def is_interest_file(file, rules) -> bool: file.get(smb_client)
# massive_wordlist = prepare_classifiers() '''
# print(massive_wordlist) interest_names = []
# for root, dirs, files in os.walk(snafflepy_path, topdown=False):
# for name in files:
# with open(os.path.join(root, name), 'rb') as tfile:
# print(toml.loads(tfile))
interest_names = ["Creds.txt"]
if file.get_shortname() in interest_names: if file.get_shortname() in interest_names:
return True return True
else: else:
return False return False
'''
def is_interest_share(share, rules: Rules): def is_interest_share(share, rules: Rules):
regex_rules = []
# Tedium City to find match in wordlist. Did not prepare rules beforehand except by putting each MatchLocation in its own list # Tedium City to find match in wordlist. Did not prepare rules beforehand except by putting each MatchLocation in its own list
# so I have to do more work here before I can find the match # so I have to do more work here before I can find the match
for rule in rules.share_classifiers: for rule in rules.share_classifiers:
regex_rules = []
share_text = termcolor.colored("[Share]", 'yellow')
if rule['WordListType'] == "Regex": if rule['WordListType'] == "Regex":
regex_rules = rule['WordList'] regex_rules = rule['WordList']
for pattern in regex_rules: for pattern in regex_rules:
if re.search(str(pattern), str(share)) is not None: if re.search(str(pattern), str(share)) is not None:
log.info(f"{share} matched {rule['RuleName']}:{rule['Description']}") if rule['MatchAction'] == "Snaffle":
color = rule['Triage']
print(share_text, termcolor.colored(f"{{{rule['Triage']}}} {share} <{rule['RuleName']}>:<{rule['Description']}>",str(color).lower(), 'on_white'))
else:
log.debug(f"{rule['MatchAction']} {share} matched rule {rule['RuleName']}:{rule['Description']}")
elif rule['WordListType'] == "EndsWith": elif rule['WordListType'] == "EndsWith":
regex_rules = rule['WordList'] regex_rules = rule['WordList']
for pattern in regex_rules: for pattern in regex_rules:
if re.search(str(pattern + "$"), str(share)) is not None: if re.search(str(pattern + "$"), str(share)) is not None:
if rule['MatchAction'] == 'Snaffle': if rule['MatchAction'] == "Snaffle":
log.info(f"{share} matched rule {rule['RuleName']}:{rule['Description']}") color = rule['Triage']
print(share_text, termcolor.colored(f"{{{rule['Triage']}}} {share} <{rule['RuleName']}>:<{rule['Description']}>",str(color).lower(), 'on_white'))
else: else:
log.debug(f"{rule['MatchAction']} {share} matched rule {rule['RuleName']}:{rule['Description']}") log.debug(f"{rule['MatchAction']} {share} matched rule {rule['RuleName']}:{rule['Description']}")
@@ -93,19 +89,29 @@ def is_interest_share(share, rules: Rules):
regex_rules = rule['WordList'] regex_rules = rule['WordList']
for pattern in regex_rules: for pattern in regex_rules:
if re.search(str("^" + pattern), str(share)) is not None: if re.search(str("^" + pattern), str(share)) is not None:
log.warning(f"{share} matched rule {rule['RuleName']}: {rule['Description']}") color = rule['Triage']
print(share_text, termcolor.colored(f"{{{rule['Triage']}}} {share} <{rule['RuleName']}>:<{rule['Description']}>",str(color).lower(), 'on_white'))
elif rule['WordListType'] == "Contains": elif rule['WordListType'] == "Contains":
regex_rules = rule['WordList'] regex_rules = rule['WordList']
for pattern in regex_rules: for pattern in regex_rules:
if re.search(str(pattern), str(share)) is not None: if re.search(str(pattern), str(share)) is not None:
log.warning(f"{share} matched rule {rule['RuleName']}:{rule['Description']}") if rule['MatchAction'] == "Snaffle":
color = rule['Triage']
print(share_text, termcolor.colored(f"{{{rule['Triage']}}} {share} <{rule['RuleName']}>:<{rule['Description']}>",str(color).lower(), 'on_white'))
else:
log.debug(f"{rule['MatchAction']} {share} matched rule {rule['RuleName']}:{rule['Description']}")
elif rule['WordListType'] == "Exact": elif rule['WordListType'] == "Exact":
regex_rules = rule['WordList'] regex_rules = rule['WordList']
for pattern in regex_rules: for pattern in regex_rules:
if re.search(str("^" + pattern + "$"), str(share)) is not None: if re.search(str("^" + pattern + "$"), str(share)) is not None:
print(f"{share} matched {rule['RuleName']}:{rule['Description']}") if rule['MatchAction'] == "Snaffle":
color = rule['Triage']
print(share_text, termcolor.colored(f"{{{rule['Triage']}}} {share} <{rule['RuleName']}>:<{rule['Description']}>",str(color).lower(), 'on_white'))
else:
log.debug(f"{rule['MatchAction']} {share} matched rule {rule['RuleName']}:{rule['Description']}")
else: else:
log.warning(f"{rule['RuleName']} has an invalid WordListType - valid values are Regex, EndsWith, StartsWith, Contains, or Exact") log.warning(f"{rule['RuleName']} has an invalid WordListType - valid values are Regex, EndsWith, StartsWith, Contains, or Exact")
raise Exception("Invalid WordListType")
snaffcore/file.py
+13 -4
View File
@@ -1,7 +1,7 @@
import io
from .utilities import * from .utilities import *
from .errors import * from .errors import *
from pathlib import Path from pathlib import Path
import os
# RT: Stolen from manspider - https://github.com/blacklanternsecurity/MANSPIDER # RT: Stolen from manspider - https://github.com/blacklanternsecurity/MANSPIDER
@@ -20,9 +20,18 @@ class RemoteFile():
self.size = size self.size = size
self.smb_client = None self.smb_client = None
file_suffix = Path(name).suffix.lower() does_exist = os.path.exists("remotefiles")
self.tmp_filename = Path('/tmp/.snafflepy') / \ if not does_exist:
(random_string(15) + file_suffix) log.info("remotefiles directory not present, creating dir")
os.makedirs("remotefiles")
# file_suffix = Path(name).suffix.lower()
self.tmp_filename = Path('./remotefiles') / \
(self.name)
# self.tmp_filename = Path('/tmp/.snafflepy') / \
# (random_string(15) + file_suffix)
def get(self, smb_client=None): def get(self, smb_client=None):
''' '''
snaffcore/go_snaffle.py
+36 -49
View File
@@ -13,19 +13,14 @@ from .classifier import *
log = logging.getLogger('snafflepy') log = logging.getLogger('snafflepy')
def begin_snaffle(options): def begin_snaffle(options):
# Prepare classifiers for use in naive_classify()
snaff_rules = Rules() snaff_rules = Rules()
snaff_rules.prepare_classifiers() snaff_rules.prepare_classifiers()
# for dict_rules in prepped_rules:
# for actual_rule in dict_rules['ClassifierRules']:
# pprint.pprint(actual_rule['Triage'])
print("Beginning the snaffle...") print("Beginning the snaffle...")
sleep(0.2)
# Automatically get domain from target if not provided
if not options.domain: if not options.domain:
log.info("Domain not provided, retrieving automatically.") log.info("Domain not provided, retrieving automatically.")
s = Server(options.targets[0], get_info=ALL) s = Server(options.targets[0], get_info=ALL)
@@ -67,7 +62,7 @@ def begin_snaffle(options):
log.warning(f"Unable to add{target} to targets to snaffle") log.warning(f"Unable to add{target} to targets to snaffle")
continue continue
log.debug(f"Targets that will be snaffled: {options.targets}") # log.debug(f"Targets that will be snaffled: {options.targets}")
# Login via SMB # Login via SMB
# log.info("Preparing classifiers...") # log.info("Preparing classifiers...")
@@ -79,13 +74,15 @@ def begin_snaffle(options):
except: except:
log.error(f"Error logging in to SMB on {options.targets[0]}") log.error(f"Error logging in to SMB on {options.targets[0]}")
if options.go_loud: if options.go_loud:
log.warning("[GO LOUD ACTIVATED] Enumerating all shares for all files...") log.warning(
"[GO LOUD ACTIVATED] Enumerating all shares for all files...")
for target in options.targets: for target in options.targets:
try: try:
smb_client = SMBClient( smb_client = SMBClient(
target, options.username, options.password, options.domain, options.hash) target, options.username, options.password, options.domain, options.hash)
if not smb_client.login(): if not smb_client.login():
log.error(f" Unable to login to{target}") log.error(f" Unable to login to{target}")
continue
for share in smb_client.shares: for share in smb_client.shares:
try: try:
if not options.go_loud: if not options.go_loud:
@@ -96,12 +93,28 @@ def begin_snaffle(options):
files = smb_client.ls(share, "") files = smb_client.ls(share, "")
for file in files: for file in files:
# filelist.append(file) size = file.get_filesize()
# Ask do they want file sizes? name = file.get_longname()
file = RemoteFile(name, share, target, size)
if options.go_loud: if options.go_loud:
log.info(f"{target}: {share}\\{file.get_longname()}") # Dont care about empty files
if size == 0:
continue
try:
file.get(smb_client)
log.info(f"{target}: {share}\\{name}")
except FileRetrievalError:
log.debug(f"Unable to download ({target}\\\\{share}\\{name})")
else: else:
classify_file(share, file, snaff_rules) if size >= options.max_file_snaffle:
pass
else:
try:
classify_file(file, snaff_rules, smb_client)
except FileRetrievalError as e:
log.debug(f"{e}")
continue
except FileListError: except FileListError:
log.error( log.error(
@@ -109,7 +122,8 @@ def begin_snaffle(options):
continue continue
except Exception as e: except Exception as e:
log.error(f"Error creating SMBClient object, {e}") log.debug(f"{e}")
def access_ldap_server(ip, username, password): def access_ldap_server(ip, username, password):
log.info("Accessing LDAP Server") log.info("Accessing LDAP Server")
@@ -119,7 +133,8 @@ def access_ldap_server(ip, username, password):
# log.debug(server.schema) # log.debug(server.schema)
if not conn.bind(): if not conn.bind():
log.critical(f"Unable to bind to {server} as {username}, ") log.critical(f"Unable to bind to {server}")
return None
return conn return conn
except Exception as e: except Exception as e:
@@ -127,38 +142,36 @@ def access_ldap_server(ip, username, password):
log.info("Trying guest session... ") log.info("Trying guest session... ")
try: try:
conn = Connection(server, username='Guest', password='') conn = Connection(server, user='Guest', password='')
if not conn.bind(): if not conn.bind():
log.critical(f"Unable to bind to {server} as {username}") log.critical(f"Unable to bind to {server} as {username}")
return None
return conn return conn
except Exception as e: except Exception as e:
log.critical(f'Error logging in to {ip}, as {username}') log.critical(f'Error logging in to {ip}, as {username}')
log.info("Trying null session... ") log.info("Trying null session... ")
conn = Connection(server, username='', password='') conn = Connection(server, user='', password='')
if not conn.bind(): if not conn.bind():
log.critical(f"Unable to bind to {server} as {username}") log.critical(f"Unable to bind to {server}")
return None return None
return conn return conn
# 2nd snaffle step, finding additional targets from original target via LDAP queries # 2nd snaffle step, finding additional targets from original target via LDAP queries
def list_computers(connection: Connection, domain): def list_computers(connection: Connection, domain):
dn = get_domain_dn(domain) dn = get_domain_dn(domain)
# filter = "(objectCategory=computer)" # filter = "(objectCategory=computer)"
if connection is None: if connection is None:
log.critical("Connection is not established") log.critical("Connection is not established")
sys.exit(2)
try: try:
connection.search(search_base=dn, search_filter='(&(objectCategory=Computer)(name=*))', connection.search(search_base=dn, search_filter='(&(objectCategory=Computer)(name=*))',
search_scope=SUBTREE, attributes=['dNSHostName'], paged_size=500) search_scope=SUBTREE, attributes=['dNSHostName'], paged_size=500)
# log.debug(connection.entries)
# connection.search(search_base=dn,search_filter=filter,search_scope=SUBTREE,attributes=ALL_ATTRIBUTES)
domain_names = [] domain_names = []
# log.debug(connection.entries)
for entry in connection.entries: for entry in connection.entries:
sep = str(entry).strip().split(':') sep = str(entry).strip().split(':')
domain_names.append(sep[6]) domain_names.append(sep[6])
@@ -172,35 +185,9 @@ def list_computers(connection: Connection, domain):
# TODO # TODO
def classify_file(share, file, rules: Rules): def classify_file(file: RemoteFile, rules: Rules, smb_client: SMBClient):
# log.info(f"{share}: {file.get_longname()}") is_interest_file(file, rules, smb_client)
if is_interest_file(file, rules):
log.info(f"Found interesting file: {share}/{file}")
def classify_share(share, rules: Rules): def classify_share(share, rules: Rules):
is_interest_share(share, rules) is_interest_share(share, rules)
# These functions resolve to public IP Address:
'''
def resolve(nameserver, host_fqdn):
resolver = dns.resolver.Resolver()
resolver.nameservers = [nameserver]
answer = resolver.query(host_fqdn, "A")
return answer
def get_ip(target):
try:
print(socket.gethostbyname(target))
except socket.gaierror:
parsed_url = urllib.parse.urlparse(target)
hostname = parsed_url.hostname
try:
answers = dns.resolver.query(hostname, 'A')
for rdata in answers:
print(rdata.address)
except dns.resolver.NXDOMAIN:
print('ip not found')
'''
snaffcore/smb.py
+3 -3
View File
@@ -78,7 +78,7 @@ class SMBClient:
assert False assert False
log.debug( log.debug(
f'{self.server}: Authenticating as "{self.domain}\\{self.username}"') f'{self.server}: Authenticating as "{self.username}"')
# pass the hash if requested # pass the hash if requested
if self.nthash and not self.password: if self.nthash and not self.password:
@@ -117,14 +117,14 @@ class SMBClient:
log.warning( log.warning(
f'{self.server}: {s}: {self.username}') f'{self.server}: {s}: {self.username}')
log.warning(f'{self.server}: Trying guest session') log.debug(f'{self.server}: Trying guest session')
self.username = 'Guest' self.username = 'Guest'
self.password = '' self.password = ''
self.domain = '' self.domain = ''
self.nthash = '' self.nthash = ''
guest_success = self.login(refresh=True, first_try=False) guest_success = self.login(refresh=True, first_try=False)
if not guest_success: if not guest_success:
log.warning(f'{self.server}: Switching to null session') log.debug(f'{self.server}: Switching to null session')
self.username = '' self.username = ''
self.login(refresh=True, first_try=False) self.login(refresh=True, first_try=False)
snaffler.py
+3 -1
View File
@@ -1,6 +1,7 @@
import argparse import argparse
import sys import sys
import logging import logging
import termcolor
from snaffcore.go_snaffle import * from snaffcore.go_snaffle import *
from snaffcore.utilities import * from snaffcore.utilities import *
@@ -31,8 +32,8 @@ def parse_arguments():
action='store_true', help="Show more info") action='store_true', help="Show more info")
parser.add_argument("--go-loud", action='store_true', parser.add_argument("--go-loud", action='store_true',
help="Don't try to find anything interesting, literally just go through every computer and every share and print out as many files as possible. Use at your own risk") help="Don't try to find anything interesting, literally just go through every computer and every share and print out as many files as possible. Use at your own risk")
# parser.add_argument("-e", "--exclude", )
parser.add_argument("-m", "--max-file-snaffle", metavar="size", type=int, default=10000, help="Max filesize to snaffle in bytes (any files over this size will be dropped)")
# TODO # TODO
parser.add_argument("-i", "--no-share-discovery", action='store_true', parser.add_argument("-i", "--no-share-discovery", action='store_true',
help="Disables share discovery (more stealthy)") help="Disables share discovery (more stealthy)")
@@ -90,6 +91,7 @@ def main():
print("\nI snaffled 'til the snafflin was done") print("\nI snaffled 'til the snafflin was done")
print("View log file at ~/.snafflepy/logs/") print("View log file at ~/.snafflepy/logs/")
print("Files snaffled from targets are available in <PATH-TO-SNAFFLEPY>/remotefiles/")
sys.exit(1) sys.exit(1)