Auto-load detection library from S1 API, improve coverage map accuracy

- Fetch detection library rules from platform-rules API at startup (falls
  back to extracted.json); adds Sync Detection Library button for refresh
- Parser column simplified to ✓ Parsed / ✗ Not Parsed
- Detection counts now use library rules only (exclude custom STAR rules)
- Add close-match suggestions for dataSource.name mismatches (e.g. CloudTrail
  → AWS CloudTrail, Microsoft 365 Collaboration → Microsoft O365)
- Exclude SentinelOne Ranger AD from coverage map (native S1 source)
- Add success feedback banners to Load SDL Parsers and Sync Library buttons
- Remove rule_counts.json manual override; extracted.json is source of truth
- Remove Load Detections button; rules auto-import on backend startup
- Add get_account_id() and get_platform_rules() to s1_client

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Mick
2026-05-20 15:14:10 -04:00
parent 6e137438b1
commit 6cd9da82da
8 changed files with 580 additions and 90 deletions
+114 -9
View File
@@ -24,16 +24,72 @@ def _iso_to_epoch_ms(iso_str: str) -> int:
return int(dt.timestamp() * 1000)
async def get_star_rules(limit: int = 200) -> list:
"""Fetch active STAR rules from the Management Console API."""
async def get_star_rules(page_size: int = 100) -> list:
"""Fetch custom STAR rules from /cloud-detection/rules, paginating via cursor."""
all_rules = []
cursor = None
async with httpx.AsyncClient(timeout=30) as client:
resp = await client.get(
f"{BASE_URL}/web/api/v2.1/cloud-detection/rules",
headers=HEADERS,
params={"limit": limit},
)
resp.raise_for_status()
return resp.json().get("data", [])
while True:
params = {"limit": page_size}
if cursor:
params["cursor"] = cursor
resp = await client.get(
f"{BASE_URL}/web/api/v2.1/cloud-detection/rules",
headers=HEADERS,
params=params,
)
resp.raise_for_status()
body = resp.json()
all_rules.extend(body.get("data", []))
cursor = body.get("pagination", {}).get("nextCursor")
if not cursor:
break
return all_rules
async def get_library_rules(page_size: int = 100) -> list:
"""
Fetch Detection Library (OOTB/Platform) rules from /web/api/v2.1/detection-library/rules.
Requires an account-level or higher API token — site-scoped tokens will receive a 400.
Returns an empty list gracefully if the token lacks sufficient scope.
"""
all_rules = []
cursor = None
async with httpx.AsyncClient(timeout=60) as client:
while True:
params: dict = {"limit": page_size}
if cursor:
params["cursor"] = cursor
resp = await client.get(
f"{BASE_URL}/web/api/v2.1/detection-library/rules",
headers=HEADERS,
params=params,
)
# 400 typically means site-scoped token — return empty rather than crash
if resp.status_code == 400:
return []
resp.raise_for_status()
body = resp.json()
batch = body.get("data", [])
all_rules.extend(batch)
cursor = body.get("pagination", {}).get("nextCursor")
if not cursor:
break
results = []
for rule in all_rules:
results.append({
"id": str(rule.get("id", "")),
"name": rule.get("name", "unnamed"),
"s1ql": rule.get("s1ql") or rule.get("query", ""),
"queryType": rule.get("queryType", "events"),
"severity": rule.get("severity", ""),
"description": rule.get("description", ""),
"gdlRuleId": rule.get("id", ""),
"creator": "SentinelOne",
"expirationMode": rule.get("expirationMode", "Permanent"),
})
return results
async def run_powerquery(query: str, from_date: str, to_date: str) -> dict:
@@ -124,6 +180,55 @@ async def get_sdl_parser(filename: str) -> dict:
return resp.json()
async def get_account_id() -> str | None:
"""Return the first account ID visible to the current token."""
async with httpx.AsyncClient(timeout=15) as client:
resp = await client.get(
f"{BASE_URL}/web/api/v2.1/accounts",
headers=HEADERS,
params={"limit": 1},
)
resp.raise_for_status()
accounts = resp.json().get("data", [])
return str(accounts[0]["id"]) if accounts else None
async def get_platform_rules(page_size: int = 1000) -> list:
"""
Fetch all Detection Library platform rules from /detection-library/platform-rules.
Requires scopeLevel + scopeId — uses account scope with the first visible account.
Returns list of rules, each with a 'sources' list (authoritative data source names).
"""
account_id = await get_account_id()
if not account_id:
return []
all_rules: list = []
cursor: str = ""
async with httpx.AsyncClient(timeout=60) as client:
while True:
params: dict = {
"scopeLevel": "account",
"scopeId": account_id,
"limit": page_size,
"cursor": cursor,
}
resp = await client.get(
f"{BASE_URL}/web/api/v2.1/detection-library/platform-rules",
headers=HEADERS,
params=params,
)
if resp.status_code == 400:
return []
resp.raise_for_status()
body = resp.json()
all_rules.extend(body.get("data", []))
cursor = body.get("pagination", {}).get("nextCursor") or ""
if not cursor:
break
return all_rules
async def get_sites() -> list:
async with httpx.AsyncClient(timeout=30) as client:
resp = await client.get(