609 lines
22 KiB
Python
609 lines
22 KiB
Python
# Copyright (C) 2025 Cetmix OÜ
|
|
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl).
|
|
|
|
import json
|
|
from unittest.mock import patch
|
|
|
|
from odoo.tests import HttpCase, tagged
|
|
|
|
|
|
@tagged("-at_install", "post_install")
|
|
class TestCxTowerWebhookController(HttpCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
super().setUpClass()
|
|
env = cls.env
|
|
# Authenticator that always allows requests
|
|
cls.authenticator = env["cx.tower.webhook.authenticator"].create(
|
|
{"name": "Always OK", "code": "result = {'allowed': True}"}
|
|
)
|
|
# POST webhook
|
|
cls.webhook_post = env["cx.tower.webhook"].create(
|
|
{
|
|
"name": "Test Webhook POST",
|
|
"endpoint": "webhook_post",
|
|
"method": "post",
|
|
"authenticator_id": cls.authenticator.id,
|
|
"code": "result = {'exit_code': 0, 'message': 'POST ok'}",
|
|
}
|
|
)
|
|
# GET webhook
|
|
cls.webhook_get = env["cx.tower.webhook"].create(
|
|
{
|
|
"name": "Test Webhook GET",
|
|
"endpoint": "webhook_get",
|
|
"method": "get",
|
|
"authenticator_id": cls.authenticator.id,
|
|
"code": "result = {'exit_code': 0, 'message': 'GET ok'}",
|
|
}
|
|
)
|
|
# Log model
|
|
cls.Log = env["cx.tower.webhook.log"]
|
|
|
|
def url_for(self, endpoint):
|
|
"""Helper to build webhook url"""
|
|
url = f"/cetmix_tower_webhooks/{endpoint}"
|
|
return self.base_url() + url
|
|
|
|
def assert_log(self, log=None, request_payload=None, **expected):
|
|
"""
|
|
Universal log checker for webhook log model.
|
|
Checks expected field values and substrings.
|
|
"""
|
|
self.assertIsNotNone(log, "Log record was not created")
|
|
if request_payload is not None:
|
|
try:
|
|
log_payload = log.request_payload
|
|
# try to convert both to Python dict for comparison
|
|
if isinstance(log_payload, str):
|
|
log_payload = log_payload.strip()
|
|
self.assertDictEqual(
|
|
json.loads(
|
|
log_payload.replace("'", '"')
|
|
), # try to make JSON from possible str(dict)
|
|
json.loads(request_payload),
|
|
)
|
|
except Exception as ex:
|
|
self.fail(
|
|
f"Payload comparison failed: {ex}\nLog: {log.request_payload}\nExpected: {request_payload}" # noqa: E501
|
|
)
|
|
for field, value in expected.items():
|
|
if field == "request_payload":
|
|
continue # Already checked
|
|
actual = getattr(log, field)
|
|
self.assertEqual(actual, value, f"{field}: expected {value}, got {actual}")
|
|
|
|
def test_post_webhook_success(self):
|
|
"""Success test for POST request with correct payload."""
|
|
data = json.dumps({"some": "data"})
|
|
response = self.url_open(
|
|
self.url_for(self.webhook_post.endpoint),
|
|
data=data,
|
|
headers={"Content-Type": "application/json"},
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertIn(b"POST ok", response.content)
|
|
|
|
log = self.Log.search([("webhook_id", "=", self.webhook_post.id)])
|
|
self.assert_log(
|
|
log,
|
|
code_status="success",
|
|
authentication_status="success",
|
|
http_status=200,
|
|
endpoint=self.webhook_post.endpoint,
|
|
request_payload=data,
|
|
)
|
|
|
|
def test_get_webhook_success(self):
|
|
"""Success test for GET request with correct payload."""
|
|
response = self.url_open(
|
|
f"{self.url_for(self.webhook_get.endpoint)}?foo=bar",
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertIn(b"GET ok", response.content)
|
|
|
|
log = self.Log.search([("webhook_id", "=", self.webhook_get.id)])
|
|
self.assert_log(
|
|
log,
|
|
code_status="success",
|
|
authentication_status="success",
|
|
http_status=200,
|
|
endpoint=self.webhook_get.endpoint,
|
|
)
|
|
self.assertIn("foo", log.request_payload)
|
|
|
|
def test_webhook_not_found(self):
|
|
"""Test request to a non-existing webhook endpoint."""
|
|
data = json.dumps({"test": 1})
|
|
response = self.url_open(
|
|
self.url_for("missing"),
|
|
data=data,
|
|
headers={"Content-Type": "application/json"},
|
|
)
|
|
self.assertEqual(response.status_code, 404)
|
|
self.assertIn(b"Webhook not found", response.content)
|
|
|
|
log = self.Log.search([("webhook_id", "=", False)])
|
|
self.assert_log(
|
|
log,
|
|
code_status="skipped",
|
|
authentication_status="failed",
|
|
http_status=404,
|
|
endpoint="missing",
|
|
error_message="Webhook not found",
|
|
request_payload=data,
|
|
)
|
|
|
|
def test_wrong_method(self):
|
|
"""
|
|
Test GET request to POST-only webhook.
|
|
"""
|
|
response = self.url_open(
|
|
self.url_for(self.webhook_post.endpoint),
|
|
)
|
|
self.assertEqual(response.status_code, 404)
|
|
self.assertIn(b"Webhook not found", response.content)
|
|
|
|
log = self.Log.search([("webhook_id", "=", False)])
|
|
self.assert_log(
|
|
log,
|
|
code_status="skipped",
|
|
authentication_status="failed",
|
|
http_status=404,
|
|
error_message="Webhook not found",
|
|
endpoint=self.webhook_post.endpoint,
|
|
request_method="get",
|
|
)
|
|
|
|
def test_missing_payload_post(self):
|
|
"""
|
|
Test POST request with empty payload.
|
|
"""
|
|
# use opener instead of url_open to avoid checking of data
|
|
response = self.opener.post(
|
|
self.url_for(self.webhook_post.endpoint),
|
|
timeout=1200000,
|
|
headers={"Content-Type": "application/json"},
|
|
allow_redirects=True,
|
|
)
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertIn(b"POST ok", response.content)
|
|
|
|
log = self.Log.search([("webhook_id", "=", self.webhook_post.id)])
|
|
self.assert_log(
|
|
log,
|
|
code_status="success",
|
|
authentication_status="success",
|
|
http_status=200,
|
|
endpoint=self.webhook_post.endpoint,
|
|
request_payload="{}",
|
|
)
|
|
|
|
def test_authentication_failed(self):
|
|
"""
|
|
Test POST request with authenticator that always denies.
|
|
"""
|
|
bad_auth = self.env["cx.tower.webhook.authenticator"].create(
|
|
{
|
|
"name": "Never OK",
|
|
"code": "result = {'allowed': False, 'custom_message': 'Forbidden'}",
|
|
}
|
|
)
|
|
webhook = self.env["cx.tower.webhook"].create(
|
|
{
|
|
"name": "Forbidden Webhook",
|
|
"endpoint": "forbidden",
|
|
"method": "post",
|
|
"authenticator_id": bad_auth.id,
|
|
"code": "result = {'exit_code': 0, 'message': 'Should not run'}",
|
|
}
|
|
)
|
|
data = json.dumps({"fail": 1})
|
|
response = self.url_open(
|
|
self.url_for(webhook.endpoint),
|
|
data=data,
|
|
headers={"Content-Type": "application/json"},
|
|
)
|
|
self.assertEqual(response.status_code, 403)
|
|
self.assertIn(b"Authentication not allowed", response.content)
|
|
|
|
log = self.Log.search([("webhook_id", "=", webhook.id)])
|
|
self.assert_log(
|
|
log,
|
|
code_status="skipped",
|
|
authentication_status="failed",
|
|
http_status=403,
|
|
endpoint=webhook.endpoint,
|
|
request_payload=data,
|
|
)
|
|
|
|
def test_webhook_code_failure(self):
|
|
"""
|
|
Test POST request to a webhook that raises an exception in code.
|
|
"""
|
|
self.webhook_post.code = "raise Exception('Some error!')"
|
|
response = self.url_open(
|
|
self.url_for(self.webhook_post.endpoint),
|
|
data=json.dumps({}),
|
|
headers={"Content-Type": "application/json"},
|
|
)
|
|
self.assertEqual(response.status_code, 500)
|
|
self.assertIn(b"Some error!", response.content)
|
|
|
|
log = self.Log.search([("webhook_id", "=", self.webhook_post.id)])
|
|
self.assert_log(
|
|
log,
|
|
code_status="failed",
|
|
authentication_status="success",
|
|
http_status=500,
|
|
endpoint=self.webhook_post.endpoint,
|
|
request_payload="{}",
|
|
)
|
|
self.assertIn("Some error!", log.error_message)
|
|
|
|
def test_json_headers_are_stored(self):
|
|
"""
|
|
Test that request headers and payload are saved in webhook log record.
|
|
"""
|
|
payload = {"foo": "bar"}
|
|
headers = {"X-Test-Header": "xxx", "Content-Type": "application/json"}
|
|
response = self.url_open(
|
|
self.url_for(self.webhook_post.endpoint),
|
|
data=json.dumps(payload),
|
|
headers=headers,
|
|
)
|
|
|
|
log = self.Log.search([("webhook_id", "=", self.webhook_post.id)])
|
|
self.assert_log(
|
|
log,
|
|
code_status="success",
|
|
authentication_status="success",
|
|
http_status=200,
|
|
endpoint=self.webhook_post.endpoint,
|
|
)
|
|
self.assertIn("foo", log.request_payload)
|
|
self.assertIn("X-Test-Header", log.request_headers)
|
|
self.assertIn(log.result_message, response.text)
|
|
|
|
def test_log_contains_ip(self):
|
|
"""
|
|
Test that the log contains the client's IP address and country (if available).
|
|
"""
|
|
payload = {"check": "ip"}
|
|
self.url_open(
|
|
self.url_for(self.webhook_post.endpoint),
|
|
data=json.dumps(payload),
|
|
headers={"Content-Type": "application/json"},
|
|
)
|
|
|
|
log = self.Log.search([("webhook_id", "=", self.webhook_post.id)])
|
|
self.assertTrue(log.ip_address)
|
|
|
|
def test_inactive_webhook(self):
|
|
"""Test that inactive webhooks are not callable."""
|
|
self.webhook_post.active = False
|
|
response = self.url_open(
|
|
self.url_for(self.webhook_post.endpoint),
|
|
data=json.dumps({"a": 1}),
|
|
headers={"Content-Type": "application/json"},
|
|
)
|
|
self.assertEqual(response.status_code, 404)
|
|
self.assertIn(b"Webhook not found", response.content)
|
|
|
|
def test_authenticator_code_raises(self):
|
|
"""
|
|
Test that if authenticator's code raises an error,
|
|
proper log is created and 403 returned.
|
|
"""
|
|
bad_auth = self.env["cx.tower.webhook.authenticator"].create(
|
|
{"name": "Broken Auth", "code": "raise Exception('auth fail')"}
|
|
)
|
|
webhook = self.env["cx.tower.webhook"].create(
|
|
{
|
|
"name": "Web with bad auth",
|
|
"endpoint": "bad_auth",
|
|
"method": "post",
|
|
"authenticator_id": bad_auth.id,
|
|
"code": "result = {'exit_code': 0, 'message': 'Should not run'}",
|
|
}
|
|
)
|
|
response = self.url_open(
|
|
self.url_for(webhook.endpoint),
|
|
data=json.dumps({"x": 1}),
|
|
headers={"Content-Type": "application/json"},
|
|
)
|
|
self.assertEqual(response.status_code, 403)
|
|
self.assertIn(b"auth fail", response.content)
|
|
|
|
log = self.Log.search([("webhook_id", "=", webhook.id)])
|
|
self.assert_log(
|
|
log,
|
|
code_status="skipped",
|
|
authentication_status="failed",
|
|
http_status=403,
|
|
endpoint=webhook.endpoint,
|
|
)
|
|
self.assertIn("auth fail", log.error_message)
|
|
|
|
def test_post_webhook_json_content_type(self):
|
|
"""
|
|
Test POST request with content_type json.
|
|
"""
|
|
self.webhook_post.content_type = "json"
|
|
self.webhook_post.code = "result = {'exit_code': 0, 'message': 'POST JSON ok'}"
|
|
|
|
data = json.dumps({"json_test": "ok"})
|
|
response = self.url_open(
|
|
self.url_for(self.webhook_post.endpoint),
|
|
data=data,
|
|
headers={"Content-Type": "application/json"},
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertIn(b"POST JSON ok", response.content)
|
|
|
|
log = self.Log.search([("webhook_id", "=", self.webhook_post.id)])
|
|
self.assert_log(
|
|
log,
|
|
code_status="success",
|
|
authentication_status="success",
|
|
http_status=200,
|
|
endpoint=self.webhook_post.endpoint,
|
|
request_payload=data,
|
|
)
|
|
|
|
def test_post_webhook_form_content_type(self):
|
|
"""
|
|
Test POST request with content_type form.
|
|
"""
|
|
self.webhook_post.content_type = "form"
|
|
self.webhook_post.code = "result = {'exit_code': 0, 'message': 'POST FORM ok'}"
|
|
|
|
data = {"form_field": "ok"}
|
|
response = self.url_open(
|
|
self.url_for(self.webhook_post.endpoint),
|
|
data=data,
|
|
headers={"Content-Type": "application/x-www-form-urlencoded"},
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertIn(b"POST FORM ok", response.content)
|
|
|
|
log = self.Log.search([("webhook_id", "=", self.webhook_post.id)])
|
|
self.assertIn("form_field", log.request_payload)
|
|
|
|
def test_authenticator_ipv4_and_ipv6(self):
|
|
"""
|
|
Test IP filter for IPv4, IPv6, and networks
|
|
by monkeypatching REMOTE_ADDR in environ.
|
|
"""
|
|
auth = self.env["cx.tower.webhook.authenticator"].create(
|
|
{
|
|
"name": "IP Test",
|
|
"allowed_ip_addresses": "203.0.113.5,2001:db8::42,198.51.100.0/24,2001:db8:abcd::/48", # noqa: E501
|
|
"code": "result = {'allowed': True}",
|
|
}
|
|
)
|
|
webhook = self.env["cx.tower.webhook"].create(
|
|
{
|
|
"name": "IP Webhook",
|
|
"endpoint": "webhook_iptest",
|
|
"method": "post",
|
|
"authenticator_id": auth.id,
|
|
"code": "result = {'exit_code': 0, 'message': 'IP OK'}",
|
|
}
|
|
)
|
|
|
|
data = json.dumps({"ip": "test"})
|
|
|
|
def do_req(ip):
|
|
# Patch _get_remote_addr to simulate requests coming
|
|
# from different IP addresses
|
|
with patch(
|
|
"odoo.addons.cetmix_tower_webhook.controllers.main.CetmixTowerWebhookController._get_remote_addr",
|
|
return_value=ip,
|
|
):
|
|
return self.url_open(
|
|
self.url_for(webhook.endpoint),
|
|
data=data,
|
|
headers={"Content-Type": "application/json"},
|
|
)
|
|
|
|
# IPv4 address allowed
|
|
resp = do_req("203.0.113.5")
|
|
self.assertEqual(resp.status_code, 200)
|
|
self.assertIn(b"IP OK", resp.content)
|
|
|
|
# IPv6 address allowed
|
|
resp = do_req("2001:db8::42")
|
|
self.assertEqual(resp.status_code, 200)
|
|
self.assertIn(b"IP OK", resp.content)
|
|
|
|
# IPv4 network allowed
|
|
resp = do_req("198.51.100.99")
|
|
self.assertEqual(resp.status_code, 200)
|
|
self.assertIn(b"IP OK", resp.content)
|
|
|
|
# IPv6 network allowed
|
|
resp = do_req("2001:db8:abcd::abcd")
|
|
self.assertEqual(resp.status_code, 200)
|
|
self.assertIn(b"IP OK", resp.content)
|
|
|
|
# Denied IPv4 address
|
|
resp = do_req("203.0.113.99")
|
|
self.assertEqual(resp.status_code, 403)
|
|
self.assertIn(b"Address not allowed", resp.content)
|
|
|
|
# Denied IPv6 address
|
|
resp = do_req("2001:db8:ffff::1")
|
|
self.assertEqual(resp.status_code, 403)
|
|
self.assertIn(b"Address not allowed", resp.content)
|
|
|
|
def _make_proxy_webhook(
|
|
self,
|
|
allowed,
|
|
trusted=None,
|
|
code="result = {'exit_code': 0, 'message': 'OK via proxy'}",
|
|
):
|
|
"""
|
|
Helper to create a webhook with a dedicated authenticator configured
|
|
for proxy-aware tests.
|
|
"""
|
|
auth = self.env["cx.tower.webhook.authenticator"].create(
|
|
{
|
|
"name": "Proxy Aware",
|
|
"allowed_ip_addresses": allowed,
|
|
"trusted_proxy_ips": trusted or "",
|
|
"code": "result = {'allowed': True}",
|
|
}
|
|
)
|
|
wh = self.env["cx.tower.webhook"].create(
|
|
{
|
|
"name": "Proxy Webhook",
|
|
"endpoint": "proxy_webhook",
|
|
"method": "post",
|
|
"authenticator_id": auth.id,
|
|
"code": code,
|
|
}
|
|
)
|
|
return wh, auth
|
|
|
|
def test_proxy_headers_ignored_without_trusted_proxy(self):
|
|
"""
|
|
When trusted_proxy_ips is empty, XFF/X-Real-IP must be ignored.
|
|
We fallback to immediate peer (proxy IP), which is not allowed -> 403.
|
|
"""
|
|
# Allow only the real client network, not the proxy itself
|
|
webhook, _auth = self._make_proxy_webhook(
|
|
allowed="203.0.113.0/24", trusted=None
|
|
)
|
|
|
|
data = json.dumps({"k": "v"})
|
|
proxy_ip = "10.0.0.5" # immediate peer (undocumented as trusted)
|
|
headers = {
|
|
"Content-Type": "application/json",
|
|
"X-Forwarded-For": "203.0.113.7, 10.0.0.5", # should be ignored
|
|
"X-Real-IP": "203.0.113.7", # should be ignored
|
|
}
|
|
with patch(
|
|
"odoo.addons.cetmix_tower_webhook.controllers.main.CetmixTowerWebhookController._get_remote_addr",
|
|
return_value=proxy_ip,
|
|
):
|
|
resp = self.url_open(
|
|
self.url_for(webhook.endpoint), data=data, headers=headers
|
|
)
|
|
|
|
self.assertEqual(resp.status_code, 403)
|
|
self.assertIn(b"Address not allowed", resp.content)
|
|
|
|
def test_proxy_xff_honored_with_trusted_proxy(self):
|
|
"""
|
|
With trusted proxy configured, take the left-most IP from X-Forwarded-For.
|
|
"""
|
|
webhook, _auth = self._make_proxy_webhook(
|
|
allowed="203.0.113.0/24",
|
|
trusted="10.0.0.5",
|
|
code="result = {'exit_code': 0, 'message': 'OK XFF'}",
|
|
)
|
|
|
|
data = json.dumps({"k": "v"})
|
|
proxy_ip = "10.0.0.5"
|
|
headers = {
|
|
"Content-Type": "application/json",
|
|
# XFF list: client, proxy
|
|
"X-Forwarded-For": "203.0.113.7, 10.0.0.5",
|
|
}
|
|
with patch(
|
|
"odoo.addons.cetmix_tower_webhook.controllers.main.CetmixTowerWebhookController._get_remote_addr",
|
|
return_value=proxy_ip,
|
|
):
|
|
resp = self.url_open(
|
|
self.url_for(webhook.endpoint), data=data, headers=headers
|
|
)
|
|
|
|
self.assertEqual(resp.status_code, 200)
|
|
self.assertIn(b"OK XFF", resp.content)
|
|
|
|
def test_proxy_x_real_ip_fallback_when_xff_missing(self):
|
|
"""
|
|
If XFF is missing/invalid but trusted proxy is set, fall back to X-Real-IP.
|
|
"""
|
|
webhook, _auth = self._make_proxy_webhook(
|
|
allowed="203.0.113.0/24",
|
|
trusted="10.0.0.5",
|
|
code="result = {'exit_code': 0, 'message': 'OK X-Real-IP'}",
|
|
)
|
|
|
|
data = json.dumps({"k": "v"})
|
|
proxy_ip = "10.0.0.5"
|
|
headers = {
|
|
"Content-Type": "application/json",
|
|
"X-Forwarded-For": "garbage, not_an_ip", # invalids should be skipped
|
|
"X-Real-IP": "203.0.113.8",
|
|
}
|
|
with patch(
|
|
"odoo.addons.cetmix_tower_webhook.controllers.main.CetmixTowerWebhookController._get_remote_addr",
|
|
return_value=proxy_ip,
|
|
):
|
|
resp = self.url_open(
|
|
self.url_for(webhook.endpoint), data=data, headers=headers
|
|
)
|
|
|
|
self.assertEqual(resp.status_code, 200)
|
|
self.assertIn(b"OK X-Real-IP", resp.content)
|
|
|
|
def test_proxy_invalid_headers_fall_back_to_immediate_peer(self):
|
|
"""
|
|
If headers are invalid even with trusted proxy, fall back to immediate peer.
|
|
Since the proxy IP is not in allowlist, the request is denied.
|
|
"""
|
|
webhook, _auth = self._make_proxy_webhook(
|
|
allowed="203.0.113.0/24", # does NOT include proxy IP
|
|
trusted="10.0.0.5",
|
|
)
|
|
|
|
data = json.dumps({"k": "v"})
|
|
proxy_ip = "10.0.0.5"
|
|
headers = {
|
|
"Content-Type": "application/json",
|
|
"X-Forwarded-For": "not_an_ip, also_bad",
|
|
"X-Real-IP": "bad_ip_value",
|
|
}
|
|
with patch(
|
|
"odoo.addons.cetmix_tower_webhook.controllers.main.CetmixTowerWebhookController._get_remote_addr",
|
|
return_value=proxy_ip,
|
|
):
|
|
resp = self.url_open(
|
|
self.url_for(webhook.endpoint), data=data, headers=headers
|
|
)
|
|
|
|
self.assertEqual(resp.status_code, 403)
|
|
self.assertIn(b"Address not allowed", resp.content)
|
|
|
|
def test_proxy_allows_via_immediate_peer_when_proxy_ip_in_allowlist(self):
|
|
"""
|
|
If headers are ignored/invalid, but the proxy IP itself is allowed,
|
|
access should be granted based on immediate peer.
|
|
"""
|
|
webhook, _auth = self._make_proxy_webhook(
|
|
allowed="10.0.0.5", # allow the proxy itself
|
|
trusted="", # no trusted proxies => headers ignored
|
|
code="result = {'exit_code': 0, 'message': 'OK immediate peer'}",
|
|
)
|
|
|
|
data = json.dumps({"k": "v"})
|
|
proxy_ip = "10.0.0.5"
|
|
headers = {
|
|
"Content-Type": "application/json",
|
|
"X-Forwarded-For": "203.0.113.7", # should be ignored
|
|
}
|
|
with patch(
|
|
"odoo.addons.cetmix_tower_webhook.controllers.main.CetmixTowerWebhookController._get_remote_addr",
|
|
return_value=proxy_ip,
|
|
):
|
|
resp = self.url_open(
|
|
self.url_for(webhook.endpoint), data=data, headers=headers
|
|
)
|
|
|
|
self.assertEqual(resp.status_code, 200)
|
|
self.assertIn(b"OK immediate peer", resp.content)
|