feat(scans): a sync that dies at login is now a failed run, on screen and by mail

A Wazuh API that refused every login for a day was visible only in the
container log. Every scan-jobs row is per asset, and a sync that fails at
authentication never reaches an asset — so /scans showed a clean history of
COMPLETED runs, the manual button said "Sync Failed — check the logs", and
nobody was told.

Ledger: every sync execution (Wazuh, Nessus, Intune, vCenter, IGEL; button
or scheduler) is one sync_runs row with start, end, status, stats and the
error that stopped it, written from its own session so the record never
depends on the job's transaction. A run where every agent failed the same
way counts as failed, not as "0 triggered". Rows left running by a restart
are closed at startup.

Visible: Scan Jobs gets a Sync Health strip (per source: state, last run,
last success, error) and a Sync Runs table; the dashboard shows a banner
while a source is failed or stale (stale follows the schedule interval, a
weekly Nessus is not stale after six days); the Sync Failed modal shows the
real reason, and the sync endpoints return it instead of "check server
logs".

Mail: a failed run mails the sync-alert recipients (own setting, else the
notification defaults) with the error and the last successful sync, i.e.
how long the coverage gap already is — one mail per source per 24h while
it keeps failing, logged as SYNC_FAILURE in the notification log.
This commit is contained in:
2026-09-03 18:11:24 +02:00
parent ece21b9a00
commit 9ca0be24f0
18 changed files with 1333 additions and 95 deletions
+50
View File
@@ -0,0 +1,50 @@
"""Sync run ledger + SYNC_FAILURE notification type
Revision ID: 050
Revises: 049
Create Date: 2026-09-03 14:00:00.000000
A Wazuh API that refused every login for a day was visible only in the
container log. Every scan-jobs row is per asset, and a sync that fails at
authentication never reaches an asset — so /scans showed a clean history of
COMPLETED runs while nothing had been synced. This table records every sync
execution (any source, button or scheduler) with its outcome, and the new
notification type carries the failure mail in the notification log.
ALTER TYPE ... ADD VALUE cannot run inside a transaction block on older
Postgres → autocommit_block. Idempotent (IF NOT EXISTS).
"""
from alembic import op
import sqlalchemy as sa
revision = "050"
down_revision = "049"
branch_labels = None
depends_on = None
def upgrade() -> None:
op.create_table(
"sync_runs",
sa.Column("id", sa.Integer(), primary_key=True),
sa.Column("source", sa.String(length=30), nullable=False),
sa.Column("trigger", sa.String(length=20), nullable=False, server_default="manual"),
sa.Column("status", sa.String(length=20), nullable=False, server_default="running"),
sa.Column("started_at", sa.DateTime(), nullable=False),
sa.Column("finished_at", sa.DateTime(), nullable=True),
sa.Column("error_message", sa.Text(), nullable=True),
sa.Column("stats", sa.Text(), nullable=True),
sa.Column("alerted_at", sa.DateTime(), nullable=True),
)
op.create_index("ix_sync_runs_source", "sync_runs", ["source"])
op.create_index("ix_sync_runs_status", "sync_runs", ["status"])
with op.get_context().autocommit_block():
op.execute("ALTER TYPE notificationtype ADD VALUE IF NOT EXISTS 'SYNC_FAILURE'")
def downgrade() -> None:
op.drop_index("ix_sync_runs_status", table_name="sync_runs")
op.drop_index("ix_sync_runs_source", table_name="sync_runs")
op.drop_table("sync_runs")
# Postgres cannot drop an enum label — see 043; deliberately left in place.
+2
View File
@@ -13,6 +13,7 @@ from app.models.policy import Policy
from app.models.scan_schedule import ScanSchedule
from app.models.notification_log import NotificationLog
from app.models.setting import Setting
from app.models.sync_run import SyncRun
from app.models.ai_report import AIReport
from app.models.cve_remediation import CveRemediation
from app.models.app_cve_cache import AppCveCache
@@ -33,6 +34,7 @@ __all__ = [
"ScanSchedule",
"NotificationLog",
"Setting",
"SyncRun",
"AIReport",
"CveRemediation",
"AppCveCache",
+2
View File
@@ -17,6 +17,8 @@ class NotificationType(str, Enum):
# Actively exploited (CISA KEV / ENISA EUVD) AND present in our inventory —
# its own type so the log can be filtered for the alerts that meant "now".
KEV_ALERT = "kev_alert"
# A sync job (Wazuh/Nessus/Intune/vCenter/IGEL) failed — coverage gap.
SYNC_FAILURE = "sync_failure"
MANUAL = "manual"
+31
View File
@@ -0,0 +1,31 @@
"""SyncRun — one row per sync job execution, whatever the source or trigger.
The `scans` table records per-asset results, so a sync that dies BEFORE it
reaches any asset (Wazuh authentication refused, Nessus unreachable, IGEL UMS
down) left no trace anywhere but the container log. This ledger is the
source-level record: started, finished, status, and the error that stopped it.
"""
from sqlalchemy import Column, Integer, String, DateTime, Text
from app.models.base import Base
class SyncRun(Base):
__tablename__ = "sync_runs"
id = Column(Integer, primary_key=True, index=True)
# wazuh | nessus | intune | vcenter | igel
source = Column(String(30), nullable=False, index=True)
# manual | scheduled
trigger = Column(String(20), nullable=False, default="manual")
# running | completed | failed (plain strings: no Postgres enum to migrate)
status = Column(String(20), nullable=False, default="running", index=True)
started_at = Column(DateTime, nullable=False)
finished_at = Column(DateTime, nullable=True)
error_message = Column(Text, nullable=True)
# JSON: the service's stats dict (agents synced, created, errors[] ...)
stats = Column(Text, nullable=True)
# When a failure mail went out for this run — the cooldown anchor.
alerted_at = Column(DateTime, nullable=True)
def __repr__(self):
return f"<SyncRun(source='{self.source}', status='{self.status}')>"
+15 -1
View File
@@ -1039,7 +1039,19 @@ def sync_wazuh_assets(
"""
Synchronizes assets from Wazuh Manager.
fetch active agents -> update/create local assets.
Recorded in the sync-run ledger as the "assets" phase; when the
scheduled scan calls this, the record merges into that run.
"""
from app.services.sync_run_service import record_sync_run
with record_sync_run("wazuh") as run:
run.stats.setdefault("phase", "assets")
stats = _sync_wazuh_assets(db)
run.stats.update(stats)
return stats
def _sync_wazuh_assets(db: Session) -> dict:
# 1. Get Wazuh Config (transparently decrypted)
from app.auth.setting_crypto import read_setting_value
raw_wazuh = read_setting_value(db, "wazuh_config")
@@ -1276,7 +1288,9 @@ def sync_wazuh_assets(
except Exception as e:
logger.error(f"Wazuh Sync Failed: {e}")
raise HTTPException(status_code=500, detail="Wazuh sync failed. Check server logs for details.")
# The real reason goes to the caller (editors only): "check the
# server logs" left an operator with a failing Wazuh API for a day.
raise HTTPException(status_code=502, detail=f"Wazuh sync failed: {e}")
return stats
+4 -1
View File
@@ -17,6 +17,7 @@ from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from app.auth.dependencies import RequireAdmin, RequireEditor
from app.services.sync_run_service import record_sync_run
from app.database import get_db, SessionLocal
from app.models.user import User
from app.services.igel_service import (
@@ -61,7 +62,9 @@ async def test_igel(
def _run_igel_sync_threaded() -> None:
db = SessionLocal()
try:
stats = run_igel_sync(db)
with record_sync_run("igel", "manual") as run:
stats = run_igel_sync(db)
run.stats.update(stats)
result = {k: v for k, v in stats.items() if k != "errors"}
_IGEL_SYNC["result"] = result
_IGEL_SYNC["error"] = None
+4 -1
View File
@@ -19,6 +19,7 @@ from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from app.auth.dependencies import RequireAdmin, RequireEditor
from app.services.sync_run_service import record_sync_run
from app.database import get_db, SessionLocal
from app.models.user import User
from app.services.intune_service import load_intune_config, run_intune_sync, _build_client
@@ -53,7 +54,9 @@ async def test_intune(
def _run_intune_sync_threaded() -> None:
db = SessionLocal()
try:
stats = run_intune_sync(db)
with record_sync_run("intune", "manual") as run:
stats = run_intune_sync(db)
run.stats.update(stats)
result = {k: v for k, v in stats.items() if k != "errors"}
_INTUNE_SYNC["result"] = result
_INTUNE_SYNC["error"] = None
+4 -1
View File
@@ -17,6 +17,7 @@ from pydantic import BaseModel, Field
from sqlalchemy.orm import Session
from app.auth.dependencies import RequireAdmin, RequireEditor
from app.services.sync_run_service import record_sync_run
from app.database import get_db
from app.integrations.nessus_client import (
NessusAPIError,
@@ -152,7 +153,9 @@ def trigger_nessus_sync(
"""
import httpx as _httpx
try:
stats = run_nessus_sync(db, scan_ids=payload.scan_ids)
with record_sync_run("nessus", "manual") as run:
stats = run_nessus_sync(db, scan_ids=payload.scan_ids)
run.stats.update(stats)
return {"message": "Nessus sync completed", **stats}
except RuntimeError as e:
# config missing
+39 -1
View File
@@ -235,7 +235,20 @@ def trigger_autoscan(
):
"""
Trigger automated scans for ALL assets connected to Wazuh.
Recorded in the sync-run ledger; a run where every agent failed
(authentication refused) counts as failed, not as 0 triggered.
"""
from app.services.sync_run_service import record_sync_run, fail_if_nothing_synced
with record_sync_run("wazuh") as run:
run.stats["phase"] = "vulnerabilities"
result = _autoscan(db)
run.stats.update({"agents_synced": result.get("scans_triggered", 0),
"errors": result.get("errors", [])})
fail_if_nothing_synced(run, result.get("scans_triggered", 0), result.get("errors", []))
return result
def _autoscan(db: Session) -> dict:
# 1. Get Wazuh Configuration from DB (transparently decrypted)
from app.auth.setting_crypto import read_setting_value
raw_wazuh = read_setting_value(db, "wazuh_config")
@@ -318,7 +331,7 @@ def trigger_autoscan(
scan_obj.error_message = f"Wazuh connection failed: {str(e)}"
scan_obj.completed_at = datetime.now()
db.commit()
raise HTTPException(status_code=500, detail="Failed to connect to Wazuh. Check server logs for details.")
raise HTTPException(status_code=502, detail=f"Failed to connect to Wazuh: {e}")
return {
"message": f"Triggered scans for {triggered_count} assets.",
@@ -436,3 +449,28 @@ async def delete_schedule(
db.delete(schedule)
db.commit()
# --- Sync run ledger (every source, manual and scheduled) ---
@router.get("/sync-runs")
async def list_sync_runs(
limit: int = Query(30, ge=1, le=200),
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
"""Latest sync executions across all sources, newest first."""
from app.services.sync_run_service import list_runs
return list_runs(db, limit)
@router.get("/sync-health")
async def get_sync_health(
db: Session = Depends(get_db),
current_user: User = Depends(get_current_user),
):
"""Per source: last run, last success, and a state the GUI can colour
(ok / failed / stale / running / unconfigured) — plus the failure-mail
settings so the panel can edit them in place."""
from app.services.sync_run_service import health_payload
return health_payload(db)
+4 -1
View File
@@ -18,6 +18,7 @@ from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from app.auth.dependencies import RequireAdmin, RequireEditor
from app.services.sync_run_service import record_sync_run
from app.database import get_db, SessionLocal
from app.models.user import User
from app.services.vcenter_service import (
@@ -54,7 +55,9 @@ async def test_vcenter(
def _run_vcenter_sync_threaded() -> None:
db = SessionLocal()
try:
stats = run_vcenter_sync(db)
with record_sync_run("vcenter", "manual") as run:
stats = run_vcenter_sync(db)
run.stats.update(stats)
result = {k: v for k, v in stats.items() if k != "errors"}
_VCENTER_SYNC["result"] = result
_VCENTER_SYNC["error"] = None
+14 -2
View File
@@ -1788,12 +1788,24 @@ def sync_vulnerabilities_from_wazuh(
try:
result = run_wazuh_vulnerability_sync(db)
return result
except HTTPException:
raise # already carries the real reason and status
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
def run_wazuh_vulnerability_sync(db: Session) -> dict:
"""Synchronous Wazuh vulnerability sync with statistics return"""
"""Synchronous Wazuh vulnerability sync with statistics return.
Recorded in the sync-run ledger as the "vulnerabilities" phase."""
from app.services.sync_run_service import record_sync_run
with record_sync_run("wazuh") as run:
run.stats.setdefault("phase", "vulnerabilities")
result = _run_wazuh_vulnerability_sync(db)
run.stats.update({k: v for k, v in result.items() if k != "message"})
return result
def _run_wazuh_vulnerability_sync(db: Session) -> dict:
# 1. Get Wazuh Config from DB (transparently decrypted)
from app.auth.setting_crypto import read_setting_value
@@ -2177,7 +2189,7 @@ def run_wazuh_vulnerability_sync(db: Session) -> dict:
db.rollback()
wazuh.close()
logger.error(f"Wazuh sync failed: {e}")
raise HTTPException(status_code=500, detail="Wazuh sync failed. Check server logs for details.")
raise HTTPException(status_code=502, detail=f"Wazuh sync failed: {e}")
async def sync_wazuh_vulnerabilities(db_session: Optional[Session] = None):
+105 -85
View File
@@ -29,6 +29,9 @@ from app.models.policy import Policy
from app.models.notification_log import NotificationLog, NotificationType, NotificationStatus
from app.models.user import User
from app.integrations.wazuh_client import WazuhClient
from app.services.sync_run_service import (
record_sync_run, mark_interrupted_runs, fail_if_nothing_synced,
)
scheduler = AsyncIOScheduler() if HAS_APSCHEDULER else None
@@ -59,7 +62,9 @@ def execute_scheduled_scan(schedule_id: int):
if scanner == "nessus":
try:
from app.services.nessus_sync import run_nessus_sync
stats = run_nessus_sync(db)
with record_sync_run("nessus", "scheduled") as run:
stats = run_nessus_sync(db)
run.stats.update(stats)
logger.info(f"Scheduled Nessus sync done: {stats}")
except Exception as e:
logger.error(f"Scheduled Nessus sync failed: {e}")
@@ -85,97 +90,104 @@ def execute_scheduled_scan(schedule_id: int):
db.commit()
return
# ---- Wazuh branch (default, existing behaviour) ----
# Wazuh Config laden (transparently decrypted)
from app.auth.setting_crypto import read_setting_value
raw_wazuh = read_setting_value(db, "wazuh_config")
if not raw_wazuh:
logger.error("Scheduled scan aborted: Wazuh configuration missing")
return
with record_sync_run("wazuh", "scheduled") as run:
# ---- Wazuh branch (default, existing behaviour) ----
# Wazuh Config laden (transparently decrypted)
from app.auth.setting_crypto import read_setting_value
raw_wazuh = read_setting_value(db, "wazuh_config")
if not raw_wazuh:
logger.error("Scheduled scan aborted: Wazuh configuration missing")
run.fail("Wazuh configuration missing")
return
config = json.loads(raw_wazuh)
config = json.loads(raw_wazuh)
# Asset sync first — the same call the manual "Sync Data (Wazuh)"
# button makes before it syncs vulnerabilities. Without it this job
# only ever sees the agent ids already in the DB: a host that
# re-registers with a NEW agent id keeps its stale id forever, so
# every CVE and EOL lookup asks Wazuh about an agent that no longer
# exists, and the asset never flips back from INACTIVE to ACTIVE
# until someone clicks sync by hand.
# ponytail: the endpoint function is called directly (current_user is
# unused in its body) rather than extracted into a service — same
# pattern as the sync_agent_vulnerabilities import below.
try:
from app.routers.assets import sync_wazuh_assets
logger.info(f"Scheduled Wazuh asset sync: {sync_wazuh_assets(db=db, current_user=None)}")
except Exception as e:
logger.error(f"Scheduled Wazuh asset sync failed: {e}")
# Asset sync first — the same call the manual "Sync Data (Wazuh)"
# button makes before it syncs vulnerabilities. Without it this job
# only ever sees the agent ids already in the DB: a host that
# re-registers with a NEW agent id keeps its stale id forever, so
# every CVE and EOL lookup asks Wazuh about an agent that no longer
# exists, and the asset never flips back from INACTIVE to ACTIVE
# until someone clicks sync by hand.
# ponytail: the endpoint function is called directly (current_user is
# unused in its body) rather than extracted into a service — same
# pattern as the sync_agent_vulnerabilities import below.
try:
from app.routers.assets import sync_wazuh_assets
logger.info(f"Scheduled Wazuh asset sync: {sync_wazuh_assets(db=db, current_user=None)}")
except Exception as e:
logger.error(f"Scheduled Wazuh asset sync failed: {e}")
run.stats.setdefault("errors", []).append(f"asset sync: {e}")
assets = db.query(Asset).filter(Asset.wazuh_agent_id.isnot(None)).all()
if not assets:
logger.info("No Wazuh assets found for scheduled scan")
schedule.last_run = datetime.now()
db.commit()
return
assets = db.query(Asset).filter(Asset.wazuh_agent_id.isnot(None)).all()
if not assets:
logger.info("No Wazuh assets found for scheduled scan")
schedule.last_run = datetime.now()
db.commit()
return
triggered = 0
errors = []
triggered = 0
errors = []
try:
with WazuhClient(
base_url=config.get("api_url"),
username=config.get("username"),
password=config.get("password"),
indexer_url=config.get("indexer_url"),
indexer_username=config.get("indexer_username"),
indexer_password=config.get("indexer_password"),
verify_ssl=bool(config.get("verify_ssl", True))
) as client:
# Sync vulnerabilities for each agent
from app.routers.vulnerabilities import (
sync_agent_vulnerabilities, reconcile_empty_agents,
)
# Shared across the run so an agent that returns nothing can be
# judged against the run as a whole (see reconcile_empty_agents).
run_stats: dict = {}
for asset in assets:
try:
with WazuhClient(
base_url=config.get("api_url"),
username=config.get("username"),
password=config.get("password"),
indexer_url=config.get("indexer_url"),
indexer_username=config.get("indexer_username"),
indexer_password=config.get("indexer_password"),
verify_ssl=bool(config.get("verify_ssl", True))
) as client:
# Sync vulnerabilities for each agent
from app.routers.vulnerabilities import (
sync_agent_vulnerabilities, reconcile_empty_agents,
)
# Shared across the run so an agent that returns nothing can be
# judged against the run as a whole (see reconcile_empty_agents).
run_stats: dict = {}
for asset in assets:
try:
scan = Scan(
asset_id=asset.id,
scan_type=ScanType.WAZUH, # source-labelled (was FULL)
status=ScanStatus.RUNNING,
started_at=datetime.now()
)
db.add(scan)
sync_agent_vulnerabilities(db, client, asset.wazuh_agent_id,
asset, run_stats=run_stats)
scan.status = ScanStatus.COMPLETED
scan.completed_at = datetime.now()
triggered += 1
except Exception as e:
scan.status = ScanStatus.FAILED
scan.error_message = str(e)
errors.append(f"{asset.hostname}: {e}")
logger.error(f"Scheduled scan error for {asset.hostname}: {e}")
# Now that the run is done, agents that returned nothing can be
# judged: real emptiness if the API answered for anyone else.
try:
scan = Scan(
asset_id=asset.id,
scan_type=ScanType.WAZUH, # source-labelled (was FULL)
status=ScanStatus.RUNNING,
started_at=datetime.now()
)
db.add(scan)
sync_agent_vulnerabilities(db, client, asset.wazuh_agent_id,
asset, run_stats=run_stats)
scan.status = ScanStatus.COMPLETED
scan.completed_at = datetime.now()
triggered += 1
reconcile_empty_agents(db, run_stats)
except Exception as e:
scan.status = ScanStatus.FAILED
scan.error_message = str(e)
errors.append(f"{asset.hostname}: {e}")
logger.error(f"Scheduled scan error for {asset.hostname}: {e}")
logger.error(f"Wazuh empty-agent reconcile failed: {e}")
# Now that the run is done, agents that returned nothing can be
# judged: real emptiness if the API answered for anyone else.
try:
reconcile_empty_agents(db, run_stats)
except Exception as e:
logger.error(f"Wazuh empty-agent reconcile failed: {e}")
except Exception as e:
logger.error(f"Wazuh connection error during scheduled scan: {e}")
run.fail(f"Wazuh connection failed: {e}")
except Exception as e:
logger.error(f"Wazuh connection error during scheduled scan: {e}")
schedule.last_run = datetime.now()
interval_delta = INTERVAL_MAP.get(schedule.interval, {"days": 1})
schedule.next_run = datetime.now() + timedelta(**interval_delta)
db.commit()
schedule.last_run = datetime.now()
interval_delta = INTERVAL_MAP.get(schedule.interval, {"days": 1})
schedule.next_run = datetime.now() + timedelta(**interval_delta)
db.commit()
logger.info(f"Scheduled scan '{schedule.name}' completed: {triggered} assets scanned, {len(errors)} errors")
logger.info(f"Scheduled scan '{schedule.name}' completed: {triggered} assets scanned, {len(errors)} errors")
run.stats.update({"phase": "assets+vulnerabilities", "agents_synced": triggered,
"errors": run.stats.get("errors", []) + errors})
fail_if_nothing_synced(run, triggered, errors)
except Exception as e:
logger.error(f"Scheduled scan error: {e}")
@@ -592,7 +604,9 @@ def intune_sync_nightly():
m365_service.fetch_security_data(db, force_refresh=True)
except Exception as e:
logger.warning("M365 page refresh failed (non-fatal, cache kept): %s", e)
stats = run_intune_sync(db)
with record_sync_run("intune", "scheduled") as run:
stats = run_intune_sync(db)
run.stats.update(stats)
logger.info("Intune nightly: %s", {k: v for k, v in stats.items() if k != "errors"})
except Exception as e:
logger.error("Intune nightly failed: %s", e)
@@ -618,7 +632,9 @@ def vcenter_sync_nightly():
if not load_vcenter_config(db):
logger.info("vCenter sync skipped — vcenter_config not set")
return
stats = run_vcenter_sync(db, refresh_catalog=True)
with record_sync_run("vcenter", "scheduled") as run:
stats = run_vcenter_sync(db, refresh_catalog=True)
run.stats.update(stats)
logger.info("vCenter nightly: %s", {k: v for k, v in stats.items() if k != "errors"})
except Exception as e:
logger.error("vCenter nightly failed: %s", e)
@@ -640,7 +656,9 @@ def igel_sync_nightly():
if not load_igel_config(db):
logger.info("IGEL sync skipped — igel_config not set")
return
stats = run_igel_sync(db)
with record_sync_run("igel", "scheduled") as run:
stats = run_igel_sync(db)
run.stats.update(stats)
logger.info("IGEL nightly: %s", {k: v for k, v in stats.items() if k != "errors"})
except Exception as e:
logger.error("IGEL nightly failed: %s", e)
@@ -1403,6 +1421,8 @@ def start_scheduler():
replace_existing=True,
)
# Rows left "running" by a restart mid-sync would show a spinner forever.
mark_interrupted_runs()
scheduler.start()
logger.info(
"Background scheduler started (SLA Breach Checker + Threat Intel Refresh + KEV Alert Hourly + Vulnrichment Nightly + Compliance SCA Nightly + URS Nightly + Audit-Log Prune)"
+429
View File
@@ -0,0 +1,429 @@
"""Sync run ledger + failure alerting.
Every sync job Wazuh, Nessus, Intune, vCenter, IGEL; button or scheduler
runs inside `record_sync_run(source, trigger)`. That writes ONE `sync_runs`
row: started, finished, status, the stats dict, and the error that stopped it.
The ledger uses its OWN session. The job's session may be mid-transaction or
already rolled back when the job dies, and the record of the failure must not
depend on it.
A failed run sends one mail per source per ALERT_COOLDOWN (24h) an hourly
schedule against a dead API must not send 24 mails and states the last
successful sync, i.e. how long the coverage gap already is.
"""
import contextlib
import html
import json
import logging
import os
import re
import threading
from datetime import datetime, timedelta
from typing import Dict, Iterable, List, Optional
from sqlalchemy.orm import Session
from app.database import SessionLocal
from app.models.sync_run import SyncRun
logger = logging.getLogger(__name__)
SOURCES = {
"wazuh": "Wazuh",
"nessus": "Tenable Nessus",
"intune": "Microsoft Intune",
"vcenter": "VMware vCenter",
"igel": "IGEL UMS",
}
# A source counts as configured when its config setting exists and is non-empty.
CONFIG_KEYS = {
"wazuh": "wazuh_config",
"nessus": "nessus_config",
"intune": "intune_config",
"vcenter": "vcenter_config",
"igel": "igel_config",
}
# Inventory syncs (Intune/vCenter/IGEL) are nightly: 36h means "missed one
# night" without flagging a slow one. Wazuh and Nessus follow their own
# ScanSchedule — see stale_thresholds(): twice the interval, at least 36h,
# and never stale without an enabled schedule (manual-only sources).
STALE_AFTER = timedelta(hours=36)
CRON_STALE_AFTER = timedelta(days=8) # a cron expression we do not parse
ALERT_COOLDOWN = timedelta(hours=24)
SETTING_ENABLED = "sync_alert_enabled"
SETTING_RECIPIENTS = "sync_alert_recipients"
MAX_ERRORS_KEPT = 50
_local = threading.local()
_alert_lock = threading.Lock()
class SyncRunHandle:
"""What the job sees: a stats dict to fill and fail() for a run that
ended without an exception but did nothing useful."""
def __init__(self, run_id: Optional[int], source: str):
self.run_id = run_id
self.source = source
self.stats: dict = {}
self.error: Optional[str] = None
def fail(self, message: str) -> None:
self.error = str(message)[:2000]
def fail_if_nothing_synced(handle: SyncRunHandle, triggered: int, errors: list) -> None:
"""Every agent failed the same way (authentication refused, API down):
that is a failed run, not a completed one with N errors. Used by the
scheduled Wazuh scan and the manual autoscan alike."""
if triggered == 0 and errors:
handle.fail(f"all {len(errors)} agents failed — {errors[0]}")
def _error_text(e: BaseException) -> str:
detail = getattr(e, "detail", None) # HTTPException carries the message there
text = str(detail) if detail else (str(e) or type(e).__name__)
return text[:2000]
@contextlib.contextmanager
def record_sync_run(source: str, trigger: str = "manual"):
"""Record one sync execution. Nested use (the scheduled Wazuh scan calls
the same asset-sync function the button calls) yields the OUTER handle so
a run is one row, not two, and one mail, not two."""
outer = getattr(_local, "active", None)
if outer is not None and outer.source == source:
yield outer
return
db = SessionLocal()
run = SyncRun(source=source, trigger=trigger, status="running",
started_at=datetime.now())
try:
db.add(run)
db.commit()
except Exception as e: # ledger unavailable: the job still has to run
logger.error("sync_runs: could not open run for %s: %s", source, e)
db.close()
yield SyncRunHandle(None, source)
return
handle = SyncRunHandle(run.id, source)
_local.active = handle
try:
yield handle
except BaseException as e:
handle.error = handle.error or _error_text(e)
raise
finally:
_local.active = outer
try:
_finish(db, run, handle)
except Exception as e:
logger.error("sync_runs: could not close run %s: %s", run.id, e)
finally:
db.close()
def _finish(db: Session, run: SyncRun, handle: SyncRunHandle) -> None:
stats = dict(handle.stats)
errs = stats.get("errors")
if isinstance(errs, list):
stats["error_count"] = len(errs)
stats["errors"] = [str(x)[:500] for x in errs[:MAX_ERRORS_KEPT]]
run.finished_at = datetime.now()
run.status = "failed" if handle.error else "completed"
run.error_message = handle.error
try:
run.stats = json.dumps(stats, default=str)
except Exception:
run.stats = None
db.commit()
level = logger.error if handle.error else logger.info
level("sync run #%s %s (%s): %s%s", run.id, run.source, run.trigger, run.status,
f"{handle.error}" if handle.error else "")
if handle.error:
try:
notify_sync_failure(db, run)
except Exception as e:
logger.error("sync failure mail for %s failed: %s", run.source, e)
def mark_interrupted_runs() -> int:
"""Close rows left 'running' by a backend restart mid-sync, else the
health panel would show a spinner forever. Called from start_scheduler."""
db = SessionLocal()
try:
rows = db.query(SyncRun).filter(SyncRun.status == "running").all()
for r in rows:
r.status = "failed"
r.finished_at = datetime.now()
r.error_message = "interrupted: backend restarted while the sync was running"
if rows:
db.commit()
logger.warning("sync_runs: closed %d interrupted run(s)", len(rows))
return len(rows)
except Exception as e:
logger.error("sync_runs: interrupted-run sweep failed: %s", e)
return 0
finally:
db.close()
# ---- health ---------------------------------------------------------------
def _setting(db: Session, key: str) -> Optional[str]:
from app.models.setting import Setting
try:
row = db.query(Setting).filter(Setting.key == key).first()
except Exception:
return None
return row.value if row and row.value is not None else None
def is_alert_enabled(db: Session) -> bool:
raw = (_setting(db, SETTING_ENABLED) or "").strip().strip('"').lower()
return raw not in ("false", "0", "no", "off")
def get_recipients(db: Session) -> List[tuple]:
"""(user_id_or_None, email, display_name). Dedicated setting wins, else
the shared notification default (configured list, else all admins)."""
from app.models.user import User
from app.services.email_service import get_default_recipients
raw = _setting(db, SETTING_RECIPIENTS)
if raw and raw.strip():
out, seen = [], set()
for em in (e.strip() for e in re.split(r"[,;\s]+", raw) if e.strip()):
if em in seen:
continue
seen.add(em)
u = db.query(User).filter(User.email == em).first()
out.append(((u.id if u else None), em, (u.username if u else em)))
if out:
return out
return get_default_recipients(db)
def configured_sources(db: Session) -> set:
return {s for s, key in CONFIG_KEYS.items() if (_setting(db, key) or "").strip()}
def _serialize(run: Optional[SyncRun]) -> Optional[dict]:
if run is None:
return None
try:
stats = json.loads(run.stats) if run.stats else {}
except Exception:
stats = {}
return {
"id": run.id,
"source": run.source,
"source_label": SOURCES.get(run.source, run.source),
"trigger": run.trigger,
"status": run.status,
"started_at": run.started_at,
"finished_at": run.finished_at,
"error_message": run.error_message,
"error_count": stats.get("error_count", 0) or 0,
"stats": stats,
"alerted_at": run.alerted_at,
}
def stale_thresholds(db: Session) -> Dict[str, Optional[timedelta]]:
"""How long without a success before a source counts as stale. None =
never (no enabled schedule drives it, so silence is not a failure)."""
from app.models.scan_schedule import ScanSchedule, ScheduleInterval
interval_of = {
ScheduleInterval.EVERY_HOUR: timedelta(hours=1),
ScheduleInterval.EVERY_6_HOURS: timedelta(hours=6),
ScheduleInterval.EVERY_12_HOURS: timedelta(hours=12),
ScheduleInterval.DAILY: timedelta(days=1),
ScheduleInterval.WEEKLY: timedelta(weeks=1),
}
out: Dict[str, Optional[timedelta]] = {s: STALE_AFTER for s in ("intune", "vcenter", "igel")}
out.update({"wazuh": None, "nessus": None})
try:
schedules = db.query(ScanSchedule).filter(ScanSchedule.enabled.is_(True)).all()
except Exception:
schedules = []
for sch in schedules:
source = (getattr(sch, "scanner_type", None) or "wazuh").lower()
if source not in ("wazuh", "nessus"):
continue
every = interval_of.get(sch.interval)
limit = max(STALE_AFTER, 2 * every) if every else CRON_STALE_AFTER
cur = out.get(source)
out[source] = limit if cur is None else min(cur, limit)
return out
def summarize_health(latest: Dict[str, SyncRun], last_success: Dict[str, SyncRun],
configured: Iterable[str],
thresholds: Optional[Dict[str, Optional[timedelta]]] = None,
now: Optional[datetime] = None) -> List[dict]:
"""Pure: one entry per known source with a state the GUI can colour.
failed the most recent run of this source failed
running a run is in progress
stale configured, but no successful run within its threshold
ok last run completed and is recent enough
unconfigured no config stored for this source
"""
now = now or datetime.now()
configured = set(configured)
thresholds = thresholds if thresholds is not None else {s: STALE_AFTER for s in SOURCES}
out = []
for source, label in SOURCES.items():
last = latest.get(source)
good = last_success.get(source)
good_at = good.finished_at if good else None
if source not in configured and last is None:
state = "unconfigured"
elif last is not None and last.status == "running":
state = "running"
elif last is not None and last.status == "failed":
state = "failed"
elif (limit := thresholds.get(source)) is not None and (good_at is None or now - good_at > limit):
state = "stale"
else:
state = "ok"
out.append({
"source": source,
"label": label,
"state": state,
"configured": source in configured,
"last_run": _serialize(last),
"last_success_at": good_at,
"error_message": last.error_message if last is not None else None,
})
return out
def sync_health(db: Session) -> List[dict]:
latest: Dict[str, SyncRun] = {}
last_success: Dict[str, SyncRun] = {}
for source in SOURCES:
latest[source] = (db.query(SyncRun).filter(SyncRun.source == source)
.order_by(SyncRun.started_at.desc()).first())
last_success[source] = (db.query(SyncRun)
.filter(SyncRun.source == source, SyncRun.status == "completed")
.order_by(SyncRun.finished_at.desc()).first())
return summarize_health(latest, last_success, configured_sources(db), stale_thresholds(db))
def health_payload(db: Session) -> dict:
"""What GET /scans/sync-health returns: per-source health plus the
failure-mail settings so the panel can edit them in place."""
return {
"sources": sync_health(db),
"alert_enabled": is_alert_enabled(db),
"recipients": [e for _, e, _ in get_recipients(db)],
"recipients_setting": _setting(db, SETTING_RECIPIENTS) or "",
}
def list_runs(db: Session, limit: int = 30) -> List[dict]:
rows = db.query(SyncRun).order_by(SyncRun.started_at.desc()).limit(limit).all()
return [_serialize(r) for r in rows]
# ---- failure mail ---------------------------------------------------------
def in_cooldown(last_alerted: Optional[SyncRun], now: Optional[datetime] = None) -> bool:
if last_alerted is None or last_alerted.alerted_at is None:
return False
return (now or datetime.now()) - last_alerted.alerted_at < ALERT_COOLDOWN
def _fmt(dt: Optional[datetime]) -> str:
return dt.strftime("%Y-%m-%d %H:%M") if dt else "never"
def notify_sync_failure(db: Session, run: SyncRun) -> dict:
"""Mail the failure to the sync-alert recipients, once per source per
cooldown window, and log it in notification_logs."""
from app.models.notification_log import (NotificationLog, NotificationStatus,
NotificationType)
from app.services.email_service import send_email
stats = {"sent": 0, "failed": 0, "skipped": None}
if not is_alert_enabled(db):
stats["skipped"] = "disabled"
return stats
with _alert_lock:
return _notify_locked(db, run, stats)
def _notify_locked(db: Session, run: SyncRun, stats: dict) -> dict:
from app.models.notification_log import (NotificationLog, NotificationStatus,
NotificationType)
from app.services.email_service import send_email
last_alerted = (db.query(SyncRun)
.filter(SyncRun.source == run.source, SyncRun.alerted_at.isnot(None))
.order_by(SyncRun.alerted_at.desc()).first())
if in_cooldown(last_alerted):
stats["skipped"] = "cooldown"
logger.info("sync failure mail for %s suppressed (cooldown until %s)",
run.source, last_alerted.alerted_at + ALERT_COOLDOWN)
return stats
recipients = get_recipients(db)
if not recipients:
stats["skipped"] = "no recipients"
return stats
good = (db.query(SyncRun)
.filter(SyncRun.source == run.source, SyncRun.status == "completed")
.order_by(SyncRun.finished_at.desc()).first())
label = SOURCES.get(run.source, run.source)
base = os.getenv("DASHBOARD_URL", "http://localhost:3000").rstrip("/")
subject = f"[TRUEVULN] Sync failed: {label} ({run.trigger}) — vulnerability coverage at risk"
gap = ("No successful sync of this source has ever been recorded."
if not (good and good.finished_at) else
f"Last successful sync: {_fmt(good.finished_at)} "
f"({(datetime.now() - good.finished_at).days} day(s) ago).")
body = f"""<!DOCTYPE html><html><body style="font-family:Arial,sans-serif;color:#111">
<h2 style="color:#b91c1c">Sync failed: {html.escape(label)}</h2>
<table cellpadding="6" style="border-collapse:collapse">
<tr><td><b>Source</b></td><td>{html.escape(label)}</td></tr>
<tr><td><b>Trigger</b></td><td>{html.escape(run.trigger)}</td></tr>
<tr><td><b>Started</b></td><td>{_fmt(run.started_at)}</td></tr>
<tr><td><b>Failed</b></td><td>{_fmt(run.finished_at)}</td></tr>
<tr><td><b>Error</b></td><td><code>{html.escape(run.error_message or "unknown")}</code></td></tr>
</table>
<p>{html.escape(gap)}</p>
<p><b>Until the next successful sync, new vulnerabilities and inventory changes
from this source are not detected.</b> Check the connection and credentials in
Settings, then run the sync again from Scan Jobs.</p>
<p><a href="{html.escape(base)}/scans">Open Scan Jobs</a></p>
<p style="color:#6b7280;font-size:12px">You receive at most one mail per source
per 24 hours while it keeps failing.</p>
</body></html>"""
for user_id, email, _name in recipients:
ok, err = send_email(db, email, subject, body)
db.add(NotificationLog(
user_id=user_id,
notification_type=NotificationType.SYNC_FAILURE,
sent_at=datetime.now(),
subject=subject[:500],
recipient_email=email,
status=NotificationStatus.SENT if ok else NotificationStatus.FAILED,
message_body=f"{label} sync ({run.trigger}) failed: {run.error_message}",
error_message=None if ok else err,
))
stats["sent" if ok else "failed"] += 1
# Only a mail somebody actually got starts the cooldown — a broken SMTP
# must not silence the alert for a day.
if stats["sent"]:
run.alerted_at = datetime.now()
db.commit()
logger.info("sync failure mail for %s: %d sent, %d failed",
run.source, stats["sent"], stats["failed"])
return stats
+2
View File
@@ -10,6 +10,7 @@ import {
} from '@heroicons/react/24/outline';
import Link from 'next/link';
import api from '../lib/api';
import { SyncHealthBanner } from '../components/shared/SyncHealth';
import { kevBadge } from '../lib/kevBadge';
import { DashboardStats, Vulnerability, AIPriorityResponse } from '../types';
import AIRecommendations from '@/components/AIRecommendations';
@@ -437,6 +438,7 @@ export default function Dashboard() {
// Full available width (minus AppShell padding) — the empty left/right
// gutters should be used on wide monitors. No max-width cap.
<div className="w-full">
<SyncHealthBanner />
{/* Header Section */}
<div className="md:flex md:items-center md:justify-between mb-8">
<div className="min-w-0 flex-1">
+18 -2
View File
@@ -4,6 +4,7 @@ import { Fragment, useEffect, useState } from 'react';
import api from '../../lib/api';
import { Asset, ScanRunSummary } from '../../types';
import { PlayIcon, ArrowPathIcon, ClockIcon, TrashIcon, ChevronDownIcon, ChevronRightIcon, PencilIcon } from '@heroicons/react/24/outline';
import { SyncHealthPanel } from '../../components/shared/SyncHealth';
interface ScanSchedule {
id: number;
@@ -36,6 +37,11 @@ export default function ScansPage() {
const [newSchedule, setNewSchedule] = useState({ name: '', interval: 'daily', cron_expression: '', scanner_type: 'wazuh' });
const [editingScheduleId, setEditingScheduleId] = useState<number | null>(null);
const [toast, setToast] = useState<{ message: string; type: 'success' | 'error' | 'info' } | null>(null);
// Bumped after every fetch/sync so the sync-health panel reloads with the rest.
const [healthKey, setHealthKey] = useState(0);
const [userRole, setUserRole] = useState('');
// The real reason a manual Wazuh sync failed — was "check the logs" before.
const [syncError, setSyncError] = useState<string | null>(null);
// Autoscan Modal State
const [isAutoscanModalOpen, setIsAutoscanModalOpen] = useState(false);
@@ -68,11 +74,13 @@ export default function ScansPage() {
console.error("Failed to fetch data:", error);
} finally {
setLoading(false);
setHealthKey((k) => k + 1);
}
};
useEffect(() => {
fetchData();
api.get('/auth/me').then((r) => setUserRole(r.data?.role || '')).catch(() => {});
}, []);
const handleAutoscan = async () => {
@@ -121,6 +129,7 @@ export default function ScansPage() {
setSyncStatus('running');
setSyncStep('assets');
setSyncResult(null);
setSyncError(null);
try {
// Sync assets first
@@ -139,8 +148,11 @@ export default function ScansPage() {
fetchData();
} catch (error: any) {
console.error("Sync failed:", error);
setToast({ message: error.response?.data?.detail || "Failed to sync with Wazuh.", type: 'error' });
const detail = error.response?.data?.detail || error.message || "Failed to sync with Wazuh.";
setToast({ message: detail, type: 'error' });
setSyncError(detail);
setSyncStatus('error');
fetchData(); // the failed run is in the ledger now — show it
}
};
@@ -301,6 +313,8 @@ export default function ScansPage() {
</div>
</div>
<SyncHealthPanel refreshKey={healthKey} canEdit={userRole === 'admin'} />
{/* Scan Runs Summary Table */}
<div className="bg-white border border-gray-200 shadow-sm rounded-sm overflow-hidden">
{scanRuns.length > 0 && (
@@ -734,7 +748,9 @@ export default function ScansPage() {
</svg>
</div>
<h4 className="text-base font-semibold leading-6 text-gray-900 font-mono">Sync Failed</h4>
<p className="text-sm text-gray-500 mt-2 font-mono">Check the logs or try again later.</p>
<p className="text-sm text-gray-500 mt-2 font-mono">Step {syncStep === 'assets' ? '1 (Importing Assets)' : '2 (Syncing Vulnerabilities)'} failed:</p>
<p className="text-sm text-red-700 mt-2 font-mono text-left bg-red-50 p-3 rounded-md break-words">{syncError || 'Unknown error'}</p>
<p className="text-xs text-gray-500 mt-2 font-mono">Recorded under Sync Runs. Check the Wazuh connection in Settings and try again.</p>
<div className="mt-6">
<button
type="button"
+229
View File
@@ -0,0 +1,229 @@
"use client";
import Link from 'next/link';
import { useEffect, useState } from 'react';
import api from '../../lib/api';
import { SyncHealthResponse, SyncHealthSource, SyncRun } from '../../types';
import { ExclamationTriangleIcon } from '@heroicons/react/24/outline';
const STATE_STYLE: Record<SyncHealthSource['state'], { badge: string; label: string }> = {
ok: { badge: 'bg-green-100 text-green-800', label: 'OK' },
failed: { badge: 'bg-red-100 text-red-800', label: 'FAILED' },
stale: { badge: 'bg-amber-100 text-amber-800', label: 'STALE' },
running: { badge: 'bg-blue-100 text-blue-800', label: 'RUNNING' },
unconfigured: { badge: 'bg-gray-100 text-gray-500', label: 'NOT CONFIGURED' },
};
const fmt = (iso: string | null | undefined) => (iso ? new Date(iso).toLocaleString() : '—');
const ago = (iso: string | null | undefined) => {
if (!iso) return 'never';
const h = Math.round((Date.now() - new Date(iso).getTime()) / 36e5);
if (h < 1) return 'less than an hour ago';
if (h < 48) return `${h}h ago`;
return `${Math.round(h / 24)}d ago`;
};
// Sources needing attention: the most recent run failed, or nothing has
// succeeded within the stale window. Shared by the banner and the panel.
const problems = (h: SyncHealthResponse | null) =>
(h?.sources || []).filter((s) => s.state === 'failed' || s.state === 'stale');
/** Dashboard strip: only rendered when a sync source is failed or stale. */
export function SyncHealthBanner() {
const [health, setHealth] = useState<SyncHealthResponse | null>(null);
useEffect(() => {
api.get('/api/v1/scans/sync-health').then((r) => setHealth(r.data)).catch(() => {});
}, []);
const bad = problems(health);
if (bad.length === 0) return null;
return (
<div className="mb-6 rounded-md border border-red-300 bg-red-50 p-4">
<div className="flex items-start gap-3">
<ExclamationTriangleIcon className="h-6 w-6 text-red-600 flex-shrink-0" />
<div className="min-w-0 flex-1">
<p className="text-sm font-bold text-red-800 font-mono">
Vulnerability coverage at risk: {bad.length} sync source{bad.length > 1 ? 's' : ''} not delivering data
</p>
<ul className="mt-1 text-sm text-red-700 font-mono space-y-0.5">
{bad.map((s) => (
<li key={s.source} className="truncate">
<b>{s.label}</b>: {s.state === 'failed'
? `last run failed ${ago(s.last_run?.finished_at)}${s.error_message || 'unknown error'}`
: `no successful sync since ${s.last_success_at ? fmt(s.last_success_at) : 'ever'}`}
{' · '}last success {ago(s.last_success_at)}
</li>
))}
</ul>
</div>
<Link href="/scans" className="text-sm font-semibold text-red-800 underline font-mono whitespace-nowrap">
Scan Jobs
</Link>
</div>
</div>
);
}
/** Scan Jobs panel: per-source health, the run ledger, and the failure-mail settings. */
export function SyncHealthPanel({ refreshKey, canEdit }: { refreshKey: number; canEdit: boolean }) {
const [health, setHealth] = useState<SyncHealthResponse | null>(null);
const [runs, setRuns] = useState<SyncRun[]>([]);
const [expanded, setExpanded] = useState<number | null>(null);
const [cfg, setCfg] = useState({ enabled: true, recipients: '' });
// An unsaved edit must survive the reloads other actions on the page trigger.
const [cfgDirty, setCfgDirty] = useState(false);
const [saving, setSaving] = useState(false);
const [saveMsg, setSaveMsg] = useState<string | null>(null);
const load = async () => {
try {
const [h, r] = await Promise.all([
api.get('/api/v1/scans/sync-health'),
api.get('/api/v1/scans/sync-runs', { params: { limit: 30 } }),
]);
setHealth(h.data);
setRuns(r.data || []);
if (!cfgDirty) setCfg({ enabled: h.data?.alert_enabled !== false, recipients: h.data?.recipients_setting || '' });
} catch (e) {
console.error('sync health load failed', e);
}
};
useEffect(() => { load(); }, [refreshKey]); // eslint-disable-line react-hooks/exhaustive-deps
const saveCfg = async () => {
setSaving(true);
setSaveMsg(null);
try {
await api.put('/api/v1/settings/sync_alert_enabled', { value: cfg.enabled ? 'true' : 'false' });
await api.put('/api/v1/settings/sync_alert_recipients', { value: cfg.recipients.trim() });
setSaveMsg('Saved.');
setCfgDirty(false);
load();
} catch (e) {
const detail = (e as { response?: { data?: { detail?: string } } })?.response?.data?.detail;
setSaveMsg(detail || 'Save failed (admin only).');
} finally {
setSaving(false);
}
};
const bad = problems(health);
return (
<div className="mb-6 space-y-4">
{/* Per-source health cards */}
<div className="bg-white border border-gray-200 shadow-sm rounded-sm">
<div className="px-6 py-3 bg-gray-50 border-b border-gray-200 flex items-center justify-between">
<span className="text-sm font-bold text-gray-900 font-mono">Sync Health</span>
{bad.length > 0 && (
<span className="text-xs font-mono text-red-700">{bad.length} source{bad.length > 1 ? 's' : ''} need attention</span>
)}
</div>
<div className="grid grid-cols-1 md:grid-cols-5 divide-y md:divide-y-0 md:divide-x divide-gray-200">
{(health?.sources || []).map((s) => {
const st = STATE_STYLE[s.state] || STATE_STYLE.unconfigured;
return (
<div key={s.source} className="p-4 font-mono text-xs space-y-1">
<div className="flex items-center justify-between">
<span className="font-bold text-sm text-gray-900">{s.label}</span>
<span className={`px-2 py-0.5 rounded-full text-[10px] font-bold ${st.badge}`}>{st.label}</span>
</div>
{s.state !== 'unconfigured' && (
<>
<div className="text-gray-500">Last run: {fmt(s.last_run?.started_at)}{s.last_run ? ` (${s.last_run.trigger})` : ''}</div>
<div className="text-gray-500">Last success: {s.last_success_at ? `${fmt(s.last_success_at)} · ${ago(s.last_success_at)}` : 'never'}</div>
{s.error_message && (
<div className="text-red-700 break-words" title={s.error_message}>{s.error_message}</div>
)}
</>
)}
</div>
);
})}
</div>
</div>
{/* Run ledger */}
<div className="bg-white border border-gray-200 shadow-sm rounded-sm overflow-hidden">
<div className="px-6 py-3 bg-gray-50 border-b border-gray-200">
<span className="text-sm font-bold text-gray-900 font-mono">Sync Runs</span>
<span className="ml-2 text-xs text-gray-500 font-mono">every source, manual and scheduled including runs that never reached an asset</span>
</div>
{runs.length === 0 ? (
<p className="p-6 text-sm text-gray-500 font-mono">No sync runs recorded yet.</p>
) : (
<div className="overflow-x-auto">
<table className="min-w-full divide-y divide-gray-200">
<thead className="bg-gray-50">
<tr>
{['Started', 'Source', 'Trigger', 'Status', 'Duration', 'Result'].map((h) => (
<th key={h} className="px-6 py-3 text-left text-xs font-mono font-medium text-gray-500 uppercase tracking-wider">{h}</th>
))}
</tr>
</thead>
<tbody className="bg-white divide-y divide-gray-200 font-mono text-sm">
{runs.map((r) => {
const dur = r.finished_at
? `${Math.max(0, Math.round((new Date(r.finished_at).getTime() - new Date(r.started_at).getTime()) / 1000))}s`
: '…';
const badge = STATE_STYLE[r.status === 'failed' ? 'failed' : r.status === 'running' ? 'running' : r.error_count > 0 ? 'stale' : 'ok'].badge;
const label = r.status === 'completed' && r.error_count > 0 ? `COMPLETED · ${r.error_count} ERR` : r.status.toUpperCase();
const isOpen = expanded === r.id;
const errs: string[] = Array.isArray(r.stats?.errors) ? r.stats.errors : [];
const summary = Object.entries(r.stats || {})
.filter(([k, v]) => k !== 'errors' && k !== 'error_count' && k !== 'phase' && (typeof v === 'number' || typeof v === 'string'))
.map(([k, v]) => `${k}=${v}`).join(' ');
return (
<tr key={r.id} className={`${r.status === 'failed' ? 'bg-red-50/40' : ''} ${errs.length || r.error_message ? 'cursor-pointer hover:bg-gray-50' : ''}`}
onClick={() => (errs.length || r.error_message) && setExpanded(isOpen ? null : r.id)}>
<td className="px-6 py-3 whitespace-nowrap text-gray-900">{fmt(r.started_at)}</td>
<td className="px-6 py-3 whitespace-nowrap text-gray-900">{r.source_label}{r.stats?.phase ? <span className="text-gray-400"> · {r.stats.phase}</span> : null}</td>
<td className="px-6 py-3 whitespace-nowrap text-gray-500">{r.trigger}</td>
<td className="px-6 py-3 whitespace-nowrap"><span className={`px-2 py-0.5 rounded-full text-[10px] font-bold ${badge}`}>{label}</span></td>
<td className="px-6 py-3 whitespace-nowrap text-gray-500">{dur}</td>
<td className="px-6 py-3 text-xs max-w-xl">
{r.error_message ? <span className="text-red-700">{r.error_message}</span> : <span className="text-gray-500">{summary}</span>}
{isOpen && errs.length > 0 && (
<ul className="mt-2 space-y-0.5 text-red-700 border-l-2 border-red-200 pl-2">
{errs.map((e, i) => <li key={i}>{e}</li>)}
</ul>
)}
</td>
</tr>
);
})}
</tbody>
</table>
</div>
)}
</div>
{/* Failure-mail settings */}
<div className="bg-white border border-gray-200 shadow-sm rounded-sm p-4 font-mono text-sm">
<div className="flex flex-wrap items-center gap-4">
<label className="flex items-center gap-2">
<input type="checkbox" checked={cfg.enabled} disabled={!canEdit}
onChange={(e) => { setCfg({ ...cfg, enabled: e.target.checked }); setCfgDirty(true); }}
className="h-4 w-4 rounded border-gray-300 text-truevuln-blue" />
<span className="font-bold text-gray-900">E-mail on sync failure</span>
</label>
<input type="text" value={cfg.recipients} disabled={!canEdit}
onChange={(e) => { setCfg({ ...cfg, recipients: e.target.value }); setCfgDirty(true); }}
placeholder={`recipients (empty = ${(health?.recipients || []).join(', ') || 'notification defaults / admins'})`}
className="flex-1 min-w-[16rem] rounded border border-gray-300 px-2 py-1 text-xs" />
{canEdit && (
<button onClick={saveCfg} disabled={saving}
className="rounded-sm bg-truevuln-blue px-3 py-1.5 text-xs font-semibold text-white hover:bg-blue-600 disabled:opacity-50">
{saving ? 'Saving…' : 'Save'}
</button>
)}
{saveMsg && <span className="text-xs text-gray-500">{saveMsg}</span>}
</div>
<p className="mt-2 text-xs text-gray-500">
One mail per source per 24h while it keeps failing, stating the error and the last successful sync. Needs SMTP in Settings.
</p>
</div>
</div>
);
}
+32
View File
@@ -230,3 +230,35 @@ export interface AIPriorityResponse {
global_strategy: string;
error?: string;
}
// --- Sync run ledger (GET /api/v1/scans/sync-runs, /sync-health) ---
export interface SyncRun {
id: number;
source: string;
source_label: string;
trigger: 'manual' | 'scheduled' | string;
status: 'running' | 'completed' | 'failed' | string;
started_at: string;
finished_at: string | null;
error_message: string | null;
error_count: number;
stats: Record<string, any>;
alerted_at: string | null;
}
export interface SyncHealthSource {
source: string;
label: string;
state: 'ok' | 'failed' | 'stale' | 'running' | 'unconfigured';
configured: boolean;
last_run: SyncRun | null;
last_success_at: string | null;
error_message: string | null;
}
export interface SyncHealthResponse {
sources: SyncHealthSource[];
alert_enabled: boolean;
recipients: string[];
recipients_setting: string;
}
+349
View File
@@ -0,0 +1,349 @@
"""Sync run ledger + failure alert — run: python -m pytest tests/test_sync_runs.py
A Wazuh API that answered every login with 500 for a day was visible only in
the container log: the scan-jobs page is per asset, and a sync that fails at
authentication never reaches an asset. So it showed a clean history of
COMPLETED runs while nothing had been synced, and nobody was told.
The ledger records every sync execution with its outcome, whatever the source
and whether a button or the scheduler started it; a failure sends ONE mail
per source per cooldown window, and a nested recorder (the scheduled Wazuh
scan calling the asset sync) merges into the outer run instead of writing a
second row.
"""
import os
import sys
from datetime import datetime, timedelta
import pytest
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from app.models.sync_run import SyncRun # noqa: E402
from app.services import sync_run_service as svc # noqa: E402
class _Q:
def __init__(self, rows):
self._rows = rows
def filter(self, *a, **k):
return self
def order_by(self, *a, **k):
return self
def limit(self, *a, **k):
return self
def first(self):
return self._rows[0] if self._rows else None
def all(self):
return list(self._rows)
class _DB:
"""Enough Session for the ledger: add() assigns ids, query(SyncRun)
returns what was added, newest first."""
def __init__(self):
self.rows = []
self.commits = 0
self.closed = False
def add(self, obj):
if isinstance(obj, SyncRun) and obj.id is None:
obj.id = len(self.rows) + 1
self.rows.append(obj)
def commit(self):
self.commits += 1
def rollback(self):
pass
def close(self):
self.closed = True
def query(self, model):
return _Q([r for r in reversed(self.rows) if isinstance(r, model)])
@pytest.fixture
def db(monkeypatch):
fake = _DB()
monkeypatch.setattr(svc, "SessionLocal", lambda: fake)
monkeypatch.setattr(svc, "notify_sync_failure", lambda db, run: None)
return fake
def _runs(db):
return [r for r in db.rows if isinstance(r, SyncRun)]
def test_success_is_recorded_with_stats(db):
with svc.record_sync_run("wazuh", "manual") as run:
run.stats.update({"agents_synced": 3})
(row,) = _runs(db)
assert row.status == "completed"
assert row.finished_at is not None
assert row.error_message is None
assert '"agents_synced": 3' in row.stats
assert db.closed
def test_exception_marks_run_failed_and_reraises(db):
with pytest.raises(RuntimeError):
with svc.record_sync_run("wazuh", "scheduled"):
raise RuntimeError("Authentication failed: Server error '500'")
(row,) = _runs(db)
assert row.status == "failed"
assert "Authentication failed" in row.error_message
assert row.finished_at is not None
def test_http_exception_detail_is_the_error(db):
from fastapi import HTTPException
with pytest.raises(HTTPException):
with svc.record_sync_run("wazuh"):
raise HTTPException(status_code=500, detail="Wazuh sync failed: 500 daemons not ready")
assert _runs(db)[0].error_message == "Wazuh sync failed: 500 daemons not ready"
def test_explicit_fail_without_exception(db):
with svc.record_sync_run("wazuh", "scheduled") as run:
run.fail("all 14 agents failed: Authentication failed")
assert _runs(db)[0].status == "failed"
def test_failure_triggers_notification(db, monkeypatch):
seen = []
monkeypatch.setattr(svc, "notify_sync_failure", lambda db, run: seen.append(run.source))
with pytest.raises(ValueError):
with svc.record_sync_run("igel"):
raise ValueError("boom")
assert seen == ["igel"]
def test_notification_crash_never_breaks_the_job(db, monkeypatch):
def _explode(db, run):
raise RuntimeError("smtp down")
monkeypatch.setattr(svc, "notify_sync_failure", _explode)
with pytest.raises(ValueError): # the job's own error, not smtp's
with svc.record_sync_run("igel"):
raise ValueError("boom")
assert _runs(db)[0].status == "failed"
def test_nested_recorder_merges_into_outer_run(db):
with svc.record_sync_run("wazuh", "scheduled") as outer:
with svc.record_sync_run("wazuh", "manual") as inner:
inner.stats["created"] = 2
assert inner is outer
rows = _runs(db)
assert len(rows) == 1
assert rows[0].trigger == "scheduled"
assert '"created": 2' in rows[0].stats
def test_nested_exception_does_not_finish_outer_early(db):
with svc.record_sync_run("wazuh", "scheduled") as outer:
try:
with svc.record_sync_run("wazuh"):
raise RuntimeError("asset sync failed")
except RuntimeError as e:
outer.stats.setdefault("errors", []).append(str(e))
outer.stats["agents_synced"] = 5
(row,) = _runs(db)
assert row.status == "completed"
assert '"agents_synced": 5' in row.stats
def test_stale_running_rows_are_closed_on_startup(db):
db.add(SyncRun(source="nessus", trigger="scheduled", status="running",
started_at=datetime.now() - timedelta(hours=3)))
svc.mark_interrupted_runs()
row = _runs(db)[0]
assert row.status == "failed"
assert "interrupted" in row.error_message
# ---- alert cooldown ------------------------------------------------------
def test_cooldown_suppresses_repeat_mail_for_same_source():
now = datetime.now()
recent = SyncRun(source="wazuh", status="failed", started_at=now,
alerted_at=now - timedelta(hours=2))
old = SyncRun(source="wazuh", status="failed", started_at=now,
alerted_at=now - timedelta(hours=30))
assert svc.in_cooldown(recent, now) is True
assert svc.in_cooldown(old, now) is False
assert svc.in_cooldown(None, now) is False
def test_health_flags_failed_and_stale_sources():
now = datetime.now()
ok = SyncRun(source="wazuh", status="completed", started_at=now,
finished_at=now - timedelta(hours=1))
failed = SyncRun(source="igel", status="failed", started_at=now,
finished_at=now, error_message="UMS login refused")
stale = SyncRun(source="nessus", status="completed", started_at=now,
finished_at=now - timedelta(days=3))
out = svc.summarize_health(
latest={"wazuh": ok, "igel": failed, "nessus": stale},
last_success={"wazuh": ok, "nessus": stale},
configured={"wazuh", "igel", "nessus"},
now=now,
)
by = {h["source"]: h for h in out}
assert by["wazuh"]["state"] == "ok"
assert by["igel"]["state"] == "failed"
assert by["igel"]["error_message"] == "UMS login refused"
assert by["igel"]["last_success_at"] is None
assert by["nessus"]["state"] == "stale"
assert by["intune"]["state"] == "unconfigured"
# ---- against real SQL (SQLite in memory): the queries, the mail, the log ---
@pytest.fixture
def sql(monkeypatch):
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool
import app.models # noqa: F401 — registers every table on Base
from app.models.base import Base
engine = create_engine("sqlite://", connect_args={"check_same_thread": False},
poolclass=StaticPool)
Base.metadata.create_all(engine)
Local = sessionmaker(bind=engine)
monkeypatch.setattr(svc, "SessionLocal", Local)
yield Local
engine.dispose()
def test_failed_run_mails_once_and_logs_it(sql, monkeypatch):
from app.models.notification_log import NotificationLog, NotificationType
from app.models.setting import Setting
from app.services import email_service
sent = []
monkeypatch.setattr(email_service, "send_email",
lambda db, to, subject, body: (sent.append((to, subject, body)) or (True, "OK")))
s = sql()
s.add(Setting(key=svc.SETTING_RECIPIENTS, value="ops@example.com"))
s.add(Setting(key="wazuh_config", value='{"api_url": "https://w"}'))
s.commit()
s.close()
for _ in range(2): # hourly schedule, API still down
with pytest.raises(RuntimeError):
with svc.record_sync_run("wazuh", "scheduled"):
raise RuntimeError("Authentication failed: Server error '500'")
assert len(sent) == 1 # second failure is inside the cooldown
to, subject, body = sent[0]
assert to == "ops@example.com"
assert "Wazuh" in subject and "scheduled" in subject
assert "Authentication failed" in body
assert "No successful sync" in body # the coverage gap, spelled out
s = sql()
logs = s.query(NotificationLog).all()
assert len(logs) == 1
assert logs[0].notification_type == NotificationType.SYNC_FAILURE
assert logs[0].recipient_email == "ops@example.com"
runs = svc.list_runs(s, 10)
assert [r["status"] for r in runs] == ["failed", "failed"]
assert runs[1]["alerted_at"] is not None and runs[0]["alerted_at"] is None
health = {h["source"]: h for h in svc.sync_health(s)}
assert health["wazuh"]["state"] == "failed"
assert health["wazuh"]["configured"] is True
assert "Authentication failed" in health["wazuh"]["error_message"]
assert health["nessus"]["state"] == "unconfigured"
s.close()
def test_success_after_failure_shows_ok(sql, monkeypatch):
monkeypatch.setattr(svc, "notify_sync_failure", lambda db, run: None)
s = sql()
s.add(__import__("app.models.setting", fromlist=["Setting"]).Setting(
key="igel_config", value='{"host": "ums"}'))
s.commit()
with pytest.raises(RuntimeError):
with svc.record_sync_run("igel"):
raise RuntimeError("UMS down")
with svc.record_sync_run("igel", "scheduled") as run:
run.stats.update({"devices": 12, "errors": []})
health = {h["source"]: h for h in svc.sync_health(s)}
assert health["igel"]["state"] == "ok"
assert health["igel"]["last_success_at"] is not None
assert health["igel"]["last_run"]["stats"]["devices"] == 12
s.close()
# ---- review follow-ups -----------------------------------------------------
def test_nested_different_source_gets_its_own_row(db):
with svc.record_sync_run("wazuh", "scheduled") as outer:
with svc.record_sync_run("igel", "manual") as inner:
assert inner is not outer
with svc.record_sync_run("wazuh") as again: # outer restored after inner
assert again is outer
rows = _runs(db)
assert sorted(r.source for r in rows) == ["igel", "wazuh"]
def test_stale_threshold_follows_the_schedule():
from app.models.scan_schedule import ScanSchedule, ScheduleInterval
class _SchedDB:
def __init__(self, rows):
self._rows = rows
def query(self, model):
return _Q(self._rows)
weekly = ScanSchedule(name="w", interval=ScheduleInterval.WEEKLY, scanner_type="nessus", enabled=True)
hourly = ScanSchedule(name="h", interval=ScheduleInterval.EVERY_HOUR, scanner_type="wazuh", enabled=True)
t = svc.stale_thresholds(_SchedDB([weekly, hourly]))
assert t["nessus"] == timedelta(weeks=2) # twice the interval
assert t["wazuh"] == svc.STALE_AFTER # never below the floor
assert t["igel"] == svc.STALE_AFTER # nightly job
assert svc.stale_thresholds(_SchedDB([]))["nessus"] is None # manual-only: never stale
now = datetime.now()
six_days = SyncRun(source="nessus", status="completed", started_at=now,
finished_at=now - timedelta(days=6))
by = {h["source"]: h for h in svc.summarize_health(
{"nessus": six_days}, {"nessus": six_days}, {"nessus"},
thresholds={"nessus": timedelta(weeks=2)}, now=now)}
assert by["nessus"]["state"] == "ok" # a weekly Nessus is not stale after 6 days
by = {h["source"]: h for h in svc.summarize_health(
{"nessus": six_days}, {"nessus": six_days}, {"nessus"},
thresholds={"nessus": None}, now=now)}
assert by["nessus"]["state"] == "ok"
def test_every_agent_failed_is_a_failed_run(db):
"""Authentication refused → each agent raises, 0 triggered, N errors.
That is the failure the user could not see anywhere. One agent through
is a completed run with errors the per-asset rows carry those."""
errs = ["srv01: Authentication failed: Server error '500'",
"srv02: Authentication failed: Server error '500'"]
with svc.record_sync_run("wazuh", "scheduled") as run:
svc.fail_if_nothing_synced(run, 0, errs)
with svc.record_sync_run("wazuh", "manual") as run:
svc.fail_if_nothing_synced(run, 1, errs)
with svc.record_sync_run("wazuh", "manual") as run:
svc.fail_if_nothing_synced(run, 0, []) # no assets at all: not a failure
dead, partial, empty = _runs(db)
assert dead.status == "failed"
assert "all 2 agents failed" in dead.error_message
assert "Authentication failed" in dead.error_message
assert partial.status == "completed"
assert empty.status == "completed"