mirror of
https://github.com/idrainformatica/PecFlow.git
synced 2026-06-16 12:45:42 +02:00
126 lines
3.5 KiB
Python
126 lines
3.5 KiB
Python
"""
|
||
Webhook sender – POST HTTP con firma HMAC-SHA256.
|
||
|
||
Config non sensibile (config):
|
||
{ "url": "https://...", "content_type": "application/json" }
|
||
|
||
Config sensibile (config_enc → config_secret):
|
||
{ "webhook_secret": "..." } # opzionale – usato per firma HMAC
|
||
|
||
Header inviati:
|
||
Content-Type: application/json
|
||
X-PEChub-Event: {event_type}
|
||
X-Hub-Signature-256: sha256={hex} (solo se webhook_secret configurato)
|
||
X-Delivery: {uuid}
|
||
User-Agent: PEChub-Webhook/1.0
|
||
"""
|
||
|
||
import hashlib
|
||
import hmac
|
||
import json
|
||
import uuid as uuid_mod
|
||
from datetime import datetime
|
||
|
||
import httpx
|
||
|
||
DEFAULT_TIMEOUT = 10.0
|
||
|
||
|
||
class WebhookError(Exception):
|
||
"""Errore durante l'invio di una notifica webhook."""
|
||
|
||
def __init__(self, message: str, http_status: int | None = None):
|
||
super().__init__(message)
|
||
self.http_status = http_status
|
||
|
||
|
||
async def send_webhook(
|
||
url: str,
|
||
payload: dict,
|
||
event_type: str = "new_message",
|
||
webhook_secret: str | None = None,
|
||
timeout: float = DEFAULT_TIMEOUT,
|
||
) -> dict:
|
||
"""
|
||
Invia un payload JSON a un webhook URL.
|
||
|
||
Args:
|
||
url: URL destinatario del webhook
|
||
payload: dict serializzato come JSON nel body
|
||
event_type: valore dell'header X-PEChub-Event
|
||
webhook_secret: segreto per firma HMAC-SHA256 (opzionale)
|
||
timeout: timeout HTTP in secondi
|
||
|
||
Returns:
|
||
dict con http_status, response_text, delivery_id
|
||
|
||
Raises:
|
||
WebhookError: in caso di timeout, errore di rete o HTTP >= 400
|
||
"""
|
||
body = json.dumps(payload, ensure_ascii=False, default=str).encode("utf-8")
|
||
delivery_id = str(uuid_mod.uuid4())
|
||
|
||
headers = {
|
||
"Content-Type": "application/json",
|
||
"X-PEChub-Event": event_type,
|
||
"X-Delivery": delivery_id,
|
||
"User-Agent": "PEChub-Webhook/1.0",
|
||
}
|
||
|
||
if webhook_secret:
|
||
sig = hmac.new(
|
||
webhook_secret.encode("utf-8"),
|
||
body,
|
||
hashlib.sha256,
|
||
).hexdigest()
|
||
headers["X-Hub-Signature-256"] = f"sha256={sig}"
|
||
|
||
async with httpx.AsyncClient(timeout=timeout) as client:
|
||
try:
|
||
response = await client.post(url, content=body, headers=headers)
|
||
except httpx.TimeoutException as exc:
|
||
raise WebhookError(
|
||
f"Timeout webhook dopo {timeout}s"
|
||
) from exc
|
||
except httpx.RequestError as exc:
|
||
raise WebhookError(f"Errore di rete webhook: {exc}") from exc
|
||
|
||
if response.status_code >= 400:
|
||
raise WebhookError(
|
||
f"Webhook ha risposto con HTTP {response.status_code}: "
|
||
f"{response.text[:200]}",
|
||
http_status=response.status_code,
|
||
)
|
||
|
||
return {
|
||
"http_status": response.status_code,
|
||
"response_text": response.text[:500],
|
||
"delivery_id": delivery_id,
|
||
}
|
||
|
||
|
||
async def send_test_webhook(
|
||
url: str,
|
||
webhook_secret: str | None = None,
|
||
channel_name: str = "PEChub",
|
||
) -> dict:
|
||
"""Invia un payload di test al webhook per verificare la configurazione."""
|
||
payload = {
|
||
"event": "test",
|
||
"channel": channel_name,
|
||
"timestamp": datetime.now().isoformat(),
|
||
"message": "Notifica di test da PEChub",
|
||
"data": {
|
||
"subject": "[TEST] PEC di prova",
|
||
"from_address": "test@pec.example.it",
|
||
"pec_type": "posta_certificata",
|
||
"direction": "inbound",
|
||
},
|
||
}
|
||
return await send_webhook(
|
||
url=url,
|
||
payload=payload,
|
||
event_type="test",
|
||
webhook_secret=webhook_secret,
|
||
)
|