1#@!#!123s

: / opt / dedrads / account_review /
Filename : cpanel.py
back
#!/usr/lib/rads/venv/bin/python3
"""Account-review: Shows relevant information about cPanel users"""


import sys
import os
import glob
import platform
from prettytable import PrettyTable
from cpapis import whmapi1
import rads
from rads import color, UserData, CpuserError

STANDARD_NS = [
    'ns1.webhostinghub.com',
    'ns2.webhostinghub.com',
    'ns1.servconfig.com',
    'ns2.servconfig.com',
    'ns.inmotionhosting.com',
    'ns1.inmotionhosting.com',  # there's a cname to ns
    'ns2.inmotionhosting.com',
]

# IOError: file unreadable
# ValueError: bad json
# KeyError: expected key is missing from dict
# AttributeError: tried to run .values(), etc on non-dict
# TypeError: tried to iterate on a None
BAD_YAML = (IOError, ValueError, KeyError, AttributeError, TypeError)
BAD_API = (ValueError, KeyError, AttributeError, TypeError)


def errprint(*errmsgs, **kwargs):
    """deprecated, copied from rads.common"""
    errmsgs = [color.red(x) for x in errmsgs]
    try:
        fatal = kwargs.pop('fatal')
    except KeyError:
        fatal = False
    print(*errmsgs, file=sys.stderr, **kwargs)
    if fatal is True:
        sys.exit(1)
    elif fatal is False:
        return
    else:
        sys.exit(fatal)


def conv_size(size):
    """Convert size, in MB, to human GB/MB notation"""
    try:
        size_int = int(float(str(size).rstrip('M')))
    except ValueError:
        return size
    if size_int > 1024:
        return f'{size_int / 1024.0:.1f} GB'
    return f'{size_int} MB'


def fake_header(table, header):
    """Supplied a prettytable object, form a fake header"""
    width = len(str(table).splitlines()[0]) - 2
    print('+', '-' * width, '+', sep='')
    print(f'|{color.bold(header.center(width))}|')


def get_users_from_args():
    """Parse a list of users to display from CLI args or working dir"""
    argv = sys.argv[1:]
    restricted = rads.OUR_RESELLERS
    restricted.extend(rads.SYS_USERS)
    secure_user = rads.SECURE_USER
    if secure_user is not None:
        restricted.append(secure_user)
    if len(sys.argv) < 2:
        try:
            argv.append(os.getcwd().split('/')[2])
        except IndexError:
            pass
    good_users = []
    for user in argv:
        if not rads.is_cpuser(user):
            errprint(user, 'is not a valid cPanel user')
        elif user in restricted:
            errprint(user, 'is a restricted user')
        else:
            good_users.append(user)
    if len(good_users) == 0:
        errprint('no users to display', fatal=True)
    return good_users


def check_modsec_custom(user):
    """Check for Apache customizations"""
    if os.path.isfile('/etc/cpanel/ea4/is_ea4'):
        basepath = '/etc/apache2/conf.d/userdata/std/2*'
    else:
        basepath = '/usr/local/apache/conf/userdata/std/2'
    conf_paths = os.path.join(basepath, user, '*/modsec.conf')
    return len(glob.glob(conf_paths)) > 0


def check_homedir(user):
    """Check to see if user's homedir exists"""
    return os.path.isdir(os.path.expanduser("~%s" % user))


def check_email_suspension(user):
    """
    Checks /etc/outgoing_mail_suspended_users for passed user
    Returns True if found, otherwise returns False.
    """

    with open('/etc/outgoing_mail_suspended_users', encoding="utf-8") as file:
        for lines in file:
            if user in lines:
                return True
    return False


def count_databases(user):
    """Get number of databases"""
    paths = []
    paths.append(os.path.join('/var/cpanel/datastore', user, 'mysql-db-count'))
    paths.append(
        os.path.join('/home', user, '.cpanel/datastore/mysql-db-count')
    )
    for path in paths:
        try:
            with open(path, encoding="utf-8") as file:
                return file.read().strip()
        except OSError:
            continue
    return '0'


def display_user(user, shared_ips):
    """Display one user. Called per argument by main()"""
    # domin info / ssl and doc roots
    try:
        udata = UserData(user)
    except CpuserError as exc:
        errprint(exc)
        print(
            'See http://wiki.inmotionhosting.com/index.php?'
            'title=Account-review#Troubleshooting'
        )
        return
    # get SSL status of primary domain
    if udata.primary.has_ssl:
        ssl_text = 'Has SSL'
    else:
        ssl_text = 'No SSL'
    # account info from WHM
    try:
        acct_summ = whmapi1('accountsummary', {"user": user})['data']['acct'][0]
    except BAD_API:
        errprint('Error running "accountsummary" in WHM API')
        return
    # Create the headers
    usrtable = PrettyTable(['1', '2', '3'])
    usrtable.align['1'] = 'r'
    usrtable.align['2'] = 'l'
    usrtable.align['3'] = 'l'
    usrtable.header = False

    # 'Created On' crow
    usrtable.add_row(['Created on', acct_summ['startdate'], ''])

    # 'Owner' row
    if acct_summ['owner'] == user:
        owner_text = color.yellow('Self')
    elif acct_summ['owner'] == 'root':
        owner_text = color.red('Issue')
    else:
        owner_text = ''
    usrtable.add_row(['Owner', acct_summ['owner'], owner_text])

    # 'Contact' row
    usrtable.add_row(['cPanel Contact', acct_summ['email'], ''])

    # 'Account Size' row
    usrtable.add_row(['Account Size', conv_size(acct_summ['diskused']), ''])

    # 'MySQL Databases' row
    usrtable.add_row(['MySQL Databases', count_databases(user), ''])

    # 'IP' row
    if acct_summ['ip'] in shared_ips:
        ip_text = 'Shared'
    else:
        ip_text = color.red('Dedicated')
    usrtable.add_row(['IP', acct_summ['ip'], ip_text])

    # 'SSL' rowdd
    usrtable.add_row(['SSL', ssl_text, ''])

    # 'Docroot' row
    if udata.primary.docroot == os.path.join('/home', user, 'public_html'):
        docroot_text = 'Standard'
    else:
        docroot_text = color.red('Custom')
    usrtable.add_row(['Docroot', udata.primary.docroot, docroot_text])

    # 'Modsec' row
    if check_modsec_custom(user):
        modsec_text = color.red('Modified')
    else:
        modsec_text = 'Enabled'
    usrtable.add_row(['Modsec', modsec_text, ''])

    # 'Homedir' row
    if check_homedir(user):
        homedir_text = os.path.expanduser("~%s" % user)
    else:
        homedir_text = color.red('Missing')
    usrtable.add_row(['Homedir', homedir_text, ''])

    # We print out our own fake header, not the one supplied by prettytable
    header = '{0.user} / {0.primary.domain} / {1[plan]}'.format(
        udata, acct_summ
    )
    fake_header(usrtable, header)
    print(usrtable)

    # Suspension notice
    if acct_summ['suspended']:
        print(color.red('Suspended:'), acct_summ['suspendreason'].strip())
    else:
        print('Suspended: No')

    # Check for outgoing email suspension
    if check_email_suspension(user):
        print(color.red('Outgoing Email Suspended: Yes'))
    else:
        print('Outgoing Email Suspended: No')

    # Print Addon/Parked/Sub domains
    # This is all handled by the same table to align them all the same
    cols = [
        'labelcol',
        'domlabel',
        'dom',
        'rootlabel',
        'root',
        'ssllabel',
        'ssl',
    ]
    dom_table = PrettyTable(cols)
    dom_table.border = False
    dom_table.header = False
    dom_table.align['labelcol'] = 'r'
    dom_table.align['dom'] = 'l'
    dom_table.align['root'] = 'l'
    table_labels = {'Addon': 'addons', 'Parked': 'parked', 'Sub': 'subs'}
    for label, lblkey in table_labels.items():
        if len(getattr(udata, lblkey)) == 0:
            continue
        # the fake "header" to display between each type of domain
        head_row = [
            color.bold(label),
            color.bold('Domains:'),
            '',
            '',
            '',
            '',
            '',
        ]
        dom_table.add_row(['', '', '', '', '', '', ''])
        dom_table.add_row(head_row)

        # iterate over each domain of this type
        for dom in getattr(udata, lblkey):
            if label == 'Parked':
                ssl_label = ''
                ssl_out = ''
            elif dom.has_ssl:
                ssl_label = color.magenta('Has SSL:')
                ssl_out = 'Yes'
            else:
                ssl_label = color.magenta('Has SSL:')
                ssl_out = 'No'
            row = [
                ' ' * 4,
                color.magenta('Domain:'),
                dom.domain,
                color.magenta('Docroot:'),
                dom.docroot,
                ssl_label,
                ssl_out,
            ]
            dom_table.add_row(row)
    print(dom_table)

    # Reseller Info Table
    reseller = None
    try:
        reseller = user in whmapi1("listresellers")['data']['reseller']
    except Exception as e:
        errprint(f'Error obtaining reseller list from WHM API. {e}', fatal=True)
    if not reseller:
        return
    try:
        res_stats = whmapi1('resellerstats', {"user": user})['data']['reseller']
    except BAD_API:
        errprint('Error running "resellerstats" API function', fatal=True)
    res_table = PrettyTable(['1', '2', '3'])
    res_table.header = False
    res_table.align['1'] = 'r'
    res_table.align['2'] = 'l'
    res_table.align['3'] = 'l'
    res_table.add_row(['Child Count', len(res_stats['acct']), ''])
    if acct_summ['owner'] != user:
        try:
            diskused = float(acct_summ['diskused'][:-1])
        except ValueError:
            diskused = None
            used_raw = acct_summ['diskused']
            errprint(
                'Unexpected value for "diskused" in accountsummary for', user
            )
            errprint(rf"Expcted to match regex '^\d+M$' but got {used_raw}")
        if diskused is None:
            combined = 'error'
        else:
            combined = conv_size(res_stats['diskused'] + diskused)
        res_table.add_row(['Combined Size', combined, ''])
    else:
        res_table.add_row(
            ['Combined Size', conv_size(res_stats['diskused']), '']
        )
    for index, (nsname, nstext) in enumerate(get_nameservers(user)):
        res_table.add_row([f'Nameserver {index + 1}', nsname, nstext])
    for index, (dip, diptext) in enumerate(get_ip_pool(user, shared_ips)):
        res_table.add_row([f'Pool IP {index + 1}', dip, diptext])
    print()
    fake_header(res_table, f'Reseller Stats for {user}')
    print(res_table)

    # Child Accounts Table
    child_table = PrettyTable(
        ['Child', 'Domain', 'Size', 'Suspended', 'Deleted']
    )
    child_table.align['Domain'] = 'l'
    child_table.align['Size'] = 'r'
    is_deleted = False
    for acct in res_stats['acct']:
        if acct['suspended']:
            susp_text = color.red('yes')
        else:
            if check_email_suspension(acct['user']):
                susp_text = color.red('Outbound Email')
            else:
                susp_text = 'no'
        if acct['deleted']:
            del_text = color.red('yes†')
            is_deleted = True
        else:
            del_text = 'no'
        if user == acct['user']:
            user_text = color.yellow(user)
        else:
            user_text = acct['user']
        child_table.add_row(
            [
                user_text,
                acct['domain'],
                conv_size(acct['diskused']),
                susp_text,
                del_text,
            ]
        )
    print()
    fake_header(child_table, f'Child accounts of {user}')
    print(child_table)
    if is_deleted:
        print(
            color.red('†')
            + "Deleted accounts cannot be recovered without a backup."
        )


def get_ip_pool(user, shared_ips):
    """Get IP pool for a reseller and status text for each"""
    dips_path = os.path.join('/var/cpanel/dips', user)
    try:
        with open(dips_path, encoding="utf-8") as file:
            dips = file.read().split()
    except OSError:
        return []
    others = set()
    for reseller in os.listdir('/var/cpanel/dips'):
        if reseller == user:
            continue
        other_path = os.path.join('/var/cpanel/dips', reseller)
        try:
            with open(other_path, encoding="utf-8") as file:
                others.update(file.read().split())
        except OSError:
            errprint('Could not read', other_path)
    res_main_path = os.path.join('/var/cpanel/mainips', user)
    try:
        with open(res_main_path, encoding="utf-8") as file:
            res_main = file.read().split()
    except OSError:
        errprint('Could not read', res_main_path)
        res_main = []
    ip_pool = []
    for dip in dips:
        ip_msgs = []
        if dip in res_main:
            ip_msgs.append('Pool Main')
        if dip in others:
            ip_msgs.append(color.red('Reseller Conflict'))
        if dip in shared_ips:
            ip_msgs.append(color.red('Server Main'))
        ip_pool.append([dip, '\n'.join(ip_msgs)])
    return ip_pool


def get_nameservers(user):
    """Get nameservers for a reseller and status text for each"""
    try:
        with open(
            '/var/cpanel/resellers-nameservers', encoding="utf-8"
        ) as file:
            data = file.read().splitlines()
    except OSError:
        errprint('could not read /var/cpanel/resellers-nameservers')
        return []
    starter = f'{user}:'
    for line in data:
        if not line.startswith(starter):
            continue
        raw_nsnames = line[len(starter) :].strip().split(',')
        nsnames = []
        for nsname in raw_nsnames:
            nsname = nsname.strip()
            if len(nsname) > 0:
                nsnames.append(nsname)
        nsinfo = []
        for nsname in nsnames:
            nsinfo.append([nsname, ns_custom(nsname)])
        return nsinfo
    errprint('user missing from /var/cpanel/resellers-nameservers')
    return []


def ns_custom(nsname):
    """Try to determine if a nameserver is custom"""
    if nsname not in STANDARD_NS:
        return color.red('Custom')
    fqdn = platform.node()
    if len(fqdn.split('.')) != 3:
        return color.yellow('Custom? Server fqdn setup wrong')
    fqdn_split = fqdn.split('.')
    fqdn_split[0] = ''
    dom = '.'.join(fqdn_split)
    if nsname.endswith(dom):
        return 'Standard'
    return color.yellow('Server FQDN mismatch')


def cpanel_display_user(users):
    """Main loop - run display_user on each user"""
    with open('/var/cpanel/mainip', encoding="utf-8") as handle:
        shared_ips = [handle.read().strip()]
    try:
        with open('/var/cpanel/mainips/root', encoding="utf-8") as handle:
            shared_ips.extend(handle.read().split())
    except OSError:
        pass  # this file may not exist; that is okay

    for user in users:
        display_user(user, shared_ips)