1#@!#!123s

: / proc / thread-self / root / opt / imunify360 / venv / lib / python3.11 / site-packages / im360 / api / server /
Filename : ipecho.py
back
"""IPEchoAPI - returns real IP address of the host (behind NAT)"""
import asyncio
import logging
import urllib

from typing import Optional

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

logger = logging.getLogger(__name__)


TIMEOUT_FOR_IPECHO_REQUEST = 5  # in seconds


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

    URL = "/api/ip"

    @classmethod
    async def get_ip(cls, ip_version: IPVersion) -> Optional[str]:
        """Return resolved IP from echo ip API, v4 supported only,
        for v6 return None"""
        if ip_version != IP.V4:
            raise APIError from NotImplementedError("IPV6 doesn't supported")
        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 _get_ip(cls):
        """Send request to API and process response"""
        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")
        return response.get("ip")