ported config store security updates from monolithic volttron. See is…

…sue - VOLTTRON/volttron#3097

pyproject.toml

@@ -37,6 +37,7 @@ watchdog-gevent = "^0.1.1"
 pip = "22.2.2"
 pytest-timeout = "^2.1.0"
 pytest-mock = "^3.10.0"
+deprecated = "^1.2.14"
 
 [tool.poetry.group.dev.dependencies]
 pytest = "^6.2.5"

src/volttron/client/commands/control.py

@@ -1553,7 +1553,7 @@ def add_config_to_store(opts):
     file_contents = opts.infile.read()
 
     call(
-        "manage_store",
+        "set_config",
         opts.identity,
         opts.name,
         file_contents,
@@ -1565,24 +1565,24 @@ def delete_config_from_store(opts):
     opts.connection.peer = CONFIGURATION_STORE
     call = opts.connection.call
     if opts.delete_store:
-        call("manage_delete_store", opts.identity)
+        call("delete_store", opts.identity)
         return
 
     if opts.name is None:
         _stderr.write("ERROR: must specify a configuration when not deleting entire store\n")
         return
 
-    call("manage_delete_config", opts.identity, opts.name)
+    call("delete_config", opts.identity, opts.name)
 
 
 def list_store(opts):
     opts.connection.peer = CONFIGURATION_STORE
     call = opts.connection.call
     results = []
     if opts.identity is None:
-        results = call("manage_list_stores")
+        results = call("list_stores")
     else:
-        results = call("manage_list_configs", opts.identity)
+        results = call("list_configs", opts.identity)
 
     for item in results:
         _stdout.write(item + "\n")
@@ -1591,7 +1591,7 @@ def list_store(opts):
 def get_config(opts):
     opts.connection.peer = CONFIGURATION_STORE
     call = opts.connection.call
-    results = call("manage_get", opts.identity, opts.name, raw=opts.raw)
+    results = call("get_config", opts.identity, opts.name, raw=opts.raw)
 
     if opts.raw:
         _stdout.write(results)
@@ -1612,7 +1612,7 @@ def edit_config(opts):
         raw_data = ""
     else:
         try:
-            results = call("manage_get_metadata", opts.identity, opts.name)
+            results = call("get_metadata", opts.identity, opts.name)
             config_type = results["type"]
             raw_data = results["data"]
         except RemoteError as e:
@@ -1648,7 +1648,7 @@ def edit_config(opts):
             return
 
         call(
-            "manage_store",
+            "set_config",
             opts.identity,
             opts.name,
             new_raw_data,

src/volttron/client/vip/agent/subsystems/configstore.py

@@ -30,7 +30,7 @@
 
 from .base import SubsystemBase
 from volttron.utils.storeutils import list_unique_links, check_for_config_link
-
+from volttron.utils import jsonapi
 # from volttron.client.storeutils import list_unique_links, check_for_config_link
 from volttron.client.vip.agent import errors
 from volttron.client.known_identities import CONFIGURATION_STORE
@@ -47,7 +47,7 @@
 
 _log = logging.getLogger(__name__)
 
-VALID_ACTIONS = set(["NEW", "UPDATE", "DELETE"])
+VALID_ACTIONS = ("NEW", "UPDATE", "DELETE")
 
 
 class ConfigStore(SubsystemBase):
@@ -68,6 +68,7 @@ def __init__(self, owner, core, rpc):
         self._initial_callbacks_called = False
 
         self._process_callbacks_code_object = self._process_callbacks.__code__
+        self.vip_identity = self._core().identity
 
         def sub_factory():
             return defaultdict(set)
@@ -77,14 +78,16 @@ def sub_factory():
         def onsetup(sender, **kwargs):
             rpc.export(self._update_config, "config.update")
             rpc.export(self._initial_update, "config.initial_update")
+            rpc.allow("config.update", "sync_agent_config")
+            rpc.allow("config.initial_update", "sync_agent_config")
 
         core.onsetup.connect(onsetup, self)
         core.configuration.connect(self._onconfig, self)
 
     def _onconfig(self, sender, **kwargs):
         if not self._initialized:
             try:
-                self._rpc().call(CONFIGURATION_STORE, "get_configs").get()
+                self._rpc().call(CONFIGURATION_STORE, "initialize_configs", self.vip_identity).get()
             except errors.Unreachable as e:
                 _log.error("Connected platform does not support the Configuration Store feature.")
                 return
@@ -144,17 +147,15 @@ def _process_links(self, config_contents, already_gathered):
                 elif isinstance(value, str):
                     config_name = check_for_config_link(value)
                     if config_name is not None:
-                        config_contents[key] = self._gather_child_configs(
-                            config_name, already_gathered)
+                        config_contents[key] = self._gather_child_configs(config_name, already_gathered)
         elif isinstance(config_contents, list):
             for i, value in enumerate(config_contents):
                 if isinstance(value, (dict, list)):
                     self._process_links(value, already_gathered)
                 elif isinstance(value, str):
                     config_name = check_for_config_link(value)
                     if config_name is not None:
-                        config_contents[i] = self._gather_child_configs(
-                            config_name, already_gathered)
+                        config_contents[i] = self._gather_child_configs(config_name, already_gathered)
 
     def _gather_child_configs(self, config_name, already_gathered):
         if config_name in already_gathered:
@@ -285,7 +286,7 @@ def list(self):
         # Handle case were we are called during "onstart".
         if not self._initialized:
             try:
-                self._rpc().call(CONFIGURATION_STORE, "get_configs").get()
+                self._rpc().call(CONFIGURATION_STORE, "initialize_configs", self.vip_identity).get()
             except errors.Unreachable as e:
                 _log.error("Connected platform does not support the Configuration Store feature.")
             except errors.VIPError as e:
@@ -317,7 +318,7 @@ def get(self, config_name="config"):
         # may be a default configuration to grab.
         if not self._initialized:
             try:
-                self._rpc().call(CONFIGURATION_STORE, "get_configs").get()
+                self._rpc().call(CONFIGURATION_STORE, "initialize_configs", self.vip_identity).get()
             except errors.Unreachable as e:
                 _log.error("Connected platform does not support the Configuration Store feature.")
             except errors.VIPError as e:
@@ -333,9 +334,7 @@ def _check_call_from_process_callbacks(self):
             # Don't create any unneeded references to frame objects.
             for frame, *_ in frame_records:
                 if self._process_callbacks_code_object is frame.f_code:
-                    raise RuntimeError(
-                        "Cannot request changes to the config store from a configuration callback."
-                    )
+                    raise RuntimeError("Cannot request changes to the config store from a configuration callback.")
         finally:
             del frame_records
 
@@ -349,21 +348,26 @@ def set(self, config_name, contents, trigger_callback=False, send_update=True):
         :param config_name: Name of configuration to add to store.
         :param contents: Contents of the configuration. May be a string, dictionary, or list.
         :param trigger_callback: Tell the platform to trigger callbacks on the agent for this change.
+        :param send_update: Boolean flag to tell the server if it should call config.update on this agent
+         after server side update is done
 
         :type config_name: str
         :type contents: str, dict, list
         :type trigger_callback: bool
         """
         self._check_call_from_process_callbacks()
 
-        self._rpc().call(
-            CONFIGURATION_STORE,
-            "set_config",
-            config_name,
-            contents,
-            trigger_callback=trigger_callback,
-            send_update=send_update,
-        ).get(timeout=10.0)
+        if isinstance(contents, (dict, list)):
+            config_type = 'json'
+            raw_data = jsonapi.dumps(contents)
+        elif isinstance(contents, str):
+            config_type = 'raw'
+            raw_data = contents
+        else:
+            raise ValueError("Unsupported configuration content type: {}".format(str(type(contents))))
+
+        self._rpc().call(CONFIGURATION_STORE, "set_config", self.vip_identity, config_name, raw_data,
+                         config_type, trigger_callback=trigger_callback, send_update=send_update).get(timeout=10.0)
 
     def set_default(self, config_name, contents):
         """Called to set the contents of a default configuration file. Default configurations are used if the
@@ -422,19 +426,16 @@ def delete(self, config_name, trigger_callback=False, send_update=True):
         """
         self._check_call_from_process_callbacks()
 
-        self._rpc().call(
-            CONFIGURATION_STORE,
-            "delete_config",
-            config_name,
-            trigger_callback=trigger_callback,
-            send_update=send_update,
-        ).get(timeout=10.0)
+        self._rpc().call(CONFIGURATION_STORE, "delete_config", self.vip_identity, config_name,
+                         trigger_callback=trigger_callback,
+                         send_update=send_update).get(timeout=10.0)
 
     def subscribe(self, callback, actions=VALID_ACTIONS, pattern="*"):
         """Subscribe to changes to a configuration.
 
         :param callback: Function to call in response to changes to a configuration.
-        :param actions: Change actions to respond to. Valid values are "NEW", "UPDATE", and "DELETE". May be a single action or a list of actions.
+        :param actions: Change actions to respond to. Valid values are "NEW", "UPDATE", and "DELETE".
+         Maybe a single action or a list of actions.
         :param pattern: Configuration name pattern to match to.  Uses Unix style filename pattern matching.
 
         :type callback: str
@@ -446,9 +447,9 @@ def subscribe(self, callback, actions=VALID_ACTIONS, pattern="*"):
 
         actions = set(action.upper() for action in actions)
 
-        invalid_actions = actions - VALID_ACTIONS
+        invalid_actions = actions - set(VALID_ACTIONS)
         if invalid_actions:
-            raise ValueError("Invalid actions: " + list(invalid_actions))
+            raise ValueError(f"Invalid actions: {invalid_actions}")
 
         pattern = pattern.lower()
 

src/volttron/services/auth/auth_service.py

@@ -1154,7 +1154,7 @@ def __init__(self, auth_file=None):
 
     @property
     def version(self):
-        return {"major": 1, "minor": 2}
+        return {"major": 1, "minor": 3}
 
     def _check_for_upgrade(self):
         allow_list, deny_list, groups, roles, version = self._read()
@@ -1290,6 +1290,10 @@ def upgrade_1_1_to_1_2(allow_list):
             version["minor"] = 1
         if version["major"] == 1 and version["minor"] == 1:
             allow_list = upgrade_1_1_to_1_2(allow_list)
+        if version["major"] == 1 and version["minor"] == 2:
+            # on start a new entry for config.store should have got created automatically
+            # so just update version
+            version["minor"] = 3
 
         allow_entries, deny_entries = self._get_entries(allow_list, deny_list)
         self._write(allow_entries, deny_entries, groups, roles)