LEARN. PRACTICE. GROW.Prepare the exercise’s trust chain
The code below generates a temporary RSA key, constructs DNSKEY and signatures for an original fixture, and configures the matching digest as Unbound’s DS anchor. The starting point is fund.test, not the public root. The validator requests DNSKEY and checks received data from that configured trust. It uses temporary loopback ports only and stops instances at the end. Provide Python 3, Unbound 1.26.1, matching unbound-checkconf, OpenSSL 3, and dig; save the code as dns-validation.py. Recorded execution used Python 3.13.1, OpenSSL 3.6.1, and dig 9.10.6. RSA/SHA-256, algorithm eight, serves this exercise; production choice needs its own policy and compatibility assessment. Do not use disposable keys outside the lab. The JSON file stores public experiment data without the private key.
Read the valid control and altered-data case
The valid experiment receives NOERROR with A and AD through the known instance. This control demonstrates that the configured path can validate data. In tampered, the fixture signs 192.0.2.80 and sends 192.0.2.81 with the earlier signature. The normal query receives SERVFAIL and the log reports cryptographic failure. The CD query returns data without AD. Compare these observations instead of merely concluding that “DNS answers”. A CD result helps investigation but neither repairs the signature nor authorizes changing consumer policy. At fictional bank Arco, pressure to recover reconciliation prompted a proposal to disable validation globally. The useful decision is to engage the signed-publication owner, assess repair or coherent rollback, and confirm normal querying before application acceptance.
Distinguish time, missing signature, and mismatched anchor
The remaining experiments introduce different causes of the same SERVFAIL. expired uses a signature that expired two days ago; future uses inception two days ahead. Record TTL remains sixty seconds. The purpose is to show that receiving data recently does not make a signature valid at checking time. missing serves A without RRSIG; the validator also requests DS to clarify context. The fixture includes one specific signed NSEC proof of DS absence at that existing name, which is not a delegation. This permits a missing-signature diagnosis but does not provide general negative-answer coverage. anchorMismatch changes the configured digest while retaining the key and signed data. Record each failure’s reason and avoid summarizing everything as “expired certificate”: DNSKEY, RRSIG, and a TLS certificate are different objects.
Reproduce and interpret evidence limits
Run the command shown in the next lesson’s guide and retain JSON. Each repetition creates another key, ports, and identifiers; compare states, flags, log causes, and checks rather than complete byte equality. The runner refuses an account with a personal.digrc to avoid implicit client configuration. Use a clean lab account without deleting working configuration. If a check fails, diagnostics include the experiment, unexpected requests, and validator log. Do not replace results with expected output. Confirmed execution passed 41 checks across six experiments and removed the temporary key. Summary: presence of A or RRSIG is insufficient to demonstrate trust. This experiment does not cover the public root, key rollover, all NSEC/NSEC3 logic, TLS, or applications. Connect the lesson to caching, delegations, and acceptance criteria.
#!/usr/bin/env python3
"""Original DNSSEC lab: disposable RSA key, local signed fixture, validating Unbound. No system DNS changes."""
import argparse,base64,hashlib,json,platform,re,socket,socketserver,struct,subprocess,tempfile,threading,time
from pathlib import Path
ZONE='fund.test.'TTL=60
def wire(name):return b''.join(bytes([len(x)])+x.lower.encode('ascii') for x in name.rstrip('.').split('.'))+b'\0'
def rr(name,kind,data):return wire(name)+struct.pack('!HHIH',kind,1,TTL,len(data))+data
def main:
p=argparse.ArgumentParser(description=__doc__);p.add_argument('--unbound',required=True);p.add_argument('--checkconf',required=True);p.add_argument('--openssl',required=True);p.add_argument('--dig',default='/usr/bin/dig');p.add_argument('--output',default='dns-validation-evidence.json');args=p.parse_args
assert not Path.home.joinpath('.digrc').exists,'Use a clean lab account without personal.digrc; do not delete working configuration.'
def run(argv,data=None):
x=subprocess.run(argv,input=data,capture_output=True,timeout=20);assert x.returncode==0,(argv,x.stdout.decode(errors='replace'),x.stderr.decode(errors='replace'));return x.stdout
ub=run([args.unbound,'-V']).decode;assert re.search(r'Version 1\.26\.1\b',ub),ub
ov=run([args.openssl,'version']).decode.strip;assert ov.startswith('OpenSSL 3.'),ov
version_result=subprocess.run([args.dig,'-v'],capture_output=True,timeout=10);assert version_result.returncode==0
dv=(version_result.stdout+version_result.stderr).decode.strip;assert dv, 'Missing dig version'
checks=[];observations={};events=[];configs=[]
def check(label,result):assert result,label;checks.append(label)
with tempfile.TemporaryDirectory(prefix='dr-dns-validation-') as tmp:
tmp=Path(tmp);private=tmp/'ephemeral.pem'
run([args.openssl,'genpkey','-algorithm','RSA','-pkeyopt','rsa_keygen_bits:2048','-out',str(private)])
private.chmod(0o600)
der=run([args.openssl,'rsa','-in',str(private),'-RSAPublicKey_out','-outform','DER'])
def tlv(data,offset):
tag=data[offset];n=data[offset+1];pos=offset+2
if n&128:k=n&127;n=int.from_bytes(data[pos:pos+k],'big');pos+=k
return tag,data[pos:pos+n],pos+n
tag,seq,end=tlv(der,0);assert tag==0x30 and end==len(der)
tag,mod,pos=tlv(seq,0);assert tag==2
tag,exp,pos=tlv(seq,pos);assert tag==2 and pos==len(seq)
mod=mod.lstrip(b'\0');exp=exp.lstrip(b'\0');assert len(exp)<256
dnskey=struct.pack('!HBB',257,3,8)+bytes([len(exp)])+exp+mod
acc=sum(b<<(8 if i%2==0 else 0) for i,b in enumerate(dnskey));acc+=(acc>>16)&65535;keytag=acc&65535
digest=hashlib.sha256(wire(ZONE)+dnskey).hexdigest.upper;now=int(time.time)
check('RSA public key is 2048 bits',len(mod)==256)
def signed(name,kind,data,mode='valid'):
inception,expiration=now-60,now+3600
if mode=='expired':inception,expiration=now-259200,now-172800
if mode=='future':inception,expiration=now+172800,now+259200
header=struct.pack('!HBBIIIH',kind,8,len(name.rstrip('.').split('.')),TTL,expiration,inception,keytag)+wire(ZONE)
canonical=rr(name,kind,data)
signature=run([args.openssl,'dgst','-sha256','-sign',str(private)],header+canonical)
return canonical+rr(name,46,header+signature)
dnskey_answer=signed(ZONE,48,dnskey)
fixtures={}
for label in ['valid','tampered','expired','future','missing']:
name=label+'.'+ZONE;data=socket.inet_aton('192.0.2.80')
answer=signed(name,1,data,label)
if label=='tampered':answer=rr(name,1,socket.inet_aton('192.0.2.81'))+answer[len(rr(name,1,data)):]
if label=='missing':answer=rr(name,1,data)
fixtures[name]=answer
soa=wire('ns.'+ZONE)+wire('hostmaster.'+ZONE)+struct.pack('!IIIII',1,60,30,600,60)
# A narrow signed NODATA proof: this existing name has A, RRSIG and NSEC, but no DS or NS.
nsec=wire('tampered.'+ZONE)+bytes([0,6,0x40,0,0,0,0,0x03])
missing_ds_proof=signed(ZONE,6,soa)+signed('missing.'+ZONE,47,nsec)
errors=[]
class Authority(socketserver.BaseRequestHandler):
def handle(self):
try:
data,sock=self.request;offset=12;labels=[]
while data[offset]:
n=data[offset];assert n<64;labels.append(data[offset+1:offset+1+n].decode('ascii').lower);offset+=1+n
offset+=1;kind,cls=struct.unpack('!HH',data[offset:offset+4]);assert cls==1;name='.'.join(labels)+'.'q=data[12:offset+4]
events.append({'name':name,'type':kind})
assert name==ZONE or name.endswith('.'+ZONE),name
authority_data=b''authority_count=0
if name=='missing.'+ZONE and kind==43:answer=b''count=0;authority_data=missing_ds_proof;authority_count=4
elif name==ZONE and kind==48:answer=dnskey_answer;count=2
elif name in fixtures and kind==1:answer=fixtures[name];count=1 if name.startswith('missing.') else 2
else:raise AssertionError('Unexpected question '+name+' '+str(kind))
flags=0x8400|(struct.unpack('!H',data[2:4])[0]&0x0100)
sock.sendto(data[:2]+struct.pack('!HHHHH',flags,1,count,authority_count,0)+q+answer+authority_data,self.client_address)
except Exception as exc:errors.append(str(exc))
authority=socketserver.UDPServer(('127.0.0.1',0),Authority);thread=threading.Thread(target=authority.serve_forever,daemon=True);thread.start;authport=authority.server_address[1]
def experiment(label,name,wrong_anchor=False):
with socket.socket as s:s.bind(('127.0.0.1',0));port=s.getsockname[1]
chosen=('0' if digest[0]!='0' else '1')+digest[1:] if wrong_anchor else digest
conf=tmp/(label+'.conf');log=tmp/(label+'.log')
config=f'''server:
interface: 127.0.0.1@{port}
outgoing-interface: 127.0.0.1
access-control: 127.0.0.0/8 allow
do-ip6: no
username: ""
chroot: ""
directory: "{tmp}"
pidfile: "{tmp}/{label}.pid"
use-syslog: no
do-daemonize: no
verbosity: 1
val-log-level: 2
num-threads: 1
module-config: "validator iterator"
trust-anchor: "{ZONE} IN DS {keytag} 8 2 {chosen}"
trust-anchor-signaling: no
root-key-sentinel: no
qname-minimisation: no
do-not-query-localhost: no
local-zone: "." refuse
local-zone: "{ZONE}" transparent
prefetch: no
serve-expired: no
val-permissive-mode: no
stub-zone:
name: "{ZONE}"
stub-addr: 127.0.0.1@{authport}
stub-first: no
stub-prime: no
remote-control:
control-enable: no
'''
conf.write_text(config);validation=run([args.checkconf,str(conf)]).decode;configs.append({'label':label,'config':config,'checkconf':validation})
def query(cd=False):
argv=[args.dig,'@127.0.0.1','-p',str(port),name,'A','+recurse','+nosearch','+dnssec','+bufsize=1232','+time=4','+tries=1']
if cd:argv.append('+cdflag')
return run(argv).decode
with log.open('w') as lf:
proc=subprocess.Popen([args.unbound,'-d','-c',str(conf)],stdout=lf,stderr=lf)
try:
for _ in range(100):
assert proc.poll is None,log.read_text
try:
with socket.create_connection(('127.0.0.1',port),timeout=.1):break
except OSError:time.sleep(.03)
else:raise AssertionError('Resolver did not listen')
before=len(events);normal=query;cd=query(True)
def flag(text,value):return value in re.search(r' flags: ([^;]+);',text).group(1).split
secure=label=='valid'
check(label+': normal status','status: '+('NOERROR' if secure else 'SERVFAIL') in normal)
check(label+': AD reflects validation',flag(normal,'ad')==secure)
check(label+': CD returns response','status: NOERROR' in cd and ('192.0.2.81' if label=='tampered' else '192.0.2.80')in cd)
if not secure:check(label+': CD does not assert AD',not flag(cd,'ad'))
check(label+': DNSKEY requested',any(x['name']==ZONE and x['type']==48 for x in events[before:]))
if label=='missing':check('missing: DS absence lookup observed',any(x['name']==name and x['type']==43 for x in events[before:]))
observations[label]={'normal':normal,'checkingDisabled':cd,'authorityRequests':events[before:]}
except Exception:
print(json.dumps({'experiment':label,'fixtureErrors':errors,'events':events,'validatorLog':log.read_text}))
raise
finally:
proc.terminate
try:proc.wait(timeout=5)
except subprocess.TimeoutExpired:proc.kill;proc.wait(timeout=5)
check(label+': process stopped',proc.poll is not None)
observations[label]['validatorLog']=log.read_text
try:
for label in ['valid','tampered','expired','future','missing']:experiment(label,label+'.'+ZONE)
experiment('anchorMismatch','valid.'+ZONE,True)
check('no unexpected fixture questions',not errors)
check('all fixture traffic stays in test zone',all(x['name']==ZONE or x['name'].endswith('.'+ZONE)for x in events))
finally:authority.shutdown;authority.server_close;thread.join(timeout=3)
check('authority thread stopped',not thread.is_alive)
public={'zone':ZONE,'keyTag':keytag,'algorithm':8,'digestType':2,'dnskeyBase64':base64.b64encode(dnskey[4:]).decode,'dsSha256':digest,'generatedAtEpoch':now}
check('temporary key and configuration removed',not tmp.exists)
result={'checkedAtEpoch':int(time.time),'unboundVersion':ub,'opensslVersion':ov,'digVersion':dv,'pythonVersion':platform.python_version,'checks':checks,'observations':observations,'configurations':configs,'publicLabKey':public,'runnerSha256':hashlib.sha256(Path(__file__).read_bytes).hexdigest,'scope':'Six actual validating Unbound experiments using an original signed UDP fixture and a locally configured DS trust anchor. Ephemeral RSA/SHA-256 key generated and removed. No public root chain, parent delegation traversal, general NSEC/NSEC3 denial coverage, key rollover, production application or system DNS changes. The fixture includes one signed NSEC proof for DS absence at an existing non-delegation name. CD is diagnostic evidence, not a production workaround.'}
Path(args.output).write_text(json.dumps(result,indent=2)+'\n');print(json.dumps({'experiments':len(observations),'passed':len(checks),'output':args.output}))
if __name__=='__main__':main
Arco receives an address with CD, but the normal query fails because of a mismatched signature. The result helps locate the cause and does not demonstrate secure recovery.
Common pitfalls
Confusing RRSIG presence with completed validation, using CD as a permanent fix, treating TTL as signature validity, or assuming a public chain from a local anchor.
Related topics: Caching and resolution failures · Transport and trust
The valid control and intentional faults show what configuration validates and rejects. The conclusion must identify the trust point and executed scope.
Reference: Unbound validator configuration and original DR DNSSEC experiments · DNS RFC 1034/1035 with RFC 2181, 2308, 3596, 4033, 7766 and 8767; dig BIND 9.20