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// authenticationif (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维护了 协议与用户、主机的映射关系，包括白名单和黑名单，支持动态刷新。