hadoop SLA

    SLA：Service Level Authorization(服務層受權)

1. sla 經過配置訪問策略來控制哪些用戶、哪些用戶組、哪些機器能夠訪問特定的服務。

2. 若是啓用了sla (hadoop.security.authorization=true)，則每次服務端收到rpc請求後，都會對請求作權限檢測。RPCServer在收到客戶端的rpc請求後會對該請求作sla檢測。

3. 核心代碼：

   流程：
   processRpcOutOfBandRequest(RpcRequestHeaderProto header, DataInputStream dis)
   -->org.apache.hadoop.ipc.Server.Connection.processConnectionContext(DataInputStream dis)
   -->org.apache.hadoop.ipc.Server.Connection.authorizeConnection()
   -->org.apache.hadoop.ipc.Server.authorize(UserGroupInformation user, String protocolName, InetAddress addr)
   -->org.apache.hadoop.security.authorize.ServiceAuthorizationManager.authorize(UserGroupInformation user, Class<?> protocol, Configuration conf, InetAddress addr)

   /**
        * Authorize proxy users to access this server
        * @throws WrappedRpcServerException - user is not allowed to proxy
        */
       private void authorizeConnection() throws WrappedRpcServerException {
         try {
           // If auth method is TOKEN, the token was obtained by the
           // real user for the effective user, therefore not required to
           // authorize real user. doAs is allowed only for simple or kerberos
           // authentication
           if (user != null && user.getRealUser() != null
               && (authMethod != AuthMethod.TOKEN)) {
             ProxyUsers.authorize(user, this.getHostAddress());
           }
           //用戶user從主機（getHostInetAddress()）發起了協議protocolName的訪問，此處對該訪問進行受權
           authorize(user, protocolName, getHostInetAddress());
           if (LOG.isDebugEnabled()) {
             LOG.debug("Successfully authorized " + connectionContext);
           }
           rpcMetrics.incrAuthorizationSuccesses();
         } catch (AuthorizationException ae) {
           LOG.info("Connection from " + this
               + " for protocol " + connectionContext.getProtocol()
               + " is unauthorized for user " + user);
           rpcMetrics.incrAuthorizationFailures();
           throw new WrappedRpcServerException(
               RpcErrorCodeProto.FATAL_UNAUTHORIZED, ae);
         }
       }
   
   
     /**
      * Authorize the incoming client connection.
      * 
      * @param user client user
      * @param protocolName - the protocol
      * @param addr InetAddress of incoming connection
      * @throws AuthorizationException when the client isn't authorized to talk the protocol
      */
     private void authorize(UserGroupInformation user, String protocolName,
         InetAddress addr) throws AuthorizationException {
       if (authorize) {
         if (protocolName == null) {
           throw new AuthorizationException("Null protocol not authorized");
         }
         Class<?> protocol = null;
         try {
           protocol = getProtocolClass(protocolName, getConf());
         } catch (ClassNotFoundException cfne) {
           throw new AuthorizationException("Unknown protocol: " + 
                                            protocolName);
         }
         serviceAuthorizationManager.authorize(user, protocol, getConf(), addr);
       }
     }
   
   
   
   /**
      * Authorize the user to access the protocol being used.
      * 
      * @param user user accessing the service 
      * @param protocol service being accessed
      * @param conf configuration to use
      * @param addr InetAddress of the client
      * @throws AuthorizationException on authorization failure
      */
     public void authorize(UserGroupInformation user, 
                                  Class<?> protocol,
                                  Configuration conf,
                                  InetAddress addr
                                  ) throws AuthorizationException {
       AccessControlList[] acls = protocolToAcls.get(protocol);
       MachineList[] hosts = protocolToMachineLists.get(protocol);
       if (acls == null || hosts == null) {
         throw new AuthorizationException("Protocol " + protocol + 
                                          " is not known.");
       }
       
       // get client principal key to verify (if available)
       KerberosInfo krbInfo = SecurityUtil.getKerberosInfo(protocol, conf);
       String clientPrincipal = null; 
       if (krbInfo != null) {
         String clientKey = krbInfo.clientPrincipal();
         if (clientKey != null && !clientKey.isEmpty()) {
           try {
             clientPrincipal = SecurityUtil.getServerPrincipal(
                 conf.get(clientKey), addr);
           } catch (IOException e) {
             throw (AuthorizationException) new AuthorizationException(
                 "Can't figure out Kerberos principal name for connection from "
                     + addr + " for user=" + user + " protocol=" + protocol)
                 .initCause(e);
           }
         }
       }
       if((clientPrincipal != null && !clientPrincipal.equals(user.getUserName())) || 
          acls.length != 2  || !acls[0].isUserAllowed(user) || acls[1].isUserAllowed(user)) {
         AUDITLOG.warn(AUTHZ_FAILED_FOR + user + " for protocol=" + protocol
             + ", expected client Kerberos principal is " + clientPrincipal);
         throw new AuthorizationException("User " + user + 
             " is not authorized for protocol " + protocol + 
             ", expected client Kerberos principal is " + clientPrincipal);
       }
       if (addr != null) {
         String hostAddress = addr.getHostAddress();
         if (hosts.length != 2 || !hosts[0].includes(hostAddress) ||
             hosts[1].includes(hostAddress)) {
           AUDITLOG.warn(AUTHZ_FAILED_FOR + " for protocol=" + protocol
               + " from host = " +  hostAddress);
           throw new AuthorizationException("Host " + hostAddress +
               " is not authorized for protocol " + protocol) ;
         }
       }
       AUDITLOG.info(AUTHZ_SUCCESSFUL_FOR + user + " for protocol="+protocol);
     }

   ServiceAuthorizationManager維護了 協議與用戶、主機的映射關係，包括白名單和黑名單，支持動態刷新。