feat: backend ingest API, Docker Compose, Ubuntu install guide
- FastAPI: POST /api/v1/events, GET /health, JSON Schema validation - PostgreSQL models, Alembic migration, bootstrap API key - deploy/docker-compose.yml, .env.example - docs/install-ubuntu-24.04.md, updated work-plan (agents first)
This commit is contained in:
@@ -0,0 +1,89 @@
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.models import Event, Host
|
||||
|
||||
|
||||
def _parse_dt(value: str) -> datetime:
|
||||
dt = datetime.fromisoformat(value.replace("Z", "+00:00"))
|
||||
if dt.tzinfo is None:
|
||||
dt = dt.replace(tzinfo=timezone.utc)
|
||||
return dt
|
||||
|
||||
|
||||
def upsert_host(db: Session, payload: dict) -> Host:
|
||||
source = payload.get("source") or {}
|
||||
host_data = payload.get("host") or {}
|
||||
agent_id = source.get("agent_instance_id")
|
||||
hostname = host_data.get("hostname") or "unknown"
|
||||
product = source.get("product") or "unknown"
|
||||
|
||||
host: Host | None = None
|
||||
if agent_id:
|
||||
host = db.scalar(select(Host).where(Host.agent_instance_id == agent_id))
|
||||
|
||||
if host is None:
|
||||
host = db.scalar(
|
||||
select(Host).where(Host.hostname == hostname, Host.product == product).limit(1)
|
||||
)
|
||||
|
||||
if host is None:
|
||||
host = Host(
|
||||
agent_instance_id=agent_id,
|
||||
hostname=hostname,
|
||||
display_name=host_data.get("display_name"),
|
||||
os_family=host_data.get("os_family", "linux"),
|
||||
os_version=host_data.get("os_version"),
|
||||
product=product,
|
||||
product_version=source.get("product_version"),
|
||||
ipv4=host_data.get("ipv4"),
|
||||
ipv6=host_data.get("ipv6"),
|
||||
tags=payload.get("tags") or [],
|
||||
)
|
||||
db.add(host)
|
||||
else:
|
||||
host.hostname = hostname
|
||||
host.display_name = host_data.get("display_name") or host.display_name
|
||||
host.os_version = host_data.get("os_version") or host.os_version
|
||||
host.product_version = source.get("product_version") or host.product_version
|
||||
host.ipv4 = host_data.get("ipv4") or host.ipv4
|
||||
host.ipv6 = host_data.get("ipv6") or host.ipv6
|
||||
if agent_id and not host.agent_instance_id:
|
||||
host.agent_instance_id = agent_id
|
||||
|
||||
host.last_seen_at = datetime.now(timezone.utc)
|
||||
db.flush()
|
||||
return host
|
||||
|
||||
|
||||
def ingest_event(db: Session, payload: dict) -> tuple[Event, bool]:
|
||||
"""
|
||||
Returns (event, created).
|
||||
created=False if event_id already exists (idempotent).
|
||||
"""
|
||||
event_id = payload["event_id"]
|
||||
existing = db.scalar(select(Event).where(Event.event_id == event_id))
|
||||
if existing:
|
||||
return existing, False
|
||||
|
||||
host = upsert_host(db, payload)
|
||||
event = Event(
|
||||
event_id=event_id,
|
||||
host_id=host.id,
|
||||
occurred_at=_parse_dt(payload["occurred_at"]),
|
||||
category=payload["category"],
|
||||
type=payload["type"],
|
||||
severity=payload["severity"],
|
||||
title=payload["title"],
|
||||
summary=payload["summary"],
|
||||
details=payload.get("details"),
|
||||
raw=payload.get("raw"),
|
||||
dedup_key=payload.get("dedup_key"),
|
||||
correlation_id=payload.get("correlation_id"),
|
||||
payload=payload,
|
||||
)
|
||||
db.add(event)
|
||||
db.flush()
|
||||
return event, True
|
||||
Reference in New Issue
Block a user