diff --git a/core/settings.py b/core/settings.py index c837795..e253962 100644 --- a/core/settings.py +++ b/core/settings.py @@ -21,6 +21,7 @@ INSTALLED_APPS = [ 'simplemde', 'markdown', # 只需要基本的markdown包 'netscan', + 'nginxmon', ] ROOT_URLCONF = 'core.urls' diff --git a/core/urls.py b/core/urls.py index 762c147..b09f086 100644 --- a/core/urls.py +++ b/core/urls.py @@ -48,6 +48,7 @@ urlpatterns = [ # Include netscan and files BEFORE links.urls to prevent the alias catch-all from intercepting them path('ui/netscan/', include('netscan.urls')), + path('ui/nginxmon/', include('nginxmon.urls')), path('ui/files/', include('links.file_urls')), # Import external image by URL — /import/images/ (also plural alias) diff --git a/data/db.sqlite3 b/data/db.sqlite3 index 0471407..4cf1b0d 100644 Binary files a/data/db.sqlite3 and b/data/db.sqlite3 differ diff --git a/justfile b/justfile index eccf33e..208f02e 100644 --- a/justfile +++ b/justfile @@ -22,16 +22,34 @@ install-browsers: # ── Local Development ───────────────────────────────────────────────────────── -# Run Django server + Tailwind CSS together (Tailwind in background) +# Run Django server + Tailwind CSS + stern nginx ingestion together dev: #!/usr/bin/env bash set -eu source .venv/bin/activate + PIDS=() + cleanup() { + echo "Stopping background processes…" + for pid in "${PIDS[@]}"; do + kill "$pid" 2>/dev/null || true + done + } + trap cleanup EXIT INT TERM + npm run dev & - TAILWIND_PID=$! - trap "kill $TAILWIND_PID 2>/dev/null" EXIT + PIDS+=($!) + + uv run manage.py nginxmon_stern & + PIDS+=($!) + uv run manage.py runserver 0.0.0.0:8000 +# Stream nginx logs via stern only (without starting Django) +stern: + #!/usr/bin/env bash + source .venv/bin/activate + uv run manage.py nginxmon_stern + # Run Tailwind CSS in watch mode (standalone) tailwind: npm run dev diff --git a/nginxmon/__init__.py b/nginxmon/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/nginxmon/admin.py b/nginxmon/admin.py new file mode 100644 index 0000000..8c08152 --- /dev/null +++ b/nginxmon/admin.py @@ -0,0 +1,39 @@ +from django.contrib import admin +from .models import NginxSettings, NginxAlertProfile, NginxAccessLog, IPGeoCache, ThreatAlert + + +@admin.register(NginxSettings) +class NginxSettingsAdmin(admin.ModelAdmin): + list_display = ['namespace', 'pod_label', 'container', 'fetch_interval_seconds', + 'log_file_path', 'enabled', 'last_fetch_at'] + + +@admin.register(NginxAlertProfile) +class NginxAlertProfileAdmin(admin.ModelAdmin): + list_display = ['name', 'alert_window_seconds', 'alert_max_requests', + 'alert_max_error_rate', 'enabled', 'created_at'] + list_filter = ['enabled'] + + +@admin.register(NginxAccessLog) +class NginxAccessLogAdmin(admin.ModelAdmin): + list_display = ['timestamp', 'remote_addr', 'method', 'request_uri', 'status', + 'request_time', 'service', 'country', 'city'] + list_filter = ['status', 'service', 'country'] + search_fields = ['remote_addr', 'request_uri', 'request_id'] + readonly_fields = ['timestamp', 'request_id'] + date_hierarchy = 'timestamp' + + +@admin.register(IPGeoCache) +class IPGeoCacheAdmin(admin.ModelAdmin): + list_display = ['ip', 'country', 'city', 'isp', 'is_private', 'fetched_at'] + search_fields = ['ip', 'country', 'city'] + + +@admin.register(ThreatAlert) +class ThreatAlertAdmin(admin.ModelAdmin): + list_display = ['detected_at', 'remote_addr', 'alert_type', 'profile', + 'request_count', 'error_count', 'country', 'notified', 'dismissed'] + list_filter = ['alert_type', 'dismissed', 'notified', 'profile'] + search_fields = ['remote_addr', 'country'] diff --git a/nginxmon/apps.py b/nginxmon/apps.py new file mode 100644 index 0000000..d4819ff --- /dev/null +++ b/nginxmon/apps.py @@ -0,0 +1,26 @@ +import logging +from django.apps import AppConfig + +logger = logging.getLogger(__name__) + + +class NginxmonConfig(AppConfig): + default_auto_field = 'django.db.models.BigAutoField' + name = 'nginxmon' + verbose_name = 'Nginx Monitor' + + def ready(self): + try: + from nginxmon.models import NginxSettings, NginxAlertProfile + from nginxmon.tasks import start_fetch_job, schedule_profile, start_cleanup_job + + settings = NginxSettings.get() + if settings.enabled: + start_fetch_job(settings.fetch_interval_seconds) + + for profile in NginxAlertProfile.objects.filter(enabled=True): + schedule_profile(profile) + + start_cleanup_job() + except Exception as exc: + logger.warning('nginxmon: could not schedule on startup: %s', exc) diff --git a/nginxmon/detector.py b/nginxmon/detector.py new file mode 100644 index 0000000..ab03e15 --- /dev/null +++ b/nginxmon/detector.py @@ -0,0 +1,86 @@ +""" +Threat detection against NginxAccessLog for each enabled NginxAlertProfile. +""" +import logging +from datetime import timedelta + +from django.db.models import Count, Q +from django.utils import timezone + +from .models import NginxAlertProfile, NginxAccessLog, ThreatAlert, IPGeoCache +from .notifications import send_alert_telegram + +logger = logging.getLogger(__name__) + + +def detect_threats(profile_id: int): + try: + profile = NginxAlertProfile.objects.get(pk=profile_id, enabled=True) + except NginxAlertProfile.DoesNotExist: + return + + window = timedelta(seconds=profile.alert_window_seconds) + now = timezone.now() + window_start = now - window + + qs = ( + NginxAccessLog.objects + .filter(timestamp__gte=window_start) + .values('remote_addr') + .annotate( + total=Count('id'), + errors=Count('id', filter=Q(status__gte=400)), + ) + ) + + for row in qs: + ip, total, errors = row['remote_addr'], row['total'], row['errors'] + + existing = ThreatAlert.objects.filter( + profile=profile, + remote_addr=ip, + window_start__gte=window_start, + dismissed=False, + ) + + if total >= profile.alert_max_requests: + if not existing.filter(alert_type='rate_limit').exists(): + _create_alert(profile, 'rate_limit', ip, total, errors, window_start, now) + + if total >= profile.alert_min_requests: + if (errors / total) >= profile.alert_max_error_rate: + if not existing.filter(alert_type='error_rate').exists(): + _create_alert(profile, 'error_rate', ip, total, errors, window_start, now) + + +def _create_alert(profile, alert_type, ip, total, errors, window_start, window_end): + geo = IPGeoCache.objects.filter(ip=ip).first() + country = geo.country if geo else '' + city = geo.city if geo else '' + + if alert_type == 'rate_limit': + detail = ( + f'{total} requests from {ip} in the last {profile.alert_window_seconds}s ' + f'(threshold: {profile.alert_max_requests})' + ) + else: + rate_pct = round(errors / total * 100) + detail = ( + f'{errors}/{total} requests from {ip} failed ' + f'({rate_pct}% error rate, threshold: {int(profile.alert_max_error_rate * 100)}%)' + ) + + alert = ThreatAlert.objects.create( + profile=profile, + alert_type=alert_type, + remote_addr=ip, + window_start=window_start, + window_end=window_end, + request_count=total, + error_count=errors, + detail=detail, + country=country, + city=city, + ) + logger.warning('nginxmon alert: %s', detail) + send_alert_telegram(profile, alert) diff --git a/nginxmon/fetcher.py b/nginxmon/fetcher.py new file mode 100644 index 0000000..2864867 --- /dev/null +++ b/nginxmon/fetcher.py @@ -0,0 +1,181 @@ +""" +Fetch nginx ingress logs — from kubectl (production) or a local file (dev/testing). +Deduplicates by request_id so overlapping windows don't double-insert. +""" +import logging +import subprocess +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 _kubectl_logs(settings) + if raw is None: + return 0 + + entries = parse_lines(raw) + if not entries: + _touch(settings) + return 0 + + 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: + 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) + + _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 + + existing_ids = set( + NginxAccessLog.objects.values_list('request_id', flat=True) + ) + new_logs, seen = [], set() + for e in entries: + 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) + + return len(new_logs) + + +def _kubectl_logs(settings: NginxSettings) -> str | None: + since = settings.fetch_interval_seconds + _OVERLAP + cmd = [ + 'kubectl', 'logs', + '-n', settings.namespace, + '-l', settings.pod_label, + '--container', settings.container, + f'--since={since}s', + '--timestamps=false', + '--prefix=true', + '--max-log-requests=10', + ] + try: + result = subprocess.run(cmd, capture_output=True, text=True, timeout=30) + if result.returncode != 0: + logger.error('kubectl logs failed: %s', result.stderr[:500]) + return None + return result.stdout + except FileNotFoundError: + logger.error('kubectl not found in PATH') + return None + except subprocess.TimeoutExpired: + logger.error('kubectl logs timed out') + return None + except Exception as exc: + logger.error('kubectl 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) diff --git a/nginxmon/forms.py b/nginxmon/forms.py new file mode 100644 index 0000000..c6a4ca6 --- /dev/null +++ b/nginxmon/forms.py @@ -0,0 +1,47 @@ +from django import forms +from .models import NginxSettings, NginxAlertProfile + + +class NginxSettingsForm(forms.ModelForm): + class Meta: + model = NginxSettings + fields = [ + 'namespace', + 'pod_label', + 'container', + 'fetch_interval_seconds', + 'log_file_path', + 'enabled', + ] + + +class NginxAlertProfileForm(forms.ModelForm): + class Meta: + model = NginxAlertProfile + fields = [ + 'name', + 'alert_window_seconds', + 'alert_max_requests', + 'alert_max_error_rate', + 'alert_min_requests', + 'telegram_bot_token', + 'telegram_chat_id', + 'enabled', + ] + widgets = { + 'telegram_bot_token': forms.PasswordInput(render_value=True), + } + help_texts = { + 'alert_max_error_rate': 'Value between 0 and 1, e.g. 0.6 = 60% errors.', + } + + +class PasteLogsForm(forms.Form): + log_text = forms.CharField( + widget=forms.Textarea(attrs={ + 'rows': 10, + 'placeholder': 'Paste raw nginx ingress log lines here…', + 'class': 'font-mono text-xs', + }), + label='Raw log lines', + ) diff --git a/nginxmon/geo.py b/nginxmon/geo.py new file mode 100644 index 0000000..336de51 --- /dev/null +++ b/nginxmon/geo.py @@ -0,0 +1,100 @@ +""" +Geo-IP enrichment using ip-api.com batch endpoint (free, no key required). +Results are cached in IPGeoCache to minimise outbound requests. +""" +import ipaddress +import logging + +import requests + +from .models import IPGeoCache, NginxAccessLog + +logger = logging.getLogger(__name__) + +_BATCH_URL = 'http://ip-api.com/batch' +_BATCH_FIELDS = 'country,countryCode,regionName,city,lat,lon,isp,status,query' +_PRIVATE_RANGES = ( + ipaddress.ip_network('10.0.0.0/8'), + ipaddress.ip_network('172.16.0.0/12'), + ipaddress.ip_network('192.168.0.0/16'), + ipaddress.ip_network('127.0.0.0/8'), + ipaddress.ip_network('::1/128'), + ipaddress.ip_network('fc00::/7'), +) + + +def _is_private(ip: str) -> bool: + try: + addr = ipaddress.ip_address(ip) + return any(addr in net for net in _PRIVATE_RANGES) + except ValueError: + return False + + +def _lookup_batch(ips: list[str]) -> dict[str, dict]: + """Query ip-api.com for up to 100 IPs. Returns {ip: geo_dict}.""" + payload = [{'query': ip, 'fields': _BATCH_FIELDS} for ip in ips[:100]] + try: + resp = requests.post(_BATCH_URL, json=payload, timeout=10) + resp.raise_for_status() + results = {} + for item in resp.json(): + q = item.get('query', '') + if item.get('status') == 'success': + results[q] = item + return results + except Exception as exc: + logger.warning('ip-api.com batch lookup failed: %s', exc) + return {} + + +def enrich_geo_batch(log_objs: list[NginxAccessLog]): + """ + Look up geo data for IPs in `log_objs`, update cache and log records. + Uses the local cache first; only queries the API for unknown IPs. + """ + unique_ips = {obj.remote_addr for obj in log_objs} + + # Split private vs public + private_ips = {ip for ip in unique_ips if _is_private(ip)} + public_ips = unique_ips - private_ips + + # Ensure private IPs are in cache + for ip in private_ips: + IPGeoCache.objects.get_or_create(ip=ip, defaults={'is_private': True}) + + # Load cached public IPs + cached = { + g.ip: g + for g in IPGeoCache.objects.filter(ip__in=public_ips) + } + missing = [ip for ip in public_ips if ip not in cached] + + # Fetch missing from API in batches of 100 + api_results: dict[str, dict] = {} + for i in range(0, len(missing), 100): + api_results.update(_lookup_batch(missing[i:i + 100])) + + # Upsert cache + for ip, data in api_results.items(): + obj, _ = IPGeoCache.objects.get_or_create(ip=ip) + obj.country = data.get('country', '') + obj.country_code = data.get('countryCode', '') + obj.region = data.get('regionName', '') + obj.city = data.get('city', '') + obj.lat = data.get('lat') + obj.lon = data.get('lon') + obj.isp = data.get('isp', '') + obj.save() + cached[ip] = obj + + # Apply geo back via per-IP queryset update (avoids needing PKs on in-memory objects) + for ip, geo in cached.items(): + NginxAccessLog.objects.filter( + remote_addr=ip, country='' + ).update( + country=geo.country or '', + country_code=geo.country_code or '', + region=geo.region or '', + city=geo.city or '', + ) diff --git a/nginxmon/management/__init__.py b/nginxmon/management/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/nginxmon/management/commands/__init__.py b/nginxmon/management/commands/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/nginxmon/management/commands/nginxmon_ingest.py b/nginxmon/management/commands/nginxmon_ingest.py new file mode 100644 index 0000000..630130a --- /dev/null +++ b/nginxmon/management/commands/nginxmon_ingest.py @@ -0,0 +1,58 @@ +""" +Management command to ingest nginx log lines into the database. + +Usage examples: + + # Read from a file + python manage.py nginxmon_ingest --file /path/to/nginx.log + + # Pipe from stdin + kubectl logs -n ingress-nginx deploy/ingress-nginx-controller | python manage.py nginxmon_ingest + + # Use the log file configured in NginxSettings + python manage.py nginxmon_ingest --from-settings +""" +import sys + +from django.core.management.base import BaseCommand + +from nginxmon.fetcher import fetch_and_store, ingest_raw + + +class Command(BaseCommand): + help = 'Ingest nginx ingress log lines into the nginxmon database' + + def add_arguments(self, parser): + group = parser.add_mutually_exclusive_group() + group.add_argument( + '--file', '-f', + metavar='PATH', + help='Path to a nginx log file to ingest', + ) + group.add_argument( + '--from-settings', + action='store_true', + help='Run a full fetch using the source configured in NginxSettings ' + '(kubectl or log file)', + ) + # No flag = read from stdin + + def handle(self, *args, **options): + if options['from_settings']: + count = fetch_and_store() + self.stdout.write(self.style.SUCCESS(f'Fetched and inserted {count} new log entries.')) + return + + if options['file']: + try: + with open(options['file'], 'r', encoding='utf-8', errors='replace') as fh: + text = fh.read() + except OSError as exc: + self.stderr.write(self.style.ERROR(f'Cannot read file: {exc}')) + return + else: + self.stdout.write('Reading from stdin… (Ctrl-D to finish)') + text = sys.stdin.read() + + count = ingest_raw(text) + self.stdout.write(self.style.SUCCESS(f'Inserted {count} new log entries.')) diff --git a/nginxmon/management/commands/nginxmon_stern.py b/nginxmon/management/commands/nginxmon_stern.py new file mode 100644 index 0000000..8965a38 --- /dev/null +++ b/nginxmon/management/commands/nginxmon_stern.py @@ -0,0 +1,145 @@ +""" +Management command that streams logs from stern and ingests them continuously. + +Usage: + python manage.py nginxmon_stern + python manage.py nginxmon_stern --namespace ingress-nginx --selector app.kubernetes.io/name=ingress-nginx + python manage.py nginxmon_stern --tail 200 --batch-size 20 + +The command checks NginxSettings.enabled every few seconds and pauses/resumes +ingestion accordingly. Stop with Ctrl-C. +""" +import subprocess +import time +import signal +import sys + +from django.core.management.base import BaseCommand + +from nginxmon.fetcher import ingest_raw +from nginxmon.models import NginxSettings + + +_STERN_BIN = '/opt/homebrew/bin/stern' +_DEFAULT_NAMESPACE = 'ingress-nginx' +_DEFAULT_SELECTOR = 'ingress-nginx-controller' +_BATCH_SIZE = 30 +_ENABLED_CHECK_INTERVAL = 5 # seconds between enabled-flag checks +_FLUSH_INTERVAL = 3 # seconds before flushing a partial batch + + +class Command(BaseCommand): + help = 'Stream nginx logs via stern and continuously ingest them into the database' + + def add_arguments(self, parser): + parser.add_argument('--namespace', '-n', default=_DEFAULT_NAMESPACE, + help='Kubernetes namespace (default: ingress-nginx)') + parser.add_argument('--selector', '-l', default=_DEFAULT_SELECTOR, + help='Pod selector passed to stern (default: ingress-nginx-controller)') + parser.add_argument('--tail', type=int, default=500, + help='Number of historical lines to retrieve on start (default: 500)') + parser.add_argument('--batch-size', type=int, default=_BATCH_SIZE, + help='Number of lines to batch before ingesting (default: 30)') + + def handle(self, *args, **options): + namespace = options['namespace'] + selector = options['selector'] + tail = options['tail'] + batch_size = options['batch_size'] + + self._shutdown = False + signal.signal(signal.SIGINT, self._handle_signal) + signal.signal(signal.SIGTERM, self._handle_signal) + + self.stdout.write(self.style.SUCCESS( + f'Starting stern: selector={selector!r} namespace={namespace!r} tail={tail}' + )) + + cmd = [ + _STERN_BIN, + selector, + '--namespace', namespace, + '--tail', str(tail), + '--color', 'never', + '--template', '{{.PodName}} {{.ContainerName}} {{.Message}}\n', + ] + + self.stdout.write(f'Command: {" ".join(cmd)}') + + proc = None + try: + proc = subprocess.Popen( + cmd, + stdout=subprocess.PIPE, + stderr=None, # pass stern's stderr straight to the terminal + text=True, + bufsize=1, + ) + self._ingest_loop(proc, batch_size) + except FileNotFoundError: + self.stderr.write(self.style.ERROR( + f'stern not found at {_STERN_BIN}. Install with: brew install stern' + )) + sys.exit(1) + finally: + if proc and proc.poll() is None: + proc.terminate() + try: + proc.wait(timeout=5) + except subprocess.TimeoutExpired: + proc.kill() + + def _handle_signal(self, signum, frame): + self.stdout.write('\nShutting down stern ingestion…') + self._shutdown = True + + def _ingest_loop(self, proc, batch_size): + batch = [] + last_flush = time.monotonic() + last_enabled_check = 0.0 + enabled = True + + import select + + while not self._shutdown: + # Periodically re-check NginxSettings.enabled + now = time.monotonic() + if now - last_enabled_check >= _ENABLED_CHECK_INTERVAL: + try: + enabled = NginxSettings.get().enabled + except Exception: + pass + last_enabled_check = now + + if proc.poll() is not None: + # stern process exited + self.stderr.write(self.style.WARNING('stern process exited.')) + break + + # Non-blocking readline with a short select timeout + ready, _, _ = select.select([proc.stdout], [], [], 0.5) + if ready: + line = proc.stdout.readline() + if not line: + break + if enabled: + batch.append(line) + + # Flush when batch is full or flush interval elapsed + if batch and (len(batch) >= batch_size or time.monotonic() - last_flush >= _FLUSH_INTERVAL): + self._flush(batch) + batch = [] + last_flush = time.monotonic() + + # Final flush + if batch: + self._flush(batch) + + def _flush(self, lines): + text = ''.join(lines) + try: + count = ingest_raw(text) + if count: + self.stdout.write(f' [stern] Ingested {count} new log entries') + except Exception as exc: + self.stderr.write(self.style.ERROR(f' [stern] Ingest error: {exc}')) diff --git a/nginxmon/migrations/0001_initial.py b/nginxmon/migrations/0001_initial.py new file mode 100644 index 0000000..dbc6c1b --- /dev/null +++ b/nginxmon/migrations/0001_initial.py @@ -0,0 +1,116 @@ +# Generated by Django 5.2.12 on 2026-04-01 03:47 + +import django.db.models.deletion +import netscan.fields +from django.db import migrations, models + + +class Migration(migrations.Migration): + + initial = True + + dependencies = [ + ] + + operations = [ + migrations.CreateModel( + name='IPGeoCache', + fields=[ + ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), + ('ip', models.GenericIPAddressField(db_index=True, unique=True)), + ('country', models.CharField(blank=True, max_length=100)), + ('country_code', models.CharField(blank=True, max_length=10)), + ('region', models.CharField(blank=True, max_length=100)), + ('city', models.CharField(blank=True, max_length=100)), + ('lat', models.FloatField(blank=True, null=True)), + ('lon', models.FloatField(blank=True, null=True)), + ('isp', models.CharField(blank=True, max_length=300)), + ('is_private', models.BooleanField(default=False)), + ('fetched_at', models.DateTimeField(auto_now=True)), + ], + ), + migrations.CreateModel( + name='NginxAlertProfile', + fields=[ + ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), + ('name', models.CharField(max_length=200)), + ('alert_window_seconds', models.IntegerField(default=60, help_text='Rolling window in seconds for rate analysis')), + ('alert_max_requests', models.IntegerField(default=200, help_text='Max requests per IP per window before triggering a rate alert')), + ('alert_max_error_rate', models.FloatField(default=0.6, help_text='Error rate threshold (0–1) to flag an IP')), + ('alert_min_requests', models.IntegerField(default=15, help_text='Minimum requests before running error-rate check')), + ('telegram_bot_token', netscan.fields.EncryptedCharField(blank=True)), + ('telegram_chat_id', netscan.fields.EncryptedCharField(blank=True)), + ('enabled', models.BooleanField(default=True)), + ('created_at', models.DateTimeField(auto_now_add=True)), + ], + ), + migrations.CreateModel( + name='NginxSettings', + fields=[ + ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), + ('namespace', models.CharField(default='ingress-nginx', max_length=200)), + ('pod_label', models.CharField(default='app.kubernetes.io/name=ingress-nginx', help_text='kubectl -l selector for the ingress-nginx pods', max_length=200)), + ('container', models.CharField(default='controller', max_length=100)), + ('fetch_interval_seconds', models.IntegerField(default=30, help_text='How often to pull new logs (seconds)')), + ('log_file_path', models.CharField(blank=True, help_text='Absolute path to a local nginx log file for testing. Leave empty to use kubectl in production.', max_length=500)), + ('enabled', models.BooleanField(default=True)), + ('last_fetch_at', models.DateTimeField(blank=True, null=True)), + ], + options={ + 'verbose_name': 'Nginx Settings', + 'verbose_name_plural': 'Nginx Settings', + }, + ), + migrations.CreateModel( + name='NginxAccessLog', + fields=[ + ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), + ('timestamp', models.DateTimeField(db_index=True)), + ('remote_addr', models.GenericIPAddressField(db_index=True)), + ('method', models.CharField(max_length=20)), + ('request_uri', models.TextField()), + ('protocol', models.CharField(max_length=20)), + ('status', models.IntegerField(db_index=True)), + ('body_bytes_sent', models.IntegerField()), + ('http_referer', models.TextField(blank=True)), + ('http_user_agent', models.TextField(blank=True)), + ('request_length', models.IntegerField(default=0)), + ('request_time', models.FloatField()), + ('service', models.CharField(blank=True, db_index=True, max_length=200)), + ('upstream_addr', models.CharField(blank=True, max_length=200)), + ('upstream_response_time', models.FloatField(blank=True, null=True)), + ('upstream_status', models.IntegerField(blank=True, null=True)), + ('request_id', models.CharField(blank=True, db_index=True, max_length=100)), + ('country', models.CharField(blank=True, max_length=100)), + ('country_code', models.CharField(blank=True, max_length=10)), + ('region', models.CharField(blank=True, max_length=100)), + ('city', models.CharField(blank=True, max_length=100)), + ], + options={ + 'ordering': ['-timestamp'], + 'indexes': [models.Index(fields=['remote_addr', 'timestamp'], name='nginxmon_ng_remote__1ceb1d_idx')], + }, + ), + migrations.CreateModel( + name='ThreatAlert', + fields=[ + ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), + ('alert_type', models.CharField(choices=[('rate_limit', 'High Request Rate'), ('error_rate', 'High Error Rate')], max_length=50)), + ('remote_addr', models.GenericIPAddressField(db_index=True)), + ('detected_at', models.DateTimeField(auto_now_add=True)), + ('window_start', models.DateTimeField()), + ('window_end', models.DateTimeField()), + ('request_count', models.IntegerField()), + ('error_count', models.IntegerField(default=0)), + ('detail', models.TextField()), + ('notified', models.BooleanField(default=False)), + ('dismissed', models.BooleanField(default=False)), + ('country', models.CharField(blank=True, max_length=100)), + ('city', models.CharField(blank=True, max_length=100)), + ('profile', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='alerts', to='nginxmon.nginxalertprofile')), + ], + options={ + 'ordering': ['-detected_at'], + }, + ), + ] diff --git a/nginxmon/migrations/0002_covering_index.py b/nginxmon/migrations/0002_covering_index.py new file mode 100644 index 0000000..3a42e12 --- /dev/null +++ b/nginxmon/migrations/0002_covering_index.py @@ -0,0 +1,17 @@ +# Generated by Django 5.2.12 on 2026-04-01 04:09 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('nginxmon', '0001_initial'), + ] + + operations = [ + migrations.AddIndex( + model_name='nginxaccesslog', + index=models.Index(fields=['timestamp', 'status'], name='nginxmon_ng_timesta_978a40_idx'), + ), + ] diff --git a/nginxmon/migrations/__init__.py b/nginxmon/migrations/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/nginxmon/models.py b/nginxmon/models.py new file mode 100644 index 0000000..5a16869 --- /dev/null +++ b/nginxmon/models.py @@ -0,0 +1,169 @@ +from django.db import models + +from netscan.fields import EncryptedCharField + + +class NginxSettings(models.Model): + """ + Singleton (pk=1) — the one-off cluster connection config. + Use NginxSettings.get() everywhere instead of pk lookups. + """ + namespace = models.CharField(max_length=200, default='ingress-nginx') + pod_label = models.CharField( + max_length=200, + default='app.kubernetes.io/name=ingress-nginx', + help_text='kubectl -l selector for the ingress-nginx pods', + ) + container = models.CharField(max_length=100, default='controller') + fetch_interval_seconds = models.IntegerField( + default=30, + help_text='How often to pull new logs (seconds)', + ) + # Local dev / testing: read from a file instead of kubectl + log_file_path = models.CharField( + max_length=500, + blank=True, + help_text=( + 'Absolute path to a local nginx log file for testing. ' + 'Leave empty to use kubectl in production.' + ), + ) + enabled = models.BooleanField(default=True) + last_fetch_at = models.DateTimeField(null=True, blank=True) + + class Meta: + verbose_name = 'Nginx Settings' + verbose_name_plural = 'Nginx Settings' + + def __str__(self): + return f'Nginx Settings ({self.namespace})' + + @classmethod + def get(cls): + obj, _ = cls.objects.get_or_create(pk=1) + return obj + + @property + def source_display(self): + if self.log_file_path: + return f'File: {self.log_file_path}' + return f'kubectl -n {self.namespace} -l {self.pod_label}' + + +class NginxAlertProfile(models.Model): + """ + One or more alert profiles — each with its own thresholds and Telegram config. + """ + name = models.CharField(max_length=200) + alert_window_seconds = models.IntegerField( + default=60, + help_text='Rolling window in seconds for rate analysis', + ) + alert_max_requests = models.IntegerField( + default=200, + help_text='Max requests per IP per window before triggering a rate alert', + ) + alert_max_error_rate = models.FloatField( + default=0.6, + help_text='Error rate threshold (0–1) to flag an IP', + ) + alert_min_requests = models.IntegerField( + default=15, + help_text='Minimum requests before running error-rate check', + ) + telegram_bot_token = EncryptedCharField(blank=True) + telegram_chat_id = EncryptedCharField(blank=True) + enabled = models.BooleanField(default=True) + created_at = models.DateTimeField(auto_now_add=True) + + def __str__(self): + return self.name + + +class NginxAccessLog(models.Model): + """Stores parsed nginx access log entries (cluster-global, no profile FK).""" + timestamp = models.DateTimeField(db_index=True) + remote_addr = models.GenericIPAddressField(db_index=True) + method = models.CharField(max_length=20) + request_uri = models.TextField() + protocol = models.CharField(max_length=20) + status = models.IntegerField(db_index=True) + body_bytes_sent = models.IntegerField() + http_referer = models.TextField(blank=True) + http_user_agent = models.TextField(blank=True) + request_length = models.IntegerField(default=0) + request_time = models.FloatField() + service = models.CharField(max_length=200, blank=True, db_index=True) + upstream_addr = models.CharField(max_length=200, blank=True) + upstream_response_time = models.FloatField(null=True, blank=True) + upstream_status = models.IntegerField(null=True, blank=True) + request_id = models.CharField(max_length=100, blank=True, db_index=True) + # Geo (populated after insert) + country = models.CharField(max_length=100, blank=True) + country_code = models.CharField(max_length=10, blank=True) + region = models.CharField(max_length=100, blank=True) + city = models.CharField(max_length=100, blank=True) + + class Meta: + ordering = ['-timestamp'] + indexes = [ + models.Index(fields=['remote_addr', 'timestamp']), + models.Index(fields=['timestamp', 'status']), # covers chart aggregation queries + ] + + @property + def geo_display(self): + parts = [p for p in [self.city, self.country] if p] + return ', '.join(parts) if parts else '—' + + +class IPGeoCache(models.Model): + ip = models.GenericIPAddressField(unique=True, db_index=True) + country = models.CharField(max_length=100, blank=True) + country_code = models.CharField(max_length=10, blank=True) + region = models.CharField(max_length=100, blank=True) + city = models.CharField(max_length=100, blank=True) + lat = models.FloatField(null=True, blank=True) + lon = models.FloatField(null=True, blank=True) + isp = models.CharField(max_length=300, blank=True) + is_private = models.BooleanField(default=False) + fetched_at = models.DateTimeField(auto_now=True) + + def __str__(self): + if self.is_private: + return f'{self.ip} (private)' + parts = [p for p in [self.city, self.country] if p] + return f'{self.ip} — {", ".join(parts)}' + + +class ThreatAlert(models.Model): + ALERT_TYPES = [ + ('rate_limit', 'High Request Rate'), + ('error_rate', 'High Error Rate'), + ] + + profile = models.ForeignKey( + NginxAlertProfile, on_delete=models.CASCADE, related_name='alerts' + ) + alert_type = models.CharField(max_length=50, choices=ALERT_TYPES) + remote_addr = models.GenericIPAddressField(db_index=True) + detected_at = models.DateTimeField(auto_now_add=True) + window_start = models.DateTimeField() + window_end = models.DateTimeField() + request_count = models.IntegerField() + error_count = models.IntegerField(default=0) + detail = models.TextField() + notified = models.BooleanField(default=False) + dismissed = models.BooleanField(default=False) + country = models.CharField(max_length=100, blank=True) + city = models.CharField(max_length=100, blank=True) + + class Meta: + ordering = ['-detected_at'] + + def __str__(self): + return f'[{self.get_alert_type_display()}] {self.remote_addr} @ {self.detected_at:%Y-%m-%d %H:%M}' + + @property + def error_rate(self): + return self.error_count / self.request_count if self.request_count else 0 diff --git a/nginxmon/notifications.py b/nginxmon/notifications.py new file mode 100644 index 0000000..25fe3de --- /dev/null +++ b/nginxmon/notifications.py @@ -0,0 +1,52 @@ +import logging +import requests + +logger = logging.getLogger(__name__) + +_ICONS = {'rate_limit': '🚨', 'error_rate': '⚠️'} + + +def send_alert_telegram(profile, alert) -> bool: + if not profile.telegram_bot_token or not profile.telegram_chat_id: + return False + + icon = _ICONS.get(alert.alert_type, '🔔') + geo = f' ({alert.city}, {alert.country})' if alert.country else '' + text = ( + f'{icon} *NginxMon Alert* — {profile.name}\n' + f'Type: {alert.get_alert_type_display()}\n' + f'IP: `{alert.remote_addr}`{geo}\n' + f'{alert.detail}\n' + f'_Detected at {alert.detected_at.strftime("%Y-%m-%d %H:%M:%S UTC") if alert.detected_at else "now"}_' + ) + url = f'https://api.telegram.org/bot{profile.telegram_bot_token}/sendMessage' + try: + resp = requests.post(url, json={ + 'chat_id': profile.telegram_chat_id, + 'text': text, + 'parse_mode': 'Markdown', + }, timeout=10) + resp.raise_for_status() + alert.notified = True + alert.save(update_fields=['notified']) + logger.info('nginxmon: Telegram alert sent for alert #%s', alert.pk) + return True + except Exception as exc: + logger.error('nginxmon: Telegram send failed: %s', exc) + return False + + +def send_test_telegram(profile) -> dict: + if not profile.telegram_bot_token or not profile.telegram_chat_id: + return {'ok': False, 'error': 'Bot token or chat ID not set.'} + try: + url = f'https://api.telegram.org/bot{profile.telegram_bot_token}/sendMessage' + resp = requests.post(url, json={ + 'chat_id': profile.telegram_chat_id, + 'text': f'✅ *NginxMon test* from profile _{profile.name}_. Alerts are working.', + 'parse_mode': 'Markdown', + }, timeout=10) + resp.raise_for_status() + return {'ok': True} + except Exception as e: + return {'ok': False, 'error': str(e)} diff --git a/nginxmon/parser.py b/nginxmon/parser.py new file mode 100644 index 0000000..4392523 --- /dev/null +++ b/nginxmon/parser.py @@ -0,0 +1,83 @@ +""" +Parse nginx ingress-controller log lines. + +Log format: + - - [] " " + "" "" [] [] + +""" +import re +import logging +from datetime import datetime, timezone + +logger = logging.getLogger(__name__) + +# The leading \S* tolerates stray chars (e.g. 'z') sometimes prepended by log shippers. +_LOG_RE = re.compile( + r'\S*ingress-nginx\S+\s+' # pod name + r'(?P\S+)\s+' # container + r'(?P[\d.a-fA-F:]+)\s+-\s+-\s+' # IP - - + r'\[(?P[^\]]+)\]\s+' # [timestamp] + r'"(?P\S+)\s+(?P\S+)\s+' # "METHOD /path + r'(?P[^"]+)"\s+' # PROTO" + r'(?P\d+)\s+' # status + r'(?P\d+)\s+' # bytes + r'"(?P[^"]*)"\s+' # "referer" + r'"(?P[^"]*)"\s+' # "ua" + r'(?P\d+)\s+' # req_len + r'(?P[\d.]+)\s+' # req_time + r'\[(?P[^\]]*)\]\s+' # [service] + r'\[(?P[^\]]*)\]\s+' # [] + r'(?P\S+)\s+' # upstream addr + r'(?P\d+)\s+' # upstream bytes + r'(?P[\d.]+)\s+' # upstream time + r'(?P\d+)\s+' # upstream status + r'(?P\S+)' # request id +) + +_TIME_FMT = '%d/%b/%Y:%H:%M:%S %z' + + +def parse_line(line: str) -> dict | None: + """Return a parsed dict or None if the line doesn't match.""" + m = _LOG_RE.search(line) + if not m: + return None + d = m.groupdict() + try: + ts = datetime.strptime(d['time_local'], _TIME_FMT) + except ValueError: + logger.debug('Bad timestamp: %s', d['time_local']) + return None + return { + 'timestamp': ts, + 'remote_addr': d['remote_addr'], + 'method': d['method'], + 'request_uri': d['request_uri'], + 'protocol': d['protocol'].strip(), + 'status': int(d['status']), + 'body_bytes_sent': int(d['body_bytes_sent']), + 'http_referer': d['http_referer'], + 'http_user_agent': d['http_user_agent'], + 'request_length': int(d['request_length']), + 'request_time': float(d['request_time']), + 'service': d['service'], + 'upstream_addr': d['upstream_addr'], + 'upstream_response_time': float(d['upstream_response_time']), + 'upstream_status': int(d['upstream_status']), + 'request_id': d['request_id'], + } + + +def parse_lines(text: str) -> list[dict]: + results = [] + for line in text.splitlines(): + line = line.strip() + if not line: + continue + parsed = parse_line(line) + if parsed: + results.append(parsed) + else: + logger.debug('Skipped unparseable line: %.120s', line) + return results diff --git a/nginxmon/tasks.py b/nginxmon/tasks.py new file mode 100644 index 0000000..014c61c --- /dev/null +++ b/nginxmon/tasks.py @@ -0,0 +1,56 @@ +import logging + +from apscheduler.triggers.interval import IntervalTrigger +from core.scheduler import scheduler +from nginxmon.fetcher import fetch_and_store, cleanup_old_logs +from nginxmon.detector import detect_threats + +logger = logging.getLogger(__name__) + +_FETCH_JOB = 'nginxmon_fetch' +_CLEANUP_JOB = 'nginxmon_cleanup' + + +def start_fetch_job(interval_seconds: int): + scheduler.add_job( + fetch_and_store, + trigger=IntervalTrigger(seconds=interval_seconds), + id=_FETCH_JOB, + replace_existing=True, + misfire_grace_time=60, + ) + logger.info('nginxmon: scheduled fetch every %ds', interval_seconds) + + +def stop_fetch_job(): + if scheduler.get_job(_FETCH_JOB): + scheduler.remove_job(_FETCH_JOB) + + +def schedule_profile(profile): + job_id = f'nginxmon_detect_{profile.pk}' + scheduler.add_job( + detect_threats, + trigger=IntervalTrigger(seconds=max(60, profile.alert_window_seconds)), + id=job_id, + args=[profile.pk], + replace_existing=True, + misfire_grace_time=120, + ) + logger.info('nginxmon: scheduled detect for profile "%s"', profile.name) + + +def unschedule_profile(profile): + job_id = f'nginxmon_detect_{profile.pk}' + if scheduler.get_job(job_id): + scheduler.remove_job(job_id) + + +def start_cleanup_job(): + if not scheduler.get_job(_CLEANUP_JOB): + scheduler.add_job( + cleanup_old_logs, + trigger=IntervalTrigger(hours=24), + id=_CLEANUP_JOB, + replace_existing=True, + ) diff --git a/nginxmon/templates/nginxmon/_alerts.html b/nginxmon/templates/nginxmon/_alerts.html new file mode 100644 index 0000000..3b68ba0 --- /dev/null +++ b/nginxmon/templates/nginxmon/_alerts.html @@ -0,0 +1,50 @@ +{% if alerts %} + + + + + + + + + + + + + + {% for alert in alerts %} + + + + + + + + + + {% endfor %} + +
DetectedTypeIPDetailAction
{{ alert.detected_at|date:"M d H:i" }} + + {% if alert.alert_type == 'rate_limit' %}🚨{% else %}⚠️{% endif %} + {{ alert.get_alert_type_display }} + + {{ alert.remote_addr }}{{ alert.detail }} + +
+{% else %} +
+ + No active threat alerts +
+{% endif %} diff --git a/nginxmon/templates/nginxmon/_live_logs.html b/nginxmon/templates/nginxmon/_live_logs.html new file mode 100644 index 0000000..a038ae7 --- /dev/null +++ b/nginxmon/templates/nginxmon/_live_logs.html @@ -0,0 +1,29 @@ +{% for log in logs %} + + {{ log.timestamp|date:"m/d H:i:s" }} + + + + {{ log.geo_display }} + + {{ log.method }} + + {{ log.request_uri }} + + {% if log.service %}{% else %}—{% endif %} + + + + + {{ log.request_time|floatformat:3 }}s + +{% empty %} +No logs yet — use "Fetch Now" or wait for the scheduler. +{% endfor %} diff --git a/nginxmon/templates/nginxmon/dashboard.html b/nginxmon/templates/nginxmon/dashboard.html new file mode 100644 index 0000000..3763314 --- /dev/null +++ b/nginxmon/templates/nginxmon/dashboard.html @@ -0,0 +1,416 @@ +{% extends 'base.html' %} + +{% block content %} + + +
+ + +
+
+

+ + Nginx Monitor + + Live + +

+

{{ nginx_settings.source_display }}

+
+ {% if nginx_settings.last_fetch_at %}Last: {{ nginx_settings.last_fetch_at|timesince }} ago{% else %}No ingestion yet{% endif %} + · + — +
+
+
+ +
+ {% for rkey, rlabel in all_ranges %} + + {{ rlabel|slice:"5:" }} + + {% endfor %} +
+ +
+
+ +
+ {% csrf_token %} + +
+
+ {% csrf_token %} + +
+ + New Alert Profile + + + Settings + +
+
+
+ + + {% for msg in messages %} +
+ {{ msg }} +
+ {% endfor %} + + +
+ {% with s=stats %} +
+
Requests
+
{{ s.total }}
+
+
+
Errors
+
{{ s.errors }}
+
+
+
Unique IPs
+
{{ s.unique_ips }}
+
+
+
Avg Response
+
{{ s.avg_resp_ms }} ms
+
+ {% endwith %} +
+ + +
+
+
{{ range_label }}
+
+
+
+
Status codes — {{ range_label }}
+
+
+
+ + +
+
+
Top IPs — {{ range_label }}
+ + + + + + + + + + + {% for ip in top_ips %} + + + + + + + {% empty %} + + {% endfor %} + +
IPLocationReqsErrors
+ + + {% if ip.city or ip.country %}{{ ip.city }}{% if ip.city and ip.country %}, {% endif %}{{ ip.country }}{% else %}—{% endif %} + {{ ip.count }}{{ ip.errors }}
No data yet
+
+ +
+
Top Services — {{ range_label }}
+ + + + + + + + + {% for svc in top_services %} + + + + + {% empty %} + + {% endfor %} + +
ServiceRequests
+ {% if svc.service %}{% else %}—{% endif %} + {{ svc.count }}
No data yet
+
+
+ + + {% if profiles %} +
+
+
Alert Profiles
+ + New +
+
+ {% for p in profiles %} +
+
+ {{ p.name }} + + window {{ p.alert_window_seconds }}s • max {{ p.alert_max_requests }} req • {{ p.alert_max_error_rate|floatformat:0 }}% err rate + +
+
+
+ {% csrf_token %} + +
+ Edit + Delete +
+
+ {% endfor %} +
+
+ {% endif %} + + +
+
+ + Threat Alerts + {% if active_alerts %}{{ active_alerts }}{% endif %} +
+
+ {% include 'nginxmon/_alerts.html' %} +
+
+ + +
+
+
+ + Live Access Log +
+ Auto-refreshes every 5s • last 60 entries +
+
+ + + + + + + + + + + + + + + {% include 'nginxmon/_live_logs.html' %} + +
Time ▼IP ⇅Method ⇅PathStatus ⇅
+
+
+ +
+ + + + + + +{% endblock %} diff --git a/nginxmon/templates/nginxmon/profile_confirm_delete.html b/nginxmon/templates/nginxmon/profile_confirm_delete.html new file mode 100644 index 0000000..0e6b897 --- /dev/null +++ b/nginxmon/templates/nginxmon/profile_confirm_delete.html @@ -0,0 +1,28 @@ +{% extends 'base.html' %} + +{% block content %} +
+
+
+

+ Delete Alert Profile +

+
+
+

+ Delete alert profile {{ object.name }}? + Its alerts history will also be removed. +

+
+ {% csrf_token %} + + + Cancel + +
+
+
+
+{% endblock %} diff --git a/nginxmon/templates/nginxmon/profile_form.html b/nginxmon/templates/nginxmon/profile_form.html new file mode 100644 index 0000000..2a9d584 --- /dev/null +++ b/nginxmon/templates/nginxmon/profile_form.html @@ -0,0 +1,36 @@ +{% extends 'base.html' %} + +{% block content %} +
+
+
+

+ {% if object %}Edit Alert Profile — {{ object.name }}{% else %}New Alert Profile{% endif %} +

+

+ Alert thresholds and Telegram notification settings for this monitoring rule. +

+
+ +
+ {% csrf_token %} + {% for field in form %} +
+ + {{ field }} + {% if field.help_text %}

{{ field.help_text }}

{% endif %} + {% for error in field.errors %}

{{ error }}

{% endfor %} +
+ {% endfor %} + +
+ + Cancel +
+
+
+
+{% endblock %} diff --git a/nginxmon/templates/nginxmon/settings.html b/nginxmon/templates/nginxmon/settings.html new file mode 100644 index 0000000..0bff8c3 --- /dev/null +++ b/nginxmon/templates/nginxmon/settings.html @@ -0,0 +1,103 @@ +{% extends 'base.html' %} + +{% block content %} +
+ + +
+ + + +
+

+ Nginx Monitor Settings +

+

Cluster connection & local testing

+
+
+ + {% for msg in messages %} +
+ {{ msg }} +
+ {% endfor %} + + +
+
+

Cluster Connection

+

+ Configured once. The scheduler uses these to run + kubectl logs on a timer. +

+
+
+ {% csrf_token %} + {% for field in form %} +
+ + {{ field }} + {% if field.help_text %}

{{ field.help_text }}

{% endif %} + {% for error in field.errors %}

{{ error }}

{% endfor %} +
+ {% endfor %} +
+ +
+
+
+ + +
+
+ +
+

Test Locally — Paste Log Lines

+

+ No kubectl needed. Paste raw nginx ingress log lines and they'll be parsed and stored + immediately, so you can preview the dashboard with real data. +

+
+
+
+ {% csrf_token %} + {% for field in paste_form %} +
+ +
{{ field }}
+ {% for error in field.errors %}

{{ error }}

{% endfor %} +
+ {% endfor %} +
+ + Duplicates are automatically skipped. +
+
+ + +
+

Or via the management command:

+ 
# From a file
+python manage.py nginxmon_ingest --file /path/to/nginx.log
+
+# Pipe from kubectl directly
+kubectl logs -n ingress-nginx \
+  -l app.kubernetes.io/name=ingress-nginx \
+  --container controller | python manage.py nginxmon_ingest
+
+# Re-run the configured source (kubectl or file)
+python manage.py nginxmon_ingest --from-settings
+
+
+ +
+{% endblock %} diff --git a/nginxmon/urls.py b/nginxmon/urls.py new file mode 100644 index 0000000..44ad6a0 --- /dev/null +++ b/nginxmon/urls.py @@ -0,0 +1,29 @@ +from django.urls import path +from . import views + +urlpatterns = [ + # Main dashboard + path('', views.DashboardView.as_view(), name='nginxmon-dashboard'), + + # One-off settings (k8s pod config + paste-logs test) + path('settings/', views.SettingsView.as_view(), name='nginxmon-settings'), + + # Alert profiles + path('profiles/new/', views.ProfileCreateView.as_view(), name='nginxmon-profile-create'), + path('profiles//edit/', views.ProfileUpdateView.as_view(), name='nginxmon-profile-edit'), + path('profiles//delete/', views.ProfileDeleteView.as_view(), name='nginxmon-profile-delete'), + path('profiles//test-telegram/', views.TestTelegramView.as_view(), name='nginxmon-test-telegram'), + + # Manual actions + path('actions/fetch/', views.TriggerFetchView.as_view(), name='nginxmon-fetch'), + path('actions/detect/', views.TriggerDetectView.as_view(), name='nginxmon-detect'), + path('alert//dismiss/', views.DismissAlertView.as_view(), name='nginxmon-dismiss-alert'), + + # HTMX partials & APIs + path('api/live-logs/', views.LiveLogsPartialView.as_view(), name='nginxmon-live-logs'), + path('api/alerts/', views.AlertsPartialView.as_view(), name='nginxmon-alerts-partial'), + path('api/chart-data/', views.ChartDataView.as_view(), name='nginxmon-chart-data'), + path('api/ingest/', views.IngestLogsView.as_view(), name='nginxmon-ingest'), + path('api/toggle-ingestion/', views.ToggleIngestionView.as_view(), name='nginxmon-toggle'), + path('api/status-stream/', views.StatusStreamView.as_view(), name='nginxmon-status-stream'), +] diff --git a/nginxmon/views.py b/nginxmon/views.py new file mode 100644 index 0000000..7ae5e2c --- /dev/null +++ b/nginxmon/views.py @@ -0,0 +1,384 @@ +import json +import logging +import threading +import time +from datetime import timedelta + +from django.contrib import messages +from django.db.models import Avg, Count, Q +from django.db.models.functions import TruncDay, TruncHour, TruncMinute +from django.http import JsonResponse, StreamingHttpResponse +from django.shortcuts import get_object_or_404, redirect, render +from django.urls import reverse_lazy +from django.utils import timezone +from django.views.generic import CreateView, DeleteView, TemplateView, UpdateView, View + +from .detector import detect_threats as run_detect +from .fetcher import fetch_and_store, ingest_raw +from .forms import NginxSettingsForm, NginxAlertProfileForm, PasteLogsForm +from .models import NginxAccessLog, NginxAlertProfile, NginxSettings, ThreatAlert +from .notifications import send_test_telegram +from .tasks import start_fetch_job, stop_fetch_job, schedule_profile, unschedule_profile + +logger = logging.getLogger(__name__) + +# ── Time range config ───────────────────────────────────────────────────────── + +_RANGES = { + '5m': {'delta': timedelta(minutes=5), 'trunc': TruncMinute, 'group_n': 1, 'label': 'Last 5 min', 'fmt': '%H:%M'}, + '30m': {'delta': timedelta(minutes=30), 'trunc': TruncMinute, 'group_n': 2, 'label': 'Last 30 min', 'fmt': '%H:%M'}, + '6h': {'delta': timedelta(hours=6), 'trunc': TruncMinute, 'group_n': 15, 'label': 'Last 6 hours', 'fmt': '%H:%M'}, + '1d': {'delta': timedelta(hours=24), 'trunc': TruncHour, 'group_n': 1, 'label': 'Last 24 hours', 'fmt': '%H:%M'}, + '7d': {'delta': timedelta(days=7), 'trunc': TruncHour, 'group_n': 4, 'label': 'Last 7 days', 'fmt': '%m/%d %H:%M'}, + '30d': {'delta': timedelta(days=30), 'trunc': TruncDay, 'group_n': 1, 'label': 'Last 30 days', 'fmt': '%m/%d'}, + '90d': {'delta': timedelta(days=90), 'trunc': TruncDay, 'group_n': 3, 'label': 'Last 90 days', 'fmt': '%m/%d'}, +} +_DEFAULT_RANGE = '6h' + + +def _get_range(request): + key = request.GET.get('range', _DEFAULT_RANGE) + if key not in _RANGES: + key = _DEFAULT_RANGE + cfg = _RANGES[key] + return key, cfg, timezone.now() - cfg['delta'] + + +def _floor_bucket_key(dt, trunc_fn, group_n, fmt): + """Return format string for the bucket boundary dt falls into.""" + if trunc_fn is TruncMinute: + m = (dt.minute // group_n) * group_n + aligned = dt.replace(minute=m, second=0, microsecond=0) + elif trunc_fn is TruncHour: + h = (dt.hour // group_n) * group_n + aligned = dt.replace(hour=h, minute=0, second=0, microsecond=0) + else: # TruncDay + if group_n <= 1: + aligned = dt.replace(hour=0, minute=0, second=0, microsecond=0) + else: + from datetime import date + epoch = date(2020, 1, 1) + d = dt.date() if hasattr(dt, 'date') else dt + days = (d - epoch).days + floored_date = epoch + timedelta(days=(days // group_n) * group_n) + aligned = dt.replace( + year=floored_date.year, month=floored_date.month, + day=floored_date.day, hour=0, minute=0, second=0, microsecond=0, + ) + return aligned.strftime(fmt) + + +# ── Dashboard — live monitoring ─────────────────────────────────────────────── + +class DashboardView(TemplateView): + template_name = 'nginxmon/dashboard.html' + + def get_context_data(self, **kwargs): + ctx = super().get_context_data(**kwargs) + range_key, cfg, since = _get_range(self.request) + + logs = NginxAccessLog.objects.filter(timestamp__gte=since) + stats = { + 'total': logs.count(), + 'errors': logs.filter(status__gte=400).count(), + 'unique_ips': logs.values('remote_addr').distinct().count(), + 'avg_resp_ms': round( + (logs.aggregate(a=Avg('request_time'))['a'] or 0) * 1000, 1 + ), + } + + top_services = list( + logs.values('service') + .annotate(count=Count('id')) + .order_by('-count')[:10] + ) + status_dist = list( + logs.values('status') + .annotate(count=Count('id')) + .order_by('status') + ) + top_ips = list( + logs.values('remote_addr', 'country', 'city') + .annotate(count=Count('id'), errors=Count('id', filter=Q(status__gte=400))) + .order_by('-count')[:10] + ) + active_alerts = ThreatAlert.objects.filter(dismissed=False).count() + profiles = NginxAlertProfile.objects.filter(enabled=True) + + ctx.update({ + 'nginx_settings': NginxSettings.get(), + 'stats': stats, + 'top_services': top_services, + 'status_dist': status_dist, + 'top_ips': top_ips, + 'active_alerts': active_alerts, + 'profiles': profiles, + 'status_dist_json': json.dumps(status_dist), + 'current_range': range_key, + 'range_label': cfg['label'], + 'all_ranges': [(k, v['label']) for k, v in _RANGES.items()], + }) + return ctx + + +# ── Settings (one-off k8s config + paste test logs) ────────────────────────── + +class SettingsView(View): + template_name = 'nginxmon/settings.html' + + def _get_settings(self): + return NginxSettings.get() + + def get(self, request): + obj = self._get_settings() + return render(request, self.template_name, { + 'form': NginxSettingsForm(instance=obj), + 'paste_form': PasteLogsForm(), + 'nginx_settings': obj, + }) + + def post(self, request): + obj = self._get_settings() + if 'save_settings' in request.POST: + form = NginxSettingsForm(request.POST, instance=obj) + if form.is_valid(): + saved = form.save() + stop_fetch_job() + if saved.enabled: + start_fetch_job(saved.fetch_interval_seconds) + messages.success(request, 'Settings saved.') + return redirect('nginxmon-settings') + return render(request, self.template_name, { + 'form': form, + 'paste_form': PasteLogsForm(), + 'nginx_settings': obj, + }) + + if 'paste_logs' in request.POST: + paste_form = PasteLogsForm(request.POST) + if paste_form.is_valid(): + count = ingest_raw(paste_form.cleaned_data['log_text']) + messages.success(request, f'Ingested {count} new log entries.') + return redirect('nginxmon-settings') + return render(request, self.template_name, { + 'form': NginxSettingsForm(instance=obj), + 'paste_form': paste_form, + 'nginx_settings': obj, + }) + + return redirect('nginxmon-settings') + + +# ── Alert profiles ──────────────────────────────────────────────────────────── + +class ProfileCreateView(CreateView): + model = NginxAlertProfile + form_class = NginxAlertProfileForm + template_name = 'nginxmon/profile_form.html' + success_url = reverse_lazy('nginxmon-dashboard') + + def form_valid(self, form): + response = super().form_valid(form) + if self.object.enabled: + schedule_profile(self.object) + messages.success(self.request, f'Alert profile "{self.object.name}" created.') + return response + + +class ProfileUpdateView(UpdateView): + model = NginxAlertProfile + form_class = NginxAlertProfileForm + template_name = 'nginxmon/profile_form.html' + success_url = reverse_lazy('nginxmon-dashboard') + + def form_valid(self, form): + response = super().form_valid(form) + unschedule_profile(self.object) + if self.object.enabled: + schedule_profile(self.object) + messages.success(self.request, f'Alert profile "{self.object.name}" updated.') + return response + + +class ProfileDeleteView(DeleteView): + model = NginxAlertProfile + template_name = 'nginxmon/profile_confirm_delete.html' + success_url = reverse_lazy('nginxmon-dashboard') + + def form_valid(self, form): + unschedule_profile(self.object) + return super().form_valid(form) + + +class TestTelegramView(View): + def post(self, request, pk): + profile = get_object_or_404(NginxAlertProfile, pk=pk) + result = send_test_telegram(profile) + if result['ok']: + messages.success(request, 'Telegram test message sent!') + else: + messages.error(request, f'Telegram error: {result["error"]}') + return redirect('nginxmon-dashboard') + + +# ── HTMX partials ───────────────────────────────────────────────────────────── + +class LiveLogsPartialView(View): + _SORT_MAP = { + 'timestamp': 'timestamp', + 'ip': 'remote_addr', + 'status': 'status', + 'method': 'method', + 'service': 'service', + 'time': 'request_time', + 'bytes': 'body_bytes_sent', + } + + def get(self, request): + qs = NginxAccessLog.objects.all() + + ip_f = request.GET.get('ip', '').strip() + status_f = request.GET.get('status', '').strip() + service_f = request.GET.get('service', '').strip() + sort_col = request.GET.get('sort', 'timestamp') + sort_ord = request.GET.get('order', 'desc') + + if ip_f: + qs = qs.filter(remote_addr=ip_f) + if status_f: + try: + qs = qs.filter(status=int(status_f)) + except ValueError: + pass + if service_f: + qs = qs.filter(service=service_f) + + db_col = self._SORT_MAP.get(sort_col, 'timestamp') + qs = qs.order_by(f'{"" if sort_ord == "asc" else "-"}{db_col}') + + return render(request, 'nginxmon/_live_logs.html', {'logs': qs[:60]}) + + +class AlertsPartialView(View): + def get(self, request): + alerts = ThreatAlert.objects.filter(dismissed=False).select_related('profile')[:20] + return render(request, 'nginxmon/_alerts.html', {'alerts': alerts}) + + +# ── Action views ────────────────────────────────────────────────────────────── + +class TriggerFetchView(View): + def post(self, request): + threading.Thread(target=fetch_and_store, daemon=True).start() + messages.success(request, 'Log fetch triggered in the background.') + return redirect('nginxmon-dashboard') + + +class TriggerDetectView(View): + def post(self, request): + for profile in NginxAlertProfile.objects.filter(enabled=True): + threading.Thread(target=run_detect, args=[profile.pk], daemon=True).start() + messages.success(request, 'Threat detection triggered for all profiles.') + return redirect('nginxmon-dashboard') + + +class DismissAlertView(View): + def post(self, request, pk): + alert = get_object_or_404(ThreatAlert, pk=pk) + alert.dismissed = True + alert.save(update_fields=['dismissed']) + if request.headers.get('HX-Request'): + return JsonResponse({'ok': True}) + return redirect('nginxmon-dashboard') + + +# ── Chart & ingest API ──────────────────────────────────────────────────────── + +class ChartDataView(View): + def get(self, request): + range_key, cfg, since = _get_range(request) + trunc_fn = cfg['trunc'] + group_n = cfg['group_n'] + fmt = cfg['fmt'] + + # Aggregate in DB using the timestamp index + trunc function. + # Returns at most ~360 rows (6h/minute) — then Python merges into final buckets. + rows = list( + NginxAccessLog.objects + .filter(timestamp__gte=since) + .annotate(bucket=trunc_fn('timestamp')) + .values('bucket') + .annotate(total=Count('id'), errors=Count('id', filter=Q(status__gte=400))) + .order_by('bucket') + ) + + # Merge adjacent DB buckets into the desired bucket size. + # Dict insertion order is preserved (Python 3.7+) so labels stay sorted. + buckets: dict = {} + for row in rows: + bkt = row['bucket'] + if bkt is None: + continue + key = _floor_bucket_key(bkt, trunc_fn, group_n, fmt) + if key not in buckets: + buckets[key] = {'total': 0, 'errors': 0} + buckets[key]['total'] += row['total'] + buckets[key]['errors'] += row['errors'] + + return JsonResponse({ + 'labels': list(buckets), + 'total': [v['total'] for v in buckets.values()], + 'errors': [v['errors'] for v in buckets.values()], + 'range': range_key, + 'label': cfg['label'], + }) + + +class IngestLogsView(View): + """POST endpoint for the paste-logs form (used from SettingsView).""" + def post(self, request): + text = request.POST.get('log_text', '') + count = ingest_raw(text) + return JsonResponse({'inserted': count}) + + +# ── Ingestion toggle & SSE status stream ───────────────────────────────────── + +class ToggleIngestionView(View): + """POST — flip NginxSettings.enabled and restart/stop the fetch job.""" + def post(self, request): + settings = NginxSettings.get() + settings.enabled = not settings.enabled + settings.save(update_fields=['enabled']) + stop_fetch_job() + if settings.enabled: + start_fetch_job(settings.fetch_interval_seconds) + return JsonResponse({'enabled': settings.enabled}) + + +class StatusStreamView(View): + """SSE endpoint — pushes ingestion status JSON every 2 seconds.""" + def get(self, request): + def event_stream(): + while True: + try: + settings = NginxSettings.get() + total = NginxAccessLog.objects.count() + last_fetch = ( + settings.last_fetch_at.isoformat() + if settings.last_fetch_at else None + ) + payload = json.dumps({ + 'enabled': settings.enabled, + 'last_fetch_at': last_fetch, + 'total_logs': total, + }) + yield f'data: {payload}\n\n' + except Exception: + pass + time.sleep(2) + + response = StreamingHttpResponse(event_stream(), content_type='text/event-stream') + response['Cache-Control'] = 'no-cache' + response['X-Accel-Buffering'] = 'no' + return response diff --git a/tailwind.config.js b/tailwind.config.js index 7e75998..7ef8fc2 100644 --- a/tailwind.config.js +++ b/tailwind.config.js @@ -3,6 +3,8 @@ module.exports = { './templates/**/*.html', './links/templates/**/*.html', './new_theme/templates/**/*.html', + './netscan/templates/**/*.html', + './nginxmon/templates/**/*.html', './static_src/**/*.{js,jsx}', './**/*.js', ], diff --git a/templates/base.html b/templates/base.html index a0823c5..8bf7060 100644 --- a/templates/base.html +++ b/templates/base.html @@ -139,6 +139,18 @@ + +
+ + + + + {% trans "Nginx Monitor" %} +
+ +