HDFSRPC安全认证Kerberos篇

本文主要阐述HDFSRPC安全认证相关的实现。主要介绍Kerberos相关的实现。

写在前面

Rpc安全认证

Rpc安全认证使用的是sasl框架，sasl框架本身无认证相关的实现，认证实现使用的Kerberos。SASL: 在jdk中定义的一种通用的基于客户端和服务端的认证框架，GSSAPI是其实现之一。

GSSAPI: 在jdk中，作为对kerberos认证实现的一部分。

Kerberos: 一种基于中心认证服务器的中心化认证协议和框架。应用程序访问服务前需使用此框架进行登录认证，以在应用程序之间形成动态可控的受信。中心化登录服务器称为KDC。

Krb5LoginModule: 在jdk中，负责从KDC获取登录凭证，是kerberos认证实现的一部分。

SASL

简单认证与安全层（Simple Authentication And Security Layer）是一个在网络协议中用来认证和数据加密的框架。它把认证机制从程序中分离开，理论上使用SASL的程序协议都可以使用SASL所支持的全部认证机制（token认证就是其中的一种认证机制）。

认证机制可支持代理认证，这让一个用户可以承担另一个用户的认证。

SASL同样提供数据安全层，这提供了数据完整验证和数据加密，例如DIGEST_MD5提供了数据加密层。

SASL是一种challenge-response的协议，由服务端发送challenge到客户端，客户端基于challenge发送response，这种交互直到服务端被满足并且不再发布新的challenge。

challenge和对应的response都是任意长度的二进制数据。其大概流程如下所示：

要注意一点sasl是一种框架，不涉及具体实现，通信时可以自己定义相关的包。

Hdfs中sasl相关实现

Negotiate：

client会发送一个saslMessage给server，其中saslstate为Negotiate，无其他信息。Server接收到Negotiate请求后，会返回一个Negotiate的saslMessage，其中包含sasl使用需要使用哪种协议，例如：

auths {
  method: "KERBEROS"
  mechanism: "GSSAPI"
  protocol: "root"
  serverId: "node17"
}

Initiate：

client接收到Server的Negotiate的saslMessage后，会根据相关的信息，初始化saslClient，并产生一个Token，发送一个saslstate为Initiate的saslMessage给server。Server接收到saslMessage以后，同样会初始化saslServer，然后evaluate token，生成新的token，返回一个saslstate为challenge的saslMessage给client。

Response-challenge：

client接收到Server的challenge的saslMessage后，会evaluate challenge token，产生一个新的token，然后发送一个saslstate为Response的saslMessage给server。。Server接收到saslMessage以后，然后evaluate token，生成新的token，如果server已经完成初始化，返回一个saslstate为success的saslMessage给client，反之则返回一个saslstate为challenge的saslMessage给client。Client接收如果为challenge的saslMessage则重复上述流程，反之如果接收到success则完成client的初始化。一般这个过程会经过两轮。

GSSAPI

在jdk中，作为对kerberos认证实现的一部分。

Kerberos

Hadoop 使用 Kerberos 作为用户和服务的强身份验证和身份传播的基础。Kerberos 是一种计算机网络认证协议，它允许某实体在非安全网络环境下通信，向另一个实体以一种安全的方式证明自己的身份。 Kerberos 是第三方认证机制，其中用户和服务依赖于第三方（Kerberos 服务器）来对彼此进行身份验证。Kerberos服务器本身称为密钥分发中心或 KDC。在较高的层面上，它有三个部分：

1）它知道的用户和服务（称为主体）及其各自的 Kerberos 密码的数据库

2）一个认证服务器（Authentication Server，简称 AS）：验证Client端的身份（确定你是身份证上的本人），验证通过就会给一张票证授予票证（Ticket Granting Ticket，简称 TGT）给 Client。

3）一个票据授权服务器（Ticket Granting Server，简称 TGS）：通过 TGT（AS 发送给 Client 的票）获取访问 Server 端的票（Server Ticket，简称 ST）。ST（Service Ticket）也有资料称为 TGS Ticket。

以平时坐火车举例：

一个用户主要来自AS请求认证。AS 返回使用用户主体的 Kerberos密码加密的 TGT ，该密码仅为用户主体和 AS 所知。用户主体使用其 Kerberos 密码在本地解密TGT，从那时起，直到ticket 到期，用户主体可以使用 TGT 从 TGS 获取服务票据。服务票证允许委托人访问服务。

Hdfs中的Kerberos

在sasl中似乎没有提到Kerberos，实际上saslclient、saslserver的初始化后需要通过Kerberos验证以后才能实现。具体流程如下：

Client：

Client使用的当前登录用户，需要使用kinit登录用户，然后代码中使用通用库登录，获取Subject。Subject有doas方法，可以使用此方法来初始化saslclient，已经使用saslclient.evaluateChallenge方法。

Server：

Server为服务类型，通常使用免密登录。需要单独的 Kerberos 账号及其 keytab 文件。有了这个以后，可以使用通用库登录，获取Subject。Subject有doas方法，可以使用此方法来初始化saslserver，已经使用saslserver.evaluateResponse方法。

为了更好的理解整个过程，附上Java代码简单实现

Java代码简单实现

SaslClient.java

public class SaslClientTest {
  private static Socket socket;

  public static void main(String[] args) throws UnknownHostException, IOException, LoginException, PrivilegedActionException {
    //kerberos login
    Configuration config = new Configuration() {
      @Override
      public AppConfigurationEntry[] getAppConfigurationEntry(String name) {
        ArrayList<AppConfigurationEntry> entries = new ArrayList<>();
        String loginModuleName = "com.sun.security.auth.module.Krb5LoginModule";
        final Map<String,String> options = new HashMap<>();
        LoginModuleControlFlag controlFlag = LoginModuleControlFlag.OPTIONAL;
        options.put("useTicketCache", "true");
        options.put("renewTGT", "true");
        options.put("doNotPrompt", "true");
        options.put("refreshKrb5Config", "true");
        AppConfigurationEntry kerberosEntry = new AppConfigurationEntry(loginModuleName, controlFlag, options);
        entries.add(kerberosEntry);
        return entries.toArray(new AppConfigurationEntry[0]);
      }
    };
    LoginContext login = new LoginContext("hadoop-kerberos", null, null, config);
    login.login();
    Subject subject = login.getSubject();
    System.out.println(subject.getPrincipals());

    Subject.doAs(subject, new PrivilegedExceptionAction<String>() {
      @Override
      public String run() throws Exception {
        String hostname = "localhost";
        int port = 19876;
        socket = new Socket(hostname, port);
        DataInputStream socketIn = new DataInputStream(socket.getInputStream());
        DataOutputStream socketOut = new DataOutputStream(socket.getOutputStream());
        System.out.println("Connected to server " + socket.getInetAddress());

        boolean done = false;
        SaslClient saslClient = null;
        byte[] responseToken;
        Packet requestPacket;
        Packet negotiatePacket = new Packet(SaslState.NEGOTIATE);
        negotiatePacket.write(socketOut);
        do {
          Packet responsePacket = Packet.read(socketIn);
          switch (responsePacket.getState()) {
            case NEGOTIATE:
              //create saslclient
              //auths { method: "KERBEROS" mechanism: "GSSAPI" protocol: "root" serverId: "node17" }
              String mechanism = "GSSAPI";
              String saslProtocol = "root";
              String saslServerName = "node17";
              String saslUser = null;
              CallbackHandler saslCallback = null;
              Map<String, String> saslProperties = new HashMap<String, String>();
              saslProperties.put("javax.security.sasl.qop", "auth");
              saslProperties.put("javax.security.sasl.server.authentication", "true");
              saslClient = Sasl.createSaslClient(new String[]{mechanism}, saslUser, saslProtocol, saslServerName, saslProperties, saslCallback);

              byte[] challengeToken = null;
              if (saslClient.hasInitialResponse()) {
                challengeToken = new byte[0];
              }
              responseToken = (challengeToken != null) ? saslClient.evaluateChallenge(challengeToken) : new byte[0];
              requestPacket = new Packet(SaslState.INITIATE, responseToken);
              requestPacket.write(socketOut);
              break;
            case CHALLENGE:
              if (saslClient == null) {
                // should probably instantiate a client to allow a server to
                // demand a specific negotiation
                throw new SaslException("Server sent unsolicited challenge");
              }
              responseToken = saslEvaluateToken(responsePacket, false, saslClient);
              requestPacket = new Packet(SaslState.RESPONSE, responseToken);
              requestPacket.write(socketOut);
              break;
            case SUCCESS:
              if (saslClient == null) {
                throw new SaslException("Server sent unsolicited success");
              } else {
                saslEvaluateToken(responsePacket, true, saslClient);
              }
              done = true;
              break;
            default:
              throw new SaslException("error state");
          }
        }while (!done);
        socket.close();
        return null;
      }
    });
  }

  private static byte[] saslEvaluateToken(Packet response, boolean serverIsDone, SaslClient saslClient) throws SaslException {
    byte[] saslToken = null;
    if (!serverIsDone && response.hasToken()) {
      saslToken = response.getToken();
      saslToken = saslClient.evaluateChallenge(saslToken);
    } else if (!serverIsDone) {
      // the server may only omit a token when it's done
      throw new SaslException("Server challenge contains no token");
    }
    if (serverIsDone) {
      // server tried to report success before our client completed
      if (!saslClient.isComplete()) {
        throw new SaslException("Client is out of sync with server");
      }
      // a client cannot generate a response to a success message
      if (saslToken != null) {
        throw new SaslException("Client generated spurious response");        
      }
    }
    return saslToken;
  }
}

Saslserver.java

public class SaslServerTest {
  private static ServerSocket ss;

  public static void main(String[] args) throws IOException, LoginException, PrivilegedActionException {
    int port = 19876;
    ss = new ServerSocket(port);
    //login kerberos
    Configuration config = new Configuration() {
      @Override
      public AppConfigurationEntry[] getAppConfigurationEntry(String name) {
        ArrayList<AppConfigurationEntry> entries = new ArrayList<>();
        String loginModuleName = "com.sun.security.auth.module.Krb5LoginModule";
        final Map<String,String> options = new HashMap<>();
        LoginModuleControlFlag controlFlag = LoginModuleControlFlag.REQUIRED;
        options.put("principal", "root/node17@LOONGOOP.COM");
        options.put("useKeyTab", "true");
        options.put("keyTab", "/var/kerberos/krb5kdc/nn.keytab");
        options.put("storeKey", "true");
        options.put("doNotPrompt", "true");
        options.put("refreshKrb5Config", "true");
        AppConfigurationEntry kerberosEntry = new AppConfigurationEntry(loginModuleName, controlFlag, options);
        entries.add(kerberosEntry);
        return entries.toArray(new AppConfigurationEntry[0]);
      }
    };
    LoginContext  login = new LoginContext("hadoop-kerberos", null, null, config);
    login.login();
    Subject subject = login.getSubject();
    System.out.println(subject.getPrincipals());
    while(true) {
      Subject.doAs(subject, new PrivilegedExceptionAction<String>() {

        @Override
        public String run() throws Exception {
          Socket socket = ss.accept();
          DataInputStream socketIn = new DataInputStream(socket.getInputStream());
          DataOutputStream socketOut = new DataOutputStream(socket.getOutputStream());
          System.out.println("Got connection from client " + socket.getInetAddress());
          boolean done = false;
          Packet responsePacket;
          SaslServer saslServer = null;
          do {
            Packet requestPacket = Packet.read(socketIn);
            switch(requestPacket.getState()) {
              case NEGOTIATE:
                responsePacket = new Packet(SaslState.NEGOTIATE);
                requestPacket.write(socketOut);
                break;
              case INITIATE:
                saslServer = Subject.doAs(subject, new PrivilegedExceptionAction<SaslServer>() {
                  @Override
                  public SaslServer run() throws Exception {
                    //auths { method: "KERBEROS" mechanism: "GSSAPI" protocol: "root" serverId: "node17" }
                    String mechanism = "GSSAPI";
                    String protocol = "root";
                    String serverId = "node17";
                    final CallbackHandler callback = new SaslGssCallbackHandler();
                    return FastSaslServerFactory.getInstance().createSaslServer(mechanism, protocol, serverId, null, callback);
                  }
                });

                responsePacket = processSaslToken(requestPacket, saslServer);
                responsePacket.write(socketOut);
                break;
              case RESPONSE:
                responsePacket = processSaslToken(requestPacket, saslServer);
                responsePacket.write(socketOut);
                if(saslServer.isComplete()) {
                  done = true;
                }
                break;
              default:
                throw new SaslException("error sasl state");
            }
          } while(!done);  
          return null;
        }
      });
    }
  }

  private static Packet processSaslToken(Packet requestPacket, SaslServer saslServer) throws SaslException {
    if (!requestPacket.hasToken()) {
      throw new SaslException("Client did not send a token");
    }
    if (saslServer == null) {
      throw new SaslException("sasl server is null");
    }
    byte[] saslToken = requestPacket.getToken();
    saslToken = saslServer.evaluateResponse(saslToken);
    return new Packet(saslServer.isComplete() ? SaslState.SUCCESS : SaslState.CHALLENGE, saslToken);
  }

  public static class SaslGssCallbackHandler implements CallbackHandler {
    @Override
    public void handle(Callback[] callbacks) throws UnsupportedCallbackException {
      AuthorizeCallback ac = null;
      for (Callback callback : callbacks) {
        if (callback instanceof AuthorizeCallback) {
          ac = (AuthorizeCallback) callback;
        } else {
          throw new UnsupportedCallbackException(callback, "Unrecognized SASL GSSAPI Callback");
        }
      }
      if (ac != null) {
        String authid = ac.getAuthenticationID();
        String authzid = ac.getAuthorizationID();
        if (authid.equals(authzid)) {
          ac.setAuthorized(true);
        } else {
          ac.setAuthorized(false);
        }
        if (ac.isAuthorized()) {
          ac.setAuthorizedID(authzid);
        }
      }
    }
  }
}

独立站原文