Compare commits
8 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 8e744c076d | |||
| cc494c6937 | |||
| e2fa96318f | |||
| 416c3c4758 | |||
| 1c7ef28c5d | |||
| 1a0218799b | |||
| 048f28a671 | |||
| b818f07c80 |
@@ -200,3 +200,5 @@ cython_debug/
|
|||||||
# Build metadata files - generated at build time
|
# Build metadata files - generated at build time
|
||||||
GIT_SHA
|
GIT_SHA
|
||||||
RUNTIME_INFO
|
RUNTIME_INFO
|
||||||
|
node_modules
|
||||||
|
frontend/node_modules
|
||||||
|
|||||||
+1
-1
@@ -1 +1 @@
|
|||||||
2026-05-17T12:40:20Z
|
2026-05-17T14:20:08Z
|
||||||
|
|||||||
@@ -10,6 +10,37 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
|
|||||||
|
|
||||||
<!-- version list -->
|
<!-- version list -->
|
||||||
|
|
||||||
|
## v0.172.12 (2026-05-17)
|
||||||
|
|
||||||
|
### Bug Fixes
|
||||||
|
|
||||||
|
- Validate webhook targets before delivery
|
||||||
|
([#846](https://github.com/christianlouis/DocuElevate/pull/846),
|
||||||
|
[`e2fa963`](https://github.com/christianlouis/DocuElevate/commit/e2fa96318f5bd45607baa0fe08a0bf14e1ca83d4))
|
||||||
|
|
||||||
|
### Testing
|
||||||
|
|
||||||
|
- Cover webhook SSRF validation ([#846](https://github.com/christianlouis/DocuElevate/pull/846),
|
||||||
|
[`e2fa963`](https://github.com/christianlouis/DocuElevate/commit/e2fa96318f5bd45607baa0fe08a0bf14e1ca83d4))
|
||||||
|
|
||||||
|
|
||||||
|
## v0.172.11 (2026-05-17)
|
||||||
|
|
||||||
|
### Bug Fixes
|
||||||
|
|
||||||
|
- Escape search result template values
|
||||||
|
([#853](https://github.com/christianlouis/DocuElevate/pull/853),
|
||||||
|
[`1a02187`](https://github.com/christianlouis/DocuElevate/commit/1a0218799b9a1eb4154e2f4fbb2572cb3922106a))
|
||||||
|
|
||||||
|
### Documentation
|
||||||
|
|
||||||
|
- **changelog**: Update changelog [skip ci]
|
||||||
|
([`048f28a`](https://github.com/christianlouis/DocuElevate/commit/048f28a6717fa7f5cf4b235f9142e625e80e5d59))
|
||||||
|
|
||||||
|
|
||||||
|
## Unreleased
|
||||||
|
|
||||||
|
|
||||||
## v0.172.10 (2026-05-17)
|
## v0.172.10 (2026-05-17)
|
||||||
|
|
||||||
### Bug Fixes
|
### Bug Fixes
|
||||||
|
|||||||
+6
-6
@@ -1,10 +1,10 @@
|
|||||||
DocuElevate Build Information
|
DocuElevate Build Information
|
||||||
==============================
|
==============================
|
||||||
Version: 0.172.10
|
Version: 0.172.12
|
||||||
Build Date: 2026-05-17T12:40:20Z
|
Build Date: 2026-05-17T14:20:08Z
|
||||||
Git Commit: 62d4ca6367a6c8a2e7909305fec56c5b2c24e312
|
Git Commit: e2fa96318f5bd45607baa0fe08a0bf14e1ca83d4
|
||||||
Git Short SHA: 62d4ca6
|
Git Short SHA: e2fa963
|
||||||
Git Branch: main
|
Git Branch: main
|
||||||
Commit Date: 2026-05-17T14:39:59+02:00
|
Commit Date: 2026-05-17T16:19:41+02:00
|
||||||
Build Timestamp: 2026-05-17T12:40:20Z
|
Build Timestamp: 2026-05-17T14:20:08Z
|
||||||
==============================
|
==============================
|
||||||
|
|||||||
@@ -12,11 +12,13 @@ import smtplib
|
|||||||
from email.mime.multipart import MIMEMultipart
|
from email.mime.multipart import MIMEMultipart
|
||||||
from email.mime.text import MIMEText
|
from email.mime.text import MIMEText
|
||||||
from typing import Any
|
from typing import Any
|
||||||
|
from urllib.parse import urlparse
|
||||||
|
|
||||||
import httpx
|
import httpx
|
||||||
|
|
||||||
from app.database import SessionLocal
|
from app.database import SessionLocal
|
||||||
from app.models import InAppNotification, UserNotificationPreference, UserNotificationTarget
|
from app.models import InAppNotification, UserNotificationPreference, UserNotificationTarget
|
||||||
|
from app.utils.network import is_private_ip
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
@@ -28,6 +30,11 @@ USER_EVENT_LABELS: dict[str, str] = {
|
|||||||
EVENT_DOCUMENT_PROCESSED: "Document Processed",
|
EVENT_DOCUMENT_PROCESSED: "Document Processed",
|
||||||
EVENT_DOCUMENT_FAILED: "Document Processing Failed",
|
EVENT_DOCUMENT_FAILED: "Document Processing Failed",
|
||||||
}
|
}
|
||||||
|
METADATA_ENDPOINTS = {
|
||||||
|
"169.254.169.254",
|
||||||
|
"169.254.169.253",
|
||||||
|
"metadata.google.internal",
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
def create_in_app_notification(
|
def create_in_app_notification(
|
||||||
@@ -128,6 +135,20 @@ def _send_webhook_notification(target_config: dict[str, Any], event_type: str, t
|
|||||||
logger.warning("Webhook notification target missing url")
|
logger.warning("Webhook notification target missing url")
|
||||||
return False
|
return False
|
||||||
|
|
||||||
|
parsed_url = urlparse(url)
|
||||||
|
if parsed_url.scheme not in {"http", "https"}:
|
||||||
|
logger.warning("Webhook notification to %s blocked: invalid scheme %s", url, parsed_url.scheme)
|
||||||
|
return False
|
||||||
|
|
||||||
|
hostname = parsed_url.hostname
|
||||||
|
if not hostname:
|
||||||
|
logger.warning("Webhook notification to %s blocked: missing hostname", url)
|
||||||
|
return False
|
||||||
|
|
||||||
|
if hostname in METADATA_ENDPOINTS or is_private_ip(hostname):
|
||||||
|
logger.warning("Webhook notification to %s blocked: private or metadata endpoint", url)
|
||||||
|
return False
|
||||||
|
|
||||||
payload = {
|
payload = {
|
||||||
"event": event_type,
|
"event": event_type,
|
||||||
"title": title,
|
"title": title,
|
||||||
|
|||||||
@@ -18,11 +18,13 @@ import json
|
|||||||
import logging
|
import logging
|
||||||
import time
|
import time
|
||||||
from typing import Any
|
from typing import Any
|
||||||
|
from urllib.parse import urlparse
|
||||||
|
|
||||||
import requests
|
import requests
|
||||||
|
|
||||||
from app.database import SessionLocal
|
from app.database import SessionLocal
|
||||||
from app.models import WebhookConfig
|
from app.models import WebhookConfig
|
||||||
|
from app.utils.network import is_private_ip
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
@@ -40,6 +42,11 @@ VALID_EVENTS: frozenset[str] = frozenset(
|
|||||||
|
|
||||||
#: Timeout (seconds) for outgoing webhook HTTP requests.
|
#: Timeout (seconds) for outgoing webhook HTTP requests.
|
||||||
WEBHOOK_TIMEOUT = 10
|
WEBHOOK_TIMEOUT = 10
|
||||||
|
METADATA_ENDPOINTS = {
|
||||||
|
"169.254.169.254",
|
||||||
|
"169.254.169.253",
|
||||||
|
"metadata.google.internal",
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
def compute_signature(payload_bytes: bytes, secret: str) -> str:
|
def compute_signature(payload_bytes: bytes, secret: str) -> str:
|
||||||
@@ -67,6 +74,20 @@ def deliver_webhook(url: str, payload: dict[str, Any], secret: str | None = None
|
|||||||
Returns:
|
Returns:
|
||||||
``True`` when the remote server responds with a 2xx status.
|
``True`` when the remote server responds with a 2xx status.
|
||||||
"""
|
"""
|
||||||
|
parsed_url = urlparse(url)
|
||||||
|
if parsed_url.scheme not in {"http", "https"}:
|
||||||
|
logger.warning("Webhook to %s blocked: invalid scheme %s", url, parsed_url.scheme)
|
||||||
|
return False
|
||||||
|
|
||||||
|
hostname = parsed_url.hostname
|
||||||
|
if not hostname:
|
||||||
|
logger.warning("Webhook to %s blocked: missing hostname", url)
|
||||||
|
return False
|
||||||
|
|
||||||
|
if hostname in METADATA_ENDPOINTS or is_private_ip(hostname):
|
||||||
|
logger.warning("Webhook to %s blocked: private or metadata endpoint", url)
|
||||||
|
return False
|
||||||
|
|
||||||
body = json.dumps(payload, default=str, sort_keys=True)
|
body = json.dumps(payload, default=str, sort_keys=True)
|
||||||
body_bytes = body.encode("utf-8")
|
body_bytes = body.encode("utf-8")
|
||||||
|
|
||||||
|
|||||||
@@ -1538,6 +1538,28 @@
|
|||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
|
function escapeHtml(str) {
|
||||||
|
if (!str) return '';
|
||||||
|
return String(str)
|
||||||
|
.replace(/&/g, '&')
|
||||||
|
.replace(/</g, '<')
|
||||||
|
.replace(/>/g, '>')
|
||||||
|
.replace(/"/g, '"')
|
||||||
|
.replace(/'/g, ''');
|
||||||
|
}
|
||||||
|
|
||||||
|
function sanitizeHighlight(html) {
|
||||||
|
if (!html) return '';
|
||||||
|
let safe = String(html)
|
||||||
|
.replace(/<mark>/gi, '\x00MARK_OPEN\x00')
|
||||||
|
.replace(/<\/mark>/gi, '\x00MARK_CLOSE\x00');
|
||||||
|
safe = escapeHtml(safe);
|
||||||
|
safe = safe
|
||||||
|
.replace(/\x00MARK_OPEN\x00/g, '<mark>')
|
||||||
|
.replace(/\x00MARK_CLOSE\x00/g, '</mark>');
|
||||||
|
return safe;
|
||||||
|
}
|
||||||
|
|
||||||
function renderSearchResults(data, q) {
|
function renderSearchResults(data, q) {
|
||||||
const panel = document.getElementById('search-results-panel');
|
const panel = document.getElementById('search-results-panel');
|
||||||
const list = document.getElementById('search-results-list');
|
const list = document.getElementById('search-results-list');
|
||||||
@@ -1555,25 +1577,31 @@
|
|||||||
|
|
||||||
list.innerHTML = results.map(hit => {
|
list.innerHTML = results.map(hit => {
|
||||||
const fmt = hit._formatted || {};
|
const fmt = hit._formatted || {};
|
||||||
const title = fmt.document_title || hit.document_title || hit.original_filename || __i18n.untitled;
|
const titleRaw = fmt.document_title || hit.document_title || hit.original_filename || __i18n.untitled;
|
||||||
const filename = fmt.original_filename || hit.original_filename || '';
|
const filenameRaw = fmt.original_filename || hit.original_filename || '';
|
||||||
const snippet = fmt.ocr_text || '';
|
const snippetRaw = fmt.ocr_text || '';
|
||||||
const tags = Array.isArray(hit.tags) ? hit.tags.join(', ') : (hit.tags || '');
|
const tagsRaw = Array.isArray(hit.tags) ? hit.tags.join(', ') : (hit.tags || '');
|
||||||
const docType = hit.document_type || '';
|
const docTypeRaw = hit.document_type || '';
|
||||||
|
|
||||||
|
const safeTitle = fmt.document_title ? sanitizeHighlight(titleRaw) : escapeHtml(titleRaw);
|
||||||
|
const safeFilename = escapeHtml(filenameRaw);
|
||||||
|
const safeSnippet = sanitizeHighlight(snippetRaw);
|
||||||
|
const safeTags = escapeHtml(tagsRaw);
|
||||||
|
const safeDocType = escapeHtml(docTypeRaw);
|
||||||
|
|
||||||
return `<div style="padding: 0.75rem 1rem; border-bottom: 1px solid #f3f4f6; display: flex; gap: 0.75rem; align-items: flex-start;">
|
return `<div style="padding: 0.75rem 1rem; border-bottom: 1px solid #f3f4f6; display: flex; gap: 0.75rem; align-items: flex-start;">
|
||||||
<div style="flex-shrink: 0; color: #3b82f6; font-size: 1.25rem; padding-top: 0.1rem;">
|
<div style="flex-shrink: 0; color: #3b82f6; font-size: 1.25rem; padding-top: 0.1rem;">
|
||||||
<i class="fas fa-file-pdf"></i>
|
<i class="fas fa-file-pdf"></i>
|
||||||
</div>
|
</div>
|
||||||
<div style="flex: 1; min-width: 0;">
|
<div style="flex: 1; min-width: 0;">
|
||||||
<div style="font-weight: 600; font-size: 0.9rem; color: #111827;">${title}</div>
|
<div style="font-weight: 600; font-size: 0.9rem; color: #111827;">${safeTitle}</div>
|
||||||
${filename ? `<div style="font-size: 0.8rem; color: #6b7280; margin-top: 0.15rem;">${filename}</div>` : ''}
|
${safeFilename ? `<div style="font-size: 0.8rem; color: #6b7280; margin-top: 0.15rem;">${safeFilename}</div>` : ''}
|
||||||
${docType ? `<span style="display: inline-block; margin-top: 0.25rem; padding: 0.1rem 0.5rem; background: #eff6ff; color: #1d4ed8; border-radius: 9999px; font-size: 0.75rem;">${docType}</span>` : ''}
|
${safeDocType ? `<span style="display: inline-block; margin-top: 0.25rem; padding: 0.1rem 0.5rem; background: #eff6ff; color: #1d4ed8; border-radius: 9999px; font-size: 0.75rem;">${safeDocType}</span>` : ''}
|
||||||
${tags ? `<span style="display: inline-block; margin-top: 0.25rem; margin-left: 0.25rem; padding: 0.1rem 0.5rem; background: #f0fdf4; color: #15803d; border-radius: 9999px; font-size: 0.75rem;">${tags}</span>` : ''}
|
${safeTags ? `<span style="display: inline-block; margin-top: 0.25rem; margin-left: 0.25rem; padding: 0.1rem 0.5rem; background: #f0fdf4; color: #15803d; border-radius: 9999px; font-size: 0.75rem;">${safeTags}</span>` : ''}
|
||||||
${snippet ? `<div style="margin-top: 0.4rem; font-size: 0.8rem; color: #374151; white-space: pre-wrap; word-break: break-word;">…${snippet}…</div>` : ''}
|
${safeSnippet ? `<div style="margin-top: 0.4rem; font-size: 0.8rem; color: #374151; white-space: pre-wrap; word-break: break-word;">…${safeSnippet}…</div>` : ''}
|
||||||
</div>
|
</div>
|
||||||
<div style="flex-shrink: 0;">
|
<div style="flex-shrink: 0;">
|
||||||
<a href="/files/${hit.file_id}" style="padding: 0.25rem 0.6rem; background: #f3f4f6; color: #374151; border-radius: 0.25rem; font-size: 0.8rem; text-decoration: none; white-space: nowrap;" title="${__i18n.viewFile}">
|
<a href="/files/${escapeHtml(hit.file_id)}" style="padding: 0.25rem 0.6rem; background: #f3f4f6; color: #374151; border-radius: 0.25rem; font-size: 0.8rem; text-decoration: none; white-space: nowrap;" title="${__i18n.viewFile}">
|
||||||
<i class="fas fa-external-link-alt"></i>
|
<i class="fas fa-external-link-alt"></i>
|
||||||
</a>
|
</a>
|
||||||
</div>
|
</div>
|
||||||
|
|||||||
+1
-1
@@ -1,5 +1,5 @@
|
|||||||
[build-system]
|
[build-system]
|
||||||
requires = ["setuptools>=45", "wheel"]
|
requires = ["setuptools>=82.0.1", "wheel"]
|
||||||
build-backend = "setuptools.build_meta"
|
build-backend = "setuptools.build_meta"
|
||||||
|
|
||||||
[project]
|
[project]
|
||||||
|
|||||||
@@ -235,7 +235,10 @@ class TestSendWebhookNotification:
|
|||||||
mock_response.status_code = 200
|
mock_response.status_code = 200
|
||||||
mock_response.raise_for_status = MagicMock()
|
mock_response.raise_for_status = MagicMock()
|
||||||
|
|
||||||
with patch("app.utils.user_notification.httpx.post", return_value=mock_response) as mock_post:
|
with (
|
||||||
|
patch("app.utils.user_notification.is_private_ip", return_value=False),
|
||||||
|
patch("app.utils.user_notification.httpx.post", return_value=mock_response) as mock_post,
|
||||||
|
):
|
||||||
result = _send_webhook_notification(
|
result = _send_webhook_notification(
|
||||||
{"url": "https://hook.example.com/test", "secret": "mysecret"},
|
{"url": "https://hook.example.com/test", "secret": "mysecret"},
|
||||||
"document.processed",
|
"document.processed",
|
||||||
@@ -256,7 +259,10 @@ class TestSendWebhookNotification:
|
|||||||
mock_response.status_code = 200
|
mock_response.status_code = 200
|
||||||
mock_response.raise_for_status = MagicMock()
|
mock_response.raise_for_status = MagicMock()
|
||||||
|
|
||||||
with patch("app.utils.user_notification.httpx.post", return_value=mock_response) as mock_post:
|
with (
|
||||||
|
patch("app.utils.user_notification.is_private_ip", return_value=False),
|
||||||
|
patch("app.utils.user_notification.httpx.post", return_value=mock_response) as mock_post,
|
||||||
|
):
|
||||||
result = _send_webhook_notification(
|
result = _send_webhook_notification(
|
||||||
{"url": "https://hook.example.com/test"},
|
{"url": "https://hook.example.com/test"},
|
||||||
"document.failed",
|
"document.failed",
|
||||||
@@ -272,9 +278,12 @@ class TestSendWebhookNotification:
|
|||||||
"""_send_webhook_notification returns False when httpx raises."""
|
"""_send_webhook_notification returns False when httpx raises."""
|
||||||
from app.utils.user_notification import _send_webhook_notification
|
from app.utils.user_notification import _send_webhook_notification
|
||||||
|
|
||||||
with patch(
|
with (
|
||||||
"app.utils.user_notification.httpx.post",
|
patch("app.utils.user_notification.is_private_ip", return_value=False),
|
||||||
side_effect=Exception("connection error"),
|
patch(
|
||||||
|
"app.utils.user_notification.httpx.post",
|
||||||
|
side_effect=Exception("connection error"),
|
||||||
|
),
|
||||||
):
|
):
|
||||||
result = _send_webhook_notification(
|
result = _send_webhook_notification(
|
||||||
{"url": "https://hook.example.com/test"},
|
{"url": "https://hook.example.com/test"},
|
||||||
@@ -300,7 +309,10 @@ class TestSendWebhookNotification:
|
|||||||
)
|
)
|
||||||
)
|
)
|
||||||
|
|
||||||
with patch("app.utils.user_notification.httpx.post", return_value=mock_response):
|
with (
|
||||||
|
patch("app.utils.user_notification.is_private_ip", return_value=False),
|
||||||
|
patch("app.utils.user_notification.httpx.post", return_value=mock_response),
|
||||||
|
):
|
||||||
result = _send_webhook_notification(
|
result = _send_webhook_notification(
|
||||||
{"url": "https://hook.example.com/test"},
|
{"url": "https://hook.example.com/test"},
|
||||||
"document.processed",
|
"document.processed",
|
||||||
@@ -310,6 +322,69 @@ class TestSendWebhookNotification:
|
|||||||
|
|
||||||
assert result is False
|
assert result is False
|
||||||
|
|
||||||
|
def test_blocks_private_webhook_target(self):
|
||||||
|
"""Webhook delivery is skipped for private network targets."""
|
||||||
|
from app.utils.user_notification import _send_webhook_notification
|
||||||
|
|
||||||
|
with (
|
||||||
|
patch("app.utils.user_notification.is_private_ip", return_value=True),
|
||||||
|
patch("app.utils.user_notification.httpx.post") as mock_post,
|
||||||
|
):
|
||||||
|
result = _send_webhook_notification(
|
||||||
|
{"url": "https://10.0.0.5/test"},
|
||||||
|
"document.processed",
|
||||||
|
"T",
|
||||||
|
"M",
|
||||||
|
)
|
||||||
|
|
||||||
|
assert result is False
|
||||||
|
mock_post.assert_not_called()
|
||||||
|
|
||||||
|
def test_blocks_metadata_webhook_target(self):
|
||||||
|
"""Webhook delivery is skipped for cloud metadata endpoints."""
|
||||||
|
from app.utils.user_notification import _send_webhook_notification
|
||||||
|
|
||||||
|
with patch("app.utils.user_notification.httpx.post") as mock_post:
|
||||||
|
result = _send_webhook_notification(
|
||||||
|
{"url": "http://169.254.169.254/latest/meta-data"},
|
||||||
|
"document.processed",
|
||||||
|
"T",
|
||||||
|
"M",
|
||||||
|
)
|
||||||
|
|
||||||
|
assert result is False
|
||||||
|
mock_post.assert_not_called()
|
||||||
|
|
||||||
|
def test_blocks_invalid_webhook_scheme(self):
|
||||||
|
"""Webhook delivery is skipped for unsupported URL schemes."""
|
||||||
|
from app.utils.user_notification import _send_webhook_notification
|
||||||
|
|
||||||
|
with patch("app.utils.user_notification.httpx.post") as mock_post:
|
||||||
|
result = _send_webhook_notification(
|
||||||
|
{"url": "file:///etc/passwd"},
|
||||||
|
"document.processed",
|
||||||
|
"T",
|
||||||
|
"M",
|
||||||
|
)
|
||||||
|
|
||||||
|
assert result is False
|
||||||
|
mock_post.assert_not_called()
|
||||||
|
|
||||||
|
def test_blocks_webhook_without_hostname(self):
|
||||||
|
"""Webhook delivery is skipped when the URL has no hostname."""
|
||||||
|
from app.utils.user_notification import _send_webhook_notification
|
||||||
|
|
||||||
|
with patch("app.utils.user_notification.httpx.post") as mock_post:
|
||||||
|
result = _send_webhook_notification(
|
||||||
|
{"url": "https:///missing-host"},
|
||||||
|
"document.processed",
|
||||||
|
"T",
|
||||||
|
"M",
|
||||||
|
)
|
||||||
|
|
||||||
|
assert result is False
|
||||||
|
mock_post.assert_not_called()
|
||||||
|
|
||||||
|
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
# dispatch_user_notification – preference loop
|
# dispatch_user_notification – preference loop
|
||||||
|
|||||||
@@ -81,6 +81,7 @@ class TestDeliverWebhook:
|
|||||||
|
|
||||||
def test_success_returns_true(self, mocker):
|
def test_success_returns_true(self, mocker):
|
||||||
"""A 200 response returns True."""
|
"""A 200 response returns True."""
|
||||||
|
mocker.patch("app.utils.webhook.is_private_ip", return_value=False)
|
||||||
mock_post = mocker.patch("app.utils.webhook.requests.post")
|
mock_post = mocker.patch("app.utils.webhook.requests.post")
|
||||||
mock_post.return_value = MagicMock(ok=True, status_code=200)
|
mock_post.return_value = MagicMock(ok=True, status_code=200)
|
||||||
|
|
||||||
@@ -90,6 +91,7 @@ class TestDeliverWebhook:
|
|||||||
|
|
||||||
def test_non_2xx_returns_false(self, mocker):
|
def test_non_2xx_returns_false(self, mocker):
|
||||||
"""A non-2xx response returns False."""
|
"""A non-2xx response returns False."""
|
||||||
|
mocker.patch("app.utils.webhook.is_private_ip", return_value=False)
|
||||||
mock_post = mocker.patch("app.utils.webhook.requests.post")
|
mock_post = mocker.patch("app.utils.webhook.requests.post")
|
||||||
mock_post.return_value = MagicMock(ok=False, status_code=500)
|
mock_post.return_value = MagicMock(ok=False, status_code=500)
|
||||||
|
|
||||||
@@ -100,6 +102,7 @@ class TestDeliverWebhook:
|
|||||||
"""A network error returns False."""
|
"""A network error returns False."""
|
||||||
import requests
|
import requests
|
||||||
|
|
||||||
|
mocker.patch("app.utils.webhook.is_private_ip", return_value=False)
|
||||||
mocker.patch("app.utils.webhook.requests.post", side_effect=requests.ConnectionError("fail"))
|
mocker.patch("app.utils.webhook.requests.post", side_effect=requests.ConnectionError("fail"))
|
||||||
|
|
||||||
result = deliver_webhook("https://example.com/hook", {"event": "test"})
|
result = deliver_webhook("https://example.com/hook", {"event": "test"})
|
||||||
@@ -107,6 +110,7 @@ class TestDeliverWebhook:
|
|||||||
|
|
||||||
def test_signature_header_included_when_secret(self, mocker):
|
def test_signature_header_included_when_secret(self, mocker):
|
||||||
"""X-Webhook-Signature header is present when a secret is supplied."""
|
"""X-Webhook-Signature header is present when a secret is supplied."""
|
||||||
|
mocker.patch("app.utils.webhook.is_private_ip", return_value=False)
|
||||||
mock_post = mocker.patch("app.utils.webhook.requests.post")
|
mock_post = mocker.patch("app.utils.webhook.requests.post")
|
||||||
mock_post.return_value = MagicMock(ok=True, status_code=200)
|
mock_post.return_value = MagicMock(ok=True, status_code=200)
|
||||||
|
|
||||||
@@ -118,6 +122,7 @@ class TestDeliverWebhook:
|
|||||||
|
|
||||||
def test_no_signature_header_without_secret(self, mocker):
|
def test_no_signature_header_without_secret(self, mocker):
|
||||||
"""X-Webhook-Signature header is absent when no secret is supplied."""
|
"""X-Webhook-Signature header is absent when no secret is supplied."""
|
||||||
|
mocker.patch("app.utils.webhook.is_private_ip", return_value=False)
|
||||||
mock_post = mocker.patch("app.utils.webhook.requests.post")
|
mock_post = mocker.patch("app.utils.webhook.requests.post")
|
||||||
mock_post.return_value = MagicMock(ok=True, status_code=200)
|
mock_post.return_value = MagicMock(ok=True, status_code=200)
|
||||||
|
|
||||||
@@ -126,6 +131,43 @@ class TestDeliverWebhook:
|
|||||||
headers = call_kwargs.kwargs.get("headers") or call_kwargs[1].get("headers")
|
headers = call_kwargs.kwargs.get("headers") or call_kwargs[1].get("headers")
|
||||||
assert "X-Webhook-Signature" not in headers
|
assert "X-Webhook-Signature" not in headers
|
||||||
|
|
||||||
|
def test_private_target_is_blocked(self, mocker):
|
||||||
|
"""Private network webhook targets are not called."""
|
||||||
|
mocker.patch("app.utils.webhook.is_private_ip", return_value=True)
|
||||||
|
mock_post = mocker.patch("app.utils.webhook.requests.post")
|
||||||
|
|
||||||
|
result = deliver_webhook("https://10.0.0.5/hook", {"event": "test"})
|
||||||
|
|
||||||
|
assert result is False
|
||||||
|
mock_post.assert_not_called()
|
||||||
|
|
||||||
|
def test_metadata_target_is_blocked(self, mocker):
|
||||||
|
"""Cloud metadata webhook targets are not called."""
|
||||||
|
mock_post = mocker.patch("app.utils.webhook.requests.post")
|
||||||
|
|
||||||
|
result = deliver_webhook("http://169.254.169.254/latest/meta-data", {"event": "test"})
|
||||||
|
|
||||||
|
assert result is False
|
||||||
|
mock_post.assert_not_called()
|
||||||
|
|
||||||
|
def test_invalid_scheme_is_blocked(self, mocker):
|
||||||
|
"""Unsupported webhook URL schemes are not called."""
|
||||||
|
mock_post = mocker.patch("app.utils.webhook.requests.post")
|
||||||
|
|
||||||
|
result = deliver_webhook("file:///etc/passwd", {"event": "test"})
|
||||||
|
|
||||||
|
assert result is False
|
||||||
|
mock_post.assert_not_called()
|
||||||
|
|
||||||
|
def test_missing_hostname_is_blocked(self, mocker):
|
||||||
|
"""Webhook URLs without a hostname are not called."""
|
||||||
|
mock_post = mocker.patch("app.utils.webhook.requests.post")
|
||||||
|
|
||||||
|
result = deliver_webhook("https:///missing-host", {"event": "test"})
|
||||||
|
|
||||||
|
assert result is False
|
||||||
|
mock_post.assert_not_called()
|
||||||
|
|
||||||
|
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
# Unit tests – get_active_webhooks_for_event (DB)
|
# Unit tests – get_active_webhooks_for_event (DB)
|
||||||
|
|||||||
Reference in New Issue
Block a user