test_wowza.py

#!/usr/bin/env python3

"""
Criticality: Normal
Checks that the streaming server (Wowza) is running correctly.
"""

from pathlib import Path
import re
import subprocess  # nosec: B404
import sys

from defusedxml.ElementTree import parse
from psutil import net_connections

sys.path.append(str(Path(__file__).parents[1].resolve()))

# pylint: disable=wrong-import-position
import utils as u  # noqa: E402

LATEST_VERSION = "4.7.7"


def main():
    """Run all checks and exits with corresponding exit code."""

    print("Checking Wowza settings:")
    warnings = 0
    errors = 0

    # check if wowza is installed
    if not check_installed():
        exit(2)

    # check wowza version
    check_warn, check_err = check_version()
    if check_err:
        errors += 1
    elif check_warn:
        warnings += 1

    # check wowza heap size
    check_warn, check_err = check_heap_size()
    if check_err:
        errors += 1
    elif check_warn:
        warnings += 1

    # check that wowza is running
    check_warn, check_err = check_running()
    if check_err:
        errors += 1
    elif check_warn:
        warnings += 1

    # check that wowza is listening
    check_warn, check_err = check_listening()
    if check_err:
        errors += 1
    elif check_warn:
        warnings += 1

    if errors:
        exit(1)
    elif warnings:
        exit(3)

    exit(0)


def check_installed() -> bool:
    """Check that Wowza is installed.

    :return: Exit return codes
    :rtype: bool
    """

    cmd = "dpkg --get-selections 'wowza*'"
    out = subprocess.getoutput(cmd)
    state = out.split()[-1]

    if state != "install":
        u.info("not installed, skip test")
        return False

    return True


def check_version() -> tuple:
    """Check the Wowza version installed.

    :return: Exit return codes
    :rtype: bool
    """

    warnings = 0
    errors = 0

    cmd = "dpkg --get-selections 'wowza*'"
    out = subprocess.getoutput(cmd)
    version = None

    for line in out.split("\n"):
        if line.split()[-1] == "install":
            if version:
                u.error("many Wowza versions installed, keep only latest")
                errors += 1
            version = ".".join(re.findall(r"\d", line))

    if not version:
        u.error("cannot find wWowza version")
        errors += 1

    if version != LATEST_VERSION:
        u.warning("using outdated version: {}".format(version))
        warnings += 1
    else:
        u.success("using recommended version: {}".format(LATEST_VERSION))

    return warnings, errors


def check_heap_size() -> tuple:
    """Check the heap size configured.

    :return: Exit return codes
    :rtype: bool
    """

    warnings = 0
    errors = 0

    cmd = "grep '<HeapSize>2000M</HeapSize>' /usr/local/WowzaStreamingEngine/conf/Tune.xml"
    check_heap, _ = subprocess.getstatusoutput(cmd)
    if check_heap != 0:
        u.warning("not using recommended heap size")
        warnings += 1
    else:
        u.success("using recommended heap size")

    return warnings, errors


def check_running() -> tuple:
    """Check that Wowza is running.

    :return: Exit return codes
    :rtype: bool
    """

    warnings = 0
    errors = 0

    cmd = "systemctl status WowzaStreamingEngine"
    out = subprocess.getoutput(cmd)
    if "Active: active (running)" not in out:
        u.error("service not running")
        errors += 1
    else:
        u.success("service running")

    return warnings, errors


def check_listening() -> tuple:
    """Check that Wowza is listening on configured port.

    :return: Exit return codes
    :rtype: bool
    """

    warnings = 0
    errors = 0

    # get port number in Wowza config
    conf = parse("/usr/local/WowzaStreamingEngine/conf/VHost.xml").getroot()
    port = conf.findall("VHost/HostPortList/HostPort/Port")[0].text

    # get listening ports
    listening = set(
        c.laddr[1] for c in net_connections(kind="inet") if c.status == "LISTEN"
    )

    # check that system is listening on this port
    if int(port) not in listening:
        u.error("not listening on port {}".format(port))
        errors += 1
    else:
        u.success("listening on port {}".format(port))

    return warnings, errors


if __name__ == "__main__":
    main()