gistfile1.txt
· 4.2 KiB · Text
Brut
def fw_events(file, filter_ip, filter_type, clients_dir, limit):
"""Format firewall drop events with dedup and counts"""
import glob
proto_map = {1: 'icmp', 6: 'tcp', 17: 'udp'}
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:
pass
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
dst = e.get('dest_ip', '')
port = e.get('dest_port', '')
proto = e.get('ip.protocol', 0)
key = (src, dst, port, proto)
ts_str = e.get('timestamp', '')
try:
from datetime import datetime
ts = datetime.fromisoformat(ts_str).timestamp()
except:
ts = 0
windows = {1: 5, 6: 30, 17: 10}
window = windows.get(proto, 10)
if key in last_seen and (ts - last_seen[key]) < window:
continue
last_seen[key] = ts
events.append(e)
except:
pass
except:
pass
# Dedup consecutive same src+dst+port within 60s with count
deduped = []
counts = []
for e in events:
ts_str = e.get('timestamp', '')
try:
from datetime import datetime
ts = datetime.fromisoformat(ts_str).timestamp()
except:
def wg_events(file, filter_client, filter_type, limit):
"""Format WireGuard events from events.log with dedup"""
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:
pass
except:
pass
# Dedup consecutive same client+event+endpoint within 60s
deduped = []
counts = []
for e in events:
ts_str = e.get('timestamp', '')
try:
from datetime import datetime
ts = datetime.fromisoformat(ts_str).timestamp()
except:
ts = 0
client = e.get('client', '')
event = e.get('event', '')
endpoint = e.get('endpoint', '')
key = (client, event, endpoint[:15])
if deduped and counts:
prev = deduped[-1]
prev_ts_str = prev.get('timestamp', '')
try:
def format_fw_event(line, clients_dir):
"""Format a single fw_event line"""
import glob
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 l in f:
if l.startswith('Address'):
ip = l.split('=')[1].strip().split('/')[0]
ip_to_name[ip] = name
except:
pass
try:
e = json.loads(line.strip())
src = e.get('src_ip', '')
if not src:
return None
ts = e.get('timestamp', '')
try:
from datetime import datetime
dt = datetime.fromisoformat(ts)
ts = dt.strftime(DATETIME_FMT)
except:
pass
| 1 | def fw_events(file, filter_ip, filter_type, clients_dir, limit): |
| 2 | """Format firewall drop events with dedup and counts""" |
| 3 | import glob |
| 4 | proto_map = {1: 'icmp', 6: 'tcp', 17: 'udp'} |
| 5 | |
| 6 | ip_to_name = {} |
| 7 | for conf in glob.glob(f"{clients_dir}/*.conf"): |
| 8 | name = os.path.basename(conf).replace('.conf', '') |
| 9 | try: |
| 10 | with open(conf) as f: |
| 11 | for line in f: |
| 12 | if line.startswith('Address'): |
| 13 | ip = line.split('=')[1].strip().split('/')[0] |
| 14 | ip_to_name[ip] = name |
| 15 | except: |
| 16 | pass |
| 17 | |
| 18 | events = [] |
| 19 | last_seen = {} |
| 20 | |
| 21 | try: |
| 22 | with open(file) as f: |
| 23 | for line in f: |
| 24 | try: |
| 25 | e = json.loads(line.strip()) |
| 26 | src = e.get('src_ip', '') |
| 27 | if not src: |
| 28 | continue |
| 29 | if filter_ip and src != filter_ip: |
| 30 | continue |
| 31 | dst = e.get('dest_ip', '') |
| 32 | port = e.get('dest_port', '') |
| 33 | proto = e.get('ip.protocol', 0) |
| 34 | key = (src, dst, port, proto) |
| 35 | ts_str = e.get('timestamp', '') |
| 36 | try: |
| 37 | from datetime import datetime |
| 38 | ts = datetime.fromisoformat(ts_str).timestamp() |
| 39 | except: |
| 40 | ts = 0 |
| 41 | windows = {1: 5, 6: 30, 17: 10} |
| 42 | window = windows.get(proto, 10) |
| 43 | if key in last_seen and (ts - last_seen[key]) < window: |
| 44 | continue |
| 45 | last_seen[key] = ts |
| 46 | events.append(e) |
| 47 | except: |
| 48 | pass |
| 49 | except: |
| 50 | pass |
| 51 | |
| 52 | # Dedup consecutive same src+dst+port within 60s with count |
| 53 | deduped = [] |
| 54 | counts = [] |
| 55 | for e in events: |
| 56 | ts_str = e.get('timestamp', '') |
| 57 | try: |
| 58 | from datetime import datetime |
| 59 | ts = datetime.fromisoformat(ts_str).timestamp() |
| 60 | except: |
| 61 | def wg_events(file, filter_client, filter_type, limit): |
| 62 | """Format WireGuard events from events.log with dedup""" |
| 63 | events = [] |
| 64 | try: |
| 65 | with open(file) as f: |
| 66 | for line in f: |
| 67 | try: |
| 68 | e = json.loads(line.strip()) |
| 69 | client = e.get('client', '') |
| 70 | if not client: |
| 71 | continue |
| 72 | if filter_client and client != filter_client: |
| 73 | continue |
| 74 | if filter_type and not client.startswith(filter_type + '-'): |
| 75 | continue |
| 76 | events.append(e) |
| 77 | except: |
| 78 | pass |
| 79 | except: |
| 80 | pass |
| 81 | |
| 82 | # Dedup consecutive same client+event+endpoint within 60s |
| 83 | deduped = [] |
| 84 | counts = [] |
| 85 | for e in events: |
| 86 | ts_str = e.get('timestamp', '') |
| 87 | try: |
| 88 | from datetime import datetime |
| 89 | ts = datetime.fromisoformat(ts_str).timestamp() |
| 90 | except: |
| 91 | ts = 0 |
| 92 | client = e.get('client', '') |
| 93 | event = e.get('event', '') |
| 94 | endpoint = e.get('endpoint', '') |
| 95 | key = (client, event, endpoint[:15]) |
| 96 | |
| 97 | if deduped and counts: |
| 98 | prev = deduped[-1] |
| 99 | prev_ts_str = prev.get('timestamp', '') |
| 100 | try: |
| 101 | def format_fw_event(line, clients_dir): |
| 102 | """Format a single fw_event line""" |
| 103 | import glob |
| 104 | proto_map = {1: 'icmp', 6: 'tcp', 17: 'udp'} |
| 105 | |
| 106 | # Build ip->name map |
| 107 | ip_to_name = {} |
| 108 | for conf in glob.glob(f"{clients_dir}/*.conf"): |
| 109 | name = os.path.basename(conf).replace('.conf', '') |
| 110 | try: |
| 111 | with open(conf) as f: |
| 112 | for l in f: |
| 113 | if l.startswith('Address'): |
| 114 | ip = l.split('=')[1].strip().split('/')[0] |
| 115 | ip_to_name[ip] = name |
| 116 | except: |
| 117 | pass |
| 118 | |
| 119 | try: |
| 120 | e = json.loads(line.strip()) |
| 121 | src = e.get('src_ip', '') |
| 122 | if not src: |
| 123 | return None |
| 124 | ts = e.get('timestamp', '') |
| 125 | try: |
| 126 | from datetime import datetime |
| 127 | dt = datetime.fromisoformat(ts) |
| 128 | ts = dt.strftime(DATETIME_FMT) |
| 129 | except: |
| 130 | pass |