vulncheck
00d22fc318
Tester: a CVE newly created by a sync appeared in the vuln list but left
NO initial audit entry ("new CVE detected on asset X at ...") — the audit
trail only began with the first status change (open -> patched etc.).
Confirmed: not intentional, simply never built; fails the revisionssicher
requirement.
- Migration 031: ALTER TYPE auditeventtype ADD VALUE
'VULNERABILITY_DETECTED' (idempotent, autocommit block).
- New app/services/audit_events.py: audit_new_vulnerabilities() writes one
System/Auto event per newly created finding:
"New finding detected: CVE-X on <hostname> (severity=..., source=...)".
- Wired into every creation path:
* Wazuh full sync (run_wazuh_vulnerability_sync) -> source=wazuh
* Wazuh per-agent sync (sync_agent_vulnerabilities, covers the
scheduler loop + per-asset rescan) -> source=wazuh
* Nessus sync (incl. Nessus EOL pseudo-vulns via
newly_created_vuln_ids) -> source=nessus
* endoflife.date upsert -> source=eol_check
* M365 Apps upsert -> source=m365_check
Full-sync and per-agent paths are independent (no double events).
- Best-effort: audit failure never breaks a sync.
Migration 031 required: alembic upgrade head.
95 lines
2.9 KiB
Python
95 lines
2.9 KiB
Python
"""
|
|
Audit Log Model for security events
|
|
"""
|
|
from enum import Enum
|
|
from sqlalchemy import Column, Integer, String, Text, ForeignKey, Enum as SQLEnum, DateTime
|
|
from sqlalchemy.orm import relationship
|
|
from app.models.base import Base
|
|
|
|
|
|
class AuditEventType(str, Enum):
|
|
"""Audit event types"""
|
|
# Authentication (local)
|
|
LOGIN_SUCCESS = "LOGIN_SUCCESS"
|
|
LOGIN_FAILED = "LOGIN_FAILED"
|
|
LOGOUT = "LOGOUT"
|
|
PASSWORD_CHANGE = "PASSWORD_CHANGE" # nosec: False positive, this is an Enum event name, not a password
|
|
|
|
# Authentication (external providers)
|
|
LOGIN_LDAP_SUCCESS = "LOGIN_LDAP_SUCCESS"
|
|
LOGIN_SSO_SUCCESS = "LOGIN_SSO_SUCCESS"
|
|
AUTH_PROVIDER_FAILED = "AUTH_PROVIDER_FAILED"
|
|
JIT_USER_CREATED = "JIT_USER_CREATED"
|
|
EXTERNAL_ROLE_MAPPED = "EXTERNAL_ROLE_MAPPED"
|
|
USER_AUTO_LINKED = "USER_AUTO_LINKED"
|
|
|
|
# MFA
|
|
MFA_ENABLED = "MFA_ENABLED"
|
|
MFA_DISABLED = "MFA_DISABLED"
|
|
MFA_VERIFIED = "MFA_VERIFIED"
|
|
MFA_FAILED = "MFA_FAILED"
|
|
|
|
# Authorization
|
|
ACCESS_DENIED = "ACCESS_DENIED"
|
|
PERMISSION_ESCALATION = "PERMISSION_ESCALATION"
|
|
USER_DELETED = "USER_DELETED"
|
|
|
|
# Data Operations
|
|
VULNERABILITY_DETECTED = "VULNERABILITY_DETECTED" # sync created a new finding
|
|
VULNERABILITY_UPDATED = "VULNERABILITY_UPDATED"
|
|
ASSET_CREATED = "ASSET_CREATED"
|
|
ASSET_UPDATED = "ASSET_UPDATED"
|
|
ASSET_DELETED = "ASSET_DELETED"
|
|
ASSET_DEACTIVATED = "ASSET_DEACTIVATED" # sync no longer reports this asset -> INACTIVE
|
|
ASSET_REACTIVATED = "ASSET_REACTIVATED" # sync reported it again -> ACTIVE
|
|
|
|
# Scans
|
|
SCAN_TRIGGERED = "SCAN_TRIGGERED"
|
|
SCAN_COMPLETED = "SCAN_COMPLETED"
|
|
|
|
# AI
|
|
AI_ANALYSIS_REQUESTED = "AI_ANALYSIS_REQUESTED"
|
|
|
|
# System
|
|
CONFIG_CHANGE = "CONFIG_CHANGE"
|
|
SECURITY_ALERT = "SECURITY_ALERT"
|
|
|
|
|
|
class AuditLog(Base):
|
|
"""
|
|
Audit Log table for security and compliance logging
|
|
|
|
All security-relevant events are logged here
|
|
"""
|
|
__tablename__ = "audit_logs"
|
|
|
|
id = Column(Integer, primary_key=True, index=True)
|
|
user_id = Column(Integer, ForeignKey("users.id"), nullable=True, index=True)
|
|
|
|
# Event Details
|
|
event_type = Column(
|
|
SQLEnum(AuditEventType),
|
|
nullable=False,
|
|
index=True
|
|
)
|
|
event_description = Column(String(500), nullable=False)
|
|
|
|
# Context
|
|
ip_address = Column(String(45), nullable=True)
|
|
user_agent = Column(String(500), nullable=True)
|
|
|
|
# Additional Data
|
|
resource_type = Column(String(50), nullable=True) # e.g. "vulnerability", "asset"
|
|
resource_id = Column(String(50), nullable=True)
|
|
old_value = Column(Text, nullable=True) # JSON for diff
|
|
new_value = Column(Text, nullable=True) # JSON for diff
|
|
|
|
# Timestamp
|
|
timestamp = Column(DateTime, nullable=False, index=True)
|
|
|
|
# Relationships
|
|
user = relationship("User", back_populates="audit_logs")
|
|
|
|
def __repr__(self):
|
|
return f"<AuditLog(event='{self.event_type}', user_id={self.user_id})>"
|