1#@!#!123s

: / opt / imunify360 / venv / lib / python3.11 / site-packages / imav / migration_utils /
Filename : plesk_sdk.py
back
"""
This program is free software: you can redistribute it and/or modify it under
the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License,
or (at your option) any later version.


This program is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. 
See the GNU General Public License for more details.


You should have received a copy of the GNU General Public License
 along with this program.  If not, see <https://www.gnu.org/licenses/>.

Copyright © 2019 Cloud Linux Software Inc.

This software is also available under ImunifyAV commercial license,
see <https://www.imunify360.com/legal/eula>
"""
import json
import re
from subprocess import PIPE, check_output


class PleskSdk:
    """PHP wrapper for Plesk SDK"""

    _cmd = (
        "plesk",
        "php",
        "-r",
    )
    _php = """
        require_once("sdk.php");
        pm_Context::init("revisium-antivirus");
        {};
        """

    @classmethod
    def _join(cls, php: str, strip_whitespaces=True) -> str:
        """
        Append a piece of code to the header and make it a single-line string
        """
        code = cls._php.format(php.strip())
        r = "".join(filter(None, (line.strip() for line in code.splitlines())))
        if strip_whitespaces:
            r = re.sub(r"(\W)\s+|\s+(\W)", r"\1\2", r)
        return r

    @classmethod
    def exec(cls, php, *args):
        cmd = (*cls._cmd, cls._join(php), *args)
        return check_output(cmd, stderr=PIPE, text=True)

    @classmethod
    def context__get_var_dir(cls):
        return cls.exec("echo pm_Context::getVarDir()")

    @classmethod
    def settings__get(cls, key, default):
        return cls.exec(
            "echo pm_Settings::get($argv[1], $argv[2])",
            key,
            json.dumps(default),
        )

    @classmethod
    def domain__get_all_domains(cls):
        php = """
            echo json_encode(
                array_map(
                    function ($domain) {
                        $props = [
                            "id" => "getId",
                            "name" => "getName",
                            "document_root" => "getDocumentRoot",
                            "owner" => "getSysUserLogin"
                        ];
                        $result = (object)[];
                        foreach ($props as $property => $method) {
                            try {
                                $result->{$property} = $domain->{$method}();
                            } catch (Throwable $ignored) {}
                        }
                        return $result;
                    },
                    pm_Domain::getAllDomains()
                )
            );
            """
        domains = json.loads(cls.exec(php))
        # because when there are no domains -> it returns empty LIST instead of dict,
        # but further code relies on DICT
        return domains if domains else {}

    @classmethod
    def domain__has_permission(cls, domain_id, permission):
        """
        For getting permissions status for domain
        e.g: domain__has_permission('1', 'ra_cleanup_allowed')
        """
        php = f"""
            $domain = pm_Domain::getByDomainId(%s);
            echo json_encode($domain->hasPermission("%s"));
        """ % (
            domain_id,
            permission,
        )
        return json.loads(cls.exec(php))