Avoid Unwanted headers/query parameters being forwarded to the backend.
### Description:

There can be headers which are not supposed to forwarded to backend as well as in the throttle publish event. 

### Steps to reproduce:
Invoke API with APIKey. 
And check the recieved set of headers in the backend. 

### Affected Product Version:
choreo-connect-1.0.0-rc1

### Environment details (with versions):
- OS:
- Client:
- Env (Docker/K8s):

---
### Optional Fields
#### Related Issues:
<!-- Any related issues from this/other repositories-->

#### Suggested Labels:
<!--Only to be used by non-members-->

#### Suggested Assignees:
<!--Only to be used by non-members-->

diff --git a/enforcer-parent/commons/src/main/java/org/wso2/choreo/connect/enforcer/commons/model/RequestContext.java b/enforcer-parent/commons/src/main/java/org/wso2/choreo/connect/enforcer/commons/model/RequestContext.java
index 38f5b63e0..fa64ad2e5 100644
--- a/enforcer-parent/commons/src/main/java/org/wso2/choreo/connect/enforcer/commons/model/RequestContext.java
+++ b/enforcer-parent/commons/src/main/java/org/wso2/choreo/connect/enforcer/commons/model/RequestContext.java
@@ -62,6 +62,10 @@ public class RequestContext {
     private WebSocketFrameContext webSocketFrameContext;
     private Map<String, String> queryParameters;
     private Map<String, String> pathParameters;
+    private ArrayList<String> queryParamsToRemove;
+    // This is used to keep protected headers like authorization header. The protected headers will not be
+    // sent to the Traffic Manager when header based rate limiting is enabled.
+    private ArrayList<String> protectedHeaders;
 
     // Request Timestamp is required for analytics
     private long requestTimeStamp;
@@ -312,6 +316,28 @@ public class RequestContext {
         return webSocketFrameContext;
     }
 
+    /**
+     * If there is a set of query parameters needs to be removed from the outbound request, those parameters should
+     * be added to the arrayList here.
+     *
+     * @return query parameters which are supposed to be removed.
+     */
+    public ArrayList<String> getQueryParamsToRemove() {
+        return queryParamsToRemove;
+    }
+
+    /**
+     * If there is a set of headers needs to be removed from the throttle publishing event, those headers should
+     * be added to the arrayList here.
+     *
+     * Ex. Authorization Header
+     *
+     * @return header names which are not supposed to be published to the traffic manager.
+     */
+    public ArrayList<String> getProtectedHeaders() {
+        return protectedHeaders;
+    }
+
     /**
      * Implements builder pattern to build an {@link RequestContext} object.
      */
@@ -412,6 +438,8 @@ public class RequestContext {
             requestContext.clientIp = this.clientIp;
             requestContext.addHeaders = new HashMap<>();
             requestContext.removeHeaders = new ArrayList<>();
+            requestContext.queryParamsToRemove = new ArrayList<>();
+            requestContext.protectedHeaders = new ArrayList<>();
             String[] queryParts = this.requestPath.split("\\\\?");
             String queryPrams = queryParts.length > 1 ? queryParts[1] : "";
 
diff --git a/enforcer-parent/enforcer/src/main/java/org/wso2/choreo/connect/enforcer/api/ResponseObject.java b/enforcer-parent/enforcer/src/main/java/org/wso2/choreo/connect/enforcer/api/ResponseObject.java
index b4d34e3c6..36e0e588d 100644
--- a/enforcer-parent/enforcer/src/main/java/org/wso2/choreo/connect/enforcer/api/ResponseObject.java
+++ b/enforcer-parent/enforcer/src/main/java/org/wso2/choreo/connect/enforcer/api/ResponseObject.java
@@ -34,6 +34,9 @@ public class ResponseObject {
     private ArrayList<String> removeHeaderMap = new ArrayList<>();
     private Map<String, String> metaDataMap;
     private boolean isDirectResponse = false;
+    private ArrayList<String> queryParamsToRemove = new ArrayList<>();
+    private Map<String, String> queryParams;
+    private String requestPath;
 
     public ArrayList<String> getRemoveHeaderMap() {
         return removeHeaderMap;
@@ -110,4 +113,28 @@ public class ResponseObject {
     public void setMetaDataMap(Map<String, String> metaDataMap) {
         this.metaDataMap = metaDataMap;
     }
+
+    public ArrayList<String> getQueryParamsToRemove() {
+        return queryParamsToRemove;
+    }
+
+    public void setQueryParamsToRemove(ArrayList<String> queryParamsToRemove) {
+        this.queryParamsToRemove = queryParamsToRemove;
+    }
+
+    public Map<String, String> getQueryParamMap() {
+        return queryParams;
+    }
+
+    public void setQueryParamMap(Map<String, String> queryParams) {
+        this.queryParams = queryParams;
+    }
+
+    public String getRequestPath() {
+        return requestPath;
+    }
+
+    public void setRequestPath(String requestPath) {
+        this.requestPath = requestPath;
+    }
 }
diff --git a/enforcer-parent/enforcer/src/main/java/org/wso2/choreo/connect/enforcer/api/RestAPI.java b/enforcer-parent/enforcer/src/main/java/org/wso2/choreo/connect/enforcer/api/RestAPI.java
index b89f6cbca..d5f998815 100644
--- a/enforcer-parent/enforcer/src/main/java/org/wso2/choreo/connect/enforcer/api/RestAPI.java
+++ b/enforcer-parent/enforcer/src/main/java/org/wso2/choreo/connect/enforcer/api/RestAPI.java
@@ -157,17 +157,15 @@ public class RestAPI implements API {
     @Override
     public ResponseObject process(RequestContext requestContext) {
         ResponseObject responseObject = new ResponseObject(requestContext.getRequestID());
+        responseObject.setRequestPath(requestContext.getRequestPath());
         boolean analyticsEnabled = ConfigHolder.getInstance().getConfig().getAnalyticsConfig().isEnabled();
 
-        // Process to-be-removed headers
-        AuthHeaderDto authHeader = ConfigHolder.getInstance().getConfig().getAuthHeader();
-        if (!authHeader.isEnableOutboundAuthHeader()) {
-            String authHeaderName = FilterUtils.getAuthHeaderName(requestContext);
-            requestContext.getRemoveHeaders().add(authHeaderName);
-        }
+        populateRemoveAndProtectedHeaders(requestContext);
 
         if (executeFilterChain(requestContext)) {
             responseObject.setRemoveHeaderMap(requestContext.getRemoveHeaders());
+            responseObject.setQueryParamsToRemove(requestContext.getQueryParamsToRemove());
+            responseObject.setQueryParamMap(requestContext.getQueryParameters());
             responseObject.setStatusCode(APIConstants.StatusCodes.OK.getCode());
             if (requestContext.getAddHeaders() != null && requestContext.getAddHeaders().size() > 0) {
                 responseObject.setHeaderMap(requestContext.getAddHeaders());
@@ -288,4 +286,43 @@ public class RestAPI implements API {
             }
         }
     }
+
+    private void populateRemoveAndProtectedHeaders(RequestContext requestContext) {
+        Map<String, SecuritySchemaConfig> securitySchemeDefinitions =
+                requestContext.getMatchedAPI().getSecuritySchemeDefinitions();
+        // API key headers are considered to be protected headers, such that the header would not be sent
+        // to backend and traffic manager.
+        // This would prevent leaking credentials, even if user is invoking unsecured resource with some
+        // credentials.
+        for (Map.Entry<String, SecuritySchemaConfig> entry : securitySchemeDefinitions.entrySet()) {
+            SecuritySchemaConfig schema = entry.getValue();
+            if (APIConstants.SWAGGER_API_KEY_AUTH_TYPE_NAME.equalsIgnoreCase(schema.getType())) {
+                if (APIConstants.SWAGGER_API_KEY_IN_HEADER.equals(schema.getIn())) {
+                    requestContext.getProtectedHeaders().add(schema.getName());
+                    requestContext.getRemoveHeaders().add(schema.getName());
+                    continue;
+                }
+                if (APIConstants.SWAGGER_API_KEY_IN_QUERY.equals(schema.getIn())) {
+                    requestContext.getQueryParamsToRemove().add(schema.getName());
+                }
+            }
+        }
+
+        // Internal-Key credential is considered to be protected headers, such that the header would not be sent
+        // to backend and traffic manager.
+        String internalKeyHeader = ConfigHolder.getInstance().getConfig().getAuthHeader()
+                .getTestConsoleHeaderName().toLowerCase();
+        requestContext.getRemoveHeaders().add(internalKeyHeader);
+        // Avoid internal key being published to the Traffic Manager
+        requestContext.getProtectedHeaders().add(internalKeyHeader);
+
+        // Remove Authorization Header
+        AuthHeaderDto authHeader = ConfigHolder.getInstance().getConfig().getAuthHeader();
+        String authHeaderName = FilterUtils.getAuthHeaderName(requestContext);
+        if (!authHeader.isEnableOutboundAuthHeader()) {
+            requestContext.getRemoveHeaders().add(authHeaderName);
+        }
+        // Authorization Header should not be included in the throttle publishing event.
+        requestContext.getProtectedHeaders().add(authHeaderName);
+    }
 }
diff --git a/enforcer-parent/enforcer/src/main/java/org/wso2/choreo/connect/enforcer/constants/APIConstants.java b/enforcer-parent/enforcer/src/main/java/org/wso2/choreo/connect/enforcer/constants/APIConstants.java
index 80bd4b0ea..ea792da28 100644
--- a/enforcer-parent/enforcer/src/main/java/org/wso2/choreo/connect/enforcer/constants/APIConstants.java
+++ b/enforcer-parent/enforcer/src/main/java/org/wso2/choreo/connect/enforcer/constants/APIConstants.java
@@ -86,6 +86,7 @@ public class APIConstants {
     public static final String APPLICATION_JSON = "application/json";
     public static final String API_TRACE_KEY = "X-TRACE-KEY";
     public static final String X_FORWARDED_FOR = "x-forwarded-for";
+    public static final String PATH_HEADER = ":path";
 
     public static final String LOG_TRACE_ID = "traceId";
 
diff --git a/enforcer-parent/enforcer/src/main/java/org/wso2/choreo/connect/enforcer/grpc/ExtAuthService.java b/enforcer-parent/enforcer/src/main/java/org/wso2/choreo/connect/enforcer/grpc/ExtAuthService.java
index 594749436..6225ef6e8 100644
--- a/enforcer-parent/enforcer/src/main/java/org/wso2/choreo/connect/enforcer/grpc/ExtAuthService.java
+++ b/enforcer-parent/enforcer/src/main/java/org/wso2/choreo/connect/enforcer/grpc/ExtAuthService.java
@@ -47,6 +47,9 @@ import org.wso2.choreo.connect.enforcer.tracing.TracingSpan;
 import org.wso2.choreo.connect.enforcer.tracing.TracingTracer;
 import org.wso2.choreo.connect.enforcer.tracing.Utils;
 
+import java.util.List;
+import java.util.Map;
+
 /**
  * This is the gRPC server written to match with the envoy ext-authz filter proto file. Envoy proxy call this service.
  * This is the entry point to the filter chain process for a request.
@@ -140,6 +143,21 @@ public class ExtAuthService extends AuthorizationGrpc.AuthorizationImplBase {
                     .build();
         } else {
             OkHttpResponse.Builder okResponseBuilder = OkHttpResponse.newBuilder();
+
+            // If the user is sending the APIKey credentials within query parameters, those query parameters should
+            // not be sent to the backend. Hence, the :path header needs to be constructed again removing the apiKey
+            // query parameter. In this scenario, apiKey query parameter is sent within the property called
+            // 'queryParamsToRemove' so that the custom filters also can utilize the method.
+            if (responseObject.getQueryParamsToRemove().size() > 0) {
+                String constructedPath = constructQueryParamString(responseObject.getRequestPath(),
+                        responseObject.getQueryParamMap(), responseObject.getQueryParamsToRemove());
+                HeaderValueOption headerValueOption = HeaderValueOption.newBuilder()
+                        .setHeader(HeaderValue.newBuilder().setKey(APIConstants.PATH_HEADER).setValue(constructedPath)
+                                .build())
+                        .build();
+                okResponseBuilder.addHeaders(headerValueOption);
+            }
+
             if (responseObject.getHeaderMap() != null) {
                 responseObject.getHeaderMap().forEach((key, value) -> {
                             HeaderValueOption headerValueOption = HeaderValueOption.newBuilder()
@@ -180,4 +198,34 @@ public class ExtAuthService extends AuthorizationGrpc.AuthorizationImplBase {
         }
         return Code.INTERNAL_VALUE;
     }
+
+    private String constructQueryParamString(String requestPath, Map<String, String> queryParamMap,
+                                             List<String> queryParamsToRemove) {
+        // If no query parameters needs to be removed, then the request path can be applied as it is.
+        if (queryParamsToRemove.size() == 0) {
+            return requestPath;
+        }
+
+        String pathWithoutQueryParams = requestPath.split("\\\\?")[0];
+        StringBuilder requestPathBuilder = new StringBuilder(pathWithoutQueryParams);
+        int count = 0;
+        if (queryParamMap.size() > 0) {
+            for (String queryParam : queryParamMap.keySet()) {
+                if (queryParamsToRemove.contains(queryParam)) {
+                    continue;
+                }
+                if (count == 0) {
+                    requestPathBuilder.append("?");
+                } else {
+                    requestPathBuilder.append("&");
+                }
+                requestPathBuilder.append(queryParam);
+                if (queryParamMap.get(queryParam) != null) {
+                    requestPathBuilder.append("=").append(queryParamMap.get(queryParam));
+                }
+                count++;
+            }
+        }
+        return requestPathBuilder.toString();
+    }
 }
diff --git a/enforcer-parent/enforcer/src/main/java/org/wso2/choreo/connect/enforcer/security/jwt/InternalAPIKeyAuthenticator.java b/enforcer-parent/enforcer/src/main/java/org/wso2/choreo/connect/enforcer/security/jwt/InternalAPIKeyAuthenticator.java
index dde3ff0bf..681d0602e 100644
--- a/enforcer-parent/enforcer/src/main/java/org/wso2/choreo/connect/enforcer/security/jwt/InternalAPIKeyAuthenticator.java
+++ b/enforcer-parent/enforcer/src/main/java/org/wso2/choreo/connect/enforcer/security/jwt/InternalAPIKeyAuthenticator.java
@@ -96,8 +96,6 @@ public class InternalAPIKeyAuthenticator extends APIKeyHandler {
                 }
                 // Extract internal from the request while removing it from the msg context.
                 String internalKey = extractInternalKey(requestContext);
-                // Remove internal key from outbound request
-                requestContext.getRemoveHeaders().add(securityParam);
 
                 String[] splitToken = internalKey.split("\\\\.");
                 SignedJWT signedJWT = SignedJWT.parse(internalKey);
diff --git a/enforcer-parent/enforcer/src/main/java/org/wso2/choreo/connect/enforcer/throttle/ThrottleFilter.java b/enforcer-parent/enforcer/src/main/java/org/wso2/choreo/connect/enforcer/throttle/ThrottleFilter.java
index 19f1a7abf..62f359b33 100644
--- a/enforcer-parent/enforcer/src/main/java/org/wso2/choreo/connect/enforcer/throttle/ThrottleFilter.java
+++ b/enforcer-parent/enforcer/src/main/java/org/wso2/choreo/connect/enforcer/throttle/ThrottleFilter.java
@@ -394,6 +394,15 @@ public class ThrottleFilter implements Filter {
         if (config.isHeaderConditionsEnabled()) {
             Map<String, String> headers = requestContext.getHeaders();
             for (String name : headers.keySet()) {
+                // To avoid publishing user token to the traffic manager.
+                if (requestContext.getProtectedHeaders().contains(name)) {
+                    continue;
+                }
+                // Sending path header is stopped as it could contain query parameters which are used
+                // to secure APIs.
+                if (name.equals(APIConstants.PATH_HEADER)) {
+                    continue;
+                }
                 jsonObMap.put(name, headers.get(name));
             }
         }
@@ -401,6 +410,10 @@ public class ThrottleFilter implements Filter {
         if (config.isQueryConditionsEnabled()) {
             Map<String, String> params = requestContext.getQueryParameters();
             for (String name : params.keySet()) {
+                // To avoid publishing apiKey to the traffic manager.
+                if (requestContext.getQueryParamsToRemove().contains(name)) {
+                    continue;
+                }
                 jsonObMap.put(name, params.get(name));
             }
         }
['enforcer-parent/enforcer/src/main/java/org/wso2/choreo/connect/enforcer/grpc/ExtAuthService.java', 'enforcer-parent/enforcer/src/main/java/org/wso2/choreo/connect/enforcer/throttle/ThrottleFilter.java', 'enforcer-parent/enforcer/src/main/java/org/wso2/choreo/connect/enforcer/security/jwt/InternalAPIKeyAuthenticator.java', 'enforcer-parent/enforcer/src/main/java/org/wso2/choreo/connect/enforcer/constants/APIConstants.java', 'enforcer-parent/commons/src/main/java/org/wso2/choreo/connect/enforcer/commons/model/RequestContext.java', 'enforcer-parent/enforcer/src/main/java/org/wso2/choreo/connect/enforcer/api/ResponseObject.java', 'enforcer-parent/enforcer/src/main/java/org/wso2/choreo/connect/enforcer/api/RestAPI.java']