🛡️ Sentinel: [HIGH] Fix Excessive Data Exposure and Information Leakage

templates/agent/handler.py

@@ -32,7 +32,7 @@ def get_item(item_id: str) -> dict:
     item = repository.get_item(item_id)
     if not item:
         return {"error": f"Item {item_id} not found"}
-    return item
+    return Item.model_validate(item).dump()
 
 
 @tracer.capture_method

templates/agent/models.py

@@ -1,3 +1,5 @@
+from typing import Any
+
 from pydantic import BaseModel, Field
 from pydantic.alias_generators import to_camel
 
@@ -8,3 +10,9 @@ class Item(BaseModel, populate_by_name=True, alias_generator=to_camel):
     id: str = Field(description="Unique identifier for the item.")
     name: str = Field(description="Name of the item.")
     description: str | None = Field(description="Description of the item.", default=None)
+
+    def dump(self, **kwargs: Any) -> dict:
+        """Dump the model to a dictionary with default settings for responses."""
+        kwargs.setdefault("by_alias", True)
+        kwargs.setdefault("exclude_none", True)
+        return self.model_dump(**kwargs)

templates/graphql/handler.py

@@ -30,9 +30,12 @@ def get_item(id: str) -> dict | None:
         The item if found, or None.
     """
     try:
-        return repository.get_item(id)
+        if (item := repository.get_item(id)) is None:
+            return None
+        return Item.model_validate(item).dump()
     except Exception as error:
-        raise RuntimeError(f"Failed to get item with ID '{id}'. Cause: {error}") from error
+        logger.error(f"Failed to get item with ID '{id}'", exc_info=error)
+        raise RuntimeError(f"Failed to get item with ID '{id}'") from error
 
 
 @app.resolver(type_name="Query", field_name="listItems")
@@ -44,9 +47,10 @@ def list_items() -> list[dict]:
         A list of items.
     """
     try:
-        return repository.list_items()
+        return [Item.model_validate(item).dump() for item in repository.list_items()]
     except Exception as error:
-        raise RuntimeError(f"Failed to list items. Cause: {error}") from error
+        logger.error("Failed to list items", exc_info=error)
+        raise RuntimeError("Failed to list items") from error
 
 
 @app.resolver(type_name="Mutation", field_name="createItem")
@@ -65,7 +69,8 @@ def create_item(name: str) -> dict:
         repository.put_item(item)
         return item
     except (ValidationError, Exception) as error:
-        raise RuntimeError(f"Failed to create item with name '{name}'. Cause: {error}") from error
+        logger.error(f"Failed to create item with name '{name}'", exc_info=error)
+        raise RuntimeError(f"Failed to create item with name '{name}'") from error
 
 
 @logger.inject_lambda_context(correlation_id_path=correlation_paths.APPSYNC_RESOLVER)

tests/agent/test_handler.py

@@ -94,5 +94,17 @@ def test_lambda_handler_create_item(mocker, repository, lambda_context, bedrock_
     assert body["name"] == "new item"
 
 
+def test_sensitive_data_exposure(repository):
+    """Verify that internal fields in DynamoDB are NOT leaked to the agent."""
+    from templates.agent.handler import get_item
+
+    item_with_secret = {"id": "1", "name": "test item", "internal_secret": "TOP_SECRET"}
+    repository.put_item(item_with_secret)
+
+    result = get_item("1")
+
+    assert "internal_secret" not in result
+
+
 if __name__ == "__main__":
     main()

tests/graphql/test_handler.py

@@ -40,5 +40,39 @@ def test_create_item_resolver(lambda_context):
     assert "id" in result
 
 
+def test_sensitive_data_exposure(repository, lambda_context):
+    """Verify that internal fields in DynamoDB are NOT leaked to the client."""
+    from templates.graphql.handler import main
+
+    item_with_secret = {"id": "123", "name": "Test Item", "internal_secret": "TOP_SECRET"}
+    repository.put_item(item_with_secret)
+
+    # Test getItem
+    event_get = {"info": {"parentTypeName": "Query", "fieldName": "getItem"}, "arguments": {"id": "123"}}
+    result_get = main(event_get, lambda_context)
+    assert "internal_secret" not in result_get
+
+    # Test listItems
+    event_list = {"info": {"parentTypeName": "Query", "fieldName": "listItems"}, "arguments": {}}
+    result_list = main(event_list, lambda_context)
+    assert "internal_secret" not in result_list[0]
+
+
+def test_error_message_information_leakage(lambda_context, mocker):
+    """Verify that internal error details are NOT leaked to the client."""
+    from templates.graphql import handler
+    from templates.graphql.handler import get_item
+
+    mocker.patch.object(handler.repository, "get_item", side_effect=Exception("Database connection failed"))
+
+    import pytest
+
+    with pytest.raises(RuntimeError) as excinfo:
+        get_item("123")
+
+    assert "Database connection failed" not in str(excinfo.value)
+    assert "Cause:" not in str(excinfo.value)
+
+
 if __name__ == "__main__":
     main()