PapaTramp/security-alert-center
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: SAC 0.7.0 display_name columns and event severity overrides

Browse Source 
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
PTah
2026-06-01 10:21:43 +10:00
parent 8de5d14cfb
commit de29270a25
25 changed files with 850 additions and 222 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/api/v1/problems.py
+351 -174
View File
@@ -1,174 +1,351 @@
from datetime import datetime
from fastapi import APIRouter, Depends, HTTPException, Query
from pydantic import BaseModel
from sqlalchemy import func, select
from sqlalchemy.orm import Session, joinedload
from app.auth.jwt_auth import get_current_user
from app.database import get_db
from app.models import Event, Host, Problem, ProblemEvent
router = APIRouter(prefix="/problems", tags=["problems"])
class ProblemSummary(BaseModel):
id: int
host_id: int | None
hostname: str | None
title: str
summary: str
severity: str
status: str
rule_id: str | None
fingerprint: str
event_count: int
last_seen_at: datetime
created_at: datetime
updated_at: datetime
class ProblemListResponse(BaseModel):
items: list[ProblemSummary]
total: int
page: int
page_size: int
class ProblemEventItem(BaseModel):
id: int
event_id: str
occurred_at: datetime
type: str
severity: str
title: str
summary: str
class ProblemDetail(ProblemSummary):
events: list[ProblemEventItem]
def _problem_summary(p: Problem) -> ProblemSummary:
return ProblemSummary(
id=p.id,
host_id=p.host_id,
hostname=p.host.hostname if p.host else None,
title=p.title,
summary=p.summary,
severity=p.severity,
status=p.status,
rule_id=p.rule_id,
fingerprint=p.fingerprint,
event_count=p.event_count,
last_seen_at=p.last_seen_at,
created_at=p.created_at,
updated_at=p.updated_at,
)
@router.get("", response_model=ProblemListResponse)
def list_problems(
page: int = Query(1, ge=1),
page_size: int = Query(50, ge=1, le=200),
status: str | None = Query(None),
severity: str | None = Query(None),
host_id: int | None = Query(None),
hostname: str | None = Query(None),
db: Session = Depends(get_db),
_user: str = Depends(get_current_user),
) -> ProblemListResponse:
stmt = select(Problem).join(Host, isouter=True).options(joinedload(Problem.host))
count_stmt = select(func.count()).select_from(Problem).join(Host, isouter=True)
if status:
stmt = stmt.where(Problem.status == status)
count_stmt = count_stmt.where(Problem.status == status)
if severity:
from datetime import datetime
from fastapi import APIRouter, Depends, HTTPException, Query
from pydantic import BaseModel
from sqlalchemy import func, select
from sqlalchemy.orm import Session, joinedload
from app.auth.jwt_auth import get_current_user
from app.database import get_db
from app.models import Event, Host, Problem, ProblemEvent
router = APIRouter(prefix="/problems", tags=["problems"])
class ProblemSummary(BaseModel):
id: int
host_id: int | None
hostname: str | None
display_name: str | None = None
title: str
summary: str
severity: str
status: str
rule_id: str | None
fingerprint: str
event_count: int
last_seen_at: datetime
created_at: datetime
updated_at: datetime
class ProblemListResponse(BaseModel):
items: list[ProblemSummary]
total: int
page: int
page_size: int
class ProblemEventItem(BaseModel):
id: int
event_id: str
occurred_at: datetime
type: str
severity: str
title: str
summary: str
class ProblemDetail(ProblemSummary):
events: list[ProblemEventItem]
def _problem_summary(p: Problem) -> ProblemSummary:
return ProblemSummary(
id=p.id,
host_id=p.host_id,
hostname=p.host.hostname if p.host else None,
display_name=p.host.display_name if p.host else None,
title=p.title,
summary=p.summary,
severity=p.severity,
status=p.status,
rule_id=p.rule_id,
fingerprint=p.fingerprint,
event_count=p.event_count,
last_seen_at=p.last_seen_at,
created_at=p.created_at,
updated_at=p.updated_at,
)
@router.get("", response_model=ProblemListResponse)
def list_problems(
page: int = Query(1, ge=1),
page_size: int = Query(50, ge=1, le=200),
status: str | None = Query(None),
severity: str | None = Query(None),
host_id: int | None = Query(None),
hostname: str | None = Query(None),
db: Session = Depends(get_db),
_user: str = Depends(get_current_user),
) -> ProblemListResponse:
stmt = select(Problem).join(Host, isouter=True).options(joinedload(Problem.host))
count_stmt = select(func.count()).select_from(Problem).join(Host, isouter=True)
if status:
stmt = stmt.where(Problem.status == status)
count_stmt = count_stmt.where(Problem.status == status)
if severity:
stmt = stmt.where(Problem.severity == severity)
count_stmt = count_stmt.where(Problem.severity == severity)
if host_id is not None:
stmt = stmt.where(Problem.host_id == host_id)
count_stmt = count_stmt.where(Problem.host_id == host_id)
if hostname:
like = f"%{hostname}%"
stmt = stmt.where(Host.hostname.ilike(like) | Host.display_name.ilike(like))
count_stmt = count_stmt.where(Host.hostname.ilike(like) | Host.display_name.ilike(like))
total = db.scalar(count_stmt) or 0
rows = db.scalars(
stmt.order_by(Problem.last_seen_at.desc()).offset((page - 1) * page_size).limit(page_size)
).all()
items = [_problem_summary(p) for p in rows]
return ProblemListResponse(items=items, total=total, page=page, page_size=page_size)
@router.get("/{problem_id}", response_model=ProblemDetail)
def get_problem(
problem_id: int,
db: Session = Depends(get_db),
_user: str = Depends(get_current_user),
) -> ProblemDetail:
problem = db.scalar(
select(Problem).where(Problem.id == problem_id).options(joinedload(Problem.host))
)
if problem is None:
raise HTTPException(status_code=404, detail="Problem not found")
event_rows = db.scalars(
select(Event)
.join(ProblemEvent, ProblemEvent.event_id == Event.id)
.where(ProblemEvent.problem_id == problem_id)
.order_by(Event.occurred_at.desc())
).all()
base = _problem_summary(problem)
return ProblemDetail(
**base.model_dump(),
events=[
ProblemEventItem(
id=e.id,
event_id=e.event_id,
occurred_at=e.occurred_at,
type=e.type,
severity=e.severity,
title=e.title,
summary=e.summary,
)
for e in event_rows
],
)
@router.post("/{problem_id}/ack")
def ack_problem(
problem_id: int,
db: Session = Depends(get_db),
_user: str = Depends(get_current_user),
) -> dict:
problem = db.get(Problem, problem_id)
if problem is None:
raise HTTPException(status_code=404, detail="Problem not found")
if problem.status == "resolved":
raise HTTPException(status_code=409, detail="Problem already resolved")
problem.status = "acknowledged"
db.commit()
return {"id": problem.id, "status": problem.status}
@router.post("/{problem_id}/resolve")
def resolve_problem(
problem_id: int,
db: Session = Depends(get_db),
_user: str = Depends(get_current_user),
) -> dict:
problem = db.get(Problem, problem_id)
if problem is None:
raise HTTPException(status_code=404, detail="Problem not found")
if problem.status == "resolved":
raise HTTPException(status_code=409, detail="Problem already resolved")
problem.status = "resolved"
db.commit()
return {"id": problem.id, "status": problem.status}