feat(RHINENG-5479): Update API to handle Service Account type (#1588)

Co-authored-by: Asa Price <[email protected]>

app/auth/identity.py

@@ -9,6 +9,7 @@
 
 from app.logging import get_logger
 from app.logging import threadctx
+from app.validators import verify_uuid_format
 
 
 __all__ = ["Identity", "from_auth_header", "from_bearer_token"]
@@ -36,6 +37,7 @@ class CertType(str, Enum):
 class IdentityType(str, Enum):
     SYSTEM = "System"
     USER = "User"
+    SERVICE_ACCOUNT = "ServiceAccount"
 
 
 class Identity:
@@ -73,6 +75,8 @@ def __init__(self, obj=None, token=None):
                 self.user = result.get("user")
             if "system" in result:
                 self.system = result.get("system")
+            if "service_account" in result:
+                self.service_account = result.get("service_account")
 
             threadctx.org_id = self.org_id
 
@@ -91,13 +95,16 @@ def _asdict(self):
         if self.identity_type == IdentityType.USER:
             if hasattr(self, "user"):
                 ident["user"] = self.user.copy()
-
             return ident
 
         if self.identity_type == IdentityType.SYSTEM:
             ident["system"] = self.system.copy()
             return ident
 
+        if self.identity_type == IdentityType.SERVICE_ACCOUNT:
+            ident["service_account"] = self.service_account.copy()
+            return ident
+
     def __eq__(self, other):
         return self.org_id == other.org_id
 
@@ -125,8 +132,10 @@ class IdentitySchema(IdentityBaseSchema):
     def user_system_check(self, in_data, **kwargs):
         if in_data["type"] == IdentityType.USER:
             result = UserIdentitySchema().load(in_data)
-        else:
+        elif in_data["type"] == IdentityType.SYSTEM:
             result = SystemIdentitySchema().load(in_data)
+        else:
+            result = ServiceAccountIdentitySchema().load(in_data)
 
         return result
 
@@ -155,6 +164,15 @@ class SystemIdentitySchema(IdentityBaseSchema):
     system = m.fields.Nested(SystemInfoIdentitySchema, required=True)
 
 
+class ServiceAccountInfoIdentitySchema(IdentityBaseSchema):
+    client_id = m.fields.Str(required=True, validate=verify_uuid_format)
+    username = m.fields.Str(required=True, validate=m.validate.Length(min=1))
+
+
+class ServiceAccountIdentitySchema(IdentityBaseSchema):
+    service_account = m.fields.Nested(ServiceAccountInfoIdentitySchema, required=True)
+
+
 # Messages from the system_profile topic don't need to provide a real Identity,
 # So this helper function creates a basic User-type identity from the host data.
 def create_mock_identity_with_org_id(org_id):

tests/helpers/test_utils.py

@@ -34,6 +34,18 @@
     "user": {"email": "[email protected]", "first_name": "test"},
 }
 
+SERVICE_ACCOUNT_IDENTITY = {
+    "account_number": "test",
+    "org_id": "456",
+    "auth_type": "jwt-auth",
+    "internal": {"auth_time": 500, "cross_access": False, "org_id": "456"},
+    "service_account": {
+        "client_id": "b9757340-f839-4541-9af6-f7535edf08db",
+        "username": "service-account-b9757340-f839-4541-9af6-f7535edf08db",
+    },
+    "type": "ServiceAccount",
+}
+
 YUM_REPO1 = {"id": "repo1", "name": "repo1", "gpgcheck": True, "enabled": True, "base_url": "http://rpms.redhat.com"}
 
 YUM_REPO2 = {"id": "repo2", "name": "repo2", "gpgcheck": True, "enabled": True, "base_url": "http://rpms.redhat.com"}

tests/test_identity_schema.py

@@ -6,6 +6,7 @@
 from app.auth.identity import CertType
 from app.auth.identity import IdentitySchema
 from app.auth.identity import IdentityType
+from tests.helpers.test_utils import SERVICE_ACCOUNT_IDENTITY
 from tests.helpers.test_utils import SYSTEM_IDENTITY
 from tests.helpers.test_utils import USER_IDENTITY
 
@@ -28,16 +29,22 @@ def identity_test_common(identity):
     if result.get("type") == IdentityType.USER:
         assert "user" in result
         assert type(result.get("user")) is dict
-    else:  # IdentityType.SYSTEM
+    elif result.get("type") == IdentityType.SYSTEM:
         assert "system" in result
         system = result.get("system")
         assert type(system) is dict
         assert "cert_type" in system
         assert system.get("cert_type") in CertType.__members__.values()
         assert "cn" in system
+    else:  # IdentityType.SERVICE_ACCOUNT
+        assert "service_account" in result
+        service_acc = result.get("service_account")
+        assert type(service_acc) is dict
+        assert "client_id" in service_acc
+        assert "username" in service_acc
 
 
-@pytest.mark.parametrize("identity", (USER_IDENTITY, SYSTEM_IDENTITY))
+@pytest.mark.parametrize("identity", (USER_IDENTITY, SYSTEM_IDENTITY, SERVICE_ACCOUNT_IDENTITY))
 def test_validate_valid_identity_schema(identity):
     identity_test_common(identity)
 
@@ -58,6 +65,10 @@ def test_validate_valid_system_identity_schema():
     identity_test_common(SYSTEM_IDENTITY)
 
 
+def test_validate_valid_service_account_identity_schema():
+    identity_test_common(SERVICE_ACCOUNT_IDENTITY)
+
+
 @pytest.mark.parametrize("required", ("type", "auth_type", "org_id"))
 def test_identity_missing_required(required):
     # Test missing values
@@ -104,6 +115,23 @@ def test_system_identity_missing_required(required):
         identity_test_common(bad_identity)
 
 
+@pytest.mark.parametrize("required", ("client_id", "username"))
+def test_service_account_identity_missing_required(required):
+    # Test blank values
+    bad_identity = copy.deepcopy(SERVICE_ACCOUNT_IDENTITY)
+    assert "service_account" in bad_identity
+    bad_identity["service_account"][required] = ""
+    with pytest.raises(ValueError):
+        identity_test_common(bad_identity)
+
+    # Test missing values
+    bad_identity = copy.deepcopy(SERVICE_ACCOUNT_IDENTITY)
+    assert "service_account" in bad_identity
+    bad_identity["service_account"].pop(required, None)
+    with pytest.raises(ValueError):
+        identity_test_common(bad_identity)
+
+
 @pytest.mark.parametrize(
     "test_field,bad_value",
     [("org_id", ""), ("org_id", "X" * 37), ("account_number", "X" * 37)],