How to add webhook events to KeyVaults, programmatically?

I am trying to implement a secret-renewal service for various shared secrets in our environment. My plan for this is to add a webhook event for "Secret expires soon" to each keyvault (there are many), pointing back to a service that knows how to deal with that.

I can do that manually, but I can't figure out which Azure SDK module will allow me to add events to a keyvault. There's no mention of it in the keyvault modules. The eventgrid module wants me to create an eventgrid domain first, and I know I don't need to, because the Portal doesn't do that.

What is managing those event subscriptions?

Solution

You can create an eventsubscription using the eventgrid module without needing to create an eventgrid domain first. Here's a simple example:

from azure.identity import DefaultAzureCredential
from azure.mgmt.eventgrid import EventGridManagementClient
from azure.mgmt.eventgrid.models import EventSubscriptionFilter, WebHookEventSubscriptionDestination, EventSubscription

    # webhook URL should respond to the eventgrid validation query as usual and have valid HTTPS config
    webhook_url = "https://kv-listener.my-app.com/kv-events"

    subscription_id = "12345678-1234-1234-1234-1234567890"
    kv_resource_id = "/subscriptions/12345678-1234-1234-1234-1234567890/resourceGroups/vault-test/providers/Microsoft.KeyVault/vaults/testvault42"

    credential = DefaultAzureCredential()
    event_client = EventGridManagementClient(credential, subscription_id=subscription_id)

    subscription_name = "expiry-event-hook"
    destination = WebHookEventSubscriptionDestination(endpoint_url=webhook_url)
    event_filter = EventSubscriptionFilter(
        included_event_types=['Microsoft.KeyVault.SecretNearExpiry', 'Microsoft.KeyVault.SecretExpired']
    )
    the_sub = EventSubscription(
        destination=destination,
        filter=event_filter
    )
    scope = kv_resource_id
    poller = event_client.event_subscriptions.begin_create_or_update(scope, subscription_name, the_sub)
    print("Requested. Waiting on job to finish")
    poller.wait()
    print(f"Finished: {poller.status()}")