test_nginx_vhosts.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
'''
Criticality: High
Tests that all webserver services (vhosts) are available and reachable.
'''
import imp
import os
import re
import requests
import socket
import sys

'''
This script checks for all enabled vhosts in Nginx conf that:
* The response status code is 200 or 403.
* The host is resolved as 127.0.0.1.
* The Wowza response is correct on /streaming/ (only for mediaserver vhosts).
'''

# check that Nginx dir exists
nginx_dir = '/etc/nginx/sites-enabled'
if not os.path.exists(nginx_dir):
    print('Nginx dir does not exists ("%s").' % nginx_dir)
    sys.exit(2)

# check that Wowza is installed
wowza_dir = '/usr/local/WowzaStreamingEngine'
if not os.path.exists(wowza_dir):
    print('Info: Wowza is not installed ("%s" does not exist).' % wowza_dir)
    wowza_dir = None
else:
    print('Info: Wowza is installed, /streaming/ will be tested on mediaserver vhosts.')

# get envsetup conf
conf = dict()
os.chdir(os.path.dirname(__file__))
if os.path.isfile('../utils.py'):
    es_utils = imp.load_source('es_utils', '../utils.py')
    conf = es_utils.load_conf()

# get enabled vhosts
resolution_ignored = conf.get('TESTER_VHOST_RESOLUTION_IGNORED', '').split(',')
requests.packages.urllib3.disable_warnings()
found = False
errors = 0
for name in os.listdir(nginx_dir):
    path = os.path.join(nginx_dir, name)
    with open(path, 'r') as fo:
        vhost = fo.read()
    vhost = vhost.replace('\t', ' ')
    matching = re.search(r'.*server_name\ +([0-9a-zA-Z\.\-\_\ ]+);.*', vhost)
    if not matching:
        print('The server_name was not found in: "%s".' % path)
        errors += 1
        continue
    domains = matching.groups()[0].strip().split(' ')
    https = re.search(r'listen +\w* +ssl;', vhost) is not None \
        or re.search(r'ssl +on;', vhost) is not None
    for domain in domains:
        if domain == 'localhost':
            continue  # status vhost
        found = True
        url = '%s://%s' % ('https' if https else 'http', domain)
        sys.stdout.write('Testing url "%s":\n' % url)
        # test domain IP
        ip_error = None
        try:
            ip = socket.gethostbyname(domain)
        except Exception as e:
            ip_error = 'domain is not resolved: %s' % e
        else:
            if ip != '127.0.0.1':
                ip_error = 'domain is resolved with %s instead of 127.0.0.1' % ip
        sys.stdout.write('  IP: ')
        if ip_error:
            if domain in resolution_ignored:
                sys.stdout.write('\033[94mIgnored (%s)\033[0m' % ip_error)
                ip_error = None
            else:
                sys.stdout.write('\033[91mKO (%s)\033[0m' % ip_error)
        else:
            sys.stdout.write('\033[92mOK (127.0.0.1)\033[0m')
        # test url
        sys.stdout.write(', status: ')
        req_error = False
        try:
            req = requests.get(url, verify=False, proxies={'http': '', 'https': ''}, timeout=10)
        except Exception as e:
            code = str(e)
        else:
            code = req.status_code
        if code not in (200, 403):
            sys.stdout.write('\033[91mKO (%s)\033[0m' % code)
            req_error = True
        else:
            sys.stdout.write('\033[92mOK (%s)\033[0m' % code)
            if 'mediaserver' in name and wowza_dir:
                # test /streaming url
                sys.stdout.write(', streaming: ')
                try:
                    req = requests.get(url + '/streaming/', verify=False, proxies={'http': '', 'https': ''}, timeout=10)
                except Exception as e:
                    code = str(e)
                else:
                    code = req.status_code
                if code != 200:
                    sys.stdout.write('\033[91mKO (%s)\033[0m' % code)
                    req_error = True
                elif 'Wowza Streaming Engine' not in req.text:
                    sys.stdout.write('\033[91mKO (%s, %s)\033[0m' % (code, req.text.replace('\n', ' ')[:200]))
                    req_error = True
                else:
                    sys.stdout.write('\033[92mOK (%s)\033[0m' % code)
        sys.stdout.write('.\n')

        if ip_error or req_error:
            errors += 1

if errors:
    print('%s vhost(s) did not correctly responded.' % errors)
    sys.exit(1)
if not found:
    print('No vhost found in Nginx sites-enabled dir.')
    sys.exit(1)