# # 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 # """Entity implementing the business logic of user connection/access policy.""" import json from .policy_util import PolicyError, HostStruct, HostAddr, PolicyAppConnectionMgr, is_ipv6_enabled class PolicyKeys: """ String constants """ # Common key words KW_IGNORED_NAME = "name" KW_IGNORED_IDENTITY = "identity" KW_IGNORED_TYPE = "type" KW_VHOST_NAME = "hostname" KW_VHOST_DEPRECATED_ID = "id" # Policy ruleset key words KW_MAXCONN = "maxConnections" KW_MAXCONNPERHOST = "maxConnectionsPerHost" KW_MAXCONNPERUSER = "maxConnectionsPerUser" KW_CONNECTION_ALLOW_DEFAULT = "allowUnknownUser" KW_GROUPS = "groups" # Policy settings key words KW_USERS = "users" KW_REMOTE_HOSTS = "remoteHosts" KW_MAX_FRAME_SIZE = "maxFrameSize" KW_MAX_MESSAGE_SIZE = "maxMessageSize" KW_MAX_SESSION_WINDOW = "maxSessionWindow" KW_MAX_SESSIONS = "maxSessions" KW_MAX_SENDERS = "maxSenders" KW_MAX_RECEIVERS = "maxReceivers" KW_ALLOW_DYNAMIC_SRC = "allowDynamicSource" KW_ALLOW_ANONYMOUS_SENDER = "allowAnonymousSender" KW_ALLOW_USERID_PROXY = "allowUserIdProxy" KW_ALLOW_WAYPOINT_LINKS = "allowWaypointLinks" KW_ALLOW_FALLBACK_LINKS = "allowFallbackLinks" KW_ALLOW_DYNAMIC_LINK_ROUTES = "allowDynamicLinkRoutes" KW_ALLOW_ADMIN_STATUS_UPDATE = "allowAdminStatusUpdate" KW_SOURCES = "sources" KW_TARGETS = "targets" KW_SOURCE_PATTERN = "sourcePattern" KW_TARGET_PATTERN = "targetPattern" KW_VHOST_ALIASES = "aliases" # Policy stats key words KW_CONNECTIONS_APPROVED = "connectionsApproved" KW_CONNECTIONS_DENIED = "connectionsDenied" KW_CONNECTIONS_CURRENT = "connectionsCurrent" KW_LINKS_DENIED = "linksDenied" KW_TOTAL_DENIALS = "totalDenials" KW_PER_USER_STATE = "perUserState" KW_PER_HOST_STATE = "perHostState" # What settings does a user get when allowed to connect but # not restricted by a user group? KW_DEFAULT_SETTINGS = "$default" # Config file separator character for two IP addresses in a range KC_CONFIG_IP_SEP = "-" # Config file separator character for names in a list KC_CONFIG_LIST_SEP = "," # user-to-group computed map in compiled ruleset RULESET_U2G_MAP = "U2G" # policy stats controlled by C code but referenced by settings KW_CSTATS = "denialCounts" # Username subsitituion token in link source and target names and patterns KC_TOKEN_USER = "${user}" # Link target/source name wildcard tuple keys KC_TUPLE_ABSENT = 'a' KC_TUPLE_PREFIX = 'p' KC_TUPLE_SUFFIX = 's' KC_TUPLE_EMBED = 'e' KC_TUPLE_WILDCARD = '*' # # class PolicyCompiler: """ Validate incoming configuration for legal schema. - Warn about section options that go unused. - Disallow negative max connection/message size numbers. - Check that connectionOrigins resolve to IP hosts. - Enforce internal consistency, """ allowed_ruleset_options = [ PolicyKeys.KW_IGNORED_NAME, PolicyKeys.KW_IGNORED_IDENTITY, PolicyKeys.KW_IGNORED_TYPE, PolicyKeys.KW_VHOST_NAME, PolicyKeys.KW_MAXCONN, PolicyKeys.KW_MAX_MESSAGE_SIZE, PolicyKeys.KW_MAXCONNPERHOST, PolicyKeys.KW_MAXCONNPERUSER, PolicyKeys.KW_CONNECTION_ALLOW_DEFAULT, PolicyKeys.KW_GROUPS, PolicyKeys.KW_VHOST_ALIASES ] allowed_settings_options = [ PolicyKeys.KW_USERS, PolicyKeys.KW_REMOTE_HOSTS, PolicyKeys.KW_MAXCONNPERHOST, PolicyKeys.KW_MAXCONNPERUSER, PolicyKeys.KW_MAX_FRAME_SIZE, PolicyKeys.KW_MAX_MESSAGE_SIZE, PolicyKeys.KW_MAX_SESSION_WINDOW, PolicyKeys.KW_MAX_SESSIONS, PolicyKeys.KW_MAX_SENDERS, PolicyKeys.KW_MAX_RECEIVERS, PolicyKeys.KW_ALLOW_DYNAMIC_SRC, PolicyKeys.KW_ALLOW_ANONYMOUS_SENDER, PolicyKeys.KW_ALLOW_USERID_PROXY, PolicyKeys.KW_ALLOW_WAYPOINT_LINKS, PolicyKeys.KW_ALLOW_FALLBACK_LINKS, PolicyKeys.KW_ALLOW_DYNAMIC_LINK_ROUTES, PolicyKeys.KW_ALLOW_ADMIN_STATUS_UPDATE, PolicyKeys.KW_SOURCES, PolicyKeys.KW_TARGETS, PolicyKeys.KW_SOURCE_PATTERN, PolicyKeys.KW_TARGET_PATTERN ] def __init__(self): """ Create a validator """ pass def validateNumber(self, val, v_min, v_max, errors): """ Range check a numeric int policy value @param[in] val policy value to check @param[in] v_min minumum value @param[in] v_max maximum value. zero disables check @param[out] errors failure message @return v_min <= val <= v_max """ try: v_int = int(val) except Exception as e: errors.append("Value '%s' does not resolve to an integer." % val) return False if v_int < v_min: errors.append("Value '%s' is below minimum '%s'." % (val, v_min)) return False if 0 < v_max < v_int: errors.append("Value '%s' is above maximum '%s'." % (val, v_max)) return False return True def compile_connection_group(self, vhostname, groupname, val, list_out, warnings, errors): """ Handle an ingressHostGroups submap. Each origin value is verified. On a successful run the submap is replaced parsed lists of HostAddr objects. @param[in] vhostname vhost name @param[in] groupname vhost/group name @param[in] val origin list as text string @param[out] list_out user inputs replaced with HostAddr objects @param[out] warnings nonfatal irregularities observed @param[out] errors descriptions of failure @return - origins is usable. If True then warnings[] may contain useful information about fields that are ignored. If False then warnings[] may contain info and errors[0] will hold the description of why the origin was rejected. """ key = PolicyKeys.KW_REMOTE_HOSTS # convert val string to list of host specs if isinstance(val, list): # ['abc', 'def', 'mytarget'] pass elif isinstance(val, str): val = [x.strip(' ') for x in val.split(PolicyKeys.KC_CONFIG_LIST_SEP)] else: errors.append( "Policy vhost '%s' user group '%s' option '%s' has illegal value '%s'. Type must be 'str' or 'list' but is '%s;" % (vhostname, groupname, key, val, type(val))) return False for coname in val: try: coha = HostAddr(coname, PolicyKeys.KC_CONFIG_IP_SEP) list_out.append(coha) except Exception as e: errors.append("Policy vhost '%s' user group '%s' option '%s' connectionOption '%s' failed to translate: '%s'." % (vhostname, groupname, key, coname, e)) return False return True def compile_app_settings(self, vhostname, usergroup, policy_in, policy_out, warnings, errors): """ Compile a vhostUserGroupSettings schema from processed json format to local internal format. @param[in] name vhost name @param[in] policy_in user config settings @param[out] policy_out validated Internal format @param[out] warnings nonfatal irregularities observed @param[out] errors descriptions of failure @return - settings are usable. If True then warnings[] may contain useful information about fields that are ignored. If False then warnings[] may contain info and errors[0] will hold the description of why the policy was rejected. """ # rulesets may not come through standard config so make nice defaults policy_out[PolicyKeys.KW_USERS] = '' policy_out[PolicyKeys.KW_REMOTE_HOSTS] = '' # DISPATCH-2305: do not provide default values for max # frame/window/sessions. The router already provides these. Setting # zero here will cause the router to use configured values unless # specifically overridden by policy: policy_out[PolicyKeys.KW_MAX_FRAME_SIZE] = 0 policy_out[PolicyKeys.KW_MAX_SESSION_WINDOW] = 0 policy_out[PolicyKeys.KW_MAX_SESSIONS] = 0 policy_out[PolicyKeys.KW_MAX_MESSAGE_SIZE] = None policy_out[PolicyKeys.KW_MAX_SENDERS] = 2147483647 policy_out[PolicyKeys.KW_MAX_RECEIVERS] = 2147483647 policy_out[PolicyKeys.KW_ALLOW_DYNAMIC_SRC] = False policy_out[PolicyKeys.KW_ALLOW_ANONYMOUS_SENDER] = False policy_out[PolicyKeys.KW_ALLOW_USERID_PROXY] = False policy_out[PolicyKeys.KW_ALLOW_WAYPOINT_LINKS] = True policy_out[PolicyKeys.KW_ALLOW_FALLBACK_LINKS] = True policy_out[PolicyKeys.KW_ALLOW_DYNAMIC_LINK_ROUTES] = True policy_out[PolicyKeys.KW_ALLOW_ADMIN_STATUS_UPDATE] = True policy_out[PolicyKeys.KW_SOURCES] = '' policy_out[PolicyKeys.KW_TARGETS] = '' policy_out[PolicyKeys.KW_SOURCE_PATTERN] = '' policy_out[PolicyKeys.KW_TARGET_PATTERN] = '' policy_out[PolicyKeys.KW_MAXCONNPERHOST] = None # optional group limit policy_out[PolicyKeys.KW_MAXCONNPERUSER] = None cerror = [] user_sources = False user_targets = False user_src_pattern = False user_tgt_pattern = False for key, val in policy_in.items(): if key not in self.allowed_settings_options: warnings.append("Policy vhost '%s' user group '%s' option '%s' is ignored." % (vhostname, usergroup, key)) if key in [PolicyKeys.KW_MAXCONNPERHOST, PolicyKeys.KW_MAXCONNPERUSER ]: if not self.validateNumber(val, 0, 65535, cerror): msg = ("Policy vhost '%s' user group '%s' option '%s' has error '%s'." % (vhostname, usergroup, key, cerror[0])) errors.append(msg) return False policy_out[key] = int(val) elif key in [PolicyKeys.KW_MAX_FRAME_SIZE, PolicyKeys.KW_MAX_MESSAGE_SIZE, PolicyKeys.KW_MAX_RECEIVERS, PolicyKeys.KW_MAX_SENDERS, PolicyKeys.KW_MAX_SESSION_WINDOW, PolicyKeys.KW_MAX_SESSIONS ]: if not self.validateNumber(val, 0, 0, cerror): errors.append("Policy vhost '%s' user group '%s' option '%s' has error '%s'." % (vhostname, usergroup, key, cerror[0])) return False policy_out[key] = int(val) elif key == PolicyKeys.KW_REMOTE_HOSTS: # Conection groups are lists of IP addresses that need to be # converted into binary structures for comparisons. val_out = [] if not self.compile_connection_group(vhostname, usergroup, val, val_out, warnings, errors): return False policy_out[key] = val_out elif key in [PolicyKeys.KW_ALLOW_ANONYMOUS_SENDER, PolicyKeys.KW_ALLOW_DYNAMIC_SRC, PolicyKeys.KW_ALLOW_USERID_PROXY, PolicyKeys.KW_ALLOW_WAYPOINT_LINKS, PolicyKeys.KW_ALLOW_FALLBACK_LINKS, PolicyKeys.KW_ALLOW_DYNAMIC_LINK_ROUTES, PolicyKeys.KW_ALLOW_ADMIN_STATUS_UPDATE ]: if isinstance(val, str) and val.lower() in ['true', 'false']: val = val == 'true' if not isinstance(val, bool): errors.append("Policy vhost '%s' user group '%s' option '%s' has illegal boolean value '%s'." % (vhostname, usergroup, key, val)) return False policy_out[key] = val elif key in [PolicyKeys.KW_USERS, PolicyKeys.KW_SOURCES, PolicyKeys.KW_TARGETS, PolicyKeys.KW_SOURCE_PATTERN, PolicyKeys.KW_TARGET_PATTERN, PolicyKeys.KW_VHOST_ALIASES ]: # accept a string or list if isinstance(val, list): # ['abc', 'def', 'mytarget'] pass elif isinstance(val, str): # 'abc, def, mytarget' val = [x.strip(' ') for x in val.split(PolicyKeys.KC_CONFIG_LIST_SEP)] else: errors.append("Policy vhost '%s' user group '%s' option '%s' has illegal value '%s'. Type must be 'str' or 'list' but is '%s;" % (vhostname, usergroup, key, val, type(val))) # deduplicate address lists val = list(set(val)) # val is CSV string with no white space between values: 'abc,def,mytarget,tmp-${user}' if key == PolicyKeys.KW_USERS: # user name list items are literal strings and need no special handling policy_out[key] = ','.join(val) else: # source and target names get special handling for the '${user}' substitution token # The literal string is translated to a (key, prefix, suffix) set of three strings. # C code does not have to search for the username token and knows with authority # how to construct match strings. # A wildcard is also signaled. utoken = PolicyKeys.KC_TOKEN_USER eVal = [] for v in val: vcount = v.count(utoken) if vcount > 1: errors.append("Policy vhost '%s' user group '%s' policy key '%s' item '%s' contains multiple user subtitution tokens" % (vhostname, usergroup, key, v)) return False elif vcount == 1: # a single token is present as a prefix, suffix, or embedded # construct cChar, S1, S2 encodings to be added to eVal description if v.startswith(utoken): # prefix eVal.append(PolicyKeys.KC_TUPLE_PREFIX) eVal.append('') eVal.append(v[v.find(utoken) + len(utoken):]) elif v.endswith(utoken): # suffix eVal.append(PolicyKeys.KC_TUPLE_SUFFIX) eVal.append(v[0:v.find(utoken)]) eVal.append('') else: # embedded if key in [PolicyKeys.KW_SOURCE_PATTERN, PolicyKeys.KW_TARGET_PATTERN]: errors.append("Policy vhost '%s' user group '%s' policy key '%s' item '%s' may contain match pattern '%s' as a prefix or a suffix only." % (vhostname, usergroup, key, v, utoken)) return False eVal.append(PolicyKeys.KC_TUPLE_EMBED) eVal.append(v[0:v.find(utoken)]) eVal.append(v[v.find(utoken) + len(utoken):]) else: # ${user} token is absent if v == PolicyKeys.KC_TUPLE_WILDCARD: eVal.append(PolicyKeys.KC_TUPLE_WILDCARD) eVal.append('') eVal.append('') else: eVal.append(PolicyKeys.KC_TUPLE_ABSENT) eVal.append(v) eVal.append('') policy_out[key] = ','.join(eVal) if key == PolicyKeys.KW_SOURCES: user_sources = True if key == PolicyKeys.KW_TARGETS: user_targets = True if key == PolicyKeys.KW_SOURCE_PATTERN: user_src_pattern = True if key == PolicyKeys.KW_TARGET_PATTERN: user_tgt_pattern = True if user_sources and user_src_pattern: errors.append("Policy vhost '%s' user group '%s' specifies conflicting 'sources' and 'sourcePattern' attributes. Use only one or the other." % (vhostname, usergroup)) return False if user_targets and user_tgt_pattern: errors.append("Policy vhost '%s' user group '%s' specifies conflicting 'targets' and 'targetPattern' attributes. Use only one or the other." % (vhostname, usergroup)) return False return True def compile_access_ruleset(self, name, policy_in, policy_out, warnings, errors): """ Compile a vhost schema from processed json format to local internal format. @param[in] name vhost name @param[in] policy_in raw policy to be validated @param[out] policy_out validated Internal format @param[out] warnings nonfatal irregularities observed @param[out] errors descriptions of failure @return - policy is usable. If True then warnings[] may contain useful information about fields that are ignored. If False then warnings[] may contain info and errors[0] will hold the description of why the policy was rejected. """ cerror = [] # rulesets may not come through standard config so make nice defaults policy_out[PolicyKeys.KW_MAXCONN] = 65535 policy_out[PolicyKeys.KW_MAXCONNPERHOST] = 65535 policy_out[PolicyKeys.KW_MAXCONNPERUSER] = 65535 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 policy_in.items(): if key not in self.allowed_ruleset_options: warnings.append("Policy vhost '%s' option '%s' is ignored." % (name, key)) if key in [PolicyKeys.KW_MAXCONN, PolicyKeys.KW_MAXCONNPERHOST, PolicyKeys.KW_MAXCONNPERUSER ]: if not self.validateNumber(val, 0, 65535, cerror): msg = ("Policy vhost '%s' option '%s' has error '%s'." % (name, key, cerror[0])) errors.append(msg) return False policy_out[key] = val elif key in [PolicyKeys.KW_MAX_MESSAGE_SIZE ]: if not self.validateNumber(val, 0, 0, cerror): msg = ("Policy vhost '%s' option '%s' has error '%s'." % (name, key, cerror[0])) errors.append(msg) return False policy_out[key] = val elif key in [PolicyKeys.KW_CONNECTION_ALLOW_DEFAULT]: if not isinstance(val, bool): errors.append("Policy vhost '%s' option '%s' must be of type 'bool' but is '%s'" % (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 isinstance(val, dict): errors.append("Policy vhost '%s' option '%s' must be of type 'dict' but is '%s'" % (name, key, type(val))) return False for skey, sval in val.items(): newsettings = {} if not self.compile_app_settings(name, skey, sval, newsettings, warnings, errors): return False policy_out[key][skey] = {} policy_out[key][skey].update(newsettings) # Verify that each user is in only one group. # Create user-to-group map for looking up user's group policy_out[PolicyKeys.RULESET_U2G_MAP] = {} if PolicyKeys.KW_GROUPS in policy_out: for group, groupsettings in policy_out[PolicyKeys.KW_GROUPS].items(): if PolicyKeys.KW_USERS in groupsettings: users = [x.strip(' ') for x in groupsettings[PolicyKeys.KW_USERS].split(PolicyKeys.KC_CONFIG_LIST_SEP)] for user in users: if user in policy_out[PolicyKeys.RULESET_U2G_MAP]: errors.append("Policy vhost '%s' user '%s' is in multiple user groups '%s' and '%s'" % (name, user, policy_out[PolicyKeys.RULESET_U2G_MAP][user], group)) return False else: policy_out[PolicyKeys.RULESET_U2G_MAP][user] = group else: warnings.append("Policy vhost '%s' user group '%s' has no defined users. This policy has no effect" % (name, group)) # Default connections require a default settings if policy_out[PolicyKeys.KW_CONNECTION_ALLOW_DEFAULT]: if PolicyKeys.KW_DEFAULT_SETTINGS not in policy_out[PolicyKeys.KW_GROUPS]: errors.append("Policy vhost '%s' allows connections by default but default settings are not defined" % (name)) return False return True # # class AppStats: """ Maintain live state and statistics for an vhost. """ def __init__(self, id, manager, ruleset): self.my_id = id self._manager = manager self.conn_mgr = PolicyAppConnectionMgr( ruleset[PolicyKeys.KW_MAXCONN], ruleset[PolicyKeys.KW_MAXCONNPERUSER], ruleset[PolicyKeys.KW_MAXCONNPERHOST]) self._cstats = self._manager.get_agent().qd.qd_dispatch_policy_c_counts_alloc() self._manager.get_agent().add_implementation(self, "vhostStats") def update_ruleset(self, ruleset): """ The parent ruleset has changed. Propagate settings into the connection manager. @param ruleset: new ruleset @return: """ self.conn_mgr.update( ruleset[PolicyKeys.KW_MAXCONN], ruleset[PolicyKeys.KW_MAXCONNPERHOST], ruleset[PolicyKeys.KW_MAXCONNPERUSER]) def refresh_entity(self, attributes): """Refresh management attributes""" entitymap = {} entitymap[PolicyKeys.KW_VHOST_NAME] = self.my_id entitymap[PolicyKeys.KW_VHOST_DEPRECATED_ID] = self.my_id entitymap[PolicyKeys.KW_CONNECTIONS_APPROVED] = self.conn_mgr.connections_approved entitymap[PolicyKeys.KW_CONNECTIONS_DENIED] = self.conn_mgr.connections_denied entitymap[PolicyKeys.KW_CONNECTIONS_CURRENT] = self.conn_mgr.connections_active entitymap[PolicyKeys.KW_PER_USER_STATE] = self.conn_mgr.per_user_state entitymap[PolicyKeys.KW_PER_HOST_STATE] = self.conn_mgr.per_host_state self._manager.get_agent().qd.qd_dispatch_policy_c_counts_refresh(self._cstats, entitymap) attributes.update(entitymap) def can_connect(self, conn_id, user, host, diags, group_max_conn_user, group_max_conn_host): return self.conn_mgr.can_connect(conn_id, user, host, diags, group_max_conn_user, group_max_conn_host) def disconnect(self, conn_id, user, host): self.conn_mgr.disconnect(conn_id, user, host) def count_other_denial(self): self.conn_mgr.count_other_denial() def get_cstats(self): return self._cstats # # class ConnectionFacts: def __init__(self, user, host, app, conn_name): self.user = user self.host = host self.app = app self.conn_name = conn_name # # class PolicyLocal: """ The local policy database. """ def __init__(self, manager): """ Create instance @params manager policy manager class """ # manager is a class # It provides access the dispatch system functions self._manager = manager # rulesetdb is a map # key : vhost name # val : ruleset for this app # created by configuration # augmented by policy compiler self.rulesetdb = {} # settingsdb is a map # key : # val : a map # key : # val : settings to use for user's connection # created by configuration self.settingsdb = {} # statsdb is a map # key : # val : AppStats object self.statsdb = {} # _policy_compiler is a function # validates incoming policy and readies it for internal use self._policy_compiler = PolicyCompiler() # _connections is a map # key : numeric connection id # val : ConnectionFacts # Entries created as connection AMQP Opens arrive # Entries destroyed as sockets closed self._connections = {} # _default_vhost is a string # holds the name of the vhost to use when the # open.hostname is not found in the rulesetdb self._default_vhost = "" # _use_hostname_patterns # holds policy setting. # When true policy ruleset definitions are propagated to C code self.use_hostname_patterns = False # _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 # def create_ruleset(self, attributes): """ Create or update named policy ruleset. @param[in] attributes: from config """ warnings = [] diag = [] candidate = {} name = attributes[PolicyKeys.KW_VHOST_NAME] result = self._policy_compiler.compile_access_ruleset(name, attributes, candidate, warnings, diag) if not result: raise PolicyError("Policy '%s' is invalid: %s" % (name, diag[0])) 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() # 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 self.rulesetdb[name] = {} self.rulesetdb[name].update(candidate) def policy_delete(self, name): """ Delete named policy @param[in] name vhost name """ if name not in self.rulesetdb: raise PolicyError("Policy '%s' does not exist" % name) # TODO: ruleset lock 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] # # db enumerator # def policy_db_get_names(self): """ Return a list of vhost names in this policy """ return list(self.rulesetdb.keys()) def set_default_vhost(self, name): """ Set the default vhost name. @param name: the name of the default vhost @return: none """ self._default_vhost = name self._manager.log_info("Policy fallback defaultVhost is defined: '%s'" % name) def default_vhost_enabled(self): """ The default vhost is enabled if the name is not blank and the vhost is defined in rulesetdb. @return: """ return self._default_vhost != "" and self._default_vhost in self.rulesetdb # # 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. Determine if a user on host accessing vhost through AMQP Open is allowed according to the policy access rules. If allowed then return the policy vhost settings name. If stats.can_connect returns true then it has registered and counted the connection. @param[in] user connection authId @param[in] rhost connection remote host numeric IP address as string @param[in] vhost_in vhost user is accessing @param[in] conn_name connection name used for tracking reports @param[in] conn_id internal connection id @return settings user-group name if allowed; "" if not allowed """ try: # choose rule set based on incoming vhost or default vhost # or potential vhost found by pattern matching 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': " "proceeds using vhost '%s' ruleset" % (user, rhost, vhost_in, vhost)) ruleset = self.rulesetdb[vhost] # look up the stats if vhost not in self.statsdb: msg = ( "DENY AMQP Open for user '%s', rhost '%s', vhost '%s': " "INTERNAL: Policy is defined but stats are missing" % (user, rhost, vhost)) raise PolicyError(msg) stats = self.statsdb[vhost] # Get settings for user in a user group or in default if user in ruleset[PolicyKeys.RULESET_U2G_MAP]: usergroup = ruleset[PolicyKeys.RULESET_U2G_MAP][user] elif "*" in ruleset[PolicyKeys.RULESET_U2G_MAP]: usergroup = ruleset[PolicyKeys.RULESET_U2G_MAP]["*"] else: if ruleset[PolicyKeys.KW_CONNECTION_ALLOW_DEFAULT]: usergroup = PolicyKeys.KW_DEFAULT_SETTINGS else: self._manager.log_info( "DENY AMQP Open for user '%s', rhost '%s', vhost '%s': " "User is not in a user group and unknown users are denied" % (user, rhost, vhost)) stats.count_other_denial() return "" groupsettings = ruleset[PolicyKeys.KW_GROUPS][usergroup] # User in usergroup allowed to connect from rhost? allowed = False if PolicyKeys.KW_REMOTE_HOSTS in groupsettings: # Users are restricted to connecting from a rhost # defined by the group's remoteHost list cglist = groupsettings[PolicyKeys.KW_REMOTE_HOSTS] uhs = HostStruct(rhost) for cohost in cglist: if cohost.match_bin(uhs): allowed = True break if not allowed: self._manager.log_info( "DENY AMQP Open for user '%s', rhost '%s', vhost '%s': " "User is not allowed to connect from this network host" % (user, rhost, vhost)) stats.count_other_denial() return "" # This user passes administrative approval. # Now check live connection counts # Extract optional usergroup connection counts group_max_conn_user = groupsettings.get(PolicyKeys.KW_MAXCONNPERUSER) group_max_conn_host = groupsettings.get(PolicyKeys.KW_MAXCONNPERHOST) diags = [] if not stats.can_connect(conn_name, user, rhost, diags, group_max_conn_user, group_max_conn_host): for diag in diags: self._manager.log_info( "DENY AMQP Open for user '%s', rhost '%s', vhost '%s': " "%s" % (user, rhost, vhost, diag)) return "" # Record facts about this connection to use during teardown facts = ConnectionFacts(user, rhost, vhost, conn_name) self._connections[conn_id] = facts # Return success return usergroup except Exception as e: self._manager.log_info( "DENY AMQP Open lookup_user failed for user '%s', rhost '%s', vhost '%s': " "Internal error: %s" % (user, rhost, vhost, e)) # return failure return "" def lookup_settings(self, vhost_in, groupname, upolicy): """ Given a settings name, return the aggregated policy blob. @param[in] vhost_in: vhost user is accessing @param[in] groupname: user group name @param[out] upolicy: dict holding policy values - the settings blob TODO: make this a c struct @return if lookup worked # Note: the upolicy output is a non-nested dict with settings of interest """ try: vhost = self.lookup_vhost_alias(vhost_in) if vhost != vhost_in: self._manager.log_debug( "AMQP Open lookup settings for vhost '%s': " "proceeds using vhost '%s' ruleset" % (vhost_in, vhost)) if vhost not in self.rulesetdb: self._manager.log_info( "lookup_settings fail for vhost '%s', user group '%s': " "No policy defined for this vhost" % (vhost, groupname)) return False ruleset = self.rulesetdb[vhost] if groupname not in ruleset[PolicyKeys.KW_GROUPS]: self._manager.log_trace( "lookup_settings fail for vhost '%s', user group '%s': " "This vhost has no settings for the user group" % (vhost, groupname)) return False upolicy.update(ruleset[PolicyKeys.KW_GROUPS][groupname]) maxsize = upolicy.get(PolicyKeys.KW_MAX_MESSAGE_SIZE, None) if maxsize is None: maxsize = ruleset.get(PolicyKeys.KW_MAX_MESSAGE_SIZE, None) if maxsize is None: maxsize = self._max_message_size upolicy[PolicyKeys.KW_MAX_MESSAGE_SIZE] = maxsize upolicy[PolicyKeys.KW_CSTATS] = self.statsdb[vhost].get_cstats() return True except Exception as e: return False def close_connection(self, conn_id): """ Close the connection. @param conn_id: @return: """ try: if conn_id in self._connections: facts = self._connections[conn_id] stats = self.statsdb[facts.app] stats.disconnect(facts.conn_name, facts.user, facts.host) del self._connections[conn_id] except Exception as e: self._manager.log_trace( "Policy internal error closing connection id %s. %s" % (conn_id, str(e))) def set_max_message_size(self, size): """ record max message size from policy config object :param size: :return:ls """ self._max_message_size = size # # def test_load_config(self): """ Test function to load a policy. @return: """ 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" },' ruleset_str += '"paidsubscribers": { "users": "p1, p2", "remoteHosts": "*", "maxFrameSize": 333333, "maxMessageSize": 333333, "maxSessionWindow": 333333, "maxSessions": 3, "maxSenders": 33, "maxReceivers": 33, "allowDynamicSource": true, "allowAnonymousSender": false, "sources": "public, private", "targets": "public, private" },' ruleset_str += '"test": { "users": "zeke, ynot", "remoteHosts": "10.48.0.0-10.48.255.255, 192.168.100.0-192.168.100.255", "maxFrameSize": 444444, "maxMessageSize": 444444, "maxSessionWindow": 444444, "maxSessions": 4, "maxSenders": 44, "maxReceivers": 44, "allowDynamicSource": true, "allowAnonymousSender": true, "sources": "private", "targets": "private" },' if is_ipv6_enabled(): ruleset_str += '"admin": { "users": "alice, bob", "remoteHosts": "10.48.0.0-10.48.255.255, 192.168.100.0-192.168.100.255, 10.18.0.0-10.18.255.255, 127.0.0.1, ::1", "maxFrameSize": 555555, "maxMessageSize": 555555, "maxSessionWindow": 555555, "maxSessions": 5, "maxSenders": 55, "maxReceivers": 55, "allowDynamicSource": true, "allowAnonymousSender": true, "sources": "public, private, management", "targets": "public, private, management" },' ruleset_str += '"superuser": { "users": "ellen", "remoteHosts": "72.135.2.9, 127.0.0.1, ::1", "maxFrameSize": 666666, "maxMessageSize": 666666, "maxSessionWindow": 666666, "maxSessions": 6, "maxSenders": 66, "maxReceivers": 66, "allowDynamicSource": false, "allowAnonymousSender": false, "sources": "public, private, management, root", "targets": "public, private, management, root" },' else: ruleset_str += '"admin": { "users": "alice, bob", "remoteHosts": "10.48.0.0-10.48.255.255, 192.168.100.0-192.168.100.255, 10.18.0.0-10.18.255.255, 127.0.0.1", "maxFrameSize": 555555, "maxMessageSize": 555555, "maxSessionWindow": 555555, "maxSessions": 5, "maxSenders": 55, "maxReceivers": 55, "allowDynamicSource": true, "allowAnonymousSender": true, "sources": "public, private, management", "targets": "public, private, management" },' ruleset_str += '"superuser": { "users": "ellen", "remoteHosts": "72.135.2.9, 127.0.0.1", "maxFrameSize": 666666, "maxMessageSize": 666666, "maxSessionWindow": 666666, "maxSessions": 6, "maxSenders": 66, "maxReceivers": 66, "allowDynamicSource": false, "allowAnonymousSender": false, "sources": "public, private, management, root", "targets": "public, private, management, root" },' ruleset_str += '"$default": { "remoteHosts": "*", "maxFrameSize": 222222, "maxMessageSize": 222222, "maxSessionWindow": 222222, "maxSessions": 2, "maxSenders": 22, "maxReceivers": 22, "allowDynamicSource": false, "allowAnonymousSender": false, "sources": "public, private", "targets": "public" }' ruleset_str += '}}]' ruleset = json.loads(ruleset_str) self.create_ruleset(ruleset[1])