mirror of
https://github.com/wahyd4/links.git
synced 2026-08-08 21:04:53 +10:00
475 lines
16 KiB
Python
475 lines
16 KiB
Python
"""
|
||
Fetch nginx ingress logs — from the Kubernetes API (production) or a local file (dev/testing).
|
||
Deduplicates by request_id so overlapping windows don't double-insert.
|
||
"""
|
||
import ipaddress
|
||
import logging
|
||
from datetime import timedelta
|
||
|
||
from django.db import transaction
|
||
from django.utils import timezone as dj_tz
|
||
|
||
from .models import NginxSettings, NginxAccessLog
|
||
from .parser import parse_lines
|
||
from .geo import enrich_geo_batch
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
_OVERLAP = 60 # extra seconds to avoid missing entries near boundaries
|
||
|
||
|
||
def fetch_and_store() -> int:
|
||
"""
|
||
Pull new log lines, parse, deduplicate, and save.
|
||
Returns the number of new rows inserted.
|
||
"""
|
||
settings = NginxSettings.get()
|
||
if not settings.enabled:
|
||
return 0
|
||
|
||
raw = _read_file(settings) if settings.log_file_path else _k8s_logs(settings)
|
||
if raw is None:
|
||
return 0
|
||
|
||
entries = parse_lines(raw)
|
||
if not entries:
|
||
_touch(settings)
|
||
return 0
|
||
|
||
excluder = _build_excluder(settings.excluded_ips)
|
||
since_seconds = settings.fetch_interval_seconds + _OVERLAP
|
||
existing_ids = set(
|
||
NginxAccessLog.objects.filter(
|
||
timestamp__gte=dj_tz.now() - timedelta(seconds=since_seconds + 10),
|
||
).values_list('request_id', flat=True)
|
||
)
|
||
|
||
new_logs, seen = [], set()
|
||
for e in entries:
|
||
if excluder(e['remote_addr']):
|
||
continue
|
||
key = e['request_id'] or (
|
||
f"{e['timestamp'].isoformat()}|{e['remote_addr']}|{e['request_uri']}"
|
||
)
|
||
if key in existing_ids or key in seen:
|
||
continue
|
||
seen.add(key)
|
||
new_logs.append(NginxAccessLog(
|
||
timestamp=e['timestamp'],
|
||
remote_addr=e['remote_addr'],
|
||
method=e['method'],
|
||
request_uri=e['request_uri'],
|
||
protocol=e['protocol'],
|
||
status=e['status'],
|
||
body_bytes_sent=e['body_bytes_sent'],
|
||
http_referer=e['http_referer'],
|
||
http_user_agent=e['http_user_agent'],
|
||
request_length=e['request_length'],
|
||
request_time=e['request_time'],
|
||
service=e['service'],
|
||
upstream_addr=e['upstream_addr'],
|
||
upstream_response_time=e['upstream_response_time'],
|
||
upstream_status=e['upstream_status'],
|
||
request_id=e['request_id'],
|
||
))
|
||
|
||
if new_logs:
|
||
with transaction.atomic():
|
||
NginxAccessLog.objects.bulk_create(new_logs, ignore_conflicts=True)
|
||
logger.info('nginxmon: inserted %d new log entries', len(new_logs))
|
||
enrich_geo_batch(new_logs)
|
||
if settings.auto_ban_enabled:
|
||
auto_ban_auth_scanners(settings)
|
||
auto_ban_php_scanners(settings)
|
||
auto_ban_404_flood(settings)
|
||
|
||
_touch(settings)
|
||
return len(new_logs)
|
||
|
||
|
||
def ingest_raw(text: str) -> int:
|
||
"""
|
||
Parse and store log lines from a raw string (used by the paste-logs UI
|
||
and the management command). Returns the number of new rows inserted.
|
||
"""
|
||
entries = parse_lines(text)
|
||
if not entries:
|
||
return 0
|
||
|
||
settings = NginxSettings.get()
|
||
excluder = _build_excluder(settings.excluded_ips)
|
||
|
||
existing_ids = set(
|
||
NginxAccessLog.objects.values_list('request_id', flat=True)
|
||
)
|
||
new_logs, seen = [], set()
|
||
for e in entries:
|
||
if excluder(e['remote_addr']):
|
||
continue
|
||
key = e['request_id'] or (
|
||
f"{e['timestamp'].isoformat()}|{e['remote_addr']}|{e['request_uri']}"
|
||
)
|
||
if key in existing_ids or key in seen:
|
||
continue
|
||
seen.add(key)
|
||
new_logs.append(NginxAccessLog(
|
||
timestamp=e['timestamp'],
|
||
remote_addr=e['remote_addr'],
|
||
method=e['method'],
|
||
request_uri=e['request_uri'],
|
||
protocol=e['protocol'],
|
||
status=e['status'],
|
||
body_bytes_sent=e['body_bytes_sent'],
|
||
http_referer=e['http_referer'],
|
||
http_user_agent=e['http_user_agent'],
|
||
request_length=e['request_length'],
|
||
request_time=e['request_time'],
|
||
service=e['service'],
|
||
upstream_addr=e['upstream_addr'],
|
||
upstream_response_time=e['upstream_response_time'],
|
||
upstream_status=e['upstream_status'],
|
||
request_id=e['request_id'],
|
||
))
|
||
|
||
if new_logs:
|
||
with transaction.atomic():
|
||
NginxAccessLog.objects.bulk_create(new_logs, ignore_conflicts=True)
|
||
logger.info('nginxmon: ingested %d log entries from raw text', len(new_logs))
|
||
enrich_geo_batch(new_logs)
|
||
if settings.auto_ban_enabled:
|
||
auto_ban_auth_scanners(settings)
|
||
auto_ban_php_scanners(settings)
|
||
auto_ban_404_flood(settings)
|
||
|
||
return len(new_logs)
|
||
|
||
|
||
# Auth-probe paths that signal someone trying to log in / brute-force OAuth
|
||
_AUTH_PROBE_PREFIXES = ('/oauth2/', '/authorize')
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Shared helpers
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def _push_bans(settings: NginxSettings, candidate_ips: list[str],
|
||
ban_source: str, make_reason, make_note,
|
||
window_start) -> list[str]:
|
||
"""
|
||
Common routine:
|
||
1. Filter out excluded / private IPs.
|
||
2. Load ConfigMap, find new offenders, write back.
|
||
3. Create BannedIP records.
|
||
4. Annotate matching log rows.
|
||
Returns list of newly banned IPs.
|
||
"""
|
||
from django.db.models import Q
|
||
from .models import BannedIP
|
||
|
||
excluder = _build_excluder(settings.excluded_ips)
|
||
candidate_ips = [ip for ip in candidate_ips
|
||
if not excluder(ip) and not _is_private_ip(ip)]
|
||
if not candidate_ips:
|
||
return []
|
||
|
||
try:
|
||
from links.mini_apps_views import _read_blocked_ips, _write_blocked_ips
|
||
blocked = _read_blocked_ips()
|
||
except Exception as exc:
|
||
logger.error('nginxmon auto_ban: could not read ConfigMap: %s', exc)
|
||
return []
|
||
|
||
already_in_db = set(
|
||
BannedIP.objects.filter(ip__in=candidate_ips).values_list('ip', flat=True)
|
||
)
|
||
newly_banned = [ip for ip in candidate_ips
|
||
if ip not in blocked and ip not in already_in_db]
|
||
if not newly_banned:
|
||
return []
|
||
|
||
# Push to ConfigMap
|
||
try:
|
||
_write_blocked_ips(blocked + newly_banned)
|
||
logger.info('nginxmon auto_ban [%s]: banned %d IPs: %s',
|
||
ban_source, len(newly_banned), newly_banned)
|
||
except Exception as exc:
|
||
logger.error('nginxmon auto_ban: could not write ConfigMap: %s', exc)
|
||
return []
|
||
|
||
# Persist BannedIP records and annotate log rows
|
||
for ip in newly_banned:
|
||
reason = make_reason(ip, window_start)
|
||
note = f'Auto banned by system. Reason: {reason}.'
|
||
|
||
# Geo from cache
|
||
from .models import IPGeoCache
|
||
geo = IPGeoCache.objects.filter(ip=ip).first()
|
||
BannedIP.objects.get_or_create(
|
||
ip=ip,
|
||
defaults=dict(
|
||
ban_source=ban_source,
|
||
reason=reason,
|
||
request_count=NginxAccessLog.objects.filter(
|
||
remote_addr=ip, timestamp__gte=window_start).count(),
|
||
country=geo.country if geo else '',
|
||
city=geo.city if geo else '',
|
||
),
|
||
)
|
||
|
||
# Annotate all matching log rows for this IP in the window
|
||
NginxAccessLog.objects.filter(
|
||
remote_addr=ip, timestamp__gte=window_start,
|
||
).update(note=note)
|
||
|
||
return newly_banned
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Detector: repeated auth-probe failures
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def auto_ban_auth_scanners(settings: NginxSettings) -> list[str]:
|
||
"""
|
||
Ban IPs with > threshold *failed* (4xx/5xx) requests to /oauth2/ or /authorize
|
||
within the rolling window.
|
||
"""
|
||
from django.db.models import Count, Q
|
||
from datetime import timedelta
|
||
|
||
window_start = dj_tz.now() - timedelta(hours=settings.auto_ban_window_hours)
|
||
|
||
probe_filter = Q()
|
||
for prefix in _AUTH_PROBE_PREFIXES:
|
||
probe_filter |= Q(request_uri__startswith=prefix)
|
||
|
||
candidates = list(
|
||
NginxAccessLog.objects
|
||
.filter(probe_filter, timestamp__gte=window_start, status__gte=400)
|
||
.values('remote_addr')
|
||
.annotate(cnt=Count('id'))
|
||
.filter(cnt__gt=settings.auto_ban_threshold)
|
||
.values_list('remote_addr', flat=True)
|
||
)
|
||
|
||
def make_reason(ip, ws):
|
||
paths = list(
|
||
NginxAccessLog.objects.filter(
|
||
probe_filter, remote_addr=ip, timestamp__gte=ws, status__gte=400,
|
||
).values_list('request_uri', flat=True)[:20]
|
||
)
|
||
hit = sorted({p for prefix in _AUTH_PROBE_PREFIXES for p in paths if p.startswith(prefix)})
|
||
return 'repeated failed auth-probe requests to: ' + (', '.join(hit) or ', '.join(_AUTH_PROBE_PREFIXES))
|
||
|
||
return _push_bans(settings, candidates, 'auto_auth_probe', make_reason, None, window_start)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Detector: PHP webshell / backdoor scanner
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def auto_ban_php_scanners(settings: NginxSettings) -> list[str]:
|
||
"""
|
||
Ban IPs probing random .php paths (classic webshell/backdoor scanners).
|
||
Threshold reuses auto_ban_threshold (default 5). Any IP making > threshold
|
||
requests to *.php paths (excluding the known auth endpoints) gets banned.
|
||
"""
|
||
from django.db.models import Count, Q
|
||
from datetime import timedelta
|
||
|
||
window_start = dj_tz.now() - timedelta(hours=settings.auto_ban_window_hours)
|
||
|
||
# Match paths ending in .php or containing .php? — but NOT auth endpoints
|
||
auth_filter = Q()
|
||
for prefix in _AUTH_PROBE_PREFIXES:
|
||
auth_filter |= Q(request_uri__startswith=prefix)
|
||
|
||
php_filter = Q(request_uri__iregex=r'\.php(\?|$|/)')
|
||
|
||
candidates = list(
|
||
NginxAccessLog.objects
|
||
.filter(php_filter, timestamp__gte=window_start)
|
||
.exclude(auth_filter)
|
||
.values('remote_addr')
|
||
.annotate(cnt=Count('id'))
|
||
.filter(cnt__gt=settings.auto_ban_threshold)
|
||
.values_list('remote_addr', flat=True)
|
||
)
|
||
|
||
def make_reason(ip, ws):
|
||
sample = list(
|
||
NginxAccessLog.objects.filter(
|
||
php_filter, remote_addr=ip, timestamp__gte=ws,
|
||
).values_list('request_uri', flat=True)[:5]
|
||
)
|
||
cnt = NginxAccessLog.objects.filter(
|
||
php_filter, remote_addr=ip, timestamp__gte=ws,
|
||
).count()
|
||
return (f'PHP webshell/backdoor scanner — {cnt} probes to .php paths, '
|
||
f'e.g.: {", ".join(sample[:3])}')
|
||
|
||
return _push_bans(settings, candidates, 'auto_php_scan', make_reason, None, window_start)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Detector: 404 flood (scraper / content scanner)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def auto_ban_404_flood(settings: NginxSettings) -> list[str]:
|
||
"""
|
||
Ban IPs generating an abnormally high number of 404s (badge scrapers,
|
||
content scanners, etc.). Threshold = 5 × auto_ban_threshold.
|
||
"""
|
||
from django.db.models import Count
|
||
from datetime import timedelta
|
||
|
||
window_start = dj_tz.now() - timedelta(hours=settings.auto_ban_window_hours)
|
||
threshold = settings.auto_ban_threshold * 5 # harsher: 25 by default
|
||
|
||
candidates = list(
|
||
NginxAccessLog.objects
|
||
.filter(status=404, timestamp__gte=window_start)
|
||
.values('remote_addr')
|
||
.annotate(cnt=Count('id'))
|
||
.filter(cnt__gt=threshold)
|
||
.values_list('remote_addr', flat=True)
|
||
)
|
||
|
||
def make_reason(ip, ws):
|
||
cnt = NginxAccessLog.objects.filter(
|
||
remote_addr=ip, status=404, timestamp__gte=ws,
|
||
).count()
|
||
sample = list(
|
||
NginxAccessLog.objects.filter(
|
||
remote_addr=ip, status=404, timestamp__gte=ws,
|
||
).values_list('request_uri', flat=True)[:3]
|
||
)
|
||
return (f'404 flood — {cnt} consecutive 404 responses, '
|
||
f'e.g.: {", ".join(sample)}')
|
||
|
||
return _push_bans(settings, candidates, 'auto_404_flood', make_reason, None, window_start)
|
||
|
||
|
||
def _k8s_logs(settings: NginxSettings) -> str | None:
|
||
"""Fetch pod logs via the Kubernetes Python client (works in-cluster and locally)."""
|
||
since = settings.fetch_interval_seconds + _OVERLAP
|
||
try:
|
||
from kubernetes import client, config
|
||
try:
|
||
config.load_incluster_config()
|
||
except config.ConfigException:
|
||
config.load_kube_config()
|
||
|
||
v1 = client.CoreV1Api()
|
||
pods = v1.list_namespaced_pod(
|
||
namespace=settings.namespace,
|
||
label_selector=settings.pod_label,
|
||
)
|
||
if not pods.items:
|
||
logger.warning('nginxmon: no pods found for selector %r in %r',
|
||
settings.pod_label, settings.namespace)
|
||
return None
|
||
|
||
lines = []
|
||
for pod in pods.items:
|
||
pod_name = pod.metadata.name
|
||
try:
|
||
log_text = v1.read_namespaced_pod_log(
|
||
name=pod_name,
|
||
namespace=settings.namespace,
|
||
container=settings.container,
|
||
since_seconds=since,
|
||
)
|
||
for line in log_text.splitlines():
|
||
if line.strip():
|
||
# Prepend "pod container" so the parser matches its expected format
|
||
lines.append(f'{pod_name} {settings.container} {line}')
|
||
except Exception as exc:
|
||
logger.warning('nginxmon: could not read logs from pod %s: %s', pod_name, exc)
|
||
|
||
return '\n'.join(lines)
|
||
|
||
except ImportError:
|
||
logger.error('nginxmon: kubernetes package not installed — run: uv add kubernetes')
|
||
return None
|
||
except Exception as exc:
|
||
logger.error('nginxmon: k8s logs error: %s', exc)
|
||
return None
|
||
|
||
|
||
def _read_file(settings: NginxSettings) -> str | None:
|
||
try:
|
||
with open(settings.log_file_path, 'r', encoding='utf-8', errors='replace') as f:
|
||
return f.read()
|
||
except FileNotFoundError:
|
||
logger.error('nginxmon: log file not found: %s', settings.log_file_path)
|
||
return None
|
||
except Exception as exc:
|
||
logger.error('nginxmon: error reading log file: %s', exc)
|
||
return None
|
||
|
||
|
||
def _touch(settings: NginxSettings):
|
||
NginxSettings.objects.filter(pk=settings.pk).update(last_fetch_at=dj_tz.now())
|
||
|
||
|
||
def cleanup_old_logs(days: int = 7):
|
||
from datetime import timedelta
|
||
cutoff = dj_tz.now() - timedelta(days=days)
|
||
deleted, _ = NginxAccessLog.objects.filter(timestamp__lt=cutoff).delete()
|
||
if deleted:
|
||
logger.info('nginxmon: pruned %d old log entries (>%d days)', deleted, days)
|
||
|
||
|
||
def _is_private_ip(ip: str) -> bool:
|
||
"""Return True for RFC-1918, loopback, link-local, and other private ranges."""
|
||
try:
|
||
return ipaddress.ip_address(ip).is_private
|
||
except ValueError:
|
||
return False
|
||
|
||
|
||
def _build_excluder(excluded_ips_text: str):
|
||
"""
|
||
Build and return a callable(ip: str) -> bool that returns True when the
|
||
given IP should be excluded from ingestion.
|
||
|
||
Supported entry formats (one per line):
|
||
192.168.1.218 exact IP
|
||
192.168.1.x wildcard — any IP whose first 3 octets match
|
||
192.168.1.0/24 CIDR range
|
||
"""
|
||
exact: set[str] = set()
|
||
wildcards: list[str] = [] # prefixes like '192.168.1.'
|
||
networks: list[ipaddress.IPv4Network | ipaddress.IPv6Network] = []
|
||
|
||
for raw_line in (excluded_ips_text or '').splitlines():
|
||
entry = raw_line.strip()
|
||
if not entry or entry.startswith('#'):
|
||
continue
|
||
if entry.endswith('.x') or entry.endswith('.*'):
|
||
# Wildcard: everything before .x/.* becomes a dotted prefix
|
||
wildcards.append(entry[:-1]) # keep trailing '.'
|
||
elif '/' in entry:
|
||
try:
|
||
networks.append(ipaddress.ip_network(entry, strict=False))
|
||
except ValueError:
|
||
logger.warning('nginxmon: invalid CIDR in excluded_ips: %r', entry)
|
||
else:
|
||
exact.add(entry)
|
||
|
||
def _is_excluded(ip: str) -> bool:
|
||
if ip in exact:
|
||
return True
|
||
for prefix in wildcards:
|
||
if ip.startswith(prefix):
|
||
return True
|
||
if networks:
|
||
try:
|
||
addr = ipaddress.ip_address(ip)
|
||
return any(addr in net for net in networks)
|
||
except ValueError:
|
||
pass
|
||
return False
|
||
|
||
return _is_excluded
|