feat(extras): receive SaaS webhooks, at flyte.extras.webhooks

Product-agnostic machinery for turning an inbound webhook into a Flyte run, so
each flyteplugins-<product> package only has to say how to verify and parse its
own deliveries.

- WebhookAppEnvironment: one app serving a setup dashboard and a verified
  receiver at /webhook/{provider}, for whichever providers it is handed.
- WebhookEvent: the normalized event every provider parses into. Its dedupe key
  folds in the product's own timestamp, so a later change to one resource gets
  its own key rather than collapsing onto the first.
- idempotent_run: launch once per event key. Identity lives on a run *label*,
  never a name -- names race against concurrent launches and cap how many runs
  one key can ever have.
- Provider plus the verification primitives, written once because they are the
  part that is easy to get subtly wrong. constant_time_equals compares bytes:
  hmac.compare_digest raises TypeError on non-ASCII str operands, and these
  headers come off the wire, so a crafted one would turn a clean 401 into a 500.
- testing.assert_provider_conforms: the conformance check every product plugin
  runs, which replays a real sample delivery through verify and parse rather
  than trusting them to agree with each other.

No new dependencies. pydantic is already a runtime dependency; fastapi stays
behind the same lazy import flyte.app.extras._fastapi uses, so importing this
package never requires it -- only building the app does. Product plugins
declare their own [app] extra.

Tested against a stub provider, so the SDK depends on no product package.
examples/apps/webhook_custom_provider.py is a complete provider you can run
with no account at all.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VKZrTNjjWVzTZUxDFbn4Nk
Signed-off-by: Niels Bantilan <niels.bantilan@gmail.com>

Author: cosmicBboy

examples/apps/webhook_custom_provider.py

@@ -0,0 +1,162 @@
+"""Support a product this family does not ship a plugin for.
+
+A provider is small: say which environment variable holds its secret, how to
+verify a delivery, and how to turn a payload into a `WebhookEvent`. Core does
+the rest — the app, the dashboard, dispatch, the scope allowlist, and idempotent
+launching.
+
+Run it without an account:
+
+    python custom_provider.py --local
+
+That posts a signed sample delivery through the app in-process, so you see
+verification, normalization, and dispatch end to end.
+
+Once it works, move it into its own `flyteplugins-webhooks-<product>` package
+beside the others and add the one-line conformance test:
+
+    from flyte.extras.webhooks.testing import assert_provider_conforms
+    import flyteplugins.webhooks.acme as plugin
+
+    def test_conformance():
+        assert_provider_conforms(plugin)
+"""
+
+import hashlib
+import hmac
+import json
+import os
+import sys
+from typing import Mapping
+
+import flyte
+from flyte.extras.webhooks import (
+    EventType,
+    Provider,
+    WebhookAppEnvironment,
+    WebhookEvent,
+    constant_time_equals,
+    hex_hmac_sha256,
+    json_body,
+    lower_headers,
+)
+
+DEFAULT_SECRET_ENV = "ACME_WEBHOOK_SECRET"
+
+
+class Ticket(EventType):
+    """Acme's ticket events. `ANY` matches every action on the type."""
+
+    ANY = "ticket"
+    OPENED = "ticket.opened"
+    CLOSED = "ticket.closed"
+
+
+def verify(body: bytes, headers: Mapping[str, str], secret: str) -> bool:
+    """Verify Acme's hex HMAC-SHA256 over the raw body.
+
+    Use `constant_time_equals` rather than `hmac.compare_digest` directly: the
+    latter raises `TypeError` on `str` operands containing non-ASCII, and this
+    header comes off the wire, so a crafted one would turn a clean 401 into a
+    500.
+    """
+    signature = lower_headers(headers).get("x-acme-signature")
+    if not signature:
+        return False
+    return constant_time_equals(hex_hmac_sha256(secret, body), signature.strip())
+
+
+def parse(headers: Mapping[str, str], body: bytes) -> WebhookEvent:
+    """Normalize an Acme delivery.
+
+    Fill in `resource_id` and `occurred_at` wherever the product gives them:
+    together they are the dedupe key, and without a timestamp every later change
+    to one resource collapses onto the first one's key and never launches.
+    """
+    payload = json_body(body)
+    ticket = payload.get("ticket") or {}
+    return WebhookEvent(
+        provider="acme",
+        event_type="ticket",
+        action=payload.get("action"),
+        delivery_id=str(payload.get("delivery_id") or ""),
+        resource_id=str(ticket.get("id")) if ticket.get("id") is not None else None,
+        occurred_at=ticket.get("updated_at"),
+        scope=ticket.get("project"),
+        title=ticket.get("subject"),
+        url=ticket.get("url"),
+        payload=payload,
+    )
+
+
+class AcmeProvider(Provider):
+    """Acme's webhook provider, with its defaults pre-wired.
+
+    Users then write `providers=[AcmeProvider()]`, with `secret_env=` there for
+    anyone storing the secret under a different name.
+    """
+
+    def __init__(self, *, secret_env: str = DEFAULT_SECRET_ENV) -> None:
+        super().__init__(
+            name="acme",
+            secret_env=secret_env,
+            verify=verify,
+            parse=parse,
+            setup_hint="Acme Settings -> Webhooks",
+        )
+
+
+app_env = WebhookAppEnvironment(
+    name="acme-webhooks",
+    providers=[AcmeProvider()],
+    image=flyte.Image.from_debian_base(python_version=(3, 12)).with_pip_packages("fastapi", "uvicorn"),
+    secrets=[flyte.Secret(DEFAULT_SECRET_ENV, as_env_var=DEFAULT_SECRET_ENV)],
+)
+
+
+@app_env.on_event(Ticket.OPENED)
+async def on_ticket_opened(event):
+    return {"saw": event.qualified_type, "resource": event.resource_id, "dedupe_key": event.dedupe_key()}
+
+
+#: A realistic delivery, the same thing a shipped plugin exports as SAMPLE_DELIVERY.
+SAMPLE_BODY = json.dumps(
+    {
+        "action": "opened",
+        "delivery_id": "d-1",
+        "ticket": {
+            "id": 42,
+            "subject": "Printer on fire",
+            "project": "SUPPORT",
+            "updated_at": "2024-01-01T00:00:00Z",
+            "url": "https://acme.example/t/42",
+        },
+    }
+).encode()
+
+
+def _try_locally() -> None:
+    from fastapi.testclient import TestClient
+
+    secret = os.environ.setdefault(DEFAULT_SECRET_ENV, "local-trial-secret")
+    headers = {"X-Acme-Signature": hmac.new(secret.encode(), SAMPLE_BODY, hashlib.sha256).hexdigest()}
+    assert app_env.app is not None  # built in __post_init__
+    client = TestClient(app_env.app)
+
+    print("POST /webhook/acme  (signed with a throwaway secret)")
+    response = client.post("/webhook/acme", content=SAMPLE_BODY, headers=headers)
+    print(f"  {response.status_code}  {response.json()}\n")
+
+    print("an unsigned delivery is refused:")
+    bad = client.post("/webhook/acme", content=SAMPLE_BODY, headers={})
+    print(f"  {bad.status_code}  {bad.json()}")
+
+
+if __name__ == "__main__":
+    if "--local" in sys.argv:
+        _try_locally()
+    else:
+        flyte.init_from_config()
+        handle = flyte.serve(app_env)
+        handle.activate(wait=True)
+        print(f"Dashboard ready at {handle.endpoint}")

plugins/README-saas-integrations.md

@@ -0,0 +1,100 @@
+# SaaS integration plugins
+
+Receive SaaS webhooks in Flyte and launch runs from them.
+
+The receiver lives in `flyte.extras.webhooks`; each product is its own package
+contributing one `Provider`:
+
+| Package | Product | Verification |
+| --- | --- | --- |
+| [`flyteplugins-github`](../github) | GitHub | HMAC-SHA256 (`X-Hub-Signature-256`) |
+| [`flyteplugins-slack`](../slack) | Slack Events API | HMAC-SHA256 with a replay window (`X-Slack-Signature`) |
+| [`flyteplugins-linear`](../linear) | Linear | HMAC-SHA256 (`X-Linear-Signature`) |
+| [`flyteplugins-clickup`](../clickup) | ClickUp | HMAC-SHA256 (`X-Clickup-Signature`) |
+| [`flyteplugins-jira`](../jira) | Jira Cloud | none — Jira does not sign; a shared token stands in |
+
+Install core plus the packages for the products you wire up — each row above is
+a distribution name:
+
+```bash
+pip install "flyteplugins-github[app]"
+```
+
+## Try one without an account
+
+Every plugin ships an example that replays a real sample delivery through the
+app in-process — verification, normalization, and dispatch, with nothing to
+configure:
+
+```bash
+python plugins/webhooks/github/examples/github_webhooks.py --local
+```
+
+## The division of labor
+
+**`flyte.extras.webhooks` owns** the app, the dashboard, dispatch, the scope
+allowlist, idempotent launching, the normalized event, and the verification
+primitives — the parts that are easy to get subtly wrong and expensive to get
+wrong once per product. It ships with flyte, and adds no runtime dependency:
+serving the app needs `fastapi`, which stays an optional extra.
+
+**A provider plugin owns** only what is specific to its product: which
+environment variable holds the secret, how to verify a delivery, how to parse
+one into a `WebhookEvent`, and the typed constants for its events. That is
+usually under 150 lines.
+
+**Calling the product's API** is where these packages have room to grow. Today
+they own webhooks only, and the recipes in `examples/external_saas_integrations`
+call `PyGithub`, `slack_sdk`, and the rest directly. A plugin earns a client
+method when it does something the vendor SDK cannot — return a `flyte.io.File`
+instead of an inline megabyte diff, render into the task report, or participate
+in caching and fan-out. Forwarding arguments and reshaping JSON does not.
+
+## One app, many products
+
+```python
+import flyte
+from flyte.extras.webhooks import WebhookAppEnvironment
+from flyteplugins.github import GitHubProvider
+from flyteplugins.github import events as github_events
+from flyteplugins.slack import SlackProvider
+
+app_env = WebhookAppEnvironment(name="saas-webhooks", providers=[GitHubProvider(), SlackProvider()])
+
+
+@app_env.on_event(github_events.PullRequest.OPENED)
+async def triage(event): ...
+```
+
+Each provider gets a route at `/webhook/<name>`; anything not configured 404s.
+The dashboard at `/` shows one row per provider with its payload URL, whether
+its secret is mounted, and how it is verified.
+
+## Conformance
+
+Every provider plugin ships the same one-line test:
+
+```python
+from flyte.extras.webhooks.testing import assert_provider_conforms
+import flyteplugins.github as plugin
+
+
+def test_conformance():
+    assert_provider_conforms(plugin)
+```
+
+CI fails if a plugin drifts. The harness checks the things that actually go
+wrong: a verifier that raises instead of returning False on a hostile header,
+event constants that render as `Class.MEMBER` rather than their wire value, a
+dedupe key that is unstable, a sample delivery that no constant spells.
+
+It leans on `SAMPLE_DELIVERY`, a real payload each plugin ships. Without one
+there is no way to assert that `verify` and `parse` agree with the product
+rather than merely with each other.
+
+## Adding a product
+
+Copy the smallest existing plugin, implement `verify` and `parse`, export a
+`Provider` subclass with its defaults pre-wired, `events`, and
+`SAMPLE_DELIVERY`, and add the conformance test. `examples/apps/webhook_custom_provider.py`
+is a complete worked version you can run.

src/flyte/extras/__init__.py

@@ -14,13 +14,19 @@
 3. Sleep: Route a task to the backend `core-sleep` plugin, which executes in leaseworker with no
                    task pod.
 
-4. Shell: Wrap a CLI tool packaged in a container image. Designed as the foundation for
+4. Webhooks: Receive SaaS webhooks (GitHub, Slack, Jira, ...) and turn them into runs.
+             `WebhookAppEnvironment` serves a verified receiver, and `idempotent_run`
+             launches once per event key so a redelivery is a no-op. Products plug in
+             through `Provider`; the `flyteplugins-<product>` packages ship those.
+             Serving the app needs `fastapi`, which stays an optional extra.
+
+5. Shell: Wrap a CLI tool packaged in a container image. Designed as the foundation for
                 bio module libraries (bedtools, samtools, bcftools, GATK, etc.) and any other case
                 where a user wants to call a pre-built binary in a published container with
                 typed inputs and outputs.
 """
 
-from . import shell
+from . import shell, webhooks
 from ._container import ContainerTask
 from ._dynamic_batcher import (
     BatchStats,
@@ -46,4 +52,5 @@
     "serialize",
     "serialize_env",
     "shell",
+    "webhooks",
 ]

src/flyte/extras/webhooks/__init__.py

@@ -0,0 +1,83 @@
+"""Receive SaaS webhooks in Flyte, and turn them into runs.
+
+This package holds the product-agnostic machinery, so each
+`flyteplugins-<product>` plugin stays thin and consistent:
+
+- `WebhookAppEnvironment` — one app serving a dashboard and a verified receiver
+  at `/webhook/{provider}`, for whichever providers you hand it.
+- `Provider` — the contract a plugin implements: which env var holds its secret,
+  how to verify a delivery, how to parse one into an event.
+- `WebhookEvent` — the normalized event every provider parses into, so handlers
+  and dedupe keys work the same regardless of which product sent it.
+- `idempotent_run` — launch a run once per event key. Webhook senders retry on
+  any non-2xx and operators re-trigger by hand; this makes that safe.
+- `EventType` — base for the typed event constants each plugin ships.
+- `flyte.extras.webhooks.testing` — `assert_provider_conforms`, the
+  CI-enforced conformance check every plugin runs.
+
+Serving the app needs `fastapi` and `uvicorn`, which flyte keeps as the `app`
+extra rather than as runtime dependencies — importing this package never
+requires them; only building the app does.
+
+The division of labor: core owns the app, dispatch, dedupe, and the verification
+primitives that are easy to get subtly wrong; a plugin owns only what is
+specific to its product.
+
+```python
+import flyte
+from flyte.extras.webhooks import DuplicateRun, WebhookAppEnvironment, idempotent_run
+from flyteplugins.github import GitHubProvider
+from flyteplugins.github import events
+
+app_env = WebhookAppEnvironment(name="saas-webhooks", providers=[GITHUB])
+
+
+@app_env.on_event(events.PullRequest.OPENED)
+async def triage(event):
+    import flyte.remote as remote
+
+    task = remote.Task.get(name="github-triage.triage_pr", auto_version="latest")
+    try:
+        run = await idempotent_run.aio(task, key=event.dedupe_key(), repo=event.scope)
+    except DuplicateRun as exc:
+        return {"skipped": str(exc)}
+    return {"run": run.name}
+```
+"""
+
+from ._app import EventHandler, WebhookAppEnvironment
+from ._errors import SignatureError, WebhookPluginError
+from ._event import WebhookEvent
+from ._event_type import EventType
+from ._idempotent_run import DUPE_LABEL_KEY, DuplicateRun, blocking_run, idempotent_run
+from ._provider import (
+    HandshakeFn,
+    ParseFn,
+    Provider,
+    VerifyFn,
+    constant_time_equals,
+    hex_hmac_sha256,
+    json_body,
+    lower_headers,
+)
+
+__all__ = [
+    "DUPE_LABEL_KEY",
+    "DuplicateRun",
+    "EventHandler",
+    "EventType",
+    "HandshakeFn",
+    "ParseFn",
+    "Provider",
+    "SignatureError",
+    "VerifyFn",
+    "WebhookAppEnvironment",
+    "WebhookEvent",
+    "WebhookPluginError",
+    "blocking_run",
+    "constant_time_equals",
+    "hex_hmac_sha256",
+    "idempotent_run",
+    "json_body",
+    "lower_headers",
+]