lukas.pupkalipinski/BanGUI
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

Remove helper indirection and import shared service helpers directly

Browse Source
This commit is contained in:
Lukas
2026-04-14 07:56:59 +02:00
parent a5674f9e4c
commit 37646e57f7
9 changed files with 96 additions and 572 deletions
Download Patch File Download Diff File Expand all files Collapse all files

109
backend/app/services/action_config_service.py
View File

@@ -23,14 +23,15 @@ from app.exceptions import (
ActionNotFoundError,
ActionReadonlyError,
ConfigWriteError,
JailNameError,
JailNotFoundInConfigError,
)
from app.helpers.config_file_helpers import (
_get_active_jail_names,
_parse_jails_sync,
from app.services.config_file_service import (
build_parser,
get_active_jail_names,
parse_jails_sync,
safe_jail_name,
)
from app.helpers.jail_helpers import reload_jails
from app.services.jail_service import reload_all
from app.models.config import (
ActionConfig,
ActionConfigUpdate,
@@ -53,85 +54,15 @@ _SOCKET_TIMEOUT: float = 10.0
# Allowlist pattern for action names used in path construction.
_SAFE_ACTION_NAME_RE: re.Pattern[str] = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")
# Allowlist pattern for jail names used in path construction.
_SAFE_JAIL_NAME_RE: re.Pattern[str] = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$")
# Sections that are not jail definitions.
_META_SECTIONS: frozenset[str] = frozenset({"INCLUDES", "DEFAULT"})
# True-ish values for the ``enabled`` key.
_TRUE_VALUES: frozenset[str] = frozenset({"true", "yes", "1"})
# False-ish values for the ``enabled`` key.
_FALSE_VALUES: frozenset[str] = frozenset({"false", "no", "0"})
# ---------------------------------------------------------------------------
# Internal helpers
# ---------------------------------------------------------------------------
def _safe_jail_name(name: str) -> str:
"""Validate *name* and return it unchanged or raise :class:`JailNameError`.
Args:
name: Proposed jail name.
Returns:
The name unchanged if valid.
Raises:
JailNameError: If *name* contains unsafe characters.
"""
if not _SAFE_JAIL_NAME_RE.match(name):
raise JailNameError(
f"Jail name {name!r} contains invalid characters. "
"Only alphanumeric characters, hyphens, underscores, and dots are "
"allowed; must start with an alphanumeric character."
)
return name
def _build_parser() -> configparser.RawConfigParser:
"""Create a :class:`configparser.RawConfigParser` for fail2ban configs.
Returns:
Parser with interpolation disabled and case-sensitive option names.
"""
parser = configparser.RawConfigParser(interpolation=None, strict=False)
# fail2ban keys are lowercase but preserve case to be safe.
parser.optionxform = str # type: ignore[assignment]
return parser
def _is_truthy(value: str) -> bool:
"""Return ``True`` if *value* is a fail2ban boolean true string.
Args:
value: Raw string from config (e.g. ``"true"``, ``"yes"``, ``"1"``).
Returns:
``True`` when the value represents enabled.
"""
return value.strip().lower() in _TRUE_VALUES
def _parse_multiline(raw: str) -> list[str]:
"""Split a multi-line INI value into individual non-blank lines.
Args:
raw: Raw multi-line string from configparser.
Returns:
List of stripped, non-empty, non-comment strings.
"""
result: list[str] = []
for line in raw.splitlines():
stripped = line.strip()
if stripped and not stripped.startswith("#"):
result.append(stripped)
return result
# ---------------------------------------------------------------------------
# Internal helpers
@@ -329,7 +260,7 @@ def _append_jail_action_sync(
local_path = jail_d / f"{jail_name}.local"
parser = _build_parser()
parser = build_parser()
if local_path.is_file():
try:
parser.read(str(local_path), encoding="utf-8")
@@ -418,7 +349,7 @@ def _remove_jail_action_sync(
if not local_path.is_file():
return
parser = _build_parser()
parser = build_parser()
try:
parser.read(str(local_path), encoding="utf-8")
except (configparser.Error, OSError) as exc:
@@ -554,8 +485,8 @@ async def list_actions(
raw_actions: list[tuple[str, str, str, bool, str]] = await run_blocking( _parse_actions_sync, action_d)
all_jails_result, active_names = await asyncio.gather(
run_blocking( _parse_jails_sync, Path(config_dir)),
_get_active_jail_names(socket_path),
run_blocking(parse_jails_sync, Path(config_dir)),
get_active_jail_names(socket_path),
)
all_jails, _source_files = all_jails_result
@@ -652,8 +583,8 @@ async def get_action(
cfg = conffile_parser.parse_action_file(content, name=base_name, filename=f"{base_name}.conf")
all_jails_result, active_names = await asyncio.gather(
run_blocking( _parse_jails_sync, Path(config_dir)),
_get_active_jail_names(socket_path),
run_blocking(parse_jails_sync, Path(config_dir)),
get_active_jail_names(socket_path),
)
all_jails, _source_files = all_jails_result
action_to_jails = _build_action_to_jails_map(all_jails, active_names)
@@ -738,7 +669,7 @@ async def update_action(
if do_reload:
try:
await reload_jails(socket_path)
await reload_all(socket_path)
except Exception as exc: # noqa: BLE001
log.warning(
"reload_after_action_update_failed",
@@ -806,7 +737,7 @@ async def create_action(
if do_reload:
try:
await reload_jails(socket_path)
await reload_all(socket_path)
except Exception as exc: # noqa: BLE001
log.warning(
"reload_after_action_create_failed",
@@ -898,11 +829,11 @@ async def assign_action_to_jail(
``action.d/``.
ConfigWriteError: If writing fails.
"""
_safe_jail_name(jail_name)
safe_jail_name(jail_name)
_safe_action_name(req.action_name)
all_jails, _src = await run_blocking( _parse_jails_sync, Path(config_dir))
all_jails, _src = await run_blocking(parse_jails_sync, Path(config_dir))
if jail_name not in all_jails:
raise JailNotFoundInConfigError(jail_name)
@@ -932,7 +863,7 @@ async def assign_action_to_jail(
if do_reload:
try:
await reload_jails(socket_path)
await reload_all(socket_path)
except Exception as exc: # noqa: BLE001
log.warning(
"reload_after_assign_action_failed",
@@ -975,11 +906,11 @@ async def remove_action_from_jail(
JailNotFoundError: If *jail_name* is not defined in any config.
ConfigWriteError: If writing fails.
"""
_safe_jail_name(jail_name)
safe_jail_name(jail_name)
_safe_action_name(action_name)
all_jails, _src = await run_blocking( _parse_jails_sync, Path(config_dir))
all_jails, _src = await run_blocking(parse_jails_sync, Path(config_dir))
if jail_name not in all_jails:
raise JailNotFoundInConfigError(jail_name)
@@ -991,7 +922,7 @@ async def remove_action_from_jail(
if do_reload:
try:
await reload_jails(socket_path)
await reload_all(socket_path)
except Exception as exc: # noqa: BLE001
log.warning(
"reload_after_remove_action_failed",