1#@!#!123s

: / opt / imunify360 / venv / lib / python3.11 / site-packages / clcommon /
Filename : clconfpars.py
back
# -*- coding: utf-8 -*-

# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2018 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENSE.TXT
#

import configparser
import locale
import os
import re
import syslog
from collections import namedtuple


class WebConfigParsingError(Exception):
    def __init__(self, message):
        self.message = message


class WebConfigMissing(Exception):
    def __init__(self, message):
        self.message = message


SECHEAD = 'asection'


def load(path, case_sensitive=False, ignore_bad_encoding=False):
    config = configparser.ConfigParser(allow_no_value=True,
                                       interpolation=None,
                                       strict=False)
    if case_sensitive:
        config.optionxform = str
    if ignore_bad_encoding:
        with open(path, 'rb') as f:
            raw = f.read().decode(locale.getpreferredencoding(), 'replace')
    else:
        with open(path, 'r', encoding='utf-8') as f:
            raw = f.read()
    config.read_string(f'[{SECHEAD}]\n' + raw, source=path)
    return dict(config.items(section=SECHEAD))


_QUOTES = "'", '"'

def _strip_escape_quotes_of_config_value(val: str) -> str:
    """
    Strips single or double quote char only if the quote present from both sides.
    """
    if val.startswith(_QUOTES) and val.endswith(_QUOTES):
        return val[1:-1]
    return val


def load_fast(path, delimiter="=", strip_quotes=False):
    data = {}
    with open(path, "r", encoding="utf-8", errors="surrogateescape") as f:
        for line in f.readlines():
            parts = line.split(delimiter, 1)
            try:
                key, value = parts
            except ValueError:
                # Skip broken lines
                continue

            value = value.strip()
            value = (
                _strip_escape_quotes_of_config_value(value)
                if strip_quotes
                else value
            )
            data[key.strip()] = value
    return data


cache = {}

def load_once(path, ignore_errors=False):
    """
    Read ini file once (cached) and return its content as dict
    """
    try:
        res = cache[path]
    except KeyError:
        try:
            res =  cache[path] = load(path)
        except (IOError, configparser.Error):
            if not ignore_errors:
                raise
            res = cache[path] = {}
    return res


def change_settings(settings_dict, path, tmp_path=None):
    if not tmp_path:
        tmp_path = path + ".tmp"

    used_keys = []

    with (open(path, 'r', encoding='utf-8') as fin,
          open(tmp_path, 'w', encoding='utf-8') as fout):
        for line in fin:
            stripped_line = line.strip()
            if stripped_line and not stripped_line.startswith('#'):
                key, _ = stripped_line.split('=', 1)
                key = key.strip()
                if key in settings_dict:
                    fout.write(f'{key}={settings_dict[key]}\n')
                    used_keys.append(key)
                    continue
            fout.write(line)

    with open(tmp_path, 'a', encoding='utf-8') as fout:
        for key in settings_dict:
            if key not in used_keys:
                fout.write(f'{key}={settings_dict[key]}\n')

    os.rename(tmp_path, path)


_NGINX_TOKENS_RE = re.compile(
    r"""
    (
      # Comments
      (:? \# .* $ )

      # Single-, double-quoted strings and bare strings without whitespaces
      | (:? "[^"\n]*?" )
      | (:? '[^'\n]*?' )
      | (:? [^"';\s\{\}]+ )

      # Structural characters
      | ;
      | \{
      | \}
      | \n
    )
    """,
    re.IGNORECASE | re.MULTILINE | re.VERBOSE,
)


def _ngx_tokenize(data):
    tokens = (
        match.group(0)
        for match in _NGINX_TOKENS_RE.finditer(data)
        if match and match.group(0)
    )
    # Explicitly ignore comments
    return (tok for tok in tokens if not tok.startswith('#'))

def _ngx_take_until(it, val):
    for tok in it:
        if tok in val:
            return

        yield tok

def _ngx_take_until_block_end(it):
    lvl = 1
    for t in it:
        if t == "{":
            lvl += 1
        elif t == "}":
            lvl -= 1
        if lvl < 1:
            return
        yield t

def _ngx_scan_block_info(block_tokens, need_fields):
    """Scan a block for required fields, skips nested blocks"""
    info = {}
    for tok in block_tokens:
        # We need to skip until the end of inner block if it occurs
        if tok == "{":
            for _ in _ngx_take_until_block_end(block_tokens):
                pass
        # Now gather the value, the last occurrence is in priority
        if tok in need_fields:
            value_tokens = _ngx_take_until(block_tokens, ";\n")
            info[tok] = list(value_tokens)

    return info


def nginx_conf_loose_parser(data):
    """
    Parse content of NGINX configuration in a manner tolerant to minor mistakes
    and extract relevant fields from all `server` directives.

    Relevant fields are:
    - `server_name`
    - `root` - returned as `document_root`
    - `ssl` - if `listen` field contains "ssl" word

    Doesn't handle interpolated values (ex. `${val}`) outside of quoted strings
    """
    tokens = _ngx_tokenize(data)
    for tok in tokens:
        if tok != "server":
            continue

        # Nothing seems to be allowed between "server" directive and
        #  the opening of his block, so we just discard everything
        #  until first block opening seen
        for _ in _ngx_take_until(tokens, "{"):
            pass

        # Limit further scan by the inside of block
        block_tokens = _ngx_take_until_block_end(tokens)

        # By using only `block_tokens` we ensure all blocks are properly delimited
        info = _ngx_scan_block_info(block_tokens, ("server_name", "root", "listen"))
        try:
            server_name = info["server_name"]
            root = info["root"]
        except KeyError:
            continue
        if not server_name and not root:
            continue

        yield {
            "server_name": _strip_escape_quotes_of_config_value(server_name[0]),
            "document_root": _strip_escape_quotes_of_config_value(root[0]),
            "ssl": "ssl" in info.get("listen", []),
        }


def nginx_conf_parser(conf_file):
    """Parse NGINX config file, see `nginx_conf_loose_parser` for more details"""
    if not os.path.isfile(conf_file):
        raise WebConfigMissing(f'File does not exists {conf_file}')

    dirty_data = read_unicode_file_with_decode_fallback(conf_file)

    return list(nginx_conf_loose_parser(dirty_data))


def apache_conf_parser(conf_file):
    if not os.path.isfile(conf_file):
        raise WebConfigMissing(f'File does not exists {conf_file}')

    conf_data = []

    data_all = read_unicode_file_with_decode_fallback(conf_file).splitlines()

    data = [i for i in data_all if re.search('^((?!#).)*$', i)]

    ID = 0
    enable = False

    result = {}
    vhost = []
    while len(data) > 0:
        out = data.pop(0)
        if "<VirtualHost" in out:
            ip_port = out.split()[1]
            port = '0'
            try:
                ip, port = ip_port.split(':')
                port = port.replace('>', '')
            except ValueError:
                ip = ip_port
            vhost.append(ip)
            vhost.append(port)
            enable = True
            continue

        if "</VirtualHost>" in out:
            result[ID] = vhost
            ID+=1
            enable = False
            vhost = []
            continue

        if enable:
            vhost.append(out)
            continue

    for value in result.values():
        # result[i][0] is an IP
        # result[i][1] is a port
        data = {
                'user' : None,
                'server_name' : '',
                'document_root' : '',
                'server_alias' : None,
                'port' : int(value[1]),
                'ssl' : False,
               }
        for line in value:
            if "ServerName" in line:
                data['server_name'] = line.split()[1].strip().replace('www.', '')
                continue
            if "DocumentRoot" in line:
                # remove all whitespaces (first strip) and also quotes (second one)
                data['document_root'] = line.split()[1].strip().strip('"')
                continue
            if "ServerAlias" in line:
                data['server_alias'] = ','.join(str(n) for n in line.split()[1:])
                continue
            if "SuexecUserGroup" in line:
                data['user'] = line.split()[1].strip()
            if "SSLEngine" in line:
                data['ssl'] = line.split()[1].strip().lower() == 'on'

        conf_data.append(data)

    return conf_data


PamLVECfg = namedtuple('PamLVECfg', ['min_uid', 'cagefs_enabled', 'groups'])

def parse_pam_lve_config(configfile):
    """
    Parse string like:
    "session      required      pam_lve.so      500      1     group1,group2"
    :param configfile: path to config file to parse
    :type configfile: str
    :return: PamLVECfg instance when pam_lve configuratiom is found, None otherwise
    :rtype: namedtuple
    :raises: IOError, ValueError
    """
    with open(configfile, 'r', encoding='utf-8') as f:
        for line in f:
            if line.startswith('#'):
                continue
            s = line.split()
            if len(s) >= 3 and s[2] == 'pam_lve.so':
                # parse config string taking pam_lve defaults into account
                min_uid = int(s[3]) if len(s) >= 4 else 500
                cagefs_enabled = bool(int(s[4])) if len(s) >= 5 else False
                groups = s[5].split(',') if len(s) >= 6 else ['wheel']
                return PamLVECfg(min_uid, cagefs_enabled, groups)
    # pam_lve line is not found in config file
    return None


def read_unicode_file_with_decode_fallback(file_path: str) -> str:
    with open(file_path, 'rb') as f:
        raw_data = f.read()
    try:
        return raw_data.decode()
    except UnicodeDecodeError:
        syslog.syslog(
            syslog.LOG_WARNING,
            f'Failed to decode "{file_path}" content as utf-8 - loading with placeholders for invalid unicode sequences'
        )
        return raw_data.decode(errors='replace')