Receive your VAULT notifications on Slack

The Vault notifications system integrates smoothly with any third-party message application. If your company communicates through Slack, this 10 minutes tutorial is worth reading. No previous development experience is needed - follow along.

Tutorial covers:

  1. How to build a minimal web server to forward notification to Slack
  2. How to configure your web server to receive notifications

Prerequisites:

  • You have configured API communication with your VAULT workspace:
  • You have a running LAM (Ledger Authenticated Module)
  • You have a registered LAM user

ℹī¸ In a near future, you will soon be able to interact with your VAULT programmatically for less sensitive reporting endpoints (transaction, account, user, whitelist exports) without any LAM setup.

Tutorial architecture

slack notifications

1. How to build a minimal web server to forward notification to Slack

Vault notification schema

Copy
Copied
from pydantic import BaseModel

class VaultNotification(BaseModel):
    payload_type: str
    event_type: str
    id: int

ℹī¸ To extract insightful information from the VaultNotification, we need to query the full Vault object using the id.

Get full Vault object details

Copy
Copied
import os
import requests

LAM_URL = os.getenv("LAM_URL")
API_USER = os.getenv("API_USER")
LAM_HEADERS = {'X-Ledger-API-User': API_USER}

def get_request_details(notification: VaultNotification):
    response = requests.get(f"{LAM_URL}/requests/{notification.id}", headers=LAM_HEADERS)
    request_details = response.json()
    return request_details 
    

def get_transaction_details(notification: VaultNotification):
    response = requests.get(f"{LAM_URL}/transactions/{notification.id}", headers=LAM_HEADERS)
    transaction_details = response.json()
    return transaction_details

Format message for slack

Copy
Copied
def is_transaction_event(notification: VaultNotification):
    transaction_events = [
        "NEW_TRANSACTION_HAS_BEEN_SIGNED",
        "NEW_TRANSACTION_HAS_BEEN_BROADCASTED",
        "NEW_TRANSACTION_HAS_FAILED",
        "NEW_TRANSACTION_HAS_BEEN_ABORTED",
        "NEW_TRANSACTION_HAS_BEEN_RECEIVED",
    ]
    return notification.event_type in transaction_events

def is_request_event(notification: VaultNotification):
    request_events = [
        "REQUEST_HAS_BEEN_CREATED",
        "REQUEST_HAS_RECEIVED_AN_APPROVAL",
        "REQUEST_HAS_REACHED_STEP",
        "REQUEST_HAS_REACHED_QUORUM",
        "REQUEST_HAS_BEEN_ABORTED",
        "REQUEST_HAS_EXPIRED"
    ]
    return notification.event_type in request_events


def format_request(request_details, event_type):
    workspace = os.getenv(VAULT_WORKSPACE)
    transaction_id = request_details["target_id"]
    lines = [
        f":rocket: TRANSACTION {event_type}",
        f"https://vault.ledger.com/{workspace}/operator/transactions/modal/transactions/details/{transaction_id}/overview",
        f"Created by {request_details['username']}",
        f"Created on {request_details['created_on']}",
        f"Expires at {request_details['expires']}",
        f"Status {request_details['status']}"
    ]

    slack_message = "\n".join(lines)
    return slack_message
    

def format_transaction(transaction_details, event_type):
    workspace = os.getenv(VAULT_WORKSPACE)
    transaction_id = transaction_details["id"]
    lines = [
        f":rocket: {event_type}",
        f"https://vault.ledger.com/{workspace}/operator/transactions/modal/transactions/details/{transaction_id}/overview",
        f"Transaction id {transaction_id}",
        f"Created on {transaction_details['account']}",
        f"Expires at {transaction_details['currency']}",
        f"Status {transaction_details['recipient']}"
    ]
    slack_message = "\n".join(lines)
    return slack_message

# NOTE: in this sample we only send transaction-related notifications 
# but you could send all types of notification received from the Vault.
# Here is an exhaustive liste: https://ledger-enterprise-api-portal.redoc.ly/openapi/le_api/tag/Notifications/
def format_slack_message(notification: VaultNotification):
    if is_transaction_event(notification):
        transaction_details = get_transaction_details(notification)
        return format_transaction(transaction_details, notification.event_type)

    if is_request_event(notification):
        request_details = get_request_details(notification)
        if request_details["request_type"] == "CREATE_TRANSACTION":
            return format_request(request_details, notification.event_type)

Minimal web server to forward Vault transaction notifications to Slack

Copy
Copied
import os
from fastapi import FastAPI
from slack_sdk import WebClient


app = FastAPI()
@app.get("/health")
def health():
    return {"status": "success"}

@app.post("/slack-notification")
def slack_notification(notification: VaultNotification):
    message = format_slack_message(notification)
    if message:
        # Follow the slack documentation: https://api.slack.com/authentication/basics
        # To create a bot application, set the appropriate access to your workspace and generate a token.
        slack_token = os.getenv("SLACK_TOKEN")
        slack_client = WebClient(token=slack_token)
        channel = os.getenv("SLACK_CHANNEL")
        slack_client.chat_postMessage(channel=channel, text=message)

2. How to configure your web server to receive notifications

Once you've deployed your webserver, you have to register your new <YOUR_HOSTNAME>/slack-notification" endpoint as a Vault webhook.

ℹī¸ You can leverage Unicorn and ngrok for a 5 minutes-ready deployment.

Copy
Copied
curl -X PUT  -H "Content-Type: application/json" -H "X-Ledger-API-User: <LAM_USER> " -d '{"all": {"webhook": "<YOUR_HOSTNAME>/slack-notification"}}' <Vault URL> 

See the documentation to create and register a LAM_USER.

Copyright Š Ledger Enterprise Platform 2023. All right reserved.