_get_auth_token() - Code Metrics - Inspection of "Add action to list packages from packagecloud" - StackStorm/st2contrib - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#451)

by W

created 2016-03-21 23:17 UTC

_get_auth_token() D

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	8
dl	0
loc	26
rs	4

#!/usr/bin/env python

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

try:
    import requests
    requests.packages.urllib3.disable_warnings()
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 = 'https://localhost/api/v1/'
ST2_AUTH_BASE_URL = 'https://localhost/auth/v1/'
ST2_USERNAME = None
ST2_PASSWORD = None
ST2_API_KEY = None
ST2_AUTH_TOKEN = None
ST2_SSL_VERIFY = False

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])

# Sensu configuration

SENSU_HOST = 'localhost'
SENSU_PORT = 4567
SENSU_USER = ''
SENSU_PASS = ''

REGISTERED_WITH_ST2 = False
UNAUTHED = False
IS_API_KEY_AUTH = False

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

TOKEN_AUTH_HEADER = 'X-Auth-Token'
API_KEY_AUTH_HEADER = 'St2-Api-Key'


def _get_sensu_request_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, verbose=False):
    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)

        if verbose:
            print('Getting sensu stash info from URL: %s' % url)

        try:
            response = requests.get(url, headers=_get_sensu_request_headers())
        except requests.exceptions.ConnectionError:
            traceback.print_exc(limit=20)
            msg = 'Couldn\'t connect to sensu to get stash info. Is sensu running on %s:%s?' % (
                SENSU_HOST, SENSU_PORT
            )
            raise Exception(msg)

        if verbose:
            print('Sensu response code: %s.' % response.status_code)

        if response.status_code == 200:
            print("Check or client is stashed")
            sys.exit(0)


def _get_st2_request_headers():
    headers = {}

    if not UNAUTHED:
        if IS_API_KEY_AUTH:
            headers[API_KEY_AUTH_HEADER] = ST2_API_KEY
        else:
            if ST2_AUTH_TOKEN:
                headers[TOKEN_AUTH_HEADER] = ST2_AUTH_TOKEN
            else:
                pass

    return headers


def _create_trigger_type(verbose=False):
    try:
        url = _get_st2_triggers_base_url()
        payload = {
            'name': ST2_TRIGGERTYPE_NAME,
            'pack': ST2_TRIGGERTYPE_PACK,
            'description': 'Trigger type for sensu event handler.'
        }

        headers = _get_st2_request_headers()
        headers['Content-Type'] = 'application/json; charset=utf-8'

        if verbose:
            print('POST to URL %s for registering trigger. Body = %s, headers = %s.' %
                  (url, payload, headers))
        post_resp = requests.post(url, data=json.dumps(payload),
                                  headers=headers, verify=ST2_SSL_VERIFY)
    except:
        traceback.print_exc(limit=20)
        raise Exception('Unable to connect to st2 endpoint %s.' % url)
    else:
        status = post_resp.status_code
        if status in UNREACHABLE_CODES:
            msg = 'Got response %s. Invalid triggers endpoint %s. Check configuration!' % (
                status,
                url
            )
            raise Exception(msg)

        if status not in OK_CODES:
            msg = 'Failed to register trigger type %s.%s with st2. HTTP_CODE: %s' % (
                ST2_TRIGGERTYPE_PACK, ST2_TRIGGERTYPE_NAME, status
            )
            raise Exception(msg)
        else:
            print('Registered trigger type with st2.')


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


def _get_auth_token(verbose=False):
    auth_url = _get_auth_url()

    if verbose:
        print('Will POST to URL %s to get auth token.' % auth_url)

    try:
        resp = requests.post(auth_url, json.dumps({'ttl': 5 * 60}),
                             auth=(ST2_USERNAME, ST2_PASSWORD), verify=ST2_SSL_VERIFY)
    except:
        traceback.print_exc(limit=20)
        raise Exception('Unable to connect to st2 endpoint %s.' % auth_url)
    else:
        if resp.status_code in UNREACHABLE_CODES:
            msg = 'Got response %s. Invalid auth endpoint %s. Check configuration!' % (
                resp.status_code,
                auth_url
            )
            raise Exception(msg)
        if resp.status_code not in OK_CODES:
            msg = 'Cannot get a valid auth token from %s. HTTP_CODE: %s' % (
                auth_url,
                resp.status_code
            )
            raise Exception(msg)
        return resp.json()['token']


def _register_trigger_with_st2(verbose=False):
    triggers_url = _get_st2_triggers_url()

    try:
        headers = _get_st2_request_headers()
        if verbose:
            print('Will GET from URL %s for detecting trigger %s.' % (
                  triggers_url, ST2_TRIGGERTYPE_REF))
            print('Request headers: %s' % headers)
        get_resp = requests.get(triggers_url, headers=headers, verify=ST2_SSL_VERIFY)

        if get_resp.status_code != httplib.OK:
            _create_trigger_type(verbose=verbose)
        else:
            body = json.loads(get_resp.text)
            if len(body) == 0:
                _create_trigger_type(verbose=verbose)
    except:
        traceback.print_exc(limit=20)
        raise Exception('Unable to connect to st2 endpoint %s.' % triggers_url)
    else:
        if verbose:
            print('Successfully registered trigger %s with st2.' % ST2_TRIGGERTYPE_REF)


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


def _get_st2_triggers_url():
    url = urljoin(_get_st2_triggers_base_url() + '/', ST2_TRIGGERTYPE_REF)
    return url


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


def _post_webhook(url, body, verbose=False):
    headers = _get_st2_request_headers()
    headers['X-ST2-Integration'] = 'sensu.'
    headers['St2-Trace-Tag'] = body['payload']['id']
    headers['Content-Type'] = 'application/json; charset=utf-8'

    try:
        if verbose:
            print('Webhook POST: url: %s, headers: %s, body: %s\n' % (url, headers, body))
        r = requests.post(url, data=json.dumps(body), headers=headers, verify=False)
    except:
        raise Exception('Cannot connect to st2 endpoint %s.' % url)
    else:
        status = r.status_code

        if status in UNREACHABLE_CODES:
            msg = 'Webhook URL %s does not exist. Check StackStorm installation!'
            raise Exception(msg)

        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 _post_event_to_st2(payload, verbose=False):
    body = {}
    body['trigger'] = ST2_TRIGGERTYPE_REF

    try:
        body['payload'] = json.loads(payload)
    except:
        print('Invalid JSON payload %s.' % payload)
        sys.exit(3)

    try:
        client = body['payload']['client']['name']
        check = body['payload']['check']['name']
    except KeyError:
        print('Invalid payload spec %s.' % payload)

    if not _check_stash(client, check, verbose=verbose):
        try:
            _post_webhook(url=_get_st2_webhooks_url(), body=body, verbose=verbose)
            return True
        except:
            traceback.print_exc(limit=20)
            print('Cannot send event to st2.')
            sys.exit(4)
    return False


def _register_with_st2(verbose=False):
    global REGISTERED_WITH_ST2
    try:
        if not REGISTERED_WITH_ST2:
            if verbose:
                print('Checking if trigger %s registered with st2.' % ST2_TRIGGERTYPE_REF)
            _register_trigger_with_st2(verbose=verbose)
            REGISTERED_WITH_ST2 = True
    except:
        traceback.print_exc(limit=20)
        sys.stderr.write(
            'Failed registering with st2. Won\'t post event.\n')
        sys.exit(2)


def _set_config_opts(config_file, verbose=False, unauthed=False, ssl_verify=False):
    global ST2_USERNAME
    global ST2_PASSWORD
    global ST2_API_KEY
    global ST2_AUTH_TOKEN
    global ST2_API_BASE_URL
    global ST2_API_BASE_URL
    global ST2_AUTH_BASE_URL
    global ST2_SSL_VERIFY
    global SENSU_HOST
    global SENSU_PORT
    global SENSU_USER
    global SENSU_PASS
    global UNAUTHED
    global IS_API_KEY_AUTH

    UNAUTHED = unauthed
    ST2_SSL_VERIFY = ssl_verify

    if not os.path.exists(config_file):
        print('Configuration file %s not found. Exiting!!!' % config_file)
        sys.exit(1)

    with open(config_file) as f:
        config = yaml.safe_load(f)

        if verbose:
            print('Contents of config file: %s' % config)

        ST2_USERNAME = config['st2_username']
        ST2_PASSWORD = config['st2_password']
        ST2_API_KEY = config['st2_api_key']
        ST2_API_BASE_URL = config['st2_api_base_url']
        if not ST2_API_BASE_URL.endswith('/'):
            ST2_API_BASE_URL += '/'
        ST2_AUTH_BASE_URL = config['st2_auth_base_url']
        if not ST2_AUTH_BASE_URL.endswith('/'):
            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 ST2_API_KEY:
        IS_API_KEY_AUTH = True

    if verbose:
        print('Unauthed? : %s' % UNAUTHED)
        print('API key auth?: %s' % IS_API_KEY_AUTH)
        print('SSL_VERIFY? : %s' % ST2_SSL_VERIFY)

    if not UNAUTHED and not IS_API_KEY_AUTH:
        try:
            if not ST2_AUTH_TOKEN:
                if verbose:
                    print('No auth token found. Let\'s get one from StackStorm!')
                ST2_AUTH_TOKEN = _get_auth_token(verbose=verbose)
        except:
            traceback.print_exc(limit=20)
            print('Unable to negotiate an auth token. Exiting!')
            sys.exit(1)


def main(config_file, payload, verbose=False, unauthed=False, ssl_verify=False):
    _set_config_opts(config_file=config_file, unauthed=unauthed, verbose=verbose,
                     ssl_verify=ssl_verify)
    _register_with_st2(verbose=verbose)
    _post_event_to_st2(payload, verbose=verbose)


if __name__ == '__main__':
    parser = argparse.ArgumentParser(description='StackStorm sensu event handler.')
    parser.add_argument('config_path',
                        help='Exchange to listen on')
    parser.add_argument('--verbose', '-v', required=False, action='store_true',
                        help='Verbose mode.')
    parser.add_argument('--unauthed', '-u', required=False, action='store_true',
                        help='Allow to post to unauthed st2. E.g. when auth is disabled ' +
                        'server side.')
    parser.add_argument('--ssl-verify', '-k', required=False, action='store_true',
                        help='Turn on SSL verification for st2 APIs.')
    args = parser.parse_args()
    payload = sys.stdin.read().strip()
    main(config_file=args.config_path, payload=payload, verbose=args.verbose,
         unauthed=args.unauthed, ssl_verify=args.ssl_verify)


1			#!/usr/bin/env python
2
3			import argparse
4			import base64
5			import httplib
6			try:
7			import simplejson as json
8			except ImportError:
9			import json
10			import os
11			import sys
12			import traceback
13			from urlparse import urljoin
14
15			try:
16			import requests
17			requests.packages.urllib3.disable_warnings()
18			except ImportError:
19			raise ImportError('Missing dependency "requests". \
20			Do ``pip install requests``.')
21
22			try:
23			import yaml
24			except ImportError:
25			raise ImportError('Missing dependency "pyyaml". \
26			Do ``pip install pyyaml``.')
27
28			# ST2 configuration
29
30			ST2_API_BASE_URL = 'https://localhost/api/v1/'
31			ST2_AUTH_BASE_URL = 'https://localhost/auth/v1/'
32			ST2_USERNAME = None
33			ST2_PASSWORD = None
34			ST2_API_KEY = None
35			ST2_AUTH_TOKEN = None
36			ST2_SSL_VERIFY = False
37
38			ST2_AUTH_PATH = 'tokens'
39			ST2_WEBHOOKS_PATH = 'webhooks/st2'
40			ST2_TRIGGERS_PATH = 'triggertypes'
41			ST2_TRIGGERTYPE_PACK = 'sensu'
42			ST2_TRIGGERTYPE_NAME = 'event_handler'
43			ST2_TRIGGERTYPE_REF = '.'.join([ST2_TRIGGERTYPE_PACK, ST2_TRIGGERTYPE_NAME])
44
45			# Sensu configuration
46
47			SENSU_HOST = 'localhost'
48			SENSU_PORT = 4567
49			SENSU_USER = ''
50			SENSU_PASS = ''
51
52			REGISTERED_WITH_ST2 = False
53			UNAUTHED = False
54			IS_API_KEY_AUTH = False
55
56			OK_CODES = [httplib.OK, httplib.CREATED, httplib.ACCEPTED, httplib.CONFLICT]
57			UNREACHABLE_CODES = [httplib.NOT_FOUND]
58
59			TOKEN_AUTH_HEADER = 'X-Auth-Token'
60			API_KEY_AUTH_HEADER = 'St2-Api-Key'
61
62
63			def _get_sensu_request_headers():
64			b64auth = base64.b64encode(
65			"%s:%s" %
66			(SENSU_USER, SENSU_PASS))
67			auth_header = "BASIC %s" % b64auth
68			content_header = "application/json"
69			return {"Authorization": auth_header, "Content-Type": content_header}
70
71
72			def _check_stash(client, check, verbose=False):
73			sensu_api = "http://%s:%i" % (SENSU_HOST, SENSU_PORT)
74			endpoints = [
75			"silence/%s" % client,
76			"silence/%s/%s" % (client, check),
77			"silence/all/%s" % check]
78
79			for endpoint in endpoints:
80			url = "%s/stashes/%s" % (sensu_api, endpoint)
81
82			if verbose:
83			print('Getting sensu stash info from URL: %s' % url)
84
85			try:
86			response = requests.get(url, headers=_get_sensu_request_headers())
87			except requests.exceptions.ConnectionError:
88			traceback.print_exc(limit=20)
89			msg = 'Couldn\'t connect to sensu to get stash info. Is sensu running on %s:%s?' % (
90			SENSU_HOST, SENSU_PORT
91			)
92			raise Exception(msg)
93
94			if verbose:
95			print('Sensu response code: %s.' % response.status_code)
96
97			if response.status_code == 200:
98			print("Check or client is stashed")
99			sys.exit(0)
100
101
102			def _get_st2_request_headers():
103			headers = {}
104
105			if not UNAUTHED:
106			if IS_API_KEY_AUTH:
107			headers[API_KEY_AUTH_HEADER] = ST2_API_KEY
108			else:
109			if ST2_AUTH_TOKEN:
110			headers[TOKEN_AUTH_HEADER] = ST2_AUTH_TOKEN
111			else:
112			pass
113
114			return headers
115
116
117			def _create_trigger_type(verbose=False):
118			try:
119			url = _get_st2_triggers_base_url()
120			payload = {
121			'name': ST2_TRIGGERTYPE_NAME,
122			'pack': ST2_TRIGGERTYPE_PACK,
123			'description': 'Trigger type for sensu event handler.'
124			}
125
126			headers = _get_st2_request_headers()
127			headers['Content-Type'] = 'application/json; charset=utf-8'
128
129			if verbose:
130			print('POST to URL %s for registering trigger. Body = %s, headers = %s.' %
131			(url, payload, headers))
132			post_resp = requests.post(url, data=json.dumps(payload),
133			headers=headers, verify=ST2_SSL_VERIFY)
134			except:
135			traceback.print_exc(limit=20)
136			raise Exception('Unable to connect to st2 endpoint %s.' % url)
137			else:
138			status = post_resp.status_code
139			if status in UNREACHABLE_CODES:
140			msg = 'Got response %s. Invalid triggers endpoint %s. Check configuration!' % (
141			status,
142			url
143			)
144			raise Exception(msg)
145
146			if status not in OK_CODES:
147			msg = 'Failed to register trigger type %s.%s with st2. HTTP_CODE: %s' % (
148			ST2_TRIGGERTYPE_PACK, ST2_TRIGGERTYPE_NAME, status
149			)
150			raise Exception(msg)
151			else:
152			print('Registered trigger type with st2.')
153
154
155			def _get_auth_url():
156			return urljoin(ST2_AUTH_BASE_URL, ST2_AUTH_PATH)
157
158
159			def _get_auth_token(verbose=False):
160			auth_url = _get_auth_url()
161
162			if verbose:
163			print('Will POST to URL %s to get auth token.' % auth_url)
164
165			try:
166			resp = requests.post(auth_url, json.dumps({'ttl': 5 * 60}),
167			auth=(ST2_USERNAME, ST2_PASSWORD), verify=ST2_SSL_VERIFY)
168			except:
169			traceback.print_exc(limit=20)
170			raise Exception('Unable to connect to st2 endpoint %s.' % auth_url)
171			else:
172			if resp.status_code in UNREACHABLE_CODES:
173			msg = 'Got response %s. Invalid auth endpoint %s. Check configuration!' % (
174			resp.status_code,
175			auth_url
176			)
177			raise Exception(msg)
178			if resp.status_code not in OK_CODES:
179			msg = 'Cannot get a valid auth token from %s. HTTP_CODE: %s' % (
180			auth_url,
181			resp.status_code
182			)
183			raise Exception(msg)
184			return resp.json()['token']
185
186
187			def _register_trigger_with_st2(verbose=False):
188			triggers_url = _get_st2_triggers_url()
189
190			try:
191			headers = _get_st2_request_headers()
192			if verbose:
193			print('Will GET from URL %s for detecting trigger %s.' % (
194			triggers_url, ST2_TRIGGERTYPE_REF))
195			print('Request headers: %s' % headers)
196			get_resp = requests.get(triggers_url, headers=headers, verify=ST2_SSL_VERIFY)
197
198			if get_resp.status_code != httplib.OK:
199			_create_trigger_type(verbose=verbose)
200			else:
201			body = json.loads(get_resp.text)
202			if len(body) == 0:
203			_create_trigger_type(verbose=verbose)
204			except:
205			traceback.print_exc(limit=20)
206			raise Exception('Unable to connect to st2 endpoint %s.' % triggers_url)
207			else:
208			if verbose:
209			print('Successfully registered trigger %s with st2.' % ST2_TRIGGERTYPE_REF)
210
211
212			def _get_st2_triggers_base_url():
213			url = urljoin(ST2_API_BASE_URL, ST2_TRIGGERS_PATH)
214			return url
215
216
217			def _get_st2_triggers_url():
218			url = urljoin(_get_st2_triggers_base_url() + '/', ST2_TRIGGERTYPE_REF)
219			return url
220
221
222			def _get_st2_webhooks_url():
223			url = urljoin(ST2_API_BASE_URL, ST2_WEBHOOKS_PATH)
224			return url
225
226
227			def _post_webhook(url, body, verbose=False):
228			headers = _get_st2_request_headers()
229			headers['X-ST2-Integration'] = 'sensu.'
230			headers['St2-Trace-Tag'] = body['payload']['id']
231			headers['Content-Type'] = 'application/json; charset=utf-8'
232
233			try:
234			if verbose:
235			print('Webhook POST: url: %s, headers: %s, body: %s\n' % (url, headers, body))
236			r = requests.post(url, data=json.dumps(body), headers=headers, verify=False)
237			except:
238			raise Exception('Cannot connect to st2 endpoint %s.' % url)
239			else:
240			status = r.status_code
241
242			if status in UNREACHABLE_CODES:
243			msg = 'Webhook URL %s does not exist. Check StackStorm installation!'
244			raise Exception(msg)
245
246			if status not in OK_CODES:
247			sys.stderr.write('Failed posting sensu event to st2. HTTP_CODE: \
248			%d\n' % status)
249			else:
250			sys.stdout.write('Sent sensu event to st2. HTTP_CODE: \
251			%d\n' % status)
252
253
254			def _post_event_to_st2(payload, verbose=False):
255			body = {}
256			body['trigger'] = ST2_TRIGGERTYPE_REF
257
258			try:
259			body['payload'] = json.loads(payload)
260			except:
261			print('Invalid JSON payload %s.' % payload)
262			sys.exit(3)
263
264			try:
265			client = body['payload']['client']['name']
266			check = body['payload']['check']['name']
267			except KeyError:
268			print('Invalid payload spec %s.' % payload)
269
270			if not _check_stash(client, check, verbose=verbose):
271			try:
272			_post_webhook(url=_get_st2_webhooks_url(), body=body, verbose=verbose)
273			return True
274			except:
275			traceback.print_exc(limit=20)
276			print('Cannot send event to st2.')
277			sys.exit(4)
278			return False
279
280
281			def _register_with_st2(verbose=False):
282			global REGISTERED_WITH_ST2
283			try:
284			if not REGISTERED_WITH_ST2:
285			if verbose:
286			print('Checking if trigger %s registered with st2.' % ST2_TRIGGERTYPE_REF)
287			_register_trigger_with_st2(verbose=verbose)
288			REGISTERED_WITH_ST2 = True
289			except:
290			traceback.print_exc(limit=20)
291			sys.stderr.write(
292			'Failed registering with st2. Won\'t post event.\n')
293			sys.exit(2)
294
295
296			def _set_config_opts(config_file, verbose=False, unauthed=False, ssl_verify=False):
297			global ST2_USERNAME
298			global ST2_PASSWORD
299			global ST2_API_KEY
300			global ST2_AUTH_TOKEN
301			global ST2_API_BASE_URL
302			global ST2_API_BASE_URL
303			global ST2_AUTH_BASE_URL
304			global ST2_SSL_VERIFY
305			global SENSU_HOST
306			global SENSU_PORT
307			global SENSU_USER
308			global SENSU_PASS
309			global UNAUTHED
310			global IS_API_KEY_AUTH
311
312			UNAUTHED = unauthed
313			ST2_SSL_VERIFY = ssl_verify
314
315			if not os.path.exists(config_file):
316			print('Configuration file %s not found. Exiting!!!' % config_file)
317			sys.exit(1)
318
319			with open(config_file) as f:
320			config = yaml.safe_load(f)
321
322			if verbose:
323			print('Contents of config file: %s' % config)
324
325			ST2_USERNAME = config['st2_username']
326			ST2_PASSWORD = config['st2_password']
327			ST2_API_KEY = config['st2_api_key']
328			ST2_API_BASE_URL = config['st2_api_base_url']
329			if not ST2_API_BASE_URL.endswith('/'):
330			ST2_API_BASE_URL += '/'
331			ST2_AUTH_BASE_URL = config['st2_auth_base_url']
332			if not ST2_AUTH_BASE_URL.endswith('/'):
333			ST2_AUTH_BASE_URL += '/'
334			SENSU_HOST = config.get('sensu_host', 'localhost')
335			SENSU_PORT = config.get('sensu_port', '4567')
336			SENSU_USER = config.get('sensu_user', None)
337			SENSU_PASS = config.get('sensu_pass', None)
338
339			if ST2_API_KEY:
340			IS_API_KEY_AUTH = True
341
342			if verbose:
343			print('Unauthed? : %s' % UNAUTHED)
344			print('API key auth?: %s' % IS_API_KEY_AUTH)
345			print('SSL_VERIFY? : %s' % ST2_SSL_VERIFY)
346
347			if not UNAUTHED and not IS_API_KEY_AUTH:
348			try:
349			if not ST2_AUTH_TOKEN:
350			if verbose:
351			print('No auth token found. Let\'s get one from StackStorm!')
352			ST2_AUTH_TOKEN = _get_auth_token(verbose=verbose)
353			except:
354			traceback.print_exc(limit=20)
355			print('Unable to negotiate an auth token. Exiting!')
356			sys.exit(1)
357
358
359			def main(config_file, payload, verbose=False, unauthed=False, ssl_verify=False):
360			_set_config_opts(config_file=config_file, unauthed=unauthed, verbose=verbose,
361			ssl_verify=ssl_verify)
362			_register_with_st2(verbose=verbose)
363			_post_event_to_st2(payload, verbose=verbose)
364
365
366			if __name__ == '__main__':
367			parser = argparse.ArgumentParser(description='StackStorm sensu event handler.')
368			parser.add_argument('config_path',
369			help='Exchange to listen on')
370			parser.add_argument('--verbose', '-v', required=False, action='store_true',
371			help='Verbose mode.')
372			parser.add_argument('--unauthed', '-u', required=False, action='store_true',
373			help='Allow to post to unauthed st2. E.g. when auth is disabled ' +
374			'server side.')
375			parser.add_argument('--ssl-verify', '-k', required=False, action='store_true',
376			help='Turn on SSL verification for st2 APIs.')
377			args = parser.parse_args()
378			payload = sys.stdin.read().strip()
379			main(config_file=args.config_path, payload=payload, verbose=args.verbose,
380			unauthed=args.unauthed, ssl_verify=args.ssl_verify)
381

StackStorm / st2contrib

Pull Request — master (#451)

_get_auth_token() D

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like