Add mass unban: DELETE /api/bans/all clears all active bans
- Send fail2ban's `unban --all` command via new `unban_all_ips()` service function; returns the count of unbanned IPs - Add `UnbanAllResponse` Pydantic model (message + count) - Add `DELETE /api/bans/all` router endpoint; handles 502 on socket error - Frontend: `bansAll` endpoint constant, `unbanAllBans()` API call, `UnbanAllResponse` type, `unbanAll` action in `useActiveBans` hook - JailsPage: "Clear All Bans" button (visible when bans > 0) with a Fluent UI confirmation Dialog before executing the operation - 7 new tests (3 service, 4 router); 440 total pass, 82% coverage
This commit is contained in:
@@ -13,6 +13,7 @@ from app.config import Settings
|
||||
from app.db import init_db
|
||||
from app.main import create_app
|
||||
from app.models.ban import ActiveBan, ActiveBanListResponse
|
||||
from app.utils.fail2ban_client import Fail2BanConnectionError
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fixtures
|
||||
@@ -270,3 +271,61 @@ class TestUnbanIp:
|
||||
)
|
||||
|
||||
assert resp.status_code == 404
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# DELETE /api/bans/all
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestUnbanAll:
|
||||
"""Tests for ``DELETE /api/bans/all``."""
|
||||
|
||||
async def test_200_clears_all_bans(self, bans_client: AsyncClient) -> None:
|
||||
"""DELETE /api/bans/all returns 200 with count when successful."""
|
||||
with patch(
|
||||
"app.routers.bans.jail_service.unban_all_ips",
|
||||
AsyncMock(return_value=3),
|
||||
):
|
||||
resp = await bans_client.request("DELETE", "/api/bans/all")
|
||||
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["count"] == 3
|
||||
assert "3" in data["message"]
|
||||
|
||||
async def test_200_with_zero_count(self, bans_client: AsyncClient) -> None:
|
||||
"""DELETE /api/bans/all returns 200 with count=0 when no bans existed."""
|
||||
with patch(
|
||||
"app.routers.bans.jail_service.unban_all_ips",
|
||||
AsyncMock(return_value=0),
|
||||
):
|
||||
resp = await bans_client.request("DELETE", "/api/bans/all")
|
||||
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["count"] == 0
|
||||
|
||||
async def test_502_when_fail2ban_unreachable(
|
||||
self, bans_client: AsyncClient
|
||||
) -> None:
|
||||
"""DELETE /api/bans/all returns 502 when fail2ban is unreachable."""
|
||||
with patch(
|
||||
"app.routers.bans.jail_service.unban_all_ips",
|
||||
AsyncMock(
|
||||
side_effect=Fail2BanConnectionError(
|
||||
"cannot connect",
|
||||
"/var/run/fail2ban/fail2ban.sock",
|
||||
)
|
||||
),
|
||||
):
|
||||
resp = await bans_client.request("DELETE", "/api/bans/all")
|
||||
|
||||
assert resp.status_code == 502
|
||||
|
||||
async def test_401_when_unauthenticated(self, bans_client: AsyncClient) -> None:
|
||||
"""DELETE /api/bans/all returns 401 without session."""
|
||||
resp = await AsyncClient(
|
||||
transport=ASGITransport(app=bans_client._transport.app), # type: ignore[attr-defined]
|
||||
base_url="http://test",
|
||||
).request("DELETE", "/api/bans/all")
|
||||
assert resp.status_code == 401
|
||||
|
||||
@@ -554,3 +554,38 @@ class TestLookupIp:
|
||||
result = await jail_service.lookup_ip(_SOCKET, "9.9.9.9")
|
||||
|
||||
assert result["currently_banned_in"] == []
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# unban_all_ips
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestUnbanAllIps:
|
||||
"""Tests for :func:`~app.services.jail_service.unban_all_ips`."""
|
||||
|
||||
async def test_unban_all_ips_returns_count(self) -> None:
|
||||
"""unban_all_ips returns the integer count from fail2ban."""
|
||||
responses = {"unban|--all": (0, 5)}
|
||||
with _patch_client(responses):
|
||||
count = await jail_service.unban_all_ips(_SOCKET)
|
||||
|
||||
assert count == 5
|
||||
|
||||
async def test_unban_all_ips_returns_zero_when_none_banned(self) -> None:
|
||||
"""unban_all_ips returns 0 when no IPs are currently banned."""
|
||||
responses = {"unban|--all": (0, 0)}
|
||||
with _patch_client(responses):
|
||||
count = await jail_service.unban_all_ips(_SOCKET)
|
||||
|
||||
assert count == 0
|
||||
|
||||
async def test_unban_all_ips_raises_on_connection_error(self) -> None:
|
||||
"""unban_all_ips propagates Fail2BanConnectionError."""
|
||||
with patch(
|
||||
"app.services.jail_service.Fail2BanClient",
|
||||
side_effect=Fail2BanConnectionError("unreachable", _SOCKET),
|
||||
):
|
||||
with pytest.raises(Fail2BanConnectionError):
|
||||
await jail_service.unban_all_ips(_SOCKET)
|
||||
|
||||
Reference in New Issue
Block a user