#444 fixed case sensitive user access to script logs

src/auth/authorization.py

@@ -7,7 +7,7 @@
 
 def _normalize_user(user):
     if user:
-        return user.lower()
+        return user.lower().strip()
     return user
 
 
@@ -186,3 +186,7 @@ def _exclude_unknown_groups_from_admin_users(admin_users, known_groups):
         result.append(user)
 
     return result
+
+
+def is_same_user(user_id1, user_id2):
+    return _normalize_user(user_id1) == _normalize_user(user_id2)

src/execution/execution_service.py

@@ -2,7 +2,7 @@
 from collections import namedtuple
 from typing import Optional, Dict, Callable, Any
 
-from auth.authorization import Authorizer
+from auth.authorization import Authorizer, is_same_user
 from auth.user import User
 from execution.executor import ScriptExecutor
 from model import script_config
@@ -140,7 +140,7 @@ def validate_execution_id(self, execution_id, user, only_active=True, allow_when
 
     @staticmethod
     def _can_access_execution(execution_info: _ExecutionInfo, user_id):
-        return (execution_info is not None) and (execution_info.owner_user.user_id == user_id)
+        return (execution_info is not None) and (is_same_user(execution_info.owner_user.user_id, user_id))
 
     def get_user_parameter_values(self, execution_id):
         return self._get_for_executor(execution_id,

src/execution/logging.py

@@ -4,6 +4,7 @@
 import re
 from string import Template
 
+from auth.authorization import is_same_user
 from execution.execution_service import ExecutionService
 from model.model_helper import AccessProhibitedException
 from utils import file_utils, audit_utils
@@ -332,7 +333,7 @@ def _can_access_entry(self, entry, user_id, system_call=False):
         if entry is None:
             return True
 
-        if entry.user_id == user_id:
+        if is_same_user(entry.user_id, user_id):
             return True
 
         if system_call:

src/tests/execution_logging_test.py

@@ -7,6 +7,8 @@
 from datetime import datetime, timedelta
 from typing import Optional
 
+from parameterized import parameterized
+
 from auth.authorization import Authorizer, EmptyGroupProvider
 from auth.user import User
 from execution import executor
@@ -266,13 +268,17 @@ def test_get_history_entries_after_delete(self):
         entries = self.logging_service.get_history_entries('userX')
         self.assertCountEqual([], entries)
 
-    def test_get_history_entries_only_for_current_user(self):
+    @parameterized.expand([
+        ('userA',),
+        ('USERa',),
+    ])
+    def test_get_history_entries_only_for_current_user(self, user_id):
         self.simulate_logging(execution_id='id1', user_id='userA')
         self.simulate_logging(execution_id='id2', user_id='userB')
         self.simulate_logging(execution_id='id3', user_id='userC')
         self.simulate_logging(execution_id='id4', user_id='userA')
 
-        entries = self._get_entries_sorted('userA')
+        entries = self._get_entries_sorted(user_id)
         self.assertEquals(2, len(entries))
 
         self.validate_history_entry(entry=entries[0], id='id1', user_id='userA')

src/tests/execution_service_test.py

@@ -66,19 +66,27 @@ def test_running_services_when_2_started_and_1_stopped(self):
 
         self.assertCountEqual([id1], execution_service.get_running_executions())
 
-    def test_active_executions_when_2_started(self):
+    @parameterized.expand([
+        (DEFAULT_USER_ID,),
+        (DEFAULT_USER_ID.upper(),),
+    ])
+    def test_active_executions_when_2_started(self, user_id):
         execution_service = self.create_execution_service()
         id1 = self._start(execution_service)
         id2 = self._start(execution_service)
 
-        self.assertCountEqual([id1, id2], execution_service.get_active_executions(DEFAULT_USER_ID))
+        self.assertCountEqual([id1, id2], execution_service.get_active_executions(user_id))
 
-    def test_active_executions_with_different_user(self):
+    @parameterized.expand([
+        ('another_user',),
+        ('ANOTHER_USER',),
+    ])
+    def test_active_executions_with_different_user(self, user_id):
         execution_service = self.create_execution_service()
         self._start(execution_service)
         self._start(execution_service)
 
-        self.assertCountEqual([], execution_service.get_active_executions('another_user'))
+        self.assertCountEqual([], execution_service.get_active_executions(user_id))
 
     def test_active_executions_when_2_started_and_1_cleanup(self):
         execution_service = self.create_execution_service()