nessi
67eee0662a
Add RuntimeSettingsRead/RuntimeSettingsUpdate schemas with auto_node_sync_enabled/auto_node_sync_interval_minutes/auto_ipam_sync_enabled/auto_ipam_sync_interval_minutes/last_node_auto_sync_at/last_ipam_auto_sync_at fields, implement firewall_log_lines_from_files to parse /var/log/pve-firewall.log with max_lines limit and error collection, extend collect_firewall_log
31 lines
1.0 KiB
Python
31 lines
1.0 KiB
Python
import importlib.util
|
|
from pathlib import Path
|
|
|
|
|
|
def load_agent_module():
|
|
path = Path(__file__).resolve().parents[1] / "app" / "agent_assets" / "nexafabric-agent.py"
|
|
spec = importlib.util.spec_from_file_location("nexafabric_agent", path)
|
|
assert spec and spec.loader
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def test_parse_proxmox_reject_firewall_log_line():
|
|
agent = load_agent_module()
|
|
line = (
|
|
"100 6 tap100i0-IN 09/Jul/2026:23:04:58 +0200 REJECT: IN=fwbr100i0 OUT=fwbr100i0 "
|
|
"PHYSIN=fwln100i0 PHYSOUT=tap100i0 SRC=172.16.155.74 DST=172.16.0.100 LEN=52 "
|
|
"TTL=128 ID=47107 PROTO=TCP SPT=58945 DPT=80"
|
|
)
|
|
|
|
flow = agent.parse_firewall_log_line(line)
|
|
|
|
assert flow["source_ip"] == "172.16.155.74"
|
|
assert flow["destination_ip"] == "172.16.0.100"
|
|
assert flow["protocol"] == "tcp"
|
|
assert flow["source_port"] == 58945
|
|
assert flow["destination_port"] == 80
|
|
assert flow["decision"] == "blocked"
|
|
assert flow["state"] == "blocked"
|