1 files changed, 150 insertions, 52 deletions

diff --git a/cpp/src/qpid/broker/ConnectionHandler.cpp b/cpp/src/qpid/broker/ConnectionHandler.cpp
index 77a4d1a3de..50a5aff2c9 100644
--- a/cpp/src/qpid/broker/ConnectionHandler.cpp
+++ b/cpp/src/qpid/broker/ConnectionHandler.cpp
@@ -8,9 +8,9 @@
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
- * 
+ *
  *   http://www.apache.org/licenses/LICENSE-2.0
- * 
+ *
  * Unless required by applicable law or agreed to in writing,
  * software distributed under the License is distributed on an
  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
@@ -20,70 +20,91 @@
  *
  */
 
-#include "config.h"
-
-#include "ConnectionHandler.h"
-#include "Connection.h"
-#include "qpid/framing/ClientInvoker.h"
-#include "qpid/framing/ServerInvoker.h"
-#include "qpid/framing/constants.h"
+#include "qpid/broker/ConnectionHandler.h"
+#include "qpid/broker/Connection.h"
+#include "qpid/broker/SecureConnection.h"
+#include "qpid/Url.h"
+#include "qpid/framing/AllInvoker.h"
+#include "qpid/framing/enum.h"
 #include "qpid/log/Statement.h"
+#include "qpid/sys/SecurityLayer.h"
+#include "qpid/broker/AclModule.h"
+#include "qmf/org/apache/qpid/broker/EventClientConnectFail.h"
 
 using namespace qpid;
 using namespace qpid::broker;
 using namespace qpid::framing;
+using qpid::sys::SecurityLayer;
+namespace _qmf = qmf::org::apache::qpid::broker;
 
-
-namespace 
+namespace
 {
 const std::string ANONYMOUS = "ANONYMOUS";
 const std::string PLAIN     = "PLAIN";
 const std::string en_US     = "en_US";
+const std::string QPID_FED_LINK = "qpid.fed_link";
+const std::string QPID_FED_TAG  = "qpid.federation_tag";
+const std::string SESSION_FLOW_CONTROL("qpid.session_flow");
+const std::string CLIENT_PROCESS_NAME("qpid.client_process");
+const std::string CLIENT_PID("qpid.client_pid");
+const std::string CLIENT_PPID("qpid.client_ppid");
+const int SESSION_FLOW_CONTROL_VER = 1;
 }
 
-void ConnectionHandler::close(ReplyCode code, const string& text, ClassId, MethodId)
+void ConnectionHandler::close(connection::CloseCode code, const string& text)
 {
-    handler->client.close(code, text);
+    handler->proxy.close(code, text);
+}
+
+void ConnectionHandler::heartbeat()
+{
+    handler->proxy.heartbeat();
 }
 
 void ConnectionHandler::handle(framing::AMQFrame& frame)
 {
     AMQMethodBody* method=frame.getBody()->getMethod();
+    Connection::ErrorListener* errorListener = handler->connection.getErrorListener();
     try{
-        bool handled = false;
-        if (handler->serverMode) {
-            handled = invoke(static_cast<AMQP_ServerOperations::ConnectionHandler&>(*handler.get()), *method);
-        } else {
-            handled = invoke(static_cast<AMQP_ClientOperations::ConnectionHandler&>(*handler.get()), *method);
-        }
-        if (!handled) {
+        if (!invoke(static_cast<AMQP_AllOperations::ConnectionHandler&>(*handler.get()), *method)) {
             handler->connection.getChannel(frame.getChannel()).in(frame);
         }
-
     }catch(ConnectionException& e){
-        handler->client.close(e.code, e.what());
+        if (errorListener) errorListener->connectionError(e.what());
+        handler->proxy.close(e.code, e.what());
     }catch(std::exception& e){
-        handler->client.close(541/*internal error*/, e.what());
+        if (errorListener) errorListener->connectionError(e.what());
+        handler->proxy.close(541/*internal error*/, e.what());
     }
 }
 
+void ConnectionHandler::setSecureConnection(SecureConnection* secured)
+{
+    handler->secured = secured;
+}
+
 ConnectionHandler::ConnectionHandler(Connection& connection, bool isClient)  : handler(new Handler(connection, isClient)) {}
 
 ConnectionHandler::Handler::Handler(Connection& c, bool isClient) :
-    client(c.getOutput()), server(c.getOutput()), 
-    connection(c), serverMode(!isClient)
+    proxy(c.getOutput()),
+    connection(c), serverMode(!isClient), acl(0), secured(0)
 {
     if (serverMode) {
+
+    	acl =  connection.getBroker().getAcl();
+
         FieldTable properties;
         Array mechanisms(0x95);
-        
+
+        properties.setString(QPID_FED_TAG, connection.getBroker().getFederationTag());
+
         authenticator = SaslAuthenticator::createAuthenticator(c);
         authenticator->getMechanisms(mechanisms);
-        
+
         Array locales(0x95);
         boost::shared_ptr<FieldValue> l(new Str16Value(en_US));
         locales.add(l);
-        client.start(properties, mechanisms, locales);
+        proxy.start(properties, mechanisms, locales);
     }
 }
 
@@ -91,65 +112,138 @@ ConnectionHandler::Handler::Handler(Connection& c, bool isClient) :
 ConnectionHandler::Handler::~Handler() {}
 
 
-void ConnectionHandler::Handler::startOk(const framing::FieldTable& /*clientProperties*/,
-                                         const string& mechanism, 
+void ConnectionHandler::Handler::startOk(const framing::FieldTable& clientProperties,
+                                         const string& mechanism,
                                          const string& response,
                                          const string& /*locale*/)
 {
-    authenticator->start(mechanism, response);
+    try {
+        authenticator->start(mechanism, response);
+    } catch (std::exception& /*e*/) {
+        management::ManagementAgent* agent = connection.getAgent();
+        if (agent) {
+            string error;
+            string uid;
+            authenticator->getError(error);
+            authenticator->getUid(uid);
+            agent->raiseEvent(_qmf::EventClientConnectFail(connection.getMgmtId(), uid, error));
+        }
+        throw;
+    }
+    connection.setFederationLink(clientProperties.get(QPID_FED_LINK));
+    connection.setFederationPeerTag(clientProperties.getAsString(QPID_FED_TAG));
+    if (connection.isFederationLink()) {
+    	if (acl && !acl->authorise(connection.getUserId(),acl::ACT_CREATE,acl::OBJ_LINK,"")){
+            proxy.close(framing::connection::CLOSE_CODE_CONNECTION_FORCED,"ACL denied creating a federation link");
+            return;
+        }
+        QPID_LOG(info, "Connection is a federation link");
+    }
+    if (clientProperties.getAsInt(SESSION_FLOW_CONTROL) == SESSION_FLOW_CONTROL_VER) {
+        connection.setClientThrottling();
+    }
+
+    if (connection.getMgmtObject() != 0) {
+        string procName = clientProperties.getAsString(CLIENT_PROCESS_NAME);
+        uint32_t pid = clientProperties.getAsInt(CLIENT_PID);
+        uint32_t ppid = clientProperties.getAsInt(CLIENT_PPID);
+
+        if (!procName.empty())
+            connection.getMgmtObject()->set_remoteProcessName(procName);
+        if (pid != 0)
+            connection.getMgmtObject()->set_remotePid(pid);
+        if (ppid != 0)
+            connection.getMgmtObject()->set_remoteParentPid(ppid);
+    }
 }
-        
+
 void ConnectionHandler::Handler::secureOk(const string& response)
 {
-    authenticator->step(response);
+    try {
+        authenticator->step(response);
+    } catch (std::exception& /*e*/) {
+        management::ManagementAgent* agent = connection.getAgent();
+        if (agent) {
+            string error;
+            string uid;
+            authenticator->getError(error);
+            authenticator->getUid(uid);
+            agent->raiseEvent(_qmf::EventClientConnectFail(connection.getMgmtId(), uid, error));
+        }
+        throw;
+    }
 }
-        
+
 void ConnectionHandler::Handler::tuneOk(uint16_t /*channelmax*/,
     uint16_t framemax, uint16_t heartbeat)
 {
     connection.setFrameMax(framemax);
-    connection.setHeartbeat(heartbeat);
+    connection.setHeartbeatInterval(heartbeat);
 }
-        
+
 void ConnectionHandler::Handler::open(const string& /*virtualHost*/,
                                       const framing::Array& /*capabilities*/, bool /*insist*/)
 {
-    framing::Array knownhosts;
-    client.openOk(knownhosts);
+    std::vector<Url> urls = connection.broker.getKnownBrokers();
+    framing::Array array(0x95); // str16 array
+    for (std::vector<Url>::iterator i = urls.begin(); i < urls.end(); ++i)
+        array.add(boost::shared_ptr<Str16Value>(new Str16Value(i->str())));
+    proxy.openOk(array);
+
+    //install security layer if one has been negotiated:
+    if (secured) {
+        std::auto_ptr<SecurityLayer> sl = authenticator->getSecurityLayer(connection.getFrameMax());
+        if (sl.get()) secured->activateSecurityLayer(sl);
+    }
 }
 
-        
+
 void ConnectionHandler::Handler::close(uint16_t replyCode, const string& replyText)
 {
     if (replyCode != 200) {
         QPID_LOG(warning, "Client closed connection with " << replyCode << ": " << replyText);
     }
 
-    if (replyCode == framing::connection::CONNECTION_FORCED)
+    if (replyCode == framing::connection::CLOSE_CODE_CONNECTION_FORCED)
         connection.notifyConnectionForced(replyText);
 
-    client.closeOk();
+    proxy.closeOk();
     connection.getOutput().close();
-} 
-        
+}
+
 void ConnectionHandler::Handler::closeOk(){
     connection.getOutput().close();
-} 
+}
 
+void ConnectionHandler::Handler::heartbeat(){
+    // For general case, do nothing - the purpose of heartbeats is
+    // just to make sure that there is some traffic on the connection
+    // within the heart beat interval, we check for the traffic and
+    // don't need to do anything in response to heartbeats.  The
+    // exception is when we are in fact the client to another broker
+    // (i.e. an inter-broker link), in which case we echo the
+    // heartbeat back to the peer
+    if (!serverMode) proxy.heartbeat();
+}
 
-void ConnectionHandler::Handler::start(const FieldTable& /*serverProperties*/,
+void ConnectionHandler::Handler::start(const FieldTable& serverProperties,
                                        const framing::Array& /*mechanisms*/,
                                        const framing::Array& /*locales*/)
 {
     string mechanism = connection.getAuthMechanism();
     string response  = connection.getAuthCredentials();
-    
-    server.startOk(FieldTable(), mechanism, response, en_US);
+
+    connection.setFederationPeerTag(serverProperties.getAsString(QPID_FED_TAG));
+
+    FieldTable ft;
+    ft.setInt(QPID_FED_LINK,1);
+    ft.setString(QPID_FED_TAG, connection.getBroker().getFederationTag());
+    proxy.startOk(ft, mechanism, response, en_US);
 }
 
 void ConnectionHandler::Handler::secure(const string& /*challenge*/)
 {
-    server.secureOk("");
+    proxy.secureOk("");
 }
 
 void ConnectionHandler::Handler::tune(uint16_t channelMax,
@@ -159,15 +253,19 @@ void ConnectionHandler::Handler::tune(uint16_t channelMax,
 {
     connection.setFrameMax(frameMax);
     connection.setHeartbeat(heartbeatMax);
-    server.tuneOk(channelMax, frameMax, heartbeatMax);
-    server.open("/", Array(), true);
+    proxy.tuneOk(channelMax, frameMax, heartbeatMax);
+    proxy.open("/", Array(), true);
 }
 
-void ConnectionHandler::Handler::openOk(const framing::Array& /*knownHosts*/)
+void ConnectionHandler::Handler::openOk(const framing::Array& knownHosts)
 {
+    for (Array::ValueVector::const_iterator i = knownHosts.begin(); i != knownHosts.end(); ++i) {
+        Url url((*i)->get<std::string>());
+        connection.getKnownHosts().push_back(url);
+    }
 }
 
 void ConnectionHandler::Handler::redirect(const string& /*host*/, const framing::Array& /*knownHosts*/)
 {
-    
+
 }