晋太元中，武陵人捕鱼为业。缘溪行，忘路之远近。忽逢桃花林，夹岸数百步，中无杂树，芳草鲜美，落英缤纷。渔人甚异之，复前行，欲穷其林。 　　林尽水源，便得一山，山有小口，仿佛若有光。便舍船，从口入。初极狭，才通人。复行数十步，豁然开朗。土地平旷，屋舍俨然，有良田、美池、桑竹之属。阡陌交通，鸡犬相闻。其中往来种作，男女衣着，悉如外人。黄发垂髫，并怡然自乐。 　　见渔人，乃大惊，问所从来。具答之。便要还家，设酒杀鸡作食。村中闻有此人，咸来问讯。自云先世避秦时乱，率妻子邑人来此绝境，不复出焉，遂与外人间隔。问今是何世，乃不知有汉，无论魏晋。此人一一为具言所闻，皆叹惋。余人各复延至其家，皆出酒食。停数日，辞去。此中人语云：“不足为外人道也。”(间隔 一作：隔绝) 　　既出，得其船，便扶向路，处处志之。及郡下，诣太守，说如此。太守即遣人随其往，寻向所志，遂迷，不复得路。 　　南阳刘子骥，高尚士也，闻之，欣然规往。未果，寻病终。后遂无问津者。 .
Prv8 Shell
Server : Apache
System : Linux srv.rainic.com 4.18.0-553.47.1.el8_10.x86_64 #1 SMP Wed Apr 2 05:45:37 EDT 2025 x86_64
User : rainic ( 1014)
PHP Version : 7.4.33
Disable Function : exec,passthru,shell_exec,system
Directory :  /opt/imunify360/venv/lib/python3.11/site-packages/imav/plugins/

Upload File :
current_dir [ Writeable ] document_root [ Writeable ]

 

Current File : //opt/imunify360/venv/lib/python3.11/site-packages/imav/plugins/generic_sensor.py
"""
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>


Generic Sensor plugin

- Creates listening unix domain socket on config.GenericSensor.SOCKET_PATH
- Expects alert data formatted as

"""
import asyncio
import base64
import json
import os
import time
from logging import getLogger

from imav.contracts.messages import MSGS_WITHOUT_IP
from defence360agent.contracts.config import (
    SimpleRpc,
    GENERIC_SENSOR_SOCKET_PATH,
)
from defence360agent.contracts.messages import MessageType
from defence360agent.contracts.plugins import Sensor
from defence360agent.internals.global_scope import g
from defence360agent.internals.logger import getNetworkLogger
from defence360agent.simple_rpc import RpcServerAV
from defence360agent.utils import Scope
from defence360agent.utils.buffer import LineBuffer

logger, network_logger = getLogger(__name__), getNetworkLogger(__name__)


class Protocol(asyncio.Protocol):
    METHOD2MSGTYPE = {
        "NOOP": MessageType.Noop,
        "MALWARE_SCAN": MessageType.MalwareScan,
        "MALWARE_SCAN_TASK": MessageType.MalwareScanTask,
        "MALWARE_SCAN_COMPLETE": MessageType.MalwareScanComplete,
        "MALWARE_CLEAN_COMPLETE": MessageType.MalwareCleanComplete,
        "MALWARE_RESTORE_COMPLETE": MessageType.MalwareRestoreComplete,
    }

    def __init__(self, loop, sink, *_):
        self._loop = loop
        self._sink = sink
        self._line_buffer = LineBuffer()

    def connection_made(self, transport):
        network_logger.debug("Connection made")

    def data_received(self, data):
        msgs = data.decode()
        if not msgs.strip():
            logger.error("Empty message received <%s>", msgs)
            return

        self._line_buffer.append(msgs)
        for msg in self._line_buffer:
            if msg:
                network_logger.debug("data_received: {!r}".format(msg))
                tokens = self._parse_msg(msg)
                if tokens:
                    tokens["timestamp"] = time.time()
                    self._process_msg(tokens)

    def _parse_msg(self, msg):
        try:
            return json.loads(msg)
        except json.JSONDecodeError:
            logger.exception("data_received(%s): unable to decode", repr(msg))

    def _process_msg(self, tokens):
        # map 'method' to appropriate Message type
        try:
            msgtype = self.METHOD2MSGTYPE[tokens["method"]]
        except KeyError as e:
            logger.error(
                "data_received(%s): Wrong or missing 'method' [%s]",
                repr(tokens),
                repr(e),
            )
            return

        if tokens["method"] == "MALWARE_SCAN_TASK":
            tokens["filelist"] = [
                os.fsdecode(base64.b64decode(f)) for f in tokens["filelist"]
            ]
        elif tokens["method"] == "SYNCLIST":  # added to avoid using pyrasite
            logger.info("Received test SynclistResponse")
        elif (
            not tokens.get("attackers_ip")
            and tokens["method"] not in MSGS_WITHOUT_IP
        ):
            logger.error(
                "Method type is %s but empty or no <attackers_ip> "
                "in message <%s>",
                tokens.get("method"),
                tokens,
            )
            return

        self._loop.create_task(self._sink.process_message(msgtype(tokens)))

    def connection_lost(self, transport):
        network_logger.debug("Disconnected")


class GenericSensor(Sensor):
    SOCKET_PATH = GENERIC_SENSOR_SOCKET_PATH
    PROTOCOL_CLASS = Protocol
    SCOPE = Scope.AV

    async def create_sensor(self, loop, sink):
        if SimpleRpc.SOCKET_ACTIVATION:

            class GenericSensorSocket(RpcServerAV):
                SOCKET_PATH = self.SOCKET_PATH
                PROTOCOL_CLASS = self.PROTOCOL_CLASS

            g.sensor_server = await GenericSensorSocket.create(loop, sink)
            return g.sensor_server
        else:
            # FIXME make sure root can write to
            os.makedirs(os.path.dirname(self.SOCKET_PATH), exist_ok=True)
            if os.path.exists(self.SOCKET_PATH):
                os.unlink(self.SOCKET_PATH)

            g.sensor_server = await loop.create_unix_server(
                lambda: self.PROTOCOL_CLASS(loop, sink), self.SOCKET_PATH
            )
            return g.sensor_server

haha - 2025