Merge pull request #1353 from major/rspeed-2652/rh-identity-field-validation

RSPEED-2652: validate type and format of rh-identity fields

Author: tisnik

src/authentication/rh_identity.py

@@ -6,7 +6,7 @@
 
 import base64
 import json
-from typing import Optional
+from typing import Any, Optional
 
 from fastapi import HTTPException, Request
 
@@ -52,7 +52,7 @@ def _validate_structure(self) -> None:
         """Validate the identity data structure.
 
         Raises:
-            HTTPException: 400 if required fields are missing
+            HTTPException: 400 if required fields are missing or malformed
         """
         if (
             "identity" not in self.identity_data
@@ -68,44 +68,111 @@ def _validate_structure(self) -> None:
 
         identity_type = identity["type"]
         if identity_type == "User":
-            if "user" not in identity:
-                logger.warning(
-                    "Identity validation failed: missing 'user' field for User type"
-                )
-                raise HTTPException(status_code=400, detail="Invalid identity data")
-            user = identity["user"]
-            if "user_id" not in user:
-                logger.warning(
-                    "Identity validation failed: missing 'user_id' in user data"
-                )
-                raise HTTPException(status_code=400, detail="Invalid identity data")
-            if "username" not in user:
-                logger.warning(
-                    "Identity validation failed: missing 'username' in user data"
-                )
-                raise HTTPException(status_code=400, detail="Invalid identity data")
+            self._validate_user_fields(identity)
         elif identity_type == "System":
-            if "system" not in identity:
-                logger.warning(
-                    "Identity validation failed: missing 'system' field for System type"
-                )
-                raise HTTPException(status_code=400, detail="Invalid identity data")
-            system = identity["system"]
-            if "cn" not in system:
-                logger.warning(
-                    "Identity validation failed: missing 'cn' in system data"
-                )
-                raise HTTPException(status_code=400, detail="Invalid identity data")
-            if "account_number" not in identity:
-                logger.warning(
-                    "Identity validation failed: "
-                    "missing 'account_number' for System type"
-                )
-                raise HTTPException(status_code=400, detail="Invalid identity data")
+            self._validate_system_fields(identity)
         else:
             logger.warning("Identity validation failed: unsupported identity type")
             raise HTTPException(status_code=400, detail="Invalid identity data")
 
+        # Validate org_id if present and non-empty
+        org_id = identity.get("org_id")
+        if org_id is not None and org_id != "":
+            self._validate_string_field("org_id", org_id)
+
+    def _validate_user_fields(self, identity: dict) -> None:
+        """Validate required fields for User identity type.
+
+        Args:
+            identity: The identity dict containing user data
+
+        Raises:
+            HTTPException: 400 if required User fields are missing or malformed
+        """
+        if "user" not in identity:
+            logger.warning(
+                "Identity validation failed: missing 'user' field for User type"
+            )
+            raise HTTPException(status_code=400, detail="Invalid identity data")
+        user = identity["user"]
+        if "user_id" not in user:
+            logger.warning("Identity validation failed: missing 'user_id' in user data")
+            raise HTTPException(status_code=400, detail="Invalid identity data")
+        if "username" not in user:
+            logger.warning(
+                "Identity validation failed: missing 'username' in user data"
+            )
+            raise HTTPException(status_code=400, detail="Invalid identity data")
+        self._validate_string_field("user_id", user["user_id"])
+        self._validate_string_field("username", user["username"])
+
+    def _validate_system_fields(self, identity: dict) -> None:
+        """Validate required fields for System identity type.
+
+        Args:
+            identity: The identity dict containing system data
+
+        Raises:
+            HTTPException: 400 if required System fields are missing or malformed
+        """
+        if "system" not in identity:
+            logger.warning(
+                "Identity validation failed: missing 'system' field for System type"
+            )
+            raise HTTPException(status_code=400, detail="Invalid identity data")
+        system = identity["system"]
+        if "cn" not in system:
+            logger.warning("Identity validation failed: missing 'cn' in system data")
+            raise HTTPException(status_code=400, detail="Invalid identity data")
+        if "account_number" not in identity:
+            logger.warning(
+                "Identity validation failed: "
+                "missing 'account_number' for System type"
+            )
+            raise HTTPException(status_code=400, detail="Invalid identity data")
+        self._validate_string_field("cn", system["cn"])
+        self._validate_string_field("account_number", identity["account_number"])
+
+    def _validate_string_field(
+        self, field_name: str, value: Any, max_length: int = 256
+    ) -> None:
+        """Validate that a field value is a well-formed string.
+
+        Args:
+            field_name: Name of the field being validated (for error messages)
+            value: The value to validate
+            max_length: Maximum allowed string length
+
+        Raises:
+            HTTPException: 400 if validation fails
+        """
+        if not isinstance(value, str):
+            logger.warning(
+                "Identity validation failed: %s must be a string, got %s",
+                field_name,
+                type(value).__name__,
+            )
+            raise HTTPException(status_code=400, detail="Invalid identity data")
+        if not value.strip():
+            logger.warning(
+                "Identity validation failed: %s must not be empty",
+                field_name,
+            )
+            raise HTTPException(status_code=400, detail="Invalid identity data")
+        if len(value) > max_length:
+            logger.warning(
+                "Identity validation failed: %s exceeds maximum length of %d",
+                field_name,
+                max_length,
+            )
+            raise HTTPException(status_code=400, detail="Invalid identity data")
+        if any(ord(c) < 32 or ord(c) == 127 for c in value):
+            logger.warning(
+                "Identity validation failed: %s contains control characters",
+                field_name,
+            )
+            raise HTTPException(status_code=400, detail="Invalid identity data")
+
     def _get_identity_type(self) -> str:
         """Get the identity type (User or System).
 

tests/unit/authentication/test_rh_identity.py

@@ -594,3 +594,139 @@ async def test_header_exceeding_limit_rejected(
         mock_warning.assert_called_once()
         assert mock_warning.call_args.args[1] == header_size
         assert mock_warning.call_args.args[2] == max_size
+
+
+class TestRHIdentityFieldValidation:
+    """Test suite for RHIdentityData string field validation."""
+
+    @pytest.mark.parametrize(
+        "field_path,bad_value",
+        [
+            (("user", "user_id"), None),
+            (("user", "user_id"), 12345),
+            (("user", "user_id"), True),
+            (("user", "user_id"), []),
+            (("user", "user_id"), {}),
+            (("user", "user_id"), 3.14),
+            (("user", "username"), None),
+            (("user", "username"), 12345),
+        ],
+    )
+    def test_user_non_string_types_rejected(
+        self, user_identity_data: dict, field_path: tuple[str, str], bad_value: object
+    ) -> None:
+        """Reject non-string values in User identity string fields."""
+        user_identity_data["identity"][field_path[0]][field_path[1]] = bad_value
+        with pytest.raises(HTTPException) as exc_info:
+            RHIdentityData(user_identity_data)
+        assert exc_info.value.status_code == 400
+
+    @pytest.mark.parametrize(
+        "field_path,bad_value",
+        [
+            (("system", "cn"), None),
+            (("system", "cn"), 12345),
+            (("system", "cn"), True),
+            (("system", "cn"), []),
+            (("system", "cn"), {}),
+            (("account_number",), None),
+            (("account_number",), 12345),
+            (("account_number",), False),
+            (("account_number",), []),
+            (("account_number",), {}),
+        ],
+    )
+    def test_system_non_string_types_rejected(
+        self,
+        system_identity_data: dict,
+        field_path: tuple[str, ...],
+        bad_value: object,
+    ) -> None:
+        """Reject non-string values in System identity string fields."""
+        identity = system_identity_data["identity"]
+        if len(field_path) == 1:
+            identity[field_path[0]] = bad_value
+        else:
+            identity[field_path[0]][field_path[1]] = bad_value
+        with pytest.raises(HTTPException) as exc_info:
+            RHIdentityData(system_identity_data)
+        assert exc_info.value.status_code == 400
+
+    @pytest.mark.parametrize("bad_value", ["", "   ", "\t", "\n"])
+    def test_empty_whitespace_rejected(
+        self, user_identity_data: dict, bad_value: str
+    ) -> None:
+        """Reject empty and whitespace-only strings."""
+        user_identity_data["identity"]["user"]["user_id"] = bad_value
+        with pytest.raises(HTTPException) as exc_info:
+            RHIdentityData(user_identity_data)
+        assert exc_info.value.status_code == 400
+
+    @pytest.mark.parametrize(
+        "bad_value",
+        ["user\x00id", "user\nid", "user\rid", "a\x1fb", "a\x7fb"],
+    )
+    def test_control_characters_rejected(
+        self, user_identity_data: dict, bad_value: str
+    ) -> None:
+        """Reject strings containing control characters."""
+        user_identity_data["identity"]["user"]["user_id"] = bad_value
+        with pytest.raises(HTTPException) as exc_info:
+            RHIdentityData(user_identity_data)
+        assert exc_info.value.status_code == 400
+
+    def test_oversized_value_rejected(self, user_identity_data: dict) -> None:
+        """Reject values longer than 256 characters."""
+        user_identity_data["identity"]["user"]["user_id"] = "a" * 257
+        with pytest.raises(HTTPException) as exc_info:
+            RHIdentityData(user_identity_data)
+        assert exc_info.value.status_code == 400
+
+    def test_max_length_boundary_accepted(self, user_identity_data: dict) -> None:
+        """Accept values exactly 256 characters long."""
+        user_identity_data["identity"]["user"]["user_id"] = "a" * 256
+        RHIdentityData(user_identity_data)
+
+    def test_org_id_missing_accepted(self, user_identity_data: dict) -> None:
+        """Allow missing org_id."""
+        user_identity_data["identity"].pop("org_id", None)
+        RHIdentityData(user_identity_data)
+
+    def test_org_id_empty_accepted(self, user_identity_data: dict) -> None:
+        """Allow empty org_id."""
+        user_identity_data["identity"]["org_id"] = ""
+        RHIdentityData(user_identity_data)
+
+    def test_org_id_non_string_rejected(self, user_identity_data: dict) -> None:
+        """Reject non-string org_id when provided and non-empty."""
+        user_identity_data["identity"]["org_id"] = 12345
+        with pytest.raises(HTTPException) as exc_info:
+            RHIdentityData(user_identity_data)
+        assert exc_info.value.status_code == 400
+
+    def test_org_id_valid_accepted(self, user_identity_data: dict) -> None:
+        """Accept valid string org_id."""
+        user_identity_data["identity"]["org_id"] = "valid-org-id"
+        RHIdentityData(user_identity_data)
+
+    def test_org_id_oversized_rejected(self, user_identity_data: dict) -> None:
+        """Reject oversized org_id."""
+        user_identity_data["identity"]["org_id"] = "a" * 257
+        with pytest.raises(HTTPException) as exc_info:
+            RHIdentityData(user_identity_data)
+        assert exc_info.value.status_code == 400
+
+    def test_org_id_control_chars_rejected(self, user_identity_data: dict) -> None:
+        """Reject org_id containing control characters."""
+        user_identity_data["identity"]["org_id"] = "org\x00id"
+        with pytest.raises(HTTPException) as exc_info:
+            RHIdentityData(user_identity_data)
+        assert exc_info.value.status_code == 400
+
+    def test_valid_user_data_still_passes(self, user_identity_data: dict) -> None:
+        """Regression: valid User identity data passes validation."""
+        RHIdentityData(user_identity_data)
+
+    def test_valid_system_data_still_passes(self, system_identity_data: dict) -> None:
+        """Regression: valid System identity data passes validation."""
+        RHIdentityData(system_identity_data)