551 lines
18 KiB
Python
551 lines
18 KiB
Python
from datetime import datetime
|
|
from typing import Optional
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from pydantic import BaseModel
|
|
|
|
from app.core.security import get_current_user
|
|
from app.db.session import get_db
|
|
from app.models.bgp_anomaly import BGPAnomaly
|
|
from app.models.bgp_incident import BGPIncident
|
|
from app.models.bgp_observation import BGPObservation
|
|
from app.models.user import User
|
|
from app.services.bgp_collector_locations import (
|
|
build_bgp_collector_location_query,
|
|
collect_bgp_collector_location_candidates,
|
|
get_bgp_collector_location_dict,
|
|
)
|
|
from app.services.bgp_collectors import build_bgp_collector_coverage
|
|
from app.services.ai_client import get_ai_provider_client
|
|
from app.api.v1.settings import get_web_search_client
|
|
from app.services.location.llm_fallback import (
|
|
collect_llm_location_fallback_candidate,
|
|
collect_location_search_evidence,
|
|
)
|
|
|
|
router = APIRouter()
|
|
|
|
BGP_SOURCES = ("ris_live_bgp", "bgpstream_bgp")
|
|
|
|
|
|
def _parse_dt(value: Optional[str]) -> Optional[datetime]:
|
|
if not value:
|
|
return None
|
|
return datetime.fromisoformat(value.replace("Z", "+00:00"))
|
|
def _event_filters(
|
|
*,
|
|
prefix: Optional[str],
|
|
origin_asn: Optional[int],
|
|
peer_asn: Optional[int],
|
|
collector: Optional[str],
|
|
event_type: Optional[str],
|
|
source: Optional[str],
|
|
time_from: Optional[datetime],
|
|
time_to: Optional[datetime],
|
|
):
|
|
filters = [BGPObservation.source.in_(BGP_SOURCES)]
|
|
if source:
|
|
filters.append(BGPObservation.source == source)
|
|
if prefix:
|
|
filters.append(BGPObservation.prefix == prefix)
|
|
if origin_asn is not None:
|
|
filters.append(BGPObservation.origin_asn == origin_asn)
|
|
if peer_asn is not None:
|
|
filters.append(BGPObservation.peer_asn == peer_asn)
|
|
if collector:
|
|
filters.append(BGPObservation.collector == collector)
|
|
if event_type:
|
|
filters.append(BGPObservation.event_type == event_type)
|
|
if time_from:
|
|
filters.append(BGPObservation.observed_at >= time_from)
|
|
if time_to:
|
|
filters.append(BGPObservation.observed_at <= time_to)
|
|
return filters
|
|
|
|
|
|
def _anomaly_filters(
|
|
*,
|
|
severity: Optional[str],
|
|
anomaly_type: Optional[str],
|
|
status: Optional[str],
|
|
prefix: Optional[str],
|
|
origin_asn: Optional[int],
|
|
time_from: Optional[datetime],
|
|
time_to: Optional[datetime],
|
|
):
|
|
filters = []
|
|
if severity:
|
|
filters.append(BGPAnomaly.severity == severity)
|
|
if anomaly_type:
|
|
filters.append(BGPAnomaly.anomaly_type == anomaly_type)
|
|
if status:
|
|
filters.append(BGPAnomaly.status == status)
|
|
if prefix:
|
|
filters.append(BGPAnomaly.prefix == prefix)
|
|
if origin_asn is not None:
|
|
filters.append(BGPAnomaly.origin_asn == origin_asn)
|
|
if time_from:
|
|
filters.append(BGPAnomaly.created_at >= time_from)
|
|
if time_to:
|
|
filters.append(BGPAnomaly.created_at <= time_to)
|
|
return filters
|
|
|
|
|
|
def _incident_filters(
|
|
*,
|
|
severity: Optional[str],
|
|
incident_type: Optional[str],
|
|
status: Optional[str],
|
|
):
|
|
filters = []
|
|
if severity:
|
|
filters.append(BGPIncident.severity == severity)
|
|
if incident_type:
|
|
filters.append(BGPIncident.incident_type == incident_type)
|
|
if status:
|
|
filters.append(BGPIncident.status == status)
|
|
return filters
|
|
|
|
|
|
async def _build_event_summary_payload(db: AsyncSession) -> dict:
|
|
base_filters = [BGPObservation.source.in_(BGP_SOURCES)]
|
|
|
|
total_result = await db.execute(
|
|
select(func.count(BGPObservation.id)).where(*base_filters)
|
|
)
|
|
collectors_result = await db.execute(
|
|
select(func.count(func.distinct(BGPObservation.collector))).where(
|
|
*base_filters, BGPObservation.collector.isnot(None)
|
|
)
|
|
)
|
|
prefixes_result = await db.execute(
|
|
select(func.count(func.distinct(BGPObservation.prefix))).where(
|
|
*base_filters, BGPObservation.prefix.isnot(None)
|
|
)
|
|
)
|
|
type_result = await db.execute(
|
|
select(BGPObservation.event_type, func.count(BGPObservation.id))
|
|
.where(*base_filters)
|
|
.group_by(BGPObservation.event_type)
|
|
)
|
|
|
|
return {
|
|
"total": total_result.scalar() or 0,
|
|
"collector_count": collectors_result.scalar() or 0,
|
|
"prefix_count": prefixes_result.scalar() or 0,
|
|
"by_type": {row[0]: row[1] for row in type_result.fetchall()},
|
|
}
|
|
|
|
|
|
async def _build_anomaly_summary_payload(db: AsyncSession) -> dict:
|
|
total_result = await db.execute(select(func.count(BGPAnomaly.id)))
|
|
type_result = await db.execute(
|
|
select(BGPAnomaly.anomaly_type, func.count(BGPAnomaly.id))
|
|
.group_by(BGPAnomaly.anomaly_type)
|
|
.order_by(func.count(BGPAnomaly.id).desc())
|
|
)
|
|
severity_result = await db.execute(
|
|
select(BGPAnomaly.severity, func.count(BGPAnomaly.id))
|
|
.group_by(BGPAnomaly.severity)
|
|
.order_by(func.count(BGPAnomaly.id).desc())
|
|
)
|
|
status_result = await db.execute(
|
|
select(BGPAnomaly.status, func.count(BGPAnomaly.id))
|
|
.group_by(BGPAnomaly.status)
|
|
.order_by(func.count(BGPAnomaly.id).desc())
|
|
)
|
|
|
|
return {
|
|
"total": total_result.scalar() or 0,
|
|
"by_type": {row[0]: row[1] for row in type_result.fetchall()},
|
|
"by_severity": {row[0]: row[1] for row in severity_result.fetchall()},
|
|
"by_status": {row[0]: row[1] for row in status_result.fetchall()},
|
|
}
|
|
|
|
|
|
async def _build_incident_summary_payload(db: AsyncSession) -> dict:
|
|
total_result = await db.execute(select(func.count(BGPIncident.id)))
|
|
type_result = await db.execute(
|
|
select(BGPIncident.incident_type, func.count(BGPIncident.id))
|
|
.group_by(BGPIncident.incident_type)
|
|
.order_by(func.count(BGPIncident.id).desc())
|
|
)
|
|
severity_result = await db.execute(
|
|
select(BGPIncident.severity, func.count(BGPIncident.id))
|
|
.group_by(BGPIncident.severity)
|
|
.order_by(func.count(BGPIncident.id).desc())
|
|
)
|
|
status_result = await db.execute(
|
|
select(BGPIncident.status, func.count(BGPIncident.id))
|
|
.group_by(BGPIncident.status)
|
|
.order_by(func.count(BGPIncident.id).desc())
|
|
)
|
|
|
|
return {
|
|
"total": total_result.scalar() or 0,
|
|
"by_type": {row[0]: row[1] for row in type_result.fetchall()},
|
|
"by_severity": {row[0]: row[1] for row in severity_result.fetchall()},
|
|
"by_status": {row[0]: row[1] for row in status_result.fetchall()},
|
|
}
|
|
|
|
|
|
@router.get("/events")
|
|
async def list_bgp_events(
|
|
prefix: Optional[str] = Query(None),
|
|
origin_asn: Optional[int] = Query(None),
|
|
peer_asn: Optional[int] = Query(None),
|
|
collector: Optional[str] = Query(None),
|
|
event_type: Optional[str] = Query(None),
|
|
source: Optional[str] = Query(None),
|
|
time_from: Optional[str] = Query(None),
|
|
time_to: Optional[str] = Query(None),
|
|
page: int = Query(1, ge=1),
|
|
page_size: int = Query(50, ge=1, le=200),
|
|
current_user: User = Depends(get_current_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
dt_from = _parse_dt(time_from)
|
|
dt_to = _parse_dt(time_to)
|
|
filters = _event_filters(
|
|
prefix=prefix,
|
|
origin_asn=origin_asn,
|
|
peer_asn=peer_asn,
|
|
collector=collector,
|
|
event_type=event_type,
|
|
source=source,
|
|
time_from=dt_from,
|
|
time_to=dt_to,
|
|
)
|
|
offset = (page - 1) * page_size
|
|
count_result = await db.execute(
|
|
select(func.count(BGPObservation.id)).where(*filters)
|
|
)
|
|
data_result = await db.execute(
|
|
select(BGPObservation)
|
|
.where(*filters)
|
|
.order_by(BGPObservation.observed_at.desc(), BGPObservation.id.desc())
|
|
.offset(offset)
|
|
.limit(page_size)
|
|
)
|
|
records = data_result.scalars().all()
|
|
|
|
return {
|
|
"total": count_result.scalar() or 0,
|
|
"page": page,
|
|
"page_size": page_size,
|
|
"data": [record.to_dict() for record in records],
|
|
}
|
|
|
|
|
|
@router.get("/events/summary")
|
|
async def get_bgp_event_summary(
|
|
current_user: User = Depends(get_current_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
return await _build_event_summary_payload(db)
|
|
|
|
|
|
@router.get("/collectors")
|
|
async def list_bgp_collectors(
|
|
current_user: User = Depends(get_current_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
data = await build_bgp_collector_coverage(db, source_filter=BGP_SOURCES)
|
|
return {
|
|
"total": len(data),
|
|
"data": data,
|
|
}
|
|
|
|
|
|
@router.get("/collectors/summary")
|
|
async def get_bgp_collector_summary(
|
|
current_user: User = Depends(get_current_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
collectors = await build_bgp_collector_coverage(db, source_filter=BGP_SOURCES)
|
|
active_collectors = [item for item in collectors if item["observation_count"] > 0]
|
|
return {
|
|
"total": len(collectors),
|
|
"active_collectors": len(active_collectors),
|
|
"observed_prefixes": sum(item["prefix_count"] for item in active_collectors),
|
|
"observed_origins": sum(item["origin_asn_count"] for item in active_collectors),
|
|
"recent_24h_events": sum(item["recent_24h_observation_count"] for item in active_collectors),
|
|
"recent_7d_events": sum(item["recent_7d_observation_count"] for item in active_collectors),
|
|
}
|
|
|
|
|
|
class CollectBGPCollectorLocationRequest(BaseModel):
|
|
city: Optional[str] = None
|
|
country: Optional[str] = None
|
|
site: Optional[str] = None
|
|
operator: Optional[str] = None
|
|
|
|
|
|
@router.post("/collectors/{collector_id}/collect-location")
|
|
async def collect_bgp_collector_location(
|
|
collector_id: str,
|
|
payload: CollectBGPCollectorLocationRequest,
|
|
current_user: User = Depends(get_current_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""Run the shared location pipeline for a BGP route collector.
|
|
|
|
Mirrors ``POST /api/v1/visualization/compute-centers/{source_id}/collect-location``.
|
|
Returns ranked candidates from source coordinates and Nominatim queries
|
|
built around the collector's stored context (IXP / city / country). Stored
|
|
collector locations provide context only; they are not emitted as
|
|
candidates.
|
|
"""
|
|
if not collector_id or not collector_id.strip():
|
|
raise HTTPException(status_code=400, detail="collector_id is required")
|
|
|
|
legacy = get_bgp_collector_location_dict(collector_id) or {}
|
|
site = payload.site or legacy.get("matched_location_name")
|
|
city = payload.city or legacy.get("city")
|
|
country = payload.country or legacy.get("country")
|
|
operator = payload.operator or "RIPE NCC"
|
|
|
|
candidates, attempted_queries = collect_bgp_collector_location_candidates(
|
|
collector=collector_id,
|
|
site=site,
|
|
city=city,
|
|
country=country,
|
|
operator=operator,
|
|
)
|
|
llm_failure_reason = None
|
|
if not candidates:
|
|
query = build_bgp_collector_location_query(
|
|
collector=collector_id,
|
|
site=site,
|
|
city=city,
|
|
country=country,
|
|
operator=operator,
|
|
)
|
|
llm_result = None
|
|
try:
|
|
web_search_client = await get_web_search_client(db)
|
|
search_result = await collect_location_search_evidence(
|
|
web_search_client=web_search_client,
|
|
query=query,
|
|
entity_type="bgp_collector",
|
|
)
|
|
attempted_queries = [*attempted_queries, *search_result.attempted_queries]
|
|
if not search_result.evidence:
|
|
llm_failure_reason = search_result.failure_reason
|
|
raise RuntimeError(search_result.failure_reason or "no WebSearch evidence")
|
|
provider_client = await get_ai_provider_client(db)
|
|
llm_result = await collect_llm_location_fallback_candidate(
|
|
provider_client=provider_client,
|
|
query=query,
|
|
entity_type="bgp_collector",
|
|
attempted_queries=attempted_queries,
|
|
search_evidence=search_result.evidence,
|
|
)
|
|
except Exception as exc:
|
|
if llm_failure_reason is None:
|
|
llm_failure_reason = f"LLM location factcheck unavailable: {exc}"
|
|
attempted_queries = [
|
|
*attempted_queries,
|
|
f"llm_factcheck:bgp_collector:{collector_id or 'unknown'}",
|
|
]
|
|
if llm_result is not None:
|
|
attempted_queries = [*attempted_queries, *llm_result.attempted_queries]
|
|
candidates = llm_result.candidates
|
|
llm_failure_reason = llm_result.failure_reason
|
|
|
|
context = {
|
|
"collector": collector_id,
|
|
"site": site,
|
|
"city": city,
|
|
"country": country,
|
|
"operator": operator,
|
|
}
|
|
|
|
if not candidates:
|
|
return {
|
|
"collector_id": collector_id,
|
|
"name": collector_id,
|
|
"success": False,
|
|
"failure_reason": (
|
|
"No source coordinates or online geocoding result reached"
|
|
" city-level precision for this collector."
|
|
),
|
|
"candidates": [],
|
|
"attempted_queries": list(attempted_queries),
|
|
"llm_failure_reason": llm_failure_reason,
|
|
"context": context,
|
|
}
|
|
|
|
return {
|
|
"collector_id": collector_id,
|
|
"name": collector_id,
|
|
"success": True,
|
|
"candidates": [candidate.to_dict() for candidate in candidates],
|
|
"best_candidate": candidates[0].to_dict(),
|
|
"attempted_queries": list(attempted_queries),
|
|
"context": context,
|
|
}
|
|
|
|
|
|
@router.get("/overview/summary")
|
|
async def get_bgp_overview_summary(
|
|
current_user: User = Depends(get_current_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
event_summary = await _build_event_summary_payload(db)
|
|
anomaly_summary = await _build_anomaly_summary_payload(db)
|
|
incident_summary = await _build_incident_summary_payload(db)
|
|
collectors = await build_bgp_collector_coverage(db, source_filter=BGP_SOURCES)
|
|
active_collectors = [item for item in collectors if item["observation_count"] > 0]
|
|
|
|
return {
|
|
"incidentSummary": incident_summary,
|
|
"anomalySummary": anomaly_summary,
|
|
"eventSummary": event_summary,
|
|
"collectorSummary": {
|
|
"total": len(collectors),
|
|
"active_collectors": len(active_collectors),
|
|
"observed_prefixes": sum(item["prefix_count"] for item in active_collectors),
|
|
"observed_origins": sum(item["origin_asn_count"] for item in active_collectors),
|
|
"recent_24h_events": sum(item["recent_24h_observation_count"] for item in active_collectors),
|
|
"recent_7d_events": sum(item["recent_7d_observation_count"] for item in active_collectors),
|
|
},
|
|
}
|
|
|
|
|
|
@router.get("/events/{event_id}")
|
|
async def get_bgp_event(
|
|
event_id: int,
|
|
current_user: User = Depends(get_current_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
record = await db.get(BGPObservation, event_id)
|
|
if not record or record.source not in BGP_SOURCES:
|
|
raise HTTPException(status_code=404, detail="BGP event not found")
|
|
return record.to_dict()
|
|
|
|
|
|
@router.get("/anomalies")
|
|
async def list_bgp_anomalies(
|
|
severity: Optional[str] = Query(None),
|
|
anomaly_type: Optional[str] = Query(None),
|
|
status: Optional[str] = Query(None),
|
|
prefix: Optional[str] = Query(None),
|
|
origin_asn: Optional[int] = Query(None),
|
|
time_from: Optional[str] = Query(None),
|
|
time_to: Optional[str] = Query(None),
|
|
page: int = Query(1, ge=1),
|
|
page_size: int = Query(50, ge=1, le=200),
|
|
current_user: User = Depends(get_current_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
dt_from = _parse_dt(time_from)
|
|
dt_to = _parse_dt(time_to)
|
|
filters = _anomaly_filters(
|
|
severity=severity,
|
|
anomaly_type=anomaly_type,
|
|
status=status,
|
|
prefix=prefix,
|
|
origin_asn=origin_asn,
|
|
time_from=dt_from,
|
|
time_to=dt_to,
|
|
)
|
|
offset = (page - 1) * page_size
|
|
total_result = await db.execute(
|
|
select(func.count(BGPAnomaly.id)).where(*filters)
|
|
)
|
|
data_result = await db.execute(
|
|
select(BGPAnomaly)
|
|
.where(*filters)
|
|
.order_by(BGPAnomaly.created_at.desc(), BGPAnomaly.id.desc())
|
|
.offset(offset)
|
|
.limit(page_size)
|
|
)
|
|
records = data_result.scalars().all()
|
|
|
|
return {
|
|
"total": total_result.scalar() or 0,
|
|
"page": page,
|
|
"page_size": page_size,
|
|
"data": [record.to_dict() for record in records],
|
|
}
|
|
|
|
|
|
@router.get("/anomalies/summary")
|
|
async def get_bgp_anomaly_summary(
|
|
current_user: User = Depends(get_current_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
return await _build_anomaly_summary_payload(db)
|
|
|
|
|
|
@router.get("/anomalies/{anomaly_id}")
|
|
async def get_bgp_anomaly(
|
|
anomaly_id: int,
|
|
current_user: User = Depends(get_current_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
record = await db.get(BGPAnomaly, anomaly_id)
|
|
if not record:
|
|
raise HTTPException(status_code=404, detail="BGP anomaly not found")
|
|
return record.to_dict()
|
|
|
|
|
|
@router.get("/incidents")
|
|
async def list_bgp_incidents(
|
|
severity: Optional[str] = Query(None),
|
|
incident_type: Optional[str] = Query(None),
|
|
status: Optional[str] = Query(None),
|
|
page: int = Query(1, ge=1),
|
|
page_size: int = Query(50, ge=1, le=200),
|
|
current_user: User = Depends(get_current_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
filters = _incident_filters(
|
|
severity=severity,
|
|
incident_type=incident_type,
|
|
status=status,
|
|
)
|
|
offset = (page - 1) * page_size
|
|
total_result = await db.execute(
|
|
select(func.count(BGPIncident.id)).where(*filters)
|
|
)
|
|
data_result = await db.execute(
|
|
select(BGPIncident)
|
|
.where(*filters)
|
|
.order_by(BGPIncident.created_at.desc(), BGPIncident.id.desc())
|
|
.offset(offset)
|
|
.limit(page_size)
|
|
)
|
|
records = data_result.scalars().all()
|
|
|
|
return {
|
|
"total": total_result.scalar() or 0,
|
|
"page": page,
|
|
"page_size": page_size,
|
|
"data": [record.to_dict() for record in records],
|
|
}
|
|
|
|
|
|
@router.get("/incidents/summary")
|
|
async def get_bgp_incident_summary(
|
|
current_user: User = Depends(get_current_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
return await _build_incident_summary_payload(db)
|
|
|
|
|
|
@router.get("/incidents/{incident_id}")
|
|
async def get_bgp_incident(
|
|
incident_id: int,
|
|
current_user: User = Depends(get_current_user),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
record = await db.get(BGPIncident, incident_id)
|
|
if not record:
|
|
raise HTTPException(status_code=404, detail="BGP incident not found")
|
|
return record.to_dict()
|