Update DefaultAuthenticationFailureHandler to handle 403s (elastic#138930)

1. Catch 403s are return them when certain metadata conditions are fulfilled. Deny 403s in response otherwise.

Author: ankit--sethi

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/authc/DefaultAuthenticationFailureHandler.java

@@ -29,6 +29,11 @@
  * response headers like 'WWW-Authenticate'
  */
 public class DefaultAuthenticationFailureHandler implements AuthenticationFailureHandler {
+    /**
+     * Metadata key to denote exceptions that are allowed to return with 403 status
+     */
+    public static final String ACCESS_DENIED_METADATA_KEY = "es.security.access_denied";
+
     private volatile Map<String, List<String>> defaultFailureResponseHeaders;
 
     /**
@@ -113,6 +118,11 @@ public ElasticsearchSecurityException exceptionProcessingRequest(HttpPreRequest
         if (e instanceof ElasticsearchAuthenticationProcessingError) {
             return (ElasticsearchAuthenticationProcessingError) e;
         }
+        if (e instanceof ElasticsearchSecurityException ese) {
+            if (ese.status() == RestStatus.FORBIDDEN && ese.getMetadata(ACCESS_DENIED_METADATA_KEY) != null) {
+                return ese;
+            }
+        }
         return createAuthenticationError("error attempting to authenticate request", e, (Object[]) null);
     }
 
@@ -128,6 +138,11 @@ public ElasticsearchSecurityException exceptionProcessingRequest(
         if (e instanceof ElasticsearchAuthenticationProcessingError) {
             return (ElasticsearchAuthenticationProcessingError) e;
         }
+        if (e instanceof ElasticsearchSecurityException ese) {
+            if (ese.status() == RestStatus.FORBIDDEN && ese.getMetadata(ACCESS_DENIED_METADATA_KEY) != null) {
+                return ese;
+            }
+        }
         return createAuthenticationError("error attempting to authenticate request", e, (Object[]) null);
     }
 
@@ -168,7 +183,7 @@ private ElasticsearchSecurityException createAuthenticationError(final String me
         final ElasticsearchSecurityException ese;
         final boolean containsNegotiateWithToken;
         if (t instanceof ElasticsearchSecurityException) {
-            assert ((ElasticsearchSecurityException) t).status() == RestStatus.UNAUTHORIZED;
+            assert ((ElasticsearchSecurityException) t).status() == RestStatus.UNAUTHORIZED : "rest status must be 401 UNAUTHORIZED";
             ese = (ElasticsearchSecurityException) t;
             if (ese.getBodyHeader("WWW-Authenticate") != null && ese.getBodyHeader("WWW-Authenticate").isEmpty() == false) {
                 /**

x-pack/plugin/core/src/test/java/org/elasticsearch/xpack/core/security/authc/DefaultAuthenticationFailureHandlerTests.java

@@ -13,6 +13,7 @@
 import org.elasticsearch.http.HttpPreRequest;
 import org.elasticsearch.rest.RestStatus;
 import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.transport.TransportRequest;
 import org.elasticsearch.xpack.core.XPackField;
 
 import java.util.Arrays;
@@ -21,7 +22,9 @@
 import java.util.List;
 import java.util.Map;
 
+import static org.elasticsearch.xpack.core.security.authc.DefaultAuthenticationFailureHandler.ACCESS_DENIED_METADATA_KEY;
 import static org.hamcrest.Matchers.contains;
+import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.is;
 import static org.hamcrest.Matchers.notNullValue;
@@ -112,6 +115,7 @@ public void testExceptionProcessingRequest() {
             } else {
                 expectThrows(
                     AssertionError.class,
+                    containsString("rest status must be 401 UNAUTHORIZED"),
                     () -> failureHandler.exceptionProcessingRequest(
                         mock(HttpPreRequest.class),
                         cause,
@@ -156,6 +160,92 @@ public void testSortsWWWAuthenticateHeaderValues() {
         assertWWWAuthenticateWithSchemes(ese, negotiateAuthScheme, bearerAuthScheme, apiKeyAuthScheme, basicAuthScheme);
     }
 
+    public void testExceptionProcessingRequestWith403SecurityException() {
+        final DefaultAuthenticationFailureHandler failureHandler = new DefaultAuthenticationFailureHandler(Collections.emptyMap());
+        final ThreadContext threadContext = new ThreadContext(Settings.builder().build());
+
+        // Test 1: 403 exception with es.security.access_denied metadata should be returned as-is
+        {
+            ElasticsearchSecurityException forbiddenWithMetadata = new ElasticsearchSecurityException(
+                "failed to authorize user for project [test-project]",
+                RestStatus.FORBIDDEN,
+                null,
+                (Object[]) null
+            );
+            forbiddenWithMetadata.addMetadata(ACCESS_DENIED_METADATA_KEY, "true");
+
+            ElasticsearchSecurityException result = failureHandler.exceptionProcessingRequest(
+                mock(HttpPreRequest.class),
+                forbiddenWithMetadata,
+                threadContext
+            );
+            assertThat(result, is(sameInstance(forbiddenWithMetadata)));
+            assertThat(result.status(), equalTo(RestStatus.FORBIDDEN));
+
+            result = failureHandler.exceptionProcessingRequest(
+                mock(TransportRequest.class),
+                "index:foo/bar",
+                forbiddenWithMetadata,
+                threadContext
+            );
+            assertThat(result, is(sameInstance(forbiddenWithMetadata)));
+            assertThat(result.status(), equalTo(RestStatus.FORBIDDEN));
+        }
+
+        // Test 2: 403 exception without metadata should fail assertion
+        {
+            ElasticsearchSecurityException forbiddenWithoutMetadata = new ElasticsearchSecurityException(
+                "some forbidden error",
+                RestStatus.FORBIDDEN,
+                null,
+                (Object[]) null
+            );
+
+            expectThrows(
+                AssertionError.class,
+                containsString("rest status must be 401 UNAUTHORIZED"),
+                () -> failureHandler.exceptionProcessingRequest(mock(HttpPreRequest.class), forbiddenWithoutMetadata, threadContext)
+            );
+            expectThrows(
+                AssertionError.class,
+                containsString("rest status must be 401 UNAUTHORIZED"),
+                () -> failureHandler.exceptionProcessingRequest(
+                    mock(TransportRequest.class),
+                    "index:foo/bar",
+                    forbiddenWithoutMetadata,
+                    threadContext
+                )
+            );
+        }
+
+        // Test 3: 403 exception with different metadata should fail assertion
+        {
+            ElasticsearchSecurityException forbiddenWithDifferentMetadata = new ElasticsearchSecurityException(
+                "some other forbidden error",
+                RestStatus.FORBIDDEN,
+                null,
+                (Object[]) null
+            );
+            forbiddenWithDifferentMetadata.addMetadata("es.some.other.metadata", "value");
+
+            expectThrows(
+                AssertionError.class,
+                containsString("rest status must be 401 UNAUTHORIZED"),
+                () -> failureHandler.exceptionProcessingRequest(mock(HttpPreRequest.class), forbiddenWithDifferentMetadata, threadContext)
+            );
+            expectThrows(
+                AssertionError.class,
+                containsString("rest status must be 401 UNAUTHORIZED"),
+                () -> failureHandler.exceptionProcessingRequest(
+                    mock(TransportRequest.class),
+                    "index:foo/bar",
+                    forbiddenWithDifferentMetadata,
+                    threadContext
+                )
+            );
+        }
+    }
+
     private void assertWWWAuthenticateWithSchemes(final ElasticsearchSecurityException ese, final String... schemes) {
         assertThat(ese.getBodyHeader("WWW-Authenticate").size(), is(schemes.length));
         assertThat(ese.getBodyHeader("WWW-Authenticate"), contains(schemes));