mirror of
https://github.com/Stirling-Tools/Stirling-PDF.git
synced 2026-09-03 05:10:16 +03:00
Anthony Stirling
3bf0019d7c
# Description of Changes Create custom webhooks as a source, allows file pushes toa custom made endpoint with custom auth ID - Adds webhook as a policy source: external systems push documents to a receiver endpoint, which stages the files locally and triggers the policy run - Requests are authenticated with HMAC signatures; receiver hardened with bounded body reads and server-minted IDs - Uses the same team-scoped IntegrationConfig connection model as the S3 source, with matching portal UI (source type, icon, wizard) - Includes a policies-gated Cucumber feature covering the receiver end-to-end --- ## Checklist ### General - [ ] I have read the [Contribution Guidelines](https://github.com/Stirling-Tools/Stirling-PDF/blob/main/CONTRIBUTING.md) - [ ] I have read the [Stirling-PDF Developer Guide](https://github.com/Stirling-Tools/Stirling-PDF/blob/main/DeveloperGuide.md) (if applicable) - [ ] I have read the [How to add new languages to Stirling-PDF](https://github.com/Stirling-Tools/Stirling-PDF/blob/main/devGuide/HowToAddNewLanguage.md) (if applicable) - [ ] I have performed a self-review of my own code - [ ] My changes generate no new warnings ### Documentation - [ ] I have updated relevant docs on [Stirling-PDF's doc repo](https://github.com/Stirling-Tools/Stirling-Tools.github.io/blob/main/docs/) (if functionality has heavily changed) - [ ] I have read the section [Add New Translation Tags](https://github.com/Stirling-Tools/Stirling-PDF/blob/main/devGuide/HowToAddNewLanguage.md#add-new-translation-tags) (for new translation tags only) ### Translations (if applicable) - [ ] I ran [`scripts/counter_translation.py`](https://github.com/Stirling-Tools/Stirling-PDF/blob/main/docs/counter_translation.md) ### UI Changes (if applicable) - [ ] Screenshots or videos demonstrating the UI changes are attached (e.g., as comments or direct attachments in the PR) ### Testing (if applicable) - [ ] I have run `task check` to verify linters, typechecks, and tests pass - [ ] I have tested my changes locally. Refer to the [Testing Guide](https://github.com/Stirling-Tools/Stirling-PDF/blob/main/DeveloperGuide.md#7-testing) for more details.
85 lines
3.1 KiB
Python
85 lines
3.1 KiB
Python
import hashlib
|
|
import hmac
|
|
|
|
import requests
|
|
from behave import given, when, then
|
|
|
|
BASE_URL = "http://localhost:8080"
|
|
API_HEADERS = {"X-API-KEY": "123456789"}
|
|
|
|
|
|
def _sign(secret, body):
|
|
digest = hmac.new(secret.encode(), body.encode(), hashlib.sha256).hexdigest()
|
|
return "sha256=" + digest
|
|
|
|
|
|
@given('I create a webhook source named "{name}"')
|
|
def step_create_webhook_source(context, name):
|
|
resp = requests.post(
|
|
f"{BASE_URL}/api/v1/sources",
|
|
headers={**API_HEADERS, "Content-Type": "application/json"},
|
|
json={"name": name, "type": "webhook", "options": {}, "enabled": True},
|
|
timeout=15,
|
|
)
|
|
assert resp.status_code == 200, f"create source failed: {resp.status_code} {resp.text}"
|
|
context.webhook_create_response = resp
|
|
body = resp.json()
|
|
context.webhook_source_id = body["id"]
|
|
context.webhook_id = body["options"]["webhookId"]
|
|
context.webhook_secret = body["options"]["signingSecret"]
|
|
|
|
|
|
@when('I deliver "{payload}" to the webhook with a valid signature')
|
|
def step_deliver_signed(context, payload):
|
|
signature = _sign(context.webhook_secret, payload)
|
|
context.webhook_response = requests.post(
|
|
f"{BASE_URL}/api/v1/webhooks/{context.webhook_id}",
|
|
headers={"Content-Type": "application/pdf", "X-Stirling-Signature": signature},
|
|
data=payload.encode(),
|
|
timeout=15,
|
|
)
|
|
|
|
|
|
@when('I deliver "{payload}" to the webhook with signature "{signature}"')
|
|
def step_deliver_with_signature(context, payload, signature):
|
|
context.webhook_response = requests.post(
|
|
f"{BASE_URL}/api/v1/webhooks/{context.webhook_id}",
|
|
headers={"Content-Type": "application/pdf", "X-Stirling-Signature": signature},
|
|
data=payload.encode(),
|
|
timeout=15,
|
|
)
|
|
|
|
|
|
@when('I deliver "{payload}" to webhook id "{webhook_id}"')
|
|
def step_deliver_to_id(context, payload, webhook_id):
|
|
context.webhook_response = requests.post(
|
|
f"{BASE_URL}/api/v1/webhooks/{webhook_id}",
|
|
headers={"Content-Type": "application/pdf", "X-Stirling-Signature": "sha256=00"},
|
|
data=payload.encode(),
|
|
timeout=15,
|
|
)
|
|
|
|
|
|
@then("the webhook response status should be {status:d}")
|
|
def step_check_status(context, status):
|
|
actual = context.webhook_response.status_code
|
|
assert actual == status, f"expected {status}, got {actual}: {context.webhook_response.text}"
|
|
|
|
|
|
@then("the webhook create response includes a signing secret")
|
|
def step_secret_present(context):
|
|
secret = context.webhook_create_response.json()["options"].get("signingSecret", "")
|
|
assert secret and secret != "********", f"expected a revealed secret, got '{secret}'"
|
|
|
|
|
|
@then("reading the webhook source back masks the signing secret")
|
|
def step_secret_masked(context):
|
|
resp = requests.get(
|
|
f"{BASE_URL}/api/v1/sources/{context.webhook_source_id}",
|
|
headers=API_HEADERS,
|
|
timeout=15,
|
|
)
|
|
assert resp.status_code == 200, f"get source failed: {resp.status_code} {resp.text}"
|
|
secret = resp.json()["options"].get("signingSecret", "")
|
|
assert secret != context.webhook_secret, "secret was returned in clear text on read"
|