我正在尝试使用代理连接器将代理的流量重定向到另一个代理。我需要在我的场景中通过证书处理身份验证。当我使用 type 的绑定 uri 时nio+ssl://192.168.1.5:61619，通信是完美的。当我使用 type 的绑定 uri 时nio+ssl://192.168.1.5:61619?needClientAuth=true，客户端无法启动连接。
我错过了什么吗？
这是我的代理的代码，由两个类组成。
ActiveMQBrokerActivator.java：
public class ActiveMQBrokerActivator {
private static BrokerService broker;
private static SslContext customSslContext;
private final static String certificatesFolder = "config" + File.separator + "certificates" + File.separator +
"amq" + File.separator;
private final static String broker_ks_file = certificatesFolder + "broker.ks";
private final static String broker_ts_file = certificatesFolder + "broker.ts";
private static String trustorepswd = "MyStrongPwd"; //TODO: change with your password
private static String keystorepswd = trustorepswd;
private static int jmxConnectorPort = 1098;
private static String jmxDomainName = "proxybroker";
private String proxyBrokerAddr = "10.0.3.41"; //TODO: change with your local IP
private String remoteBrokerAddr = "10.0.3.13"; //TODO: change with IP of the remote broker
private String[] proxyNames = new String[] { "proxy_nio", "proxy_nio_ssl", "proxy_tcp", "proxy_tcp_ssl" };
private int[] proxyBrokerPorts = new int[] { 61716, 61719, 61726, 61729 };
private String[] proxyParams = new String[] { "", "?needClientAuth=true", "", "?needClientAuth=true" };
//private String[] proxyParams = new String[] { "", "", "", "" };
private String[] protocols = new String[] { "nio", "nio+ssl", "tcp", "ssl" };
private int[] remoteBrokerPorts = new int[] { 61616, 61619, 61626, 61629 };
public static void main(String[] args) throws Exception {
ActiveMQBrokerActivator activator = new ActiveMQBrokerActivator();
activator.init();
String input = "";
while(!input.equalsIgnoreCase("quit")) {
System.out.print("Type 'reload' to reload certificates; 'quit' to exit");
BufferedReader br = new BufferedReader(new InputStreamReader(System.in));
input = br.readLine();
if (input.equalsIgnoreCase("reload"))
ActiveMQBrokerActivator.updateSSLcontext();
}
}
public void init() throws Exception {
System.out.println("Starting ActiveMQ Broker...");
//setup broker:
Properties properties = System.getProperties();
broker = new BrokerService();
broker.setUseJmx(true);
broker.setBrokerName(jmxDomainName);
//set JMX connector port and domain name in order not to retrieve the correct info
broker.getManagementContext().setConnectorPort(jmxConnectorPort);
broker.getManagementContext().setJmxDomainName(jmxDomainName);
//set key-store and trust-store properties:
properties.put("javax.net.ssl.keyStore", broker_ks_file);
properties.put("javax.net.ssl.keyStorePassword", keystorepswd);
properties.put("javax.net.ssl.trustStore", broker_ts_file);
properties.put("javax.net.ssl.trustStorePassword", trustorepswd);
System.out.println(" - keystore is " + properties.getProperty("javax.net.ssl.keyStore"));
System.out.println(" - truststore is " + properties.getProperty("javax.net.ssl.trustStore"));
System.out.println("Setting custom SSLContext for dynamic cert loading...");
customSslContext = getSslContext();
broker.setSslContext(customSslContext);
System.out.println("...done!");
//Add proxy connectors to the broker:
for (int i = 0; i < 4; i++) {
addProxyConn(protocols[i] + "://" + proxyBrokerAddr + ":" + proxyBrokerPorts[i] + proxyParams[i],
proxyNames[i], protocols[i] + "://" + remoteBrokerAddr + ":" + remoteBrokerPorts[i]);
}
tuning();
// broker must be activated here if we want to add plugins...
broker.start();
System.out.println("Broker AMQ is starting...");
broker.waitUntilStarted();
while(!broker.isStarted());
System.out.println("Broker AMQ started: " + broker.getAdminView().getVMURL() +
" broker id: " + broker.getAdminView().getBrokerId() + " v. " + broker.getAdminView().getBrokerVersion());
/* TimeStampingBrokerPlugin handles the possible miss alignments among broker and producers
* timestamps. Besides, it overides the zero-TTL specified by the producers
*/
TimeStampingBrokerPlugin tsbp = new TimeStampingBrokerPlugin();
tsbp.setZeroExpirationOverride(200000);
tsbp.installPlugin(broker.getBroker());
tsbp.start();
System.out.println("Broker [" + broker.getBrokerName() + "] has correctly started");
}
private static void addProxyConn(String bindUri, String name, String remoteUri) throws Exception {
System.out.println("Adding AMQ proxy connector...");
ProxyConnector proxy_conn = new ProxyConnector();
proxy_conn.setBind(new URI(bindUri));
proxy_conn.setName(name);
proxy_conn.setProxyToLocalBroker(false);
proxy_conn.setRemote(new URI(remoteUri));
System.out.println(" AMQ proxy connector " + proxy_conn.getName() + " has been added successfully.");
broker.addProxyConnector(proxy_conn);
}
private static synchronized void updateSSLcontext(){
try {
ReloadableX509TrustManager tm = (ReloadableX509TrustManager)customSslContext.getTrustManagersAsArray()[0];
tm.reloadTrustManager();
}catch(Exception e){
System.err.println("Error while reloading trust manager due to " + e.toString());
}
}
private SslContext getSslContext() throws Exception {
KeyManagerFactory kmf =
KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
KeyStore ks = KeyStore.getInstance("jks");
KeyManager[] keystoreManagers = null;
ks.load(new FileInputStream(new File(broker_ks_file)), keystorepswd.toCharArray());
kmf.init(ks, keystorepswd.toCharArray());
keystoreManagers = kmf.getKeyManagers();
TrustManager[] trustStoreManagers = new TrustManager[] {
new ReloadableX509TrustManager(broker_ts_file, trustorepswd)
};
SslContext context = new SslContext(keystoreManagers, trustStoreManagers, null);
return context;
}
private void tuning() {
/* We want to delete destinations that are inactive for a period of time. Since ActiveMQ version 5.4.0,
* it's possible to do that using destination policy entries and broker attribute
* schedulePeriodForDestinationPurge > 0
*/
broker.setSchedulePeriodForDestinationPurge(120000); //2 minutes
/* This is to set how the broker should dispatch outgoing messages:
* At the time being, PolicyEntry items do not aggregate (see at
* http://activemq.2283324.n4.nabble.com/Do-policy-map-entries-aggregate-tt4297601.html#a4300231)
* There is a best match, and the default takes the unmatched case: all topics starting with
* "stream." will use the stream_entry policy while all other topics will default on std_entry
* policy.
*/
PolicyMap map = new PolicyMap();
PolicyEntry std_entry = new PolicyEntry();
PolicyEntry stream_entry = new PolicyEntry();
// All topics:
std_entry.setTopic(">");
stream_entry.setTopic("stream.>");
std_entry.setDispatchPolicy(new StrictOrderDispatchPolicy());
std_entry.setOptimizedDispatch(true);
stream_entry.setOptimizedDispatch(true);
final long EXPIRE_MSG_PERIOD = 200000; //200 seconds
final long STREAM_EXPIRE_MSG_PERIOD = 3000; //3 seconds
/* Sets the strategy to calculate the maximum number of messages that are allowed
* to be pending on consumers (in addition to their prefetch sizes).
* Once the limit is reached, non-durable topics can then start discarding old
* messages. This allows us to keep dispatching messages to slow consumers while
* not blocking fast consumers and discarding the messages oldest first.
*/
final long PENDING_MSG_LIMIT = 200000; //after this, start check TTL
ConstantPendingMessageLimitStrategy pendMsgStrategy = new ConstantPendingMessageLimitStrategy();
pendMsgStrategy.setLimit((int)PENDING_MSG_LIMIT);
std_entry.setPendingMessageLimitStrategy(pendMsgStrategy);
final long STREAM_PENDING_MSG_LIMIT = 200000; //after this, start check TTL
ConstantPendingMessageLimitStrategy streamPendMsgStrategy = new ConstantPendingMessageLimitStrategy();
streamPendMsgStrategy.setLimit((int)STREAM_PENDING_MSG_LIMIT);
stream_entry.setPendingMessageLimitStrategy(streamPendMsgStrategy);
/* See: http://activemq.apache.org/producer-flow-control.html
* With producer flow control disabled, messages for slow consumers will be off-lined to
* temporary storage by default, enabling the producers and the rest of the consumers to
* run at a much faster rate:
*/
std_entry.setProducerFlowControl(false);
/* See http://activemq.apache.org/manage-durable-subscribers.html
* Some applications send message with specified time to live. If those messages are kept on
* the broker for the offline durable subscriber we need to remove them when they reach their
* expiry time. Just as AMQ does with queues, now AMQ checks for those messages every
* EXPIRE_MSG_PERIOD.
* This configuration complements the Timestampplugin (http://activemq.apache.org/timestampplugin.html)
*/
std_entry.setExpireMessagesPeriod((int)EXPIRE_MSG_PERIOD);
stream_entry.setProducerFlowControl(false);
stream_entry.setExpireMessagesPeriod((int)STREAM_EXPIRE_MSG_PERIOD);
/* This will check for inactive destination and it will delete all queues (gcInactiveDestinations option)
* if they are empty for "OfflineDurableSubscriberTimeout" millis
*/
std_entry.setGcInactiveDestinations(true);
std_entry.setInactiveTimoutBeforeGC(1200000);
stream_entry.setGcInactiveDestinations(true);
stream_entry.setInactiveTimoutBeforeGC(1200000);
map.setDefaultEntry(std_entry);
LinkedList<PolicyEntry> policies = new LinkedList<PolicyEntry>();
policies.add(stream_entry);
map.setPolicyEntries(policies);
broker.setDestinationPolicy(map);
// Set memory manager: refer to the following links for further documentation:
// - http://activemq.2283324.n4.nabble.com/StoreUsage-TempUsage-and-MemoryUsage-td2356734.html
// And also (NOT OFFIAL but rather clear explanations):
// - http://tmielke.blogspot.it/2011/02/observations-on-activemqs-temp-storage.html
// - http://java.dzone.com/articles/activemq-understanding-memory
// Default values (see http://activemq.apache.org/producer-flow-control.html#ProducerFlowControl-Systemusage):
// - Default Memory limit is 64 mb
// - Default Temp Usage limit is 100 gb
// - Default Store Usage limit is 10 gb
SystemUsage usage = broker.getSystemUsage();
long memLimit = 1024L * 1024L,
tempLimit = 1024L * 1024L,
storeLimit = 1024L * 1024L;
memLimit *= 64;
tempLimit *= 10000;
storeLimit *= 50000;
MemoryUsage memUsage = usage.getMemoryUsage();
memUsage.setLimit(memLimit);
usage.setMemoryUsage(memUsage);
TempUsage tmpUsage = usage.getTempUsage();
tmpUsage.setLimit(tempLimit);
usage.setTempUsage(tmpUsage);
StoreUsage storeUsage = usage.getStoreUsage();
storeUsage.setLimit(storeLimit);
usage.setStoreUsage(storeUsage);
boolean sendFailIfNoSpace=false;
if(sendFailIfNoSpace)
usage.setSendFailIfNoSpace(sendFailIfNoSpace);
broker.setSystemUsage(usage);
manageAdditionalPolicies();
}
private void manageAdditionalPolicies() {
PolicyEntry std_entry = new PolicyEntry();
std_entry.setTopic(">");
std_entry.setEnableAudit(false);
ConditionalNetworkBridgeFilterFactory bff=new ConditionalNetworkBridgeFilterFactory();
bff.setReplayWhenNoConsumers(true);
std_entry.setNetworkBridgeFilterFactory(bff);
LinkedList<PolicyEntry> policies = new LinkedList<PolicyEntry>();
policies.add(std_entry);
broker.getDestinationPolicy().setPolicyEntries(policies);
}
protected static int getJmxConnectorPort() {
return jmxConnectorPort;
}
protected static String getJmxDomainName() {
return jmxDomainName;
}
}
ReloadableX509TrustManager.java：
public class ReloadableX509TrustManager implements X509TrustManager {
private final String trustStorePath;
private final String tspassword;
private X509TrustManager trustManager;
public ReloadableX509TrustManager(String tspath, String tspassword) throws Exception {
this.trustStorePath = tspath;
this.tspassword = tspassword;
reloadTrustManager();
}
@Override
public void checkClientTrusted(X509Certificate[] chain, String authType) throws CertificateException {
try{
trustManager.checkClientTrusted(chain, authType);
}catch(Exception e){
try{
reloadTrustManager();
}catch(Exception ex){
throw new CertificateException(ex);
}
trustManager.checkClientTrusted(chain, authType);
}
}
@Override
public void checkServerTrusted(X509Certificate[] chain, String authType) throws CertificateException {
try {
trustManager.checkServerTrusted(chain, authType);
} catch (CertificateException cx) {
try{
reloadTrustManager();
}catch(Exception e){
throw new CertificateException(e);
}
trustManager.checkServerTrusted(chain, authType);
}
}
@Override
public X509Certificate[] getAcceptedIssuers() {
X509Certificate[] issuers = trustManager.getAcceptedIssuers();
return issuers;
}
public void reloadTrustManager() throws Exception {
// load keystore from specified cert store (or default)
KeyStore ts = KeyStore.getInstance(KeyStore.getDefaultType());
InputStream in = new FileInputStream(trustStorePath);
try {
ts.load(in, null);
}catch(Exception e){
e.printStackTrace();
}finally {
in.close();
}
// initialize a new TMF with the ts we just loaded
TrustManagerFactory tmf = TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm());
tmf.init(ts);
// acquire X509 trust manager from factory
TrustManager tms[] = tmf.getTrustManagers();
for (int i = 0; i < tms.length; i++) {
if (tms[i] instanceof X509TrustManager) {
trustManager = (X509TrustManager)tms[i];
return;
}
}
throw new NoSuchAlgorithmException("No X509TrustManager in TrustManagerFactory");
}
protected void addServerCertAndReload(X509Certificate cert) {
try {
// import the cert into file trust store
File tsfile = new File(this.trustStorePath);
java.io.FileInputStream fis = null;
if(tsfile.exists()){
fis = new FileInputStream(tsfile);
}else{
System.err.println("Truststore " + tsfile.getAbsolutePath() + " does not exist!");
throw new Exception("Truststore " +tsfile.getAbsolutePath() + " does not exist!");
}
KeyStore ts = KeyStore.getInstance(KeyStore.getDefaultType());
char[] keystorePass = this.tspassword.toCharArray();
try {
ts.load(fis, keystorePass);
}catch(Exception e){
e.printStackTrace();
}finally {
fis.close();
}
ts.setCertificateEntry("", cert);
ts.store(new FileOutputStream(this.trustStorePath), keystorePass);
reloadTrustManager();
} catch (Exception ex) {
ex.printStackTrace();
}
}
}