Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 28 additions & 1 deletion keepercommander/commands/discoveryrotation.py
Original file line number Diff line number Diff line change
Expand Up @@ -3294,7 +3294,34 @@ def execute(self, params, **kwargs):
if rri_status_name == 'RRS_ONLINE':

configuration_uid = utils.base64_url_encode(rri.configurationUid)
gateway_name = rri.controllerName if rri.controllerName else '-'
gateway_name = rri.controllerName
if not gateway_name and rri.controllerUid:
def _normalize_uid(uid):
if uid is None:
return None
if isinstance(uid, (bytes, bytearray)):
return utils.base64_url_encode(uid)
return str(uid)

target_uid = _normalize_uid(rri.controllerUid)
if target_uid is None:
gateway_name = None
else:
try:
all_gateways = gateway_helper.get_all_gateways(params) or []
except (Exception,) as ex:
logging.debug(f"Failed to retrieve gateway list for name resolution: {ex}")
all_gateways = []

if all_gateways:
matched = next((g for g in all_gateways
if _normalize_uid(getattr(g, 'controllerUid', None)) == target_uid), None)
if matched:
gateway_name = getattr(matched, 'controllerName', None)
if gateway_name:
logging.debug(f"Resolved gateway name from controllerUid {target_uid} -> {gateway_name}")

gateway_name = gateway_name if gateway_name else '-'
gateway_uid = utils.base64_url_encode(rri.controllerUid) if rri.controllerUid else '-'

def is_resource_ok(resource_id, params, configuration_uid):
Expand Down
189 changes: 189 additions & 0 deletions unit-tests/pam/test_pam_rotation.py
Original file line number Diff line number Diff line change
Expand Up @@ -860,6 +860,195 @@ def test_table_mode_returns_none(self, mock_rrg, mock_schedules):
result = cmd.execute(mock_params, record_uid=record_uid, format='table')
self.assertIsNone(result)

@patch('keepercommander.commands.discoveryrotation.gateway_helper.get_all_gateways')
@patch('keepercommander.commands.discoveryrotation.router_get_rotation_schedules')
@patch('keepercommander.commands.discoveryrotation.record_rotation_get')
def test_gateway_name_resolved_from_uid_when_empty(self, mock_rrg, mock_schedules, mock_get_gateways):
"""When controllerName is empty, it should be resolved from gateway list using controllerUid."""
from keeper_secrets_manager_core.utils import url_safe_str_to_bytes
record_uid = 'test_record_uid_'
record_uid_bytes = url_safe_str_to_bytes(record_uid)

rri = self._make_rri('RRS_ONLINE')
rri.controllerName = ''

mock_rrg.return_value = rri

sched_mock = MagicMock()
sched_mock.schedules = [self._make_schedule(record_uid_bytes)]
mock_schedules.return_value = sched_mock

mock_gateway = MagicMock()
mock_gateway.controllerUid = rri.controllerUid
mock_gateway.controllerName = 'gw-test-resolved'
mock_get_gateways.return_value = [mock_gateway]

mock_params = create_mock_params()
mock_params.record_cache = {}

cmd = PAMRouterGetRotationInfo()
result = cmd.execute(mock_params, record_uid=record_uid, format='json')

self.assertIsNotNone(result, "Expected JSON string, got None")
data = json.loads(result)
self.assertEqual(data['gateway_name'], 'gw-test-resolved')

@patch('keepercommander.commands.discoveryrotation.gateway_helper.get_all_gateways')
@patch('keepercommander.commands.discoveryrotation.router_get_rotation_schedules')
@patch('keepercommander.commands.discoveryrotation.record_rotation_get')
def test_gateway_name_present_unchanged(self, mock_rrg, mock_schedules, mock_get_gateways):
"""When controllerName is present, it should be used without looking up gateways."""
from keeper_secrets_manager_core.utils import url_safe_str_to_bytes
record_uid = 'test_record_uid_'
record_uid_bytes = url_safe_str_to_bytes(record_uid)

rri = self._make_rri('RRS_ONLINE')
rri.controllerName = 'gw-original'

mock_rrg.return_value = rri

sched_mock = MagicMock()
sched_mock.schedules = [self._make_schedule(record_uid_bytes)]
mock_schedules.return_value = sched_mock

mock_params = create_mock_params()
mock_params.record_cache = {}

cmd = PAMRouterGetRotationInfo()
result = cmd.execute(mock_params, record_uid=record_uid, format='json')

self.assertIsNotNone(result)
data = json.loads(result)
self.assertEqual(data['gateway_name'], 'gw-original')
mock_get_gateways.assert_not_called()

@patch('keepercommander.commands.discoveryrotation.gateway_helper.get_all_gateways')
@patch('keepercommander.commands.discoveryrotation.router_get_rotation_schedules')
@patch('keepercommander.commands.discoveryrotation.record_rotation_get')
def test_gateway_name_empty_no_match_falls_back_to_dash(self, mock_rrg, mock_schedules, mock_get_gateways):
"""When controllerName is empty and no gateway matches, should fall back to '-'."""
from keeper_secrets_manager_core.utils import url_safe_str_to_bytes
record_uid = 'test_record_uid_'
record_uid_bytes = url_safe_str_to_bytes(record_uid)

rri = self._make_rri('RRS_ONLINE')
rri.controllerName = ''

mock_rrg.return_value = rri

sched_mock = MagicMock()
sched_mock.schedules = [self._make_schedule(record_uid_bytes)]
mock_schedules.return_value = sched_mock

mock_gateway = MagicMock()
mock_gateway.controllerUid = b'different_uid_'
mock_gateway.controllerName = 'gw-other'
mock_get_gateways.return_value = [mock_gateway]

mock_params = create_mock_params()
mock_params.record_cache = {}

cmd = PAMRouterGetRotationInfo()
result = cmd.execute(mock_params, record_uid=record_uid, format='json')

self.assertIsNotNone(result)
data = json.loads(result)
self.assertEqual(data['gateway_name'], '-')

@patch('keepercommander.commands.discoveryrotation.gateway_helper.get_all_gateways')
@patch('keepercommander.commands.discoveryrotation.router_get_rotation_schedules')
@patch('keepercommander.commands.discoveryrotation.record_rotation_get')
def test_gateway_name_resolved_with_different_uid_types(self, mock_rrg, mock_schedules, mock_get_gateways):
"""When controllerUid types differ (bytes vs string), should still resolve correctly."""
from keeper_secrets_manager_core.utils import url_safe_str_to_bytes
from keepercommander import utils
record_uid = 'test_record_uid_'
record_uid_bytes = url_safe_str_to_bytes(record_uid)

rri = self._make_rri('RRS_ONLINE')
rri.controllerName = ''

mock_rrg.return_value = rri

sched_mock = MagicMock()
sched_mock.schedules = [self._make_schedule(record_uid_bytes)]
mock_schedules.return_value = sched_mock

mock_gateway = MagicMock()
mock_gateway.controllerUid = utils.base64_url_encode(rri.controllerUid)
mock_gateway.controllerName = 'gw-test-resolved'
mock_get_gateways.return_value = [mock_gateway]

mock_params = create_mock_params()
mock_params.record_cache = {}

cmd = PAMRouterGetRotationInfo()
result = cmd.execute(mock_params, record_uid=record_uid, format='json')

self.assertIsNotNone(result)
data = json.loads(result)
self.assertEqual(data['gateway_name'], 'gw-test-resolved')

@patch('keepercommander.commands.discoveryrotation.gateway_helper.get_all_gateways')
@patch('keepercommander.commands.discoveryrotation.router_get_rotation_schedules')
@patch('keepercommander.commands.discoveryrotation.record_rotation_get')
def test_gateway_list_returns_none_falls_back_gracefully(self, mock_rrg, mock_schedules, mock_get_gateways):
"""When get_all_gateways returns None, should fall back to '-' gracefully."""
from keeper_secrets_manager_core.utils import url_safe_str_to_bytes
record_uid = 'test_record_uid_'
record_uid_bytes = url_safe_str_to_bytes(record_uid)

rri = self._make_rri('RRS_ONLINE')
rri.controllerName = ''

mock_rrg.return_value = rri

sched_mock = MagicMock()
sched_mock.schedules = [self._make_schedule(record_uid_bytes)]
mock_schedules.return_value = sched_mock

mock_get_gateways.return_value = None

mock_params = create_mock_params()
mock_params.record_cache = {}

cmd = PAMRouterGetRotationInfo()
result = cmd.execute(mock_params, record_uid=record_uid, format='json')

self.assertIsNotNone(result)
data = json.loads(result)
self.assertEqual(data['gateway_name'], '-')

@patch('keepercommander.commands.discoveryrotation.gateway_helper.get_all_gateways')
@patch('keepercommander.commands.discoveryrotation.router_get_rotation_schedules')
@patch('keepercommander.commands.discoveryrotation.record_rotation_get')
def test_gateway_list_raises_exception_falls_back_gracefully(self, mock_rrg, mock_schedules, mock_get_gateways):
"""When get_all_gateways raises an exception, should fall back to '-' without crashing."""
from keeper_secrets_manager_core.utils import url_safe_str_to_bytes
record_uid = 'test_record_uid_'
record_uid_bytes = url_safe_str_to_bytes(record_uid)

rri = self._make_rri('RRS_ONLINE')
rri.controllerName = ''

mock_rrg.return_value = rri

sched_mock = MagicMock()
sched_mock.schedules = [self._make_schedule(record_uid_bytes)]
mock_schedules.return_value = sched_mock

mock_get_gateways.side_effect = RuntimeError("Gateway service unavailable")

mock_params = create_mock_params()
mock_params.record_cache = {}

cmd = PAMRouterGetRotationInfo()
result = cmd.execute(mock_params, record_uid=record_uid, format='json')

self.assertIsNotNone(result)
data = json.loads(result)
self.assertEqual(data['gateway_name'], '-')


class TestUsesDefaultRotationSchedule(unittest.TestCase):

Expand Down