*/*: reorganize for easy testing using setuptools or tox.

[djbdns-logparse.git] / djbdns / tinydns.py

diff --git a/djbdns/tinydns.py b/djbdns/tinydns.py
new file mode 100644 (file)
index 0000000..52b06a4
--- /dev/null
+++ b/djbdns/tinydns.py
@@ -0,0 +1,95 @@
+from re import compile
+from typing import Optional
+from djbdns.common import *
+
+# 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}'
+
+# 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 = compile(
+    rf'({timestamp_pat}) ({ip_pat}):({hex4_pat}):({hex4_pat}) ([\+\-IC/]) ({hex4_pat}) (.*)'
+)
+
+# 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. We include the "+" case here, indicating that the query was
+# NOT dropped, to avoid a special case later on when we're formatting
+# the human-readable output.
+query_drop_reason = {
+    "+": None,
+    "-": "no authority",
+    "I": "invalid query",
+    "C": "invalid class",
+    "/": "couldn't parse"
+}
+
+
+def handle_tinydns_log(line : str) -> Optional[str]:
+    r"""
+    Handle a single log line if it matches the ``tinydns_log_re`` regex.
+
+    Parameters
+    ----------
+
+    line : string
+        The log line that might match ``tinydns_log_re``.
+
+    Returns
+    -------
+
+    Either the human-readable string if the log line was handled (that
+    is, if it was really a tinydns log line), or ``None`` if it was
+    not.
+
+    Examples
+    --------
+
+        >>> line = "2022-09-14 21:04:40.206516500 7f000001:9d61:be69 - 0001 www.example.com"
+        >>> handle_tinydns_log(line)
+        '2022-09-14 21:04:40.206516500 dropped query (no authority) from 127.0.0.1:40289 (id 48745): a www.example.com'
+
+        >>> line = "this line is nonsense"
+        >>> handle_tinydns_log(line)
+
+    """
+    match = tinydns_log_re.match(line)
+    if not match:
+        return None
+
+    (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"
+
+    line_tpl = "{timestamp} "
+
+    reason = query_drop_reason[code]
+    if code == "+":
+        line_tpl += "sent response to {ip}:{port} (id {id}): {type} {name}"
+    else:
+        line_tpl += "dropped query ({reason}) from {ip}:{port}"
+        if code != "/":
+            # If the query can actually be parsed, the log line is a
+            # bit more informative than it would have been otherwise.
+            line_tpl += " (id {id}): {type} {name}"
+
+    return line_tpl.format(timestamp=timestamp,
+                           reason=reason,
+                           ip=ip,
+                           port=port,
+                           id=id,
+                           type=type,
+                           name=name)