gistfile1.txt
· 4.7 KiB · Text
Raw
def fw_events(file, filter_ip, filter_type, clients_dir, net_file, limit):
"""
Format firewall drop events with dedup, counts, and service annotation.
Output per line: ts|client|dest_ip|dest_port|proto|service_name|count
"""
import glob
from datetime import datetime
proto_map = {1: 'icmp', 6: 'tcp', 17: 'udp'}
# Build ip->name map
ip_to_name = {}
for conf in glob.glob(f"{clients_dir}/*.conf"):
name = os.path.basename(conf).replace('.conf', '')
try:
with open(conf) as f:
for line in f:
if line.startswith('Address'):
ip = line.split('=')[1].strip().split('/')[0]
ip_to_name[ip] = name
except Exception:
pass
# Load net services for reverse lookup — independent of rest of function
net_data = {}
if net_file and os.path.exists(net_file):
try:
with open(net_file) as f:
net_data = json.load(f)
except Exception:
pass
def reverse_lookup(dest_ip, dest_port, proto):
for svc_name, svc in net_data.items():
if not isinstance(svc, dict):
continue
if svc.get('ip', '') != dest_ip:
continue
ports = svc.get('ports', {})
if dest_port:
for port_name, port_def in ports.items():
if not isinstance(port_def, dict):
continue
if (str(port_def.get('port', '')) == str(dest_port) and
port_def.get('proto', 'tcp') == proto):
return f"{svc_name}:{port_name}"
return svc_name
return svc_name
return ''
# Parse and first-pass dedup (within time window per key)
events = []
last_seen = {}
try:
with open(file) as f:
for line in f:
try:
e = json.loads(line.strip())
src = e.get('src_ip', '')
if not src:
continue
if filter_ip and src != filter_ip:
continue
proto_num = int(e.get('ip.protocol', 0))
proto = proto_map.get(proto_num, str(proto_num))
dst = e.get('dest_ip', '')
port = str(e.get('dest_port', ''))
key = (src, dst, port, proto_num)
ts_str = e.get('timestamp', '')
try:
ts = datetime.fromisoformat(ts_str).timestamp()
except Exception:
ts = 0
windows = {1: 5, 6: 30, 17: 10}
window = windows.get(proto_num, 10)
if key in last_seen and (ts - last_seen[key]) < window:
continue
def wg_events(file, filter_client, filter_type, limit):
"""
Format WireGuard events with dedup and counts.
Output per line: ts|client|endpoint|event|count
"""
from datetime import datetime
events = []
try:
with open(file) as f:
for line in f:
try:
e = json.loads(line.strip())
client = e.get('client', '')
if not client:
continue
if filter_client and client != filter_client:
continue
if filter_type and not client.startswith(filter_type + '-'):
continue
events.append(e)
except Exception:
pass
except Exception:
pass
# Dedup consecutive same client+event+endpoint within 60s
deduped = []
counts = []
for e in events:
ts_str = e.get('timestamp', '')
try:
ts = datetime.fromisoformat(ts_str).timestamp()
except Exception:
ts = 0
client = e.get('client', '')
event = e.get('event', '')
endpoint = e.get('endpoint', '')
key = (client, event, endpoint[:15])
if deduped:
prev = deduped[-1]
prev_ts_str = prev.get('timestamp', '')
try:
prev_ts = datetime.fromisoformat(prev_ts_str).timestamp()
except Exception:
prev_ts = 0
prev_key = (
prev.get('client', ''),
prev.get('event', ''),
prev.get('endpoint', '')[:15]
)
if key == prev_key and (ts - prev_ts) < 300:
counts[-1] += 1
continue
deduped.append(e)
counts.append(1)
limit = int(limit) if limit else 50
| 1 | def fw_events(file, filter_ip, filter_type, clients_dir, net_file, limit): |
| 2 | """ |
| 3 | Format firewall drop events with dedup, counts, and service annotation. |
| 4 | Output per line: ts|client|dest_ip|dest_port|proto|service_name|count |
| 5 | """ |
| 6 | import glob |
| 7 | from datetime import datetime |
| 8 | proto_map = {1: 'icmp', 6: 'tcp', 17: 'udp'} |
| 9 | |
| 10 | # Build ip->name map |
| 11 | ip_to_name = {} |
| 12 | for conf in glob.glob(f"{clients_dir}/*.conf"): |
| 13 | name = os.path.basename(conf).replace('.conf', '') |
| 14 | try: |
| 15 | with open(conf) as f: |
| 16 | for line in f: |
| 17 | if line.startswith('Address'): |
| 18 | ip = line.split('=')[1].strip().split('/')[0] |
| 19 | ip_to_name[ip] = name |
| 20 | except Exception: |
| 21 | pass |
| 22 | |
| 23 | # Load net services for reverse lookup — independent of rest of function |
| 24 | net_data = {} |
| 25 | if net_file and os.path.exists(net_file): |
| 26 | try: |
| 27 | with open(net_file) as f: |
| 28 | net_data = json.load(f) |
| 29 | except Exception: |
| 30 | pass |
| 31 | |
| 32 | def reverse_lookup(dest_ip, dest_port, proto): |
| 33 | for svc_name, svc in net_data.items(): |
| 34 | if not isinstance(svc, dict): |
| 35 | continue |
| 36 | if svc.get('ip', '') != dest_ip: |
| 37 | continue |
| 38 | ports = svc.get('ports', {}) |
| 39 | if dest_port: |
| 40 | for port_name, port_def in ports.items(): |
| 41 | if not isinstance(port_def, dict): |
| 42 | continue |
| 43 | if (str(port_def.get('port', '')) == str(dest_port) and |
| 44 | port_def.get('proto', 'tcp') == proto): |
| 45 | return f"{svc_name}:{port_name}" |
| 46 | return svc_name |
| 47 | return svc_name |
| 48 | return '' |
| 49 | |
| 50 | # Parse and first-pass dedup (within time window per key) |
| 51 | events = [] |
| 52 | last_seen = {} |
| 53 | |
| 54 | try: |
| 55 | with open(file) as f: |
| 56 | for line in f: |
| 57 | try: |
| 58 | e = json.loads(line.strip()) |
| 59 | src = e.get('src_ip', '') |
| 60 | if not src: |
| 61 | continue |
| 62 | if filter_ip and src != filter_ip: |
| 63 | continue |
| 64 | |
| 65 | proto_num = int(e.get('ip.protocol', 0)) |
| 66 | proto = proto_map.get(proto_num, str(proto_num)) |
| 67 | dst = e.get('dest_ip', '') |
| 68 | port = str(e.get('dest_port', '')) |
| 69 | key = (src, dst, port, proto_num) |
| 70 | |
| 71 | ts_str = e.get('timestamp', '') |
| 72 | try: |
| 73 | ts = datetime.fromisoformat(ts_str).timestamp() |
| 74 | except Exception: |
| 75 | ts = 0 |
| 76 | |
| 77 | windows = {1: 5, 6: 30, 17: 10} |
| 78 | window = windows.get(proto_num, 10) |
| 79 | if key in last_seen and (ts - last_seen[key]) < window: |
| 80 | continue |
| 81 | def wg_events(file, filter_client, filter_type, limit): |
| 82 | """ |
| 83 | Format WireGuard events with dedup and counts. |
| 84 | Output per line: ts|client|endpoint|event|count |
| 85 | """ |
| 86 | from datetime import datetime |
| 87 | |
| 88 | events = [] |
| 89 | try: |
| 90 | with open(file) as f: |
| 91 | for line in f: |
| 92 | try: |
| 93 | e = json.loads(line.strip()) |
| 94 | client = e.get('client', '') |
| 95 | if not client: |
| 96 | continue |
| 97 | if filter_client and client != filter_client: |
| 98 | continue |
| 99 | if filter_type and not client.startswith(filter_type + '-'): |
| 100 | continue |
| 101 | events.append(e) |
| 102 | except Exception: |
| 103 | pass |
| 104 | except Exception: |
| 105 | pass |
| 106 | |
| 107 | # Dedup consecutive same client+event+endpoint within 60s |
| 108 | deduped = [] |
| 109 | counts = [] |
| 110 | for e in events: |
| 111 | ts_str = e.get('timestamp', '') |
| 112 | try: |
| 113 | ts = datetime.fromisoformat(ts_str).timestamp() |
| 114 | except Exception: |
| 115 | ts = 0 |
| 116 | client = e.get('client', '') |
| 117 | event = e.get('event', '') |
| 118 | endpoint = e.get('endpoint', '') |
| 119 | key = (client, event, endpoint[:15]) |
| 120 | |
| 121 | if deduped: |
| 122 | prev = deduped[-1] |
| 123 | prev_ts_str = prev.get('timestamp', '') |
| 124 | try: |
| 125 | prev_ts = datetime.fromisoformat(prev_ts_str).timestamp() |
| 126 | except Exception: |
| 127 | prev_ts = 0 |
| 128 | prev_key = ( |
| 129 | prev.get('client', ''), |
| 130 | prev.get('event', ''), |
| 131 | prev.get('endpoint', '')[:15] |
| 132 | ) |
| 133 | if key == prev_key and (ts - prev_ts) < 300: |
| 134 | counts[-1] += 1 |
| 135 | continue |
| 136 | |
| 137 | deduped.append(e) |
| 138 | counts.append(1) |
| 139 | |
| 140 | limit = int(limit) if limit else 50 |
| 141 |