Repository: PrefectHQ/prefect
Issue category: Feature Request

Allow prefect Webhook to use insecure connections
### First check

- [X] I added a descriptive title to this issue.
- [X] I used the GitHub search to find a similar request and didn't find it.
- [X] I searched the Prefect documentation for this feature.

### Prefect Version

2.x

### Describe the current behavior

Currently prefect Webhook is instantiated with `verify=True` with no option to change this. This is caused by using a global AsyncHTTPTransport object in prefect.blocks.webhook, i.e.
```python
# Use a global HTTP transport to maintain a process-wide connection pool for
# interservice requests
_http_transport = AsyncHTTPTransport()
```


### Describe the proposed behavior

It would be convenient for development to be able to instantiate an insecure webhook like so:
```python
from prefect.blocks.webhook import Webhook
insecure_webhook = Webhook(method="POST", url="https://some-server.domain/captain/hook",
                             headers={"Content-Type": "application/json"},verify=False)
```
and receive a webhook that has ssl-verification turned off.

### Example Use

This is convenient in environments with self-signed certificates and the system admin being on holiday or over-worked.

### Additional context

Currently I have patched the block_initialization in webhook.py like so:
```python
    def block_initialization(self):
        if not getattr(self, "verify", True):
            self._client = AsyncClient(transport=AsyncHTTPTransport(verify=False))
        else:
            self._client = AsyncClient(transport=_http_transport)
```
which allows for instantiating a webhook like proposed above.
I do not know about potential unwanted side effects.



Functions to edit:
src/prefect/blocks/webhook.py:Webhook.block_initialization
src/prefect/blocks/webhook.py:Webhook
Functions added: src/prefect/blocks/webhook.py:Webhook

Patch:
diff --git a/src/prefect/blocks/webhook.py b/src/prefect/blocks/webhook.py
index 13b477e8c712..0c95428a44b9 100644
--- a/src/prefect/blocks/webhook.py
+++ b/src/prefect/blocks/webhook.py
@@ -11,6 +11,7 @@
 # Use a global HTTP transport to maintain a process-wide connection pool for
 # interservice requests
 _http_transport = AsyncHTTPTransport()
+_insecure_http_transport = AsyncHTTPTransport(verify=False)
 
 
 class Webhook(Block):
@@ -44,9 +45,16 @@ class Webhook(Block):
         default=True,
         description="Whether to allow notifications to private URLs. Defaults to True.",
     )
+    verify: bool = Field(
+        default=True,
+        description="Whether or not to enforce a secure connection to the webhook.",
+    )
 
     def block_initialization(self):
-        self._client = AsyncClient(transport=_http_transport)
+        if self.verify:
+            self._client = AsyncClient(transport=_http_transport)
+        else:
+            self._client = AsyncClient(transport=_insecure_http_transport)
 
     async def call(self, payload: Optional[dict] = None) -> Response:
         """

Test patch: diff --git a/tests/blocks/test_webhook.py b/tests/blocks/test_webhook.py
index a3cbc2143917..9d8679e07eda 100644
--- a/tests/blocks/test_webhook.py
+++ b/tests/blocks/test_webhook.py
@@ -44,19 +44,40 @@ class TestWebhook:
     def test_webhook_raises_error_on_bad_request_method(self):
         for method in ["GET", "PUT", "POST", "PATCH", "DELETE"]:
             Webhook(method=method, url="http://google.com")
+            Webhook(method=method, url="http://google.com", verify=False)
 
         for bad_method in ["get", "BLAH", ""]:
             with pytest.raises(ValueError):
                 Webhook(method=bad_method, url="http://google.com")
+                Webhook(method=bad_method, url="http://google.com", verify=False)
+
+    def test_insecure_webhook_set(self):
+        insecure_webhook = Webhook(method="GET", url="http://google.com", verify=False)
+        assert not getattr(insecure_webhook, "verify")
+        assert (
+            insecure_webhook._client._transport._pool._ssl_context.verify_mode.name
+            == "CERT_NONE"
+        )
+
+        secure_webhook = Webhook(method="GET", url="http://google.com")
+        assert (
+            secure_webhook._client._transport._pool._ssl_context.verify_mode.name
+            == "CERT_REQUIRED"
+        )
 
     @pytest.mark.parametrize("value, reason", RESTRICTED_URLS)
     async def test_webhook_must_not_point_to_restricted_urls(
         self, value: str, reason: str
     ):
-        webhook = Webhook(url=value, allow_private_urls=False)
+        secure_webhook = Webhook(url=value, allow_private_urls=False)
 
         with pytest.raises(ValueError, match=f"is not a valid URL.*{reason}"):
-            await webhook.call(payload="some payload")
+            await secure_webhook.call(payload="some payload")
+
+        insecure_webhook = Webhook(url=value, allow_private_urls=False, verify=False)
+
+        with pytest.raises(ValueError, match=f"is not a valid URL.*{reason}"):
+            await insecure_webhook.call(payload="some payload")
 
     async def test_webhook_sends(self, monkeypatch):
         send_mock = AsyncMock()
@@ -73,6 +94,20 @@ async def test_webhook_sends(self, monkeypatch):
             json={"event_id": "123"},
         )
 
+        await Webhook(
+            method="POST",
+            url="http://yahoo.com",
+            headers={"authorization": "password"},
+            verify=False,
+        ).call(payload={"event_id": "123"})
+
+        send_mock.assert_called_with(
+            method="POST",
+            url="http://yahoo.com",
+            headers={"authorization": "password"},
+            json={"event_id": "123"},
+        )
+
     async def test_webhook_sends_get_request_with_no_payload(self, monkeypatch):
         send_mock = AsyncMock()
         monkeypatch.setattr("httpx.AsyncClient.request", send_mock)
@@ -84,12 +119,29 @@ async def test_webhook_sends_get_request_with_no_payload(self, monkeypatch):
             method="GET", url="http://google.com", headers={"foo": "bar"}, json=None
         )
 
+        await Webhook(
+            method="GET", url="http://google.com", headers={"foo": "bar"}, verify=False
+        ).call(payload=None)
+
+        send_mock.assert_called_with(
+            method="GET", url="http://google.com", headers={"foo": "bar"}, json=None
+        )
+
     async def test_save_and_load_webhook(self):
         await Webhook(
             method="GET", url="http://google.com", headers={"foo": "bar"}
-        ).save(name="webhook-test")
+        ).save(name="secure-webhook-test")
+
+        secure_webhook = await Webhook.load(name="secure-webhook-test")
+        assert secure_webhook.url.get_secret_value() == "http://google.com"
+        assert secure_webhook.method == "GET"
+        assert secure_webhook.headers.get_secret_value() == {"foo": "bar"}
+
+        await Webhook(
+            method="GET", url="http://google.com", headers={"foo": "bar"}, verify=False
+        ).save(name="insecure-webhook-test")
 
-        webhook = await Webhook.load(name="webhook-test")
-        assert webhook.url.get_secret_value() == "http://google.com"
-        assert webhook.method == "GET"
-        assert webhook.headers.get_secret_value() == {"foo": "bar"}
+        insecure_webhook = await Webhook.load(name="insecure-webhook-test")
+        assert insecure_webhook.url.get_secret_value() == "http://google.com"
+        assert insecure_webhook.method == "GET"
+        assert insecure_webhook.headers.get_secret_value() == {"foo": "bar"}