You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ch...@apache.org on 2020/07/08 17:49:26 UTC
[qpid-dispatch] branch master updated: DISPATCH-1585: Define vhost
aliases for configuation sharing
This is an automated email from the ASF dual-hosted git repository.
chug pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/qpid-dispatch.git
The following commit(s) were added to refs/heads/master by this push:
new 5f3eb3f DISPATCH-1585: Define vhost aliases for configuation sharing
5f3eb3f is described below
commit 5f3eb3f8a7b004a6404d7afe1552fedac8340335
Author: Chuck Rolke <ch...@apache.org>
AuthorDate: Wed Jul 8 13:40:10 2020 -0400
DISPATCH-1585: Define vhost aliases for configuation sharing
Given a vhost definition:
vhost {
hostname: example.com
aliases: example.org, example.net
... }
Vhost _aliases_ are alternative literal hostnames or patterns that direct
the router to use the settings in this vhost. Alias hostnames that match an
incoming connection will use the settings defined in the contained vhost.
In a multi-tenant configuration a connection to a vhost alias will use the
base vhost hostname for the tenant namespace. In this example if a connection
is directed to vhost `example.org` then the settings from the base vhost
hostname `example.com` will apply and `example.com` will be the tenant
namespace.
Vhost `hostname` and `aliases` settings from all vhosts are shared in a
single list and must be unique.
This closes #770
---
config.sh | 47 +-
.../user-guide/creating-vhost-policies.adoc | 6 +
python/qpid_dispatch/management/qdrouter.json | 7 +
.../qpid_dispatch_internal/policy/policy_local.py | 133 ++-
.../policy/policy_manager.py | 25 +
src/policy.c | 66 ++
src/policy.h | 15 +
src/router_node.c | 4 +-
tests/CMakeLists.txt | 1 +
tests/router_policy_test.py | 128 +++
tests/system_tests_multi_tenancy_policy.py | 931 +++++++++++++++++++++
tests/system_tests_policy.py | 134 +++
12 files changed, 1439 insertions(+), 58 deletions(-)
diff --git a/config.sh b/config.sh
index fff867d..b083c63 100755
--- a/config.sh
+++ b/config.sh
@@ -1,33 +1,14 @@
-#
-# Licensed to the Apache Software Foundation (ASF) under one
-# or more contributor license agreements. See the NOTICE file
-# distributed with this work for additional information
-# regarding copyright ownership. The ASF licenses this file
-# 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
-# KIND, either express or implied. See the License for the
-# specific language governing permissions and limitations
-# under the License.
-#
-
-if [[ ! -f config.sh ]]; then
- echo "You must source config.sh from within its own directory"
- return
-fi
-
-export SOURCE_DIR=$(pwd)
-export BUILD_DIR=$SOURCE_DIR/${1:-build}
-export INSTALL_DIR=$SOURCE_DIR/${2:-install}
-
-PYTHON_BIN=`type -P python || type -P python3`
-PYTHON_LIB=$(${PYTHON_BIN} -c "from distutils.sysconfig import get_python_lib; print(get_python_lib(prefix='$INSTALL_DIR'))")
-
-export PYTHONPATH=$PYTHON_LIB:$PYTHONPATH
-export PATH=$INSTALL_DIR/sbin:$INSTALL_DIR/bin:$SOURCE_DIR/bin:$PATH
+PYTHONPATH="/home/chug/git/qpid-dispatch/python:/home/chug/git/qpid-dispatch/build/python:/home/chug/git/qpid-dispatch/tests:/opt/local/lib/python3.7/site-packages:/opt/local/lib64/proton/bindings/python:"
+PATH="/home/chug/git/qpid-dispatch/build:/home/chug/git/qpid-dispatch/build/tests:/home/chug/git/qpid-dispatch/build/router:/home/chug/git/qpid-dispatch/tools:/home/chug/git/qpid-dispatch/build/tools:/home/chug/git/qpid-dispatch/bin:/opt/local/sbin:/opt/local/bin:/usr/share/Modules/bin:/usr/lib64/ccache:/usr/local/bin:/usr/local/sbin:/usr/bin:/usr/sbin:/home/chug/.dotnet/tools:/home/chug/bin"
+MANPATH="/home/chug/git/qpid-dispatch/build/doc/man::"
+SOURCE_DIR="/home/chug/git/qpid-dispatch"
+BUILD_DIR="/home/chug/git/qpid-dispatch/build"
+QPID_DISPATCH_HOME="/home/chug/git/qpid-dispatch"
+QPID_DISPATCH_LIB="/home/chug/git/qpid-dispatch/build/src/"
+QPID_DISPATCH_RUNNER=""
+MALLOC_CHECK_="3"
+MALLOC_PERTURB_="153"
+TSAN_OPTIONS=""
+ASAN_OPTIONS=""
+LSAN_OPTIONS=""
+export PYTHONPATH PATH MANPATH SOURCE_DIR BUILD_DIR QPID_DISPATCH_HOME QPID_DISPATCH_LIB QPID_DISPATCH_RUNNER MALLOC_CHECK_ MALLOC_PERTURB_ TSAN_OPTIONS ASAN_OPTIONS LSAN_OPTIONS
diff --git a/docs/books/modules/user-guide/creating-vhost-policies.adoc b/docs/books/modules/user-guide/creating-vhost-policies.adoc
index 7849382..a7e9101 100644
--- a/docs/books/modules/user-guide/creating-vhost-policies.adoc
+++ b/docs/books/modules/user-guide/creating-vhost-policies.adoc
@@ -41,6 +41,7 @@ The connection limits apply to all users that are connected to the vhost. These
----
vhost {
hostname: example.com
+ aliases: example.org, example.net
maxConnections: 10000
maxMessageSize: 500000
maxConnectionsPerUser: 100
@@ -54,6 +55,11 @@ The literal hostname of the vhost (the messaging endpoint) or a pattern that mat
+
If `enableVhostNamePatterns` is set to `true`, you can use wildcards to specify a pattern that matches a range of hostnames. For more information, see xref:vhost-policy-hostname-pattern-matching-rules-{context}[].
+`aliases`::
+Alternative literal hostnames or patterns that direct the router to use the settings in this vhost. Alias hostnames that match an incoming connection will use the settings defined in the contained vhost. In a multi-tenant configuration a connection to a vhost alias will use the base vhost hostname for the tenant namespace. In this example if a connection is directed to vhost `example.org` then the settings from the base vhost hostname `example.com` will apply and `example.com` will be th [...]
++
+If `enableVhostNamePatterns` is set to `true`, you can use wildcards to specify a pattern that matches a range of hostname aliases. For more information, see xref:vhost-policy-hostname-pattern-matching-rules-{context}[].
+
`maxConnections`::
The global maximum number of concurrent client connections allowed for this vhost. The default is 65535.
diff --git a/python/qpid_dispatch/management/qdrouter.json b/python/qpid_dispatch/management/qdrouter.json
index 0cea967..aa1e13d 100644
--- a/python/qpid_dispatch/management/qdrouter.json
+++ b/python/qpid_dispatch/management/qdrouter.json
@@ -1945,6 +1945,13 @@
"required": true,
"create": true
},
+ "aliases": {
+ "type": "string",
+ "description": "Alternate hostnames that share this vhost configuation. Hosts named in this attribute are treated as if this vhost was defined with the alias name in the vhost 'hostname' attribute. This attribute is implemented to help with multitenant configurations where multiple vhosts share a common configuration. The string is a comma- or space-separated list of literal hostnames or hostname patterns. A vhost aliases hostname must be unique across all vhost hostn [...]
+ "required": false,
+ "create": true,
+ "update": true
+ },
"maxConnections": {
"type": "integer",
"default": 65535,
diff --git a/python/qpid_dispatch_internal/policy/policy_local.py b/python/qpid_dispatch_internal/policy/policy_local.py
index cfe3400..b75163e 100644
--- a/python/qpid_dispatch_internal/policy/policy_local.py
+++ b/python/qpid_dispatch_internal/policy/policy_local.py
@@ -27,6 +27,7 @@ from __future__ import print_function
import json
import pdb
+import sys
from .policy_util import PolicyError, HostStruct, HostAddr, PolicyAppConnectionMgr, is_ipv6_enabled
from ..compat import PY_STRING_TYPE
from ..compat import PY_TEXT_TYPE
@@ -78,6 +79,7 @@ class PolicyKeys(object):
KW_TARGETS = "targets"
KW_SOURCE_PATTERN = "sourcePattern"
KW_TARGET_PATTERN = "targetPattern"
+ KW_VHOST_ALIASES = "aliases"
# Policy stats key words
KW_CONNECTIONS_APPROVED = "connectionsApproved"
@@ -135,7 +137,8 @@ class PolicyCompiler(object):
PolicyKeys.KW_MAXCONNPERHOST,
PolicyKeys.KW_MAXCONNPERUSER,
PolicyKeys.KW_CONNECTION_ALLOW_DEFAULT,
- PolicyKeys.KW_GROUPS
+ PolicyKeys.KW_GROUPS,
+ PolicyKeys.KW_VHOST_ALIASES
]
allowed_settings_options = [
@@ -324,7 +327,8 @@ class PolicyCompiler(object):
PolicyKeys.KW_SOURCES,
PolicyKeys.KW_TARGETS,
PolicyKeys.KW_SOURCE_PATTERN,
- PolicyKeys.KW_TARGET_PATTERN
+ PolicyKeys.KW_TARGET_PATTERN,
+ PolicyKeys.KW_VHOST_ALIASES
]:
# accept a string or list
if isinstance(val, list):
@@ -431,6 +435,7 @@ class PolicyCompiler(object):
policy_out[PolicyKeys.KW_CONNECTION_ALLOW_DEFAULT] = False
policy_out[PolicyKeys.KW_GROUPS] = {}
policy_out[PolicyKeys.KW_MAX_MESSAGE_SIZE] = None
+ policy_out[PolicyKeys.KW_VHOST_ALIASES] = []
# validate the options
for key, val in dict_iteritems(policy_in):
@@ -461,6 +466,22 @@ class PolicyCompiler(object):
(name, key, type(val)))
return False
policy_out[key] = val
+ elif key in [PolicyKeys.KW_VHOST_ALIASES]:
+ # vhost aliases is a CSV string. convert to a list
+ val0 = [x.strip(' ') for x in val.split(PolicyKeys.KC_CONFIG_LIST_SEP)]
+ # Reject aliases that duplicate the vhost itself or other aliases
+ val = []
+ for vtest in val0:
+ if vtest == name:
+ errors.append("Policy vhost '%s' option '%s' value '%s' duplicates vhost name" %
+ (name, key, vtest))
+ return False
+ if vtest in val:
+ errors.append("Policy vhost '%s' option '%s' value '%s' is duplicated" %
+ (name, key, vtest))
+ return False
+ val.append(vtest)
+ policy_out[key] = val
elif key in [PolicyKeys.KW_GROUPS]:
if not type(val) is dict:
errors.append("Policy vhost '%s' option '%s' must be of type 'dict' but is '%s'" %
@@ -622,6 +643,11 @@ class PolicyLocal(object):
# _max_message_size
# holds global value from policy config object
self._max_message_size = 0
+
+ # _vhost_aliases is a map
+ # key : alias vhost name
+ # val : actual vhost to which alias refers
+ self._vhost_aliases = {}
#
# Service interfaces
#
@@ -641,16 +667,64 @@ class PolicyLocal(object):
if len(warnings) > 0:
for warning in warnings:
self._manager.log_warning(warning)
+
+ # Reject if any vhost alias name conflicts
+ if name in self._vhost_aliases:
+ # hostname is an alias
+ raise PolicyError(
+ "Policy is creating vhost '%s' but that name is already an alias for vhost '%s'" % (name, self._vhost_aliases[name]))
+ for vhost_alias in candidate[PolicyKeys.KW_VHOST_ALIASES]:
+ # alias is a hostname
+ if vhost_alias in self.rulesetdb.keys():
+ raise PolicyError(
+ "Policy for vhost '%s' defines alias '%s' which conflicts with an existing vhost named '%s'" % (name, vhost_alias, vhost_alias))
+ if name not in self.rulesetdb:
+ # Creating new ruleset. Vhost aliases cannot overlap
+ for vhost_alias in candidate[PolicyKeys.KW_VHOST_ALIASES]:
+ if vhost_alias in self._vhost_aliases:
+ raise PolicyError(
+ "Policy for vhost '%s' alias '%s' conflicts with existing alias for vhost '%s'" % (name, vhost_alias, self._vhost_aliases[vhost_alias]))
+ else:
+ # Updating an existing ruleset.
+ # Vhost aliases still cannot overlap but replacement is allowed
+ for vhost_alias in candidate[PolicyKeys.KW_VHOST_ALIASES]:
+ if vhost_alias in self._vhost_aliases and not self._vhost_aliases[vhost_alias] == name:
+ raise PolicyError(
+ "Policy for vhost '%s' alias '%s' conflicts with existing alias for vhost '%s'" % (name, vhost_alias, self._vhost_aliases[vhost_alias]))
+
# Reject if parse tree optimized name collision
+ # Coincidently add name and aliases to parse tree
if self.use_hostname_patterns:
agent = self._manager.get_agent()
- if not agent.qd.qd_dispatch_policy_host_pattern_add(agent.dispatch, name):
- raise PolicyError("Policy '%s' optimized pattern conflicts with existing pattern" % name)
+ # construct a list of names to be added
+ tnames = []
+ tnames.append(name)
+ tnames += candidate[PolicyKeys.KW_VHOST_ALIASES]
+ # create a list of names to undo in case a subsequent name does not work
+ snames = []
+ for tname in tnames:
+ if not agent.qd.qd_dispatch_policy_host_pattern_add(agent.dispatch, tname):
+ # undo the snames list
+ for sname in snames:
+ agent.qd.qd_dispatch_policy_host_pattern_del(agent.dispatch, sname)
+ raise PolicyError("Policy for vhost '%s' alias '%s' optimized pattern conflicts with existing pattern" % (name, tname))
+ snames.append(tname)
+ # Names pass administrative approval
if name not in self.rulesetdb:
+ # add new aliases
+ for nname in candidate[PolicyKeys.KW_VHOST_ALIASES]:
+ self._vhost_aliases[nname] = name
if name not in self.statsdb:
self.statsdb[name] = AppStats(name, self._manager, candidate)
self._manager.log_info("Created policy rules for vhost %s" % name)
else:
+ # remove old aliases
+ old_aliases = self.rulesetdb[name][PolicyKeys.KW_VHOST_ALIASES]
+ for oname in old_aliases:
+ del self._vhost_aliases[oname]
+ # add new aliases
+ for nname in candidate[PolicyKeys.KW_VHOST_ALIASES]:
+ self._vhost_aliases[nname] = name
self.statsdb[name].update_ruleset(candidate)
self._manager.log_info("Updated policy rules for vhost %s" % name)
# TODO: ruleset lock
@@ -668,6 +742,9 @@ class PolicyLocal(object):
if self.use_hostname_patterns:
agent = self._manager.get_agent()
agent.qd.qd_dispatch_policy_host_pattern_remove(agent.dispatch, name)
+ anames = self.rulesetdb[name][PolicyKeys.KW_VHOST_ALIASES]
+ for aname in anames:
+ agent.qd.qd_dispatch_policy_host_pattern_remove(agent.dispatch, aname)
del self.rulesetdb[name]
#
@@ -699,6 +776,26 @@ class PolicyLocal(object):
#
# Runtime query interface
#
+ def lookup_vhost_alias(self, vhost_in):
+ """
+ Resolve given vhost name to vhost settings name.
+ If the incoming name is a vhost hostname then return the same name.
+ If the incoming name is a vhost alias hostname then return the containing vhost name.
+ If a default vhost is defined then return its name.
+ :param vhost_in: vhost name to test
+ :return: name of policy settings vhost to be applied or blank if lookup failed.
+ """
+ vhost = vhost_in
+ if self.use_hostname_patterns:
+ agent = self._manager.get_agent()
+ vhost = agent.qd.qd_dispatch_policy_host_pattern_lookup(agent.dispatch, vhost)
+ # Translate an aliased vhost to a concrete vhost. If no alias then use current vhost.
+ vhost = self._vhost_aliases.get(vhost, vhost)
+ # If no usable vhost yet then try default vhost
+ if vhost not in self.rulesetdb:
+ vhost = self._default_vhost if self.default_vhost_enabled() else ""
+ return vhost
+
def lookup_user(self, user, rhost, vhost_in, conn_name, conn_id):
"""
Lookup function called from C.
@@ -716,18 +813,12 @@ class PolicyLocal(object):
try:
# choose rule set based on incoming vhost or default vhost
# or potential vhost found by pattern matching
- vhost = vhost_in
- if self.use_hostname_patterns:
- agent = self._manager.get_agent()
- vhost = agent.qd.qd_dispatch_policy_host_pattern_lookup(agent.dispatch, vhost)
- if vhost not in self.rulesetdb:
- if self.default_vhost_enabled():
- vhost = self._default_vhost
- else:
- self._manager.log_info(
- "DENY AMQP Open for user '%s', rhost '%s', vhost '%s': "
- "No policy defined for vhost" % (user, rhost, vhost_in))
- return ""
+ vhost = self.lookup_vhost_alias(vhost_in)
+ if vhost == "":
+ self._manager.log_info(
+ "DENY AMQP Open for user '%s', rhost '%s', vhost '%s': "
+ "No policy defined for vhost" % (user, rhost, vhost_in))
+ return ""
if vhost != vhost_in:
self._manager.log_debug(
"AMQP Open for user '%s', rhost '%s', vhost '%s': "
@@ -815,13 +906,7 @@ class PolicyLocal(object):
# Note: the upolicy output is a non-nested dict with settings of interest
"""
try:
- vhost = vhost_in
- if self.use_hostname_patterns:
- agent = self._manager.get_agent()
- vhost = agent.qd.qd_dispatch_policy_host_pattern_lookup(agent.dispatch, vhost)
- if vhost not in self.rulesetdb:
- if self.default_vhost_enabled():
- vhost = self._default_vhost
+ vhost = self.lookup_vhost_alias(vhost_in)
if vhost != vhost_in:
self._manager.log_debug(
"AMQP Open lookup settings for vhost '%s': "
@@ -887,7 +972,7 @@ class PolicyLocal(object):
Test function to load a policy.
@return:
"""
- ruleset_str = '["vhost", {"hostname": "photoserver", "maxConnections": 50, "maxConnectionsPerUser": 5, "maxConnectionsPerHost": 20, "allowUnknownUser": true,'
+ ruleset_str = '["vhost", {"hostname": "photoserver", "maxConnections": 50, "maxConnectionsPerUser": 5, "maxConnectionsPerHost": 20, "allowUnknownUser": true, "aliases": "antialias",'
ruleset_str += '"groups": {'
ruleset_str += '"anonymous": { "users": "anonymous", "remoteHosts": "*", "maxFrameSize": 111111, "maxMessageSize": 111111, "maxSessionWindow": 111111, "maxSessions": 1, "maxSenders": 11, "maxReceivers": 11, "allowDynamicSource": false, "allowAnonymousSender": false, "sources": "public", "targets": "" },'
ruleset_str += '"users": { "users": "u1, u2", "remoteHosts": "*", "maxFrameSize": 222222, "maxMessageSize": 222222, "maxSessionWindow": 222222, "maxSessions": 2, "maxSenders": 22, "maxReceivers": 22, "allowDynamicSource": false, "allowAnonymousSender": false, "sources": "public, private", "targets": "public" },'
diff --git a/python/qpid_dispatch_internal/policy/policy_manager.py b/python/qpid_dispatch_internal/policy/policy_manager.py
index f7f35fb..b44239b 100644
--- a/python/qpid_dispatch_internal/policy/policy_manager.py
+++ b/python/qpid_dispatch_internal/policy/policy_manager.py
@@ -128,6 +128,17 @@ class PolicyManager(object):
#
# Runtime query interface
#
+ def lookup_vhost_alias(self, vhost_in):
+ """
+ Resolve given vhost name to vhost settings name.
+ If the incoming name is a vhost hostname then return the same name.
+ If the incoming name is a vhost alias hostname then return the containing vhost name.
+ If a default vhost is defined then return its name.
+ :param vhost_in: vhost name to test
+ :return: name of policy settings vhost to be applied. Or blank if not defined.
+ """
+ return self._policy_local.lookup_vhost_alias(vhost_in)
+
def lookup_user(self, user, rhost, vhost, conn_name, conn_id):
"""
Lookup function called from C.
@@ -169,6 +180,20 @@ class PolicyManager(object):
:return: none
"""
self._policy_local.set_max_message_size(size)
+
+#
+#
+#
+def policy_lookup_vhost_alias(mgr, vhost):
+ """
+ Look up a vhost in the policy database
+ Called by C code
+ @param mgr: policy_manager
+ @param vhost: Incoming vhost from an AMQP Open
+ @return: name of policy settings vhost to be applied or blank if lookup failed.
+ """
+ return mgr.lookup_vhost_alias(vhost)
+
#
#
#
diff --git a/src/policy.c b/src/policy.c
index 7378243..ae093b1 100644
--- a/src/policy.c
+++ b/src/policy.c
@@ -406,6 +406,55 @@ qd_parse_tree_t * qd_policy_parse_tree(const char *config_spec)
// * If allowed then return the settings from the python vhost database.
//
+/** Look up vhost in python vhost aliases database
+ * * Return false if the mechanics of calling python fails or if returned name buf is blank.
+ * * Return true if a name was returned.
+ * @param[in] policy pointer to policy
+ * @param[in] vhost application name received in remote AMQP Open.hostname
+ * @param[out] name_buf pointer to return name buffer
+ * @param[in] name_buf_size size of name_buf
+ **/
+bool qd_policy_lookup_vhost_alias(
+ qd_policy_t *policy,
+ const char *vhost,
+ char *name_buf,
+ int name_buf_size)
+{
+ bool res = false;
+ name_buf[0] = 0;
+ qd_python_lock_state_t lock_state = qd_python_lock();
+ {
+ PyObject *lookup_vhost_alias = PyObject_GetAttrString(module, "policy_lookup_vhost_alias");
+ if (lookup_vhost_alias) {
+ PyObject *result = PyObject_CallFunction(lookup_vhost_alias, "(Os)",
+ (PyObject *)policy->py_policy_manager,
+ vhost);
+ if (result) {
+ char *res_string = py_obj_2_c_string(result);
+ const size_t res_len = res_string ? strlen(res_string) : 0;
+ if (res_string && res_len < name_buf_size) {
+ strcpy(name_buf, res_string);
+ } else {
+ qd_log(policy->log_source, QD_LOG_ERROR,
+ "Internal: lookup_vhost_alias: insufficient buffer for name");
+ }
+ Py_XDECREF(result);
+ free(res_string);
+ res = !!name_buf[0]; // settings name returned
+ } else {
+ qd_log(policy->log_source, QD_LOG_DEBUG, "Internal: lookup_vhost_alias: result");
+ }
+ Py_XDECREF(lookup_vhost_alias);
+ } else {
+ qd_log(policy->log_source, QD_LOG_DEBUG, "Internal: lookup_vhost_alias: lookup_vhost_alias");
+ }
+ }
+ qd_python_unlock(lock_state);
+
+ return res;
+}
+
+
/** Look up user/host/vhost in python vhost database and give the AMQP Open
* a go-no_go decision.
* * Return false if the mechanics of calling python fails or if name buf is blank.
@@ -1241,6 +1290,22 @@ void qd_policy_amqp_open(qd_connection_t *qd_conn) {
pn_transport_set_max_frame(pn_trans, qd_conn->policy_settings->maxFrameSize);
if (qd_conn->policy_settings->maxSessions > 0)
pn_transport_set_channel_max(pn_trans, qd_conn->policy_settings->maxSessions - 1);
+ const qd_server_config_t *cf = qd_connection_config(qd_conn);
+ if (cf && cf->multi_tenant) {
+ char vhost_name_buf[SETTINGS_NAME_SIZE];
+ if (qd_policy_lookup_vhost_alias(policy, vhost, vhost_name_buf, SETTINGS_NAME_SIZE)) {
+ if (!strcmp(pcrh, vhost_name_buf)) {
+ // Default condition: use proton connection value; no action here
+ } else {
+ // Policy used a name different from what came in the AMQP Open hostname.
+ // Memorize it for multitenant namespace
+ qd_conn->policy_settings->vhost_name = (char*)malloc(strlen(vhost_name_buf) + 1);
+ strcpy(qd_conn->policy_settings->vhost_name, vhost_name_buf);
+ }
+ }
+ } else {
+ // not multi-tenant: don't look for vhost
+ }
} else {
// failed to fetch settings
connection_allowed = false;
@@ -1319,6 +1384,7 @@ void qd_policy_settings_free(qd_policy_settings_t *settings)
if (settings->targetPattern) free(settings->targetPattern);
if (settings->sourceParseTree) qd_parse_tree_free(settings->sourceParseTree);
if (settings->targetParseTree) qd_parse_tree_free(settings->targetParseTree);
+ if (settings->vhost_name) free(settings->vhost_name);
free_qd_policy_settings_t(settings);
}
diff --git a/src/policy.h b/src/policy.h
index 5ad937b..ea6ecd7 100644
--- a/src/policy.h
+++ b/src/policy.h
@@ -66,6 +66,7 @@ struct qd_policy__settings_s {
qd_parse_tree_t *sourceParseTree;
qd_parse_tree_t *targetParseTree;
qd_policy_denial_counts_t *denialCounts;
+ char *vhost_name;
};
typedef struct qd_policy__settings_s qd_policy_settings_t;
@@ -126,6 +127,20 @@ bool qd_policy_socket_accept(qd_policy_t *context, const char *hostname);
void qd_policy_socket_close(qd_policy_t *context, const qd_connection_t *conn);
+/** Look up vhost in python vhost aliases database
+ * * Return false if the mechanics of calling python fails or if returned name buf is blank.
+ * * Return true if a name was returned.
+ * @param[in] policy pointer to policy
+ * @param[in] vhost vhost name received in remote AMQP Open.hostname
+ * @param[out] name_buf pointer to result name buffer
+ * @param[in] name_buf_size size of name_buf
+ **/
+bool qd_policy_lookup_vhost_alias(
+ qd_policy_t *policy,
+ const char *vhost,
+ char *name_buf,
+ int name_buf_size);
+
/** Approve a new session based on connection's policy.
* Sessions denied are closed and counted.
*
diff --git a/src/router_node.c b/src/router_node.c
index 9579241..0597ba0 100644
--- a/src/router_node.c
+++ b/src/router_node.c
@@ -1161,7 +1161,9 @@ static void AMQP_opened_handler(qd_router_t *router, qd_connection_t *conn, bool
if (multi_tenant)
- vhost = pn_connection_remote_hostname(pn_conn);
+ vhost = (conn->policy_settings && conn->policy_settings->vhost_name) ?
+ conn->policy_settings->vhost_name :
+ pn_connection_remote_hostname(pn_conn);
char proto[50];
memset(proto, 0, 50);
diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt
index 81e22f8..92cc8b7 100644
--- a/tests/CMakeLists.txt
+++ b/tests/CMakeLists.txt
@@ -118,6 +118,7 @@ foreach(py_test_module
system_tests_interior_sync_up
system_tests_distribution
system_tests_multi_tenancy
+ system_tests_multi_tenancy_policy
system_tests_dynamic_terminus
system_tests_log_message_components
system_tests_failover_list
diff --git a/tests/router_policy_test.py b/tests/router_policy_test.py
index 76108c5..f39db54 100644
--- a/tests/router_policy_test.py
+++ b/tests/router_policy_test.py
@@ -22,6 +22,9 @@ from __future__ import division
from __future__ import absolute_import
from __future__ import print_function
+import json
+import sys
+
from system_test import unittest
from qpid_dispatch_internal.policy.policy_util import HostAddr, is_ipv6_enabled
@@ -129,17 +132,27 @@ class MockAgent(object):
class MockPolicyManager(object):
def __init__(self):
self.agent = MockAgent()
+ self.logs = []
def log_debug(self, text):
print("DEBUG: %s" % text)
+ self.logs.append(text)
+
def log_info(self, text):
print("INFO: %s" % text)
+ self.logs.append(text)
+
def log_trace(self, text):
print("TRACE: %s" % text)
+ self.logs.append(text)
+
def log_error(self, text):
print("ERROR: %s" % text)
+ self.logs.append(text)
+
def log_warning(self, text):
print("WARNING: %s" % text)
+ self.logs.append(text)
def get_agent(self):
return self.agent
@@ -333,5 +346,120 @@ class PolicyAppConnectionMgrTests(TestCase):
self.assertTrue(stats.connections_denied == 1)
+class PolicyAliases(TestCase):
+
+ #
+ def test_AliasesRenameOwnVhost(self):
+ config_str="""
+[{
+ "hostname": "$default",
+ "allowUnknownUser": true,
+ "aliases": "$default",
+ "groups": {
+ "$default": {
+ "remoteHosts": "*",
+ "allowDynamicSource": true,
+ "allowAnonymousSender": true,
+ "sources": "$management, examples, q1",
+ "targets": "$management, examples, q1",
+ "maxSessions": 1
+ }
+ }
+}]
+"""
+ manager = MockPolicyManager()
+ policy = PolicyLocal(manager)
+ ruleset = json.loads(config_str)
+ denied = False
+ try:
+ policy.create_ruleset(ruleset[0])
+ except PolicyError:
+ denied = True
+ self.assertTrue(denied, "Ruleset duplicates vhost and alias but condition not detected.")
+
+ #
+ def test_SameAliasOnTwoVhosts(self):
+ config_str="""
+[{
+ "hostname": "$default",
+ "aliases": "a,b,c,d,e",
+ "groups": {
+ "$default": {
+ "maxSessions": 1
+ }
+ }
+},
+{
+ "hostname": "doshormigas",
+ "aliases": "i,h,g,f,e",
+ "groups": {
+ "$default": {
+ "maxSessions": 1
+ }
+ }
+}]
+"""
+ manager = MockPolicyManager()
+ policy = PolicyLocal(manager)
+ ruleset = json.loads(config_str)
+ denied = False
+ try:
+ policy.create_ruleset(ruleset[0])
+ policy.create_ruleset(ruleset[1])
+ except PolicyError as e:
+ denied = True
+ self.assertTrue(denied, "Rulesets duplicate same alias in two vhosts but condition not detected.")
+
+ #
+ def test_AliasConflictsWithVhost(self):
+ config_str="""
+[{
+ "hostname": "$default",
+ "groups": {
+ "$default": {
+ "maxSessions": 1
+ }
+ }
+},
+{
+ "hostname": "conflict-with-vhost",
+ "aliases": "$default",
+ "groups": {
+ "$default": {
+ "maxSessions": 1
+ }
+ }
+}]
+"""
+ manager = MockPolicyManager()
+ policy = PolicyLocal(manager)
+ ruleset = json.loads(config_str)
+ denied = False
+ try:
+ policy.create_ruleset(ruleset[0])
+ policy.create_ruleset(ruleset[1])
+ except PolicyError as e:
+ denied = True
+ self.assertTrue(denied, "Ruleset alias names other vhost but condition not detected.")
+
+ #
+ def test_AliasOperationalLookup(self):
+ manager = MockPolicyManager()
+ policy = PolicyLocal(manager)
+ policy.test_load_config()
+
+ # For this test the test config defines vhost 'photoserver'.
+ # This test accesses that vhost using the alias name 'antialias'.
+ settingsname = policy.lookup_user('zeke', '192.168.100.5', 'antialias', "connid", 5)
+ self.assertTrue(settingsname == 'test')
+
+ upolicy = {}
+ self.assertTrue(
+ policy.lookup_settings('antialias', settingsname, upolicy)
+ )
+ self.assertTrue(upolicy['maxFrameSize'] == 444444)
+ self.assertTrue(upolicy['sources'] == 'a,private,')
+
+
if __name__ == '__main__':
unittest.main(main_module())
diff --git a/tests/system_tests_multi_tenancy_policy.py b/tests/system_tests_multi_tenancy_policy.py
new file mode 100644
index 0000000..4b40bc7
--- /dev/null
+++ b/tests/system_tests_multi_tenancy_policy.py
@@ -0,0 +1,931 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# 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
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+from __future__ import unicode_literals
+from __future__ import division
+from __future__ import absolute_import
+from __future__ import print_function
+
+from proton import Message, Timeout
+from system_test import TestCase, Qdrouterd, main_module, TIMEOUT, unittest, TestTimeout, PollTimeout
+from proton.handlers import MessagingHandler
+from proton.reactor import Container, DynamicNodeProperties
+from qpid_dispatch_internal.compat import UNICODE
+
+
+class RouterMultitenantPolicyTest(TestCase):
+
+ inter_router_port = None
+
+ @classmethod
+ def setUpClass(cls):
+ """Start a router"""
+ super(RouterMultitenantPolicyTest, cls).setUpClass()
+
+ def router(name, connection):
+
+ config = [
+ ('router', {'mode': 'interior', 'id': name}),
+ ('listener', {'port': cls.tester.get_port(), 'stripAnnotations': 'no'}),
+ ('listener', {'port': cls.tester.get_port(), 'stripAnnotations': 'no', 'multiTenant': 'yes'}),
+ ('listener', {'port': cls.tester.get_port(), 'stripAnnotations': 'no', 'role': 'route-container'}),
+ ('linkRoute', {'prefix': 'hosted-group-1/link', 'direction': 'in', 'containerId': 'LRC'}),
+ ('linkRoute', {'prefix': 'hosted-group-1/link', 'direction': 'out', 'containerId': 'LRC'}),
+ ('autoLink', {'address': 'hosted-group-1/queue.waypoint', 'containerId': 'ALC', 'direction': 'in'}),
+ ('autoLink', {'address': 'hosted-group-1/queue.waypoint', 'containerId': 'ALC', 'direction': 'out'}),
+ ('autoLink', {'address': 'hosted-group-1/queue.ext', 'containerId': 'ALCE', 'direction': 'in', 'externalAddress': 'EXT'}),
+ ('autoLink', {'address': 'hosted-group-1/queue.ext', 'containerId': 'ALCE', 'direction': 'out', 'externalAddress': 'EXT'}),
+ ('address', {'prefix': 'closest', 'distribution': 'closest'}),
+ ('address', {'prefix': 'spread', 'distribution': 'balanced'}),
+ ('address', {'prefix': 'multicast', 'distribution': 'multicast'}),
+ ('address', {'prefix': 'hosted-group-1/queue', 'waypoint': 'yes'}),
+ ('policy', {'enableVhostPolicy': 'true'}),
+ ('vhost', {'hostname': 'hosted-group-1',
+ 'allowUnknownUser': 'true',
+ 'aliases': '0.0.0.0',
+ 'groups': {
+ '$default': {
+ 'users': '*',
+ 'maxConnections': 100,
+ 'remoteHosts': '*',
+ 'sources': '*',
+ 'targets': '*',
+ 'allowAnonymousSender': 'true',
+ 'allowWaypointLinks': 'true',
+ 'allowDynamicSource': 'true'
+ }
+ }
+ }),
+ connection
+ ]
+
+ config = Qdrouterd.Config(config)
+
+ cls.routers.append(cls.tester.qdrouterd(name, config, wait=True))
+
+ cls.routers = []
+
+ inter_router_port = cls.tester.get_port()
+
+ router('A', ('listener', {'role': 'inter-router', 'port': inter_router_port}))
+ router('B', ('connector', {'name': 'connectorToA', 'role': 'inter-router', 'port': inter_router_port, 'verifyHostname': 'no'}))
+
+ cls.routers[0].wait_router_connected('B')
+ cls.routers[1].wait_router_connected('A')
+
+
+ def test_01_one_router_targeted_sender_no_tenant(self):
+ test = MessageTransferTest(self.routers[0].addresses[0],
+ self.routers[0].addresses[0],
+ "anything/addr_01",
+ "anything/addr_01",
+ self.routers[0].addresses[0],
+ "M0anything/addr_01")
+ test.run()
+ self.assertEqual(None, test.error)
+
+
+ def test_02_one_router_targeted_sender_tenant_on_sender(self):
+ test = MessageTransferTest(self.routers[0].addresses[1],
+ self.routers[0].addresses[0],
+ "addr_02",
+ "hosted-group-1/addr_02",
+ self.routers[0].addresses[0],
+ "M0hosted-group-1/addr_02")
+ test.run()
+ self.assertEqual(None, test.error)
+
+
+ def test_03_one_router_targeted_sender_tenant_on_receiver(self):
+ test = MessageTransferTest(self.routers[0].addresses[0],
+ self.routers[0].addresses[1],
+ "hosted-group-1/addr_03",
+ "addr_03",
+ self.routers[0].addresses[0],
+ "M0hosted-group-1/addr_03")
+ test.run()
+ self.assertEqual(None, test.error)
+
+
+ def test_04_one_router_targeted_sender_tenant_on_both(self):
+ test = MessageTransferTest(self.routers[0].addresses[1],
+ self.routers[0].addresses[1],
+ "addr_04",
+ "addr_04",
+ self.routers[0].addresses[0],
+ "M0hosted-group-1/addr_04")
+ test.run()
+ self.assertEqual(None, test.error)
+
+
+ def test_05_two_router_targeted_sender_no_tenant(self):
+ test = MessageTransferTest(self.routers[0].addresses[0],
+ self.routers[1].addresses[0],
+ "hosted-group-1/addr_05",
+ "hosted-group-1/addr_05",
+ self.routers[0].addresses[0],
+ "M0hosted-group-1/addr_05")
+ test.run()
+ self.assertEqual(None, test.error)
+
+
+ def test_06_two_router_targeted_sender_tenant_on_sender(self):
+ test = MessageTransferTest(self.routers[0].addresses[1],
+ self.routers[1].addresses[0],
+ "addr_06",
+ "hosted-group-1/addr_06",
+ self.routers[0].addresses[0],
+ "M0hosted-group-1/addr_06")
+ test.run()
+ self.assertEqual(None, test.error)
+
+
+ def test_07_two_router_targeted_sender_tenant_on_receiver(self):
+ test = MessageTransferTest(self.routers[0].addresses[0],
+ self.routers[1].addresses[1],
+ "hosted-group-1/addr_07",
+ "addr_07",
+ self.routers[0].addresses[0],
+ "M0hosted-group-1/addr_07")
+ test.run()
+ self.assertEqual(None, test.error)
+
+
+ def test_08_two_router_targeted_sender_tenant_on_both(self):
+ test = MessageTransferTest(self.routers[0].addresses[1],
+ self.routers[1].addresses[1],
+ "addr_08",
+ "addr_08",
+ self.routers[0].addresses[0],
+ "M0hosted-group-1/addr_08")
+ test.run()
+ self.assertEqual(None, test.error)
+
+
+ def test_09_one_router_anonymous_sender_no_tenant(self):
+ test = MessageTransferAnonTest(self.routers[0].addresses[0],
+ self.routers[0].addresses[0],
+ "anything/addr_09",
+ "anything/addr_09",
+ self.routers[0].addresses[0],
+ "M0anything/addr_09")
+ test.run()
+ self.assertEqual(None, test.error)
+
+
+ def test_10_one_router_anonymous_sender_tenant_on_sender(self):
+ test = MessageTransferAnonTest(self.routers[0].addresses[1],
+ self.routers[0].addresses[0],
+ "addr_10",
+ "hosted-group-1/addr_10",
+ self.routers[0].addresses[0],
+ "M0hosted-group-1/addr_10")
+ test.run()
+ self.assertEqual(None, test.error)
+
+
+ def test_11_one_router_anonymous_sender_tenant_on_receiver(self):
+ test = MessageTransferAnonTest(self.routers[0].addresses[0],
+ self.routers[0].addresses[1],
+ "hosted-group-1/addr_11",
+ "addr_11",
+ self.routers[0].addresses[0],
+ "M0hosted-group-1/addr_11")
+ test.run()
+ self.assertEqual(None, test.error)
+
+
+ def test_12_one_router_anonymous_sender_tenant_on_both(self):
+ test = MessageTransferAnonTest(self.routers[0].addresses[1],
+ self.routers[0].addresses[1],
+ "addr_12",
+ "addr_12",
+ self.routers[0].addresses[0],
+ "M0hosted-group-1/addr_12")
+ test.run()
+ self.assertEqual(None, test.error)
+
+
+ def test_13_two_router_anonymous_sender_no_tenant(self):
+ test = MessageTransferAnonTest(self.routers[0].addresses[0],
+ self.routers[1].addresses[0],
+ "anything/addr_13",
+ "anything/addr_13",
+ self.routers[0].addresses[0],
+ "M0anything/addr_13")
+ test.run()
+ self.assertEqual(None, test.error)
+
+
+ def test_14_two_router_anonymous_sender_tenant_on_sender(self):
+ test = MessageTransferAnonTest(self.routers[0].addresses[1],
+ self.routers[1].addresses[0],
+ "addr_14",
+ "hosted-group-1/addr_14",
+ self.routers[0].addresses[0],
+ "M0hosted-group-1/addr_14")
+ test.run()
+ self.assertEqual(None, test.error)
+
+
+ def test_15_two_router_anonymous_sender_tenant_on_receiver(self):
+ test = MessageTransferAnonTest(self.routers[0].addresses[0],
+ self.routers[1].addresses[1],
+ "hosted-group-1/addr_15",
+ "addr_15",
+ self.routers[0].addresses[0],
+ "M0hosted-group-1/addr_15")
+ test.run()
+ self.assertEqual(None, test.error)
+
+
+ def test_16_two_router_anonymous_sender_tenant_on_both(self):
+ test = MessageTransferAnonTest(self.routers[0].addresses[1],
+ self.routers[1].addresses[1],
+ "addr_16",
+ "addr_16",
+ self.routers[0].addresses[0],
+ "M0hosted-group-1/addr_16")
+ test.run()
+ self.assertEqual(None, test.error)
+
+
+ def test_17_one_router_link_route_targeted(self):
+ test = LinkRouteTest(self.routers[0].addresses[1],
+ self.routers[0].addresses[2],
+ "link.addr_17",
+ "hosted-group-1/link.addr_17",
+ False,
+ self.routers[0].addresses[0])
+ test.run()
+ self.assertEqual(None, test.error)
+
+
+ def test_18_one_router_link_route_targeted_no_tenant(self):
+ test = LinkRouteTest(self.routers[0].addresses[0],
+ self.routers[0].addresses[2],
+ "hosted-group-1/link.addr_18",
+ "hosted-group-1/link.addr_18",
+ False,
+ self.routers[0].addresses[0])
+ test.run()
+ self.assertEqual(None, test.error)
+
+
+ def test_19_one_router_link_route_dynamic(self):
+ test = LinkRouteTest(self.routers[0].addresses[1],
+ self.routers[0].addresses[2],
+ "link.addr_19",
+ "hosted-group-1/link.addr_19",
+ True,
+ self.routers[0].addresses[0])
+ test.run()
+ self.assertEqual(None, test.error)
+
+
+ def test_20_one_router_link_route_dynamic_no_tenant(self):
+ test = LinkRouteTest(self.routers[0].addresses[0],
+ self.routers[0].addresses[2],
+ "hosted-group-1/link.addr_20",
+ "hosted-group-1/link.addr_20",
+ True,
+ self.routers[0].addresses[0])
+ test.run()
+ self.assertEqual(None, test.error)
+
+
+ def test_21_two_router_link_route_targeted(self):
+ test = LinkRouteTest(self.routers[0].addresses[1],
+ self.routers[1].addresses[2],
+ "link.addr_21",
+ "hosted-group-1/link.addr_21",
+ False,
+ self.routers[0].addresses[0])
+ test.run()
+ self.assertEqual(None, test.error)
+
+
+ def test_22_two_router_link_route_targeted_no_tenant(self):
+ test = LinkRouteTest(self.routers[0].addresses[0],
+ self.routers[1].addresses[2],
+ "hosted-group-1/link.addr_22",
+ "hosted-group-1/link.addr_22",
+ False,
+ self.routers[0].addresses[0])
+ test.run()
+ self.assertEqual(None, test.error)
+
+
+ def test_23_two_router_link_route_dynamic(self):
+ test = LinkRouteTest(self.routers[0].addresses[1],
+ self.routers[1].addresses[2],
+ "link.addr_23",
+ "hosted-group-1/link.addr_23",
+ True,
+ self.routers[0].addresses[0])
+ test.run()
+ self.assertEqual(None, test.error)
+
+
+ def test_24_two_router_link_route_dynamic_no_tenant(self):
+ test = LinkRouteTest(self.routers[0].addresses[0],
+ self.routers[1].addresses[2],
+ "hosted-group-1/link.addr_24",
+ "hosted-group-1/link.addr_24",
+ True,
+ self.routers[0].addresses[0])
+ test.run()
+ self.assertEqual(None, test.error)
+
+
+ def test_25_one_router_anonymous_sender_non_mobile(self):
+ test = MessageTransferAnonTest(self.routers[0].addresses[1],
+ self.routers[0].addresses[0],
+ "_local/addr_25",
+ "_local/addr_25",
+ self.routers[0].addresses[0],
+ "Laddr_25")
+ test.run()
+ self.assertEqual(None, test.error)
+
+
+ def test_26_one_router_targeted_sender_non_mobile(self):
+ test = MessageTransferTest(self.routers[0].addresses[1],
+ self.routers[0].addresses[0],
+ "_local/addr_26",
+ "_local/addr_26",
+ self.routers[0].addresses[0],
+ "Laddr_26")
+ test.run()
+ self.assertEqual(None, test.error)
+
+
+ def test_27_two_router_anonymous_sender_non_mobile(self):
+ test = MessageTransferAnonTest(self.routers[0].addresses[1],
+ self.routers[1].addresses[0],
+ "_topo/0/B/addr_27",
+ "_local/addr_27",
+ self.routers[1].addresses[0],
+ "Laddr_27")
+ test.run()
+ self.assertEqual(None, test.error)
+
+
+ def test_28_two_router_targeted_sender_non_mobile(self):
+ test = MessageTransferTest(self.routers[0].addresses[1],
+ self.routers[1].addresses[0],
+ "_topo/0/B/addr_28",
+ "_local/addr_28",
+ self.routers[1].addresses[0],
+ "Laddr_28")
+ test.run()
+ self.assertEqual(None, test.error)
+
+
+ def test_29_one_router_waypoint_no_tenant(self):
+ test = WaypointTest(self.routers[0].addresses[0],
+ self.routers[0].addresses[2],
+ "hosted-group-1/queue.waypoint",
+ "hosted-group-1/queue.waypoint")
+ test.run()
+ self.assertEqual(None, test.error)
+
+
+ def test_30_one_router_waypoint(self):
+ test = WaypointTest(self.routers[0].addresses[1],
+ self.routers[0].addresses[2],
+ "queue.waypoint",
+ "hosted-group-1/queue.waypoint")
+ test.run()
+ self.assertEqual(None, test.error)
+
+
+ def test_31_two_router_waypoint_no_tenant(self):
+ test = WaypointTest(self.routers[0].addresses[0],
+ self.routers[1].addresses[2],
+ "hosted-group-1/queue.waypoint",
+ "hosted-group-1/queue.waypoint")
+ test.run()
+ self.assertEqual(None, test.error)
+
+
+ def test_32_two_router_waypoint(self):
+ test = WaypointTest(self.routers[0].addresses[1],
+ self.routers[1].addresses[2],
+ "queue.waypoint",
+ "hosted-group-1/queue.waypoint")
+ test.run()
+ self.assertEqual(None, test.error)
+
+
+ def test_33_one_router_waypoint_no_tenant_external_addr(self):
+ test = WaypointTest(self.routers[0].addresses[0],
+ self.routers[0].addresses[2],
+ "hosted-group-1/queue.ext",
+ "EXT",
+ "ALCE")
+ test.run()
+ self.assertEqual(None, test.error)
+
+
+ def test_34_one_router_waypoint_external_addr(self):
+ test = WaypointTest(self.routers[0].addresses[1],
+ self.routers[0].addresses[2],
+ "queue.ext",
+ "EXT",
+ "ALCE")
+ test.run()
+ self.assertEqual(None, test.error)
+
+
+ def test_35_two_router_waypoint_no_tenant_external_addr(self):
+ test = WaypointTest(self.routers[0].addresses[0],
+ self.routers[1].addresses[2],
+ "hosted-group-1/queue.ext",
+ "EXT",
+ "ALCE")
+ test.run()
+ self.assertEqual(None, test.error)
+
+
+ def test_36_two_router_waypoint_external_addr(self):
+ test = WaypointTest(self.routers[0].addresses[1],
+ self.routers[1].addresses[2],
+ "queue.ext",
+ "EXT",
+ "ALCE")
+ test.run()
+ self.assertEqual(None, test.error)
+
+
+class Entity(object):
+ def __init__(self, status_code, status_description, attrs):
+ self.status_code = status_code
+ self.status_description = status_description
+ self.attrs = attrs
+
+ def __getattr__(self, key):
+ return self.attrs[key]
+
+
+class RouterProxy(object):
+ def __init__(self, reply_addr):
+ self.reply_addr = reply_addr
+
+ def response(self, msg):
+ ap = msg.properties
+ return Entity(ap['statusCode'], ap['statusDescription'], msg.body)
+
+ def read_address(self, name):
+ ap = {'operation': 'READ', 'type': 'org.apache.qpid.dispatch.router.address', 'name': name}
+ return Message(properties=ap, reply_to=self.reply_addr)
+
+ def query_addresses(self):
+ ap = {'operation': 'QUERY', 'type': 'org.apache.qpid.dispatch.router.address'}
+ return Message(properties=ap, reply_to=self.reply_addr)
+
+
+class MessageTransferTest(MessagingHandler):
+ def __init__(self, sender_host, receiver_host, sender_address, receiver_address, lookup_host, lookup_address):
+ super(MessageTransferTest, self).__init__()
+ self.sender_host = sender_host
+ self.receiver_host = receiver_host
+ self.sender_address = sender_address
+ self.receiver_address = receiver_address
+ self.lookup_host = lookup_host
+ self.lookup_address = lookup_address
+
+ self.sender_conn = None
+ self.receiver_conn = None
+ self.lookup_conn = None
+ self.error = None
+ self.sender = None
+ self.receiver = None
+ self.proxy = None
+
+ self.count = 10
+ self.n_sent = 0
+ self.n_rcvd = 0
+ self.n_accepted = 0
+
+ self.n_receiver_opened = 0
+ self.n_sender_opened = 0
+
+ def timeout(self):
+ self.error = "Timeout Expired: n_sent=%d n_rcvd=%d n_accepted=%d n_receiver_opened=%d n_sender_opened=%d" %\
+ (self.n_sent, self.n_rcvd, self.n_accepted, self.n_receiver_opened, self.n_sender_opened)
+ self.sender_conn.close()
+ self.receiver_conn.close()
+ self.lookup_conn.close()
+
+ def on_start(self, event):
+ self.timer = event.reactor.schedule(TIMEOUT, TestTimeout(self))
+ self.sender_conn = event.container.connect(self.sender_host)
+ self.receiver_conn = event.container.connect(self.receiver_host)
+ self.lookup_conn = event.container.connect(self.lookup_host)
+ self.reply_receiver = event.container.create_receiver(self.lookup_conn, dynamic=True)
+ self.agent_sender = event.container.create_sender(self.lookup_conn, "$management")
+
+ def send(self):
+ while self.sender.credit > 0 and self.n_sent < self.count:
+ self.n_sent += 1
+ m = Message(body="Message %d of %d" % (self.n_sent, self.count))
+ self.sender.send(m)
+
+ def on_link_opened(self, event):
+ if event.receiver:
+ self.n_receiver_opened += 1
+ else:
+ self.n_sender_opened += 1
+
+ if event.receiver == self.reply_receiver:
+ self.proxy = RouterProxy(self.reply_receiver.remote_source.address)
+ self.sender = event.container.create_sender(self.sender_conn, self.sender_address)
+ self.receiver = event.container.create_receiver(self.receiver_conn, self.receiver_address)
+
+ def on_sendable(self, event):
+ if event.sender == self.sender:
+ self.send()
+
+ def on_message(self, event):
+ if event.receiver == self.receiver:
+ self.n_rcvd += 1
+ if event.receiver == self.reply_receiver:
+ response = self.proxy.response(event.message)
+ if response.status_code != 200:
+ self.error = "Unexpected error code from agent: %d - %s" % (response.status_code, response.status_description)
+ if self.n_sent != self.count or self.n_rcvd != self.count:
+ self.error = "Unexpected counts: n_sent=%d n_rcvd=%d n_accepted=%d" % (self.n_sent, self.n_rcvd, self.n_accepted)
+ self.sender_conn.close()
+ self.receiver_conn.close()
+ self.lookup_conn.close()
+ self.timer.cancel()
+
+ def on_accepted(self, event):
+ if event.sender == self.sender:
+ self.n_accepted += 1
+ if self.n_accepted == self.count:
+ request = self.proxy.read_address(self.lookup_address)
+ self.agent_sender.send(request)
+
+ def run(self):
+ Container(self).run()
+
+
+class MessageTransferAnonTest(MessagingHandler):
+ def __init__(self, sender_host, receiver_host, sender_address, receiver_address, lookup_host, lookup_address):
+ super(MessageTransferAnonTest, self).__init__()
+ self.sender_host = sender_host
+ self.receiver_host = receiver_host
+ self.sender_address = sender_address
+ self.receiver_address = receiver_address
+ self.lookup_host = lookup_host
+ self.lookup_address = lookup_address
+
+ self.sender_conn = None
+ self.receiver_conn = None
+ self.lookup_conn = None
+ self.error = None
+ self.sender = None
+ self.receiver = None
+ self.proxy = None
+
+ self.count = 10
+ self.n_sent = 0
+ self.n_rcvd = 0
+ self.n_accepted = 0
+
+ self.n_agent_reads = 0
+ self.n_receiver_opened = 0
+ self.n_sender_opened = 0
+
+ def timeout(self):
+ self.error = "Timeout Expired: n_sent=%d n_rcvd=%d n_accepted=%d n_agent_reads=%d n_receiver_opened=%d n_sender_opened=%d" %\
+ (self.n_sent, self.n_rcvd, self.n_accepted, self.n_agent_reads, self.n_receiver_opened, self.n_sender_opened)
+ self.sender_conn.close()
+ self.receiver_conn.close()
+ self.lookup_conn.close()
+ if self.poll_timer:
+ self.poll_timer.cancel()
+
+ def poll_timeout(self):
+ self.poll()
+
+ def on_start(self, event):
+ self.timer = event.reactor.schedule(TIMEOUT, TestTimeout(self))
+ self.poll_timer = None
+ self.sender_conn = event.container.connect(self.sender_host)
+ self.receiver_conn = event.container.connect(self.receiver_host)
+ self.lookup_conn = event.container.connect(self.lookup_host)
+ self.reply_receiver = event.container.create_receiver(self.lookup_conn, dynamic=True)
+ self.agent_sender = event.container.create_sender(self.lookup_conn, "$management")
+ self.receiver = event.container.create_receiver(self.receiver_conn, self.receiver_address)
+
+ def send(self):
+ while self.sender.credit > 0 and self.n_sent < self.count:
+ self.n_sent += 1
+ m = Message(body="Message %d of %d" % (self.n_sent, self.count))
+ m.address = self.sender_address
+ self.sender.send(m)
+
+ def poll(self):
+ request = self.proxy.read_address(self.lookup_address)
+ self.agent_sender.send(request)
+ self.n_agent_reads += 1
+
+ def on_link_opened(self, event):
+ if event.receiver:
+ self.n_receiver_opened += 1
+ else:
+ self.n_sender_opened += 1
+
+ if event.receiver == self.reply_receiver:
+ self.proxy = RouterProxy(self.reply_receiver.remote_source.address)
+ self.poll()
+
+ def on_sendable(self, event):
+ if event.sender == self.sender:
+ self.send()
+
+ def on_message(self, event):
+ if event.receiver == self.receiver:
+ self.n_rcvd += 1
+
+ if event.receiver == self.reply_receiver:
+ response = self.proxy.response(event.message)
+ if response.status_code == 200 and (response.remoteCount + response.subscriberCount) > 0:
+ self.sender = event.container.create_sender(self.sender_conn, None)
+ if self.poll_timer:
+ self.poll_timer.cancel()
+ self.poll_timer = None
+ else:
+ self.poll_timer = event.reactor.schedule(0.25, PollTimeout(self))
+
+ def on_accepted(self, event):
+ if event.sender == self.sender:
+ self.n_accepted += 1
+ if self.n_accepted == self.count:
+ self.sender_conn.close()
+ self.receiver_conn.close()
+ self.lookup_conn.close()
+ self.timer.cancel()
+
+ def run(self):
+ Container(self).run()
+
+
+class LinkRouteTest(MessagingHandler):
+ def __init__(self, first_host, second_host, first_address, second_address, dynamic, lookup_host):
+ super(LinkRouteTest, self).__init__(prefetch=0)
+ self.first_host = first_host
+ self.second_host = second_host
+ self.first_address = first_address
+ self.second_address = second_address
+ self.dynamic = dynamic
+ self.lookup_host = lookup_host
+
+ self.first_conn = None
+ self.second_conn = None
+ self.error = None
+ self.first_sender = None
+ self.first_receiver = None
+ self.second_sender = None
+ self.second_receiver = None
+ self.poll_timer = None
+
+ self.count = 10
+ self.n_sent = 0
+ self.n_rcvd = 0
+ self.n_settled = 0
+
+ def timeout(self):
+ self.error = "Timeout Expired: n_sent=%d n_rcvd=%d n_settled=%d" % (self.n_sent, self.n_rcvd, self.n_settled)
+ self.first_conn.close()
+ self.second_conn.close()
+ self.lookup_conn.close()
+ if self.poll_timer:
+ self.poll_timer.cancel()
+
+ def poll_timeout(self):
+ self.poll()
+
+ def fail(self, text):
+ self.error = text
+ self.second_conn.close()
+ self.first_conn.close()
+ self.timer.cancel()
+ self.lookup_conn.close()
+ if self.poll_timer:
+ self.poll_timer.cancel()
+
+ def send(self):
+ while self.first_sender.credit > 0 and self.n_sent < self.count:
+ self.n_sent += 1
+ m = Message(body="Message %d of %d" % (self.n_sent, self.count))
+ self.first_sender.send(m)
+
+ def poll(self):
+ request = self.proxy.read_address("Dhosted-group-1/link")
+ self.agent_sender.send(request)
+
+ def setup_first_links(self, event):
+ self.first_sender = event.container.create_sender(self.first_conn, self.first_address)
+ if self.dynamic:
+ self.first_receiver = event.container.create_receiver(self.first_conn,
+ dynamic=True,
+ options=DynamicNodeProperties({"x-opt-qd.address":
+ UNICODE(self.first_address)}))
+ else:
+ self.first_receiver = event.container.create_receiver(self.first_conn, self.first_address)
+
+
+ def on_start(self, event):
+ self.timer = event.reactor.schedule(TIMEOUT, TestTimeout(self))
+ self.first_conn = event.container.connect(self.first_host)
+ self.second_conn = event.container.connect(self.second_host)
+ self.lookup_conn = event.container.connect(self.lookup_host)
+ self.reply_receiver = event.container.create_receiver(self.lookup_conn, dynamic=True)
+ self.agent_sender = event.container.create_sender(self.lookup_conn, "$management")
+
+
+ def on_link_opening(self, event):
+ if event.sender:
+ self.second_sender = event.sender
+ if self.dynamic:
+ if event.sender.remote_source.dynamic:
+ event.sender.source.address = self.second_address
+ event.sender.open()
+ else:
+ self.fail("Expected dynamic source on sender")
+ else:
+ if event.sender.remote_source.address == self.second_address:
+ event.sender.source.address = self.second_address
+ event.sender.open()
+ else:
+ self.fail("Incorrect address on incoming sender: got %s, expected %s" %
+ (event.sender.remote_source.address, self.second_address))
+
+ elif event.receiver:
+ self.second_receiver = event.receiver
+ if event.receiver.remote_target.address == self.second_address:
+ event.receiver.target.address = self.second_address
+ event.receiver.open()
+ else:
+ self.fail("Incorrect address on incoming receiver: got %s, expected %s" %
+ (event.receiver.remote_target.address, self.second_address))
+
+
+ def on_link_opened(self, event):
+ if event.receiver:
+ event.receiver.flow(self.count)
+
+ if event.receiver == self.reply_receiver:
+ self.proxy = RouterProxy(self.reply_receiver.remote_source.address)
+ self.poll()
+
+ def on_sendable(self, event):
+ if event.sender == self.first_sender:
+ self.send()
+
+ def on_message(self, event):
+ if event.receiver == self.first_receiver:
+ self.n_rcvd += 1
+
+ if event.receiver == self.reply_receiver:
+ response = self.proxy.response(event.message)
+ if response.status_code == 200 and (response.remoteCount + response.containerCount) > 0:
+ if self.poll_timer:
+ self.poll_timer.cancel()
+ self.poll_timer = None
+ self.setup_first_links(event)
+ else:
+ self.poll_timer = event.reactor.schedule(0.25, PollTimeout(self))
+
+ def on_settled(self, event):
+ if event.sender == self.first_sender:
+ self.n_settled += 1
+ if self.n_settled == self.count:
+ self.fail(None)
+
+ def run(self):
+ container = Container(self)
+ container.container_id = 'LRC'
+ container.run()
+
+
+class WaypointTest(MessagingHandler):
+ def __init__(self, first_host, second_host, first_address, second_address, container_id="ALC"):
+ super(WaypointTest, self).__init__()
+ self.first_host = first_host
+ self.second_host = second_host
+ self.first_address = first_address
+ self.second_address = second_address
+ self.container_id = container_id
+
+ self.first_conn = None
+ self.second_conn = None
+ self.error = None
+ self.first_sender = None
+ self.first_receiver = None
+ self.waypoint_sender = None
+ self.waypoint_receiver = None
+ self.waypoint_queue = []
+
+ self.count = 10
+ self.n_sent = 0
+ self.n_rcvd = 0
+ self.n_thru = 0
+
+ def timeout(self):
+ self.error = "Timeout Expired: n_sent=%d n_rcvd=%d n_thru=%d" % (self.n_sent, self.n_rcvd, self.n_thru)
+ self.first_conn.close()
+ self.second_conn.close()
+
+ def fail(self, text):
+ self.error = text
+ self.second_conn.close()
+ self.first_conn.close()
+ self.timer.cancel()
+
+ def send_client(self):
+ while self.first_sender.credit > 0 and self.n_sent < self.count:
+ self.n_sent += 1
+ m = Message(body="Message %d of %d" % (self.n_sent, self.count))
+ self.first_sender.send(m)
+
+ def send_waypoint(self):
+ while self.waypoint_sender.credit > 0 and len(self.waypoint_queue) > 0:
+ self.n_thru += 1
+ m = self.waypoint_queue.pop()
+ self.waypoint_sender.send(m)
+
+ def on_start(self, event):
+ self.timer = event.reactor.schedule(TIMEOUT, TestTimeout(self))
+ self.first_conn = event.container.connect(self.first_host)
+ self.second_conn = event.container.connect(self.second_host)
+
+ def on_connection_opened(self, event):
+ if event.connection == self.first_conn:
+ self.first_sender = event.container.create_sender(self.first_conn, self.first_address)
+ self.first_receiver = event.container.create_receiver(self.first_conn, self.first_address)
+
+ def on_link_opening(self, event):
+ if event.sender:
+ self.waypoint_sender = event.sender
+ if event.sender.remote_source.address == self.second_address:
+ event.sender.source.address = self.second_address
+ event.sender.open()
+ else:
+ self.fail("Incorrect address on incoming sender: got %s, expected %s" %
+ (event.sender.remote_source.address, self.second_address))
+
+ elif event.receiver:
+ self.waypoint_receiver = event.receiver
+ if event.receiver.remote_target.address == self.second_address:
+ event.receiver.target.address = self.second_address
+ event.receiver.open()
+ else:
+ self.fail("Incorrect address on incoming receiver: got %s, expected %s" %
+ (event.receiver.remote_target.address, self.second_address))
+
+
+ def on_sendable(self, event):
+ if event.sender == self.first_sender:
+ self.send_client()
+ elif event.sender == self.waypoint_sender:
+ self.send_waypoint()
+
+ def on_message(self, event):
+ if event.receiver == self.first_receiver:
+ self.n_rcvd += 1
+ if self.n_rcvd == self.count and self.n_thru == self.count:
+ self.fail(None)
+ elif event.receiver == self.waypoint_receiver:
+ m = Message(body=event.message.body)
+ self.waypoint_queue.append(m)
+ self.send_waypoint()
+
+ def run(self):
+ container = Container(self)
+ container.container_id = self.container_id
+ container.run()
+
+
+if __name__ == '__main__':
+ unittest.main(main_module())
diff --git a/tests/system_tests_policy.py b/tests/system_tests_policy.py
index df606c6..ca8a57d 100644
--- a/tests/system_tests_policy.py
+++ b/tests/system_tests_policy.py
@@ -1858,5 +1858,139 @@ class VhostPolicyConfigHashPattern(TestCase):
self.assertEqual(False, VhostPolicyConfigHashPattern.timed_out)
+class PolicyConnectionAliasTest(MessagingHandler):
+ """
+ This test tries to send an AMQP Open with a selectable hostname.
+ The hostname is expected to be an alias for a vhost. When the alias selects
+ the vhost then the connection is allowed.
+ """
+ def __init__(self, test_host, target_hostname, send_address, print_to_console=False):
+ super(PolicyConnectionAliasTest, self).__init__()
+ self.test_host = test_host # router listener
+ self.target_hostname = target_hostname # vhost name for AMQP Open
+ self.send_address = send_address # dummy address allowed by policy
+
+ self.test_conn = None
+ self.dummy_sender = None
+ self.dummy_receiver = None
+ self.error = None
+ self.shut_down = False
+ self.connection_open_seen = False
+
+ self.logger = Logger(title=("PolicyConnectionAliasTest - use virtual_host '%s'" % (self.target_hostname)), print_to_console=print_to_console)
+ self.log_unhandled = False
+
+ def timeout(self):
+ self.error = "Timeout Expired"
+ self.logger.log("self.timeout " + self.error)
+ self._shut_down_test()
+
+ def on_start(self, event):
+ self.logger.log("on_start")
+ self.timer = event.reactor.schedule(TIMEOUT, TestTimeout(self))
+ self.test_conn = event.container.connect(self.test_host.addresses[0],
+ virtual_host=self.target_hostname)
+ self.logger.log("on_start: done")
+
+ def on_connection_opened(self, event):
+ # This happens even if the connection is rejected.
+ # If the connection is rejected then it is immediately closed.
+ # Create a sender and receiver.
+ # If the sender gets on_sendable then the connection stayed up as expected.
+ self.logger.log("on_connection_opened")
+ self.connection_open_seen = True
+ self.dummy_sender = event.container.create_sender(self.test_conn, self.send_address)
+ self.dummy_receiver = event.container.create_receiver(self.test_conn, self.send_address)
+
+ def on_sendable(self, event):
+ # Success
+ self.logger.log("on_sendable: test is a success")
+ self._shut_down_test()
+
+ def on_connection_remote_close(self, event):
+ self.logger.log("on_connection_remote_close")
+ if self.connection_open_seen and not self.shut_down:
+ self.error = "Policy enforcement fail: expected connection was denied."
+ self._shut_down_test()
+
+ def on_unhandled(self, method, *args):
+ pass # self.logger.log("on_unhandled %s" % (method))
+
+ def _shut_down_test(self):
+ self.shut_down = True
+ if self.timer:
+ self.timer.cancel()
+ self.timer = None
+ if self.test_conn:
+ self.test_conn.close()
+ self.test_conn = None
+
+ def run(self):
+ try:
+ Container(self).run()
+ except Exception as e:
+ self.error = "Container run exception: %s" % (e)
+ self.logger.log(self.error)
+ self.logger.dump()
+
+
+class PolicyVhostAlias(TestCase):
+ """
+ Verify vhost aliases.
+ * A policy defines vhost A with alias B.
+ * A client opens a connection with hostname B in the AMQP Open.
+ * The test expects the connection to succeed using vhost A policy settings.
+ """
+ @classmethod
+ def setUpClass(cls):
+ """Start the router"""
+ super(PolicyVhostAlias, cls).setUpClass()
+
+ def router(name, mode, extra=None):
+ config = [
+ ('router', {'mode': mode,
+ 'id': name}),
+ ('listener', {'role': 'normal',
+ 'port': cls.tester.get_port()}),
+ ('policy', {'enableVhostPolicy': 'true'}),
+ ('vhost', {'hostname': 'A',
+ 'allowUnknownUser': 'true',
+ 'aliases': 'B',
+ 'groups': {
+ '$default': {
+ 'users': '*',
+ 'maxConnections': 100,
+ 'remoteHosts': '*',
+ 'sources': '*',
+ 'targets': '*',
+ 'allowAnonymousSender': 'true',
+ 'allowWaypointLinks': 'true',
+ 'allowDynamicSource': 'true'
+ }
+ }
+ })
+ ]
+
+ config = Qdrouterd.Config(config)
+ cls.routers.append(cls.tester.qdrouterd(name, config, wait=True))
+ return cls.routers[-1]
+
+ cls.routers = []
+
+ router('A', 'interior')
+ cls.INT_A = cls.routers[0]
+ cls.INT_A.listener = cls.INT_A.addresses[0]
+
+ def test_100_policy_aliases(self):
+ test = PolicyConnectionAliasTest(PolicyVhostAlias.INT_A,
+ "B",
+ "address-B")
+ test.run()
+ if test.error is not None:
+ test.logger.log("test_100 test error: %s" % (test.error))
+ test.logger.dump()
+ self.assertTrue(test.error is None)
+
+
if __name__ == '__main__':
unittest.main(main_module())
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org