File Manager

X7ROOT File Manager

Current Path: /opt/imunify360/venv/lib/python3.11/site-packages/defence360agent/utils

opt / imunify360 / venv / lib / python3.11 / site-packages / defence360agent / utils /

?? ..
?? __init__.py(55.96 KB)
?? __pycache__
?? _shutil.py(795 B)
?? antivirus_mode.py(497 B)
?? async_utils.py(718 B)
?? benchmark.py(538 B)
?? buffer.py(1.24 KB)
?? check_db.py(7.72 KB)
?? check_lock.py(636 B)
?? cli.py(7.39 KB)
?? common.py(14.41 KB)
?? config.py(999 B)
?? cronjob.py(902 B)
?? doctor.py(1 KB)
?? hyperscan.py(149 B)
?? importer.py(2.67 KB)
?? ipecho.py(3.17 KB)
?? json.py(953 B)
?? kwconfig.py(1.56 KB)
?? parsers.py(11.12 KB)
?? resource_limits.py(2.29 KB)
?? safe_fileops.py(7.99 KB)
?? safe_sequence.py(363 B)
?? serialization.py(1.72 KB)
?? sshutil.py(7.94 KB)
?? subprocess.py(1.53 KB)
?? support.py(5.2 KB)
?? threads.py(1005 B)
?? validate.py(4.27 KB)
?? whmcs.py(7.6 KB)
?? wordpress_mu_plugin.py(1.41 KB)

Editing: ipecho.py

"""IPEchoAPI - returns real IP address of the host (behind NAT)"""
import asyncio
import functools
import logging
import time
import urllib
from pathlib import Path
from typing import Optional

from async_lru import alru_cache

from defence360agent.api.server import API, APIError
from defence360agent.utils import atomic_rewrite
from defence360agent.utils.validate import IP, IPVersion

logger = logging.getLogger(__name__)

TIMEOUT_FOR_IPECHO_REQUEST = 5  # in seconds
CACHE_TTL_SECONDS = 3 * 60 * 60
CACHE_FILE_PATH = Path("/var/imunify360") / "ipecho_cache"

class IPEchoAPI(API):
    """Make requests to the API for obtain own IP address"""

URL = "/api/ip"

@classmethod
    @alru_cache(maxsize=3)
    async def get_ip(cls, ip_version: IPVersion = None) -> Optional[str]:
        """Return cached result for resolved IP from echo ip API"""

return await cls.ip_for_version(ip_version)

@classmethod
    @functools.lru_cache(maxsize=1)
    def server_ip(cls):
        """Return cached result for resolved IP from echo ip API"""
        try:
            return cls._get_ip()
        except Exception as e:
            raise APIError from e

@classmethod
    async def ip_for_version(
        cls, ip_version: IPVersion = None
    ) -> Optional[str]:
        """Return resolved IP from echo ip API"""

loop = asyncio.get_event_loop()
        try:
            ip = await asyncio.wait_for(
                loop.run_in_executor(None, cls._get_ip),
                timeout=TIMEOUT_FOR_IPECHO_REQUEST,
            )
            if IP.type_of(ip) != ip_version:
                raise ValueError("Wrong ip type")
            return ip
        except (asyncio.TimeoutError, ValueError) as e:
            raise APIError from e

@classmethod
    def _load_cache(cls) -> Optional[str]:
        try:
            if not CACHE_FILE_PATH.exists():
                return None

mtime = CACHE_FILE_PATH.stat().st_mtime
            cache_age = time.time() - mtime

if cache_age < 0:
                return None

if cache_age < CACHE_TTL_SECONDS:
                ip = CACHE_FILE_PATH.read_text().strip()
                return ip
            else:
                return None
        except Exception as e:
            logger.error("IPEchoAPI cache read error: %s", e)
            return None

@classmethod
    def _save_cache(cls, ip: str) -> None:
        try:
            atomic_rewrite(
                CACHE_FILE_PATH,
                ip,
                backup=False,
                permissions=0o644,
            )
        except Exception as e:
            logger.error("IPEchoAPI cache write error: %s", e)

@classmethod
    def _get_ip(cls):
        """Get IP from file-based cache or send request to API and process response."""
        cached_ip = cls._load_cache()
        if cached_ip is not None:
            return cached_ip

request = urllib.request.Request(cls._BASE_URL + cls.URL)
        response = cls.request(request)
        if response.get("status") != "ok":
            # time inside sync executor
            raise APIError("Unexpected API error")
        ip = response.get("ip")
        if ip:
            cls._save_cache(ip)

return ip

X7ROOT File Manager

Editing: ipecho.py

Upload File

Create Folder