From 7b5339b9d3cb7c0fea2c2fe7cb8296393b11232d Mon Sep 17 00:00:00 2001 From: Craig <3979063+craig8@users.noreply.github.com> Date: Wed, 8 May 2024 14:17:54 -0700 Subject: [PATCH] Releases/9.0.1 (#3171) * Update readthedocs requirements.txt * Update conf.py * Update requirements_demo.txt Add missing pandas requirement for demo * work around for issue #3154 * Fix for security issue #3168 (#3169) * Fix for security issue #3168 * handling clean up errors in test * testing group commands in different test module * moved group and role test to different module * moved group and role test to different module * Added a cache for agent names since platform start * Fixes process overload from file events * fixed issue with variable definition. * Remove PersistentDict from web-user.json file. * Update admin_endpoints.py Handle behavior of removing PersistentDict * Update version to 9.0.1 --------- Co-authored-by: Chandrika Sivaramakrishnan Co-authored-by: Chandrika Co-authored-by: Andrew Rodgers --- docs/source/conf.py | 4 +- .../core/IEEE_2030_5/requirements_demo.txt | 1 + volttron/platform/__init__.py | 2 +- volttron/platform/aip.py | 4 + volttron/platform/main.py | 3 - .../platform/vip/agent/subsystems/auth.py | 6 +- volttron/platform/vip/agent/subsystems/rpc.py | 4 +- volttron/platform/web/admin_endpoints.py | 15 +- volttron/utils/__init__.py | 4 +- .../platform/auth_tests/test_auth_control.py | 14 -- .../auth_tests/test_auth_group_roles.py | 174 ++++++++++++++ .../auth_tests/test_auth_integration.py | 224 ++++++++++++++++++ volttrontesting/utils/platformwrapper.py | 25 +- 13 files changed, 437 insertions(+), 43 deletions(-) create mode 100644 volttrontesting/platform/auth_tests/test_auth_group_roles.py create mode 100644 volttrontesting/platform/auth_tests/test_auth_integration.py diff --git a/docs/source/conf.py b/docs/source/conf.py index dc75f60eb2..10c31ff659 100644 --- a/docs/source/conf.py +++ b/docs/source/conf.py @@ -90,9 +90,9 @@ def __getattr__(cls, name): author = 'The VOLTTRON Community' # The short X.Y version -version = '9.0' +version = '9.0.1' # The full version, including alpha/beta/rc tags -release = '9.0' +release = '9.0.1' # -- General configuration --------------------------------------------------- diff --git a/services/core/IEEE_2030_5/requirements_demo.txt b/services/core/IEEE_2030_5/requirements_demo.txt index cf84bba01d..0fdf400704 100644 --- a/services/core/IEEE_2030_5/requirements_demo.txt +++ b/services/core/IEEE_2030_5/requirements_demo.txt @@ -2,3 +2,4 @@ nicegui requests xsdata>=23.8 blinker +pandas diff --git a/volttron/platform/__init__.py b/volttron/platform/__init__.py index d6ce14cdd2..5a7525b81e 100644 --- a/volttron/platform/__init__.py +++ b/volttron/platform/__init__.py @@ -35,7 +35,7 @@ from urllib.parse import urlparse from ..utils.frozendict import FrozenDict -__version__ = '9.0rc0' +__version__ = '9.0.1' _log = logging.getLogger(__name__) diff --git a/volttron/platform/aip.py b/volttron/platform/aip.py index dc7859ec89..07e8219126 100644 --- a/volttron/platform/aip.py +++ b/volttron/platform/aip.py @@ -247,6 +247,7 @@ def __init__(self, env, **kwargs): if self.message_bus == 'rmq': self.rmq_mgmt = RabbitMQMgmt() self.instance_name = get_platform_instance_name() + self.agent_uuid_name_map = {} def add_agent_user_group(self): user = pwd.getpwuid(os.getuid()) @@ -682,11 +683,14 @@ def remove_agent(self, agent_uuid, remove_auth=True): self.remove_agent_user(volttron_agent_user) def agent_name(self, agent_uuid): + if cached_name := self.agent_uuid_name_map.get(agent_uuid): + return cached_name agent_path = os.path.join(self.install_dir, agent_uuid) for agent_name in os.listdir(agent_path): dist_info = os.path.join( agent_path, agent_name, agent_name + '.dist-info') if os.path.exists(dist_info): + self.agent_uuid_name_map[agent_uuid] = agent_name return agent_name raise KeyError(agent_uuid) diff --git a/volttron/platform/main.py b/volttron/platform/main.py index a0fdd7ecf4..16db2285d1 100644 --- a/volttron/platform/main.py +++ b/volttron/platform/main.py @@ -435,9 +435,6 @@ def issue(self, topic, frames, extra=None): # return result def handle_subsystem(self, frames, user_id): - _log.debug( - f"Handling subsystem with frames: {frames} user_id: {user_id}") - subsystem = frames[5] if subsystem == 'quit': sender = frames[0] diff --git a/volttron/platform/vip/agent/subsystems/auth.py b/volttron/platform/vip/agent/subsystems/auth.py index 8ba4e2dcba..9337d0e513 100644 --- a/volttron/platform/vip/agent/subsystems/auth.py +++ b/volttron/platform/vip/agent/subsystems/auth.py @@ -265,6 +265,7 @@ def update_rpc_method_capabilities(self): """ rpc_method_authorizations = {} rpc_methods = self.get_rpc_exports() + updated_rpc_authorizations = None for method in rpc_methods: if len(method.split(".")) > 1: pass @@ -295,9 +296,7 @@ def update_rpc_method_capabilities(self): _log.info( f"Skipping updating rpc auth capabilities for agent " f"{self._core().identity} connecting to remote address: {self._core().address} ") - updated_rpc_authorizations = None except gevent.timeout.Timeout: - updated_rpc_authorizations = None _log.warning(f"update_id_rpc_authorization rpc call timed out for {self._core().identity} {rpc_method_authorizations}") except MethodNotFound: _log.warning("update_id_rpc_authorization method is missing from " @@ -306,7 +305,6 @@ def update_rpc_method_capabilities(self): "dynamic RPC authorizations.") return except Exception as e: - updated_rpc_authorizations = None _log.exception(f"Exception when calling rpc method update_id_rpc_authorizations for identity: " f"{self._core().identity} Exception:{e}") if updated_rpc_authorizations is None: @@ -318,7 +316,7 @@ def update_rpc_method_capabilities(self): f"the identity of the agent" ) return - if rpc_method_authorizations != updated_rpc_authorizations: + if rpc_method_authorizations != updated_rpc_authorizations and updated_rpc_authorizations is not None: for method in updated_rpc_authorizations: self.set_rpc_authorizations( method, updated_rpc_authorizations[method] diff --git a/volttron/platform/vip/agent/subsystems/rpc.py b/volttron/platform/vip/agent/subsystems/rpc.py index 510ce9d099..d77108a481 100644 --- a/volttron/platform/vip/agent/subsystems/rpc.py +++ b/volttron/platform/vip/agent/subsystems/rpc.py @@ -278,8 +278,8 @@ def _iterate_exports(self): for method_name in self._exports: method = self._exports[method_name] caps = annotations(method, set, "rpc.allow_capabilities") - # if caps: - # self._exports[method_name] = self._add_auth_check(method, caps) + if caps: + self._exports[method_name] = self._add_auth_check(method, caps) def _add_auth_check(self, method, required_caps): """ diff --git a/volttron/platform/web/admin_endpoints.py b/volttron/platform/web/admin_endpoints.py index 9de8d05a5c..d307b2f353 100644 --- a/volttron/platform/web/admin_endpoints.py +++ b/volttron/platform/web/admin_endpoints.py @@ -46,7 +46,6 @@ from volttron.platform import get_home from volttron.platform import jsonapi from volttron.utils import VolttronHomeFileReloader -from volttron.utils.persistance import PersistentDict _log = logging.getLogger(__name__) @@ -84,7 +83,7 @@ def __init__(self, rmq_mgmt=None, ssl_public_key: bytes = None, rpc_caller=None) else: self._ssl_public_key = None - self._userdict = None + self._userdict = {} self.reload_userdict() self._observer = Observer() @@ -96,7 +95,14 @@ def __init__(self, rmq_mgmt=None, ssl_public_key: bytes = None, rpc_caller=None) def reload_userdict(self): webuserpath = os.path.join(get_home(), 'web-users.json') - self._userdict = PersistentDict(webuserpath, format="json") + if os.path.exists(webuserpath): + with open(webuserpath) as fp: + try: + self._userdict = jsonapi.loads(fp.read()) + except json.decoder.JSONDecodeError: + self._userdict = {} + # Keep same behavior as with PersistentDict + raise ValueError("File not in a supported format") def get_routes(self): """ @@ -339,4 +345,5 @@ def add_user(self, username, unencrypted_pw, groups=None, overwrite=False): groups=groups ) - self._userdict.sync() + with open(os.path.join(get_home(), 'web-users.json'), 'w') as fp: + fp.write(jsonapi.dumps(self._userdict, indent=2)) diff --git a/volttron/utils/__init__.py b/volttron/utils/__init__.py index daa1e1aaa7..e236d5185f 100644 --- a/volttron/utils/__init__.py +++ b/volttron/utils/__init__.py @@ -108,7 +108,7 @@ def __init__(self, filetowatch, callback): _log.debug("patterns is {}".format([get_home() + '/' + filetowatch])) self._callback = callback - def on_any_event(self, event): + def on_closed(self, event): _log.debug("Calling callback on event {}. Calling {}".format(event, self._callback)) try: self._callback() @@ -133,7 +133,7 @@ def __init__(self, filetowatch, callback): def watchfile(self): return self._filetowatch - def on_any_event(self, event): + def on_closed(self, event): _log.debug("Calling callback on event {}. Calling {}".format(event, self._callback)) try: self._callback(self._filetowatch) diff --git a/volttrontesting/platform/auth_tests/test_auth_control.py b/volttrontesting/platform/auth_tests/test_auth_control.py index e76d63df8f..a17c6bf1b0 100644 --- a/volttrontesting/platform/auth_tests/test_auth_control.py +++ b/volttrontesting/platform/auth_tests/test_auth_control.py @@ -375,20 +375,6 @@ def test_auth_rpc_method_remove(auth_instance): assert entries[-1]['rpc_method_authorizations'] != {'test_method': ["test_auth"]} -@pytest.mark.control -def test_group_cmds(auth_instance): - """Test add-group, list-groups, update-group, and remove-group""" - _run_group_or_role_cmds(auth_instance, _add_group, _list_groups, - _update_group, _remove_group) - - -@pytest.mark.control -def test_role_cmds(auth_instance): - """Test add-role, list-roles, update-role, and remove-role""" - _run_group_or_role_cmds(auth_instance, _add_role, _list_roles, - _update_role, _remove_role) - - def _run_group_or_role_cmds(platform, add_fn, list_fn, update_fn, remove_fn): expected = [] key = '0' diff --git a/volttrontesting/platform/auth_tests/test_auth_group_roles.py b/volttrontesting/platform/auth_tests/test_auth_group_roles.py new file mode 100644 index 0000000000..bfcad699c5 --- /dev/null +++ b/volttrontesting/platform/auth_tests/test_auth_group_roles.py @@ -0,0 +1,174 @@ + +import os +import re +import subprocess + +import gevent +import pytest +from mock import MagicMock +from volttron.platform.auth.auth_protocols.auth_zmq import ZMQAuthorization, ZMQServerAuthentication + +from volttrontesting.platform.auth_tests.conftest import assert_auth_entries_same +from volttrontesting.utils.platformwrapper import with_os_environ +from volttrontesting.utils.utils import AgentMock +from volttron.platform.vip.agent import Agent +from volttron.platform.auth import AuthService +from volttron.platform.auth import AuthEntry +from volttron.platform import jsonapi + +@pytest.fixture(autouse=True) +def auth_instance(volttron_instance): + if not volttron_instance.auth_enabled: + pytest.skip("AUTH tests are not applicable if auth is disabled") + with open(os.path.join(volttron_instance.volttron_home, "auth.json"), 'r') as f: + auth_file = jsonapi.load(f) + print(auth_file) + try: + yield volttron_instance + finally: + with with_os_environ(volttron_instance.env): + with open(os.path.join(volttron_instance.volttron_home, "auth.json"), 'w') as f: + jsonapi.dump(auth_file, f) + + +def _run_group_or_role_cmds(platform, add_fn, list_fn, update_fn, remove_fn): + expected = [] + key = '0' + values = ['0', '1'] + expected.extend(values) + + add_fn(platform, key, values) + gevent.sleep(4) + keys = list_fn(platform) + assert set(keys[key]) == set(expected) + + # Update add single value + values = ['2'] + expected.extend(values) + update_fn(platform, key, values) + gevent.sleep(2) + keys = list_fn(platform) + assert set(keys[key]) == set(expected) + + # Update add multiple values + values = ['3', '4'] + expected.extend(values) + update_fn(platform, key, values) + gevent.sleep(2) + keys = list_fn(platform) + assert set(keys[key]) == set(expected) + + # Update remove single value + value = '0' + expected.remove(value) + update_fn(platform, key, [value], remove=True) + gevent.sleep(2) + keys = list_fn(platform) + assert set(keys[key]) == set(expected) + + # Update remove single value + values = ['1', '2'] + for value in values: + expected.remove(value) + update_fn(platform, key, values, remove=True) + gevent.sleep(2) + keys = list_fn(platform) + assert set(keys[key]) == set(expected) + + # Remove key + remove_fn(platform, key) + gevent.sleep(2) + keys = list_fn(platform) + assert key not in keys + + + +def _add_group_or_role(platform, cmd, name, list_): + with with_os_environ(platform.env): + args = ['volttron-ctl', 'auth', cmd, name] + args.extend(list_) + p = subprocess.Popen(args, env=platform.env, stdin=subprocess.PIPE, universal_newlines=True) + p.communicate() + assert p.returncode == 0 + + +def _add_group(platform, group, roles): + _add_group_or_role(platform, 'add-group', group, roles) + + +def _add_role(platform, role, capabilities): + _add_group_or_role(platform, 'add-role', role, capabilities) + + +def _list_groups_or_roles(platform, cmd): + with with_os_environ(platform.env): + output = subprocess.check_output(['volttron-ctl', 'auth', cmd], + env=platform.env, universal_newlines=True) + # For these tests don't use names that contain space, [, comma, or ' + output = output.replace('[', '').replace("'", '').replace(']', '') + output = output.replace(',', '') + lines = output.split('\n') + + dict_ = {} + for line in lines[2:-1]: # skip two header lines and last (empty) line + list_ = ' '.join(line.split()).split() # combine multiple spaces + dict_[list_[0]] = list_[1:] + return dict_ + + +def _list_groups(platform): + return _list_groups_or_roles(platform, 'list-groups') + + +def _list_roles(platform): + return _list_groups_or_roles(platform, 'list-roles') + + +def _update_group_or_role(platform, cmd, key, values, remove): + with with_os_environ(platform.env): + args = ['volttron-ctl', 'auth', cmd, key] + args.extend(values) + if remove: + args.append('--remove') + p = subprocess.Popen(args, env=platform.env, stdin=subprocess.PIPE, universal_newlines=True) + p.communicate() + assert p.returncode == 0 + + +def _update_group(platform, group, roles, remove=False): + _update_group_or_role(platform, 'update-group', group, roles, remove) + + +def _update_role(platform, role, caps, remove=False): + _update_group_or_role(platform, 'update-role', role, caps, remove) + + +def _remove_group_or_role(platform, cmd, key): + with with_os_environ(platform.env): + args = ['volttron-ctl', 'auth', cmd, key] + p = subprocess.Popen(args, env=platform.env, stdin=subprocess.PIPE, universal_newlines=True) + p.communicate() + assert p.returncode == 0 + + +def _remove_group(platform, group): + _remove_group_or_role(platform, 'remove-group', group) + + +def _remove_role(platform, role): + _remove_group_or_role(platform, 'remove-role', role) + + +@pytest.mark.control +def test_group_cmds(auth_instance): + """Test add-group, list-groups, update-group, and remove-group""" + _run_group_or_role_cmds(auth_instance, _add_group, _list_groups, + _update_group, _remove_group) + + +@pytest.mark.control +def test_role_cmds(auth_instance): + """Test add-role, list-roles, update-role, and remove-role""" + _run_group_or_role_cmds(auth_instance, _add_role, _list_roles, + _update_role, _remove_role) + diff --git a/volttrontesting/platform/auth_tests/test_auth_integration.py b/volttrontesting/platform/auth_tests/test_auth_integration.py new file mode 100644 index 0000000000..bc9dedc9b3 --- /dev/null +++ b/volttrontesting/platform/auth_tests/test_auth_integration.py @@ -0,0 +1,224 @@ +import os +import subprocess +import sys +import tempfile +import gevent +import pytest +from volttron.platform.agent.known_identities import AUTH +from volttron.platform import jsonrpc +from volttron.platform.messaging.health import STATUS_BAD + +called_agent_src = """ +import sys +from volttron.platform.agent import utils +from volttron.platform.vip.agent import Agent, Core +from volttron.platform.vip.agent.subsystems import RPC +import gevent +class CalledAgent(Agent): + def __init__(self, config_path, **kwargs): + super(CalledAgent, self).__init__(**kwargs) + @RPC.export + @RPC.allow("can_call_method") + def restricted_method(self, sender, **kwargs): + print("test") +def main(argv=sys.argv): + try: + utils.vip_main(CalledAgent, version='0.1') + except Exception as e: + print('unhandled exception: {}'.format(e)) +if __name__ == '__main__': + # Entry point for script + sys.exit(main()) +""" + +called_agent_setup = """ +from setuptools import setup +setup( + name='calledagent', + version='0.1', + install_requires=['volttron'], + packages=['calledagent'], + entry_points={ + 'setuptools.installation': [ + 'eggsecutable=calledagent.calledagent:main', + ] + } +) +""" + +caller_agent_src = """ +import sys +import gevent +import logging +from volttron.platform.agent import utils +from volttron.platform.vip.agent import Agent, Core +from volttron.platform.vip.agent.subsystems import RPC +from volttron.platform.scheduling import periodic +from volttron.platform.messaging.health import (STATUS_BAD, + STATUS_GOOD, Status) +from volttron.platform.agent.known_identities import AUTH +from volttron.platform import jsonrpc +from volttron.platform.messaging.health import STATUS_BAD + +_log = logging.getLogger(__name__) +class CallerAgent(Agent): + def __init__(self, config_path, **kwargs): + super(CallerAgent, self).__init__(**kwargs) + + # @Core.schedule(periodic(3)) + # def call_rpc_method(self): + @Core.receiver("onstart") + def onstart(self, sender, **kwargs): + try: + self.vip.rpc.call('called_agent', 'restricted_method').get(timeout=3) + except Exception as e: + self.vip.health.set_status(STATUS_BAD, f"{e}") +def main(argv=sys.argv): + try: + utils.vip_main(CallerAgent, version='0.1') + except Exception as e: + print('unhandled exception: {}'.format(e)) +if __name__ == '__main__': + # Entry point for script + sys.exit(main()) +""" + +caller_agent_setup = """ +from setuptools import setup +setup( + name='calleragent', + version='0.1', + install_requires=['volttron'], + packages=['calleragent'], + entry_points={ + 'setuptools.installation': [ + 'eggsecutable=calleragent.calleragent:main', + ] + } +) +""" + +@pytest.fixture +def install_two_agents(volttron_instance): + """Returns two agents for testing authorization + + The first agent is the "RPC callee." + The second agent is the unauthorized "RPC caller." + """ + """ + Test if control agent periodically monitors and restarts any crashed agents + :param volttron_instance: + :return: + """ + + tmpdir = volttron_instance.volttron_home+"/tmpdir" + os.mkdir(tmpdir) + tmpdir = volttron_instance.volttron_home+"/tmpdir" + "/called" + os.mkdir(tmpdir) + os.chdir(tmpdir) + + os.mkdir("calledagent") + with open(os.path.join("calledagent", "__init__.py"), "w") as file: + pass + with open(os.path.join("calledagent", "calledagent.py"), "w") as file: + file.write(called_agent_src) + with open(os.path.join("setup.py"), "w") as file: + file.write(called_agent_setup) + p = subprocess.Popen( + [sys.executable, "setup.py", "bdist_wheel"], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + ) + stdout, stderr = p.communicate() + # print("out {}".format(stdout)) + # print("err {}".format(stderr)) + + wheel = os.path.join(tmpdir, "dist", "calledagent-0.1-py3-none-any.whl") + assert os.path.exists(wheel) + called_uuid = volttron_instance.install_agent(agent_wheel=wheel, + vip_identity="called_agent", + start=False) + assert called_uuid + gevent.sleep(1) + + + tmpdir = volttron_instance.volttron_home+"/tmpdir" + "/caller" + os.mkdir(tmpdir) + os.chdir(tmpdir) + os.mkdir("calleragent") + with open(os.path.join("calleragent", "__init__.py"), "w") as file: + pass + with open(os.path.join("calleragent", "calleragent.py"), "w") as file: + file.write(caller_agent_src) + with open(os.path.join("setup.py"), "w") as file: + file.write(caller_agent_setup) + p = subprocess.Popen( + [sys.executable, "setup.py", "bdist_wheel"], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + ) + stdout, stderr = p.communicate() + # print("out {}".format(stdout)) + # print("err {}".format(stderr)) + + wheel = os.path.join(tmpdir, "dist", "calleragent-0.1-py3-none-any.whl") + assert os.path.exists(wheel) + caller_uuid = volttron_instance.install_agent(agent_wheel=wheel, + vip_identity="caller_agent", + start=False) + assert caller_uuid + gevent.sleep(1) + + try: + yield caller_uuid, called_uuid + finally: + #volttron_instance.remove_agent(caller_uuid) + #volttron_instance.remove_agent(called_uuid) + # TODO if we have to wait for auth propagation anyways why do we create new agents for each test case + # we should just update capabilities, at least we will save on agent creation and tear down time + gevent.sleep(1) + + +@pytest.fixture(autouse=True) +def build_volttron_instance(volttron_instance): + if not volttron_instance.auth_enabled: + pytest.skip("AUTH tests are not applicable if auth is disabled") + + +@pytest.mark.auth +def test_unauthorized_rpc_call(volttron_instance, install_two_agents): + """Tests an agent with no capabilities calling a method that + requires one capability ("can_call_foo") + """ + (caller_agent_uuid, called_agent_uuid) = install_two_agents + + # check auth error for newly installed agents + check_auth_error(volttron_instance, caller_agent_uuid, called_agent_uuid) + + volttron_instance.restart_platform() + gevent.sleep(3) + + # check auth error for already installed agent + check_auth_error(volttron_instance, caller_agent_uuid, called_agent_uuid) + +def check_auth_error(volttron_instance, caller_agent_uuid, called_agent_uuid): + + expected_auth_err = ('volttron.platform.jsonrpc.Error(' + '-32001, "method \'restricted_method\' ' + 'requires capabilities {\'can_call_method\'}, ' + 'but capability {\'edit_config_store\': {\'identity\': \'caller_agent\'}}' + ' was provided for user caller_agent")') + volttron_instance.start_agent(called_agent_uuid) + gevent.sleep(1) + volttron_instance.start_agent(caller_agent_uuid) + + # If the agent is not authorized health status is updated + health = volttron_instance.dynamic_agent.vip.rpc.call( + "caller_agent", "health.get_status").get(timeout=2) + + assert health.get('status') == STATUS_BAD + assert health.get('context') == expected_auth_err + + + + diff --git a/volttrontesting/utils/platformwrapper.py b/volttrontesting/utils/platformwrapper.py index 5fc82bf4b0..79cd4e4267 100644 --- a/volttrontesting/utils/platformwrapper.py +++ b/volttrontesting/utils/platformwrapper.py @@ -1564,17 +1564,20 @@ def shutdown_platform(self): return running_pids = [] - if self.dynamic_agent: # because we are not creating dynamic agent in setupmode - for agnt in self.list_agents(): - pid = self.agent_pid(agnt['uuid']) - if pid is not None and int(pid) > 0: - running_pids.append(int(pid)) - if not self.skip_cleanup: - self.remove_all_agents() - # don't wait indefinetly as shutdown will not throw an error if RMQ is down/has cert errors - self.dynamic_agent.vip.rpc(CONTROL, 'shutdown').get(timeout=10) - self.dynamic_agent.core.stop() - self.dynamic_agent = None + if self.dynamic_agent: + try:# because we are not creating dynamic agent in setupmode + for agnt in self.list_agents(): + pid = self.agent_pid(agnt['uuid']) + if pid is not None and int(pid) > 0: + running_pids.append(int(pid)) + if not self.skip_cleanup: + self.remove_all_agents() + # don't wait indefinetly as shutdown will not throw an error if RMQ is down/has cert errors + self.dynamic_agent.vip.rpc(CONTROL, 'shutdown').get(timeout=10) + self.dynamic_agent.core.stop() + self.dynamic_agent = None + except BaseException as e: + self.logit(f"Exception while shutting down. {e}") if self.p_process is not None: try: