I've been working on a Python program that searches through multiple ranges of IP addresses and checks for an SSL certificate/reads certain info off of it and then writes it to a CSV file.
The trouble I'm having with it now is that some of the IP addresses I'm going to be searching through may have multiple domain names on the same IP address. I've been looking around to try and come up with some type of solution for this, but there doesn't seem to be much documentation on checking IP addresses with SNI in OpenSSL
Also, another issue I'm having is that I want to grab the server type(eg. apache, nginx, etc..) from the certificate, but OpenSSL doesn't seem to provide that info. Curl on the other hand does, and I'm actually using curl to establish whether or not a cert is found on an IP address -- I just don't know how I would be able to grab that information from curl's output.
If anyone can give me some input as to how I would implement the two features listed above, I'd really appreciate it
Here is the code I have so far(excuse the really long hardcoded threading -- I haven't had a chance to clean it up):
from subprocess import *
import time
import ssl
import OpenSSL
import threading
import sys
import csv
def main():
global IP_list
IP_list = []
global certInfoHolder
certInfoHolder = []
getRanges()
def getRanges():
print 'Enter a range(ex. From: 192.168.1.0 , To: 192.168.1.10)'
starting_IP = raw_input('From: ')
ending_IP = raw_input('To: ')
ip_ranges = ipRange(starting_IP, ending_IP)
addMore = raw_input('Do you want to add another IP range?Y/N')
addmore = addMore.lower()
if addMore == 'y' or addMore == 'yes':
getRanges()
elif addMore == 'n' or addMore =='no':
print 'Done gathering IP Addresses'
createdThreads = 0
threadSplit = len(IP_list) / 10
#Splitting the work up between the threads
thread_1_list = IP_list[0:threadSplit]
thread_2_list = IP_list[threadSplit:(threadSplit*2)]
thread_3_list = IP_list[(threadSplit*2):(threadSplit*3)]
thread_4_list = IP_list[(threadSplit*3):(threadSplit*4)]
thread_5_list = IP_list[(threadSplit*4):(threadSplit*5)]
thread_6_list = IP_list[(threadSplit*5):(threadSplit*6)]
thread_7_list = IP_list[(threadSplit*6):(threadSplit*7)]
thread_8_list = IP_list[(threadSplit*7):(threadSplit*8)]
thread_9_list = IP_list[(threadSplit*8):(threadSplit*9)]
thread_10_list = IP_list[(threadSplit*9):(threadSplit*10)]
print '5 Threads..Splitting the work up to: ' , threadSplit
for address in range(threadSplit):
print address
thread_1 = getCertInfo(thread_1_list[address])
thread_2 = getCertInfo(thread_2_list[address])
thread_3 = getCertInfo(thread_3_list[address])
thread_4 = getCertInfo(thread_4_list[address])
thread_5 = getCertInfo(thread_5_list[address])
thread_6 = getCertInfo(thread_6_list[address])
thread_7 = getCertInfo(thread_7_list[address])
thread_8 = getCertInfo(thread_8_list[address])
thread_9 = getCertInfo(thread_9_list[address])
thread_10 = getCertInfo(thread_10_list[address])
thread_1.start()
thread_2.start()
thread_3.start()
thread_4.start()
thread_5.start()
thread_6.start()
thread_7.start()
thread_8.start()
thread_9.start()
thread_10.start()
thread_1.join()
thread_2.join()
thread_3.join()
thread_4.join()
thread_5.join()
thread_6.join()
thread_7.join()
thread_8.join()
thread_9.join()
thread_10.join()
if address == threadSplit-1:
for address in range(threadSplit,len(thread_5_list)):
thread_10 = getCertInfo(thread_10_list[address])
thread_10.start()
thread_10.join()
c = csv.writer(open("AInfo.csv", "a"))
CIH = certInfoHolder
for info in CIH:
c.writerow([info[0], info[1], info[2], info[3], info[4]])
print 'DONE'
def ipRange(start_ip, end_ip):
start = list(map(int, start_ip.split(".")))
end = list(map(int, end_ip.split(".")))
temp = start
IP_list.append(start_ip)
while temp != end:
start[3] += 1
for i in (3, 2, 1):
if temp[i] == 256:
temp[i] = 0
temp[i-1] += 1
IP = ".".join(map(str, temp))
IP_list.append(IP)
print 'IP_List size', len(IP_list)
class getCertInfo(threading.Thread):
def __init__(self,IP):
threading.Thread.__init__(self)
self.IP_check=IP
def run(self):
try:
#Using curl to check for a timeout
self.getCert = check_output('curl --max-time 2 '+self.IP_check,
stderr=STDOUT,
shell=True)
#OpenSSL get server certificate
self.server_certificate = ssl.get_server_certificate((self.IP_check, 443))
self.x509 = OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM, self.server_certificate)
#Formatting expiration date and time
self.eDateT = str(self.x509.get_notAfter())
#173.252.110.27
self.certInfo = self.x509.get_subject()
self.commonName = self.certInfo.commonName
self.companyName = self.certInfo.O
self.serialNumber = self.x509.get_serial_number()
self.x = [str(self.IP_check),str(self.companyName),str(self.eDateT),
str(self.commonName),str(self.serialNumber)]
certInfoHolder.append(self.x)
except CalledProcessError as detail:
print str(detail) + ' ' + self.IP_check +'\n'
pass
