/usr/lib/imh-ultrastack-common/venv/lib/python3.13/site-packages/ngxconf
Edit: /usr/lib/imh-ultrastack-common/venv/lib/python3.13/site-packages/ngxconf/util.py (13248B)
# vim: set ts=4 sw=4 expandtab syntax=python:
"""
ngxconf.util
Shared utility functions
Copyright (c) 2018-2020 InMotion Hosting, Inc.
http://www.inmotionhosting.com/
@author J. Hipps
"""
import os
import sys
import json
import traceback
import re
import subprocess
import hashlib
import logging
import requests
import yaml
from yaml import CDumper
from netaddr import IPAddress
from yaml import CLoader
from jinja2 import Environment
from ngxconf import __version__, __date__, default_conf, gconf_defaults
# disable warnings
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
g_profiles = {}
CP_TASK_QUEUE = "/var/cpanel/taskqueue/servers_queue.json"
CP_QUEUE_CMD = "/usr/local/cpanel/bin/servers_queue"
class GConf(object):
"""
Global configuration
"""
_conf = {}
def __init__(self, cpath='/opt/ngxconf/config.yaml'):
self.configpath = cpath
self._conf = self.parse_config(cpath)
def __getattr__(self, val):
if val in self._conf:
return self._conf.get(val)
else:
raise KeyError(val)
def __getitem__(self, val):
return self._conf.get(val)
def default(self, val):
gconf_defaults.get(val)
def parse_config(self, cpath):
"""
Parse Ngxconf config
"""
try:
with open(cpath, 'r') as f:
conf = yaml.load(f, Loader=CLoader)
logger.debug("Loaded global config from %s", cpath)
except Exception as e:
logger.warning("Failed to load configuration file: %s", str(e))
conf = gconf_defaults
gconf_defaults.update(conf)
return gconf_defaults
def __repr__(self):
return ""
class DomainCache(object):
"""
Domain list object
"""
domlist = {}
def __init__(self, domfile="/etc/userdomains"):
try:
with open(domfile) as f:
for tline in f.readlines():
try:
k, v = tline.split(': ', 1)
self.domlist[k.strip()] = v.strip()
except:
continue
except:
logger.error("Failed to read domain list from %s", domfile)
def exists(self, dom):
"""
Determine if a domain exists
"""
return bool(self.domlist.get(dom))
class WHMAPI(object):
"""
WHM/cPanel API connector
"""
_rq = None
_user = 'root'
_hash = None
prefix = None
def __init__(self, server):
# Grab root accesshash; gen if doesn't exist
if not self._getRootHash():
if not self._genHash():
logger.error("Unable to retrieve WHMAPI credentials; giving up")
return
# Create requests session
self._rq = requests.Session()
self._rq.headers = {'Authorization': "WHM %s:%s" % (self._user, self._hash)}
self._rq.verify = False
self.prefix = "https://%s:2087/json-api" % (server)
def whm(self, cmd, **kwargs):
"""run whmapi1 command"""
try:
resp = self._rq.get(self.prefix+'/'+cmd, params=kwargs)
except Exception as e:
logger.error("whmapi request [%s] failed: %s", self.prefix+'/'+cmd, str(e))
return None
try:
rjson = resp.json()
except:
rjson = None
return rjson
def cpanel(self, mod, cmd, user, **kwargs):
"""run cpapi2 command"""
cargs = kwargs
cargs.update({'cpanel_jsonapi_user': user, 'cpanel_jsonapi_apiversion': "2",
'cpanel_jsonapi_module': mod, 'cpanel_jsonapi_func': cmd})
try:
resp = self._rq.get(self.prefix + '/cpanel', params=cargs)
except Exception as e:
logger.error("cpapi2 request [%s] failed: %s", self.prefix+'/cpanel', str(e))
return None
try:
rjson = resp.json()
rdata = rjson['cpanelresult']['data']
except:
rdata = None
return rdata
def _getRootHash(self, hashfile='/root/.accesshash'):
"""
Retrieve root hash (requires this script to be running with effective root permissions)
"""
try:
with open(hashfile) as ahf:
hashraw = ahf.read()
self._hash = hashraw.replace('\n', '')
return self._hash
except Exception as e:
logger.error("Failed to read root accesshash: %s", str(e))
return None
def _genHash(self, hashfile='/root/.accesshash'):
"""
Generate access hash. 29 lines x 32 chars, first line empty (newline only). Grab 1Kbyte
from /dev/urandom and generate an MD5 hash (32 chars) * 29 times
"""
# pylint: disable=unused-variable
hashout = ''
for ll in range(1, 30):
hashout += '\n' + hashlib.md5(os.urandom(1024)).hexdigest()
try:
with open(hashfile, 'w') as f:
f.write(hashout)
os.chmod(hashfile, 0o0600)
logger.warning("New root access hash generated OK")
self._hash = hashout
return hashout
except Exception as e:
logger.error("Failed to generate new root accesshash: %s", str(e))
return None
class NGXDumper(CDumper):
def represent_none(self, data):
return self.represent_scalar('tag:yaml.org,2002:null', '~')
logger = logging.getLogger('ngxconf')
gconf = GConf()
whm = None
pkgdata = None
NGXDumper.add_representer(type(None), NGXDumper.represent_none)
def load_template(tname, basepath=None):
"""
Load template @basepath/@tname.j2; return template object
"""
if basepath is None:
basepath = gconf.template_basepath
tpath = "%s/%s.j2" % (basepath, tname)
# Add custom filters
tenv = Environment(trim_blocks=True, lstrip_blocks=True)
def _ngx_safe_regex(instr, fallback):
try:
re.compile(instr)
return instr
except Exception as e:
try:
fbstr = '|'.join(default_conf[fallback])
except:
fbstr = 'INVALID_REGEX_AND_FALLBACK'
logger.warning("Invalid user-generated regex (%s): '%s' -- falling back to '%s'", str(e), instr, fbstr)
return fbstr
def _ngx_regex_is_valid(instr):
try:
re.compile(instr)
return True
except Exception as e:
logger.warning("Invalid user-generated regex (%s): '%s'", str(e), instr)
return False
def _ipaddr(instr):
try:
IPAddress(instr)
return True
except Exception as e:
return False
tenv.filters['ngx_safe_regex'] = _ngx_safe_regex
tenv.filters['ngx_regex_is_valid'] = _ngx_regex_is_valid
tenv.globals['valid_ipaddr'] = _ipaddr
# Load template from file
try:
with open(tpath, 'r') as f:
traw = f.read()
except Exception as e:
logger.error("Failed to read template from %s: %s", tpath, str(e))
return None
# Parse with Jinja2
try:
templ = tenv.from_string(traw)
logger.debug("Parsed template '%s' from file %s", tname, tpath)
except Exception as e:
logger.error("Failed to parse template %s: %s", tname, str(e))
return None
return templ
def parse_user_default_conf(cpath='/opt/ngxconf/defaults.yaml'):
"""
Parse user default config file
Requires `apply_user_default_config: true` in global conf
"""
try:
with open(cpath, 'r') as f:
conf = yaml.load(f, Loader=CLoader)
logger.debug("Loaded user default config from %s", cpath)
except Exception as e:
logger.warning("Failed to load user default config file: %s", str(e))
conf = None
return conf
def parse_profiles(basepath="/opt/ngxconf/profiles"):
"""
Parses all profiles located at @basepath
"""
global g_profiles
try:
plist = [x for x in os.listdir(basepath) if x.endswith('.yaml')]
except Exception as e:
logger.error("Failed to read profile directory: %s", str(e))
return False
for tpath in plist:
trealpath = os.path.realpath(os.path.join(basepath, tpath))
try:
with open(trealpath, 'r') as f:
tprof = yaml.load(f, Loader=CLoader)
g_profiles[tprof['id']] = tprof
logger.debug("Parsed profile: %s [%s]", tprof['id'], trealpath)
except Exception as e:
logger.error("Failed to parse profile %s: %s", trealpath, str(e))
return True
def get_profile(prof_id):
"""
Generates a combined profile for @prof_id
Returns a dict object
"""
global g_profiles
if prof_id not in g_profiles:
return None
tprof = g_profiles[prof_id]
if tprof.get('extends'):
pprofile = get_profile(tprof['extends'])
else:
pprofile = default_conf
pprofile.update(tprof['config'])
return pprofile
def get_cp_queued_tasks(command):
"""
Reads from cPanel's task queue JSON file, and
returns a list of any pending actions for @command
"""
try:
with open(CP_TASK_QUEUE) as f:
rawq = json.load(f)[2]
except Exception as e:
logger.error("get_cp_queued_tasks: failed to read servers_queue.json: %s", str(e))
return None
tasks = []
try:
for ttask in rawq.get('waiting_queue', []) + rawq.get('deferral_queue', []):
if ttask.get('_command') == command:
tasks.append(ttask.get('_uuid'))
except Exception as e:
logger.error("get_cp_queued_tasks: failed to parse queue data: %s", str(e))
return None
return tasks
def cancel_cp_pending_tasks(command):
"""
Cancel any cPanel tasks pending for @command
Returns number of tasks de-queued
"""
tkilled = 0
tasks = get_cp_queued_tasks(command)
if tasks is None:
return 0
for ttask in tasks:
try:
ret = subprocess.run(
[CP_QUEUE_CMD, 'unqueue', ttask],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
timeout=10,
)
except subprocess.TimeoutExpired:
pass
if ret.returncode != 0:
logger.error("cancel_cp_pending_tasks: failed to de-queue task %s (return code %d)", ttask, ret.returncode)
else:
tkilled += 1
logger.warning("cancel_cp_pending_tasks: de-queued %s task %s", command, ttask)
return tkilled
def get_min_uid():
"""
Return minimum UID for current system
"""
try:
with open('/etc/login.defs') as f:
llines = f.readlines()
except Exception as e:
logger.error("Failed to read login.defs: %s", str(e))
return None
ldefs = {}
for tline in llines:
try:
var, val = re.match(r'^(?P[A-Z_0-9]+)\s+(?P.+)$', tline).groups()
ldefs[var] = val
except:
continue
uid_min = int(ldefs.get('UID_MIN', '1000'))
logger.debug("Detected UID_MIN = %d", uid_min)
return uid_min
def cp_get_package_data(pkg):
"""
Return info for specified package
Retrieves a list of packages and package limits, then caches it
"""
global whm
global pkgdata
if whm is None:
whm = WHMAPI('localhost')
if pkgdata is None:
pkgdata = {}
packraw = whm.whm('listpkgs').get('package')
if packraw:
for tpack in packraw:
pkgdata[tpack['name']] = tpack
else:
logger.error("Failed to retrieve list of packages; packraw = %s", packraw)
return {}
if not pkgdata.get(pkg):
logger.warning("get_package_data: no matching package '%s'", pkg)
return {}
else:
return pkgdata.get(pkg)
def cp_get_userinfo(username):
"""
Parse file from /var/cpanel/users/@username, injects [_mtime] field from stat result
and return a dict of values or None on error
"""
cpath = os.path.join('/var/cpanel/users', username)
try:
with open(cpath) as f:
cdata = f.readlines()
except Exception as e:
logger.error("Failed to read %s: %s", cpath, str(e))
return None
udata = {}
for tline in cdata:
try:
if tline.startswith('#'): continue
k, v = tline.strip().split('=', 1)
udata[k.strip()] = v.strip()
except:
continue
try:
udata['_mtime'] = int(os.stat(cpath).st_mtime)
except:
udata['_mtime'] = None
return udata
def dict_strip_mtime(idata):
"""
Strip _mtime from dict
"""
return {x: y for x, y in idata.items() if x != '_mtime'}
def excepthook(etype, evalue, etraceback):
"""
Default exception hook
"""
logger.critical("Unhandled exception caught:\n%s",
'\n'.join(traceback.format_exception(etype, evalue, etraceback)))