add analysis base api

backend/src/api/analysis_api.py

@@ -0,0 +1,70 @@
+"""Analysis endpoints derived from captured packet history."""
+
+from datetime import datetime, timedelta, timezone
+from typing import Any, Dict, List, Optional
+
+from fastapi import APIRouter, HTTPException, Query
+from pydantic import BaseModel, Field
+
+import src.shared_objects as shared
+
+router = APIRouter()
+
+
+class InterfaceHostEvidence(BaseModel):
+    ip_address: Optional[str] = Field(None, description="Observed IP address for the host.")
+    mac_address: Optional[str] = Field(None, description="Observed MAC address for the host.")
+    packet_count: int = Field(..., description="How many packet observations supported this mapping.")
+    last_seen: datetime = Field(..., description="Most recent packet timestamp supporting this mapping.")
+    source_on_ingress_count: int = Field(..., description="Packets where this endpoint appeared as the source on ingress.")
+    destination_on_egress_count: int = Field(..., description="Packets where this endpoint appeared as the destination on egress.")
+
+
+class InterfaceAttachment(BaseModel):
+    interface: str = Field(..., description="MITM machine interface name.")
+    hosts: List[InterfaceHostEvidence] = Field(default_factory=list, description="Endpoints inferred to be attached to this interface.")
+
+
+class InterfaceHostAnalysisResponse(BaseModel):
+    since: Optional[datetime] = Field(None, description="Only packets at or after this timestamp were analyzed.")
+    interfaces: List[InterfaceAttachment] = Field(default_factory=list)
+    notes: List[str] = Field(
+        default_factory=lambda: [
+            "This is an inference from observed packet direction, not a kernel neighbor-table lookup.",
+            "A host is inferred on an interface when it appears as source on ingress or as destination on egress on that interface.",
+            "Broadcast and obviously incomplete endpoint records are ignored.",
+        ]
+    )
+
+
+@router.get("/interface-hosts", response_model=InterfaceHostAnalysisResponse)
+async def analysis_interface_hosts(
+    since_minutes: Optional[int] = Query(
+        60,
+        ge=1,
+        le=60 * 24 * 30,
+        description="Analyze only packets seen within the last N minutes. Set to a large value to cover more history.",
+    ),
+    limit_per_interface: int = Query(
+        100,
+        ge=1,
+        le=1000,
+        description="Maximum number of inferred hosts returned per interface.",
+    ),
+) -> InterfaceHostAnalysisResponse:
+    """Infer which IP/MAC endpoints are likely attached to each MITM-side interface."""
+    db = shared.db
+    if db is None:
+        raise HTTPException(status_code=503, detail="Database not available")
+
+    since: Optional[datetime] = None
+    if since_minutes is not None:
+        since = datetime.now(timezone.utc) - timedelta(minutes=since_minutes)
+
+    try:
+        rows = await db.infer_interface_hosts(since=since, limit_per_interface=limit_per_interface)
+    except Exception as exc:
+        raise HTTPException(status_code=500, detail="Failed to infer interface host mapping") from exc
+
+    interfaces = [InterfaceAttachment(**row) for row in rows]
+    return InterfaceHostAnalysisResponse(since=since, interfaces=interfaces)