Automation Anywhere

Blu Pantheon ──webhook firmato──► Listener pubblico
                                       │ verifica HMAC, risponde 202
                                       ▼
                        Control Room API /v3/automations/deploy
                                       │
                                       ▼
                                  Bot Runner
                                       ├──► Gestionale (UI automation)
                                       └──► API Blu Pantheon (REST Web Service)

{"body":"__NOTA__","visibility":"internal","author":"automation-anywhere"}

import hashlib
import hmac
import json
import os
import time

import requests
from fastapi import FastAPI, Request, Response

app = FastAPI()

TOLERANCE_SECONDS = 300
SECRET = os.environ["BP_WEBHOOK_SECRET"]
CR_URL = os.environ["AA_CONTROL_ROOM_URL"]
CR_TOKEN = os.environ["AA_API_TOKEN"]
BOT_FILE_ID = os.environ["AA_BOT_FILE_ID"]
RUNNER_USER_ID = os.environ["AA_RUNNER_USER_ID"]

seen_deliveries: set[str] = set()


def verify(raw_body: bytes, header: str | None, now: int) -> tuple[bool, str]:
    if not header:
        return False, "header-mancante"

    timestamp = None
    signatures = []
    for segment in header.split(","):
        key, separator, value = segment.partition("=")
        if not separator:
            continue
        if key.strip() == "t":
            try:
                timestamp = int(value.strip())
            except ValueError:
                return False, "timestamp-non-valido"
        elif key.strip() == "v1":
            signatures.append(value.strip())

    if timestamp is None:
        return False, "timestamp-non-valido"
    if abs(now - timestamp) > TOLERANCE_SECONDS:
        return False, "timestamp-fuori-tolleranza"
    if not signatures:
        return False, "firma-assente"

    signed = f"{timestamp}.".encode("utf-8") + raw_body
    expected = hmac.new(SECRET.encode("utf-8"), signed, hashlib.sha256).hexdigest()

    for candidate in signatures:
        if hmac.compare_digest(candidate, expected):
            return True, "ok"

    return False, "firma-non-corrispondente"


def deploy_bot(event: dict, delivery_id: str) -> None:
    payload = {
        "fileId": BOT_FILE_ID,
        "runAsUserIds": [int(RUNNER_USER_ID)],
        "poolIds": [],
        "overrideDefaultDevice": False,
        "botInput": {
            "vEventType": {"type": "STRING", "string": event["event"]},
            "vDeliveryId": {"type": "STRING", "string": delivery_id},
            "vEventId": {"type": "STRING", "string": event["id"]},
            "vPayload": {"type": "STRING", "string": json.dumps(event, ensure_ascii=False)},
        },
    }

    response = requests.post(
        f"{CR_URL}/v3/automations/deploy",
        headers={"X-Authorization": CR_TOKEN, "Content-Type": "application/json"},
        json=payload,
        timeout=10,
    )
    response.raise_for_status()


@app.post("/webhooks/blupantheon")
async def receive(request: Request) -> Response:
    raw_body = await request.body()
    valid, reason = verify(
        raw_body,
        request.headers.get("X-BP-Signature"),
        int(time.time()),
    )

    if not valid:
        status = 400 if reason == "timestamp-fuori-tolleranza" else 401
        return Response(
            content=json.dumps({"error": reason}),
            status_code=status,
            media_type="application/json",
        )

    delivery_id = request.headers.get("X-BP-Delivery", "")
    if delivery_id in seen_deliveries:
        return Response(status_code=202)
    seen_deliveries.add(delivery_id)

    event = json.loads(raw_body)
    deploy_bot(event, delivery_id)

    return Response(status_code=202)