diff --git a/README.md b/README.md index 473a0f4..e2bdcae 100644 --- a/README.md +++ b/README.md @@ -0,0 +1,71 @@ +# fgAI + +Local FortiGate log and policy inspection agent. It parses FortiGate syslog/JSONL logs, audits FortiOS policy exports, highlights UTM events, and can quarantine malicious source IPs through the FortiGate API when explicitly enabled. + +Autoblocking is dry-run by default. The tool will not block RFC1918, loopback, multicast, link-local, reserved, or allowlisted addresses unless you change the code. + +## Quick Start + +```bash +python -m venv .venv +source .venv/bin/activate +pip install -e . +``` + +Analyze local logs: + +```bash +fgai analyze-logs --logs logs/fg_syslog.jsonl +``` + +Audit a FortiGate policy export: + +```bash +fgai audit-policies --config exports/fortigate.conf +``` + +Find block candidates without changing the firewall: + +```bash +fgai suggest-blocks --logs logs/fg_syslog.jsonl +``` + +Execute guarded quarantine actions: + +```bash +export FORTIGATE_HOST=192.0.2.10 +export FORTIGATE_API_TOKEN='...' +fgai suggest-blocks --logs logs/fg_syslog.jsonl --execute --expiry-minutes 60 +``` + +Optional local LLM summary through Ollama: + +```bash +ollama pull llama3.3 +fgai analyze-logs --logs logs/fg_syslog.jsonl --llm +``` + +## FortiGate Inputs + +For logs, configure FortiGate syslog to write into a local file such as `logs/fg_syslog.jsonl`. The parser supports common key/value syslog lines and JSONL. + +For policies, export a FortiOS config backup and pass it to `audit-policies`. + +## Environment + +- `FORTIGATE_HOST`: firewall hostname or IP. +- `FORTIGATE_API_TOKEN`: REST API token. +- `FORTIGATE_VERIFY_TLS`: `true` or `false`, defaults to `true`. +- `FGAI_ALLOWLIST`: comma-separated IPs/CIDRs never to block. +- `OLLAMA_HOST`: defaults to `http://127.0.0.1:11434`. +- `OLLAMA_MODEL`: defaults to `llama3.3`. + +## Safety Model + +The agent separates detection from enforcement: + +- UTM events are scored from FortiGate logs (`ips`, `virus`, `anomaly`, `ddos`, `webfilter`, `app-ctrl`, `waf`, `dns`). +- Source IPs must be globally routable and outside the allowlist. +- Blocking requires `--execute`. +- The FortiGate API call is limited to the quarantine/banned user monitor endpoint. + diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..8c03269 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,9 @@ +[project] +name = "fgai" +version = "0.1.0" +description = "Local FortiGate log and policy inspection agent with guarded UTM autoblocking." +requires-python = ">=3.11" +dependencies = [] + +[project.scripts] +fgai = "fgai.cli:main" diff --git a/src/fgai/__init__.py b/src/fgai/__init__.py new file mode 100644 index 0000000..abd4e91 --- /dev/null +++ b/src/fgai/__init__.py @@ -0,0 +1,4 @@ +"""Local FortiGate AI/ML inspection helpers.""" + +__all__ = ["__version__"] +__version__ = "0.1.0" diff --git a/src/fgai/__pycache__/__init__.cpython-312.pyc b/src/fgai/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000..328dd20 Binary files /dev/null and b/src/fgai/__pycache__/__init__.cpython-312.pyc differ diff --git a/src/fgai/__pycache__/cli.cpython-312.pyc b/src/fgai/__pycache__/cli.cpython-312.pyc new file mode 100644 index 0000000..74b1f41 Binary files /dev/null and b/src/fgai/__pycache__/cli.cpython-312.pyc differ diff --git a/src/fgai/__pycache__/llm.cpython-312.pyc b/src/fgai/__pycache__/llm.cpython-312.pyc new file mode 100644 index 0000000..749320b Binary files /dev/null and b/src/fgai/__pycache__/llm.cpython-312.pyc differ diff --git a/src/fgai/__pycache__/logs.cpython-312.pyc b/src/fgai/__pycache__/logs.cpython-312.pyc new file mode 100644 index 0000000..1c98026 Binary files /dev/null and b/src/fgai/__pycache__/logs.cpython-312.pyc differ diff --git a/src/fgai/__pycache__/mitigation.cpython-312.pyc b/src/fgai/__pycache__/mitigation.cpython-312.pyc new file mode 100644 index 0000000..83077f1 Binary files /dev/null and b/src/fgai/__pycache__/mitigation.cpython-312.pyc differ diff --git a/src/fgai/__pycache__/models.cpython-312.pyc b/src/fgai/__pycache__/models.cpython-312.pyc new file mode 100644 index 0000000..cd5f73d Binary files /dev/null and b/src/fgai/__pycache__/models.cpython-312.pyc differ diff --git a/src/fgai/__pycache__/policies.cpython-312.pyc b/src/fgai/__pycache__/policies.cpython-312.pyc new file mode 100644 index 0000000..93e6b60 Binary files /dev/null and b/src/fgai/__pycache__/policies.cpython-312.pyc differ diff --git a/src/fgai/cli.py b/src/fgai/cli.py new file mode 100644 index 0000000..35e70b8 --- /dev/null +++ b/src/fgai/cli.py @@ -0,0 +1,115 @@ +from __future__ import annotations + +import argparse +import json +import sys + +from .llm import ollama_summary +from .logs import read_events, summarize_events +from .mitigation import FortiGateClient, parse_allowlist, suggest_block_candidates +from .policies import audit_policies, read_policies + + +def _print_json(data: object) -> None: + print(json.dumps(data, indent=2, sort_keys=True)) + + +def analyze_logs(args: argparse.Namespace) -> int: + events = read_events(args.logs) + candidates = suggest_block_candidates( + events, + min_events=args.min_events, + min_score=args.min_score, + allowlist=parse_allowlist(args.allowlist), + ) + _print_json( + { + "summary": summarize_events(events), + "block_candidates": [ + {"src_ip": candidate.src_ip, "score": candidate.score, "reasons": candidate.reasons} + for candidate in candidates + ], + } + ) + if args.llm: + print("\nLLM summary:") + print(ollama_summary([], candidates, args.model)) + return 0 + + +def audit_policy_file(args: argparse.Namespace) -> int: + findings = audit_policies(read_policies(args.config)) + _print_json([finding.__dict__ for finding in findings]) + if args.llm: + print("\nLLM summary:") + print(ollama_summary(findings, [], args.model)) + return 0 + + +def suggest_blocks(args: argparse.Namespace) -> int: + events = read_events(args.logs) + candidates = suggest_block_candidates( + events, + min_events=args.min_events, + min_score=args.min_score, + allowlist=parse_allowlist(args.allowlist), + ) + if not candidates: + print("No block candidates met the current thresholds.") + return 0 + + for candidate in candidates: + print(f"{candidate.src_ip} score={candidate.score} reasons={', '.join(candidate.reasons)}") + if not args.execute: + print(" dry-run: not blocked") + continue + client = FortiGateClient.from_env() + reason = f"fgAI UTM mitigation score={candidate.score}" + response = client.quarantine_ip(candidate.src_ip, args.expiry_minutes, reason) + print(f" blocked: {response}") + return 0 + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description="Local FortiGate AI/ML inspection tool") + subparsers = parser.add_subparsers(required=True) + + logs = subparsers.add_parser("analyze-logs", help="Analyze local FortiGate logs") + logs.add_argument("--logs", required=True, help="Path to syslog JSONL or key/value log file") + logs.add_argument("--min-events", type=int, default=3) + logs.add_argument("--min-score", type=int, default=7) + logs.add_argument("--allowlist", default=None, help="Comma-separated IPs/CIDRs never to block") + logs.add_argument("--llm", action="store_true", help="Ask local Ollama to summarize results") + logs.add_argument("--model", default=None, help="Ollama model name") + logs.set_defaults(func=analyze_logs) + + policies = subparsers.add_parser("audit-policies", help="Audit FortiOS firewall policy config") + policies.add_argument("--config", required=True, help="Path to FortiOS config backup") + policies.add_argument("--llm", action="store_true", help="Ask local Ollama to summarize results") + policies.add_argument("--model", default=None, help="Ollama model name") + policies.set_defaults(func=audit_policy_file) + + blocks = subparsers.add_parser("suggest-blocks", help="Suggest or execute guarded source IP blocks") + blocks.add_argument("--logs", required=True, help="Path to syslog JSONL or key/value log file") + blocks.add_argument("--min-events", type=int, default=3) + blocks.add_argument("--min-score", type=int, default=7) + blocks.add_argument("--allowlist", default=None, help="Comma-separated IPs/CIDRs never to block") + blocks.add_argument("--execute", action="store_true", help="Actually call the FortiGate quarantine API") + blocks.add_argument("--expiry-minutes", type=int, default=60) + blocks.set_defaults(func=suggest_blocks) + + return parser + + +def main(argv: list[str] | None = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + try: + return args.func(args) + except Exception as exc: + print(f"fgai: {exc}", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/fgai/llm.py b/src/fgai/llm.py new file mode 100644 index 0000000..5530280 --- /dev/null +++ b/src/fgai/llm.py @@ -0,0 +1,34 @@ +from __future__ import annotations + +import json +import os +from urllib import request + +from .models import BlockCandidate, Finding + + +def ollama_summary(findings: list[Finding], candidates: list[BlockCandidate], model: str | None = None) -> str: + host = os.getenv("OLLAMA_HOST", "http://127.0.0.1:11434").rstrip("/") + selected_model = model or os.getenv("OLLAMA_MODEL", "llama3.3") + prompt = { + "findings": [finding.__dict__ for finding in findings], + "block_candidates": [ + {"src_ip": candidate.src_ip, "score": candidate.score, "reasons": candidate.reasons} + for candidate in candidates + ], + } + body = json.dumps( + { + "model": selected_model, + "stream": False, + "prompt": ( + "You are a local FortiGate security analyst. Summarize these policy findings " + "and UTM block candidates. Be concise, include risk, likely cause, and next action. " + f"Data: {json.dumps(prompt)}" + ), + } + ).encode("utf-8") + req = request.Request(f"{host}/api/generate", data=body, method="POST", headers={"Content-Type": "application/json"}) + with request.urlopen(req, timeout=60) as response: + data = json.loads(response.read().decode("utf-8")) + return str(data.get("response", "")).strip() diff --git a/src/fgai/logs.py b/src/fgai/logs.py new file mode 100644 index 0000000..c0be0ef --- /dev/null +++ b/src/fgai/logs.py @@ -0,0 +1,119 @@ +from __future__ import annotations + +import json +import shlex +from collections.abc import Iterable +from pathlib import Path + +from .models import LogEvent + +UTM_SUBTYPES = { + "ips", + "virus", + "av", + "anomaly", + "ddos", + "dos", + "webfilter", + "app-ctrl", + "application", + "waf", + "dns", +} + +THREAT_ACTIONS = { + "blocked", + "block", + "dropped", + "drop", + "reset", + "reset-client", + "reset-server", + "detected", + "deny", + "quarantine", +} + +SEVERITY_SCORE = { + "emergency": 6, + "alert": 6, + "critical": 5, + "high": 4, + "warning": 3, + "medium": 3, + "notice": 2, + "low": 1, + "information": 1, + "info": 1, +} + + +def parse_log_line(line: str) -> LogEvent: + stripped = line.strip() + if not stripped: + return LogEvent(raw=line, fields={}) + + if stripped.startswith("{"): + try: + data = json.loads(stripped) + except json.JSONDecodeError: + data = {} + else: + return LogEvent(raw=line, fields={str(k).lower(): str(v) for k, v in data.items()}) + + fields: dict[str, str] = {} + try: + parts = shlex.split(stripped) + except ValueError: + parts = stripped.split() + + for part in parts: + if "=" not in part: + continue + key, value = part.split("=", 1) + fields[key.strip().lower()] = value.strip().strip('"') + + return LogEvent(raw=line, fields=fields) + + +def read_events(path: str | Path) -> list[LogEvent]: + with Path(path).open("r", encoding="utf-8", errors="replace") as handle: + return [event for line in handle if (event := parse_log_line(line)).fields] + + +def is_utm_event(event: LogEvent) -> bool: + subtype = event.subtype + event_type = event.fields.get("type", "").lower() + return event_type == "utm" or subtype in UTM_SUBTYPES + + +def event_score(event: LogEvent) -> int: + score = SEVERITY_SCORE.get(event.severity, 0) + if is_utm_event(event): + score += 2 + if event.action in THREAT_ACTIONS: + score += 2 + if event.fields.get("attack") or event.fields.get("attackid") or event.fields.get("signature"): + score += 1 + return score + + +def summarize_events(events: Iterable[LogEvent]) -> dict[str, int]: + summary = { + "total": 0, + "utm": 0, + "threat_actions": 0, + "with_src_ip": 0, + "critical_or_high": 0, + } + for event in events: + summary["total"] += 1 + if is_utm_event(event): + summary["utm"] += 1 + if event.action in THREAT_ACTIONS: + summary["threat_actions"] += 1 + if event.src_ip: + summary["with_src_ip"] += 1 + if event.severity in {"critical", "high", "alert", "emergency"}: + summary["critical_or_high"] += 1 + return summary diff --git a/src/fgai/mitigation.py b/src/fgai/mitigation.py new file mode 100644 index 0000000..7713a6c --- /dev/null +++ b/src/fgai/mitigation.py @@ -0,0 +1,109 @@ +from __future__ import annotations + +import ipaddress +import os +from collections import defaultdict +from dataclasses import dataclass +from typing import Iterable +from urllib import error, request +import json +import ssl + +from .logs import event_score, is_utm_event +from .models import BlockCandidate, LogEvent + + +def parse_allowlist(value: str | None = None) -> list[ipaddress._BaseNetwork]: + raw = value if value is not None else os.getenv("FGAI_ALLOWLIST", "") + networks: list[ipaddress._BaseNetwork] = [] + for item in raw.split(","): + item = item.strip() + if not item: + continue + networks.append(ipaddress.ip_network(item, strict=False)) + return networks + + +def is_blockable_public_ip(ip: str, allowlist: Iterable[ipaddress._BaseNetwork] = ()) -> bool: + try: + address = ipaddress.ip_address(ip) + except ValueError: + return False + if not address.is_global: + return False + return not any(address in network for network in allowlist) + + +def suggest_block_candidates( + events: Iterable[LogEvent], + *, + min_events: int = 3, + min_score: int = 7, + allowlist: Iterable[ipaddress._BaseNetwork] = (), +) -> list[BlockCandidate]: + grouped: dict[str, list[LogEvent]] = defaultdict(list) + for event in events: + if not event.src_ip or not is_utm_event(event): + continue + if not is_blockable_public_ip(event.src_ip, allowlist): + continue + if event_score(event) > 0: + grouped[event.src_ip].append(event) + + candidates: list[BlockCandidate] = [] + for src_ip, ip_events in grouped.items(): + score = sum(event_score(event) for event in ip_events) + if len(ip_events) < min_events and score < min_score: + continue + reasons = sorted( + { + f"{event.subtype or 'utm'}:{event.action or 'observed'}:{event.severity or 'unknown'}" + for event in ip_events + } + ) + candidates.append(BlockCandidate(src_ip, score, reasons, ip_events[-5:])) + + return sorted(candidates, key=lambda candidate: candidate.score, reverse=True) + + +@dataclass(frozen=True) +class FortiGateClient: + host: str + api_token: str + verify_tls: bool = True + + @classmethod + def from_env(cls) -> "FortiGateClient": + host = os.environ["FORTIGATE_HOST"] + token = os.environ["FORTIGATE_API_TOKEN"] + verify_tls = os.getenv("FORTIGATE_VERIFY_TLS", "true").lower() not in {"0", "false", "no"} + return cls(host=host, api_token=token, verify_tls=verify_tls) + + def quarantine_ip(self, src_ip: str, expiry_minutes: int, reason: str) -> str: + payload = { + "ip_address": src_ip, + "expiry": expiry_minutes, + "source": "fgai-local-agent", + "reason": reason, + } + url = f"https://{self.host}/api/v2/monitor/user/banned/create" + body = json.dumps(payload).encode("utf-8") + req = request.Request( + url, + data=body, + method="POST", + headers={ + "Authorization": f"Bearer {self.api_token}", + "Content-Type": "application/json", + "Accept": "application/json", + }, + ) + context = None if self.verify_tls else ssl._create_unverified_context() + try: + with request.urlopen(req, timeout=15, context=context) as response: + return response.read().decode("utf-8", errors="replace") + except error.HTTPError as exc: + detail = exc.read().decode("utf-8", errors="replace") + raise RuntimeError(f"FortiGate API returned HTTP {exc.code}: {detail}") from exc + except error.URLError as exc: + raise RuntimeError(f"Could not reach FortiGate API: {exc.reason}") from exc diff --git a/src/fgai/models.py b/src/fgai/models.py new file mode 100644 index 0000000..90fc248 --- /dev/null +++ b/src/fgai/models.py @@ -0,0 +1,54 @@ +from __future__ import annotations + +from dataclasses import dataclass, field + + +@dataclass(frozen=True) +class LogEvent: + raw: str + fields: dict[str, str] + + @property + def src_ip(self) -> str | None: + return self.fields.get("srcip") or self.fields.get("src_ip") or self.fields.get("source_ip") + + @property + def subtype(self) -> str: + return self.fields.get("subtype", "").lower() + + @property + def action(self) -> str: + return self.fields.get("action", "").lower() + + @property + def severity(self) -> str: + return self.fields.get("severity", self.fields.get("level", "")).lower() + + +@dataclass(frozen=True) +class PolicyRule: + policy_id: str + settings: dict[str, list[str]] = field(default_factory=dict) + + def values(self, key: str) -> list[str]: + return self.settings.get(key, []) + + def first(self, key: str, default: str = "") -> str: + values = self.values(key) + return values[0] if values else default + + +@dataclass(frozen=True) +class Finding: + severity: str + title: str + detail: str + reference: str | None = None + + +@dataclass(frozen=True) +class BlockCandidate: + src_ip: str + score: int + reasons: list[str] + sample_events: list[LogEvent] diff --git a/src/fgai/policies.py b/src/fgai/policies.py new file mode 100644 index 0000000..3a48d5d --- /dev/null +++ b/src/fgai/policies.py @@ -0,0 +1,106 @@ +from __future__ import annotations + +import shlex +from pathlib import Path + +from .models import Finding, PolicyRule + +ANY_VALUES = {"all", "any", "0.0.0.0/0"} +UTM_KEYS = {"ips-sensor", "av-profile", "webfilter-profile", "application-list", "waf-profile", "dnsfilter-profile"} + + +def _parse_set_values(line: str) -> tuple[str, list[str]] | None: + try: + parts = shlex.split(line) + except ValueError: + parts = line.split() + if len(parts) < 3 or parts[0] != "set": + return None + return parts[1], parts[2:] + + +def parse_policy_config(text: str) -> list[PolicyRule]: + in_policy = False + current_id: str | None = None + current_settings: dict[str, list[str]] = {} + policies: list[PolicyRule] = [] + + for raw_line in text.splitlines(): + line = raw_line.strip() + if line == "config firewall policy": + in_policy = True + continue + if in_policy and line == "end": + if current_id is not None: + policies.append(PolicyRule(current_id, current_settings)) + break + if not in_policy: + continue + if line.startswith("edit "): + if current_id is not None: + policies.append(PolicyRule(current_id, current_settings)) + current_id = line.split(maxsplit=1)[1].strip('"') + current_settings = {} + continue + if line == "next": + if current_id is not None: + policies.append(PolicyRule(current_id, current_settings)) + current_id = None + current_settings = {} + continue + parsed = _parse_set_values(line) + if current_id is not None and parsed: + key, values = parsed + current_settings[key] = values + + return policies + + +def read_policies(path: str | Path) -> list[PolicyRule]: + return parse_policy_config(Path(path).read_text(encoding="utf-8", errors="replace")) + + +def audit_policies(policies: list[PolicyRule]) -> list[Finding]: + findings: list[Finding] = [] + for policy in policies: + if policy.first("status", "enable") == "disable": + continue + + action = policy.first("action") + srcaddr = {value.lower() for value in policy.values("srcaddr")} + dstaddr = {value.lower() for value in policy.values("dstaddr")} + service = {value.lower() for value in policy.values("service")} + policy_ref = f"policy {policy.policy_id}" + + if action == "accept" and srcaddr & ANY_VALUES and dstaddr & ANY_VALUES and service & ANY_VALUES: + findings.append( + Finding( + severity="critical", + title="Broad allow policy", + detail="Accepts traffic from any source to any destination for any service.", + reference=policy_ref, + ) + ) + + if action == "accept" and policy.first("logtraffic", "disable") in {"disable", "utm"}: + findings.append( + Finding( + severity="medium", + title="Insufficient traffic logging", + detail="Accepted traffic is not fully logged, reducing investigation value.", + reference=policy_ref, + ) + ) + + has_utm = policy.first("utm-status") == "enable" or any(policy.values(key) for key in UTM_KEYS) + if action == "accept" and not has_utm: + findings.append( + Finding( + severity="high", + title="Accepted traffic lacks UTM inspection", + detail="No IPS, AV, web filter, application, DNS, or WAF profile is attached.", + reference=policy_ref, + ) + ) + + return findings diff --git a/tests/__pycache__/test_logs.cpython-312.pyc b/tests/__pycache__/test_logs.cpython-312.pyc new file mode 100644 index 0000000..32b4068 Binary files /dev/null and b/tests/__pycache__/test_logs.cpython-312.pyc differ diff --git a/tests/__pycache__/test_mitigation.cpython-312.pyc b/tests/__pycache__/test_mitigation.cpython-312.pyc new file mode 100644 index 0000000..c7e3508 Binary files /dev/null and b/tests/__pycache__/test_mitigation.cpython-312.pyc differ diff --git a/tests/__pycache__/test_policies.cpython-312.pyc b/tests/__pycache__/test_policies.cpython-312.pyc new file mode 100644 index 0000000..7f647c8 Binary files /dev/null and b/tests/__pycache__/test_policies.cpython-312.pyc differ diff --git a/tests/test_logs.py b/tests/test_logs.py new file mode 100644 index 0000000..15066f6 --- /dev/null +++ b/tests/test_logs.py @@ -0,0 +1,23 @@ +import unittest + +from fgai.logs import event_score, is_utm_event, parse_log_line, summarize_events + + +class LogTests(unittest.TestCase): + def test_parse_key_value_log_line(self): + event = parse_log_line('date=2026-06-18 type="utm" subtype="ips" srcip=8.8.8.8 action="blocked" severity="critical"') + + self.assertEqual(event.src_ip, "8.8.8.8") + self.assertEqual(event.subtype, "ips") + self.assertTrue(is_utm_event(event)) + self.assertGreaterEqual(event_score(event), 9) + + def test_parse_json_log_line(self): + event = parse_log_line('{"type":"utm","subtype":"virus","srcip":"1.1.1.1","action":"detected","severity":"high"}') + + self.assertEqual(event.src_ip, "1.1.1.1") + self.assertEqual(summarize_events([event])["utm"], 1) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_mitigation.py b/tests/test_mitigation.py new file mode 100644 index 0000000..82de0f3 --- /dev/null +++ b/tests/test_mitigation.py @@ -0,0 +1,27 @@ +import unittest + +from fgai.logs import parse_log_line +from fgai.mitigation import is_blockable_public_ip, suggest_block_candidates + + +class MitigationTests(unittest.TestCase): + def test_private_ips_are_not_blockable(self): + self.assertFalse(is_blockable_public_ip("192.168.1.10")) + self.assertFalse(is_blockable_public_ip("10.0.0.5")) + self.assertTrue(is_blockable_public_ip("8.8.8.8")) + + def test_suggests_repeated_public_utm_offender(self): + events = [ + parse_log_line('type=utm subtype=ips srcip=8.8.8.8 action=blocked severity=high attack="scan"'), + parse_log_line('type=utm subtype=ips srcip=8.8.8.8 action=blocked severity=high attack="scan"'), + parse_log_line('type=utm subtype=ips srcip=8.8.8.8 action=blocked severity=high attack="scan"'), + parse_log_line('type=utm subtype=ips srcip=192.168.1.5 action=blocked severity=critical attack="scan"'), + ] + + candidates = suggest_block_candidates(events) + + self.assertEqual([candidate.src_ip for candidate in candidates], ["8.8.8.8"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_policies.py b/tests/test_policies.py new file mode 100644 index 0000000..3bacfde --- /dev/null +++ b/tests/test_policies.py @@ -0,0 +1,30 @@ +import unittest + +from fgai.policies import audit_policies, parse_policy_config + + +class PolicyTests(unittest.TestCase): + def test_policy_audit_finds_broad_unprotected_allow(self): + policies = parse_policy_config( + """ +config firewall policy + edit 1 + set srcaddr "all" + set dstaddr "all" + set service "ALL" + set action accept + set logtraffic disable + next +end +""" + ) + + findings = audit_policies(policies) + + titles = {finding.title for finding in findings} + self.assertIn("Broad allow policy", titles) + self.assertIn("Accepted traffic lacks UTM inspection", titles) + + +if __name__ == "__main__": + unittest.main()