for each record from dnsmaster | do DNS resolution | ping record | get certificate details | get http(s) response status | get DNS CNAMES RR

gistfile1.txt

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

## PING
#Ref: https://www.ictshore.com/python/python-ping-tutorial/
#Ref: https://stackoverflow.com/questions/58330533/how-to-handle-error-exceptions-with-pythonping

## CSV
#Ref: https://kite.com/python/docs/csv.DictWriter.writerows

## DIG
#Ref: https://pypi.org/project/pydig/

## CERTS
#https://stackoverflow.com/questions/30862099/how-can-i-get-certificate-issuer-information-in-python
#https://gist.github.com/ryansb/d8c333eb4a74168474c4
#https://kite.com/python/docs/ssl.SSLSocket.getpeercert

## REQUESTS
# https://realpython.com/python-requests/

## .format() 
# https://pyformat.info/

'''
@author: bockor

Created on Wed Apr 08 08:43:37 2020

Basically, the script reads the public DNS RR's from a CSV file
(typically the source dnsmaster) and try to define whether the RR's are public
Internet facing hosts.

If the RR is identified to be a public host the script will endeavour to get:

- DNS resolution
- a ping reply on one of the found IP addresses
- DNS CNAME
- http(s) response status
- TLS certificate info

Else, if the RR is identified to be a domain name the script will try to
provide:

- DNS resolution
- DNS NS info
- DNS MX info

Exit Statuses:

(1) ICMP_REPLIED : Host name is advertised on the Internet and received a ping
                response.  

(2) ICMP_IGNORED : Host name is advertised on the Internet, but for one or
                other reason ping requests are ignored or timed out.

(3) DNS_FOUND_RR : Defenitely the entry is not a host however a domain.
       The script found relevant NS/MX Resource Records on the public Internet.

(4) DNS_NO_RESOLUTION : Host name is apparently not advertised on the Internet.

(5) DNS_DIG_ERROR : During the execution of dig an error happened.
        (Could be temporary)

So, if it turns out that the given DNS RR is not a host, however a DNS domain
        the script will try to find any NS and/or MX RR for this entry.

Requirements:
- python 3.6
- pythonping (via pip3)
- pydig (via pip3)
- requests
- need root (superadmin) credentials to run

Be patient whilst running the script.  Grab yourself a beer !

Sample output CSV (input CSV from dnsmaster is in fact the HOST column)

HOST,EXIT_STATUS,IP,CNAMES,HTTPS_RESP,HTTP_RESP,NAME_SERVERS,MAIL_SERVERS,ISSUED_TO,ISSUED_BY,EXPIRES_AT
www.vrt.be,ICMP_REPLIED,13.225.233.3,d3jkmvw9bskevs.cloudfront.net.,200,200,,,www.vrt.be,Amazon,Sep  9 12:00:00 2020 GMT
jftc.nato.int,ICMP_REPLIED,46.37.13.145,,HTTPS_NO_CONN,200,,,,,
vtm.be,ICMP_REPLIED,81.243.1.187,,200,200,,,persgroep.com,Let's Encrypt Authority X3,May 27 14:01:32 2020 GMT
xhamster.com,ICMP_REPLIED,104.18.156.3,,200,200,,,ssl893711.cloudflaressl.com,COMODO ECC Domain Validation Secure Server CA 2,Aug 26 23:59:59 2020 GMT
acci.nato.int,DNS_FOUND_RR,,,,,,30 mail.lp.nato.int. | 10 mail.sh.nato.int.,,,
mail.google.com,ICMP_REPLIED,216.58.211.101,googlemail.l.google.com.,200,200,,,mail.google.com,GTS CA 1O1,May 26 09:45:55 2020 GMT
val-val.be,DNS_NO_RESOLUTION,,,,,,,,,

NOTICE:
HOSTs with CNAME *.cloudfront.net || *.amazonaws.com || *.edgekey.net do not provide a single public static IP address.
The IP address you read in the respective IP column for these hosts is in fact the IP address of a successful ping during execution of the script.
'''

from pythonping import ping
import pydig
import socket
import csv
import ssl
import requests
from requests.exceptions import ConnectionError
from datetime import datetime

CSV_IN_FILE = 'nato_public_dns_names_SMALL.csv'
DEBUG = True

'''
# pydig: If needed, uncomment and eventually change the Internet DNS resolvers

resolver = pydig.Resolver(
    nameservers=[
        '185.45.52.149',
        '185.45.53.149'
    ]
)
'''

# pydig: Set time out parameter
resolver = pydig.Resolver(
    additional_args=[
    '+time=2',
    ]
)

def get_csv_out_filename(csv_in_filename):
    parts = csv_in_filename.split('.')
    return "{}-REPORT-{}.{}".format(\
            parts[0],\
            datetime.now().strftime("%d%m%Y-%H%M"),\
            parts[1])

def get_cames(hostname):
    cnames=''
    try:
        cnames = ' | '.join(resolver.query(hostname, 'CNAME'))
        if cnames:
            if DEBUG:
                print('{:<40}{:<20}'.format(hostname, 'DNS_FOUND_CNAME'))
    except:
        pass
    return cnames

def get_http_status_explecit(hostname, protocol='https'):
    url = '{}://{}'.format(protocol, hostname)
    try:
        response = requests.get(url, timeout=(2, 5))
        if DEBUG:
            print('{:<40}{:<20}'.format(url, response.status_code))
        return (response.status_code)
    except ConnectionError:
        conn_err_msg = '{}_NO_CONN'.format(protocol.upper())
        if DEBUG:
            print('{:<40}{:<20}'.format(url, conn_err_msg))
        return (conn_err_msg)

def get_http_status(hostname):
    url = 'https://{}'.format(hostname)
    try:
        response = requests.get(url, timeout=(2, 5))
        if response:
            # status code between 200 and 400
            if DEBUG:
                print('{:<40}{:<20}'.format(url, 'HTTP_200_TO_400'))
            return ('HTTP_200_TO_400')
        else:
            if DEBUG:
                print('{:<40}{:<20}'.format(url, 'HTTP_400_PLUS'))
            return ('HTTP_400_PLUS')
    except ConnectionError:
        if DEBUG:
            print('{:<40}{:<20}'.format(url, 'HTTP_NO_CONN'))
        return ('HTTP_NO_CONN')

def get_cert_info(hostname):
    issued_to, issued_by, expires_at = '','',''
    try:
        ctx = ssl.create_default_context()
        s = ctx.wrap_socket(socket.socket(), server_hostname=hostname)
        s.settimeout(3.0)
        s.connect((hostname, 443))
        cert = s.getpeercert()

        subject = dict(x[0] for x in cert['subject'])
        issued_to = subject['commonName']
        issuer = dict(x[0] for x in cert['issuer'])
        issued_by = issuer['commonName']
        expires_at = cert['notAfter']
    except:
        pass
    return issued_to, issued_by, expires_at

def record_is_host(record):
    exit_status=''
    ip = socket.gethostbyname(record)
    ping_result = ping(ip, count=1)
    # the record responds the ICMP request
    if ping_result.success():
        if DEBUG:
            print('{:<40}{:<20}'.format(record, 'ICMP_REPLIED'))
        exit_status ='ICMP_REPLIED'
    #the record does NOT respond the ICMP request
    else:
        if DEBUG:
            print('{:<40}{:<20}'.format(record, 'ICMP_IGNORED'))
        exit_status ='ICMP_IGNORED' 
    return exit_status, ip

def record_is_domain(record):
    exit_status = 'DNS_NO_RESOLUTION'
    name_servers, mail_servers = '',''
    # Can DIG find any Name Server for the record ?
    try:
        name_servers = ' | '.join(resolver.query(record, 'NS'))
        if name_servers:
            exit_status = 'DNS_FOUND_RR'
            if DEBUG:
                print('{:<40}{:<20}'.format(record, 'DNS_FOUND_RR'))
    except:
        # Catch CalledProcessError (non-zero exit status 9
        exit_status = "DNS_DIG_ERROR"
        if DEBUG:
            print('{:<40}{:<20}'.format(record, 'DNS_DIG_ERROR'))
    # Can DIG any Mail Server for the record ?
    try:
        mail_servers = ' | '.join(resolver.query(record, 'MX'))
        if mail_servers:
            exit_status = 'DNS_FOUND_RR'
            if DEBUG:
                print('{:<40}{:<20}'.format(record, 'DNS_FOUND_RR'))
    except:
        # Catch CalledProcessError (non-zero exit status 9
        exit_status = "DNS_DIG_ERROR"
        if DEBUG:
            print('{:<40}{:<20}'.format(record, 'DNS_DIG_ERROR'))
    return exit_status, name_servers, mail_servers

def main():
    with open(CSV_IN_FILE, 'r') as csv_in_file, \
            open(get_csv_out_filename(CSV_IN_FILE), 'w') as csv_out_file:
        csv_in = csv.reader(csv_in_file) 
        records = [] 
        report= []
        for row in csv_in:
            records.append(row[0])
        for record in records:
            try:
                #Is the record a DNS resolvable public host ?
                exit_status, ip = record_is_host(record)
                #try to get cert info for the record
                issued_to, issued_by, expires_at = get_cert_info(record)
                #try to get https response status code
                https_resp_status = get_http_status_explecit(record)
                #try to get http response status code
                http_resp_status = get_http_status_explecit(record,'http')
                #try to get CNAME RR's
                cnames = get_cames(record)

                report.append({'HOST': record,\
                        'IP': ip,\
                        'EXIT_STATUS': exit_status,\
                        'HTTPS_RESP': https_resp_status,\
                        'HTTP_RESP': http_resp_status,\
                        'ISSUED_TO' : issued_to,\
                        'ISSUED_BY' : issued_by,\
                        'EXPIRES_AT' : expires_at,\
                        'CNAMES': cnames}) 

            #DNS resolution for 'record' failed, maybe in fact 'host' is a domain instead ?
            except socket.error:
                exit_status, name_servers, mail_servers = record_is_domain(record)

                report.append({'HOST': record,\
                        'EXIT_STATUS': exit_status,\
                        'NAME_SERVERS': name_servers,\
                        'MAIL_SERVERS': mail_servers})

        #print(report)      
        writer = csv.DictWriter( csv_out_file, fieldnames = \
                    ['HOST',\
                    'EXIT_STATUS',\
                    'IP',\
                    'CNAMES',\
                    'HTTPS_RESP',\
                    'HTTP_RESP',\
                    'ISSUED_TO',\
                    'ISSUED_BY',\
                    'EXPIRES_AT',\
                    'NAME_SERVERS',\
                    'MAIL_SERVERS'])
        writer.writeheader()
        writer.writerows(report)
        print('DONE')

if __name__ == '__main__':
    main()