Public/fgAI
SHA256
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fixed the bad classification issue.

Browse Source
This commit is contained in:
larssand
2026-06-30 16:49:11 +02:00
parent b24b5d48a6
commit 5a9588fe31
2 changed files with 27 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
src/fgai/profile_suggestions.py
View File

@@ -164,6 +164,7 @@ IGNORED_DISCOVERY_FIELDS = {
"fgai_stream_id",
"fgai_stream_name",
}
EMPTY_DISCOVERY_VALUES = {"", "-", "--", "unknown", "n/a", "none", "null", "nil", "undefined", "[]", "{}"}
def _stream_id(event: LogEvent) -> str:
@@ -182,6 +183,11 @@ def _is_number(value: str) -> bool:
return False
def _has_discovery_value(value: object) -> bool:
text = str(value).strip()
return text.lower() not in EMPTY_DISCOVERY_VALUES
def _pick_present(priority: tuple[str, ...], coverage: Counter[str], total: int, *, min_ratio: float = 0.05, limit: int = 6) -> list[str]:
selected = []
for field in priority:
@@ -436,7 +442,7 @@ def suggest_stream_profiles(events: list[LogEvent], *, existing_profiles: dict[s
detector_counts: Counter[str] = Counter()
for event in stream_events:
for field, value in event.fields.items():
if not value:
if not _has_discovery_value(value):
continue
coverage[field] += 1
if len(unique_values[field]) < 200:

20
tests/test_profile_suggestions.py
View File

@@ -22,6 +22,26 @@ class ProfileSuggestionTests(unittest.TestCase):
self.assertIn("hitcount", profile["numeric_fields"])
self.assertIn("deny_action", profile["detectors"])
def test_placeholder_winlog_fields_do_not_make_firewall_windows(self):
events = [
parse_log_line(
f"fgai_stream_id=firewall fgai_stream='Firewall Logs' srcip=10.0.0.{index} dstip=8.8.8.8 "
f"dstport=443 action=deny policyid=12 eventid=- winlog.channel=- winlog.event_id=- "
f"winlog.event_data.targetusername=- hostname=- username=- timestamp=2026-06-29T10:00:{index:02d}Z"
)
for index in range(1, 25)
]
suggestion = suggest_stream_profiles(events)[0]
profile = suggestion["profile"]
self.assertEqual(suggestion["detected_log_type"], "firewall")
self.assertNotIn("username", profile["entity_fields"])
self.assertNotIn("hostname", profile["entity_fields"])
self.assertIn("srcip", profile["entity_fields"])
self.assertIn("policyid", profile["categorical_fields"])
self.assertIn("deny_action", profile["detectors"])
def test_suggests_windows_event_fields(self):
events = [
parse_log_line(