hash/FunguyBot
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

refactor: async I/O, input sanitisation, and shared utilities cleanup

Browse Source
This commit is contained in:
Hash Borgir
2026-05-08 22:59:31 -05:00
parent 52a9621d50
commit f822d6a450
21 changed files with 1351 additions and 2709 deletions
Download Patch File Download Diff File Expand all files Collapse all files
plugins/headers.py
+63 -125
View File
@@ -3,27 +3,18 @@ This plugin provides comprehensive HTTP security header analysis.
"""
import logging
import requests
import aiohttp
import asyncio
import simplematrixbotlib as botlib
from urllib.parse import urlparse
import ssl
import socket
from plugins.utils import is_public_destination
import datetime
from plugins.common import is_public_destination, collapsible_summary, html_escape
async def handle_command(room, message, bot, prefix, config):
"""
Function to handle !headers command for HTTP security header analysis.
Args:
room (Room): The Matrix room where the command was invoked.
message (RoomMessage): The message object containing the command.
bot (Bot): The bot object.
prefix (str): The command prefix.
config (dict): Configuration parameters.
Returns:
None
"""
match = botlib.MessageMatch(room, message, bot, prefix)
if match.is_not_from_this_bot() and match.prefix() and match.command("headers"):
@@ -74,7 +65,7 @@ async def show_usage(room, bot):
async def analyze_headers(room, bot, url):
"""Perform comprehensive HTTP security header analysis."""
try:
await bot.api.send_text_message(room.room_id, f"🔍 Analyzing security headers for: {url}")
await bot.api.send_text_message(room.room_id, f"🔍 Analyzing security headers for: {html_escape(url)}")
results = {
'url': url,
@@ -121,65 +112,60 @@ async def analyze_headers(room, bot, url):
async def analyze_http_response(results, url):
"""Analyze HTTP response and redirect chain."""
try:
session = requests.Session()
session.max_redirects = 5
response = session.get(url, timeout=10, allow_redirects=True)
results['final_url'] = response.url
results['status_code'] = response.status_code
results['http_headers'] = dict(response.headers)
# Check if redirects to HTTPS
results['redirects_to_https'] = response.url.startswith('https://')
# Store redirect history
results['redirect_chain'] = [{
'url': resp.url,
'status_code': resp.status_code,
'headers': dict(resp.headers)
} for resp in response.history]
except requests.exceptions.SSLError:
results['ssl_error'] = True
except requests.exceptions.RequestException as e:
async with aiohttp.ClientSession() as session:
async with session.get(url, allow_redirects=True, timeout=aiohttp.ClientTimeout(total=10)) as response:
results['final_url'] = str(response.url)
results['status_code'] = response.status
results['http_headers'] = dict(response.headers)
results['redirects_to_https'] = response.url.scheme == 'https'
# aiohttp doesn't give access to redirect history easily, so we'll mark if final URL differs
if str(response.url) != url:
results['redirect_chain'] = [{'url': url, 'status_code': 301}] # simplified
except aiohttp.ClientError as e:
results['http_error'] = str(e)
async def analyze_https_response(results, url):
"""Analyze HTTPS response headers."""
try:
response = requests.get(url, timeout=10, allow_redirects=False)
results['https_headers'] = dict(response.headers)
results['https_status'] = response.status_code
except requests.exceptions.RequestException as e:
async with aiohttp.ClientSession() as session:
async with session.get(url, allow_redirects=False, timeout=aiohttp.ClientTimeout(total=10)) as response:
results['https_headers'] = dict(response.headers)
results['https_status'] = response.status
except aiohttp.ClientError as e:
results['https_error'] = str(e)
async def analyze_ssl_certificate(results, domain):
"""Analyze SSL certificate information."""
try:
context = ssl.create_default_context()
with socket.create_connection((domain, 443), timeout=10) as sock:
with context.wrap_socket(sock, server_hostname=domain) as ssock:
cert = ssock.getpeercert()
"""Analyze SSL certificate information (run in thread to avoid event loop blocking)."""
def _get_cert():
try:
context = ssl.create_default_context()
with socket.create_connection((domain, 443), timeout=10) as sock:
with context.wrap_socket(sock, server_hostname=domain) as ssock:
cert = ssock.getpeercert()
return {
'subject': dict(x[0] for x in cert['subject']),
'issuer': dict(x[0] for x in cert['issuer']),
'not_before': cert['notBefore'],
'not_after': cert['notAfter'],
'san': cert.get('subjectAltName', []),
'version': cert.get('version'),
'serial_number': cert.get('serialNumber')
}
except Exception as e:
return f"Error: {e}"
results['ssl_info'] = {
'subject': dict(x[0] for x in cert['subject']),
'issuer': dict(x[0] for x in cert['issuer']),
'not_before': cert['notBefore'],
'not_after': cert['notAfter'],
'san': cert.get('subjectAltName', []),
'version': cert.get('version'),
'serial_number': cert.get('serialNumber')
}
except Exception as e:
results['ssl_error'] = str(e)
loop = asyncio.get_running_loop()
ssl_data = await loop.run_in_executor(None, _get_cert)
if isinstance(ssl_data, str):
results['ssl_error'] = ssl_data
else:
results['ssl_info'] = ssl_data
async def calculate_security_score(results):
"""Calculate overall security score based on headers and configuration."""
score = 100
missing_headers = []
# Critical security headers
critical_headers = [
'Strict-Transport-Security',
'Content-Security-Policy',
@@ -239,7 +225,6 @@ async def generate_recommendations(results):
recommendations = []
headers = results.get('https_headers') or results.get('http_headers', {})
# HSTS recommendations
if 'Strict-Transport-Security' not in headers:
recommendations.append("🔒 Implement HSTS header with max-age=31536000, includeSubDomains, and preload")
else:
@@ -251,35 +236,21 @@ async def generate_recommendations(results):
if 'preload' not in hsts:
recommendations.append("🔒 Consider adding preload directive to HSTS for browser preloading")
# CSP recommendations
if 'Content-Security-Policy' not in headers:
recommendations.append("🛡️ Implement Content Security Policy to prevent XSS attacks")
else:
csp = headers['Content-Security-Policy']
if "default-src 'self'" not in csp and "default-src 'none'" not in csp:
recommendations.append("🛡️ Restrict CSP default-src to 'self' or specific origins")
# Frame options
if 'X-Frame-Options' not in headers:
recommendations.append("🚫 Add X-Frame-Options header to prevent clickjacking (DENY or SAMEORIGIN)")
# Content type options
if 'X-Content-Type-Options' not in headers:
recommendations.append("📄 Add X-Content-Type-Options: nosniff to prevent MIME type sniffing")
# Referrer policy
if 'Referrer-Policy' not in headers:
recommendations.append("🔗 Implement Referrer-Policy to control referrer information leakage")
# Feature policy
if 'Feature-Policy' not in headers and 'Permissions-Policy' not in headers:
recommendations.append("⚙️ Implement Feature-Policy/Permissions-Policy to restrict browser features")
# Remove server information
if 'Server' in headers or 'X-Powered-By' in headers:
recommendations.append("🕵️ Remove Server and X-Powered-By headers to avoid information disclosure")
# HTTPS enforcement
if not results.get('redirects_to_https') and not results['url'].startswith('https://'):
recommendations.append("🔐 Implement HTTP to HTTPS redirects")
@@ -287,7 +258,8 @@ async def generate_recommendations(results):
async def format_header_analysis(results):
"""Format the header analysis results for display."""
output = f"<strong>🔒 Security Headers Analysis: {results['url']}</strong><br><br>"
safe_url = html_escape(results['url'])
output = f"<strong>🔒 Security Headers Analysis: {safe_url}</strong><br><br>"
# Security Score
score = results['security_score']
@@ -296,13 +268,12 @@ async def format_header_analysis(results):
# Basic Information
output += "<strong>📊 Basic Information</strong><br>"
output += f" • <strong>Final URL:</strong> {results.get('final_url', 'N/A')}<br>"
output += f" • <strong>Final URL:</strong> {html_escape(results.get('final_url', 'N/A'))}<br>"
output += f" • <strong>Status Code:</strong> {results.get('status_code', 'N/A')}<br>"
if results.get('redirects_to_https'):
output += f" • <strong>HTTPS Redirect:</strong> ✅ Enforced<br>"
else:
output += f" • <strong>HTTPS Redirect:</strong> ❌ Not enforced<br>"
output += f" • <strong>Redirect Chain:</strong> {len(results.get('redirect_chain', []))} hops<br>"
output += "<br>"
# Security Headers Analysis
@@ -310,10 +281,10 @@ async def format_header_analysis(results):
output += "<strong>🛡️ Security Headers Analysis</strong><br>"
security_headers = {
'Strict-Transport-Security': ('🔒', 'HSTS - HTTP Strict Transport Security'),
'Content-Security-Policy': ('🛡️', 'CSP - Content Security Policy'),
'Strict-Transport-Security': ('🔒', 'HSTS'),
'Content-Security-Policy': ('🛡️', 'CSP'),
'X-Frame-Options': ('🚫', 'Clickjacking Protection'),
'X-Content-Type-Options': ('📄', 'MIME Type Sniffing Protection'),
'X-Content-Type-Options': ('📄', 'MIME Sniffing'),
'X-XSS-Protection': ('❌', 'XSS Protection (Deprecated)'),
'Referrer-Policy': ('🔗', 'Referrer Policy'),
'Feature-Policy': ('⚙️', 'Feature Policy'),
@@ -322,88 +293,55 @@ async def format_header_analysis(results):
for header, (emoji, description) in security_headers.items():
if header in headers:
value = headers[header]
if len(value) > 100:
value = value[:100] + "..."
value = html_escape(str(headers[header]))[:100]
output += f" • {emoji} <strong>{header}:</strong> ✅ {value}<br>"
else:
output += f" • {emoji} <strong>{header}:</strong> ❌ Missing<br>"
output += "<br>"
# Other Headers (Information Disclosure)
output += "<strong>📋 Other Headers</strong><br>"
info_headers = ['Server', 'X-Powered-By', 'X-AspNet-Version']
for header in info_headers:
for header in ['Server', 'X-Powered-By']:
if header in headers:
output += f" • 🔍 <strong>{header}:</strong> {headers[header]}<br>"
output += f" • 🔍 <strong>{header}:</strong> {html_escape(str(headers[header]))}<br>"
output += "<br>"
# SSL Certificate Information (if available)
if results.get('ssl_info'):
if results.get('ssl_info') and 'subject' in results['ssl_info']:
output += "<strong>🔐 SSL Certificate</strong><br>"
ssl_info = results['ssl_info']
if ssl_info.get('subject'):
output += f" • <strong>Subject:</strong> {ssl_info['subject'].get('commonName', 'N/A')}<br>"
output += f" • <strong>Subject:</strong> {html_escape(ssl_info['subject'].get('commonName', 'N/A'))}<br>"
if ssl_info.get('issuer'):
output += f" • <strong>Issuer:</strong> {ssl_info['issuer'].get('organizationName', 'N/A')}<br>"
output += f" • <strong>Issuer:</strong> {html_escape(ssl_info['issuer'].get('organizationName', 'N/A'))}<br>"
if ssl_info.get('not_after'):
output += f" • <strong>Expires:</strong> {ssl_info['not_after']}<br>"
if ssl_info.get('san'):
san_count = len([san for san in ssl_info['san'] if san[0] == 'DNS'])
output += f" • <strong>SAN Entries:</strong> {san_count}<br>"
output += f" • <strong>Expires:</strong> {html_escape(ssl_info['not_after'])}<br>"
output += "<br>"
# Recommendations
if results.get('recommendations'):
output += "<strong>💡 Security Recommendations</strong><br>"
for rec in results['recommendations'][:8]: # Show first 8 recommendations
for rec in results['recommendations'][:8]:
output += f" • {rec}<br>"
if len(results['recommendations']) > 8:
output += f" • ... and {len(results['recommendations']) - 8} more recommendations<br>"
output += "<br>"
# Missing Headers Summary
if results.get('missing_headers'):
output += "<strong>⚠️ Critical Headers Missing</strong><br>"
for header in results['missing_headers']:
output += f" • ❌ {header}<br>"
output += "<br>"
# Security Rating
score = results['security_score']
# Final rating
if score >= 80:
rating = "🟢 Excellent"
description = "Strong security headers configuration"
elif score >= 60:
rating = "🟡 Good"
description = "Moderate security, room for improvement"
elif score >= 40:
rating = "🟠 Fair"
description = "Basic security, significant improvements needed"
else:
rating = "🔴 Poor"
description = "Weak security headers configuration"
output += f"<strong>📈 Security Rating:</strong> {rating}<br>"
output += f"<strong>📝 Assessment:</strong> {description}<br>"
# Wrap in collapsible if content is large
if len(output) > 1000:
output = f"<details><summary><strong>🔒 Security Headers Analysis: {results['url']}</strong></summary>{output}</details>"
# Wrap in collapsible details
return collapsible_summary(f"🔒 Security Headers Analysis: {safe_url} (Score: {score}/100)", output)
return output
# ---------------------------------------------------------------------------
# Plugin Metadata
# ---------------------------------------------------------------------------
__version__ = "1.0.1"
__version__ = "1.0.2"
__author__ = "Funguy Bot"
__description__ = "HTTP security header analysis"
__description__ = "HTTP security header analysis (SSRF‑safe, async)"
__help__ = """
<details>
<summary><strong>!headers</strong> – HTTP security header scanner</summary>