import org.apache.thrift.transport.TSaslServerTransport; //导入依赖的package包/类
private TTransportFactory getSASLTransportFactory() {
  String[] names;
  try {
    names = FlumeAuthenticationUtil.splitKerberosName(principal);
  } catch (IOException e) {
    throw new FlumeException(
            "Error while trying to resolve Principal name - " + principal, e);
  }
  Map<String, String> saslProperties = new HashMap<String, String>();
  saslProperties.put(Sasl.QOP, "auth");
  TSaslServerTransport.Factory saslTransportFactory =
          new TSaslServerTransport.Factory();
  saslTransportFactory.addServerDefinition(
          "GSSAPI", names[0], names[1], saslProperties,
          FlumeAuthenticationUtil.getSaslGssCallbackHandler());
  return saslTransportFactory;
}
 

import org.apache.thrift.transport.TSaslServerTransport; //导入依赖的package包/类
public boolean process(final TProtocol inProt, final TProtocol outProt) throws TException {
    // populating request context
    ReqContext req_context = ReqContext.context();

    TTransport trans = inProt.getTransport();
    // Sasl transport
    TSaslServerTransport saslTrans = (TSaslServerTransport) trans;
    // remote address
    TSocket tsocket = (TSocket) saslTrans.getUnderlyingTransport();
    Socket socket = tsocket.getSocket();
    req_context.setRemoteAddress(socket.getInetAddress());

    // remote subject
    SaslServer saslServer = saslTrans.getSaslServer();
    String authId = saslServer.getAuthorizationID();
    Subject remoteUser = new Subject();
    remoteUser.getPrincipals().add(new User(authId));
    req_context.setSubject(remoteUser);

    // invoke service handler
    return wrapped.process(inProt, outProt);
}
 

import org.apache.thrift.transport.TSaslServerTransport; //导入依赖的package包/类
public boolean process(final TProtocol inProt, final TProtocol outProt) throws TException {
    //populating request context 
    ReqContext req_context = ReqContext.context();

    TTransport trans = inProt.getTransport();
    //Sasl transport
    TSaslServerTransport saslTrans = (TSaslServerTransport)trans;

    //remote address
    TSocket tsocket = (TSocket)saslTrans.getUnderlyingTransport();
    Socket socket = tsocket.getSocket();
    req_context.setRemoteAddress(socket.getInetAddress());

    //remote subject 
    SaslServer saslServer = saslTrans.getSaslServer();
    String authId = saslServer.getAuthorizationID();
    Subject remoteUser = new Subject();
    remoteUser.getPrincipals().add(new User(authId));
    req_context.setSubject(remoteUser);
    
    //invoke service handler
    return wrapped.process(inProt, outProt);
}
 

import org.apache.thrift.transport.TSaslServerTransport; //导入依赖的package包/类
protected TTransportFactory getServerTransportFactory() throws IOException {
    // create an authentication callback handler
    CallbackHandler serer_callback_handler = new ServerCallbackHandler(login_conf);

    // create a transport factory that will invoke our auth callback for digest
    TSaslServerTransport.Factory factory = new TSaslServerTransport.Factory();
    factory.addServerDefinition(DIGEST, AuthUtils.SERVICE, "localhost", null, serer_callback_handler);

    LOG.info("SASL DIGEST-MD5 transport factory will be used");
    return factory;
}
 

import org.apache.thrift.transport.TSaslServerTransport; //导入依赖的package包/类
protected TTransportFactory getServerTransportFactory() throws IOException {        
    //create an authentication callback handler
    CallbackHandler serer_callback_handler = new ServerCallbackHandler(login_conf);

    //create a transport factory that will invoke our auth callback for digest
    TSaslServerTransport.Factory factory = new TSaslServerTransport.Factory();
    factory.addServerDefinition(DIGEST, AuthUtils.SERVICE, "localhost", null, serer_callback_handler);

    LOG.info("SASL DIGEST-MD5 transport factory will be used");
    return factory;
}
 

import org.apache.thrift.transport.TSaslServerTransport; //导入依赖的package包/类
public static void setImpersonator(final TProtocol in) {
  try {
    TTransport transport = in.getTransport();
    if (transport instanceof TSaslServerTransport) {
      String impersonator = ((TSaslServerTransport) transport).getSaslServer()
          .getAuthorizationID();
      setImpersonator(impersonator);
    }
  } catch (Exception e) {
    // If there has exception when get impersonator info, log the error information.
    LOGGER.warn("There is an error when get the impersonator:" + e.getMessage());
  }
}
 

import org.apache.thrift.transport.TSaslServerTransport; //导入依赖的package包/类
/**
 * Returns the underlying TSocket from the transport, or null of the transport type is unknown.
 */
private static TSocket getUnderlyingSocketFromTransport(TTransport transport) {
  Preconditions.checkNotNull(transport);
  if (transport instanceof TSaslServerTransport) {
    return (TSocket) ((TSaslServerTransport) transport).getUnderlyingTransport();
  } else if (transport instanceof TSaslClientTransport) {
    return (TSocket) ((TSaslClientTransport) transport).getUnderlyingTransport();
  } else if (transport instanceof TSocket) {
    return (TSocket) transport;
  }
  return null;
}
 

import org.apache.thrift.transport.TSaslServerTransport; //导入依赖的package包/类
public TTransportFactory getServerTransportFactory() throws IOException {
    // create an authentication callback handler
    CallbackHandler server_callback_handler = new ServerCallbackHandler(login_conf, storm_conf);

    // login our principal
    Subject subject = null;
    try {
        // specify a configuration object to be used
        Configuration.setConfiguration(login_conf);
        // now login
        Login login = new Login(AuthUtils.LOGIN_CONTEXT_SERVER, server_callback_handler);
        subject = login.getSubject();
    } catch (LoginException ex) {
        LOG.error("Server failed to login in principal:" + ex, ex);
        throw new RuntimeException(ex);
    }

    // check the credential of our principal
    if (subject.getPrivateCredentials(KerberosTicket.class).isEmpty()) {
        throw new RuntimeException("Fail to verify user principal with section \"" + AuthUtils.LOGIN_CONTEXT_SERVER + "\" in login configuration file "
                + login_conf);
    }

    String principal = AuthUtils.get(login_conf, AuthUtils.LOGIN_CONTEXT_SERVER, "principal");
    LOG.debug("principal:" + principal);
    KerberosName serviceKerberosName = new KerberosName(principal);
    String serviceName = serviceKerberosName.getServiceName();
    String hostName = serviceKerberosName.getHostName();
    Map<String, String> props = new TreeMap<String, String>();
    props.put(Sasl.QOP, "auth");
    props.put(Sasl.SERVER_AUTH, "false");

    // create a transport factory that will invoke our auth callback for digest
    TSaslServerTransport.Factory factory = new TSaslServerTransport.Factory();
    factory.addServerDefinition(KERBEROS, serviceName, hostName, props, server_callback_handler);

    // create a wrap transport factory so that we could apply user credential during connections
    TUGIAssumingTransportFactory wrapFactory = new TUGIAssumingTransportFactory(factory, subject);

    LOG.info("SASL GSSAPI transport factory will be used");
    return wrapFactory;
}
 

import org.apache.thrift.transport.TSaslServerTransport; //导入依赖的package包/类
public static void main(String[] args) throws Exception {
  Opts opts = new Opts();
  
  opts.parseArgs(Server.class, args);
  
  Configuration conf = new Configuration();
  FileSystem fs = FileSystem.get(conf);
  
  // Parse out the primary/[email protected] from the principal
  String principal = SecurityUtil.getServerPrincipal(opts.principal, InetAddress.getLocalHost().getCanonicalHostName());
  HadoopKerberosName name = new HadoopKerberosName(principal);
  String primary = name.getServiceName();
  String instance = name.getHostName();
  
  // Log in using the keytab
  UserGroupInformation.loginUserFromKeytab(principal, opts.keytab);
  
  // Get the info from our login
  UserGroupInformation serverUser = UserGroupInformation.getLoginUser();
  log.info("Current user: {}", serverUser);
  
  // Open the server using the provide dport
  TServerSocket serverTransport = new TServerSocket(opts.port);
  
  // Wrap our implementation with the interface's processor
  HdfsService.Processor<Iface> processor = new HdfsService.Processor<Iface>(new HdfsServiceImpl(fs));
  
  // Use authorization and confidentiality
  Map<String,String> saslProperties = new HashMap<String,String>();
  saslProperties.put(Sasl.QOP, "auth-conf");
  
  // Creating the server definition
  TSaslServerTransport.Factory saslTransportFactory = new TSaslServerTransport.Factory();
      saslTransportFactory.addServerDefinition("GSSAPI", // tell SASL to use GSSAPI, which supports Kerberos
      primary, // kerberos primary for server - "myprincipal" in myprincipal/[email protected]
      instance, // kerberos instance for server - "my.server.com" in myprincipal/[email protected]
      saslProperties, // Properties set, above
      new SaslRpcServer.SaslGssCallbackHandler()); // Ensures that authenticated user is the same as the authorized user
  
  // Make sure the TTransportFactory is performing a UGI.doAs
  TTransportFactory ugiTransportFactory = new TUGIAssumingTransportFactory(saslTransportFactory, serverUser);
  
  // Processor which takes the UGI for the RPC call, proxy that user on the server login, and then run as the proxied user
  TUGIAssumingProcessor ugiProcessor = new TUGIAssumingProcessor(processor);
  
  // Make a simple TTheadPoolServer with the processor and transport factory
  TServer server = new TThreadPoolServer(new TThreadPoolServer.Args(serverTransport).transportFactory(ugiTransportFactory).processor(ugiProcessor));
  
  // Start the thrift server
  server.serve();
}
 

import org.apache.thrift.transport.TSaslServerTransport; //导入依赖的package包/类
@Override
public boolean process(final TProtocol inProt, final TProtocol outProt) throws TException {
  TTransport trans = inProt.getTransport();
  if (!(trans instanceof TSaslServerTransport)) {
    throw new TException("Unexpected non-SASL transport " + trans.getClass());
  }
  TSaslServerTransport saslTrans = (TSaslServerTransport) trans;
  SaslServer saslServer = saslTrans.getSaslServer();
  String authId = saslServer.getAuthorizationID();
  String endUser = authId;

  UserGroupInformation clientUgi = null;
  try {
    clientUgi = UserGroupInformation.createProxyUser(endUser, UserGroupInformation.getLoginUser());
    final String remoteUser = clientUgi.getShortUserName();
    log.debug("Executing action as {}", remoteUser);
    return clientUgi.doAs(new PrivilegedExceptionAction<Boolean>() {
      @Override
      public Boolean run() {
        try {
          return wrapped.process(inProt, outProt);
        } catch (TException te) {
          throw new RuntimeException(te);
        }
      }
    });
  } catch (RuntimeException rte) {
    if (rte.getCause() instanceof TException) {
      log.error("Failed to invoke wrapped processor", rte.getCause());
      throw (TException) rte.getCause();
    }
    throw rte;
  } catch (InterruptedException | IOException e) {
    log.error("Failed to invoke wrapped processor", e);
    throw new RuntimeException(e);
  } finally {
    if (clientUgi != null) {
      try {
        FileSystem.closeAllForUGI(clientUgi);
      } catch (IOException exception) {
        log.error("Could not clean up file-system handles for UGI: {}", clientUgi, exception);
      }
    }
  }
}