Skip to content

KAFKA-8404: Add HttpHeader to RestClient HTTP Request and Connector REST API #6791

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
rhauch merged 16 commits into from
Jun 4, 2019
Merged

KAFKA-8404: Add HttpHeader to RestClient HTTP Request and Connector REST API #6791

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
16 commits
Select commit Hold shift + click to select a range
470fb6b
Add HttpHeader to RestClient HTTP Request and Connector REST API
May 22, 2019
6a6c094
Clearer function name
May 23, 2019
36a8240
KAFKA-8404: Not allow authentication with POST /connectors/{connName}…
May 25, 2019
f4b050a
KAFKA-8404: Minor fix
May 28, 2019
c7bb262
Add HttpHeader to RestClient HTTP Request and Connector REST API
May 22, 2019
17ad54e
Clearer function name
May 23, 2019
ff412d2
KAFKA-8404: Not allow authentication with POST /connectors/{connName}…
May 25, 2019
97c1e04
KAFKA-8404: Minor fix
May 28, 2019
4c6524c
KAFKA-8404:Minor fix
May 30, 2019
422d517
Add HttpHeader to RestClient HTTP Request and Connector REST API
May 22, 2019
fd15a03
Clearer function name
May 23, 2019
68e6801
KAFKA-8404: Not allow authentication with POST /connectors/{connName}…
May 25, 2019
b040f84
KAFKA-8404: Minor fix
May 28, 2019
a0cdf32
Resolve rebase conflict
Jun 3, 2019
4e2c989
Minor change in String regex for task request pattern
Jun 3, 2019
d30d6ce
Change checkstyle
Jun 3, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 9 additions & 7 deletions ...src/main/java/org/apache/kafka/connect/rest/basic/auth/extension/JaasBasicAuthFilter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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)
Expand Down
34 changes: 33 additions & 1 deletion ...test/java/org/apache/kafka/connect/rest/basic/auth/extension/JaasBasicAuthFilterTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -52,6 +55,9 @@ public class JaasBasicAuthFilterTest {
private String previousJaasConfig;
private Configuration previousConfiguration;

@MockStrict
private UriInfo uriInfo;

@Before
public void setup() {
EasyMock.reset(requestContext);
Expand Down Expand Up @@ -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);
Expand All @@ -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<String> lines;
lines = new ArrayList<>();
lines.add(loginModule + " { org.apache.kafka.connect.rest.basic.auth.extension.PropertyFileLoginModule required ");
Expand Down
2 changes: 1 addition & 1 deletion ...runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);

@C0urante C0urante May 23, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If the REST API requires authorization, won't this request fail due to lack of headers?

@rhauch rhauch Jun 3, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Any answer to this?

@haidangdam haidangdam Jun 3, 2019 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hi. Pattern will filter out this endpoint so that authorization will not fail.

@mgarmes mgarmes Jun 18, 2019

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hi @haidangdam, you mean there is no Authorization for POST /connectors/xx/tasks, so anyone can change tasks configuration ??

@rayokota rayokota Jun 19, 2019

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Currently this endpoint should be secured through other means, such as mutual TLS.

cb.onCompletion(null, null);
} catch (ConnectException e) {
log.error("Request to leader to reconfigure connector tasks failed", e);
Expand Down
21 changes: 20 additions & 1 deletion connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/RestClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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 <T> 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 <T> HttpResponse<T> httpRequest(String url, String method, Object requestBodyData,
public static <T> HttpResponse<T> httpRequest(String url, String method, HttpHeaders headers, Object requestBodyData,
TypeReference<T> responseFormat, WorkerConfig config) {
HttpClient client;

Expand All @@ -82,6 +84,8 @@ public static <T> HttpResponse<T> 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");
}
Expand Down Expand Up @@ -116,6 +120,21 @@ public static <T> HttpResponse<T> 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);
}
}
}
Comment thread
rhauch marked this conversation as resolved.
Outdated

/**
* Convert response parameters from Jetty format (HttpFields)
* @param httpFields
Expand Down
45 changes: 29 additions & 16 deletions ...ime/src/main/java/org/apache/kafka/connect/runtime/rest/resources/ConnectorsResource.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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<String, Map<String, Object>> out = new HashMap<>();
Expand Down Expand Up @@ -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
Expand All @@ -132,7 +135,7 @@ public Response createConnector(final @QueryParam("forward") Boolean forward,

FutureCallback<Herder.Created<ConnectorInfo>> cb = new FutureCallback<>();
herder.putConnectorConfig(name, configs, false, cb);
Herder.Created<ConnectorInfo> info = completeOrForwardRequest(cb, "/connectors", "POST", createRequest,
Herder.Created<ConnectorInfo> info = completeOrForwardRequest(cb, "/connectors", "POST", headers, createRequest,
new TypeReference<ConnectorInfo>() { }, new CreatedConnectorInfoTranslator(), forward);

URI location = UriBuilder.fromUri("/connectors").path(name).build();
Expand All @@ -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<ConnectorInfo> 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<String, String> getConnectorConfig(final @PathParam("connector") String connector,
final @Context HttpHeaders headers,
final @QueryParam("forward") Boolean forward) throws Throwable {
FutureCallback<Map<String, String>> 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
Expand All @@ -166,14 +171,15 @@ 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<String, String> connectorConfig) throws Throwable {
FutureCallback<Herder.Created<ConnectorInfo>> cb = new FutureCallback<>();
checkAndPutConnectorConfigName(connector, connectorConfig);

herder.putConnectorConfig(connector, connectorConfig, true, cb);
Herder.Created<ConnectorInfo> createdInfo = completeOrForwardRequest(cb, "/connectors/" + connector + "/config",
"PUT", connectorConfig, new TypeReference<ConnectorInfo>() { }, new CreatedConnectorInfoTranslator(), forward);
"PUT", headers, connectorConfig, new TypeReference<ConnectorInfo>() { }, new CreatedConnectorInfoTranslator(), forward);
Response.ResponseBuilder response;
if (createdInfo.created()) {
URI location = UriBuilder.fromUri("/connectors").path(connector).build();
Expand All @@ -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<Void> 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();
}
Expand All @@ -210,26 +217,29 @@ public Response resumeConnector(@PathParam("connector") String connector) {
@GET
@Path("/{connector}/tasks")
public List<TaskInfo> getTaskConfigs(final @PathParam("connector") String connector,
final @Context HttpHeaders headers,
final @QueryParam("forward") Boolean forward) throws Throwable {
FutureCallback<List<TaskInfo>> cb = new FutureCallback<>();
herder.taskConfigs(connector, cb);
return completeOrForwardRequest(cb, "/connectors/" + connector + "/tasks", "GET", null, new TypeReference<List<TaskInfo>>() {
return completeOrForwardRequest(cb, "/connectors/" + connector + "/tasks", "GET", headers, null, new TypeReference<List<TaskInfo>>() {
}, forward);
}

@POST
@Path("/{connector}/tasks")
public void putTaskConfigs(final @PathParam("connector") String connector,
final @Context HttpHeaders headers,
final @QueryParam("forward") Boolean forward,
final List<Map<String, String>> taskConfigs) throws Throwable {
FutureCallback<Void> 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));
}
Expand All @@ -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<Void> 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<Herder.Created<ConnectorInfo>> 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
Expand All @@ -271,6 +283,7 @@ private void checkAndPutConnectorConfigName(String connectorName, Map<String, St
private <T, U> T completeOrForwardRequest(FutureCallback<T> cb,
String path,
String method,
HttpHeaders headers,
Object body,
TypeReference<U> resultType,
Translator<T, U> translator,
Expand All @@ -293,7 +306,7 @@ private <T, U> T completeOrForwardRequest(FutureCallback<T> 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.
Expand All @@ -315,14 +328,14 @@ private <T, U> T completeOrForwardRequest(FutureCallback<T> cb,
}
}

private <T> T completeOrForwardRequest(FutureCallback<T> cb, String path, String method, Object body,
private <T> T completeOrForwardRequest(FutureCallback<T> cb, String path, String method, HttpHeaders headers, Object body,
TypeReference<T> resultType, Boolean forward) throws Throwable {
return completeOrForwardRequest(cb, path, method, body, resultType, new IdentityTranslator<T>(), forward);
return completeOrForwardRequest(cb, path, method, headers, body, resultType, new IdentityTranslator<T>(), forward);
}

private <T> T completeOrForwardRequest(FutureCallback<T> cb, String path, String method,
private <T> T completeOrForwardRequest(FutureCallback<T> cb, String path, String method, HttpHeaders headers,
Object body, Boolean forward) throws Throwable {
return completeOrForwardRequest(cb, path, method, body, null, new IdentityTranslator<T>(), forward);
return completeOrForwardRequest(cb, path, method, headers, body, null, new IdentityTranslator<T>(), forward);
}

private interface Translator<T, U> {
Expand Down
Loading