This is an automated email from the ASF dual-hosted git repository. asf-gitbox-commits pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/allura.git
commit 6defa33fbc1672916216fb87e87cb25cc0a24113 Author: Dave Brondsema <[email protected]> AuthorDate: Tue May 5 11:42:25 2026 -0400 [#8603] use NonPrivateUrl on webhooks (does allow for IP addrs now also) --- Allura/allura/tests/test_webhooks.py | 64 +++++++++++++++++++++++------------- Allura/allura/webhooks.py | 9 +++-- 2 files changed, 48 insertions(+), 25 deletions(-) diff --git a/Allura/allura/tests/test_webhooks.py b/Allura/allura/tests/test_webhooks.py index dd373327d..922f41a88 100644 --- a/Allura/allura/tests/test_webhooks.py +++ b/Allura/allura/tests/test_webhooks.py @@ -291,10 +291,29 @@ def test_create_validation(self): r = self.app.post(self.url + '/repo-push/create', data) self.find_error(r, 'url', 'Please enter a value') - data = {'url': 'qwer', 'secret': 'qwe'} - r = self.app.post(self.url + '/repo-push/create', data) - self.find_error(r, 'url', - 'You must provide a full domain name (like qwer.com)') + @pytest.mark.parametrize('url', [ + 'http://localhost/hook', + 'http://127.0.0.1/hook', + 'http://10.0.0.1/hook', + 'http://192.168.1.1/hook', + 'https://169.254.169.254/', + ]) + def test_create_ssrf_private_url_rejected(self, url): + r = self.app.post(self.url + '/repo-push/create', {'url': url, 'secret': ''}) + self.find_error(r, 'url', 'Invalid URL') + + @pytest.mark.parametrize('url', [ + 'http://localhost/hook', + 'http://127.0.0.1/hook', + 'https://10.0.0.1/hook', + ]) + def test_edit_ssrf_private_url_rejected(self, url): + data = {'url': 'http://httpbin.org/post', 'secret': 'secret'} + self.create_webhook(data).follow() + wh = M.Webhook.query.get(hook_url=data['url'], type='repo-push') + edit_data = {'url': url, 'secret': '', 'webhook': str(wh._id)} + r = self.app.post(self.url + '/repo-push/edit', edit_data) + self.find_error(r, 'url', 'Invalid URL', 'edit') def test_edit_validation(self): invalid = M.Webhook( @@ -321,11 +340,6 @@ def test_edit_validation(self): r = self.app.post(self.url + '/repo-push/edit', data) self.find_error(r, 'url', 'Please enter a value', 'edit') - data = {'url': 'qwe', 'secret': 'qwe', 'webhook': str(wh._id)} - r = self.app.post(self.url + '/repo-push/edit', data) - self.find_error(r, 'url', - 'You must provide a full domain name (like qwe.com)', 'edit') - def test_delete(self): data = {'url': 'http://httpbin.org/post', 'secret': 'secret'} @@ -362,9 +376,9 @@ def test_list_webhooks(self): url2 = str(git2.admin_url + 'webhooks') data1 = {'url': 'http://httpbin.org/post', 'secret': 'secret'} - data2 = {'url': 'http://another-host.org/', + data2 = {'url': 'http://example.com/', 'secret': 'secret2'} - data3 = {'url': 'http://another-app.org/', + data3 = {'url': 'http://1.2.3.4/', 'secret': 'secret3'} self.create_webhook(data1).follow() self.create_webhook(data2).follow() @@ -515,6 +529,20 @@ def test_send_error_no_retries(self, log, requests, time): requests.post.return_value.text, requests.post.return_value.headers)) + @pytest.mark.parametrize('url', [ + 'http://localhost/hook', + 'http://127.0.0.1/hook', + 'http://10.0.0.1/hook', + 'http://192.168.1.1/hook', + ]) + @patch('allura.webhooks.requests', autospec=True) + @patch('allura.webhooks.log', autospec=True) + def test_send_ssrf_private_url_blocked(self, log, requests, url): + result = self.h._send(url, json.dumps(self.payload), {}) + assert result is False + requests.post.assert_not_called() + log.error.assert_called_once() + class TestRepoPushWebhookSender(TestWebhookBase): @patch('allura.webhooks.send_webhook', autospec=True) @@ -745,16 +773,6 @@ def test_create_validation(self): 'error': {'url': 'Please enter a value'}, } assert r.json == expected - - data = {'url': 'qwer', 'secret': 'qwe'} - r = self.api_post(self.url + '/repo-push', status=400, **data) - expected = { - 'result': 'error', - 'error': { - 'url': 'You must provide a full domain name (like qwer.com)' - }, - } - assert r.json == expected assert M.Webhook.query.find().count() == len(self.webhooks) def test_create(self): @@ -806,12 +824,12 @@ def test_create_limit_reached(self): def test_edit_validation(self): webhook = self.webhooks[0] url = f'{self.url}/repo-push/{webhook._id}' - data = {'url': 'qwe', 'secret': 'qwe'} + data = {'url': 'http://10.0.0.1/hook', 'secret': ''} r = self.api_post(url, status=400, **data) expected = { 'result': 'error', 'error': { - 'url': 'You must provide a full domain name (like qwe.com)' + 'url': 'Invalid URL.' }, } assert r.json == expected diff --git a/Allura/allura/webhooks.py b/Allura/allura/webhooks.py index 68098f84d..ed5a16ebb 100644 --- a/Allura/allura/webhooks.py +++ b/Allura/allura/webhooks.py @@ -70,7 +70,7 @@ def _convert_to_python(self, value, state): class WebhookCreateForm(schema.Schema): - url = fev.URL(not_empty=True) + url = v.NonPrivateUrl(not_empty=True) secret = v.UnicodeString() @@ -377,7 +377,12 @@ def send(self): if ok: return - def _send(self, url, data, headers): + def _send(self, url, data, headers) -> bool: + try: + v.NonPrivateUrl().to_python(url) + except Invalid as e: + log.error(self.log_msg(f'Webhook send error: resolved to private address: {e!r}')) + return False try: r = requests.post( url,
