← Back
Editing: iaid.py
import asyncio import grp import json import os import random import time from dataclasses import dataclass from logging import getLogger from pathlib import Path from typing import Callable 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, safe_cancel_task from defence360agent.utils.common import DAY from defence360agent.internals.global_scope import g from defence360agent.internals.deadlock_detecting_lock import ( DeadlockDetectingLock, DeadlockError, ) logger = getLogger(__name__) _MAX_TRIES = 10 _TIMEOUT_MULTIPLICATOR = 2 """ >>> _MAX_TRIES_FOR_DOWNLOAD = 10 >>> _TIMEOUT_MULTIPLICATOR = 2 >>> [(1 << i) * _TIMEOUT_MULTIPLICATOR for i in range(1, _MAX_TRIES_FOR_DOWNLOAD)] # noqa [4, 8, 16, 32, 64, 128, 256, 512, 1024] """ _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": [], } _register_lock = DeadlockDetectingLock() _activate_lock = asyncio.Lock() @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: Callable, *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: Callable, *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 def add_initial_task(cls): cls._add_task("activate", cls.activate, attempt=0) @classmethod async def shutdown(cls): for type, tasks in cls._tasks.items(): for task in tasks: if not task.done(): await safe_cancel_task(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() if IndependentAgentIDAPI.is_token_expired(): raise IAIDTokenError("IAID token is expired") try: token = cls.IAID_TOKEN_FILE.read_text(encoding="ascii").strip() if not token: raise IAIDTokenError("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) try: return cls.TokenInfo(**token) except TypeError as e: raise APIError("incomplete token_info %r: %s" % (token, e)) from e @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, tried_credentials_ts=None, attempt=1): # In case of Unauthorized 401 for login/activate, iaid initiates force registration. # Prevent multiple force registrations by checking if the lock was already acquired. # This approach also works if the lock was already acquired by non-force registration, # as the non-force registration is currently only triggered if iaid wasn't registered. was_waiting = cls._register_lock.locked() try: async with cls._register_lock: if cls.is_registered(): if not force or was_waiting: return # If credentials were already updated - no need to register again. # This check makes the above `was_waiting` check obsolete in most cases, # but some old filesystems have second precision of for a file mtime. # Both checks are kept to decrease a chance of race conditions. if ( tried_credentials_ts is not None and tried_credentials_ts < cls._get_credentials_ts() ): 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, tried_credentials_ts, 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() except DeadlockError: logger.error( "Received incorrect credentials on register after %s attempts", attempt, ) @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 def _get_credentials_ts(cls): return cls.IAID_PASSWORD_FILE.stat().st_mtime @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 async with cls._activate_lock: # A concurrent activate() may have already completed # while we were waiting for the lock. if cls.IAID_ACTIVATED_FILE.exists(): g["iaid"] = cls.get_iaid() return iaid = cls.IAID_FILE.read_text() password = cls.IAID_PASSWORD_FILE.read_text() credentials_ts = cls._get_credentials_ts() request = cls._request( cls.ACTIVATE_URL, iaid=iaid, password=password, license=lic ) g["iaid"] = iaid need_to_register = False 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: # need to register again, do it outside of lock need_to_register = 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() if need_to_register: await cls.register(force=True, tried_credentials_ts=credentials_ts) @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() credentials_ts = cls._get_credentials_ts() 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, tried_credentials_ts=credentials_ts ) 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, )
Save File
Cancel