runpentests.py

#!/usr/bin/env python3

import socket
import os,sys
import cameras.dlink_dcs932
import base64
import subprocess
import paramiko
import argparse
from boofuzz import *
from devices import wd_mycloud_tests
from devices import dcs930
from openvas_lib import VulnscanManager, VulnscanException
from openvas_lib import report_parser
from threading import Semaphore
from functools import partial
from xml.etree import ElementTree


#Global Vars
HTTP_PORT = 80
DEFAULT_PATH = '/'
HTTP_STATUS_OK = '200 OK';

#TODO: nmap scan, parse results for ports for scanning
#TODO: look for device name in nmap output
#nmap_scan(ip_addr)
#performs a nmap scan of a provided ip address, calls the known tests if the
#device is fingerprinted
#return values:
#known: if nmap can determine likely device return true

def my_print_status(i):
    print(str(i))

def write_openvas_report(manager, report_id, ip):
	result_dir = os.path.dirname(os.path.abspath(__file__)) + "/results"
	if not os.path.exists(result_dir):
		os.makedirs(result_dir)

	try:
		report = manager.get_report_xml(report_id)
	except Exception as e:
		print(e)
		return
	else:
		fout_path = result_dir + "/xml/"
		if not os.path.exists(fout_path):
			os.makedirs(fout_path)

		fout = open(fout_path + ip + ".xml", "wb")
		fout.write(ElementTree.tostring(report, encoding='utf-8', method='xml'))
		fout.close()

	try:
		report = manager.get_report_html(report_id)
	except Exception as e:
		print(e)
		return
	else:
		fout_path = result_dir + "/html/"
		if not os.path.exists(fout_path):
			os.makedirs(fout_path)

		html_text = report.find("report").text
		if not html_text:
			html_text = report.find("report").find("report_format").tail

		fout = open(fout_path + ip + ".html", "wb")
		fout.write(base64.b64decode(html_text))
                fout.close()


def openvas_scan(ip):
    print("running OpanVAS scan")
    try:
        sem = Semaphore(0)
        scanner = VulnscanManager("localhost", "admin", "ddb246a9-2c6f-4100-a7e6-b91793b81e58")
        scan_id, target_id = scanner.launch_scan(target = ip, profile = "Full and fast", callback_end = partial(lambda x: x.release(), sem), callback_progress = my_print_status)
        print("launched")
        sem.acquire()
        print("finished")
        report_id = scanner.get_report_id(scan_id)
        write_openvas_report(scanner, report_id, ip)
    except VulnscanException as e:
        print("Error:")
        print(e)

def nmap_scan(ip_addr):
    print("running nmap scan")
    nmap_res = subprocess.check_output("nmap -A {}".format(ip_addr), shell=True)
    ftp = ("21/" in nmap_res)
    ssh = ("22/" in nmap_res)
    http = ("80/" in nmap_res)
    https = ("443/" in nmap_res)
    smb = ("445/" in nmap_res)
    smb135 = ("135/" in nmap_res)
    filename = "results/{}/{}-nmap.txt".format(ip_addr, ip_addr)
    f = open(filename, "w+")
    f.write(nmap_res)
    f.close()
    print(nmap_res)
    #check for known devices
    if("WDMyCloud" in nmap_res):
        print("*****Potential Device Identified******")
        print("Possible Device: WDMyCloud")
        wd_mycloud_tests.test_wd_mycloud(ip_addr)
        return True, ftp, ssh, http, https, smb, smb135
    elif("DCS-930LB1" in nmap_res):
        print("*****Potential Device Identified******")
        print("Possible Device: DCS-930LB1")
        #TODO: device specific tests

        return True, ftp, ssh, http, https, smb, smb135
    elif ("Belkin Wemo" in nmap_res):
        print("*****Potential Device Identified******")
        print("Possible Device: Belkin Wemo")
        # TODO: device specific tests

        return True, ftp, ssh, http, https, smb, smb135
    else:
        print("results")
        print(nmap_res)


    return False, ftp, ssh, http, https, smb, smb135

def nikto_scan(ip_addr):
    print("Running nikto scan")
    nikto_res = subprocess.check_output("nikto -host http://{}".format(ip_addr), shell=True)
    #os.system("nikto -host http://{}".format(ip_addr))
    f = open("results/{}/{}-nikto.txt".format(ip_addr, ip_addr), "w+")
    f.write(nikto_res)
    f.close()
    print("***Scan complete***")


# Generic method to test a username & password combo against HTTP Basic Auth.
# Can be used to test default passwords/weak passwords, etc.
# TODO: Add more common Auth methods (ex. Digest Authentication, OAuth 1 Auth)
def test_basic_auth(ip_addr, username, password):
    test_failure = False
    token = base64.encodestring('%s:%s' % (username, password)).strip()
    lines = [
            'GET %s HTTP/1.1' % DEFAULT_PATH,
            'Host: %s' % ip_addr,
            'Authorization: Basic %s' % token,
            'Connection: close',
    ]
    try:
        s = socket.socket()
        s.connect((ip_addr, 80))
        f = s.makefile('rwb', bufsize=0)
        f.write('\r\n'.join(lines)+'\r\n\r\n')
        response = f.read()
        if(HTTP_STATUS_OK in response):
            test_failure = True;
        f.close()
        s.close()
    except:
        return test_failure
    return test_failure

def fuzz_device(ip_addr, port):
    print("Fuzzing on port %s\n" % port)
    #filename = "results
    f = open('results/{}/fuzz.csv'.format(ip_addr), 'w+') #fuzzing has a huge amount of outputi
    loggers = [FuzzLoggerCsv(file_handle=f)]
    logger=FuzzLogger(loggers)
    session = Session(
            target=Target(
                connection=SocketConnection(ip_addr, port, proto='tcp')
            ),
            fuzz_loggers=[logger]
    )

    s_initialize(name="Request")
    with s_block("Request-Line"):
        s_group("Method", ['GET', 'HEAD', 'POST', 'PUT', 'DELETE', 'CONNECT', 'OPTIONS', 'TRACE'])
        s_delim(" ", name='space-1')
        s_string("/", name='Request-URI')
        s_delim(" ", name='space-2')
        s_string('HTTP/1.1', name='HTTP-Version')
        s_static("\r\n", name="Request-Line-CRLF")
    s_static("\r\n", "Request-CRLF")

    session.connect(s_get("Request"))
    
    session.fuzz()

# test_ssh: tests to see if a login can suceed using given creds
# return value: true/false
def test_ssh(ip_addr, username, pw):
    try:
        ssh = paramiko.SSHClient()
        ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
        ssh.connect(hostname=ip_addr, username=username, password=pw)
        ssh.close()
        return True

    except:
        return False

# ssh_test_default_passwords: tests a list of default creds against ssh login
# return n/a
def ssh_test_passwords(ip_addr):
    print("Checking default credentials for SSH...\n")
    with open("defaultpasswords.txt") as defaultPWfile:
        for line in defaultPWfile:
            username, pw = line.strip().split(':')
            if(test_ssh(ip_addr, username, pw)):
                print ('Failure: Login succeeded with default user:pw \'%s\':\'%s\'\n' % (username, pw))
    print("Checking for weak passwords...\n")
    with open("weakpasswords.txt") as weakPWfile:
        for line in weakPWfile:
            username, pw = line.strip().split(':')
            if (test_ssh(ip_addr, username, pw)):
                print 'Failure: Login succeeded with weak user:pw \'%s\':\'%s\'\n' % (username, pw)

def test_default_passwords(ip_addr):
    print("Checking default credentials...\n")
    with open("defaultpasswords.txt") as defaultPWfile:
        for line in defaultPWfile: 
            username, pw = line.strip().split(':')
            if(test_basic_auth(ip_addr, username, pw)):
                print ('Failure: Login succeeded with default user:pw \'%s\':\'%s\'\n' % (username, pw))

def test_weak_passwords(ip_addr):
    print("Checking for weak passwords...\n")
    with open("weakpasswords.txt") as weakPWfile:
        for line in weakPWfile:
            username, pw = line.strip().split(':')    
            if(test_basic_auth(ip_addr, username, pw)):
                print 'Failure: Login succeeded with weak user:pw \'%s\':\'%s\'\n' % (username, pw)

def known():
    ip_addr = raw_input("Enter the device IP:\n")
    print("You entered {}".format(ip_addr))
    print("Scan web interface for vulnerabilities")
    nikto_scan(ip_addr)
    test_default_passwords(ip_addr)
    test_weak_passwords(ip_addr)


def blackbox():
    print("doing black box tests")
    ip_addr = raw_input("Enter the device IP:\n")
    print("You entered {}".format(ip_addr))
    print("Creating directory")
    os.system("mkdir results/{}".format(ip_addr))
    os.system("rm -rf results/{}/*".format(ip_addr))
    known_device, ftp, ssh, http, https, smb, smb135 = nmap_scan(ip_addr)
    nikto_scan(ip_addr)
    openvas_scan(ip_addr)
    test_default_passwords(ip_addr)
    test_weak_passwords(ip_addr)
    if(ssh):
        ssh_test_passwords(ip_addr)
    #fuzz_device(ip_addr, 80) #TODO: Use ports detected during nmap?

def menu():
    while(1):
        print("Starting IoT Scan Test Prototype\n\n\n")
        print("1. Run scan against known device")
        print("2. Run scan against unknown device")
        print("3. exit")
        user_input = raw_input("Enter your selection\n")

        if(user_input == '1'):
            known()
        elif(user_input == '2'):
            blackbox()
        else:
            print("exiting")
            exit()

if __name__ == "__main__":
	menu()