120 lines
4.9 KiB
Python
120 lines
4.9 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from typing import Any
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.analyzer import create_or_update_alert
|
|
from app.config import Settings
|
|
from app.llm import LLMClient
|
|
from app.models import Alert, TLSReport
|
|
|
|
|
|
def _json_list(value: str | None) -> list:
|
|
try:
|
|
parsed = json.loads(value or "[]")
|
|
except json.JSONDecodeError:
|
|
return []
|
|
return parsed if isinstance(parsed, list) else []
|
|
|
|
|
|
def _report_evidence(report: TLSReport) -> dict[str, Any]:
|
|
return {
|
|
"report_type": "tlsrpt",
|
|
"report_db_id": report.id,
|
|
"report_id": report.report_id,
|
|
"reporting_orgs": [report.org_name] if report.org_name else [],
|
|
"contact_info": report.contact_info,
|
|
"date_range": {
|
|
"begin": report.date_begin.isoformat() if report.date_begin else None,
|
|
"end": report.date_end.isoformat() if report.date_end else None,
|
|
},
|
|
}
|
|
|
|
|
|
def _severity(failures: int, total: int) -> str:
|
|
rate = failures / total * 100 if total else 0
|
|
if failures >= 100 or rate >= 20:
|
|
return "critical"
|
|
return "warning"
|
|
|
|
|
|
def analyze_tls_report(session: Session, settings: Settings, report: TLSReport, llm: LLMClient | None = None) -> list[tuple[Alert, bool, bool]]:
|
|
created: list[tuple[Alert, bool, bool]] = []
|
|
evidence = _report_evidence(report)
|
|
for policy in report.policies:
|
|
successes = policy.total_successful_session_count
|
|
failures = policy.total_failure_session_count
|
|
total = successes + failures
|
|
if failures <= 0:
|
|
continue
|
|
failure_rate = failures / total * 100 if total else 100
|
|
mx_hosts = _json_list(policy.mx_hosts_json)
|
|
details = {
|
|
**evidence,
|
|
"policy_type": policy.policy_type,
|
|
"policy_domain": policy.policy_domain,
|
|
"policy_strings": _json_list(policy.policy_strings_json),
|
|
"mx_hosts": mx_hosts,
|
|
"successful_sessions": successes,
|
|
"failed_sessions": failures,
|
|
"total_sessions": total,
|
|
"failure_rate_percent": round(failure_rate, 2),
|
|
}
|
|
created.append(
|
|
create_or_update_alert(
|
|
session,
|
|
inbox_id=report.inbox_id,
|
|
domain=report.domain,
|
|
severity=_severity(failures, total),
|
|
alert_type="tls_delivery_failures",
|
|
key=f"{policy.policy_domain}:{policy.policy_type or 'unknown'}",
|
|
title=f"TLS delivery failures reported for {policy.policy_domain}",
|
|
summary=(
|
|
f"{report.org_name or 'A reporter'} observed {failures} failed SMTP TLS sessions for "
|
|
f"{policy.policy_domain} ({failure_rate:.1f}% of {total or failures} reported sessions)."
|
|
),
|
|
details=details,
|
|
)
|
|
)
|
|
for failure in policy.failures:
|
|
if failure.failed_session_count <= 0:
|
|
continue
|
|
failure_details = {
|
|
**details,
|
|
"result_type": failure.result_type,
|
|
"sending_mta_ip": failure.sending_mta_ip,
|
|
"receiving_mx_hostname": failure.receiving_mx_hostname,
|
|
"receiving_mx_helo": failure.receiving_mx_helo,
|
|
"receiving_ip": failure.receiving_ip,
|
|
"failed_session_count": failure.failed_session_count,
|
|
"additional_information": failure.additional_information,
|
|
"failure_reason_code": failure.failure_reason_code,
|
|
}
|
|
target = failure.receiving_mx_hostname or ", ".join(mx_hosts) or policy.policy_domain
|
|
created.append(
|
|
create_or_update_alert(
|
|
session,
|
|
inbox_id=report.inbox_id,
|
|
domain=report.domain,
|
|
severity=_severity(failure.failed_session_count, total),
|
|
alert_type="tls_failure_detail",
|
|
key=f"{policy.policy_domain}:{failure.result_type or 'unknown'}:{target}:{failure.sending_mta_ip or 'unknown'}",
|
|
title=f"TLS failure detail for {target}",
|
|
summary=(
|
|
f"{failure.failed_session_count} sessions failed with "
|
|
f"{failure.result_type or 'an unspecified TLS result'} for {target}."
|
|
),
|
|
details=failure_details,
|
|
)
|
|
)
|
|
if llm and settings.llm.generate_alert_explanations:
|
|
for alert, is_new, severity_increased in created:
|
|
if (is_new or severity_increased) and alert.severity in {"warning", "critical"}:
|
|
explanation = llm.explain_alert(alert)
|
|
alert.llm_summary = explanation.summary
|
|
alert.llm_risk = explanation.risk
|
|
alert.llm_recommended_action = explanation.recommended_action
|
|
return created
|