MAINHACK

Viewing / Editing File: iaid.py

File is not writable. Editing disabled.
import asyncio
import grp
import json
import os
import random
import time
from contextlib import suppress
from dataclasses import dataclass
from logging import getLogger
from pathlib import Path
from urllib.parse import urljoin
from urllib.request import Request

from defence360agent.api.server import API, APIError
from defence360agent.contracts.license import LicenseCLN
from defence360agent.utils import atomic_rewrite
from defence360agent.utils.common import DAY
from defence360agent.internals.global_scope import g

logger = getLogger(__name__)


_MAX_TRIES = 10
_TIMEOUT_MULTIPLICATOR = 0.5
"""
>>> _MAX_TRIES_FOR_DOWNLOAD = 10
>>> _TIMEOUT_MULTIPLICATOR = 0.5
>>> [(1 << i) * _TIMEOUT_MULTIPLICATOR for i in range(1, _MAX_TRIES_FOR_DOWNLOAD)]  # noqa
[1.0, 2.0, 4.0, 8.0, 16.0, 32.0, 64.0, 128.0, 256.0]
"""
_ACTIVATE_MINIMUM_TIMEOUT = 60


class IAIDTokenError(RuntimeError):
    """Can't get iaid token for any reason."""


class IndependentAgentIDAPI(API):
    API_PATH = "/api/auth/agent/{}"
    REGISTER_URL = urljoin(API._BASE_URL, API_PATH.format("register"))
    ACTIVATE_URL = urljoin(API._BASE_URL, API_PATH.format("activate"))
    LOGIN_URL = urljoin(API._BASE_URL, API_PATH.format("login"))
    TOKEN_INFO = urljoin(API._BASE_URL, API_PATH.format("token-info"))

    IAID_DIR = Path("/var/imunify360")
    IAID_FILE = IAID_DIR / "iaid"
    IAID_PASSWORD_FILE = IAID_DIR / "iaid-password"
    IAID_TOKEN_FILE = IAID_DIR / "iaid-token"
    IAID_ACTIVATED_FILE = IAID_DIR / "iaid-activated"
    _tasks = {
        "register": [],
        "activate": [],
        "login": [],
    }

    @dataclass(frozen=True)
    class TokenInfo:
        __slots__ = [
            "valid",
            "iaid",
            "license_status",
            "server_id",
            "need_renew",
        ]
        valid: bool
        iaid: str
        license_status: str  # "ok", "ok-av", "ok-avp", "ok-trial"
        server_id: str
        need_renew: bool

    @staticmethod
    async def _retry_on_error(coro, *args, attempt, timeout=0):
        # Exponential backoff retry
        await asyncio.sleep(
            timeout + random.randrange(1 << attempt) * _TIMEOUT_MULTIPLICATOR
        )
        await coro(*args)

    @classmethod
    def _add_task(cls, type, coro, *args, attempt, timeout=0):
        cls._tasks[type] = [
            task for task in cls._tasks[type] if not task.done()
        ]
        if len(cls._tasks[type]) <= 1:
            loop = asyncio.get_event_loop()
            cls._tasks[type].append(
                loop.create_task(
                    cls._retry_on_error(
                        coro, *args, attempt=attempt, timeout=timeout
                    )
                )
            )
        else:
            logger.info("Task %s already in retry queue", type)

    @classmethod
    async def shutdown(cls):
        for type, tasks in cls._tasks.items():
            for task in tasks:
                if not task.done():
                    task.cancel()
                    with suppress(asyncio.CancelledError):
                        await task
                    logger.info("Retry task %s was canceled.", type)

    @staticmethod
    def _gid():
        return grp.getgrnam("_imunify").gr_gid

    @classmethod
    def get_iaid(cls):
        if cls.IAID_FILE.exists():
            return cls.IAID_FILE.read_text()
        return None

    @staticmethod
    def _request(url, headers=None, method="POST", **kwargs):
        _headers = {"Content-Type": "application/json"}
        if headers is not None:
            _headers.update(headers)
        return Request(
            url,
            method=method,
            headers=_headers,
            data=json.dumps(kwargs).encode() if kwargs else None,
        )

    @classmethod
    def is_registered(cls):
        return all(
            iaid_file.exists()
            for iaid_file in (cls.IAID_FILE, cls.IAID_PASSWORD_FILE)
        )

    @classmethod
    async def get_token(cls):
        """Ensure that iaid token is up to date
        Return iaid token or raise IAIDTokenError."""
        if IndependentAgentIDAPI.is_token_expired():
            await IndependentAgentIDAPI.login()
        try:
            token = cls.IAID_TOKEN_FILE.read_text(encoding="ascii")
            if not token:
                raise IAIDTokenError(f"IAID_TOKEN_FILE is empty")
            return token
        except Exception as e:
            raise IAIDTokenError(f"Can't get iaid token, reason: {e}") from e

    @classmethod
    async def _get_token_info(cls) -> TokenInfo:
        iaid_token = await cls.get_token()
        headers = {"X-Auth": iaid_token}
        request = cls._request(cls.TOKEN_INFO, headers=headers, method="GET")
        result = await cls.async_request(request)
        token = result.get("token_info")
        if token is None:
            raise APIError("wrong response %r", result)
        return cls.TokenInfo(**token)

    @classmethod
    def is_token_expired(cls):
        try:
            stat = os.stat(cls.IAID_TOKEN_FILE)
        except FileNotFoundError:
            st_mtime = 0.0
        else:
            st_mtime = stat.st_mtime
        return time.time() - st_mtime > DAY

    @classmethod
    async def register(cls, force=False, attempt=1):
        if not force and cls.is_registered():
            return

        payload = dict()
        server_id = LicenseCLN.get_server_id()
        if server_id:
            payload["server_id"] = server_id

        request = cls._request(cls.REGISTER_URL, **payload)
        try:
            result = await cls.async_request(request)
            cls.IAID_ACTIVATED_FILE.unlink(missing_ok=True)
        except APIError as e:
            logger.warning(
                "Something went wrong on register %r - attempt %s", e, attempt
            )
            if (
                e.status_code is None
                or e.status_code >= 500
                or e.status_code == 402
            ) and attempt < _MAX_TRIES:
                # internal error we may try again
                cls._add_task(
                    "register",
                    cls.register,
                    force,
                    attempt + 1,
                    attempt=attempt,
                )
            else:
                logger.error(
                    "Failed to register (%s) after %s attempts: %r",
                    request.full_url,
                    attempt,
                    e,
                )
            return
        else:
            atomic_rewrite(
                str(cls.IAID_FILE),
                result["iaid"],
                backup=cls.IAID_FILE.exists(),
                uid=-1,
                gid=cls._gid(),
                permissions=0o640,
            )
            atomic_rewrite(
                str(cls.IAID_PASSWORD_FILE),
                result["password"],
                backup=cls.IAID_PASSWORD_FILE.exists(),
                permissions=0o600,
            )
            await cls.activate()

    @classmethod
    async def ensure_is_activated_and_valid(cls):
        """Check whether the agent activated"""

        if not cls.IAID_ACTIVATED_FILE.exists():
            await cls.activate()
            return
        lic = LicenseCLN.get_token()
        token = await cls._get_token_info()
        if token.license_status != lic.get(
            "status"
        ) or token.server_id != lic.get("id"):
            logger.error("Got a corrupted token: %r", token)
            await cls.reactivate()
            return
        iaid = cls.IAID_FILE.read_text()
        if not token.valid or token.iaid != iaid or token.need_renew:
            await cls.login()

    @classmethod
    async def activate(cls, attempt=1):
        if cls.IAID_ACTIVATED_FILE.exists():
            g["iaid"] = cls.get_iaid()
            return
        if not cls.is_registered():
            logger.warning("need to register first before activate")
            await cls.register()
            return
        if LicenseCLN.is_free():
            g["iaid"] = cls.get_iaid()
            if cls.is_token_expired():
                await cls.login()
            return
        lic = LicenseCLN.get_token()
        if not lic:
            logger.warning(
                "Can't continue iaid activation: no valid license is found"
            )
            return
        iaid = cls.IAID_FILE.read_text()
        password = cls.IAID_PASSWORD_FILE.read_text()
        request = cls._request(
            cls.ACTIVATE_URL, iaid=iaid, password=password, license=lic
        )
        g["iaid"] = iaid
        try:
            await cls.async_request(request)
            cls.IAID_ACTIVATED_FILE.touch()
        except APIError as e:
            logger.warning(
                "Something went wrong on activate %r attempt %s", e, attempt
            )
            if e.status_code and e.status_code == 401:
                await cls.register(force=True)
            elif (
                e.status_code
                and (e.status_code >= 500 or e.status_code == 402)
                and attempt < _MAX_TRIES
            ):
                # internal error we may try again
                # 402 - if it is fresh registration it may take
                # time to sync CLN db
                cls._add_task(
                    "activate",
                    cls.activate,
                    attempt + 1,
                    attempt=attempt,
                    timeout=_ACTIVATE_MINIMUM_TIMEOUT,
                )
            else:
                logger.error(
                    "Failed to activate (%s) after %s attempts: %r",
                    request.full_url,
                    attempt,
                    e,
                )
        else:
            cls.IAID_TOKEN_FILE.unlink(missing_ok=True)
            await cls.login()

    @classmethod
    async def reactivate(cls):
        cls.IAID_ACTIVATED_FILE.unlink(missing_ok=True)
        await cls.activate()

    @classmethod
    async def login(cls, attempt=1):
        if not cls.is_registered():
            logger.error("need to register first before login")
            return
        iaid = cls.IAID_FILE.read_text()
        password = cls.IAID_PASSWORD_FILE.read_text()

        request = cls._request(cls.LOGIN_URL, iaid=iaid, password=password)
        try:
            result = await cls.async_request(request)
        except APIError as e:
            logger.warning(
                "Something wrong happened on login %r attempt %s", e, attempt
            )
            if attempt < _MAX_TRIES:
                if e.status_code is None or e.status_code >= 500:
                    # internal error we may try again
                    cls._add_task(
                        "login", cls.login, attempt + 1, attempt=attempt
                    )
                elif e.status_code == 401:
                    await cls.register(force=True)
            else:
                logger.error(
                    "Failed to login (%s) after %s attempts: %r",
                    request.full_url,
                    attempt,
                    e,
                )
        else:
            atomic_rewrite(
                str(cls.IAID_TOKEN_FILE),
                result["token"],
                backup=cls.IAID_TOKEN_FILE.exists(),
                uid=-1,
                gid=cls._gid(),
                permissions=0o640,
            )
Cancel / Back
[ MAINHACK ]

Viewing / Editing File: iaid.py

Server Info

System Features

User Info
