Lukas
6e76711940
_is_not_found_error in jail_service did not match the concatenated form 'unknownjailexception' that fail2ban produces when it serialises UnknownJailException, so JailOperationError was raised instead of JailNotFoundError and every ban attempt in the import loop failed individually, skipping all 27 840 IPs before returning an error. Two changes: - Add 'unknownjail' to the phrase list in _is_not_found_error so that UnknownJailException is correctly mapped to JailNotFoundError. - In blocklist_service.import_source, catch JailNotFoundError explicitly and break out of the loop immediately with a warning log instead of retrying on every IP.
537 lines
21 KiB
Python
537 lines
21 KiB
Python
"""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 <HOST>"]),
|
|
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 <HOST>" 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 the reload command without error."""
|
|
with _patch_client({"reload|sshd|[]|[]": (0, "OK")}):
|
|
await jail_service.reload_jail(_SOCKET, "sshd") # should not raise
|
|
|
|
async def test_reload_all_success(self) -> None:
|
|
"""reload_all sends the reload --all command without error."""
|
|
with _patch_client({"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_operation_error_raises(self) -> None:
|
|
"""stop_jail raises JailOperationError on fail2ban error code."""
|
|
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"] == []
|