1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269
|
#!/usr/bin/python
# dane is a tool to generate and verify HASTLS and TLSA records
# By Paul Wouters <paul@xelerance.com> and Christopher Olah <chris@xelerance.com>
# Copyright 2011 by Xelerance http://www.xelerance.com/
# License: GNU GENERAL PUBLIC LICENSE Version 2 or later
#
# https://datatracker.ietf.org/wg/dane/charter/
# https://datatracker.ietf.org/doc/draft-ietf-dane-protocol/
import sys
import binascii
import ssl, socket
import hashlib
import warnings
try:
import ldnsx
except:
import daneldnsx as ldnsx
try:
import sys
import argparse
except ImportError , e:
module = str(e)[16:]
print >> sys.stderr, "dane requires the python module " + module
if module in ["argparse", "ipcalc", "ldns"]:
print >> sys.stderr, "Fedora/CentOS: yum install " \
+ {"argparse":"python-argparse", "ipcalc": "python-ipcalc", "ldns": "ldns-python"}
print >> sys.stderr, "Debian/Ubuntu: apt-get install " \
+ {"argparse":"python-argparse", "ipcalc": "python-ipcalc", "ldns": "python-ldns"}
print >> sys.stderr, "openSUSE: zypper in " \
+ {"argparse":"python-argparse", "ipcalc": "python-ipcalc", "ldns": "python-ldns"}
sys.exit(1)
def create_txt(hostname, pubkey):
""" Kaminsky / Gilmore type TLS pubkey in TXT RRtype -- testing version"""
warnings.warn("create_txt is untested...")
hashobj = hashlib.sha1() #Are there other valid hashes?
hashobj.update(pubkey) #Should we try and get the pubkey ourselves, like in create_TLSA?
result = hashobj.hexdigest().upper()
return "%s IN TXT \"v=key1 ha=sha1 h=%s\"" % (hostname, result)
def create_hastls(hostname, fallback_default, services):
""" Creates a HASTLS RRtype """
def checkExistingTLSA(hostname, certtype, reftype):
""" check if a TLSA record already exists, to compare and notify if update is needed
UNTESTED!!!! """
warnings.warn("checkExistingTLSA is untested...")
res = ldnsx.resolver() #Get the appropriate resource records...
pkt = res.query(hostname, "TYPE65468", tries = 3)
pres_TLSAs=set(pkt.answer(rr_type="TYPE65468"))
pres_TLSAs = map(lambda rr: str(rr).strip(), pres_TLSAs)
TLSAs = set(create_TLSAs(hostname, certtype,reftype).split('\n'))
return TLSAs == pres_TLSAs # are things the same as before?
def checkExistingHASTLS(address):
""" check if a HASTLS record already exists, to compare and notify if update is needed """
def create_tlsa(hostname, certtype, reftype):
"""Creates a TLSA RRtype"""
""" get all A/AAAA records for hostname """
if secure:
pkt = ldnsx.secure_query(hostname, "ANY", flex=True)
else:
pkt = ldnsx.query(hostname, "ANY")
records = pkt.answer( rr_type = { "ipv4":"A", "ipv6":"AAAA", "both":"A|AAAA" }[transport] )
drafts, rfcs = [], []
for rr in records:
draft = genTLSA(hostname, rr.ip(), certtype, reftype, draft=True)
rfc = genTLSA(hostname, rr.ip(), certtype, reftype, draft=False)
if draft and not (draft in drafts):
drafts.append(draft.strip())
if rfc and not (rfc in rfcs):
rfcs.append(rfc.strip())
ret = ""
if not fmt == "rfc":
ret += "\n".join(drafts)
if not fmt == "draft":
ret += "\n".join(rfcs)
return ret
def genTLSA(hostname, address, certtype, reftype, draft=True):
try:
# errors will be thrown already before we get here
dercert = get_cert(hostname, address)
except:
return
if not dercert:
return
if certtype != 1:
raise Exception("Only EE-cert supported right now")
certhex = hashCert(reftype, dercert)
if draft:
# octet length is half of the string length; remember certtype and reftype are part of the length so +2
return "_443._tcp.%s IN TYPE65468 \# %s 0%s0%s%s"%(hostname, len(certhex)/2+2, certtype, reftype, certhex )
else:
return "_443._tcp.%s IN TLSA %s %s %s"%(hostname, certtype, reftype, certhex)
def get_cert(hostname, address):
# We don't use ssl.get_server_certificate because it does not support IPv6, and it converts DER to PEM, which
# we would just have to convert back to DER using ssl.PEM_cert_to_DER_cert()
try:
# kinda ugly kludge
if ":" in address:
conn = socket.socket(socket.AF_INET6)
else:
conn = socket.socket(socket.AF_INET)
conn.connect((address, 443))
except socket.error , e:
#raise Exception("%s (%s): %s"%(hostname, address, str(e)))
print >> sys.stderr, "%s (%s): %s"%(hostname, address, str(e))
return
try:
sock = ssl.wrap_socket(conn)
except ssl.SSLError , e:
#raise Exception("%s (%s): %s"%(hostname, address, str(e)))
print >> sys.stderr, "%s (%s): %s"%(hostname, address, str(e))
return
try:
dercert = sock.getpeercert(True)
except AttributeError , e:
#print >> sys.stderr, "%s (%s): %s"%(hostname, address, str(e))
return
sock.close()
conn.close()
return dercert
# take PEM encoded EE-cert and DER encode it, then sha256 it
def hashCert(reftype,certblob):
if reftype == 0:
return binascii.b2a_hex(certblob).upper()
elif reftype == 1:
hashobj = hashlib.sha256()
hashobj.update(certblob)
elif reftype == 2:
hashobj = hashlib.sha512()
hashobj.update(certblob)
else:
return 0
return hashobj.hexdigest().upper()
secure, transport, quiet, fmt = True, "both", False, "draft"
# create the parser
parser = argparse.ArgumentParser(description='Create TLS related DNS records for hosts or an entire zone. version 1.2.1')
# AXFR
parser.add_argument('-n', '--nameserver', metavar="nameserver", action='append', help='nameserver to query')
parser.add_argument('--axfr', action='store_true', help='use AXFR (all A/AAAA records will be scanned)')
# IETF status related, currently --draft is the default
parser.add_argument('--draft', action='store_true',help='output in draft private rrtype (65468/65469) format (default)')
parser.add_argument('--rfc', action='store_true',help='output in rfc (TLSA/HASTLS) rrtype format')
# TLSA related
parser.add_argument('--tlsa', action='store_true',help='generate TLSA record (default:yes)')
parser.add_argument('--eecert', action='store_true',help='use EEcert for TLSA record (default)')
parser.add_argument('--cacert', action='store_true',help='use CAcert for TLSA record (not supported yet)')
parser.add_argument('--pubkey', action='store_true',help='use pubkey for TLSA record (not supported yet)')
parser.add_argument('--txt', action='store_true',help='generate Kaminsky style TXT record (not supported yet)')
parser.add_argument('--sha256', action='store_true',help='use SHA256 for the TLSA cert type')
parser.add_argument('--sha512', action='store_true',help='use SHA512 for the TLSA cert type')
parser.add_argument('--full', action='store_true',help='use full certificate for the TLSA cert type')
# allow non-dnssec answers
parser.add_argument('--insecure', action='store_true',help='allow use of non-dnssec answers to find SSL hosts')
# limit networking to ipv4 or ipv6 only
parser.add_argument('-4', dest='ipv4', action='store_true',help='use ipv4 networking only')
parser.add_argument('-6', dest='ipv6', action='store_true',help='use ipv6 networking only')
parser.add_argument('-q', '--quiet', action='store_true',help='suppress warnings and errors')
parser.add_argument('-v', '--version', action='store_true',help='show version and exit')
# finally, the host list
parser.add_argument('hosts', metavar="hostname", nargs='+')
args = parser.parse_args(sys.argv[1:])
if args.version:
sys.exit("dane: version 1.2.2")
if not args.rfc:
args.draft = True
if args.cacert:
sys.exit("TLSA CAcert type record not yet supported")
if args.pubkey:
sys.exit("TLSA Pubkey type record not yet supported")
if args.sha512:
reftype=2
elif args.full:
reftype=0
else:
reftype=1
if args.quiet:
quiet = True
if args.tlsa:
args.eecert = True
if args.insecure:
secure = False
if args.ipv4 and not args.ipv6:
transport = "ipv4"
if args.ipv6 and not args.ipv4:
transport = "ipv6"
# filter the non-options arguments for an "@argument" and convert it for the axfr option.
filterHost= []
if not args.nameserver:
args.nameserver = []
for host in args.hosts:
if host[0] == "@":
args.nameserver.append(host[1:])
args.hosts.remove(host)
args.axfr=True
if args.rfc and args.draft:
fmt = "both"
elif args.rfc:
fmt = "rfc"
else:
fmt = "draft"
if not args.hosts:
sys.exit("Host are needed.")
# main("--help")
for host in args.hosts:
if host[-1] != ".":
host += "."
if not args.axfr:
for host in args.hosts:
print create_tlsa(host,1,reftype)
if args.axfr:
# Try and AXFR it
if len(args.nameserver) == 0:
sys.exit("nameserver needed. syntax: -n nameserver")
resolver = ldnsx.resolver(args.nameserver[0], dnssec=True)
for host in args.hosts:
ipv4 = []
ipv6 = []
for rr in resolver.AXFR(host):
if rr.rr_type() == "A" and rr.owner() not in ipv4:
ipv4.append(rr.owner())
if rr.rr_type() == "AAAA" and rr.owner() not in ipv6:
ipv6.append(rr.owner())
if transport != "ipv6":
for host in ipv4:
print create_tlsa(host,1,reftype)
if transport != "ipv4":
for host in ipv6:
print create_tlsa(host,1,reftype)
|