Lukas
0d5882b32f
Issue #3 - Unbounded Query Results (OOM): - get_all_archived_history() now uses keyset pagination with bounded max_rows (50k default) - Added 'id' field to records from get_archived_history() and get_archived_history_keyset() - Protocol signature updated with page_size, max_rows, last_ban_id params Issue #7 - Docker Health Check Fails: - Added curl to Dockerfile.backend runtime image - HEALTHCHECK now uses 'curl -f http://localhost:8000/api/health' - compose.prod.yml: increased start_period to 40s, timeout to 10s - Frontend healthcheck proxies to backend /api/health Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
134 lines
4.1 KiB
Python
134 lines
4.1 KiB
Python
"""Jail response mappers.
|
|
|
|
Convert domain models (from jail_service) to response models (for HTTP API).
|
|
|
|
This is the mapping layer at the router boundary, ensuring the service layer
|
|
remains independent of HTTP response shapes.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from app.models.ban import ActiveBan, JailBannedIpsResponse
|
|
from app.models.ban_domain import DomainActiveBan
|
|
from app.models.jail import (
|
|
Jail,
|
|
JailDetailResponse,
|
|
JailListResponse,
|
|
JailStatus,
|
|
JailSummary,
|
|
)
|
|
from app.models.jail_domain import (
|
|
DomainJailBannedIps,
|
|
DomainBantimeEscalation,
|
|
DomainJail,
|
|
DomainJailDetail,
|
|
DomainJailList,
|
|
DomainJailStatus,
|
|
DomainJailSummary,
|
|
)
|
|
from app.utils.pagination import create_pagination_metadata
|
|
|
|
|
|
def _map_domain_jail_status(domain: DomainJailStatus) -> JailStatus:
|
|
"""Convert domain jail status to response model."""
|
|
return JailStatus(
|
|
currently_banned=domain.currently_banned,
|
|
total_banned=domain.total_banned,
|
|
currently_failed=domain.currently_failed,
|
|
total_failed=domain.total_failed,
|
|
)
|
|
|
|
|
|
def _map_domain_bantime_escalation(domain: DomainBantimeEscalation) -> object:
|
|
"""Convert domain bantime escalation to response model."""
|
|
from app.models.config import BantimeEscalation
|
|
|
|
return BantimeEscalation(
|
|
increment=domain.increment,
|
|
factor=domain.factor,
|
|
formula=domain.formula,
|
|
multipliers=domain.multipliers,
|
|
max_time=domain.max_time,
|
|
rnd_time=domain.rnd_time,
|
|
overall_jails=domain.overall_jails,
|
|
)
|
|
|
|
|
|
def map_domain_jail_summary_to_response(domain: DomainJailSummary) -> JailSummary:
|
|
"""Convert domain jail summary to response model."""
|
|
return JailSummary(
|
|
name=domain.name,
|
|
enabled=domain.enabled,
|
|
running=domain.running,
|
|
idle=domain.idle,
|
|
backend=domain.backend,
|
|
find_time=domain.find_time,
|
|
ban_time=domain.ban_time,
|
|
max_retry=domain.max_retry,
|
|
status=_map_domain_jail_status(domain.status) if domain.status else None,
|
|
)
|
|
|
|
|
|
def map_domain_jail_list_to_response(domain_list: DomainJailList) -> JailListResponse:
|
|
"""Convert domain jail list to response model."""
|
|
return JailListResponse(
|
|
items=[map_domain_jail_summary_to_response(j) for j in domain_list.items],
|
|
total=domain_list.total,
|
|
)
|
|
|
|
|
|
def map_domain_jail_to_response(domain: DomainJail) -> Jail:
|
|
"""Convert domain jail to response model."""
|
|
return Jail(
|
|
name=domain.name,
|
|
enabled=domain.enabled,
|
|
running=domain.running,
|
|
idle=domain.idle,
|
|
backend=domain.backend,
|
|
log_paths=domain.log_paths,
|
|
fail_regex=domain.fail_regex,
|
|
ignore_regex=domain.ignore_regex,
|
|
ignore_ips=domain.ignore_ips,
|
|
date_pattern=domain.date_pattern,
|
|
log_encoding=domain.log_encoding,
|
|
find_time=domain.find_time,
|
|
ban_time=domain.ban_time,
|
|
max_retry=domain.max_retry,
|
|
actions=domain.actions,
|
|
bantime_escalation=(
|
|
_map_domain_bantime_escalation(domain.bantime_escalation)
|
|
if domain.bantime_escalation
|
|
else None
|
|
),
|
|
status=_map_domain_jail_status(domain.status) if domain.status else None,
|
|
)
|
|
|
|
|
|
def map_domain_jail_detail_to_response(domain: DomainJailDetail) -> JailDetailResponse:
|
|
"""Convert domain jail detail to response model."""
|
|
return JailDetailResponse(
|
|
jail=map_domain_jail_to_response(domain.jail),
|
|
ignore_list=domain.ignore_list,
|
|
ignore_self=domain.ignore_self,
|
|
)
|
|
|
|
|
|
def map_domain_jail_banned_ips_to_response(
|
|
domain: DomainJailBannedIps,
|
|
) -> JailBannedIpsResponse:
|
|
"""Convert domain jail banned IPs to response model."""
|
|
return JailBannedIpsResponse(
|
|
items=[
|
|
ActiveBan(
|
|
ip=ban.ip,
|
|
jail=ban.jail,
|
|
banned_at=ban.banned_at,
|
|
expires_at=ban.expires_at,
|
|
ban_count=ban.ban_count,
|
|
country=ban.country,
|
|
)
|
|
for ban in domain.items
|
|
],
|
|
pagination=create_pagination_metadata(domain.total, domain.page, domain.page_size),
|
|
)
|