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

feat: SAC 0.7.4 sidebar system stats block with UI toggle

Browse Source 
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
PTah
2026-06-02 09:15:15 +10:00
parent cf5a6f995c
commit 3053058cdf
20 changed files with 1082 additions and 255 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/alembic/versions/012_ui_settings.py
+31
View File
@@ -0,0 +1,31 @@
"""ui_settings singleton for SAC UI preferences
Revision ID: 012
Revises: 011
"""
from typing import Sequence, Union
import sqlalchemy as sa
from alembic import op
revision: str = "012"
down_revision: Union[str, None] = "011"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
op.create_table(
"ui_settings",
sa.Column("id", sa.Integer(), nullable=False),
sa.Column("show_sidebar_system_stats", sa.Boolean(), nullable=False, server_default=sa.text("true")),
sa.PrimaryKeyConstraint("id"),
)
op.execute(
"INSERT INTO ui_settings (id, show_sidebar_system_stats) VALUES (1, true)"
)
def downgrade() -> None:
op.drop_table("ui_settings")
backend/app/api/v1/auth.py
+156 -78
View File
@@ -1,78 +1,156 @@
from fastapi import APIRouter, Depends, HTTPException, Request
from pydantic import BaseModel, Field
from sqlalchemy.orm import Session
from app.auth.jwt_auth import CurrentUser, create_access_token, get_current_user
from app.config import get_settings
from app.database import get_db
from app.services.login_rate_limit import (
ensure_login_allowed,
record_login_failure,
record_login_success,
)
from app.services.user_auth import authenticate_user
router = APIRouter(prefix="/auth", tags=["auth"])
class LoginRequest(BaseModel):
username: str
password: str
class TokenResponse(BaseModel):
access_token: str
token_type: str = "bearer"
username: str
role: str
class MeResponse(BaseModel):
username: str
role: str
@router.post("/login", response_model=TokenResponse)
def login(body: LoginRequest, request: Request, db: Session = Depends(get_db)) -> TokenResponse:
settings = get_settings()
if not settings.sac_admin_password and not _has_any_user(db):
raise HTTPException(
from fastapi import APIRouter, Depends, HTTPException, Request
from pydantic import BaseModel, Field
from sqlalchemy.orm import Session
from app.auth.jwt_auth import CurrentUser, create_access_token, get_current_user
from app.config import get_settings
from app.database import get_db
from app.services.login_rate_limit import (
ensure_login_allowed,
record_login_failure,
record_login_success,
)
from app.services.user_auth import authenticate_user
router = APIRouter(prefix="/auth", tags=["auth"])
class LoginRequest(BaseModel):
username: str
password: str
class TokenResponse(BaseModel):
access_token: str
token_type: str = "bearer"
username: str
role: str
class MeResponse(BaseModel):
username: str
role: str
@router.post("/login", response_model=TokenResponse)
def login(body: LoginRequest, request: Request, db: Session = Depends(get_db)) -> TokenResponse:
settings = get_settings()
if not settings.sac_admin_password and not _has_any_user(db):
raise HTTPException(
status_code=503,
detail="SAC UI users are not configured (run alembic upgrade and set SAC_ADMIN_PASSWORD for bootstrap)",
)
ip_address = ensure_login_allowed(db, request)
user = authenticate_user(db, body.username, body.password)
if user is None:
record_login_failure(db, ip_address=ip_address, username=body.username)
db.commit()
raise HTTPException(status_code=401, detail="Invalid username or password")
record_login_success(db, ip_address=ip_address, username=user.username)
db.commit()
token = create_access_token(user.username, user.role)
return TokenResponse(
access_token=token,
username=user.username,
role=user.role,
)
@router.get("/me", response_model=MeResponse)
def me(current_user: CurrentUser = Depends(get_current_user)) -> MeResponse:
return MeResponse(username=current_user.username, role=current_user.role)
def _has_any_user(db: Session) -> bool:
from sqlalchemy import func, select
from app.models.user import User
try:
count = db.scalar(select(func.count()).select_from(User)) or 0
return count > 0
except Exception:
db.rollback()
return False
backend/app/api/v1/router.py
+2 -1
View File
@@ -1,6 +1,6 @@
from fastapi import APIRouter
from app.api.v1 import auth, dashboards, events, health, hosts, problems, settings, stream, users
from app.api.v1 import auth, dashboards, events, health, hosts, problems, settings, stream, system, users
api_router = APIRouter()
api_router.include_router(health.router)
@@ -10,5 +10,6 @@ api_router.include_router(hosts.router)
api_router.include_router(problems.router)
api_router.include_router(dashboards.router)
api_router.include_router(settings.router)
api_router.include_router(system.router)
api_router.include_router(users.router)
api_router.include_router(stream.router)
backend/app/api/v1/settings.py
+38
View File
@@ -28,6 +28,10 @@ from app.services.event_severity_overrides import (
list_severity_override_items,
replace_severity_overrides,
)
from app.services.ui_settings import (
get_effective_ui_settings,
upsert_ui_settings,
)
router = APIRouter(prefix="/settings", tags=["settings"])
@@ -392,3 +396,37 @@ def update_severity_overrides(
],
source=SOURCE_DB,
)
class UiSettingsResponse(BaseModel):
show_sidebar_system_stats: bool
source: str = Field(description="db или default")
class UiSettingsUpdate(BaseModel):
show_sidebar_system_stats: bool
@router.get("/ui", response_model=UiSettingsResponse)
def get_ui_settings(
db: Session = Depends(get_db),
_user=Depends(require_admin),
) -> UiSettingsResponse:
cfg = get_effective_ui_settings(db)
return UiSettingsResponse(
show_sidebar_system_stats=cfg.show_sidebar_system_stats,
source=cfg.source,
)
@router.put("/ui", response_model=UiSettingsResponse)
def update_ui_settings(
body: UiSettingsUpdate,
db: Session = Depends(get_db),
_user=Depends(require_admin),
) -> UiSettingsResponse:
cfg = upsert_ui_settings(db, show_sidebar_system_stats=body.show_sidebar_system_stats)
return UiSettingsResponse(
show_sidebar_system_stats=cfg.show_sidebar_system_stats,
source=cfg.source,
)
backend/app/api/v1/system.py
+30
View File
@@ -0,0 +1,30 @@
from fastapi import APIRouter, Depends
from pydantic import BaseModel, Field
from sqlalchemy.orm import Session
from app.auth.jwt_auth import get_current_user
from app.database import get_db
from app.services.system_stats import collect_system_stats
from app.services.ui_settings import get_effective_ui_settings
router = APIRouter(prefix="/system", tags=["system"])
class SystemStatsResponse(BaseModel):
visible: bool = Field(description="Показывать блок в боковой панели (глобальная настройка)")
collected_at: str
cpu_percent: float | None = None
memory: dict | None = None
disk: dict | None = None
database: dict
app: dict
@router.get("/stats", response_model=SystemStatsResponse)
def get_system_stats(
db: Session = Depends(get_db),
_user=Depends(get_current_user),
) -> SystemStatsResponse:
ui = get_effective_ui_settings(db)
payload = collect_system_stats(db)
return SystemStatsResponse(visible=ui.show_sidebar_system_stats, **payload)
backend/app/auth/jwt_auth.py
+341 -171
View File
@@ -1,171 +1,341 @@
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from jose import JWTError, jwt
from sqlalchemy import func, select
from sqlalchemy.orm import Session
from app.config import get_settings
from app.database import get_db
from app.models.user import USER_ROLE_ADMIN, User
bearer_scheme = HTTPBearer(auto_error=False)
@dataclass(frozen=True)
class CurrentUser:
username: str
role: str
@property
def is_admin(self) -> bool:
return self.role == USER_ROLE_ADMIN
def create_access_token(subject: str, role: str) -> str:
settings = get_settings()
expire = datetime.now(timezone.utc) + timedelta(minutes=settings.jwt_expire_minutes)
payload = {"sub": subject, "role": role, "exp": expire}
return jwt.encode(payload, settings.jwt_secret, algorithm=settings.jwt_algorithm)
def _user_from_db(db: Session, username: str) -> User:
normalized = username.strip()
user = db.scalar(select(User).where(func.lower(User.username) == normalized.lower()))
if user is None or not user.is_active:
raise HTTPException(status_code=401, detail="Invalid token")
return user
def _decode_current_user(token: str, db: Session | None = None) -> CurrentUser:
settings = get_settings()
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from jose import JWTError, jwt
from sqlalchemy import func, select
from sqlalchemy.orm import Session
from app.config import get_settings
from app.database import get_db
from app.models.user import USER_ROLE_ADMIN, User
bearer_scheme = HTTPBearer(auto_error=False)
@dataclass(frozen=True)
class CurrentUser:
username: str
role: str
@property
def is_admin(self) -> bool:
return self.role == USER_ROLE_ADMIN
def create_access_token(subject: str, role: str) -> str:
settings = get_settings()
expire = datetime.now(timezone.utc) + timedelta(minutes=settings.jwt_expire_minutes)
payload = {"sub": subject, "role": role, "exp": expire}
return jwt.encode(payload, settings.jwt_secret, algorithm=settings.jwt_algorithm)
def _user_from_db(db: Session, username: str) -> User:
normalized = username.strip()
user = db.scalar(select(User).where(func.lower(User.username) == normalized.lower()))
if user is None or not user.is_active:
raise HTTPException(status_code=401, detail="Invalid token")
return user
def _decode_current_user(token: str, db: Session | None = None) -> CurrentUser:
settings = get_settings()
try:
payload = jwt.decode(
token,
settings.jwt_secret,
algorithms=[settings.jwt_algorithm],
)
username = payload.get("sub")
if not username:
raise HTTPException(status_code=401, detail="Invalid token")
if db is not None:
user = _user_from_db(db, str(username))
return CurrentUser(username=user.username, role=user.role)
role = str(payload.get("role") or USER_ROLE_ADMIN)
return CurrentUser(username=str(username), role=role)
except JWTError as exc:
raise HTTPException(status_code=401, detail="Invalid token") from exc
def verify_access_token(token: str, db: Session | None = None) -> str:
"""Проверка JWT (в т.ч. query-параметр для SSE). Возвращает username."""
return _decode_current_user(token, db=db).username
def get_current_user(
credentials: HTTPAuthorizationCredentials | None = Depends(bearer_scheme),
db: Session = Depends(get_db),
) -> CurrentUser:
if credentials is None or credentials.scheme.lower() != "bearer":
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Not authenticated",
)
return _decode_current_user(credentials.credentials, db=db)
def require_admin(user: CurrentUser = Depends(get_current_user)) -> CurrentUser:
if not user.is_admin:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Admin access required",
)
return user
backend/app/models/__init__.py
+2
View File
@@ -6,6 +6,7 @@ from app.models.notification_cooldown import NotificationCooldown
from app.models.notification_policy import NotificationPolicy
from app.models.problem import Problem, ProblemEvent
from app.models.login_attempt import LoginAttempt
from app.models.ui_settings import UiSettings
from app.models.event_severity_override import EventSeverityOverride
from app.models.user import User
@@ -21,4 +22,5 @@ __all__ = [
"User",
"EventSeverityOverride",
"LoginAttempt",
"UiSettings",
]
backend/app/models/ui_settings.py
+15
View File
@@ -0,0 +1,15 @@
from sqlalchemy import Boolean
from sqlalchemy.orm import Mapped, mapped_column
from app.database import Base
UI_SETTINGS_ROW_ID = 1
class UiSettings(Base):
"""Singleton: глобальные настройки интерфейса SAC."""
__tablename__ = "ui_settings"
id: Mapped[int] = mapped_column(primary_key=True)
show_sidebar_system_stats: Mapped[bool] = mapped_column(Boolean, default=True, nullable=False)
backend/app/services/system_stats.py
+115
View File
@@ -0,0 +1,115 @@
"""Host and DB metrics for SAC sidebar (best-effort, no heavy queries)."""
from __future__ import annotations
import os
import shutil
import time
from datetime import datetime, timedelta, timezone
from pathlib import Path
from sqlalchemy import func, select, text
from sqlalchemy.orm import Session
from app.models import Event, Problem
def _disk_path() -> str:
explicit = os.environ.get("SAC_STATS_DISK_PATH", "").strip()
if explicit:
return explicit
return str(Path("/").resolve())
def _read_cpu_percent() -> float | None:
try:
import psutil # type: ignore[import-untyped]
return round(float(psutil.cpu_percent(interval=None)), 1)
except Exception:
return None
def _read_memory() -> dict[str, float | int] | None:
try:
import psutil # type: ignore[import-untyped]
vm = psutil.virtual_memory()
return {
"used_mb": int(vm.used / (1024 * 1024)),
"total_mb": int(vm.total / (1024 * 1024)),
"percent": round(float(vm.percent), 1),
}
except Exception:
return None
def _read_disk(path: str) -> dict[str, float | int | str] | None:
try:
usage = shutil.disk_usage(path)
total = int(usage.total)
used = int(usage.used)
if total <= 0:
return None
return {
"path": path,
"used_gb": round(used / (1024**3), 1),
"total_gb": round(total / (1024**3), 1),
"percent": round(100.0 * used / total, 1),
}
except Exception:
return None
def collect_system_stats(db: Session) -> dict:
collected_at = datetime.now(timezone.utc).isoformat()
disk_path = _disk_path()
db_status = "ok"
db_latency_ms: float | None = None
db_connections: int | None = None
t0 = time.perf_counter()
try:
db.execute(text("SELECT 1"))
db_latency_ms = round((time.perf_counter() - t0) * 1000, 1)
try:
db_connections = int(
db.scalar(
text(
"SELECT count(*) FROM pg_stat_activity "
"WHERE datname = current_database() AND pid <> pg_backend_pid()"
)
)
or 0
)
except Exception:
db_connections = None
except Exception:
db_status = "error"
since = datetime.now(timezone.utc) - timedelta(hours=1)
events_last_hour = 0
problems_open = 0
if db_status == "ok":
events_last_hour = int(
db.scalar(select(func.count()).select_from(Event).where(Event.received_at >= since)) or 0
)
problems_open = int(
db.scalar(select(func.count()).select_from(Problem).where(Problem.status == "open")) or 0
)
return {
"collected_at": collected_at,
"cpu_percent": _read_cpu_percent(),
"memory": _read_memory(),
"disk": _read_disk(disk_path),
"database": {
"status": db_status,
"latency_ms": db_latency_ms,
"active_connections": db_connections,
},
"app": {
"events_last_hour": events_last_hour,
"problems_open": problems_open,
},
}
backend/app/services/ui_settings.py
+30
View File
@@ -0,0 +1,30 @@
from dataclasses import dataclass
from sqlalchemy.orm import Session
from app.models.ui_settings import UI_SETTINGS_ROW_ID, UiSettings
@dataclass(frozen=True)
class UiSettingsConfig:
show_sidebar_system_stats: bool
source: str = "db"
def get_effective_ui_settings(db: Session) -> UiSettingsConfig:
row = db.get(UiSettings, UI_SETTINGS_ROW_ID)
if row is None:
return UiSettingsConfig(show_sidebar_system_stats=True, source="default")
return UiSettingsConfig(show_sidebar_system_stats=bool(row.show_sidebar_system_stats), source="db")
def upsert_ui_settings(db: Session, *, show_sidebar_system_stats: bool) -> UiSettingsConfig:
row = db.get(UiSettings, UI_SETTINGS_ROW_ID)
if row is None:
row = UiSettings(id=UI_SETTINGS_ROW_ID, show_sidebar_system_stats=show_sidebar_system_stats)
db.add(row)
else:
row.show_sidebar_system_stats = show_sidebar_system_stats
db.commit()
db.refresh(row)
return get_effective_ui_settings(db)
backend/app/version.py
+1 -1
View File
@@ -1,5 +1,5 @@
"""Единый источник версии SAC (API, health, логи, OpenAPI)."""
APP_NAME = "Security Alert Center"
APP_VERSION = "0.7.3"
APP_VERSION = "0.7.4"
APP_VERSION_LABEL = f"{APP_NAME} v.{APP_VERSION}"
backend/requirements.txt
+1
View File
@@ -10,6 +10,7 @@ jsonschema>=4.23.0,<5
passlib[bcrypt]>=1.7.4,<2
python-jose[cryptography]>=3.3.0,<4
httpx>=0.28.0,<1
psutil>=6.1.0,<8
# dev
pytest>=8.3.0,<9
backend/tests/test_health.py
+2 -2
View File
@@ -4,6 +4,6 @@ from app.version import APP_NAME, APP_VERSION, APP_VERSION_LABEL
def test_version_constants():
assert APP_VERSION == "0.7.3"
assert APP_VERSION == "0.7.4"
assert APP_NAME == "Security Alert Center"
assert APP_VERSION_LABEL == "Security Alert Center v.0.7.3"
assert APP_VERSION_LABEL == "Security Alert Center v.0.7.4"
backend/tests/test_system_stats.py
+30
View File
@@ -0,0 +1,30 @@
"""GET /api/v1/system/stats"""
from app.services.ui_settings import upsert_ui_settings
def test_system_stats_requires_auth(client):
response = client.get("/api/v1/system/stats")
assert response.status_code == 401
def test_system_stats_visible_default(jwt_headers, client):
response = client.get("/api/v1/system/stats", headers=jwt_headers)
assert response.status_code == 200
body = response.json()
assert body["visible"] is True
assert "collected_at" in body
assert "database" in body
assert body["database"]["status"] in ("ok", "error")
assert "app" in body
assert "events_last_hour" in body["app"]
assert "problems_open" in body["app"]
def test_system_stats_respects_ui_setting(jwt_headers, client, db_session):
upsert_ui_settings(db_session, show_sidebar_system_stats=False)
db_session.commit()
response = client.get("/api/v1/system/stats", headers=jwt_headers)
assert response.status_code == 200
assert response.json()["visible"] is False
backend/tests/test_ui_settings.py
+32
View File
@@ -0,0 +1,32 @@
"""GET/PUT /api/v1/settings/ui"""
from app.models.ui_settings import UI_SETTINGS_ROW_ID, UiSettings
def test_get_ui_settings_default(jwt_headers, client):
response = client.get("/api/v1/settings/ui", headers=jwt_headers)
assert response.status_code == 200
body = response.json()
assert body["show_sidebar_system_stats"] is True
assert body["source"] == "default"
def test_put_ui_settings_persists(jwt_headers, client, db_session):
response = client.put(
"/api/v1/settings/ui",
headers=jwt_headers,
json={"show_sidebar_system_stats": False},
)
assert response.status_code == 200
body = response.json()
assert body["show_sidebar_system_stats"] is False
assert body["source"] == "db"
row = db_session.get(UiSettings, UI_SETTINGS_ROW_ID)
assert row is not None
assert row.show_sidebar_system_stats is False
def test_ui_settings_requires_admin(jwt_monitor_headers, client):
response = client.get("/api/v1/settings/ui", headers=jwt_monitor_headers)
assert response.status_code == 403