Extended coverage with multiple corner-cases. Some tests are disabled because

they either require additonal error handling in the auth module or fail when
run as privileged user in the default docker CI.

Author: jonasbardino

tests/test_mig_shared_auth.py

@@ -28,11 +28,13 @@
 """Unit tests for authentication functionality in mig/shared/auth.py"""
 
 import datetime
+import fcntl
 import http.cookies
 import os
 import pickle
 import time
 import unittest
+from unittest.mock import patch
 
 from tests.support import MigTestCase, testmain, ensure_dirs_exist
 
@@ -45,6 +47,7 @@
 GDP_USER_DN = f'{TEST_USER_DN}/GDP=projectx'
 TEST_CLIENT_PREFIX = \
     '2e1c3d78bddf637ed6b83067c15ac9b9893545ff6a549519178cb4a252ed38b5'
+DEFAULT_INTERVAL = 30
 
 
 class MigSharedAuth__twofactor(MigTestCase):
@@ -299,11 +302,8 @@ def test_custom_token_interval(self):
 
         # Save custom interval
         client_dir = client_id_dir(TEST_USER_DN)
-        interval_path = os.path.join(
-            self.configuration.user_settings,
-            client_dir,
-            'twofactor_interval'
-        )
+        interval_path = os.path.join(self.configuration.user_settings,
+                                     client_dir, 'twofactor_interval')
         with open(interval_path, 'w') as fh:
             fh.write('60')  # 1 minute interval
 
@@ -332,10 +332,9 @@ def test_strict_address_session_handling(self):
         )
 
         # Should have address-linked file
-        addr_file = os.path.join(
-            self.configuration.twofactor_home,
-            f"{user_addr}_{session_key}"
-        )
+        addr_file = os.path.join(self.configuration.twofactor_home,
+                                 f"{user_addr}_{session_key}"
+                                 )
         self.assertTrue(os.path.exists(addr_file),
                         'should create address-linked session file')
 
@@ -367,6 +366,179 @@ def test_check_twofactor_active_missing_session(self):
                                              TEST_USER_DN, '127.0.0.1', environ)
         self.assertFalse(result, 'should reject invalid session cookie')
 
+    def test_session_prefix_hashing_consistency(self):
+        """Test session prefix hash consistency across calls"""
+        client_id = '/C=US/O=Example/CN=Test User'
+        prefix1 = auth.generate_session_prefix(self.configuration, client_id)
+        prefix2 = auth.generate_session_prefix(self.configuration, client_id)
+        self.assertEqual(prefix1, prefix2,
+                         'session prefix should be consistent for same client_id')
+
+        # Verify prefix changes with different client_id
+        prefix3 = auth.generate_session_prefix(
+            self.configuration, TEST_USER_DN)
+        self.assertNotEqual(prefix1, prefix3,
+                            'prefix should change with client_id')
+
+    def test_malformed_session_file_handling(self):
+        """Test graceful handling of corrupted session files"""
+        session_key = auth.generate_session_key(self.configuration,
+                                                TEST_USER_DN)
+        session_path = os.path.join(self.configuration.twofactor_home,
+                                    session_key)
+
+        # Create malformed pickle file
+        with open(session_path, 'wb') as fh:
+            fh.write(b'invalid pickle content')
+
+        # Attempt to load should return empty dict
+        session_data = auth.load_twofactor_session(self.configuration,
+                                                   session_key)
+        self.assertEqual(session_data.get('client_id', ''), 'UNKNOWN',
+                         'should handle malformed session files gracefully')
+        self.assertEqual(session_data.get('user_addr', ''), 'UNKNOWN',
+                         'should handle malformed session files gracefully')
+
+    @ unittest.skipIf(os.getuid() == 0, "Permissions don't work for priv users")
+    def test_session_file_permissions(self):
+        """Test secure session file permissions"""
+        session_key = auth.generate_session_key(self.configuration,
+                                                TEST_USER_DN)
+        user_addr = '192.168.1.1'
+
+        auth.save_twofactor_session(self.configuration, TEST_USER_DN,
+                                    session_key, user_addr, 'TestAgent',
+                                    time.time())
+
+        session_path = os.path.join(self.configuration.twofactor_home,
+                                    session_key)
+        mode = os.stat(session_path).st_mode
+        # TODO: tighten permissions in tested function and enable next
+        # expected = 0o600
+        expected = 0o644
+        self.assertEqual(mode & 0o777, expected,
+                         'session files should have restrictive permissions')
+
+    def test_gdp_oid_alias_handling(self):
+        """Test GDP client ID with OID alias expansion"""
+        self.configuration.site_enable_gdp = True
+        self.configuration.user_openid_alias = 'email'
+
+        # Generate session key should resolve to base GDP ID
+        session_key = auth.generate_session_key(self.configuration,
+                                                TEST_USER_DN)
+        base_prefix = auth.generate_session_prefix(self.configuration,
+                                                   GDP_USER_DN)
+        self.assertTrue(session_key.startswith(base_prefix),
+                        'should resolve OID alias to base GDP ID')
+
+    def test_token_window_boundaries(self):
+        """Test token validation at window boundaries"""
+        self._provision_test_user(self, TEST_USER_DN)
+        b32_key, _, _ = auth.get_twofactor_secrets(self.configuration,
+                                                   TEST_USER_DN)
+
+        # Create TOTP with small test window
+        with patch.object(auth, 'valid_otp_window', 1):
+            totp = auth.get_totp(TEST_USER_DN, b32_key, self.configuration)
+
+            # Generate token slightly outside window
+            test_time = datetime.datetime.fromtimestamp(
+                int(time.time() - DEFAULT_INTERVAL * 3))
+            boundary_token = totp.generate_otp(totp.timecode(test_time))
+
+            # Verify should accept within window but reject outside
+            is_valid = auth.verify_twofactor_token(self.configuration, TEST_USER_DN,
+                                                   b32_key, boundary_token)
+            self.assertFalse(is_valid, 'should reject tokens outside window')
+
+            # Generate valid boundary token
+            test_time = datetime.datetime.fromtimestamp(
+                int(time.time() - DEFAULT_INTERVAL * 1))
+
+            valid_boundary = totp.generate_otp(totp.timecode(test_time))
+            is_valid = auth.verify_twofactor_token(self.configuration, TEST_USER_DN,
+                                                   b32_key, valid_boundary)
+            self.assertTrue(is_valid, 'should accept tokens within window')
+
+    def test_nested_session_directories(self):
+        """Test session listing with nested directories"""
+        session_key = auth.generate_session_key(
+            self.configuration, TEST_USER_DN)
+
+        # Create session in subdirectory
+        nested_path = os.path.join(self.configuration.twofactor_home, 'subdir')
+        ensure_dirs_exist(nested_path)
+        session_path = os.path.join(nested_path, session_key)
+
+        with open(session_path, 'wb') as fh:
+            pickle.dump({'client_id': TEST_USER_DN}, fh)
+
+        # Should ignore nested structure
+        sessions = auth.list_twofactor_sessions(
+            self.configuration, TEST_USER_DN)
+        self.assertEqual(len(sessions), 0,
+                         'should ignore sessions in subdirectories')
+
+    @unittest.skip("TODO: enable once tested function handles errors")
+    def test_failed_file_operations(self):
+        """Test graceful handling of file operation failures"""
+        client_dir = self._provision_test_user(self, TEST_USER_DN)
+
+        # Make directory read-only
+        user_settings_path = os.path.join(self.configuration.user_settings,
+                                          client_dir)
+        os.chmod(user_settings_path, 0o555)
+
+        # Attempt key generation should fail
+        with self.assertRaises(Exception) as cm:
+            auth.get_twofactor_secrets(self.configuration, TEST_USER_DN)
+        self.assertIn('failed to reset 2FA key', str(cm.exception))
+
+    @ unittest.skip("TODO: implement locking and enable")
+    def test_session_lock_conflicts(self):
+        """Test concurrent session file access"""
+        session_key = auth.generate_session_key(self.configuration,
+                                                TEST_USER_DN)
+        session_path = os.path.join(self.configuration.twofactor_home,
+                                    session_key)
+
+        # Create test session
+        auth.save_twofactor_session(self.configuration, TEST_USER_DN,
+                                    session_key, '127.0.0.1', 'TestAgent',
+                                    time.time())
+
+        # Lock session file
+        with open(session_path, 'rb') as fh:
+            fcntl.flock(fh, fcntl.LOCK_EX | fcntl.LOCK_NB)
+
+            # Attempt loading should handle locked file
+            session_data = auth.load_twofactor_session(
+                self.configuration, session_key)
+            self.assertEqual(session_data, {},
+                             'should return empty data for locked files')
+
+    def test_custom_interval_fallback(self):
+        """Test default interval fallback when custom fails"""
+        self._provision_test_user(self, TEST_USER_DN)
+
+        # Save invalid interval
+        client_dir = client_id_dir(TEST_USER_DN)
+        interval_path = os.path.join(self.configuration.user_settings,
+                                     client_dir, 'twofactor_interval')
+        with open(interval_path, 'w') as fh:
+            fh.write('invalid-number')
+
+        b32_key, _, _ = auth.get_twofactor_secrets(
+            self.configuration, TEST_USER_DN)
+        totp = auth.get_totp(TEST_USER_DN, b32_key, self.configuration)
+        token = totp.now()
+
+        # Verification should fallback to default interval
+        result = auth.verify_twofactor_token(
+            self.configuration, TEST_USER_DN, b32_key, token)
+        self.assertTrue(result, 'should fallback to default interval')
+
 
 if __name__ == '__main__':
     testmain()