Custom Integrations¶

import hashlib
import hmac
import time


def verify_header_signature(raw_body: bytes, header_value: str, secret: str) -> bool:
    """X-Webhook-Signature is HMAC-SHA256 over the exact request bytes."""
    expected = hmac.new(secret.encode(), raw_body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(header_value, f"sha256={expected}")


def verify_inline_signature(payload: dict, secret: str, max_age: int = 900) -> bool:
    """payload['signature'] is HMAC-SHA256 over str(timestamp) + token."""
    sig = payload.get("signature", {})
    if abs(time.time() - sig.get("timestamp", 0)) > max_age:
        return False  # replay protection: reject events older than 15 minutes
    expected = hmac.new(
        secret.encode(),
        (str(sig["timestamp"]) + sig["token"]).encode(),
        hashlib.sha256,
    ).hexdigest()
    return hmac.compare_digest(sig.get("signature", ""), expected)

import hashlib
import hmac
import json


def verify_endpoint_webhook(payload: dict, header_value: str, secret: str) -> bool:
    payload_json = json.dumps(payload, sort_keys=True)
    expected = hmac.new(secret.encode(), payload_json.encode(), hashlib.sha256).hexdigest()
    return hmac.compare_digest(header_value, f"sha256={expected}")

from flask import Flask, request, jsonify
import requests

app = Flask(__name__)

HUBSPOT_TOKEN = "pat-xxx"
WEBHOOK_SECRET = "your-secret"


@app.route("/webhooks/mailyte", methods=["POST"])
def handle_mailyte_webhook():
    if not verify_header_signature(
        request.get_data(), request.headers.get("X-Webhook-Signature", ""), WEBHOOK_SECRET
    ):
        return jsonify({"status": "invalid signature"}), 401

    payload = request.get_json()
    if payload.get("event") == "email.inbound":
        data = payload.get("data", {})
        sender = data.get("from") or data.get("sender")
        if sender:
            requests.post(
                "https://api.hubapi.com/crm/v3/objects/contacts",
                headers={
                    "Authorization": f"Bearer {HUBSPOT_TOKEN}",
                    "Content-Type": "application/json",
                },
                json={"properties": {"email": sender}},
            )
    return jsonify({"status": "ok"})

@app.route("/webhooks/mailyte", methods=["POST"])
def handle_support_email():
    payload = request.get_json()
    if payload.get("event") != "email.inbound":
        return jsonify({"status": "skipped"})

    data = payload.get("data", {})
    recipient = data.get("recipient", "")
    if not recipient.startswith("support@"):
        return jsonify({"status": "skipped"})

    requests.post(
        "https://your-org.atlassian.net/rest/api/3/issue",
        auth=("email@company.com", "JIRA_API_TOKEN"),
        json={
            "fields": {
                "project": {"key": "SUP"},
                "summary": data.get("subject", "(no subject)"),
                "issuetype": {"name": "Task"},
            }
        },
    )
    return jsonify({"status": "created"})

import requests

SLACK_WEBHOOK = "https://hooks.slack.com/services/T.../B.../xxx"


@app.route("/webhooks/mailyte", methods=["POST"])
def slack_notify():
    payload = request.get_json()
    event = payload.get("event")
    data = payload.get("data", {})

    messages = {
        "email.bounced": f"Bounce for {data.get('recipient', 'unknown')}",
        "delivery.bounce.hard": f"Hard bounce: {data.get('recipient', 'unknown')}",
        "email.delivered": f"Delivered to {data.get('recipient', 'unknown')}",
    }
    msg = messages.get(event)
    if msg:
        requests.post(SLACK_WEBHOOK, json={"text": msg})
    return jsonify({"status": "ok"})

import requests

API = "https://api.yourdomain.com/api/v1"
HEADERS = {"X-API-Key": "YOUR_ADMIN_API_KEY", "Content-Type": "application/json"}


def provision_customer(org_id: str, org_name: str, domain: str, admin_email: str):
    """Set up everything a new customer needs."""

    # 1. Create organization (admin platform key required)
    requests.post(
        f"{API}/organizations/",
        headers=HEADERS,
        json={"id": org_id, "name": org_name, "admin_email": admin_email},
    )

    # 2. Add domain — DKIM is generated automatically and the
    #    response carries every DNS record the customer must publish
    domain_resp = requests.post(
        f"{API}/domains/",
        headers=HEADERS,
        json={"domain": domain, "organization_id": org_id},
    ).json()["data"]
    domain_id = domain_resp["domain_id"]

    # 3. Create the admin mailbox
    requests.post(
        f"{API}/mailboxes/email-accounts",
        headers=HEADERS,
        json={
            "email": admin_email,
            "password": generate_temp_password(),
            "name": "Admin",
            "domain_id": domain_id,
        },
    )

    # 4. Mint an SMTP credential so their application can send
    smtp_cred = requests.post(
        f"{API}/smtp-credentials/",
        headers=HEADERS,
        json={"domain_id": domain_id, "name": f"{org_id} app sending"},
    ).json()["data"]
    # smtp_cred["secret"] is shown exactly once — hand it over securely

    return {
        "org_id": org_id,
        "domain_id": domain_id,
        "dns_records": domain_resp["dns_records"],
        "dkim_record": domain_resp.get("dkim_record"),
        "smtp_username": smtp_cred["username"],
    }

def get_org_usage(org_id: str) -> dict:
    """Quota limits and current usage for an organization."""
    quotas = requests.get(f"{API}/organizations/{org_id}/quotas", headers=HEADERS).json()["data"]
    return quotas

def check_domain_health(domain_id: str) -> dict:
    """Live MX/SPF/DKIM/DMARC verification for a domain."""
    return requests.get(f"{API}/domains/{domain_id}/verify-dns", headers=HEADERS).json()


def rotate_dkim(domain_id: str) -> dict:
    """Step one of a safe DKIM rotation — publish the returned record,
    then call the activate endpoint once it resolves."""
    return requests.post(
        f"{API}/domains/{domain_id}/dkim/rotate",
        headers=HEADERS,
        json={"reason": "scheduled rotation"},
    ).json()