upstream/knot-dns
1
0
Fork 0
mirror of https://gitlab.nic.cz/knot/knot-dns.git synced 2026-02-03 18:49:28 -05:00
Code Issues Projects Releases Packages Wiki Activity Actions

samples/probe_dump.py: use python idiomatic formatting

Browse source
This commit is contained in:
Jan Doskočil 2026-01-14 12:44:48 +01:00 committed by Daniel Salzman
parent 1729a53709
commit 863f32d8f0
1 changed files with 25 additions and 20 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

45
samples/probe_dump.py
View file

@ -12,55 +12,60 @@ import libknot
import libknot.probe import libknot.probe
import sys import sys
def extract_dns_header(probe, fieldname): def extract_dns_header(probe, fieldname):
val = getattr(probe, fieldname) val = getattr(probe, fieldname)
return { return {
"additionals": val.additionals, "additionals": val.additionals,
"answers": val.answers, "answers": val.answers,
"authorities": val.authorities, "authorities": val.authorities,
"flag_aa": val.flag_aa, "flag_aa": val.flag_aa,
"flag_ad": val.flag_ad, "flag_ad": val.flag_ad,
"flag_cd": val.flag_cd, "flag_cd": val.flag_cd,
"flag_qr": val.flag_qr, "flag_qr": val.flag_qr,
"flag_ra": val.flag_ra, "flag_ra": val.flag_ra,
"flag_rd": val.flag_rd, "flag_rd": val.flag_rd,
"flag_tc": val.flag_tc, "flag_tc": val.flag_tc,
"flag_z": val.flag_z, "flag_z": val.flag_z,
"id": val.id, "id": val.id,
"opcode": val.opcode, "opcode": val.opcode,
"questions": val.questions, "questions": val.questions,
"rcode": val.rcode, "rcode": val.rcode,
} }
def extract_addr(probe, fieldname): def extract_addr(probe, fieldname):
val = getattr(probe, fieldname) val = getattr(probe, fieldname)
return probe.addr_str(val) return probe.addr_str(val)
def extract_safe(probe, fieldname): def extract_safe(probe, fieldname):
convert_func = getattr convert_func = getattr
try: try:
convert_func = { convert_func = {
'local_addr': extract_addr, "local_addr": extract_addr,
'remote_addr': extract_addr, "remote_addr": extract_addr,
'query_hdr': extract_dns_header, "query_hdr": extract_dns_header,
'reply_hdr': extract_dns_header, "reply_hdr": extract_dns_header,
'query_name': lambda probe, _: probe.qname_str(), "query_name": lambda probe, _: probe.qname_str(),
}[fieldname] }[fieldname]
except KeyError: except KeyError:
pass pass
return convert_func(probe, fieldname) return convert_func(probe, fieldname)
def convert_json(probe): def convert_json(probe):
probe_as_dict = {} probe_as_dict = {}
for field in probe._fields_: for field in probe._fields_:
fieldname = field[0] fieldname = field[0]
fieldtype = field[1] fieldtype = field[1]
fieldlen = field[2] if len(field) > 2 else None fieldlen = field[2] if len(field) > 2 else None
probe_as_dict[fieldname] = extract_safe(probe, fieldname) probe_as_dict[fieldname] = extract_safe(probe, fieldname)
return probe_as_dict return probe_as_dict
def probe_loop(args): def probe_loop(args):
try: try:
libknot.Knot(args.libknot_path) libknot.Knot(args.libknot_path)
@ -72,7 +77,7 @@ def probe_loop(args):
data = libknot.probe.KnotProbeDataArray(8) data = libknot.probe.KnotProbeDataArray(8)
try: try:
while (True): while True:
if probe.consume(data, 1000) > 0: if probe.consume(data, 1000) > 0:
for item in data: for item in data:
if args.json: if args.json: