SkullCyberArmy Bypass Shell

Uname : Linux server1.cgrithy.com 3.10.0-1160.95.1.el7.x86_64 #1 SMP Mon Jul 24 13:59:37 UTC 2023 x86_64
Soft : Apache
Ip : 203.161.60.200
Port : 443

File Name : svcctl.py

import asyncio
import logging
import os
import subprocess as su
from typing import Iterable, Union

from defence360agent.contracts.config import Core
from defence360agent.utils import check_run, run

logger = logging.getLogger(__name__)

def _apply_cmd(func):
    async def wrapper(*args, **kwargs):
        cmd = func(*args, **kwargs)
        logger.debug('check_call(%r)', cmd)
        await check_run(cmd)

return wrapper

async def _reset_failed_state(
    services: Iterable[Union["_CentOs6", "_SystemctlBased"]]
):
    for s in services:
        await s.reset_failed()
        await s.restart()
        for _ in range(10):
            if await s.is_active():
                break
            logger.warning(
                "Service %s is still not active, sleep for %s seconds", s, 1
            )
            await asyncio.sleep(1)

class _CentOs6:
    SVC_CTL_BIN = '/sbin/service'
    _CHKCONFIG = '/sbin/chkconfig'

def __init__(self, service_name):
        self._service_name = service_name

@_apply_cmd
    def start(self):
        return [self.SVC_CTL_BIN, self._service_name, 'start']

@_apply_cmd
    def stop(self):
        return [self.SVC_CTL_BIN, self._service_name, 'stop']

@_apply_cmd
    def restart(self):
        return [self.SVC_CTL_BIN, self._service_name, 'restart']

async def reset_failed(self):
        """Not implemented for Centos6"""
        pass

@_apply_cmd
    def enable(self):
        return [self._CHKCONFIG, '--add', self._service_name]

async def is_enabled(self):
        cmd = [self._CHKCONFIG, '--list', self._service_name]
        proc = await asyncio.create_subprocess_exec(
            *cmd, stdout=su.PIPE, stderr=su.DEVNULL)
        out, _ = await proc.communicate()
        rc = await proc.wait()
        return rc == 0 and b':on' in out

def is_enabled_sync(self):
        cmd = [self._CHKCONFIG, '--list', self._service_name]
        cp = su.run(cmd, stdout=su.PIPE, stderr=su.DEVNULL)
        return cp.returncode == 0 and b':on' in cp.stdout

@_apply_cmd
    def disable(self):
        return [self._CHKCONFIG, '--del', self._service_name]

mask = disable
    unmask = enable

async def is_active(self):
        cmd = [self.SVC_CTL_BIN, self._service_name, 'status']
        exit_code, _, _ = await run(cmd)
        return exit_code == 0

async def activate_socket_service(self):
        if await self.is_enabled() and not await self.is_active():
            await _reset_failed_state((self,))

class _SystemctlBased:
    SVC_CTL_BIN = 'systemctl'

def __init__(self, service_name):
        self._service_name = service_name

@_apply_cmd
    def start(self):
        return [self.SVC_CTL_BIN, 'start', self._service_name]

@_apply_cmd
    def stop(self):
        return [self.SVC_CTL_BIN, 'stop', self._service_name]

@_apply_cmd
    def restart(self):
        return [self.SVC_CTL_BIN, 'restart', self._service_name]

@_apply_cmd
    def enable(self):
        return [self.SVC_CTL_BIN, 'enable', self._service_name]

async def is_enabled(self):
        cmd = [self.SVC_CTL_BIN, 'is-enabled', self._service_name]
        proc = await asyncio.create_subprocess_exec(
            *cmd, stdout=su.DEVNULL, stderr=su.DEVNULL)
        await proc.communicate()
        rc = await proc.wait()
        return rc == 0

def is_enabled_sync(self):
        cmd = [self.SVC_CTL_BIN, 'is-enabled', self._service_name]
        rc = su.call(cmd, stdout=su.DEVNULL, stderr=su.DEVNULL)
        return rc == 0

@_apply_cmd
    def disable(self):
        return [self.SVC_CTL_BIN, 'disable', self._service_name]

@_apply_cmd
    def mask(self):
        """
        It was created for imunify360-webshield which required masking as far
        This is no more relevant but let it stay
        is started by 'Wants=' in imunify360.service
        """
        return [self.SVC_CTL_BIN, 'mask', self._service_name]

@_apply_cmd
    def unmask(self):
        """
        It was created for imunify360-webshield which required masking as far
        This is no more relevant but let it stay
        is started by 'Wants=' in imunify360.service
        """
        return [self.SVC_CTL_BIN, 'unmask', self._service_name]

async def is_active(self):
        cmd = [self.SVC_CTL_BIN, 'is-active', self._service_name]
        exit_code, _, _ = await run(cmd)
        return exit_code == 0

@_apply_cmd
    def reset_failed(self):
        return [self.SVC_CTL_BIN, 'reset-failed', self._service_name]

async def activate_socket_service(self):
        agent_service_socket = adaptor(f"{self._service_name}.socket")
        if (
            await agent_service_socket.is_enabled() and
            not await agent_service_socket.is_active()
        ):
            await _reset_failed_state((self, agent_service_socket))

class _CentOs7(_SystemctlBased):
    SVC_CTL_BIN = '/usr/bin/systemctl'

class _DebianUbuntu(_SystemctlBased):
    SVC_CTL_BIN = '/bin/systemctl'

def adaptor(service_name):
    for a in (_DebianUbuntu, _CentOs7, _CentOs6):
        if os.path.exists(a.SVC_CTL_BIN):
            return a(service_name)
    else:
        raise RuntimeError("Cannot instantiate appropriate adaptor.")

def imunify360_service():
    return adaptor(Core.SVC_NAME)

Name	Type	Size	Permission	Last Modified	Actions
.	dir	-	0755	2023-09-13 09:02:44
..	dir	-	0755	2023-09-13 09:02:44
__pycache__	dir	-	0755	2023-09-13 09:02:44
panels	dir	-	0755	2023-09-13 09:02:44
__init__.py	application/x-empty	0 B	0644	2023-08-30 02:51:01
backup_systems.py	text/x-script.python	11.18 KB	0644	2023-08-30 02:51:01
notifier.py	text/plain	1.82 KB	0644	2023-08-30 02:51:01
svcctl.py	text/x-script.python	5.17 KB	0644	2023-08-30 02:51:01
sysctl.py	text/x-script.python	379 B	0644	2023-08-30 02:51:01
systemd_notifier.py	text/plain	1.58 KB	0644	2023-08-30 02:51:01
web_server.py	text/x-script.python	20.51 KB	0644	2023-08-30 02:51:01