_get_auth_token() - Code Metrics - Inspection of "[RFR]: Update for nagios handler code" - StackStorm/st2contrib - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#502)

unknown

created 2016-06-13 21:13 UTC

_get_auth_token() B

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	2
Bugs	0	Features	0

Metric	Value
cc	6
c	2
b	0
f	0
dl	0
loc	26
rs	7.5384

#!/usr/bin/env python

import httplib
try:
    import simplejson as json
except ImportError:
    import json
import os
import sys
import traceback
from urlparse import urljoin
import argparse
try:
    import requests
except ImportError:
    raise ImportError('Missing dependency requests. \
                      Do ``pip install requests``.')

try:
    import yaml
    requests.packages.urllib3.disable_warnings()
except ImportError:
    raise ImportError('Missing dependency pyyaml. \
                      Do ``pip install pyyaml``.')

# ST2 configuration

ST2_API_BASE_URL = None
ST2_AUTH_BASE_URL = None
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 = 'nagios'
ST2_TRIGGERTYPE_NAME = 'service_state_change'
ST2_TRIGGERTYPE_REF = '.'.join([ST2_TRIGGERTYPE_PACK, ST2_TRIGGERTYPE_NAME])

STATE_MESSAGE = {
    'OK': 'All is well on the Western front.',
    'WARNING': 'We gots a warning yo!',
    'UNKNOWN': 'It be unknown...',
    'CRITICAL': 'Critical!'
}

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 _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 nagios event handler.'
        }

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

        if verbose:
            print('POST to URL {0} for registering trigger. Body = {1}, '
                  'headers = {2}.\n'.format(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 {0}.'.format(url))
    else:
        status = post_resp.status_code
        if status in UNREACHABLE_CODES:
            msg = 'Got response {0}. Invalid triggers endpoint {1}.' \
                'Check configuration!'.format(status, url)
            raise Exception(msg)

        if status not in OK_CODES:
            msg = 'Failed to register trigger type {0}.{1} with st2. ' \
                'HTTP_CODE: {2}'.format(ST2_TRIGGERTYPE_PACK,
                                        ST2_TRIGGERTYPE_NAME, status)
            raise Exception(msg)
        else:
            print('Registered trigger type with st2.\n')


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 {0} to get auth token.\n'.format(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 {0}.'.
                        format(auth_url))
    else:
        if resp.status_code in UNREACHABLE_CODES:
            msg = 'Got response {0}. Invalid auth endpoint {1}. '\
                'Check configuration!'.format(resp.status_code, auth_url)
            raise Exception(msg)

        if resp.status_code not in OK_CODES:
            msg = 'Cannot get a valid auth token from {0}. '\
                'HTTP_CODE: {1}'.format(auth_url, resp.status_code)
            raise Exception(msg)

        return resp.json()['token']


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 _register_with_st2(verbose=False):
    global REGISTERED_WITH_ST2
    try:
        if not REGISTERED_WITH_ST2:
            if verbose:
                print('Checking if trigger "{0}" registered with st2.'
                      .format(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 _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 {0} for detecting trigger {1}.\n'
                  .format(triggers_url, ST2_TRIGGERTYPE_REF))
            print('Request headers: {0}\n'.format(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 {0}.\n'
                        .format(triggers_url))
    else:
        if verbose:
            print('Successfully registered trigger {0} with st2.\n'
                  .format(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'] = 'nagios.'
    headers['Content-Type'] = 'application/json; charset=utf-8'

    try:
        if verbose:
            print('Webhook POST: url: {0}, headers: {1}, body: {2}\n'
                  .format(url, headers, body))
        r = requests.post(url, data=json.dumps(body), headers=headers,
                          verify=False)
    except:
        traceback.print_exc(20)
        raise Exception('Cannot connect to st2 endpoint {0}.'.format(url))
    else:
        status = r.status_code

        if status in UNREACHABLE_CODES:
            msg = 'Webhook URL {0} does not exist. Check StackStorm '\
                'installation!'.format(url)
            raise Exception(msg)

        if status not in OK_CODES:
            sys.stderr.write('Failed posting nagio event to st2. HTTP_CODE: '
                             '{0}\n'.format(status))
        else:
            sys.stdout.write('Sent nagio event to st2. HTTP_CODE: '
                             '{0}\n'.format(status))


def _post_event_to_st2(payload, verbose=False):
    body = {}
    body['trigger'] = ST2_TRIGGERTYPE_REF
    try:
        body['payload'] = payload
    except:
        traceback.print_exc(limit=20)
        print('Invalid JSON payload {0}.'.format(payload))
        sys.exit(3)

    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)


def _set_config_opts(config_file, verbose=False):
    global ST2_USERNAME
    global ST2_PASSWORD
    global ST2_API_KEY
    global ST2_AUTH_TOKEN
    global ST2_API_BASE_URL
    global ST2_AUTH_BASE_URL
    global ST2_SSL_VERIFY
    global UNAUTHED
    global IS_API_KEY_AUTH
    if not os.path.exists(config_file):
        print('Configuration file "{0}" not found. Exiting!!!'
              .format(config_file))
        sys.exit(1)

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

        if verbose:
            print('Contents of config file: {0}'.format(config))

        ST2_USERNAME = config['st2_username']
        ST2_PASSWORD = config['st2_password']
        ST2_API_KEY = config.get('st2_api_key', None)
        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 += '/'
        if type(config['unauthed']) is bool:
            UNAUTHED = config['unauthed']
        if type(config['ssl_verify']) is bool:
            ST2_SSL_VERIFY = config['ssl_verify']

    if ST2_API_KEY:
        IS_API_KEY_AUTH = True

    if verbose:
        print('Unauthed? : {0}\nAPI key auth?: {1}\nSSL Verify? : {2}\n'
              .format(UNAUTHED, IS_API_KEY_AUTH, 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 _get_payload(host, service, event_id, state, state_id, state_type, attempt):
    payload = {}
    payload['host'] = host
    payload['service'] = service
    payload['event_id'] = event_id
    payload['state'] = state
    payload['state_id'] = state_id
    payload['state_type'] = state_type
    payload['attempt'] = attempt
    payload['msg'] = STATE_MESSAGE.get(state, 'Undefined state.')
    return payload


def main(config_file, payload, verbose=False):

    _set_config_opts(config_file=config_file, verbose=verbose)

    _register_with_st2(verbose=verbose)

    _post_event_to_st2(payload, verbose=verbose)


if __name__ == '__main__':
    description = '\nStackStorm nagios event handler. Please provide args '\
        'in following order after the config_path:\n\n event_id, service, '\
        'state,state_id, state_type, attempt, host\n'
    parser = argparse.ArgumentParser(description)
    parser.add_argument('config_path',
                        help='Exchange to listen on')
    parser.add_argument('--verbose', '-v', required=False, action='store_true',
                        help='Verbose mode.')

    args, nagio_args = parser.parse_known_args()

    event_id = nagio_args[0]
    service = nagio_args[1]
    state = nagio_args[2]
    state_id = nagio_args[3]
    state_type = nagio_args[4]
    attempt = nagio_args[5]
    host = nagio_args[6]

    payload = _get_payload(host, service, event_id, state, state_id, state_type, attempt)
    main(config_file=args.config_path, payload=payload, verbose=args.verbose)


1			#!/usr/bin/env python
2
3			import httplib
4			try:
5			import simplejson as json
6			except ImportError:
7			import json
8			import os
9			import sys
10			import traceback
11			from urlparse import urljoin
12			import argparse
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			requests.packages.urllib3.disable_warnings()
22			except ImportError:
23			raise ImportError('Missing dependency pyyaml. \
24			Do ``pip install pyyaml``.')
25
26			# ST2 configuration
27
28			ST2_API_BASE_URL = None
29			ST2_AUTH_BASE_URL = None
30			ST2_USERNAME = None
31			ST2_PASSWORD = None
32			ST2_API_KEY = None
33			ST2_AUTH_TOKEN = None
34			ST2_SSL_VERIFY = False
35
36			ST2_AUTH_PATH = 'tokens'
37			ST2_WEBHOOKS_PATH = 'webhooks/st2'
38			ST2_TRIGGERS_PATH = 'triggertypes'
39			ST2_TRIGGERTYPE_PACK = 'nagios'
40			ST2_TRIGGERTYPE_NAME = 'service_state_change'
41			ST2_TRIGGERTYPE_REF = '.'.join([ST2_TRIGGERTYPE_PACK, ST2_TRIGGERTYPE_NAME])
42
43			STATE_MESSAGE = {
44			'OK': 'All is well on the Western front.',
45			'WARNING': 'We gots a warning yo!',
46			'UNKNOWN': 'It be unknown...',
47			'CRITICAL': 'Critical!'
48			}
49
50			REGISTERED_WITH_ST2 = False
51			UNAUTHED = False
52			IS_API_KEY_AUTH = False
53
54			OK_CODES = [httplib.OK, httplib.CREATED, httplib.ACCEPTED, httplib.CONFLICT]
55			UNREACHABLE_CODES = [httplib.NOT_FOUND]
56
57			TOKEN_AUTH_HEADER = 'X-Auth-Token'
58			API_KEY_AUTH_HEADER = 'St2-Api-Key'
59
60
61			def _create_trigger_type(verbose=False):
62			try:
63			url = _get_st2_triggers_base_url()
64			payload = {
65			'name': ST2_TRIGGERTYPE_NAME,
66			'pack': ST2_TRIGGERTYPE_PACK,
67			'description': 'Trigger type for nagios event handler.'
68			}
69
70			headers = _get_st2_request_headers()
71			headers['Content-Type'] = 'application/json; charset=utf-8'
72
73			if verbose:
74			print('POST to URL {0} for registering trigger. Body = {1}, '
75			'headers = {2}.\n'.format(url, payload, headers))
76
77			post_resp = requests.post(url, data=json.dumps(payload),
78			headers=headers,
79			verify=ST2_SSL_VERIFY)
80			except:
81			traceback.print_exc(limit=20)
82			raise Exception('Unable to connect to st2 endpoint {0}.'.format(url))
83			else:
84			status = post_resp.status_code
85			if status in UNREACHABLE_CODES:
86			msg = 'Got response {0}. Invalid triggers endpoint {1}.' \
87			'Check configuration!'.format(status, url)
88			raise Exception(msg)
89
90			if status not in OK_CODES:
91			msg = 'Failed to register trigger type {0}.{1} with st2. ' \
92			'HTTP_CODE: {2}'.format(ST2_TRIGGERTYPE_PACK,
93			ST2_TRIGGERTYPE_NAME, status)
94			raise Exception(msg)
95			else:
96			print('Registered trigger type with st2.\n')
97
98
99			def _get_auth_url():
100			return urljoin(ST2_AUTH_BASE_URL, ST2_AUTH_PATH)
101
102
103			def _get_auth_token(verbose=False):
104			auth_url = _get_auth_url()
105
106			if verbose:
107			print('Will POST to URL {0} to get auth token.\n'.format(auth_url))
108
109			try:
110			resp = requests.post(auth_url, json.dumps({'ttl': 5 * 60}),
111			auth=(ST2_USERNAME, ST2_PASSWORD),
112			verify=ST2_SSL_VERIFY)
113			except:
114			traceback.print_exc(limit=20)
115			raise Exception('Unable to connect to st2 endpoint {0}.'.
116			format(auth_url))
117			else:
118			if resp.status_code in UNREACHABLE_CODES:
119			msg = 'Got response {0}. Invalid auth endpoint {1}. '\
120			'Check configuration!'.format(resp.status_code, auth_url)
121			raise Exception(msg)
122
123			if resp.status_code not in OK_CODES:
124			msg = 'Cannot get a valid auth token from {0}. '\
125			'HTTP_CODE: {1}'.format(auth_url, resp.status_code)
126			raise Exception(msg)
127
128			return resp.json()['token']
129
130
131			def _get_st2_request_headers():
132			headers = {}
133
134			if not UNAUTHED:
135			if IS_API_KEY_AUTH:
136			headers[API_KEY_AUTH_HEADER] = ST2_API_KEY
137			else:
138			if ST2_AUTH_TOKEN:
139			headers[TOKEN_AUTH_HEADER] = ST2_AUTH_TOKEN
140			else:
141			pass
142
143			return headers
144
145
146			def _register_with_st2(verbose=False):
147			global REGISTERED_WITH_ST2
148			try:
149			if not REGISTERED_WITH_ST2:
150			if verbose:
151			print('Checking if trigger "{0}" registered with st2.'
152			.format(ST2_TRIGGERTYPE_REF))
153			_register_trigger_with_st2(verbose=verbose)
154			REGISTERED_WITH_ST2 = True
155			except:
156			traceback.print_exc(limit=20)
157			sys.stderr.write(
158			'Failed registering with st2. Won\'t post event.\n')
159			sys.exit(2)
160
161
162			def _register_trigger_with_st2(verbose=False):
163			triggers_url = _get_st2_triggers_url()
164
165			try:
166			headers = _get_st2_request_headers()
167			if verbose:
168			print('Will GET from URL {0} for detecting trigger {1}.\n'
169			.format(triggers_url, ST2_TRIGGERTYPE_REF))
170			print('Request headers: {0}\n'.format(headers))
171
172			get_resp = requests.get(triggers_url, headers=headers,
173			verify=ST2_SSL_VERIFY)
174
175			if get_resp.status_code != httplib.OK:
176			_create_trigger_type(verbose=verbose)
177			else:
178			body = json.loads(get_resp.text)
179			if len(body) == 0:
180			_create_trigger_type(verbose=verbose)
181			except:
182			traceback.print_exc(limit=20)
183			raise Exception('Unable to connect to st2 endpoint {0}.\n'
184			.format(triggers_url))
185			else:
186			if verbose:
187			print('Successfully registered trigger {0} with st2.\n'
188			.format(ST2_TRIGGERTYPE_REF))
189
190
191			def _get_st2_triggers_base_url():
192			url = urljoin(ST2_API_BASE_URL, ST2_TRIGGERS_PATH)
193			return url
194
195
196			def _get_st2_triggers_url():
197			url = urljoin(_get_st2_triggers_base_url() + '/', ST2_TRIGGERTYPE_REF)
198			return url
199
200
201			def _get_st2_webhooks_url():
202			url = urljoin(ST2_API_BASE_URL, ST2_WEBHOOKS_PATH)
203			return url
204
205
206			def _post_webhook(url, body, verbose=False):
207			headers = _get_st2_request_headers()
208			headers['X-ST2-Integration'] = 'nagios.'
209			headers['Content-Type'] = 'application/json; charset=utf-8'
210
211			try:
212			if verbose:
213			print('Webhook POST: url: {0}, headers: {1}, body: {2}\n'
214			.format(url, headers, body))
215			r = requests.post(url, data=json.dumps(body), headers=headers,
216			verify=False)
217			except:
218			traceback.print_exc(20)
219			raise Exception('Cannot connect to st2 endpoint {0}.'.format(url))
220			else:
221			status = r.status_code
222
223			if status in UNREACHABLE_CODES:
224			msg = 'Webhook URL {0} does not exist. Check StackStorm '\
225			'installation!'.format(url)
226			raise Exception(msg)
227
228			if status not in OK_CODES:
229			sys.stderr.write('Failed posting nagio event to st2. HTTP_CODE: '
230			'{0}\n'.format(status))
231			else:
232			sys.stdout.write('Sent nagio event to st2. HTTP_CODE: '
233			'{0}\n'.format(status))
234
235
236			def _post_event_to_st2(payload, verbose=False):
237			body = {}
238			body['trigger'] = ST2_TRIGGERTYPE_REF
239			try:
240			body['payload'] = payload
241			except:
242			traceback.print_exc(limit=20)
243			print('Invalid JSON payload {0}.'.format(payload))
244			sys.exit(3)
245
246			try:
247			_post_webhook(url=_get_st2_webhooks_url(), body=body, verbose=verbose)
248			return True
249			except:
250			traceback.print_exc(limit=20)
251			print('Cannot send event to st2.')
252			sys.exit(4)
253
254
255			def _set_config_opts(config_file, verbose=False):
256			global ST2_USERNAME
257			global ST2_PASSWORD
258			global ST2_API_KEY
259			global ST2_AUTH_TOKEN
260			global ST2_API_BASE_URL
261			global ST2_AUTH_BASE_URL
262			global ST2_SSL_VERIFY
263			global UNAUTHED
264			global IS_API_KEY_AUTH
265			if not os.path.exists(config_file):
266			print('Configuration file "{0}" not found. Exiting!!!'
267			.format(config_file))
268			sys.exit(1)
269
270			with open(config_file) as f:
271			config = yaml.safe_load(f)
272
273			if verbose:
274			print('Contents of config file: {0}'.format(config))
275
276			ST2_USERNAME = config['st2_username']
277			ST2_PASSWORD = config['st2_password']
278			ST2_API_KEY = config.get('st2_api_key', None)
279			ST2_API_BASE_URL = config['st2_api_base_url']
280			if not ST2_API_BASE_URL.endswith('/'):
281			ST2_API_BASE_URL += '/'
282			ST2_AUTH_BASE_URL = config['st2_auth_base_url']
283			if not ST2_AUTH_BASE_URL.endswith('/'):
284			ST2_AUTH_BASE_URL += '/'
285			if type(config['unauthed']) is bool:
286			UNAUTHED = config['unauthed']
287			if type(config['ssl_verify']) is bool:
288			ST2_SSL_VERIFY = config['ssl_verify']
289
290			if ST2_API_KEY:
291			IS_API_KEY_AUTH = True
292
293			if verbose:
294			print('Unauthed? : {0}\nAPI key auth?: {1}\nSSL Verify? : {2}\n'
295			.format(UNAUTHED, IS_API_KEY_AUTH, ST2_SSL_VERIFY))
296
297			if not UNAUTHED and not IS_API_KEY_AUTH:
298			try:
299			if not ST2_AUTH_TOKEN:
300			if verbose:
301			print('No auth token found. Let\'s get one from'
302			'StackStorm!')
303			ST2_AUTH_TOKEN = _get_auth_token(verbose=verbose)
304			except:
305			traceback.print_exc(limit=20)
306			print('Unable to negotiate an auth token. Exiting!')
307			sys.exit(1)
308
309
310			def _get_payload(host, service, event_id, state, state_id, state_type, attempt):
311			payload = {}
312			payload['host'] = host
313			payload['service'] = service
314			payload['event_id'] = event_id
315			payload['state'] = state
316			payload['state_id'] = state_id
317			payload['state_type'] = state_type
318			payload['attempt'] = attempt
319			payload['msg'] = STATE_MESSAGE.get(state, 'Undefined state.')
320			return payload
321
322
323			def main(config_file, payload, verbose=False):
324
325			_set_config_opts(config_file=config_file, verbose=verbose)
326
327			_register_with_st2(verbose=verbose)
328
329			_post_event_to_st2(payload, verbose=verbose)
330
331
332			if __name__ == '__main__':
333			description = '\nStackStorm nagios event handler. Please provide args '\
334			'in following order after the config_path:\n\n event_id, service, '\
335			'state,state_id, state_type, attempt, host\n'
336			parser = argparse.ArgumentParser(description)
337			parser.add_argument('config_path',
338			help='Exchange to listen on')
339			parser.add_argument('--verbose', '-v', required=False, action='store_true',
340			help='Verbose mode.')
341
342			args, nagio_args = parser.parse_known_args()
343
344			event_id = nagio_args[0]
345			service = nagio_args[1]
346			state = nagio_args[2]
347			state_id = nagio_args[3]
348			state_type = nagio_args[4]
349			attempt = nagio_args[5]
350			host = nagio_args[6]
351
352			payload = _get_payload(host, service, event_id, state, state_id, state_type, attempt)
353			main(config_file=args.config_path, payload=payload, verbose=args.verbose)
354

StackStorm / st2contrib

Pull Request — master (#502)

_get_auth_token() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like