fix: security page - logfmt parser and disable Caddy log fetching #3
@@ -26,25 +26,47 @@ _INTERNAL_PATHS = re.compile(
|
||||
)
|
||||
|
||||
|
||||
_LOGFMT_RE = re.compile(r'(\w+)="([^"]*)"|\b(\w+)=(\S+)')
|
||||
|
||||
|
||||
def _parse_logfmt(line: str) -> dict:
|
||||
"""Parse a logfmt line into a dict."""
|
||||
result = {}
|
||||
for m in _LOGFMT_RE.finditer(line):
|
||||
if m.group(1):
|
||||
result[m.group(1)] = m.group(2)
|
||||
elif m.group(3):
|
||||
result[m.group(3)] = m.group(4)
|
||||
return result
|
||||
|
||||
|
||||
def _parse_authelia_logs(lines: str, limit: int) -> list[dict]:
|
||||
"""Extract failed auth attempts from Authelia JSON logs."""
|
||||
"""Extract failed auth attempts from Authelia logs (logfmt or JSON)."""
|
||||
entries = []
|
||||
for line in lines.strip().splitlines():
|
||||
# Authelia outputs JSON log lines
|
||||
try:
|
||||
obj = json.loads(line.strip())
|
||||
except (json.JSONDecodeError, ValueError):
|
||||
line = line.strip()
|
||||
if not line:
|
||||
continue
|
||||
|
||||
# Try JSON first, fall back to logfmt
|
||||
try:
|
||||
obj = json.loads(line)
|
||||
except (json.JSONDecodeError, ValueError):
|
||||
obj = _parse_logfmt(line)
|
||||
if not obj:
|
||||
continue
|
||||
|
||||
level = obj.get("level", "")
|
||||
msg = obj.get("msg", "")
|
||||
|
||||
# Failed auth: level=error or warning with auth-related messages
|
||||
is_failed = (
|
||||
level in ("error", "warning")
|
||||
and any(kw in msg.lower() for kw in ("unsuccessful", "banned", "failed", "invalid", "denied"))
|
||||
)
|
||||
if not is_failed:
|
||||
# Also catch "not authorized" for anonymous access attempts
|
||||
is_unauthorized = "is not authorized" in msg.lower()
|
||||
|
||||
if not is_failed and not is_unauthorized:
|
||||
continue
|
||||
|
||||
ts_raw = obj.get("time", "")
|
||||
@@ -54,7 +76,7 @@ def _parse_authelia_logs(lines: str, limit: int) -> list[dict]:
|
||||
ts = ts_raw[:19] if ts_raw else ""
|
||||
|
||||
entries.append({
|
||||
"type": "auth_failure",
|
||||
"type": "auth_failure" if is_failed else "unauthorized",
|
||||
"timestamp": ts,
|
||||
"message": msg,
|
||||
"username": obj.get("username", obj.get("user", "")),
|
||||
|
||||
Reference in New Issue
Block a user