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
125
126
127
128
129
130
131
|
""" Test creation of a PCP web daemon via the requests module """
#
# Copyright (C) 2013-2014 Red Hat.
#
# 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 2 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.
#
import requests, argparse, subprocess, os
parser = argparse.ArgumentParser(description='test_webapi.py pmwebapi check')
parser.add_argument('--version', action='version', version='1')
parser.add_argument('--host', required=True)
parser.add_argument('--port', default=44323)
args = parser.parse_args()
url = 'http://' + args.host + ':' + str(args.port) + '/'
devnull = os.open(os.devnull, os.O_RDWR)
os.unsetenv('http_proxy')
os.unsetenv('HTTP_PROXY')
# ------------------------------------------------------------------------
# test - create contexts
req = requests.get(url=url + 'pmapi/context?local=foo')
resp = req.json()
ctx_local = resp['context']
print('Received PM_CONTEXT_LOCAL #' + str(ctx_local))
req = requests.get(url=url + 'pmapi/context?hostname=' + args.host)
resp = req.json()
ctx_host = resp['context']
print('Received PM_CONTEXT_HOST #' + str(ctx_host))
# ------------------------------------------------------------------------
# all these should get an error
req = requests.get(url=url + 'pmapi/context?archivefile=/dev/null')
print('bad archive /dev/null response code ' + str(req.status_code))
req = requests.get(url=url + 'pmapi/context?archivefile=../etc/passwd')
print('bad archive ../etc/passwd response code ' + str(req.status_code))
req = requests.get(url=url + 'pmapi/context?archivefile=../../etc/shadow')
print('bad archive ../../etc/shadow response code ' + str(req.status_code))
req = requests.get(url=url + 'pmapi/NOSUCHAPI')
print('command NOSUCHAPI response code ' + str(req.status_code))
req = requests.get(url=url + 'pmapi/NOSUCHCONTEXT/_metric')
print('context NOSUCHCONTEXT response code ' + str(req.status_code))
req = requests.get(url=url + 'pmapi/0/_metric')
print('context 0 response code ' + str(req.status_code))
req = requests.get(url=url + 'random_nonpmwebapi_url')
print('non-pmwebapi response code ' + str(req.status_code))
# ------------------------------------------------------------------------
def test_metric_enumeration(ctx, prefix):
ctxurl = url + 'pmapi/' + str(ctx) + '/'
if (ctx == ctx_local):
procargs = ['pminfo', '-L', '-t']
dbg = 'local'
else:
procargs = ['pminfo', '-h', args.host, '-t']
dbg = 'host'
if (prefix != ''):
procargs.append(prefix)
dbg = dbg + '-' + prefix
proc = subprocess.Popen(procargs,
stdout=subprocess.PIPE,
stderr=devnull)
num_metrics = 0
# f = open('/tmp/pmcd-' + dbg, 'w')
while True:
line = proc.stdout.readline()
if (line == ''):
break # eof
if line.strip() == '':
continue # blank line pminfo sometimes does that
if line.find('Error: No PMCD agent') == -1 and \
line.find('Not known to the PMDA') == -1:
num_metrics = num_metrics + 1
# f.write(line)
# f.write('Total ' + str(num_metrics) + ' metrics\n')
# f.close()
testprefix='test #'+str(ctx)+' metric '+prefix+'.*'
print(testprefix + ' enumeration with pminfo #'+str(num_metrics))
req = requests.get(url=ctxurl + '_metric' + \
('?prefix='+prefix if prefix != '' else ''))
resp = req.json()
print(testprefix + ' enumeration with pmwebinfo #'+str(len(resp['metrics'])))
# web_metrics = 0
# f = open('/tmp/webd-' + dbg, 'w')
# for metric in resp['metrics']:
# web_metrics = web_metrics + 1
# f.write(metric['name'] + '\n')
# f.write('Total ' + str(web_metrics) + ' metrics\n')
# f.close()
if (abs(len(resp['metrics']) - num_metrics) < 10): # allow some variation
print(testprefix + ' enumeration match count PASS')
else:
print(testprefix + ' enumeration match count FAIL')
test_metric_enumeration(ctx_local,'')
test_metric_enumeration(ctx_host,'')
test_metric_enumeration(ctx_local,'kernel')
test_metric_enumeration(ctx_host,'kernel')
# ------------------------------------------------------------------------
# empty _fetch should get an error
req = requests.get(url=url + 'pmapi/'+str(ctx_host)+'/_fetch')
print('context #'+str(ctx_host)+' response code ' + str(req.status_code))
# ------------------------------------------------------------------------
|
