A Flask SimpleMDM webhook receiver for Slack

main.py

import os
import json
import requests
from flask import Flask, request
from slack_sdk import WebClient
from slack_sdk.errors import SlackApiError

# Initialize the Flask app
flask_app = Flask(__name__)

# Initialize the Slack client with environment variables
slack_client = WebClient(token=os.environ.get("SLACK_BOT_TOKEN"))

# SimpleMDM API URL and token
SIMPLEMDM_API_URL = "https://a.simplemdm.com/api/v1/device_groups"
SIMPLEMDM_API_KEY = os.environ.get("SIMPLEMDM_API_KEY")

# Channel ID for posting messages
CHANNEL_ID = os.environ.get("SLACK_CHANNEL_ID")

# Function to fetch device group name from SimpleMDM API
def fetch_device_group_name(group_id):
    response = requests.get(f"{SIMPLEMDM_API_URL}/{group_id}", auth=(SIMPLEMDM_API_KEY,''))
    if response.status_code == 200:
        group = response.json()
        return group['data']['attributes']['name']
    else:
        print(f"Error fetching device group {group_id} name: {response.status_code}")
        return "Unknown Group"

# Define a function to send Slack messages
def send_slack_message(text):
    try:
        slack_client.chat_postMessage(
            channel=CHANNEL_ID,
            text=text,
            unfurl_links=False
        )
    except SlackApiError as e:
        print(f"Error sending message to Slack: {e.response['error']}")

# Define a function to process the webhook payload
def process_webhook_event(event):
    event_type = event.get("type")
    data = event.get("data", {})
    device = data.get("device", {})
    devices = data.get("devices", [])

    messages = {
        "abm.device.added": lambda: (
            f"A new device was added to the *{data['enrollment']['name']}* enrollment via the {data['enrollment']['server_name']} assignment on ABM: "
            f"*{device['serial_number']}* ({device['model']}, {device['color']})"
        ),
        "device.enrolled": lambda: (
            f"A new device has enrolled: *<https://a.simplemdm.com/admin/devices/{device['id']}|{device['name']}>* with serial: *{device['serial_number']}* in group: *{fetch_device_group_name(device['device_group_id'])}*"
        ),
        "device.unenrolled": lambda: (
            f"A device has unenrolled: *{device['name']}* with serial: *{device['serial_number']}*"
        ),
        "device.lock.enabled": lambda: (
            f"A device has been locked: *<https://a.simplemdm.com/admin/devices/{device['id']}|{device['name']}>* with serial: *{device['serial_number']}*"
        ),
        "device.changed_group": lambda: (
            f"Devices have changed groups:\n" +
            "\n".join([
                f"- *<https://a.simplemdm.com/admin/devices/{device['id']}|{device['name']}>* with serial: *{device['serial_number']}*, new group: *{fetch_device_group_name(device['device_group_id'])}*"
                for device in devices
            ])
        ),
    }

    if event_type in messages:
        send_slack_message(messages[event_type]())
    else:
        send_slack_message(f"Received an unknown webhook event. Here it is in its entirety: ```{json.dumps(event)}```")

# Define route to handle SimpleMDM webhooks
@flask_app.route("/simplemdm/webhook", methods=["POST"])
def simplemdm_webhook():
    event = request.get_json()
    if event:
        process_webhook_event(event)
        return 'OK', 200
    return 'Server error', 500

@flask_app.route("/readiness", methods=["GET"])
@flask_app.route("/liveness", methods=["GET"])
def healthcheck():
    return 'OK', 200

# Run the Flask app
if __name__ == "__main__":
    print("🚀 Initializing...")
    flask_app.run(host='0.0.0.0',port=int(os.environ.get("PORT", 80)))