Malin/krawl.es
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Linted code iwht black tool

Browse Source
This commit is contained in:
Lorenzo Venerandi
2026-01-23 22:00:21 +01:00
parent 25384585d9
commit 4450d3a4e3
22 changed files with 1387 additions and 868 deletions
Download Patch File Download Diff File Expand all files Collapse all files

258
src/tracker.py
View File

@@ -17,7 +17,13 @@ class AccessTracker:
Maintains in-memory structures for fast dashboard access and
persists data to SQLite for long-term storage and analysis.
"""
def __init__(self, max_pages_limit, ban_duration_seconds, db_manager: Optional[DatabaseManager] = None):
def __init__(
self,
max_pages_limit,
ban_duration_seconds,
db_manager: Optional[DatabaseManager] = None,
):
"""
Initialize the access tracker.
@@ -32,14 +38,32 @@ class AccessTracker:
self.user_agent_counts: Dict[str, int] = defaultdict(int)
self.access_log: List[Dict] = []
self.credential_attempts: List[Dict] = []
# Track pages visited by each IP (for good crawler limiting)
self.ip_page_visits: Dict[str, Dict[str, object]] = defaultdict(dict)
self.suspicious_patterns = [
'bot', 'crawler', 'spider', 'scraper', 'curl', 'wget', 'python-requests',
'scanner', 'nikto', 'sqlmap', 'nmap', 'masscan', 'nessus', 'acunetix',
'burp', 'zap', 'w3af', 'metasploit', 'nuclei', 'gobuster', 'dirbuster'
"bot",
"crawler",
"spider",
"scraper",
"curl",
"wget",
"python-requests",
"scanner",
"nikto",
"sqlmap",
"nmap",
"masscan",
"nessus",
"acunetix",
"burp",
"zap",
"w3af",
"metasploit",
"nuclei",
"gobuster",
"dirbuster",
]
# Load attack patterns from wordlists
@@ -49,11 +73,11 @@ class AccessTracker:
# Fallback if wordlists not loaded
if not self.attack_types:
self.attack_types = {
'path_traversal': r'\.\.',
'sql_injection': r"('|--|;|\bOR\b|\bUNION\b|\bSELECT\b|\bDROP\b)",
'xss_attempt': r'(<script|javascript:|onerror=|onload=)',
'common_probes': r'(wp-admin|phpmyadmin|\.env|\.git|/admin|/config)',
'shell_injection': r'(\||;|`|\$\(|&&)',
"path_traversal": r"\.\.",
"sql_injection": r"('|--|;|\bOR\b|\bUNION\b|\bSELECT\b|\bDROP\b)",
"xss_attempt": r"(<script|javascript:|onerror=|onload=)",
"common_probes": r"(wp-admin|phpmyadmin|\.env|\.git|/admin|/config)",
"shell_injection": r"(\||;|`|\$\(|&&)",
}
# Track IPs that accessed honeypot paths from robots.txt
@@ -94,14 +118,22 @@ class AccessTracker:
parsed = urllib.parse.parse_qs(post_data)
# Common username field names
username_fields = ['username', 'user', 'login', 'email', 'log', 'userid', 'account']
username_fields = [
"username",
"user",
"login",
"email",
"log",
"userid",
"account",
]
for field in username_fields:
if field in parsed and parsed[field]:
username = parsed[field][0]
break
# Common password field names
password_fields = ['password', 'pass', 'passwd', 'pwd', 'passphrase']
password_fields = ["password", "pass", "passwd", "pwd", "passphrase"]
for field in password_fields:
if field in parsed and parsed[field]:
password = parsed[field][0]
@@ -109,8 +141,12 @@ class AccessTracker:
except Exception:
# If parsing fails, try simple regex patterns
username_match = re.search(r'(?:username|user|login|email|log)=([^&\s]+)', post_data, re.IGNORECASE)
password_match = re.search(r'(?:password|pass|passwd|pwd)=([^&\s]+)', post_data, re.IGNORECASE)
username_match = re.search(
r"(?:username|user|login|email|log)=([^&\s]+)", post_data, re.IGNORECASE
)
password_match = re.search(
r"(?:password|pass|passwd|pwd)=([^&\s]+)", post_data, re.IGNORECASE
)
if username_match:
username = urllib.parse.unquote_plus(username_match.group(1))
@@ -119,29 +155,30 @@ class AccessTracker:
return username, password
def record_credential_attempt(self, ip: str, path: str, username: str, password: str):
def record_credential_attempt(
self, ip: str, path: str, username: str, password: str
):
"""
Record a credential login attempt.
Stores in both in-memory list and SQLite database.
"""
# In-memory storage for dashboard
self.credential_attempts.append({
'ip': ip,
'path': path,
'username': username,
'password': password,
'timestamp': datetime.now().isoformat()
})
self.credential_attempts.append(
{
"ip": ip,
"path": path,
"username": username,
"password": password,
"timestamp": datetime.now().isoformat(),
}
)
# Persist to database
if self.db:
try:
self.db.persist_credential(
ip=ip,
path=path,
username=username,
password=password
ip=ip, path=path, username=username, password=password
)
except Exception:
# Don't crash if database persistence fails
@@ -151,9 +188,9 @@ class AccessTracker:
self,
ip: str,
path: str,
user_agent: str = '',
body: str = '',
method: str = 'GET'
user_agent: str = "",
body: str = "",
method: str = "GET",
):
"""
Record an access attempt.
@@ -180,9 +217,9 @@ class AccessTracker:
attack_findings.extend(self.detect_attack_type(body))
is_suspicious = (
self.is_suspicious_user_agent(user_agent) or
self.is_honeypot_path(path) or
len(attack_findings) > 0
self.is_suspicious_user_agent(user_agent)
or self.is_honeypot_path(path)
or len(attack_findings) > 0
)
is_honeypot = self.is_honeypot_path(path)
@@ -191,15 +228,17 @@ class AccessTracker:
self.honeypot_triggered[ip].append(path)
# In-memory storage for dashboard
self.access_log.append({
'ip': ip,
'path': path,
'user_agent': user_agent,
'suspicious': is_suspicious,
'honeypot_triggered': self.is_honeypot_path(path),
'attack_types':attack_findings,
'timestamp': datetime.now().isoformat()
})
self.access_log.append(
{
"ip": ip,
"path": path,
"user_agent": user_agent,
"suspicious": is_suspicious,
"honeypot_triggered": self.is_honeypot_path(path),
"attack_types": attack_findings,
"timestamp": datetime.now().isoformat(),
}
)
# Persist to database
if self.db:
@@ -211,13 +250,13 @@ class AccessTracker:
method=method,
is_suspicious=is_suspicious,
is_honeypot_trigger=is_honeypot,
attack_types=attack_findings if attack_findings else None
attack_types=attack_findings if attack_findings else None,
)
except Exception:
# Don't crash if database persistence fails
pass
def detect_attack_type(self, data:str) -> list[str]:
def detect_attack_type(self, data: str) -> list[str]:
"""
Returns a list of all attack types found in path data
"""
@@ -230,27 +269,37 @@ class AccessTracker:
def is_honeypot_path(self, path: str) -> bool:
"""Check if path is one of the honeypot traps from robots.txt"""
honeypot_paths = [
'/admin',
'/admin/',
'/backup',
'/backup/',
'/config',
'/config/',
'/private',
'/private/',
'/database',
'/database/',
'/credentials.txt',
'/passwords.txt',
'/admin_notes.txt',
'/api_keys.json',
'/.env',
'/wp-admin',
'/wp-admin/',
'/phpmyadmin',
'/phpMyAdmin/'
"/admin",
"/admin/",
"/backup",
"/backup/",
"/config",
"/config/",
"/private",
"/private/",
"/database",
"/database/",
"/credentials.txt",
"/passwords.txt",
"/admin_notes.txt",
"/api_keys.json",
"/.env",
"/wp-admin",
"/wp-admin/",
"/phpmyadmin",
"/phpMyAdmin/",
]
return path in honeypot_paths or any(hp in path.lower() for hp in ['/backup', '/admin', '/config', '/private', '/database', 'phpmyadmin'])
return path in honeypot_paths or any(
hp in path.lower()
for hp in [
"/backup",
"/admin",
"/config",
"/private",
"/database",
"phpmyadmin",
]
)
def is_suspicious_user_agent(self, user_agent: str) -> bool:
"""Check if user agent matches suspicious patterns"""
@@ -263,34 +312,36 @@ class AccessTracker:
"""
Check if an IP has been categorized as a 'good crawler' in the database.
Uses the IP category from IpStats table.
Args:
client_ip: The client IP address (will be sanitized)
Returns:
True if the IP is categorized as 'good crawler', False otherwise
"""
try:
from sanitizer import sanitize_ip
# Sanitize the IP address
safe_ip = sanitize_ip(client_ip)
# Query the database for this IP's category
db = self.db
if not db:
return False
ip_stats = db.get_ip_stats_by_ip(safe_ip)
if not ip_stats or not ip_stats.get('category'):
if not ip_stats or not ip_stats.get("category"):
return False
# Check if category matches "good crawler"
category = ip_stats.get('category', '').lower().strip()
category = ip_stats.get("category", "").lower().strip()
return category
except Exception as e:
# Log but don't crash on database errors
import logging
logging.error(f"Error checking IP category for {client_ip}: {str(e)}")
return False
@@ -298,10 +349,10 @@ class AccessTracker:
"""
Increment page visit counter for an IP and return the new count.
If ban timestamp exists and 60+ seconds have passed, reset the counter.
Args:
client_ip: The client IP address
Returns:
The updated page visit count for this IP
"""
@@ -309,55 +360,58 @@ class AccessTracker:
# Initialize if not exists
if client_ip not in self.ip_page_visits:
self.ip_page_visits[client_ip] = {"count": 0, "ban_timestamp": None}
# Increment count
self.ip_page_visits[client_ip]["count"] += 1
# Set ban if reached limit
if self.ip_page_visits[client_ip]["count"] >= self.max_pages_limit:
self.ip_page_visits[client_ip]["ban_timestamp"] = datetime.now().isoformat()
self.ip_page_visits[client_ip][
"ban_timestamp"
] = datetime.now().isoformat()
return self.ip_page_visits[client_ip]["count"]
except Exception:
return 0
def is_banned_ip(self, client_ip: str) -> bool:
"""
Check if an IP is currently banned due to exceeding page visit limits.
Args:
client_ip: The client IP address
Returns:
True if the IP is banned, False otherwise
"""
"""
try:
if client_ip in self.ip_page_visits:
ban_timestamp = self.ip_page_visits[client_ip]["ban_timestamp"]
if ban_timestamp is not None:
banned = True
#Check if ban period has expired (> 60 seconds)
ban_time = datetime.fromisoformat(self.ip_page_visits[client_ip]["ban_timestamp"])
# Check if ban period has expired (> 60 seconds)
ban_time = datetime.fromisoformat(
self.ip_page_visits[client_ip]["ban_timestamp"]
)
time_diff = datetime.now() - ban_time
if time_diff.total_seconds() > self.ban_duration_seconds:
self.ip_page_visits[client_ip]["count"] = 0
self.ip_page_visits[client_ip]["ban_timestamp"] = None
banned = False
return banned
except Exception:
return False
def get_page_visit_count(self, client_ip: str) -> int:
"""
Get the current page visit count for an IP.
Args:
client_ip: The client IP address
Returns:
The page visit count for this IP
"""
@@ -372,20 +426,24 @@ class AccessTracker:
def get_top_paths(self, limit: int = 10) -> List[Tuple[str, int]]:
"""Get top N paths by access count"""
return sorted(self.path_counts.items(), key=lambda x: x[1], reverse=True)[:limit]
return sorted(self.path_counts.items(), key=lambda x: x[1], reverse=True)[
:limit
]
def get_top_user_agents(self, limit: int = 10) -> List[Tuple[str, int]]:
"""Get top N user agents by access count"""
return sorted(self.user_agent_counts.items(), key=lambda x: x[1], reverse=True)[:limit]
return sorted(self.user_agent_counts.items(), key=lambda x: x[1], reverse=True)[
:limit
]
def get_suspicious_accesses(self, limit: int = 20) -> List[Dict]:
"""Get recent suspicious accesses"""
suspicious = [log for log in self.access_log if log.get('suspicious', False)]
suspicious = [log for log in self.access_log if log.get("suspicious", False)]
return suspicious[-limit:]
def get_attack_type_accesses(self, limit: int = 20) -> List[Dict]:
"""Get recent accesses with detected attack types"""
attacks = [log for log in self.access_log if log.get('attack_types')]
attacks = [log for log in self.access_log if log.get("attack_types")]
return attacks[-limit:]
def get_honeypot_triggered_ips(self) -> List[Tuple[str, List[str]]]:
@@ -401,12 +459,12 @@ class AccessTracker:
stats = self.db.get_dashboard_counts()
# Add detailed lists from database
stats['top_ips'] = self.db.get_top_ips(10)
stats['top_paths'] = self.db.get_top_paths(10)
stats['top_user_agents'] = self.db.get_top_user_agents(10)
stats['recent_suspicious'] = self.db.get_recent_suspicious(20)
stats['honeypot_triggered_ips'] = self.db.get_honeypot_triggered_ips()
stats['attack_types'] = self.db.get_recent_attacks(20)
stats['credential_attempts'] = self.db.get_credential_attempts(limit=50)
stats["top_ips"] = self.db.get_top_ips(10)
stats["top_paths"] = self.db.get_top_paths(10)
stats["top_user_agents"] = self.db.get_top_user_agents(10)
stats["recent_suspicious"] = self.db.get_recent_suspicious(20)
stats["honeypot_triggered_ips"] = self.db.get_honeypot_triggered_ips()
stats["attack_types"] = self.db.get_recent_attacks(20)
stats["credential_attempts"] = self.db.get_credential_attempts(limit=50)
return stats