diff --git a/awx/main/tests/unit/utils/test_validate_rh.py b/awx/main/tests/unit/utils/test_validate_rh.py index 65052bbdef..df4adbc44f 100644 --- a/awx/main/tests/unit/utils/test_validate_rh.py +++ b/awx/main/tests/unit/utils/test_validate_rh.py @@ -1,3 +1,6 @@ +import configparser + +import pytest from unittest.mock import patch from awx.main.utils.licensing import Licenser @@ -6,21 +9,25 @@ def test_validate_rh_basic_auth_rhsm(): """ Assert get_rhsm_subs is called when - basic_auth=True + - REDHAT_CANDLEPIN_HOST is not set - host is subscription.rhsm.redhat.com """ licenser = Licenser() - with patch.object(licenser, 'get_host_from_rhsm_config', return_value='https://subscription.rhsm.redhat.com') as mock_get_host, patch.object( - licenser, 'get_rhsm_subs', return_value=[] - ) as mock_get_rhsm, patch.object(licenser, 'get_satellite_subs') as mock_get_satellite, patch.object( + with patch('awx.main.utils.licensing.settings') as mock_settings, patch.object( + licenser, 'get_host_from_rhsm_config', return_value='https://subscription.rhsm.redhat.com' + ) as mock_get_host, patch.object(licenser, 'get_rhsm_subs', return_value=[]) as mock_get_rhsm, patch.object( + licenser, 'get_satellite_subs' + ) as mock_get_satellite, patch.object( licenser, 'get_crc_subs' ) as mock_get_crc, patch.object( licenser, 'generate_license_options_from_entitlements' ) as mock_generate: + mock_settings.REDHAT_CANDLEPIN_HOST = None + licenser.validate_rh('testuser', 'testpass', basic_auth=True) - # Assert the correct methods were called mock_get_host.assert_called_once() mock_get_rhsm.assert_called_once_with('https://subscription.rhsm.redhat.com', 'testuser', 'testpass') mock_get_satellite.assert_not_called() @@ -32,21 +39,25 @@ def test_validate_rh_basic_auth_satellite(): """ Assert get_satellite_subs is called when - basic_auth=True - - custom satellite host + - REDHAT_CANDLEPIN_HOST is not set + - rhsm.conf points to a non-RHSM host """ licenser = Licenser() - with patch.object(licenser, 'get_host_from_rhsm_config', return_value='https://satellite.example.com') as mock_get_host, patch.object( - licenser, 'get_rhsm_subs' - ) as mock_get_rhsm, patch.object(licenser, 'get_satellite_subs', return_value=[]) as mock_get_satellite, patch.object( + with patch('awx.main.utils.licensing.settings') as mock_settings, patch.object( + licenser, 'get_host_from_rhsm_config', return_value='https://satellite.example.com' + ) as mock_get_host, patch.object(licenser, 'get_rhsm_subs') as mock_get_rhsm, patch.object( + licenser, 'get_satellite_subs', return_value=[] + ) as mock_get_satellite, patch.object( licenser, 'get_crc_subs' ) as mock_get_crc, patch.object( licenser, 'generate_license_options_from_entitlements' ) as mock_generate: + mock_settings.REDHAT_CANDLEPIN_HOST = None + licenser.validate_rh('testuser', 'testpass', basic_auth=True) - # Assert the correct methods were called mock_get_host.assert_called_once() mock_get_rhsm.assert_not_called() mock_get_satellite.assert_called_once_with('https://satellite.example.com', 'testuser', 'testpass') @@ -73,7 +84,6 @@ def test_validate_rh_service_account_crc(): licenser.validate_rh('client_id', 'client_secret', basic_auth=False) - # Assert the correct methods were called mock_get_host.assert_not_called() mock_get_rhsm.assert_not_called() mock_get_satellite.assert_not_called() @@ -81,74 +91,127 @@ def test_validate_rh_service_account_crc(): mock_generate.assert_called_once_with([], is_candlepin=False) +def test_validate_rh_candlepin_host_prioritized_over_rhsm_config(): + """Test REDHAT_CANDLEPIN_HOST takes priority over rhsm.conf + - basic_auth=True + - REDHAT_CANDLEPIN_HOST is set + - rhsm.conf should NOT be consulted + """ + licenser = Licenser() + + with patch('awx.main.utils.licensing.settings') as mock_settings, patch.object(licenser, 'get_host_from_rhsm_config') as mock_get_host, patch.object( + licenser, 'get_rhsm_subs' + ) as mock_get_rhsm, patch.object(licenser, 'get_satellite_subs', return_value=[]) as mock_get_satellite, patch.object( + licenser, 'get_crc_subs' + ) as mock_get_crc, patch.object( + licenser, 'generate_license_options_from_entitlements' + ) as mock_generate: + + mock_settings.REDHAT_CANDLEPIN_HOST = 'https://satellite.example.com' + licenser.validate_rh('testuser', 'testpass', basic_auth=True) + + mock_get_host.assert_not_called() + mock_get_rhsm.assert_not_called() + mock_get_satellite.assert_called_once_with('https://satellite.example.com', 'testuser', 'testpass') + mock_get_crc.assert_not_called() + mock_generate.assert_called_once_with([], is_candlepin=True) + + +def test_validate_rh_prepends_scheme_when_missing(): + """REDHAT_CANDLEPIN_HOST without a scheme gets https:// prepended""" + licenser = Licenser() + + with patch('awx.main.utils.licensing.settings') as mock_settings, patch.object(licenser, 'get_host_from_rhsm_config'), patch.object( + licenser, 'get_rhsm_subs' + ) as mock_get_rhsm, patch.object(licenser, 'get_satellite_subs', return_value=[]) as mock_get_satellite, patch.object( + licenser, 'get_crc_subs' + ), patch.object( + licenser, 'generate_license_options_from_entitlements' + ): + + mock_settings.REDHAT_CANDLEPIN_HOST = 'satellite.example.com' + licenser.validate_rh('testuser', 'testpass', basic_auth=True) + + mock_get_satellite.assert_called_once_with('https://satellite.example.com', 'testuser', 'testpass') + mock_get_rhsm.assert_not_called() + + def test_validate_rh_missing_user_raises_error(): """Test validate_rh raises ValueError when user is missing""" licenser = Licenser() - with patch.object(licenser, 'get_host_from_rhsm_config', return_value='https://subscription.rhsm.redhat.com'): - try: + with patch('awx.main.utils.licensing.settings') as mock_settings, patch.object( + licenser, 'get_host_from_rhsm_config', return_value='https://subscription.rhsm.redhat.com' + ): + mock_settings.REDHAT_CANDLEPIN_HOST = None + with pytest.raises(ValueError, match='subscriptions_client_id or subscriptions_username is required'): licenser.validate_rh(None, 'testpass', basic_auth=True) - assert False, "Expected ValueError to be raised" - except ValueError as e: - assert 'subscriptions_client_id or subscriptions_username is required' in str(e) def test_validate_rh_missing_password_raises_error(): """Test validate_rh raises ValueError when password is missing""" licenser = Licenser() - with patch.object(licenser, 'get_host_from_rhsm_config', return_value='https://subscription.rhsm.redhat.com'): - try: + with patch('awx.main.utils.licensing.settings') as mock_settings, patch.object( + licenser, 'get_host_from_rhsm_config', return_value='https://subscription.rhsm.redhat.com' + ): + mock_settings.REDHAT_CANDLEPIN_HOST = None + with pytest.raises(ValueError, match='subscriptions_client_secret or subscriptions_password is required'): licenser.validate_rh('testuser', None, basic_auth=True) - assert False, "Expected ValueError to be raised" - except ValueError as e: - assert 'subscriptions_client_secret or subscriptions_password is required' in str(e) -def test_validate_rh_no_host_fallback_to_candlepin(): - """Test validate_rh falls back to REDHAT_CANDLEPIN_HOST when no host from config +@pytest.mark.parametrize( + 'host_input, rhsm_port, expected_in_url, not_expected_in_url', + [ + ('https://satellite.example.com:8443', '443', ':8443', ':8443:443'), + ('https://satellite.example.com', '8443', ':8443', None), + ('https://satellite.example.com/', '8443', ':8443', '/:'), + ], + ids=['skip-port-when-present', 'append-port-when-missing', 'strip-trailing-slash'], +) +def test_get_satellite_subs_port_handling(host_input, rhsm_port, expected_in_url, not_expected_in_url): + licenser = Licenser() + licenser.config = configparser.ConfigParser() + licenser.config.read_string(f"[server]\nhostname=satellite.example.com\nport={rhsm_port}\n[rhsm]\nrepo_ca_cert=/etc/rhsm/ca/redhat-uep.pem\n") + + with patch('awx.main.utils.licensing.settings') as mock_settings, patch('awx.main.utils.licensing.requests') as mock_requests: + mock_settings.REDHAT_CANDLEPIN_VERIFY = None + mock_orgs = mock_requests.get.return_value + mock_orgs.json.return_value = {'results': []} + + licenser.get_satellite_subs(host_input, 'user', 'pw') + + called_url = mock_requests.get.call_args[0][0] + assert expected_in_url in called_url + if not_expected_in_url: + assert not_expected_in_url not in called_url + + +def test_get_satellite_subs_uses_candlepin_verify_setting(): + """REDHAT_CANDLEPIN_VERIFY should take priority over rhsm.conf ca_cert""" + licenser = Licenser() + licenser.config = configparser.ConfigParser() + licenser.config.read_string("[server]\nhostname=satellite.example.com\n[rhsm]\nrepo_ca_cert=/etc/rhsm/ca/redhat-uep.pem\n") + + with patch('awx.main.utils.licensing.settings') as mock_settings, patch('awx.main.utils.licensing.requests') as mock_requests: + mock_settings.REDHAT_CANDLEPIN_VERIFY = False + mock_orgs = mock_requests.get.return_value + mock_orgs.json.return_value = {'results': []} + + licenser.get_satellite_subs('https://satellite.example.com', 'user', 'pw') + + assert mock_requests.get.call_args[1]['verify'] is False + + +def test_validate_rh_no_host_raises_error(): + """Test validate_rh raises ValueError when no host is available - basic_auth=True - - no host from config - - REDHAT_CANDLEPIN_HOST is set + - REDHAT_CANDLEPIN_HOST is not set + - rhsm.conf returns None """ licenser = Licenser() - with patch('awx.main.utils.licensing.settings') as mock_settings, patch.object( - licenser, 'get_host_from_rhsm_config', return_value=None - ) as mock_get_host, patch.object(licenser, 'get_rhsm_subs', return_value=[]) as mock_get_rhsm, patch.object( - licenser, 'get_satellite_subs', return_value=[] - ) as mock_get_satellite, patch.object( - licenser, 'get_crc_subs' - ) as mock_get_crc, patch.object( - licenser, 'generate_license_options_from_entitlements' - ) as mock_generate: - - mock_settings.REDHAT_CANDLEPIN_HOST = 'https://candlepin.example.com' - licenser.validate_rh('testuser', 'testpass', basic_auth=True) - - # Assert the correct methods were called - mock_get_host.assert_called_once() - mock_get_rhsm.assert_not_called() - mock_get_satellite.assert_called_once_with('https://candlepin.example.com', 'testuser', 'testpass') - mock_get_crc.assert_not_called() - mock_generate.assert_called_once_with([], is_candlepin=True) - - -def test_validate_rh_empty_credentials_basic_auth(): - """Test validate_rh with empty string credentials raises ValueError""" - licenser = Licenser() - - with patch.object(licenser, 'get_host_from_rhsm_config', return_value='https://subscription.rhsm.redhat.com'): - # Test empty user - try: - licenser.validate_rh(None, 'testpass', basic_auth=True) - assert False, "Expected ValueError to be raised" - except ValueError as e: - assert 'subscriptions_client_id or subscriptions_username is required' in str(e) - - # Test empty password - try: - licenser.validate_rh('testuser', None, basic_auth=True) - assert False, "Expected ValueError to be raised" - except ValueError as e: - assert 'subscriptions_client_secret or subscriptions_password is required' in str(e) + with patch('awx.main.utils.licensing.settings') as mock_settings, patch.object(licenser, 'get_host_from_rhsm_config', return_value=None): + mock_settings.REDHAT_CANDLEPIN_HOST = None + with pytest.raises(ValueError, match='Could not get host url for subscriptions'): + licenser.validate_rh('testuser', 'testpass', basic_auth=True) diff --git a/awx/main/utils/licensing.py b/awx/main/utils/licensing.py index 20417940b8..b95dcdc32e 100644 --- a/awx/main/utils/licensing.py +++ b/awx/main/utils/licensing.py @@ -19,6 +19,7 @@ import json import logging import re import requests +from urllib.parse import urlparse import time import zipfile @@ -228,18 +229,16 @@ class Licenser(object): return host def validate_rh(self, user, pw, basic_auth): - # if basic auth is True, host is read from rhsm.conf (subscription.rhsm.redhat.com) - # if basic auth is False, host is settings.SUBSCRIPTIONS_RHSM_URL (console.redhat.com) - # if rhsm.conf is not found, host is settings.REDHAT_CANDLEPIN_HOST (satellite server) if basic_auth: - host = self.get_host_from_rhsm_config() - if not host: - host = getattr(settings, 'REDHAT_CANDLEPIN_HOST', None) + if not (host := getattr(settings, 'REDHAT_CANDLEPIN_HOST', None)): + host = self.get_host_from_rhsm_config() else: host = settings.SUBSCRIPTIONS_RHSM_URL if not host: raise ValueError('Could not get host url for subscriptions') + if not host.startswith(('https://', 'http://')): + host = 'https://' + host if not user: raise ValueError('subscriptions_client_id or subscriptions_username is required') @@ -308,13 +307,20 @@ class Licenser(object): def get_satellite_subs(self, host, user, pw): port = None + if (verify := getattr(settings, 'REDHAT_CANDLEPIN_VERIFY', None)) is None: + try: + verify = str(self.config.get("rhsm", "repo_ca_cert")) + except Exception as e: + logger.exception(f'Unable to read rhsm config to get ca_cert location. {e}') + verify = True try: - verify = str(self.config.get("rhsm", "repo_ca_cert")) port = str(self.config.get("server", "port")) - except Exception as e: - logger.exception('Unable to read rhsm config to get ca_cert location. {}'.format(str(e))) - verify = True - if port: + except Exception: + port = None + host = host.rstrip('/') + # Append port from rhsm.conf only if the host URL doesn't already include one + # (REDHAT_CANDLEPIN_HOST may already contain a port) + if port and not urlparse(host).port: host = ':'.join([host, port]) json = [] try: