Merge pull request #5205 from kobotoolbox/task-1200-SSRFProtect-bugfix

Fix SSRFProtect validation options to properly handle Constance settings

kobo/apps/hook/models/service_definition_interface.py

@@ -9,6 +9,7 @@
 from ssrf_protect.ssrf_protect import SSRFProtect, SSRFProtectException
 
 from kpi.utils.log import logging
+from kpi.utils.strings import split_lines_to_list
 from .hook import Hook
 from .hook_log import HookLog
 from ..constants import (
@@ -108,18 +109,24 @@ def send(self):
 
                 ssrf_protect_options = {}
                 if constance.config.SSRF_ALLOWED_IP_ADDRESS.strip():
-                    ssrf_protect_options['allowed_ip_addresses'] = constance.\
-                        config.SSRF_ALLOWED_IP_ADDRESS.strip().split('\r\n')
+                    ssrf_protect_options['allowed_ip_addresses'] = (
+                        split_lines_to_list(
+                            constance.config.SSRF_ALLOWED_IP_ADDRESS
+                        )
+                    )
 
                 if constance.config.SSRF_DENIED_IP_ADDRESS.strip():
-                    ssrf_protect_options['denied_ip_addresses'] = constance.\
-                        config.SSRF_DENIED_IP_ADDRESS.strip().split('\r\n')
+                    ssrf_protect_options['denied_ip_addresses'] = (
+                        split_lines_to_list(
+                            constance.config.SSRF_DENIED_IP_ADDRESS
+                        )
+                    )
 
-                SSRFProtect.validate(self._hook.endpoint,
-                                     options=ssrf_protect_options)
+                SSRFProtect.validate(self._hook.endpoint, options=ssrf_protect_options)
 
-                response = requests.post(self._hook.endpoint, timeout=30,
-                                         **request_kwargs)
+                response = requests.post(
+                    self._hook.endpoint, timeout=30, **request_kwargs
+                )
                 response.raise_for_status()
                 self.save_log(response.status_code, response.text, True)
                 success = True

kobo/apps/hook/tests/test_ssrf.py

@@ -1,8 +1,7 @@
-# coding: utf-8
+from unittest.mock import patch
 
 import responses
 from constance.test import override_config
-from mock import patch
 from rest_framework import status
 
 from kobo.apps.hook.constants import (
@@ -27,11 +26,12 @@ def test_send_with_ssrf_options(self):
         submissions = self.asset.deployment.get_submissions(self.asset.owner)
         submission_id = submissions[0]['_id']
         service_definition = ServiceDefinition(hook, submission_id)
-        first_mock_response = {'error': 'not found'}
-
-        responses.add(responses.POST, hook.endpoint,
-                      status=status.HTTP_200_OK,
-                      content_type='application/json')
+        responses.add(
+            responses.POST,
+            hook.endpoint,
+            status=status.HTTP_200_OK,
+            content_type='application/json',
+        )
 
         # Try to send data to external endpoint
         success = service_definition.send()

kpi/tests/test_utils.py

@@ -1,4 +1,3 @@
-# coding: utf-8
 import os
 import re
 from copy import deepcopy
@@ -17,6 +16,7 @@
 from kpi.utils.pyxform_compatibility import allow_choice_duplicates
 from kpi.utils.query_parser import parse
 from kpi.utils.sluggify import sluggify, sluggify_label
+from kpi.utils.strings import split_lines_to_list
 from kpi.utils.xml import (
     edit_submission_xml,
     fromstring_preserve_root_xmlns,
@@ -301,6 +301,12 @@ def test_allow_choice_duplicates(self):
             == 'no'
         )
 
+    def test_split_lines_to_list(self):
+
+        value = '\r\nfoo\r\nbar\n\n'
+        expected = ['foo', 'bar']
+        assert split_lines_to_list(value) == expected
+
 
 class XmlUtilsTestCase(TestCase):
 

kpi/utils/strings.py

@@ -1,4 +1,3 @@
-# coding: utf-8
 import base64
 
 
@@ -10,3 +9,8 @@ def to_str(obj):
     if isinstance(obj, bytes):
         return obj.decode()
     return obj
+
+
+def split_lines_to_list(value: str) -> list:
+    values = value.strip().split('\n')
+    return [ip.strip() for ip in values if ip.strip()]