AAP-25344: AAP 2.5 Gateway SSO for Lightspeed (#1299)

ansible_ai_connect/users/auth.py

@@ -52,7 +52,7 @@ def get_user_details(self, response):
 
  def user_data(self, access_token, *args, **kwargs):
  """Loads user data from service"""
- url = f"{settings.AAP_API_URL}/v2/me/"
+ url = self.get_me_endpoint(settings.AAP_API_URL)
  resp_data = self.get_json(url, headers={"Authorization": f"bearer {access_token}"})
  return resp_data.get("results")[0]
 
@@ -67,10 +67,24 @@ def extra_data(self, user, uid, response, details=None, *args, **kwargs):
  return data
 
  def user_has_valid_license(self, access_token):
- url = f"{settings.AAP_API_URL}/v2/config/"
+ url = self.get_config_endpoint(settings.AAP_API_URL)
  data = self.get_json(url, headers={"Authorization": f"bearer {access_token}"})
  return not data["license_info"]["date_expired"] if "license_info" in data else False
 
+ def get_me_endpoint(self, api_url):
+ """Creates me link to the AAP API depending on the Auth platform"""
+
+ # AAP Controller has /api at the end for API link, AAP Gateway doesn't
+ url = api_url.rstrip("/")
+ return f"{url}/v2/me/" if url.endswith("/api") else f"{url}/api/gateway/v1/me/"
+
+ def get_config_endpoint(self, api_url):
+ """Creates config link to the AAP API depending on the Auth platform"""
+
+ # AAP Controller has /api at the end for API link, AAP Gateway doesn't
+ url = api_url.rstrip("/")
+ return f"{url}/v2/config/" if url.endswith("/api") else f"{url}/api/controller/v2/config/"
+
 
 class RHSSOAuthentication(authentication.BaseAuthentication):
  """Red Hat SSO Access Token authentication backend"""

ansible_ai_connect/users/tests/test_auth.py

@@ -68,7 +68,8 @@ class TestAAPOAuth2(WisdomServiceLogAwareTestCase):
  }
  ),
  )
- def test_date_expired_checked_and_is_true_during_auth(self):
+ @patch("django.conf.settings.AAP_API_URL")
+ def test_date_expired_checked_and_is_true_during_auth(self, AAP_API_URL):
  self.authentication = AAPOAuth2()
  user = MagicMock()
  response = {"is_system_auditor": True, "is_superuser": True}
@@ -87,7 +88,8 @@ def test_date_expired_checked_and_is_true_during_auth(self):
  }
  ),
  )
- def test_date_expired_checked_and_is_false_during_auth(self):
+ @patch("django.conf.settings.AAP_API_URL")
+ def test_date_expired_checked_and_is_false_during_auth(self, AAP_API_URL):
  self.authentication = AAPOAuth2()
  user = MagicMock()
  response = {"is_system_auditor": False, "is_superuser": False}
@@ -102,7 +104,8 @@ def test_date_expired_checked_and_is_false_during_auth(self):
  extra_data=MagicMock(return_value={"test": "data"}),
  get_json=MagicMock(return_value={}),
  )
- def test_missing_values(self):
+ @patch("django.conf.settings.AAP_API_URL")
+ def test_missing_values(self, AAP_API_URL):
  self.authentication = AAPOAuth2()
  user = MagicMock()
  response = {}
@@ -112,6 +115,50 @@ def test_missing_values(self):
  self.assertFalse(data["aap_system_auditor"])
  self.assertFalse(data["aap_superuser"])
 
+ def test_get_me_endpoint_controller(self):
+ authentication = AAPOAuth2()
+ api_url = "http://controller.test/api"
+ self.assertEqual(
+ "http://controller.test/api/v2/me/", authentication.get_me_endpoint(api_url)
+ )
+
+ def test_get_me_endpoint_controller_ended(self):
+ authentication = AAPOAuth2()
+ api_url = "http://controller.test/api/"
+ self.assertEqual(
+ "http://controller.test/api/v2/me/", authentication.get_me_endpoint(api_url)
+ )
+
+ def test_get_me_endpoint_gateway(self):
+ authentication = AAPOAuth2()
+ api_url = "http://controller.test"
+ self.assertEqual(
+ "http://controller.test/api/gateway/v1/me/", authentication.get_me_endpoint(api_url)
+ )
+
+ def test_get_config_endpoint_controller(self):
+ authentication = AAPOAuth2()
+ api_url = "http://controller.test/api"
+ self.assertEqual(
+ "http://controller.test/api/v2/config/", authentication.get_config_endpoint(api_url)
+ )
+
+ def test_get_config_endpoint_gateway(self):
+ authentication = AAPOAuth2()
+ api_url = "http://controller.test"
+ self.assertEqual(
+ "http://controller.test/api/controller/v2/config/",
+ authentication.get_config_endpoint(api_url),
+ )
+
+ def test_get_config_endpoint_gateway_ended(self):
+ authentication = AAPOAuth2()
+ api_url = "http://controller.test/"
+ self.assertEqual(
+ "http://controller.test/api/controller/v2/config/",
+ authentication.get_config_endpoint(api_url),
+ )
+
 
 class TestRHSSOAuthentication(WisdomServiceLogAwareTestCase):
  def setUp(self):