#331 added support for ip4 subnets in trusted_ips

Author: bugy

src/auth/identification.py

@@ -4,6 +4,7 @@
 
 import tornado.websocket
 
+from model.trusted_ips import TrustedIpValidator
 from utils import tornado_utils, date_utils, audit_utils
 from utils.date_utils import days_to_ms
 
@@ -39,13 +40,13 @@ class IpBasedIdentification(Identification):
     COOKIE_KEY = 'client_id_token'
     EMPTY_TOKEN = (None, None)
 
-    def __init__(self, trusted_ips, user_header_name) -> None:
-        self._trusted_ips = set(trusted_ips)
+    def __init__(self, ip_validator: TrustedIpValidator, user_header_name) -> None:
+        self._ip_validator = ip_validator
         self._user_header_name = user_header_name
 
     def identify(self, request_handler):
         remote_ip = request_handler.request.remote_ip
-        new_trusted = remote_ip in self._trusted_ips
+        new_trusted = self._ip_validator.is_trusted(remote_ip)
 
         if new_trusted:
             if request_handler.get_cookie(self.COOKIE_KEY):
@@ -77,7 +78,7 @@ def identify(self, request_handler):
 
     def identify_for_audit(self, request_handler):
         remote_ip = request_handler.request.remote_ip
-        if (remote_ip in self._trusted_ips) and (self._user_header_name):
+        if self._ip_validator.is_trusted(remote_ip) and (self._user_header_name):
             return request_handler.request.headers.get(self._user_header_name, None)
         return None
 

src/execution/logging.py

@@ -190,8 +190,8 @@ def find_history_entry(self, execution_id, user_id):
             LOGGER.warning('find_history_entry: cannot parse file for %s', execution_id)
 
         elif not self._can_access_entry(entry, user_id):
-            message = 'User ' + user_id + ' has not access to execution #' + str(execution_id)
-            LOGGER.warning('%s. Original user: %s', message, execution_id)
+            message = 'User ' + user_id + ' has no access to execution #' + str(execution_id)
+            LOGGER.warning('%s. Original user: %s', message, entry.user_id)
             raise AccessProhibitedException(message)
 
         return entry

src/model/__init__.py

@@ -6,6 +6,7 @@
 from auth.authorization import ANY_USER
 from model import model_helper
 from model.model_helper import read_list, read_int_from_config, read_bool_from_config
+from model.trusted_ips import TrustedIpValidator
 from utils.string_utils import strip
 
 LOGGER = logging.getLogger('server_conf')
@@ -25,7 +26,7 @@ def __init__(self) -> None:
         self.admin_config = None
         self.title = None
         self.enable_script_titles = None
-        self.trusted_ips = []
+        self.ip_validator = TrustedIpValidator([])
         self.user_groups = None
         self.admin_users = []
         self.full_history_users = []
@@ -113,11 +114,11 @@ def from_json(conf_path, temp_folder):
         def_admins = def_trusted_ips
 
     if access_config:
-        config.trusted_ips = strip(read_list(access_config, 'trusted_ips', default=def_trusted_ips))
+        trusted_ips = strip(read_list(access_config, 'trusted_ips', default=def_trusted_ips))
         admin_users = _parse_admin_users(access_config, default_admins=def_admins)
         full_history_users = _parse_history_users(access_config)
     else:
-        config.trusted_ips = def_trusted_ips
+        trusted_ips = def_trusted_ips
         admin_users = def_admins
         full_history_users = []
 
@@ -129,6 +130,7 @@ def from_json(conf_path, temp_folder):
     config.admin_users = admin_users
     config.full_history_users = full_history_users
     config.user_header_name = user_header_name
+    config.ip_validator = TrustedIpValidator(trusted_ips)
 
     config.max_request_size_mb = read_int_from_config('max_request_size', json_object, default=10)
 

src/model/server_conf.py

@@ -0,0 +1,19 @@
+import ipaddress
+
+
+class TrustedIpValidator:
+    def __init__(self, trusted_ips) -> None:
+        self._simple_ips = {ip for ip in trusted_ips if '/' not in ip}
+        self._networks = [ipaddress.ip_network(ip) for ip in trusted_ips if '/' in ip]
+
+    def is_trusted(self, ip):
+        if ip in self._simple_ips:
+            return True
+
+        if self._networks:
+            address = ipaddress.ip_address(ip)
+            for network in self._networks:
+                if address in network:
+                    return True
+
+        return False

src/model/trusted_ips.py

@@ -2,6 +2,7 @@
 
 from auth.identification import IpBasedIdentification
 from auth.tornado_auth import TornadoAuth
+from model.trusted_ips import TrustedIpValidator
 from tests.test_utils import mock_object
 from utils import date_utils
 
@@ -13,7 +14,7 @@ def mock_request_handler(ip=None, x_forwarded_for=None, x_real_ip=None, saved_to
 
     handler_mock.application = mock_object()
     handler_mock.application.auth = TornadoAuth(None)
-    handler_mock.application.identification = IpBasedIdentification(['127.0.0.1'], user_header_name)
+    handler_mock.application.identification = IpBasedIdentification(TrustedIpValidator(['127.0.0.1']), user_header_name)
 
     handler_mock.request = mock_object()
     handler_mock.request.headers = {}
@@ -55,22 +56,22 @@ def clear_cookie(key):
 class IpIdentificationTest(unittest.TestCase):
 
     def test_localhost_ip_trusted_identification(self):
-        identification = IpBasedIdentification(['127.0.0.1'], None)
+        identification = IpBasedIdentification(TrustedIpValidator(['127.0.0.1']), None)
         id = identification.identify(mock_request_handler(ip='127.0.0.1'))
         self.assertEqual('127.0.0.1', id)
 
     def test_some_ip_trusted_identification(self):
-        identification = IpBasedIdentification(['192.168.21.13'], None)
+        identification = IpBasedIdentification(TrustedIpValidator(['192.168.21.13']), None)
         id = identification.identify(mock_request_handler(ip='192.168.21.13'))
         self.assertEqual('192.168.21.13', id)
 
     def test_ip_untrusted_identification(self):
-        identification = IpBasedIdentification([], None)
+        identification = IpBasedIdentification(TrustedIpValidator([]), None)
         id = identification.identify(mock_request_handler(ip='192.168.21.13'))
         self.assertNotEqual('192.168.21.13', id)
 
     def test_ip_untrusted_identification_for_different_connections(self):
-        identification = IpBasedIdentification([], None)
+        identification = IpBasedIdentification(TrustedIpValidator([]), None)
 
         ids = set()
         for _ in range(0, 100):
@@ -79,22 +80,22 @@ def test_ip_untrusted_identification_for_different_connections(self):
         self.assertEqual(100, len(ids))
 
     def test_ip_untrusted_identification_same_connection(self):
-        identification = IpBasedIdentification([], None)
+        identification = IpBasedIdentification(TrustedIpValidator([]), None)
 
         request_handler = mock_request_handler(ip='192.168.21.13')
         id1 = identification.identify(request_handler)
         id2 = identification.identify(request_handler)
         self.assertEqual(id1, id2)
 
     def test_proxied_ip_behind_trusted(self):
-        identification = IpBasedIdentification(['127.0.0.1'], None)
+        identification = IpBasedIdentification(TrustedIpValidator(['127.0.0.1']), None)
 
         request_handler = mock_request_handler(ip='127.0.0.1', x_forwarded_for='192.168.21.13')
         id = identification.identify(request_handler)
         self.assertEqual('192.168.21.13', id)
 
     def test_proxied_ip_behind_untrusted(self):
-        identification = IpBasedIdentification([], None)
+        identification = IpBasedIdentification(TrustedIpValidator([]), None)
 
         request_handler = mock_request_handler(ip='127.0.0.1', x_forwarded_for='192.168.21.13')
         id = identification.identify(request_handler)
@@ -104,9 +105,9 @@ def test_proxied_ip_behind_untrusted(self):
     def test_change_to_trusted(self):
         request_handler = mock_request_handler(ip='192.168.21.13')
 
-        old_id = IpBasedIdentification([], None).identify(request_handler)
+        old_id = IpBasedIdentification(TrustedIpValidator([]), None).identify(request_handler)
 
-        trusted_identification = IpBasedIdentification(['192.168.21.13'], None)
+        trusted_identification = IpBasedIdentification(TrustedIpValidator(['192.168.21.13']), None)
         new_id = trusted_identification.identify(request_handler)
 
         self.assertNotEqual(old_id, new_id)
@@ -116,10 +117,10 @@ def test_change_to_trusted(self):
     def test_change_to_untrusted(self):
         request_handler = mock_request_handler(ip='192.168.21.13')
 
-        trusted_identification = IpBasedIdentification(['192.168.21.13'], None)
+        trusted_identification = IpBasedIdentification(TrustedIpValidator(['192.168.21.13']), None)
         old_id = trusted_identification.identify(request_handler)
 
-        new_id = IpBasedIdentification([], None).identify(request_handler)
+        new_id = IpBasedIdentification(TrustedIpValidator([]), None).identify(request_handler)
 
         self.assertNotEqual(old_id, new_id)
         self.assertNotEqual(new_id, '192.168.21.13')
@@ -128,7 +129,7 @@ def test_change_to_untrusted(self):
     def test_no_cookie_change_for_same_user(self):
         request_handler = mock_request_handler(ip='192.168.21.13')
 
-        identification = IpBasedIdentification([], None)
+        identification = IpBasedIdentification(TrustedIpValidator([]), None)
 
         identification.identify(request_handler)
         cookie1 = request_handler.get_cookie(COOKIE_KEY)
@@ -140,7 +141,7 @@ def test_no_cookie_change_for_same_user(self):
     def test_refresh_old_cookie_with_same_id(self):
         request_handler = mock_request_handler(ip='192.168.21.13')
 
-        identification = IpBasedIdentification([], None)
+        identification = IpBasedIdentification(TrustedIpValidator([]), None)
 
         id = '1234567'
         token_expiry = str(date_utils.get_current_millis() + date_utils.days_to_ms(2))
@@ -157,7 +158,7 @@ def test_broken_token_structure(self):
         request_handler = mock_request_handler(ip='192.168.21.13')
         request_handler.set_secure_cookie(COOKIE_KEY, 'something')
 
-        IpBasedIdentification([], None).identify(request_handler)
+        IpBasedIdentification(TrustedIpValidator([]), None).identify(request_handler)
 
         new_token = request_handler.get_cookie(COOKIE_KEY)
 
@@ -167,7 +168,7 @@ def test_broken_token_timestamp(self):
         request_handler = mock_request_handler(ip='192.168.21.13')
         request_handler.set_secure_cookie(COOKIE_KEY, 'something&hello')
 
-        id = IpBasedIdentification([], None).identify(request_handler)
+        id = IpBasedIdentification(TrustedIpValidator([]), None).identify(request_handler)
 
         new_token = request_handler.get_cookie(COOKIE_KEY)
 
@@ -178,7 +179,7 @@ def test_old_token_timestamp(self):
         request_handler = mock_request_handler(ip='192.168.21.13')
         request_handler.set_secure_cookie(COOKIE_KEY, 'something&100000')
 
-        id = IpBasedIdentification([], None).identify(request_handler)
+        id = IpBasedIdentification(TrustedIpValidator([]), None).identify(request_handler)
 
         new_token = request_handler.get_cookie(COOKIE_KEY)
 

src/tests/ip_idenfication_test.py

@@ -0,0 +1,31 @@
+from unittest import TestCase
+
+from parameterized import parameterized
+
+from model.trusted_ips import TrustedIpValidator
+
+
+class TestTrustedIpValidator(TestCase):
+    @parameterized.expand([
+        (['192.168.0.15'], '192.168.0.15', True),
+        (['192.168.0.0'], '192.168.0.15', False),
+        (['192.168.0.1'], '192.168.0.15', False),
+        (['127.0.0.1'], '127.0.0.1', True),
+        (['::1'], '::1', True),
+        (['192.168.0.15/32'], '192.168.0.15', True),
+        (['192.168.0.16/32'], '192.168.0.15', False),
+        (['192.168.0.14/31'], '192.168.0.15', True),
+        (['192.168.0.16/31'], '192.168.0.15', False),
+        (['192.168.0.0/28'], '192.168.0.15', True),
+        (['192.168.0.0/28'], '192.168.0.15', True),
+        (['192.168.0.0/28'], '192.168.0.16', False),
+        (['192.168.0.16/28'], '192.168.0.15', False),
+        (['192.168.0.16/28'], '192.168.0.16', True),
+        (['192.168.0.0/24'], '192.168.0.16', True),
+        (['192.168.0.0/16'], '192.168.32.127', True),
+    ])
+    def test_is_trusted(self, configured_ips, user_ip, expected_result):
+        validator = TrustedIpValidator(configured_ips)
+        trusted = validator.is_trusted(user_ip)
+        self.assertEqual(expected_result, trusted, user_ip + ' is trusted=' + str(trusted)
+                         + ' but should be ' + str(expected_result) + ' for ' + str(configured_ips))

src/tests/model/__init__.py

@@ -931,7 +931,7 @@ def init(server_config: ServerConfig,
     if auth.is_enabled():
         identification = AuthBasedIdentification(auth)
     else:
-        identification = IpBasedIdentification(server_config.trusted_ips, server_config.user_header_name)
+        identification = IpBasedIdentification(server_config.ip_validator, server_config.user_header_name)
 
     downloads_folder = file_download_feature.get_result_files_folder()