|
|
@ -24,10 +24,13 @@ |
|
|
|
# Based on |
|
|
|
# http://backreference.org/2010/11/17/dnssec-verification-with-dig/ |
|
|
|
# https://github.com/rthalley/dnspython/blob/master/tests/test_dnssec.py |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
import traceback |
|
|
|
import sys |
|
|
|
import time |
|
|
|
import struct |
|
|
|
|
|
|
|
|
|
|
|
import dns.name |
|
|
|
import dns.query |
|
|
@ -51,6 +54,127 @@ import dns.rdtypes.IN.AAAA |
|
|
|
from dns.exception import DNSException |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
""" |
|
|
|
Pure-Python version of dns.dnssec._validate_rsig |
|
|
|
Uses tlslite instead of PyCrypto |
|
|
|
""" |
|
|
|
def python_validate_rrsig(rrset, rrsig, keys, origin=None, now=None): |
|
|
|
from dns.dnssec import ValidationFailure |
|
|
|
from dns.dnssec import _find_candidate_keys, _make_hash, _is_rsa, _to_rdata, _make_algorithm_id |
|
|
|
|
|
|
|
if isinstance(origin, (str, unicode)): |
|
|
|
origin = dns.name.from_text(origin, dns.name.root) |
|
|
|
|
|
|
|
for candidate_key in _find_candidate_keys(keys, rrsig): |
|
|
|
if not candidate_key: |
|
|
|
raise ValidationFailure, 'unknown key' |
|
|
|
|
|
|
|
# For convenience, allow the rrset to be specified as a (name, rdataset) |
|
|
|
# tuple as well as a proper rrset |
|
|
|
if isinstance(rrset, tuple): |
|
|
|
rrname = rrset[0] |
|
|
|
rdataset = rrset[1] |
|
|
|
else: |
|
|
|
rrname = rrset.name |
|
|
|
rdataset = rrset |
|
|
|
|
|
|
|
if now is None: |
|
|
|
now = time.time() |
|
|
|
if rrsig.expiration < now: |
|
|
|
raise ValidationFailure, 'expired' |
|
|
|
if rrsig.inception > now: |
|
|
|
raise ValidationFailure, 'not yet valid' |
|
|
|
|
|
|
|
hash = _make_hash(rrsig.algorithm) |
|
|
|
|
|
|
|
if _is_rsa(rrsig.algorithm): |
|
|
|
from tlslite.utils.keyfactory import _createPublicRSAKey |
|
|
|
from tlslite.utils.cryptomath import bytesToNumber |
|
|
|
keyptr = candidate_key.key |
|
|
|
(bytes,) = struct.unpack('!B', keyptr[0:1]) |
|
|
|
keyptr = keyptr[1:] |
|
|
|
if bytes == 0: |
|
|
|
(bytes,) = struct.unpack('!H', keyptr[0:2]) |
|
|
|
keyptr = keyptr[2:] |
|
|
|
rsa_e = keyptr[0:bytes] |
|
|
|
rsa_n = keyptr[bytes:] |
|
|
|
keylen = len(rsa_n) * 8 |
|
|
|
n = bytesToNumber(bytearray(rsa_n)) |
|
|
|
e = bytesToNumber(bytearray(rsa_e)) |
|
|
|
pubkey = _createPublicRSAKey(n, e) |
|
|
|
sig = rrsig.signature |
|
|
|
|
|
|
|
elif _is_ecdsa(rrsig.algorithm): |
|
|
|
if rrsig.algorithm == ECDSAP256SHA256: |
|
|
|
curve = ecdsa.curves.NIST256p |
|
|
|
key_len = 32 |
|
|
|
digest_len = 32 |
|
|
|
elif rrsig.algorithm == ECDSAP384SHA384: |
|
|
|
curve = ecdsa.curves.NIST384p |
|
|
|
key_len = 48 |
|
|
|
digest_len = 48 |
|
|
|
else: |
|
|
|
# shouldn't happen |
|
|
|
raise ValidationFailure, 'unknown ECDSA curve' |
|
|
|
keyptr = candidate_key.key |
|
|
|
x = ecdsa.util.string_to_number(keyptr[0:key_len]) |
|
|
|
y = ecdsa.util.string_to_number(keyptr[key_len:key_len * 2]) |
|
|
|
assert ecdsa.ecdsa.point_is_valid(curve.generator, x, y) |
|
|
|
point = ecdsa.ellipticcurve.Point(curve.curve, x, y, curve.order) |
|
|
|
verifying_key = ecdsa.keys.VerifyingKey.from_public_point(point, |
|
|
|
curve) |
|
|
|
pubkey = ECKeyWrapper(verifying_key, key_len) |
|
|
|
r = rrsig.signature[:key_len] |
|
|
|
s = rrsig.signature[key_len:] |
|
|
|
sig = ecdsa.ecdsa.Signature(ecdsa.util.string_to_number(r), |
|
|
|
ecdsa.util.string_to_number(s)) |
|
|
|
|
|
|
|
else: |
|
|
|
raise ValidationFailure, 'unknown algorithm %u' % rrsig.algorithm |
|
|
|
|
|
|
|
hash.update(_to_rdata(rrsig, origin)[:18]) |
|
|
|
hash.update(rrsig.signer.to_digestable(origin)) |
|
|
|
|
|
|
|
if rrsig.labels < len(rrname) - 1: |
|
|
|
suffix = rrname.split(rrsig.labels + 1)[1] |
|
|
|
rrname = dns.name.from_text('*', suffix) |
|
|
|
rrnamebuf = rrname.to_digestable(origin) |
|
|
|
rrfixed = struct.pack('!HHI', rdataset.rdtype, rdataset.rdclass, |
|
|
|
rrsig.original_ttl) |
|
|
|
rrlist = sorted(rdataset); |
|
|
|
for rr in rrlist: |
|
|
|
hash.update(rrnamebuf) |
|
|
|
hash.update(rrfixed) |
|
|
|
rrdata = rr.to_digestable(origin) |
|
|
|
rrlen = struct.pack('!H', len(rrdata)) |
|
|
|
hash.update(rrlen) |
|
|
|
hash.update(rrdata) |
|
|
|
|
|
|
|
digest = hash.digest() |
|
|
|
|
|
|
|
if _is_rsa(rrsig.algorithm): |
|
|
|
digest = _make_algorithm_id(rrsig.algorithm) + digest |
|
|
|
if pubkey.verify(bytearray(sig), bytearray(digest)): |
|
|
|
return |
|
|
|
|
|
|
|
elif _is_ecdsa(rrsig.algorithm): |
|
|
|
if pubkey.verify(digest, sig): |
|
|
|
return |
|
|
|
|
|
|
|
else: |
|
|
|
raise ValidationFailure, 'unknown algorithm %u' % rrsig.algorithm |
|
|
|
|
|
|
|
raise ValidationFailure, 'verify failure' |
|
|
|
|
|
|
|
|
|
|
|
# replace validate_rrsig |
|
|
|
dns.dnssec._validate_rrsig = python_validate_rrsig |
|
|
|
dns.dnssec.validate_rrsig = python_validate_rrsig |
|
|
|
dns.dnssec.validate = dns.dnssec._validate |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
from util import print_error |
|
|
|
|
|
|
|
|
|
|
|