diff --git a/examples/slack-events.pipe b/examples/slack-events.pipe
new file mode 100644
index 000000000..a2402fb29
--- /dev/null
+++ b/examples/slack-events.pipe
@@ -0,0 +1,77 @@
+{
+ "components": [
+ {
+ "id": "slack_1",
+ "provider": "slack",
+ "config": {
+ "parameters": {
+ "dedupTtlSeconds": 600,
+ "queueCapacity": 1000,
+ "signingSecret": ""
+ },
+ "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
+}
diff --git a/nodes/src/nodes/slack/IEndpoint.py b/nodes/src/nodes/slack/IEndpoint.py
new file mode 100644
index 000000000..0f5a5967e
--- /dev/null
+++ b/nodes/src/nodes/slack/IEndpoint.py
@@ -0,0 +1,208 @@
+# =============================================================================
+# 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
+
+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):
+ """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):
+ 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:
+ """Write one routed event to its target pipe."""
+ 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:
+ """Deliver queued Slack events to the target pipe."""
+ while True:
+ routed = await self._queue.get()
+ try:
+ if routed is None:
+ return
+ try:
+ 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())
+ self._consumer_task = asyncio.create_task(self._consume_queue())
+ 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:
+ 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
+ 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)
+ 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]):
+ """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
new file mode 100644
index 000000000..f205e6f42
--- /dev/null
+++ b/nodes/src/nodes/slack/IGlobal.py
@@ -0,0 +1,51 @@
+# =============================================================================
+# 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):
+ """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', {})
+ 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:
+ """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
new file mode 100644
index 000000000..8429c1d46
--- /dev/null
+++ b/nodes/src/nodes/slack/IInstance.py
@@ -0,0 +1,30 @@
+# =============================================================================
+# 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):
+ """Represent a Slack Events node instance."""
+
+ pass
diff --git a/nodes/src/nodes/slack/README.md b/nodes/src/nodes/slack/README.md
new file mode 100644
index 000000000..04068da81
--- /dev/null
+++ b/nodes/src/nodes/slack/README.md
@@ -0,0 +1,138 @@
+# 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`; 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
+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. 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
+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..e7b67b379
--- /dev/null
+++ b/nodes/src/nodes/slack/slack_events.py
@@ -0,0 +1,141 @@
+# =============================================================================
+# 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:
+ """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()
+ while len(self._entries) > self.MAX_ENTRIES:
+ 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()))
+ if added_at > expired_before:
+ break
+ self._entries.popitem(last=False)
+
+
+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()
+
+
+def verify_slack_signature(
+ signing_secret: str,
+ timestamp: str,
+ provided_signature: str,
+ raw_body: bytes,
+ *,
+ 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():
+ 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:
+ """Represent a classified Slack event routed to an output lane."""
+
+ 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:
+ """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']))
+ 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':
+ 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':
+ 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..539c6ab9c
--- /dev/null
+++ b/nodes/test/slack/test_endpoint.py
@@ -0,0 +1,532 @@
+# =============================================================================
+# 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
+
+ async def stream(self):
+ yield 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_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()
+ 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', '')]
+
+
+@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 = []
+
+ 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..81e6b84dd
--- /dev/null
+++ b/nodes/test/slack/test_routing.py
@@ -0,0 +1,106 @@
+# =============================================================================
+# 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'}},
+ {'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):
+ 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 == ''