lukas.pupkalipinski/BanGUI
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
8d9d63b86611622c208aaa6cb1c5323a719aa70c
BanGUI/backend/tests/test_services/test_config_file_service.py
Lukas 8d9d63b866 feat(stage-1): inactive jail discovery and activation 
- Backend: config_file_service.py parses jail.conf/jail.local/jail.d/*
  following fail2ban merge order; discovers jails not running in fail2ban
- Backend: 3 new API endpoints (GET /jails/inactive, POST /jails/{name}/activate,
  POST /jails/{name}/deactivate); moved /jails/inactive before /jails/{name}
  to fix route-ordering conflict
- Frontend: ActivateJailDialog component with optional parameter overrides
- Frontend: JailsTab extended with inactive jail list and InactiveJailDetail pane
- Frontend: JailsPage JailOverviewSection shows inactive jails with toggle
- Tests: 57 service tests + 16 router tests for all new endpoints (all pass)
- Docs: Features.md, Architekture.md, Tasks.md updated; Tasks 1.1-1.5 marked done
2026-03-13 15:44:36 +01:00

560 lines
21 KiB
Python
Raw Blame History

"""Tests for config_file_service — fail2ban jail config parser and activator."""
from __future__ import annotations
from pathlib import Path
from unittest.mock import AsyncMock, patch
import pytest
from app.services.config_file_service import (
JailAlreadyActiveError,
JailAlreadyInactiveError,
JailNameError,
JailNotFoundInConfigError,
_build_inactive_jail,
_ordered_config_files,
_parse_jails_sync,
_resolve_filter,
_safe_jail_name,
_write_local_override_sync,
activate_jail,
deactivate_jail,
list_inactive_jails,
)
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _write(path: Path, content: str) -> None:
"""Write text to *path*, creating parent directories if needed."""
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(content, encoding="utf-8")
# ---------------------------------------------------------------------------
# _safe_jail_name
# ---------------------------------------------------------------------------
class TestSafeJailName:
def test_valid_simple(self) -> None:
assert _safe_jail_name("sshd") == "sshd"
def test_valid_with_hyphen(self) -> None:
assert _safe_jail_name("apache-auth") == "apache-auth"
def test_valid_with_dot(self) -> None:
assert _safe_jail_name("nginx.http") == "nginx.http"
def test_valid_with_underscore(self) -> None:
assert _safe_jail_name("my_jail") == "my_jail"
def test_invalid_path_traversal(self) -> None:
with pytest.raises(JailNameError):
_safe_jail_name("../evil")
def test_invalid_slash(self) -> None:
with pytest.raises(JailNameError):
_safe_jail_name("a/b")
def test_invalid_starts_with_dash(self) -> None:
with pytest.raises(JailNameError):
_safe_jail_name("-bad")
def test_invalid_empty(self) -> None:
with pytest.raises(JailNameError):
_safe_jail_name("")
# ---------------------------------------------------------------------------
# _resolve_filter
# ---------------------------------------------------------------------------
class TestResolveFilter:
def test_name_substitution(self) -> None:
result = _resolve_filter("%(__name__)s", "sshd", "normal")
assert result == "sshd"
def test_mode_substitution(self) -> None:
result = _resolve_filter("%(__name__)s[mode=%(mode)s]", "sshd", "aggressive")
assert result == "sshd[mode=aggressive]"
def test_no_substitution_needed(self) -> None:
result = _resolve_filter("my-filter", "sshd", "normal")
assert result == "my-filter"
def test_empty_raw(self) -> None:
result = _resolve_filter("", "sshd", "normal")
assert result == ""
# ---------------------------------------------------------------------------
# _ordered_config_files
# ---------------------------------------------------------------------------
class TestOrderedConfigFiles:
def test_empty_dir(self, tmp_path: Path) -> None:
result = _ordered_config_files(tmp_path)
assert result == []
def test_jail_conf_only(self, tmp_path: Path) -> None:
_write(tmp_path / "jail.conf", "[sshd]\nenabled=true\n")
result = _ordered_config_files(tmp_path)
assert result == [tmp_path / "jail.conf"]
def test_full_merge_order(self, tmp_path: Path) -> None:
_write(tmp_path / "jail.conf", "[DEFAULT]\n")
_write(tmp_path / "jail.local", "[DEFAULT]\n")
_write(tmp_path / "jail.d" / "custom.conf", "[sshd]\n")
_write(tmp_path / "jail.d" / "custom.local", "[sshd]\n")
result = _ordered_config_files(tmp_path)
assert result[0] == tmp_path / "jail.conf"
assert result[1] == tmp_path / "jail.local"
assert result[2] == tmp_path / "jail.d" / "custom.conf"
assert result[3] == tmp_path / "jail.d" / "custom.local"
def test_jail_d_sorted_alphabetically(self, tmp_path: Path) -> None:
(tmp_path / "jail.d").mkdir()
for name in ("zzz.conf", "aaa.conf", "mmm.conf"):
_write(tmp_path / "jail.d" / name, "")
result = _ordered_config_files(tmp_path)
names = [p.name for p in result]
assert names == ["aaa.conf", "mmm.conf", "zzz.conf"]
# ---------------------------------------------------------------------------
# _parse_jails_sync
# ---------------------------------------------------------------------------
JAIL_CONF = """\
[DEFAULT]
bantime = 10m
findtime = 5m
maxretry = 5
[sshd]
enabled = true
filter = sshd
port = ssh
logpath = /var/log/auth.log
[apache-auth]
enabled = false
filter = apache-auth
port = http,https
logpath = /var/log/apache2/error.log
"""
JAIL_LOCAL = """\
[sshd]
bantime = 1h
"""
JAIL_D_CUSTOM = """\
[nginx-http-auth]
enabled = false
filter = nginx-http-auth
port = http,https
logpath = /var/log/nginx/error.log
"""
class TestParseJailsSync:
def test_parses_all_jails(self, tmp_path: Path) -> None:
_write(tmp_path / "jail.conf", JAIL_CONF)
jails, _ = _parse_jails_sync(tmp_path)
assert "sshd" in jails
assert "apache-auth" in jails
def test_enabled_flag_parsing(self, tmp_path: Path) -> None:
_write(tmp_path / "jail.conf", JAIL_CONF)
jails, _ = _parse_jails_sync(tmp_path)
assert jails["sshd"]["enabled"] == "true"
assert jails["apache-auth"]["enabled"] == "false"
def test_default_inheritance(self, tmp_path: Path) -> None:
_write(tmp_path / "jail.conf", JAIL_CONF)
jails, _ = _parse_jails_sync(tmp_path)
# DEFAULT values should flow into each jail via configparser
assert jails["sshd"]["bantime"] == "10m"
assert jails["apache-auth"]["maxretry"] == "5"
def test_local_override_wins(self, tmp_path: Path) -> None:
_write(tmp_path / "jail.conf", JAIL_CONF)
_write(tmp_path / "jail.local", JAIL_LOCAL)
jails, _ = _parse_jails_sync(tmp_path)
# jail.local overrides bantime for sshd from 10m → 1h
assert jails["sshd"]["bantime"] == "1h"
def test_jail_d_conf_included(self, tmp_path: Path) -> None:
_write(tmp_path / "jail.conf", JAIL_CONF)
_write(tmp_path / "jail.d" / "custom.conf", JAIL_D_CUSTOM)
jails, _ = _parse_jails_sync(tmp_path)
assert "nginx-http-auth" in jails
def test_source_file_tracked(self, tmp_path: Path) -> None:
_write(tmp_path / "jail.conf", JAIL_CONF)
_write(tmp_path / "jail.d" / "custom.conf", JAIL_D_CUSTOM)
_, source_files = _parse_jails_sync(tmp_path)
# sshd comes from jail.conf; nginx-http-auth from jail.d/custom.conf
assert source_files["sshd"] == str(tmp_path / "jail.conf")
assert source_files["nginx-http-auth"] == str(tmp_path / "jail.d" / "custom.conf")
def test_source_file_local_override_tracked(self, tmp_path: Path) -> None:
_write(tmp_path / "jail.conf", JAIL_CONF)
_write(tmp_path / "jail.local", JAIL_LOCAL)
_, source_files = _parse_jails_sync(tmp_path)
# jail.local defines [sshd] again → that file wins
assert source_files["sshd"] == str(tmp_path / "jail.local")
def test_default_section_excluded(self, tmp_path: Path) -> None:
_write(tmp_path / "jail.conf", JAIL_CONF)
jails, _ = _parse_jails_sync(tmp_path)
assert "DEFAULT" not in jails
def test_includes_section_excluded(self, tmp_path: Path) -> None:
content = "[INCLUDES]\nbefore = paths-debian.conf\n" + JAIL_CONF
_write(tmp_path / "jail.conf", content)
jails, _ = _parse_jails_sync(tmp_path)
assert "INCLUDES" not in jails
def test_corrupt_file_skipped_gracefully(self, tmp_path: Path) -> None:
_write(tmp_path / "jail.conf", "[[bad section\n")
# Should not raise; bad section just yields no jails
jails, _ = _parse_jails_sync(tmp_path)
assert isinstance(jails, dict)
# ---------------------------------------------------------------------------
# _build_inactive_jail
# ---------------------------------------------------------------------------
class TestBuildInactiveJail:
def test_basic_fields(self) -> None:
settings = {
"enabled": "false",
"filter": "sshd",
"port": "ssh",
"logpath": "/var/log/auth.log",
"bantime": "10m",
"findtime": "5m",
"maxretry": "5",
"action": "",
}
jail = _build_inactive_jail("sshd", settings, "/etc/fail2ban/jail.d/sshd.conf")
assert jail.name == "sshd"
assert jail.filter == "sshd"
assert jail.port == "ssh"
assert jail.logpath == ["/var/log/auth.log"]
assert jail.bantime == "10m"
assert jail.findtime == "5m"
assert jail.maxretry == 5
assert jail.enabled is False
assert "sshd.conf" in jail.source_file
def test_filter_name_substitution(self) -> None:
settings = {"enabled": "false", "filter": "%(__name__)s"}
jail = _build_inactive_jail("myservice", settings, "/etc/fail2ban/jail.conf")
assert jail.filter == "myservice"
def test_missing_optional_fields(self) -> None:
jail = _build_inactive_jail("minimal", {}, "/etc/fail2ban/jail.conf")
assert jail.filter == "minimal" # falls back to name
assert jail.port is None
assert jail.logpath == []
assert jail.bantime is None
assert jail.maxretry is None
def test_multiline_logpath(self) -> None:
settings = {"logpath": "/var/log/app.log\n/var/log/app2.log"}
jail = _build_inactive_jail("app", settings, "/etc/fail2ban/jail.conf")
assert "/var/log/app.log" in jail.logpath
assert "/var/log/app2.log" in jail.logpath
def test_multiline_actions(self) -> None:
settings = {"action": "iptables-multiport\niptables-ipset"}
jail = _build_inactive_jail("app", settings, "/etc/fail2ban/jail.conf")
assert len(jail.actions) == 2
def test_enabled_true(self) -> None:
settings = {"enabled": "true"}
jail = _build_inactive_jail("active-jail", settings, "/etc/fail2ban/jail.conf")
assert jail.enabled is True
# ---------------------------------------------------------------------------
# _write_local_override_sync
# ---------------------------------------------------------------------------
class TestWriteLocalOverrideSync:
def test_creates_local_file(self, tmp_path: Path) -> None:
_write_local_override_sync(tmp_path, "sshd", True, {})
local = tmp_path / "jail.d" / "sshd.local"
assert local.is_file()
def test_enabled_true_written(self, tmp_path: Path) -> None:
_write_local_override_sync(tmp_path, "sshd", True, {})
content = (tmp_path / "jail.d" / "sshd.local").read_text()
assert "enabled = true" in content
def test_enabled_false_written(self, tmp_path: Path) -> None:
_write_local_override_sync(tmp_path, "sshd", False, {})
content = (tmp_path / "jail.d" / "sshd.local").read_text()
assert "enabled = false" in content
def test_section_header_written(self, tmp_path: Path) -> None:
_write_local_override_sync(tmp_path, "apache-auth", True, {})
content = (tmp_path / "jail.d" / "apache-auth.local").read_text()
assert "[apache-auth]" in content
def test_override_bantime(self, tmp_path: Path) -> None:
_write_local_override_sync(tmp_path, "sshd", True, {"bantime": "1h"})
content = (tmp_path / "jail.d" / "sshd.local").read_text()
assert "bantime" in content
assert "1h" in content
def test_override_findtime(self, tmp_path: Path) -> None:
_write_local_override_sync(tmp_path, "sshd", True, {"findtime": "30m"})
content = (tmp_path / "jail.d" / "sshd.local").read_text()
assert "findtime" in content
assert "30m" in content
def test_override_maxretry(self, tmp_path: Path) -> None:
_write_local_override_sync(tmp_path, "sshd", True, {"maxretry": 3})
content = (tmp_path / "jail.d" / "sshd.local").read_text()
assert "maxretry" in content
assert "3" in content
def test_override_port(self, tmp_path: Path) -> None:
_write_local_override_sync(tmp_path, "sshd", True, {"port": "2222"})
content = (tmp_path / "jail.d" / "sshd.local").read_text()
assert "2222" in content
def test_override_logpath_list(self, tmp_path: Path) -> None:
_write_local_override_sync(
tmp_path, "sshd", True, {"logpath": ["/var/log/auth.log", "/var/log/secure"]}
)
content = (tmp_path / "jail.d" / "sshd.local").read_text()
assert "/var/log/auth.log" in content
assert "/var/log/secure" in content
def test_bang_gui_header_comment(self, tmp_path: Path) -> None:
_write_local_override_sync(tmp_path, "sshd", True, {})
content = (tmp_path / "jail.d" / "sshd.local").read_text()
assert "BanGUI" in content
def test_overwrites_existing_file(self, tmp_path: Path) -> None:
local = tmp_path / "jail.d" / "sshd.local"
local.parent.mkdir()
local.write_text("old content")
_write_local_override_sync(tmp_path, "sshd", True, {})
assert "old content" not in local.read_text()
# ---------------------------------------------------------------------------
# list_inactive_jails
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
class TestListInactiveJails:
async def test_returns_only_inactive(self, tmp_path: Path) -> None:
_write(tmp_path / "jail.conf", JAIL_CONF)
# sshd is enabled=true; apache-auth is enabled=false
with patch(
"app.services.config_file_service._get_active_jail_names",
new=AsyncMock(return_value={"sshd"}),
):
result = await list_inactive_jails(str(tmp_path), "/fake.sock")
names = [j.name for j in result.jails]
assert "sshd" not in names
assert "apache-auth" in names
async def test_total_matches_jails_count(self, tmp_path: Path) -> None:
_write(tmp_path / "jail.conf", JAIL_CONF)
with patch(
"app.services.config_file_service._get_active_jail_names",
new=AsyncMock(return_value={"sshd"}),
):
result = await list_inactive_jails(str(tmp_path), "/fake.sock")
assert result.total == len(result.jails)
async def test_empty_config_dir(self, tmp_path: Path) -> None:
with patch(
"app.services.config_file_service._get_active_jail_names",
new=AsyncMock(return_value=set()),
):
result = await list_inactive_jails(str(tmp_path), "/fake.sock")
assert result.jails == []
assert result.total == 0
async def test_all_active_returns_empty(self, tmp_path: Path) -> None:
_write(tmp_path / "jail.conf", JAIL_CONF)
with patch(
"app.services.config_file_service._get_active_jail_names",
new=AsyncMock(return_value={"sshd", "apache-auth"}),
):
result = await list_inactive_jails(str(tmp_path), "/fake.sock")
assert result.jails == []
async def test_fail2ban_unreachable_shows_all(self, tmp_path: Path) -> None:
# When fail2ban is unreachable, _get_active_jail_names returns empty set,
# so every config-defined jail appears as inactive.
_write(tmp_path / "jail.conf", JAIL_CONF)
with patch(
"app.services.config_file_service._get_active_jail_names",
new=AsyncMock(return_value=set()),
):
result = await list_inactive_jails(str(tmp_path), "/fake.sock")
names = {j.name for j in result.jails}
assert "sshd" in names
assert "apache-auth" in names
# ---------------------------------------------------------------------------
# activate_jail
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
class TestActivateJail:
async def test_activates_known_inactive_jail(self, tmp_path: Path) -> None:
_write(tmp_path / "jail.conf", JAIL_CONF)
from app.models.config import ActivateJailRequest
req = ActivateJailRequest()
with (
patch(
"app.services.config_file_service._get_active_jail_names",
new=AsyncMock(return_value=set()),
),
patch("app.services.config_file_service.jail_service") as mock_js,
):
mock_js.reload_all = AsyncMock()
result = await activate_jail(str(tmp_path), "/fake.sock", "apache-auth", req)
assert result.active is True
assert "apache-auth" in result.name
local = tmp_path / "jail.d" / "apache-auth.local"
assert local.is_file()
assert "enabled = true" in local.read_text()
async def test_raises_not_found_for_unknown_jail(self, tmp_path: Path) -> None:
_write(tmp_path / "jail.conf", JAIL_CONF)
from app.models.config import ActivateJailRequest
req = ActivateJailRequest()
with (
patch(
"app.services.config_file_service._get_active_jail_names",
new=AsyncMock(return_value=set()),
),
):
with pytest.raises(JailNotFoundInConfigError):
await activate_jail(str(tmp_path), "/fake.sock", "nonexistent", req)
async def test_raises_already_active(self, tmp_path: Path) -> None:
_write(tmp_path / "jail.conf", JAIL_CONF)
from app.models.config import ActivateJailRequest
req = ActivateJailRequest()
with (
patch(
"app.services.config_file_service._get_active_jail_names",
new=AsyncMock(return_value={"sshd"}),
),
):
with pytest.raises(JailAlreadyActiveError):
await activate_jail(str(tmp_path), "/fake.sock", "sshd", req)
async def test_raises_name_error_for_bad_name(self, tmp_path: Path) -> None:
from app.models.config import ActivateJailRequest
req = ActivateJailRequest()
with pytest.raises(JailNameError):
await activate_jail(str(tmp_path), "/fake.sock", "../etc/passwd", req)
async def test_writes_overrides_to_local_file(self, tmp_path: Path) -> None:
_write(tmp_path / "jail.conf", JAIL_CONF)
from app.models.config import ActivateJailRequest
req = ActivateJailRequest(bantime="2h", maxretry=3)
with (
patch(
"app.services.config_file_service._get_active_jail_names",
new=AsyncMock(return_value=set()),
),
patch("app.services.config_file_service.jail_service") as mock_js,
):
mock_js.reload_all = AsyncMock()
await activate_jail(str(tmp_path), "/fake.sock", "apache-auth", req)
content = (tmp_path / "jail.d" / "apache-auth.local").read_text()
assert "2h" in content
assert "3" in content
# ---------------------------------------------------------------------------
# deactivate_jail
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
class TestDeactivateJail:
async def test_deactivates_active_jail(self, tmp_path: Path) -> None:
_write(tmp_path / "jail.conf", JAIL_CONF)
with (
patch(
"app.services.config_file_service._get_active_jail_names",
new=AsyncMock(return_value={"sshd"}),
),
patch("app.services.config_file_service.jail_service") as mock_js,
):
mock_js.reload_all = AsyncMock()
result = await deactivate_jail(str(tmp_path), "/fake.sock", "sshd")
assert result.active is False
local = tmp_path / "jail.d" / "sshd.local"
assert local.is_file()
assert "enabled = false" in local.read_text()
async def test_raises_not_found(self, tmp_path: Path) -> None:
_write(tmp_path / "jail.conf", JAIL_CONF)
with (
patch(
"app.services.config_file_service._get_active_jail_names",
new=AsyncMock(return_value={"sshd"}),
),
):
with pytest.raises(JailNotFoundInConfigError):
await deactivate_jail(str(tmp_path), "/fake.sock", "nonexistent")
async def test_raises_already_inactive(self, tmp_path: Path) -> None:
_write(tmp_path / "jail.conf", JAIL_CONF)
with (
patch(
"app.services.config_file_service._get_active_jail_names",
new=AsyncMock(return_value=set()),
),
):
with pytest.raises(JailAlreadyInactiveError):
await deactivate_jail(str(tmp_path), "/fake.sock", "apache-auth")
async def test_raises_name_error(self, tmp_path: Path) -> None:
with pytest.raises(JailNameError):
await deactivate_jail(str(tmp_path), "/fake.sock", "a/b")
Reference in New Issue View Git Blame Copy Permalink