feat: add dedicated Decoy Fleet inventory page and API

This commit is contained in:
2026-04-07 23:15:20 -04:00
parent 1a2ad27eca
commit eb4be44c9a
8 changed files with 261 additions and 4 deletions

View File

@@ -166,8 +166,14 @@ class StatsResponse(BaseModel):
total_logs: int
unique_attackers: int
active_deckies: int
deployed_deckies: int
@app.get("/api/v1/stats", response_model=StatsResponse)
async def get_stats(current_user: str = Depends(get_current_user)) -> dict[str, Any]:
return await repo.get_stats_summary()
@app.get("/api/v1/deckies")
async def get_deckies(current_user: str = Depends(get_current_user)) -> list[dict[str, Any]]:
return await repo.get_deckies()

View File

@@ -35,6 +35,11 @@ class BaseRepository(ABC):
"""Retrieve high-level dashboard metrics."""
pass
@abstractmethod
async def get_deckies(self) -> list[dict[str, Any]]:
"""Retrieve the list of currently deployed deckies."""
pass
@abstractmethod
async def get_user_by_username(self, username: str) -> Optional[dict[str, Any]]:
"""Retrieve a user by their username."""

View File

@@ -1,6 +1,7 @@
import aiosqlite
from typing import Any, Optional
from decnet.web.repository import BaseRepository
from decnet.config import load_state
class SQLiteRepository(BaseRepository):
@@ -128,16 +129,36 @@ class SQLiteRepository(BaseRepository):
_row = await _cursor.fetchone()
_unique_attackers: int = _row["unique_attackers"] if _row else 0
# Active deckies are those that HAVE interaction logs
async with _db.execute("SELECT COUNT(DISTINCT decky) as active_deckies FROM logs") as _cursor:
_row = await _cursor.fetchone()
_active_deckies: int = _row["active_deckies"] if _row else 0
# Deployed deckies are all those in the state file
_state = load_state()
_deployed_deckies: int = 0
if _state:
_deployed_deckies = len(_state[0].deckies)
return {
"total_logs": _total_logs,
"unique_attackers": _unique_attackers,
"active_deckies": _active_deckies
"active_deckies": _active_deckies,
"deployed_deckies": _deployed_deckies
}
async def get_deckies(self) -> list[dict[str, Any]]:
_state = load_state()
if not _state:
return []
# We can also enrich this with interaction counts/last seen from DB
_deckies: list[dict[str, Any]] = []
for _d in _state[0].deckies:
_deckies.append(_d.model_dump())
return _deckies
async def get_user_by_username(self, username: str) -> Optional[dict[str, Any]]:
async with aiosqlite.connect(self.db_path) as _db:
_db.row_factory = aiosqlite.Row