"""Tests for jail_service functions.""" from __future__ import annotations from typing import Any from unittest.mock import AsyncMock, patch import pytest from app.models.ban import ActiveBanListResponse from app.models.jail import JailDetailResponse, JailListResponse from app.services import jail_service from app.services.jail_service import JailNotFoundError, JailOperationError from app.utils.fail2ban_client import Fail2BanConnectionError # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- _SOCKET = "/fake/fail2ban.sock" _JAIL_NAMES = "sshd, nginx" def _make_global_status(names: str = _JAIL_NAMES) -> tuple[int, list[Any]]: return (0, [("Number of jail", 2), ("Jail list", names)]) def _make_short_status( banned: int = 2, total_banned: int = 10, failed: int = 3, total_failed: int = 20, ) -> tuple[int, list[Any]]: return ( 0, [ ("Filter", [("Currently failed", failed), ("Total failed", total_failed)]), ("Actions", [("Currently banned", banned), ("Total banned", total_banned)]), ], ) def _make_send(responses: dict[str, Any]) -> AsyncMock: """Build an ``AsyncMock`` for ``Fail2BanClient.send``. Responses are keyed by the command joined with a pipe, e.g. ``"status"`` or ``"status|sshd|short"``. """ async def _side_effect(command: list[Any]) -> Any: key = "|".join(str(c) for c in command) if key in responses: return responses[key] # Fall back to partial key matching. for resp_key, resp_value in responses.items(): if key.startswith(resp_key): return resp_value raise KeyError(f"Unexpected command key {key!r}") return AsyncMock(side_effect=_side_effect) def _patch_client(responses: dict[str, Any]) -> Any: """Return a ``patch`` context manager that mocks ``Fail2BanClient``.""" mock_send = _make_send(responses) class _FakeClient: def __init__(self, **_kw: Any) -> None: self.send = mock_send return patch("app.services.jail_service.Fail2BanClient", _FakeClient) # --------------------------------------------------------------------------- # list_jails # --------------------------------------------------------------------------- class TestListJails: """Unit tests for :func:`~app.services.jail_service.list_jails`.""" async def test_returns_jail_list_response(self) -> None: """list_jails returns a JailListResponse.""" responses = { "status": _make_global_status("sshd"), "status|sshd|short": _make_short_status(), "get|sshd|bantime": (0, 600), "get|sshd|findtime": (0, 600), "get|sshd|maxretry": (0, 5), "get|sshd|backend": (0, "polling"), "get|sshd|idle": (0, False), } with _patch_client(responses): result = await jail_service.list_jails(_SOCKET) assert isinstance(result, JailListResponse) assert result.total == 1 assert result.jails[0].name == "sshd" async def test_empty_jail_list(self) -> None: """list_jails returns empty response when no jails are active.""" responses = {"status": (0, [("Number of jail", 0), ("Jail list", "")])} with _patch_client(responses): result = await jail_service.list_jails(_SOCKET) assert result.total == 0 assert result.jails == [] async def test_jail_status_populated(self) -> None: """list_jails populates JailStatus with failed/banned counters.""" responses = { "status": _make_global_status("sshd"), "status|sshd|short": _make_short_status(banned=5, total_banned=50), "get|sshd|bantime": (0, 600), "get|sshd|findtime": (0, 600), "get|sshd|maxretry": (0, 5), "get|sshd|backend": (0, "polling"), "get|sshd|idle": (0, False), } with _patch_client(responses): result = await jail_service.list_jails(_SOCKET) jail = result.jails[0] assert jail.status is not None assert jail.status.currently_banned == 5 assert jail.status.total_banned == 50 async def test_jail_config_populated(self) -> None: """list_jails populates ban_time, find_time, max_retry, backend.""" responses = { "status": _make_global_status("sshd"), "status|sshd|short": _make_short_status(), "get|sshd|bantime": (0, 3600), "get|sshd|findtime": (0, 300), "get|sshd|maxretry": (0, 3), "get|sshd|backend": (0, "systemd"), "get|sshd|idle": (0, True), } with _patch_client(responses): result = await jail_service.list_jails(_SOCKET) jail = result.jails[0] assert jail.ban_time == 3600 assert jail.find_time == 300 assert jail.max_retry == 3 assert jail.backend == "systemd" assert jail.idle is True async def test_multiple_jails_returned(self) -> None: """list_jails fetches all jails listed in the global status.""" responses = { "status": _make_global_status("sshd, nginx"), "status|sshd|short": _make_short_status(), "status|nginx|short": _make_short_status(banned=0), "get|sshd|bantime": (0, 600), "get|sshd|findtime": (0, 600), "get|sshd|maxretry": (0, 5), "get|sshd|backend": (0, "polling"), "get|sshd|idle": (0, False), "get|nginx|bantime": (0, 1800), "get|nginx|findtime": (0, 600), "get|nginx|maxretry": (0, 5), "get|nginx|backend": (0, "polling"), "get|nginx|idle": (0, False), } with _patch_client(responses): result = await jail_service.list_jails(_SOCKET) assert result.total == 2 names = {j.name for j in result.jails} assert names == {"sshd", "nginx"} async def test_connection_error_propagates(self) -> None: """list_jails raises Fail2BanConnectionError when socket unreachable.""" async def _raise(*_: Any, **__: Any) -> None: raise Fail2BanConnectionError("no socket", _SOCKET) class _FailClient: def __init__(self, **_kw: Any) -> None: self.send = AsyncMock(side_effect=Fail2BanConnectionError("no socket", _SOCKET)) with patch("app.services.jail_service.Fail2BanClient", _FailClient), pytest.raises(Fail2BanConnectionError): await jail_service.list_jails(_SOCKET) # --------------------------------------------------------------------------- # get_jail # --------------------------------------------------------------------------- class TestGetJail: """Unit tests for :func:`~app.services.jail_service.get_jail`.""" def _full_responses(self, name: str = "sshd") -> dict[str, Any]: return { f"status|{name}|short": _make_short_status(), f"get|{name}|logpath": (0, ["/var/log/auth.log"]), f"get|{name}|failregex": (0, ["^.*Failed.*from "]), f"get|{name}|ignoreregex": (0, []), f"get|{name}|ignoreip": (0, ["127.0.0.1"]), f"get|{name}|datepattern": (0, None), f"get|{name}|logencoding": (0, "UTF-8"), f"get|{name}|bantime": (0, 600), f"get|{name}|findtime": (0, 600), f"get|{name}|maxretry": (0, 5), f"get|{name}|backend": (0, "polling"), f"get|{name}|idle": (0, False), f"get|{name}|actions": (0, ["iptables-multiport"]), } async def test_returns_jail_detail_response(self) -> None: """get_jail returns a JailDetailResponse.""" with _patch_client(self._full_responses()): result = await jail_service.get_jail(_SOCKET, "sshd") assert isinstance(result, JailDetailResponse) assert result.jail.name == "sshd" async def test_log_paths_parsed(self) -> None: """get_jail populates log_paths from fail2ban.""" with _patch_client(self._full_responses()): result = await jail_service.get_jail(_SOCKET, "sshd") assert result.jail.log_paths == ["/var/log/auth.log"] async def test_fail_regex_parsed(self) -> None: """get_jail populates fail_regex list.""" with _patch_client(self._full_responses()): result = await jail_service.get_jail(_SOCKET, "sshd") assert "^.*Failed.*from " in result.jail.fail_regex async def test_ignore_ips_parsed(self) -> None: """get_jail populates ignore_ips list.""" with _patch_client(self._full_responses()): result = await jail_service.get_jail(_SOCKET, "sshd") assert "127.0.0.1" in result.jail.ignore_ips async def test_actions_parsed(self) -> None: """get_jail populates actions list.""" with _patch_client(self._full_responses()): result = await jail_service.get_jail(_SOCKET, "sshd") assert result.jail.actions == ["iptables-multiport"] async def test_jail_not_found_raises(self) -> None: """get_jail raises JailNotFoundError when jail is unknown.""" not_found_response = (1, Exception("Unknown jail: 'ghost'")) with _patch_client({r"status|ghost|short": not_found_response}), pytest.raises(JailNotFoundError): await jail_service.get_jail(_SOCKET, "ghost") # --------------------------------------------------------------------------- # Jail control commands # --------------------------------------------------------------------------- class TestJailControls: """Unit tests for start, stop, idle, reload commands.""" async def test_start_jail_success(self) -> None: """start_jail sends the start command without error.""" with _patch_client({"start|sshd": (0, None)}): await jail_service.start_jail(_SOCKET, "sshd") # should not raise async def test_stop_jail_success(self) -> None: """stop_jail sends the stop command without error.""" with _patch_client({"stop|sshd": (0, None)}): await jail_service.stop_jail(_SOCKET, "sshd") # should not raise async def test_set_idle_on(self) -> None: """set_idle sends idle=on when on=True.""" with _patch_client({"set|sshd|idle|on": (0, True)}): await jail_service.set_idle(_SOCKET, "sshd", on=True) # should not raise async def test_set_idle_off(self) -> None: """set_idle sends idle=off when on=False.""" with _patch_client({"set|sshd|idle|off": (0, True)}): await jail_service.set_idle(_SOCKET, "sshd", on=False) # should not raise async def test_reload_jail_success(self) -> None: """reload_jail sends a reload command with a minimal start-stream.""" with _patch_client({"reload|sshd|[]|[['start', 'sshd']]": (0, "OK")}): await jail_service.reload_jail(_SOCKET, "sshd") # should not raise async def test_reload_all_success(self) -> None: """reload_all fetches jail names then sends reload --all with a start-stream.""" with _patch_client( { "status": _make_global_status("sshd, nginx"), "reload|--all|[]|[['start', 'sshd'], ['start', 'nginx']]": (0, "OK"), } ): await jail_service.reload_all(_SOCKET) # should not raise async def test_reload_all_no_jails_still_sends_reload(self) -> None: """reload_all works with an empty jail list (sends an empty stream).""" with _patch_client( { "status": (0, [("Number of jail", 0), ("Jail list", "")]), "reload|--all|[]|[]": (0, "OK"), } ): await jail_service.reload_all(_SOCKET) # should not raise async def test_start_not_found_raises(self) -> None: """start_jail raises JailNotFoundError for unknown jail.""" with _patch_client({"start|ghost": (1, Exception("Unknown jail: 'ghost'"))}), pytest.raises(JailNotFoundError): await jail_service.start_jail(_SOCKET, "ghost") async def test_stop_jail_already_stopped_is_noop(self) -> None: """stop_jail silently succeeds when the jail is not found (idempotent).""" with _patch_client({"stop|sshd": (1, Exception("UnknownJailException('sshd')"))}): await jail_service.stop_jail(_SOCKET, "sshd") # should not raise async def test_stop_operation_error_raises(self) -> None: """stop_jail raises JailOperationError on a non-not-found fail2ban error.""" with _patch_client({"stop|sshd": (1, Exception("cannot stop"))}), pytest.raises(JailOperationError): await jail_service.stop_jail(_SOCKET, "sshd") # --------------------------------------------------------------------------- # ban_ip / unban_ip # --------------------------------------------------------------------------- class TestBanUnban: """Unit tests for :func:`~app.services.jail_service.ban_ip` and :func:`~app.services.jail_service.unban_ip`. """ async def test_ban_ip_success(self) -> None: """ban_ip sends the banip command for a valid IP.""" with _patch_client({"set|sshd|banip|1.2.3.4": (0, 1)}): await jail_service.ban_ip(_SOCKET, "sshd", "1.2.3.4") # should not raise async def test_ban_ip_invalid_raises(self) -> None: """ban_ip raises ValueError for a non-IP value.""" with pytest.raises(ValueError, match="Invalid IP"): await jail_service.ban_ip(_SOCKET, "sshd", "not-an-ip") async def test_ban_ip_unknown_jail_exception_raises_jail_not_found(self) -> None: """ban_ip raises JailNotFoundError when fail2ban returns UnknownJailException. fail2ban serialises the exception without a space (``UnknownJailException`` rather than ``Unknown JailException``), so _is_not_found_error must match the concatenated form ``"unknownjail``". """ response = (1, Exception("UnknownJailException('blocklist-import')")) with ( _patch_client({"set|missing-jail|banip|1.2.3.4": response}), pytest.raises(JailNotFoundError, match="missing-jail"), ): await jail_service.ban_ip(_SOCKET, "missing-jail", "1.2.3.4") async def test_ban_ipv6_success(self) -> None: """ban_ip accepts an IPv6 address.""" with _patch_client({"set|sshd|banip|::1": (0, 1)}): await jail_service.ban_ip(_SOCKET, "sshd", "::1") # should not raise async def test_unban_ip_all_jails(self) -> None: """unban_ip with jail=None uses the global unban command.""" with _patch_client({"unban|1.2.3.4": (0, 1)}): await jail_service.unban_ip(_SOCKET, "1.2.3.4") # should not raise async def test_unban_ip_specific_jail(self) -> None: """unban_ip with a jail sends the set unbanip command.""" with _patch_client({"set|sshd|unbanip|1.2.3.4": (0, 1)}): await jail_service.unban_ip(_SOCKET, "1.2.3.4", jail="sshd") # should not raise async def test_unban_invalid_ip_raises(self) -> None: """unban_ip raises ValueError for an invalid IP.""" with pytest.raises(ValueError, match="Invalid IP"): await jail_service.unban_ip(_SOCKET, "bad-ip") # --------------------------------------------------------------------------- # get_active_bans # --------------------------------------------------------------------------- class TestGetActiveBans: """Unit tests for :func:`~app.services.jail_service.get_active_bans`.""" async def test_returns_active_ban_list_response(self) -> None: """get_active_bans returns an ActiveBanListResponse.""" responses = { "status": _make_global_status("sshd"), "get|sshd|banip|--with-time": ( 0, ["1.2.3.4 \t2025-01-01 12:00:00 + 3600 = 2025-01-01 13:00:00"], ), } with _patch_client(responses): result = await jail_service.get_active_bans(_SOCKET) assert isinstance(result, ActiveBanListResponse) assert result.total == 1 assert result.bans[0].ip == "1.2.3.4" assert result.bans[0].jail == "sshd" async def test_empty_when_no_jails(self) -> None: """get_active_bans returns empty list when no jails are active.""" responses = {"status": (0, [("Number of jail", 0), ("Jail list", "")])} with _patch_client(responses): result = await jail_service.get_active_bans(_SOCKET) assert result.total == 0 assert result.bans == [] async def test_empty_when_no_bans(self) -> None: """get_active_bans returns empty list when all jails have zero bans.""" responses = { "status": _make_global_status("sshd"), "get|sshd|banip|--with-time": (0, []), } with _patch_client(responses): result = await jail_service.get_active_bans(_SOCKET) assert result.total == 0 async def test_ban_time_parsed(self) -> None: """get_active_bans populates banned_at and expires_at from the entry.""" responses = { "status": _make_global_status("sshd"), "get|sshd|banip|--with-time": ( 0, ["10.0.0.1 \t2025-03-01 08:00:00 + 7200 = 2025-03-01 10:00:00"], ), } with _patch_client(responses): result = await jail_service.get_active_bans(_SOCKET) ban = result.bans[0] assert ban.banned_at is not None assert "2025-03-01" in ban.banned_at assert ban.expires_at is not None assert "2025-03-01" in ban.expires_at async def test_error_in_jail_tolerated(self) -> None: """get_active_bans skips a jail that errors during the ban-list fetch.""" responses = { "status": _make_global_status("sshd, nginx"), "get|sshd|banip|--with-time": ( 0, ["1.2.3.4 \t2025-01-01 10:00:00 + 600 = 2025-01-01 10:10:00"], ), "get|nginx|banip|--with-time": Fail2BanConnectionError("no nginx", _SOCKET), } async def _side(*args: Any) -> Any: key = "|".join(str(a) for a in args[0]) resp = responses.get(key) if isinstance(resp, Exception): raise resp if resp is None: raise KeyError(f"Unexpected key {key!r}") return resp class _FakeClientPartial: def __init__(self, **_kw: Any) -> None: self.send = AsyncMock(side_effect=_side) with patch("app.services.jail_service.Fail2BanClient", _FakeClientPartial): result = await jail_service.get_active_bans(_SOCKET) # Only sshd ban returned (nginx silently skipped) assert result.total == 1 assert result.bans[0].jail == "sshd" # --------------------------------------------------------------------------- # Ignore list # --------------------------------------------------------------------------- class TestIgnoreList: """Unit tests for ignore list operations.""" async def test_get_ignore_list(self) -> None: """get_ignore_list returns a list of IP strings.""" with _patch_client({"get|sshd|ignoreip": (0, ["127.0.0.1", "10.0.0.0/8"])}): result = await jail_service.get_ignore_list(_SOCKET, "sshd") assert "127.0.0.1" in result assert "10.0.0.0/8" in result async def test_add_ignore_ip(self) -> None: """add_ignore_ip sends addignoreip for a valid CIDR.""" with _patch_client({"set|sshd|addignoreip|192.168.0.0/24": (0, "OK")}): await jail_service.add_ignore_ip(_SOCKET, "sshd", "192.168.0.0/24") async def test_add_ignore_ip_invalid_raises(self) -> None: """add_ignore_ip raises ValueError for an invalid CIDR.""" with pytest.raises(ValueError, match="Invalid IP"): await jail_service.add_ignore_ip(_SOCKET, "sshd", "not-a-cidr") async def test_del_ignore_ip(self) -> None: """del_ignore_ip sends delignoreip command.""" with _patch_client({"set|sshd|delignoreip|127.0.0.1": (0, "OK")}): await jail_service.del_ignore_ip(_SOCKET, "sshd", "127.0.0.1") async def test_get_ignore_self(self) -> None: """get_ignore_self returns a boolean.""" with _patch_client({"get|sshd|ignoreself": (0, True)}): result = await jail_service.get_ignore_self(_SOCKET, "sshd") assert result is True async def test_set_ignore_self_on(self) -> None: """set_ignore_self sends ignoreself=true.""" with _patch_client({"set|sshd|ignoreself|true": (0, True)}): await jail_service.set_ignore_self(_SOCKET, "sshd", on=True) # --------------------------------------------------------------------------- # lookup_ip # --------------------------------------------------------------------------- class TestLookupIp: """Unit tests for :func:`~app.services.jail_service.lookup_ip`.""" async def test_basic_lookup(self) -> None: """lookup_ip returns currently_banned_in list.""" responses = { "get|--all|banned|1.2.3.4": (0, []), "status": _make_global_status("sshd"), "get|sshd|banip": (0, ["1.2.3.4", "5.6.7.8"]), } with _patch_client(responses): result = await jail_service.lookup_ip(_SOCKET, "1.2.3.4") assert result["ip"] == "1.2.3.4" assert "sshd" in result["currently_banned_in"] async def test_invalid_ip_raises(self) -> None: """lookup_ip raises ValueError for invalid IP.""" with pytest.raises(ValueError, match="Invalid IP"): await jail_service.lookup_ip(_SOCKET, "not-an-ip") async def test_not_banned_returns_empty_list(self) -> None: """lookup_ip returns empty currently_banned_in when IP is not banned.""" responses = { "get|--all|banned|9.9.9.9": (0, []), "status": _make_global_status("sshd"), "get|sshd|banip": (0, ["1.2.3.4"]), } with _patch_client(responses): result = await jail_service.lookup_ip(_SOCKET, "9.9.9.9") assert result["currently_banned_in"] == []