summaryrefslogtreecommitdiff
path: root/python
diff options
context:
space:
mode:
Diffstat (limited to 'python')
-rw-r--r--python/vyos/config_mgmt.py7
-rw-r--r--python/vyos/configtree.py3
-rw-r--r--python/vyos/load_config.py200
3 files changed, 207 insertions, 3 deletions
diff --git a/python/vyos/config_mgmt.py b/python/vyos/config_mgmt.py
index 950f14d4f..fd0fa7a75 100644
--- a/python/vyos/config_mgmt.py
+++ b/python/vyos/config_mgmt.py
@@ -125,6 +125,7 @@ class ConfigMgmt:
get_first_key=True)
self.max_revisions = int(d.get('commit_revisions', 0))
+ self.num_revisions = 0
self.locations = d.get('commit_archive', {}).get('location', [])
self.source_address = d.get('commit_archive',
{}).get('source_address', '')
@@ -233,7 +234,7 @@ Proceed ?'''
msg = ''
if not self._check_revision_number(rev):
- msg = f'Invalid revision number {rev}: must be 0 < rev < {self.max_revisions}'
+ msg = f'Invalid revision number {rev}: must be 0 < rev < {self.num_revisions}'
return msg, 1
prompt_str = 'Proceed with reboot ?'
@@ -560,8 +561,8 @@ Proceed ?'''
return len(l)
def _check_revision_number(self, rev: int) -> bool:
- maxrev = self._get_number_of_revisions()
- if not 0 <= rev < maxrev:
+ self.num_revisions = self._get_number_of_revisions()
+ if not 0 <= rev < self.num_revisions:
return False
return True
diff --git a/python/vyos/configtree.py b/python/vyos/configtree.py
index 09cfd43d3..d048901f0 100644
--- a/python/vyos/configtree.py
+++ b/python/vyos/configtree.py
@@ -160,6 +160,9 @@ class ConfigTree(object):
def _get_config(self):
return self.__config
+ def get_version_string(self):
+ return self.__version
+
def to_string(self, ordered_values=False):
config_string = self.__to_string(self.__config, ordered_values).decode()
config_string = "{0}\n{1}".format(config_string, self.__version)
diff --git a/python/vyos/load_config.py b/python/vyos/load_config.py
new file mode 100644
index 000000000..af563614d
--- /dev/null
+++ b/python/vyos/load_config.py
@@ -0,0 +1,200 @@
+# Copyright 2023 VyOS maintainers and contributors <maintainers@vyos.io>
+#
+# This library is free software; you can redistribute it and/or
+# modify it under the terms of the GNU Lesser General Public
+# License as published by the Free Software Foundation; either
+# version 2.1 of the License, or (at your option) any later version.
+#
+# This library is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+# Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public License
+# along with this library. If not, see <http://www.gnu.org/licenses/>.
+
+"""This module abstracts the loading of a config file into the running
+config. It provides several varieties of loading a config file, from the
+legacy version to the developing versions, as a means of offering
+alternatives for competing use cases, and a base for profiling the
+performance of each.
+"""
+
+import sys
+from pathlib import Path
+from tempfile import NamedTemporaryFile
+from typing import Union, Literal, TypeAlias, get_type_hints, get_args
+
+from vyos.config import Config
+from vyos.configtree import ConfigTree, DiffTree
+from vyos.configsource import ConfigSourceSession, VyOSError
+from vyos.component_version import from_string as version_from_string
+from vyos.component_version import from_system as version_from_system
+from vyos.migrator import Migrator, VirtualMigrator, MigratorError
+from vyos.utils.process import popen, DEVNULL
+
+Variety: TypeAlias = Literal['explicit', 'batch', 'tree', 'legacy']
+ConfigObj: TypeAlias = Union[str, ConfigTree]
+
+thismod = sys.modules[__name__]
+
+class LoadConfigError(Exception):
+ """Raised when an error occurs loading a config file.
+ """
+
+# utility functions
+
+def get_running_config(config: Config) -> ConfigTree:
+ return config.get_config_tree(effective=True)
+
+def get_proposed_config(config_file: str = None) -> ConfigTree:
+ config_str = Path(config_file).read_text()
+ return ConfigTree(config_str)
+
+def migration_needed(config_obj: ConfigObj) -> bool:
+ """Check if a migration is needed for the config object.
+ """
+ if not isinstance(config_obj, ConfigTree):
+ atree = get_proposed_config(config_obj)
+ else:
+ atree = config_obj
+ version_str = atree.get_version_string()
+ if not version_str:
+ return True
+ aversion = version_from_string(version_str.splitlines()[1])
+ bversion = version_from_system()
+ return aversion != bversion
+
+def check_session(strict: bool, switch: Variety) -> None:
+ """Check if we are in a config session, with no uncommitted changes, if
+ strict. This is not needed for legacy load, as these checks are
+ implicit.
+ """
+
+ if switch == 'legacy':
+ return
+
+ context = ConfigSourceSession()
+
+ if not context.in_session():
+ raise LoadConfigError('not in a config session')
+
+ if strict and context.session_changed():
+ raise LoadConfigError('commit or discard changes before loading config')
+
+# methods to call for each variety
+
+# explicit
+def diff_to_commands(ctree: ConfigTree, ntree: ConfigTree) -> list:
+ """Calculate the diff between the current and proposed config."""
+ # Calculate the diff between the current and new config tree
+ commands = DiffTree(ctree, ntree).to_commands()
+ # on an empty set of 'add' or 'delete' commands, to_commands
+ # returns '\n'; prune below
+ command_list = commands.splitlines()
+ command_list = [c for c in command_list if c]
+ return command_list
+
+def set_commands(cmds: list) -> None:
+ """Set commands in the config session."""
+ if not cmds:
+ print('no commands to set')
+ return
+ error_out = []
+ for op in cmds:
+ out, rc = popen(f'/opt/vyatta/sbin/my_{op}', shell=True, stderr=DEVNULL)
+ if rc != 0:
+ error_out.append(out)
+ continue
+ if error_out:
+ out = '\n'.join(error_out)
+ raise LoadConfigError(out)
+
+# legacy
+class LoadConfig(ConfigSourceSession):
+ """A subclass for calling 'loadFile'.
+ """
+ def load_config(self, file_name):
+ return self._run(['/bin/cli-shell-api','loadFile', file_name])
+
+# end methods to call for each variety
+
+def migrate(config_obj: ConfigObj) -> ConfigObj:
+ """Migrate a config object to the current version.
+ """
+ if isinstance(config_obj, ConfigTree):
+ config_file = NamedTemporaryFile(delete=False).name
+ Path(config_file).write_text(config_obj.to_string())
+ else:
+ config_file = config_obj
+
+ virtual_migration = VirtualMigrator(config_file)
+ migration = Migrator(config_file)
+ try:
+ virtual_migration.run()
+ migration.run()
+ except MigratorError as e:
+ raise LoadConfigError(e) from e
+ else:
+ if isinstance(config_obj, ConfigTree):
+ return ConfigTree(Path(config_file).read_text())
+ return config_file
+ finally:
+ if isinstance(config_obj, ConfigTree):
+ Path(config_file).unlink()
+
+def load_explicit(config_obj: ConfigObj):
+ """Explicit load from file or configtree.
+ """
+ config = Config()
+ ctree = get_running_config(config)
+ if isinstance(config_obj, ConfigTree):
+ ntree = config_obj
+ else:
+ ntree = get_proposed_config(config_obj)
+ # Calculate the diff between the current and proposed config
+ cmds = diff_to_commands(ctree, ntree)
+ # Set the commands in the config session
+ set_commands(cmds)
+
+def load_batch(config_obj: ConfigObj):
+ # requires legacy backend patch
+ raise NotImplementedError('batch loading not implemented')
+
+def load_tree(config_obj: ConfigObj):
+ # requires vyconf backend patch
+ raise NotImplementedError('tree loading not implemented')
+
+def load_legacy(config_obj: ConfigObj):
+ """Legacy load from file or configtree.
+ """
+ if isinstance(config_obj, ConfigTree):
+ config_file = NamedTemporaryFile(delete=False).name
+ Path(config_file).write_text(config_obj.to_string())
+ else:
+ config_file = config_obj
+
+ config = LoadConfig()
+
+ try:
+ config.load_config(config_file)
+ except VyOSError as e:
+ raise LoadConfigError(e) from e
+ finally:
+ if isinstance(config_obj, ConfigTree):
+ Path(config_file).unlink()
+
+def load(config_obj: ConfigObj, strict: bool = True,
+ switch: Variety = 'legacy'):
+ type_hints = get_type_hints(load)
+ switch_choice = get_args(type_hints['switch'])
+ if switch not in switch_choice:
+ raise ValueError(f'invalid switch: {switch}')
+
+ check_session(strict, switch)
+
+ if migration_needed(config_obj):
+ config_obj = migrate(config_obj)
+
+ func = getattr(thismod, f'load_{switch}')
+ func(config_obj)