TASK-016: Validate delete_log_path query parameter with allowlist
- Extract path validation logic into shared helper function in
backend/app/utils/path_utils.py (validate_log_path)
- Refactor AddLogPathRequest to use the helper function
- Apply the same validation to DELETE /api/config/jails/{name}/logpath
endpoint by validating the log_path query parameter
- Return HTTP 422 with descriptive error if validation fails
- Add comprehensive unit tests for path validation
- Update Backend-Development.md with usage examples
This prevents path-traversal attacks on the delete_log_path endpoint
by ensuring all log paths are within allowlisted directories.
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
@@ -10,6 +10,7 @@ from typing import Literal
|
||||
from pydantic import BaseModel, ConfigDict, Field, field_validator
|
||||
|
||||
from app.config import get_settings
|
||||
from app.utils.path_utils import validate_log_path
|
||||
|
||||
DNSMode = Literal["yes", "warn", "no", "raw"]
|
||||
LogEncoding = Literal["auto", "ascii", "utf-8", "UTF-8", "latin-1"]
|
||||
@@ -295,12 +296,9 @@ class AddLogPathRequest(BaseModel):
|
||||
|
||||
@field_validator("log_path", mode="after")
|
||||
@classmethod
|
||||
def validate_log_path(cls, value: str) -> str:
|
||||
def validate_log_path_field(cls, value: str) -> str:
|
||||
"""Validate that the log path is within allowed directories.
|
||||
|
||||
Resolves the path to its canonical form (resolving symlinks) and checks
|
||||
that it is relative to one of the allowed log directories from settings.
|
||||
|
||||
Args:
|
||||
value: The log path to validate.
|
||||
|
||||
@@ -310,24 +308,7 @@ class AddLogPathRequest(BaseModel):
|
||||
Raises:
|
||||
ValueError: If the path is outside allowed log directories.
|
||||
"""
|
||||
settings = get_settings()
|
||||
try:
|
||||
resolved_path = Path(value).resolve()
|
||||
except (OSError, RuntimeError) as e:
|
||||
raise ValueError(f"Cannot resolve path {value!r}: {e}") from e
|
||||
|
||||
for allowed_dir in settings.allowed_log_dirs:
|
||||
allowed_path = Path(allowed_dir).resolve()
|
||||
try:
|
||||
resolved_path.relative_to(allowed_path)
|
||||
return value
|
||||
except ValueError:
|
||||
continue
|
||||
|
||||
allowed_dirs_str = ", ".join(settings.allowed_log_dirs)
|
||||
raise ValueError(
|
||||
f"Log path {value!r} is outside allowed directories: {allowed_dirs_str}"
|
||||
)
|
||||
return validate_log_path(value)
|
||||
|
||||
|
||||
|
||||
|
||||
@@ -3,7 +3,7 @@ from __future__ import annotations
|
||||
import shlex
|
||||
from typing import Annotated
|
||||
|
||||
from fastapi import APIRouter, Path, Query, Request, status
|
||||
from fastapi import APIRouter, HTTPException, Path, Query, Request, status
|
||||
|
||||
from app.dependencies import (
|
||||
AppDep,
|
||||
@@ -33,6 +33,7 @@ from app.services import (
|
||||
filter_config_service,
|
||||
jail_config_service,
|
||||
)
|
||||
from app.utils.path_utils import validate_log_path
|
||||
from app.utils.runtime_state import (
|
||||
clear_activation_record,
|
||||
clear_pending_recovery,
|
||||
@@ -248,10 +249,19 @@ async def delete_log_path(
|
||||
log_path: Absolute path to the log file to remove (query parameter).
|
||||
|
||||
Raises:
|
||||
HTTPException: 422 when the log path is outside allowed directories.
|
||||
HTTPException: 404 when the jail does not exist.
|
||||
HTTPException: 400 when the command is rejected.
|
||||
HTTPException: 502 when fail2ban is unreachable.
|
||||
"""
|
||||
try:
|
||||
validate_log_path(log_path)
|
||||
except ValueError as e:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
||||
detail=str(e),
|
||||
) from e
|
||||
|
||||
await config_service.delete_log_path(socket_path, name, log_path)
|
||||
|
||||
|
||||
|
||||
40
backend/app/utils/path_utils.py
Normal file
40
backend/app/utils/path_utils.py
Normal file
@@ -0,0 +1,40 @@
|
||||
"""Path validation utilities."""
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
from app.config import get_settings
|
||||
|
||||
|
||||
def validate_log_path(log_path: str) -> str:
|
||||
"""Validate that a log path is within allowed directories.
|
||||
|
||||
Resolves the path to its canonical form (resolving symlinks) and checks
|
||||
that it is relative to one of the allowed log directories from settings.
|
||||
|
||||
Args:
|
||||
log_path: The log path to validate.
|
||||
|
||||
Returns:
|
||||
The validated log path (unchanged).
|
||||
|
||||
Raises:
|
||||
ValueError: If the path is outside allowed log directories.
|
||||
"""
|
||||
settings = get_settings()
|
||||
try:
|
||||
resolved_path = Path(log_path).resolve()
|
||||
except (OSError, RuntimeError) as e:
|
||||
raise ValueError(f"Cannot resolve path {log_path!r}: {e}") from e
|
||||
|
||||
for allowed_dir in settings.allowed_log_dirs:
|
||||
allowed_path = Path(allowed_dir).resolve()
|
||||
try:
|
||||
resolved_path.relative_to(allowed_path)
|
||||
return log_path
|
||||
except ValueError:
|
||||
continue
|
||||
|
||||
allowed_dirs_str = ", ".join(settings.allowed_log_dirs)
|
||||
raise ValueError(
|
||||
f"Log path {log_path!r} is outside allowed directories: {allowed_dirs_str}"
|
||||
)
|
||||
Reference in New Issue
Block a user