139 lines
4.6 KiB
Python
139 lines
4.6 KiB
Python
"""Audit rule management endpoints."""
|
|
from typing import Optional, List
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from sqlalchemy import select, func
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.database import get_db
|
|
from app.core.security import require_role
|
|
from app.models.user import User
|
|
from app.models.audit_rule import AuditRule
|
|
from app.schemas.audit_rule import AuditRuleCreate, AuditRuleUpdate, AuditRuleResponse
|
|
|
|
router = APIRouter(prefix="/audit-rules", tags=["审核规则"])
|
|
|
|
|
|
@router.get("", response_model=List[AuditRuleResponse])
|
|
async def list_audit_rules(
|
|
rule_type: Optional[str] = None,
|
|
severity: Optional[str] = None,
|
|
db_type: Optional[str] = None,
|
|
is_enabled: Optional[bool] = None,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(require_role("admin", "dba")),
|
|
):
|
|
stmt = select(AuditRule).order_by(AuditRule.id)
|
|
if rule_type:
|
|
stmt = stmt.where(AuditRule.rule_type == rule_type)
|
|
if severity:
|
|
stmt = stmt.where(AuditRule.severity == severity)
|
|
if db_type:
|
|
stmt = stmt.where((AuditRule.db_type == db_type) | (AuditRule.db_type == "all"))
|
|
if is_enabled is not None:
|
|
stmt = stmt.where(AuditRule.is_enabled == is_enabled)
|
|
|
|
result = await db.execute(stmt)
|
|
return result.scalars().all()
|
|
|
|
|
|
@router.post("", response_model=AuditRuleResponse, status_code=201)
|
|
async def create_audit_rule(
|
|
data: AuditRuleCreate,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(require_role("admin")),
|
|
):
|
|
exists = await db.execute(select(AuditRule).where(AuditRule.name == data.name))
|
|
if exists.scalar_one_or_none():
|
|
raise HTTPException(status_code=400, detail="规则名称已存在")
|
|
|
|
rule = AuditRule(**data.model_dump())
|
|
db.add(rule)
|
|
await db.commit()
|
|
await db.refresh(rule)
|
|
return rule
|
|
|
|
|
|
@router.get("/{rule_id}", response_model=AuditRuleResponse)
|
|
async def get_audit_rule(
|
|
rule_id: int,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(require_role("admin", "dba")),
|
|
):
|
|
result = await db.execute(select(AuditRule).where(AuditRule.id == rule_id))
|
|
rule = result.scalar_one_or_none()
|
|
if not rule:
|
|
raise HTTPException(status_code=404, detail="规则不存在")
|
|
return rule
|
|
|
|
|
|
@router.put("/{rule_id}", response_model=AuditRuleResponse)
|
|
async def update_audit_rule(
|
|
rule_id: int,
|
|
data: AuditRuleUpdate,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(require_role("admin")),
|
|
):
|
|
result = await db.execute(select(AuditRule).where(AuditRule.id == rule_id))
|
|
rule = result.scalar_one_or_none()
|
|
if not rule:
|
|
raise HTTPException(status_code=404, detail="规则不存在")
|
|
|
|
update_data = data.model_dump(exclude_unset=True)
|
|
for key, value in update_data.items():
|
|
setattr(rule, key, value)
|
|
await db.commit()
|
|
await db.refresh(rule)
|
|
return rule
|
|
|
|
|
|
@router.delete("/{rule_id}")
|
|
async def delete_audit_rule(
|
|
rule_id: int,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(require_role("admin")),
|
|
):
|
|
result = await db.execute(select(AuditRule).where(AuditRule.id == rule_id))
|
|
rule = result.scalar_one_or_none()
|
|
if not rule:
|
|
raise HTTPException(status_code=404, detail="规则不存在")
|
|
|
|
await db.delete(rule)
|
|
await db.commit()
|
|
return {"message": "规则已删除"}
|
|
|
|
|
|
@router.post("/{rule_id}/toggle", response_model=AuditRuleResponse)
|
|
async def toggle_audit_rule(
|
|
rule_id: int,
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(require_role("admin", "dba")),
|
|
):
|
|
result = await db.execute(select(AuditRule).where(AuditRule.id == rule_id))
|
|
rule = result.scalar_one_or_none()
|
|
if not rule:
|
|
raise HTTPException(status_code=404, detail="规则不存在")
|
|
|
|
rule.is_enabled = not rule.is_enabled
|
|
await db.commit()
|
|
await db.refresh(rule)
|
|
return rule
|
|
|
|
|
|
@router.get("/stats/summary")
|
|
async def rule_stats(
|
|
db: AsyncSession = Depends(get_db),
|
|
current_user: User = Depends(require_role("admin", "dba")),
|
|
):
|
|
total_stmt = select(func.count(AuditRule.id))
|
|
total = (await db.execute(total_stmt)).scalar()
|
|
|
|
enabled_stmt = select(func.count(AuditRule.id)).where(AuditRule.is_enabled == True)
|
|
enabled = (await db.execute(enabled_stmt)).scalar()
|
|
|
|
type_stmt = select(AuditRule.rule_type, func.count(AuditRule.id)).group_by(AuditRule.rule_type)
|
|
type_result = (await db.execute(type_stmt)).all()
|
|
by_type = {row[0]: row[1] for row in type_result}
|
|
|
|
return {"total": total, "enabled": enabled, "disabled": total - enabled, "by_type": by_type}
|