Convert tinydns and dnscache logs to human-readable form
"""
-import re
+import re, typing
from struct import pack
from time import strftime, gmtime
from subprocess import Popen, PIPE
-# common components of line-matching regexes
-timestamp_pat = r'[\d-]+ [\d:\.]+' # output of tai64nlocal
+## Regular expressions for matching tinydns/dnscache log lines. We
+## compile these once here rather than within the corresponding
+## matching functions, because the latter get executed repeatedly.
+
+# This first pattern is used to match the timestamp format that the
+# tai64nlocal program produces. It appears in both dnscache and
+# tinydns lines, after they've been piped through tai64nlocal, of
+# course.
+timestamp_pat = r'[\d-]+ [\d:\.]+'
+
+# The regex to match dnscache log lines.
+dnscache_log_re = re.compile(fr'({timestamp_pat}) (\w+)(.*)')
+
+# The "hex4" pattern matches a string of four hexadecimal digits. This
+# is used, for example, by tinydns to encode the query type
+# identifier.
hex4_pat = r'[0-9a-f]{4}'
-ip_pat = r'[0-9a-f]{8,32}' # IPv4 or IPv6 addresses in hex
-# discriminate between dnscache and tinydns log lines
-tinydns_log_re = re.compile(
- r'(%s) (%s):(%s):(%s) ([\+\-IC/]) (%s) (.*)'
- % (timestamp_pat, ip_pat, hex4_pat, hex4_pat, hex4_pat))
-dnscache_log_re = re.compile(r'(%s) (\w+)(.*)' % timestamp_pat)
+# The IP pattern matches a string of either 8 or 32 hexadecimal
+# characters, which correspond to IPv4 and IPv6 addresses,
+# respectively, in tinydns logs.
+ip_pat = r'[0-9a-f]{8,32}'
+# The regex to match tinydns log lines.
+tinydns_log_re = re.compile(
+ rf'({timestamp_pat}) ({ip_pat}):({hex4_pat}):({hex4_pat}) ([\+\-IC/]) ({hex4_pat}) (.*)'
+)
+
+# A dictionary mapping query type identifiers, in decimal, to their
+# friendly names for tinydns. Reference:
+#
+# https://en.wikipedia.org/wiki/List_of_DNS_record_types
+#
+# Note that mapping here is non-exhaustive, and that tinydns will
+# log responses for record types that it does not know about.
query_type = {
1: "a",
2: "ns",
257: "caa"
}
-# for tinydns only
+# tinydns can drop a query for one of three reasons; this dictionary
+# maps the symbol that gets logged in each case to a human-readable
+# reason.
query_drop_reason = {
"-": "no authority",
"I": "invalid query",
"C": "invalid class",
- }
+}
def convert_ip(ip : str):
return ":".join([ip[(4*i) : (4*i+4)] for i in range(8)])
-def _cvt_ip(match):
- return convert_ip(match.group(1))
-
-def _cvt_port(match):
- return ":" + str(int(match.group(1), 16))
-
def decode_client(words, i):
chunks = words[i].split(":")
if len(chunks) == 2: # ip:port
print(timestamp, event, " ".join(words))
-def handle_tinydns_log(line, match):
+def handle_tinydns_log(line : str, match: re.Match):
+ """
+ Handle a line that matched the ``tinydns_log_re`` regex.
+
+ Parameters
+ ----------
+
+ line : string
+ The tinydns log line that matched ``tinydns_log_re``.
+
+ match : re.Match
+ The match object that was returned when ``line`` was
+ tested against ``tinydns_log_re``.
+
+ Examples
+ --------
+
+ >>> line = "2022-09-14 21:04:40.206516500 7f000001:9d61:be69 - 0001 www.example.com"
+ >>> match = tinydns_log_re.match(line)
+ >>> handle_tinydns_log(line, match)
+ 2022-09-14 21:04:40.206516500 dropped query (no authority) from 127.0.0.1:40289 (id 48745): a www.example.com
+
+ """
(timestamp, ip, port, id, code, type, name) = match.groups()
ip = convert_ip(ip)
port = int(port, 16)
id = int(id, 16)
+
+ # Convert the "type" field to a human-readable record type name
+ # using the query_type dictionary. If the right name isn't present
+ # in the dictionary, we use the (decimal) type id instead.
type = int(type, 16) # "001c" -> 28
type = query_type.get(type, type) # 28 -> "aaaa"
% (code, ip, port, id, type, name))
-def parse_logfile(file):
+def parse_logfile(file : typing.TextIO):
+ """
+ Process a single log ``file``.
+
+ Parameters
+ ----------
+
+ file : typing.TextIO
+ An open log file, or stdin.
+
+ """
# Open pipe to tai64nlocal: we will write lines of our input (the
# raw log file) to it, and read log lines with readable timestamps
# from it.
default=[stdin],
help="djbdns logfile to process (default: stdin)")
+ # Warning: argparse automatically opens its file arguments here,
+ # and they only get closed when the program terminates. There's no
+ # real benefit to closing them one-at-a-time after calling
+ # parse_logfile(), because the "scarce" resource of open file
+ # descriptors gets consumed immediately, before any processing has
+ # happened. In other words, if you're going to run out of file
+ # descriptors, it's going to happen right now.
+ #
+ # So anyway, don't run this on several million logfiles.
args = parser.parse_args()
for f in args.logfiles:
parse_logfile(f)
-
-
if __name__ == "__main__":
main()
projects / djbdns-logparse.git / blobdiff