source: nscp/scripts/python/test_nsca.py @ 6fbb689

0.4.00.4.10.4.2
Last change on this file since 6fbb689 was 6fbb689, checked in by Michael Medin <michael@…>, 2 years ago

strange...

  • Property mode set to 100644
File size: 9.6 KB
Line 
1from NSCP import Settings, Registry, Core, log, status, log_error, sleep
2from test_helper import BasicTest, TestResult, Callable, setup_singleton, install_testcases, init_testcases, shutdown_testcases
3import plugin_pb2
4from types import *
5import socket
6import uuid
7import unicodedata
8#import _thread
9#sync = _thread.allocate_lock()
10
11import threading
12sync = threading.RLock()
13
14core = Core.get()
15
16def isOpen(ip, port):
17        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
18        try:
19                s.connect((ip, int(port)))
20                s.shutdown(2)
21                return True
22        except:
23                return False
24
25class NSCAMessage:
26        uuid = None
27        source = None
28        command = None
29        status = None
30        message = None
31        perfdata = None
32        got_response = False
33        got_simple_response = False
34
35        def __init__(self, command):
36                if type(command) == 'unicode':
37                        try:
38                                self.uuid = command.decode('ascii', 'replace')
39                        except UnicodeDecodeError:
40                                self.uuid = command
41                else:
42                        self.uuid = command
43                #self.uuid = unicodedata.normalize('NFKD', command).encode('ascii','ignore')
44                self.command = command
45                self.source = None
46                self.status = None
47                self.message = None
48                self.perfdata = None
49                self.got_response = False
50                self.got_simple_response = False
51               
52               
53        def copy_changed_attributes(self, other):
54                if other.source:
55                        self.source = other.source
56                if other.command:
57                        self.command = other.command
58                if other.status:
59                        self.status = other.status
60                if other.message:
61                        self.message = other.message
62                if other.perfdata:
63                        self.perfdata = other.perfdata
64                if other.got_simple_response:
65                        self.got_simple_response = True
66       
67        def __str__(self):
68                return 'Message: %s (%s, %s, %s)'%(self.uuid, self.source, self.command, self.status)
69
70class NSCAServerTest(BasicTest):
71        instance = None
72        key = ''
73        reg = None
74        _responses = {}
75       
76        def has_response(self, id):
77                with sync:
78                        return id in self._responses
79       
80        def get_response(self, id):
81                with sync:
82                        if id in self._responses:
83                                return self._responses[id]
84                        msg = NSCAMessage(id)
85                        self._responses[id] = msg
86                        return msg
87
88        def set_response(self, msg):
89                with sync:
90                        if msg.uuid in self._responses:
91                                self._responses[msg.uuid].copy_changed_attributes(msg)
92                        else:
93                                self._responses[msg.uuid] = msg
94                       
95
96        def del_response(self, id):
97                with sync:
98                        del self._responses[id]
99                       
100       
101        def desc(self):
102                return 'Testcase for NSCA protocol'
103
104        def title(self):
105                return 'NSCA Server test'
106
107        def setup(self, plugin_id, prefix):
108                self.key = '_%stest_command'%prefix
109                self.reg = Registry.get(plugin_id)
110                self.reg.simple_subscription('nsca_test_inbox', NSCAServerTest.simple_inbox_handler)
111                self.reg.subscription('nsca_test_inbox', NSCAServerTest.inbox_handler)
112
113        def simple_inbox_handler(channel, source, command, code, message, perf):
114                instance = NSCAServerTest.getInstance()
115                return instance.simple_inbox_handler_wrapped(channel, source, command, code, message, perf)
116        simple_inbox_handler = Callable(simple_inbox_handler)
117
118        def inbox_handler(channel, request):
119                instance = NSCAServerTest.getInstance()
120                return instance.inbox_handler_wrapped(channel, request)
121        inbox_handler = Callable(inbox_handler)
122       
123        def simple_inbox_handler_wrapped(self, channel, source, command, status, message, perf):
124                log('Got simple message %s on %s'%(command, channel))
125                msg = NSCAMessage(command)
126                msg.source = source
127                msg.status = status
128                msg.message = message
129                msg.perfdata = perf
130                msg.got_simple_response = True
131                self.set_response(msg)
132                return True
133
134        def inbox_handler_wrapped(self, channel, request):
135                message = plugin_pb2.SubmitRequestMessage()
136                message.ParseFromString(request)
137                command = message.payload[0].command
138                log('Got message %s on %s'%(command, channel))
139               
140                msg = NSCAMessage(command)
141                msg.got_response = True
142                self.set_response(msg)
143                return None
144               
145        def teardown(self):
146                None
147               
148        def wait_and_validate(self, uuid, result, msg, perf, tag):
149                found = False
150                for i in range(0,10):
151                        if self.has_response(uuid):
152                                rmsg = self.get_response(uuid)
153                                if not rmsg.got_simple_response or not rmsg.got_response:
154                                        for j in range(0,3):
155                                                rmsg = self.get_response(uuid)
156                                                if rmsg.got_simple_response and rmsg.got_response:
157                                                        log('Got delayed response %s'%uuid)
158                                                else:
159                                                        log('Waiting for delayed response %s (%d/10)'%(uuid, j+1))
160                                       
161                                result.add_message(rmsg.got_response, 'Testing to recieve message using %s'%tag)
162                                result.add_message(rmsg.got_simple_response, 'Testing to recieve simple message using %s'%tag)
163                                result.assert_equals(rmsg.command, uuid, 'Verify that command is sent through using %s'%tag)
164                                result.assert_contains(rmsg.message, msg, 'Verify that message is sent through using %s'%tag)
165                               
166                                #result.assert_equals(rmsg.last_source, source, 'Verify that source is sent through')
167                                #result.assert_equals(rmsg.perfdata, perf, 'Verify that performance data is sent through using %s'%tag)
168                                self.del_response(uuid)
169                                return True
170                        else:
171                                log('Waiting for %s (%d/10)'%(uuid, i+1))
172                                sleep(1)
173                result.add_message(False, 'Failed to recieve message %s using %s'%(uuid, tag))
174                return False
175       
176               
177        def submit_payload(self, encryption, source, status, msg, perf, tag):
178                message = plugin_pb2.SubmitRequestMessage()
179               
180                message.header.version = plugin_pb2.Common.VERSION_1
181                message.header.recipient_id = "test_rp"
182                message.channel = 'nsca_test_outbox'
183                host = message.header.hosts.add()
184                host.address = "127.0.0.1:15667"
185                host.id = "test_rp"
186                enc = host.metadata.add()
187                enc.key = "encryption"
188                enc.value = encryption
189                enc = host.metadata.add()
190                enc.key = "password"
191                enc.value = 'pwd-%s'%encryption
192
193                uid = str(uuid.uuid4())
194                payload = message.payload.add()
195                payload.result = status
196                payload.command = uid
197                payload.message = '%s - %s'%(uid, msg)
198                payload.source = source
199                (result_code, err) = core.submit('nsca_test_outbox', message.SerializeToString())
200
201                result = TestResult('Testing payload submission (via API): %s'%tag)
202                result.add_message(len(err) == 0, 'Testing to send message using %s/sbp'%tag, err)
203                self.wait_and_validate(uid, result, msg, perf, '%s/spb'%tag)
204                return result
205               
206        def submit_via_exec(self, encryption, source, status, msg, perf, tag):
207                uid = str(uuid.uuid4())
208       
209                args = [
210                        #'--exec', 'submit',
211                        '--alias', uid,
212                        '--result', '%d'%status,
213                        '--message', '%s - %s'%(uid, msg),
214                        '--encryption', encryption,
215                        '--password', 'pwd-%s'%encryption,
216                        '--host', '127.0.0.1:15667'
217                        ]
218                (result_code, result_message) = core.simple_exec('any', 'nsca_submit', args)
219                result = TestResult('Testing payload submission (via command line exec): %s'%tag)
220               
221                result.add_message(result_code == 0, 'Testing to send message using %s/exec:1'%tag)
222                result.add_message(len(result_message) == 1, 'Testing to send message using %s/exec:2'%tag, len(result_message))
223                result.add_message(len(result_message[0]) == 0, 'Testing to send message using %s/exec:3'%tag, result_message[0])
224                self.wait_and_validate(uid, result, msg, perf, '%s/exec'%tag)
225                return result
226
227        def test_one_crypto_full(self, encryption, state, key):
228                result = TestResult('Testing %s/%s'%(encryption, key))
229                result.add(self.submit_payload(encryption, '%ssrc%s'%(key, key), state, '%smsg%s'%(key, key), '', '%s/%s'%(state, encryption)))
230                result.add(self.submit_via_exec(encryption, '%ssrc%s'%(key, key), state, '%smsg%s'%(key, key), '', '%s/%s'%(state, encryption)))
231                return result
232
233        def test_one_crypto(self, crypto):
234                conf = Settings.get()
235                conf.set_string('/settings/NSCA/test_nsca_server', 'encryption', '%s'%crypto)
236                conf.set_string('/settings/NSCA/test_nsca_server', 'password', 'pwd-%s'%crypto)
237                core.reload('test_nsca_server')
238                result = TestResult('Testing encryption algorith: %s'%crypto)
239                result.add_message(isOpen('localhost', 15667), 'Checking that port is open')
240                result.add(self.test_one_crypto_full(crypto, status.UNKNOWN, 'unknown'))
241                result.add(self.test_one_crypto_full(crypto, status.OK, 'ok'))
242                result.add(self.test_one_crypto_full(crypto, status.WARNING, 'warn'))
243                result.add(self.test_one_crypto_full(crypto, status.CRITICAL, 'crit'))
244                return result
245
246        def run_test(self):
247                result = TestResult()
248                cryptos = ["none", "xor", "des", "3des", "cast128", "xtea", "blowfish", "twofish", "rc2", "aes", "aes256", "aes192", "aes128", "serpent", "gost", "3way"]
249                for c in cryptos:
250                        result.add_message(True, 'Testing crypto: %s'%c)
251                        result.add(self.test_one_crypto(c))
252               
253                return result
254               
255        def install(self, arguments):
256                conf = Settings.get()
257                conf.set_string('/modules', 'test_nsca_server', 'NSCAServer')
258                conf.set_string('/modules', 'test_nsca_client', 'NSCAClient')
259                conf.set_string('/modules', 'pytest', 'PythonScript')
260
261                conf.set_string('/settings/pytest/scripts', 'test_nsca', 'test_nsca.py')
262               
263                conf.set_string('/settings/NSCA/test_nsca_server', 'port', '15667')
264                conf.set_string('/settings/NSCA/test_nsca_server', 'inbox', 'nsca_test_inbox')
265                conf.set_string('/settings/NSCA/test_nsca_server', 'encryption', '1')
266
267                conf.set_string('/settings/NSCA/test_nsca_client/targets', 'nsca_test_local', 'nsca://127.0.0.1:15667')
268                conf.set_string('/settings/NSCA/test_nsca_client', 'channel', 'nsca_test_outbox')
269               
270                conf.save()
271
272        def uninstall(self):
273                None
274
275        def help(self):
276                None
277
278        def init(self, plugin_id):
279                None
280
281        def shutdown(self):
282                None
283               
284        def require_boot(self):
285                return True
286               
287
288setup_singleton(NSCAServerTest)
289
290all_tests = [NSCAServerTest]
291
292def __main__():
293        install_testcases(all_tests)
294       
295def init(plugin_id, plugin_alias, script_alias):
296        init_testcases(plugin_id, plugin_alias, script_alias, all_tests)
297
298def shutdown():
299        shutdown_testcases()
Note: See TracBrowser for help on using the repository browser.