import backtype.storm.security.auth.AuthUtils; //导入依赖的package包/类
/**
 * Constructor based on a JAAS configuration
 * 
 * For digest, you should have a pair of user name and password defined.
 * 
 * @param configuration
 * @throws IOException
 */
public ClientCallbackHandler(Configuration configuration)
		throws IOException {
	if (configuration == null)
		return;
	AppConfigurationEntry configurationEntries[] = configuration
			.getAppConfigurationEntry(AuthUtils.LOGIN_CONTEXT_CLIENT);
	if (configurationEntries == null) {
		String errorMessage = "Could not find a '"
				+ AuthUtils.LOGIN_CONTEXT_CLIENT
				+ "' entry in this configuration: Client cannot start.";
		throw new IOException(errorMessage);
	}

	_password = "";
	for (AppConfigurationEntry entry : configurationEntries) {
		if (entry.getOptions().get(USERNAME) != null) {
			_username = (String) entry.getOptions().get(USERNAME);
		}
		if (entry.getOptions().get(PASSWORD) != null) {
			_password = (String) entry.getOptions().get(PASSWORD);
		}
	}
}
 

import backtype.storm.security.auth.AuthUtils; //导入依赖的package包/类
/**
 * Invoked once immediately after construction
 * 
 * @param conf Storm configuration
 */
@Override
public void prepare(Map conf) {
    _admins = new HashSet<String>();
    _supervisors = new HashSet<String>();
    _nimbusUsers = new HashSet<String>();
    _nimbusGroups = new HashSet<String>();

    if (conf.containsKey(Config.NIMBUS_ADMINS)) {
        _admins.addAll((Collection<String>) conf.get(Config.NIMBUS_ADMINS));
    }
    if (conf.containsKey(Config.NIMBUS_SUPERVISOR_USERS)) {
        _supervisors.addAll((Collection<String>) conf.get(Config.NIMBUS_SUPERVISOR_USERS));
    }
    if (conf.containsKey(Config.NIMBUS_USERS)) {
        _nimbusUsers.addAll((Collection<String>) conf.get(Config.NIMBUS_USERS));
    }

    if (conf.containsKey(Config.NIMBUS_GROUPS)) {
        _nimbusGroups.addAll((Collection<String>) conf.get(Config.NIMBUS_GROUPS));
    }

    _ptol = AuthUtils.GetPrincipalToLocalPlugin(conf);
    _groupMappingProvider = AuthUtils.GetGroupMappingServiceProviderPlugin(conf);
}
 

import backtype.storm.security.auth.AuthUtils; //导入依赖的package包/类
public ServerCallbackHandler(Configuration configuration) throws IOException {
    if (configuration == null)
        return;

    AppConfigurationEntry configurationEntries[] = configuration.getAppConfigurationEntry(AuthUtils.LOGIN_CONTEXT_SERVER);
    if (configurationEntries == null) {
        String errorMessage = "Could not find a '" + AuthUtils.LOGIN_CONTEXT_SERVER + "' entry in this configuration: Server cannot start.";
        throw new IOException(errorMessage);
    }
    credentials.clear();
    for (AppConfigurationEntry entry : configurationEntries) {
        Map<String, ?> options = entry.getOptions();
        // Populate DIGEST-MD5 user -> password map with JAAS configuration entries from the "Server" section.
        // Usernames are distinguished from other options by prefixing the username with a "user_" prefix.
        for (Map.Entry<String, ?> pair : options.entrySet()) {
            String key = pair.getKey();
            if (key.startsWith(USER_PREFIX)) {
                String userName = key.substring(USER_PREFIX.length());
                credentials.put(userName, (String) pair.getValue());
            }
        }
    }
}
 

import backtype.storm.security.auth.AuthUtils; //导入依赖的package包/类
/**
 * Constructor based on a JAAS configuration
 * 
 * For digest, you should have a pair of user name and password defined.
 * 
 * @param configuration
 * @throws IOException
 */
public ClientCallbackHandler(Configuration configuration) throws IOException {
    if (configuration == null)
        return;
    AppConfigurationEntry configurationEntries[] = configuration.getAppConfigurationEntry(AuthUtils.LOGIN_CONTEXT_CLIENT);
    if (configurationEntries == null) {
        String errorMessage = "Could not find a '" + AuthUtils.LOGIN_CONTEXT_CLIENT + "' entry in this configuration: Client cannot start.";
        throw new IOException(errorMessage);
    }

    _password = "";
    for (AppConfigurationEntry entry : configurationEntries) {
        if (entry.getOptions().get(USERNAME) != null) {
            _username = (String) entry.getOptions().get(USERNAME);
        }
        if (entry.getOptions().get(PASSWORD) != null) {
            _password = (String) entry.getOptions().get(PASSWORD);
        }
    }
}
 

import backtype.storm.security.auth.AuthUtils; //导入依赖的package包/类
public ServerCallbackHandler(Configuration configuration) throws IOException {
    if (configuration==null) return;

    AppConfigurationEntry configurationEntries[] = configuration.getAppConfigurationEntry(AuthUtils.LOGIN_CONTEXT_SERVER);
    if (configurationEntries == null) {
        String errorMessage = "Could not find a '"+AuthUtils.LOGIN_CONTEXT_SERVER+"' entry in this configuration: Server cannot start.";
        throw new IOException(errorMessage);
    }
    credentials.clear();
    for(AppConfigurationEntry entry: configurationEntries) {
        Map<String,?> options = entry.getOptions();
        // Populate DIGEST-MD5 user -> password map with JAAS configuration entries from the "Server" section.
        // Usernames are distinguished from other options by prefixing the username with a "user_" prefix.
        for(Map.Entry<String, ?> pair : options.entrySet()) {
            String key = pair.getKey();
            if (key.startsWith(USER_PREFIX)) {
                String userName = key.substring(USER_PREFIX.length());
                credentials.put(userName,(String)pair.getValue());
            }
        }
    }
}
 

import backtype.storm.security.auth.AuthUtils; //导入依赖的package包/类
/**
 * Constructor based on a JAAS configuration
 * 
 * For digest, you should have a pair of user name and password defined.
 * 
 * @param configuration
 * @throws IOException
 */
public ClientCallbackHandler(Configuration configuration) throws IOException {
    if (configuration == null) return;
    AppConfigurationEntry configurationEntries[] = configuration.getAppConfigurationEntry(AuthUtils.LOGIN_CONTEXT_CLIENT);
    if (configurationEntries == null) {
        String errorMessage = "Could not find a '"+AuthUtils.LOGIN_CONTEXT_CLIENT
                + "' entry in this configuration: Client cannot start.";
        throw new IOException(errorMessage);
    }

    _password = "";
    for(AppConfigurationEntry entry: configurationEntries) {
        if (entry.getOptions().get(USERNAME) != null) {
            _username = (String)entry.getOptions().get(USERNAME);
        }
        if (entry.getOptions().get(PASSWORD) != null) {
            _password = (String)entry.getOptions().get(PASSWORD);
        }
    }
}
 

import backtype.storm.security.auth.AuthUtils; //导入依赖的package包/类
public ServerCallbackHandler(Configuration configuration)
		throws IOException {
	if (configuration == null)
		return;

	AppConfigurationEntry configurationEntries[] = configuration
			.getAppConfigurationEntry(AuthUtils.LOGIN_CONTEXT_SERVER);
	if (configurationEntries == null) {
		String errorMessage = "Could not find a '"
				+ AuthUtils.LOGIN_CONTEXT_SERVER
				+ "' entry in this configuration: Server cannot start.";
		throw new IOException(errorMessage);
	}
	credentials.clear();
	for (AppConfigurationEntry entry : configurationEntries) {
		Map<String, ?> options = entry.getOptions();
		// Populate DIGEST-MD5 user -> password map with JAAS configuration
		// entries from the "Server" section.
		// Usernames are distinguished from other options by prefixing the
		// username with a "user_" prefix.
		for (Map.Entry<String, ?> pair : options.entrySet()) {
			String key = pair.getKey();
			if (key.startsWith(USER_PREFIX)) {
				String userName = key.substring(USER_PREFIX.length());
				credentials.put(userName, (String) pair.getValue());
			}
		}
	}
}
 

import backtype.storm.security.auth.AuthUtils; //导入依赖的package包/类
@Override
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 backtype.storm.security.auth.AuthUtils; //导入依赖的package包/类
@Override
public TTransport connect(TTransport transport, String serverHost)
		throws TTransportException, IOException {
	ClientCallbackHandler client_callback_handler = new ClientCallbackHandler(
			login_conf);
	TSaslClientTransport wrapper_transport = new TSaslClientTransport(
			DIGEST, null, AuthUtils.SERVICE, serverHost, null,
			client_callback_handler, transport);

	wrapper_transport.open();
	LOG.debug("SASL DIGEST-MD5 client transport has been established");

	return wrapper_transport;
}
 

import backtype.storm.security.auth.AuthUtils; //导入依赖的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 backtype.storm.security.auth.AuthUtils; //导入依赖的package包/类
public TTransport connect(TTransport transport, String serverHost)
		throws TTransportException, IOException {
	ClientCallbackHandler client_callback_handler = new ClientCallbackHandler(
			login_conf);
	TSaslClientTransport wrapper_transport = new TSaslClientTransport(
			DIGEST, null, AuthUtils.SERVICE, serverHost, null,
			client_callback_handler, transport);

	wrapper_transport.open();
	LOG.debug("SASL DIGEST-MD5 client transport has been established");

	return wrapper_transport;
}
 

import backtype.storm.security.auth.AuthUtils; //导入依赖的package包/类
@Override
public void prepare(Map conf) {
    Boolean isStrict = (Boolean) conf.get(Config.DRPC_AUTHORIZER_ACL_STRICT);
    _permitWhenMissingFunctionEntry = (isStrict != null && !isStrict) ? true : false;
    _aclFileName = (String) conf.get(Config.DRPC_AUTHORIZER_ACL_FILENAME);
    _ptol = AuthUtils.GetPrincipalToLocalPlugin(conf);
}
 

import backtype.storm.security.auth.AuthUtils; //导入依赖的package包/类
public ServerCallbackHandler(Configuration configuration, Map stormConf) throws IOException {
    if (configuration == null)
        return;

    AppConfigurationEntry configurationEntries[] = configuration.getAppConfigurationEntry(AuthUtils.LOGIN_CONTEXT_SERVER);
    if (configurationEntries == null) {
        String errorMessage = "Could not find a '" + AuthUtils.LOGIN_CONTEXT_SERVER + "' entry in this configuration: Server cannot start.";
        LOG.error(errorMessage);
        throw new IOException(errorMessage);
    }

}
 

import backtype.storm.security.auth.AuthUtils; //导入依赖的package包/类
@Override
public void populateCredentials(Map<String, String> credentials) {
    // Log the user in and get the TGT
    try {
        Configuration login_conf = AuthUtils.GetConfiguration(conf);
        ClientCallbackHandler client_callback_handler = new ClientCallbackHandler(login_conf);

        // login our user
        Configuration.setConfiguration(login_conf);
        LoginContext lc = new LoginContext(AuthUtils.LOGIN_CONTEXT_CLIENT, client_callback_handler);
        try {
            lc.login();
            final Subject subject = lc.getSubject();
            KerberosTicket tgt = getTGT(subject);

            if (tgt == null) { // error
                throw new RuntimeException("Fail to verify user principal with section \"" + AuthUtils.LOGIN_CONTEXT_CLIENT
                        + "\" in login configuration file " + login_conf);
            }

            if (!tgt.isForwardable()) {
                throw new RuntimeException("The TGT found is not forwardable");
            }

            if (!tgt.isRenewable()) {
                throw new RuntimeException("The TGT found is not renewable");
            }

            LOG.info("Pushing TGT for " + tgt.getClient() + " to topology.");
            saveTGT(tgt, credentials);
        } finally {
            lc.logout();
        }
    } catch (Exception e) {
        throw new RuntimeException(e);
    }
}
 

import backtype.storm.security.auth.AuthUtils; //导入依赖的package包/类
/**
 * Constructor based on a JAAS configuration
 * 
 * For digest, you should have a pair of user name and password defined in this figgure.
 * 
 * @param configuration
 * @throws IOException
 */
public ClientCallbackHandler(Configuration configuration) throws IOException {
    if (configuration == null)
        return;
    AppConfigurationEntry configurationEntries[] = configuration.getAppConfigurationEntry(AuthUtils.LOGIN_CONTEXT_CLIENT);
    if (configurationEntries == null) {
        String errorMessage = "Could not find a '" + AuthUtils.LOGIN_CONTEXT_CLIENT + "' entry in this configuration: Client cannot start.";
        LOG.error(errorMessage);
        throw new IOException(errorMessage);
    }
}
 

import backtype.storm.security.auth.AuthUtils; //导入依赖的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 backtype.storm.security.auth.AuthUtils; //导入依赖的package包/类
@Override
public TTransport connect(TTransport transport, String serverHost, String asUser) throws TTransportException, IOException {
    ClientCallbackHandler client_callback_handler = new ClientCallbackHandler(login_conf);
    TSaslClientTransport wrapper_transport =
            new TSaslClientTransport(DIGEST, null, AuthUtils.SERVICE, serverHost, null, client_callback_handler, transport);

    wrapper_transport.open();
    LOG.debug("SASL DIGEST-MD5 client transport has been established");

    return wrapper_transport;
}
 

import backtype.storm.security.auth.AuthUtils; //导入依赖的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 backtype.storm.security.auth.AuthUtils; //导入依赖的package包/类
public TTransport connect(TTransport transport, String serverHost) throws TTransportException, IOException {
    ClientCallbackHandler client_callback_handler = new ClientCallbackHandler(login_conf);
    TSaslClientTransport wrapper_transport = new TSaslClientTransport(DIGEST, 
            null, 
            AuthUtils.SERVICE, 
            serverHost,
            null,
            client_callback_handler, 
            transport);

    wrapper_transport.open();
    LOG.debug("SASL DIGEST-MD5 client transport has been established");

    return wrapper_transport;
}
 

import backtype.storm.security.auth.AuthUtils; //导入依赖的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;
}