This adds the user_id checking described in QPID-943.

If a user_id is supplied in message properties it will be checked against the id used for authentication.
This check is disabled for federation links.


git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk@699047 13f79535-47bb-0310-9956-ffa450edef68

6 files changed, 54 insertions, 20 deletions

diff --git a/qpid/cpp/src/qpid/broker/Connection.cpp b/qpid/cpp/src/qpid/broker/Connection.cpp
index 17f58adc78..320efc42f1 100644
--- a/qpid/cpp/src/qpid/broker/Connection.cpp
+++ b/qpid/cpp/src/qpid/broker/Connection.cpp
@@ -157,6 +157,13 @@ void Connection::setUserId(const string& userId)
         mgmtObject->set_authIdentity(userId);
 }
 
+void Connection::setFederationLink(bool b)
+{
+    ConnectionState::setFederationLink(b);
+    if (mgmtObject != 0)
+            mgmtObject->set_federationLink(b);
+}
+
 void Connection::close(
     ReplyCode code, const string& text, ClassId classId, MethodId methodId)
 {
diff --git a/qpid/cpp/src/qpid/broker/Connection.h b/qpid/cpp/src/qpid/broker/Connection.h
index cf0b4bc5c0..a38c89156e 100644
--- a/qpid/cpp/src/qpid/broker/Connection.h
+++ b/qpid/cpp/src/qpid/broker/Connection.h
@@ -96,6 +96,7 @@ class Connection : public sys::ConnectionInputHandler,
     std::string getAuthCredentials();
     void notifyConnectionForced(const std::string& text);
     void setUserId(const string& uid);
+    void setFederationLink(bool b);
 
     template <class F> void eachSessionHandler(const F& f) {
         for (ChannelMap::iterator i = channels.begin(); i != channels.end(); ++i)
diff --git a/qpid/cpp/src/qpid/broker/ConnectionHandler.cpp b/qpid/cpp/src/qpid/broker/ConnectionHandler.cpp
index 4feda3950b..7281fb53c6 100644
--- a/qpid/cpp/src/qpid/broker/ConnectionHandler.cpp
+++ b/qpid/cpp/src/qpid/broker/ConnectionHandler.cpp
@@ -34,11 +34,12 @@ using namespace qpid::broker;
 using namespace qpid::framing;
 
 
-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";
 }
 
 void ConnectionHandler::close(ReplyCode code, const string& text, ClassId, MethodId)
@@ -70,16 +71,16 @@ void ConnectionHandler::handle(framing::AMQFrame& frame)
 ConnectionHandler::ConnectionHandler(Connection& connection, bool isClient)  : handler(new Handler(connection, isClient)) {}
 
 ConnectionHandler::Handler::Handler(Connection& c, bool isClient) :
-    client(c.getOutput()), server(c.getOutput()), 
+    client(c.getOutput()), server(c.getOutput()),
     connection(c), serverMode(!isClient)
 {
     if (serverMode) {
         FieldTable properties;
         Array mechanisms(0x95);
-        
+
         authenticator = SaslAuthenticator::createAuthenticator(c);
         authenticator->getMechanisms(mechanisms);
-        
+
         Array locales(0x95);
         boost::shared_ptr<FieldValue> l(new Str16Value(en_US));
         locales.add(l);
@@ -91,26 +92,30 @@ 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);
+    connection.setFederationLink(clientProperties.get(QPID_FED_LINK));
+    if (connection.isFederationLink()){
+		QPID_LOG(info, "Connection is a federation link");
+    }
 }
-        
+
 void ConnectionHandler::Handler::secureOk(const string& response)
 {
     authenticator->step(response);
 }
-        
+
 void ConnectionHandler::Handler::tuneOk(uint16_t /*channelmax*/,
     uint16_t framemax, uint16_t heartbeat)
 {
     connection.setFrameMax(framemax);
     connection.setHeartbeat(heartbeat);
 }
-        
+
 void ConnectionHandler::Handler::open(const string& /*virtualHost*/,
                                       const framing::Array& /*capabilities*/, bool /*insist*/)
 {
@@ -143,8 +148,10 @@ void ConnectionHandler::Handler::start(const FieldTable& /*serverProperties*/,
 {
     string mechanism = connection.getAuthMechanism();
     string response  = connection.getAuthCredentials();
-    
-    server.startOk(FieldTable(), mechanism, response, en_US);
+
+    FieldTable ft;
+    ft.setInt(QPID_FED_LINK,1);
+    server.startOk(ft, mechanism, response, en_US);
 }
 
 void ConnectionHandler::Handler::secure(const string& /*challenge*/)
@@ -169,5 +176,5 @@ void ConnectionHandler::Handler::openOk(const framing::Array& /*knownHosts*/)
 
 void ConnectionHandler::Handler::redirect(const string& /*host*/, const framing::Array& /*knownHosts*/)
 {
-    
+
 }
diff --git a/qpid/cpp/src/qpid/broker/ConnectionState.h b/qpid/cpp/src/qpid/broker/ConnectionState.h
index 97055f8b2e..aac31bbf96 100644
--- a/qpid/cpp/src/qpid/broker/ConnectionState.h
+++ b/qpid/cpp/src/qpid/broker/ConnectionState.h
@@ -40,11 +40,12 @@ class ConnectionState : public ConnectionToken, public management::Manageable
   public:
     ConnectionState(qpid::sys::ConnectionOutputHandler* o, Broker& b) :
         out(o),
-        broker(b), 
+        broker(b),
         outputTasks(out),
-        framemax(65535), 
+        framemax(65535),
         heartbeat(0),
-        stagingThreshold(broker.getStagingThreshold())
+        stagingThreshold(broker.getStagingThreshold()),
+        federationLink(true)
         {}
 
 
@@ -61,12 +62,15 @@ class ConnectionState : public ConnectionToken, public management::Manageable
 
     virtual void setUserId(const string& uid) {  userId = uid; }
     const string& getUserId() const { return userId; }
-    
+
+    void setFederationLink(bool b) {  federationLink = b; }
+    bool isFederationLink() const { return federationLink; }
+
     Broker& getBroker() { return broker; }
 
     Broker& broker;
     std::vector<Queue::shared_ptr> exclusiveQueues;
-    
+
     //contained output tasks
     sys::AggregateOutput outputTasks;
 
@@ -81,6 +85,7 @@ class ConnectionState : public ConnectionToken, public management::Manageable
     uint16_t heartbeat;
     uint64_t stagingThreshold;
     string userId;
+    bool federationLink;
 };
 
 }}
diff --git a/qpid/cpp/src/qpid/broker/SemanticState.cpp b/qpid/cpp/src/qpid/broker/SemanticState.cpp
index 7907b47854..065e15543b 100644
--- a/qpid/cpp/src/qpid/broker/SemanticState.cpp
+++ b/qpid/cpp/src/qpid/broker/SemanticState.cpp
@@ -61,7 +61,9 @@ SemanticState::SemanticState(DeliveryAdapter& da, SessionContext& ss)
       deliveryAdapter(da),
       tagGenerator("sgen"),
       dtxSelected(false),
-      outputTasks(ss)
+      outputTasks(ss),
+      authMsg(getSession().getBroker().getOptions().auth && !getSession().getConnection().isFederationLink()),
+      userID(getSession().getConnection().getUserId().substr(0,getSession().getConnection().getUserId().find('@')))
 {
     acl = getSession().getBroker().getAcl();
 }
@@ -348,13 +350,23 @@ void SemanticState::route(intrusive_ptr<Message> msg, Deliverable& strategy) {
     // The client library ensures this is always empty for messages from normal clients.
     if (msg->isA<MessageTransferBody>()) {
         if (!msg->hasProperties<DeliveryProperties>() ||
-            msg->getProperties<DeliveryProperties>()->getExchange().empty()) 
+            msg->getProperties<DeliveryProperties>()->getExchange().empty())
             msg->getProperties<DeliveryProperties>()->setExchange(exchangeName);
     }
     if (!cacheExchange || cacheExchange->getName() != exchangeName){
         cacheExchange = session.getBroker().getExchanges().get(exchangeName);
     }
 
+    /* verify the userid if specified: */
+    std::string id =
+    	msg->hasProperties<MessageProperties>()? msg->getProperties<MessageProperties>()->getUserId():"";
+
+    if (authMsg &&  !id.empty() && id != userID )
+    {
+        QPID_LOG(debug, "user id : " << userID << " msgProps.getUserID() " << msg->getProperties<MessageProperties>()->getUserId());
+        throw UnauthorizedAccessException("user id in the message is not the same id used to authenticate the connection");
+    }
+
     if (acl && acl->doTransferAcl())
     {
         if (!acl->authorise(getSession().getConnection().getUserId(),acl::PUBLISH,acl::EXCHANGE,exchangeName, msg->getRoutingKey() ))
diff --git a/qpid/cpp/src/qpid/broker/SemanticState.h b/qpid/cpp/src/qpid/broker/SemanticState.h
index 2170fe4e2e..df631883f6 100644
--- a/qpid/cpp/src/qpid/broker/SemanticState.h
+++ b/qpid/cpp/src/qpid/broker/SemanticState.h
@@ -135,7 +135,9 @@ class SemanticState : public sys::OutputTask,
     boost::shared_ptr<Exchange> cacheExchange;
     sys::AggregateOutput outputTasks;
     AclModule* acl;
-	
+    const bool authMsg;
+    const string userID;
+
     void route(boost::intrusive_ptr<Message> msg, Deliverable& strategy);
     void record(const DeliveryRecord& delivery);
     void checkDtxTimeout();