Fix IAM user link permission-check rotation request shape.

keepercommander/commands/tunnel/port_forward/TunnelGraph.py

@@ -379,8 +379,18 @@ def _permission_check_iam_user_link(self, user_uid):
         server enforces edit-access on the pamUser record at the call boundary, so 
         the per-flag check that other Layer-B endpoints do is not needed here."""
         from ...pam.router_helper import router_set_record_rotation_information
+        current_record_rotation = self.params.record_rotation_cache.get(user_uid)
+        revision = (
+            current_record_rotation.get('revision', 0)
+            if current_record_rotation else 0
+        )
+        # IAM link: resourceUid must stay empty so krouter sets isIAM=true
+        # (resourceUid.isEmpty && noop=False). Never copy resource_uid from cache.
         rq = router_pb2.RouterRecordRotationRequest(
             recordUid=url_safe_str_to_bytes(user_uid),
+            configurationUid=url_safe_str_to_bytes(self.record.record_uid),
+            revision=revision,
+            resourceUid=b'',
             noop=False,
         )
         router_set_record_rotation_information(self.params, rq)

unit-tests/pam/test_dag_layer_b_migration.py

@@ -22,6 +22,7 @@
 from unittest.mock import MagicMock, patch
 
 import pytest
+from keeper_secrets_manager_core.utils import url_safe_str_to_bytes
 
 sys.path.insert(0, os.path.dirname(__file__))
 
@@ -697,6 +698,7 @@ def _build_tg():
 
         tg = TunnelDAG.__new__(TunnelDAG)
         tg.params = MagicMock()
+        tg.params.record_rotation_cache = {}
         tg.record = MagicMock()
         tg.record.record_uid = CONFIG_UID_STR
         tg.linking_dag = MagicMock()
@@ -723,6 +725,7 @@ def _capture(params, rq, *args, **kwargs):
         assert isinstance(rq, router_pb2.RouterRecordRotationRequest)
         # recordUid is the pamUser record (the target of the permission check)
         assert len(rq.recordUid) == 16
+        assert rq.configurationUid == url_safe_str_to_bytes(CONFIG_UID_STR)
         # noop=False is what triggers the is_iam_user write server-side
         assert rq.noop is False
         # NO resourceUid, NO saasConfiguration — these would change the semantics
@@ -731,6 +734,52 @@ def _capture(params, rq, *args, **kwargs):
         # After permission check succeeds, the legacy link_user mutation still runs
         link_user_mock.assert_called_once()
 
+    def test_link_user_to_config_passes_revision_from_rotation_cache(self):
+        """When pamUser already has rotation metadata, KeeperApp requires matching revision."""
+        tg, _ = self._build_tg()
+        tg.params.record_rotation_cache = {
+            USER_UID_STR: {'revision': 3, 'configuration_uid': CONFIG_UID_STR},
+        }
+        captured = {}
+
+        def _capture(params, rq, *args, **kwargs):
+            captured['rq'] = rq
+
+        with patch('keepercommander.commands.pam.router_helper.router_set_record_rotation_information',
+                   side_effect=_capture), \
+             patch.object(tg, 'link_user'):
+            tg.link_user_to_config(USER_UID_STR)
+
+        rq = captured.get('rq')
+        assert rq is not None
+        assert rq.revision == 3
+        assert rq.resourceUid == b''
+
+    def test_link_user_to_config_clears_stale_resource_uid_from_rotation_cache(self):
+        """IAM permission-check must not send cached resource_uid (non-IAM semantics)."""
+        tg, _ = self._build_tg()
+        tg.params.record_rotation_cache = {
+            USER_UID_STR: {
+                'revision': 2,
+                'configuration_uid': CONFIG_UID_STR,
+                'resource_uid': 'AAAAAAAAAAAAAAAAAAAAAA',
+            },
+        }
+        captured = {}
+
+        def _capture(params, rq, *args, **kwargs):
+            captured['rq'] = rq
+
+        with patch('keepercommander.commands.pam.router_helper.router_set_record_rotation_information',
+                   side_effect=_capture), \
+             patch.object(tg, 'link_user'):
+            tg.link_user_to_config(USER_UID_STR)
+
+        rq = captured.get('rq')
+        assert rq is not None
+        assert rq.revision == 2
+        assert rq.resourceUid == b''
+
     def test_link_user_to_config_no_fallback_on_permission_denial(self):
         """If set_record_rotation fails (permission denied), DO NOT fall back to
         the legacy DAG write - propagate so the unauthorized link is never
@@ -770,6 +819,8 @@ def _capture(params, rq, *args, **kwargs):
         rq = captured.get('rq')
         assert rq is not None
         assert isinstance(rq, router_pb2.RouterRecordRotationRequest)
+        assert rq.configurationUid == url_safe_str_to_bytes(CONFIG_UID_STR)
+        assert rq.resourceUid == b''
         assert rq.noop is False
 
     def test_link_user_to_config_with_options_iam_user_not_true_skips_set_rotation(self):