zwitschi
9f8ab0f61c
- Implemented a new utility for querying WHOIS information and consolidating IP addresses by subnet. - Added caching mechanism for WHOIS responses to improve performance. - Introduced functions for loading and saving the WHOIS cache, querying subnets, and consolidating IPs based on a threshold. feat: create Flask web UI for intrusion detection system - Developed a Flask web application to manage the intrusion detection system. - Implemented routes for scanning, reports, blocklist management, and firewall configuration. - Added helper functions for reading files and handling requests. feat: define routes for web UI - Registered routes for various web pages and API endpoints in the Flask application. - Implemented functionality for scanning, reporting, blocklist management, and firewall state retrieval. feat: implement background scan runner for web UI - Created a background scan runner to handle scan jobs in a separate thread. - Managed an in-memory job store for tracking scan jobs and their statuses. - Added persistence for scan jobs to maintain state across application restarts. chore: update start script for Flask application - Enhanced the PowerShell start script to check for the existence of the virtual environment. - Added comments and structured the script for better readability and usability.
48 lines
1.7 KiB
Python
48 lines
1.7 KiB
Python
#!/usr/bin/env python3
|
|
"""Flagging rules for the intrusion-detection system.
|
|
|
|
Detection functions that identify hostile IPs based on aggregated stats.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from . import bad_patterns
|
|
from .utils.constants import DEFAULT_ERROR_RATIO, DEFAULT_MIN_DENIED, DEFAULT_MIN_ERRORS
|
|
from .utils.models import IPStats
|
|
|
|
|
|
def flag_scanners(stats: dict[str, IPStats], min_errors: int = DEFAULT_MIN_ERRORS,
|
|
min_ratio: float = DEFAULT_ERROR_RATIO) -> dict[str, IPStats]:
|
|
"""IPs with many errors AND a high error ratio."""
|
|
return {
|
|
ip: s for ip, s in stats.items()
|
|
if s.errors >= min_errors and s.error_ratio >= min_ratio
|
|
}
|
|
|
|
|
|
def flag_denied_only(stats: dict[str, IPStats],
|
|
min_denied: int = DEFAULT_MIN_DENIED) -> dict[str, IPStats]:
|
|
"""IPs that only ever hit denied paths and never authenticated."""
|
|
return {
|
|
ip: s for ip, s in stats.items()
|
|
if s.denied >= min_denied and not s.authenticated
|
|
}
|
|
|
|
|
|
def flag_malicious_patterns(stats: dict[str, IPStats],
|
|
min_severity: int = 2,
|
|
min_hits: int = 1) -> dict[str, IPStats]:
|
|
"""IPs with hits against malicious path patterns at or above severity threshold."""
|
|
return {
|
|
ip: s for ip, s in stats.items()
|
|
if s.malicious_hits >= min_hits and any(
|
|
bad_patterns.PATTERNS_BY_NAME[name].severity >= min_severity
|
|
for name in s.malicious_patterns
|
|
)
|
|
}
|
|
|
|
|
|
def by_total(stats: dict[str, IPStats]) -> list[IPStats]:
|
|
"""Stats sorted by request count, descending."""
|
|
return sorted(stats.values(), key=lambda s: (-s.total, s.ip))
|