mirror of
https://github.com/wahyd4/links.git
synced 2026-08-08 21:04:53 +10:00
225 lines
8.1 KiB
Python
225 lines
8.1 KiB
Python
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)
|
||
excluded_ips = models.TextField(
|
||
blank=True,
|
||
default='',
|
||
help_text=(
|
||
'One entry per line. Supports exact IPs (192.168.1.218), '
|
||
'wildcards (192.168.1.x), and CIDR ranges (192.168.1.0/24). '
|
||
'Matching requests will not be ingested.'
|
||
),
|
||
)
|
||
auto_ban_enabled = models.BooleanField(
|
||
default=True,
|
||
help_text='Automatically ban IPs that repeatedly probe auth endpoints.',
|
||
)
|
||
auto_ban_threshold = models.IntegerField(
|
||
default=5,
|
||
help_text='Number of auth-probe requests from one IP within the window before auto-ban.',
|
||
)
|
||
auto_ban_window_hours = models.IntegerField(
|
||
default=24,
|
||
help_text='Rolling window (hours) used to count auth-probe requests.',
|
||
)
|
||
|
||
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)
|
||
# Auto-ban annotation
|
||
note = models.TextField(blank=True, default='')
|
||
# 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
|
||
|
||
|
||
class BannedIP(models.Model):
|
||
"""
|
||
Permanent record of every IP banned by the system (auto or manual).
|
||
The IP is also pushed to the Kubernetes nginx ConfigMap block list.
|
||
"""
|
||
BAN_SOURCES = [
|
||
('manual', 'Manual'),
|
||
('auto_auth_probe', 'Auto: Auth Probe'),
|
||
('auto_php_scan', 'Auto: PHP Scanner'),
|
||
('auto_404_flood', 'Auto: 404 Flood'),
|
||
]
|
||
|
||
ip = models.GenericIPAddressField(unique=True, db_index=True)
|
||
ban_source = models.CharField(max_length=30, choices=BAN_SOURCES, default='manual')
|
||
reason = models.TextField()
|
||
request_count = models.IntegerField(default=0)
|
||
banned_at = models.DateTimeField(auto_now_add=True)
|
||
unbanned_at = models.DateTimeField(null=True, blank=True)
|
||
country = models.CharField(max_length=100, blank=True)
|
||
city = models.CharField(max_length=100, blank=True)
|
||
|
||
class Meta:
|
||
ordering = ['-banned_at']
|
||
|
||
@property
|
||
def is_unbanned(self):
|
||
return self.unbanned_at is not None
|
||
|
||
def __str__(self):
|
||
return f'{self.ip} [{self.get_ban_source_display()}] @ {self.banned_at:%Y-%m-%d %H:%M}'
|