1#@!#!123s

: / opt / imh-python / lib / python2.7 / site-packages / rads /
Filename : reseller.py
back
"""Shared RADS reseller functions"""

import os
import re
import json
from rads.shared import is_cpanel_user
from rads.cpanel_api import whm_api, api_success

# We abuse this APIPA address to temporarily assign to resellers during moves
# Per RFC 3927 this should be pseudorandomly generated, but we don't care
APIPA = '169.254.100.100'

with open('/etc/rads/rads.json', 'r') as rads_conf:
    OUR_RESELLERS = json.load(rads_conf)['users']['our_resellers']

def get_owner(user):
    """get the reseller of an account or return None on error"""
    # read /var/cpanel/users/userna5
    user_file_path = os.path.join('/var/cpanel/users', user)
    try:
        with open(user_file_path, 'r') as user_file:
            user_info = user_file.read().splitlines()
    except IOError:
        return None
    owner = None
    for line in user_info:
        if line.startswith('OWNER='):
            owner = line.split('=')[1].strip()
    if owner is None or (owner != 'root' and not is_cpanel_user(owner)):
        # root is the only valid reseller which is not a cpanel user
        return None
    # If owner in trueuserowners != owner user file, don't trust either of them
    # One will be correct, but we can't be sure which
    with open('/etc/trueuserowners', 'r') as trueuserowners_file:
        trueuserowners_data = trueuserowners_file.read().splitlines()
        expected = '%s: %s' % (user, owner)
        if expected not in trueuserowners_data:
            return None
    return owner

def all_resellers():
    """Get all reseller usernames. returns None on error"""
    response = whm_api('listresellers', version=1)
    try:
        return response['data']['reseller']
    except (KeyError, TypeError):
        return None

def all_main_users(res_bool=False, skip=None):
    """Get a all non-child, non-system, "main" cPanel users.
    If on shared, pass the secure# user to skip<list>"""
    to_skip = {'system', 'root'}
    to_skip.update(OUR_RESELLERS)
    if skip:
        to_skip.update(skip)
    resellers = all_resellers()
    if resellers is None:
        return None
    users = [
        x for x in os.listdir('/var/cpanel/users')
        if not x in to_skip and is_cpanel_user(x)
    ]
    for user in users[:]: # loop over a new copy in mem so we can mutate
        if is_child(user, resellers) and not user == get_owner(user):
            users.remove(user)
    if res_bool:
        return {x: x in resellers for x in users}
    return users

def is_reseller(user):
    """True or False if account is a reseller. returns None on error.
    If checking multiple users, use all_resellers() instead"""
    resellers = all_resellers()
    if resellers is None:
        return None
    return user in resellers

def list_reseller_acls():
    """return a list of reseller acl names or None on error"""
    try:
        response = whm_api('listacls', version=1)['data']['acl']
        return [x['name'] for x in response]
    except (TypeError, KeyError):
        return None

def is_child(user, resellers=None):
    """True or False if account is a child of a reseller or None on error.
    A list of all resellers can be passed if checking multiple users to
    speed up this test"""
    owner = get_owner(user)
    if owner is None:
        return None
    if owner in OUR_RESELLERS:
        return False
    if resellers:
        return owner in resellers
    return is_reseller(owner)

def get_reseller_acl(user):
    """Gets the reseller ACL name for a user on the assumption package = acl"""
    userfile_path = os.path.join('/var/cpanel/users', user)
    with open(userfile_path, 'r') as userfile_handle:
        userfile_data = userfile_handle.read().splitlines()
    package = None
    for line in userfile_data:
        if line.startswith('PLAN='):
            try:
                package = line.split('=')[1].strip()
            except IndexError:
                continue
    if package is None:
        return None
    valid_acls = list_reseller_acls()
    return package if package in valid_acls else None

def add_reseller_perms(user, debug=False):
    """Add reseller permissions (without applying an ACL yet)
    Return success as a boolean"""
    result = whm_api('setupreseller', version=1, user=user, makeowner=0)
    other_valid = (
        'tried to make a reseller out of a reseller '
        'when a normal user was expected'
    )
    if debug:
        if result is None:
            return False, None
        success = api_success(result)
        if success:
            return True, result
        else:
            try:
                statmsg = result['metadata']['reason']
            except (ValueError, IndexError):
                return False, result
            return other_valid in statmsg, result
    else:
        if result is None:
            return False
        elif api_success(result):
            return True
        else:
            try:
                statmsg = result['metadata']['reason']
            except (ValueError, IndexError):
                return False
            return other_valid in statmsg

def set_reseller_acl(user, acl_name, debug=False):
    """Set a reseller to a specific ACL. Return success as a boolean"""
    # look before we leap due to cPanel's return status not always being sane
    # with the 'setacls' API function
    valid_acls = list_reseller_acls()
    if valid_acls is None or acl_name not in valid_acls:
        return False
    result = whm_api('setacls', version=1, user=user, acllist=acl_name)
    if debug:
        return api_success(result), result
    return api_success(result)

def get_reseller_ip_pool(user):
    """Get IP pool of a reseller as a list"""
    try:
        result = whm_api('getresellerips', version=1, user=user)['data']
    except (TypeError, KeyError): # None or invalid response
        return None
    if 'all' in result:
        # this reseller is set to open delegation
        return None
    try:
        return result['ip']
    except KeyError:
        return []

def set_reseller_ip_pool(user, ips):
    """Set reseller IP pool"""
    if isinstance(ips, basestring):
        ips = [ips]
    try:
        os.mkdir('/var/cpanel/dips', 0o700)
    except OSError:
        pass
    with open(os.path.join('/var/cpanel/dips', user), 'w') as handle:
        handle.write('\n'.join(ips) + '\n')
    with open(os.path.join('/var/cpanel/mainips', user), 'w') as handle:
        handle.write('\n'.join(ips) + '\n')

def set_reseller_main_ip(user, ip_address, debug=False):
    """Set reseller's main IP. Returns success as a boolean"""
    result = whm_api('setresellermainip', version=1, user=user, ip=ip_address)
    if debug:
        return api_success(result), result
    return api_success(result)

def set_owner(user, owner, debug=False):
    """Set an account's owner. If already owner, still reports success"""
    result = whm_api('modifyacct', version=1, user=user, owner=owner)
    if debug:
        return api_success(result), result
    return api_success(result)

def get_reseller_main_ip(user):
    """Get the "shared" ip of a reseller"""
    try:
        with open(os.path.join('/var/cpanel/mainips', user)) as mainips_file:
            mainip = mainips_file.read().splitlines()[0]
        if mainip == '':
            return None
    except IOError:
        return None
    return mainip

def get_children(owner):
    """get a list of child accounts for a reseller"""
    children = []
    matcher = re.compile(r'(?P<child>\w+):\s*(?P<owner>\w+)')
    with open('/etc/trueuserowners', 'r') as trueuserowners_file:
        trueuserdata = trueuserowners_file.read().splitlines()
    for line in trueuserdata:
        match = matcher.match(line)
        if match is None:
            continue
        gdict = match.groupdict()
        if gdict['owner'] == owner and gdict['child'] != owner:
            children.append(match.groupdict()['child'])
    return children