[Backport 2.x] Feature OnBehalfOf Authentication (#3563)

### Description
Backport feature OnBehalfOf Authentication into `2.x` branch

* Category (Enhancement, New feature, Bug fix, Test fix, Refactoring,
Maintenance, Documentation)
New feature

### Issues Resolved
* Relate #3555 

_For completely resolve the above issue, we still need to backport #3421
into `2.x` branch. However, since that change also includes something
unrelated to OBO Auth, we will backport that separately._

### Testing
Unit Tests + Integration Tests included

### Check List
- [x] New functionality includes testing
- [ ] New functionality has been documented
- [x] Commits are signed per the DCO using --signoff

By submitting this pull request, I confirm that my contribution is made
under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and
signing off your commits, please check
[here](https://github.com/opensearch-project/OpenSearch/blob/main/CONTRIBUTING.md#developer-certificate-of-origin).

---------

Signed-off-by: Ryan Liang <jiallian@amazon.com>

src/integrationTest/java/org/opensearch/security/http/OnBehalfOfJwtAuthenticationTest.java

@@ -0,0 +1,236 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ *
+ * Modifications Copyright OpenSearch Contributors. See
+ * GitHub history for details.
+ */
+
+package org.opensearch.security.http;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.Base64;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import com.carrotsearch.randomizedtesting.annotations.ThreadLeakScope;
+import io.jsonwebtoken.Claims;
+import io.jsonwebtoken.Jwts;
+import org.apache.http.Header;
+import org.apache.http.HttpStatus;
+import org.apache.http.message.BasicHeader;
+import org.junit.Assert;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import org.opensearch.security.authtoken.jwt.EncryptionDecryptionUtil;
+import org.opensearch.test.framework.OnBehalfOfConfig;
+import org.opensearch.test.framework.RolesMapping;
+import org.opensearch.test.framework.TestSecurityConfig;
+import org.opensearch.test.framework.cluster.ClusterManager;
+import org.opensearch.test.framework.cluster.LocalCluster;
+import org.opensearch.test.framework.cluster.TestRestClient;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.aMapWithSize;
+import static org.hamcrest.Matchers.allOf;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.hasKey;
+import static org.junit.Assert.assertTrue;
+import static org.opensearch.security.support.ConfigConstants.SECURITY_ALLOW_DEFAULT_INIT_SECURITYINDEX;
+import static org.opensearch.security.support.ConfigConstants.SECURITY_RESTAPI_ROLES_ENABLED;
+import static org.opensearch.test.framework.TestSecurityConfig.AuthcDomain.AUTHC_HTTPBASIC_INTERNAL;
+import static org.opensearch.test.framework.TestSecurityConfig.Role.ALL_ACCESS;
+
+@RunWith(com.carrotsearch.randomizedtesting.RandomizedRunner.class)
+@ThreadLeakScope(ThreadLeakScope.Scope.NONE)
+public class OnBehalfOfJwtAuthenticationTest {
+
+    public static final String POINTER_USERNAME = "/user_name";
+
+    static final TestSecurityConfig.User ADMIN_USER = new TestSecurityConfig.User("admin").roles(ALL_ACCESS);
+
+    private static Boolean oboEnabled = true;
+    private static final String signingKey = Base64.getEncoder()
+        .encodeToString(
+            "jwt signing key for an on behalf of token authentication backend for testing of OBO authentication".getBytes(
+                StandardCharsets.UTF_8
+            )
+        );
+    private static final String encryptionKey = Base64.getEncoder().encodeToString("encryptionKey".getBytes(StandardCharsets.UTF_8));
+    public static final String ADMIN_USER_NAME = "admin";
+    public static final String OBO_USER_NAME_WITH_PERM = "obo_user";
+    public static final String OBO_USER_NAME_NO_PERM = "obo_user_no_perm";
+    public static final String DEFAULT_PASSWORD = "secret";
+    public static final String NEW_PASSWORD = "testPassword123!!";
+    public static final String OBO_TOKEN_REASON = "{\"description\":\"Test generation\"}";
+    public static final String OBO_ENDPOINT_PREFIX = "_plugins/_security/api/generateonbehalfoftoken";
+    public static final String OBO_DESCRIPTION = "{\"description\":\"Testing\", \"service\":\"self-issued\"}";
+
+    public static final String OBO_DESCRIPTION_WITH_INVALID_DURATIONSECONDS =
+        "{\"description\":\"Testing\", \"service\":\"self-issued\", \"durationSeconds\":\"invalid-seconds\"}";
+
+    public static final String OBO_DESCRIPTION_WITH_INVALID_PARAMETERS =
+        "{\"description\":\"Testing\", \"service\":\"self-issued\", \"invalidParameter\":\"invalid-parameter\"}";
+
+    public static final String HOST_MAPPING_IP = "127.0.0.1";
+    public static final String OBO_USER_NAME_WITH_HOST_MAPPING = "obo_user_with_ip_role_mapping";
+    public static final String CURRENT_AND_NEW_PASSWORDS = "{ \"current_password\": \""
+        + DEFAULT_PASSWORD
+        + "\", \"password\": \""
+        + NEW_PASSWORD
+        + "\" }";
+
+    private static final TestSecurityConfig.Role ROLE_WITH_OBO_PERM = new TestSecurityConfig.Role("obo_access_role").clusterPermissions(
+        "security:obo/create"
+    );
+
+    private static final TestSecurityConfig.Role ROLE_WITH_NO_OBO_PERM = new TestSecurityConfig.Role("obo_user_no_perm");
+
+    protected final static TestSecurityConfig.User OBO_USER = new TestSecurityConfig.User(OBO_USER_NAME_WITH_PERM).roles(
+        ROLE_WITH_OBO_PERM
+    );
+
+    protected final static TestSecurityConfig.User OBO_USER_NO_PERM = new TestSecurityConfig.User(OBO_USER_NAME_NO_PERM).roles(
+        ROLE_WITH_NO_OBO_PERM
+    );
+
+    private static final TestSecurityConfig.Role HOST_MAPPING_ROLE = new TestSecurityConfig.Role("host_mapping_role");
+
+    protected final static TestSecurityConfig.User HOST_MAPPING_OBO_USER = new TestSecurityConfig.User(OBO_USER_NAME_WITH_HOST_MAPPING)
+        .roles(HOST_MAPPING_ROLE, ROLE_WITH_OBO_PERM);
+
+    @ClassRule
+    public static final LocalCluster cluster = new LocalCluster.Builder().clusterManager(ClusterManager.SINGLENODE)
+        .anonymousAuth(false)
+        .users(ADMIN_USER, OBO_USER, OBO_USER_NO_PERM, HOST_MAPPING_OBO_USER)
+        .nodeSettings(
+            Map.of(SECURITY_ALLOW_DEFAULT_INIT_SECURITYINDEX, true, SECURITY_RESTAPI_ROLES_ENABLED, List.of("user_admin__all_access"))
+        )
+        .authc(AUTHC_HTTPBASIC_INTERNAL)
+        .rolesMapping(new RolesMapping(HOST_MAPPING_ROLE).hostIPs(HOST_MAPPING_IP))
+        .onBehalfOf(new OnBehalfOfConfig().oboEnabled(oboEnabled).signingKey(signingKey).encryptionKey(encryptionKey))
+        .build();
+
+    @Test
+    public void shouldAuthenticateWithOBOTokenEndPoint() {
+        String oboToken = generateOboToken(ADMIN_USER_NAME, DEFAULT_PASSWORD);
+        Header adminOboAuthHeader = new BasicHeader("Authorization", "Bearer " + oboToken);
+        authenticateWithOboToken(adminOboAuthHeader, ADMIN_USER_NAME, HttpStatus.SC_OK);
+    }
+
+    @Test
+    public void shouldNotAuthenticateWithATemperedOBOToken() {
+        String oboToken = generateOboToken(ADMIN_USER_NAME, DEFAULT_PASSWORD);
+        oboToken = oboToken.substring(0, oboToken.length() - 1); // tampering the token
+        Header adminOboAuthHeader = new BasicHeader("Authorization", "Bearer " + oboToken);
+        authenticateWithOboToken(adminOboAuthHeader, ADMIN_USER_NAME, HttpStatus.SC_UNAUTHORIZED);
+    }
+
+    @Test
+    public void shouldNotAuthenticateForUsingOBOTokenToAccessOBOEndpoint() {
+        String oboToken = generateOboToken(ADMIN_USER_NAME, DEFAULT_PASSWORD);
+        Header adminOboAuthHeader = new BasicHeader("Authorization", "Bearer " + oboToken);
+
+        try (TestRestClient client = cluster.getRestClient(adminOboAuthHeader)) {
+            TestRestClient.HttpResponse response = client.getOnBehalfOfToken(OBO_DESCRIPTION, adminOboAuthHeader);
+            response.assertStatusCode(HttpStatus.SC_UNAUTHORIZED);
+        }
+    }
+
+    @Test
+    public void shouldNotAuthenticateForUsingOBOTokenToAccessAccountEndpoint() {
+        String oboToken = generateOboToken(ADMIN_USER_NAME, DEFAULT_PASSWORD);
+        Header adminOboAuthHeader = new BasicHeader("Authorization", "Bearer " + oboToken);
+
+        try (TestRestClient client = cluster.getRestClient(adminOboAuthHeader)) {
+            TestRestClient.HttpResponse response = client.changeInternalUserPassword(CURRENT_AND_NEW_PASSWORDS, adminOboAuthHeader);
+            response.assertStatusCode(HttpStatus.SC_UNAUTHORIZED);
+        }
+    }
+
+    @Test
+    public void shouldAuthenticateForNonAdminUserWithOBOPermission() {
+        String oboToken = generateOboToken(OBO_USER_NAME_WITH_PERM, DEFAULT_PASSWORD);
+        Header oboAuthHeader = new BasicHeader("Authorization", "Bearer " + oboToken);
+        authenticateWithOboToken(oboAuthHeader, OBO_USER_NAME_WITH_PERM, HttpStatus.SC_OK);
+    }
+
+    @Test
+    public void shouldNotAuthenticateForNonAdminUserWithoutOBOPermission() {
+        try (TestRestClient client = cluster.getRestClient(OBO_USER_NO_PERM)) {
+            assertThat(client.post(OBO_ENDPOINT_PREFIX).getStatusCode(), equalTo(HttpStatus.SC_UNAUTHORIZED));
+        }
+    }
+
+    @Test
+    public void shouldNotIncludeRolesFromHostMappingInOBOToken() {
+        String oboToken = generateOboToken(OBO_USER_NAME_WITH_HOST_MAPPING, DEFAULT_PASSWORD);
+
+        Claims claims = Jwts.parserBuilder().setSigningKey(signingKey).build().parseClaimsJws(oboToken).getBody();
+
+        Object er = claims.get("er");
+        EncryptionDecryptionUtil encryptionDecryptionUtil = new EncryptionDecryptionUtil(encryptionKey);
+        String rolesClaim = encryptionDecryptionUtil.decrypt(er.toString());
+        List<String> roles = Arrays.stream(rolesClaim.split(","))
+            .map(String::trim)
+            .filter(s -> !s.isEmpty())
+            .collect(Collectors.toUnmodifiableList());
+
+        Assert.assertFalse(roles.contains("host_mapping_role"));
+    }
+
+    @Test
+    public void shouldNotAuthenticateWithInvalidDurationSeconds() {
+        try (TestRestClient client = cluster.getRestClient(ADMIN_USER_NAME, DEFAULT_PASSWORD)) {
+            client.assertCorrectCredentials(ADMIN_USER_NAME);
+            TestRestClient.HttpResponse response = client.postJson(OBO_ENDPOINT_PREFIX, OBO_DESCRIPTION_WITH_INVALID_DURATIONSECONDS);
+            response.assertStatusCode(HttpStatus.SC_BAD_REQUEST);
+            Map<String, Object> oboEndPointResponse = (Map<String, Object>) response.getBodyAs(Map.class);
+            assertTrue(oboEndPointResponse.containsValue("durationSeconds must be an integer."));
+        }
+    }
+
+    @Test
+    public void shouldNotAuthenticateWithInvalidAPIParameter() {
+        try (TestRestClient client = cluster.getRestClient(ADMIN_USER_NAME, DEFAULT_PASSWORD)) {
+            client.assertCorrectCredentials(ADMIN_USER_NAME);
+            TestRestClient.HttpResponse response = client.postJson(OBO_ENDPOINT_PREFIX, OBO_DESCRIPTION_WITH_INVALID_PARAMETERS);
+            response.assertStatusCode(HttpStatus.SC_BAD_REQUEST);
+            Map<String, Object> oboEndPointResponse = (Map<String, Object>) response.getBodyAs(Map.class);
+            assertTrue(oboEndPointResponse.containsValue("Unrecognized parameter: invalidParameter"));
+        }
+    }
+
+    private String generateOboToken(String username, String password) {
+        try (TestRestClient client = cluster.getRestClient(username, password)) {
+            client.assertCorrectCredentials(username);
+            TestRestClient.HttpResponse response = client.postJson(OBO_ENDPOINT_PREFIX, OBO_TOKEN_REASON);
+            response.assertStatusCode(HttpStatus.SC_OK);
+            Map<String, Object> oboEndPointResponse = (Map<String, Object>) response.getBodyAs(Map.class);
+            assertThat(
+                oboEndPointResponse,
+                allOf(aMapWithSize(3), hasKey("user"), hasKey("authenticationToken"), hasKey("durationSeconds"))
+            );
+            return oboEndPointResponse.get("authenticationToken").toString();
+        }
+    }
+
+    private void authenticateWithOboToken(Header authHeader, String expectedUsername, int expectedStatusCode) {
+        try (TestRestClient client = cluster.getRestClient(authHeader)) {
+            TestRestClient.HttpResponse response = client.getAuthInfo();
+            response.assertStatusCode(expectedStatusCode);
+            assertThat(response.getStatusCode(), equalTo(expectedStatusCode));
+            if (expectedStatusCode == HttpStatus.SC_OK) {
+                String username = response.getTextFromJsonBody(POINTER_USERNAME);
+                assertThat(username, equalTo(expectedUsername));
+            }
+        }
+    }
+}

src/integrationTest/java/org/opensearch/test/framework/TestSecurityConfig.java

@@ -114,6 +114,11 @@ public TestSecurityConfig xff(XffConfig xffConfig) {
         return this;
     }
 
+    public TestSecurityConfig onBehalfOf(OnBehalfOfConfig onBehalfOfConfig) {
+        config.onBehalfOfConfig(onBehalfOfConfig);
+        return this;
+    }
+
     public TestSecurityConfig authc(AuthcDomain authcDomain) {
         config.authc(authcDomain);
         return this;
@@ -170,6 +175,7 @@ public static class Config implements ToXContentObject {
 
         private Boolean doNotFailOnForbidden;
         private XffConfig xffConfig;
+        private OnBehalfOfConfig onBehalfOfConfig;
         private Map<String, AuthcDomain> authcDomainMap = new LinkedHashMap<>();
 
         private AuthFailureListeners authFailureListeners;
@@ -190,6 +196,11 @@ public Config xffConfig(XffConfig xffConfig) {
             return this;
         }
 
+        public Config onBehalfOfConfig(OnBehalfOfConfig onBehalfOfConfig) {
+            this.onBehalfOfConfig = onBehalfOfConfig;
+            return this;
+        }
+
         public Config authc(AuthcDomain authcDomain) {
             authcDomainMap.put(authcDomain.id, authcDomain);
             return this;
@@ -210,6 +221,10 @@ public XContentBuilder toXContent(XContentBuilder xContentBuilder, Params params
             xContentBuilder.startObject();
             xContentBuilder.startObject("dynamic");
 
+            if (onBehalfOfConfig != null) {
+                xContentBuilder.field("on_behalf_of", onBehalfOfConfig);
+            }
+
             if (anonymousAuth || (xffConfig != null)) {
                 xContentBuilder.startObject("http");
                 xContentBuilder.field("anonymous_auth_enabled", anonymousAuth);

src/integrationTest/java/org/opensearch/test/framework/cluster/LocalCluster.java

@@ -55,6 +55,7 @@
 import org.opensearch.test.framework.AuditConfiguration;
 import org.opensearch.test.framework.AuthFailureListeners;
 import org.opensearch.test.framework.AuthzDomain;
+import org.opensearch.test.framework.OnBehalfOfConfig;
 import org.opensearch.test.framework.RolesMapping;
 import org.opensearch.test.framework.TestIndex;
 import org.opensearch.test.framework.TestSecurityConfig;
@@ -471,6 +472,11 @@ public Builder xff(XffConfig xffConfig) {
             return this;
         }
 
+        public Builder onBehalfOf(OnBehalfOfConfig onBehalfOfConfig) {
+            testSecurityConfig.onBehalfOf(onBehalfOfConfig);
+            return this;
+        }
+
         public Builder loadConfigurationIntoIndex(boolean loadConfigurationIntoIndex) {
             this.loadConfigurationIntoIndex = loadConfigurationIntoIndex;
             return this;

src/integrationTest/resources/config.yml

@@ -15,3 +15,8 @@ config:
         authentication_backend:
           type: "internal"
           config: {}
+    on_behalf_of:
+      # The decoded signing key is: This is the jwt signing key for an on behalf of token authentication backend for testing of extensions
+      # The decoded encryption key is: encryptionKey
+      signing_key: "VGhpcyBpcyB0aGUgand0IHNpZ25pbmcga2V5IGZvciBhbiBvbiBiZWhhbGYgb2YgdG9rZW4gYXV0aGVudGljYXRpb24gYmFja2VuZCBmb3IgdGVzdGluZyBvZiBleHRlbnNpb25z"
+      encryption_key: "ZW5jcnlwdGlvbktleQ=="