lidavidm commented on a change in pull request #7994:
URL: https://github.com/apache/arrow/pull/7994#discussion_r478675271
##########
File path:
java/flight/flight-core/src/main/java/org/apache/arrow/flight/example/integration/AuthBasicProtoScenario.java
##########
@@ -58,40 +58,37 @@ public void doAction(CallContext context, Action action,
StreamListener<Result>
public void buildServer(FlightServer.Builder builder) {
builder.authHandler(new BasicServerAuthHandler(new
BasicServerAuthHandler.BasicAuthValidator() {
@Override
- public byte[] getToken(String username, String password) throws
Exception {
+ public Optional<String> validateCredentials(String username, String
password) throws Exception {
+ if (Strings.isNullOrEmpty(username)) {
+ throw CallStatus.UNAUTHORIZED.withDescription("Credentials not
supplied").toRuntimeException();
+ }
+
if (!USERNAME.equals(username) || !PASSWORD.equals(password)) {
- throw CallStatus.UNAUTHENTICATED.withDescription("Username or
password is invalid.").toRuntimeException();
+ throw CallStatus.UNAUTHORIZED.withDescription("Username or password
is invalid.").toRuntimeException();
}
- return ("valid:" + username).getBytes(StandardCharsets.UTF_8);
+ return Optional.of("valid:" + username);
}
@Override
- public Optional<String> isValid(byte[] token) {
- if (token != null) {
- final String credential = new String(token, StandardCharsets.UTF_8);
- if (credential.startsWith("valid:")) {
- return Optional.of(credential.substring(6));
- }
+ public Optional<String> isValid(String token) {
+ if (token.startsWith("valid:")) {
+ return Optional.of(token.substring(6));
}
return Optional.empty();
}
}));
}
@Override
- public void client(BufferAllocator allocator, Location location,
FlightClient client) {
- final FlightRuntimeException e =
IntegrationAssertions.assertThrows(FlightRuntimeException.class, () -> {
- client.listActions().forEach(act -> {
- });
- });
- if (!FlightStatusCode.UNAUTHENTICATED.equals(e.status().code())) {
- throw new AssertionError("Expected UNAUTHENTICATED but found " +
e.status().code(), e);
- }
-
- client.authenticate(new BasicClientAuthHandler(USERNAME, PASSWORD));
- final Result result = client.doAction(new Action("")).next();
- if (!USERNAME.equals(new String(result.getBody(),
StandardCharsets.UTF_8))) {
- throw new AssertionError("Expected " + USERNAME + " but got " +
Arrays.toString(result.getBody()));
+ public void client(BufferAllocator allocator, Location location,
FlightClient.Builder clientBuilder) {
Review comment:
We should still test what happens if the client tries to make an
unauthenticated request.
##########
File path:
java/flight/flight-core/src/main/java/org/apache/arrow/flight/grpc/ContextAdapter.java
##########
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.arrow.flight.grpc;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.arrow.flight.CallContext;
+import org.apache.arrow.flight.auth.AuthConstants;
+
+import io.grpc.Context;
+
+/**
+ * Adapter class for gRPC contexts.
+ */
+public class ContextAdapter implements CallContext {
+ // gRPC uses reference equality when looking up keys in a Context. Cache
used keys in this static map
+ // so that look ups can succeed.
+ private static final Map<String, Context.Key<String>> usedKeys = new
HashMap<>();
Review comment:
An internal synchronized cache may be surprising...I'd say if we have to
have a cache, it should just use ConcurrentHashMap, and preferably, we should
just wrap Context.Key<T> ourselves.
##########
File path:
java/flight/flight-core/src/main/java/org/apache/arrow/flight/auth/ClientHandshakeWrapper.java
##########
@@ -0,0 +1,97 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.arrow.flight.auth;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+
+import org.apache.arrow.flight.grpc.StatusUtils;
+import org.apache.arrow.flight.impl.Flight.HandshakeRequest;
+import org.apache.arrow.flight.impl.Flight.HandshakeResponse;
+import org.apache.arrow.flight.impl.FlightServiceGrpc.FlightServiceStub;
+
+import io.grpc.StatusRuntimeException;
+import io.grpc.stub.StreamObserver;
+
+/**
+ * Utility class for executing a handshake with a FlightServer.
+ */
+public class ClientHandshakeWrapper {
+ private static final org.slf4j.Logger logger =
org.slf4j.LoggerFactory.getLogger(ClientHandshakeWrapper.class);
+
+ /**
+ * Do handshake for a client. The stub will be authenticated after this
method returns.
+ *
+ * @param stub The service stub.
+ */
+ public static void doClientHandshake(FlightServiceStub stub) {
+ final HandshakeObserver observer = new HandshakeObserver();
+ try {
+ observer.requestObserver = stub.handshake(observer);
+ observer.requestObserver.onNext(HandshakeRequest.newBuilder().build());
+ observer.requestObserver.onCompleted();
+ try {
+ if (!observer.completed.get()) {
+ // TODO: ARROW-5681
+ throw new RuntimeException("Unauthenticated");
+ }
+ } catch (InterruptedException ex) {
+ Thread.currentThread().interrupt();
+ throw ex;
Review comment:
Does this actually build? We're throwing a checked exception here
##########
File path:
java/flight/flight-core/src/main/java/org/apache/arrow/flight/auth/ServerAuthMiddleware.java
##########
@@ -0,0 +1,105 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.arrow.flight.auth;
+
+import org.apache.arrow.flight.CallContext;
+import org.apache.arrow.flight.CallHeaders;
+import org.apache.arrow.flight.CallInfo;
+import org.apache.arrow.flight.CallStatus;
+import org.apache.arrow.flight.FlightServerMiddleware;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Middleware that's used to validate credentials during the handshake and
verify
+ * the bearer token in subsequent requests.
+ */
+public class ServerAuthMiddleware implements FlightServerMiddleware {
+ private static final Logger logger =
LoggerFactory.getLogger(ServerAuthMiddleware.class);
+
+ /**
+ * Factory for accessing ServerAuthMiddleware.
+ */
+ public static class Factory implements
FlightServerMiddleware.Factory<ServerAuthMiddleware> {
+ private final ServerAuthHandler authHandler;
+ private final GeneratedBearerTokenAuthHandler bearerTokenAuthHandler;
+
+ /**
+ * Construct a factory with the given auth handler.
+ * @param authHandler The auth handler what will be used for
authenticating requests.
+ */
+ public Factory(ServerAuthHandler authHandler) {
+ this.authHandler = authHandler;
+ bearerTokenAuthHandler = authHandler.enableCachedCredentials() ?
+ new GeneratedBearerTokenAuthHandler() : null;
+ }
+
+ @Override
+ public ServerAuthMiddleware onCallStarted(CallInfo callInfo, CallHeaders
incomingHeaders, CallContext context) {
+ logger.debug("Call name: {}", callInfo.method().name());
Review comment:
Let's maybe not spam logs here; logging should go into dedicated logging
middleware.
##########
File path:
java/flight/flight-core/src/main/java/org/apache/arrow/flight/auth/ServerAuthMiddleware.java
##########
@@ -0,0 +1,105 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.arrow.flight.auth;
+
+import org.apache.arrow.flight.CallContext;
+import org.apache.arrow.flight.CallHeaders;
+import org.apache.arrow.flight.CallInfo;
+import org.apache.arrow.flight.CallStatus;
+import org.apache.arrow.flight.FlightServerMiddleware;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Middleware that's used to validate credentials during the handshake and
verify
+ * the bearer token in subsequent requests.
+ */
+public class ServerAuthMiddleware implements FlightServerMiddleware {
+ private static final Logger logger =
LoggerFactory.getLogger(ServerAuthMiddleware.class);
+
+ /**
+ * Factory for accessing ServerAuthMiddleware.
+ */
+ public static class Factory implements
FlightServerMiddleware.Factory<ServerAuthMiddleware> {
+ private final ServerAuthHandler authHandler;
+ private final GeneratedBearerTokenAuthHandler bearerTokenAuthHandler;
+
+ /**
+ * Construct a factory with the given auth handler.
+ * @param authHandler The auth handler what will be used for
authenticating requests.
+ */
+ public Factory(ServerAuthHandler authHandler) {
+ this.authHandler = authHandler;
+ bearerTokenAuthHandler = authHandler.enableCachedCredentials() ?
+ new GeneratedBearerTokenAuthHandler() : null;
+ }
+
+ @Override
+ public ServerAuthMiddleware onCallStarted(CallInfo callInfo, CallHeaders
incomingHeaders, CallContext context) {
+ logger.debug("Call name: {}", callInfo.method().name());
+ // Check if bearer token auth is being used, and if we've enabled use of
server-generated
+ // bearer tokens.
+ if (authHandler.enableCachedCredentials()) {
Review comment:
Is it possible for logic like this to be handled by the auth handler
itself? I don't see the value of hardcoding a particular caching strategy here.
##########
File path:
java/flight/flight-core/src/main/java/org/apache/arrow/flight/auth/ClientHandshakeWrapper.java
##########
@@ -0,0 +1,97 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.arrow.flight.auth;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+
+import org.apache.arrow.flight.grpc.StatusUtils;
+import org.apache.arrow.flight.impl.Flight.HandshakeRequest;
+import org.apache.arrow.flight.impl.Flight.HandshakeResponse;
+import org.apache.arrow.flight.impl.FlightServiceGrpc.FlightServiceStub;
+
+import io.grpc.StatusRuntimeException;
+import io.grpc.stub.StreamObserver;
+
+/**
+ * Utility class for executing a handshake with a FlightServer.
+ */
+public class ClientHandshakeWrapper {
+ private static final org.slf4j.Logger logger =
org.slf4j.LoggerFactory.getLogger(ClientHandshakeWrapper.class);
+
+ /**
+ * Do handshake for a client. The stub will be authenticated after this
method returns.
+ *
+ * @param stub The service stub.
+ */
+ public static void doClientHandshake(FlightServiceStub stub) {
+ final HandshakeObserver observer = new HandshakeObserver();
+ try {
+ observer.requestObserver = stub.handshake(observer);
+ observer.requestObserver.onNext(HandshakeRequest.newBuilder().build());
+ observer.requestObserver.onCompleted();
+ try {
+ if (!observer.completed.get()) {
+ // TODO: ARROW-5681
+ throw new RuntimeException("Unauthenticated");
+ }
+ } catch (InterruptedException ex) {
+ Thread.currentThread().interrupt();
+ throw ex;
Review comment:
Oh I see, it gets caught and rethrown below
##########
File path:
java/flight/flight-core/src/main/java/org/apache/arrow/flight/auth/ServerAuthMiddleware.java
##########
@@ -0,0 +1,105 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.arrow.flight.auth;
+
+import org.apache.arrow.flight.CallContext;
+import org.apache.arrow.flight.CallHeaders;
+import org.apache.arrow.flight.CallInfo;
+import org.apache.arrow.flight.CallStatus;
+import org.apache.arrow.flight.FlightServerMiddleware;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Middleware that's used to validate credentials during the handshake and
verify
+ * the bearer token in subsequent requests.
+ */
+public class ServerAuthMiddleware implements FlightServerMiddleware {
+ private static final Logger logger =
LoggerFactory.getLogger(ServerAuthMiddleware.class);
+
+ /**
+ * Factory for accessing ServerAuthMiddleware.
+ */
+ public static class Factory implements
FlightServerMiddleware.Factory<ServerAuthMiddleware> {
+ private final ServerAuthHandler authHandler;
+ private final GeneratedBearerTokenAuthHandler bearerTokenAuthHandler;
+
+ /**
+ * Construct a factory with the given auth handler.
+ * @param authHandler The auth handler what will be used for
authenticating requests.
+ */
+ public Factory(ServerAuthHandler authHandler) {
+ this.authHandler = authHandler;
+ bearerTokenAuthHandler = authHandler.enableCachedCredentials() ?
+ new GeneratedBearerTokenAuthHandler() : null;
+ }
+
+ @Override
+ public ServerAuthMiddleware onCallStarted(CallInfo callInfo, CallHeaders
incomingHeaders, CallContext context) {
+ logger.debug("Call name: {}", callInfo.method().name());
+ // Check if bearer token auth is being used, and if we've enabled use of
server-generated
+ // bearer tokens.
+ if (authHandler.enableCachedCredentials()) {
+ final String bearerTokenFromHeaders =
+ AuthUtilities.getValueFromAuthHeader(incomingHeaders,
AuthConstants.BEARER_PREFIX);
+ if (bearerTokenFromHeaders != null) {
+ final ServerAuthHandler.AuthResult result =
bearerTokenAuthHandler.authenticate(incomingHeaders);
+ context.put(AuthConstants.PEER_IDENTITY_KEY,
result.getPeerIdentity());
+ return new ServerAuthMiddleware(result.getPeerIdentity(),
result.getBearerToken().get());
+ }
+ }
+
+ // Delegate to server auth handler to do the validation.
+ final ServerAuthHandler.AuthResult result =
authHandler.authenticate(incomingHeaders);
+ final String bearerToken;
+ if (authHandler.enableCachedCredentials()) {
+ bearerToken = bearerTokenAuthHandler.registerBearer(result);
+ } else {
+ bearerToken = result.getBearerToken().get();
+ }
+ context.put(AuthConstants.PEER_IDENTITY_KEY, result.getPeerIdentity());
+ return new ServerAuthMiddleware(result.getPeerIdentity(), bearerToken);
+ }
+ }
+
+ private final String bearerToken;
+ private final String peerIdentity;
+
+ public ServerAuthMiddleware(String peerIdentity, String bearerToken) {
+ this.peerIdentity = peerIdentity;
+ this.bearerToken = bearerToken;
+ }
+
+ @Override
+ public void onBeforeSendingHeaders(CallHeaders outgoingHeaders) {
+ if (bearerToken != null &&
Review comment:
The same goes for outbound headers - an auth handler should be able to
send whatever outbound headers it might need, instead of hardcoding a
particular header here. SPNEGO is an example of an auth scheme that uses
WWW-Negotiate instead of Authorization in responses, for example.
##########
File path:
java/flight/flight-core/src/main/java/org/apache/arrow/flight/auth/ClientHandshakeWrapper.java
##########
@@ -0,0 +1,97 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.arrow.flight.auth;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+
+import org.apache.arrow.flight.grpc.StatusUtils;
+import org.apache.arrow.flight.impl.Flight.HandshakeRequest;
+import org.apache.arrow.flight.impl.Flight.HandshakeResponse;
+import org.apache.arrow.flight.impl.FlightServiceGrpc.FlightServiceStub;
+
+import io.grpc.StatusRuntimeException;
+import io.grpc.stub.StreamObserver;
+
+/**
+ * Utility class for executing a handshake with a FlightServer.
+ */
+public class ClientHandshakeWrapper {
+ private static final org.slf4j.Logger logger =
org.slf4j.LoggerFactory.getLogger(ClientHandshakeWrapper.class);
+
+ /**
+ * Do handshake for a client. The stub will be authenticated after this
method returns.
+ *
+ * @param stub The service stub.
+ */
+ public static void doClientHandshake(FlightServiceStub stub) {
+ final HandshakeObserver observer = new HandshakeObserver();
+ try {
+ observer.requestObserver = stub.handshake(observer);
+ observer.requestObserver.onNext(HandshakeRequest.newBuilder().build());
+ observer.requestObserver.onCompleted();
+ try {
+ if (!observer.completed.get()) {
+ // TODO: ARROW-5681
+ throw new RuntimeException("Unauthenticated");
+ }
+ } catch (InterruptedException ex) {
+ Thread.currentThread().interrupt();
+ throw ex;
+ } catch (ExecutionException ex) {
+ logger.error("Failed on completing future", ex.getCause());
+ throw ex.getCause();
Review comment:
This should probably be `StatusUtils.fromThrowable(ex.getCause())`
##########
File path:
java/flight/flight-core/src/main/java/org/apache/arrow/flight/example/integration/AuthBasicProtoScenario.java
##########
@@ -58,40 +58,37 @@ public void doAction(CallContext context, Action action,
StreamListener<Result>
public void buildServer(FlightServer.Builder builder) {
builder.authHandler(new BasicServerAuthHandler(new
BasicServerAuthHandler.BasicAuthValidator() {
@Override
- public byte[] getToken(String username, String password) throws
Exception {
+ public Optional<String> validateCredentials(String username, String
password) throws Exception {
+ if (Strings.isNullOrEmpty(username)) {
+ throw CallStatus.UNAUTHORIZED.withDescription("Credentials not
supplied").toRuntimeException();
Review comment:
Unauthorized != Unauthenticated, this should be Unauthenticated. The
same goes on like 67 below.
##########
File path:
java/flight/flight-core/src/main/java/org/apache/arrow/flight/auth/BearerTokenAuthHandler.java
##########
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.arrow.flight.auth;
+
+import java.util.Optional;
+
+import org.apache.arrow.flight.CallHeaders;
+import org.apache.arrow.flight.CallStatus;
+import org.apache.arrow.flight.FlightRuntimeException;
+
+/**
+ * Partial implementation of ServerAuthHandler for bearer-token based
authentication.
+ */
+abstract class BearerTokenAuthHandler implements ServerAuthHandler {
+ @Override
+ public AuthResult authenticate(CallHeaders headers) {
+ final String bearerToken = AuthUtilities.getValueFromAuthHeader(headers,
AuthConstants.BEARER_PREFIX);
+ if (bearerToken == null) {
+ throw new FlightRuntimeException(CallStatus.UNAUTHENTICATED);
+ }
+
+ if (!validateBearer(bearerToken)) {
+ throw new FlightRuntimeException(CallStatus.UNAUTHORIZED);
Review comment:
This should also be Unauthenticated
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
