[C++] Implement auth API for Flight

cpp/src/arrow/flight/CMakeLists.txt

@@ -81,6 +81,7 @@ set(ARROW_FLIGHT_SRCS
     protocol-internal.cc
     serialization-internal.cc
     server.cc
+    server_auth.cc
     types.cc)
 
 add_arrow_lib(arrow_flight

cpp/src/arrow/flight/api.h

@@ -18,5 +18,7 @@
 #pragma once
 
 #include "arrow/flight/client.h"
+#include "arrow/flight/client_auth.h"
 #include "arrow/flight/server.h"
+#include "arrow/flight/server_auth.h"
 #include "arrow/flight/types.h"

cpp/src/arrow/flight/client.cc

@@ -38,6 +38,7 @@
 #include "arrow/type.h"
 #include "arrow/util/logging.h"
 
+#include "arrow/flight/client_auth.h"
 #include "arrow/flight/internal.h"
 #include "arrow/flight/serialization-internal.h"
 #include "arrow/flight/types.h"
@@ -63,6 +64,60 @@ struct ClientRpc {
     ss << error_message << context.debug_error_string();
     return Status::IOError(ss.str());
   }
+
+  /// \brief Add an auth token via an auth handler
+  Status SetToken(ClientAuthHandler* auth_handler) {
+    if (auth_handler) {
+      std::string token;
+      RETURN_NOT_OK(auth_handler->GetToken(&token));
+      context.AddMetadata(internal::AUTH_HEADER, token);
+    }
+    return Status::OK();
+  }
+};
+
+class GrpcClientAuthSender : public ClientAuthSender {
+ public:
+  explicit GrpcClientAuthSender(
+      std::shared_ptr<
+          grpc::ClientReaderWriter<pb::HandshakeRequest, pb::HandshakeResponse>>
+          stream)
+      : stream_(stream) {}
+
+  Status Write(const std::string& token) const override {
+    pb::HandshakeRequest response;
+    response.set_payload(token);
+    if (stream_->Write(response)) {
+      return Status::OK();
+    }
+    return internal::FromGrpcStatus(stream_->Finish());
+  }
+
+ private:
+  std::shared_ptr<grpc::ClientReaderWriter<pb::HandshakeRequest, pb::HandshakeResponse>>
+      stream_;
+};
+
+class GrpcClientAuthReader : public ClientAuthReader {
+ public:
+  explicit GrpcClientAuthReader(
+      std::shared_ptr<
+          grpc::ClientReaderWriter<pb::HandshakeRequest, pb::HandshakeResponse>>
+          stream)
+      : stream_(stream) {}
+
+  Status Read(std::string* token) const override {
+    pb::HandshakeResponse request;
+    if (stream_->Read(&request)) {
+      *token = std::move(*request.release_payload());
+      return Status::OK();
+    }
+    return internal::FromGrpcStatus(stream_->Finish());
+  }
+
+ private:
+  std::shared_ptr<grpc::ClientReaderWriter<pb::HandshakeRequest, pb::HandshakeResponse>>
+      stream_;
 };
 
 class FlightIpcMessageReader : public ipc::MessageReader {
@@ -182,11 +237,24 @@ class FlightClient::FlightClientImpl {
     return Status::OK();
   }
 
+  Status Authenticate(std::unique_ptr<ClientAuthHandler> auth_handler) {
+    auth_handler_ = std::move(auth_handler);
+    grpc::ClientContext context{};
+    std::shared_ptr<grpc::ClientReaderWriter<pb::HandshakeRequest, pb::HandshakeResponse>>
+        stream = stub_->Handshake(&context);
+    GrpcClientAuthSender outgoing{stream};
+    GrpcClientAuthReader incoming{stream};
+    RETURN_NOT_OK(auth_handler_->Authenticate(outgoing, incoming));
+    RETURN_NOT_OK(internal::FromGrpcStatus(stream->Finish()));
+    return Status::OK();
+  }
+
   Status ListFlights(const Criteria& criteria, std::unique_ptr<FlightListing>* listing) {
     // TODO(wesm): populate criteria
     pb::Criteria pb_criteria;
 
     ClientRpc rpc;
+    RETURN_NOT_OK(rpc.SetToken(auth_handler_.get()));
     std::unique_ptr<grpc::ClientReader<pb::FlightGetInfo>> stream(
         stub_->ListFlights(&rpc.context, pb_criteria));
 
@@ -208,6 +276,7 @@ class FlightClient::FlightClientImpl {
     RETURN_NOT_OK(internal::ToProto(action, &pb_action));
 
     ClientRpc rpc;
+    RETURN_NOT_OK(rpc.SetToken(auth_handler_.get()));
     std::unique_ptr<grpc::ClientReader<pb::Result>> stream(
         stub_->DoAction(&rpc.context, pb_action));
 
@@ -229,6 +298,7 @@ class FlightClient::FlightClientImpl {
     pb::Empty empty;
 
     ClientRpc rpc;
+    RETURN_NOT_OK(rpc.SetToken(auth_handler_.get()));
     std::unique_ptr<grpc::ClientReader<pb::ActionType>> stream(
         stub_->ListActions(&rpc.context, empty));
 
@@ -249,6 +319,7 @@ class FlightClient::FlightClientImpl {
     RETURN_NOT_OK(internal::ToProto(descriptor, &pb_descriptor));
 
     ClientRpc rpc;
+    RETURN_NOT_OK(rpc.SetToken(auth_handler_.get()));
     Status s = internal::FromGrpcStatus(
         stub_->GetFlightInfo(&rpc.context, pb_descriptor, &pb_response));
     RETURN_NOT_OK(s);
@@ -264,6 +335,7 @@ class FlightClient::FlightClientImpl {
     internal::ToProto(ticket, &pb_ticket);
 
     std::unique_ptr<ClientRpc> rpc(new ClientRpc);
+    RETURN_NOT_OK(rpc->SetToken(auth_handler_.get()));
     std::unique_ptr<grpc::ClientReader<pb::FlightData>> stream(
         stub_->DoGet(&rpc->context, pb_ticket));
 
@@ -275,6 +347,7 @@ class FlightClient::FlightClientImpl {
   Status DoPut(const FlightDescriptor& descriptor, const std::shared_ptr<Schema>& schema,
                std::unique_ptr<ipc::RecordBatchWriter>* out) {
     std::unique_ptr<ClientRpc> rpc(new ClientRpc);
+    RETURN_NOT_OK(rpc->SetToken(auth_handler_.get()));
     std::unique_ptr<protocol::PutResult> response(new protocol::PutResult);
     std::unique_ptr<grpc::ClientWriter<pb::FlightData>> writer(
         stub_->DoPut(&rpc->context, response.get()));
@@ -288,6 +361,7 @@ class FlightClient::FlightClientImpl {
 
  private:
   std::unique_ptr<pb::FlightService::Stub> stub_;
+  std::shared_ptr<ClientAuthHandler> auth_handler_;
 };
 
 FlightClient::FlightClient() { impl_.reset(new FlightClientImpl); }
@@ -300,6 +374,10 @@ Status FlightClient::Connect(const std::string& host, int port,
   return (*client)->impl_->Connect(host, port);
 }
 
+Status FlightClient::Authenticate(std::unique_ptr<ClientAuthHandler> auth_handler) {
+  return impl_->Authenticate(std::move(auth_handler));
+}
+
 Status FlightClient::DoAction(const Action& action,
                               std::unique_ptr<ResultStream>* results) {
   return impl_->DoAction(action, results);

cpp/src/arrow/flight/client.h

@@ -39,6 +39,8 @@ class Schema;
 
 namespace flight {
 
+class ClientAuthHandler;
+
 /// \brief Client class for Arrow Flight RPC services (gRPC-based).
 /// API experimental for now
 class ARROW_EXPORT FlightClient {
@@ -54,6 +56,10 @@ class ARROW_EXPORT FlightClient {
   static Status Connect(const std::string& host, int port,
                         std::unique_ptr<FlightClient>* client);
 
+  /// \brief Authenticate to the server using the given handler.
+  /// \return Status OK if the client authenticated successfully
+  Status Authenticate(std::unique_ptr<ClientAuthHandler> auth_handler);
+
   /// \brief Perform the indicated action, returning an iterator to the stream
   /// of results, if any
   /// \param[in] action the action to be performed

cpp/src/arrow/flight/client_auth.h

@@ -0,0 +1,63 @@
+// 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.
+
+#pragma once
+
+#include <string>
+
+#include "arrow/status.h"
+#include "arrow/util/visibility.h"
+
+namespace arrow {
+
+namespace flight {
+
+/// \brief A reader for messages from the server during an
+/// authentication handshake.
+class ARROW_EXPORT ClientAuthReader {
+ public:
+  virtual ~ClientAuthReader() = default;
+  virtual Status Read(std::string* response) const = 0;
+};
+
+/// \brief A writer for messages to the server during an
+/// authentication handshake.
+class ARROW_EXPORT ClientAuthSender {
+ public:
+  virtual ~ClientAuthSender() = default;
+  virtual Status Write(const std::string& token) const = 0;
+};
+
+/// \brief An authentication implementation for a Flight service.
+/// Authentication includes both an initial negotiation and a per-call
+/// token validation. Implementations may choose to use either or both
+/// mechanisms.
+class ARROW_EXPORT ClientAuthHandler {
+ public:
+  virtual ~ClientAuthHandler() = default;
+  /// \brief Authenticate the client on initial connection. The client
+  /// can send messages to/read responses from the server at any time.
+  /// \return Status OK if authenticated successfully
+  virtual Status Authenticate(const ClientAuthSender& outgoing,
+                              const ClientAuthReader& incoming) = 0;
+  /// \brief Get a per-call token.
+  /// \param[out] token The token to send to the server.
+  virtual Status GetToken(std::string* token) = 0;
+};
+
+}  // namespace flight
+}  // namespace arrow

cpp/src/arrow/flight/flight-test.cc

@@ -207,6 +207,37 @@ class TestFlightClient : public ::testing::Test {
   std::unique_ptr<TestServer> server_;
 };
 
+class AuthTestServer : public FlightServerBase {
+  Status DoAction(const ServerCallContext& context, const Action& action,
+                  std::unique_ptr<ResultStream>* result) {
+    std::shared_ptr<Buffer> buf;
+    RETURN_NOT_OK(Buffer::FromString(context.peer_identity(), &buf));
+    *result = std::unique_ptr<ResultStream>(new SimpleResultStream({Result{buf}}));
+    return Status::OK();
+  }
+};
+
+class TestAuthHandler : public ::testing::Test {
+ public:
+  void SetUp() {
+    port_ = 30000;
+    server_.reset(new InProcessTestServer(
+        std::unique_ptr<FlightServerBase>(new AuthTestServer), port_));
+    ASSERT_OK(server_->Start(std::unique_ptr<ServerAuthHandler>(
+        new TestServerAuthHandler("user", "p4ssw0rd"))));
+    ASSERT_OK(ConnectClient());
+  }
+
+  void TearDown() { server_->Stop(); }
+
+  Status ConnectClient() { return FlightClient::Connect("localhost", port_, &client_); }
+
+ protected:
+  int port_;
+  std::unique_ptr<FlightClient> client_;
+  std::unique_ptr<InProcessTestServer> server_;
+};
+
 TEST_F(TestFlightClient, ListFlights) {
   std::unique_ptr<FlightListing> listing;
   ASSERT_OK(client_->ListFlights(&listing));
@@ -328,5 +359,99 @@ TEST_F(TestFlightClient, Issue5095) {
   ASSERT_THAT(status.message(), ::testing::HasSubstr("No data"));
 }
 
+TEST_F(TestAuthHandler, PassAuthenticatedCalls) {
+  ASSERT_OK(client_->Authenticate(
+      std::unique_ptr<ClientAuthHandler>(new TestClientAuthHandler("user", "p4ssw0rd"))));
+
+  Status status;
+  std::unique_ptr<FlightListing> listing;
+  status = client_->ListFlights(&listing);
+  ASSERT_RAISES(NotImplemented, status);
+
+  std::unique_ptr<ResultStream> results;
+  Action action;
+  action.type = "";
+  action.body = Buffer::FromString("");
+  status = client_->DoAction(action, &results);
+  ASSERT_OK(status);
+
+  std::vector<ActionType> actions;
+  status = client_->ListActions(&actions);
+  ASSERT_RAISES(NotImplemented, status);
+
+  std::unique_ptr<FlightInfo> info;
+  status = client_->GetFlightInfo(FlightDescriptor{}, &info);
+  ASSERT_RAISES(NotImplemented, status);
+
+  std::unique_ptr<RecordBatchReader> stream;
+  status = client_->DoGet(Ticket{}, &stream);
+  ASSERT_RAISES(NotImplemented, status);
+
+  std::unique_ptr<ipc::RecordBatchWriter> writer;
+  std::shared_ptr<Schema> schema = arrow::schema({});
+  status = client_->DoPut(FlightDescriptor{}, schema, &writer);
+  ASSERT_OK(status);
+  status = writer->Close();
+  ASSERT_RAISES(NotImplemented, status);
+}
+
+TEST_F(TestAuthHandler, FailUnauthenticatedCalls) {
+  Status status;
+  std::unique_ptr<FlightListing> listing;
+  status = client_->ListFlights(&listing);
+  ASSERT_RAISES(IOError, status);
+  ASSERT_THAT(status.message(), ::testing::HasSubstr("Invalid token"));
+
+  std::unique_ptr<ResultStream> results;
+  Action action;
+  action.type = "";
+  action.body = Buffer::FromString("");
+  status = client_->DoAction(action, &results);
+  ASSERT_RAISES(IOError, status);
+  ASSERT_THAT(status.message(), ::testing::HasSubstr("Invalid token"));
+
+  std::vector<ActionType> actions;
+  status = client_->ListActions(&actions);
+  ASSERT_RAISES(IOError, status);
+  ASSERT_THAT(status.message(), ::testing::HasSubstr("Invalid token"));
+
+  std::unique_ptr<FlightInfo> info;
+  status = client_->GetFlightInfo(FlightDescriptor{}, &info);
+  ASSERT_RAISES(IOError, status);
+  ASSERT_THAT(status.message(), ::testing::HasSubstr("Invalid token"));
+
+  std::unique_ptr<RecordBatchReader> stream;
+  status = client_->DoGet(Ticket{}, &stream);
+  ASSERT_RAISES(IOError, status);
+  ASSERT_THAT(status.message(), ::testing::HasSubstr("Invalid token"));
+
+  std::unique_ptr<ipc::RecordBatchWriter> writer;
+  std::shared_ptr<Schema> schema(
+      (new arrow::Schema(std::vector<std::shared_ptr<Field>>())));
+  status = client_->DoPut(FlightDescriptor{}, schema, &writer);
+  ASSERT_OK(status);
+  status = writer->Close();
+  ASSERT_RAISES(IOError, status);
+  ASSERT_THAT(status.message(), ::testing::HasSubstr("Invalid token"));
+}
+
+TEST_F(TestAuthHandler, CheckPeerIdentity) {
+  ASSERT_OK(client_->Authenticate(
+      std::unique_ptr<ClientAuthHandler>(new TestClientAuthHandler("user", "p4ssw0rd"))));
+
+  Action action;
+  action.type = "who-am-i";
+  action.body = Buffer::FromString("");
+  std::unique_ptr<ResultStream> results;
+  ASSERT_OK(client_->DoAction(action, &results));
+  ASSERT_NE(results, nullptr);
+
+  std::unique_ptr<Result> result;
+  ASSERT_OK(results->Next(&result));
+  ASSERT_NE(result, nullptr);
+  // Action returns the peer identity as the result.
+  ASSERT_EQ(result->body->ToString(), "user");
+}
+
 }  // namespace flight
 }  // namespace arrow

cpp/src/arrow/flight/internal.h

@@ -45,16 +45,26 @@ class Message;
 
 namespace flight {
 
-#define GRPC_RETURN_NOT_OK(s)                             \
+#define GRPC_RETURN_NOT_OK(expr)                          \
   do {                                                    \
-    ::arrow::Status _s = (s);                             \
+    ::arrow::Status _s = (expr);                          \
     if (ARROW_PREDICT_FALSE(!_s.ok())) {                  \
       return ::arrow::flight::internal::ToGrpcStatus(_s); \
     }                                                     \
   } while (0)
 
+#define GRPC_RETURN_NOT_GRPC_OK(expr)    \
+  do {                                   \
+    ::grpc::Status _s = (expr);          \
+    if (ARROW_PREDICT_FALSE(!_s.ok())) { \
+      return _s;                         \
+    }                                    \
+  } while (0)
+
 namespace internal {
 
+static const char* AUTH_HEADER = "auth-token-bin";
+
 Status SchemaToString(const Schema& schema, std::string* out);
 
 Status FromProto(const pb::ActionType& pb_type, ActionType* type);

cpp/src/arrow/flight/perf-server.cc

@@ -136,7 +136,7 @@ class FlightPerfServer : public FlightServerBase {
                            field("d", int64())});
   }
 
-  Status GetFlightInfo(const FlightDescriptor& request,
+  Status GetFlightInfo(const ServerCallContext& context, const FlightDescriptor& request,
                        std::unique_ptr<FlightInfo>* info) override {
     perf::Perf perf_request;
     CHECK_PARSE(perf_request.ParseFromString(request.cmd));
@@ -166,7 +166,7 @@ class FlightPerfServer : public FlightServerBase {
     return Status::OK();
   }
 
-  Status DoGet(const Ticket& request,
+  Status DoGet(const ServerCallContext& context, const Ticket& request,
                std::unique_ptr<FlightDataStream>* data_stream) override {
     perf::Token token;
     CHECK_PARSE(token.ParseFromString(request.ticket));
@@ -194,7 +194,8 @@ int main(int argc, char** argv) {
 
   g_server.reset(new arrow::flight::FlightPerfServer);
 
-  ARROW_CHECK_OK(g_server->Init(FLAGS_port));
+  ARROW_CHECK_OK(
+      g_server->Init(std::unique_ptr<arrow::flight::NoOpAuthHandler>(), FLAGS_port));
   // Exit with a clean error code (0) on SIGTERM
   ARROW_CHECK_OK(g_server->SetShutdownOnSignals({SIGTERM}));
   std::cout << "Server port: " << FLAGS_port << std::endl;

cpp/src/arrow/flight/server.cc

@@ -41,6 +41,7 @@
 
 #include "arrow/flight/internal.h"
 #include "arrow/flight/serialization-internal.h"
+#include "arrow/flight/server_auth.h"
 #include "arrow/flight/types.h"
 
 using FlightService = arrow::flight::protocol::FlightService;
@@ -105,11 +106,61 @@ class FlightMessageReaderImpl : public FlightMessageReader {
   bool stream_finished_;
 };
 
+class GrpcServerAuthReader : public ServerAuthReader {
+ public:
+  explicit GrpcServerAuthReader(
+      grpc::ServerReaderWriter<pb::HandshakeResponse, pb::HandshakeRequest>* stream)
+      : stream_(stream) {}
+
+  Status Read(std::string* token) const override {
+    pb::HandshakeRequest request;
+    if (stream_->Read(&request)) {
+      *token = std::move(*request.release_payload());
+      return Status::OK();
+    }
+    return Status::UnknownError("Could not read client handshake request.");
+  }
+
+ private:
+  grpc::ServerReaderWriter<pb::HandshakeResponse, pb::HandshakeRequest>* stream_;
+};
+
+class GrpcServerAuthSender : public ServerAuthSender {
+ public:
+  explicit GrpcServerAuthSender(
+      grpc::ServerReaderWriter<pb::HandshakeResponse, pb::HandshakeRequest>* stream)
+      : stream_(stream) {}
+
+  Status Write(const std::string& token) const override {
+    pb::HandshakeResponse response;
+    response.set_payload(token);
+    if (stream_->Write(response)) {
+      return Status::OK();
+    }
+    return Status::IOError("Stream was closed.");
+  }
+
+ private:
+  grpc::ServerReaderWriter<pb::HandshakeResponse, pb::HandshakeRequest>* stream_;
+};
+
+class FlightServiceImpl;
+class GrpcServerCallContext : public ServerCallContext {
+  const std::string& peer_identity() const override { return peer_identity_; }
+
+ private:
+  friend class FlightServiceImpl;
+  ServerContext* context_;
+  std::string peer_identity_;
+};
+
 // This class glues an implementation of FlightServerBase together with the
 // gRPC service definition, so the latter is not exposed in the public API
 class FlightServiceImpl : public FlightService::Service {
  public:
-  explicit FlightServiceImpl(FlightServerBase* server) : server_(server) {}
+  explicit FlightServiceImpl(std::shared_ptr<ServerAuthHandler> auth_handler,
+                             FlightServerBase* server)
+      : auth_handler_(auth_handler), server_(server) {}
 
   template <typename UserType, typename Iterator, typename ProtoType>
   grpc::Status WriteStream(Iterator* iterator, ServerWriter<ProtoType>* writer) {
@@ -151,8 +202,45 @@ class FlightServiceImpl : public FlightService::Service {
     return grpc::Status::OK;
   }
 
+  // Authenticate the client (if applicable) and construct the call context
+  grpc::Status CheckAuth(ServerContext* context, GrpcServerCallContext& flight_context) {
+    flight_context.context_ = context;
+    if (!auth_handler_) {
+      flight_context.peer_identity_ = "";
+      return grpc::Status::OK;
+    }
+
+    const auto client_metadata = context->client_metadata();
+    const auto auth_header = client_metadata.find(internal::AUTH_HEADER);
+    std::string token;
+    if (auth_header == client_metadata.end()) {
+      token = "";
+    } else {
+      token = std::string(auth_header->second.data(), auth_header->second.length());
+    }
+    GRPC_RETURN_NOT_OK(auth_handler_->IsValid(token, &flight_context.peer_identity_));
+    return grpc::Status::OK;
+  }
+
+  grpc::Status Handshake(
+      ServerContext* context,
+      grpc::ServerReaderWriter<pb::HandshakeResponse, pb::HandshakeRequest>* stream) {
+    if (!auth_handler_) {
+      return grpc::Status(
+          grpc::StatusCode::UNIMPLEMENTED,
+          "This service does not have an authentication mechanism enabled.");
+    }
+    GrpcServerAuthSender outgoing{stream};
+    GrpcServerAuthReader incoming{stream};
+    GRPC_RETURN_NOT_OK(auth_handler_->Authenticate(outgoing, incoming));
+    return grpc::Status::OK;
+  }
+
   grpc::Status ListFlights(ServerContext* context, const pb::Criteria* request,
                            ServerWriter<pb::FlightGetInfo>* writer) {
+    GrpcServerCallContext flight_context;
+    GRPC_RETURN_NOT_GRPC_OK(CheckAuth(context, flight_context));
+
     // Retrieve the listing from the implementation
     std::unique_ptr<FlightListing> listing;
 
@@ -160,7 +248,7 @@ class FlightServiceImpl : public FlightService::Service {
     if (request) {
       GRPC_RETURN_NOT_OK(internal::FromProto(*request, &criteria));
     }
-    GRPC_RETURN_NOT_OK(server_->ListFlights(&criteria, &listing));
+    GRPC_RETURN_NOT_OK(server_->ListFlights(flight_context, &criteria, &listing));
     if (!listing) {
       // Treat null listing as no flights available
       return grpc::Status::OK;
@@ -171,12 +259,14 @@ class FlightServiceImpl : public FlightService::Service {
   grpc::Status GetFlightInfo(ServerContext* context, const pb::FlightDescriptor* request,
                              pb::FlightGetInfo* response) {
     CHECK_ARG_NOT_NULL(request, "FlightDescriptor cannot be null");
+    GrpcServerCallContext flight_context;
+    GRPC_RETURN_NOT_GRPC_OK(CheckAuth(context, flight_context));
 
     FlightDescriptor descr;
     GRPC_RETURN_NOT_OK(internal::FromProto(*request, &descr));
 
     std::unique_ptr<FlightInfo> info;
-    GRPC_RETURN_NOT_OK(server_->GetFlightInfo(descr, &info));
+    GRPC_RETURN_NOT_OK(server_->GetFlightInfo(flight_context, descr, &info));
 
     if (!info) {
       // Treat null listing as no flights available
@@ -190,12 +280,14 @@ class FlightServiceImpl : public FlightService::Service {
   grpc::Status DoGet(ServerContext* context, const pb::Ticket* request,
                      ServerWriter<pb::FlightData>* writer) {
     CHECK_ARG_NOT_NULL(request, "ticket cannot be null");
+    GrpcServerCallContext flight_context;
+    GRPC_RETURN_NOT_GRPC_OK(CheckAuth(context, flight_context));
 
     Ticket ticket;
     GRPC_RETURN_NOT_OK(internal::FromProto(*request, &ticket));
 
     std::unique_ptr<FlightDataStream> data_stream;
-    GRPC_RETURN_NOT_OK(server_->DoGet(ticket, &data_stream));
+    GRPC_RETURN_NOT_OK(server_->DoGet(flight_context, ticket, &data_stream));
 
     if (!data_stream) {
       return grpc::Status(grpc::StatusCode::NOT_FOUND, "No data in this flight");
@@ -234,6 +326,8 @@ class FlightServiceImpl : public FlightService::Service {
 
   grpc::Status DoPut(ServerContext* context, grpc::ServerReader<pb::FlightData>* reader,
                      pb::PutResult* response) {
+    GrpcServerCallContext flight_context;
+    GRPC_RETURN_NOT_GRPC_OK(CheckAuth(context, flight_context));
     // Get metadata
     internal::FlightData data;
     if (internal::ReadPayload(reader, &data)) {
@@ -253,7 +347,8 @@ class FlightServiceImpl : public FlightService::Service {
 
         auto message_reader = std::unique_ptr<FlightMessageReader>(
             new FlightMessageReaderImpl(*data.descriptor.get(), schema, reader));
-        return internal::ToGrpcStatus(server_->DoPut(std::move(message_reader)));
+        return internal::ToGrpcStatus(
+            server_->DoPut(flight_context, std::move(message_reader)));
       }
     } else {
       return internal::ToGrpcStatus(
@@ -264,20 +359,24 @@ class FlightServiceImpl : public FlightService::Service {
 
   grpc::Status ListActions(ServerContext* context, const pb::Empty* request,
                            ServerWriter<pb::ActionType>* writer) {
+    GrpcServerCallContext flight_context;
+    GRPC_RETURN_NOT_GRPC_OK(CheckAuth(context, flight_context));
     // Retrieve the listing from the implementation
     std::vector<ActionType> types;
-    GRPC_RETURN_NOT_OK(server_->ListActions(&types));
+    GRPC_RETURN_NOT_OK(server_->ListActions(flight_context, &types));
     return WriteStream<ActionType>(types, writer);
   }
 
   grpc::Status DoAction(ServerContext* context, const pb::Action* request,
                         ServerWriter<pb::Result>* writer) {
+    GrpcServerCallContext flight_context;
+    GRPC_RETURN_NOT_GRPC_OK(CheckAuth(context, flight_context));
     CHECK_ARG_NOT_NULL(request, "Action cannot be null");
     Action action;
     GRPC_RETURN_NOT_OK(internal::FromProto(*request, &action));
 
     std::unique_ptr<ResultStream> results;
-    GRPC_RETURN_NOT_OK(server_->DoAction(action, &results));
+    GRPC_RETURN_NOT_OK(server_->DoAction(flight_context, action, &results));
 
     if (!results) {
       return grpc::Status::CANCELLED;
@@ -301,6 +400,7 @@ class FlightServiceImpl : public FlightService::Service {
   }
 
  private:
+  std::shared_ptr<ServerAuthHandler> auth_handler_;
   FlightServerBase* server_;
 };
 
@@ -341,9 +441,10 @@ FlightServerBase::FlightServerBase() { impl_.reset(new Impl); }
 
 FlightServerBase::~FlightServerBase() {}
 
-Status FlightServerBase::Init(int port) {
+Status FlightServerBase::Init(std::unique_ptr<ServerAuthHandler> auth_handler, int port) {
   impl_->address_ = "localhost:" + std::to_string(port);
-  impl_->service_.reset(new FlightServiceImpl(this));
+  std::shared_ptr<ServerAuthHandler> handler = std::move(auth_handler);
+  impl_->service_.reset(new FlightServiceImpl(handler, this));
 
   grpc::ServerBuilder builder;
   // Allow uploading messages of any length
@@ -409,31 +510,35 @@ void FlightServerBase::Shutdown() {
   impl_->server_->Shutdown();
 }
 
-Status FlightServerBase::ListFlights(const Criteria* criteria,
+Status FlightServerBase::ListFlights(const ServerCallContext& context,
+                                     const Criteria* criteria,
                                      std::unique_ptr<FlightListing>* listings) {
   return Status::NotImplemented("NYI");
 }
 
-Status FlightServerBase::GetFlightInfo(const FlightDescriptor& request,
+Status FlightServerBase::GetFlightInfo(const ServerCallContext& context,
+                                       const FlightDescriptor& request,
                                        std::unique_ptr<FlightInfo>* info) {
   return Status::NotImplemented("NYI");
 }
 
-Status FlightServerBase::DoGet(const Ticket& request,
+Status FlightServerBase::DoGet(const ServerCallContext& context, const Ticket& request,
                                std::unique_ptr<FlightDataStream>* data_stream) {
   return Status::NotImplemented("NYI");
 }
 
-Status FlightServerBase::DoPut(std::unique_ptr<FlightMessageReader> reader) {
+Status FlightServerBase::DoPut(const ServerCallContext& context,
+                               std::unique_ptr<FlightMessageReader> reader) {
   return Status::NotImplemented("NYI");
 }
 
-Status FlightServerBase::DoAction(const Action& action,
+Status FlightServerBase::DoAction(const ServerCallContext& context, const Action& action,
                                   std::unique_ptr<ResultStream>* result) {
   return Status::NotImplemented("NYI");
 }
 
-Status FlightServerBase::ListActions(std::vector<ActionType>* actions) {
+Status FlightServerBase::ListActions(const ServerCallContext& context,
+                                     std::vector<ActionType>* actions) {
   return Status::NotImplemented("NYI");
 }
 

cpp/src/arrow/flight/server.h

@@ -21,6 +21,7 @@
 #pragma once
 
 #include <memory>
+#include <string>
 #include <vector>
 
 #include "arrow/util/visibility.h"
@@ -36,6 +37,8 @@ class Status;
 
 namespace flight {
 
+class ServerAuthHandler;
+
 /// \brief Interface that produces a sequence of IPC payloads to be sent in
 /// FlightData protobuf messages
 class ARROW_EXPORT FlightDataStream {
@@ -72,6 +75,14 @@ class ARROW_EXPORT FlightMessageReader : public RecordBatchReader {
   virtual const FlightDescriptor& descriptor() const = 0;
 };
 
+/// \brief Call state/contextual data.
+class ARROW_EXPORT ServerCallContext {
+ public:
+  virtual ~ServerCallContext() = default;
+  /// \brief The name of the authenticated peer (may be the empty string)
+  virtual const std::string& peer_identity() const = 0;
+};
+
 /// \brief Skeleton RPC server implementation which can be used to create
 /// custom servers by implementing its abstract methods
 class ARROW_EXPORT FlightServerBase {
@@ -84,7 +95,10 @@ class ARROW_EXPORT FlightServerBase {
   /// \brief Initialize an insecure TCP server listening on localhost
   /// at the given port.
   /// This method must be called before any other method.
-  Status Init(int port);
+  /// \param[in] port The port to serve on.
+  /// \param[in] auth_handler The authentication handler. May be
+  /// nullptr if no authentication is desired.
+  Status Init(std::unique_ptr<ServerAuthHandler> auth_handler, int port);
 
   /// \brief Set the server to stop when receiving any of the given signal
   /// numbers.
@@ -113,41 +127,52 @@ class ARROW_EXPORT FlightServerBase {
 
   /// \brief Retrieve a list of available fields given an optional opaque
   /// criteria
+  /// \param[in] context The call context.
   /// \param[in] criteria may be null
   /// \param[out] listings the returned listings iterator
   /// \return Status
-  virtual Status ListFlights(const Criteria* criteria,
+  virtual Status ListFlights(const ServerCallContext& context, const Criteria* criteria,
                              std::unique_ptr<FlightListing>* listings);
 
   /// \brief Retrieve the schema and an access plan for the indicated
   /// descriptor
+  /// \param[in] context The call context.
   /// \param[in] request may be null
   /// \param[out] info the returned flight info provider
   /// \return Status
-  virtual Status GetFlightInfo(const FlightDescriptor& request,
+  virtual Status GetFlightInfo(const ServerCallContext& context,
+                               const FlightDescriptor& request,
                                std::unique_ptr<FlightInfo>* info);
 
   /// \brief Get a stream of IPC payloads to put on the wire
+  /// \param[in] context The call context.
   /// \param[in] request an opaque ticket
   /// \param[out] stream the returned stream provider
   /// \return Status
-  virtual Status DoGet(const Ticket& request, std::unique_ptr<FlightDataStream>* stream);
+  virtual Status DoGet(const ServerCallContext& context, const Ticket& request,
+                       std::unique_ptr<FlightDataStream>* stream);
 
   /// \brief Process a stream of IPC payloads sent from a client
+  /// \param[in] context The call context.
   /// \param[in] reader a sequence of uploaded record batches
   /// \return Status
-  virtual Status DoPut(std::unique_ptr<FlightMessageReader> reader);
+  virtual Status DoPut(const ServerCallContext& context,
+                       std::unique_ptr<FlightMessageReader> reader);
 
   /// \brief Execute an action, return stream of zero or more results
+  /// \param[in] context The call context.
   /// \param[in] action the action to execute, with type and body
   /// \param[out] result the result iterator
   /// \return Status
-  virtual Status DoAction(const Action& action, std::unique_ptr<ResultStream>* result);
+  virtual Status DoAction(const ServerCallContext& context, const Action& action,
+                          std::unique_ptr<ResultStream>* result);
 
   /// \brief Retrieve the list of available actions
+  /// \param[in] context The call context.
   /// \param[out] actions a vector of available action types
   /// \return Status
-  virtual Status ListActions(std::vector<ActionType>* actions);
+  virtual Status ListActions(const ServerCallContext& context,
+                             std::vector<ActionType>* actions);
 
  private:
   struct Impl;

cpp/src/arrow/flight/server_auth.cc

@@ -0,0 +1,37 @@
+// 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.
+
+#include "arrow/flight/server_auth.h"
+
+namespace arrow {
+namespace flight {
+
+ServerAuthHandler::~ServerAuthHandler() {}
+
+NoOpAuthHandler::~NoOpAuthHandler() {}
+Status NoOpAuthHandler::Authenticate(const ServerAuthSender& outgoing,
+                                     const ServerAuthReader& incoming) {
+  return Status::OK();
+}
+
+Status NoOpAuthHandler::IsValid(const std::string& token, std::string* peer_identity) {
+  *peer_identity = "";
+  return Status::OK();
+}
+
+}  // namespace flight
+}  // namespace arrow

cpp/src/arrow/flight/server_auth.h

@@ -0,0 +1,80 @@
+// 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.
+
+/// \brief Server-side APIs to implement authentication for Flight.
+
+#pragma once
+
+#include <string>
+
+#include "arrow/status.h"
+#include "arrow/util/visibility.h"
+
+namespace arrow {
+
+namespace flight {
+
+/// \brief A reader for messages from the client during an
+/// authentication handshake.
+class ARROW_EXPORT ServerAuthReader {
+ public:
+  virtual ~ServerAuthReader() = default;
+  virtual Status Read(std::string* token) const = 0;
+};
+
+/// \brief A writer for messages to the client during an
+/// authentication handshake.
+class ARROW_EXPORT ServerAuthSender {
+ public:
+  virtual ~ServerAuthSender() = default;
+  virtual Status Write(const std::string& message) const = 0;
+};
+
+/// \brief An authentication implementation for a Flight service.
+/// Authentication includes both an initial negotiation and a per-call
+/// token validation. Implementations may choose to use either or both
+/// mechanisms.
+/// An implementation may need to track some state, e.g. a mapping of
+/// client tokens to authenticated identities.
+class ARROW_EXPORT ServerAuthHandler {
+ public:
+  virtual ~ServerAuthHandler();
+  /// \brief Authenticate the client on initial connection. The server
+  /// can send and read responses from the client at any time.
+  virtual Status Authenticate(const ServerAuthSender& outgoing,
+                              const ServerAuthReader& incoming) = 0;
+  /// \brief Validate a per-call client token.
+  /// \param[in] token The client token. May be the empty string if
+  /// the client does not provide a token.
+  /// \param[out] peer_identity The identity of the peer, if this
+  /// authentication method supports it.
+  /// \return Status OK if the token is valid, any other status if
+  /// validation failed
+  virtual Status IsValid(const std::string& token, std::string* peer_identity) = 0;
+};
+
+/// \brief An authentication mechanism that does nothing.
+class ARROW_EXPORT NoOpAuthHandler : public ServerAuthHandler {
+ public:
+  ~NoOpAuthHandler();
+  Status Authenticate(const ServerAuthSender& outgoing,
+                      const ServerAuthReader& incoming) override;
+  Status IsValid(const std::string& token, std::string* peer_identity) override;
+};
+
+}  // namespace flight
+}  // namespace arrow

cpp/src/arrow/flight/test-integration-server.cc

@@ -32,6 +32,7 @@
 
 #include "arrow/flight/internal.h"
 #include "arrow/flight/server.h"
+#include "arrow/flight/server_auth.h"
 #include "arrow/flight/test-util.h"
 
 DEFINE_int32(port, 31337, "Server port to listen on");
@@ -40,7 +41,7 @@ namespace arrow {
 namespace flight {
 
 class FlightIntegrationTestServer : public FlightServerBase {
-  Status GetFlightInfo(const FlightDescriptor& request,
+  Status GetFlightInfo(const ServerCallContext& context, const FlightDescriptor& request,
                        std::unique_ptr<FlightInfo>* info) override {
     if (request.type == FlightDescriptor::PATH) {
       if (request.path.size() == 0) {
@@ -70,7 +71,7 @@ class FlightIntegrationTestServer : public FlightServerBase {
     }
   }
 
-  Status DoGet(const Ticket& request,
+  Status DoGet(const ServerCallContext& context, const Ticket& request,
                std::unique_ptr<FlightDataStream>* data_stream) override {
     auto data = uploaded_chunks.find(request.ticket);
     if (data == uploaded_chunks.end()) {
@@ -84,7 +85,8 @@ class FlightIntegrationTestServer : public FlightServerBase {
     return Status::OK();
   }
 
-  Status DoPut(std::unique_ptr<FlightMessageReader> reader) override {
+  Status DoPut(const ServerCallContext& context,
+               std::unique_ptr<FlightMessageReader> reader) override {
     const FlightDescriptor& descriptor = reader->descriptor();
 
     if (descriptor.type != FlightDescriptor::DescriptorType::PATH) {
@@ -122,7 +124,7 @@ int main(int argc, char** argv) {
   gflags::ParseCommandLineFlags(&argc, &argv, true);
 
   g_server.reset(new arrow::flight::FlightIntegrationTestServer);
-  ARROW_CHECK_OK(g_server->Init(FLAGS_port));
+  ARROW_CHECK_OK(g_server->Init(nullptr, FLAGS_port));
   // Exit with a clean error code (0) on SIGTERM
   ARROW_CHECK_OK(g_server->SetShutdownOnSignals({SIGTERM}));
 

cpp/src/arrow/flight/test-server.cc

@@ -30,6 +30,7 @@
 #include "arrow/util/logging.h"
 
 #include "arrow/flight/server.h"
+#include "arrow/flight/server_auth.h"
 #include "arrow/flight/test-util.h"
 
 DEFINE_int32(port, 31337, "Server port to listen on");
@@ -54,14 +55,14 @@ Status GetBatchForFlight(const Ticket& ticket, std::shared_ptr<RecordBatchReader
 }
 
 class FlightTestServer : public FlightServerBase {
-  Status ListFlights(const Criteria* criteria,
+  Status ListFlights(const ServerCallContext& context, const Criteria* criteria,
                      std::unique_ptr<FlightListing>* listings) override {
     std::vector<FlightInfo> flights = ExampleFlightInfo();
     *listings = std::unique_ptr<FlightListing>(new SimpleFlightListing(flights));
     return Status::OK();
   }
 
-  Status GetFlightInfo(const FlightDescriptor& request,
+  Status GetFlightInfo(const ServerCallContext& context, const FlightDescriptor& request,
                        std::unique_ptr<FlightInfo>* out) override {
     std::vector<FlightInfo> flights = ExampleFlightInfo();
 
@@ -74,7 +75,7 @@ class FlightTestServer : public FlightServerBase {
     return Status::Invalid("Flight not found: ", request.ToString());
   }
 
-  Status DoGet(const Ticket& request,
+  Status DoGet(const ServerCallContext& context, const Ticket& request,
                std::unique_ptr<FlightDataStream>* data_stream) override {
     // Test for ARROW-5095
     if (request.ticket == "ARROW-5095-fail") {
@@ -109,7 +110,8 @@ class FlightTestServer : public FlightServerBase {
     return Status::OK();
   }
 
-  Status DoAction(const Action& action, std::unique_ptr<ResultStream>* out) override {
+  Status DoAction(const ServerCallContext& context, const Action& action,
+                  std::unique_ptr<ResultStream>* out) override {
     if (action.type == "action1") {
       return RunAction1(action, out);
     } else if (action.type == "action2") {
@@ -119,7 +121,8 @@ class FlightTestServer : public FlightServerBase {
     }
   }
 
-  Status ListActions(std::vector<ActionType>* out) override {
+  Status ListActions(const ServerCallContext& context,
+                     std::vector<ActionType>* out) override {
     std::vector<ActionType> actions = ExampleActionTypes();
     *out = std::move(actions);
     return Status::OK();
@@ -135,7 +138,8 @@ int main(int argc, char** argv) {
   gflags::ParseCommandLineFlags(&argc, &argv, true);
 
   g_server.reset(new arrow::flight::FlightTestServer);
-  ARROW_CHECK_OK(g_server->Init(FLAGS_port));
+  ARROW_CHECK_OK(
+      g_server->Init(std::unique_ptr<arrow::flight::NoOpAuthHandler>(), FLAGS_port));
   // Exit with a clean error code (0) on SIGTERM
   ARROW_CHECK_OK(g_server->SetShutdownOnSignals({SIGTERM}));
 

cpp/src/arrow/flight/test-util.cc

@@ -118,6 +118,26 @@ bool TestServer::IsRunning() { return server_process_->running(); }
 
 int TestServer::port() const { return port_; }
 
+Status InProcessTestServer::Start(std::unique_ptr<ServerAuthHandler> auth_handler) {
+  RETURN_NOT_OK(server_->Init(std::move(auth_handler), port_));
+  thread_ = std::thread([this]() { ARROW_EXPECT_OK(server_->Serve()); });
+  return Status::OK();
+}
+
+void InProcessTestServer::Stop() {
+  server_->Shutdown();
+  thread_.join();
+}
+
+int InProcessTestServer::port() const { return port_; }
+
+InProcessTestServer::~InProcessTestServer() {
+  // Make sure server shuts down properly
+  if (thread_.joinable()) {
+    Stop();
+  }
+}
+
 Status MakeFlightInfo(const Schema& schema, const FlightDescriptor& descriptor,
                       const std::vector<FlightEndpoint>& endpoints, int64_t total_records,
                       int64_t total_bytes, FlightInfo::Data* out) {
@@ -193,5 +213,49 @@ std::vector<ActionType> ExampleActionTypes() {
   return {{"drop", "drop a dataset"}, {"cache", "cache a dataset"}};
 }
 
+TestServerAuthHandler::TestServerAuthHandler(const std::string& username,
+                                             const std::string& password)
+    : username_(username), password_(password) {}
+TestServerAuthHandler::~TestServerAuthHandler() {}
+Status TestServerAuthHandler::Authenticate(const ServerAuthSender& outgoing,
+                                           const ServerAuthReader& incoming) {
+  std::string token;
+  RETURN_NOT_OK(incoming.Read(&token));
+  if (token != password_) {
+    return Status::Invalid("Invalid password");
+  }
+  RETURN_NOT_OK(outgoing.Write(username_));
+  return Status::OK();
+}
+
+Status TestServerAuthHandler::IsValid(const std::string& token,
+                                      std::string* peer_identity) {
+  if (token != password_) {
+    return Status::Invalid("Invalid token");
+  }
+  *peer_identity = username_;
+  return Status::OK();
+}
+
+TestClientAuthHandler::TestClientAuthHandler(const std::string& username,
+                                             const std::string& password)
+    : username_(username), password_(password) {}
+TestClientAuthHandler::~TestClientAuthHandler() {}
+Status TestClientAuthHandler::Authenticate(const ClientAuthSender& outgoing,
+                                           const ClientAuthReader& incoming) {
+  RETURN_NOT_OK(outgoing.Write(password_));
+  std::string username;
+  RETURN_NOT_OK(incoming.Read(&username));
+  if (username != username_) {
+    return Status::Invalid("Invalid username");
+  }
+  return Status::OK();
+}
+
+Status TestClientAuthHandler::GetToken(std::string* token) {
+  *token = password_;
+  return Status::OK();
+}
+
 }  // namespace flight
 }  // namespace arrow

cpp/src/arrow/flight/test-util.h

@@ -18,10 +18,14 @@
 #include <cstdint>
 #include <memory>
 #include <string>
+#include <thread>
+#include <utility>
 #include <vector>
 
 #include "arrow/status.h"
 
+#include "arrow/flight/client_auth.h"
+#include "arrow/flight/server_auth.h"
 #include "arrow/flight/types.h"
 
 namespace boost {
@@ -57,6 +61,21 @@ class ARROW_EXPORT TestServer {
   std::shared_ptr<::boost::process::child> server_process_;
 };
 
+class ARROW_EXPORT InProcessTestServer {
+ public:
+  explicit InProcessTestServer(std::unique_ptr<FlightServerBase> server, int port)
+      : server_(std::move(server)), port_(port), thread_() {}
+  ~InProcessTestServer();
+  Status Start(std::unique_ptr<ServerAuthHandler> auth_handler);
+  void Stop();
+  int port() const;
+
+ private:
+  std::unique_ptr<FlightServerBase> server_;
+  int port_;
+  std::thread thread_;
+};
+
 // ----------------------------------------------------------------------
 // A RecordBatchReader for serving a sequence of in-memory record batches
 
@@ -111,5 +130,37 @@ Status MakeFlightInfo(const Schema& schema, const FlightDescriptor& descriptor,
                       const std::vector<FlightEndpoint>& endpoints, int64_t total_records,
                       int64_t total_bytes, FlightInfo::Data* out);
 
+// ----------------------------------------------------------------------
+// A pair of authentication handlers that check for a predefined password
+// and set the peer identity to a predefined username.
+
+class ARROW_EXPORT TestServerAuthHandler : public ServerAuthHandler {
+ public:
+  explicit TestServerAuthHandler(const std::string& username,
+                                 const std::string& password);
+  ~TestServerAuthHandler();
+  Status Authenticate(const ServerAuthSender& outgoing,
+                      const ServerAuthReader& incoming) override;
+  Status IsValid(const std::string& token, std::string* peer_identity) override;
+
+ private:
+  std::string username_;
+  std::string password_;
+};
+
+class ARROW_EXPORT TestClientAuthHandler : public ClientAuthHandler {
+ public:
+  explicit TestClientAuthHandler(const std::string& username,
+                                 const std::string& password);
+  ~TestClientAuthHandler();
+  Status Authenticate(const ClientAuthSender& outgoing,
+                      const ClientAuthReader& incoming) override;
+  Status GetToken(std::string* token) override;
+
+ private:
+  std::string username_;
+  std::string password_;
+};
+
 }  // namespace flight
 }  // namespace arrow

cpp/src/arrow/python/flight.cc

@@ -33,6 +33,7 @@ PyFlightServer::PyFlightServer(PyObject* server, PyFlightServerVtable vtable)
 }
 
 Status PyFlightServer::ListFlights(
+    const arrow::flight::ServerCallContext& context,
     const arrow::flight::Criteria* criteria,
     std::unique_ptr<arrow::flight::FlightListing>* listings) {
   PyAcquireGIL lock;

cpp/src/arrow/python/flight.h

@@ -60,7 +60,8 @@ class ARROW_PYTHON_EXPORT PyFlightServer : public arrow::flight::FlightServerBas
   // if necessary.  This function may return with a Python exception set.
   Status ServeWithSignals();
 
-  Status ListFlights(const arrow::flight::Criteria* criteria,
+  Status ListFlights(const arrow::flight::ServerCallContext& context,
+                     const arrow::flight::Criteria* criteria,
                      std::unique_ptr<arrow::flight::FlightListing>* listings) override;
   Status GetFlightInfo(const arrow::flight::FlightDescriptor& request,
                        std::unique_ptr<arrow::flight::FlightInfo>* info) override;