Monitoring for Website Security Vulnerabilities: A Defensive Guide

#!/usr/bin/env python3
import ssl
import socket
import datetime
import OpenSSL
import requests
from cryptography import x509
from cryptography. hazmat.backends import default_backend

def check_ssl_configuration (hostname, port=443):
"""Check SSL/TLS configuration for a given hostname."""
results = {
'hostname': hostname,
'port': port,
'timestamp': datetime.datetime .now().isoformat(),
'errors': [],
'warnings': [],
'certificate': {},
'protocols': {},
'cipher_suites': [],
'security_headers': {}
}

# Check certificate details
try:
# Connect and get certificate
context = ssl.create_ default_context()
with socket.create_ connection ((hostname, port)) as sock:
with context.wrap_ socket(sock, server_ hostname= hostname) as ssock:
cert_binary = ssock.getpeercert (binary_form= True)
cert = x509.load_der _x509_certificate (cert_binary, default_backend())

# Certificate details
results ['certificate'] ['subject'] = str (cert.subject)
results ['certificate'] ['issuer'] = str (cert.issuer)
results ['certificate'] ['not_valid_before'] = cert .not_valid _before.isoformat()
results ['certificate'] ['not_valid_after'] = cert.not_valid _after.isoformat()
results ['certificate'] ['serial_number'] = cert.serial_number

# Certificate expiration check
days_to_expiry = (cert .not _valid_after - datetime.datetime .now()).days
results ['certificate'] ['days_to_expiry'] = days_to_expiry

if days_to_expiry < 7:
results ['errors'].append (f"Certificate expires in {days_to_expiry} days")
elif days_to_expiry < 30:
results ['warnings']. append (f"Certificate expires in {days_to_expiry} days")

# Protocol version
results ['protocols'] ['version'] = ssock.version()

# Check for weak protocols
if results ['protocols'] ['version'] in ['SSLv2', 'SSLv3', 'TLSv1', 'TLSv1.1']:
results ['errors'] .append (f"Weak protocol in use: {results ['protocols'] ['version']}")

# Get cipher suite
results ['cipher_suites'] .append (ssock.cipher())

except ssl.SSLError as e:
results['errors'] .append (f"SSL Error: {str(e)}")
except socket.error as e:
results['errors'] .append (f"Connection Error: {str(e)}")

# Check security headers
try:
response = requests.get (f"https://{hostname}")
headers = response.headers

# Check HSTS
if 'Strict-Transport-Security' in headers:
results ['security_headers'] ['hsts'] = headers ['Strict-Transport-Security']
else:
results['warnings'] .append ("HSTS header not implemented")

# Check other security headers
security_headers = {
'Content-Security-Policy': headers.get ('Content-Security-Policy', None),
'X-Content-Type-Options': headers.get ('X-Content-Type-Options', None),
'X-Frame-Options': headers.get ('X-Frame-Options', None),
'X-XSS-Protection': headers.get ('X-XSS-Protection', None),
}
results ['security_headers'] .update (security_headers)

# Validate security headers
if not security_headers ['X-Content- Type-Options']:
results ['warnings']. append ("X-Content- Type-Options header not implemented")
if not security_headers ['X-Frame-Options']:
results['warnings'] .append ("X-Frame-Options header not implemented")

except requests.exceptions .RequestException as e:
results ['errors'] .append (f"Request Error: {str(e)}")

return results

# Example usage
if name == "main":
results = check_ssl_ configuration ("example.com")
print (f"Certificate expires in {results ['certificate'] ['days_to_expiry']} days")
print (f"Protocol version: {results ['protocols'] ['version']}")
print (f"Warnings: {results ['warnings']}")
print (f"Errors: {results ['errors']}")

import requests
import json
from urllib.parse import urlparse

def check_csp_ configuration(url):
"""Check Content Security Policy configuration and effectiveness."""
results = {
'url': url,
'csp_found': False,
'csp_report_only': False,
'directives': {},
'missing_directives': [],
'warnings': [],
'recommendations': []
}

# Essential CSP directives to check for
essential_directives = [
'default-src',
'script-src',
'style-src',
'img-src',
'connect-src',
'font-src',
'object-src',
'frame-src',
'frame-ancestors'
]

try:
response = requests.get(url)
headers = response.headers

# Check for CSP header in enforcement mode
if 'Content-Security-Policy' in headers:
results['csp_found'] = True
csp = headers ['Content-Security-Policy']
results['policy'] = csp

# Parse directives
directives = csp.split(';')
for directive in directives:
directive = directive.strip()
if not directive:
continue

parts = directive.split(None, 1)
directive_name = parts[0]
directive_value = parts[1] if len(parts) > 1 else ""
results ['directives'] [directive_name] = directive_value

# Check for missing essential directives
for directive in essential_directives:
if directive not in results['directives']:
results ['missing_directives'] .append(directive)

# Check for unsafe practices
if 'script-src' in results ['directives']:
script_src = results ['directives'] ['script-src']
if "'unsafe-inline'" in script_src:
results ['warnings'].append ("'unsafe-inline' in script-src reduces XSS protection")
if "'unsafe-eval'" in script_src:
results ['warnings'] .append ("'unsafe-eval' in script-src reduces XSS protection")
if '' in script_src:
results ['warnings'] .append ("Wildcard () in script-src reduces security")

# Check object-src and frame-ancestors for clickjacking protection
if 'object-src' not in results['directives']:
results ['recommendations'].append ("Add 'object-src' directive to prevent object injections")

if 'frame-ancestors' not in results['directives']:
results ['recommendations'].append ("Add 'frame-ancestors' directive to prevent clickjacking")

# Check for Report-Only mode
elif 'Content-Security- Policy-Report-Only' in headers:
results ['csp_found'] = True
results ['csp_report_only'] = True
results ['policy'] = headers ['Content-Security- Policy-Report-Only']
results ['warnings'].append ("CSP is in report-only mode, not enforcing policies")
else:
results ['warnings'].append ("No Content Security Policy implemented")
results ['recommendations']. append ("Implement CSP to prevent XSS and code injection attacks")

except requests.exceptions. RequestException as e:
results['errors'] = [f"Request Error: {str(e)}"]

return results

from flask import Flask, request, jsonify
import json
import logging

app = Flask(name)
logging.basicConfig (level= logging.INFO)
logger = logging. getLogger(name)

@app.route ('/csp-report', methods=['POST'])
def csp_report():
"""Endpoint to collect CSP violation reports."""
try:
report = json.loads (request.data. decode('utf-8'))

# Extract relevant details
if 'csp-report' in report:
csp_report = report ['csp-report']

# Log violation details
logger.warning (f"CSP Violation: {csp_report.get ('violated-directive', 'unknown')} - " +
f"Blocked: {csp_report.get ('blocked-uri', 'unknown')} - " +
f"Document: {csp_report.get ('document-uri', 'unknown')}")

# Store report for analysis
store_csp_ violation (csp_report)

# Check for potential attack patterns
if is_potential_attack (csp_report):
trigger_security_alert (csp_report)

return jsonify ({'status': 'received'}), 204
except Exception as e:
logger.error (f"Error processing CSP report: {str(e)}")
return jsonify ({'error': str(e)}), 500

def store_csp_violation (violation):
"""Store CSP violation for analysis."""
# Implementation would store in database or log system
pass

def is_potential_attack (violation):
"""Analyze if violation might indicate an attack."""
# Implementation would check for attack signatures
return False

def trigger_security_alert (violation):
"""Trigger security alert for suspicious violations."""
# Implementation would send alert
pass

if name == 'main':
app.run (debug=True, port=8000)

import hashlib
import requests
import re
import difflib
from bs4 import BeautifulSoup
import json
import time

class ContentChangeMonitor:
def init(self, config_file):
"""Initialize content change monitor with configuration."""
with open (config_file, 'r') as f:
self.config = json.load(f)

# Load baseline data if available
self.baselines = self._ load_baselines()

def _load_baselines (self):
"""Load baseline content for monitored pages."""
try:
with open (self.config ['baseline_file'], 'r') as f:
return json.load(f)
except (FileNotFoundError, json. JSONDecodeError):
return {}

def _save_baselines(self):
"""Save updated baseline content."""
with open (self.config ['baseline_file'], 'w') as f:
json.dump (self.baselines, f, indent=2)

def establish_baseline (self, url, force=False):
"""Create or update baseline for a URL."""
if url in self.baselines and not force:
return False

try:
response = requests.get (url)
content = response.text

# Parse content
soup = BeautifulSoup (content, 'html.parser')

# Remove dynamic elements based on config
for selector in self.config.get ('dynamic_elements', []):
for element in soup.select (selector):
element.decompose()

# Create clean content for hashing
clean_content = soup.prettify()

# Calculate hashes
full_hash = hashlib.sha256 (clean_content.encode ('utf-8')). hexdigest()

# Extract critical elements if specified
critical_elements = {}
for element_name, selector in self.config. get ('critical_elements', {}).items():
elements = soup.select (selector)
if elements:
critical_elements [element_name] = hashlib. sha256(
elements[0]. prettify().encode ('utf-8')). hexdigest()

# Store baseline
self. baselines[url] = {
'full_hash': full_hash,
'critical_ elements': critical_elements,
'last_updated': time.time()
}

self._save_baselines()
return True

except requests.exceptions. RequestException as e:
print(f"Error establishing baseline for {url}: {str(e)}")
return False

def check_for_changes(self, url):
"""Check a URL for unauthorized content changes."""
if url not in self.baselines:
print(f"No baseline established for {url}")
return {'status': 'error', 'message': 'No baseline established'}

results = {
'url': url,
'timestamp': time.time(),
'changes_detected': False,
'full_content_changed': False,
'critical_element _changes': [],
'details': {}
}

try:
response = requests.get(url)
content = response.text

# Parse content
soup = BeautifulSoup (content, 'html.parser')

# Remove dynamic elements based on config
for selector in self.config.get ('dynamic_elements', []):
for element in soup .select (selector):
element. decompose()

# Create clean content for hashing
clean_content = soup.prettify()

# Calculate full content hash
current_hash = hashlib .sha256 (clean_content.encode ('utf-8')). hexdigest()

# Check if full content has changed
if current_hash != self.baselines [url] ['full_hash']:
results ['changes_detected'] = True
results ['full_content_changed'] = True

# Calculate diff for details
baseline_content = BeautifulSoup (requests.get (url).text, 'html.parser'). prettify()
diff = difflib.unified_diff(
baseline_ content.splitlines(),
clean_content. splitlines(),
lineterm=''
)
results ['details'] ['diff'] = list (diff)

# Check critical elements
for element_name, selector in self.config. get ('critical_elements', {}).items():
if element_name in self.baselines [url] ['critical_elements']:
elements = soup.select (selector)
if elements:
current_ element_hash = hashlib.sha256 (
elements[0]. prettify().encode ('utf-8')). hexdigest()

if current_ element_hash != self.baselines [url] ['critical_elements'] [element_name]:
results ['changes_detected'] = True
results ['critical_element_changes'] .append (element_name)

return results

except requests. exceptions.RequestException as e:
return {
'status': 'error',
'message': f"Error checking {url}: {str(e)}"
}

def scan_all(self):
"""Scan all URLs in the baseline."""
results = []
for url in self.baselines:
results.append (self.check_for_changes (url))
return results

# Example usage
if name == "main":
monitor = ContentChangeMonitor ("content_monitor_ config.json")

# Establish baselines for key pages
monitor. establish_baseline ("https://example.com/")
monitor. establish_baseline ("https://example.com/login")

# Check for changes
changes = monitor. check_for_changes ("https://example.com/")
if changes ['changes_detected']:
print ("Changes detected!")
print (f"Critical elements changed: {changes ['critical_element _changes']}")
else:
print ("No changes detected.")

import subprocess
import json
import csv
import datetime
import requests
import io

def run_zap_scan (target_url, api_key):
"""Run a basic ZAP security scan against a target."""
results = {
'target': target_url,
'timestamp': datetime. datetime.now(). isoformat(),
'scan_completed': False,
'alerts': [],
'errors': []
}

try:
# Assuming ZAP is running as a daemon
zap_api_url = "http://localhost:8080"

# Start the spider
spider_scan_id = start_zap_spider (zap_api_url, api_key, target_url)
if not spider_scan_id:
results ['errors']. append ("Failed to start spider scan")
return results

# Wait for spider to complete
if not wait_for_zap_spider (zap_api_url, api_key, spider_scan_id):
results['errors'] .append ("Spider scan failed to complete")
return results

# Start the active scan
active_scan_id = start_zap_ active_scan (zap_api_url, api_key, target_url)
if not active_scan_id:
results ['errors'] .append("Failed to start active scan")
return results

# Wait for active scan to complete
if not wait_for_zap_ active_scan (zap_api_url, api_key, active_scan_id):
results ['errors'] .append ("Active scan failed to complete")
return results

# Get the alerts
alerts = get_zap_alerts (zap_api_url, api_key, target_url)
results ['alerts'] = alerts
results ['scan_completed'] = True

return results

except Exception as e:
results ['errors'] .append (f"Scan error: {str(e)}")
return results

def start_zap_spider(zap_api_url, api_key, target_url):
"""Start a ZAP spider scan."""
response = requests.get(
f"{zap_api_url} /JSON/spider /action/scan/",
params={
'apikey': api_key,
'url': target_url,
'recurse': 'true',
'inScopeOnly': 'true'
}
)

if response. status_code == 200:
data = response.json()
return data.get ('scan')
return None

def wait_for_zap_spider (zap_api_url, api_key, scan_id):
"""Wait for ZAP spider to complete."""
while True:
response = requests.get(
f" {zap_api_url} /JSON/spider/view /status/",
params={
'apikey': api_key,
'scanId': scan_id
}
)

if response. status_code == 200:
data = response. json()
status = int (data.get ('status', 0))
if status >= 100:
return True

# Sleep to avoid hammering the API
import time
time.sleep (5)
else:
return False

def start_zap_ active_scan (zap_api_url, api_key, target_url):
"""Start a ZAP active scan."""
response = requests.get(
f"{zap_api_url} /JSON/ascan /action/scan/",
params={
'apikey': api_key,
'url': target_url,
'recurse': 'true',
'inScopeOnly': 'true'
}
)

if response. status_code == 200:
data = response. json()
return data.get ('scan')
return None

def wait_for_zap_ active_scan (zap_api_url, api_key, scan_id):
"""Wait for ZAP active scan to complete."""
while True:
response = requests.get(
f"{zap_api_url} /JSON/ascan /view/status/",
params={
'apikey': api_key,
'scanId': scan_id
}
)

if response. status_code == 200:
data = response. json()
status = int(data.get ('status', 0))
if status >= 100:
return True

# Sleep to avoid hammering the API
import time
time.sleep (10)
else:
return False

def get_zap_ alerts (zap_api_url, api_key, target_url):
"""Get alerts from ZAP scan."""
response = requests.get(
f"{zap_api_url} /JSON/core /view/alerts/",
params={
'apikey': api_key,
'baseurl': target_url
}
)

if response. status_code == 200:
data = response.json()
return data.get ('alerts', [])
return []

# Example usage
if name == "main":
results = run_zap_scan ("https://example.com", "your-zap-api-key")

if results ['scan_completed']:
print (f"Scan completed with {len(results ['alerts'])} alerts")

# Group alerts by risk level
risk_levels = {"High": [], "Medium": [], "Low": [], "Informational": []}
for alert in results ['alerts']:
risk = alert.get ('risk', 'Informational')
risk_levels [risk].append (alert)

# Report high and medium risk findings
for risk in ["High", "Medium"]:
if risk_levels [risk]:
print (f"\n{risk} Risk Findings ({len(risk_levels [risk])}):")
for alert in risk_levels [risk]:
print (f"- {alert.get('name')}: {alert.get ('url')}")
else:
print (f"Scan failed: {results ['errors']}")

def check_for_injections(url):
"""Check a page for common injection patterns."""
results = {
'url': url,
'timestamp': str (datetime. datetime.now()),
'injections_detected': [],
'errors': []
}

# Common injection patterns to check for
injection_patterns = {
'javascript_injection': [
r'<script>. ?(alert|eval|document .cookie|window .location)',
r'javascript:.? (alert|eval|document .cookie|window. location)',
r'on(load |click|mouseover |error)=.? (alert|eval| document.cookie)'
],
'iframe_injection': [
r'<iframe.? src=.?>'
],
'redirect_injection': [
r'window. location.href\s=',
r'document. location\s*=',
r'location. replace('
],
'data_exfiltration': [
r'new Image\ ().src\s*=.? document.cookie',
r'fetch (.?document. cookie',
r'navigator .sendBeacon (.?'
],
'hidden_content': [
r'display: \snone.? <div.?>.?< /div>',
r'visibility: \shidden.?< div.?>.*? </div>'
],
'known_malware': [
r'googletag .pubads',
r'coinhive',
r'cryptonight',
r'minero.cc'
]
}

try:
response = requests. get(url)
content = response. text

# Check each pattern type
for injection_type, patterns in injection_patterns .items():
for pattern in patterns:
matches = re.findall (pattern, content, re.IGNORECASE)
if matches:
# Don't include the actual matches as they might contain malicious code
results ['injections_detected'] .append({
'type': injection_type,
'count': len(matches),
'pattern': pattern
})

# Check for suspicious domains
soup = BeautifulSoup (content, 'html.parser')

# Check script sources
for script in soup.find_all ('script', src= True):
src = script ['src']
if self._is_suspicious domain (src):
results ['injections_detected'] .append({
'type': 'suspicious script_source',
'value': src
})

# Check iframe sources
for iframe in soup.find_all ('iframe', src= True):
src = iframe['src']
if self._is_suspicious domain(src):
results ['injections_detected'] .append({
'type': 'suspicious iframe_source',
'value': src
})

# Check for obfuscated JavaScript
obfuscation _indicators = [
'eval(',
'atob(',
'String. fromCharCode(',
'unescape (',
'decodeURIComponent (',
'\x',
'\u00'
]

scripts = soup.find_all('script')
for script in scripts:
if script.string:
script_content = script.string
for indicator in obfuscation _indicators:
if indicator in script_content:
results ['injections_detected'] .append({
'type': 'obfuscated_javascript',
'indicator': indicator
})
break

return results

except Exception as e:
results ['errors'] .append(str(e))
return results

def is_suspicious domain (self, url):
"""Check if a URL points to a suspicious domain."""
if not url.startswith (('http://', 'https://', '//')):
return False

# Extract domain
if url. startswith('//'):
url = 'https:' + url

domain = urlparse (url).netloc

# Check against whitelist
if domain in self.config.get ('whitelisted_ domains', []):
return False

# Check against blacklist
if domain in self.config. get('blacklisted_ domains', []):
return True

# Check for suspicious TLDs
suspicious_tlds = ['.xyz', '.top', '.club', '.gq', '.tk', '.ml']
if any (domain. endswith(tld) for tld in suspicious_tlds):
return True

# Check for domain typosquatting of common services
common_services = ['google', 'facebook', 'microsoft', 'apple', 'amazon']
for service in common_services:
if service in domain and service not in self.config.get ('whitelisted_ domains', []):
# Check for slight variations
if self.levenshtein distance (service, domain) <= 2:
return True

return False

def levenshtein_distance (self, s1, s2):
"""Calculate the Levenshtein distance between two strings."""
if len(s1) < len(s2):
return self. levenshtein_distance (s2, s1)

if len(s2) == 0:
return len(s1)

previous_row = range(len (s2) + 1)
for i, c1 in enumerate (s1):
current_row = [i + 1]
for j, c2 in enumerate (s2):
insertions = previous_row [j + 1] + 1
deletions = current_row [j] + 1
substitutions = previous_row [j] + (c1 != c2)
current_row.append (min (insertions, deletions, substitutions))
previous_row = current_row
return previous_row [-1]

import json
import requests
import time
import hashlib
from enum import Enum

class AlertSeverity(Enum):
CRITICAL = 1
HIGH = 2
MEDIUM = 3
LOW = 4
INFO = 5

class AlertStatus (Enum):
NEW = "new"
ACKNOWLEDGED = "acknowledged"
INVESTIGATING = "investigating"
RESOLVED = "resolved"
FALSE_POSITIVE = "false_positive"

class SecurityAlert:
def init (self, alert_type, severity, source, details, target):
self.id = self._generate_id (alert_type, source, target, details)
self.type = alert_type
self.severity = severity
self.source = source
self.details = details
self.target = target
self.status = AlertStatus.NEW
self.created_time = int (time.time())
self.updated_time = self.created_time
self.assigned_to = None
self.resolution_notes = None
self.related_alerts = []

def _generate_id (self, alert_type, source, target, details):
"""Generate a unique ID for the alert."""
hash_input = f"{alert_type} :{source}: {target}: {json.dumps (details, sort_keys= True)}"
return hashlib.sha256 (hash_input. encode ('utf-8')). hexdigest() [:12]

def update_status (self, status, user, notes=None):
"""Update alert status."""
self.status = status
self.updated_time = int(time.time())
if status == AlertStatus.INVESTIGATING:
self. assigned_to = user
if notes:
if self. resolution_notes:
self. resolution_notes += f"\n {int(time.time ())} - {user}: {notes}"
else:
self. resolution_notes = f" {int(time.time ())} - {user}: {notes}"

def add_related_alert (self, alert_id):
"""Add a related alert ID."""
if alert_id not in self.related_alerts:
self.related _alerts.append (alert_id)

def to_dict (self):
"""Convert alert to dictionary."""
return {
'id': self.id,
'type': self.type,
'severity': self.severity .name,
'source': self.source,
'details': self.details,
'target': self.target,
'status': self.status .value,
'created_time': self.created _time,
'updated_time': self.updated _time,
'assigned_to': self.assigned _to,
'resolution_notes': self.resolution _notes,
'related_alerts': self.related _alerts
}

class SecurityAlertManager:
def init (self, config_file):
"""Initialize the security alert manager."""
with open (config_file, 'r') as f:
self.config = json.load(f)

self.alerts = {}
self.load alerts()

def _load_alerts (self):
"""Load existing alerts from storage."""
try:
with open (self.config ['alert_storage _file'], 'r') as f:
alerts_data = json.load(f)

for alert_data in alerts_data:
alert = SecurityAlert(
alert_data ['type'],
AlertSeverity [alert_data['severity']],
alert_data ['source'],
alert_data ['details'],
alert_data ['target']
)
alert.id = alert_data ['id']
alert.status = AlertStatus (alert_data['status'])
alert. created_time = alert_data ['created_time']
alert. updated_time = alert_data ['updated_time']
alert. assigned_to = alert_data ['assigned_to']
alert. resolution_notes = alert_data ['resolution_notes']
alert. related_alerts = alert_data ['related_alerts']

self.alerts [alert.id] = alert

except (FileNotFoundError, json. JSONDecodeError):
# No existing alerts or invalid format
pass

def _save_alerts (self):
"""Save alerts to storage."""
alerts_data = [alert.to_dict () for alert in self.alerts. values()]
with open (self.config ['alert_storage_file'], 'w') as f:
json.dump (alerts_data, f, indent=2)

def create_alert (self, alert_type, severity, source, details, target):
"""Create a new security alert."""
alert = SecurityAlert (alert_type, severity, source, details, target)

# Check for duplicate alerts
if alert.id in self.alerts:
existing_alert = self.alerts [alert.id]
# Update if this is newer
if existing_ alert.status in [AlertStatus. RESOLVED, AlertStatus. FALSE_POSITIVE]:
# Create a new alert but reference the old one
alert. add_related_alert (existing_alert.id)
self. alerts [alert.id] = alert
self. _save_alerts()
return alert.id
else:
# Just update the existing alert
return existing _alert.id

# Store new alert
self.alerts [alert.id] = alert
self. _save_alerts()

# Send notifications
self. send_alert notifications (alert)

return alert.id

def update_alert (self, alert_id, status, user, notes=None):
"""Update an existing alert."""
if alert_id not in self.alerts:
return False

alert = self.alerts [alert_id]
alert. update_status (status, user, notes)
self. _save_alerts ()

# Send update notifications if needed
if status in [AlertStatus .RESOLVED, AlertStatus .FALSE_POSITIVE]:
self.send resolution_ notifications (alert)

return True

def get_alert (self, alert_id):
"""Get an alert by ID."""
return self.alerts .get (alert_id)

def get_ active_alerts (self, severity=None):
"""Get all active alerts, optionally filtered by severity."""
active_statuses = [AlertStatus.NEW, AlertStatus.ACKNOWLEDGED, AlertStatus.INVESTIGATING]

if severity:
return [
alert for alert in self.alerts.values()
if alert.status in active_statuses and alert.severity == severity
]
else:
return [
alert for alert in self.alerts.values()
if alert.status in active_statuses
]

def send_alert notifications (self, alert):
"""Send notifications for a new alert."""
# Determine notification channels based on severity
channels = []

if alert.severity in [AlertSeverity.CRITICAL, AlertSeverity.HIGH]:
channels = self.config ['notification_channels'] ['high_priority']
elif alert.severity == AlertSeverity.MEDIUM:
channels = self.config ['notification_channels'] ['medium_priority']
else:
channels = self.config ['notification_channels'] ['low_priority']

# Build notification message
message = self.format alert_notification (alert)

# Send to each channel
for channel in channels:
self. _send_to_channel (channel, message, alert)

def send_resolution notifications (self, alert):
"""Send notifications for a resolved alert."""
# Typically only send resolution notifications for higher severity alerts
if alert.severity in [AlertSeverity.CRITICAL, AlertSeverity.HIGH]:
channels = self.config ['notification_channels'] ['resolutions']

# Build resolution message
message = self._format _resolution_notification (alert)

# Send to each channel
for channel in channels:
self. _send_to_channel (channel, message, alert)

def _format_alert _notification (self, alert):
"""Format an alert notification message."""
return {
'title': f"{alert.severity .name} Security Alert: {alert.type}",
'message': f"Target: {alert.target} \nSource: {alert.source} \nDetails: {json.dumps (alert.details, indent=2)}",
'alert_id': alert.id,
'severity': alert.severity.name,
'created_time': alert.created_time
}

def _format_resolution _notification (self, alert):
"""Format a resolution notification message."""
return {
'title': f"Security Alert Resolved: {alert.type}",
'message': f"Alert ID: {alert.id} \nTarget: {alert.target} \nResolution: {alert. resolution_notes}",
'alert_id': alert.id,
'severity': alert.severity.name,
'resolution_time': alert.updated_time
}

def _send_to_channel(self, channel, message, alert):
"""Send notification to a specific channel."""
if channel['type'] == 'email':
self. send_email notification (channel, message, alert)
elif channel['type'] == 'slack':
self. send_slack notification (channel, message, alert)
elif channel['type'] == 'pagerduty':
self. send_pagerduty notification(channel, message, alert)
elif channel ['type'] == 'webhook':
self. send_webhook notification (channel, message, alert)

def _send_email _notification(self, channel, message, alert):
"""Send email notification."""
# Implementation depends on email service
pass

def _send_slack _notification(self, channel, message, alert):
"""Send Slack notification."""
try:
webhook_url = channel['webhook_url']

# Format Slack message
slack_message = {
'text': message['title'],
'blocks': [
{
'type': 'header',
'text': {
'type': 'plain_text',
'text': message['title']
}
},
{
'type': 'section',
'text': {
'type': 'mrkdwn',
'text': message ['message'].replace('\n', '\n>')
}
},
{
'type': 'context',
'elements': [
{
'type': 'mrkdwn',
'text': f"Alert ID: {alert.id} | Severity: {alert.severity.name} | Time: {time.strftime ('%Y-%m-%d %H:%M:%S', time.localtime (alert.created_time))}"
}
]
}
]
}

# Add actions if this is a new alert
if alert.status == AlertStatus.NEW:
slack_message ['blocks'] .append({
'type': 'actions',
'elements': [
{
'type': 'button',
'text': {
'type': 'plain_text',
'text': 'Acknowledge'
},
'style': 'primary',
'value': f"acknowledge: {alert.id}"
},
{
'type': 'button',
'text': {
'type': 'plain_text',
'text': 'False Positive'
},
'style': 'danger',
'value': f"false_ positive: {alert.id}"
}
]
})

# Send to Slack
response = requests.post(
webhook_url,
json= slack_message
)

if response.status _code != 200:
print(f"Error sending Slack notification: {response.text}")

except Exception as e:
print(f"Error sending Slack notification: {str(e)}")

def _send_pagerduty _notification (self, channel, message, alert):
"""Send PagerDuty notification."""
# Implementation depends on PagerDuty API
pass

def _send_webhook _notification (self, channel, message, alert):
"""Send webhook notification."""
try:
webhook_url = channel['url']

# Send to webhook
response = requests.post(
webhook_url,
json ={
'alert': alert.to_dict(),
'notification': message
}
)

if response.status_code not in [200, 201, 202, 204]:
print(f"Error sending webhook notification: {response.text}")

except Exception as e:
print(f"Error sending webhook notification: {str(e)}")

# Example usage
if name == "main":
alert_manager = Security AlertManager ("security_alert_ config.json")

# Create a new critical alert
alert_id = alert_manager. create_alert(
"ssl_configuration _issue",
AlertSeverity .CRITICAL,
"ssl_monitor",
{
"issue": "Weak cipher suite detected",
"details": "The server supports RC4 cipher suites which are considered insecure"
},
"https://example.com"
)

print(f"Created alert with ID: {alert_id}")

# Update the alert
alert_manager .update_alert(
alert_id,
AlertStatus .INVESTIGATING,
"security_analyst",
"Investigating the SSL configuration issue"
)

from flask import Flask, render_template, jsonify, request
import json
import time
import os

app = Flask(__name__)

# Load configuration
with open ('security_dashboard _config.json', 'r') as f:
config = json.load(f)

# Initialize alert manager
alert_manager = Security AlertManager (config['alert_ manager_config'])

@app.route('/')
def dashboard():
"""Render the main security dashboard."""
return render_template ('dashboard.html')

@app.route('/api/alerts')
def get_alerts():
"""API endpoint to get security alerts."""
status = request.args.get ('status')
severity = request.args.get ('severity')

alerts = []
if status == 'active':
if severity:
alerts = alert_manager .get_active_alerts (AlertSeverity [severity.upper()])
else:
alerts = alert_manager .get_active_alerts ()
else:
alerts = list (alert_manager. alerts.values())

alert_data = [alert.to_dict() for alert in alerts]

alert_data.sort (key= lambda x: x['created_time'], reverse=True)

return jsonify (alert_data)

@app.route ('/api/alerts/', methods=['GET'])
def get_alert (alert_id):
"""API endpoint to get a specific alert."""
alert = alert_ manager.get_ alert(alert_id)
if alert:
return jsonify (alert.to_dict())
else:
return jsonify ({'error': 'Alert not found'}), 404

@app.route ('/api/alerts/', methods=['PUT'])
def update_alert (alert_id):
"""API endpoint to update an alert."""
data = request.json
if not data:
return jsonify ({'error': 'No update data provided'}), 400

status = data.get ('status')
user = data.get ('user')
notes = data.get ('notes')

if not status or not user:
return jsonify ({'error': 'Status and user are required'}), 400

try:
status_enum = AlertStatus (status)
except ValueError:
return jsonify ({'error': f"Invalid status value: {status}"}), 400

success = alert_manager .update_alert (alert_id, status_enum, user, notes)
if success:
return jsonify ({'status': 'success'})
else:
return jsonify ({'error': 'Alert update failed'}), 500

@app.route('/api/stats')
def get_stats():
"""API endpoint to get security monitoring statistics."""
stats = {
'alerts_by _severity': {},
'alerts_by _status': {},
'alerts_by _type': {},
'recent _activity': []
}

for severity in AlertSeverity:
stats ['alerts_by_ severity'] [severity.name] = 0

for status in AlertStatus:
stats ['alerts_by _status'] [status.value] = 0

now = int (time.time())
seven_days_ago = now - (7 * 24 * 60 * 60)
daily_alerts = [0] * 7

for alert in alert_manager .alerts.values ():
stats ['alerts_by_severity'] [alert.severity.name] += 1
stats ['alerts_by_status'] [alert.status.value] += 1

alert_type = alert.type
if alert_type not in stats ['alerts_by_type']:
stats ['alerts_by_type'] [alert_type] = 0
stats ['alerts_by_type'] [alert_type] += 1

if alert.created_time > seven_days_ago:
day_index = min(6, (now - alert. created_time) // (24 * 60 * 60))
daily_alerts [6 - day_index] += 1

stats ['daily_alert_data'] = {
'labels': [
time.strftime ('%Y-%m-%d', time.localtime (now - (6 - i) * 24 * 60 * 60))
for i in range(7)
],
'data': daily_alerts
}

recent_alerts = list (alert_manager.alerts .values())
recent_alerts .sort (key= lambda x: max (x.created_time, x.updated_time), reverse=True)

for alert in recent_alerts [:10]:
if alert.updated _time > alert.created_time:
stats ['recent_activity'] .append({
'type': 'update',
'alert_id': alert.id,
'alert_type': alert.type,
'severity': alert. severity.name,
'status': alert. status.value,
'timestamp': alert. updated_time,
'user': alert. assigned_to
})
else:
stats ['recent_activity'] .append({
'type': 'new',
'alert_id': alert.id,
'alert_type': alert.type,
'severity': alert.severity .name,
'timestamp': alert.created _time
})

return jsonify (stats)

@app.route('/api/security-checks')
def get_security _checks():
"""API endpoint to get status of security checks."""
try:
with open (config ['security_ checks_file'], 'r') as f:
checks = json.load(f)
return jsonify (checks)
except (FileNotFoundError, json. JSONDecodeError):
return jsonify([])

if __name__ == '__main__':
app.run (debug=True)