New parser: conntrack #646
Labels
Comments
Closed
|
This is a draft-implementation created using #!/usr/bin/env python3
import re
def parse_conntrack_line(line: str) -> dict:
"""
Parse a single line of conntrack output and return a dictionary
with keys:
- status (e.g. "NEW", "UPDATE", "DESTROY")
- protocol (e.g. "udp", "tcp", "unknown")
- proto_num (numeric protocol, e.g. 17 for UDP, 6 for TCP)
- timeout (integer, if present; else None)
- state (TCP state like SYN_SENT, ESTABLISHED, etc. if present; else None)
- origin (dict of key-value pairs for the original direction, e.g. src=..., dst=..., sport=..., dport=...)
- reply (dict of key-value pairs for the reply direction, similarly)
- flags (list of bracketed flags seen, e.g. ["UNREPLIED", "ASSURED"])
Returns an empty dictionary {} if the line is not parseable.
"""
# Strip and skip blank lines
line = line.strip()
if not line:
return {}
# Split tokens by whitespace
tokens = line.split()
# The first token should be something like "[NEW]" or "[UPDATE]" or "[DESTROY]"
# If it does not look like a bracket, skip/return empty.
if not tokens[0].startswith("[") or not tokens[0].endswith("]"):
return {}
status = tokens[0].strip("[]")
# Next token is the textual protocol: "tcp", "udp", "unknown", etc.
if len(tokens) < 2:
return {}
protocol = tokens[1]
# Third token is typically the numeric protocol (e.g. "17", "6", or "2")
# Bail out if we don't have enough tokens
if len(tokens) < 3:
return {}
proto_num_str = tokens[2]
if not proto_num_str.isdigit():
return {}
proto_num = int(proto_num_str)
# Fourth token may be the timeout (if it's numeric) or might be a TCP state (e.g. SYN_SENT).
idx = 3
timeout = None
state = None
# Check if the current token is a digit => it is the timeout
if idx < len(tokens) and tokens[idx].isdigit():
timeout = int(tokens[idx])
idx += 1
# After that, we *might* have a TCP state (like SYN_SENT, SYN_RECV, ESTABLISHED, etc.)
# or we might already be at key/value pairs: src=..., dst=..., etc.
if idx < len(tokens) and not tokens[idx].startswith("src="):
# Assume it's a TCP/conntrack state
# (For UDP lines, there is often no such extra state token.)
possible_state = tokens[idx]
# Heuristic: if it's bracketed like "[UNREPLIED]" we treat that as a flag, not a state
# but typically you see bracketed tokens separate from these fields.
if not possible_state.startswith("["):
state = possible_state
idx += 1
# Now we expect pairs (like src=..., dst=..., sport=..., dport=...)
# possibly interspersed with bracketed flags such as [UNREPLIED], [ASSURED].
# We also expect to see two "directions": "origin" and "reply".
def parse_keyvalue_pairs(tokens, start_idx):
"""
Parse key=value pairs from tokens starting at start_idx
until we hit a bracket token (e.g. [UNREPLIED]) or run out of tokens
or find a token that doesn't look like key=value.
Returns (dict_of_parsed_pairs, new_index)
"""
pairs = {}
i = start_idx
while i < len(tokens):
# If we see a bracket, stop (those are flags or other markers)
if tokens[i].startswith("["):
break
# If it's "key=value", parse it
if '=' in tokens[i]:
key, value = tokens[i].split('=', 1)
pairs[key] = value
i += 1
else:
# Something else that doesn't match key=value, break out
break
return pairs, i
flags = []
# Parse origin part
origin, idx = parse_keyvalue_pairs(tokens, idx)
# Collect bracketed flags (if any)
while idx < len(tokens) and tokens[idx].startswith("["):
flags.append(tokens[idx].strip("[]"))
idx += 1
# Parse reply part
reply, idx = parse_keyvalue_pairs(tokens, idx)
# Collect any trailing bracketed flags
while idx < len(tokens) and tokens[idx].startswith("["):
flags.append(tokens[idx].strip("[]"))
idx += 1
# Build the final parsed structure
parsed = {
"status": status,
"protocol": protocol,
"proto_num": proto_num,
"timeout": timeout,
"state": state,
"origin": origin,
"reply": reply,
"flags": flags
}
return parsed
def parse_conntrack_output(conntrack_text: str) -> list:
"""
Given the full text output from `conntrack -L`,
split it into lines and parse each line into a dictionary structure.
Return a list of the parsed results (one dict per valid line).
"""
results = []
for line in conntrack_text.splitlines():
line = line.strip()
if not line:
continue
parsed = parse_conntrack_line(line)
if parsed:
results.append(parsed)
return results
if __name__ == "__main__":
# Example usage:
example_output = r"""
[NEW] unknown 2 600 src=192.168.136.128 dst=224.0.0.22 [UNREPLIED] src=224.0.0.22 dst=192.168.136.128
[NEW] udp 17 30 src=192.168.136.128 dst=8.8.8.8 sport=36037 dport=53 [UNREPLIED] src=8.8.8.8 dst=192.168.136.128 sport=53 dport=36037
[UPDATE] udp 17 30 src=192.168.136.128 dst=8.8.8.8 sport=36037 dport=53 src=8.8.8.8 dst=192.168.136.128 sport=53 dport=36037
[NEW] tcp 6 120 SYN_SENT src=192.168.136.128 dst=91.189.91.48 sport=38784 dport=80 [UNREPLIED] src=91.189.91.48 dst=192.168.136.128 sport=80 dport=38784
[UPDATE] tcp 6 60 SYN_RECV src=192.168.136.128 dst=91.189.91.48 sport=38784 dport=80 src=91.189.91.48 dst=192.168.136.128 sport=80 dport=38784
[UPDATE] tcp 6 432000 ESTABLISHED src=192.168.136.128 dst=91.189.91.48 sport=38784 dport=80 src=91.189.91.48 dst=192.168.136.128 sport=80 dport=38784 [ASSURED]
[DESTROY] udp 17 src=192.168.136.128 dst=8.8.8.8 sport=42284 dport=53 src=8.8.8.8 dst=192.168.136.128 sport=53 dport=42284
"""
parsed_lines = parse_conntrack_output(example_output)
for entry in parsed_lines:
print(entry) |
|
Thank you for the parser suggestion and the reference code! I'll start working on the next release soon and we can probably get this in. |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
and others
Currently
jchas support for several networking related CLI's, but it doesn't have support forconntrack. I can't find many conntrack parsers online, most of them haven't been updated in +5 years.Package name:
conntrackExample Output: from running
conntrack -E/conntrack -Land connecting/disconnecting the VM to the internet:The text was updated successfully, but these errors were encountered: