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

added file download capability

Browse Source
This commit is contained in:
robert-todora
2023-07-25 13:56:38 -04:00
parent 389591ac5b
commit 0602a8b4dc
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"
RuleName = "KeepDollarShares"
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"
WordListType = "EndsWith"
WordListType = "Exact"
MatchLength = 0
WordList = ["\\\\C\\$",
"\\\\ADMIN\\$", "ADMIN\\$", "C\\$"]
snaffcore/classifier.py
+32 -26
View File
@@ -2,10 +2,11 @@ import re
import toml
import os
import logging
import pprint
# import pprint
import termcolor
from impacket.smbconnection import SessionError, SMBConnection
from .file import *
log = logging.getLogger('snafflepy.classifier')
@@ -45,47 +46,42 @@ class Rules:
log.warning(
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
def is_interest_file(file, rules) -> bool:
# massive_wordlist = prepare_classifiers()
# print(massive_wordlist)
# 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"]
def is_interest_file(file:RemoteFile, rules, smb_client) -> bool:
file.get(smb_client)
'''
interest_names = []
if file.get_shortname() in interest_names:
return True
else:
return False
'''
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
# so I have to do more work here before I can find the match
for rule in rules.share_classifiers:
regex_rules = []
share_text = termcolor.colored("[Share]", 'yellow')
if rule['WordListType'] == "Regex":
regex_rules = rule['WordList']
for pattern in regex_rules:
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":
regex_rules = rule['WordList']
for pattern in regex_rules:
if re.search(str(pattern + "$"), str(share)) is not None:
if rule['MatchAction'] == 'Snaffle':
log.info(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']}")
@@ -93,19 +89,29 @@ def is_interest_share(share, rules: Rules):
regex_rules = rule['WordList']
for pattern in regex_rules:
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":
regex_rules = rule['WordList']
for pattern in regex_rules:
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":
regex_rules = rule['WordList']
for pattern in regex_rules:
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:
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 .errors import *
from pathlib import Path
import os
# RT: Stolen from manspider - https://github.com/blacklanternsecurity/MANSPIDER
@@ -20,9 +20,18 @@ class RemoteFile():
self.size = size
self.smb_client = None
file_suffix = Path(name).suffix.lower()
self.tmp_filename = Path('/tmp/.snafflepy') / \
(random_string(15) + file_suffix)
does_exist = os.path.exists("remotefiles")
if not does_exist:
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):
'''
snaffcore/go_snaffle.py
+36 -49
View File
@@ -13,19 +13,14 @@ from .classifier import *
log = logging.getLogger('snafflepy')
def begin_snaffle(options):
# Prepare classifiers for use in naive_classify()
snaff_rules = Rules()
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...")
sleep(0.2)
# Automatically get domain from target if not provided
if not options.domain:
log.info("Domain not provided, retrieving automatically.")
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")
continue
log.debug(f"Targets that will be snaffled: {options.targets}")
# log.debug(f"Targets that will be snaffled: {options.targets}")
# Login via SMB
# log.info("Preparing classifiers...")
@@ -79,13 +74,15 @@ def begin_snaffle(options):
except:
log.error(f"Error logging in to SMB on {options.targets[0]}")
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:
try:
smb_client = SMBClient(
target, options.username, options.password, options.domain, options.hash)
if not smb_client.login():
log.error(f" Unable to login to{target}")
continue
for share in smb_client.shares:
try:
if not options.go_loud:
@@ -96,12 +93,28 @@ def begin_snaffle(options):
files = smb_client.ls(share, "")
for file in files:
# filelist.append(file)
# Ask do they want file sizes?
size = file.get_filesize()
name = file.get_longname()
file = RemoteFile(name, share, target, size)
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:
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:
log.error(
@@ -109,7 +122,8 @@ def begin_snaffle(options):
continue
except Exception as e:
log.error(f"Error creating SMBClient object, {e}")
log.debug(f"{e}")
def access_ldap_server(ip, username, password):
log.info("Accessing LDAP Server")
@@ -119,7 +133,8 @@ def access_ldap_server(ip, username, password):
# log.debug(server.schema)
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
except Exception as e:
@@ -127,38 +142,36 @@ def access_ldap_server(ip, username, password):
log.info("Trying guest session... ")
try:
conn = Connection(server, username='Guest', password='')
conn = Connection(server, user='Guest', password='')
if not conn.bind():
log.critical(f"Unable to bind to {server} as {username}")
return None
return conn
except Exception as e:
log.critical(f'Error logging in to {ip}, as {username}')
log.info("Trying null session... ")
conn = Connection(server, username='', password='')
conn = Connection(server, user='', password='')
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
# 2nd snaffle step, finding additional targets from original target via LDAP queries
def list_computers(connection: Connection, domain):
dn = get_domain_dn(domain)
# filter = "(objectCategory=computer)"
if connection is None:
log.critical("Connection is not established")
sys.exit(2)
try:
connection.search(search_base=dn, search_filter='(&(objectCategory=Computer)(name=*))',
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 = []
# log.debug(connection.entries)
for entry in connection.entries:
sep = str(entry).strip().split(':')
domain_names.append(sep[6])
@@ -172,35 +185,9 @@ def list_computers(connection: Connection, domain):
# TODO
def classify_file(share, file, rules: Rules):
# log.info(f"{share}: {file.get_longname()}")
if is_interest_file(file, rules):
log.info(f"Found interesting file: {share}/{file}")
def classify_file(file: RemoteFile, rules: Rules, smb_client: SMBClient):
is_interest_file(file, rules, smb_client)
def classify_share(share, rules: 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
log.debug(
f'{self.server}: Authenticating as "{self.domain}\\{self.username}"')
f'{self.server}: Authenticating as "{self.username}"')
# pass the hash if requested
if self.nthash and not self.password:
@@ -117,14 +117,14 @@ class SMBClient:
log.warning(
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.password = ''
self.domain = ''
self.nthash = ''
guest_success = self.login(refresh=True, first_try=False)
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.login(refresh=True, first_try=False)
snaffler.py
+3 -1
View File
@@ -1,6 +1,7 @@
import argparse
import sys
import logging
import termcolor
from snaffcore.go_snaffle import *
from snaffcore.utilities import *
@@ -31,8 +32,8 @@ def parse_arguments():
action='store_true', help="Show more info")
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")
# 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
parser.add_argument("-i", "--no-share-discovery", action='store_true',
help="Disables share discovery (more stealthy)")
@@ -90,6 +91,7 @@ def main():
print("\nI snaffled 'til the snafflin was done")
print("View log file at ~/.snafflepy/logs/")
print("Files snaffled from targets are available in <PATH-TO-SNAFFLEPY>/remotefiles/")
sys.exit(1)