1#@!#!123s

: / proc / 3 / root / opt / tier1adv / bin /
Filename : server_overview
back
#!/opt/support/venv/bin/python3
"""Displays an overview of the server configuration"""
# Original py2 version written by Joe C <joe@cartonia.net>
import argparse
from pathlib import Path
import os
import re
import platform
import pwd
from glob import glob
import shlex
import subprocess
import sys
from collections.abc import Generator
import distro
import psutil
import pymysql
import rads

sys.path.insert(0, '/opt/support/lib')
from output import header as header_print, warn
from server import ROLE
from arg_types import any_cpuser_arg, cpuser_safe_arg

header = lambda x: header_print(
    f" {x} ".center(80, '='),
    color=rads.color.blue,
    flush=True,
)

MEGACLI = '/opt/MegaRAID/MegaCli/MegaCli64'

if Path('/etc/cpanel/ea4/is_ea4').exists():
    HTTPD_CONF = '/etc/apache2/conf/httpd.conf'
    HTTPD_BIN = '/usr/sbin/httpd'
else:
    HTTPD_CONF = '/usr/local/apache/conf/httpd.conf'
    HTTPD_BIN = '/usr/local/apache/bin/httpd'

IGNORE_DIRS = ['mail', 'etc']
CONFIG_MATCHES = [
    'wp-config.php',
    'configuration.php',
    'local.xml',
    'config.php',
    'core.php',
    'settings.php',
    'settings.inc.php',
    'header.tpl',
    'general.php',
]


def parse_args() -> list[str]:
    """Parse CLI arguments"""
    parser = argparse.ArgumentParser(description=__doc__)
    group = parser.add_mutually_exclusive_group()
    group.add_argument(
        "-u",
        "--users",
        default=[],
        nargs='*',
        type=cpuser_safe_arg if ROLE else any_cpuser_arg,
        help="list of users to analyze software for",
    )
    if not ROLE:  # v/ded
        group.add_argument(
            '-a',
            '--all-users',
            action='store_true',
            help="Analyze software for all users",
        )
    args = parser.parse_args()
    if getattr(args, 'all_users', False):
        args.users = rads.all_cpusers()
    return args.users


def show_server_ips():
    try:
        print(
            'cPanel main IP:',
            Path('/var/cpanel/mainip').read_text(encoding='utf-8'),
        )
    except FileNotFoundError:
        pass
    try:
        print(
            'cPanel shared IP(s):',
            Path('/var/cpanel/mainips/root').read_text(encoding='utf-8'),
        )
    except FileNotFoundError:
        try:
            with open('/etc/wwwacct.conf', encoding='utf-8') as handle:
                for line in handle:
                    if line.startswith('ADDR '):
                        print('cPanel shared IP:', line.split()[1])
                        return
        except (FileNotFoundError, IndexError):
            pass
    if distro.major_version() == '6':
        cmd = ['ip', 'address', 'show', 'scope', 'global']
    else:
        cmd = ['ip', '-c', 'address', 'show', 'scope', 'global']
    print('Interfaces:')
    subprocess.call(cmd)


def cpu_info():
    print('CPU cores (logical):', psutil.cpu_count(logical=True))
    print('CPU cores (physical):', psutil.cpu_count(logical=False))
    cpu_model = 'UNKNOWN'
    with open('/proc/cpuinfo', encoding='utf-8') as cpuinfo:
        for line in cpuinfo:
            if line.startswith('model name'):
                cpu_model = line.split(':', maxsplit=1)[1].strip()
    print('CPU Model:', cpu_model)


def mem_info():
    mem = psutil.virtual_memory()
    for attr in 'used free available total'.split():
        print(
            f"Memory {attr.title(): <9}: {int(getattr(mem, attr) / 2**20): >6} MB"
        )


def display_server_specs():
    header("Server specs")
    print(f"Hostname: {platform.node()}")
    show_server_ips()
    print("kernel:", end=' ', flush=True)
    subprocess.call(['uname', '-r'])
    cpu_info()
    mem_info()


def out_lines(cmd: list[str], stderr=subprocess.DEVNULL) -> list[str]:
    try:
        return subprocess.check_output(
            cmd, encoding='utf-8', stderr=stderr
        ).splitlines()
    except (FileNotFoundError, subprocess.CalledProcessError):
        return []


def display_disk_info():
    header("Disk information")
    subprocess.call(['df', '-h'])
    try:
        with open('/proc/scsi/sg/device_strs', encoding='utf-8') as handle:
            unique = {x for x in handle if 'PERC' in x or 'Virtual Disk' in x}
            for line in unique:
                print(line, end='')
    except FileNotFoundError:
        pass
    for line in out_lines(['lspci', '-vvv']):
        if re.search('RAID', line, re.I):
            print(re.sub(r'^[\s]+Subsystem: ', '', line))
    found_raid = False
    for line in out_lines(['sas2ircu', '0', 'DISPLAY']):
        if re.search(r'(RAID level|Volume Name)', line):
            print(re.sub(r'^[\s]+', '', line))
            found_raid = True
    for line in out_lines(['sas2ircu', '0', 'STATUS']):
        if 'Volume state' in line:
            print(re.sub(r'^[\s]+', '', line))
    if not found_raid:
        found_raid = check_megacli()
    if not found_raid:
        for line in out_lines(['/usr/bin/lsiutil', '-s']):
            if re.search(r'(Vendor|Disk)', line):
                print(line)
                found_raid = True
    if not found_raid:
        print("RAID not found")


def check_megacli() -> bool:
    found_raid = False
    for line in out_lines([MEGACLI, '-AdpAllInfo', '-aALL', '-NoLog']):
        if 'Product Name' in line:
            print(line)
            found_raid = True
    for line in out_lines([MEGACLI, '-AdpBbuCmd', '-aALL', '-NoLog']):
        if re.match(r'(?:Relative State of Charge|Charger Status)', line):
            print(line)
    for line in out_lines([MEGACLI, '-PDList', '-aALL', '-NoLog']):
        if 'Enclosure Device ID' in line:
            enc_id = re.sub(r'Enclosure Device ID: ', '', line)
        elif 'Slot Number' in line:
            slot_id = re.sub(r'Slot Number: ', '', line)
        elif 'Device Id' in line:
            dev_id = re.sub(r'Device Id: ', '', line)
        elif 'PD Type' in line:
            pd_type = re.sub(r'PD Type: ', '', line)
        elif 'Raw Size' in line:
            raw_size = re.sub(r'Raw Size: ', '', line)
            raw_size = re.sub(r' \[.*$', '', raw_size)
        elif 'Inquiry Data' in line:
            drive = re.sub(r'Inquiry Data: ', '', line)
        if 'Foreign State' in line:
            print(f"[{enc_id}:{slot_id}:{dev_id}] {pd_type} {raw_size} {drive}")
            found_raid = True
    for line in out_lines([MEGACLI, '-LDInfo', '-Lall', '-aALL', '-NoLog']):
        if re.search(
            r'^(RAID Level:|Size:|State:|Number Of Drives:|Current Cache Policy:)',
            line,
        ):
            if re.search(r'^(Size:|Number Of Drives:)', line):
                line = re.sub(r':', ': ', line)
                found_raid = True
            print(line)
    return found_raid


def display_apache_info():
    header("Apache information")
    try:
        version = subprocess.check_output([HTTPD_BIN, '-v'], encoding='utf-8')
        print(version.replace('\n', '  '))
    except FileNotFoundError:
        warn('Could not find Apache binary at', HTTPD_BIN)
    except subprocess.CalledProcessError:
        pass
    for line in out_lines([HTTPD_BIN, '-M'], stderr=None):
        if 'mpm_prefork_module' in line:
            print("Using MPM Prefork")
        elif 'mpm_worker_module' in line:
            print("Using MPM Worker")
    try:
        with open(HTTPD_CONF, encoding='utf-8') as conf:
            for line in conf:
                if re.search(r'MaxClients|ServerLimit|KeepAlive', line):
                    print(line, end='')
    except FileNotFoundError:
        warn('Could not find Apache configuration at', HTTPD_CONF)


def display_php_info():
    header("PHP information")
    try:
        subprocess.call(['/usr/local/bin/php', '-v'])
    except FileNotFoundError:
        pass
    try:
        subprocess.call(
            ['/usr/local/cpanel/bin/rebuild_phpconf', '--current'],
            stderr=subprocess.DEVNULL,
        )
    except FileNotFoundError:
        warn("Could not find /usr/local/cpanel/bin/rebuild_phpconf")


def display_mysql_info():
    header("MySQL information")
    try:
        subprocess.call(['mysql', '-V'])
    except FileNotFoundError:
        warn('Could not find mysql client')
        return
    try:
        subprocess.call(['mysqladmin', 'status'])
    except FileNotFoundError:
        warn('could not find mysqladmin client')
        return
    conn_kargs = {
        'host': 'localhost',
        'read_default_file': '/root/.my.cnf',
        'database': 'mysql',
    }
    try:
        with pymysql.connect(**conn_kargs) as conn, conn.cursor() as cur:
            cur.execute("SHOW VARIABLES LIKE 'max_%conn%'")
            for row in cur.fetchall():
                print(*row, sep=' = ')
    except pymysql.Error as exc:
        warn(str(exc))


def display_cpanel_info():
    header("cPanel information")
    try:
        ver = Path('/usr/local/cpanel/version').read_text(encoding='utf-8')
    except FileNotFoundError:
        warn('Could not find cPanel version')
    else:
        print('cPanel version:', ver.strip())
    try:
        user_count = len(rads.all_cpusers())
    except (FileNotFoundError, rads.CpuserError) as exc:
        warn(str(exc))
    else:
        print('cPanel accounts:', user_count)


def check_optimizations():
    header("Optimizations")
    for line in out_lines(['/usr/local/bin/php', '-m']):
        if re.match(r'apc$', line):
            print("Found PHP PECL: apc")
        if re.match(r'memcache$', line):
            print("Found PHP PECL: memcache")
    if not Path('/usr/local/bin/memcached').is_file():
        return
    try:
        memcached = out_lines(['/usr/local/bin/memcached', '-i'])[0]
    except IndexError:
        warn("Found '/usr/local/bin/memcached' but could not determine version")
    else:
        print('Found', memcached)


def check_mds_exists() -> bool:
    """Check if /mds exists and has contents"""
    try:
        return any(os.scandir('/mds'))
    except (NotADirectoryError, FileNotFoundError):
        return False


def check_recent_high_loads():
    header("Recent high 1-minute load averages")
    if not check_sa():
        return
    data = {
        'file': "File",
        'time': "Time",
        'runq': "Run Queue",
        'plist': "Task List",
        'one': "Load Avg 1",
        'five': "Load Avg 5",
        'fifteen': "Load Avg 15",
    }
    template = (
        "%(file)20s%(time)15s%(runq)14s%(plist)14s"
        "%(one)14s%(five)14s%(fifteen)14s"
    )
    print(template % data)
    for file_name, data in get_highest_sa_load_entries():
        if file_name and data:
            data['file'] = file_name
            print(template % data)


def get_highest_sa_load_entries() -> list[tuple[str, dict]]:
    sar_entry = re.compile(
        r"(?P<time>\d{2}:\d{2}:\d{2}\s(?:A|P)M)\s+"
        r"(?P<runq>\d+)\s+(?P<plist>\d+)\s+"
        r"(?P<one>[\d\.]+)\s+(?P<five>[\d\.]+)\s+(?P<fifteen>[\d\.]+)"
    )
    highest_entries = []
    # iterating over sa files, newest first
    for sa_file in sorted(
        glob('/var/log/sa/sa[0-9][0-9]'),
        key=lambda x: os.stat(x).st_mtime,
    ):
        highest = 0.0
        high_data = None
        for line in iter_sar('-q', '-f', sa_file):
            if match := sar_entry.match(line):
                data = match.groupdict()
                if float(data['one']) > highest:
                    highest = float(data['one'])
                    high_data = data
        if high_data:
            highest_entries.append((sa_file, high_data))
    return highest_entries


def check_sa() -> bool:
    if not Path('/var/log/sa').is_dir():
        warn("Could not find sysstat log directory /var/log/sa")
        return False
    if not any(os.scandir('/var/log/sa')):
        warn('/var/log/sa is empty')
        return False
    return True


def iter_sar(*args) -> Generator[str, None, None]:
    cmd = ['sar']
    cmd.extend(args)
    try:
        with subprocess.Popen(
            cmd,
            encoding='utf-8',
            stdout=subprocess.PIPE,
            stderr=subprocess.DEVNULL,
        ) as sar_proc:
            yield from sar_proc.stdout
    except FileNotFoundError:
        warn('could not find sar')
    except subprocess.CalledProcessError:
        warn('error running', shlex.join(cmd))


def check_recent_high_mem():
    header("Recent high mem usage averages")
    if not check_sa():
        return
    colnames = []
    for line in iter_sar('-r'):
        if 'mem' in line:
            colnames = line.split()[2::]
            break
    if not colnames:
        warn('could not read columns from sar -r')
        return
    for index, value in enumerate(colnames):
        colnames[index] = re.sub(r'%', 'p', value)
    data = {'file': "File", 'time': "Time"}
    template_code = "%(file)20s%(time)15s"
    for item in colnames:
        data[str(item)] = str(item)
        template_code += "%(" + str(item) + ")14s"
    template = template_code
    del template_code
    print(template % data)
    for file_name, data in get_highest_sa_mem_averages(colnames):
        if file_name and data:
            data['file'] = file_name
            print(template % data)


def get_highest_sa_mem_averages(colnames):
    pattern = r'(?P<time>\d{2}:\d{2}:\d{2}\s(?:A|P)M)\s+'
    for item in colnames:
        pattern += '(?P<' + re.escape(item) + r'>[\d\.]+)\s+'
    sar_entry = re.compile(pattern)
    del pattern
    highest_entries = []
    for sa_file in sorted(
        glob('/var/log/sa/sa[0-9][0-9]'),
        key=lambda x: os.stat(x).st_mtime,
    ):
        highest = 0.0
        high_data = None
        for line in iter_sar('-r', '-f', sa_file):
            if match := sar_entry.match(line):
                data = match.groupdict()
                if float(data['pmemused']) > highest:
                    highest = float(data['pmemused'])
                    high_data = data
        if high_data:
            highest_entries.append((sa_file, high_data))
    return highest_entries


def display_listeners():
    header("Listening ports (HTTP and special)")
    for line in out_lines(['netstat', '-plnt']):
        if not re.search(r'(httpd|nginx|varnish|memcached)', line):
            continue
        try:
            cols = line.split()
            print(cols[3], cols[6])
        except IndexError:
            continue


def find_configs(users: list[str]) -> Generator[Path, None, None]:
    homedirs = sorted(pwd.getpwnam(x).pw_dir for x in users)
    for homedir in homedirs:
        for dirname, dirnames, filenames in os.walk(homedir, topdown=True):
            if dirname == homedir:
                for ig_dir in IGNORE_DIRS:
                    if ig_dir in dirnames:
                        dirnames.remove(ig_dir)
            for filename in filenames:
                if filename in CONFIG_MATCHES:
                    yield Path(dirname, filename)


def display_software(users: list[str]):
    header("Software installations found")
    for conf_path in find_configs(users):
        if conf_path.name == 'wp-config.php':
            display_wordpress(conf_path.parent)
        elif conf_path.name == 'configuration.php':
            display_joomla(conf_path.parent)
        elif str(conf_path).endswith('/sites/default/settings.php'):
            display_drupal(conf_path.parent.parent.parent)
        elif str(conf_path).endswith('/app/etc/local.xml'):
            display_magento(conf_path.parent.parent.parent)
        elif str(conf_path).endswith('/config/settings.inc.php'):
            display_prestashop(conf_path.parent.parent)
        elif str(conf_path).endswith('/view/template/common/header.tpl'):
            display_opencart(conf_path.parent.parent.parent.parent)
        elif str(conf_path).endswith('/applications/settings/general.php'):
            display_socialengine(conf_path.parent.parent.parent)
        elif str(conf_path).endswith('/ow_includes/config.php'):
            display_oxwall(conf_path.parent.parent)
        elif str(conf_path).endswith('/app/Config/core.php'):
            display_cakephp(conf_path.parent.parent.parent)


def display_wordpress(root: Path):
    version = None
    try:
        with open(root / 'wp-includes/version.php', encoding='utf-8') as handle:
            for line in handle:
                if re.search(r'\$wp_version\s*?=', line):
                    version = line.split("'")[1]
                    break
    except FileNotFoundError:
        pass
    print_software('WordPress', version, root)


def display_joomla(root: Path):
    version = None
    ver_paths = [
        'includes/version.php',
        'libraries/joomla/version.php',
        'libraries/cms/version/version.php',
    ]
    version_file = None
    for path in ver_paths:
        if root.joinpath(path).is_file():
            version_file = root / path
            break
    if version_file:
        version = ''
        with version_file.open(encoding='utf-8') as handle:
            for line in handle:
                if re.search(r'\$RELEASE\s*?=', line):
                    version = line.split("'")[1]
                elif re.search(r'\$DEV_LEVEL\s*?=', line):
                    version += '.' + line.split("'")[1]
                    break
    print_software('Joomla', version, root)


def display_drupal(root: Path):
    version = None
    try:
        with open(root / 'includes/bootstrap.inc', encoding='utf-8') as handle:
            for line in handle:
                if re.search(r'\'VERSION\',\s*?\'', line):
                    version = line.split("'")[3]
                    break
    except FileNotFoundError:
        pass
    print_software('Drupal', version, root)


def display_magento(root: Path):
    version = None
    try:
        with open(
            root / 'app/code/core/Mage/Admin/etc/config.xml', encoding='utf-8'
        ) as handle:
            for line in handle:
                if re.search(r'<version>[0-9\.]+</version>', line):
                    version = re.search(r'[0-9\.]+', line).group(0)
                    break
    except FileNotFoundError:
        pass
    print_software('Magento', version, root)


def display_prestashop(root: Path):
    version = None
    try:
        with open(root / 'docs/readme_en.txt', encoding='utf-8') as handle:
            for line in handle:
                if re.search(r'VERSION: [0-9\.]+', line):
                    version = re.search(r'[0-9\.]+', line).group(0)
                    break
    except FileNotFoundError:
        pass
    print_software('PrestaShop', version, root)


def display_opencart(root: Path):
    version = None
    try:
        with open(root / 'index.php', encoding='utf-8') as handle:
            for line in handle:
                if re.search(r'\'VERSION\',\s*?\'', line):
                    version = line.split("'")[3]
                    break
    except FileNotFoundError:
        pass
    print_software('OpenCart', version, root)


def display_socialengine(root: Path):
    version = None
    try:
        with open(
            root / 'application/libraries/Engine/manifest.php', encoding='utf-8'
        ) as handle:
            for line in handle:
                if re.search(r'\'version\'.*?\'[0-9\.]+\'', line):
                    version = line.split("'")[3]
                    break
    except FileNotFoundError:
        pass
    print_software('SocialEngine', version, root)


def display_oxwall(root: Path):
    version = None
    try:
        with open(root / 'ow_version.xml', encoding='utf-8') as handle:
            for line in handle:
                if re.search(r'<version>[0-9\.]+</version>', line):
                    version = re.search(r'[0-9\.]+', line).group(0)
                    break
    except FileNotFoundError:
        pass
    print_software('Oxwall', version, root)


def display_cakephp(root: Path):
    version = None
    try:
        with open(root / 'lib/Cake/VERSION.txt', encoding='utf-8') as handle:
            for line in handle:
                if re.search(r'^[0-9\.]+', line):
                    version = re.search(r'^[0-9\.]+', line).group(0)
                    break
    except FileNotFoundError:
        pass
    print_software('CakePHP', version, root)


def print_software(cms: str, version: str | None, path: Path):
    if version:
        print(f"{cms} {version}:", path)
    else:
        print(cms, path)


def main():
    users = parse_args()
    header("SERVER OVERVIEW")
    display_server_specs()
    display_disk_info()
    display_apache_info()
    display_php_info()
    display_mysql_info()
    display_cpanel_info()
    check_optimizations()
    check_recent_high_loads()
    check_recent_high_mem()
    display_listeners()
    if users:
        display_software(users)
    header("END OF REPORT")


if __name__ == "__main__":
    main()