Lukas
b8f3a1c562
- history.py models: HistoryBanItem, HistoryListResponse, IpTimelineEvent, IpDetailResponse
- history_service.py: list_history() with dynamic WHERE clauses (range/jail/ip
prefix/all-time), get_ip_detail() with timeline aggregation
- history.py router: GET /api/history + GET /api/history/{ip} (404 for unknown)
- Fixed latent bug in ban_service._parse_data_json: json.loads('null') -> None
-> AttributeError; now checks isinstance(parsed, dict) before assigning obj
- 317 tests pass (27 new), ruff + mypy clean (46 files)
- types/history.ts, api/history.ts, hooks/useHistory.ts created
- HistoryPage.tsx: filter bar (time range/jail/IP), DataGrid table,
high-ban-count row highlighting, per-IP IpDetailView with timeline,
pagination
- Frontend tsc + ESLint clean (0 errors/warnings)
- Tasks.md Stage 9 marked done
342 lines
12 KiB
Python
342 lines
12 KiB
Python
"""Tests for history_service.list_history() and history_service.get_ip_detail()."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
import aiosqlite
|
|
import pytest
|
|
|
|
from app.services import history_service
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Time helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
_NOW: int = int(time.time())
|
|
_ONE_HOUR_AGO: int = _NOW - 3600
|
|
_TWO_DAYS_AGO: int = _NOW - 2 * 24 * 3600
|
|
_THIRTY_DAYS_AGO: int = _NOW - 30 * 24 * 3600
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# DB fixture helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def _create_f2b_db(path: str, rows: list[dict[str, Any]]) -> None:
|
|
"""Create a minimal fail2ban SQLite database with the given ban rows."""
|
|
async with aiosqlite.connect(path) as db:
|
|
await db.execute(
|
|
"CREATE TABLE jails ("
|
|
"name TEXT NOT NULL UNIQUE, "
|
|
"enabled INTEGER NOT NULL DEFAULT 1"
|
|
")"
|
|
)
|
|
await db.execute(
|
|
"CREATE TABLE bans ("
|
|
"jail TEXT NOT NULL, "
|
|
"ip TEXT, "
|
|
"timeofban INTEGER NOT NULL, "
|
|
"bantime INTEGER NOT NULL, "
|
|
"bancount INTEGER NOT NULL DEFAULT 1, "
|
|
"data JSON"
|
|
")"
|
|
)
|
|
for row in rows:
|
|
await db.execute(
|
|
"INSERT INTO bans (jail, ip, timeofban, bantime, bancount, data) "
|
|
"VALUES (?, ?, ?, ?, ?, ?)",
|
|
(
|
|
row["jail"],
|
|
row["ip"],
|
|
row["timeofban"],
|
|
row.get("bantime", 3600),
|
|
row.get("bancount", 1),
|
|
json.dumps(row["data"]) if "data" in row else None,
|
|
),
|
|
)
|
|
await db.commit()
|
|
|
|
|
|
@pytest.fixture
|
|
async def f2b_db_path(tmp_path: Path) -> str: # type: ignore[misc]
|
|
"""Return the path to a test fail2ban SQLite database."""
|
|
path = str(tmp_path / "fail2ban_test.sqlite3")
|
|
await _create_f2b_db(
|
|
path,
|
|
[
|
|
{
|
|
"jail": "sshd",
|
|
"ip": "1.2.3.4",
|
|
"timeofban": _ONE_HOUR_AGO,
|
|
"bantime": 3600,
|
|
"bancount": 3,
|
|
"data": {
|
|
"matches": ["Mar 1 sshd[1]: Failed password for root"],
|
|
"failures": 5,
|
|
},
|
|
},
|
|
{
|
|
"jail": "nginx",
|
|
"ip": "5.6.7.8",
|
|
"timeofban": _ONE_HOUR_AGO,
|
|
"bantime": 7200,
|
|
"bancount": 1,
|
|
"data": {"matches": ["GET /admin HTTP/1.1"], "failures": 3},
|
|
},
|
|
{
|
|
"jail": "sshd",
|
|
"ip": "1.2.3.4",
|
|
"timeofban": _TWO_DAYS_AGO,
|
|
"bantime": 3600,
|
|
"bancount": 2,
|
|
"data": {"failures": 5},
|
|
},
|
|
{
|
|
"jail": "sshd",
|
|
"ip": "9.0.0.1",
|
|
"timeofban": _THIRTY_DAYS_AGO,
|
|
"bantime": 3600,
|
|
"bancount": 1,
|
|
"data": None,
|
|
},
|
|
],
|
|
)
|
|
return path
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# list_history tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestListHistory:
|
|
"""history_service.list_history()."""
|
|
|
|
async def test_returns_all_bans_with_no_filter(
|
|
self, f2b_db_path: str
|
|
) -> None:
|
|
"""No filter returns every record in the database."""
|
|
with patch(
|
|
"app.services.history_service._get_fail2ban_db_path",
|
|
new=AsyncMock(return_value=f2b_db_path),
|
|
):
|
|
result = await history_service.list_history("fake_socket")
|
|
assert result.total == 4
|
|
assert len(result.items) == 4
|
|
|
|
async def test_time_range_filter_excludes_old_bans(
|
|
self, f2b_db_path: str
|
|
) -> None:
|
|
"""The ``range_`` filter excludes bans older than the window."""
|
|
with patch(
|
|
"app.services.history_service._get_fail2ban_db_path",
|
|
new=AsyncMock(return_value=f2b_db_path),
|
|
):
|
|
# "24h" window should include only the two recent bans
|
|
result = await history_service.list_history(
|
|
"fake_socket", range_="24h"
|
|
)
|
|
assert result.total == 2
|
|
|
|
async def test_jail_filter(self, f2b_db_path: str) -> None:
|
|
"""Jail filter restricts results to bans from that jail."""
|
|
with patch(
|
|
"app.services.history_service._get_fail2ban_db_path",
|
|
new=AsyncMock(return_value=f2b_db_path),
|
|
):
|
|
result = await history_service.list_history("fake_socket", jail="nginx")
|
|
assert result.total == 1
|
|
assert result.items[0].jail == "nginx"
|
|
|
|
async def test_ip_prefix_filter(self, f2b_db_path: str) -> None:
|
|
"""IP prefix filter restricts results to matching IPs."""
|
|
with patch(
|
|
"app.services.history_service._get_fail2ban_db_path",
|
|
new=AsyncMock(return_value=f2b_db_path),
|
|
):
|
|
result = await history_service.list_history(
|
|
"fake_socket", ip_filter="1.2.3"
|
|
)
|
|
assert result.total == 2
|
|
for item in result.items:
|
|
assert item.ip.startswith("1.2.3")
|
|
|
|
async def test_combined_filters(self, f2b_db_path: str) -> None:
|
|
"""Jail + IP prefix filters applied together narrow the result set."""
|
|
with patch(
|
|
"app.services.history_service._get_fail2ban_db_path",
|
|
new=AsyncMock(return_value=f2b_db_path),
|
|
):
|
|
result = await history_service.list_history(
|
|
"fake_socket", jail="sshd", ip_filter="1.2.3.4"
|
|
)
|
|
# 2 sshd bans for 1.2.3.4
|
|
assert result.total == 2
|
|
|
|
async def test_unknown_ip_returns_empty(self, f2b_db_path: str) -> None:
|
|
"""Filtering by a non-existent IP returns an empty result set."""
|
|
with patch(
|
|
"app.services.history_service._get_fail2ban_db_path",
|
|
new=AsyncMock(return_value=f2b_db_path),
|
|
):
|
|
result = await history_service.list_history(
|
|
"fake_socket", ip_filter="99.99.99.99"
|
|
)
|
|
assert result.total == 0
|
|
assert result.items == []
|
|
|
|
async def test_failures_extracted_from_data(
|
|
self, f2b_db_path: str
|
|
) -> None:
|
|
"""``failures`` field is parsed from the JSON ``data`` column."""
|
|
with patch(
|
|
"app.services.history_service._get_fail2ban_db_path",
|
|
new=AsyncMock(return_value=f2b_db_path),
|
|
):
|
|
result = await history_service.list_history(
|
|
"fake_socket", ip_filter="5.6.7.8"
|
|
)
|
|
assert result.total == 1
|
|
assert result.items[0].failures == 3
|
|
|
|
async def test_matches_extracted_from_data(
|
|
self, f2b_db_path: str
|
|
) -> None:
|
|
"""``matches`` list is parsed from the JSON ``data`` column."""
|
|
with patch(
|
|
"app.services.history_service._get_fail2ban_db_path",
|
|
new=AsyncMock(return_value=f2b_db_path),
|
|
):
|
|
result = await history_service.list_history(
|
|
"fake_socket", ip_filter="1.2.3.4", range_="24h"
|
|
)
|
|
# Most recent record for 1.2.3.4 has a matches list
|
|
recent = result.items[0]
|
|
assert len(recent.matches) == 1
|
|
assert "Failed password" in recent.matches[0]
|
|
|
|
async def test_null_data_column_handled_gracefully(
|
|
self, f2b_db_path: str
|
|
) -> None:
|
|
"""Records with ``data=NULL`` produce failures=0 and matches=[]."""
|
|
with patch(
|
|
"app.services.history_service._get_fail2ban_db_path",
|
|
new=AsyncMock(return_value=f2b_db_path),
|
|
):
|
|
result = await history_service.list_history(
|
|
"fake_socket", ip_filter="9.0.0.1"
|
|
)
|
|
assert result.total == 1
|
|
item = result.items[0]
|
|
assert item.failures == 0
|
|
assert item.matches == []
|
|
|
|
async def test_pagination(self, f2b_db_path: str) -> None:
|
|
"""Pagination returns the correct slice."""
|
|
with patch(
|
|
"app.services.history_service._get_fail2ban_db_path",
|
|
new=AsyncMock(return_value=f2b_db_path),
|
|
):
|
|
result = await history_service.list_history(
|
|
"fake_socket", page=1, page_size=2
|
|
)
|
|
assert result.total == 4
|
|
assert len(result.items) == 2
|
|
assert result.page == 1
|
|
assert result.page_size == 2
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# get_ip_detail tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestGetIpDetail:
|
|
"""history_service.get_ip_detail()."""
|
|
|
|
async def test_returns_none_for_unknown_ip(
|
|
self, f2b_db_path: str
|
|
) -> None:
|
|
"""Returns ``None`` when the IP has no records in the database."""
|
|
with patch(
|
|
"app.services.history_service._get_fail2ban_db_path",
|
|
new=AsyncMock(return_value=f2b_db_path),
|
|
):
|
|
result = await history_service.get_ip_detail("fake_socket", "99.99.99.99")
|
|
assert result is None
|
|
|
|
async def test_returns_ip_detail_for_known_ip(
|
|
self, f2b_db_path: str
|
|
) -> None:
|
|
"""Returns an IpDetailResponse with correct totals for a known IP."""
|
|
with patch(
|
|
"app.services.history_service._get_fail2ban_db_path",
|
|
new=AsyncMock(return_value=f2b_db_path),
|
|
):
|
|
result = await history_service.get_ip_detail("fake_socket", "1.2.3.4")
|
|
|
|
assert result is not None
|
|
assert result.ip == "1.2.3.4"
|
|
assert result.total_bans == 2
|
|
assert result.total_failures == 10 # 5 + 5
|
|
|
|
async def test_timeline_ordered_newest_first(
|
|
self, f2b_db_path: str
|
|
) -> None:
|
|
"""Timeline events are ordered newest-first."""
|
|
with patch(
|
|
"app.services.history_service._get_fail2ban_db_path",
|
|
new=AsyncMock(return_value=f2b_db_path),
|
|
):
|
|
result = await history_service.get_ip_detail("fake_socket", "1.2.3.4")
|
|
|
|
assert result is not None
|
|
assert len(result.timeline) == 2
|
|
# First event should be the most recent
|
|
assert result.timeline[0].banned_at > result.timeline[1].banned_at
|
|
|
|
async def test_last_ban_at_is_most_recent(self, f2b_db_path: str) -> None:
|
|
"""``last_ban_at`` matches the banned_at of the first timeline event."""
|
|
with patch(
|
|
"app.services.history_service._get_fail2ban_db_path",
|
|
new=AsyncMock(return_value=f2b_db_path),
|
|
):
|
|
result = await history_service.get_ip_detail("fake_socket", "1.2.3.4")
|
|
|
|
assert result is not None
|
|
assert result.last_ban_at == result.timeline[0].banned_at
|
|
|
|
async def test_geo_enrichment_applied_when_provided(
|
|
self, f2b_db_path: str
|
|
) -> None:
|
|
"""Geolocation is applied when a geo_enricher is provided."""
|
|
from app.services.geo_service import GeoInfo
|
|
|
|
mock_geo = GeoInfo(
|
|
country_code="US",
|
|
country_name="United States",
|
|
asn="AS15169",
|
|
org="Google",
|
|
)
|
|
fake_enricher = AsyncMock(return_value=mock_geo)
|
|
|
|
with patch(
|
|
"app.services.history_service._get_fail2ban_db_path",
|
|
new=AsyncMock(return_value=f2b_db_path),
|
|
):
|
|
result = await history_service.get_ip_detail(
|
|
"fake_socket", "1.2.3.4", geo_enricher=fake_enricher
|
|
)
|
|
|
|
assert result is not None
|
|
assert result.country_code == "US"
|
|
assert result.country_name == "United States"
|
|
assert result.asn == "AS15169"
|
|
assert result.org == "Google"
|