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