Validar dados DNSSEC num laboratório isolado

#!/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()

Tópicos relacionados: Cache e falhas de resolução · Transporte e confiança

Referência: 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