snmp.py
No OneTemporary
Actions

Size

12 KB

Referenced Files

None

Subscribers

None

snmp.py
View Options

	#!/usr/bin/env python3
	#
	# Copyright (C) 2018-2021 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

	from sys import exit

	from vyos.base import Warning
	from vyos.config import Config
	from vyos.configdict import dict_merge
	from vyos.configverify import verify_vrf
	from vyos.snmpv3_hashgen import plaintext_to_md5
	from vyos.snmpv3_hashgen import plaintext_to_sha1
	from vyos.snmpv3_hashgen import random
	from vyos.template import render
	from vyos.util import call
	from vyos.util import chmod_755
	from vyos.util import dict_search
	from vyos.validate import is_addr_assigned
	from vyos.version import get_version_data
	from vyos.xml import defaults
	from vyos import ConfigError
	from vyos import airbag
	airbag.enable()

	config_file_client = r'/etc/snmp/snmp.conf'
	config_file_daemon = r'/etc/snmp/snmpd.conf'
	config_file_access = r'/usr/share/snmp/snmpd.conf'
	config_file_user = r'/var/lib/snmp/snmpd.conf'
	systemd_override = r'/etc/systemd/system/snmpd.service.d/override.conf'
	systemd_service = 'snmpd.service'

	def get_config(config=None):
	if config:
	conf = config
	else:
	conf = Config()
	base = ['service', 'snmp']

	snmp = conf.get_config_dict(base, key_mangling=('-', '_'),
	get_first_key=True, no_tag_node_value_mangle=True)
	if not conf.exists(base):
	snmp.update({'deleted' : ''})

	if conf.exists(['service', 'lldp', 'snmp', 'enable']):
	snmp.update({'lldp_snmp' : ''})

	if 'deleted' in snmp:
	return snmp

	version_data = get_version_data()
	snmp['version'] = version_data['version']

	# create an internal snmpv3 user of the form 'vyosxxxxxxxxxxxxxxxx'
	snmp['vyos_user'] = 'vyos' + random(8)
	snmp['vyos_user_pass'] = random(16)

	# We have gathered the dict representation of the CLI, but there are default
	# options which we need to update into the dictionary retrived.
	default_values = defaults(base)

	# We can not merge defaults for tagNodes - those need to be blended in
	# per tagNode instance
	if 'listen_address' in default_values:
	del default_values['listen_address']
	if 'community' in default_values:
	del default_values['community']
	if 'trap_target' in default_values:
	del default_values['trap_target']
	if 'v3' in default_values:
	del default_values['v3']
	snmp = dict_merge(default_values, snmp)

	if 'listen_address' in snmp:
	default_values = defaults(base + ['listen-address'])
	for address in snmp['listen_address']:
	snmp['listen_address'][address] = dict_merge(
	default_values, snmp['listen_address'][address])

	# Always listen on localhost if an explicit address has been configured
	# This is a safety measure to not end up with invalid listen addresses
	# that are not configured on this system. See https://phabricator.vyos.net/T850
	if '127.0.0.1' not in snmp['listen_address']:
	tmp = {'127.0.0.1': {'port': '161'}}
	snmp['listen_address'] = dict_merge(tmp, snmp['listen_address'])

	if '::1' not in snmp['listen_address']:
	tmp = {'::1': {'port': '161'}}
	snmp['listen_address'] = dict_merge(tmp, snmp['listen_address'])

	if 'community' in snmp:
	default_values = defaults(base + ['community'])
	for community in snmp['community']:
	snmp['community'][community] = dict_merge(
	default_values, snmp['community'][community])

	if 'trap_target' in snmp:
	default_values = defaults(base + ['trap-target'])
	for trap in snmp['trap_target']:
	snmp['trap_target'][trap] = dict_merge(
	default_values, snmp['trap_target'][trap])

	if 'v3' in snmp:
	default_values = defaults(base + ['v3'])
	# tagNodes need to be merged in individually later on
	for tmp in ['user', 'group', 'trap_target']:
	del default_values[tmp]
	snmp['v3'] = dict_merge(default_values, snmp['v3'])

	for user_group in ['user', 'group']:
	if user_group in snmp['v3']:
	default_values = defaults(base + ['v3', user_group])
	for tmp in snmp['v3'][user_group]:
	snmp['v3'][user_group][tmp] = dict_merge(
	default_values, snmp['v3'][user_group][tmp])

	if 'trap_target' in snmp['v3']:
	default_values = defaults(base + ['v3', 'trap-target'])
	for trap in snmp['v3']['trap_target']:
	snmp['v3']['trap_target'][trap] = dict_merge(
	default_values, snmp['v3']['trap_target'][trap])

	return snmp

	def verify(snmp):
	if not snmp:
	return None

	if {'deleted', 'lldp_snmp'} <= set(snmp):
	raise ConfigError('Can not delete SNMP service, as LLDP still uses SNMP!')

	### check if the configured script actually exist
	if 'script_extensions' in snmp and 'extension_name' in snmp['script_extensions']:
	for extension, extension_opt in snmp['script_extensions']['extension_name'].items():
	if 'script' not in extension_opt:
	raise ConfigError(f'Script extension "{extension}" requires an actual script to be configured!')

	tmp = extension_opt['script']
	if not os.path.isfile(tmp):
	Warning(f'script "{tmp}" does not exist!')
	else:
	chmod_755(extension_opt['script'])

	if 'listen_address' in snmp:
	for address in snmp['listen_address']:
	# We only wan't to configure addresses that exist on the system.
	# Hint the user if they don't exist
	if not is_addr_assigned(address):
	Warning(f'SNMP listen address "{address}" not configured!')

	if 'trap_target' in snmp:
	for trap, trap_config in snmp['trap_target'].items():
	if 'community' not in trap_config:
	raise ConfigError(f'Trap target "{trap}" requires a community to be set!')

	verify_vrf(snmp)

	# bail out early if SNMP v3 is not configured
	if 'v3' not in snmp:
	return None

	if 'user' in snmp['v3']:
	for user, user_config in snmp['v3']['user'].items():
	if 'group' not in user_config:
	raise ConfigError(f'Group membership required for user "{user}"!')

	if 'plaintext_password' not in user_config['auth'] and 'encrypted_password' not in user_config['auth']:
	raise ConfigError(f'Must specify authentication encrypted-password or plaintext-password for user "{user}"!')

	if 'plaintext_password' not in user_config['privacy'] and 'encrypted_password' not in user_config['privacy']:
	raise ConfigError(f'Must specify privacy encrypted-password or plaintext-password for user "{user}"!')

	if 'group' in snmp['v3']:
	for group, group_config in snmp['v3']['group'].items():
	if 'seclevel' not in group_config:
	raise ConfigError(f'Must configure "seclevel" for group "{group}"!')
	if 'view' not in group_config:
	raise ConfigError(f'Must configure "view" for group "{group}"!')

	# Check if 'view' exists
	view = group_config['view']
	if 'view' not in snmp['v3'] or view not in snmp['v3']['view']:
	raise ConfigError(f'You must create view "{view}" first!')

	if 'view' in snmp['v3']:
	for view, view_config in snmp['v3']['view'].items():
	if 'oid' not in view_config:
	raise ConfigError(f'Must configure an "oid" for view "{view}"!')

	if 'trap_target' in snmp['v3']:
	for trap, trap_config in snmp['v3']['trap_target'].items():
	if 'plaintext_password' not in trap_config['auth'] and 'encrypted_password' not in trap_config['auth']:
	raise ConfigError(f'Must specify one of authentication encrypted-password or plaintext-password for trap "{trap}"!')

	if {'plaintext_password', 'encrypted_password'} <= set(trap_config['auth']):
	raise ConfigError(f'Can not specify both authentication encrypted-password and plaintext-password for trap "{trap}"!')

	if 'plaintext_password' not in trap_config['privacy'] and 'encrypted_password' not in trap_config['privacy']:
	raise ConfigError(f'Must specify one of privacy encrypted-password or plaintext-password for trap "{trap}"!')

	if {'plaintext_password', 'encrypted_password'} <= set(trap_config['privacy']):
	raise ConfigError(f'Can not specify both privacy encrypted-password and plaintext-password for trap "{trap}"!')

	if 'type' not in trap_config:
	raise ConfigError('SNMP v3 trap "type" must be specified!')

	return None

	def generate(snmp):

	#
	# As we are manipulating the snmpd user database we have to stop it first!
	# This is even save if service is going to be removed
	call(f'systemctl stop {systemd_service}')
	# Clean config files
	config_files = [config_file_client, config_file_daemon,
	config_file_access, config_file_user, systemd_override]
	for file in config_files:
	if os.path.isfile(file):
	os.unlink(file)

	if not snmp:
	return None

	if 'v3' in snmp:
	# net-snmp is now regenerating the configuration file in the background
	# thus we need to re-open and re-read the file as the content changed.
	# After that we can no read the encrypted password from the config and
	# replace the CLI plaintext password with its encrypted version.
	os.environ['vyos_libexec_dir'] = '/usr/libexec/vyos'

	if 'user' in snmp['v3']:
	for user, user_config in snmp['v3']['user'].items():
	if dict_search('auth.type', user_config) == 'sha':
	hash = plaintext_to_sha1
	else:
	hash = plaintext_to_md5

	if dict_search('auth.plaintext_password', user_config) is not None:
	tmp = hash(dict_search('auth.plaintext_password', user_config),
	dict_search('v3.engineid', snmp))

	snmp['v3']['user'][user]['auth']['encrypted_password'] = tmp
	del snmp['v3']['user'][user]['auth']['plaintext_password']

	call(f'/opt/vyatta/sbin/my_set service snmp v3 user "{user}" auth encrypted-password "{tmp}" > /dev/null')
	call(f'/opt/vyatta/sbin/my_delete service snmp v3 user "{user}" auth plaintext-password > /dev/null')

	if dict_search('privacy.plaintext_password', user_config) is not None:
	tmp = hash(dict_search('privacy.plaintext_password', user_config),
	dict_search('v3.engineid', snmp))

	snmp['v3']['user'][user]['privacy']['encrypted_password'] = tmp
	del snmp['v3']['user'][user]['privacy']['plaintext_password']

	call(f'/opt/vyatta/sbin/my_set service snmp v3 user "{user}" privacy encrypted-password "{tmp}" > /dev/null')
	call(f'/opt/vyatta/sbin/my_delete service snmp v3 user "{user}" privacy plaintext-password > /dev/null')

	# Write client config file
	render(config_file_client, 'snmp/etc.snmp.conf.j2', snmp)
	# Write server config file
	render(config_file_daemon, 'snmp/etc.snmpd.conf.j2', snmp)
	# Write access rights config file
	render(config_file_access, 'snmp/usr.snmpd.conf.j2', snmp)
	# Write access rights config file
	render(config_file_user, 'snmp/var.snmpd.conf.j2', snmp)
	# Write daemon configuration file
	render(systemd_override, 'snmp/override.conf.j2', snmp)

	return None

	def apply(snmp):
	# Always reload systemd manager configuration
	call('systemctl daemon-reload')

	if not snmp:
	return None

	# start SNMP daemon
	call(f'systemctl restart {systemd_service}')

	# Enable AgentX in FRR
	# This should be done for each daemon individually because common command
	# works only if all the daemons started with SNMP support
	frr_daemons_list = [
	'bgpd', 'ospf6d', 'ospfd', 'ripd', 'ripngd', 'isisd', 'ldpd', 'zebra'
	]
	for frr_daemon in frr_daemons_list:
	call(
	f'vtysh -c "configure terminal" -d {frr_daemon} -c "agentx" >/dev/null'
	)

	return None

	if __name__ == '__main__':
	try:
	c = get_config()
	verify(c)
	generate(c)
	apply(c)
	except ConfigError as e:
	print(e)
	exit(1)

File Metadata

Mime Type: text/x-script.python
Expires: Sat, Oct 17, 12:11 AM (1 d, 10 h)
Storage Engine: blob
Storage Format: Raw Data
Storage Handle: 4351150
Default Alt Text: snmp.py (12 KB)

snmp.pyNo OneTemporaryActions

snmp.pyView Options

File Metadata

Event Timeline

snmp.py
No OneTemporary
Actions

snmp.py
View Options