Code Scanning security alert fixes

CONTRIBUTING.md

@@ -35,7 +35,7 @@ a proposal for your work first, to be sure that we can use it.
 ## <a name="submit"></a> Submission Guidelines
 
 ### <a name="submit-issue"></a> Submitting an Issue
-Before you submit an issue, search the archive, maybe your question was already answered.
+Before you submit an issue, search the archive, maybe your question has already been answered.
 
 If your issue appears to be a bug, and hasn't been reported, open a new issue.
 Help us to maximize the effort we can spend fixing issues and adding new

End_to_end_Solutions/AOAISearchDemo/app/backend/app.py

@@ -340,7 +340,7 @@ def chat():
         return jsonify(response.to_item()), 400
     except Exception as e:
         logger.exception(f"Exception in /chat: {e}", extra=properties)
-        response = ChatResponse(answer=Answer(), error=str(e), show_retry=True)
+        response = ChatResponse(answer=Answer(), error="An internal error has occurred.", show_retry=True)
         return jsonify(response.to_item()), 500
 
 
@@ -352,7 +352,7 @@ def get_all_user_profiles():
         return jsonify(user_profiles_dict)
     except Exception as e:
         logger.exception(f"Exception in /user-profiles: {e}")
-        return jsonify({"error": str(e)}), 500
+        return jsonify({"error": "An internal error has occurred."}), 500
 
 
 @app.route("/chat-sessions/<user_id>/<conversation_id>", methods=["DELETE"])
@@ -369,7 +369,7 @@ def clear_chat_session(user_id: str, conversation_id: str):
         logger.exception(
             f"Exception in /chat-sessions/<user_id>/<conversation_id>: {e}"
         )
-        return jsonify({"error": str(e)}), 500
+        return jsonify({"error": "An internal error has occurred."}), 500
 
 
 @app.route("/search-settings", methods=["GET"])
@@ -391,7 +391,7 @@ def get_search_settings():
         return jsonify(search_settings.to_item())
     except Exception as e:
         logger.exception(f"Exception in /search-settings: {e}")
-        return jsonify({"error": str(e)}), 500
+        return jsonify({"error": "An internal error has occurred."}), 500
 
 
 if __name__ == "__main__":

End_to_end_Solutions/AOAISearchDemo/app/backend/data_client/data_client.py

@@ -12,6 +12,26 @@
 from typing import List, Optional
 
 from common.logging.log_helper import CustomLogger
+from urllib.parse import urlparse
+from urllib.parse import urlparse, urljoin
+
+def _validate_base_uri(self, base_uri: str):
+    parsed_uri = urlparse(base_uri)
+    if parsed_uri.scheme not in ["http", "https"]:
+        raise ValueError("Invalid URI scheme")
+    if not parsed_uri.netloc:
+        raise ValueError("Invalid URI netloc")
+
+def _validate_base_uri(self, base_uri: str):
+    parsed_uri = urlparse(base_uri)
+    if parsed_uri.scheme not in ["http", "https"]:
+        raise ValueError("Invalid URI scheme")
+    if not parsed_uri.netloc:
+        raise ValueError("Invalid URI netloc")
+
+def _validate_path(self, path: str):
+    if not path.startswith("/"):
+        raise ValueError("Invalid path")
 
 class DataClient:
     class HttpMethod(Enum):
@@ -21,8 +41,22 @@ class HttpMethod(Enum):
         DELETE="DELETE"
 
     def __init__(self, base_uri: str, logger: CustomLogger):
+        self._validate_base_uri(base_uri)
         self.base_uri = base_uri
+        self.base_uri = self._validate_base_uri(base_uri)
         self.logger = logger
+
+    def _validate_base_uri(self, base_uri: str) -> str:
+        # Ensure the base_uri is a trusted URL
+        if not base_uri.startswith("https://trusted-domain.com"):
+            raise ValueError("Invalid base URI")
+        return base_uri
+
+    def _sanitize_path(self, path: str) -> str:
+        # Sanitize the path to prevent malicious input
+        if ".." in path or path.startswith("/"):
+            raise ValueError("Invalid path")
+        return path
 
     def check_chat_session(self, user_id: str, conversation_id: str) -> bool:
         path = f"/check-chat-session/{user_id}/{conversation_id}"
@@ -113,6 +147,8 @@ def get_user_resources(self, user_id: str) -> List[ResourceProfile]:
 
     @retry(reraise=True, stop = stop_after_attempt(3), wait = wait_exponential(multiplier = 1, max = 60))
     def _make_request(self, path: str, method: HttpMethod, payload: Optional[dict] = None) -> str:
+        path = self._sanitize_path(path)
+        self._validate_path(path)
 
         headers = self.logger.get_converation_and_dialog_ids()
         properties = self.logger.get_updated_properties(headers)