LEARN. PRACTICE. GROW.Prepare an experiment that can explain its result
The runner below creates an original authoritative fixture and starts Unbound on a temporary loopback port. It does not change the computer’s DNS configuration. Every dig query explicitly identifies its destination and uses an absolute name under fund.test. The fixture records each request’s name, type, and time; this record distinguishes reused answers from fresh retrieval. Before running, provide Python 3, Unbound 1.26.1, unbound-checkconf from the same version, and dig. Save the code as dns-cache.py and pass executable paths as arguments. The recorded execution used Python 3.13.1 and dig 9.10.6 on macOS. The script refuses an account with a personal.digrc to avoid implicit options. Use a clean lab account instead of deleting working configuration. Confirm versions before interpreting differences.
Compare publication with a cached answer
In the first experiment, positive.fund.test points to 192.0.2.10 with a four-second TTL. After the first query, the runner changes the fixture to 192.0.2.20. The immediate query still receives the earlier address without adding an upstream request. After waiting, the new address appears and the counter increases. The conclusion belongs to this instance and configuration: reuse occurred, followed by fresh retrieval after expiry. Organization-wide convergence was not measured. Response time alone would not be sufficient evidence because a local fixture can also respond quickly. During an incident, preserve the same evidence elements: name, type, server, origin, time, and published state. Avoid changing the zone again simply because a resolver retains data acquired earlier. First establish which entry is actually being observed.
Creating names and adding types are different experiments
The second experiment queries new.fund.test while the name does not yet exist. The fixture sends NXDOMAIN and SOA with a four-second negative lifetime. Creating A immediately does not change the answer already stored by Unbound; new publication appears only after refresh. In the third experiment, dual.fund.test exists with A but not yet AAAA. The AAAA result is NOERROR without that answer type and with SOA. Adding AAAA does not remove the earlier absence, while A is still obtained successfully. Record these situations separately in the report. At fictional bank Ria, labeling every empty answer “name nonexistent” prompted a proposal to delete a working name. The exercise asks you to correct classification, track the affected type, and retain application validation as a separate step. No case represents an employer’s procedures.
Read the evidence and prepare a bounded conclusion
The generated JSON contains complete answers, counters, configuration, and checks. In the zero-TTL experiment, two independent queries obtain different values and generate two fixture requests. That does not prove a real application never retains an address or connection. Save the file and write one conclusion per experiment with three parts: observation, interpretation, and limitation. If an assertion fails, preserve the failure and investigate versions, configuration, timing, and requests; do not replace output with the expected result. Temporary processes are terminated at the end and their working directory is removed. JSON remains at the chosen output path. Summary: publication, caching, and application consumption are distinct states. Connect this lesson to negative answers, message diagnosis, and rollback planning. The next lesson uses the same instrumentation to show the effect of local policies.
#!/usr/bin/env python3
"""Original local cache experiment. Requires Unbound 1.26.1 and dig; never changes system DNS."""
import argparse, hashlib, json, os, re, socket, socketserver, struct, subprocess, tempfile, threading, time
from pathlib import Path
def name_bytes(name):
return b''.join(bytes([len(x)]) + x.encode('ascii') for x in name.rstrip('.').split('.')) + b'\0'
def question(data):
i, labels = 12, []
while data[i]:
n=data[i]; labels.append(data[i+1:i+1+n].decode('ascii')); i+=n+1
i+=1
kind, cls=struct.unpack('!HH',data[i:i+4]); assert cls==1
return '.'.join(labels)+'.',kind,data[12:i+4]
def rr(name, kind, ttl, payload):
return name_bytes(name)+struct.pack('!HHIH',kind,1,ttl,len(payload))+payload
def main:
parser=argparse.ArgumentParser(description=__doc__)
parser.add_argument('--unbound',required=True); parser.add_argument('--checkconf',required=True)
parser.add_argument('--dig',default='/usr/bin/dig'); parser.add_argument('--output',default='dns-cache-evidence.json')
args=parser.parse_args; assert not Path.home.joinpath('.digrc').exists, 'Personal.digrc detected; use an isolated lab account.'
def cmd(argv):
p=subprocess.run(argv,text=True,capture_output=True,timeout=15)
assert p.returncode==0,(argv,p.stdout,p.stderr)
return p.stdout+p.stderr
version=cmd([args.unbound,'-V']); assert re.search(r'Version 1\.26\.1\b',version),version
dig_version=cmd([args.dig,'-v']).strip
records={}; hits=[]; observations={}; checks=[]; configurations=[]
def check(label, condition):
assert condition,label
checks.append(label)
class Authority(socketserver.BaseRequestHandler):
def handle(self):
data,sock=self.request
name,kind,q=question(data)
assert name.endswith('.fund.test.'),name
hits.append({'name':name,'type':kind,'at':round(time.monotonic,6)})
row=records.get(name)
answer=b'' auth=b'' rcode=0
if row is None: rcode=3
elif kind in row:
value,ttl=row[kind]
answer=rr(name,kind,ttl,socket.inet_pton(socket.AF_INET if kind==1 else socket.AF_INET6,value))
if not answer:
soa=name_bytes('ns.fund.test.')+name_bytes('hostmaster.fund.test.')+struct.pack('!IIIII',1,60,30,600,4)
# RFC 2308: SOA negative TTL is min(original SOA TTL 9, MINIMUM 4).
auth=rr('fund.test.',6,4,soa)
flags=0x8000|0x0400|(struct.unpack('!H',data[2:4])[0]&0x0100)|rcode
reply=data[:2]+struct.pack('!HHHHH',flags,1,bool(answer),bool(auth),0)+q+answer+auth
sock.sendto(reply,self.client_address)
with tempfile.TemporaryDirectory(prefix='dr-dns-cache-') as tmp:
authority=socketserver.UDPServer(('127.0.0.1',0),Authority)
thread=threading.Thread(target=authority.serve_forever,daemon=True);thread.start
auth_port=authority.server_address[1]
def count(name,kind=1):return sum(h['name']==name and h['type']==kind for h in hits)
def reserve_port:
with socket.socket as s:s.bind(('127.0.0.1',0));return s.getsockname[1]
def run_resolver(label, action, extra=''):
port=reserve_port;conf=Path(tmp)/(label+'.conf');log=Path(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: 0
num-threads: 1
module-config: "iterator"
qname-minimisation: no
do-not-query-localhost: no
local-zone: "." refuse
local-zone: "fund.test." transparent
prefetch: no
serve-expired: no
cache-min-ttl: 0
cache-max-ttl: 86400
cache-min-negative-ttl: 0
cache-max-negative-ttl: 3600
{extra}
stub-zone:
name: "fund.test."
stub-addr: 127.0.0.1@{auth_port}
stub-first: no
stub-prime: no
remote-control:
control-enable: no
'''
# Explicit policy replacement avoids duplicate settings.
if extra:
for line in extra.strip.splitlines:
key=line.strip.split(':')[0]
config=re.sub(r'^ '+re.escape(key)+r':[^\n]*\n','',config,flags=re.M)
config=config.replace('stub-zone:',extra+'\nstub-zone:')
conf.write_text(config); validation=cmd([args.checkconf,str(conf)])
configurations.append({'label':label,'config':config,'checkconf':validation})
def query(name,kind='A'):
out=cmd([args.dig,'@127.0.0.1','-p',str(port),name,kind,'+recurse','+nosearch','+noedns','+time=2','+tries=1'])
assert 'status:' in out,out
return out
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('Unbound did not listen')
action(query)
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)
try:
def baseline(query):
n='positive.fund.test.'records[n]={1:('192.0.2.10',4)};first=query(n);loaded=time.monotonic;before=count(n)
records[n]={1:('192.0.2.20',4)};cached=query(n)
check('positive initially fetched',before==1 and '192.0.2.10' in first)
check('positive cache retains old address',count(n)==before and '192.0.2.10' in cached)
time.sleep(max(0,loaded+4.3-time.monotonic));fresh=query(n)
check('positive expiry causes fetch',count(n)==before+1 and '192.0.2.20' in fresh)
check('recursive response flags',bool(re.search(r'flags: qr rd ra;',cached)))
observations['positive']={'first':first,'cached':cached,'expired':fresh,'upstreamRequests':count(n)}
n='new.fund.test.'first=query(n);loaded=time.monotonic;before=count(n);records[n]={1:('192.0.2.30',30)};cached=query(n)
check('NXDOMAIN initially fetched',before==1 and 'status: NXDOMAIN' in first)
check('creation does not invalidate negative cache',count(n)==before and 'status: NXDOMAIN' in cached)
time.sleep(max(0,loaded+4.3-time.monotonic));fresh=query(n)
check('negative expiry reveals created name',count(n)==before+1 and '192.0.2.30' in fresh)
observations['negative']={'first':first,'cached':cached,'expired':fresh,'upstreamRequests':count(n)}
n='dual.fund.test.'records[n]={1:('192.0.2.40',30)};first=query(n,'AAAA');loaded=time.monotonic;before=count(n,28)
records[n][28]=('2001:db8::40',30);cached=query(n,'AAAA');a=query(n,'A')
check('NODATA has NOERROR and SOA', 'status: NOERROR' in first and 'ANSWER: 0, AUTHORITY: 1' in first and 'SOA' in first)
check('AAAA absence stays cached',count(n,28)==before and 'ANSWER: 0, AUTHORITY: 1' in cached)
check('AAAA NODATA does not suppress A',count(n,1)==1 and '192.0.2.40' in a)
time.sleep(max(0,loaded+4.3-time.monotonic));fresh=query(n,'AAAA')
check('AAAA expiry reveals published address',count(n,28)==before+1 and '2001:db8::40' in fresh)
observations['nodata']={'first':first,'cached':cached,'a':a,'expired':fresh,'aaaaRequests':count(n,28),'aRequests':count(n,1)}
n='zero.fund.test.'records[n]={1:('192.0.2.50',0)};first=query(n);records[n]={1:('192.0.2.51',0)};second=query(n)
check('zero TTL returns first data','192.0.2.50' in first)
check('zero TTL refetches next independent query',count(n)==2 and '192.0.2.51' in second)
observations['zeroTTL']={'first':first,'second':second,'upstreamRequests':count(n)}
run_resolver('baseline',baseline)
def cap(query):
n='cap.fund.test.'records[n]={1:('192.0.2.60',30)};first=query(n);loaded=time.monotonic;records[n]={1:('192.0.2.61',30)}
check('maximum TTL lowers client TTL',bool(re.search(r'cap\.fund\.test\.\s+[012]\s+IN\s+A',first)))
time.sleep(max(0,loaded+2.3-time.monotonic));fresh=query(n)
check('maximum TTL refreshes before original 30 seconds',count(n)==2 and '192.0.2.61' in fresh)
observations['maximumTTL']={'first':first,'expired':fresh,'upstreamRequests':count(n)}
run_resolver('cap',cap,' cache-max-ttl: 2\n')
def floor(query):
n='floor.fund.test.'records[n]={1:('192.0.2.70',1)};first=query(n);loaded=time.monotonic;records[n]={1:('192.0.2.71',1)}
time.sleep(max(0,loaded+1.5-time.monotonic));cached=query(n)
check('minimum TTL extends local retention',count(n)==1 and '192.0.2.70' in cached)
time.sleep(max(0,loaded+4.3-time.monotonic));fresh=query(n)
check('minimum TTL eventually refreshes',count(n)==2 and '192.0.2.71' in fresh)
observations['minimumTTL']={'first':first,'afterAuthoritativeTTL':cached,'expired':fresh,'upstreamRequests':count(n)}
run_resolver('floor',floor,' cache-min-ttl: 4\n')
check('all authority requests stay in fixture zone',all(h['name'].endswith('.fund.test.') for h in hits))
finally:
authority.shutdown;authority.server_close;thread.join(timeout=3)
check('authority thread stopped',not thread.is_alive)
evidence={'unboundVersion':version,'digVersion':dig_version,'pythonVersion':__import__('platform').python_version,'observations':observations,'checks':checks,'authorityRequests':hits,'configurations':configurations,'runnerSha256':hashlib.sha256(Path(__file__).read_bytes).hexdigest,'scope':'Actual Unbound 1.26.1 cache with an original UDP authoritative fixture and dig over loopback. Six experiments, three isolated resolver processes. No external DNS, DNSSEC validator, live delegation traversal, operating-system DNS changes or production application. Cache policy experiments are deliberate lab settings, not recommended production defaults.'}
Path(args.output).write_text(json.dumps(evidence,indent=2)+'\n')
print(json.dumps({'passed':len(checks),'experiments':len(observations),'output':args.output}))
if __name__=='__main__':main
Cais published a name after the batch received NXDOMAIN. Compare authority with the batch resolver and observe refresh before repeating the operation.
Common pitfalls
Using latency as sole cache evidence, confusing NODATA with NXDOMAIN, or closing an application failure solely because DNS answers again.
Related topics: Negative answers and SOA · Consumer diagnosis
A useful measurement connects answer, configuration, and upstream request. Lab success does not replace application acceptance.
Reference: Unbound runtime manual and original DR cache experiments · DNS RFC 1034/1035 with RFC 2181, 2308, 3596, 4033, 7766 and 8767; dig BIND 9.20