Centralize fail2ban metadata resolution and cache DB path discovery

This commit is contained in:
2026-04-12 19:48:33 +02:00
parent e221cd414f
commit 72488b14b2
4 changed files with 189 additions and 18 deletions

View File

@@ -0,0 +1,94 @@
"""Fail2Ban runtime metadata resolution and caching service."""
from __future__ import annotations
import asyncio
import structlog
from app.utils.fail2ban_client import (
Fail2BanClient,
Fail2BanConnectionError,
Fail2BanProtocolError,
)
log: structlog.stdlib.BoundLogger = structlog.get_logger()
class Fail2BanMetadataService:
"""Resolve and cache runtime fail2ban metadata.
This service is responsible for querying fail2ban for metadata such as the
path to the fail2ban SQLite database and caching the result so repeated
requests do not require a socket discovery round-trip.
"""
def __init__(self) -> None:
self._cache: dict[str, str] = {}
self._lock: asyncio.Lock[None] = asyncio.Lock()
async def get_db_path(self, socket_path: str, *, force_refresh: bool = False) -> str:
"""Return the cached fail2ban database path or resolve it if missing.
Args:
socket_path: The fail2ban Unix domain socket path.
force_refresh: If ``True``, attempt to resolve the DB path again even
when a cached value exists.
Returns:
The resolved fail2ban SQLite database path.
"""
cached = self._cache.get(socket_path)
if cached is not None and not force_refresh:
return cached
async with self._lock:
cached = self._cache.get(socket_path)
if cached is not None and not force_refresh:
return cached
try:
db_path = await self._resolve_db_path(socket_path)
except (Fail2BanConnectionError, Fail2BanProtocolError) as exc:
if cached is not None:
log.warning(
"fail2ban_metadata_service_fallback_to_cache",
socket_path=socket_path,
error=str(exc),
)
return cached
raise
self._cache[socket_path] = db_path
return db_path
async def _resolve_db_path(self, socket_path: str) -> str:
"""Query fail2ban for the configured database path."""
socket_timeout: float = 5.0
async with Fail2BanClient(socket_path, timeout=socket_timeout) as client:
response = await client.send(["get", "dbfile"])
if not isinstance(response, tuple) or len(response) != 2:
raise RuntimeError(f"Unexpected response from fail2ban: {response!r}")
code, data = response
if code != 0:
raise RuntimeError(f"fail2ban error code {code}: {data!r}")
if data is None:
raise RuntimeError("fail2ban has no database configured (dbfile is None)")
db_path = str(data)
log.info(
"fail2ban_metadata_service_resolved_db_path",
socket_path=socket_path,
db_path=db_path,
)
return db_path
def invalidate_db_path(self, socket_path: str) -> None:
"""Clear a cached DB path so it is re-resolved on next request."""
self._cache.pop(socket_path, None)
default_fail2ban_metadata_service = Fail2BanMetadataService()

View File

@@ -5,32 +5,24 @@ from __future__ import annotations
import json
from datetime import UTC, datetime
from app.services.fail2ban_metadata_service import default_fail2ban_metadata_service
def ts_to_iso(unix_ts: int) -> str:
"""Convert a Unix timestamp to an ISO 8601 UTC string."""
return datetime.fromtimestamp(unix_ts, tz=UTC).isoformat()
async def get_fail2ban_db_path(socket_path: str) -> str:
"""Query fail2ban for the path to its SQLite database file."""
from app.utils.fail2ban_client import Fail2BanClient # pragma: no cover
async def get_fail2ban_db_path(socket_path: str, *, force_refresh: bool = False) -> str:
"""Return the fail2ban database path, using cached metadata when available."""
return await default_fail2ban_metadata_service.get_db_path(
socket_path, force_refresh=force_refresh
)
socket_timeout: float = 5.0
async with Fail2BanClient(socket_path, timeout=socket_timeout) as client:
response = await client.send(["get", "dbfile"])
if not isinstance(response, tuple) or len(response) != 2:
raise RuntimeError(f"Unexpected response from fail2ban: {response!r}")
code, data = response
if code != 0:
raise RuntimeError(f"fail2ban error code {code}: {data!r}")
if data is None:
raise RuntimeError("fail2ban has no database configured (dbfile is None)")
return str(data)
def invalidate_fail2ban_db_path(socket_path: str) -> None:
"""Invalidate the cached fail2ban database path for the given socket."""
default_fail2ban_metadata_service.invalidate_db_path(socket_path)
def parse_data_json(raw: object) -> tuple[list[str], int]:

View File

@@ -0,0 +1,84 @@
"""Tests for the fail2ban metadata service."""
from __future__ import annotations
from unittest.mock import AsyncMock, patch
from app.services.fail2ban_metadata_service import Fail2BanMetadataService
from app.utils.fail2ban_client import Fail2BanConnectionError
async def test_get_db_path_caches_result() -> None:
"""Same socket path should be resolved only once and reused from cache."""
service = Fail2BanMetadataService()
client = AsyncMock()
client.send = AsyncMock(return_value=(0, "/tmp/fail2ban.sqlite3"))
client.__aenter__.return_value = client
client.__aexit__.return_value = None
with patch(
"app.services.fail2ban_metadata_service.Fail2BanClient",
return_value=client,
) as mock_client_cls:
result1 = await service.get_db_path("/tmp/fail2ban.sock")
result2 = await service.get_db_path("/tmp/fail2ban.sock")
assert result1 == "/tmp/fail2ban.sqlite3"
assert result2 == "/tmp/fail2ban.sqlite3"
assert mock_client_cls.call_count == 1
assert client.send.call_count == 1
async def test_get_db_path_uses_cached_path_when_refresh_fails() -> None:
"""When explicit refresh fails, the previously cached path is returned."""
service = Fail2BanMetadataService()
first_client = AsyncMock()
first_client.send = AsyncMock(return_value=(0, "/tmp/fail2ban.sqlite3"))
first_client.__aenter__.return_value = first_client
first_client.__aexit__.return_value = None
second_client = AsyncMock()
second_client.send = AsyncMock(side_effect=Fail2BanConnectionError("socket down", "/tmp/fail2ban.sock"))
second_client.__aenter__.return_value = second_client
second_client.__aexit__.return_value = None
with patch(
"app.services.fail2ban_metadata_service.Fail2BanClient",
side_effect=[first_client, second_client],
) as mock_client_cls:
initial_path = await service.get_db_path("/tmp/fail2ban.sock")
refreshed_path = await service.get_db_path(
"/tmp/fail2ban.sock",
force_refresh=True,
)
assert initial_path == "/tmp/fail2ban.sqlite3"
assert refreshed_path == "/tmp/fail2ban.sqlite3"
assert mock_client_cls.call_count == 2
assert second_client.send.call_count == 1
async def test_invalidate_db_path_forces_re_resolution() -> None:
"""Invalidating the cache forces a new metadata resolution."""
service = Fail2BanMetadataService()
first_client = AsyncMock()
first_client.send = AsyncMock(return_value=(0, "/tmp/fail2ban-v1.sqlite3"))
first_client.__aenter__.return_value = first_client
first_client.__aexit__.return_value = None
second_client = AsyncMock()
second_client.send = AsyncMock(return_value=(0, "/tmp/fail2ban-v2.sqlite3"))
second_client.__aenter__.return_value = second_client
second_client.__aexit__.return_value = None
with patch(
"app.services.fail2ban_metadata_service.Fail2BanClient",
side_effect=[first_client, second_client],
) as mock_client_cls:
first_path = await service.get_db_path("/tmp/fail2ban.sock")
service.invalidate_db_path("/tmp/fail2ban.sock")
second_path = await service.get_db_path("/tmp/fail2ban.sock")
assert first_path == "/tmp/fail2ban-v1.sqlite3"
assert second_path == "/tmp/fail2ban-v2.sqlite3"
assert mock_client_cls.call_count == 2