Gecko [ dataharbor.africa ]

Name	Size	Permission
?;__pycache__	[ DIR ]	drwxr-xr-x
dashboard_malfunctions.py	1.36 KB	-rw-r--r--
exec_command.py	1.62 KB	-rwxr-xr-x
lsphpchecker.ini	1.55 KB	-rw-r--r--
lsphpchecker.py	11.02 KB	-rwxr-xr-x
lve_diagnostic.py	13.42 KB	-rwxr-xr-x
mod_lsapi_stat.py	14.7 KB	-rw-r--r--
selector_usage_lib.py	12.87 KB	-rw-r--r--
stat_utils.py	9.17 KB	-rw-r--r--
switch_lsapi	27.54 KB	-rwxr-xr-x

Code Editor : lsphpchecker.py

#!/usr/bin/python3

# Copyright (c) Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2018 All Rights Reserved
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

#coding:utf-8
import glob
import os
import pwd
import sys
from subprocess import Popen, PIPE

CHECK = {
    "gt": lambda x, y: x > y,
    "gte": lambda x, y: x >= y,
    "lt": lambda x, y: x < y,
    "lte": lambda x, y: x <= y,
    "eq": lambda x, y: x == y,
}
CHECK_KEYS = {"gt": "more", "gte": "more or equal", "lt": "less",
              "lte": "less or equal", "eq": "equal"}
BYTES_CONVERSION_TABLE = {'M': 1, 'G': 1024, 'T': 1024 * 1024}

quick = 0

def log_error(msg):
    """
    Wrapper for logging errors.
    Simple logging to stderr.
    """
    print(msg, file=sys.stderr)

class Cast:
    """
    Class with functions for cast to any type
    """
    @staticmethod
    def to_bool(_v):
        return {"Off": 0, "On": 1}.get(_v, Cast.to_number(_v))

@staticmethod
    def to_mb(_v):
        if not len(_v):
            return None

if _v[-1].isdigit():
            _v = "%sM" % _v

_num = Cast.to_number(_v[:-1])
        if _num is None:
            return None

return _num * BYTES_CONVERSION_TABLE.get(_v[-1].upper(), 1)

@staticmethod
    def to_number(_v):
        try:
            return int(_v)
        except (ValueError, TypeError):
            return None

class PhpChecker:
    """
    docstring for PhpChecker
    """
    SAMPLE_CONFIG_PATH = "/usr/share/lve/modlscapi/user/lsphpchecker.ini"
    __php_binary = None
    __sample = None

def __init__(self):
        """
        Initialize php versions list
        """
        global quick
        super(PhpChecker, self).__init__()
        php_list = glob.glob("/usr/local/bin/lsphp")
        if quick != 1:
            php_list.extend(glob.glob("/opt/alt/php*/usr/bin/lsphp"))
            php_list.extend(glob.glob("/opt/cpanel/ea-php*/root/usr/bin/lsphp"))
            php_list.extend(glob.glob("/usr/bin/lsphp"))
            php_list = sorted(php_list)
        self.__php_binary = php_list
        
        self._load_sample_options()

def check_user(self, user):
        """
        Check configurations for user
        """
        for php_path in self.__php_binary:
            if os.path.exists(php_path):
                check_result = self._check_php_options(php_path, user)
                if check_result:
                    for message in check_result:
                        print("%s: %s: %s" % (user, php_path, message))

def _check_php_options(self, php_path, user):
        """
        Load and check specified php version options
        @param `php_path` str: path to php binary
        @param `user` str: username
        """
        warnings = []
        options = self._load_php_options(php_path, user)
        modules = self._detect_danger_modules(php_path, user)

warnings += self._check_options(options, "apc")
        warnings += self._check_options(options, "suhosin")
        #if "zend_guard_loader" in modules:
        #    warnings.append("Unstable module Zend Guard detected. Please disable module")

#if "ioncube_loader" in modules:
        #    warnings.append("Unstable module ionCube detected. Please disable module")

return warnings

def _load_php_options(self, php_path, user):
        """
        Load php options from CLI phpinfo output
        """
        #print "%s: %s" % (user, php_path)
        if user == "":
            p = Popen([php_path,  "-i"], stdout=PIPE,
                  stderr=PIPE)
        else:
            p = Popen(["/bin/su", user, "-c", "[ ! -f %s ] || %s -i" % (php_path, php_path)], stdout=PIPE,
                  stderr=PIPE)
        out, err = p.communicate()

#if p.returncode != 0:
        #    log_error(err)

options = {}
        option_value = False
        for line in out.decode().split("\n"):
            if "Directive => Local Value => Master Value" == line:
                option_value = True
                continue

if option_value:
                if not line:
                    option_value = False
                    continue

values = line.split(" => ")
                if len(values) < 3:
                    log_error("Invalid option line - %s" % line)
                    continue

if "." not in values[0]:
                    module = "__common__"
                    key = values[0]
                else:
                    module, key = values[0].split(".", 1)

if module not in options:
                    options[module] = {}

options[module][values[0]] = values[1]

return options

def _detect_danger_modules(self, php_path, user):
        """
        Detect unstable and potential danger php modules from php version output
        """
        if user == "":
            p = Popen([php_path, "-i"], stdout=PIPE,
                  stderr=PIPE)
        else:
            p = Popen(["/bin/su", user, "-c", "[ ! -f %s ] || %s -i" % (php_path, php_path)], stdout=PIPE,
                  stderr=PIPE)
        out, err = p.communicate()

#if p.returncode != 0:
        #    log_error(err)

modules = {}
        for line in out.decode().split("\n"):
            line = line.strip()
            if line.startswith("with the ionCube PHP Loader"):
                modules["ioncube_loader"] = True
            elif line.startswith("with Zend Guard Loader"):
                modules["zend_guard_loader"] = True

return modules

def _check_options(self, config, module):
        """
        Check php options based on sample config
        """
        if not config or not isinstance(config, dict) or \
                not isinstance(module, str) or \
                not isinstance(config.get(module), dict):
            return []

if module not in self.__sample:
            return []

result = []
        options = config[module]
        for key, check_info in self.__sample[module].items():
            if key in options and not self._validate_value(options[key], check_info):
                result.append("%s must be %s %s (current value: %s) (no value means Off)" %
                    (key, CHECK_KEYS.get(check_info["check"], ""),
                        check_info["value"], options[key]))

if len(result):
            result.insert(0, "[%s]" % module)
            result.insert(0, "change %s options to default" % module)

return result

def _load_sample_options(self):
        """
        Load sample options
        """
        self.__sample = {}
        try:
            f = open(self.SAMPLE_CONFIG_PATH, "r")
            for line in f:
                line = line.strip()
                value_type = "number"
                check_type = "gte"
                try:
                    key, value = line.split("=", 1)
                    if "," in key:
                        # get additional information about value type and check method
                        type_info, key = key.split(",")
                        value_type, check_type = type_info.split(":")
                    module = key.split(".", 1)[0]
                except (ValueError, IndexError):
                    print("Invalid sample string %s" % line)
                    continue
    
                if module not in self.__sample:
                    self.__sample[module] = {}
                self.__sample[module][key.strip()] = {"value": value.strip(),
                    "type": value_type, "check": check_type}
            f.close()
        except (OSError, IOError):
            log_error("Error read %s" % self.SAMPLE_CONFIG_PATH)

def _validate_value(self, value, rule):
        """
        Validate option value.
        @param value_type `str`|default:"number" : value type
        @param value1 `str|int`: value1 for compare
        @param value2 `str|int`: value2 for compare
        @return int : -1, 0, 1
        """
        if not isinstance(rule, dict):
            return False

value_type = rule.get("type", "number")
        check_type = rule.get("check", "gte")
        check_value = rule["value"]

cast_func = "to_%s" % value_type
        if hasattr(Cast, cast_func):
            value = getattr(Cast, cast_func)(value)
            check_value = getattr(Cast, cast_func)(check_value)

if check_type not in CHECK:
            return False

return CHECK[check_type](value, check_value)

def load_min_max_uid():
    """
    Load min and max UID from /etc/login.defs config
    """
    min_uid = max_uid = None
    try:
        f = open("/etc/login.defs", "r")
        for line in f:
            if not line.startswith("UID_MIN") and not line.startswith("UID_MAX"):
                continue

data = line.split()
            if not data:
                continue

try:
                if data[0] == "UID_MIN":
                    min_uid = int(data[1])
                elif data[0] == "UID_MAX":
                    max_uid = int(data[1])
            except (ValueError, IndexError):
                log_error("Invalid UID_MIN/UID_MAX values")
                break

f.close()
    except (IOError, OSError):
        print("Can`t read UID_MIN and UID_MAX from /etc/login.defs file", file=sys.stderr)

return min_uid, max_uid

def main(users_list):
    """
    Run check
    """
    global quick
    
    if len(users_list)>0 and users_list[0] == "help":
        print("%s [help] [quick] [users list...]" % sys.argv[0])
        print("   help       - show this help")
        print("   fast       - check all lsphp without switching to user")
        print("   medium     - check all users but only /usr/local/bin/lsphp config")
        print("   users list - list of needed users or empty. i this case users list will be taken from passwd")
        return
    if len(users_list)>0 and (users_list[0] == "fast" or users_list[0] == "medium"):
        if users_list[0] == "fast":
            quick = 2
        else:
            quick = 1
        users_list = users_list[1:]
    checker = PhpChecker()
    if quick == 2:
        checker.check_user("")
    elif users_list:
        for username in users_list:
            try:
                user = pwd.getpwnam(username)
                checker.check_user(username)
            except KeyError:
                log_error("User %s doesn`t exists" % username)
    else:
        UID_MIN, UID_MAX = load_min_max_uid()
        # get username list
        for user in pwd.getpwall():
            if  user.pw_uid >= UID_MIN and user.pw_uid <= UID_MAX:
                # for each user run check_user
                checker.check_user(user.pw_name)

if "__main__" == __name__:
    main(sys.argv[1:])

${this.title}

Code Editor : lsphpchecker.py