diff --git a/connect/basic-auth-extension/src/main/java/org/apache/kafka/connect/rest/basic/auth/extension/JaasBasicAuthFilter.java b/connect/basic-auth-extension/src/main/java/org/apache/kafka/connect/rest/basic/auth/extension/JaasBasicAuthFilter.java index 6167434b98031..d5b15c6c65a5f 100644 --- a/connect/basic-auth-extension/src/main/java/org/apache/kafka/connect/rest/basic/auth/extension/JaasBasicAuthFilter.java +++ b/connect/basic-auth-extension/src/main/java/org/apache/kafka/connect/rest/basic/auth/extension/JaasBasicAuthFilter.java @@ -17,6 +17,8 @@ package org.apache.kafka.connect.rest.basic.auth.extension; +import java.util.regex.Pattern; +import javax.ws.rs.HttpMethod; import org.apache.kafka.common.config.ConfigException; import java.io.IOException; @@ -35,18 +37,18 @@ import javax.ws.rs.core.Response; public class JaasBasicAuthFilter implements ContainerRequestFilter { - private static final String CONNECT_LOGIN_MODULE = "KafkaConnect"; static final String AUTHORIZATION = "Authorization"; - + private static final Pattern TASK_REQUEST_PATTERN = Pattern.compile("/?connectors/([^/]+)/tasks/?"); @Override public void filter(ContainerRequestContext requestContext) throws IOException { - try { - LoginContext loginContext = - new LoginContext(CONNECT_LOGIN_MODULE, new BasicAuthCallBackHandler( - requestContext.getHeaderString(AUTHORIZATION))); - loginContext.login(); + if (!(requestContext.getMethod().equals(HttpMethod.POST) && TASK_REQUEST_PATTERN.matcher(requestContext.getUriInfo().getPath()).matches())) { + LoginContext loginContext = + new LoginContext(CONNECT_LOGIN_MODULE, new BasicAuthCallBackHandler( + requestContext.getHeaderString(AUTHORIZATION))); + loginContext.login(); + } } catch (LoginException | ConfigException e) { requestContext.abortWith( Response.status(Response.Status.UNAUTHORIZED) diff --git a/connect/basic-auth-extension/src/test/java/org/apache/kafka/connect/rest/basic/auth/extension/JaasBasicAuthFilterTest.java b/connect/basic-auth-extension/src/test/java/org/apache/kafka/connect/rest/basic/auth/extension/JaasBasicAuthFilterTest.java index 835bef0f37c14..fe5f8b9de5e4e 100644 --- a/connect/basic-auth-extension/src/test/java/org/apache/kafka/connect/rest/basic/auth/extension/JaasBasicAuthFilterTest.java +++ b/connect/basic-auth-extension/src/test/java/org/apache/kafka/connect/rest/basic/auth/extension/JaasBasicAuthFilterTest.java @@ -17,12 +17,15 @@ package org.apache.kafka.connect.rest.basic.auth.extension; +import javax.ws.rs.HttpMethod; +import javax.ws.rs.core.UriInfo; import org.apache.kafka.common.security.JaasUtils; import org.easymock.EasyMock; import org.junit.After; import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; +import org.powermock.api.easymock.PowerMock; import org.powermock.api.easymock.annotation.MockStrict; import org.powermock.core.classloader.annotations.PowerMockIgnore; import org.powermock.modules.junit4.PowerMockRunner; @@ -52,6 +55,9 @@ public class JaasBasicAuthFilterTest { private String previousJaasConfig; private Configuration previousConfiguration; + @MockStrict + private UriInfo uriInfo; + @Before public void setup() { EasyMock.reset(requestContext); @@ -137,7 +143,34 @@ public void testNoFileOption() throws IOException { jaasBasicAuthFilter.filter(requestContext); } + @Test + public void testPostWithoutAppropriateCredential() throws IOException { + EasyMock.expect(requestContext.getMethod()).andReturn(HttpMethod.POST); + EasyMock.expect(requestContext.getUriInfo()).andReturn(uriInfo); + EasyMock.expect(uriInfo.getPath()).andReturn("connectors/connName/tasks"); + + PowerMock.replayAll(); + jaasBasicAuthFilter.filter(requestContext); + EasyMock.verify(requestContext); + } + + @Test + public void testPostNotChangingConnectorTask() throws IOException { + EasyMock.expect(requestContext.getMethod()).andReturn(HttpMethod.POST); + EasyMock.expect(requestContext.getUriInfo()).andReturn(uriInfo); + EasyMock.expect(uriInfo.getPath()).andReturn("local:randomport/connectors/connName"); + String authHeader = "Basic" + Base64.getEncoder().encodeToString(("user" + ":" + "password").getBytes()); + EasyMock.expect(requestContext.getHeaderString(JaasBasicAuthFilter.AUTHORIZATION)) + .andReturn(authHeader); + requestContext.abortWith(EasyMock.anyObject(Response.class)); + EasyMock.expectLastCall(); + PowerMock.replayAll(); + jaasBasicAuthFilter.filter(requestContext); + EasyMock.verify(requestContext); + } + private void setMock(String authorization, String username, String password, boolean exceptionCase) { + EasyMock.expect(requestContext.getMethod()).andReturn(HttpMethod.GET); String authHeader = authorization + " " + Base64.getEncoder().encodeToString((username + ":" + password).getBytes()); EasyMock.expect(requestContext.getHeaderString(JaasBasicAuthFilter.AUTHORIZATION)) .andReturn(authHeader); @@ -152,7 +185,6 @@ private void setupJaasConfig(String loginModule, String credentialFilePath, bool File jaasConfigFile = File.createTempFile("ks-jaas-", ".conf"); jaasConfigFile.deleteOnExit(); previousJaasConfig = System.setProperty(JaasUtils.JAVA_LOGIN_CONFIG_PARAM, jaasConfigFile.getPath()); - List lines; lines = new ArrayList<>(); lines.add(loginModule + " { org.apache.kafka.connect.rest.basic.auth.extension.PropertyFileLoginModule required "); diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java index 52709f7856fbb..8e59d87153a33 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java @@ -1202,7 +1202,7 @@ public void run() { return; } String reconfigUrl = RestServer.urlJoin(leaderUrl, "/connectors/" + connName + "/tasks"); - RestClient.httpRequest(reconfigUrl, "POST", rawTaskProps, null, config); + RestClient.httpRequest(reconfigUrl, "POST", null, rawTaskProps, null, config); cb.onCompletion(null, null); } catch (ConnectException e) { log.error("Request to leader to reconfigure connector tasks failed", e); diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/RestClient.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/RestClient.java index c1b6036b28563..7ef1c78cf0031 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/RestClient.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/RestClient.java @@ -19,6 +19,7 @@ import com.fasterxml.jackson.core.type.TypeReference; import com.fasterxml.jackson.databind.ObjectMapper; +import javax.ws.rs.core.HttpHeaders; import org.apache.kafka.connect.runtime.WorkerConfig; import org.apache.kafka.connect.runtime.rest.entities.ErrorMessage; import org.apache.kafka.connect.runtime.rest.errors.ConnectRestException; @@ -50,12 +51,13 @@ public class RestClient { * * @param url HTTP connection will be established with this url. * @param method HTTP method ("GET", "POST", "PUT", etc.) + * @param headers HTTP headers from REST endpoint * @param requestBodyData Object to serialize as JSON and send in the request body. * @param responseFormat Expected format of the response to the HTTP request. * @param The type of the deserialized response to the HTTP request. * @return The deserialized response to the HTTP request, or null if no data is expected. */ - public static HttpResponse httpRequest(String url, String method, Object requestBodyData, + public static HttpResponse httpRequest(String url, String method, HttpHeaders headers, Object requestBodyData, TypeReference responseFormat, WorkerConfig config) { HttpClient client; @@ -82,6 +84,8 @@ public static HttpResponse httpRequest(String url, String method, Object req.method(method); req.accept("application/json"); req.agent("kafka-connect"); + addHeadersToRequest(headers, req); + if (serializedBody != null) { req.content(new StringContentProvider(serializedBody, StandardCharsets.UTF_8), "application/json"); } @@ -116,6 +120,21 @@ public static HttpResponse httpRequest(String url, String method, Object } } + + /** + * Extract headers from REST call and add to client request + * @param headers Headers from REST endpoint + * @param req The client request to modify + */ + private static void addHeadersToRequest(HttpHeaders headers, Request req) { + if (headers != null) { + String credentialAuthorization = headers.getHeaderString(HttpHeaders.AUTHORIZATION); + if (credentialAuthorization != null) { + req.header(HttpHeaders.AUTHORIZATION, credentialAuthorization); + } + } + } + /** * Convert response parameters from Jetty format (HttpFields) * @param httpFields diff --git a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/resources/ConnectorsResource.java b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/resources/ConnectorsResource.java index 61cf5da83a190..96a4f51eec312 100644 --- a/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/resources/ConnectorsResource.java +++ b/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/resources/ConnectorsResource.java @@ -18,6 +18,7 @@ import com.fasterxml.jackson.core.type.TypeReference; +import javax.ws.rs.core.HttpHeaders; import org.apache.kafka.connect.errors.NotFoundException; import org.apache.kafka.connect.runtime.ConnectorConfig; import org.apache.kafka.connect.runtime.Herder; @@ -85,7 +86,8 @@ public ConnectorsResource(Herder herder, WorkerConfig config) { @GET @Path("/") public Response listConnectors( - final @Context UriInfo uriInfo + final @Context UriInfo uriInfo, + final @Context HttpHeaders headers ) throws Throwable { if (uriInfo.getQueryParameters().containsKey("expand")) { Map> out = new HashMap<>(); @@ -121,6 +123,7 @@ public Response listConnectors( @POST @Path("/") public Response createConnector(final @QueryParam("forward") Boolean forward, + final @Context HttpHeaders headers, final CreateConnectorRequest createRequest) throws Throwable { // Trim leading and trailing whitespaces from the connector name, replace null with empty string // if no name element present to keep validation within validator (NonEmptyStringWithoutControlChars @@ -132,7 +135,7 @@ public Response createConnector(final @QueryParam("forward") Boolean forward, FutureCallback> cb = new FutureCallback<>(); herder.putConnectorConfig(name, configs, false, cb); - Herder.Created info = completeOrForwardRequest(cb, "/connectors", "POST", createRequest, + Herder.Created info = completeOrForwardRequest(cb, "/connectors", "POST", headers, createRequest, new TypeReference() { }, new CreatedConnectorInfoTranslator(), forward); URI location = UriBuilder.fromUri("/connectors").path(name).build(); @@ -142,19 +145,21 @@ public Response createConnector(final @QueryParam("forward") Boolean forward, @GET @Path("/{connector}") public ConnectorInfo getConnector(final @PathParam("connector") String connector, + final @Context HttpHeaders headers, final @QueryParam("forward") Boolean forward) throws Throwable { FutureCallback cb = new FutureCallback<>(); herder.connectorInfo(connector, cb); - return completeOrForwardRequest(cb, "/connectors/" + connector, "GET", null, forward); + return completeOrForwardRequest(cb, "/connectors/" + connector, "GET", headers, null, forward); } @GET @Path("/{connector}/config") public Map getConnectorConfig(final @PathParam("connector") String connector, + final @Context HttpHeaders headers, final @QueryParam("forward") Boolean forward) throws Throwable { FutureCallback> cb = new FutureCallback<>(); herder.connectorConfig(connector, cb); - return completeOrForwardRequest(cb, "/connectors/" + connector + "/config", "GET", null, forward); + return completeOrForwardRequest(cb, "/connectors/" + connector + "/config", "GET", headers, null, forward); } @GET @@ -166,6 +171,7 @@ public ConnectorStateInfo getConnectorStatus(final @PathParam("connector") Strin @PUT @Path("/{connector}/config") public Response putConnectorConfig(final @PathParam("connector") String connector, + final @Context HttpHeaders headers, final @QueryParam("forward") Boolean forward, final Map connectorConfig) throws Throwable { FutureCallback> cb = new FutureCallback<>(); @@ -173,7 +179,7 @@ public Response putConnectorConfig(final @PathParam("connector") String connecto herder.putConnectorConfig(connector, connectorConfig, true, cb); Herder.Created createdInfo = completeOrForwardRequest(cb, "/connectors/" + connector + "/config", - "PUT", connectorConfig, new TypeReference() { }, new CreatedConnectorInfoTranslator(), forward); + "PUT", headers, connectorConfig, new TypeReference() { }, new CreatedConnectorInfoTranslator(), forward); Response.ResponseBuilder response; if (createdInfo.created()) { URI location = UriBuilder.fromUri("/connectors").path(connector).build(); @@ -187,15 +193,16 @@ public Response putConnectorConfig(final @PathParam("connector") String connecto @POST @Path("/{connector}/restart") public void restartConnector(final @PathParam("connector") String connector, + final @Context HttpHeaders headers, final @QueryParam("forward") Boolean forward) throws Throwable { FutureCallback cb = new FutureCallback<>(); herder.restartConnector(connector, cb); - completeOrForwardRequest(cb, "/connectors/" + connector + "/restart", "POST", null, forward); + completeOrForwardRequest(cb, "/connectors/" + connector + "/restart", "POST", headers, null, forward); } @PUT @Path("/{connector}/pause") - public Response pauseConnector(@PathParam("connector") String connector) { + public Response pauseConnector(@PathParam("connector") String connector, final @Context HttpHeaders headers) { herder.pauseConnector(connector); return Response.accepted().build(); } @@ -210,26 +217,29 @@ public Response resumeConnector(@PathParam("connector") String connector) { @GET @Path("/{connector}/tasks") public List getTaskConfigs(final @PathParam("connector") String connector, + final @Context HttpHeaders headers, final @QueryParam("forward") Boolean forward) throws Throwable { FutureCallback> cb = new FutureCallback<>(); herder.taskConfigs(connector, cb); - return completeOrForwardRequest(cb, "/connectors/" + connector + "/tasks", "GET", null, new TypeReference>() { + return completeOrForwardRequest(cb, "/connectors/" + connector + "/tasks", "GET", headers, null, new TypeReference>() { }, forward); } @POST @Path("/{connector}/tasks") public void putTaskConfigs(final @PathParam("connector") String connector, + final @Context HttpHeaders headers, final @QueryParam("forward") Boolean forward, final List> taskConfigs) throws Throwable { FutureCallback cb = new FutureCallback<>(); herder.putTaskConfigs(connector, taskConfigs, cb); - completeOrForwardRequest(cb, "/connectors/" + connector + "/tasks", "POST", taskConfigs, forward); + completeOrForwardRequest(cb, "/connectors/" + connector + "/tasks", "POST", headers, taskConfigs, forward); } @GET @Path("/{connector}/tasks/{task}/status") public ConnectorStateInfo.TaskState getTaskStatus(final @PathParam("connector") String connector, + final @Context HttpHeaders headers, final @PathParam("task") Integer task) throws Throwable { return herder.taskStatus(new ConnectorTaskId(connector, task)); } @@ -238,20 +248,22 @@ public ConnectorStateInfo.TaskState getTaskStatus(final @PathParam("connector") @Path("/{connector}/tasks/{task}/restart") public void restartTask(final @PathParam("connector") String connector, final @PathParam("task") Integer task, + final @Context HttpHeaders headers, final @QueryParam("forward") Boolean forward) throws Throwable { FutureCallback cb = new FutureCallback<>(); ConnectorTaskId taskId = new ConnectorTaskId(connector, task); herder.restartTask(taskId, cb); - completeOrForwardRequest(cb, "/connectors/" + connector + "/tasks/" + task + "/restart", "POST", null, forward); + completeOrForwardRequest(cb, "/connectors/" + connector + "/tasks/" + task + "/restart", "POST", headers, null, forward); } @DELETE @Path("/{connector}") public void destroyConnector(final @PathParam("connector") String connector, + final @Context HttpHeaders headers, final @QueryParam("forward") Boolean forward) throws Throwable { FutureCallback> cb = new FutureCallback<>(); herder.deleteConnectorConfig(connector, cb); - completeOrForwardRequest(cb, "/connectors/" + connector, "DELETE", null, forward); + completeOrForwardRequest(cb, "/connectors/" + connector, "DELETE", headers, null, forward); } // Check whether the connector name from the url matches the one (if there is one) provided in the connectorconfig @@ -271,6 +283,7 @@ private void checkAndPutConnectorConfigName(String connectorName, Map T completeOrForwardRequest(FutureCallback cb, String path, String method, + HttpHeaders headers, Object body, TypeReference resultType, Translator translator, @@ -293,7 +306,7 @@ private T completeOrForwardRequest(FutureCallback cb, .build() .toString(); log.debug("Forwarding request {} {} {}", forwardUrl, method, body); - return translator.translate(RestClient.httpRequest(forwardUrl, method, body, resultType, config)); + return translator.translate(RestClient.httpRequest(forwardUrl, method, headers, body, resultType, config)); } else { // we should find the right target for the query within two hops, so if // we don't, it probably means that a rebalance has taken place. @@ -315,14 +328,14 @@ private T completeOrForwardRequest(FutureCallback cb, } } - private T completeOrForwardRequest(FutureCallback cb, String path, String method, Object body, + private T completeOrForwardRequest(FutureCallback cb, String path, String method, HttpHeaders headers, Object body, TypeReference resultType, Boolean forward) throws Throwable { - return completeOrForwardRequest(cb, path, method, body, resultType, new IdentityTranslator(), forward); + return completeOrForwardRequest(cb, path, method, headers, body, resultType, new IdentityTranslator(), forward); } - private T completeOrForwardRequest(FutureCallback cb, String path, String method, + private T completeOrForwardRequest(FutureCallback cb, String path, String method, HttpHeaders headers, Object body, Boolean forward) throws Throwable { - return completeOrForwardRequest(cb, path, method, body, null, new IdentityTranslator(), forward); + return completeOrForwardRequest(cb, path, method, headers, body, null, new IdentityTranslator(), forward); } private interface Translator { diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/rest/resources/ConnectorPluginsResourceTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/rest/resources/ConnectorPluginsResourceTest.java index a3aee6a407d2e..2882355eccbec 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/rest/resources/ConnectorPluginsResourceTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/rest/resources/ConnectorPluginsResourceTest.java @@ -18,6 +18,7 @@ import com.fasterxml.jackson.core.type.TypeReference; import com.fasterxml.jackson.databind.ObjectMapper; +import javax.ws.rs.core.HttpHeaders; import org.apache.kafka.common.config.Config; import org.apache.kafka.common.config.ConfigDef; import org.apache.kafka.common.config.ConfigDef.Importance; @@ -182,7 +183,7 @@ public class ConnectorPluginsResourceTest { @Before public void setUp() throws Exception { PowerMock.mockStatic(RestClient.class, - RestClient.class.getMethod("httpRequest", String.class, String.class, Object.class, TypeReference.class, WorkerConfig.class)); + RestClient.class.getMethod("httpRequest", String.class, String.class, HttpHeaders.class, Object.class, TypeReference.class, WorkerConfig.class)); plugins = PowerMock.createMock(Plugins.class); herder = PowerMock.createMock(AbstractHerder.class); diff --git a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/rest/resources/ConnectorsResourceTest.java b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/rest/resources/ConnectorsResourceTest.java index 5dc7f1ee687b9..5490df03afb6d 100644 --- a/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/rest/resources/ConnectorsResourceTest.java +++ b/connect/runtime/src/test/java/org/apache/kafka/connect/runtime/rest/resources/ConnectorsResourceTest.java @@ -18,6 +18,7 @@ import com.fasterxml.jackson.core.type.TypeReference; +import javax.ws.rs.core.HttpHeaders; import org.apache.kafka.connect.errors.AlreadyExistsException; import org.apache.kafka.connect.errors.NotFoundException; import org.apache.kafka.connect.runtime.ConnectorConfig; @@ -79,6 +80,7 @@ public class ConnectorsResourceTest { private static final String CONNECTOR_NAME_PADDING_WHITESPACES = " " + CONNECTOR_NAME + " \n "; private static final Boolean FORWARD = true; private static final Map CONNECTOR_CONFIG_SPECIAL_CHARS = new HashMap<>(); + private static final HttpHeaders NULL_HEADERS = null; static { CONNECTOR_CONFIG_SPECIAL_CHARS.put("name", CONNECTOR_NAME_SPECIAL_CHARS); CONNECTOR_CONFIG_SPECIAL_CHARS.put("sample_config", "test_config"); @@ -130,7 +132,7 @@ public class ConnectorsResourceTest { @Before public void setUp() throws NoSuchMethodException { PowerMock.mockStatic(RestClient.class, - RestClient.class.getMethod("httpRequest", String.class, String.class, Object.class, TypeReference.class, WorkerConfig.class)); + RestClient.class.getMethod("httpRequest", String.class, String.class, HttpHeaders.class, Object.class, TypeReference.class, WorkerConfig.class)); connectorsResource = new ConnectorsResource(herder, null); forward = EasyMock.mock(UriInfo.class); MultivaluedMap queryParams = new MultivaluedHashMap<>(); @@ -151,7 +153,7 @@ public void testListConnectors() throws Throwable { PowerMock.replayAll(); - Collection connectors = (Collection) connectorsResource.listConnectors(forward).getEntity(); + Collection connectors = (Collection) connectorsResource.listConnectors(forward, NULL_HEADERS).getEntity(); // Ordering isn't guaranteed, compare sets assertEquals(new HashSet<>(Arrays.asList(CONNECTOR_NAME, CONNECTOR2_NAME)), new HashSet<>(connectors)); @@ -174,7 +176,7 @@ public void testExpandConnectorsStatus() throws Throwable { PowerMock.replayAll(); - Map> expanded = (Map>) connectorsResource.listConnectors(forward).getEntity(); + Map> expanded = (Map>) connectorsResource.listConnectors(forward, NULL_HEADERS).getEntity(); // Ordering isn't guaranteed, compare sets assertEquals(new HashSet<>(Arrays.asList(CONNECTOR_NAME, CONNECTOR2_NAME)), expanded.keySet()); assertEquals(connector2, expanded.get(CONNECTOR2_NAME).get("status")); @@ -198,7 +200,7 @@ public void testExpandConnectorsInfo() throws Throwable { PowerMock.replayAll(); - Map> expanded = (Map>) connectorsResource.listConnectors(forward).getEntity(); + Map> expanded = (Map>) connectorsResource.listConnectors(forward, NULL_HEADERS).getEntity(); // Ordering isn't guaranteed, compare sets assertEquals(new HashSet<>(Arrays.asList(CONNECTOR_NAME, CONNECTOR2_NAME)), expanded.keySet()); assertEquals(connector2, expanded.get(CONNECTOR2_NAME).get("info")); @@ -226,7 +228,7 @@ public void testFullExpandConnectors() throws Throwable { PowerMock.replayAll(); - Map> expanded = (Map>) connectorsResource.listConnectors(forward).getEntity(); + Map> expanded = (Map>) connectorsResource.listConnectors(forward, NULL_HEADERS).getEntity(); // Ordering isn't guaranteed, compare sets assertEquals(new HashSet<>(Arrays.asList(CONNECTOR_NAME, CONNECTOR2_NAME)), expanded.keySet()); assertEquals(connectorInfo2, expanded.get(CONNECTOR2_NAME).get("info")); @@ -252,7 +254,7 @@ public void testExpandConnectorsWithConnectorNotFound() throws Throwable { PowerMock.replayAll(); - Map> expanded = (Map>) connectorsResource.listConnectors(forward).getEntity(); + Map> expanded = (Map>) connectorsResource.listConnectors(forward, NULL_HEADERS).getEntity(); // Ordering isn't guaranteed, compare sets assertEquals(Collections.singleton(CONNECTOR2_NAME), expanded.keySet()); assertEquals(connector2, expanded.get(CONNECTOR2_NAME).get("status")); @@ -271,7 +273,7 @@ public void testCreateConnector() throws Throwable { PowerMock.replayAll(); - connectorsResource.createConnector(FORWARD, body); + connectorsResource.createConnector(FORWARD, NULL_HEADERS, body); PowerMock.verifyAll(); } @@ -284,19 +286,57 @@ public void testCreateConnectorNotLeader() throws Throwable { herder.putConnectorConfig(EasyMock.eq(CONNECTOR_NAME), EasyMock.eq(body.config()), EasyMock.eq(false), EasyMock.capture(cb)); expectAndCallbackNotLeaderException(cb); // Should forward request - EasyMock.expect(RestClient.httpRequest(EasyMock.eq("http://leader:8083/connectors?forward=false"), EasyMock.eq("POST"), EasyMock.eq(body), EasyMock.anyObject(), EasyMock.anyObject(WorkerConfig.class))) + EasyMock.expect(RestClient.httpRequest(EasyMock.eq("http://leader:8083/connectors?forward=false"), EasyMock.eq("POST"), EasyMock.isNull(), EasyMock.eq(body), EasyMock.anyObject(), EasyMock.anyObject(WorkerConfig.class))) .andReturn(new RestClient.HttpResponse<>(201, new HashMap(), new ConnectorInfo(CONNECTOR_NAME, CONNECTOR_CONFIG, CONNECTOR_TASK_NAMES, ConnectorType.SOURCE))); PowerMock.replayAll(); - connectorsResource.createConnector(FORWARD, body); + connectorsResource.createConnector(FORWARD, NULL_HEADERS, body); PowerMock.verifyAll(); } + @Test + public void testCreateConnectorWithHeaderAuthorization() throws Throwable { + CreateConnectorRequest body = new CreateConnectorRequest(CONNECTOR_NAME, Collections.singletonMap(ConnectorConfig.NAME_CONFIG, CONNECTOR_NAME)); + final Capture>> cb = Capture.newInstance(); + HttpHeaders httpHeaders = EasyMock.mock(HttpHeaders.class); + EasyMock.expect(httpHeaders.getHeaderString("Authorization")).andReturn("Basic YWxhZGRpbjpvcGVuc2VzYW1l").times(1); + EasyMock.replay(httpHeaders); + herder.putConnectorConfig(EasyMock.eq(CONNECTOR_NAME), EasyMock.eq(body.config()), EasyMock.eq(false), EasyMock.capture(cb)); + expectAndCallbackResult(cb, new Herder.Created<>(true, new ConnectorInfo(CONNECTOR_NAME, CONNECTOR_CONFIG, + CONNECTOR_TASK_NAMES, ConnectorType.SOURCE))); + + PowerMock.replayAll(); + + connectorsResource.createConnector(FORWARD, httpHeaders, body); + + PowerMock.verifyAll(); + } + + + + @Test + public void testCreateConnectorWithoutHeaderAuthorization() throws Throwable { + CreateConnectorRequest body = new CreateConnectorRequest(CONNECTOR_NAME, Collections.singletonMap(ConnectorConfig.NAME_CONFIG, CONNECTOR_NAME)); + final Capture>> cb = Capture.newInstance(); + HttpHeaders httpHeaders = EasyMock.mock(HttpHeaders.class); + EasyMock.expect(httpHeaders.getHeaderString("Authorization")).andReturn(null).times(1); + EasyMock.replay(httpHeaders); + herder.putConnectorConfig(EasyMock.eq(CONNECTOR_NAME), EasyMock.eq(body.config()), EasyMock.eq(false), EasyMock.capture(cb)); + expectAndCallbackResult(cb, new Herder.Created<>(true, new ConnectorInfo(CONNECTOR_NAME, CONNECTOR_CONFIG, + CONNECTOR_TASK_NAMES, ConnectorType.SOURCE))); + + PowerMock.replayAll(); + + connectorsResource.createConnector(FORWARD, httpHeaders, body); + + PowerMock.verifyAll(); + } + @Test(expected = AlreadyExistsException.class) public void testCreateConnectorExists() throws Throwable { CreateConnectorRequest body = new CreateConnectorRequest(CONNECTOR_NAME, Collections.singletonMap(ConnectorConfig.NAME_CONFIG, CONNECTOR_NAME)); @@ -307,7 +347,7 @@ public void testCreateConnectorExists() throws Throwable { PowerMock.replayAll(); - connectorsResource.createConnector(FORWARD, body); + connectorsResource.createConnector(FORWARD, NULL_HEADERS, body); PowerMock.verifyAll(); } @@ -326,7 +366,7 @@ public void testCreateConnectorNameTrimWhitespaces() throws Throwable { PowerMock.replayAll(); - connectorsResource.createConnector(FORWARD, bodyIn); + connectorsResource.createConnector(FORWARD, NULL_HEADERS, bodyIn); PowerMock.verifyAll(); } @@ -345,7 +385,7 @@ public void testCreateConnectorNameAllWhitespaces() throws Throwable { PowerMock.replayAll(); - connectorsResource.createConnector(FORWARD, bodyIn); + connectorsResource.createConnector(FORWARD, NULL_HEADERS, bodyIn); PowerMock.verifyAll(); } @@ -364,7 +404,7 @@ public void testCreateConnectorNoName() throws Throwable { PowerMock.replayAll(); - connectorsResource.createConnector(FORWARD, bodyIn); + connectorsResource.createConnector(FORWARD, NULL_HEADERS, bodyIn); PowerMock.verifyAll(); } @@ -377,7 +417,7 @@ public void testDeleteConnector() throws Throwable { PowerMock.replayAll(); - connectorsResource.destroyConnector(CONNECTOR_NAME, FORWARD); + connectorsResource.destroyConnector(CONNECTOR_NAME, NULL_HEADERS, FORWARD); PowerMock.verifyAll(); } @@ -388,12 +428,12 @@ public void testDeleteConnectorNotLeader() throws Throwable { herder.deleteConnectorConfig(EasyMock.eq(CONNECTOR_NAME), EasyMock.capture(cb)); expectAndCallbackNotLeaderException(cb); // Should forward request - EasyMock.expect(RestClient.httpRequest("http://leader:8083/connectors/" + CONNECTOR_NAME + "?forward=false", "DELETE", null, null, null)) + EasyMock.expect(RestClient.httpRequest("http://leader:8083/connectors/" + CONNECTOR_NAME + "?forward=false", "DELETE", NULL_HEADERS, null, null, null)) .andReturn(new RestClient.HttpResponse<>(204, new HashMap(), null)); PowerMock.replayAll(); - connectorsResource.destroyConnector(CONNECTOR_NAME, FORWARD); + connectorsResource.destroyConnector(CONNECTOR_NAME, NULL_HEADERS, FORWARD); PowerMock.verifyAll(); } @@ -407,7 +447,7 @@ public void testDeleteConnectorNotFound() throws Throwable { PowerMock.replayAll(); - connectorsResource.destroyConnector(CONNECTOR_NAME, FORWARD); + connectorsResource.destroyConnector(CONNECTOR_NAME, NULL_HEADERS, FORWARD); PowerMock.verifyAll(); } @@ -421,7 +461,7 @@ public void testGetConnector() throws Throwable { PowerMock.replayAll(); - ConnectorInfo connInfo = connectorsResource.getConnector(CONNECTOR_NAME, FORWARD); + ConnectorInfo connInfo = connectorsResource.getConnector(CONNECTOR_NAME, NULL_HEADERS, FORWARD); assertEquals(new ConnectorInfo(CONNECTOR_NAME, CONNECTOR_CONFIG, CONNECTOR_TASK_NAMES, ConnectorType.SOURCE), connInfo); @@ -436,7 +476,7 @@ public void testGetConnectorConfig() throws Throwable { PowerMock.replayAll(); - Map connConfig = connectorsResource.getConnectorConfig(CONNECTOR_NAME, FORWARD); + Map connConfig = connectorsResource.getConnectorConfig(CONNECTOR_NAME, NULL_HEADERS, FORWARD); assertEquals(CONNECTOR_CONFIG, connConfig); PowerMock.verifyAll(); @@ -450,7 +490,7 @@ public void testGetConnectorConfigConnectorNotFound() throws Throwable { PowerMock.replayAll(); - connectorsResource.getConnectorConfig(CONNECTOR_NAME, FORWARD); + connectorsResource.getConnectorConfig(CONNECTOR_NAME, NULL_HEADERS, FORWARD); PowerMock.verifyAll(); } @@ -464,7 +504,7 @@ public void testPutConnectorConfig() throws Throwable { PowerMock.replayAll(); - connectorsResource.putConnectorConfig(CONNECTOR_NAME, FORWARD, CONNECTOR_CONFIG); + connectorsResource.putConnectorConfig(CONNECTOR_NAME, NULL_HEADERS, FORWARD, CONNECTOR_CONFIG); PowerMock.verifyAll(); } @@ -480,7 +520,7 @@ public void testCreateConnectorWithSpecialCharsInName() throws Throwable { PowerMock.replayAll(); - String rspLocation = connectorsResource.createConnector(FORWARD, body).getLocation().toString(); + String rspLocation = connectorsResource.createConnector(FORWARD, NULL_HEADERS, body).getLocation().toString(); String decoded = new URI(rspLocation).getPath(); Assert.assertEquals("/connectors/" + CONNECTOR_NAME_SPECIAL_CHARS, decoded); @@ -498,7 +538,7 @@ public void testCreateConnectorWithControlSequenceInName() throws Throwable { PowerMock.replayAll(); - String rspLocation = connectorsResource.createConnector(FORWARD, body).getLocation().toString(); + String rspLocation = connectorsResource.createConnector(FORWARD, NULL_HEADERS, body).getLocation().toString(); String decoded = new URI(rspLocation).getPath(); Assert.assertEquals("/connectors/" + CONNECTOR_NAME_CONTROL_SEQUENCES1, decoded); @@ -515,7 +555,7 @@ public void testPutConnectorConfigWithSpecialCharsInName() throws Throwable { PowerMock.replayAll(); - String rspLocation = connectorsResource.putConnectorConfig(CONNECTOR_NAME_SPECIAL_CHARS, FORWARD, CONNECTOR_CONFIG_SPECIAL_CHARS).getLocation().toString(); + String rspLocation = connectorsResource.putConnectorConfig(CONNECTOR_NAME_SPECIAL_CHARS, NULL_HEADERS, FORWARD, CONNECTOR_CONFIG_SPECIAL_CHARS).getLocation().toString(); String decoded = new URI(rspLocation).getPath(); Assert.assertEquals("/connectors/" + CONNECTOR_NAME_SPECIAL_CHARS, decoded); @@ -532,7 +572,7 @@ public void testPutConnectorConfigWithControlSequenceInName() throws Throwable { PowerMock.replayAll(); - String rspLocation = connectorsResource.putConnectorConfig(CONNECTOR_NAME_CONTROL_SEQUENCES1, FORWARD, CONNECTOR_CONFIG_CONTROL_SEQUENCES).getLocation().toString(); + String rspLocation = connectorsResource.putConnectorConfig(CONNECTOR_NAME_CONTROL_SEQUENCES1, NULL_HEADERS, FORWARD, CONNECTOR_CONFIG_CONTROL_SEQUENCES).getLocation().toString(); String decoded = new URI(rspLocation).getPath(); Assert.assertEquals("/connectors/" + CONNECTOR_NAME_CONTROL_SEQUENCES1, decoded); @@ -543,7 +583,7 @@ public void testPutConnectorConfigWithControlSequenceInName() throws Throwable { public void testPutConnectorConfigNameMismatch() throws Throwable { Map connConfig = new HashMap<>(CONNECTOR_CONFIG); connConfig.put(ConnectorConfig.NAME_CONFIG, "mismatched-name"); - connectorsResource.putConnectorConfig(CONNECTOR_NAME, FORWARD, connConfig); + connectorsResource.putConnectorConfig(CONNECTOR_NAME, NULL_HEADERS, FORWARD, connConfig); } @Test(expected = BadRequestException.class) @@ -551,7 +591,7 @@ public void testCreateConnectorConfigNameMismatch() throws Throwable { Map connConfig = new HashMap<>(); connConfig.put(ConnectorConfig.NAME_CONFIG, "mismatched-name"); CreateConnectorRequest request = new CreateConnectorRequest(CONNECTOR_NAME, connConfig); - connectorsResource.createConnector(FORWARD, request); + connectorsResource.createConnector(FORWARD, NULL_HEADERS, request); } @Test @@ -562,7 +602,7 @@ public void testGetConnectorTaskConfigs() throws Throwable { PowerMock.replayAll(); - List taskInfos = connectorsResource.getTaskConfigs(CONNECTOR_NAME, FORWARD); + List taskInfos = connectorsResource.getTaskConfigs(CONNECTOR_NAME, NULL_HEADERS, FORWARD); assertEquals(TASK_INFOS, taskInfos); PowerMock.verifyAll(); @@ -576,7 +616,7 @@ public void testGetConnectorTaskConfigsConnectorNotFound() throws Throwable { PowerMock.replayAll(); - connectorsResource.getTaskConfigs(CONNECTOR_NAME, FORWARD); + connectorsResource.getTaskConfigs(CONNECTOR_NAME, NULL_HEADERS, FORWARD); PowerMock.verifyAll(); } @@ -589,7 +629,7 @@ public void testPutConnectorTaskConfigs() throws Throwable { PowerMock.replayAll(); - connectorsResource.putTaskConfigs(CONNECTOR_NAME, FORWARD, TASK_CONFIGS); + connectorsResource.putTaskConfigs(CONNECTOR_NAME, NULL_HEADERS, FORWARD, TASK_CONFIGS); PowerMock.verifyAll(); } @@ -602,7 +642,7 @@ public void testPutConnectorTaskConfigsConnectorNotFound() throws Throwable { PowerMock.replayAll(); - connectorsResource.putTaskConfigs(CONNECTOR_NAME, FORWARD, TASK_CONFIGS); + connectorsResource.putTaskConfigs(CONNECTOR_NAME, NULL_HEADERS, FORWARD, TASK_CONFIGS); PowerMock.verifyAll(); } @@ -615,7 +655,7 @@ public void testRestartConnectorNotFound() throws Throwable { PowerMock.replayAll(); - connectorsResource.restartConnector(CONNECTOR_NAME, FORWARD); + connectorsResource.restartConnector(CONNECTOR_NAME, NULL_HEADERS, FORWARD); PowerMock.verifyAll(); } @@ -627,12 +667,12 @@ public void testRestartConnectorLeaderRedirect() throws Throwable { expectAndCallbackNotLeaderException(cb); EasyMock.expect(RestClient.httpRequest(EasyMock.eq("http://leader:8083/connectors/" + CONNECTOR_NAME + "/restart?forward=true"), - EasyMock.eq("POST"), EasyMock.isNull(), EasyMock.anyObject(), EasyMock.anyObject(WorkerConfig.class))) + EasyMock.eq("POST"), EasyMock.isNull(), EasyMock.isNull(), EasyMock.anyObject(), EasyMock.anyObject(WorkerConfig.class))) .andReturn(new RestClient.HttpResponse<>(202, new HashMap(), null)); PowerMock.replayAll(); - connectorsResource.restartConnector(CONNECTOR_NAME, null); + connectorsResource.restartConnector(CONNECTOR_NAME, NULL_HEADERS, null); PowerMock.verifyAll(); } @@ -645,12 +685,12 @@ public void testRestartConnectorOwnerRedirect() throws Throwable { expectAndCallbackException(cb, new NotAssignedException("not owner test", ownerUrl)); EasyMock.expect(RestClient.httpRequest(EasyMock.eq("http://owner:8083/connectors/" + CONNECTOR_NAME + "/restart?forward=false"), - EasyMock.eq("POST"), EasyMock.isNull(), EasyMock.anyObject(), EasyMock.anyObject(WorkerConfig.class))) + EasyMock.eq("POST"), EasyMock.isNull(), EasyMock.isNull(), EasyMock.anyObject(), EasyMock.anyObject(WorkerConfig.class))) .andReturn(new RestClient.HttpResponse<>(202, new HashMap(), null)); PowerMock.replayAll(); - connectorsResource.restartConnector(CONNECTOR_NAME, true); + connectorsResource.restartConnector(CONNECTOR_NAME, NULL_HEADERS, true); PowerMock.verifyAll(); } @@ -664,7 +704,7 @@ public void testRestartTaskNotFound() throws Throwable { PowerMock.replayAll(); - connectorsResource.restartTask(CONNECTOR_NAME, 0, FORWARD); + connectorsResource.restartTask(CONNECTOR_NAME, 0, NULL_HEADERS, FORWARD); PowerMock.verifyAll(); } @@ -678,12 +718,12 @@ public void testRestartTaskLeaderRedirect() throws Throwable { expectAndCallbackNotLeaderException(cb); EasyMock.expect(RestClient.httpRequest(EasyMock.eq("http://leader:8083/connectors/" + CONNECTOR_NAME + "/tasks/0/restart?forward=true"), - EasyMock.eq("POST"), EasyMock.isNull(), EasyMock.anyObject(), EasyMock.anyObject(WorkerConfig.class))) + EasyMock.eq("POST"), EasyMock.isNull(), EasyMock.isNull(), EasyMock.anyObject(), EasyMock.anyObject(WorkerConfig.class))) .andReturn(new RestClient.HttpResponse<>(202, new HashMap(), null)); PowerMock.replayAll(); - connectorsResource.restartTask(CONNECTOR_NAME, 0, null); + connectorsResource.restartTask(CONNECTOR_NAME, 0, NULL_HEADERS, null); PowerMock.verifyAll(); } @@ -698,12 +738,12 @@ public void testRestartTaskOwnerRedirect() throws Throwable { expectAndCallbackException(cb, new NotAssignedException("not owner test", ownerUrl)); EasyMock.expect(RestClient.httpRequest(EasyMock.eq("http://owner:8083/connectors/" + CONNECTOR_NAME + "/tasks/0/restart?forward=false"), - EasyMock.eq("POST"), EasyMock.isNull(), EasyMock.anyObject(), EasyMock.anyObject(WorkerConfig.class))) + EasyMock.eq("POST"), EasyMock.isNull(), EasyMock.isNull(), EasyMock.anyObject(), EasyMock.anyObject(WorkerConfig.class))) .andReturn(new RestClient.HttpResponse<>(202, new HashMap(), null)); PowerMock.replayAll(); - connectorsResource.restartTask(CONNECTOR_NAME, 0, true); + connectorsResource.restartTask(CONNECTOR_NAME, 0, NULL_HEADERS, true); PowerMock.verifyAll(); }