File size: 8,674 Bytes
5db9737 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 | #!/usr/bin/env python3
# ==============================================================================
# BRIDGE Modul X - Production Webhook Alert Templates & Signing Dispatcher
# ==============================================================================
# This module defines JSON alert templates and manages cryptographically signed
# webhook dispatches to client notification systems. It uses HMAC-SHA256 signatures
# so receiving targets can securely verify payload authenticity [106, 111].
# ==============================================================================
import hmac
import hashlib
import time
import json
import os
import asyncio
from typing import Dict, Any, Optional
import httpx
import structlog
logger = structlog.get_logger("webhook_alerts")
# Try to load app settings if available
try:
from app.config import settings
except ImportError:
class FallbackSettings:
# Secret string used to generate HMAC signatures [106]
WEBHOOK_SIGNING_SECRET = os.getenv("WEBHOOK_SIGNING_SECRET", "modulx_secret_sec_auth_key_102938")
SYSTEM_TIMEOUT = 10.0
settings = FallbackSettings()
class WebhookDispatcher:
"""
Formulates and dispatches signed webhook event payloads to client systems [106].
"""
@staticmethod
def calculate_signature(payload: str, secret: str) -> str:
"""
Computes HMAC-SHA256 signature for verification checks [106, 111].
"""
return hmac.new(
secret.encode("utf-8"),
payload.encode("utf-8"),
hashlib.sha256
).hexdigest()
@classmethod
async def dispatch(cls, url: str, event_type: str, data: Dict[str, Any], custom_secret: Optional[str] = None) -> bool:
"""
Asynchronously sends signed JSON payloads with Retry logic and structured logs.
"""
secret = custom_secret or settings.WEBHOOK_SIGNING_SECRET
timestamp = str(int(time.time()))
# Structure the standardized wrapper envelope [106]
envelope = {
"event_id": f"evt_{timestamp}_{hash(event_type) & 0xfffffff}",
"event_type": event_type,
"timestamp": timestamp,
"data": data
}
serialized_payload = json.dumps(envelope, sort_keys=True)
# Construct HMAC signature mixing body payload + timestamp to prevent replay attacks [106]
signature_base = f"t={timestamp},v1={serialized_payload}"
signature = cls.calculate_signature(signature_base, secret)
headers = {
"Content-Type": "application/json",
"X-ModulX-Signature": f"t={timestamp},v1={signature}",
"User-Agent": "BRIDGE-ModulX-Webhook-Engine/2.0.0"
}
# Dispatch using HTTPX AsyncClient with exponential backoff on server throttles [106]
async with httpx.AsyncClient(timeout=10.0) as client:
retries = 3
backoff = 1.0
for attempt in range(1, retries + 1):
try:
logger.info("Dispatching webhook event...", url=url, event=event_type, attempt=attempt)
response = await client.post(url, headers=headers, content=serialized_payload)
if response.status_code in (200, 201, 202, 204):
logger.info("Webhook acknowledged successfully.", url=url, status_code=response.status_code)
return True
elif 400 <= response.status_code < 500:
logger.error("Client rejected payload. Aborting dispatch.", url=url, status_code=response.status_code)
return False
else:
logger.warning("Server threw transient error code. Scheduling backoff.", url=url, status_code=response.status_code)
except httpx.RequestError as req_err:
logger.warning("Transport network error during dispatch attempt.", url=url, error=str(req_err))
if attempt < retries:
await asyncio.sleep(backoff)
backoff *= 2.0
logger.error("All webhook retry attempts exhausted. Delivery failed.", url=url, event=event_type)
return False
# ==============================================================================
# high-fidelity JSON Webhook Payload Templates
# ==============================================================================
def get_campaign_completed_template(job_id: str, total_count: int, valid_count: int, invalid_count: int, risky_count: int, duration_sec: float) -> Dict[str, Any]:
"""
Template payload emitted when a bulk verification campaign completes [106, 111].
"""
return {
"job_id": job_id,
"summary": {
"total_records": total_count,
"processed_records": valid_count + invalid_count + risky_count,
"valid_deliverable": valid_count,
"invalid_blocked": invalid_count,
"risky_catchall": risky_count,
"accuracy_ratio": round((valid_count / total_count * 100), 2) if total_count > 0 else 100.0,
"elapsed_seconds": round(duration_sec, 2)
},
"metadata": {
"completed_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
"platform_cluster": "modulx-celery-pool-01"
}
}
def get_security_breach_template(email: str, risk_score: int, violation_reason: str) -> Dict[str, Any]:
"""
Template payload emitted immediately when a high-risk provider block list or
burn domain spam pattern gets processed in real-time [106, 111].
"""
return {
"security_incident_id": f"sec_alert_{int(time.time())}",
"severity": "CRITICAL",
"incident_type": "HIGH_RISK_RECIPIENT_SCRUTINIZED",
"details": {
"target_recipient": email,
"risk_score": risk_score,
"risk_level": "HIGH",
"reason": violation_reason,
"suggested_action": "REJECT_TRANSACTION_AND_FLAG_ACCOUNT"
},
"logged_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
}
def get_system_alert_template(metric_name: str, threshold: float, current_value: float) -> Dict[str, Any]:
"""
Template payload emitted to system administrators if queue backlog sizes or API latencies
breach acceptable performance thresholds [106].
"""
return {
"sys_alert_id": f"sys_warn_{int(time.time())}",
"severity": "WARNING",
"system_status": "DEGRADED",
"metric_violated": metric_name,
"alert_conditions": {
"threshold_limit": threshold,
"active_value": current_value,
"deviation_percentage": round(((current_value - threshold) / threshold * 100), 1) if threshold > 0 else 100.0
},
"triggered_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
}
# Quick self-test demonstrating webhook signing mechanics
if __name__ == "__main__":
print("\n" + "="*80)
print(" BRIDGE Modul X - Webhook Payload Template Demonstrations")
print("="*80 + "\n")
# 1. Generate Campaign Completed Mock Data
camp_payload = get_campaign_completed_template(
job_id="9b1deb4d-3b7d-4bad-9bdd-2b0d7b3dcb6d",
total_count=1000,
valid_count=820,
invalid_count=120,
risky_count=60,
duration_sec=14.85
)
print("--- 1. [CAMPAIGN_COMPLETED] payload template ---")
print(json.dumps(camp_payload, indent=2))
print("\n" + "-"*80 + "\n")
# 2. Generate Security Real-time Breach Alert
sec_payload = get_security_breach_template(
email="spammer@tempburner-provider.org",
risk_score=15,
violation_reason="disposable_burner_blocklist_match"
)
print("--- 2. [SECURITY_HIGH_RISK_DETECTED] payload template ---")
print(json.dumps(sec_payload, indent=2))
print("\n" + "-"*80 + "\n")
# 3. Signature Sample
test_secret = "my_super_secure_key"
payload_str = json.dumps(sec_payload, sort_keys=True)
ts = str(int(time.time()))
sig_base = f"t={ts},v1={payload_str}"
sig = WebhookDispatcher.calculate_signature(sig_base, test_secret)
print("--- HMAC-SHA256 Signature Header Template ---")
print(f"X-ModulX-Signature: t={ts},v1={sig}")
print("="*80 + "\n")
|