| Server IP : 157.90.209.209 / Your IP : 216.73.216.17 [ Web Server : Apache System : Linux hcomm124.dns-wk.info 4.18.0-553.64.1.el8_10.x86_64 #1 SMP Mon Jul 28 12:01:56 EDT 2025 x86_64 User : evidenciarevista ( 1049) PHP Version : 7.2.34 Disable Function : exec,passthru,shell_exec,system Domains : 216 Domains MySQL : OFF | cURL : ON | WGET : ON | Perl : ON | Python : OFF | Sudo : ON | Pkexec : ON Directory : /opt/imunify360/venv/lib/python3.11/site-packages/imav/plugins/ |
Upload File : |
"""
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 logging
from typing import Any
from defence360agent.contracts.messages import MessageType, ServerConfig
from defence360agent.contracts.plugins import (
MessageSink,
expect,
)
from defence360agent.utils import Scope
from imav.contracts.imunify_patch_id import (
get_imunify_patch_id,
get_imunify_patch_purchase_url,
)
from defence360agent.subsys.panels.hosting_panel import HostingPanel
UserData = dict[str, Any]
logger = logging.getLogger(__name__)
class EnrichServerConfigWithImunifyPatchId(MessageSink):
SCOPE = Scope.AV_IM360
PROCESSING_ORDER = MessageSink.ProcessingOrder.PRE_PROCESS_MESSAGE
async def create_sink(self, loop):
pass
@expect(MessageType.ServerConfig)
async def enrich_server_config(
self, message: ServerConfig
) -> ServerConfig:
if "users" in message:
users = message["users"]
else:
users = [
{"username": user} for user in await HostingPanel().get_users()
]
message["users"] = [
await self._enrich_user_data(user) for user in users
]
return message
async def _enrich_user_data(self, user: UserData) -> UserData:
if "username" not in user:
logger.warning(f"User data missing 'username' field: {user=}")
return user
user["imunify_patch_user_id"] = await get_imunify_patch_id(
user["username"]
)
user[
"imunify_patch_purchase_url"
] = await get_imunify_patch_purchase_url(user["username"])
return user