Viewing file:  cldetectlib.py (18.4 KB) -rw-r--r--
Select action/file-type:
 (+) |  (+) |  (+) | Code (+) | Session (+) |  (+) | SDB (+) |  (+) |  (+) |  (+) |  (+) |  (+) |
# -*- coding: utf-8 -*-
# CLDETECT python lib
#
# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2019 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENSE.TXT
# Detection:
#
# Control Panel name & version
# Control Panel name
# Control Panel admin email
# CXS is installed
# mod_suphp is enabled for easyapache on cPanel
# get apache gid
# Detect LiteSpeed
# Detect PostGreSQL
# Detect admin user for DirectAdmin control panel
# Detect CloudLinux instalation process
# Detect Nagios
# Detect if cloudlinux=yes is present for DirectAdmin
# Get fs.enforce_symlinksifowner from /etc/sysctl.conf
# Detect suEXEC
# Detect suPHP
# Check suEXEC or suPHP for SecureLVE jail
# Check /etc/ssh/sshd_config for UsePAM yes
# Separate functions for detect machines: is_da, is_isp, etc
# Detect cagefs installed
import os
import pwd
import re
import subprocess
import sys
from configparser import ConfigParser, NoOptionError, NoSectionError
from clcommon import cpapi
from clcommon.sysctl import SYSCTL_CL_CONF_FILE, SysCtlConf
# Control panel name
CP_NAME = None
# Control panel version
CP_VERSION = None
# If CP_NAME is "ISPManager" and CP_VERSION is "5.xx" ISP5 Type: "Master" or "Node".
# else - always None
CP_ISP_TYPE = None
CP_ADMIN_EMAIL = None
NAGIOS_GID = 0
APACHE_GID = 48
APACHE_UNAME = "apache"
LITESPEED_CONFIG_FILE = "/usr/local/lsws/conf/httpd_config.xml"
LITESPEED_OPEN_CONFIG_FILE = "/usr/local/lsws/conf/httpd_config.conf"
LITESPEED_VERSION_FILE = "/usr/local/lsws/VERSION"
POSTGRE_SERVER_FILE = None
POSTGRE_SYSTEMD_PATH = "/usr/lib/systemd/system/postgresql.service"
POSTGRE_INITD_PATH = "/etc/rc.d/init.d/postgresql"
CL_SETUP_LOCK_FILE = "/var/lock/cldeploy.lck"
CL_CONFIG_FILE = "/etc/sysconfig/cloudlinux"
USEPAM_FILE = "/etc/ssh/sshd_config"
SUEXEC_ENABLED = None
SUPHP_ENABLED = None
SHARED_PRO_EDITION_HUMAN_READABLE = "CloudLinux OS Shared Pro"
SHARED_EDITION_HUMAN_READABLE = "CloudLinux OS Shared"
SOLO_EDITION_HUMAN_READABLE = "CloudLinux OS Solo"
if os.path.isfile(POSTGRE_SYSTEMD_PATH):
POSTGRE_SERVER_FILE = POSTGRE_SYSTEMD_PATH
else:
POSTGRE_SERVER_FILE = POSTGRE_INITD_PATH
def is_ea4():
return os.path.exists("/etc/cpanel/ea4/is_ea4")
# This function get CP name and CP version
def getCP():
global CP_NAME
global CP_VERSION
global CP_ISP_TYPE
CP_NAME = "Unknown"
CP_VERSION = "0"
CP_ISP_TYPE = None
####################################################################
# Try to detect panels supported by CL and custom panel with cpapi plugin
try:
panel_data = cpapi.get_cp_description()
if panel_data:
CP_NAME = panel_data["name"]
CP_VERSION = panel_data["version"]
CP_ISP_TYPE = panel_data["additional_info"]
except Exception:
pass
# Try to detect some other panels without retrieving info about them
####################################################################
# H-Sphere
try:
with open("/hsphere/shared/version", encoding="utf-8") as f:
data = f.read()
release = re.findall(r"Release:\s+(.+)", data)[0]
version = re.findall(r"Version:\s+(.+)", data)[0]
CP_NAME = "H-Sphere"
CP_VERSION = f"{release}.{version}"
return True
except Exception:
pass
####################################################################
# HostingNG check
if os.path.isfile("/lib64/libnss_ng.so"):
CP_NAME = "HostingNG"
CP_VERSION = "none"
return True
####################################################################
# CentOS Web Panel check
if os.path.isdir("/usr/local/cwpsrv"):
CP_NAME = "CentOS_WEB_Panel"
CP_VERSION = "none"
return True
# Atomia check: (what is atomia you can see at www.atomia.com)
# Atomia is more than just CP inside the CloudLinux,
# So we just check presence of Atomia program agent
# by its footprints - config files, which agent created.
if os.path.isfile("/etc/httpd/conf.d/atomia-pa-apache.conf") or os.path.isdir("/storage/configuration/cloudlinux"):
CP_NAME = "Atomia_agent"
CP_VERSION = "none"
return True
# Cyber Panel
if os.path.isdir("/usr/local/CyberCP"):
CP_NAME = "Cyberpanel"
CP_VERSION = "none"
return True
# Planet Hoster
if os.path.isdir("/var/phmgr"):
CP_NAME = "PlaneHoster"
CP_VERSION = "none"
return True
# Vesta CP, check it`s main dir
# can install from https://vestacp.com/install/
if os.path.isdir("/usr/local/vesta"):
CP_NAME = "Vesta"
CP_VERSION = "none"
return True
# We can check if VirtualminWebmin is installed by checking the license file.
# That file is always present, license serial and key are predefined
# in the beginning of the installation script
if os.path.isfile("/etc/virtualmin-license"):
CP_NAME = "VirtualminWebmin"
CP_VERSION = "none"
return True
# Detect Webuzo panel
if os.path.isfile("/usr/local/webuzo/universal.php"):
CP_NAME = "Webuzo"
CP_VERSION = "none"
return True
# No panel detected
return False
# Get params value from file
def get_param_from_file(file_name, param_name, separator=None, default_val=""):
try:
with open(file_name, encoding="utf-8") as f:
content = f.readlines()
except OSError:
return default_val
for line in content:
line = line.strip()
if line.startswith(param_name):
lineParts = line.split(separator)
if (len(lineParts) == 2) and (lineParts[0].strip() == param_name):
return lineParts[1].strip()
return default_val
# This function get CP name only
def getCPName():
global CP_NAME
if CP_NAME:
return CP_NAME
# cPanel check
if os.path.isfile("/usr/local/cpanel/cpanel"):
CP_NAME = "cPanel"
# Plesk check
elif os.path.isfile("/usr/local/psa/version"):
CP_NAME = "Plesk"
# DirectAdmin check
elif os.path.isfile("/usr/local/directadmin/directadmin"):
CP_NAME = "DirectAdmin"
# ISPmanager v4 or v5 check
elif os.path.isfile("/usr/local/ispmgr/bin/ispmgr") or os.path.isdir("/usr/local/mgr5"):
CP_NAME = "ISPManager"
# InterWorx check
elif os.path.isdir("/usr/local/interworx"):
CP_NAME = "InterWorx"
# HSphere check
elif os.path.isdir("/hsphere/shared"):
CP_NAME = "H-Sphere"
elif os.path.isfile("/lib64/libnss_ng.so"):
CP_NAME = "HostingNG"
# CentOS Web Panel check
elif os.path.isdir("/usr/local/cwpsrv"):
CP_NAME = "CentOS_WEB_Panel"
elif os.path.isfile("/etc/httpd/conf.d/atomia-pa-apache.conf") or os.path.isdir(
"/storage/configuration/cloudlinux"
):
CP_NAME = "Atomia_agent"
elif os.path.isdir("/usr/local/vesta"):
CP_NAME = "Vesta"
elif os.path.isfile("/etc/virtualmin-license"):
CP_NAME = "VirtualminWebmin"
elif os.path.isdir("/var/phmgr"):
CP_NAME = "PlaneHoster"
elif os.path.isdir("/usr/local/CyberCP"):
CP_NAME = "Cyberpanel"
elif os.path.isfile("/usr/local/webuzo/universal.php"):
CP_NAME = "Webuzo"
else:
# Detect custom panel name
panel_data = cpapi.get_cp_description()
# If panel data retreived, use its name
CP_NAME = panel_data["name"] if panel_data else "Unknown"
return CP_NAME
def add_server_stats(status_report):
"""
Add server statistics to status_report dict
:param status_report: dict to add statistics to
:type status_report: dict
"""
from clcommon import ClPwd # pylint: disable=import-outside-toplevel
res = {}
cp_name = getCPName()
if cp_name != "Unknown":
res["cp"] = cp_name
if cp_name == "Plesk":
clpwd = ClPwd(10000)
else:
clpwd = ClPwd()
d = clpwd.get_uid_dict()
users = 0
sys_users = {
"nfsnobody",
"avahi-autoipd",
"exim",
"clamav",
"varnish",
"nagios",
"saslauth",
"mysql",
"lsadm",
"systemd-bus-proxy",
"systemd-network",
"polkitd",
"firebird",
"nginx",
"dovecot",
"dovenull",
"roundcube_sysuser",
"cpanel",
"cpanelhorde",
"cpanelphpmyadmin",
"cpanelphppgadmin",
"cpanelroundcube",
"mailman",
"cpaneleximfilter",
"cpanellogaholic",
"cpanellogin",
"munin",
"cpaneleximscanner",
"cpanelphpgadmin",
"cpses",
"cpanelconnecttrack",
"cpanelrrdtool",
"admin",
"webapps",
"apache",
"diradmin",
"majordomo",
"viapm",
"iworx",
"iworx-web",
"iworx-pma",
"iworx-backup",
"iworx-horde",
"iworx-roundcube",
"iworx-sqmail",
"iworx_support_user",
"psaadm",
"popuser",
"psaftp",
"drweb",
"sw-cp-server",
"horde_sysuser",
}
for pw_entries in d.values():
found = False
for entry in pw_entries:
if entry.pw_name in sys_users:
found = True
break
if not found:
users += 1
res["users"] = users
status_report["cln"] = res
# Control Panel admin email
def getCPAdminEmail():
global CP_ADMIN_EMAIL
if CP_ADMIN_EMAIL:
return CP_ADMIN_EMAIL
if not os.path.isfile(CL_CONFIG_FILE):
print("Error: missing " + CL_CONFIG_FILE + " config file.")
sys.exit(1)
try:
parser = ConfigParser(interpolation=None, strict=False)
parser.read(CL_CONFIG_FILE)
if parser.get("license_check", "EMAIL").strip().find("@") != -1:
CP_ADMIN_EMAIL = parser.get("license_check", "EMAIL").strip()
else:
try:
getCPName()
get_email_script = parser.get("license_check", CP_NAME + "_getemail_script")
if not os.path.isfile(get_email_script):
raise FileNotFoundError
with subprocess.Popen(
[get_email_script],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
) as proc:
out, _ = proc.communicate()
CP_ADMIN_EMAIL = out.strip()
except (NoSectionError, NoOptionError, FileNotFoundError):
CP_ADMIN_EMAIL = "root@localhost.localdomain"
return CP_ADMIN_EMAIL
except Exception:
print("Error: bad " + CL_CONFIG_FILE + " config file.")
sys.exit(1)
# Check is CXS installed
def CXS_check():
return os.path.isdir("/etc/cxs")
# Check is mod_suphp is enabled in easyapache on cPanel
# TODO check cagefs_posteasyapache_hook.sh for suPHP check via /usr/local/cpanel/bin/rebuild_phpconf --available
def mod_suPHP_check():
getCPName()
if CP_NAME != "cPanel":
return False
return os.path.isfile("/usr/local/apache/modules/mod_suphp.so")
# Get Apache gid
def get_apache_gid():
getCPName()
global APACHE_GID
global APACHE_UNAME
if CP_VERSION == "0":
return False
if CP_NAME == "cPanel":
APACHE_UNAME = "nobody"
if CP_NAME == "H-Sphere":
APACHE_UNAME = "httpd"
# line 24 | APACHE_UNAME = 'apache' - for others control panel (DA,ISP,IWorx,Plesk)
try:
APACHE_GID = pwd.getpwnam(APACHE_UNAME).pw_gid
except Exception:
pass
return True
# Detect LiteSpeed
def detect_litespeed():
"""
LiteSpeed can be enterprise or open source, and each of them
stores config in different formats
So this checker will search for one of them
"""
return detect_enterprise_litespeed() or detect_open_litespeed()
def detect_enterprise_litespeed():
"""
Detect LSWS Enterprise presence
"""
return os.path.isfile(LITESPEED_CONFIG_FILE)
def detect_open_litespeed():
"""
Detect OpenLiteSpeed presence
"""
return os.path.isfile(LITESPEED_OPEN_CONFIG_FILE)
def get_litespeed_version():
"""
Determine Litespeed version.
Works for both LSWS Enterprise and OpenLiteSpeed.
"""
try:
# Content of LITESPEED_VERSION_FILE: '5.4.12'
with open(LITESPEED_VERSION_FILE, encoding="utf-8") as f:
return f.read().strip()
except (FileNotFoundError, OSError):
return ""
# Detect PostGreSQL
def detect_postgresql():
return os.path.isfile(POSTGRE_SERVER_FILE)
# Detect DirectAdmin admin user
def detect_DA_admin():
getCPName()
if CP_NAME != "DirectAdmin":
return False
try:
with open("/usr/local/directadmin/conf/directadmin.conf", encoding="utf-8") as f:
out = f.read()
return out.split("admindir=")[1].split("\n")[0].split("/")[-1].strip()
except Exception:
return "admin"
# Detect CloudLinux instalation process
def check_CL_installing():
if not os.path.isfile(CL_SETUP_LOCK_FILE):
return False
try:
with open(CL_SETUP_LOCK_FILE, encoding="utf-8") as f:
pid = int(f.read())
return os.path.isdir(f"/proc/{pid}")
except Exception:
return False
# Detect Nagios
def get_nagios():
if not os.path.isdir("/usr/local/nagios"):
return False
global NAGIOS_GID
try:
NAGIOS_GID = pwd.getpwnam("nagios").pw_gid
return True
except Exception:
return False
# Detect if cloudlinux=yes is present for DirectAdmin
def da_check_options():
check_result = get_param_from_file("/usr/local/directadmin/custombuild/options.conf", "cloudlinux", "=")
return check_result == "yes"
def get_symlinksifowner():
"""get fs.enforce_symlinksifowner from sysctl conf"""
sysctl = SysCtlConf(config_file=SYSCTL_CL_CONF_FILE, mute_errors=False)
value = sysctl.get("fs.enforce_symlinksifowner")
return int(value) if value is not None else value
# Get suEXEC status
def get_suEXEC_status():
global SUEXEC_ENABLED
if SUEXEC_ENABLED is None:
detect_suEXEC_suPHP()
return SUEXEC_ENABLED
# Get suPHP status():
def get_suPHP_status():
global SUPHP_ENABLED
if SUPHP_ENABLED is None:
detect_suEXEC_suPHP()
return SUPHP_ENABLED
# Detect suEXEC and suPHP
def detect_suEXEC_suPHP():
global SUEXEC_ENABLED
global SUPHP_ENABLED
# This helps us to avoid double check when we checks both suEXEC and suPHP
SUEXEC_ENABLED = False
SUPHP_ENABLED = False
modules = get_apache_modules()
if modules is None:
return
SUEXEC_ENABLED = "suexec_module" in modules
SUPHP_ENABLED = "suphp_module" in modules
def get_apache_modules():
# path to httpd is the same on the panels
bin_exec = "/usr/sbin/httpd"
try:
with subprocess.Popen(
[bin_exec, "-M"],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
) as proc:
out, _ = proc.communicate()
modules = []
out = out.split("\n")
# clean the output from 1st line 'Loaded modules'
for line in out[1:]:
if not line:
continue
# core_module (static) so_module (static) http_module (static) mpm_worker_module (shared)...
# --> ['core_module', 'so_module', 'http_module', 'mpm_worker_module']
try:
mod = line.strip().split(" ")[0]
except IndexError:
mod = ""
if mod == "":
continue
modules.append(mod)
except OSError:
return None
return modules
def execute(command):
"""
Execute command with bash interpreter
"""
with subprocess.Popen(
command,
shell=True,
executable="/bin/bash",
stdout=subprocess.PIPE,
text=True,
bufsize=-1,
) as proc:
return proc.communicate()[0]
# check suPHP or suEXEC binary for jail
def check_binary_has_jail(location):
try:
if is_ea4():
result = execute("/usr/bin/strings " + str(location[getCPName() + "_ea4"]) + " | grep jail")
else:
result = execute("/usr/bin/strings " + str(location[getCPName()]) + " | grep jail")
return result.find("jail error") != -1
except KeyError:
return None
except OSError:
return False
# Check sshd -T output for usepam yes
def check_SSHd_UsePAM():
try:
result = execute("/usr/sbin/sshd -T | grep usepam")
return result.find("usepam yes") != -1
except OSError:
return None
def init_cp_name():
if CP_NAME is None:
getCPName()
# NOTE: This section of code is deprecated and should not be added to.
# Detect DirectAdmin machine
def is_da():
init_cp_name()
return CP_NAME == "DirectAdmin"
# Detect ISP Manager machine
def is_ispmanager():
init_cp_name()
return CP_NAME == "ISPManager"
# Detect ISP Manager v5 machine type: "Master" or "Node"
# If not ISP5 - always None
def ispmanager5_type():
init_cp_name()
return CP_ISP_TYPE
# Detect ISP Manager v5 machine is Master
def ispmanager5_is_master():
return CP_ISP_TYPE == "Master"
# Detect cPanel machine
def is_cpanel():
init_cp_name()
return CP_NAME == "cPanel"
# Detect Plesk machine
def is_plesk():
init_cp_name()
return CP_NAME == "Plesk"
# Detect InterWorx machine
def is_internetworx():
init_cp_name()
return CP_NAME == "InterWorx"
# Detect H-Sphere machine
def is_hsphere():
init_cp_name()
return CP_NAME == "H-Sphere"
# Detect HostingNG machine
def is_hostingng():
init_cp_name()
return CP_NAME == "HostingNG"
# Detect unknown machine
def is_unknown():
init_cp_name()
return CP_NAME == "Unknown"
def is_openvz():
"""
Return 0 if there is no OpenVZ, otherwise return node ID (envID)
"""
pid = os.getpid()
with open(f"/proc/{pid}/status", encoding="utf-8") as f:
for line in f:
if line.startswith("envID:"):
env_id = line.split(":")[1].strip()
return int(env_id)
return 0 # no openvz found
def is_cagefs_installed():
return os.path.exists("/usr/sbin/cagefsctl")
def get_boolean_param(file_name, param_name, separator="=", default_val=True):
config_val = get_param_from_file(file_name, param_name, separator, default_val=None)
if config_val is None:
return default_val
return config_val.lower() in ("true", "1", "yes", "on")
|