diff --git a/src/fgai/baseline.py b/src/fgai/baseline.py index 282237e..d9c21af 100644 --- a/src/fgai/baseline.py +++ b/src/fgai/baseline.py @@ -39,6 +39,11 @@ class BaselineStore: source_ip text not null, kind text not null, value text not null, seen_count integer not null, primary key (source_ip, kind, value) ); + create table if not exists profile_buckets ( + stream_id text not null, entity text not null, field text not null, bucket_start integer not null, + events integer not null, numeric_sum real not null, numeric_sum_squares real not null, + primary key (stream_id, entity, field, bucket_start) + ); """ ) @@ -81,6 +86,33 @@ class BaselineStore: ) return inserted + def ingest_profile_fields(self, events: list[LogEvent], profiles: dict[str, object], *, observed_at: int | None = None) -> int: + observed_at = observed_at or int(time.time()) + bucket = observed_at - (observed_at % self.bucket_seconds) + pending: dict[tuple[str, str, str], list[float]] = defaultdict(lambda: [0, 0.0, 0.0]) + for event in events: + stream_id = event.fields.get("fgai_stream_id", "") + profile = profiles.get(stream_id) + if not profile: + continue + entity_field = str(getattr(profile, "entity_field", "")).lower() + entity = event.fields.get(entity_field) + if not entity: + continue + fields = [*getattr(profile, "categorical_fields", ()), *getattr(profile, "numeric_fields", ())] + numeric = {str(field).lower() for field in getattr(profile, "numeric_fields", ())} + for field in fields: + key = (stream_id, entity, str(field).lower()) + value = _number(event.fields.get(key[2])) if key[2] in numeric else 0 + pending[key][0] += 1 + pending[key][1] += value + pending[key][2] += value * value + with self._connect() as connection: + for (stream_id, entity, field), values in pending.items(): + connection.execute("""insert into profile_buckets values (?, ?, ?, ?, ?, ?, ?) + on conflict(stream_id, entity, field, bucket_start) do update set events=events+excluded.events,numeric_sum=numeric_sum+excluded.numeric_sum,numeric_sum_squares=numeric_sum_squares+excluded.numeric_sum_squares""", (stream_id, entity, field, bucket, *values)) + return len(pending) + def profiles(self, source_ips: set[str]) -> dict[str, dict[str, object]]: profiles: dict[str, dict[str, object]] = {} with self._connect() as connection: diff --git a/src/fgai/graylog_source.py b/src/fgai/graylog_source.py index ffc9e41..e84584b 100644 --- a/src/fgai/graylog_source.py +++ b/src/fgai/graylog_source.py @@ -90,6 +90,7 @@ class GraylogStreamSource: def _event(self, record: dict[str, object]) -> LogEvent: fields = {str(key).lower(): str(value) for key, value in record.items() if value is not None} fields["fgai_stream"] = self.stream_label + fields["fgai_stream_id"] = self.stream for canonical, candidates in DEFAULT_FIELD_MAP.items(): mapped = self.mapping.get(canonical) candidates = (str(mapped),) if mapped else candidates diff --git a/src/fgai/monitor.py b/src/fgai/monitor.py index d6f752a..cf07f1d 100644 --- a/src/fgai/monitor.py +++ b/src/fgai/monitor.py @@ -64,6 +64,7 @@ def build_status( 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 + profile_baseline_fields = baseline.ingest_profile_fields(events, stream_profiles) if baseline else 0 intel_ips = sorted( { ip @@ -97,7 +98,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}, + "baseline": {"enabled": bool(baseline), "sources_ready": len(profiles), "new_events_recorded": baseline_events, "profile_fields_recorded": profile_baseline_fields}, "capabilities": {"threat_intel": threat_intel_status, "graylog_mcp": mcp_status}, "configuration": runtime_config, "stream_profiles": [{"stream_id": item.stream_id, "entity_field": item.entity_field, "timestamp_field": item.timestamp_field, "categorical_fields": list(item.categorical_fields), "numeric_fields": list(item.numeric_fields)} for item in stream_profiles.values()],