[ PARENT ]

EDIT :: packages.py

import logging import datetime from typing import List from defence360agent.utils import timed_cache from defence360agent.subsys.panels.cpanel.whm import WHMAPIException, whmapi1 logger = logging.getLogger(__name__) class PackageNotExistError(WHMAPIException): pass class PkgInfo(dict): def extensions(self) -> List[str]: return self.get("_PACKAGE_EXTENSIONS", "").split() def has_extension(self, name: str) -> bool: return name in self.extensions() def name(self) -> str: return self["name"] @timed_cache(datetime.timedelta(seconds=90), maxsize=100) async def get_package_info(name: str) -> PkgInfo: try: data = await whmapi1("getpkginfo", pkg=name) except WHMAPIException as e: if "No such file or directory" in str(e): raise PackageNotExistError(e) else: raise info = PkgInfo(data["pkg"]) info["name"] = name return info async def list_packages(want="all") -> List[PkgInfo]: data = await whmapi1("listpkgs", want=want) return [PkgInfo(item) for item in data["pkg"]] async def remove_extension(extension_name: str, package_info: PkgInfo) -> None: """Removes extension from a package described by package_info.""" name = package_info.name() if package_info.has_extension(extension_name): await whmapi1( "delpkgext", name=name, _DELETE_EXTENSIONS=extension_name ) logger.info( "Extension %s disabled for package %s", extension_name, name ) return logger.info( "Extension %s was already disabled for package %s", extension_name, name, ) async def add_extension( extension_name: str, package_info: PkgInfo, **kwargs ) -> None: """Adds extension to a package described by package_info. kwargs holds extra variables to set for the extension.""" name = package_info.name() if not package_info.has_extension(extension_name): await whmapi1( "addpkgext", name=name, _PACKAGE_EXTENSIONS=extension_name, **kwargs ) logger.info( "Extension %s enabled for package %s", extension_name, name ) return logger.info( "Extension %s was already enabled for package %s", extension_name, name ) async def add_extension_for_all(extension_name: str, **kwargs) -> None: """Add given extension to all cPanel packages.""" for pkg in await list_packages(): try: await add_extension(extension_name, pkg, **kwargs) except WHMAPIException: logger.exception( "Unable to add extension %s to package %s", extension_name, pkg["name"], ) async def remove_extension_from_all(extension_name: str) -> None: """Remove given extension from all cPanel packages.""" for pkg in await list_packages(): try: await remove_extension(extension_name, pkg) except WHMAPIException: logger.exception( "Unable to remove extension %s from package %s", extension_name, pkg["name"], ) logger.info( "Imunify360 package extensions have been removed from all packages." )

SAVE CANCEL