_get_auth_token() - Code Metrics - Inspection of "WIP AlertLogic pack" - StackStorm/st2contrib - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#423)

unknown

created 2016-03-03 11:42 UTC

_get_auth_token() A

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	3
dl	0
loc	10
rs	9.4285

#!/usr/bin/env python

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_CONFIG_FILE = './config.yaml'

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 = '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

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


def _create_trigger_type():
    try:
        url = _get_st2_triggers_url()
        payload = {
            'name': ST2_TRIGGERTYPE_NAME,
            'pack': ST2_TRIGGERTYPE_PACK,
            'description': 'Trigger type for nagios 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:
        raise Exception('Cannot get auth token from st2. Will try unauthed.')
    else:
        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'] = 'nagios.'
    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 nagios event to st2. HTTP_CODE: %d\n' % status)
        else:
            sys.stdout.write('Sent nagios event to st2. HTTP_CODE: %d\n' % status)


def _get_payload(host, service, event_id, state, state_type, attempt):
    payload = {}
    payload['host'] = host
    payload['service'] = service
    payload['event_id'] = event_id
    payload['state'] = state
    payload['state_type'] = state_type
    payload['attempt'] = attempt
    payload['msg'] = STATE_MESSAGE.get(state, 'Undefined state.')
    return payload


def main(args):
    event_id = args[1]
    service = args[2]
    state = args[3]
    state_type = args[4]
    attempt = args[5]
    host = args[6]

    payload = _get_payload(host, service, event_id, state, state_type, attempt)
    body = {}
    body['trigger'] = ST2_TRIGGERTYPE_REF
    body['payload'] = payload
    _post_event_to_st2(_get_st2_webhooks_url(), body)


if __name__ == '__main__':
    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']

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


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			from urlparse import urljoin
11
12			try:
13			import requests
14			except ImportError:
15			raise ImportError('Missing dependency requests. Do ``pip install requests``.')
16
17			try:
18			import yaml
19			except ImportError:
20			raise ImportError('Missing dependency pyyaml. Do ``pip install pyyaml``.')
21
22			# ST2 configuration
23			ST2_CONFIG_FILE = './config.yaml'
24
25			ST2_API_BASE_URL = 'http://localhost:9101/v1'
26			ST2_AUTH_BASE_URL = 'http://localhost:9100'
27			ST2_USERNAME = None
28			ST2_PASSWORD = None
29			ST2_AUTH_TOKEN = None
30
31			ST2_AUTH_PATH = 'tokens'
32			ST2_WEBHOOKS_PATH = 'webhooks/st2/'
33			ST2_TRIGGERS_PATH = 'triggertypes/'
34			ST2_TRIGGERTYPE_PACK = 'nagios'
35			ST2_TRIGGERTYPE_NAME = 'service-state-change'
36			ST2_TRIGGERTYPE_REF = '.'.join([ST2_TRIGGERTYPE_PACK, ST2_TRIGGERTYPE_NAME])
37
38			STATE_MESSAGE = {
39			'OK': 'All is well on the Western front.',
40			'WARNING': 'We gots a warning yo!',
41			'UNKNOWN': 'It be unknown...',
42			'CRITICAL': 'Critical!'
43			}
44
45			REGISTERED_WITH_ST2 = False
46
47			OK_CODES = [httplib.OK, httplib.CREATED, httplib.ACCEPTED, httplib.CONFLICT]
48
49
50			def _create_trigger_type():
51			try:
52			url = _get_st2_triggers_url()
53			payload = {
54			'name': ST2_TRIGGERTYPE_NAME,
55			'pack': ST2_TRIGGERTYPE_PACK,
56			'description': 'Trigger type for nagios event handler.'
57			}
58			# sys.stdout.write('POST: %s: Body: %s\n' % (url, payload))
59			headers = {}
60			headers['Content-Type'] = 'application/json; charset=utf-8'
61
62			if ST2_AUTH_TOKEN:
63			headers['X-Auth-Token'] = ST2_AUTH_TOKEN
64
65			post_resp = requests.post(url, data=json.dumps(payload), headers=headers)
66			except:
67			sys.stderr.write('Unable to register trigger type with st2.')
68			raise
69			else:
70			status = post_resp.status_code
71			if status not in OK_CODES:
72			sys.stderr.write('Failed to register trigger type with st2. HTTP_CODE: %d\n' %
73			status)
74			raise
75			else:
76			sys.stdout.write('Registered trigger type with st2.\n')
77
78
79			def _get_auth_url():
80			return urljoin(ST2_AUTH_BASE_URL, ST2_AUTH_PATH)
81
82
83			def _get_auth_token():
84			global ST2_AUTH_TOKEN
85			auth_url = _get_auth_url()
86			try:
87			resp = requests.post(auth_url, json.dumps({'ttl': 5 * 60}),
88			auth=(ST2_USERNAME, ST2_PASSWORD))
89			except:
90			raise Exception('Cannot get auth token from st2. Will try unauthed.')
91			else:
92			ST2_AUTH_TOKEN = resp.json()['token']
93
94
95			def _register_with_st2():
96			global REGISTERED_WITH_ST2
97			try:
98			url = urljoin(_get_st2_triggers_url(), ST2_TRIGGERTYPE_REF)
99			# sys.stdout.write('GET: %s\n' % url)
100			if not ST2_AUTH_TOKEN:
101			_get_auth_token()
102
103			if ST2_AUTH_TOKEN:
104			get_resp = requests.get(url, headers={'X-Auth-Token': ST2_AUTH_TOKEN})
105			else:
106			get_resp = requests.get(url)
107
108			if get_resp.status_code != httplib.OK:
109			_create_trigger_type()
110			else:
111			body = json.loads(get_resp.text)
112			if len(body) == 0:
113			_create_trigger_type()
114			except:
115			raise
116			else:
117			REGISTERED_WITH_ST2 = True
118
119
120			def _get_st2_triggers_url():
121			url = urljoin(ST2_API_BASE_URL, ST2_TRIGGERS_PATH)
122			return url
123
124
125			def _get_st2_webhooks_url():
126			url = urljoin(ST2_API_BASE_URL, ST2_WEBHOOKS_PATH)
127			return url
128
129
130			def _post_event_to_st2(url, body):
131			headers = {}
132			headers['X-ST2-Integration'] = 'nagios.'
133			headers['Content-Type'] = 'application/json; charset=utf-8'
134			if ST2_AUTH_TOKEN:
135			headers['X-Auth-Token'] = ST2_AUTH_TOKEN
136			try:
137			# sys.stdout.write('POST: url: %s, body: %s\n' % (url, body))
138			r = requests.post(url, data=json.dumps(body), headers=headers)
139			except:
140			sys.stderr.write('Cannot connect to st2 endpoint.')
141			else:
142			status = r.status_code
143			if status not in OK_CODES:
144			sys.stderr.write('Failed posting nagios event to st2. HTTP_CODE: %d\n' % status)
145			else:
146			sys.stdout.write('Sent nagios event to st2. HTTP_CODE: %d\n' % status)
147
148
149			def _get_payload(host, service, event_id, state, state_type, attempt):
150			payload = {}
151			payload['host'] = host
152			payload['service'] = service
153			payload['event_id'] = event_id
154			payload['state'] = state
155			payload['state_type'] = state_type
156			payload['attempt'] = attempt
157			payload['msg'] = STATE_MESSAGE.get(state, 'Undefined state.')
158			return payload
159
160
161			def main(args):
162			event_id = args[1]
163			service = args[2]
164			state = args[3]
165			state_type = args[4]
166			attempt = args[5]
167			host = args[6]
168
169			payload = _get_payload(host, service, event_id, state, state_type, attempt)
170			body = {}
171			body['trigger'] = ST2_TRIGGERTYPE_REF
172			body['payload'] = payload
173			_post_event_to_st2(_get_st2_webhooks_url(), body)
174
175
176			if __name__ == '__main__':
177			try:
178			if not os.path.exists(ST2_CONFIG_FILE):
179			sys.stderr.write('Configuration file not found. Exiting.\n')
180			sys.exit(1)
181
182			with open(ST2_CONFIG_FILE) as f:
183			config = yaml.safe_load(f)
184			ST2_USERNAME = config['st2_username']
185			ST2_PASSWORD = config['st2_password']
186			ST2_API_BASE_URL = config['st2_api_base_url']
187			ST2_AUTH_BASE_URL = config['st2_auth_base_url']
188
189			if not REGISTERED_WITH_ST2:
190			_register_with_st2()
191			except:
192			sys.stderr.write('Failed registering with st2. Won\'t post event.\n')
193			else:
194			main(sys.argv)
195

StackStorm / st2contrib

Pull Request — master (#423)

_get_auth_token() A

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like