diff --git a/README.md b/README.md
index 36fd741..ec1a6d0 100644
--- a/README.md
+++ b/README.md
@@ -54,6 +54,11 @@ files are gzip-compressed and 14 archives are retained. Override this when neede
FGAI_LOG_ROTATE_BYTES=$((100 * 1024 * 1024)) FGAI_LOG_ROTATE_COUNT=30 ./start.sh restart
```
+The continuous monitor also stores a local SQLite behavior baseline at
+`state/fgai-baseline.sqlite3`. A source becomes baseline-ready after 12 completed
+five-minute windows. Historical rate and hitcount-rate deviations then contribute
+to its anomaly score. Set `FGAI_BASELINE_DB` to use another location.
+
Analyze local logs:
```bash
diff --git a/src/fgai/anomaly.py b/src/fgai/anomaly.py
index d6db5db..ce4b072 100644
--- a/src/fgai/anomaly.py
+++ b/src/fgai/anomaly.py
@@ -74,7 +74,10 @@ def _rate_per_minute(events: list[LogEvent]) -> tuple[float | None, float]:
return len(timestamps) * 60 / max(1.0, duration_seconds), duration_seconds
-def detect_source_anomalies(events: list[LogEvent], *, limit: int = 20) -> list[AnomalyFinding]:
+def detect_source_anomalies(
+ events: list[LogEvent], *, limit: int = 20, baselines: dict[str, dict[str, float | int]] | None = None
+) -> list[AnomalyFinding]:
+ baselines = baselines or {}
by_src: dict[str, list[LogEvent]] = defaultdict(list)
for event in events:
if event.src_ip:
@@ -137,6 +140,12 @@ def detect_source_anomalies(events: list[LogEvent], *, limit: int = 20) -> list[
points = min(25, 10 + int(rate_z * 5))
score += points
reasons.append(f"unusually high log rate ({event_rate:.1f} events/min, z={rate_z:.1f})")
+ baseline = baselines.get(src_ip)
+ if baseline:
+ historical_z = (event_rate - float(baseline["event_rate_mean"])) / float(baseline["event_rate_stddev"])
+ if historical_z >= 3:
+ score += min(25, 10 + int(historical_z * 3))
+ reasons.append(f"log rate exceeds its {baseline['samples']}-window baseline (z={historical_z:.1f})")
dst_z = (distinct_dst - avg_dst) / std_dst
if distinct_dst >= 10 and dst_z >= 2:
@@ -155,6 +164,13 @@ def detect_source_anomalies(events: list[LogEvent], *, limit: int = 20) -> list[
points = min(15, 5 + int(hitcount_z * 3))
score += points
reasons.append(f"unusually high policy hitcount ({total_hitcount}, max event value {max_hitcount})")
+ if event_rate is not None and observed_duration > 0 and src_ip in baselines:
+ hit_rate = total_hitcount * 60 / max(1.0, observed_duration)
+ baseline = baselines[src_ip]
+ historical_z = (hit_rate - float(baseline["hitcount_rate_mean"])) / float(baseline["hitcount_rate_stddev"])
+ if total_hitcount >= 10 and historical_z >= 3:
+ score += min(15, 5 + int(historical_z * 2))
+ reasons.append(f"hitcount rate exceeds its historical baseline (z={historical_z:.1f})")
if event_count >= 5:
deny_rate = deny_count / event_count
diff --git a/src/fgai/baseline.py b/src/fgai/baseline.py
new file mode 100644
index 0000000..8278768
--- /dev/null
+++ b/src/fgai/baseline.py
@@ -0,0 +1,91 @@
+from __future__ import annotations
+
+import hashlib
+import sqlite3
+import time
+from collections import defaultdict
+from pathlib import Path
+from statistics import mean, pstdev
+
+from .logs import THREAT_ACTIONS, is_utm_event
+from .models import LogEvent
+
+
+def _number(value: str | None) -> int:
+ try:
+ return int(float(value or 0))
+ except ValueError:
+ return 0
+
+
+class BaselineStore:
+ """Persistent five-minute behavior baseline, implemented with stdlib SQLite."""
+
+ def __init__(self, path: str, *, bucket_seconds: int = 300) -> None:
+ self.path = Path(path)
+ self.path.parent.mkdir(parents=True, exist_ok=True)
+ self.bucket_seconds = bucket_seconds
+ with self._connect() as connection:
+ connection.executescript(
+ """
+ create table if not exists seen_events (fingerprint text primary key);
+ create table if not exists source_buckets (
+ source_ip text not null, bucket_start integer not null,
+ events integer not null, bytes integer not null, hitcount integer not null,
+ denies integer not null, utm integer not null,
+ primary key (source_ip, bucket_start)
+ );
+ """
+ )
+
+ def _connect(self) -> sqlite3.Connection:
+ return sqlite3.connect(self.path)
+
+ def ingest(self, events: list[LogEvent], *, observed_at: int | None = None) -> int:
+ observed_at = observed_at or int(time.time())
+ pending: dict[tuple[str, int], list[int]] = defaultdict(lambda: [0, 0, 0, 0, 0])
+ inserted = 0
+ with self._connect() as connection:
+ for event in events:
+ if not event.src_ip:
+ continue
+ fingerprint = hashlib.sha256(event.raw.encode("utf-8", errors="replace")).hexdigest()
+ if connection.execute("insert or ignore into seen_events values (?)", (fingerprint,)).rowcount != 1:
+ continue
+ bucket = observed_at - (observed_at % self.bucket_seconds)
+ values = pending[(event.src_ip, bucket)]
+ values[0] += 1
+ values[1] += _number(event.fields.get("sentbyte")) + _number(event.fields.get("rcvdbyte"))
+ values[2] += _number(event.fields.get("hitcount"))
+ values[3] += int(event.action in THREAT_ACTIONS)
+ values[4] += int(is_utm_event(event))
+ inserted += 1
+ for (source_ip, bucket), values in pending.items():
+ connection.execute(
+ """insert into source_buckets values (?, ?, ?, ?, ?, ?, ?)
+ on conflict(source_ip, bucket_start) do update set
+ events=events+excluded.events, bytes=bytes+excluded.bytes,
+ hitcount=hitcount+excluded.hitcount, denies=denies+excluded.denies, utm=utm+excluded.utm""",
+ (source_ip, bucket, *values),
+ )
+ return inserted
+
+ def profiles(self, source_ips: set[str]) -> dict[str, dict[str, float | int]]:
+ profiles: dict[str, dict[str, float | int]] = {}
+ with self._connect() as connection:
+ for source_ip in source_ips:
+ rows = connection.execute(
+ "select events, bytes, hitcount from source_buckets where source_ip=? order by bucket_start", (source_ip,)
+ ).fetchall()
+ # Do not compare the current (often incomplete) bucket to itself.
+ rows = rows[:-1]
+ if len(rows) < 12:
+ continue
+ rates = [row[0] * 60 / self.bucket_seconds for row in rows]
+ hit_rates = [row[2] * 60 / self.bucket_seconds for row in rows]
+ profiles[source_ip] = {
+ "samples": len(rows),
+ "event_rate_mean": mean(rates), "event_rate_stddev": pstdev(rates) or 1.0,
+ "hitcount_rate_mean": mean(hit_rates), "hitcount_rate_stddev": pstdev(hit_rates) or 1.0,
+ }
+ return profiles
diff --git a/src/fgai/cli.py b/src/fgai/cli.py
index 441fade..14f1fbb 100644
--- a/src/fgai/cli.py
+++ b/src/fgai/cli.py
@@ -203,6 +203,7 @@ def run_monitor(args: argparse.Namespace) -> int:
llm_interval=args.llm_interval,
llm_model=args.model,
llm_timeout=args.llm_timeout,
+ baseline_path=args.baseline_db,
)
return 0
@@ -285,6 +286,7 @@ def build_parser() -> argparse.ArgumentParser:
monitor.add_argument("--policies", default=None, help="Optional FortiGate policy JSON/config file to audit continuously")
monitor.add_argument("--interval", type=int, default=10, help="Seconds between analysis runs")
monitor.add_argument("--anomaly-limit", type=int, default=20, help="Maximum anomaly findings to include")
+ monitor.add_argument("--baseline-db", default="state/fgai-baseline.sqlite3", help="SQLite database for historical behavior baselines")
monitor.add_argument("--llm", action="store_true", help="Generate cached Ollama analyst note for dashboard")
monitor.add_argument("--llm-interval", type=int, default=300, help="Seconds between Ollama dashboard assessments")
monitor.add_argument("--model", default=None, help="Ollama model name")
diff --git a/src/fgai/dashboard.py b/src/fgai/dashboard.py
index 6a42659..3edf67b 100644
--- a/src/fgai/dashboard.py
+++ b/src/fgai/dashboard.py
@@ -83,7 +83,8 @@ async function refresh() {
`Log file: ${esc(data.log_path || '')}`,
`Policy file: ${esc(data.policy_path || 'none')}`,
`Critical anomalies: ${esc((a.critical || 0))}`,
- `High anomalies: ${esc((a.high || 0))}`
+ `High anomalies: ${esc((a.high || 0))}`,
+ `Baseline sources ready: ${esc((data.baseline || {}).sources_ready || 0)}`
].join('
');
const llm = data.llm_assessment || {};
const llmText = llm.text ? esc(llm.text).replace(/\\n/g, '
') : esc(llm.error || 'LLM assessment disabled or waiting for first run.');
diff --git a/src/fgai/monitor.py b/src/fgai/monitor.py
index 2474234..d25d73e 100644
--- a/src/fgai/monitor.py
+++ b/src/fgai/monitor.py
@@ -5,6 +5,7 @@ import time
from pathlib import Path
from .anomaly import anomaly_summary, detect_source_anomalies
+from .baseline import BaselineStore
from .llm import ollama_dashboard_assessment
from .logs import local_in_failures, read_events, summarize_events, top_field_values
from .mitigation import parse_allowlist, suggest_block_candidates
@@ -20,9 +21,13 @@ def build_status(
min_block_events: int = 3,
min_block_score: int = 7,
anomaly_limit: int = 20,
+ baseline_path: str | None = None,
) -> dict[str, object]:
events = read_events(log_path) if Path(log_path).exists() else []
- anomalies = detect_source_anomalies(events, limit=anomaly_limit)
+ baseline = BaselineStore(baseline_path) if baseline_path else None
+ profiles = baseline.profiles({event.src_ip for event in events if event.src_ip}) if baseline else {}
+ anomalies = detect_source_anomalies(events, limit=anomaly_limit, baselines=profiles)
+ baseline_events = baseline.ingest(events) if baseline else 0
intel_ips = sorted(
{
ip
@@ -54,6 +59,7 @@ def build_status(
"policy_path": policy_path,
"summary": summarize_events(events),
"anomaly_summary": anomaly_summary(anomalies),
+ "baseline": {"enabled": bool(baseline), "sources_ready": len(profiles), "new_events_recorded": baseline_events},
"diagnostics": {
"top_source_ips": top_field_values(events, "srcip", limit=10),
"top_destination_ports": top_field_values(events, "dstport", limit=10),
@@ -134,13 +140,14 @@ def monitor_loop(
llm_interval: int = 300,
llm_model: str | None = None,
llm_timeout: int | None = None,
+ baseline_path: str | None = None,
) -> None:
print(f"Monitoring {log_path}")
print(f"Writing status to {output}")
last_llm_at = 0
last_llm_text: str | None = None
while True:
- status = build_status(log_path, policy_path=policy_path, anomaly_limit=anomaly_limit)
+ status = build_status(log_path, policy_path=policy_path, anomaly_limit=anomaly_limit, baseline_path=baseline_path)
if llm:
now = int(time.time())
if now - last_llm_at >= llm_interval:
diff --git a/start.sh b/start.sh
index dcaef0c..ab641f3 100755
--- a/start.sh
+++ b/start.sh
@@ -11,6 +11,7 @@ LOG_ROTATE_COUNT="${FGAI_LOG_ROTATE_COUNT:-14}"
LISTENER_LOG="${FGAI_LISTENER_LOG:-$ROOT_DIR/logs/fgai-listener.log}"
POLICY_FILE="${FGAI_POLICY_FILE:-$ROOT_DIR/exports/policies.json}"
STATE_FILE="${FGAI_STATE_FILE:-$ROOT_DIR/state/fgai-status.json}"
+BASELINE_DB="${FGAI_BASELINE_DB:-$ROOT_DIR/state/fgai-baseline.sqlite3}"
MONITOR_INTERVAL="${FGAI_MONITOR_INTERVAL:-10}"
LLM_ENABLED="${FGAI_LLM:-0}"
LLM_INTERVAL="${FGAI_LLM_INTERVAL:-300}"
@@ -118,7 +119,7 @@ start_monitor() {
return 0
fi
- monitor_args=(monitor --logs "$LOG_FILE" --output "$STATE_FILE" --interval "$MONITOR_INTERVAL")
+ monitor_args=(monitor --logs "$LOG_FILE" --output "$STATE_FILE" --interval "$MONITOR_INTERVAL" --baseline-db "$BASELINE_DB")
if [ -f "$POLICY_FILE" ]; then
monitor_args+=(--policies "$POLICY_FILE")
fi
diff --git a/tests/test_baseline.py b/tests/test_baseline.py
new file mode 100644
index 0000000..08f42b1
--- /dev/null
+++ b/tests/test_baseline.py
@@ -0,0 +1,16 @@
+import tempfile
+import unittest
+from pathlib import Path
+
+from fgai.baseline import BaselineStore
+from fgai.logs import parse_log_line
+
+
+class BaselineTests(unittest.TestCase):
+ def test_creates_profile_after_twelve_completed_windows(self):
+ with tempfile.TemporaryDirectory() as directory:
+ store = BaselineStore(str(Path(directory) / "baseline.sqlite3"))
+ for index in range(13):
+ store.ingest([parse_log_line(f"srcip=10.0.0.1 dstport={1000 + index} hitcount=2 sentbyte=5")], observed_at=1_700_000_000 + index * 300)
+ profiles = store.profiles({"10.0.0.1"})
+ self.assertEqual(profiles["10.0.0.1"]["samples"], 12)