vulncheck
6969d0c62e
VulnCheck - Open Source Vulnerability Management for Wazuh Features: - Vulnerability management with Wazuh integration - AI-powered CVE analysis (OpenAI, Anthropic, Google, DeepSeek, Ollama, Infomaniak) - SLA policy enforcement with automated email alerts - Automated patch verification via Wazuh Syscollector - Role-based access control (Admin, Editor, Readonly) - PDF/CSV reporting for compliance workflows - Full audit trail https://gitea.isuit.ch/vulncheck/vulncheck
74 lines
2.2 KiB
Python
74 lines
2.2 KiB
Python
from typing import List, Optional
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy import desc
|
|
|
|
from app.database import get_db
|
|
from app.models.user import User, UserRole
|
|
from app.models.audit_log import AuditLog
|
|
from app.routers.auth import get_current_user
|
|
from pydantic import BaseModel
|
|
from datetime import datetime
|
|
|
|
router = APIRouter(
|
|
prefix="/audit",
|
|
tags=["audit"],
|
|
responses={404: {"description": "Not found"}},
|
|
)
|
|
|
|
class AuditLogResponse(BaseModel):
|
|
id: int
|
|
user_id: Optional[int]
|
|
username: Optional[str]
|
|
event_type: str
|
|
event_description: str
|
|
ip_address: Optional[str]
|
|
resource_type: Optional[str]
|
|
resource_id: Optional[str]
|
|
timestamp: datetime
|
|
|
|
class Config:
|
|
orm_mode = True
|
|
|
|
@router.get("/logs", response_model=List[AuditLogResponse])
|
|
async def get_audit_logs(
|
|
skip: int = 0,
|
|
limit: int = 50,
|
|
user_id: Optional[int] = None,
|
|
db: Session = Depends(get_db),
|
|
current_user: User = Depends(get_current_user)
|
|
):
|
|
"""
|
|
Get audit logs. Restricted to Admins.
|
|
"""
|
|
if current_user.role != UserRole.ADMIN:
|
|
raise HTTPException(status_code=403, detail="Not authorized to view audit logs")
|
|
|
|
query = db.query(AuditLog)
|
|
|
|
if user_id:
|
|
query = query.filter(AuditLog.user_id == user_id)
|
|
|
|
# Join with User to get usernames efficiently?
|
|
# Actually SQLAlchemy does lazy loading, but for performance joinedload is better.
|
|
# For now simple query is enough.
|
|
|
|
logs = query.order_by(desc(AuditLog.timestamp)).offset(skip).limit(limit).all()
|
|
|
|
# Map manually to handle missing users gracefully
|
|
response = []
|
|
for log in logs:
|
|
response.append({
|
|
"id": log.id,
|
|
"user_id": log.user_id,
|
|
"username": log.user.username if log.user else "System/Deleted",
|
|
"event_type": log.event_type.value if hasattr(log.event_type, 'value') else str(log.event_type),
|
|
"event_description": log.event_description,
|
|
"ip_address": log.ip_address,
|
|
"resource_type": log.resource_type,
|
|
"resource_id": log.resource_id,
|
|
"timestamp": log.timestamp
|
|
})
|
|
|
|
return response
|