import quincy 17.2.0

[ceph.git] / ceph / src / arrow / java / flight / flight-core / src / main / java / org / apache / arrow / flight / example / integration / AuthBasicProtoScenario.java

/*
 * 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.example.integration;

import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.Optional;

import org.apache.arrow.flight.Action;
import org.apache.arrow.flight.CallStatus;
import org.apache.arrow.flight.FlightClient;
import org.apache.arrow.flight.FlightProducer;
import org.apache.arrow.flight.FlightRuntimeException;
import org.apache.arrow.flight.FlightServer;
import org.apache.arrow.flight.FlightStatusCode;
import org.apache.arrow.flight.Location;
import org.apache.arrow.flight.NoOpFlightProducer;
import org.apache.arrow.flight.Result;
import org.apache.arrow.flight.auth.BasicClientAuthHandler;
import org.apache.arrow.flight.auth.BasicServerAuthHandler;
import org.apache.arrow.memory.BufferAllocator;

/**
 * A scenario testing the built-in basic authentication Protobuf.
 */
final class AuthBasicProtoScenario implements Scenario {

  static final String USERNAME = "arrow";
  static final String PASSWORD = "flight";

  @Override
  public FlightProducer producer(BufferAllocator allocator, Location location) {
    return new NoOpFlightProducer() {
      @Override
      public void doAction(CallContext context, Action action, StreamListener<Result> listener) {
        listener.onNext(new Result(context.peerIdentity().getBytes(StandardCharsets.UTF_8)));
        listener.onCompleted();
      }
    };
  }

  @Override
  public void buildServer(FlightServer.Builder builder) {
    builder.authHandler(new BasicServerAuthHandler(new BasicServerAuthHandler.BasicAuthValidator() {
      @Override
      public byte[] getToken(String username, String password) throws Exception {
        if (!USERNAME.equals(username) || !PASSWORD.equals(password)) {
          throw CallStatus.UNAUTHENTICATED.withDescription("Username or password is invalid.").toRuntimeException();
        }
        return ("valid:" + username).getBytes(StandardCharsets.UTF_8);
      }

      @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));
          }
        }
        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()));
    }
  }
}