From 476b71532b649c60a0622444b84265a55f3b3040 Mon Sep 17 00:00:00 2001 From: Krish Garg Date: Tue, 14 Jul 2026 21:07:53 -0700 Subject: [PATCH 1/2] feat(nodes): add Slack Events source node Add slack, an ingress source node that authenticates Slack Events API callbacks and routes approved message and reaction events into RocketRide pipelines. It follows the existing source endpoint and response-lane patterns. - Verify request signatures and timestamp freshness before parsing - Route message text and reaction JSON with bounded queueing and deduplication - Add documentation, a native Slack icon, an example pipeline, and focused tests Tests: 61 Slack tests and 304 contract tests passed; live Cursor canvas verification delivered a Slack message and reaction without errors. --- examples/slack-events.pipe | 78 +++++ nodes/src/nodes/slack/IEndpoint.py | 176 +++++++++++ nodes/src/nodes/slack/IGlobal.py | 46 +++ nodes/src/nodes/slack/IInstance.py | 28 ++ nodes/src/nodes/slack/README.md | 136 ++++++++ nodes/src/nodes/slack/__init__.py | 32 ++ nodes/src/nodes/slack/requirements.txt | 1 + nodes/src/nodes/slack/services.json | 128 ++++++++ nodes/src/nodes/slack/slack-events.svg | 6 + nodes/src/nodes/slack/slack_events.py | 127 ++++++++ nodes/test/slack/__init__.py | 103 ++++++ nodes/test/slack/test_endpoint.py | 421 +++++++++++++++++++++++++ nodes/test/slack/test_queue_dedup.py | 40 +++ nodes/test/slack/test_routing.py | 100 ++++++ nodes/test/slack/test_signature.py | 172 ++++++++++ 15 files changed, 1594 insertions(+) create mode 100644 examples/slack-events.pipe create mode 100644 nodes/src/nodes/slack/IEndpoint.py create mode 100644 nodes/src/nodes/slack/IGlobal.py create mode 100644 nodes/src/nodes/slack/IInstance.py create mode 100644 nodes/src/nodes/slack/README.md create mode 100644 nodes/src/nodes/slack/__init__.py create mode 100644 nodes/src/nodes/slack/requirements.txt create mode 100644 nodes/src/nodes/slack/services.json create mode 100644 nodes/src/nodes/slack/slack-events.svg create mode 100644 nodes/src/nodes/slack/slack_events.py create mode 100644 nodes/test/slack/__init__.py create mode 100644 nodes/test/slack/test_endpoint.py create mode 100644 nodes/test/slack/test_queue_dedup.py create mode 100644 nodes/test/slack/test_routing.py create mode 100644 nodes/test/slack/test_signature.py diff --git a/examples/slack-events.pipe b/examples/slack-events.pipe new file mode 100644 index 000000000..317790683 --- /dev/null +++ b/examples/slack-events.pipe @@ -0,0 +1,78 @@ +{ + "components": [ + { + "id": "slack_1", + "provider": "slack", + "config": { + "parameters": { + "dedupTtlSeconds": 600, + "queueCapacity": 1000, + "signingSecret": "", + "google": {} + }, + "type": "slack", + "hideForm": true, + "mode": "Source" + }, + "ui": { + "position": { + "x": 20, + "y": 200 + }, + "nodeType": "default", + "formDataValid": true + } + }, + { + "id": "response_text_1", + "provider": "response_text", + "config": { + "laneName": "slack_messages" + }, + "ui": { + "position": { + "x": 280, + "y": 120 + }, + "nodeType": "default", + "formDataValid": true + }, + "input": [ + { + "lane": "text", + "from": "slack_1" + } + ] + }, + { + "id": "response_json_1", + "provider": "response_json", + "config": { + "laneName": "slack_reactions" + }, + "ui": { + "position": { + "x": 280, + "y": 280 + }, + "nodeType": "default", + "formDataValid": true + }, + "input": [ + { + "lane": "json", + "from": "slack_1" + } + ] + } + ], + "project_id": "32e2808e-9557-4080-8b23-a0e9818f8cf4", + "version": 1, + "isLocked": false, + "snapToGrid": true, + "snapGridSize": [ + 10, + 10 + ], + "docRevision": 8 +} \ No newline at end of file diff --git a/nodes/src/nodes/slack/IEndpoint.py b/nodes/src/nodes/slack/IEndpoint.py new file mode 100644 index 000000000..81a3487a3 --- /dev/null +++ b/nodes/src/nodes/slack/IEndpoint.py @@ -0,0 +1,176 @@ +# ============================================================================= +# MIT License +# Copyright (c) 2026 Aparavi Software AG +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the "Software"), to deal +# in the Software without restriction, including without limitation the rights +# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +# copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in +# all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +# SOFTWARE. +# ============================================================================= + +import argparse +import asyncio +import json +import sys +from collections.abc import Callable +from typing import Any + +from ai.web import WebServer +from fastapi import Request +from fastapi.responses import PlainTextResponse +from rocketlib import ( + OPEN_MODE, + IEndpointBase, + IJson, + getObject, + monitorCompleted, + monitorFailed, + monitorOther, + monitorStatus, +) + +from .slack_events import TtlDedupCache, classify_event, verify_slack_signature + + +class IEndpoint(IEndpointBase): + target: IEndpointBase | None = None + _signing_secret: str = '' + _queue: asyncio.Queue | None = None + _dedup: TtlDedupCache | None = None + _consumer_task: asyncio.Task | None = None + _accepting: bool = False + + async def _request_handler(self, request: Request): + raw_body = await request.body() + timestamp = request.headers.get('X-Slack-Request-Timestamp', '') + signature = request.headers.get('X-Slack-Signature', '') + if not verify_slack_signature(self._signing_secret, timestamp, signature, raw_body): + return PlainTextResponse('', status_code=401) + try: + envelope = json.loads(raw_body) + except (TypeError, ValueError, json.JSONDecodeError): + return PlainTextResponse('', status_code=400) + if not isinstance(envelope, dict): + return PlainTextResponse('', status_code=400) + if envelope.get('type') == 'url_verification': + return PlainTextResponse(str(envelope.get('challenge', ''))) + routed = classify_event(envelope) + if routed is None: + return PlainTextResponse('') + if not self._accepting or self._queue is None or self._dedup is None: + return PlainTextResponse('', status_code=503) + if self._dedup.contains(envelope['event_id']): + return PlainTextResponse('') + try: + self._queue.put_nowait(routed) + except asyncio.QueueFull: + return PlainTextResponse('', status_code=503) + self._dedup.add(envelope['event_id']) + return PlainTextResponse('') + + def _emit_event(self, routed) -> None: + envelope = routed.envelope + event_id = envelope['event_id'] + team_id = envelope.get('team_id', '') + entry = getObject( + obj={ + 'url': f'slack://{team_id}/{event_id}', + 'name': event_id, + } + ) + entry.fromDict({'metadata': envelope}) + payload_size = ( + len(routed.payload.encode('utf-8')) + if routed.lane == 'text' + else len(json.dumps(routed.payload, separators=(',', ':'), ensure_ascii=False).encode('utf-8')) + ) + pipe = self.target.getPipe() + try: + pipe.open(entry) + if routed.lane == 'text': + pipe.writeText(routed.payload) + else: + pipe.writeJson(IJson(routed.payload)) + pipe.close() + monitorCompleted(payload_size) + except Exception: + monitorFailed(payload_size) + finally: + self.target.putPipe(pipe) + + async def _consume_queue(self) -> None: + while True: + routed = await self._queue.get() + try: + if routed is None: + return + try: + await asyncio.to_thread(self._emit_event, routed) + except Exception: + monitorFailed(0) + finally: + self._queue.task_done() + + def _queue_capacity(self) -> int: + parameters = getattr(self.endpoint, 'serviceConfig', {}).get('parameters', {}) + return min(10000, max(1, int(parameters.get('queueCapacity', 1000)))) + + def _dedup_ttl(self) -> int: + parameters = getattr(self.endpoint, 'serviceConfig', {}).get('parameters', {}) + return min(3600, max(300, int(parameters.get('dedupTtlSeconds', TtlDedupCache.TTL_SECONDS)))) + + async def _startup(self) -> None: + self._accepting = True + self._queue = asyncio.Queue(maxsize=self._queue_capacity()) + self._dedup = TtlDedupCache(ttl_seconds=self._dedup_ttl()) + self._consumer_task = asyncio.create_task(self._consume_queue()) + monitorStatus('Slack Events ready - waiting for events') + + async def _shutdown(self) -> None: + self._accepting = False + if self._queue is not None and self._consumer_task is not None: + try: + await asyncio.wait_for(self._queue.join(), timeout=5) + await self._queue.put(None) + await asyncio.wait_for(self._consumer_task, timeout=5) + except (asyncio.TimeoutError, asyncio.CancelledError): + self._consumer_task.cancel() + try: + await self._consumer_task + except asyncio.CancelledError: + pass + self._consumer_task = None + self._signing_secret = '' + monitorOther('usr', '') + + def _run(self) -> None: + parser = argparse.ArgumentParser(add_help=False) + parser.add_argument('--data_host', type=str, default='localhost') + parser.add_argument('--data_port', type=int, default=5567) + args, _ = parser.parse_known_args(sys.argv) + self._server = WebServer( + config={'host': args.data_host, 'port': args.data_port}, + on_startup=self._startup, + on_shutdown=self._shutdown, + ) + self._server.app.state.target = self.target + self._server.add_route('/slack/events', self._request_handler, ['POST'], public=True) + self._server.run() + + def scanObjects(self, _path: str, _scan_callback: Callable[[dict[str, Any]], None]): + self.target = self.endpoint.target + if self.endpoint.openMode != OPEN_MODE.CONFIG: + self._run() diff --git a/nodes/src/nodes/slack/IGlobal.py b/nodes/src/nodes/slack/IGlobal.py new file mode 100644 index 000000000..483d3752d --- /dev/null +++ b/nodes/src/nodes/slack/IGlobal.py @@ -0,0 +1,46 @@ +# ============================================================================= +# MIT License +# Copyright (c) 2026 Aparavi Software AG +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the "Software"), to deal +# in the Software without restriction, including without limitation the rights +# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +# copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in +# all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +# SOFTWARE. +# ============================================================================= + +from rocketlib import IGlobalBase, OPEN_MODE, warning + +from .slack_events import resolve_signing_secret + + +class IGlobal(IGlobalBase): + def validateConfig(self) -> None: + config = self.IEndpoint.endpoint.serviceConfig.get('parameters', {}) + if not resolve_signing_secret(config): + warning('Slack signing secret is missing.') + + def beginGlobal(self) -> None: + if self.IEndpoint.endpoint.openMode == OPEN_MODE.CONFIG: + return + config = self.IEndpoint.endpoint.serviceConfig.get('parameters', {}) + self.signing_secret = resolve_signing_secret(config) + self.IEndpoint._signing_secret = self.signing_secret + if not self.signing_secret: + warning('Slack signing secret is missing.') + + def endGlobal(self) -> None: + self.signing_secret = '' + self.IEndpoint._signing_secret = '' diff --git a/nodes/src/nodes/slack/IInstance.py b/nodes/src/nodes/slack/IInstance.py new file mode 100644 index 000000000..71acb6364 --- /dev/null +++ b/nodes/src/nodes/slack/IInstance.py @@ -0,0 +1,28 @@ +# ============================================================================= +# MIT License +# Copyright (c) 2026 Aparavi Software AG +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the "Software"), to deal +# in the Software without restriction, including without limitation the rights +# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +# copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in +# all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +# SOFTWARE. +# ============================================================================= + +from rocketlib import IInstanceBase + + +class IInstance(IInstanceBase): + pass diff --git a/nodes/src/nodes/slack/README.md b/nodes/src/nodes/slack/README.md new file mode 100644 index 000000000..29229b5d4 --- /dev/null +++ b/nodes/src/nodes/slack/README.md @@ -0,0 +1,136 @@ +# Slack Events + +## What it does + +Slack Events is a `source` node (`slack://`) that receives Slack Events API HTTP callbacks at `/slack/events`. It authenticates callbacks, acknowledges them quickly, and injects approved event content into a RocketRide pipeline. It makes no outbound Slack API calls. + +## Lanes and routing + +The `_source` lane emits `text` and `json`. These are the only five approved +event routes: + +| Slack callback | Output lane | Payload | +|---|---|---| +| `app_mention` | `text` | Exact `event.text` | +| `message.channels` | `text` | Exact `event.text` | +| `message.groups` | `text` | Exact `event.text` | +| `message.im` | `text` | Exact `event.text` | +| `reaction_added` | `json` | Exact inner `event` object | + +Each emitted entry retains the complete outer Slack envelope as native +`Entry.metadata` (`IJson`); it is not serialized into either output lane. + +Slack-marked bot/app text events are authenticated and acknowledged, then +ignored to prevent feedback loops. This applies when `event.bot_id` or +`event.app_id` is a nonempty string, or `event.subtype` is exactly +`bot_message`; it does not apply to `reaction_added` or other user-authored +message subtypes. +Unsupported or incomplete authenticated events are acknowledged and ignored. + +## Setup + +1. Create a Slack app and enable **Event Subscriptions**. +2. Set Slack's Request URL to the runtime-published `/slack/events` URL. +3. Subscribe to `app_mention`, `message.channels`, `message.groups`, + `message.im`, and `reaction_added`. Grant their matching scopes: + `app_mentions:read`, `channels:history`, `groups:history`, `im:history`, + and `reactions:read`. +4. Copy the app's signing secret to the secure **Signing Secret** field, or set + `SLACK_SIGNING_SECRET` in the runtime environment. + +The listener is available only while the pipeline is running. Slack must be able to reach the published URL over HTTPS. + +## Signing secret + +The secure `slack.signingSecret` configuration takes precedence when nonempty. +Otherwise the node reads `SLACK_SIGNING_SECRET`. Requests are verified against +their exact raw body with Slack's HMAC-SHA256 signature and a five-minute +timestamp freshness window before their JSON is parsed. + +## Queue and deduplication + +Accepted events enter a bounded in-memory queue. `slack.queueCapacity` defaults +to 1000 events and accepts values from 1 through 10000. A full queue returns +`503` without recording the event ID, allowing Slack to retry. + +Event IDs are deduplicated in process memory for 600 seconds by default. +`slack.dedupTtlSeconds` accepts values from 300 through 3600 seconds. The +deduplication cache holds at most 10000 IDs and is intentionally process-local: +a restart clears it, so a duplicate can be processed after a restart. IDs are +recorded only after successful enqueue. + +## Acknowledgement, retries, and delivery behavior + +Slack expects an acknowledgement within three seconds. After authenticating and +classifying a supported callback, the node queues it and responds without +waiting for pipeline processing. URL verification is authenticated before its +challenge is returned and is never emitted. Healthy and duplicate deliveries +receive an empty `200` response. Invalid signatures receive `401`, malformed +authenticated JSON receives `400`, and unavailable or saturated intake receives +`503`; Slack can retry the latter. + +After successful enqueue, the node does not provide a durable retry for +downstream pipe-emission failures: it records the failure and does not requeue +the delivery. + +The queue and deduplication cache are process-local and not durable. On an +orderly shutdown, intake stops and the node waits up to five seconds for queued +and in-flight work, then allows up to five seconds for the consumer to exit +before cancelling it. An abrupt process termination can lose accepted queued +work, and a restart clears deduplication state. + +Only the five approved callback shapes are emitted. Socket Mode, OAuth app +installation or provisioning, subscription discovery or management, and +outbound Slack Web API calls are outside this node. All other Slack event types +and incomplete approved shapes are ignored after authentication. + +## Example + +[`examples/slack-events.pipe`](../../../../examples/slack-events.pipe) connects +the `text` lane to `response_text` and the `json` lane to `response_json`. +Configure the signing secret outside the pipeline file and source control. + +## Upstream documentation + +- [Slack Events API](https://api.slack.com/apis/events-api) +- [Request URLs](https://api.slack.com/apis/events-api/using-http-request-urls) +- [Signing secret verification](https://docs.slack.dev/authentication/verifying-requests-from-slack/) +- [URL verification](https://api.slack.com/events/url_verification) +- [Events reference](https://api.slack.com/events) + +## Troubleshooting + +- **URL verification fails:** confirm the pipeline is running, the published + `/slack/events` URL is reachable, and the signing secret is correct. +- **Invalid signature:** check that a proxy does not alter the raw request body and that the configured secret belongs to the same Slack app. +- **Stale local clock:** synchronize the runtime host's clock; Slack signatures + outside the five-minute window are rejected. +- **Missing scopes:** confirm Event Subscriptions, all five individual event + types, and their matching scopes are configured in Slack. +- **Queue saturation:** reduce downstream work or raise queue capacity within + the 1--10000 limit; Slack retries `503` responses. +- **Duplicate deliveries:** duplicates are suppressed only while their event ID + remains in this process's TTL cache; a restart resets that cache. +- **Pipeline is not running:** start the pipeline before Slack sends callbacks; + the public listener exists only while the pipeline runs. + + + + +## Schema + +| Field | Type | Description | Default | +|---|---|---|---| +| `Pipe.source.parameters` | | **Slack Events Configuration** | | +| `slack.dedupTtlSeconds` | `integer` | **Deduplication TTL (seconds)**
How long accepted Slack event IDs are retained to prevent duplicate emission. | `600` | +| `slack.queueCapacity` | `integer` | **Queue Capacity**
Maximum accepted Slack events held in memory before Slack is asked to retry. | `1000` | +| `slack.signingSecret` | `string` | **Signing Secret**
Slack signing secret. If empty, SLACK_SIGNING_SECRET is used. | | + +## Dependencies + +- `fastapi` + +## Source + +[ View source](https://github.com/rocketride-org/rocketride-server/tree/develop/nodes/src/nodes/slack) + diff --git a/nodes/src/nodes/slack/__init__.py b/nodes/src/nodes/slack/__init__.py new file mode 100644 index 000000000..e1db387ce --- /dev/null +++ b/nodes/src/nodes/slack/__init__.py @@ -0,0 +1,32 @@ +# ============================================================================= +# MIT License +# Copyright (c) 2026 Aparavi Software AG +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the "Software"), to deal +# in the Software without restriction, including without limitation the rights +# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +# copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in +# all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +# SOFTWARE. +# ============================================================================= + +from .IEndpoint import IEndpoint +from .IGlobal import IGlobal +from .IInstance import IInstance + +__all__ = [ + 'IEndpoint', + 'IGlobal', + 'IInstance', +] diff --git a/nodes/src/nodes/slack/requirements.txt b/nodes/src/nodes/slack/requirements.txt new file mode 100644 index 000000000..6b0b9396e --- /dev/null +++ b/nodes/src/nodes/slack/requirements.txt @@ -0,0 +1 @@ +fastapi diff --git a/nodes/src/nodes/slack/services.json b/nodes/src/nodes/slack/services.json new file mode 100644 index 000000000..e38b135c4 --- /dev/null +++ b/nodes/src/nodes/slack/services.json @@ -0,0 +1,128 @@ +{ + // + // Required: + // The displayable name of this node + // + "title": "Slack Events", + // + // Required: + // The protocol is the endpoint protocol + // + "protocol": "slack://", + // + // Required: + // Class type of the node - what it does + // + "classType": ["source"], + // + // Required: + // Capabilities are flags that change the behavior of the underlying + // engine + // + "capabilities": ["noinclude"], + // + // Optional: + // Register is either filter, endpoint or ignored if not specified. If the + // type is specified, a factory is registered of that given type + // + "register": "endpoint", + // + // Optional: + // The node is the actual physical node to instantiate - if + // not specified, the protocol will be used + // + "node": "python", + // + // Optional: + // The path is the executable/script code - it is node dependent + // and is optional for most nodes + // + "path": "nodes.slack", + // + // Required: + // The prefix map when added/removed when converting URLs <=> paths + // + "prefix": "Slack", + // + // Optional: + // Description of this driver + // + "description": ["A Slack Events API source node that receives authenticated callbacks and routes approved events to text and JSON lanes."], + // + // Optional: + // The icon is the icon to display in the UI for this node + // + "icon": "slack-events.svg", + // + // Optional: + // The tile is the displayable name of this node + // + "tile": [], + // + // Optional: + // As a pipe component, define what this pipe component takes + // and what it produces + // + "lanes": { + "_source": ["text", "json"] + }, + // + // Optional: + // Profile section are configuration options used by the driver itself + // + "preconfig": { + "default": "default", + "profiles": { + "default": {} + } + }, + // + // Optional: + // Local field definitions - these define fields only for the + // current service. + // + "fields": { + "slack.signingSecret": { + "type": "string", + "title": "Signing Secret", + "description": "Slack signing secret. If empty, SLACK_SIGNING_SECRET is used.", + "optional": false, + "secure": true, + "ui": { + "ui:widget": "ApiKeyWidget" + } + }, + "slack.queueCapacity": { + "type": "integer", + "title": "Queue Capacity", + "description": "Maximum accepted Slack events held in memory before Slack is asked to retry.", + "default": 1000, + "minimum": 1, + "maximum": 10000 + }, + "slack.dedupTtlSeconds": { + "type": "integer", + "title": "Deduplication TTL (seconds)", + "description": "How long accepted Slack event IDs are retained to prevent duplicate emission.", + "default": 600, + "minimum": 300, + "maximum": 3600 + }, + "Pipe.source.parameters": { + "section": "parameters", + "title": "Slack Events Configuration", + "properties": ["slack.signingSecret", "slack.queueCapacity", "slack.dedupTtlSeconds"] + } + }, + // + // Required: + // Defines the fields (shape) of the service. + // + "shape": [ + { + "section": "Pipe", + "title": "Slack Events", + "properties": ["type", "Pipe.source.parameters"] + } + ] +} diff --git a/nodes/src/nodes/slack/slack-events.svg b/nodes/src/nodes/slack/slack-events.svg new file mode 100644 index 000000000..86f521f37 --- /dev/null +++ b/nodes/src/nodes/slack/slack-events.svg @@ -0,0 +1,6 @@ + + + + + + diff --git a/nodes/src/nodes/slack/slack_events.py b/nodes/src/nodes/slack/slack_events.py new file mode 100644 index 000000000..c44be3749 --- /dev/null +++ b/nodes/src/nodes/slack/slack_events.py @@ -0,0 +1,127 @@ +# ============================================================================= +# MIT License +# Copyright (c) 2026 Aparavi Software AG +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the "Software"), to deal +# in the Software without restriction, including without limitation the rights +# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +# copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in +# all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +# SOFTWARE. +# ============================================================================= + +import hashlib +import hmac +import os +import time +from collections import OrderedDict +from collections.abc import Mapping +from dataclasses import dataclass +from typing import Any + + +class TtlDedupCache: + TTL_SECONDS = 600 + MAX_ENTRIES = 10_000 + + def __init__(self, *, ttl_seconds: int = TTL_SECONDS, clock: Any = time.time): + self.ttl_seconds = ttl_seconds + self._clock = clock + self._entries: OrderedDict[str, float] = OrderedDict() + + def contains(self, event_id: str) -> bool: + self._prune() + return event_id in self._entries + + def add(self, event_id: str) -> None: + self._prune() + self._entries.pop(event_id, None) + self._entries[event_id] = self._clock() + while len(self._entries) > self.MAX_ENTRIES: + self._entries.popitem(last=False) + + def _prune(self) -> None: + expired_before = self._clock() - self.ttl_seconds + while self._entries: + _event_id, added_at = next(iter(self._entries.items())) + if added_at > expired_before: + break + self._entries.popitem(last=False) + + +def resolve_signing_secret(config: Mapping, environ: Mapping | None = None) -> str: + env = os.environ if environ is None else environ + configured = config.get('signingSecret', '') if hasattr(config, 'get') else '' + return str(configured or env.get('SLACK_SIGNING_SECRET', '') or '').strip() + + +def verify_slack_signature( + signing_secret: str, + timestamp: str, + provided_signature: str, + raw_body: bytes, + *, + now: float | None = None, +) -> bool: + if not signing_secret or not timestamp or not provided_signature: + return False + if not timestamp.isascii() or not timestamp.isdecimal(): + return False + request_time = int(timestamp) + current = time.time() if now is None else now + if abs(current - request_time) > 300: + return False + base = b'v0:' + timestamp.encode('ascii') + b':' + raw_body + expected = 'v0=' + hmac.new(signing_secret.encode(), base, hashlib.sha256).hexdigest() + return hmac.compare_digest(expected, provided_signature) + + +@dataclass(frozen=True) +class RoutedEvent: + event_type: str + lane: str + payload: Any + envelope: dict + + +_MESSAGE_TYPES = { + 'channel': 'message.channels', + 'group': 'message.groups', + 'im': 'message.im', +} + + +def classify_event(envelope: dict) -> RoutedEvent | None: + if envelope.get('type') != 'event_callback' or not isinstance(envelope.get('event_id'), str): + return None + inner = envelope.get('event') + if not isinstance(inner, dict): + return None + inner_type = inner.get('type') + is_bot_or_app_text = ( + (isinstance(inner.get('bot_id'), str) and bool(inner['bot_id'])) + or (isinstance(inner.get('app_id'), str) and bool(inner['app_id'])) + or inner.get('subtype') == 'bot_message' + ) + if inner_type in {'app_mention', 'message'} and is_bot_or_app_text: + return None + if inner_type == 'app_mention' and isinstance(inner.get('text'), str) and inner['text']: + return RoutedEvent('app_mention', 'text', inner['text'], envelope) + if inner_type == 'message': + logical = _MESSAGE_TYPES.get(inner.get('channel_type')) + if logical and isinstance(inner.get('text'), str) and inner['text']: + return RoutedEvent(logical, 'text', inner['text'], envelope) + if inner_type == 'reaction_added': + return RoutedEvent('reaction_added', 'json', inner, envelope) + return None diff --git a/nodes/test/slack/__init__.py b/nodes/test/slack/__init__.py new file mode 100644 index 000000000..95b8f13ec --- /dev/null +++ b/nodes/test/slack/__init__.py @@ -0,0 +1,103 @@ +# ============================================================================= +# MIT License +# Copyright (c) 2026 Aparavi Software AG +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the "Software"), to deal +# in the Software without restriction, including without limitation the rights +# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +# copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in +# all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +# SOFTWARE. +# ============================================================================= + +import sys +import types +from pathlib import Path + + +NODES_SRC = Path(__file__).resolve().parents[2] / 'src' +if str(NODES_SRC) not in sys.path: + sys.path.insert(0, str(NODES_SRC)) + +if 'depends' not in sys.modules: + depends = types.ModuleType('depends') + depends.depends = lambda _requirements: None + sys.modules['depends'] = depends + +if 'rocketlib' not in sys.modules: + rocketlib = types.ModuleType('rocketlib') + rocketlib.IEndpointBase = type('IEndpointBase', (), {}) + rocketlib.IGlobalBase = type('IGlobalBase', (), {}) + rocketlib.IInstanceBase = type('IInstanceBase', (), {}) + rocketlib.OPEN_MODE = types.SimpleNamespace(CONFIG='config', SOURCE='source') + rocketlib.warning = lambda _message: None + + class IJson(dict): + def toDict(self): + return dict(self) + + class Entry: + def __init__(self, *, url, name): + self.url = url + self.name = name + self._metadata = IJson() + + @property + def metadata(self): + return self._metadata + + def fromDict(self, value): + self._metadata = IJson(value['metadata']) + + rocketlib.IJson = IJson + rocketlib.getObject = lambda *, obj: Entry(**obj) + rocketlib.monitorCompleted = lambda _count: None + rocketlib.monitorFailed = lambda _count: None + rocketlib.monitorOther = lambda *_args: None + rocketlib.monitorStatus = lambda _message: None + sys.modules['rocketlib'] = rocketlib + +if 'ai.common.config' not in sys.modules: + ai = sys.modules.setdefault('ai', types.ModuleType('ai')) + common = types.ModuleType('ai.common') + config = types.ModuleType('ai.common.config') + config.Config = type('Config', (), {'getNodeConfig': staticmethod(lambda _logical_type, _config: {})}) + ai.common = common + common.config = config + sys.modules['ai.common'] = common + sys.modules['ai.common.config'] = config + +if 'ai.web' not in sys.modules: + web = types.ModuleType('ai.web') + web.WebServer = type('WebServer', (), {}) + sys.modules['ai'].web = web + sys.modules['ai.web'] = web + +if 'fastapi.responses' not in sys.modules: + fastapi = types.ModuleType('fastapi') + responses = types.ModuleType('fastapi.responses') + + class Request: + pass + + class _Response: + def __init__(self, content='', status_code=200): + self.body = content.encode() if isinstance(content, str) else content + self.status_code = status_code + + responses.PlainTextResponse = _Response + fastapi.Request = Request + fastapi.responses = responses + sys.modules['fastapi'] = fastapi + sys.modules['fastapi.responses'] = responses diff --git a/nodes/test/slack/test_endpoint.py b/nodes/test/slack/test_endpoint.py new file mode 100644 index 000000000..c98b680d5 --- /dev/null +++ b/nodes/test/slack/test_endpoint.py @@ -0,0 +1,421 @@ +# ============================================================================= +# MIT License +# ============================================================================= + +import asyncio +import hashlib +import hmac +import importlib +import inspect +import json +import threading +import time +import types + +import pytest +from fastapi import Request + +from nodes.slack.slack_events import RoutedEvent, TtlDedupCache + + +SECRET = 'endpoint-secret' + + +class FakeRequest: + def __init__(self, body, *, headers=None): + self._body = body + self.headers = headers or {} + + async def body(self): + return self._body + + +class FakePipe: + def __init__(self, *, fail_at=None): + self.fail_at = fail_at + self.opened = [] + self.text = [] + self.json = [] + self.closed = 0 + + def open(self, entry): + self.opened.append(entry) + if self.fail_at == 'open': + raise RuntimeError('open failed') + + def writeText(self, text): + self.text.append(text) + if self.fail_at == 'text': + raise RuntimeError('text failed') + + def writeJson(self, value): + self.json.append(value) + if self.fail_at == 'json': + raise RuntimeError('json failed') + + def close(self): + self.closed += 1 + if self.fail_at == 'close': + raise RuntimeError('close failed') + + +class FakeTarget: + def __init__(self, pipe): + self.pipe = pipe + self.acquired = 0 + self.released = [] + + def getPipe(self): + self.acquired += 1 + return self.pipe + + def putPipe(self, pipe): + self.released.append(pipe) + + +def _module(): + return importlib.import_module('nodes.slack.IEndpoint') + + +def _signature(body, timestamp): + base = b'v0:' + str(timestamp).encode() + b':' + body + return 'v0=' + hmac.new(SECRET.encode(), base, hashlib.sha256).hexdigest() + + +def _request(payload, *, timestamp=None, extra_headers=None): + body = payload if isinstance(payload, bytes) else json.dumps(payload, separators=(',', ':')).encode() + timestamp = int(time.time()) if timestamp is None else timestamp + headers = { + 'X-Slack-Request-Timestamp': str(timestamp), + 'X-Slack-Signature': _signature(body, timestamp), + } + headers.update(extra_headers or {}) + return FakeRequest(body, headers=headers) + + +def _endpoint(): + endpoint = _module().IEndpoint() + endpoint._signing_secret = SECRET + endpoint._queue = asyncio.Queue(maxsize=1) + endpoint._dedup = TtlDedupCache() + endpoint._accepting = True + return endpoint + + +def test_request_handler_declares_fastapi_request_parameter(): + parameter = inspect.signature(_module().IEndpoint._request_handler).parameters['request'] + + assert parameter.annotation is Request + + +def _event(event_id='Ev1', *, inner=None): + return { + 'type': 'event_callback', + 'event_id': event_id, + 'team_id': 'T1', + 'event': inner or {'type': 'app_mention', 'text': '<@A> hello'}, + } + + +@pytest.mark.asyncio +async def test_rejects_invalid_signature_before_json_or_queue(monkeypatch): + endpoint = _endpoint() + request = FakeRequest( + b'{not json', headers={'X-Slack-Request-Timestamp': str(int(time.time())), 'X-Slack-Signature': 'v0=bad'} + ) + monkeypatch.setattr(_module().json, 'loads', lambda _value: pytest.fail('decoded')) + + response = await endpoint._request_handler(request) + + assert response.status_code == 401 + assert endpoint._queue.empty() + + +@pytest.mark.asyncio +async def test_rejects_stale_timestamp_before_json_or_queue(monkeypatch): + endpoint = _endpoint() + request = _request(b'{not json', timestamp=int(time.time()) - 301) + monkeypatch.setattr(_module().json, 'loads', lambda _value: pytest.fail('decoded')) + + response = await endpoint._request_handler(request) + + assert response.status_code == 401 + assert endpoint._queue.empty() + + +@pytest.mark.asyncio +async def test_verified_challenge_returns_exact_text_without_queueing(): + endpoint = _endpoint() + + response = await endpoint._request_handler(_request({'type': 'url_verification', 'challenge': 'challenge-value'})) + + assert response.status_code == 200 + assert response.body == b'challenge-value' + assert endpoint._queue.empty() + + +@pytest.mark.asyncio +async def test_supported_callback_enqueues_once_and_duplicate_returns_ok(): + endpoint = _endpoint() + request = _request(_event()) + + assert (await endpoint._request_handler(request)).status_code == 200 + assert (await endpoint._request_handler(request)).status_code == 200 + + routed = endpoint._queue.get_nowait() + assert routed.event_type == 'app_mention' + assert endpoint._queue.empty() + + +@pytest.mark.asyncio +async def test_full_queue_returns_503_without_dedup_marking_so_retry_enqueues(): + endpoint = _endpoint() + endpoint._queue.put_nowait('full') + request = _request(_event()) + + assert (await endpoint._request_handler(request)).status_code == 503 + assert not endpoint._dedup.contains('Ev1') + endpoint._queue.get_nowait() + + assert (await endpoint._request_handler(request)).status_code == 200 + assert endpoint._dedup.contains('Ev1') + + +@pytest.mark.asyncio +async def test_malformed_authenticated_json_returns_400_without_enqueue(): + endpoint = _endpoint() + + response = await endpoint._request_handler(_request(b'{not json')) + + assert response.status_code == 400 + assert endpoint._queue.empty() + + +@pytest.mark.asyncio +@pytest.mark.parametrize('envelope', [[], 'not-an-object', 7, None]) +async def test_authenticated_non_object_json_returns_400_without_queue_access(envelope): + endpoint = _endpoint() + + class NoQueueAccess: + def __getattr__(self, _name): + pytest.fail('queue accessed') + + endpoint._queue = NoQueueAccess() + response = await endpoint._request_handler(_request(envelope)) + + assert response.status_code == 400 + + +@pytest.mark.asyncio +async def test_unsupported_callback_and_retry_headers_do_not_bypass_deduplication(): + endpoint = _endpoint() + unsupported = _event(inner={'type': 'reaction_removed'}) + assert (await endpoint._request_handler(_request(unsupported))).status_code == 200 + assert endpoint._queue.empty() + + request = _request(_event(), extra_headers={'X-Slack-Retry-Num': '1', 'X-Slack-Retry-Reason': 'http_timeout'}) + assert (await endpoint._request_handler(request)).status_code == 200 + assert (await endpoint._request_handler(request)).status_code == 200 + assert endpoint._queue.qsize() == 1 + + +def test_emit_text_sets_native_metadata_and_releases_pipe(monkeypatch): + endpoint = _endpoint() + pipe = FakePipe() + endpoint.target = FakeTarget(pipe) + completed = [] + monkeypatch.setattr(_module(), 'monitorCompleted', completed.append) + routed = RoutedEvent('app_mention', 'text', 'hé', _event()) + + endpoint._emit_event(routed) + + entry = pipe.opened[0] + assert entry.url == 'slack://T1/Ev1' + assert entry.metadata.toDict() == routed.envelope + assert pipe.text == ['hé'] + assert pipe.json == [] + assert endpoint.target.acquired == len(endpoint.target.released) == 1 + assert completed == [len('hé'.encode())] + + +@pytest.mark.parametrize( + ('lane', 'fail_at'), + [('text', 'open'), ('text', 'text'), ('json', 'json'), ('text', 'close')], +) +def test_emit_releases_pipe_for_every_open_write_and_close_failure(monkeypatch, lane, fail_at): + endpoint = _endpoint() + pipe = FakePipe(fail_at=fail_at) + endpoint.target = FakeTarget(pipe) + failed = [] + monkeypatch.setattr(_module(), 'monitorFailed', failed.append) + inner = {'type': 'reaction_added', 'reaction': 'eyes'} + routed = RoutedEvent( + 'reaction_added' if lane == 'json' else 'app_mention', + lane, + inner if lane == 'json' else 'text', + _event(inner=inner), + ) + + endpoint._emit_event(routed) + + assert endpoint.target.acquired == len(endpoint.target.released) == 1 + assert failed + + +@pytest.mark.asyncio +async def test_consumer_survives_pipe_acquisition_failure_and_processes_next_event(monkeypatch): + endpoint = _endpoint() + endpoint._queue = asyncio.Queue() + pipe = FakePipe() + + class FlakyTarget(FakeTarget): + def getPipe(self): + if self.acquired == 0: + self.acquired += 1 + raise RuntimeError('temporary pool failure') + return super().getPipe() + + endpoint.target = FlakyTarget(pipe) + failed = [] + monkeypatch.setattr(_module(), 'monitorFailed', failed.append) + monkeypatch.setattr(_module(), 'monitorCompleted', lambda _size: None) + endpoint._queue.put_nowait(RoutedEvent('app_mention', 'text', 'first', _event())) + endpoint._queue.put_nowait(RoutedEvent('app_mention', 'text', 'second', _event('Ev2'))) + endpoint._queue.put_nowait(None) + + await endpoint._consume_queue() + + assert failed == [0] + assert pipe.text == ['second'] + + +@pytest.mark.parametrize( + ('parameters', 'expected_capacity', 'expected_ttl'), + [ + ({'queueCapacity': 0, 'dedupTtlSeconds': 0}, 1, 300), + ({'queueCapacity': 10001, 'dedupTtlSeconds': 3601}, 10000, 3600), + ], +) +def test_runtime_clamps_queue_and_dedup_settings_to_schema_bounds(parameters, expected_capacity, expected_ttl): + endpoint = _endpoint() + endpoint.endpoint = types.SimpleNamespace(serviceConfig={'parameters': parameters}) + + assert endpoint._queue_capacity() == expected_capacity + assert endpoint._dedup_ttl() == expected_ttl + + +@pytest.mark.asyncio +async def test_startup_and_shutdown_manage_queue_worker_route_state_and_secret(monkeypatch): + endpoint = _endpoint() + endpoint.endpoint = types.SimpleNamespace( + serviceConfig={'parameters': {'queueCapacity': 3, 'dedupTtlSeconds': 700}} + ) + endpoint.target = FakeTarget(FakePipe()) + statuses = [] + monkeypatch.setattr(_module(), 'monitorStatus', statuses.append) + + await endpoint._startup() + + assert endpoint._queue.maxsize == 3 + assert endpoint._dedup.ttl_seconds == 700 + assert endpoint._consumer_task is not None + assert statuses[-1] == 'Slack Events ready - waiting for events' + + await endpoint._shutdown() + + assert endpoint._signing_secret == '' + assert endpoint._accepting is False + + +@pytest.mark.asyncio +async def test_shutdown_stops_intake_and_drains_queued_and_inflight_work(monkeypatch): + endpoint = _endpoint() + endpoint.endpoint = types.SimpleNamespace(serviceConfig={'parameters': {}}) + loop = asyncio.get_running_loop() + started = asyncio.Event() + release = threading.Event() + emitted = [] + + def emit(routed): + emitted.append(routed) + if len(emitted) == 1: + loop.call_soon_threadsafe(started.set) + release.wait() + + endpoint._emit_event = emit + await endpoint._startup() + endpoint._queue.put_nowait(RoutedEvent('app_mention', 'text', 'in-flight', _event())) + await started.wait() + endpoint._queue.put_nowait(RoutedEvent('app_mention', 'text', 'queued', _event('Ev2'))) + shutdown = asyncio.create_task(endpoint._shutdown()) + await asyncio.sleep(0) + + assert not endpoint._accepting + assert not shutdown.done() + assert (await endpoint._request_handler(_request(_event('Ev3')))).status_code == 503 + release.set() + await shutdown + + assert [item.payload for item in emitted] == ['in-flight', 'queued'] + + +@pytest.mark.asyncio +async def test_shutdown_cancels_consumer_after_bounded_drain_timeout(monkeypatch): + endpoint = _endpoint() + endpoint._queue = asyncio.Queue() + endpoint._accepting = True + monitor = [] + + async def owned_consumer(): + await asyncio.Event().wait() + + async def timeout(awaitable, *, timeout): + awaitable.close() + raise asyncio.TimeoutError + + owned = asyncio.create_task(owned_consumer()) + endpoint._consumer_task = owned + monkeypatch.setattr(_module().asyncio, 'wait_for', timeout) + monkeypatch.setattr(_module(), 'monitorOther', lambda *args: monitor.append(args)) + + await endpoint._shutdown() + + assert endpoint._consumer_task is None + assert owned.cancelled() + assert endpoint._signing_secret == '' + assert monitor == [('usr', '')] + + +def test_execution_lifecycle_binds_only_the_public_slack_events_post_route(monkeypatch): + module = _module() + created = [] + + class FakeServer: + def __init__(self, *, config, on_startup, on_shutdown): + self.config = config + self.on_startup = on_startup + self.on_shutdown = on_shutdown + self.app = types.SimpleNamespace(state=types.SimpleNamespace()) + self.routes = [] + self.ran = False + created.append(self) + + def add_route(self, path, handler, methods, *, public): + self.routes.append((path, handler, methods, public)) + + def run(self): + self.ran = True + + monkeypatch.setattr(module, 'WebServer', FakeServer) + endpoint = module.IEndpoint() + endpoint.endpoint = types.SimpleNamespace(target='target', openMode=module.OPEN_MODE.SOURCE) + + endpoint.scanObjects('', lambda _item: None) + + assert created[0].routes == [('/slack/events', endpoint._request_handler, ['POST'], True)] + assert created[0].ran + + config_endpoint = module.IEndpoint() + config_endpoint.endpoint = types.SimpleNamespace(target='target', openMode=module.OPEN_MODE.CONFIG) + config_endpoint.scanObjects('', lambda _item: None) + assert len(created) == 1 diff --git a/nodes/test/slack/test_queue_dedup.py b/nodes/test/slack/test_queue_dedup.py new file mode 100644 index 000000000..fb8f283eb --- /dev/null +++ b/nodes/test/slack/test_queue_dedup.py @@ -0,0 +1,40 @@ +# ============================================================================= +# MIT License +# ============================================================================= + +from nodes.slack.slack_events import TtlDedupCache + + +def test_dedup_cache_tracks_ids_until_the_ttl_expires(): + now = [100.0] + cache = TtlDedupCache(clock=lambda: now[0]) + + assert not cache.contains('Ev1') + cache.add('Ev1') + assert cache.contains('Ev1') + + now[0] += 600 + assert not cache.contains('Ev1') + + +def test_dedup_cache_uses_a_supplied_ttl_while_defaulting_to_600_seconds(): + now = [100.0] + configured = TtlDedupCache(ttl_seconds=7, clock=lambda: now[0]) + default = TtlDedupCache(clock=lambda: now[0]) + configured.add('Ev1') + default.add('Ev2') + + now[0] += 7 + + assert not configured.contains('Ev1') + assert default.contains('Ev2') + + +def test_dedup_cache_evicts_the_oldest_id_beyond_its_hard_cap(): + now = [100.0] + cache = TtlDedupCache(clock=lambda: now[0]) + for index in range(10_001): + cache.add(f'Ev{index}') + + assert not cache.contains('Ev0') + assert cache.contains('Ev10000') diff --git a/nodes/test/slack/test_routing.py b/nodes/test/slack/test_routing.py new file mode 100644 index 000000000..2e1e9425a --- /dev/null +++ b/nodes/test/slack/test_routing.py @@ -0,0 +1,100 @@ +# ============================================================================= +# MIT License +# Copyright (c) 2026 Aparavi Software AG +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the "Software"), to deal +# in the Software without restriction, including without limitation the rights +# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +# copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in +# all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +# SOFTWARE. +# ============================================================================= + +import pytest + +from nodes.slack.slack_events import RoutedEvent, classify_event + + +@pytest.mark.parametrize( + ('inner', 'expected_type'), + [ + ({'type': 'app_mention', 'text': '<@A1> hello'}, 'app_mention'), + ({'type': 'message', 'channel_type': 'channel', 'text': 'public'}, 'message.channels'), + ( + { + 'type': 'message', + 'channel_type': 'channel', + 'subtype': 'thread_broadcast', + 'user': 'U1', + 'text': 'public thread', + }, + 'message.channels', + ), + ({'type': 'message', 'channel_type': 'group', 'text': 'private'}, 'message.groups'), + ({'type': 'message', 'channel_type': 'im', 'text': 'direct'}, 'message.im'), + ], +) +def test_routes_supported_text_without_normalizing(inner, expected_type): + envelope = {'type': 'event_callback', 'event_id': 'Ev1', 'event': inner} + routed = classify_event(envelope) + assert routed == RoutedEvent(expected_type, 'text', inner['text'], envelope) + + +@pytest.mark.parametrize( + 'inner', + [ + {'type': 'app_mention', 'text': '<@A1> bot', 'bot_id': 'B1'}, + {'type': 'app_mention', 'text': '<@A1> app', 'app_id': 'A1'}, + {'type': 'app_mention', 'text': '<@A1> subtype', 'subtype': 'bot_message'}, + {'type': 'message', 'channel_type': 'channel', 'text': 'bot', 'bot_id': 'B1'}, + {'type': 'message', 'channel_type': 'channel', 'text': 'app', 'app_id': 'A1'}, + { + 'type': 'message', + 'channel_type': 'channel', + 'text': 'subtype', + 'subtype': 'bot_message', + }, + ], +) +def test_ignores_slack_marked_bot_or_app_text_events(inner): + envelope = {'type': 'event_callback', 'event_id': 'EvBot', 'event': inner} + assert classify_event(envelope) is None + + +def test_routes_reaction_as_exact_json(): + inner = {'type': 'reaction_added', 'reaction': 'eyes', 'item': {'type': 'message'}} + envelope = {'type': 'event_callback', 'event_id': 'Ev2', 'event': inner} + assert classify_event(envelope) == RoutedEvent('reaction_added', 'json', inner, envelope) + + +@pytest.mark.parametrize( + 'envelope', + [ + {}, + {'type': 'url_verification', 'challenge': 'x'}, + { + 'type': 'event_callback', + 'event_id': 'Ev3', + 'event': {'type': 'message', 'channel_type': 'mpim', 'text': 'x'}, + }, + { + 'type': 'event_callback', + 'event_id': 'Ev4', + 'event': {'type': 'message', 'channel_type': 'channel'}, + }, + {'type': 'event_callback', 'event_id': 'Ev5', 'event': {'type': 'reaction_removed'}}, + ], +) +def test_ignores_unapproved_or_incomplete_events(envelope): + assert classify_event(envelope) is None diff --git a/nodes/test/slack/test_signature.py b/nodes/test/slack/test_signature.py new file mode 100644 index 000000000..c37034e07 --- /dev/null +++ b/nodes/test/slack/test_signature.py @@ -0,0 +1,172 @@ +# ============================================================================= +# MIT License +# Copyright (c) 2026 Aparavi Software AG +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the "Software"), to deal +# in the Software without restriction, including without limitation the rights +# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +# copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in +# all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +# SOFTWARE. +# ============================================================================= + +import hashlib +import hmac +import importlib + +import pytest + +from nodes.slack.IGlobal import IGlobal +from nodes.slack.slack_events import resolve_signing_secret, verify_slack_signature + + +SECRET = 'test-signing-secret' +NOW = 1_700_000_000 +BODY = b'{"type":"event_callback","event_id":"Ev1"}' + + +def signature(body: bytes = BODY, timestamp: int = NOW) -> str: + base = b'v0:' + str(timestamp).encode() + b':' + body + return 'v0=' + hmac.new(SECRET.encode(), base, hashlib.sha256).hexdigest() + + +def test_accepts_exact_raw_body_signature(): + assert verify_slack_signature(SECRET, str(NOW), signature(), BODY, now=NOW) + + +def test_signature_base_preserves_the_exact_timestamp_header(): + timestamp = f'0{NOW}' + base = b'v0:' + timestamp.encode() + b':' + BODY + provided = 'v0=' + hmac.new(SECRET.encode(), base, hashlib.sha256).hexdigest() + + assert verify_slack_signature(SECRET, timestamp, provided, BODY, now=NOW) + + +@pytest.mark.parametrize( + ('secret', 'timestamp', 'provided', 'body'), + [ + ('', str(NOW), signature(), BODY), + (SECRET, '', signature(), BODY), + (SECRET, str(NOW), '', BODY), + (SECRET, str(NOW), 'v0=wrong', BODY), + (SECRET, str(NOW - 301), signature(timestamp=NOW - 301), BODY), + (SECRET, str(NOW + 301), signature(timestamp=NOW + 301), BODY), + (SECRET, str(NOW), signature(), BODY + b' '), + ], +) +def test_rejects_invalid_or_replayed_request(secret, timestamp, provided, body): + assert not verify_slack_signature(secret, timestamp, provided, body, now=NOW) + + +def test_secret_config_precedes_environment(): + assert resolve_signing_secret({'signingSecret': ' config '}, {'SLACK_SIGNING_SECRET': 'env'}) == 'config' + + +def test_secret_falls_back_to_environment(): + assert resolve_signing_secret({}, {'SLACK_SIGNING_SECRET': ' env '}) == 'env' + + +class _FakeEndpoint: + def __init__(self, open_mode): + self.endpoint = type( + 'EndpointState', + (), + { + 'openMode': open_mode, + 'serviceConfig': {'parameters': {}}, + }, + )() + self._signing_secret = 'unchanged' + + +def _global(open_mode): + instance = IGlobal() + instance.IEndpoint = _FakeEndpoint(open_mode) + instance.glb = type('GlobalState', (), {'logicalType': 'slack', 'connConfig': {}})() + return instance + + +def test_config_mode_does_not_load_or_transfer_runtime_secret(): + global_module = importlib.import_module('nodes.slack.IGlobal') + + instance = _global(global_module.OPEN_MODE.CONFIG) + + instance.beginGlobal() + + assert not hasattr(instance, 'signing_secret') + assert instance.IEndpoint._signing_secret == 'unchanged' + + +def test_execution_uses_config_secret_before_environment(monkeypatch): + global_module = importlib.import_module('nodes.slack.IGlobal') + + instance = _global(global_module.OPEN_MODE.SOURCE) + instance.IEndpoint.endpoint.serviceConfig = {'parameters': {'signingSecret': 'configured'}} + monkeypatch.setenv('SLACK_SIGNING_SECRET', 'environment') + + instance.beginGlobal() + + assert instance.signing_secret == 'configured' + assert instance.IEndpoint._signing_secret == 'configured' + + +def test_execution_reads_signing_secret_from_endpoint_service_config_parameters(): + global_module = importlib.import_module('nodes.slack.IGlobal') + + instance = _global(global_module.OPEN_MODE.SOURCE) + instance.IEndpoint.endpoint.serviceConfig = {'parameters': {'signingSecret': 'canvas-secret'}} + + instance.beginGlobal() + + assert instance.signing_secret == 'canvas-secret' + assert instance.IEndpoint._signing_secret == 'canvas-secret' + + +def test_validate_config_reads_signing_secret_from_endpoint_service_config_parameters(monkeypatch): + global_module = importlib.import_module('nodes.slack.IGlobal') + + instance = _global(global_module.OPEN_MODE.CONFIG) + instance.IEndpoint.endpoint.serviceConfig = {'parameters': {'signingSecret': 'canvas-secret'}} + warnings = [] + monkeypatch.setattr(global_module, 'warning', warnings.append) + + instance.validateConfig() + + assert warnings == [] + + +def test_missing_secret_warns_instead_of_raising(monkeypatch): + global_module = importlib.import_module('nodes.slack.IGlobal') + + instance = _global(global_module.OPEN_MODE.SOURCE) + warnings = [] + monkeypatch.delenv('SLACK_SIGNING_SECRET', raising=False) + monkeypatch.setattr(global_module, 'warning', warnings.append) + + instance.beginGlobal() + + assert warnings + + +def test_end_global_clears_global_and_endpoint_secrets(): + global_module = importlib.import_module('nodes.slack.IGlobal') + + instance = _global(global_module.OPEN_MODE.SOURCE) + instance.signing_secret = 'secret' + instance.IEndpoint._signing_secret = 'secret' + + instance.endGlobal() + + assert instance.signing_secret == '' + assert instance.IEndpoint._signing_secret == '' From 9d9b74a3ffb9442c7cd1e70523fbb2f95d4ebdc6 Mon Sep 17 00:00:00 2001 From: Krish Garg Date: Tue, 14 Jul 2026 21:26:13 -0700 Subject: [PATCH 2/2] fix(nodes): address Slack Events review feedback Bound public callback bodies, reject malformed hash keys, document node lifecycle symbols, and keep in-flight pipe delivery tracked through shutdown. Clean the example configuration and add regression coverage for request limits and shutdown races. Tests: 67 Slack tests and 304 node contract tests passed; ruff, formatting, and gitleaks passed. --- examples/slack-events.pipe | 5 +- nodes/src/nodes/slack/IEndpoint.py | 36 ++++++++- nodes/src/nodes/slack/IGlobal.py | 5 ++ nodes/src/nodes/slack/IInstance.py | 2 + nodes/src/nodes/slack/README.md | 8 +- nodes/src/nodes/slack/slack_events.py | 16 +++- nodes/test/slack/test_endpoint.py | 111 ++++++++++++++++++++++++++ nodes/test/slack/test_routing.py | 6 ++ 8 files changed, 180 insertions(+), 9 deletions(-) diff --git a/examples/slack-events.pipe b/examples/slack-events.pipe index 317790683..a2402fb29 100644 --- a/examples/slack-events.pipe +++ b/examples/slack-events.pipe @@ -7,8 +7,7 @@ "parameters": { "dedupTtlSeconds": 600, "queueCapacity": 1000, - "signingSecret": "", - "google": {} + "signingSecret": "" }, "type": "slack", "hideForm": true, @@ -75,4 +74,4 @@ 10 ], "docRevision": 8 -} \ No newline at end of file +} diff --git a/nodes/src/nodes/slack/IEndpoint.py b/nodes/src/nodes/slack/IEndpoint.py index 81a3487a3..0f5a5967e 100644 --- a/nodes/src/nodes/slack/IEndpoint.py +++ b/nodes/src/nodes/slack/IEndpoint.py @@ -44,17 +44,30 @@ from .slack_events import TtlDedupCache, classify_event, verify_slack_signature +MAX_SLACK_REQUEST_BODY_BYTES = 1024 * 1024 + class IEndpoint(IEndpointBase): + """Receive verified Slack Events API callbacks and route them to output lanes.""" + target: IEndpointBase | None = None _signing_secret: str = '' _queue: asyncio.Queue | None = None _dedup: TtlDedupCache | None = None _consumer_task: asyncio.Task | None = None + _delivery_task: asyncio.Task | None = None _accepting: bool = False async def _request_handler(self, request: Request): - raw_body = await request.body() + """Read, authenticate, and enqueue one Slack callback request.""" + body_chunks = [] + body_size = 0 + async for chunk in request.stream(): + body_size += len(chunk) + if body_size > MAX_SLACK_REQUEST_BODY_BYTES: + return PlainTextResponse('', status_code=413) + body_chunks.append(chunk) + raw_body = b''.join(body_chunks) timestamp = request.headers.get('X-Slack-Request-Timestamp', '') signature = request.headers.get('X-Slack-Signature', '') if not verify_slack_signature(self._signing_secret, timestamp, signature, raw_body): @@ -82,6 +95,7 @@ async def _request_handler(self, request: Request): return PlainTextResponse('') def _emit_event(self, routed) -> None: + """Write one routed event to its target pipe.""" envelope = routed.envelope event_id = envelope['event_id'] team_id = envelope.get('team_id', '') @@ -112,27 +126,35 @@ def _emit_event(self, routed) -> None: self.target.putPipe(pipe) async def _consume_queue(self) -> None: + """Deliver queued Slack events to the target pipe.""" while True: routed = await self._queue.get() try: if routed is None: return try: - await asyncio.to_thread(self._emit_event, routed) + self._delivery_task = asyncio.create_task(asyncio.to_thread(self._emit_event, routed)) + await asyncio.shield(self._delivery_task) except Exception: monitorFailed(0) + finally: + if self._delivery_task.done(): + self._delivery_task = None finally: self._queue.task_done() def _queue_capacity(self) -> int: + """Return the configured bounded event queue capacity.""" parameters = getattr(self.endpoint, 'serviceConfig', {}).get('parameters', {}) return min(10000, max(1, int(parameters.get('queueCapacity', 1000)))) def _dedup_ttl(self) -> int: + """Return the configured event deduplication TTL.""" parameters = getattr(self.endpoint, 'serviceConfig', {}).get('parameters', {}) return min(3600, max(300, int(parameters.get('dedupTtlSeconds', TtlDedupCache.TTL_SECONDS)))) async def _startup(self) -> None: + """Initialize the callback queue and start its delivery worker.""" self._accepting = True self._queue = asyncio.Queue(maxsize=self._queue_capacity()) self._dedup = TtlDedupCache(ttl_seconds=self._dedup_ttl()) @@ -140,6 +162,7 @@ async def _startup(self) -> None: monitorStatus('Slack Events ready - waiting for events') async def _shutdown(self) -> None: + """Stop intake, drain queued work, and finish active delivery safely.""" self._accepting = False if self._queue is not None and self._consumer_task is not None: try: @@ -152,11 +175,19 @@ async def _shutdown(self) -> None: await self._consumer_task except asyncio.CancelledError: pass + if self._delivery_task is not None: + try: + await asyncio.shield(self._delivery_task) + except Exception: + monitorFailed(0) + finally: + self._delivery_task = None self._consumer_task = None self._signing_secret = '' monitorOther('usr', '') def _run(self) -> None: + """Start the WebServer hosting the public Slack events endpoint.""" parser = argparse.ArgumentParser(add_help=False) parser.add_argument('--data_host', type=str, default='localhost') parser.add_argument('--data_port', type=int, default=5567) @@ -171,6 +202,7 @@ def _run(self) -> None: self._server.run() def scanObjects(self, _path: str, _scan_callback: Callable[[dict[str, Any]], None]): + """Start source execution when this endpoint is not in configuration mode.""" self.target = self.endpoint.target if self.endpoint.openMode != OPEN_MODE.CONFIG: self._run() diff --git a/nodes/src/nodes/slack/IGlobal.py b/nodes/src/nodes/slack/IGlobal.py index 483d3752d..f205e6f42 100644 --- a/nodes/src/nodes/slack/IGlobal.py +++ b/nodes/src/nodes/slack/IGlobal.py @@ -27,12 +27,16 @@ class IGlobal(IGlobalBase): + """Manage Slack signing-secret configuration for the endpoint lifecycle.""" + def validateConfig(self) -> None: + """Warn when no Slack signing secret is configured.""" config = self.IEndpoint.endpoint.serviceConfig.get('parameters', {}) if not resolve_signing_secret(config): warning('Slack signing secret is missing.') def beginGlobal(self) -> None: + """Resolve and assign the signing secret before source execution.""" if self.IEndpoint.endpoint.openMode == OPEN_MODE.CONFIG: return config = self.IEndpoint.endpoint.serviceConfig.get('parameters', {}) @@ -42,5 +46,6 @@ def beginGlobal(self) -> None: warning('Slack signing secret is missing.') def endGlobal(self) -> None: + """Clear the signing secret after source execution ends.""" self.signing_secret = '' self.IEndpoint._signing_secret = '' diff --git a/nodes/src/nodes/slack/IInstance.py b/nodes/src/nodes/slack/IInstance.py index 71acb6364..8429c1d46 100644 --- a/nodes/src/nodes/slack/IInstance.py +++ b/nodes/src/nodes/slack/IInstance.py @@ -25,4 +25,6 @@ class IInstance(IInstanceBase): + """Represent a Slack Events node instance.""" + pass diff --git a/nodes/src/nodes/slack/README.md b/nodes/src/nodes/slack/README.md index 29229b5d4..04068da81 100644 --- a/nodes/src/nodes/slack/README.md +++ b/nodes/src/nodes/slack/README.md @@ -67,7 +67,7 @@ waiting for pipeline processing. URL verification is authenticated before its challenge is returned and is never emitted. Healthy and duplicate deliveries receive an empty `200` response. Invalid signatures receive `401`, malformed authenticated JSON receives `400`, and unavailable or saturated intake receives -`503`; Slack can retry the latter. +`503`; request bodies larger than 1 MiB receive `413`. Slack can retry `503` responses. After successful enqueue, the node does not provide a durable retry for downstream pipe-emission failures: it records the failure and does not requeue @@ -76,8 +76,10 @@ the delivery. The queue and deduplication cache are process-local and not durable. On an orderly shutdown, intake stops and the node waits up to five seconds for queued and in-flight work, then allows up to five seconds for the consumer to exit -before cancelling it. An abrupt process termination can lose accepted queued -work, and a restart clears deduplication state. +before cancelling it. After that cancellation window, shutdown still waits for +an already-running downstream pipe write before pipe teardown. An abrupt +process termination can lose accepted queued work, and a restart clears +deduplication state. Only the five approved callback shapes are emitted. Socket Mode, OAuth app installation or provisioning, subscription discovery or management, and diff --git a/nodes/src/nodes/slack/slack_events.py b/nodes/src/nodes/slack/slack_events.py index c44be3749..e7b67b379 100644 --- a/nodes/src/nodes/slack/slack_events.py +++ b/nodes/src/nodes/slack/slack_events.py @@ -32,19 +32,24 @@ class TtlDedupCache: + """Maintain a bounded time-to-live cache of Slack event IDs.""" + TTL_SECONDS = 600 MAX_ENTRIES = 10_000 def __init__(self, *, ttl_seconds: int = TTL_SECONDS, clock: Any = time.time): + """Initialize the cache with its expiry period and clock.""" self.ttl_seconds = ttl_seconds self._clock = clock self._entries: OrderedDict[str, float] = OrderedDict() def contains(self, event_id: str) -> bool: + """Return whether an event ID is currently cached.""" self._prune() return event_id in self._entries def add(self, event_id: str) -> None: + """Record an event ID and enforce the cache size limit.""" self._prune() self._entries.pop(event_id, None) self._entries[event_id] = self._clock() @@ -52,6 +57,7 @@ def add(self, event_id: str) -> None: self._entries.popitem(last=False) def _prune(self) -> None: + """Remove expired event IDs from the cache.""" expired_before = self._clock() - self.ttl_seconds while self._entries: _event_id, added_at = next(iter(self._entries.items())) @@ -61,6 +67,7 @@ def _prune(self) -> None: def resolve_signing_secret(config: Mapping, environ: Mapping | None = None) -> str: + """Resolve the Slack signing secret from configuration or the environment.""" env = os.environ if environ is None else environ configured = config.get('signingSecret', '') if hasattr(config, 'get') else '' return str(configured or env.get('SLACK_SIGNING_SECRET', '') or '').strip() @@ -74,6 +81,7 @@ def verify_slack_signature( *, now: float | None = None, ) -> bool: + """Return whether a Slack request signature is valid and timely.""" if not signing_secret or not timestamp or not provided_signature: return False if not timestamp.isascii() or not timestamp.isdecimal(): @@ -89,6 +97,8 @@ def verify_slack_signature( @dataclass(frozen=True) class RoutedEvent: + """Represent a classified Slack event routed to an output lane.""" + event_type: str lane: str payload: Any @@ -103,12 +113,15 @@ class RoutedEvent: def classify_event(envelope: dict) -> RoutedEvent | None: + """Classify a Slack event envelope or return none when unsupported.""" if envelope.get('type') != 'event_callback' or not isinstance(envelope.get('event_id'), str): return None inner = envelope.get('event') if not isinstance(inner, dict): return None inner_type = inner.get('type') + if not isinstance(inner_type, str): + return None is_bot_or_app_text = ( (isinstance(inner.get('bot_id'), str) and bool(inner['bot_id'])) or (isinstance(inner.get('app_id'), str) and bool(inner['app_id'])) @@ -119,7 +132,8 @@ def classify_event(envelope: dict) -> RoutedEvent | None: if inner_type == 'app_mention' and isinstance(inner.get('text'), str) and inner['text']: return RoutedEvent('app_mention', 'text', inner['text'], envelope) if inner_type == 'message': - logical = _MESSAGE_TYPES.get(inner.get('channel_type')) + channel_type = inner.get('channel_type') + logical = _MESSAGE_TYPES.get(channel_type) if isinstance(channel_type, str) else None if logical and isinstance(inner.get('text'), str) and inner['text']: return RoutedEvent(logical, 'text', inner['text'], envelope) if inner_type == 'reaction_added': diff --git a/nodes/test/slack/test_endpoint.py b/nodes/test/slack/test_endpoint.py index c98b680d5..539c6ab9c 100644 --- a/nodes/test/slack/test_endpoint.py +++ b/nodes/test/slack/test_endpoint.py @@ -29,6 +29,9 @@ def __init__(self, body, *, headers=None): async def body(self): return self._body + async def stream(self): + yield self._body + class FakePipe: def __init__(self, *, fail_at=None): @@ -131,6 +134,50 @@ async def test_rejects_invalid_signature_before_json_or_queue(monkeypatch): assert endpoint._queue.empty() +@pytest.mark.asyncio +async def test_request_handler_preserves_exact_under_limit_raw_body_for_signature(monkeypatch): + endpoint = _endpoint() + module = _module() + body = b'{"type":"url_verification","challenge":"\xc3\xa9"}' + timestamp = int(time.time()) + request = FakeRequest( + body, + headers={ + 'X-Slack-Request-Timestamp': str(timestamp), + 'X-Slack-Signature': _signature(body, timestamp), + }, + ) + verified = [] + monkeypatch.setattr( + module, + 'verify_slack_signature', + lambda secret, timestamp, signature, raw: verified.append(raw) or True, + ) + + response = await endpoint._request_handler(request) + + assert response.status_code == 200 + assert verified == [body] + + +@pytest.mark.asyncio +async def test_request_handler_rejects_body_over_hard_limit_without_signature_check(monkeypatch): + endpoint = _endpoint() + module = _module() + chunks = [b'a' * module.MAX_SLACK_REQUEST_BODY_BYTES, b'b'] + + class ChunkedRequest(FakeRequest): + async def stream(self): + for chunk in chunks: + yield chunk + + monkeypatch.setattr(module, 'verify_slack_signature', lambda *args: pytest.fail('signature checked')) + + response = await endpoint._request_handler(ChunkedRequest(b'', headers={})) + + assert response.status_code == 413 + + @pytest.mark.asyncio async def test_rejects_stale_timestamp_before_json_or_queue(monkeypatch): endpoint = _endpoint() @@ -386,6 +433,70 @@ async def timeout(awaitable, *, timeout): assert monitor == [('usr', '')] +@pytest.mark.asyncio +async def test_shutdown_records_delivery_failure_after_cancelling_consumer(monkeypatch): + endpoint = _endpoint() + endpoint._queue = asyncio.Queue() + endpoint._accepting = True + loop = asyncio.get_running_loop() + started = asyncio.Event() + release = threading.Event() + failed = [] + + def emit(_routed): + loop.call_soon_threadsafe(started.set) + release.wait() + raise RuntimeError('delivery failed') + + async def timeout(awaitable, *, timeout): + awaitable.close() + raise asyncio.TimeoutError + + endpoint._emit_event = emit + monkeypatch.setattr(_module().asyncio, 'wait_for', timeout) + monkeypatch.setattr(_module(), 'monitorFailed', failed.append) + endpoint._consumer_task = asyncio.create_task(endpoint._consume_queue()) + endpoint._queue.put_nowait(RoutedEvent('app_mention', 'text', 'blocked', _event())) + await started.wait() + + shutdown = asyncio.create_task(endpoint._shutdown()) + await asyncio.sleep(0) + release.set() + await shutdown + + assert failed == [0] + assert endpoint._delivery_task is None + + +@pytest.mark.asyncio +async def test_shutdown_waits_for_delivery_thread_after_drain_timeout(monkeypatch): + endpoint = _endpoint() + endpoint.endpoint = types.SimpleNamespace(serviceConfig={'parameters': {}}) + loop = asyncio.get_running_loop() + started = asyncio.Event() + release = threading.Event() + + def emit(_routed): + loop.call_soon_threadsafe(started.set) + release.wait() + + endpoint._emit_event = emit + await endpoint._startup() + endpoint._queue.put_nowait(RoutedEvent('app_mention', 'text', 'blocked', _event())) + await started.wait() + + async def timeout(awaitable, *, timeout): + awaitable.close() + raise asyncio.TimeoutError + + monkeypatch.setattr(_module().asyncio, 'wait_for', timeout) + shutdown = asyncio.create_task(endpoint._shutdown()) + await asyncio.sleep(0) + assert not shutdown.done() + release.set() + await shutdown + + def test_execution_lifecycle_binds_only_the_public_slack_events_post_route(monkeypatch): module = _module() created = [] diff --git a/nodes/test/slack/test_routing.py b/nodes/test/slack/test_routing.py index 2e1e9425a..81e6b84dd 100644 --- a/nodes/test/slack/test_routing.py +++ b/nodes/test/slack/test_routing.py @@ -94,6 +94,12 @@ def test_routes_reaction_as_exact_json(): 'event': {'type': 'message', 'channel_type': 'channel'}, }, {'type': 'event_callback', 'event_id': 'Ev5', 'event': {'type': 'reaction_removed'}}, + {'type': 'event_callback', 'event_id': 'Ev6', 'event': {'type': [], 'text': 'x'}}, + { + 'type': 'event_callback', + 'event_id': 'Ev7', + 'event': {'type': 'message', 'channel_type': {}, 'text': 'x'}, + }, ], ) def test_ignores_unapproved_or_incomplete_events(envelope):