mirror of
https://github.com/cisagov/snafflepy
synced 2026-09-24 18:22:23 +00:00
Merge pull request #6 from robert-todora/dev
Stole more code from manspider, project now will print out shares tha…
This commit is contained in:
Robert Todora
@@ -0,0 +1,6 @@
|
||||
# created by virtualenv automatically
|
||||
snaffcore/__pycache__/
|
||||
active_directory/
|
||||
.idea
|
||||
.vscode
|
||||
snafflepy/
|
||||
Generated
-3
@@ -1,3 +0,0 @@
|
||||
# Default ignored files
|
||||
/shelf/
|
||||
/workspace.xml
|
||||
@@ -0,0 +1,51 @@
|
||||
import io
|
||||
from .utilities import *
|
||||
from .errors import *
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
class RemoteFile():
|
||||
'''
|
||||
Represents a file on an SMB share
|
||||
Passed from a spiderling up to its parent spide
|
||||
r '''
|
||||
|
||||
def __init__(self, name, share, target, size=0):
|
||||
|
||||
self.share = share
|
||||
self.target = target
|
||||
self.name = name
|
||||
self.size = size
|
||||
self.smb_client = None
|
||||
|
||||
file_suffix = Path(name).suffix.lower()
|
||||
self.tmp_filename = Path('/tmp/.snafflepy') / (random_string(15) + file_suffix)
|
||||
|
||||
|
||||
def get(self, smb_client=None):
|
||||
'''
|
||||
Downloads file to self.tmp_filename
|
||||
|
||||
NOTE: SMBConnection() can't be passed through a multiprocessing queue
|
||||
This means that smb_client must be set after the file arrives at Spider()
|
||||
'''
|
||||
|
||||
if smb_client is None and self.smb_client is None:
|
||||
raise FileRetrievalError('Please specify smb_client')
|
||||
|
||||
#memfile = io.BytesIO()
|
||||
with open(str(self.tmp_filename), 'wb') as f:
|
||||
|
||||
try:
|
||||
smb_client.conn.getFile(self.share, self.name, f.write)
|
||||
except Exception as e:
|
||||
handle_impacket_error(e, smb_client, self.share, self.name)
|
||||
raise FileRetrievalError(f'Error retrieving file "{str(self)}": {str(e)[:150]}')
|
||||
|
||||
# reset cursor back to zero so .read() will return the whole file
|
||||
#memfile.seek(0)
|
||||
|
||||
|
||||
def __str__(self):
|
||||
|
||||
return f'{self.target}\\{self.share}\\{self.name}'
|
||||
+20
-14
@@ -3,6 +3,8 @@ from ldap3 import *
|
||||
from time import sleep
|
||||
from .smb import *
|
||||
|
||||
log = logging.getLogger('snafflepy')
|
||||
|
||||
def begin_snaffle(options):
|
||||
print("Beginning the snaffle...")
|
||||
sleep(0.2)
|
||||
@@ -25,21 +27,25 @@ def begin_snaffle(options):
|
||||
sys.exit(2)
|
||||
else:
|
||||
|
||||
# Login through LDAP
|
||||
server = Server(options.dcip, get_info=ALL)
|
||||
conn = Connection(server)
|
||||
# Login through LDAP
|
||||
#server = Server(options.targets[0], get_info=ALL)
|
||||
#conn = Connection(server)
|
||||
|
||||
if not conn.bind():
|
||||
print("Error connecting to domain")
|
||||
print(conn.result)
|
||||
sys.exit(2)
|
||||
# if not conn.bind():
|
||||
# print("Error connecting to domain")
|
||||
# print(conn.result)
|
||||
# sys.exit(2)
|
||||
|
||||
print(server.info)
|
||||
#print(server.info)
|
||||
|
||||
# Login via SMB
|
||||
smb_client = SMBClient(options.targets, options.username, options.password, options.domain)
|
||||
|
||||
logon_result = smb_client.login()
|
||||
|
||||
print(logon_result)
|
||||
# Login via SMB
|
||||
for target in options.targets:
|
||||
try:
|
||||
smb_client = SMBClient(target, options.username, options.password, options.domain, options.hash)
|
||||
smb_client.login()
|
||||
|
||||
for share in smb_client.shares:
|
||||
print(share)
|
||||
|
||||
except Exception as e:
|
||||
print("Exception: ", e)
|
||||
@@ -0,0 +1,100 @@
|
||||
import logging
|
||||
from copy import copy
|
||||
from sys import stdout
|
||||
from pathlib import Path
|
||||
from datetime import datetime
|
||||
from multiprocessing import Queue
|
||||
from logging.handlers import QueueHandler, QueueListener
|
||||
|
||||
|
||||
### PRETTY COLORS ###
|
||||
|
||||
|
||||
class ColoredFormatter(logging.Formatter):
|
||||
|
||||
color_mapping = {
|
||||
'DEBUG': 69, # blue
|
||||
'INFO': 118, # green
|
||||
'WARNING': 208, # orange
|
||||
'ERROR': 196, # red
|
||||
'CRITICAL': 196, # red
|
||||
}
|
||||
|
||||
char_mapping = {
|
||||
'DEBUG': '*',
|
||||
'INFO': '+',
|
||||
'WARNING': '-',
|
||||
'ERROR': '!',
|
||||
'CRITICAL': '!!!',
|
||||
}
|
||||
|
||||
prefix = '\033[1;38;5;'
|
||||
suffix = '\033[0m'
|
||||
|
||||
def __init__(self, pattern):
|
||||
|
||||
super().__init__(pattern)
|
||||
|
||||
|
||||
def format(self, record):
|
||||
|
||||
colored_record = copy(record)
|
||||
levelname = colored_record.levelname
|
||||
levelchar = self.char_mapping.get(levelname, '+')
|
||||
seq = self.color_mapping.get(levelname, 15) # default white
|
||||
colored_levelname = f'{self.prefix}{seq}m[{levelchar}]{self.suffix}'
|
||||
colored_record.levelname = colored_levelname
|
||||
|
||||
return logging.Formatter.format(self, colored_record)
|
||||
|
||||
|
||||
@classmethod
|
||||
def green(cls, s):
|
||||
|
||||
return cls.color(s)
|
||||
|
||||
|
||||
@classmethod
|
||||
def red(cls, s):
|
||||
|
||||
return cls.color(s, level='ERROR')
|
||||
|
||||
|
||||
@classmethod
|
||||
def color(cls, s, level='INFO'):
|
||||
|
||||
color = cls.color_mapping.get(level)
|
||||
return f'{cls.prefix}{color}m{s}{cls.suffix}'
|
||||
|
||||
|
||||
|
||||
class CustomQueueListener(QueueListener):
|
||||
'''
|
||||
Ignore errors in the monitor thread that result from a race condition when the program exits
|
||||
'''
|
||||
def _monitor(self):
|
||||
try:
|
||||
super()._monitor()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
### LOG TO STDERR ###
|
||||
|
||||
console = logging.StreamHandler(stdout)
|
||||
# tell the handler to use this format
|
||||
console.setFormatter(ColoredFormatter('%(levelname)s %(message)s'))
|
||||
|
||||
### LOG TO FILE ###
|
||||
|
||||
log_queue = Queue()
|
||||
listener = CustomQueueListener(log_queue, console)
|
||||
sender = QueueHandler(log_queue)
|
||||
logging.getLogger('snafflepy').handlers = [sender]
|
||||
|
||||
logdir = Path.home() / '.snafflepy' / 'logs'
|
||||
logdir.mkdir(parents=True, exist_ok=True)
|
||||
logfile = f'snafflepy_{datetime.now().strftime("%m-%d-%Y")}.log'
|
||||
handler = logging.FileHandler(str(logdir / logfile))
|
||||
handler.setFormatter(logging.Formatter('%(asctime)s %(levelname)s %(message)s'))
|
||||
logging.getLogger('snafflepy').addHandler(handler)
|
||||
+3
-2
@@ -39,7 +39,7 @@ class SMBClient:
|
||||
resp = self.conn.listShares()
|
||||
for i in range(len(resp)):
|
||||
sharename = resp[i]['shi1_netname'][:-1]
|
||||
log.debug(f'{self.server}: Found share: {sharename}')
|
||||
log.info(f'{self.server}: Found share: {sharename}')
|
||||
yield sharename
|
||||
|
||||
except Exception as e:
|
||||
@@ -59,8 +59,9 @@ class SMBClient:
|
||||
if self.conn is None or refresh:
|
||||
try:
|
||||
self.conn = SMBConnection(self.server, self.server, sess_port=445, timeout=20)
|
||||
#print("Here: ", self.conn)
|
||||
except Exception as e:
|
||||
log.debug(impacket_error(e))
|
||||
print("Exception: ", impacket_error(e))
|
||||
return None
|
||||
|
||||
try:
|
||||
|
||||
@@ -0,0 +1,141 @@
|
||||
import os
|
||||
# import magic
|
||||
import string
|
||||
import random
|
||||
import logging
|
||||
import ipaddress
|
||||
from pathlib import Path
|
||||
|
||||
# Stolen from https://github.com/blacklanternsecurity/MANSPIDER
|
||||
|
||||
log = logging.getLogger('snafflerpy.util')
|
||||
|
||||
|
||||
def str_to_list(s):
|
||||
|
||||
l = set()
|
||||
# try to open as file
|
||||
try:
|
||||
with open(s) as f:
|
||||
lines = set([l.strip() for l in f.readlines()])
|
||||
for line in lines:
|
||||
if line:
|
||||
l.add(line)
|
||||
except OSError:
|
||||
l.add(s)
|
||||
|
||||
return list(l)
|
||||
|
||||
|
||||
def make_targets(s):
|
||||
'''
|
||||
Accepts filename, CIDR, IP, hostname, file, or folder
|
||||
Returns list of targets as IPs, hostnames, or Path() objects
|
||||
'''
|
||||
|
||||
targets = set()
|
||||
|
||||
p = Path(s)
|
||||
if p.is_dir():
|
||||
targets.add(p)
|
||||
|
||||
else:
|
||||
for i in str_to_list(s):
|
||||
try:
|
||||
for ip in ipaddress.ip_network(i, strict=False):
|
||||
targets.add(str(ip))
|
||||
except ValueError:
|
||||
targets.add(i)
|
||||
|
||||
return list(targets)
|
||||
|
||||
|
||||
def human_to_int(h):
|
||||
'''
|
||||
converts human-readable number to integer
|
||||
e.g. 1K --> 1000
|
||||
'''
|
||||
|
||||
if type(h) == int:
|
||||
return h
|
||||
|
||||
units = {'': 1, 'K': 1024, 'M': 1024**2, 'G': 1024**3, 'T': 1024**4}
|
||||
|
||||
try:
|
||||
h = h.upper().strip()
|
||||
i = float(''.join(c for c in h if c in string.digits + '.'))
|
||||
unit = ''.join([c for c in h if c in units.keys()])
|
||||
except (ValueError, KeyError):
|
||||
raise ValueError(f'Invalid filesize "{h}"')
|
||||
|
||||
return int(i * units[unit])
|
||||
|
||||
|
||||
def bytes_to_human(_bytes):
|
||||
'''
|
||||
converts bytes to human-readable filesize
|
||||
e.g. 1024 --> 1KB
|
||||
'''
|
||||
|
||||
sizes = ['B', 'KB', 'MB', 'GB', 'TB', 'PB', 'EB', 'ZB']
|
||||
units = {}
|
||||
count = 0
|
||||
for size in sizes:
|
||||
units[size] = pow(1024, count)
|
||||
count +=1
|
||||
|
||||
for size in sizes:
|
||||
if abs(_bytes) < 1024.0:
|
||||
if size == sizes[0]:
|
||||
_bytes = str(int(_bytes))
|
||||
else:
|
||||
_bytes = '{:.2f}'.format(_bytes)
|
||||
return '{}{}'.format(_bytes, size)
|
||||
_bytes /= 1024
|
||||
|
||||
raise ValueError
|
||||
|
||||
'''
|
||||
def better_decode(b):
|
||||
|
||||
# detect encoding with libmagic
|
||||
m = magic.Magic(mime_encoding=True)
|
||||
encoding = m.from_buffer(b)
|
||||
|
||||
try:
|
||||
return b.decode(encoding)
|
||||
except Exception:
|
||||
return str(b)[2:-1]
|
||||
'''
|
||||
|
||||
def random_string(length):
|
||||
|
||||
return ''.join(random.choice(string.ascii_lowercase + string.ascii_uppercase + string.digits) for i in range(length))
|
||||
|
||||
|
||||
def list_files(path):
|
||||
|
||||
path = Path(path)
|
||||
|
||||
if path.is_file() and not path.is_symlink():
|
||||
yield path
|
||||
|
||||
elif path.is_dir():
|
||||
for dir_name, dirnames, filenames in os.walk(path):
|
||||
for file in filenames:
|
||||
file = Path(dir_name) / file
|
||||
if file.is_file() and not file.is_symlink():
|
||||
yield file
|
||||
|
||||
|
||||
def rmdir(directory):
|
||||
'''
|
||||
Recursively remove directory
|
||||
'''
|
||||
directory = Path(directory)
|
||||
for item in directory.iterdir():
|
||||
if item.is_dir():
|
||||
rmdir(item)
|
||||
else:
|
||||
item.unlink()
|
||||
directory.rmdir()
|
||||
+19
-10
@@ -3,26 +3,27 @@ import sys
|
||||
import logging
|
||||
from snaffcore.go_snaffle import *
|
||||
from snaffcore.utilities import *
|
||||
from snaffcore.logger import *
|
||||
|
||||
log = logging.getLogger('snafflepy')
|
||||
log.setLevel(logging.INFO)
|
||||
log.setLevel(logging.DEBUG)
|
||||
|
||||
def parse_arguments():
|
||||
syntax_error = False
|
||||
print("SnafflePy by @robert-todora")
|
||||
|
||||
parser = argparse.ArgumentParser(add_help=True, prog='snafflepy', description='A "port" of Snaffler in python')
|
||||
parser.add_argument("targets", nargs='+',type=make_targets,required=True, help="IPs, hostnames, CIDR ranges, or files contains targets to snaffle")
|
||||
parser = argparse.ArgumentParser(add_help=True, prog='snaffler.py', description='A "port" of Snaffler in python')
|
||||
parser.add_argument("targets", nargs='+',type=make_targets, help="IPs, hostnames, CIDR ranges, or files contains targets to snaffle")
|
||||
parser.add_argument("-u","--username", metavar='username',type=str, help="domain username")
|
||||
parser.add_argument("-p","--password", metavar='password',type=str, help="password for domain user")
|
||||
#parser.add_argument('--dcip', metavar='[IP addr]', help="IP address of domain controller")
|
||||
parser.add_argument("-d", "--domain", metavar='domain', default="", help="FQDN domain to authenticate to")
|
||||
parser.add_argument("-H", "--hash", metavar='hash', default="", help="NT hash for authentication")
|
||||
parser.add_argument("-v", "--verbose", action='store_true', help="Show more info")
|
||||
parser.add_argument("--test", metavar='test', type=bool, default=False, help="switch to testing mode")
|
||||
parser.add_argument("-f", "--file", help="path to file with list of targets")
|
||||
options = parser.parse_args()
|
||||
|
||||
|
||||
try:
|
||||
if len(sys.argv) == 1:
|
||||
if len(sys.argv) <= 1:
|
||||
parser.print_help()
|
||||
sys.exit(1)
|
||||
|
||||
@@ -36,8 +37,16 @@ def parse_arguments():
|
||||
parser.print_help()
|
||||
sys.exit(2)
|
||||
else:
|
||||
return options
|
||||
|
||||
options = parser.parse_args()
|
||||
|
||||
if options.verbose:
|
||||
log.setLevel('DEBUG')
|
||||
|
||||
targets = set()
|
||||
[[targets.add(t) for t in g] for g in options.targets]
|
||||
options.targets = list(targets)
|
||||
|
||||
return options
|
||||
|
||||
def print_banner():
|
||||
print(r'''
|
||||
@@ -59,7 +68,7 @@ def main():
|
||||
begin_snaffle(snaffle_options)
|
||||
|
||||
|
||||
print("I snaffled 'til the snafflin was done")
|
||||
print("\nI snaffled 'til the snafflin was done")
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
||||
Reference in New Issue
Block a user