Skip to content

Commit f2b11a4

Browse files
cosmicBboyclaude
cosmicBboy
and
claude
committed
feat(webhooks): add flyteplugins-webhooks-core
The shared contract for receiving SaaS webhooks in Flyte, in the shape of plugins/agents: core holds the product-agnostic machinery, and each product gets its own thin package implementing one interface. Core owns the app (one dashboard, one verified receiver at /webhook/{provider}), the normalized WebhookEvent, idempotent_run, the Provider contract, the verification primitives, and the conformance harness every product plugin runs. Two pieces worth calling out. idempotent_run keys identity on a run *label*, never a name: names race against concurrent launches and cap how many runs one key can ever have. And constant_time_equals compares bytes, because hmac.compare_digest raises TypeError on non-ASCII str operands and these headers come off the wire — a crafted one would otherwise turn a clean 401 into a 500. Core is tested against a stub provider, so it depends on no product package. 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>
1 parent 33548bf commit f2b11a4

16 files changed

Lines changed: 3296 additions & 0 deletions

File tree

‎plugins/webhooks/README.md‎

Lines changed: 83 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,83 @@
1+
# Flyte webhook plugins
2+
3+
Receive SaaS webhooks in Flyte and launch runs from them.
4+
5+
Each product is its own package, sharing one contract from `core`:
6+
7+
| Package | Product | Verification |
8+
| --- | --- | --- |
9+
| [`flyteplugins-webhooks-core`](core) | — | the shared app, event model, and contract |
10+
| [`flyteplugins-webhooks-github`](github) | GitHub | HMAC-SHA256 (`X-Hub-Signature-256`) |
11+
| [`flyteplugins-webhooks-slack`](slack) | Slack Events API | HMAC-SHA256 with a replay window (`X-Slack-Signature`) |
12+
| [`flyteplugins-webhooks-linear`](linear) | Linear | HMAC-SHA256 (`X-Linear-Signature`) |
13+
| [`flyteplugins-webhooks-clickup`](clickup) | ClickUp | HMAC-SHA256 (`X-Clickup-Signature`) |
14+
| [`flyteplugins-webhooks-jira`](jira) | Jira Cloud | none — Jira does not sign; a shared token stands in |
15+
16+
Install only what you wire up:
17+
18+
```bash
19+
pip install "flyteplugins-webhooks-core[app]" flyteplugins-webhooks-github
20+
```
21+
22+
## The division of labor
23+
24+
**Core owns** the app, the dashboard, dispatch, the scope allowlist, idempotent
25+
launching, the normalized event, and the verification primitives — the parts
26+
that are easy to get subtly wrong and expensive to get wrong once per product.
27+
28+
**A provider plugin owns** only what is specific to its product: which
29+
environment variable holds the secret, how to verify a delivery, how to parse
30+
one into a `WebhookEvent`, and the typed constants for its events. That is
31+
usually under 150 lines.
32+
33+
**Neither owns calling the product's API.** `PyGithub`, `slack_sdk`, and the
34+
rest are maintained by people with live API access; a task calling them directly
35+
needs nothing in between. See `examples/external_saas_integrations`.
36+
37+
## One app, many products
38+
39+
```python
40+
import flyte
41+
from flyteplugins.webhooks.core import WebhookAppEnvironment
42+
from flyteplugins.webhooks.github import PROVIDER as GITHUB
43+
from flyteplugins.webhooks.github import events as github_events
44+
from flyteplugins.webhooks.slack import PROVIDER as SLACK
45+
46+
app_env = WebhookAppEnvironment(name="saas-webhooks", providers=[GITHUB, SLACK])
47+
48+
49+
@app_env.on_event(github_events.PullRequest.OPENED)
50+
async def triage(event): ...
51+
```
52+
53+
Each provider gets a route at `/webhook/<name>`; anything not configured 404s.
54+
The dashboard at `/` shows one row per provider with its payload URL, whether
55+
its secret is mounted, and how it is verified.
56+
57+
## Conformance
58+
59+
Every provider plugin ships the same one-line test:
60+
61+
```python
62+
from flyteplugins.webhooks.core.testing import assert_provider_conforms
63+
import flyteplugins.webhooks.github as plugin
64+
65+
66+
def test_conformance():
67+
assert_provider_conforms(plugin)
68+
```
69+
70+
CI fails if a plugin drifts. The harness checks the things that actually go
71+
wrong: a verifier that raises instead of returning False on a hostile header,
72+
event constants that render as `Class.MEMBER` rather than their wire value, a
73+
dedupe key that is unstable, a sample delivery that no constant spells.
74+
75+
It leans on `SAMPLE_DELIVERY`, a real payload each plugin ships. Without one
76+
there is no way to assert that `verify` and `parse` agree with the product
77+
rather than merely with each other.
78+
79+
## Adding a product
80+
81+
Copy the smallest existing plugin, implement `verify` and `parse`, export
82+
`PROVIDER`, `events`, and `SAMPLE_DELIVERY`, and add the conformance test.
83+
`core/README.md` has the full contract.

‎plugins/webhooks/core/README.md‎

Lines changed: 128 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,128 @@
1+
# flyteplugins-webhooks-core
2+
3+
The shared contract every `flyteplugins-webhooks-<product>` plugin implements.
4+
5+
This package holds the product-agnostic machinery, so each provider plugin stays
6+
thin: which environment variable holds its secret, how to verify a delivery, how
7+
to parse one into an event. Everything else lives here.
8+
9+
```bash
10+
pip install "flyteplugins-webhooks-core[app]" flyteplugins-webhooks-github
11+
```
12+
13+
## What core owns
14+
15+
| | |
16+
| --- | --- |
17+
| `WebhookAppEnvironment` | One app: a setup dashboard and a verified receiver at `/webhook/{provider}`, for whichever providers you hand it |
18+
| `Provider` | The contract a plugin implements |
19+
| `WebhookEvent` | The normalized event every provider parses into |
20+
| `idempotent_run` | Launch a run once per event key |
21+
| `EventType` | Base for the typed event constants each plugin ships |
22+
| `constant_time_equals`, `hex_hmac_sha256`, `json_body`, `lower_headers` | Verification primitives, written once |
23+
| `testing.assert_provider_conforms` | The conformance check every plugin runs |
24+
25+
## Using it
26+
27+
```python
28+
import flyte
29+
from flyteplugins.webhooks.core import DuplicateRun, WebhookAppEnvironment, idempotent_run
30+
from flyteplugins.webhooks.github import PROVIDER, events
31+
32+
app_env = WebhookAppEnvironment(
33+
name="github-webhooks",
34+
providers=[PROVIDER],
35+
secrets=[flyte.Secret("GITHUB_WEBHOOK_SECRET", as_env_var="GITHUB_WEBHOOK_SECRET")],
36+
)
37+
38+
39+
@app_env.on_event(events.PullRequest.OPENED)
40+
async def triage(event):
41+
import flyte.remote as remote
42+
43+
task = remote.Task.get(name="github-triage.triage_pr", auto_version="latest")
44+
try:
45+
run = await idempotent_run.aio(task, key=event.dedupe_key(), repo=event.scope)
46+
except DuplicateRun as exc:
47+
return {"skipped": str(exc)}
48+
return {"run": run.name}
49+
50+
51+
flyte.serve(app_env)
52+
```
53+
54+
Handlers must `await idempotent_run.aio(...)`. The blocking form stalls the
55+
app's event loop, and webhook senders time deliveries out in seconds.
56+
57+
## The normalized event
58+
59+
Five payload shapes, one model. Handlers match on `qualified_type` and read the
60+
fields they need; `payload` always carries the provider's original JSON.
61+
62+
| Field | Meaning |
63+
| --- | --- |
64+
| `provider` | Which plugin parsed it |
65+
| `event_type` / `action` | The product's type and action; `qualified_type` joins them |
66+
| `resource_id` | Issue key, task id, message timestamp — what the event is about |
67+
| `occurred_at` | The product's timestamp, when it sends one |
68+
| `scope` | Repository, channel, team, list, or project key |
69+
| `title`, `url`, `actor` | For dashboards and messages |
70+
| `payload` | The original JSON, verbatim |
71+
72+
`event.dedupe_key()` combines provider, qualified type, resource, and timestamp.
73+
The timestamp is what makes it usable for `update`-shaped events: without it,
74+
every later change to one resource would collapse onto the first one's key and
75+
never launch. The key is just a string — build your own and pass it to
76+
`idempotent_run` when you want a different scope, such as one run per Slack
77+
thread rather than per message.
78+
79+
## Idempotent launching
80+
81+
`idempotent_run` refuses to launch when a run carrying the same `dedupe` label
82+
is live or has succeeded. Failed, aborted, and timed-out runs do not block:
83+
re-triggering after a failure is a retry.
84+
85+
Identity lives on the label, never a run name. Names race against concurrent
86+
launches and cap how many runs one key can ever have; the control plane assigns
87+
them.
88+
89+
The label check is a read followed by a launch, so two *simultaneous* deliveries
90+
of one event can both launch. Redeliveries are seconds to minutes apart and
91+
dedupe reliably; closing the concurrent case needs a compare-and-set the control
92+
plane does not expose.
93+
94+
## Writing a provider plugin
95+
96+
Implement `Provider` and export three names:
97+
98+
```python
99+
from flyteplugins.webhooks.core import Provider, WebhookEvent, constant_time_equals, hex_hmac_sha256
100+
101+
def verify(body, headers, secret) -> bool: ...
102+
def parse(headers, body) -> WebhookEvent: ...
103+
104+
PROVIDER = Provider(name="acme", secret_env="ACME_WEBHOOK_SECRET", verify=verify, parse=parse)
105+
SAMPLE_DELIVERY = (build_headers, b'{"real": "payload"}')
106+
```
107+
108+
Then add the one-line conformance test:
109+
110+
```python
111+
from flyteplugins.webhooks.core.testing import assert_provider_conforms
112+
import flyteplugins.webhooks.acme as plugin
113+
114+
115+
def test_conformance():
116+
assert_provider_conforms(plugin)
117+
```
118+
119+
`SAMPLE_DELIVERY` is what makes the rest checkable. The harness signs and
120+
replays it, so `verify` and `parse` are checked against a real payload rather
121+
than against each other. It also asserts what is easy to get wrong: that a
122+
verifier returns False rather than raising on attacker-controlled headers, that
123+
event constants render as wire values rather than enum names, and that the
124+
sample parses to something the constants actually spell.
125+
126+
Use `constant_time_equals` rather than `hmac.compare_digest` directly. The
127+
latter raises `TypeError` on `str` operands containing non-ASCII, and these
128+
headers come off the wire — a crafted one would turn a clean 401 into a 500.

‎plugins/webhooks/core/pyproject.toml‎

Lines changed: 58 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,58 @@
1+
[project]
2+
name = "flyteplugins-webhooks-core"
3+
dynamic = ["version"]
4+
description = "Shared contract for Flyte SaaS webhook receivers."
5+
readme = "README.md"
6+
authors = [{ name = "Flyte Contributors" }]
7+
requires-python = ">=3.10"
8+
dependencies = ["flyte", "pydantic>=2"]
9+
10+
[project.optional-dependencies]
11+
app = ["fastapi>=0.115", "uvicorn>=0.30"]
12+
13+
[build-system]
14+
requires = ["setuptools", "setuptools_scm"]
15+
build-backend = "setuptools.build_meta"
16+
17+
[dependency-groups]
18+
dev = [
19+
"pytest>=8.3.5",
20+
"pytest-asyncio>=0.26.0",
21+
"fastapi>=0.115",
22+
"uvicorn>=0.30",
23+
"httpx>=0.27",
24+
]
25+
26+
[tool.setuptools]
27+
include-package-data = true
28+
29+
[tool.setuptools.packages.find]
30+
where = ["src"]
31+
include = ["flyteplugins*"]
32+
33+
[tool.setuptools_scm]
34+
root = "../../.."
35+
36+
[tool.pytest.ini_options]
37+
norecursedirs = []
38+
log_cli = true
39+
log_cli_level = 20
40+
markers = []
41+
asyncio_mode = "auto"
42+
asyncio_default_fixture_loop_scope = "function"
43+
44+
[tool.coverage.run]
45+
branch = true
46+
47+
[tool.ruff]
48+
line-length = 120
49+
50+
[tool.ruff.lint]
51+
select = ["E", "W", "F", "I", "PLW", "YTT", "ASYNC", "C4", "T10", "EXE", "ISC", "LOG", "PIE", "Q", "RSE", "FLY", "PGH", "PLC", "PLE", "FURB", "RUF"]
52+
ignore = ["PGH003", "PLC0415"]
53+
54+
[tool.ruff.lint.per-file-ignores]
55+
"examples/*" = ["E402"]
56+
57+
[tool.uv.sources]
58+
flyte = { path = "../../..", editable = true }

‎plugins/webhooks/core/src/flyteplugins/webhooks/core/__init__.py‎

Lines changed: 79 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,79 @@
1+
"""flyteplugins-webhooks-core — the shared contract every webhook plugin implements.
2+
3+
This package holds the product-agnostic machinery, so each
4+
`flyteplugins-webhooks-<product>` plugin stays thin and consistent:
5+
6+
- `WebhookAppEnvironment` — one app serving a dashboard and a verified receiver
7+
at `/webhook/{provider}`, for whichever providers you hand it.
8+
- `Provider` — the contract a plugin implements: which env var holds its secret,
9+
how to verify a delivery, how to parse one into an event.
10+
- `WebhookEvent` — the normalized event every provider parses into, so handlers
11+
and dedupe keys work the same regardless of which product sent it.
12+
- `idempotent_run` — launch a run once per event key. Webhook senders retry on
13+
any non-2xx and operators re-trigger by hand; this makes that safe.
14+
- `EventType` — base for the typed event constants each plugin ships.
15+
- `flyteplugins.webhooks.core.testing` — `assert_provider_conforms`, the
16+
CI-enforced conformance check every plugin runs.
17+
18+
The division of labor: core owns the app, dispatch, dedupe, and the verification
19+
primitives that are easy to get subtly wrong; a plugin owns only what is
20+
specific to its product.
21+
22+
```python
23+
import flyte
24+
from flyteplugins.webhooks.core import DuplicateRun, WebhookAppEnvironment, idempotent_run
25+
from flyteplugins.webhooks.github import PROVIDER as GITHUB
26+
from flyteplugins.webhooks.github import events
27+
28+
app_env = WebhookAppEnvironment(name="saas-webhooks", providers=[GITHUB])
29+
30+
31+
@app_env.on_event(events.PullRequest.OPENED)
32+
async def triage(event):
33+
import flyte.remote as remote
34+
35+
task = remote.Task.get(name="github-triage.triage_pr", auto_version="latest")
36+
try:
37+
run = await idempotent_run.aio(task, key=event.dedupe_key(), repo=event.scope)
38+
except DuplicateRun as exc:
39+
return {"skipped": str(exc)}
40+
return {"run": run.name}
41+
```
42+
"""
43+
44+
from ._app import EventHandler, WebhookAppEnvironment
45+
from ._errors import SignatureError, WebhookPluginError
46+
from ._event import WebhookEvent
47+
from ._event_type import EventType
48+
from ._idempotent_run import DUPE_LABEL_KEY, DuplicateRun, blocking_run, idempotent_run
49+
from ._provider import (
50+
HandshakeFn,
51+
ParseFn,
52+
Provider,
53+
VerifyFn,
54+
constant_time_equals,
55+
hex_hmac_sha256,
56+
json_body,
57+
lower_headers,
58+
)
59+
60+
__all__ = [
61+
"DUPE_LABEL_KEY",
62+
"DuplicateRun",
63+
"EventHandler",
64+
"EventType",
65+
"HandshakeFn",
66+
"ParseFn",
67+
"Provider",
68+
"SignatureError",
69+
"VerifyFn",
70+
"WebhookAppEnvironment",
71+
"WebhookEvent",
72+
"WebhookPluginError",
73+
"blocking_run",
74+
"constant_time_equals",
75+
"hex_hmac_sha256",
76+
"idempotent_run",
77+
"json_body",
78+
"lower_headers",
79+
]

0 commit comments

Comments
 (0)