An AI-assisted webhook request/test component adds response readback to an earlier outbound-request primitive. The shipped SQL-backed test path can cross the intended destination boundary when an admitted client can configure and test webhooks.

@@ -2,10 +2,11 @@ import hashlib import hmac import json import logging+from typing import Optional  import requests -from mlflow.entities.webhook import WebhookEvent+from mlflow.entities.webhook import WebhookEvent, WebhookTestResult from mlflow.store.model_registry.abstract_store import AbstractStore from mlflow.webhooks.constants import WEBHOOK_SIGNATURE_HEADER from mlflow.webhooks.types import WebhookPayload@@ -27,6 +28,44 @@ def _generate_hmac_signature(secret: str, payload_bytes: bytes) -> str:

+) -> WebhookTestResult:+    """Send a webhook request to the specified URL.++    Args:+        url: The webhook URL to send the request to+        payload: The payload to send+        secret: Optional secret for HMAC signature++    Returns:+        WebhookTestResult indicating success/failure and response details+    """+    try:+        payload_bytes = json.dumps(payload).encode("utf-8")+        headers = {"Content-Type": "application/json"}++        # Add HMAC signature if secret is configured+        if secret:+            signature = _generate_hmac_signature(secret, payload_bytes)+            headers[WEBHOOK_SIGNATURE_HEADER] = signature++        response = requests.post(url, data=payload_bytes, headers=headers, timeout=30)+

+            success=False,+            error_message=str(e)[:500],  # Truncate error message+        )++ def _dispatch_webhook_impl(     *,     event: WebhookEvent,@@ -36,14 +75,7 @@ def _dispatch_webhook_impl(     # TODO: Make this non-blocking     for webhook in store.list_webhooks():         if event in webhook.events:-            payload_bytes = json.dumps(payload).encode("utf-8")-            headers = {"Content-Type": "application/json"}-            # Add HMAC signature if secret is configured-            if webhook.secret:-                signature = _generate_hmac_signature(webhook.secret, payload_bytes)-                headers[WEBHOOK_SIGNATURE_HEADER] = signature--            requests.post(webhook.url, data=payload_bytes, headers=headers)+            _send_webhook_request(webhook.url, payload, webhook.secret)   def dispatch_webhook(@@ -59,3 +91,61 @@ def dispatch_webhook(             f"Failed to dispatch webhook for event {event}: {e}",             exc_info=True,

++    Args:+        webhook_id: The ID of the webhook to test+        store: The model registry store to retrieve webhook details+        event: Optional event type to test. If not specified, uses the first event from webhook.++    Returns:+        WebhookTestResult indicating success/failure and response details+    """+    try:+        webhook = store.get_webhook(webhook_id)++        # Use provided event or the first event type for testing+        test_event = event or webhook.events[0]++        # Generate example payload based on the event type+        if test_event == WebhookEvent.REGISTERED_MODEL_CREATED:+            from mlflow.webhooks.types import RegisteredModelCreatedPayload++            test_payload = RegisteredModelCreatedPayload.example()+        elif test_event == WebhookEvent.MODEL_VERSION_CREATED:+            from mlflow.webhooks.types import ModelVersionCreatedPayload++            test_payload = ModelVersionCreatedPayload.example()+        elif test_event == WebhookEvent.MODEL_VERSION_TAG_SET:+            from mlflow.webhooks.types import ModelVersionTagSetPayload++            test_payload = ModelVersionTagSetPayload.example()+        elif test_event == WebhookEvent.MODEL_VERSION_TAG_DELETED:+            from mlflow.webhooks.types import ModelVersionTagDeletedPayload++            test_payload = ModelVersionTagDeletedPayload.example()+        elif test_event == WebhookEvent.MODEL_VERSION_ALIAS_CREATED:+            from mlflow.webhooks.types import ModelVersionAliasCreatedPayload++            test_payload = ModelVersionAliasCreatedPayload.example()+        elif test_event == WebhookEvent.MODEL_VERSION_ALIAS_DELETED:+            from mlflow.webhooks.types import ModelVersionAliasDeletedPayload++            test_payload = ModelVersionAliasDeletedPayload.example()+        else:+            # Default to MODEL_VERSION_CREATED payload+            from mlflow.webhooks.types import ModelVersionCreatedPayload++            test_payload = ModelVersionCreatedPayload.example()++        return _send_webhook_request(webhook.url, test_payload, webhook.secret)+    except Exception as e:+        return WebhookTestResult(+            success=False,+            error_message=f"Failed to test webhook: {str(e)[:500]}",+        )

@@ -898,6 +898,32 @@ def _validate_webhook_name(name: str) -> None:         )  

++    This normalizes IPv6-transition addresses (IPv4-mapped, 6to4, NAT64) to their+    embedded IPv4 address so that a wrapped private/CGNAT/link-local address cannot+    masquerade as global.+    """+    if isinstance(ip, ipaddress.IPv6Address):+        embedded = ip.ipv4_mapped or ip.sixtofour+        if embedded is None and ip in _NAT64_PREFIX:+            embedded = ipaddress.IPv4Address(int(ip) & 0xFFFFFFFF)+        if embedded is not None:+            return embedded

     acquired = _HOSTNAME_RESOLUTION_SEMAPHORE.acquire(timeout=_HOSTNAME_RESOLUTION_TIMEOUT_SECONDS)     if not acquired:@@ -944,7 +970,7 @@ def _validate_hostname_resolves_to_public_ips(hostname: str, field_name: str) ->             raise MlflowException.invalid_parameter_value(

@@ -27,6 +27,7 @@ from urllib3.connectionpool import HTTPConnectionPool, HTTPSConnectionPool

@@ -58,7 +59,7 @@ def _assert_public_peer(sock: socket.socket) -> None:             f"Webhook connection resolved to an invalid IP address: {peer_ip!r}"