AAP-38623: Internal RH user check.

Signed-off-by: romartin <[email protected]>

ansible_ai_connect/ai/api/views.py

@@ -97,7 +97,7 @@
 )
 from ansible_ai_connect.users.models import User
 
-from ...main.permissions import IsRHEmployee, IsTestUser
+from ...main.permissions import IsRHInternalUser, IsTestUser
 from ...users.throttling import EndpointRateThrottle
 from ..feature_flags import FeatureFlags
 from .data.data_model import ContentMatchPayloadData, ContentMatchResponseDto
@@ -950,7 +950,7 @@ class ChatEndpointThrottle(EndpointRateThrottle):
     permission_classes = [
         permissions.IsAuthenticated,
         IsAuthenticatedOrTokenHasScope,
-        IsRHEmployee | IsTestUser,
+        IsRHInternalUser | IsTestUser,
     ]
     required_scopes = ["read", "write"]
     schema1_event = schema1.ChatBotOperationalEvent

ansible_ai_connect/main/permissions.py

@@ -14,11 +14,12 @@
 from rest_framework.permissions import BasePermission
 
 
-class IsRHEmployee(BasePermission):
+class IsRHInternalUser(BasePermission):
     """
-    Allow access only to users who are Red Hat employees
+    Allow access only to users who are Red Hat internal users.
     """
 
+    # TODO: Rename? Integ. tests or clients may be affected.
     code = "permission_denied__user_not_rh_employee"
     message = "The User is not a Red Hat employee."
 

ansible_ai_connect/main/tests/test_permissions.py

@@ -17,14 +17,14 @@
 from django.test import RequestFactory, TestCase
 from django.urls import resolve, reverse
 
-from ansible_ai_connect.main.permissions import IsRHEmployee, IsTestUser
+from ansible_ai_connect.main.permissions import IsRHInternalUser, IsTestUser
 
 
-class TestIsRHEmployee(TestCase):
+class TestIsRHInternalUser(TestCase):
     def setUp(self):
         super().setUp()
 
-        self.permission = IsRHEmployee()
+        self.permission = IsRHInternalUser()
 
         payload = {
             "query": "Hello",

ansible_ai_connect/main/views.py

@@ -32,7 +32,7 @@
     IsOrganisationLightspeedSubscriber,
 )
 from ansible_ai_connect.main.base_views import ProtectedTemplateView
-from ansible_ai_connect.main.permissions import IsRHEmployee, IsTestUser
+from ansible_ai_connect.main.permissions import IsRHInternalUser, IsTestUser
 from ansible_ai_connect.main.settings.base import SOCIAL_AUTH_OIDC_KEY
 
 logger = logging.getLogger(__name__)
@@ -116,7 +116,7 @@ class ChatbotView(ProtectedTemplateView):
     template_name = "chatbot/index.html"
 
     permission_classes = [
-        IsRHEmployee | IsTestUser,
+        IsRHInternalUser | IsTestUser,
     ]
 
     def get(self, request):

ansible_ai_connect/users/migrations/0011_user_rh_employee.py

@@ -12,6 +12,7 @@ class Migration(migrations.Migration):
     operations = [
         migrations.AddField(
             model_name="user",
+            # TODO: Another migration script for renaming the field to rh_internal?.
             name="rh_employee",
             field=models.BooleanField(default=False),
         ),

ansible_ai_connect/users/models.py

@@ -61,6 +61,7 @@ class User(ExportModelOperationsMixin("user"), AbstractUser):
         on_delete=models.CASCADE,
     )
     rh_user_is_org_admin = models.BooleanField(default=False)
+    # TODO: Rename to rh_internal?
     rh_employee = models.BooleanField(default=False)
     external_username = models.CharField(default="", null=False)
     name = models.CharField(default=None, null=True)

ansible_ai_connect/users/pipeline.py

@@ -84,6 +84,17 @@ def github_get_username(strategy, details, backend, user=None, *args, **kwargs):
         return get_username(strategy, details, backend, user, *args, **kwargs)
 
 
+def is_rh_email_domain(email):
+    email_domain: str = email[email.find("@") + 1 :] if email else None
+    return email_domain.lower() == "redhat.com" if email_domain else False
+
+
+def is_rh_internal_user(user):
+    return is_rh_email_domain(user.email) and (
+        user.email_verified if user.email_verified is not None else False
+    )
+
+
 def redhat_organization(backend, user, response, *args, **kwargs):
     if backend.name != "oidc":
         return
@@ -106,7 +117,8 @@ def redhat_organization(backend, user, response, *args, **kwargs):
     user.email = payload.get("email")
     user.email_verified = payload.get("email_verified")
     user.rh_user_is_org_admin = "admin:org:all" in roles
-    user.rh_employee = "redhat:employees" in roles
+    # TODO: Refactor by rh_internal?
+    user.rh_employee = is_rh_internal_user(user)
 
     if settings.AUTHZ_BACKEND_TYPE == "dummy":
         if settings.AUTHZ_DUMMY_RH_ORG_ADMINS == "*":

ansible_ai_connect/users/tests/test_pipeline.py

@@ -25,7 +25,11 @@
 
 from ansible_ai_connect.test_utils import WisdomServiceLogAwareTestCase
 from ansible_ai_connect.users.constants import RHSSO_LIGHTSPEED_SCOPE
-from ansible_ai_connect.users.pipeline import load_extra_data, redhat_organization
+from ansible_ai_connect.users.pipeline import (
+    is_rh_email_domain,
+    load_extra_data,
+    redhat_organization,
+)
 
 
 def build_access_token(private_key, payload):
@@ -310,9 +314,10 @@ def test_rh_employee_field(self):
             "access_token": build_access_token(
                 self.rsa_private_key,
                 {
-                    "realm_access": {"roles": ["redhat:employees"]},
                     "preferred_username": "jean-michel",
                     "organization": {"id": "345"},
+                    "email": "[email protected]",
+                    "email_verified": True,
                 },
             )
         }
@@ -324,6 +329,59 @@ def test_rh_employee_field(self):
         )
         self.assertEqual(answer["rh_employee"], True)
 
+    def test_not_rh_employee_fields(self):
+        response = {
+            "access_token": build_access_token(
+                self.rsa_private_key,
+                {
+                    "preferred_username": "jean-michel",
+                    "organization": {"id": "345"},
+                },
+            )
+        }
+        answer = redhat_organization(
+            backend=DummyRHBackend(public_key=self.jwk_public_key),
+            user=self.rh_user,
+            response=response,
+        )
+        self.assertEqual(answer["rh_employee"], False)
+
+    def test_rh_employee_field_not_rh_email(self):
+        response = {
+            "access_token": build_access_token(
+                self.rsa_private_key,
+                {
+                    "email": "[email protected]",
+                    "preferred_username": "jean-michel",
+                    "organization": {"id": "345"},
+                },
+            )
+        }
+        answer = redhat_organization(
+            backend=DummyRHBackend(public_key=self.jwk_public_key),
+            user=self.rh_user,
+            response=response,
+        )
+        self.assertEqual(answer["rh_employee"], False)
+
+    def test_rh_employee_field_not_email_verified(self):
+        response = {
+            "access_token": build_access_token(
+                self.rsa_private_key,
+                {
+                    "email": "[email protected]",
+                    "preferred_username": "jean-michel",
+                    "organization": {"id": "345"},
+                },
+            )
+        }
+        answer = redhat_organization(
+            backend=DummyRHBackend(public_key=self.jwk_public_key),
+            user=self.rh_user,
+            response=response,
+        )
+        self.assertEqual(answer["rh_employee"], False)
+
     def test_rhoss_user_and_email(self):
         response = {
             "access_token": build_access_token(
@@ -348,3 +406,12 @@ def test_rhoss_user_and_email(self):
         self.assertEqual(self.rh_user.email, "[email protected]")
         self.assertEqual(self.rh_user.given_name, "Francis")
         self.assertEqual(self.rh_user.name, "Francis Drake")
+
+    def test_is_rh_email_domain(self):
+        self.assertFalse(is_rh_email_domain(None))
+        self.assertFalse(is_rh_email_domain(""))
+        self.assertFalse(is_rh_email_domain("user.com"))
+        self.assertFalse(is_rh_email_domain("[email protected]"))
+        self.assertTrue(is_rh_email_domain("[email protected]"))
+        self.assertTrue(is_rh_email_domain("[email protected]"))
+        self.assertTrue(is_rh_email_domain("@REDHAT.com"))