vulncheck
236c00ada7
Groups, policies, scan schedules, finding assignment and suppression, and
every sync, scan, refresh and connection-test button wrote no audit row,
so "who did this, and what changed" had no answer for them.
Records that change now write one row naming each field old -> new
(audit_trail.record_change; a create passes before={}, a delete after={},
nothing changed writes nothing), in the same transaction as the change.
Defer, undefer and false-positive also record the date, reason and
suppression they set or clear; an admin's email change is logged.
Buttons take the route dependency audited("What"): one row once the
endpoint returned, before the response goes out, so a refused or raising
call writes none and a failing audit write is a 500 rather than a gap. It
keeps the query parameters and only the named body fields, because request
bodies carry passwords: the test-mail recipient and SMTP host, the syslog
target, the AI provider and base URL, the Nessus asset and scan.
A guard test fails for any new POST/PUT/PATCH/DELETE route that does not
audit itself. No new event types, so no migration: groups, policies and
schedules log as CONFIG_CHANGE, buttons as SCAN_TRIGGERED, told apart by
resource_type and the description.
1519 lines
64 KiB
Python
1519 lines
64 KiB
Python
"""
|
|
Asset/Inventory Management Router
|
|
|
|
Endpoints for asset management (Wazuh Agents + manual systems).
|
|
"""
|
|
from typing import Optional, List, Any
|
|
from datetime import datetime
|
|
from fastapi import APIRouter, Depends, HTTPException, status, Query, Response
|
|
from fastapi.responses import StreamingResponse
|
|
from sqlalchemy.orm import Session, selectinload
|
|
from sqlalchemy import asc, desc, func, nulls_last
|
|
from pydantic import BaseModel, field_validator
|
|
|
|
from app.database import get_db
|
|
from app.models.user import User
|
|
from app.models.asset import Asset, AssetSource, AssetStatus, ContainerPlacement
|
|
from app.models.policy import Policy
|
|
from app.models.vulnerability import Vulnerability, VulnerabilityStatus
|
|
from app.models.audit_log import AuditLog, AuditEventType
|
|
from app.auth.dependencies import get_current_user, RequireEditor, RequireAdmin, audited
|
|
from app.services.audit_trail import record_change
|
|
import csv
|
|
import io
|
|
import logging
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
router = APIRouter(prefix="/api/v1/assets", tags=["Assets"])
|
|
|
|
|
|
# ============================================
|
|
# Pydantic Schemas
|
|
# ============================================
|
|
|
|
class AssetResponse(BaseModel):
|
|
id: int
|
|
hostname: str
|
|
ip_address: Optional[str]
|
|
wazuh_agent_id: Optional[str]
|
|
nessus_host_uuid: Optional[str] = None
|
|
# Intune/Defender ids — surfaced so the UI can offer the software/app-scan
|
|
# actions on MDM-managed (iOS/macOS/Android) assets, not just Wazuh hosts.
|
|
intune_device_id: Optional[str] = None
|
|
defender_machine_id: Optional[str] = None
|
|
# Entra/AAD device id — the one identifier that survives a device being
|
|
# deleted in Intune while Defender TVM keeps reporting it under another
|
|
# name. Read-only in the UI; it is what an operator needs to exclude the
|
|
# right machine in Defender (field report 2026-08-30).
|
|
aad_device_id: Optional[str] = None
|
|
# The remaining per-source identity pins, read-only in the edit dialog so
|
|
# an operator can see what a sync reconnects the asset by (and the serial
|
|
# to find the hardware with).
|
|
vmware_uuid: Optional[str] = None
|
|
vmware_serial: Optional[str] = None
|
|
igel_unit_id: Optional[str] = None
|
|
igel_serial: Optional[str] = None
|
|
netdisco_device_ip: Optional[str] = None
|
|
netdisco_serial: Optional[str] = None
|
|
xen_host_uuid: Optional[str] = None
|
|
intune_serial: Optional[str] = None
|
|
operating_system: Optional[str]
|
|
os_version: Optional[str]
|
|
# VMware VMkernel/vpxd build number — the precise fix identifier for
|
|
# ESXi/vCenter (e.g. "24784735"), surfaced in the OS column.
|
|
vmware_build: Optional[str] = None
|
|
source: AssetSource
|
|
status: AssetStatus
|
|
last_scan: Optional[datetime]
|
|
last_seen: Optional[datetime]
|
|
# Which sync wrote those stamps — the same attribution the reactivation
|
|
# audit entry quotes.
|
|
last_seen_source: Optional[str] = None
|
|
vulnerability_count: Optional[int] = 0
|
|
policy_id: Optional[int] = None
|
|
description: Optional[str] = None
|
|
location: Optional[str] = None
|
|
owner: Optional[str] = None
|
|
assigned_user_id: Optional[int] = None
|
|
assigned_user_name: Optional[str] = None
|
|
# assigned_group_id obsolete, replaced by groups list
|
|
groups: List[str] = [] # List of Team Names for display
|
|
policy_name: Optional[str] = None
|
|
criticality: Optional[str] = None
|
|
# Network-exposure risk dimension
|
|
network_exposure_score: Optional[float] = None
|
|
exposed_services: Optional[List[dict]] = None
|
|
exposure_updated_at: Optional[datetime] = None
|
|
# Risk Dimensions (crown-jewel roles)
|
|
high_value_score: Optional[float] = None
|
|
risk_dimensions: Optional[List[dict]] = None
|
|
# Docker host ↔ image links from the Dockhand sync (container_placements):
|
|
# a host's number of images, an image's hosts as {id, hostname}.
|
|
container_image_count: int = 0
|
|
runs_on: List[dict] = []
|
|
|
|
@field_validator('exposed_services', 'risk_dimensions', mode='before')
|
|
@classmethod
|
|
def parse_json_list(cls, v: Any) -> Optional[List[dict]]:
|
|
if v is None or isinstance(v, list):
|
|
return v
|
|
if isinstance(v, str):
|
|
try:
|
|
import json as _j
|
|
parsed = _j.loads(v)
|
|
return parsed if isinstance(parsed, list) else None
|
|
except Exception:
|
|
return None
|
|
return None
|
|
|
|
@field_validator('groups', mode='before')
|
|
@classmethod
|
|
def map_groups_to_names(cls, v: Any) -> List[str]:
|
|
if not v:
|
|
return []
|
|
# Wenn es bereits Strings sind (z.B. manuell gesetzt)
|
|
if isinstance(v, list) and all(isinstance(x, str) for x in v):
|
|
return v
|
|
# Wenn es SQLAlchemy-Objekte sind
|
|
return [g.name for g in v if hasattr(g, 'name')]
|
|
|
|
class Config:
|
|
from_attributes = True
|
|
|
|
|
|
class AssetAssignRequest(BaseModel):
|
|
user_id: Optional[int] = None
|
|
group_id: Optional[int] = None
|
|
|
|
|
|
class AssetCreateRequest(BaseModel):
|
|
hostname: str
|
|
ip_address: Optional[str] = None
|
|
operating_system: Optional[str] = None
|
|
os_version: Optional[str] = None
|
|
description: Optional[str] = None
|
|
location: Optional[str] = None
|
|
owner: Optional[str] = None
|
|
policy_id: Optional[int] = None
|
|
|
|
|
|
class AssetUpdateRequest(BaseModel):
|
|
hostname: Optional[str] = None
|
|
ip_address: Optional[str] = None
|
|
operating_system: Optional[str] = None
|
|
os_version: Optional[str] = None
|
|
status: Optional[AssetStatus] = None
|
|
description: Optional[str] = None
|
|
location: Optional[str] = None
|
|
owner: Optional[str] = None
|
|
policy_id: Optional[int] = None
|
|
# Plan E — URS inputs (criticality multiplier + framework list)
|
|
criticality: Optional[str] = None # low | normal | high | critical
|
|
compliance_frameworks: Optional[List[str]] = None
|
|
|
|
|
|
class BulkAssetUpdateRequest(BaseModel):
|
|
asset_ids: List[int]
|
|
assigned_user_id: Optional[int] = None
|
|
assigned_group_id: Optional[int] = None
|
|
policy_id: Optional[int] = None
|
|
|
|
|
|
def _asset_state(asset: Asset, fields) -> dict:
|
|
"""The audited values of `fields`; group membership by name."""
|
|
return {f: sorted(g.name for g in asset.groups) if f == "groups" else getattr(asset, f)
|
|
for f in fields}
|
|
|
|
|
|
def _audit_asset_update(db: Session, user: User, asset: Asset, before: dict) -> None:
|
|
"""One ASSET_UPDATED row naming every changed field, old → new. Nothing
|
|
changed, no row."""
|
|
record_change(db, user.id, AuditEventType.ASSET_UPDATED, "asset", asset.id,
|
|
f"Asset updated: {asset.hostname}", before, _asset_state(asset, before))
|
|
|
|
|
|
def _assignment(asset: Asset) -> dict:
|
|
return {"assigned_user": asset.assigned_user.username if asset.assigned_user else None,
|
|
"groups": sorted(g.name for g in asset.groups)}
|
|
|
|
|
|
# ============================================
|
|
# Endpoints
|
|
# ============================================
|
|
|
|
@router.post("/bulk-update")
|
|
def bulk_update_assets(
|
|
update_data: BulkAssetUpdateRequest,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(RequireEditor)
|
|
):
|
|
"""Updates multiple assets at once (assignment, policy)"""
|
|
assets = db.query(Asset).filter(Asset.id.in_(update_data.asset_ids)).all()
|
|
|
|
if not assets:
|
|
raise HTTPException(status_code=404, detail="No assets found")
|
|
|
|
update_fields = {}
|
|
if update_data.assigned_user_id is not None:
|
|
update_fields["assigned_user_id"] = update_data.assigned_user_id if update_data.assigned_user_id != -1 else None
|
|
if update_data.assigned_group_id is not None:
|
|
update_fields["assigned_group_id"] = update_data.assigned_group_id if update_data.assigned_group_id != -1 else None
|
|
if update_data.policy_id is not None:
|
|
update_fields["policy_id"] = update_data.policy_id if update_data.policy_id != -1 else None
|
|
|
|
if not update_fields:
|
|
raise HTTPException(status_code=400, detail="No update fields specified")
|
|
|
|
count = 0
|
|
from app.models.group import Group
|
|
|
|
audited = ["groups" if f == "assigned_group_id" else f for f in update_fields]
|
|
for asset in assets:
|
|
before = _asset_state(asset, audited)
|
|
for field, value in update_fields.items():
|
|
if field == "assigned_group_id":
|
|
# Special handling for legacy group assignment
|
|
if value is None:
|
|
asset.groups = []
|
|
else:
|
|
group = db.query(Group).filter(Group.id == value).first()
|
|
if group:
|
|
asset.groups = [group]
|
|
else:
|
|
setattr(asset, field, value)
|
|
_audit_asset_update(db, current_user, asset, before)
|
|
count += 1
|
|
|
|
db.commit()
|
|
|
|
# Summary row for the overview; the per-asset rows above carry the diff.
|
|
audit_log = AuditLog(
|
|
user_id=current_user.id,
|
|
event_type=AuditEventType.ASSET_UPDATED,
|
|
event_description=f"Bulk update performed for {count} assets",
|
|
resource_type="asset",
|
|
resource_id="multiple",
|
|
timestamp=datetime.now()
|
|
)
|
|
db.add(audit_log)
|
|
db.commit()
|
|
|
|
return {"message": f"{count} assets successfully updated", "count": count}
|
|
|
|
|
|
@router.post("/reconcile-lifecycle", dependencies=[audited("Asset lifecycle reconcile")])
|
|
def reconcile_lifecycle(
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(RequireEditor),
|
|
):
|
|
"""Manually run the soft-inactive reconcile (also runs 04:15 nightly).
|
|
|
|
Flips assets no source reported within asset_inactive_after_days to
|
|
INACTIVE; revives recently-seen INACTIVE ones. History + audit kept.
|
|
"""
|
|
from app.services.asset_lifecycle import reconcile_asset_lifecycle
|
|
return reconcile_asset_lifecycle(db)
|
|
|
|
|
|
@router.get("/{asset_id}/coverage-gap")
|
|
def asset_coverage_gap(
|
|
asset_id: int,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""Coverage-gap report: installed packages with NO vuln finding.
|
|
|
|
Field report: Wazuh's vuln-detector misses some products (MS365,
|
|
mRemoteNG, …). This surfaces the GAP — packages present in
|
|
syscollector but not matched to any open finding — so the operator
|
|
can investigate manually. Deliberately makes NO automatic CVE
|
|
assignment (zero false positives). Annotates each gap package with
|
|
its endoflife.date status when known.
|
|
"""
|
|
import json as _json
|
|
from app.auth.setting_crypto import read_setting_value
|
|
from app.integrations.wazuh_client import WazuhClient
|
|
from app.services import eol_service
|
|
|
|
asset = db.query(Asset).filter(Asset.id == asset_id).first()
|
|
if not asset:
|
|
raise HTTPException(404, "Asset not found")
|
|
if not asset.wazuh_agent_id:
|
|
raise HTTPException(422, "Asset has no Wazuh agent — no syscollector inventory")
|
|
|
|
raw = read_setting_value(db, "wazuh_config")
|
|
if not raw:
|
|
raise HTTPException(400, "Wazuh is not configured")
|
|
cfg = _json.loads(raw)
|
|
wazuh = WazuhClient(
|
|
base_url=cfg.get("api_url"), username=cfg.get("username"),
|
|
password=cfg.get("password"), indexer_url=cfg.get("indexer_url"),
|
|
indexer_username=cfg.get("indexer_username"),
|
|
indexer_password=cfg.get("indexer_password"),
|
|
verify_ssl=bool(cfg.get("verify_ssl", True)),
|
|
)
|
|
try:
|
|
packages = wazuh.get_packages(asset.wazuh_agent_id) or []
|
|
except Exception as e:
|
|
raise HTTPException(502, f"Could not fetch packages from Wazuh: {e}")
|
|
|
|
# Build the "covered" set: package names that already have an open
|
|
# finding on this asset (parent package_name + child rows).
|
|
covered_blob = " ".join(
|
|
(v.package_name or "").lower()
|
|
for v in db.query(Vulnerability)
|
|
.filter(Vulnerability.asset_id == asset_id,
|
|
Vulnerability.status == VulnerabilityStatus.open)
|
|
.all()
|
|
)
|
|
|
|
seen: set = set()
|
|
gap: list[dict] = []
|
|
covered_count = 0
|
|
for pkg in packages:
|
|
name = (pkg.get("name") or "").strip()
|
|
version = (pkg.get("version") or "").strip()
|
|
if not name:
|
|
continue
|
|
key = (name.lower(), version)
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
# Covered if the package name (token) appears in any finding's
|
|
# package_name blob. Token match avoids substring noise on very
|
|
# short names by requiring length >= 3.
|
|
nl = name.lower()
|
|
is_covered = len(nl) >= 3 and nl in covered_blob
|
|
if is_covered:
|
|
covered_count += 1
|
|
continue
|
|
# Annotate EOL status if endoflife knows the product (cheap —
|
|
# cached). Only adds a hint, never a CVE claim.
|
|
eol_hint = None
|
|
try:
|
|
st = eol_service.check_eol(db, name, version) if eol_service.resolve_product_slug(name) else None
|
|
if st:
|
|
if st.is_eol:
|
|
eol_hint = "EOL"
|
|
elif st.is_eol_soon:
|
|
eol_hint = f"EOL SOON ({st.days_to_eol}d)"
|
|
elif st.is_eoas:
|
|
eol_hint = "out-of-active-support"
|
|
except Exception:
|
|
pass
|
|
gap.append({"name": name[:255], "version": version[:100], "eol_hint": eol_hint})
|
|
|
|
gap.sort(key=lambda g: (g["eol_hint"] is None, g["name"].lower()))
|
|
return {
|
|
"asset_id": asset_id,
|
|
"hostname": asset.hostname,
|
|
"total_packages": len(seen),
|
|
"covered": covered_count,
|
|
"gap_count": len(gap),
|
|
"gap": gap,
|
|
}
|
|
|
|
|
|
@router.get("/{asset_id}/software")
|
|
def asset_software(
|
|
asset_id: int,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""On-demand installed-software inventory for one asset.
|
|
|
|
Pulled LIVE (not persisted) from whichever source backs the asset:
|
|
Wazuh syscollector packages, else Intune detectedApps. This is the same
|
|
inventory the app-CVE scanner consumes — surfaced read-only so the operator
|
|
can see what's installed without a DB table.
|
|
"""
|
|
import json as _json
|
|
from app.auth.setting_crypto import read_setting_value
|
|
|
|
asset = db.query(Asset).filter(Asset.id == asset_id).first()
|
|
if not asset:
|
|
raise HTTPException(404, "Asset not found")
|
|
|
|
software: list[dict] = []
|
|
source = None
|
|
from app.services.audit_events import asset_inventory
|
|
if asset_inventory(asset) == "wazuh":
|
|
from app.integrations.wazuh_client import WazuhClient
|
|
raw = read_setting_value(db, "wazuh_config")
|
|
if not raw:
|
|
raise HTTPException(400, "Wazuh is not configured")
|
|
cfg = _json.loads(raw)
|
|
wazuh = WazuhClient(
|
|
base_url=cfg.get("api_url"), username=cfg.get("username"),
|
|
password=cfg.get("password"), indexer_url=cfg.get("indexer_url"),
|
|
indexer_username=cfg.get("indexer_username"),
|
|
indexer_password=cfg.get("indexer_password"),
|
|
verify_ssl=bool(cfg.get("verify_ssl", True)),
|
|
)
|
|
try:
|
|
pkgs = wazuh.get_packages(asset.wazuh_agent_id) or []
|
|
except Exception as e:
|
|
raise HTTPException(502, f"Could not fetch packages from Wazuh: {e}")
|
|
source = "wazuh"
|
|
software = [{"name": (p.get("name") or "").strip(),
|
|
"version": (p.get("version") or "").strip(),
|
|
"vendor": (p.get("vendor") or p.get("format") or "").strip()} for p in pkgs]
|
|
elif asset.intune_device_id:
|
|
from app.services.intune_service import load_intune_config, _build_client
|
|
icfg = load_intune_config(db)
|
|
if not icfg:
|
|
raise HTTPException(400, "Intune is not configured")
|
|
client = _build_client(icfg)
|
|
try:
|
|
apps = client.get_detected_apps(asset.intune_device_id) or []
|
|
except Exception as e:
|
|
raise HTTPException(502, f"Could not fetch detectedApps from Intune: {e}")
|
|
finally:
|
|
client.close()
|
|
source = "intune"
|
|
software = [{"name": (a.get("name") or "").strip(),
|
|
"version": (a.get("version") or "").strip(),
|
|
"vendor": (a.get("vendor") or "").strip()} for a in apps]
|
|
else:
|
|
raise HTTPException(422, "Asset has no Wazuh agent or Intune device — no live software inventory")
|
|
|
|
# Dedup (name, version) + drop nameless, sort by name.
|
|
seen: set = set()
|
|
out: list[dict] = []
|
|
for s in software:
|
|
if not s["name"]:
|
|
continue
|
|
key = (s["name"].lower(), s["version"])
|
|
if key in seen:
|
|
continue
|
|
seen.add(key)
|
|
out.append(s)
|
|
out.sort(key=lambda s: s["name"].lower())
|
|
return {"asset_id": asset_id, "hostname": asset.hostname, "source": source,
|
|
"count": len(out), "software": out}
|
|
|
|
|
|
@router.get("/{asset_id}/containers")
|
|
def asset_containers(
|
|
asset_id: int,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""The container images Dockhand reports on this Docker host, with their
|
|
containers and open findings. Only images whose Docker host's hostname
|
|
names this asset (container_placements); clean images are not in
|
|
Dockhand's export and so not here either."""
|
|
host = db.query(Asset).filter(Asset.id == asset_id).first()
|
|
if not host:
|
|
raise HTTPException(status_code=404, detail="Asset not found")
|
|
open_count = (db.query(func.count(Vulnerability.id))
|
|
.filter(Vulnerability.asset_id == Asset.id,
|
|
Vulnerability.status == VulnerabilityStatus.open)
|
|
.correlate(Asset).scalar_subquery())
|
|
rows = (db.query(Asset, ContainerPlacement.containers, open_count)
|
|
.join(ContainerPlacement, ContainerPlacement.image_asset_id == Asset.id)
|
|
.filter(ContainerPlacement.host_asset_id == asset_id)
|
|
.order_by(func.lower(Asset.hostname)).all())
|
|
return {"hostname": host.hostname, "images": [
|
|
{"id": a.id, "image": a.hostname, "status": a.status, "containers": containers,
|
|
"vulnerability_count": n or 0, "last_scan": a.last_scan}
|
|
for a, containers, n in rows]}
|
|
|
|
|
|
@router.post("/refresh-exposure", dependencies=[audited("Asset exposure refresh")])
|
|
def refresh_exposure(
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(RequireEditor),
|
|
):
|
|
"""Refresh network-exposure scores from Wazuh syscollector ports.
|
|
|
|
Classifies risky listeners (VNC/RDP/Telnet/SMB/…) per asset into a
|
|
0-100 exposure score + service list. Informational risk dimension —
|
|
does not auto-change criticality.
|
|
"""
|
|
import json as _json
|
|
from app.auth.setting_crypto import read_setting_value
|
|
from app.integrations.wazuh_client import WazuhClient
|
|
from app.services.exposure_service import refresh_all_exposure
|
|
|
|
raw = read_setting_value(db, "wazuh_config")
|
|
if not raw:
|
|
raise HTTPException(400, "Wazuh is not configured")
|
|
try:
|
|
cfg = _json.loads(raw)
|
|
except _json.JSONDecodeError:
|
|
raise HTTPException(400, "Invalid Wazuh configuration format")
|
|
if not all([cfg.get("api_url"), cfg.get("username"), cfg.get("password")]):
|
|
raise HTTPException(400, "Incomplete Wazuh configuration")
|
|
wazuh = WazuhClient(
|
|
base_url=cfg.get("api_url"),
|
|
username=cfg.get("username"),
|
|
password=cfg.get("password"),
|
|
indexer_url=cfg.get("indexer_url"),
|
|
indexer_username=cfg.get("indexer_username"),
|
|
indexer_password=cfg.get("indexer_password"),
|
|
verify_ssl=bool(cfg.get("verify_ssl", True)),
|
|
)
|
|
return refresh_all_exposure(db, wazuh)
|
|
|
|
|
|
def _apply_asset_filters(query, db, status=None, source=None, search=None, include_inactive=False,
|
|
include_containers=True):
|
|
"""Shared filter logic for the list endpoint and the CSV export.
|
|
|
|
Kept in one place so the two views can never drift apart. Default shows
|
|
ACTIVE + INACTIVE and hides only operator-retired DECOMMISSIONED assets;
|
|
include_inactive=True (or an explicit status) overrides that.
|
|
"""
|
|
if status:
|
|
query = query.filter(Asset.status == status)
|
|
elif include_inactive:
|
|
# everything (active + inactive + decommissioned)
|
|
pass
|
|
else:
|
|
# active + inactive visible; hide only operator-retired
|
|
query = query.filter(Asset.status != AssetStatus.DECOMMISSIONED)
|
|
|
|
if source:
|
|
# Filter by the actual sync linkage, not the creation-time source
|
|
# enum: an asset first created by Wazuh and later matched by Nessus
|
|
# keeps source=WAZUH but carries a nessus_host_uuid. The per-scanner
|
|
# id columns are the multi-source truth, so a merged asset correctly
|
|
# appears under every scanner that sees it.
|
|
def _has_vuln_source(tag: str):
|
|
# Asset has >=1 finding reported by tag (the vuln-level sources
|
|
# list is the maintained multi-source truth). Catches Nessus
|
|
# imports whose host had no nessus_host_uuid (matched by hostname).
|
|
return Asset.id.in_(
|
|
db.query(Vulnerability.asset_id)
|
|
.filter(Vulnerability.sources.contains(f'"{tag}"'))
|
|
)
|
|
|
|
if source == AssetSource.WAZUH:
|
|
query = query.filter(Asset.wazuh_agent_id.isnot(None) | _has_vuln_source("wazuh"))
|
|
elif source == AssetSource.NESSUS:
|
|
query = query.filter(Asset.nessus_host_uuid.isnot(None) | _has_vuln_source("nessus"))
|
|
elif source == AssetSource.INTUNE:
|
|
# Intune devices often carry no findings so id columns are the signal.
|
|
query = query.filter(
|
|
Asset.intune_device_id.isnot(None) | Asset.defender_machine_id.isnot(None)
|
|
| _has_vuln_source("intune") | _has_vuln_source("defender"))
|
|
elif source == AssetSource.VCENTER:
|
|
# An ESXi host reached only over vCenter carries no findings until
|
|
# the first scan, so the pin column is the signal - same as Intune.
|
|
query = query.filter(
|
|
Asset.vmware_uuid.isnot(None) | (Asset.source == AssetSource.VCENTER))
|
|
elif source == AssetSource.IGEL:
|
|
# A thin client carries no findings until the first scan either,
|
|
# so the pin column is the signal - same as vCenter and Intune.
|
|
query = query.filter(
|
|
Asset.igel_unit_id.isnot(None) | (Asset.source == AssetSource.IGEL))
|
|
elif source == AssetSource.NETDISCO:
|
|
# A switch carries no findings until the first firmware scan, and
|
|
# one another source registered first keeps that source - so the
|
|
# pin column is the signal, same as vCenter and IGEL.
|
|
query = query.filter(
|
|
Asset.netdisco_device_ip.isnot(None) | (Asset.source == AssetSource.NETDISCO))
|
|
elif source == AssetSource.XENSERVER:
|
|
# A fully patched host carries no findings, so the pin column is
|
|
# the signal - same as vCenter.
|
|
query = query.filter(
|
|
Asset.xen_host_uuid.isnot(None) | (Asset.source == AssetSource.XENSERVER))
|
|
elif source == AssetSource.MANUAL:
|
|
# genuinely manual = no scanner linkage AND no scanner-sourced vuln
|
|
query = query.filter(
|
|
Asset.wazuh_agent_id.is_(None), Asset.nessus_host_uuid.is_(None),
|
|
Asset.intune_device_id.is_(None), Asset.defender_machine_id.is_(None),
|
|
Asset.vmware_uuid.is_(None), Asset.source != AssetSource.VCENTER,
|
|
Asset.igel_unit_id.is_(None), Asset.source != AssetSource.IGEL,
|
|
Asset.netdisco_device_ip.is_(None), Asset.source != AssetSource.NETDISCO,
|
|
Asset.xen_host_uuid.is_(None), Asset.source != AssetSource.XENSERVER,
|
|
Asset.source != AssetSource.CONTAINER,
|
|
~_has_vuln_source("wazuh"), ~_has_vuln_source("nessus"),
|
|
~_has_vuln_source("intune"), ~_has_vuln_source("defender"))
|
|
else:
|
|
query = query.filter(Asset.source == source)
|
|
elif not include_containers:
|
|
query = query.filter(Asset.source != AssetSource.CONTAINER)
|
|
|
|
if search:
|
|
search_pattern = f"%{search}%"
|
|
query = query.filter(
|
|
(Asset.hostname.ilike(search_pattern)) |
|
|
(Asset.ip_address.ilike(search_pattern))
|
|
)
|
|
|
|
return query
|
|
|
|
|
|
@router.get("", response_model=List[AssetResponse])
|
|
def list_assets(
|
|
status: Optional[AssetStatus] = Query(None),
|
|
source: Optional[AssetSource] = Query(None),
|
|
search: Optional[str] = Query(None, description="Suche in Hostname, IP"),
|
|
include_inactive: bool = Query(False, description="Also include DECOMMISSIONED assets (INACTIVE are shown by default)"),
|
|
include_containers: bool = Query(True, description="false hides container images (unless source=CONTAINER)"),
|
|
sort_by: str = Query("hostname", description="hostname, ip_address, operating_system, status, last_scan, network_exposure_score, policy_name, assigned_user_name"),
|
|
sort_order: str = Query("asc", description="asc, desc"),
|
|
limit: int = Query(100, le=1000),
|
|
offset: int = Query(0),
|
|
response: Response = None,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user)
|
|
):
|
|
"""Liste aller Assets mit Filterung.
|
|
|
|
Default zeigt ACTIVE **und** INACTIVE — der Sinn der Status-Spalte ist
|
|
ja, soft-inactive Assets (kein source-sync innerhalb
|
|
asset_inactive_after_days) mit Badge sichtbar zu halten, nicht sie zu
|
|
verstecken. Nur DECOMMISSIONED (operator-final) ist per default
|
|
ausgeblendet; include_inactive=true zeigt auch die.
|
|
Vuln-historie + audit bleiben erhalten.
|
|
"""
|
|
query = db.query(Asset)
|
|
query = _apply_asset_filters(query, db, status=status, source=source,
|
|
search=search, include_inactive=include_inactive,
|
|
include_containers=include_containers)
|
|
|
|
# Whitelist-driven ORDER BY. Always map to a real SQLAlchemy column;
|
|
# never pass `sort_by` straight to .order_by() (SQL-injection surface).
|
|
# Joins are added lazily so non-join sorts don't pay the join cost.
|
|
_SORT_MAP_DIRECT = {
|
|
"hostname": Asset.hostname,
|
|
"ip_address": Asset.ip_address,
|
|
"operating_system": Asset.operating_system,
|
|
"status": Asset.status,
|
|
"last_scan": Asset.last_scan,
|
|
"network_exposure_score": Asset.network_exposure_score,
|
|
"high_value_score": Asset.high_value_score,
|
|
}
|
|
sort_dir = desc if sort_order == "desc" else asc
|
|
if sort_by in _SORT_MAP_DIRECT:
|
|
col = _SORT_MAP_DIRECT[sort_by]
|
|
if sort_by == "last_scan":
|
|
# NULL-safe on BOTH directions — never-scanned assets land at
|
|
# the bottom regardless of asc/desc (PG: NULLS LAST is independent
|
|
# of the primary direction).
|
|
query = query.order_by(nulls_last(sort_dir(col)))
|
|
elif sort_by == "hostname":
|
|
# Case-insensitive alpha sort — a host called "alpine" must
|
|
# not outrank "Webserver" because the literal `A` > `W`.
|
|
query = query.order_by(nulls_last(sort_dir(func.lower(col))))
|
|
else:
|
|
# NULLs always last (both directions) — sorting Exposure/Risk
|
|
# desc must surface the real high scores first, not the empty
|
|
# ("—") rows. (Observed: desc showed blanks before real values.)
|
|
query = query.order_by(nulls_last(sort_dir(col)))
|
|
elif sort_by == "policy_name":
|
|
query = query.outerjoin(Policy, Asset.policy_id == Policy.id).order_by(sort_dir(Policy.name))
|
|
elif sort_by == "assigned_user_name":
|
|
query = query.outerjoin(User, Asset.assigned_user_id == User.id).order_by(sort_dir(User.username))
|
|
else:
|
|
# Unknown / unsupported sort_by — fall back to hostname asc.
|
|
query = query.order_by(asc(Asset.hostname))
|
|
|
|
# Total (pre-pagination) so the UI can page; returned as a header to
|
|
# keep the response body a plain array (other consumers expect a list).
|
|
total_count = query.order_by(None).count()
|
|
response.headers["X-Total-Count"] = str(total_count)
|
|
response.headers["Access-Control-Expose-Headers"] = "X-Total-Count"
|
|
|
|
assets = query.offset(offset).limit(limit).all()
|
|
|
|
# Vulnerability-Counts + assigned user info.
|
|
#
|
|
# ONE grouped query, not one COUNT per asset. This loop used to issue a
|
|
# separate `SELECT count(*) FROM vulnerabilities WHERE asset_id = ?` per
|
|
# row, so a page of 1000 assets meant 1000 round trips and the request
|
|
# took the better part of a minute (observed in the backend log:
|
|
# GET /api/v1/assets?limit=1000 dominating the page load).
|
|
counts = {}
|
|
if assets:
|
|
counts = dict(
|
|
db.query(Vulnerability.asset_id, func.count(Vulnerability.id))
|
|
.filter(Vulnerability.asset_id.in_([a.id for a in assets]),
|
|
Vulnerability.status == VulnerabilityStatus.open)
|
|
.group_by(Vulnerability.asset_id).all())
|
|
# Docker host ↔ image links, both directions in one query.
|
|
image_counts, runs_on = {}, {}
|
|
if assets:
|
|
ids = [a.id for a in assets]
|
|
for image_id, host_id, host_name in (
|
|
db.query(ContainerPlacement.image_asset_id, Asset.id, Asset.hostname)
|
|
.join(Asset, Asset.id == ContainerPlacement.host_asset_id)
|
|
.filter(ContainerPlacement.host_asset_id.in_(ids)
|
|
| ContainerPlacement.image_asset_id.in_(ids))
|
|
.order_by(func.lower(Asset.hostname))):
|
|
image_counts[host_id] = image_counts.get(host_id, 0) + 1
|
|
runs_on.setdefault(image_id, []).append({"id": host_id, "hostname": host_name})
|
|
for asset in assets:
|
|
asset.vulnerability_count = counts.get(asset.id, 0)
|
|
asset.container_image_count = image_counts.get(asset.id, 0)
|
|
asset.runs_on = runs_on.get(asset.id, [])
|
|
asset.assigned_user_name = asset.assigned_user.username if asset.assigned_user else None
|
|
asset.policy_name = asset.policy.name if asset.policy else None
|
|
|
|
return assets
|
|
|
|
|
|
def _csv_json_list(raw):
|
|
"""Parse a TEXT JSON-list column into a list (best-effort, non-fatal)."""
|
|
if not raw:
|
|
return []
|
|
try:
|
|
import json as _json
|
|
parsed = _json.loads(raw)
|
|
return parsed if isinstance(parsed, list) else []
|
|
except (ValueError, TypeError):
|
|
return []
|
|
|
|
|
|
def _fmt_exposed_services(services):
|
|
"""exposed_services JSON -> 'service:port/proto (risk N)' joined by ' | '."""
|
|
parts = []
|
|
for s in services:
|
|
if not isinstance(s, dict):
|
|
continue
|
|
seg = s.get("service") or ""
|
|
port = s.get("port")
|
|
proto = s.get("proto") or ""
|
|
risk = s.get("risk")
|
|
if port is not None:
|
|
seg = f"{seg}:{port}" if seg else str(port)
|
|
if proto:
|
|
seg += f"/{proto}"
|
|
if risk is not None:
|
|
seg += f" (risk {risk})"
|
|
if seg:
|
|
parts.append(seg)
|
|
return " | ".join(parts)
|
|
|
|
|
|
def _fmt_risk_dimensions(dims):
|
|
"""risk_dimensions JSON -> 'label (weight)' joined by ' | '."""
|
|
parts = []
|
|
for d in dims:
|
|
if not isinstance(d, dict):
|
|
continue
|
|
label = d.get("label") or d.get("role") or ""
|
|
weight = d.get("weight")
|
|
if weight is not None:
|
|
label = f"{label} ({weight})"
|
|
if label:
|
|
parts.append(label)
|
|
return " | ".join(parts)
|
|
|
|
|
|
@router.get("/names")
|
|
def list_asset_names(
|
|
include_inactive: bool = Query(False),
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""Every asset as {id, hostname} — for host pickers. No cap, no counts.
|
|
|
|
The host dropdown on the Vulnerabilities page used to fetch the full asset
|
|
list at `limit=1000`, which is the endpoint's hard maximum. With 1194
|
|
assets in the estate the list simply stopped at the letter "M" and the
|
|
remaining 194 hosts could not be selected at all — a silent cap on a
|
|
picker, which is the worst place for one.
|
|
|
|
Fetching the full rows was also far more than a picker needs: a whole
|
|
AssetResponse each, with the open-vulnerability count behind it. Two
|
|
columns, one query, ordered the same way the table is (case-insensitively,
|
|
so "alpine" does not outrank "Webserver").
|
|
|
|
Declared BEFORE the `/{asset_id}` route on purpose — FastAPI matches in
|
|
declaration order, and "names" would otherwise be read as an asset id.
|
|
"""
|
|
q = _apply_asset_filters(db.query(Asset.id, Asset.hostname), db,
|
|
include_inactive=include_inactive)
|
|
rows = q.order_by(func.lower(Asset.hostname)).all()
|
|
return [{"id": r.id, "hostname": r.hostname} for r in rows]
|
|
|
|
|
|
@router.get("/export")
|
|
def export_assets_csv(
|
|
status: Optional[AssetStatus] = Query(None),
|
|
source: Optional[AssetSource] = Query(None),
|
|
search: Optional[str] = Query(None, description="Suche in Hostname, IP"),
|
|
include_inactive: bool = Query(False, description="Also include DECOMMISSIONED assets (INACTIVE are shown by default)"),
|
|
include_containers: bool = Query(True, description="false hides container images (unless source=CONTAINER)"),
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user),
|
|
):
|
|
"""Stream ALL matching assets as CSV - full export, no pagination cap.
|
|
|
|
Same filters as GET /assets (search, source, status/include_inactive), but
|
|
returns every row at once with the complete field set. Streamed via
|
|
StreamingResponse + yield_per so large inventories do not sit in memory.
|
|
"""
|
|
query = db.query(Asset)
|
|
query = _apply_asset_filters(query, db, status=status, source=source,
|
|
search=search, include_inactive=include_inactive,
|
|
include_containers=include_containers)
|
|
# Eager-load the related columns the CSV reads (policy / assignee / groups)
|
|
# so a full export does not fall back to a lazy SELECT per row under
|
|
# yield_per. selectinload is yield_per-safe (joinedload is not).
|
|
query = query.options(
|
|
selectinload(Asset.policy),
|
|
selectinload(Asset.assigned_user),
|
|
selectinload(Asset.groups),
|
|
)
|
|
query = query.order_by(asc(func.lower(Asset.hostname)))
|
|
|
|
# One grouped query for the open-vulnerability counts (avoids an N+1
|
|
# per-row COUNT over the whole export).
|
|
vuln_counts = dict(
|
|
db.query(Vulnerability.asset_id, func.count(Vulnerability.id))
|
|
.filter(Vulnerability.status == VulnerabilityStatus.open)
|
|
.group_by(Vulnerability.asset_id)
|
|
.all()
|
|
)
|
|
|
|
def _fmt_dt(value):
|
|
return value.isoformat() if value else ""
|
|
|
|
def row_iter():
|
|
buf = io.StringIO()
|
|
writer = csv.writer(buf, quoting=csv.QUOTE_MINIMAL)
|
|
writer.writerow([
|
|
"ID", "Hostname", "IP Address", "OS", "OS Version", "Kernel Version",
|
|
"VMware Build", "Status", "Source", "Criticality",
|
|
"Compliance Frameworks", "Last Scan", "Last Seen",
|
|
"Policy", "Assigned User", "Groups", "Open Vulnerabilities",
|
|
"Network Exposure Score", "Exposed Services", "Exposure Updated At",
|
|
"High Value Score", "Risk Dimensions", "Risk Dimensions Updated At",
|
|
"Description", "Location", "Owner",
|
|
"Wazuh Agent ID", "Nessus Host UUID", "Intune Device ID",
|
|
"Defender Machine ID", "AAD Device ID", "VMware UUID",
|
|
"Created At", "Updated At",
|
|
"Intune Serial", "IGEL Unit ID", "IGEL Serial",
|
|
"Netdisco Device IP", "Netdisco Serial", "VMware Serial", "XenServer Host UUID",
|
|
])
|
|
yield buf.getvalue()
|
|
buf.seek(0); buf.truncate(0)
|
|
|
|
for asset in query.yield_per(500):
|
|
writer.writerow([
|
|
asset.id,
|
|
asset.hostname or "",
|
|
asset.ip_address or "",
|
|
asset.operating_system or "",
|
|
asset.os_version or "",
|
|
asset.kernel_version or "",
|
|
asset.vmware_build or "",
|
|
asset.status.value if hasattr(asset.status, "value") else str(asset.status),
|
|
asset.source.value if hasattr(asset.source, "value") else str(asset.source),
|
|
asset.criticality or "",
|
|
" | ".join(str(x) for x in _csv_json_list(asset.compliance_frameworks)),
|
|
_fmt_dt(asset.last_scan),
|
|
_fmt_dt(asset.last_seen),
|
|
asset.policy.name if asset.policy else "",
|
|
asset.assigned_user.username if asset.assigned_user else "",
|
|
" | ".join(g.name for g in asset.groups) if asset.groups else "",
|
|
vuln_counts.get(asset.id, 0),
|
|
asset.network_exposure_score if asset.network_exposure_score is not None else "",
|
|
_fmt_exposed_services(_csv_json_list(asset.exposed_services)),
|
|
_fmt_dt(asset.exposure_updated_at),
|
|
asset.high_value_score if asset.high_value_score is not None else "",
|
|
_fmt_risk_dimensions(_csv_json_list(asset.risk_dimensions)),
|
|
_fmt_dt(asset.risk_dimensions_updated_at),
|
|
asset.description or "",
|
|
asset.location or "",
|
|
asset.owner or "",
|
|
asset.wazuh_agent_id or "",
|
|
asset.nessus_host_uuid or "",
|
|
asset.intune_device_id or "",
|
|
asset.defender_machine_id or "",
|
|
asset.aad_device_id or "",
|
|
asset.vmware_uuid or "",
|
|
_fmt_dt(asset.created_at),
|
|
_fmt_dt(asset.updated_at),
|
|
asset.intune_serial or "",
|
|
asset.igel_unit_id or "",
|
|
asset.igel_serial or "",
|
|
asset.netdisco_device_ip or "",
|
|
asset.netdisco_serial or "",
|
|
asset.vmware_serial or "",
|
|
asset.xen_host_uuid or "",
|
|
])
|
|
yield buf.getvalue()
|
|
buf.seek(0); buf.truncate(0)
|
|
|
|
fname = f"assets_{datetime.now().strftime('%Y%m%d_%H%M%S')}.csv"
|
|
return StreamingResponse(
|
|
row_iter(),
|
|
media_type="text/csv; charset=utf-8",
|
|
headers={"Content-Disposition": f'attachment; filename="{fname}"'},
|
|
)
|
|
|
|
|
|
@router.get("/{asset_id}", response_model=AssetResponse)
|
|
def get_asset_detail(
|
|
asset_id: int,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user)
|
|
):
|
|
"""Asset-Details"""
|
|
asset = db.query(Asset).filter(Asset.id == asset_id).first()
|
|
|
|
if not asset:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Asset not found"
|
|
)
|
|
|
|
vuln_count = db.query(Vulnerability).filter(
|
|
Vulnerability.asset_id == asset_id,
|
|
Vulnerability.status == VulnerabilityStatus.open
|
|
).count()
|
|
|
|
asset.vulnerability_count = vuln_count
|
|
return asset
|
|
|
|
|
|
@router.post("", status_code=status.HTTP_201_CREATED, response_model=AssetResponse)
|
|
def create_asset(
|
|
asset_data: AssetCreateRequest,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(RequireEditor)
|
|
):
|
|
"""
|
|
Erstellt manuelles Asset (für Systeme ohne Wazuh-Agent)
|
|
"""
|
|
# Kein Duplikat-Check auf Hostname oder IP-Adresse: beides sind keine
|
|
# Identifikatoren. Ab Werk vergebene Namen ("RackPDU") stehen mehrfach im
|
|
# Bestand; Gastnetze, VPN-Pools, VLAN-Segmente und recycelte DHCP-Leases
|
|
# fuehren dieselbe Adresse mehrfach. Der Operator konnte die betroffenen
|
|
# Systeme sonst gar nicht erst anlegen. Was ein Asset identifiziert, steht
|
|
# in services/asset_matching.py.
|
|
|
|
# Erstelle Asset
|
|
asset = Asset(
|
|
**asset_data.model_dump(),
|
|
source=AssetSource.MANUAL,
|
|
status=AssetStatus.ACTIVE
|
|
)
|
|
|
|
db.add(asset)
|
|
db.commit()
|
|
db.refresh(asset)
|
|
|
|
# Audit-Log
|
|
audit_log = AuditLog(
|
|
user_id=current_user.id,
|
|
event_type=AuditEventType.ASSET_CREATED,
|
|
event_description=f"Manual asset created: {asset.hostname}",
|
|
resource_type="asset",
|
|
resource_id=str(asset.id),
|
|
timestamp=datetime.now()
|
|
)
|
|
db.add(audit_log)
|
|
db.commit()
|
|
|
|
asset.vulnerability_count = 0
|
|
return asset
|
|
|
|
|
|
@router.put("/{asset_id}", response_model=AssetResponse)
|
|
def update_asset(
|
|
asset_id: int,
|
|
update_data: AssetUpdateRequest,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(RequireEditor)
|
|
):
|
|
"""Aktualisiert Asset-Daten"""
|
|
asset = db.query(Asset).filter(Asset.id == asset_id).first()
|
|
|
|
if not asset:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Asset not found"
|
|
)
|
|
|
|
# Update nur gesetzte Felder
|
|
import json as _json
|
|
# Lifecycle transitions do not go through setattr: apply_status() writes the
|
|
# dedicated ASSET_DEACTIVATED / ASSET_REACTIVATED audit entry, so a manual
|
|
# decommission reads the same way in the trail as a sync-driven one.
|
|
status_change: Optional[AssetStatus] = None
|
|
changes = update_data.model_dump(exclude_unset=True)
|
|
before = _asset_state(asset, [f for f in changes if f != "status"])
|
|
for field, value in changes.items():
|
|
if field == "status":
|
|
if value is not None:
|
|
status_change = value if isinstance(value, AssetStatus) else AssetStatus(value)
|
|
continue
|
|
if field == "compliance_frameworks":
|
|
# Stored as TEXT JSON list
|
|
value = _json.dumps(value or []) if value is not None else None
|
|
if field == "criticality" and value is not None:
|
|
value = value.lower()
|
|
if value not in ("low", "normal", "high", "critical"):
|
|
raise HTTPException(400, "criticality must be one of: low, normal, high, critical")
|
|
setattr(asset, field, value)
|
|
|
|
if status_change is not None:
|
|
from app.services.asset_lifecycle import apply_status
|
|
# force=True: the operator-final guard exists to stop a sync from
|
|
# reviving a retired asset, not the operator who retired it.
|
|
apply_status(
|
|
db, asset, status_change,
|
|
f"manual status change by {current_user.username}",
|
|
force=True,
|
|
user_id=current_user.id,
|
|
)
|
|
|
|
_audit_asset_update(db, current_user, asset, before)
|
|
db.commit()
|
|
db.refresh(asset)
|
|
|
|
vuln_count = db.query(Vulnerability).filter(
|
|
Vulnerability.asset_id == asset_id,
|
|
Vulnerability.status == VulnerabilityStatus.open
|
|
).count()
|
|
|
|
asset.vulnerability_count = vuln_count
|
|
return asset
|
|
|
|
|
|
@router.delete("/{asset_id}", status_code=status.HTTP_204_NO_CONTENT)
|
|
def delete_asset(
|
|
asset_id: int,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(RequireAdmin)
|
|
):
|
|
"""
|
|
Löscht Asset (Admin-only)
|
|
|
|
ACHTUNG: Löscht auch alle zugehörigen Vulnerabilities (CASCADE)
|
|
"""
|
|
asset = db.query(Asset).filter(Asset.id == asset_id).first()
|
|
|
|
if not asset:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Asset not found"
|
|
)
|
|
|
|
# Revisionssicher audit — snapshot the full asset + cascade impact
|
|
# BEFORE deletion so the trail says WHO deleted WHAT (and how much
|
|
# history it took with it), analog to the CVE status-change entries.
|
|
import json as _json
|
|
vuln_count = db.query(Vulnerability).filter(
|
|
Vulnerability.asset_id == asset.id
|
|
).count()
|
|
snapshot = {
|
|
"hostname": asset.hostname,
|
|
"ip_address": asset.ip_address,
|
|
"operating_system": asset.operating_system,
|
|
"source": asset.source.value if hasattr(asset.source, "value") else str(asset.source),
|
|
"status": asset.status.value if hasattr(asset.status, "value") else str(asset.status),
|
|
"wazuh_agent_id": asset.wazuh_agent_id,
|
|
"vulnerabilities_cascade_deleted": vuln_count,
|
|
"deleted_by": current_user.username,
|
|
}
|
|
audit_log = AuditLog(
|
|
user_id=current_user.id,
|
|
event_type=AuditEventType.ASSET_DELETED,
|
|
event_description=(
|
|
f"Asset '{asset.hostname}' ({asset.ip_address or 'no-ip'}) deleted by "
|
|
f"{current_user.username} — {vuln_count} vulnerabilities cascade-removed"
|
|
)[:500],
|
|
resource_type="asset",
|
|
resource_id=str(asset.id),
|
|
old_value=_json.dumps(snapshot),
|
|
new_value="deleted",
|
|
timestamp=datetime.now()
|
|
)
|
|
db.add(audit_log)
|
|
db.commit() # persist audit first so it survives even if delete fails
|
|
|
|
db.delete(asset)
|
|
db.commit()
|
|
|
|
return None
|
|
|
|
|
|
@router.patch("/{asset_id}/assign")
|
|
def assign_asset(
|
|
asset_id: int,
|
|
assign_data: AssetAssignRequest,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(RequireEditor)
|
|
):
|
|
"""Assigns a user to an asset"""
|
|
asset = db.query(Asset).filter(Asset.id == asset_id).first()
|
|
if not asset:
|
|
raise HTTPException(status_code=404, detail="Asset not found")
|
|
|
|
if assign_data.user_id is not None:
|
|
user = db.query(User).filter(User.id == assign_data.user_id).first()
|
|
if not user:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
|
|
before = _assignment(asset)
|
|
asset.assigned_user_id = assign_data.user_id
|
|
|
|
# Handle single group assignment (legacy support)
|
|
from app.models.group import Group
|
|
if assign_data.group_id is None:
|
|
# Clear groups when explicitly set to null
|
|
asset.groups = []
|
|
else:
|
|
group = db.query(Group).filter(Group.id == assign_data.group_id).first()
|
|
if group:
|
|
asset.groups = [group]
|
|
|
|
db.flush()
|
|
db.expire(asset, ["assigned_user"]) # reload it from the new id
|
|
record_change(db, current_user.id, AuditEventType.ASSET_UPDATED, "asset", asset.id,
|
|
f"Asset updated: {asset.hostname}", before, _assignment(asset))
|
|
db.commit()
|
|
db.refresh(asset)
|
|
|
|
vuln_count = db.query(Vulnerability).filter(
|
|
Vulnerability.asset_id == asset_id,
|
|
Vulnerability.status == VulnerabilityStatus.open
|
|
).count()
|
|
asset.vulnerability_count = vuln_count
|
|
asset.assigned_user_name = asset.assigned_user.username if asset.assigned_user else None
|
|
return asset
|
|
|
|
|
|
@router.get("/{asset_id}/vulnerabilities")
|
|
def get_asset_vulnerabilities(
|
|
asset_id: int,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user)
|
|
):
|
|
"""Alle Vulnerabilities eines Assets"""
|
|
asset = db.query(Asset).filter(Asset.id == asset_id).first()
|
|
|
|
if not asset:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail="Asset not found"
|
|
)
|
|
|
|
vulns = db.query(Vulnerability).filter(
|
|
Vulnerability.asset_id == asset_id
|
|
).all()
|
|
|
|
return [{
|
|
"id": v.id,
|
|
"cve_id": v.cve_id,
|
|
"severity": v.severity,
|
|
"status": v.status,
|
|
"cvss_score": v.cvss_score,
|
|
"package_name": v.package_name,
|
|
"detected_at": v.detected_at
|
|
} for v in vulns]
|
|
|
|
# --- Wazuh Sync Endpoint ---
|
|
import json
|
|
from app.integrations.wazuh_client import WazuhClient, WazuhAPIError
|
|
from app.models.setting import Setting
|
|
from app.services.asset_matching import like_literal, pick_one
|
|
|
|
@router.post("/sync_wazuh", response_model=dict, dependencies=[audited("Wazuh asset sync")])
|
|
def sync_wazuh_assets(
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(RequireEditor)
|
|
):
|
|
"""
|
|
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")
|
|
if not raw_wazuh:
|
|
raise HTTPException(status_code=400, detail="Wazuh configuration not found.")
|
|
|
|
try:
|
|
config = json.loads(raw_wazuh)
|
|
api_url = config.get("api_url")
|
|
username = config.get("username")
|
|
password = config.get("password")
|
|
except json.JSONDecodeError:
|
|
raise HTTPException(status_code=500, detail="Invalid Wazuh configuration format.")
|
|
|
|
if not all([api_url, username, password]):
|
|
raise HTTPException(status_code=400, detail="Incomplete Wazuh configuration.")
|
|
|
|
indexer_url = config.get("indexer_url")
|
|
indexer_username = config.get("indexer_username")
|
|
indexer_password = config.get("indexer_password")
|
|
|
|
stats = {"created": 0, "updated": 0, "total_from_wazuh": 0, "assets_inactivated": 0, "assets_reactivated": 0}
|
|
seen_wazuh_agent_ids: set = set() # for sync-driven asset reconciliation
|
|
|
|
# 2. Fetch Agents from Wazuh
|
|
try:
|
|
with WazuhClient(
|
|
base_url=api_url,
|
|
username=username,
|
|
password=password,
|
|
indexer_url=indexer_url,
|
|
indexer_username=indexer_username,
|
|
indexer_password=indexer_password,
|
|
verify_ssl=bool(config.get("verify_ssl", True))
|
|
) as client:
|
|
# ALL agents, disconnected ones included. Fetching active-only was
|
|
# a data-loss bug, not a filter: the reconcile below inactivates
|
|
# every agent id this sync does not report, and an INACTIVE asset
|
|
# is hidden from every CVE view — so a laptop switched off over
|
|
# the weekend lost its findings within one sync, while Wazuh
|
|
# itself kept showing them. How long an offline host stays visible
|
|
# is now the grace window's decision (asset_inactive_after_days),
|
|
# not the TCP connection's.
|
|
from app.services.asset_lifecycle import (
|
|
advance_last_seen,
|
|
apply_status,
|
|
created,
|
|
parse_source_timestamp,
|
|
status_for_last_seen,
|
|
threshold_days,
|
|
)
|
|
grace_days = threshold_days(db)
|
|
|
|
agents = []
|
|
while True:
|
|
page = client.get_agents(limit=500, offset=len(agents))
|
|
agents.extend(page)
|
|
# ponytail: 20k-agent ceiling guards against a server that
|
|
# ignores `offset` and would otherwise loop forever. Raise it
|
|
# if a deployment ever gets that big.
|
|
if len(page) < 500 or len(agents) >= 20000:
|
|
break
|
|
|
|
stats["total_from_wazuh"] = len(agents)
|
|
stats["never_connected_skipped"] = 0
|
|
# Every agent id this run reports — the discriminator for the
|
|
# hostname fallback below.
|
|
live_agent_ids = {a.get("id") for a in agents if a.get("id")}
|
|
|
|
for agent in agents:
|
|
agent_id = agent.get("id")
|
|
if agent_id == "000": continue # Skip manager itself
|
|
|
|
# Registered but never checked in: no inventory, no keepalive,
|
|
# nothing to scan. Left out of seen_ids on purpose so a stale
|
|
# asset behind such an id still ages out normally.
|
|
last_keepalive = parse_source_timestamp(agent.get("lastKeepAlive"))
|
|
if last_keepalive is None and agent.get("status") == "never_connected":
|
|
stats["never_connected_skipped"] += 1
|
|
continue
|
|
|
|
seen_wazuh_agent_ids.add(agent_id)
|
|
|
|
hostname = agent.get("name")
|
|
ip = agent.get("ip")
|
|
os_info = agent.get("os", {})
|
|
os_name = os_info.get("name")
|
|
os_version = os_info.get("version")
|
|
|
|
# Check if exists by Wazuh ID
|
|
asset = db.query(Asset).filter(Asset.wazuh_agent_id == agent_id).first()
|
|
|
|
if not asset and hostname:
|
|
# Check by hostname to avoid duplicates if manually added
|
|
# …but only if the name is free. This stage is what keeps
|
|
# a re-registered agent on its asset: authd's <force>
|
|
# replaces an agent of the same name (default since 4.3),
|
|
# the host comes back under a NEW id, and the name is the
|
|
# only way back to its history. A replaced agent is gone
|
|
# from the list, so its id is not live — while two agents
|
|
# that are BOTH in this run are two hosts, and adopting
|
|
# the first one's asset would move the second one's
|
|
# findings onto it and take the id away (two APC PDUs, both
|
|
# named "RackPDU" from the factory, is all it takes).
|
|
# Several free same-named assets: the address decides, or
|
|
# the agent gets its own (pick_one).
|
|
# Case-insensitive like every other source: Intune listed
|
|
# "PC-LAB01", the agent reports "pc-lab01", and an exact
|
|
# compare forked the device into two assets.
|
|
free = [a for a in db.query(Asset)
|
|
.filter(Asset.hostname.ilike(like_literal(hostname),
|
|
escape="\\")).order_by(Asset.id)
|
|
if not (a.wazuh_agent_id and a.wazuh_agent_id in live_agent_ids)]
|
|
asset = pick_one(free, ip, hostname)
|
|
|
|
if asset:
|
|
# Update
|
|
asset.wazuh_agent_id = agent_id
|
|
asset.ip_address = ip or asset.ip_address
|
|
asset.operating_system = os_name or asset.operating_system
|
|
asset.os_version = os_version or asset.os_version
|
|
# Wazuh's own clock, not ours: `datetime.now()` recorded
|
|
# when TrueVuln ran a sync, which made every registered
|
|
# agent look permanently fresh and left no usable stamp to
|
|
# measure a grace window against.
|
|
# Forward only: another source (Intune) may have seen the
|
|
# host after the agent died.
|
|
advance_last_seen(asset, last_keepalive or datetime.now(), "wazuh")
|
|
asset.wazuh_last_seen = last_keepalive
|
|
asset.source = AssetSource.WAZUH
|
|
# Audited, not a bare assignment: this line revives an
|
|
# asset the reconcile deactivated one sync earlier (agent
|
|
# merely disconnected, keepalive still inside the window),
|
|
# and a silent revival left the audit log ending on a
|
|
# deactivation for an asset the UI showed as ACTIVE.
|
|
apply_status(
|
|
db, asset,
|
|
status_for_last_seen(asset.last_seen, grace_days),
|
|
f"Wazuh agent sync (last keepalive {last_keepalive}, "
|
|
f"last seen {asset.last_seen} by {asset.last_seen_source}, "
|
|
f"grace window {grace_days}d)",
|
|
)
|
|
stats["updated"] += 1
|
|
else:
|
|
# Create
|
|
# Default Group Logic
|
|
default_group_setting = db.query(Setting).filter(Setting.key == "default_group_id").first()
|
|
group_id = int(default_group_setting.value) if default_group_setting and default_group_setting.value else None
|
|
|
|
first_seen = last_keepalive or datetime.now()
|
|
new_asset = Asset(
|
|
hostname=hostname,
|
|
ip_address=ip,
|
|
wazuh_agent_id=agent_id,
|
|
operating_system=os_name,
|
|
os_version=os_version,
|
|
source=AssetSource.WAZUH,
|
|
# A host discovered for the first time but offline for
|
|
# months arrives INACTIVE — it should not import as a
|
|
# live system just because we happened to meet it now.
|
|
status=status_for_last_seen(first_seen, grace_days),
|
|
last_seen=first_seen,
|
|
last_seen_source="wazuh",
|
|
wazuh_last_seen=last_keepalive,
|
|
)
|
|
|
|
# Handle default group
|
|
if group_id:
|
|
from app.models.group import Group
|
|
default_group = db.query(Group).filter(Group.id == group_id).first()
|
|
if default_group:
|
|
new_asset.groups.append(default_group)
|
|
|
|
db.add(new_asset)
|
|
created(db, new_asset, "Wazuh")
|
|
stats["created"] += 1
|
|
|
|
# The manager nodes themselves. They run no agent — they
|
|
# inventory other hosts and nobody inventories them — so they were
|
|
# never assets and their own CVEs (54 at NVD, several of them RCE
|
|
# on the server) went unseen. The API reports its own version
|
|
# without an agent; no extra credentials, same session.
|
|
try:
|
|
# An all-in-one install reports no IP for its single node —
|
|
# there is no cluster to address. The API URL points at exactly
|
|
# that machine, so it is the honest fallback rather than
|
|
# leaving the column empty.
|
|
api_host = None
|
|
try:
|
|
import ipaddress
|
|
import socket
|
|
from urllib.parse import urlparse
|
|
host = urlparse(config.get("api_url") or "").hostname
|
|
if host:
|
|
try:
|
|
ipaddress.ip_address(host)
|
|
api_host = host # already an address
|
|
except ValueError:
|
|
# A hostname in a column called IP ADDRESS breaks
|
|
# sorting and filtering, so resolve it. If DNS has
|
|
# no answer the field stays empty rather than
|
|
# holding something that is not an address.
|
|
api_host = socket.gethostbyname(host)
|
|
except Exception as e:
|
|
logger.debug("could not resolve the Wazuh API host: %s", e)
|
|
api_host = None
|
|
nodes = client.get_cluster_nodes()
|
|
for node in nodes:
|
|
nid = f"node:{node['name']}"[:64]
|
|
if not node.get("ip") and len(nodes) == 1:
|
|
node["ip"] = api_host
|
|
seen_wazuh_agent_ids.add(nid)
|
|
a = db.query(Asset).filter(Asset.wazuh_agent_id == nid).first()
|
|
if not a:
|
|
# Never an agent's asset: node names ship as "node01",
|
|
# and taking it overwrote a real agent's id.
|
|
# Several case variants of the name: the address decides.
|
|
a = pick_one(db.query(Asset).filter(
|
|
Asset.hostname.ilike(like_literal(node["name"]), escape="\\"),
|
|
Asset.wazuh_agent_id.is_(None)).order_by(Asset.id).all(),
|
|
node.get("ip"), node["name"])
|
|
label = f"Wazuh {node.get('type') or 'node'} node"
|
|
if a:
|
|
a.wazuh_agent_id = nid
|
|
a.operating_system = label
|
|
a.os_version = node.get("version") or a.os_version
|
|
a.ip_address = node.get("ip") or a.ip_address
|
|
a.last_seen = datetime.now()
|
|
a.last_seen_source = "wazuh"
|
|
apply_status(db, a, AssetStatus.ACTIVE,
|
|
"Wazuh manager node reported by the cluster sync")
|
|
stats["updated"] += 1
|
|
else:
|
|
a = Asset(hostname=node["name"], ip_address=node.get("ip"),
|
|
wazuh_agent_id=nid, operating_system=label,
|
|
os_version=node.get("version"),
|
|
source=AssetSource.WAZUH,
|
|
status=AssetStatus.ACTIVE,
|
|
last_seen=datetime.now(),
|
|
last_seen_source="wazuh")
|
|
db.add(a)
|
|
created(db, a, "Wazuh")
|
|
stats["created"] += 1
|
|
# Flush inside the guard: the agent loop's commit comes
|
|
# after this block, so anything rejected by the database here
|
|
# would surface there instead and take the whole sync down
|
|
# with a 500 — which is exactly what a too-long node id did.
|
|
db.flush()
|
|
except Exception as e:
|
|
db.rollback()
|
|
logger.warning("wazuh cluster-node sync failed: %s", e)
|
|
|
|
db.commit()
|
|
|
|
# Sync-driven reconciliation: any ACTIVE WAZUH-sourced asset whose
|
|
# wazuh_agent_id was not returned by this sync (deleted on the
|
|
# Wazuh server, or no longer in any agent list) is flipped to
|
|
# INACTIVE. Vice-versa for re-activation. Audit-logged.
|
|
from app.services.asset_lifecycle import reconcile_missing_from_sync
|
|
recon = reconcile_missing_from_sync(
|
|
db,
|
|
source=AssetSource.WAZUH,
|
|
seen_ids=seen_wazuh_agent_ids,
|
|
id_field=Asset.wazuh_agent_id,
|
|
reason="not in latest Wazuh agents list",
|
|
)
|
|
stats["assets_inactivated"] = recon["inactivated"]
|
|
stats["assets_reactivated"] = recon["reactivated"]
|
|
from app.services.asset_lifecycle import unlink_gone_wazuh_agents
|
|
stats["wazuh_unlinked"] = unlink_gone_wazuh_agents(db, seen_wazuh_agent_ids)
|
|
if recon["inactivated"] or recon["reactivated"] or stats["wazuh_unlinked"]:
|
|
db.commit()
|
|
|
|
except Exception as e:
|
|
logger.error(f"Wazuh Sync Failed: {e}")
|
|
# 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
|
|
|
|
|
|
@router.post("/{asset_id}/rescan", response_model=dict)
|
|
def rescan_asset(
|
|
asset_id: int,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(RequireEditor)
|
|
):
|
|
"""
|
|
Triggert einen gezielten Rescan eines einzelnen Assets via Wazuh.
|
|
"""
|
|
from app.routers.vulnerabilities import sync_agent_vulnerabilities
|
|
|
|
asset = db.query(Asset).filter(Asset.id == asset_id).first()
|
|
if not asset:
|
|
raise HTTPException(status_code=404, detail="Asset not found")
|
|
|
|
if asset.source != AssetSource.WAZUH or not asset.wazuh_agent_id:
|
|
raise HTTPException(status_code=400, detail="Rescan only possible for Wazuh-managed assets")
|
|
|
|
# 1. Wazuh Config holen (transparently decrypted)
|
|
from app.auth.setting_crypto import read_setting_value
|
|
raw_wazuh = read_setting_value(db, "wazuh_config")
|
|
if not raw_wazuh:
|
|
raise HTTPException(status_code=400, detail="Wazuh configuration missing")
|
|
|
|
config = json.loads(raw_wazuh)
|
|
wazuh = 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))
|
|
)
|
|
|
|
try:
|
|
# Sync Vulnerabilities for this specific agent
|
|
sync_agent_vulnerabilities(db, wazuh, asset.wazuh_agent_id, asset)
|
|
|
|
# Audit Log
|
|
audit = AuditLog(
|
|
user_id=current_user.id,
|
|
event_type=AuditEventType.SCAN_TRIGGERED,
|
|
event_description=f"Manual rescan started for asset {asset.hostname} (ID: {asset.wazuh_agent_id})",
|
|
resource_type="asset",
|
|
resource_id=str(asset_id),
|
|
timestamp=datetime.now()
|
|
)
|
|
db.add(audit)
|
|
db.commit()
|
|
|
|
return {"message": f"Rescan for {asset.hostname} successfully completed.", "last_scan": asset.last_scan}
|
|
except Exception as e:
|
|
logger.error(f"Rescan error: {e}")
|
|
raise HTTPException(status_code=500, detail=f"Rescan failed: {str(e)}")
|