68 lines
2.6 KiB
Python
68 lines
2.6 KiB
Python
"""Blocklist-suggestion generation (Chapter 10 follow-up): no earlier
|
|
chapter assigned ownership of populating blocklist_suggestions or setting
|
|
ip_registry.is_flagged. Runs in the background aggregator pass (batch, not
|
|
request-time, per Ch02/03), reusing severity_scoring.py so there's exactly
|
|
one scoring model between the dashboard display and the flagging decision.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from collections import defaultdict
|
|
from datetime import date, datetime, timedelta
|
|
|
|
from app.extensions import db
|
|
from app.models.blocklist_suggestion import BlocklistSuggestion
|
|
from app.models.ip_registry import IPRegistry
|
|
from app.models.suspicious_event import SuspiciousEvent
|
|
from app.services.severity_scoring import SeverityInputs, compute_effective_severity
|
|
|
|
_RANK = {"low": 0, "medium": 1, "high": 2}
|
|
|
|
|
|
def refresh_blocklist_suggestions(day: date) -> None:
|
|
"""Flag an IP (is_flagged + a suggestion row) if its escalated severity
|
|
for `day` reaches 'high'. Idempotent — skips IPs already suggested.
|
|
"""
|
|
start = datetime.combine(day, datetime.min.time())
|
|
end = start + timedelta(days=1)
|
|
|
|
events = (
|
|
db.session.query(SuspiciousEvent.ip, SuspiciousEvent.timestamp, SuspiciousEvent.severity)
|
|
.filter(SuspiciousEvent.timestamp >= start, SuspiciousEvent.timestamp < end)
|
|
.all()
|
|
)
|
|
if not events:
|
|
return
|
|
|
|
by_ip: dict[str, list] = defaultdict(list)
|
|
for ip, ts, sev in events:
|
|
by_ip[ip].append((ts, sev))
|
|
|
|
already_suggested = {ip for (ip,) in db.session.query(BlocklistSuggestion.ip).distinct().all()}
|
|
|
|
for ip, ip_events in by_ip.items():
|
|
if ip in already_suggested:
|
|
continue
|
|
timestamps = sorted(ts for ts, _ in ip_events)
|
|
avg_interval = (
|
|
(timestamps[-1] - timestamps[0]).total_seconds() / (len(timestamps) - 1)
|
|
if len(timestamps) > 1 else None
|
|
)
|
|
worst_base = max((sev for _, sev in ip_events), key=lambda s: _RANK.get(s, 0))
|
|
effective = compute_effective_severity(
|
|
SeverityInputs(base_severity=worst_base, ip_event_count=len(ip_events), avg_interval_seconds=avg_interval)
|
|
)
|
|
if effective != "high":
|
|
continue
|
|
|
|
db.session.add(BlocklistSuggestion(
|
|
ip=ip,
|
|
reason=f"{len(ip_events)} suspicious event(s) on {day.isoformat()}, escalated to high severity",
|
|
created_at=datetime.utcnow(),
|
|
exported=False,
|
|
))
|
|
ip_row = db.session.get(IPRegistry, ip)
|
|
if ip_row is not None:
|
|
ip_row.is_flagged = True
|
|
|
|
db.session.commit()
|