1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
|
#!/usr/bin/env python3
#
# Copyright (C) 2019-2020 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 unittest
from vyos.validate import is_ipv4
from psutil import process_iter
from vyos.config import Config
from vyos.configsession import ConfigSession, ConfigSessionError
from vyos.util import read_file
SNMPD_CONF = '/etc/snmp/snmpd.conf'
base_path = ['service', 'snmp']
def get_config_value(key):
tmp = read_file(SNMPD_CONF)
return re.findall(r'\n?{}\s+(.*)'.format(key), tmp)
class TestSNMPService(unittest.TestCase):
def setUp(self):
self.session = ConfigSession(os.getpid())
env = self.session.get_session_env()
self.config = Config(session_env=env)
def tearDown(self):
# Delete SNNP configuration
self.session.delete(base_path)
self.session.commit()
del self.session
def test_snmp(self):
""" Check if SNMP can be configured and service runs """
clients = ['192.0.2.1', '2001:db8::1']
networks = ['192.0.2.128/25', '2001:db8:babe::/48']
listen = ['127.0.0.1', '::1']
for auth in ['ro', 'rw']:
community = 'VyOS' + auth
self.session.set(base_path + ['community', community, 'authorization', auth])
for client in clients:
self.session.set(base_path + ['community', community, 'client', client])
for network in networks:
self.session.set(base_path + ['community', community, 'network', network])
for addr in listen:
self.session.set(base_path + ['listen-address', addr])
self.session.set(base_path + ['contact', 'maintainers@vyos.io'])
self.session.set(base_path + ['location', 'qemu'])
self.session.commit()
# verify listen address, it will be returned as
# ['unix:/run/snmpd.socket,udp:127.0.0.1:161,udp6:[::1]:161']
# thus we need to transfor this into a proper list
config = get_config_value('agentaddress')[0]
expected = 'unix:/run/snmpd.socket'
for addr in listen:
if is_ipv4(addr):
expected += ',udp:{}:161'.format(addr)
else:
expected += ',udp6:[{}]:161'.format(addr)
try:
self.assertTrue(expected in config)
except:
print("expected: {}".format(expected))
print("config: {}".format(config))
# Check for running process
self.assertTrue("snmpd" in (p.name() for p in process_iter()))
def test_snmpv3(self):
""" Check if SNMPv3 can be configured and service runs"""
self.session.set(base_path + ['v3', 'engineid', '0xaffedeadbeef'])
self.session.set(base_path + ['v3', 'group', 'default', 'mode', 'ro'])
# check validate() - a view must be created before this can be comitted
with self.assertRaises(ConfigSessionError):
self.session.commit()
self.session.set(base_path + ['v3', 'view', 'default', 'oid', '1'])
self.session.set(base_path + ['v3', 'group', 'default', 'view', 'default'])
self.session.commit()
# create user
for authpriv in ['auth', 'privacy']:
self.session.set(base_path + ['v3', 'user', 'vyos', authpriv, 'plaintext-key', 'vyos1234'])
self.session.set(base_path + ['v3', 'user', 'vyos', 'group', 'default'])
# TODO: read in config file and check values
# Check for running process
self.assertTrue("snmpd" in (p.name() for p in process_iter()))
if __name__ == '__main__':
unittest.main()
|
