diff --git a/python/vyos/config.py b/python/vyos/config.py
index 6fececd76..0ca41718f 100644
--- a/python/vyos/config.py
+++ b/python/vyos/config.py
@@ -1,568 +1,573 @@
# Copyright 2017, 2019-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/>.
"""
A library for reading VyOS running config data.
This library is used internally by all config scripts of VyOS,
but its API should be considered stable and safe to use
in user scripts.
Note that this module will not work outside VyOS.
Node taxonomy
#############
There are multiple types of config tree nodes in VyOS, each requires
its own set of operations.
*Leaf nodes* (such as "address" in interfaces) can have values, but cannot
have children.
Leaf nodes can have one value, multiple values, or no values at all.
For example, "system host-name" is a single-value leaf node,
"system name-server" is a multi-value leaf node (commonly abbreviated "multi node"),
and "system ip disable-forwarding" is a valueless leaf node.
Non-leaf nodes cannot have values, but they can have child nodes. They are divided into
two classes depending on whether the names of their children are fixed or not.
For example, under "system", the names of all valid child nodes are predefined
("login", "name-server" etc.).
To the contrary, children of the "system task-scheduler task" node can have arbitrary names.
Such nodes are called *tag nodes*. This terminology is confusing but we keep using it for lack
of a better word. No one remembers if the "tag" in "task Foo" is "task" or "Foo",
but the distinction is irrelevant in practice.
Configuration modes
###################
VyOS has two distinct modes: operational mode and configuration mode. When a user logins,
the CLI is in the operational mode. In this mode, only the running (effective) config is accessible for reading.
When a user enters the "configure" command, a configuration session is setup. Every config session
has its *proposed* (or *session*) config built on top of the current running config. When changes are commited, if commit succeeds,
the proposed config is merged into the running config.
In configuration mode, "base" functions like `exists`, `return_value` return values from the session config,
while functions prefixed "effective" return values from the running config.
In operational mode, all functions return values from the running config.
"""
import re
import json
from copy import deepcopy
from typing import Union
import vyos.configtree
from vyos.xml_ref import multi_to_list
from vyos.xml_ref import from_source
from vyos.xml_ref import ext_dict_merge
from vyos.xml_ref import relative_defaults
from vyos.utils.dict import get_sub_dict
from vyos.utils.dict import mangle_dict_keys
from vyos.configsource import ConfigSource
from vyos.configsource import ConfigSourceSession
class ConfigDict(dict):
_from_defaults = {}
_dict_kwargs = {}
def from_defaults(self, path: list[str]) -> bool:
return from_source(self._from_defaults, path)
@property
def kwargs(self) -> dict:
return self._dict_kwargs
def config_dict_merge(src: dict, dest: Union[dict, ConfigDict]) -> ConfigDict:
if not isinstance(dest, ConfigDict):
dest = ConfigDict(dest)
return ext_dict_merge(src, dest)
class Config(object):
"""
The class of config access objects.
Internally, in the current implementation, this object is *almost* stateless,
the only state it keeps is relative *config path* for convenient access to config
subtrees.
"""
def __init__(self, session_env=None, config_source=None):
if config_source is None:
self._config_source = ConfigSourceSession(session_env)
else:
if not isinstance(config_source, ConfigSource):
raise TypeError("config_source not of type ConfigSource")
self._config_source = config_source
self._level = []
self._dict_cache = {}
(self._running_config,
self._session_config) = self._config_source.get_configtree_tuple()
+ def get_config_tree(self, effective=False):
+ if effective:
+ return self._running_config
+ return self._session_config
+
def _make_path(self, path):
# Backwards-compatibility stuff: original implementation used string paths
# libvyosconfig paths are lists, but since node names cannot contain whitespace,
# splitting at whitespace is reasonably safe.
# It may cause problems with exists() when it's used for checking values,
# since values may contain whitespace.
if isinstance(path, str):
path = re.split(r'\s+', path)
elif isinstance(path, list):
pass
else:
raise TypeError("Path must be a whitespace-separated string or a list")
return (self._level + path)
def set_level(self, path):
"""
Set the *edit level*, that is, a relative config tree path.
Once set, all operations will be relative to this path,
for example, after ``set_level("system")``, calling
``exists("name-server")`` is equivalent to calling
``exists("system name-server"`` without ``set_level``.
Args:
path (str|list): relative config path
"""
# Make sure there's always a space between default path (level)
# and path supplied as method argument
# XXX: for small strings in-place concatenation is not a problem
if isinstance(path, str):
if path:
self._level = re.split(r'\s+', path)
else:
self._level = []
elif isinstance(path, list):
self._level = path.copy()
else:
raise TypeError("Level path must be either a whitespace-separated string or a list")
def get_level(self):
"""
Gets the current edit level.
Returns:
str: current edit level
"""
return(self._level.copy())
def exists(self, path):
"""
Checks if a node or value with given path exists in the proposed config.
Args:
path (str): Configuration tree path
Returns:
True if node or value exists in the proposed config, False otherwise
Note:
This function should not be used outside of configuration sessions.
In operational mode scripts, use ``exists_effective``.
"""
if self._session_config is None:
return False
# Assume the path is a node path first
if self._session_config.exists(self._make_path(path)):
return True
else:
# If that check fails, it may mean the path has a value at the end.
# libvyosconfig exists() works only for _nodes_, not _values_
# libvyattacfg also worked for values, so we emulate that case here
if isinstance(path, str):
path = re.split(r'\s+', path)
path_without_value = path[:-1]
try:
# return_values() is safe to use with single-value nodes,
# it simply returns a single-item list in that case.
values = self._session_config.return_values(self._make_path(path_without_value))
# If we got this far, the node does exist and has values,
# so we need to check if it has the value in question among its values.
return (path[-1] in values)
except vyos.configtree.ConfigTreeError:
# Even the parent node doesn't exist at all
return False
def session_changed(self):
"""
Returns:
True if the config session has uncommited changes, False otherwise.
"""
return self._config_source.session_changed()
def in_session(self):
"""
Returns:
True if called from a configuration session, False otherwise.
"""
return self._config_source.in_session()
def show_config(self, path=[], default=None, effective=False):
"""
Args:
path (str list): Configuration tree path, or empty
default (str): Default value to return
Returns:
str: working configuration
"""
return self._config_source.show_config(path, default, effective)
def get_cached_root_dict(self, effective=False):
cached = self._dict_cache.get(effective, {})
if cached:
return cached
if effective:
config = self._running_config
else:
config = self._session_config
if config:
config_dict = json.loads(config.to_json())
else:
config_dict = {}
self._dict_cache[effective] = config_dict
return config_dict
def verify_mangling(self, key_mangling):
if not (isinstance(key_mangling, tuple) and \
(len(key_mangling) == 2) and \
isinstance(key_mangling[0], str) and \
isinstance(key_mangling[1], str)):
raise ValueError("key_mangling must be a tuple of two strings")
def get_config_dict(self, path=[], effective=False, key_mangling=None,
get_first_key=False, no_multi_convert=False,
no_tag_node_value_mangle=False,
with_defaults=False, with_recursive_defaults=False):
"""
Args:
path (str list): Configuration tree path, can be empty
effective=False: effective or session config
key_mangling=None: mangle dict keys according to regex and replacement
get_first_key=False: if k = path[:-1], return sub-dict d[k] instead of {k: d[k]}
no_multi_convert=False: if convert, return single value of multi node as list
Returns: a dict representation of the config under path
"""
kwargs = locals().copy()
del kwargs['self']
del kwargs['no_multi_convert']
del kwargs['with_defaults']
del kwargs['with_recursive_defaults']
lpath = self._make_path(path)
root_dict = self.get_cached_root_dict(effective)
conf_dict = get_sub_dict(root_dict, lpath, get_first_key=get_first_key)
rpath = lpath if get_first_key else lpath[:-1]
if not no_multi_convert:
conf_dict = multi_to_list(rpath, conf_dict)
if key_mangling is not None:
self.verify_mangling(key_mangling)
conf_dict = mangle_dict_keys(conf_dict,
key_mangling[0], key_mangling[1],
abs_path=rpath,
no_tag_node_value_mangle=no_tag_node_value_mangle)
if with_defaults or with_recursive_defaults:
defaults = self.get_config_defaults(**kwargs,
recursive=with_recursive_defaults)
conf_dict = config_dict_merge(defaults, conf_dict)
else:
conf_dict = ConfigDict(conf_dict)
# save optional args for a call to get_config_defaults
setattr(conf_dict, '_dict_kwargs', kwargs)
return conf_dict
def get_config_defaults(self, path=[], effective=False, key_mangling=None,
no_tag_node_value_mangle=False, get_first_key=False,
recursive=False) -> dict:
lpath = self._make_path(path)
root_dict = self.get_cached_root_dict(effective)
conf_dict = get_sub_dict(root_dict, lpath, get_first_key)
defaults = relative_defaults(lpath, conf_dict,
get_first_key=get_first_key,
recursive=recursive)
rpath = lpath if get_first_key else lpath[:-1]
if key_mangling is not None:
self.verify_mangling(key_mangling)
defaults = mangle_dict_keys(defaults,
key_mangling[0], key_mangling[1],
abs_path=rpath,
no_tag_node_value_mangle=no_tag_node_value_mangle)
return defaults
def merge_defaults(self, config_dict: ConfigDict, recursive=False):
if not isinstance(config_dict, ConfigDict):
raise TypeError('argument is not of type ConfigDict')
if not config_dict.kwargs:
raise ValueError('argument missing metadata')
args = config_dict.kwargs
d = self.get_config_defaults(**args, recursive=recursive)
config_dict = config_dict_merge(d, config_dict)
return config_dict
def is_multi(self, path):
"""
Args:
path (str): Configuration tree path
Returns:
True if a node can have multiple values, False otherwise.
Note:
It also returns False if node doesn't exist.
"""
self._config_source.set_level(self.get_level)
return self._config_source.is_multi(path)
def is_tag(self, path):
"""
Args:
path (str): Configuration tree path
Returns:
True if a node is a tag node, False otherwise.
Note:
It also returns False if node doesn't exist.
"""
self._config_source.set_level(self.get_level)
return self._config_source.is_tag(path)
def is_leaf(self, path):
"""
Args:
path (str): Configuration tree path
Returns:
True if a node is a leaf node, False otherwise.
Note:
It also returns False if node doesn't exist.
"""
self._config_source.set_level(self.get_level)
return self._config_source.is_leaf(path)
def return_value(self, path, default=None):
"""
Retrieve a value of single-value leaf node in the running or proposed config
Args:
path (str): Configuration tree path
default (str): Default value to return if node does not exist
Returns:
str: Node value, if it has any
None: if node is valueless *or* if it doesn't exist
Note:
Due to the issue with treatment of valueless nodes by this function,
valueless nodes should be checked with ``exists`` instead.
This function cannot be used outside a configuration session.
In operational mode scripts, use ``return_effective_value``.
"""
if self._session_config:
try:
value = self._session_config.return_value(self._make_path(path))
except vyos.configtree.ConfigTreeError:
value = None
else:
value = None
if not value:
return(default)
else:
return(value)
def return_values(self, path, default=[]):
"""
Retrieve all values of a multi-value leaf node in the running or proposed config
Args:
path (str): Configuration tree path
Returns:
str list: Node values, if it has any
[]: if node does not exist
Note:
This function cannot be used outside a configuration session.
In operational mode scripts, use ``return_effective_values``.
"""
if self._session_config:
try:
values = self._session_config.return_values(self._make_path(path))
except vyos.configtree.ConfigTreeError:
values = []
else:
values = []
if not values:
return(default.copy())
else:
return(values)
def list_nodes(self, path, default=[]):
"""
Retrieve names of all children of a tag node in the running or proposed config
Args:
path (str): Configuration tree path
Returns:
string list: child node names
"""
if self._session_config:
try:
nodes = self._session_config.list_nodes(self._make_path(path))
except vyos.configtree.ConfigTreeError:
nodes = []
else:
nodes = []
if not nodes:
return(default.copy())
else:
return(nodes)
def exists_effective(self, path):
"""
Checks if a node or value exists in the running (effective) config.
Args:
path (str): Configuration tree path
Returns:
True if node exists in the running config, False otherwise
Note:
This function is safe to use in operational mode. In configuration mode,
it ignores uncommited changes.
"""
if self._running_config is None:
return False
# Assume the path is a node path first
if self._running_config.exists(self._make_path(path)):
return True
else:
# If that check fails, it may mean the path has a value at the end.
# libvyosconfig exists() works only for _nodes_, not _values_
# libvyattacfg also worked for values, so we emulate that case here
if isinstance(path, str):
path = re.split(r'\s+', path)
path_without_value = path[:-1]
try:
# return_values() is safe to use with single-value nodes,
# it simply returns a single-item list in that case.
values = self._running_config.return_values(self._make_path(path_without_value))
# If we got this far, the node does exist and has values,
# so we need to check if it has the value in question among its values.
return (path[-1] in values)
except vyos.configtree.ConfigTreeError:
# Even the parent node doesn't exist at all
return False
def return_effective_value(self, path, default=None):
"""
Retrieve a values of a single-value leaf node in a running (effective) config
Args:
path (str): Configuration tree path
default (str): Default value to return if node does not exist
Returns:
str: Node value
"""
if self._running_config:
try:
value = self._running_config.return_value(self._make_path(path))
except vyos.configtree.ConfigTreeError:
value = None
else:
value = None
if not value:
return(default)
else:
return(value)
def return_effective_values(self, path, default=[]):
"""
Retrieve all values of a multi-value node in a running (effective) config
Args:
path (str): Configuration tree path
Returns:
str list: A list of values
"""
if self._running_config:
try:
values = self._running_config.return_values(self._make_path(path))
except vyos.configtree.ConfigTreeError:
values = []
else:
values = []
if not values:
return(default.copy())
else:
return(values)
def list_effective_nodes(self, path, default=[]):
"""
Retrieve names of all children of a tag node in the running config
Args:
path (str): Configuration tree path
Returns:
str list: child node names
"""
if self._running_config:
try:
nodes = self._running_config.list_nodes(self._make_path(path))
except vyos.configtree.ConfigTreeError:
nodes = []
else:
nodes = []
if not nodes:
return(default.copy())
else:
return(nodes)
diff --git a/python/vyos/config_mgmt.py b/python/vyos/config_mgmt.py
index 4ddabd6c2..0fc72e660 100644
--- a/python/vyos/config_mgmt.py
+++ b/python/vyos/config_mgmt.py
@@ -1,698 +1,698 @@
# 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/>.
import os
import re
import sys
import gzip
import logging
from typing import Optional, Tuple, Union
from filecmp import cmp
from datetime import datetime
from textwrap import dedent
from pathlib import Path
from tabulate import tabulate
from vyos.config import Config
from vyos.configtree import ConfigTree, ConfigTreeError, show_diff
from vyos.defaults import directories
from vyos.version import get_full_version_data
from vyos.utils.io import ask_yes_no
from vyos.utils.process import is_systemd_service_active
from vyos.utils.process import rc_cmd
-SAVE_CONFIG = '/opt/vyatta/sbin/vyatta-save-config.pl'
+SAVE_CONFIG = '/usr/libexec/vyos/vyos-save-config.py'
# created by vyatta-cfg-postinst
commit_post_hook_dir = '/etc/commit/post-hooks.d'
commit_hooks = {'commit_revision': '01vyos-commit-revision',
'commit_archive': '02vyos-commit-archive'}
DEFAULT_TIME_MINUTES = 10
timer_name = 'commit-confirm'
config_file = os.path.join(directories['config'], 'config.boot')
archive_dir = os.path.join(directories['config'], 'archive')
archive_config_file = os.path.join(archive_dir, 'config.boot')
commit_log_file = os.path.join(archive_dir, 'commits')
logrotate_conf = os.path.join(archive_dir, 'lr.conf')
logrotate_state = os.path.join(archive_dir, 'lr.state')
rollback_config = os.path.join(archive_dir, 'config.boot-rollback')
prerollback_config = os.path.join(archive_dir, 'config.boot-prerollback')
tmp_log_entry = '/tmp/commit-rev-entry'
logger = logging.getLogger('config_mgmt')
logger.setLevel(logging.INFO)
ch = logging.StreamHandler()
formatter = logging.Formatter('%(funcName)s: %(levelname)s:%(message)s')
ch.setFormatter(formatter)
logger.addHandler(ch)
def save_config(target):
cmd = f'{SAVE_CONFIG} {target}'
rc, out = rc_cmd(cmd)
if rc != 0:
logger.critical(f'save config failed: {out}')
def unsaved_commits() -> bool:
if get_full_version_data()['boot_via'] == 'livecd':
return False
tmp_save = '/tmp/config.running'
save_config(tmp_save)
ret = not cmp(tmp_save, config_file, shallow=False)
os.unlink(tmp_save)
return ret
def get_file_revision(rev: int):
revision = os.path.join(archive_dir, f'config.boot.{rev}.gz')
try:
with gzip.open(revision) as f:
r = f.read().decode()
except FileNotFoundError:
logger.warning(f'commit revision {rev} not available')
return ''
return r
def get_config_tree_revision(rev: int):
c = get_file_revision(rev)
return ConfigTree(c)
def is_node_revised(path: list = [], rev1: int = 1, rev2: int = 0) -> bool:
from vyos.configtree import DiffTree
left = get_config_tree_revision(rev1)
right = get_config_tree_revision(rev2)
diff_tree = DiffTree(left, right)
if diff_tree.add.exists(path) or diff_tree.sub.exists(path):
return True
return False
class ConfigMgmtError(Exception):
pass
class ConfigMgmt:
def __init__(self, session_env=None, config=None):
if session_env:
self._session_env = session_env
else:
self._session_env = None
if config is None:
config = Config()
d = config.get_config_dict(['system', 'config-management'],
key_mangling=('-', '_'),
get_first_key=True)
self.max_revisions = int(d.get('commit_revisions', 0))
self.locations = d.get('commit_archive', {}).get('location', [])
self.source_address = d.get('commit_archive',
{}).get('source_address', '')
if config.exists(['system', 'host-name']):
self.hostname = config.return_value(['system', 'host-name'])
else:
self.hostname = 'vyos'
# upload only on existence of effective values, notably, on boot.
# one still needs session self.locations (above) for setting
# post-commit hook in conf_mode script
path = ['system', 'config-management', 'commit-archive', 'location']
if config.exists_effective(path):
self.effective_locations = config.return_effective_values(path)
else:
self.effective_locations = []
# a call to compare without args is edit_level aware
edit_level = os.getenv('VYATTA_EDIT_LEVEL', '')
self.edit_path = [l for l in edit_level.split('/') if l]
self.active_config = config._running_config
self.working_config = config._session_config
# Console script functions
#
def commit_confirm(self, minutes: int=DEFAULT_TIME_MINUTES,
no_prompt: bool=False) -> Tuple[str,int]:
"""Commit with reboot to saved config in 'minutes' minutes if
'confirm' call is not issued.
"""
if is_systemd_service_active(f'{timer_name}.timer'):
msg = 'Another confirm is pending'
return msg, 1
if unsaved_commits():
W = '\nYou should save previous commits before commit-confirm !\n'
else:
W = ''
prompt_str = f'''
commit-confirm will automatically reboot in {minutes} minutes unless changes
are confirmed.\n
Proceed ?'''
prompt_str = W + prompt_str
if not no_prompt and not ask_yes_no(prompt_str, default=True):
msg = 'commit-confirm canceled'
return msg, 1
action = 'sg vyattacfg "/usr/bin/config-mgmt revert"'
cmd = f'sudo systemd-run --quiet --on-active={minutes}m --unit={timer_name} {action}'
rc, out = rc_cmd(cmd)
if rc != 0:
raise ConfigMgmtError(out)
# start notify
cmd = f'sudo -b /usr/libexec/vyos/commit-confirm-notify.py {minutes}'
os.system(cmd)
msg = f'Initialized commit-confirm; {minutes} minutes to confirm before reboot'
return msg, 0
def confirm(self) -> Tuple[str,int]:
"""Do not reboot to saved config following 'commit-confirm'.
Update commit log and archive.
"""
if not is_systemd_service_active(f'{timer_name}.timer'):
msg = 'No confirm pending'
return msg, 0
cmd = f'sudo systemctl stop --quiet {timer_name}.timer'
rc, out = rc_cmd(cmd)
if rc != 0:
raise ConfigMgmtError(out)
# kill notify
cmd = 'sudo pkill -f commit-confirm-notify.py'
rc, out = rc_cmd(cmd)
if rc != 0:
raise ConfigMgmtError(out)
entry = self._read_tmp_log_entry()
self._add_log_entry(**entry)
if self._archive_active_config():
self._update_archive()
msg = 'Reboot timer stopped'
return msg, 0
def revert(self) -> Tuple[str,int]:
"""Reboot to saved config, dropping commits from 'commit-confirm'.
"""
_ = self._read_tmp_log_entry()
# archived config will be reverted on boot
rc, out = rc_cmd('sudo systemctl reboot')
if rc != 0:
raise ConfigMgmtError(out)
return '', 0
def rollback(self, rev: int, no_prompt: bool=False) -> Tuple[str,int]:
"""Reboot to config revision 'rev'.
"""
from shutil import copy
msg = ''
if not self._check_revision_number(rev):
msg = f'Invalid revision number {rev}: must be 0 < rev < {maxrev}'
return msg, 1
prompt_str = 'Proceed with reboot ?'
if not no_prompt and not ask_yes_no(prompt_str, default=True):
msg = 'Canceling rollback'
return msg, 0
rc, out = rc_cmd(f'sudo cp {archive_config_file} {prerollback_config}')
if rc != 0:
raise ConfigMgmtError(out)
path = os.path.join(archive_dir, f'config.boot.{rev}.gz')
with gzip.open(path) as f:
config = f.read()
try:
with open(rollback_config, 'wb') as f:
f.write(config)
copy(rollback_config, config_file)
except OSError as e:
raise ConfigMgmtError from e
rc, out = rc_cmd('sudo systemctl reboot')
if rc != 0:
raise ConfigMgmtError(out)
return msg, 0
def compare(self, saved: bool=False, commands: bool=False,
rev1: Optional[int]=None,
rev2: Optional[int]=None) -> Tuple[str,int]:
"""General compare function for config file revisions:
revision n vs. revision m; working version vs. active version;
or working version vs. saved version.
"""
ct1 = self.active_config
ct2 = self.working_config
msg = 'No changes between working and active configurations.\n'
if saved:
ct1 = self._get_saved_config_tree()
ct2 = self.working_config
msg = 'No changes between working and saved configurations.\n'
if rev1 is not None:
if not self._check_revision_number(rev1):
return f'Invalid revision number {rev1}', 1
ct1 = self._get_config_tree_revision(rev1)
ct2 = self.working_config
msg = f'No changes between working and revision {rev1} configurations.\n'
if rev2 is not None:
if not self._check_revision_number(rev2):
return f'Invalid revision number {rev2}', 1
# compare older to newer
ct2 = ct1
ct1 = self._get_config_tree_revision(rev2)
msg = f'No changes between revisions {rev2} and {rev1} configurations.\n'
out = ''
path = [] if commands else self.edit_path
try:
if commands:
out = show_diff(ct1, ct2, path=path, commands=True)
else:
out = show_diff(ct1, ct2, path=path)
except ConfigTreeError as e:
return e, 1
if out:
msg = out
return msg, 0
def wrap_compare(self, options) -> Tuple[str,int]:
"""Interface to vyatta-cfg-run: args collected as 'options' to parse
for compare.
"""
cmnds = False
r1 = None
r2 = None
if 'commands' in options:
cmnds=True
options.remove('commands')
for i in options:
if not i.isnumeric():
options.remove(i)
if len(options) > 0:
r1 = int(options[0])
if len(options) > 1:
r2 = int(options[1])
return self.compare(commands=cmnds, rev1=r1, rev2=r2)
# Initialization and post-commit hooks for conf-mode
#
def initialize_revision(self):
"""Initialize config archive, logrotate conf, and commit log.
"""
mask = os.umask(0o002)
os.makedirs(archive_dir, exist_ok=True)
self._add_logrotate_conf()
if (not os.path.exists(commit_log_file) or
self._get_number_of_revisions() == 0):
user = self._get_user()
via = 'init'
comment = ''
self._add_log_entry(user, via, comment)
# add empty init config before boot-config load for revision
# and diff consistency
if self._archive_active_config():
self._update_archive()
os.umask(mask)
def commit_revision(self):
"""Update commit log and rotate archived config.boot.
commit_revision is called in post-commit-hooks, if
['commit-archive', 'commit-revisions'] is configured.
"""
if os.getenv('IN_COMMIT_CONFIRM', ''):
self._new_log_entry(tmp_file=tmp_log_entry)
return
self._add_log_entry()
if self._archive_active_config():
self._update_archive()
def commit_archive(self):
"""Upload config to remote archive.
"""
from vyos.remote import upload
hostname = self.hostname
t = datetime.now()
timestamp = t.strftime('%Y%m%d_%H%M%S')
remote_file = f'config.boot-{hostname}.{timestamp}'
source_address = self.source_address
for location in self.effective_locations:
upload(archive_config_file, f'{location}/{remote_file}',
source_host=source_address)
# op-mode functions
#
def get_raw_log_data(self) -> list:
"""Return list of dicts of log data:
keys: [timestamp, user, commit_via, commit_comment]
"""
log = self._get_log_entries()
res_l = []
for line in log:
d = self._get_log_entry(line)
res_l.append(d)
return res_l
@staticmethod
def format_log_data(data: list) -> str:
"""Return formatted log data as str.
"""
res_l = []
for l_no, l in enumerate(data):
time_d = datetime.fromtimestamp(int(l['timestamp']))
time_str = time_d.strftime("%Y-%m-%d %H:%M:%S")
res_l.append([l_no, time_str,
f"by {l['user']}", f"via {l['commit_via']}"])
if l['commit_comment'] != 'commit': # default comment
res_l.append([None, l['commit_comment']])
ret = tabulate(res_l, tablefmt="plain")
return ret
@staticmethod
def format_log_data_brief(data: list) -> str:
"""Return 'brief' form of log data as str.
Slightly compacted format used in completion help for
'rollback'.
"""
res_l = []
for l_no, l in enumerate(data):
time_d = datetime.fromtimestamp(int(l['timestamp']))
time_str = time_d.strftime("%Y-%m-%d %H:%M:%S")
res_l.append(['\t', l_no, time_str,
f"{l['user']}", f"by {l['commit_via']}"])
ret = tabulate(res_l, tablefmt="plain")
return ret
def show_commit_diff(self, rev: int, rev2: Optional[int]=None,
commands: bool=False) -> str:
"""Show commit diff at revision number, compared to previous
revision, or to another revision.
"""
if rev2 is None:
out, _ = self.compare(commands=commands, rev1=rev, rev2=(rev+1))
return out
out, _ = self.compare(commands=commands, rev1=rev, rev2=rev2)
return out
def show_commit_file(self, rev: int) -> str:
return self._get_file_revision(rev)
# utility functions
#
@staticmethod
def _strip_version(s):
return re.split(r'(^//)', s, maxsplit=1, flags=re.MULTILINE)[0]
def _get_saved_config_tree(self):
with open(config_file) as f:
c = self._strip_version(f.read())
return ConfigTree(c)
def _get_file_revision(self, rev: int):
if rev not in range(0, self._get_number_of_revisions()):
raise ConfigMgmtError('revision not available')
revision = os.path.join(archive_dir, f'config.boot.{rev}.gz')
with gzip.open(revision) as f:
r = f.read().decode()
return r
def _get_config_tree_revision(self, rev: int):
c = self._strip_version(self._get_file_revision(rev))
return ConfigTree(c)
def _add_logrotate_conf(self):
conf: str = dedent(f"""\
{archive_config_file} {{
su root vyattacfg
rotate {self.max_revisions}
start 0
compress
copy
}}
""")
conf_file = Path(logrotate_conf)
conf_file.write_text(conf)
conf_file.chmod(0o644)
def _archive_active_config(self) -> bool:
mask = os.umask(0o113)
ext = os.getpid()
tmp_save = f'/tmp/config.boot.{ext}'
save_config(tmp_save)
try:
if cmp(tmp_save, archive_config_file, shallow=False):
# this will be the case on boot, as well as certain
# re-initialiation instances after delete/set
os.unlink(tmp_save)
return False
except FileNotFoundError:
pass
rc, out = rc_cmd(f'sudo mv {tmp_save} {archive_config_file}')
os.umask(mask)
if rc != 0:
logger.critical(f'mv file to archive failed: {out}')
return False
return True
@staticmethod
def _update_archive():
cmd = f"sudo logrotate -f -s {logrotate_state} {logrotate_conf}"
rc, out = rc_cmd(cmd)
if rc != 0:
logger.critical(f'logrotate failure: {out}')
@staticmethod
def _get_log_entries() -> list:
"""Return lines of commit log as list of strings
"""
entries = []
if os.path.exists(commit_log_file):
with open(commit_log_file) as f:
entries = f.readlines()
return entries
def _get_number_of_revisions(self) -> int:
l = self._get_log_entries()
return len(l)
def _check_revision_number(self, rev: int) -> bool:
# exclude init revision:
maxrev = self._get_number_of_revisions()
if not 0 <= rev < maxrev - 1:
return False
return True
@staticmethod
def _get_user() -> str:
import pwd
try:
user = os.getlogin()
except OSError:
try:
user = pwd.getpwuid(os.geteuid())[0]
except KeyError:
user = 'unknown'
return user
def _new_log_entry(self, user: str='', commit_via: str='',
commit_comment: str='', timestamp: Optional[int]=None,
tmp_file: str=None) -> Optional[str]:
# Format log entry and return str or write to file.
#
# Usage is within a post-commit hook, using env values. In case of
# commit-confirm, it can be written to a temporary file for
# inclusion on 'confirm'.
from time import time
if timestamp is None:
timestamp = int(time())
if not user:
user = self._get_user()
if not commit_via:
commit_via = os.getenv('COMMIT_VIA', 'other')
if not commit_comment:
commit_comment = os.getenv('COMMIT_COMMENT', 'commit')
# the commit log reserves '|' as field demarcation, so replace in
# comment if present; undo this in _get_log_entry, below
if re.search(r'\|', commit_comment):
commit_comment = commit_comment.replace('|', '%%')
entry = f'|{timestamp}|{user}|{commit_via}|{commit_comment}|\n'
mask = os.umask(0o113)
if tmp_file is not None:
try:
with open(tmp_file, 'w') as f:
f.write(entry)
except OSError as e:
logger.critical(f'write to {tmp_file} failed: {e}')
os.umask(mask)
return None
os.umask(mask)
return entry
@staticmethod
def _get_log_entry(line: str) -> dict:
log_fmt = re.compile(r'\|.*\|\n?$')
keys = ['user', 'commit_via', 'commit_comment', 'timestamp']
if not log_fmt.match(line):
logger.critical(f'Invalid log format {line}')
return {}
timestamp, user, commit_via, commit_comment = (
tuple(line.strip().strip('|').split('|')))
commit_comment = commit_comment.replace('%%', '|')
d = dict(zip(keys, [user, commit_via,
commit_comment, timestamp]))
return d
def _read_tmp_log_entry(self) -> dict:
try:
with open(tmp_log_entry) as f:
entry = f.read()
os.unlink(tmp_log_entry)
except OSError as e:
logger.critical(f'error on file {tmp_log_entry}: {e}')
return self._get_log_entry(entry)
def _add_log_entry(self, user: str='', commit_via: str='',
commit_comment: str='', timestamp: Optional[int]=None):
mask = os.umask(0o113)
entry = self._new_log_entry(user=user, commit_via=commit_via,
commit_comment=commit_comment,
timestamp=timestamp)
log_entries = self._get_log_entries()
log_entries.insert(0, entry)
if len(log_entries) > self.max_revisions:
log_entries = log_entries[:-1]
try:
with open(commit_log_file, 'w') as f:
f.writelines(log_entries)
except OSError as e:
logger.critical(e)
os.umask(mask)
# entry_point for console script
#
def run():
from argparse import ArgumentParser, REMAINDER
config_mgmt = ConfigMgmt()
for s in list(commit_hooks):
if sys.argv[0].replace('-', '_').endswith(s):
func = getattr(config_mgmt, s)
try:
func()
except Exception as e:
print(f'{s}: {e}')
sys.exit(0)
parser = ArgumentParser()
subparsers = parser.add_subparsers(dest='subcommand')
commit_confirm = subparsers.add_parser('commit_confirm',
help="Commit with opt-out reboot to saved config")
commit_confirm.add_argument('-t', dest='minutes', type=int,
default=DEFAULT_TIME_MINUTES,
help="Minutes until reboot, unless 'confirm'")
commit_confirm.add_argument('-y', dest='no_prompt', action='store_true',
help="Execute without prompt")
subparsers.add_parser('confirm', help="Confirm commit")
subparsers.add_parser('revert', help="Revert commit-confirm")
rollback = subparsers.add_parser('rollback',
help="Rollback to earlier config")
rollback.add_argument('--rev', type=int,
help="Revision number for rollback")
rollback.add_argument('-y', dest='no_prompt', action='store_true',
help="Excute without prompt")
compare = subparsers.add_parser('compare',
help="Compare config files")
compare.add_argument('--saved', action='store_true',
help="Compare session config with saved config")
compare.add_argument('--commands', action='store_true',
help="Show difference between commands")
compare.add_argument('--rev1', type=int, default=None,
help="Compare revision with session config or other revision")
compare.add_argument('--rev2', type=int, default=None,
help="Compare revisions")
wrap_compare = subparsers.add_parser('wrap_compare',
help="Wrapper interface for vyatta-cfg-run")
wrap_compare.add_argument('--options', nargs=REMAINDER)
args = vars(parser.parse_args())
func = getattr(config_mgmt, args['subcommand'])
del args['subcommand']
res = ''
try:
res, rc = func(**args)
except ConfigMgmtError as e:
print(e)
sys.exit(1)
if res:
print(res)
sys.exit(rc)
diff --git a/python/vyos/configsession.py b/python/vyos/configsession.py
index e8918d577..6d4b2af59 100644
--- a/python/vyos/configsession.py
+++ b/python/vyos/configsession.py
@@ -1,237 +1,237 @@
# configsession -- the write API for the VyOS running config
# Copyright (C) 2019-2023 VyOS maintainers and contributors
#
# 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, write to the Free Software Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
import os
import re
import sys
import subprocess
from vyos.utils.process import is_systemd_service_running
from vyos.utils.dict import dict_to_paths
CLI_SHELL_API = '/bin/cli-shell-api'
SET = '/opt/vyatta/sbin/my_set'
DELETE = '/opt/vyatta/sbin/my_delete'
COMMENT = '/opt/vyatta/sbin/my_comment'
COMMIT = '/opt/vyatta/sbin/my_commit'
DISCARD = '/opt/vyatta/sbin/my_discard'
SHOW_CONFIG = ['/bin/cli-shell-api', 'showConfig']
LOAD_CONFIG = ['/bin/cli-shell-api', 'loadFile']
MIGRATE_LOAD_CONFIG = ['/usr/libexec/vyos/vyos-load-config.py']
-SAVE_CONFIG = ['/opt/vyatta/sbin/vyatta-save-config.pl']
+SAVE_CONFIG = ['/usr/libexec/vyos/vyos-save-config.py']
INSTALL_IMAGE = ['/opt/vyatta/sbin/install-image', '--url']
REMOVE_IMAGE = ['/opt/vyatta/bin/vyatta-boot-image.pl', '--del']
GENERATE = ['/opt/vyatta/bin/vyatta-op-cmd-wrapper', 'generate']
SHOW = ['/opt/vyatta/bin/vyatta-op-cmd-wrapper', 'show']
RESET = ['/opt/vyatta/bin/vyatta-op-cmd-wrapper', 'reset']
OP_CMD_ADD = ['/opt/vyatta/bin/vyatta-op-cmd-wrapper', 'add']
OP_CMD_DELETE = ['/opt/vyatta/bin/vyatta-op-cmd-wrapper', 'delete']
# Default "commit via" string
APP = "vyos-http-api"
# When started as a service rather than from a user shell,
# the process lacks the VyOS-specific environment that comes
# from bash configs, so we have to inject it
# XXX: maybe it's better to do via a systemd environment file
def inject_vyos_env(env):
env['VYATTA_CFG_GROUP_NAME'] = 'vyattacfg'
env['VYATTA_USER_LEVEL_DIR'] = '/opt/vyatta/etc/shell/level/admin'
env['VYATTA_PROCESS_CLIENT'] = 'gui2_rest'
env['VYOS_HEADLESS_CLIENT'] = 'vyos_http_api'
env['vyatta_bindir']= '/opt/vyatta/bin'
env['vyatta_cfg_templates'] = '/opt/vyatta/share/vyatta-cfg/templates'
env['vyatta_configdir'] = '/opt/vyatta/config'
env['vyatta_datadir'] = '/opt/vyatta/share'
env['vyatta_datarootdir'] = '/opt/vyatta/share'
env['vyatta_libdir'] = '/opt/vyatta/lib'
env['vyatta_libexecdir'] = '/opt/vyatta/libexec'
env['vyatta_op_templates'] = '/opt/vyatta/share/vyatta-op/templates'
env['vyatta_prefix'] = '/opt/vyatta'
env['vyatta_sbindir'] = '/opt/vyatta/sbin'
env['vyatta_sysconfdir'] = '/opt/vyatta/etc'
env['vyos_bin_dir'] = '/usr/bin'
env['vyos_cfg_templates'] = '/opt/vyatta/share/vyatta-cfg/templates'
env['vyos_completion_dir'] = '/usr/libexec/vyos/completion'
env['vyos_configdir'] = '/opt/vyatta/config'
env['vyos_conf_scripts_dir'] = '/usr/libexec/vyos/conf_mode'
env['vyos_datadir'] = '/opt/vyatta/share'
env['vyos_datarootdir']= '/opt/vyatta/share'
env['vyos_libdir'] = '/opt/vyatta/lib'
env['vyos_libexec_dir'] = '/usr/libexec/vyos'
env['vyos_op_scripts_dir'] = '/usr/libexec/vyos/op_mode'
env['vyos_op_templates'] = '/opt/vyatta/share/vyatta-op/templates'
env['vyos_prefix'] = '/opt/vyatta'
env['vyos_sbin_dir'] = '/usr/sbin'
env['vyos_validators_dir'] = '/usr/libexec/vyos/validators'
# if running the vyos-configd daemon, inject the vyshim env var
if is_systemd_service_running('vyos-configd.service'):
env['vyshim'] = '/usr/sbin/vyshim'
return env
class ConfigSessionError(Exception):
pass
class ConfigSession(object):
"""
The write API of VyOS.
"""
def __init__(self, session_id, app=APP):
"""
Creates a new config session.
Args:
session_id (str): Session identifier
app (str): Application name, purely informational
Note:
The session identifier MUST be globally unique within the system.
The best practice is to only have one ConfigSession object per process
and used the PID for the session identifier.
"""
env_str = subprocess.check_output([CLI_SHELL_API, 'getSessionEnv', str(session_id)])
self.__session_id = session_id
# Extract actual variables from the chunk of shell it outputs
# XXX: it's better to extend cli-shell-api to provide easily readable output
env_list = re.findall(r'([A-Z_]+)=([^;\s]+)', env_str.decode())
session_env = os.environ
session_env = inject_vyos_env(session_env)
for k, v in env_list:
session_env[k] = v
self.__session_env = session_env
self.__session_env["COMMIT_VIA"] = app
self.__run_command([CLI_SHELL_API, 'setupSession'])
def __del__(self):
try:
output = subprocess.check_output([CLI_SHELL_API, 'teardownSession'], env=self.__session_env).decode().strip()
if output:
print("cli-shell-api teardownSession output for sesion {0}: {1}".format(self.__session_id, output), file=sys.stderr)
except Exception as e:
print("Could not tear down session {0}: {1}".format(self.__session_id, e), file=sys.stderr)
def __run_command(self, cmd_list):
p = subprocess.Popen(cmd_list, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, env=self.__session_env)
(stdout_data, stderr_data) = p.communicate()
output = stdout_data.decode()
result = p.wait()
if result != 0:
raise ConfigSessionError(output)
return output
def get_session_env(self):
return self.__session_env
def set(self, path, value=None):
if not value:
value = []
else:
value = [value]
self.__run_command([SET] + path + value)
def set_section(self, path: list, d: dict):
try:
for p in dict_to_paths(d):
self.set(path + p)
except (ValueError, ConfigSessionError) as e:
raise ConfigSessionError(e)
def delete(self, path, value=None):
if not value:
value = []
else:
value = [value]
self.__run_command([DELETE] + path + value)
def load_section(self, path: list, d: dict):
try:
self.delete(path)
if d:
for p in dict_to_paths(d):
self.set(path + p)
except (ValueError, ConfigSessionError) as e:
raise ConfigSessionError(e)
def comment(self, path, value=None):
if not value:
value = [""]
else:
value = [value]
self.__run_command([COMMENT] + path + value)
def commit(self):
out = self.__run_command([COMMIT])
return out
def discard(self):
self.__run_command([DISCARD])
def show_config(self, path, format='raw'):
config_data = self.__run_command(SHOW_CONFIG + path)
if format == 'raw':
return config_data
def load_config(self, file_path):
out = self.__run_command(LOAD_CONFIG + [file_path])
return out
def migrate_and_load_config(self, file_path):
out = self.__run_command(MIGRATE_LOAD_CONFIG + [file_path])
return out
def save_config(self, file_path):
out = self.__run_command(SAVE_CONFIG + [file_path])
return out
def install_image(self, url):
out = self.__run_command(INSTALL_IMAGE + [url])
return out
def remove_image(self, name):
out = self.__run_command(REMOVE_IMAGE + [name])
return out
def generate(self, path):
out = self.__run_command(GENERATE + path)
return out
def show(self, path):
out = self.__run_command(SHOW + path)
return out
def reset(self, path):
out = self.__run_command(RESET + path)
return out
def add_container_image(self, name):
out = self.__run_command(OP_CMD_ADD + ['container', 'image'] + [name])
return out
def delete_container_image(self, name):
out = self.__run_command(OP_CMD_DELETE + ['container', 'image'] + [name])
return out
def show_container_image(self):
out = self.__run_command(SHOW + ['container', 'image'])
return out
diff --git a/src/helpers/vyos-save-config.py b/src/helpers/vyos-save-config.py
new file mode 100755
index 000000000..2812155e8
--- /dev/null
+++ b/src/helpers/vyos-save-config.py
@@ -0,0 +1,55 @@
+#!/usr/bin/env python3
+#
+# Copyright (C) 2023 VyOS maintainers and contributors
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License version 2 or later as
+# published by the Free Software Foundation.
+#
+# This program 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 General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program. If not, see <http://www.gnu.org/licenses/>.
+#
+#
+import os
+import re
+import sys
+from tempfile import NamedTemporaryFile
+
+from vyos.config import Config
+from vyos.remote import urlc
+from vyos.component_version import system_footer
+from vyos.defaults import directories
+
+DEFAULT_CONFIG_PATH = os.path.join(directories['config'], 'config.boot')
+remote_save = None
+
+if len(sys.argv) > 1:
+ save_file = sys.argv[1]
+else:
+ save_file = DEFAULT_CONFIG_PATH
+
+if re.match(r'\w+:/', save_file):
+ try:
+ remote_save = urlc(save_file)
+ except ValueError as e:
+ sys.exit(e)
+
+config = Config()
+ct = config.get_config_tree(effective=True)
+
+write_file = save_file if remote_save is None else NamedTemporaryFile(delete=False).name
+with open(write_file, 'w') as f:
+ f.write(ct.to_string())
+ f.write("\n")
+ f.write(system_footer())
+
+if remote_save is not None:
+ try:
+ remote_save.upload(write_file)
+ finally:
+ os.remove(write_file)