QPID-3858: Updated code to include recent refactoring by Gordon (gsim) - see QPID-4178.

git-svn-id: https://svn.apache.org/repos/asf/qpid/branches/asyncstore@1377715 13f79535-47bb-0310-9956-ffa450edef68

7 files changed, 246 insertions, 12 deletions

diff --git a/cpp/src/qpid/acl/Acl.cpp b/cpp/src/qpid/acl/Acl.cpp
index d941577f6a..89c4b3402a 100644
--- a/cpp/src/qpid/acl/Acl.cpp
+++ b/cpp/src/qpid/acl/Acl.cpp
@@ -129,6 +129,13 @@ bool Acl::approveConnection(const qpid::broker::Connection& conn)
     return connectionCounter->approveConnection(conn);
 }
 
+
+void Acl::setUserId(const qpid::broker::Connection& connection, const std::string& username)
+{
+    connectionCounter->setUserId(connection, username);
+}
+
+
 bool Acl::result(
     const AclResult&   aclreslt,
     const std::string& id,
diff --git a/cpp/src/qpid/acl/Acl.h b/cpp/src/qpid/acl/Acl.h
index 4893f71ef2..4787934275 100644
--- a/cpp/src/qpid/acl/Acl.h
+++ b/cpp/src/qpid/acl/Acl.h
@@ -94,6 +94,8 @@ public:
 
     virtual bool approveConnection(const broker::Connection& connection);
 
+    virtual void setUserId(const broker::Connection& connection, const std::string& username);
+
     virtual ~Acl();
 private:
     bool result(
diff --git a/cpp/src/qpid/acl/AclConnectionCounter.cpp b/cpp/src/qpid/acl/AclConnectionCounter.cpp
index 052fa3c222..8c6e3eef6e 100644
--- a/cpp/src/qpid/acl/AclConnectionCounter.cpp
+++ b/cpp/src/qpid/acl/AclConnectionCounter.cpp
@@ -296,6 +296,47 @@ bool ConnectionCounter::approveConnection(const broker::Connection& connection)
     }
 }
 
+
+//
+// setUserId
+//  On cluster shadow connections, track a new user id for this connection.
+//
+void ConnectionCounter::setUserId(const broker::Connection& connection,
+                                  const std::string& username)
+{
+    Mutex::ScopedLock locker(dataLock);
+
+    connectCountsMap_t::iterator eRef = connectProgressMap.find(connection.getMgmtId());
+    if (eRef != connectProgressMap.end()) {
+        if ((*eRef).second == C_OPENED){
+            // Connection has been opened so that current user has been counted
+            if (connection.isShadow()) {
+                // This is a shadow connection and therefore receives userId changes
+                QPID_LOG(debug, "Changing User ID for cluster connection: "
+                    << connection.getMgmtId() << ", old user:'" << connection.getUserId()
+                    << "', new user:'" << username << "'");
+
+                // Decrement user in-use count for old userId
+                releaseLH(connectByNameMap,
+                        connection.getUserId(),
+                        nameLimit);
+                // Increment user in-use count for new userId
+                (void) countConnectionLH(connectByNameMap, username, nameLimit, false);
+            } else {
+                QPID_LOG(warning, "Changing User ID for non-cluster connections is not supported: "
+                    << connection.getMgmtId() << ", old user " << connection.getUserId()
+                    << ", new user " << username);
+            }
+        } else {
+            // connection exists  but has not been opened.
+            // setUserId is called in normal course. The user gets counted when connection is opened.
+        }
+    } else {
+        // Connection does not exist.
+    }
+}
+
+
 //
 // getClientIp - given a connection's mgmtId return the client host part.
 //
diff --git a/cpp/src/qpid/acl/AclConnectionCounter.h b/cpp/src/qpid/acl/AclConnectionCounter.h
index eec8e90256..54fa6933ff 100644
--- a/cpp/src/qpid/acl/AclConnectionCounter.h
+++ b/cpp/src/qpid/acl/AclConnectionCounter.h
@@ -94,6 +94,7 @@ public:
 
     // Connection counting
     bool approveConnection(const broker::Connection& conn);
+    void setUserId(const broker::Connection& connection, const std::string& username);
 };
 
 }} // namespace qpid::ha
diff --git a/cpp/src/qpid/acl/AclData.cpp b/cpp/src/qpid/acl/AclData.cpp
index a07176dc76..7c14d0985d 100644
--- a/cpp/src/qpid/acl/AclData.cpp
+++ b/cpp/src/qpid/acl/AclData.cpp
@@ -25,6 +25,13 @@ namespace qpid {
 namespace acl {
 
     //
+    // Instantiate the substitution keyword string
+    //
+    const std::string AclData::USER_SUBSTITUTION_KEYWORD       = "${user}";
+    const std::string AclData::DOMAIN_SUBSTITUTION_KEYWORD     = "${domain}";
+    const std::string AclData::USERDOMAIN_SUBSTITUTION_KEYWORD = "${userdomain}";
+
+    //
     // constructor
     //
     AclData::AclData():
@@ -147,7 +154,17 @@ namespace acl {
                         // the calling args and not in the param map.
                         if (rulePropMapItr->first == acl::SPECPROP_NAME)
                         {
-                            if (matchProp(rulePropMapItr->second, name))
+                            // substitute user name into object name
+                            bool result;
+                            if (rsItr->ruleHasUserSub[PROP_NAME]) {
+                                std::string sName(rulePropMapItr->second);
+                                substituteUserId(sName, id);
+                                result = matchProp(sName, name);
+                            } else {
+                                result = matchProp(rulePropMapItr->second, name);
+                            }
+
+                            if (result)
                             {
                                 QPID_LOG(debug, "ACL: lookup name '" << name
                                     << "' matched with rule name '"
@@ -222,7 +239,20 @@ namespace acl {
                                         break;
 
                                     default:
-                                        if (matchProp(rulePropMapItr->second, lookupParamItr->second))
+                                        bool result;
+                                        if ((SPECPROP_ALTERNATE  == rulePropMapItr->first && rsItr->ruleHasUserSub[PROP_ALTERNATE])  ||
+                                            (SPECPROP_ROUTINGKEY == rulePropMapItr->first && rsItr->ruleHasUserSub[PROP_ROUTINGKEY]) ||
+                                            (SPECPROP_QUEUENAME  == rulePropMapItr->first && rsItr->ruleHasUserSub[PROP_QUEUENAME]))
+                                        {
+                                            // These properties are allowed to have username substitution
+                                            std::string sName(rulePropMapItr->second);
+                                            substituteUserId(sName, id);
+                                            result = matchProp(sName, lookupParamItr->second);
+                                        } else {
+                                            result = matchProp(rulePropMapItr->second, lookupParamItr->second);
+                                        }
+
+                                        if (result)
                                         {
                                             QPID_LOG(debug, "ACL: the pair("
                                                 << AclHelper::getPropertyStr(lookupParamItr->first)
@@ -346,7 +376,18 @@ namespace acl {
                     bool match =true;
                     if (rsItr->pubExchNameInRule)
                     {
-                        if (matchProp(rsItr->pubExchName, name))
+                        // substitute user name into object name
+                        bool result;
+
+                        if (rsItr->ruleHasUserSub[PROP_NAME]) {
+                            std::string sName(rsItr->pubExchName);
+                            substituteUserId(sName, id);
+                            result = matchProp(sName, name);
+                        } else {
+                            result = matchProp(rsItr->pubExchName, name);
+                        }
+
+                        if (result)
                         {
                             QPID_LOG(debug, "ACL: Rule: " << rsItr->rawRuleNum << " lookup exchange name '"
                                 << name << "' matched with rule name '"
@@ -364,18 +405,40 @@ namespace acl {
 
                     if (match && rsItr->pubRoutingKeyInRule)
                     {
-                        if (rsItr->matchRoutingKey(routingKey))
+                        if ((routingKey.find(USER_SUBSTITUTION_KEYWORD, 0)       != std::string::npos) ||
+                            (routingKey.find(DOMAIN_SUBSTITUTION_KEYWORD, 0)     != std::string::npos) ||
+                            (routingKey.find(USERDOMAIN_SUBSTITUTION_KEYWORD, 0) != std::string::npos))
                         {
-                            QPID_LOG(debug, "ACL: Rule: " << rsItr->rawRuleNum << " lookup key name '"
-                                << routingKey << "' matched with rule routing key '"
-                                << rsItr->pubRoutingKey << "'");
+                            // The user is not allowed to present a routing key with the substitution key in it
+                            QPID_LOG(debug, "ACL: Rule: " << rsItr->rawRuleNum <<
+                                " User-specified routing key has substitution wildcard:" << routingKey
+                                << ". Rule match prohibited.");
+                            match = false;
                         }
                         else
                         {
-                            QPID_LOG(debug, "ACL: Rule: " << rsItr->rawRuleNum << " lookup key name '"
-                                << routingKey << "' did not match with rule routing key '"
-                                << rsItr->pubRoutingKey << "'");
-                            match = false;
+                            bool result;
+                            if (rsItr->ruleHasUserSub[PROP_ROUTINGKEY]) {
+                                std::string sKey(routingKey);
+                                substituteKeywords(sKey, id);
+                                result = rsItr->matchRoutingKey(sKey);
+                            } else {
+                                result = rsItr->matchRoutingKey(routingKey);
+                            }
+
+                            if (result)
+                            {
+                                QPID_LOG(debug, "ACL: Rule: " << rsItr->rawRuleNum << " lookup key name '"
+                                    << routingKey << "' matched with rule routing key '"
+                                    << rsItr->pubRoutingKey << "'");
+                            }
+                            else
+                            {
+                                QPID_LOG(debug, "ACL: Rule: " << rsItr->rawRuleNum << " lookup key name '"
+                                    << routingKey << "' did not match with rule routing key '"
+                                    << rsItr->pubRoutingKey << "'");
+                                match = false;
+                            }
                         }
                     }
 
@@ -501,4 +564,102 @@ namespace acl {
         return true;
     }
 
+    const std::string DOMAIN_SEPARATOR("@");
+    const std::string PERIOD(".");
+    const std::string UNDERSCORE("_");
+    //
+    // substituteString
+    //   Given a name string from an Acl rule, substitute the replacement into it
+    //   wherever the placeholder directs.
+    //
+    void AclData::substituteString(std::string& targetString,
+                                   const std::string& placeholder,
+                                   const std::string& replacement)
+    {
+        assert (!placeholder.empty());
+        if (placeholder.empty())
+            return;
+        size_t start_pos(0);
+        while((start_pos = targetString.find(placeholder, start_pos)) != std::string::npos)
+        {
+            targetString.replace(start_pos, placeholder.length(), replacement);
+            start_pos += replacement.length();
+        }
+    }
+
+
+    //
+    // normalizeUserId
+    //   Given a name string return it in a form usable as topic keys:
+    //     change "@" and "." to "_".
+    //
+    std::string AclData::normalizeUserId(const std::string& userId)
+    {
+        std::string normalId(userId);
+        substituteString(normalId, DOMAIN_SEPARATOR, UNDERSCORE);
+        substituteString(normalId, PERIOD,           UNDERSCORE);
+        return normalId;
+    }
+
+
+    //
+    // substituteUserId
+    //   Given an Acl rule and an authenticated userId
+    //   do the keyword substitutions on the rule.
+    //
+    void AclData::AclData::substituteUserId(std::string& ruleString,
+                                            const std::string& userId)
+    {
+        size_t locDomSeparator(0);
+        std::string user("");
+        std::string domain("");
+        std::string userdomain = normalizeUserId(userId);
+
+        locDomSeparator = userId.find(DOMAIN_SEPARATOR);
+        if (std::string::npos == locDomSeparator) {
+            // "@" not found. There's just a user name
+            user   = normalizeUserId(userId);
+        } else {
+            // "@" found, split the names. Domain may be blank.
+            user   = normalizeUserId(userId.substr(0,locDomSeparator));
+            domain = normalizeUserId(userId.substr(locDomSeparator+1));
+        }
+
+        substituteString(ruleString, USER_SUBSTITUTION_KEYWORD,       user);
+        substituteString(ruleString, DOMAIN_SUBSTITUTION_KEYWORD,     domain);
+        substituteString(ruleString, USERDOMAIN_SUBSTITUTION_KEYWORD, userdomain);
+    }
+
+
+    //
+    // substituteKeywords
+    //   Given an Acl rule and an authenticated userId
+    //   do reverse keyword substitutions on the rule.
+    //   That is, replace the normalized name in the rule string with
+    //   the keyword that represents it. This stragegy is used for
+    //   topic key lookups where the keyword string proper is in the
+    //   topic key search tree.
+    //
+    void AclData::AclData::substituteKeywords(std::string& ruleString,
+                                              const std::string& userId)
+    {
+        size_t locDomSeparator(0);
+        std::string user("");
+        std::string domain("");
+        std::string userdomain = normalizeUserId(userId);
+
+        locDomSeparator = userId.find(DOMAIN_SEPARATOR);
+        if (std::string::npos == locDomSeparator) {
+            // "@" not found. There's just a user name
+            user   = normalizeUserId(userId);
+        } else {
+            // "@" found, split the names
+            user   = normalizeUserId(userId.substr(0,locDomSeparator));
+            domain = normalizeUserId(userId.substr(locDomSeparator+1));
+        }
+        std::string oRule(ruleString);
+        substituteString(ruleString, userdomain, USERDOMAIN_SUBSTITUTION_KEYWORD);
+        substituteString(ruleString, user,       USER_SUBSTITUTION_KEYWORD);
+        substituteString(ruleString, domain,     DOMAIN_SUBSTITUTION_KEYWORD);
+    }
 }}
diff --git a/cpp/src/qpid/acl/AclData.h b/cpp/src/qpid/acl/AclData.h
index ca0a676a1c..b4b13c44b6 100644
--- a/cpp/src/qpid/acl/AclData.h
+++ b/cpp/src/qpid/acl/AclData.h
@@ -62,6 +62,7 @@ public:
         boost::shared_ptr<topicTester> pTTest;
         bool                  pubExchNameInRule;
         std::string           pubExchName;
+        std::vector<bool>     ruleHasUserSub;
 
         Rule (int ruleNum, qpid::acl::AclResult res, specPropertyMap& p) :
             rawRuleNum(ruleNum),
@@ -71,7 +72,8 @@ public:
             pubRoutingKey(),
             pTTest(boost::shared_ptr<topicTester>(new topicTester())),
             pubExchNameInRule(false),
-            pubExchName()
+            pubExchName(),
+            ruleHasUserSub(PROPERTYSIZE, false)
             {}
 
 
@@ -132,6 +134,17 @@ public:
 
     bool matchProp(const std::string & src, const std::string& src1);
     void clear ();
+    static const std::string USER_SUBSTITUTION_KEYWORD;
+    static const std::string DOMAIN_SUBSTITUTION_KEYWORD;
+    static const std::string USERDOMAIN_SUBSTITUTION_KEYWORD;
+    void substituteString(std::string& targetString,
+                          const std::string& placeholder,
+                          const std::string& replacement);
+    std::string normalizeUserId(const std::string& userId);
+    void substituteUserId(std::string& ruleString,
+                          const std::string& userId);
+    void substituteKeywords(std::string& ruleString,
+                            const std::string& userId);
 
     AclData();
     virtual ~AclData();
diff --git a/cpp/src/qpid/acl/AclReader.cpp b/cpp/src/qpid/acl/AclReader.cpp
index f9be49b88d..fae67d0325 100644
--- a/cpp/src/qpid/acl/AclReader.cpp
+++ b/cpp/src/qpid/acl/AclReader.cpp
@@ -103,6 +103,15 @@ namespace acl {
             } else {
                 AclData::Rule rule(cnt, (*i)->res, (*i)->props);
 
+                // Record which properties have the user substitution string
+                for (pmCitr pItr=rule.props.begin(); pItr!=rule.props.end(); pItr++) {
+                    if ((pItr->second.find(AclData::USER_SUBSTITUTION_KEYWORD, 0)       != std::string::npos) ||
+                        (pItr->second.find(AclData::DOMAIN_SUBSTITUTION_KEYWORD, 0)     != std::string::npos) ||
+                        (pItr->second.find(AclData::USERDOMAIN_SUBSTITUTION_KEYWORD, 0) != std::string::npos)) {
+                        rule.ruleHasUserSub[pItr->first] = true;
+                    }
+                }
+
                 // Action -> Object -> map<user -> set<Rule> >
                 std::ostringstream actionstr;
                 for (int acnt = ((*i)->actionAll ? 0 : (*i)->action);