Optimize scheduled tasks to reduce unnecessary processing

- Add conditional execution to export-malicious-ips task: only runs
    when honeypot was accessed in last 5 minutes
  - Add since_minutes parameter to get_access_logs() for time filtering
  - Optimize analyze-ips task to only process IPs with activity in the
    last minute, fetching full history per-IP instead of all logs
  - Exclude RFC1918 private addresses and non-routable IPs from IP
    reputation enrichment (10.x, 172.16-31.x, 192.168.x, 127.x, 169.254.x)
This commit is contained in:
Phillip Tarrant
2026-01-15 13:30:35 -06:00
parent 554bd486da
commit 541b5d0f1b
4 changed files with 48 additions and 9 deletions

View File

@@ -1,6 +1,8 @@
# tasks/export_malicious_ips.py
import os
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo
from logger import get_app_logger
from database import get_database
from models import AccessLog
@@ -24,6 +26,15 @@ OUTPUT_FILE = os.path.join(EXPORTS_DIR, "malicious_ips.txt")
# ----------------------
# TASK LOGIC
# ----------------------
def has_recent_honeypot_access(session, minutes: int = 5) -> bool:
"""Check if honeypot was accessed in the last N minutes."""
cutoff_time = datetime.now(tz=ZoneInfo('UTC')) - timedelta(minutes=minutes)
count = session.query(AccessLog).filter(
AccessLog.is_honeypot_trigger == True,
AccessLog.timestamp >= cutoff_time
).count()
return count > 0
def main():
"""
Export all IPs flagged as suspicious to a text file.
@@ -36,6 +47,11 @@ def main():
db = get_database()
session = db.session
# Check for recent honeypot activity
if not has_recent_honeypot_access(session):
app_logger.info(f"[Background Task] {task_name} skipped - no honeypot access in last 5 minutes")
return
# Query distinct suspicious IPs
results = session.query(distinct(AccessLog.ip)).filter(
AccessLog.is_suspicious == True