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
117
118
119
120
121
122
123
124
|
#!/usr/bin/python -u
# -*- coding:utf-8 -*-
'''@package snmp-asterisk.py
Configure Asterisk SIP settings using SNMP get/set commands
Author: Robert Oliveira <olivecoder@gmail.com>
'''
from builtins import str
from builtins import object
###########################################################################
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# 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 snmp_passpersist
import settings
import syslog
POLLING_INTERVAL = 6
NET_SNMP_EXTEND_MIB = "1.3.6.1.4.1.8072.1.3.1"
BASE_OID = NET_SNMP_EXTEND_MIB + '.2'
class SipSettings(Settings):
FNAME = '/etc/asterisk/sip.conf'
class SnmpExtBase(object):
POLLING_INTERVAL = 60
BASE_OID = None
def __init__(self, base_oid=None):
if base_oid:
self.BASE_OID = base_oid
self.snmp = snmp_passpersist.PassPersist(self.BASE_OID)
self.update()
def update(self):
raise NotImplemented
def start(self):
self.snmp.start(self.update, self.POLLING_INTERVAL)
class SipSnmpExt(SnmpExtBase):
NAME_SUFIX = '.1.0'
VALUE_SUFIX = '.2.0'
def __init__(self):
self.settings = settings.Tmip()
self.assign_oids()
super(TmipSnmpExt, self).__init__()
self.snmp.register_setter(self.BASE_OID, self.setter)
def assign_oids(self):
self.oids = dict()
for i, section in enumerate(sorted(self.settings.sections())):
for j, attr in enumerate(sorted(self.settings.items(section))):
attr_name, value = attr
oid = '%i.%i' % (i + 1, j + 1)
self.oids[oid] = (section, attr_name)
def getNameOid(self, oid):
return '%s%s' % (oid, self.NAME_SUFIX)
def getValueOid(self, oid):
return '%s%s' % (oid, self.VALUE_SUFIX)
def update(self):
if self.settings.updateFromFile():
syslog.syslog(syslog.LOG_INFO, 'SipSnmp: reading settings')
self.assign_oids()
for oid, setting in list(self.oids.items()):
section, attr_name = setting
attr_value = self.settings.get(section, attr_name)
self.snmp.add_str(self.getNameOid(oid), attr_name)
self.snmp.add_str(self.getValueOid(oid), attr_value)
def getSectionAttr(self, oid):
key = oid.strip()
if key.endswith(self.VALUE_SUFIX):
key = key[:-len(self.VALUE_SUFIX)].rstrip('.')
if key.startswith(self.BASE_OID):
key = key[len(self.BASE_OID):].lstrip('.')
if key in self.oids:
return self.oids[key]
def setter(self, oid, type_, value):
try:
section, attr_name = self.getSectionAttr(oid)
except:
syslog.syslog(syslog.LOG_ERR, 'SipSnmp: unknown OID: %s' % oid)
return False
try:
self.settings.set(section, attr_name, str(value))
self.settings.write()
self.update()
self.snmp.commit()
return True
except Exception as e:
syslog.syslog(syslog.LOG_ERR,
'SipSnmp: error calling set(%s, %s, %s): %s' %
(section, attr_name, str(value), str(e)))
return False
if __name__ == '__main__':
sipext = SipSnmpExt()
sipext.start()
|
