main() - Code Metrics - Inspection of "Add mailbox metadata to IMAP sensor" - StackStorm/st2contrib - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#352)

by James

created 2015-12-23 20:42 UTC

main() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	2
dl	0
loc	8
rs	9.4286

#!/usr/bin/env python

import base64
import httplib
try:
    import simplejson as json
except ImportError:
    import json
import os
import sys
from urlparse import urljoin

try:
    import requests
except ImportError:
    raise ImportError('Missing dependency "requests". \
        Do ``pip install requests``.')

try:
    import yaml
except ImportError:
    raise ImportError('Missing dependency "pyyaml". \
        Do ``pip install pyyaml``.')

# ST2 configuration

ST2_API_BASE_URL = 'http://localhost:9101/v1'
ST2_AUTH_BASE_URL = 'http://localhost:9100'
ST2_USERNAME = None
ST2_PASSWORD = None
ST2_AUTH_TOKEN = None

ST2_AUTH_PATH = 'tokens'
ST2_WEBHOOKS_PATH = 'webhooks/st2/'
ST2_TRIGGERS_PATH = 'triggertypes/'
ST2_TRIGGERTYPE_PACK = 'sensu'
ST2_TRIGGERTYPE_NAME = 'event_handler'
ST2_TRIGGERTYPE_REF = '.'.join([ST2_TRIGGERTYPE_PACK, ST2_TRIGGERTYPE_NAME])

REGISTERED_WITH_ST2 = False

OK_CODES = [httplib.OK, httplib.CREATED, httplib.ACCEPTED, httplib.CONFLICT]


def _get_headers():
    b64auth = base64.b64encode(
        "%s:%s" %
        (SENSU_USER, SENSU_PASS))
    auth_header = "BASIC %s" % b64auth
    content_header = "application/json"
    return {"Authorization": auth_header, "Content-Type": content_header}


def _check_stash(client, check):
    sensu_api = "http://%s:%i" % (SENSU_HOST, SENSU_PORT)
    endpoints = [
        "silence/%s" % client,
        "silence/%s/%s" % (client, check),
        "silence/all/%s" % check]

    for endpoint in endpoints:
        url = "%s/stashes/%s" % (sensu_api, endpoint)
        response = requests.get(url, headers=_get_headers())
        # print "%s %s" % (url, str(response.status_code))
        if response.status_code == 200:
            print "Check or client is stashed"
            sys.exit(0)


def _create_trigger_type():
    try:
        url = _get_st2_triggers_url()
        payload = {
            'name': ST2_TRIGGERTYPE_NAME,
            'pack': ST2_TRIGGERTYPE_PACK,
            'description': 'Trigger type for sensu event handler.'
        }
        # sys.stdout.write('POST: %s: Body: %s\n' % (url, payload))
        headers = {}
        headers['Content-Type'] = 'application/json; charset=utf-8'

        if ST2_AUTH_TOKEN:
            headers['X-Auth-Token'] = ST2_AUTH_TOKEN

        post_resp = requests.post(url, data=json.dumps(payload),
                                  headers=headers)
    except:
        sys.stderr.write('Unable to register trigger type with st2.')
        raise
    else:
        status = post_resp.status_code
        if status not in OK_CODES:
            sys.stderr.write('Failed to register trigger type with st2. \
                HTTP_CODE: %d\n' %
                             status)
            raise
        else:
            sys.stdout.write('Registered trigger type with st2.\n')


def _get_auth_url():
    return urljoin(ST2_AUTH_BASE_URL, ST2_AUTH_PATH)


def _get_auth_token():
    global ST2_AUTH_TOKEN
    auth_url = _get_auth_url()
    try:
        resp = requests.post(auth_url, json.dumps({'ttl': 5 * 60}),
                             auth=(ST2_USERNAME, ST2_PASSWORD))
    except:
        sys.stderr.write('Cannot get auth token from st2. Will try unauthed.')
    else:
        if resp.status_code not in OK_CODES:
            sys.stderr.write("Cannot authenticate. Will try unauthed.")
            return
        ST2_AUTH_TOKEN = resp.json()['token']


def _register_with_st2():
    global REGISTERED_WITH_ST2
    try:
        url = urljoin(_get_st2_triggers_url(), ST2_TRIGGERTYPE_REF)
        # sys.stdout.write('GET: %s\n' % url)
        if not ST2_AUTH_TOKEN:
            _get_auth_token()

        if ST2_AUTH_TOKEN:
            get_resp = requests.get(url, headers={'X-Auth-Token':
                                                  ST2_AUTH_TOKEN})
        else:
            get_resp = requests.get(url)

        if get_resp.status_code != httplib.OK:
            _create_trigger_type()
        else:
            body = json.loads(get_resp.text)
            if len(body) == 0:
                _create_trigger_type()
    except:
        raise
    else:
        REGISTERED_WITH_ST2 = True


def _get_st2_triggers_url():
    url = urljoin(ST2_API_BASE_URL, ST2_TRIGGERS_PATH)
    return url


def _get_st2_webhooks_url():
    url = urljoin(ST2_API_BASE_URL, ST2_WEBHOOKS_PATH)
    return url


def _post_event_to_st2(url, body):
    headers = {}
    headers['X-ST2-Integration'] = 'sensu.'
    headers['Content-Type'] = 'application/json; charset=utf-8'
    if ST2_AUTH_TOKEN:
        headers['X-Auth-Token'] = ST2_AUTH_TOKEN
    try:
        # sys.stdout.write('POST: url: %s, body: %s\n' % (url, body))
        r = requests.post(url, data=json.dumps(body), headers=headers)
    except:
        sys.stderr.write('Cannot connect to st2 endpoint.')
    else:
        status = r.status_code
        if status not in OK_CODES:
            sys.stderr.write('Failed posting sensu event to st2. HTTP_CODE: \
                %d\n' % status)
        else:
            sys.stdout.write('Sent sensu event to st2. HTTP_CODE: \
                %d\n' % status)


def main(args):
    body = {}
    body['trigger'] = ST2_TRIGGERTYPE_REF
    body['payload'] = json.loads(sys.stdin.read().strip())
    client = body['payload']['client']['name']
    check = body['payload']['check']['name']
    if not _check_stash(client, check):
        _post_event_to_st2(_get_st2_webhooks_url(), body)

if __name__ == '__main__':
    if len(sys.argv) > 1:
        st2_config_file = sys.argv[1]
    else:
        sys.stderr.write('Error: config file missing.\n')
        sys.stderr.write('Usage: %s ST2_CONFIG_FILE\n' % sys.argv[0])
        exit(-1)

    try:
        if not os.path.exists(st2_config_file):
            sys.stderr.write('Configuration file not found. Exiting.\n')
            sys.exit(1)

        with open(st2_config_file) as f:
            config = yaml.safe_load(f)
            ST2_USERNAME = config['st2_username']
            ST2_PASSWORD = config['st2_password']
            ST2_API_BASE_URL = config['st2_api_base_url']
            ST2_AUTH_BASE_URL = config['st2_auth_base_url']
            SENSU_HOST = config.get('sensu_host', 'localhost')
            SENSU_PORT = config.get('sensu_port', '4567')
            SENSU_USER = config.get('sensu_user', None)
            SENSU_PASS = config.get('sensu_pass', None)

        if not REGISTERED_WITH_ST2:
            _register_with_st2()
    except Exception as e:
        sys.stderr.write(
            'Failed registering with st2. Won\'t post event.\n%s' % e)
    else:
        main(sys.argv)


1			#!/usr/bin/env python
2
3			import base64
4			import httplib
5			try:
6			import simplejson as json
7			except ImportError:
8			import json
9			import os
10			import sys
11			from urlparse import urljoin
12
13			try:
14			import requests
15			except ImportError:
16			raise ImportError('Missing dependency "requests". \
17			Do ``pip install requests``.')
18
19			try:
20			import yaml
21			except ImportError:
22			raise ImportError('Missing dependency "pyyaml". \
23			Do ``pip install pyyaml``.')
24
25			# ST2 configuration
26
27			ST2_API_BASE_URL = 'http://localhost:9101/v1'
28			ST2_AUTH_BASE_URL = 'http://localhost:9100'
29			ST2_USERNAME = None
30			ST2_PASSWORD = None
31			ST2_AUTH_TOKEN = None
32
33			ST2_AUTH_PATH = 'tokens'
34			ST2_WEBHOOKS_PATH = 'webhooks/st2/'
35			ST2_TRIGGERS_PATH = 'triggertypes/'
36			ST2_TRIGGERTYPE_PACK = 'sensu'
37			ST2_TRIGGERTYPE_NAME = 'event_handler'
38			ST2_TRIGGERTYPE_REF = '.'.join([ST2_TRIGGERTYPE_PACK, ST2_TRIGGERTYPE_NAME])
39
40			REGISTERED_WITH_ST2 = False
41
42			OK_CODES = [httplib.OK, httplib.CREATED, httplib.ACCEPTED, httplib.CONFLICT]
43
44
45			def _get_headers():
46			b64auth = base64.b64encode(
47			"%s:%s" %
48			(SENSU_USER, SENSU_PASS))
49			auth_header = "BASIC %s" % b64auth
50			content_header = "application/json"
51			return {"Authorization": auth_header, "Content-Type": content_header}
52
53
54			def _check_stash(client, check):
55			sensu_api = "http://%s:%i" % (SENSU_HOST, SENSU_PORT)
56			endpoints = [
57			"silence/%s" % client,
58			"silence/%s/%s" % (client, check),
59			"silence/all/%s" % check]
60
61			for endpoint in endpoints:
62			url = "%s/stashes/%s" % (sensu_api, endpoint)
63			response = requests.get(url, headers=_get_headers())
64			# print "%s %s" % (url, str(response.status_code))
65			if response.status_code == 200:
66			print "Check or client is stashed"
67			sys.exit(0)
68
69
70			def _create_trigger_type():
71			try:
72			url = _get_st2_triggers_url()
73			payload = {
74			'name': ST2_TRIGGERTYPE_NAME,
75			'pack': ST2_TRIGGERTYPE_PACK,
76			'description': 'Trigger type for sensu event handler.'
77			}
78			# sys.stdout.write('POST: %s: Body: %s\n' % (url, payload))
79			headers = {}
80			headers['Content-Type'] = 'application/json; charset=utf-8'
81
82			if ST2_AUTH_TOKEN:
83			headers['X-Auth-Token'] = ST2_AUTH_TOKEN
84
85			post_resp = requests.post(url, data=json.dumps(payload),
86			headers=headers)
87			except:
88			sys.stderr.write('Unable to register trigger type with st2.')
89			raise
90			else:
91			status = post_resp.status_code
92			if status not in OK_CODES:
93			sys.stderr.write('Failed to register trigger type with st2. \
94			HTTP_CODE: %d\n' %
95			status)
96			raise
97			else:
98			sys.stdout.write('Registered trigger type with st2.\n')
99
100
101			def _get_auth_url():
102			return urljoin(ST2_AUTH_BASE_URL, ST2_AUTH_PATH)
103
104
105			def _get_auth_token():
106			global ST2_AUTH_TOKEN
107			auth_url = _get_auth_url()
108			try:
109			resp = requests.post(auth_url, json.dumps({'ttl': 5 * 60}),
110			auth=(ST2_USERNAME, ST2_PASSWORD))
111			except:
112			sys.stderr.write('Cannot get auth token from st2. Will try unauthed.')
113			else:
114			if resp.status_code not in OK_CODES:
115			sys.stderr.write("Cannot authenticate. Will try unauthed.")
116			return
117			ST2_AUTH_TOKEN = resp.json()['token']
118
119
120			def _register_with_st2():
121			global REGISTERED_WITH_ST2
122			try:
123			url = urljoin(_get_st2_triggers_url(), ST2_TRIGGERTYPE_REF)
124			# sys.stdout.write('GET: %s\n' % url)
125			if not ST2_AUTH_TOKEN:
126			_get_auth_token()
127
128			if ST2_AUTH_TOKEN:
129			get_resp = requests.get(url, headers={'X-Auth-Token':
130			ST2_AUTH_TOKEN})
131			else:
132			get_resp = requests.get(url)
133
134			if get_resp.status_code != httplib.OK:
135			_create_trigger_type()
136			else:
137			body = json.loads(get_resp.text)
138			if len(body) == 0:
139			_create_trigger_type()
140			except:
141			raise
142			else:
143			REGISTERED_WITH_ST2 = True
144
145
146			def _get_st2_triggers_url():
147			url = urljoin(ST2_API_BASE_URL, ST2_TRIGGERS_PATH)
148			return url
149
150
151			def _get_st2_webhooks_url():
152			url = urljoin(ST2_API_BASE_URL, ST2_WEBHOOKS_PATH)
153			return url
154
155
156			def _post_event_to_st2(url, body):
157			headers = {}
158			headers['X-ST2-Integration'] = 'sensu.'
159			headers['Content-Type'] = 'application/json; charset=utf-8'
160			if ST2_AUTH_TOKEN:
161			headers['X-Auth-Token'] = ST2_AUTH_TOKEN
162			try:
163			# sys.stdout.write('POST: url: %s, body: %s\n' % (url, body))
164			r = requests.post(url, data=json.dumps(body), headers=headers)
165			except:
166			sys.stderr.write('Cannot connect to st2 endpoint.')
167			else:
168			status = r.status_code
169			if status not in OK_CODES:
170			sys.stderr.write('Failed posting sensu event to st2. HTTP_CODE: \
171			%d\n' % status)
172			else:
173			sys.stdout.write('Sent sensu event to st2. HTTP_CODE: \
174			%d\n' % status)
175
176
177			def main(args):
178			body = {}
179			body['trigger'] = ST2_TRIGGERTYPE_REF
180			body['payload'] = json.loads(sys.stdin.read().strip())
181			client = body['payload']['client']['name']
182			check = body['payload']['check']['name']
183			if not _check_stash(client, check):
184			_post_event_to_st2(_get_st2_webhooks_url(), body)
185
186			if __name__ == '__main__':
187			if len(sys.argv) > 1:
188			st2_config_file = sys.argv[1]
189			else:
190			sys.stderr.write('Error: config file missing.\n')
191			sys.stderr.write('Usage: %s ST2_CONFIG_FILE\n' % sys.argv[0])
192			exit(-1)
193
194			try:
195			if not os.path.exists(st2_config_file):
196			sys.stderr.write('Configuration file not found. Exiting.\n')
197			sys.exit(1)
198
199			with open(st2_config_file) as f:
200			config = yaml.safe_load(f)
201			ST2_USERNAME = config['st2_username']
202			ST2_PASSWORD = config['st2_password']
203			ST2_API_BASE_URL = config['st2_api_base_url']
204			ST2_AUTH_BASE_URL = config['st2_auth_base_url']
205			SENSU_HOST = config.get('sensu_host', 'localhost')
206			SENSU_PORT = config.get('sensu_port', '4567')
207			SENSU_USER = config.get('sensu_user', None)
208			SENSU_PASS = config.get('sensu_pass', None)
209
210			if not REGISTERED_WITH_ST2:
211			_register_with_st2()
212			except Exception as e:
213			sys.stderr.write(
214			'Failed registering with st2. Won\'t post event.\n%s' % e)
215			else:
216			main(sys.argv)
217

StackStorm / st2contrib

Pull Request — master (#352)

main() A

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like