Spaces:

sitcod3
/

modul-x

Running

App Files Files Community

modul-x / app /utils /webhook_alerts.py

sitcod3

uranusmediaconn

chore: squash history after secret scrub

5db9737 about 1 month ago

Raw History Blame Contribute Delete

8.67 kB

#!/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")

	#!/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")