+r"""
+Functions and data specific to tinydns logs.
+"""
# Don't clobber the global compile() with a named import.
import re
from typing import Optional
-from djbdns.common import *
+from djbdns.common import QUERY_TYPE_NAME, TIMESTAMP_PAT, convert_ip
# 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}'
+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}'
+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}) (.*)'
+TINYDNS_LOG_RE = 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
# 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 = {
+QUERY_DROP_REASON = {
"+": None,
"-": "no authority",
"I": "invalid query",
def handle_tinydns_log(line : str) -> Optional[str]:
r"""
- Handle a single log line if it matches the ``tinydns_log_re`` regex.
+ 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``.
+ The log line that might match ``TINYDNS_LOG_RE``.
Returns
-------
>>> handle_tinydns_log(line)
"""
- match = tinydns_log_re.match(line)
+ match = TINYDNS_LOG_RE.match(line)
if not match:
return None
- (timestamp, ip, port, id, code, type, name) = match.groups()
+ (timestamp, ip, port, request_id, code, query_type, name) = match.groups()
ip = convert_ip(ip)
port = int(port, 16)
- id = int(id, 16)
+ request_id = int(request_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"
+ # using the query_type dictionary.
+ query_type = int(query_type, 16) # "001c" -> 28
+ query_type = QUERY_TYPE_NAME.get(query_type) # 28 -> "aaaa"
line_tpl = "{timestamp} "
- reason = query_drop_reason[code]
+ reason = QUERY_DROP_REASON[code]
if code == "+":
- line_tpl += "sent response to {ip}:{port} (id {id}): {type} {name}"
+ line_tpl += "sent response to {ip}:{port} (id {request_id}): "
+ line_tpl += "{query_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}"
+ line_tpl += " (id {request_id}): {query_type} {name}"
return line_tpl.format(timestamp=timestamp,
reason=reason,
ip=ip,
port=port,
- id=id,
- type=type,
+ request_id=request_id,
+ query_type=query_type,
name=name)