diff --git a/alembic/versions/050_sync_runs.py b/alembic/versions/050_sync_runs.py new file mode 100644 index 0000000..b1cb89b --- /dev/null +++ b/alembic/versions/050_sync_runs.py @@ -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. diff --git a/app/models/__init__.py b/app/models/__init__.py index 0583e7c..9956d13 100644 --- a/app/models/__init__.py +++ b/app/models/__init__.py @@ -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", diff --git a/app/models/notification_log.py b/app/models/notification_log.py index 17dd227..4f6f7e6 100644 --- a/app/models/notification_log.py +++ b/app/models/notification_log.py @@ -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" diff --git a/app/models/sync_run.py b/app/models/sync_run.py new file mode 100644 index 0000000..343dcaf --- /dev/null +++ b/app/models/sync_run.py @@ -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"" diff --git a/app/routers/assets.py b/app/routers/assets.py index 289babf..79c7c38 100644 --- a/app/routers/assets.py +++ b/app/routers/assets.py @@ -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 diff --git a/app/routers/igel.py b/app/routers/igel.py index ef280a6..42a586a 100644 --- a/app/routers/igel.py +++ b/app/routers/igel.py @@ -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 diff --git a/app/routers/intune.py b/app/routers/intune.py index 3778c72..b04bae5 100644 --- a/app/routers/intune.py +++ b/app/routers/intune.py @@ -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 diff --git a/app/routers/nessus.py b/app/routers/nessus.py index 640685b..5b3c525 100644 --- a/app/routers/nessus.py +++ b/app/routers/nessus.py @@ -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 diff --git a/app/routers/scans.py b/app/routers/scans.py index 1a12c83..d636bf9 100644 --- a/app/routers/scans.py +++ b/app/routers/scans.py @@ -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) diff --git a/app/routers/vcenter.py b/app/routers/vcenter.py index 89f416e..df7a78e 100644 --- a/app/routers/vcenter.py +++ b/app/routers/vcenter.py @@ -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 diff --git a/app/routers/vulnerabilities.py b/app/routers/vulnerabilities.py index af59a21..633ed3c 100644 --- a/app/routers/vulnerabilities.py +++ b/app/routers/vulnerabilities.py @@ -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): diff --git a/app/scheduler.py b/app/scheduler.py index 33f4594..167611a 100644 --- a/app/scheduler.py +++ b/app/scheduler.py @@ -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)" diff --git a/app/services/sync_run_service.py b/app/services/sync_run_service.py new file mode 100644 index 0000000..a056bd9 --- /dev/null +++ b/app/services/sync_run_service.py @@ -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""" +

Sync failed: {html.escape(label)}

+ + + + + + +
Source{html.escape(label)}
Trigger{html.escape(run.trigger)}
Started{_fmt(run.started_at)}
Failed{_fmt(run.finished_at)}
Error{html.escape(run.error_message or "unknown")}
+

{html.escape(gap)}

+

Until the next successful sync, new vulnerabilities and inventory changes +from this source are not detected. Check the connection and credentials in +Settings, then run the sync again from Scan Jobs.

+

Open Scan Jobs

+

You receive at most one mail per source +per 24 hours while it keeps failing.

+""" + + 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 diff --git a/frontend/app/page.tsx b/frontend/app/page.tsx index 7317977..39626fa 100644 --- a/frontend/app/page.tsx +++ b/frontend/app/page.tsx @@ -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.
+ {/* Header Section */}
diff --git a/frontend/app/scans/page.tsx b/frontend/app/scans/page.tsx index f3987e3..2cd3176 100644 --- a/frontend/app/scans/page.tsx +++ b/frontend/app/scans/page.tsx @@ -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(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(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() {
+ + {/* Scan Runs Summary Table */}
{scanRuns.length > 0 && ( @@ -734,7 +748,9 @@ export default function ScansPage() {

Sync Failed

-

Check the logs or try again later.

+

Step {syncStep === 'assets' ? '1 (Importing Assets)' : '2 (Syncing Vulnerabilities)'} failed:

+

{syncError || 'Unknown error'}

+

Recorded under Sync Runs. Check the Wazuh connection in Settings and try again.

+ )} + {saveMsg && {saveMsg}} +
+

+ One mail per source per 24h while it keeps failing, stating the error and the last successful sync. Needs SMTP in Settings. +

+
+ + ); +} diff --git a/frontend/types/index.ts b/frontend/types/index.ts index 7d0e8af..05f11ee 100644 --- a/frontend/types/index.ts +++ b/frontend/types/index.ts @@ -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; + 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; +} diff --git a/tests/test_sync_runs.py b/tests/test_sync_runs.py new file mode 100644 index 0000000..9cdb1de --- /dev/null +++ b/tests/test_sync_runs.py @@ -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"