Source code for axonius_api_client.connect

# -*- coding: utf-8 -*-
"""Easy all-in-one connection handler."""
import logging
import pathlib
import re
from typing import List, Optional, Union

import requests

from .api import (
    ActivityLogs,
    Adapters,
    Dashboard,
    Devices,
    Enforcements,
    Instances,
    Meta,
    RemoteSupport,
    SettingsGlobal,
    SettingsGui,
    SettingsIdentityProviders,
    SettingsLifecycle,
    Signup,
    SystemRoles,
    SystemUsers,
    Users,
    OpenAPISpec,
)
from .auth import ApiKey
from .constants.api import TIMEOUT_CONNECT, TIMEOUT_RESPONSE
from .constants.logs import (LOG_FILE_MAX_FILES, LOG_FILE_MAX_MB,
                             LOG_FILE_NAME, LOG_FILE_PATH, LOG_FMT_BRIEF,
                             LOG_FMT_VERBOSE, LOG_LEVEL_API, LOG_LEVEL_AUTH,
                             LOG_LEVEL_CONSOLE, LOG_LEVEL_FILE, LOG_LEVEL_HTTP,
                             LOG_LEVEL_PACKAGE)
from .exceptions import ConnectError, InvalidCredentials
from .http import Http
from .logs import LOG, add_file, add_stderr, get_obj_log, set_log_level
from .setup_env import get_env_ax
from .tools import coerce_bool, coerce_int, json_dump, json_reload, sysinfo
from .version import __version__ as VERSION


[docs]class Connect: """Easy all-in-one connection handler for using the API client. Examples: >>> #!/usr/bin/env python >>> # -*- coding: utf-8 -*- >>> '''Base example for setting up the API client.''' >>> import axonius_api_client as axonapi >>> >>> # get the URL, API key, API secret, & certwarn from the default ".env" file >>> client_args = axonapi.get_env_connect() >>> >>> # OR override OS env vars with the values from a custom .env file >>> # client_args = axonapi.get_env_connect(ax_env="/path/to/envfile", override=True) >>> >>> # create a client using the url, key, and secret from OS env >>> client = axonapi.Connect(**client_args) >>> >>> j = client.jdump # json dump helper >>> >>> client.start() # connect to axonius >>> >>> # client.activity_logs # get audit logs >>> # client.adapters # get adapters and update adapter settings >>> # client.adapters.cnx # CRUD for adapter connections >>> # client.dashboard # get/start/stop discovery cycles >>> # client.devices # get device assets >>> # client.devices.fields # get field schemas for device assets >>> # client.devices.labels # add/remove/get tags for device assets >>> # client.devices.saved_queries # CRUD for saved queries for device assets >>> # client.enforcements # CRUD for enforcements >>> # client.instances # get instances and instance meta data >>> # client.meta # get product meta data >>> # client.remote_support # enable/disable remote support settings >>> # client.settings_global # get/update global system settings >>> # client.settings_gui # get/update gui system settings >>> # client.settings_ip # get/update identity provider system settings >>> # client.settings_lifecycle # get/update lifecycle system settings >>> # client.signup # perform initial signup and use password reset tokens >>> # client.system_roles # CRUD for system roles >>> # client.system_users # CRUD for system users >>> # client.users # get user assets >>> # client.users.fields # get field schemas for user assets >>> # client.users.labels # add/remove/get tags for user assets >>> # client.users.saved_queries # CRUD for saved queries for user assets """
[docs] def __init__( self, url: str, key: str, secret: str, log_console: bool = False, log_file: bool = False, certpath: Optional[Union[str, pathlib.Path]] = None, certverify: bool = False, certwarn: bool = True, proxy: Optional[str] = None, **kwargs, ): """Easy all-in-one connection handler. Args: url: URL, hostname, or IP address of Axonius instance key: API Key from account page in Axonius instance secret: API Secret from account page in Axonius instance log_console: enable logging to console log_file: enable logging to file certpath: path to CA bundle file to use when verifying certs offered by :attr:`url` certverify: raise exception if cert is self-signed or only if cert is invalid certwarn: show insecure warning once or never show insecure warning proxy: proxy to use when making https requests to :attr:`url` **kwargs: documented as properties """ self.url: str = url """URL of Axonius instance to use""" certwarn = coerce_bool(certwarn) certverify = coerce_bool(certverify) log_console = coerce_bool(log_console) log_file = coerce_bool(log_file) self.TIMEOUT_CONNECT: int = coerce_int(kwargs.get("timeout_connect", TIMEOUT_CONNECT)) """Seconds to wait for connections to open to :attr:`url` ``kwargs=timeout_connect``""" self.TIMEOUT_RESPONSE: int = coerce_int(kwargs.get("timeout_response", TIMEOUT_RESPONSE)) """Seconds to wait for responses from :attr:`url` ``kwargs=timeout_response``""" self.CERT_CLIENT_KEY: Optional[Union[str, pathlib.Path]] = kwargs.get( "cert_client_key", None ) """Private key file for cert_client_cert ``kwargs=cert_client_key``""" self.CERT_CLIENT_CERT: Optional[Union[str, pathlib.Path]] = kwargs.get( "cert_client_cert", None ) """cert file to offer to :attr:`url` ``kwargs=cert_client_cert``""" self.CERT_CLIENT_BOTH: Optional[Union[str, pathlib.Path]] = kwargs.get( "cert_client_both", None ) """cert file with both private key and cert to offer to :attr:`url` ``kwargs=cert_client_both``""" self.SAVE_HISTORY: bool = kwargs.get("save_history", False) """append responses to :attr:`axonius_api_client.http.Http.HISTORY` ``kwargs=save_history``""" self.LOG_LEVEL: Union[str, int] = kwargs.get("log_level", "debug") """log level for this class ``kwargs=log_level``""" self.LOG_REQUEST_ATTRS: Optional[List[str]] = kwargs.get("log_request_attrs", None) """request attrs to log :attr:`axonius_api_client.constants.logs.REQUEST_ATTR_MAP` ``kwargs=log_request_attrs``""" self.LOG_RESPONSE_ATTRS: Optional[List[str]] = kwargs.get("log_response_attrs", None) """response attrs to log :attr:`axonius_api_client.constants.logs.RESPONSE_ATTR_MAP` ``kwargs=log_response_attrs``""" self.LOG_REQUEST_BODY: bool = kwargs.get("log_request_body", False) """log request bodies ``kwargs=log_request_body``""" self.LOG_RESPONSE_BODY: bool = kwargs.get("log_response_body", False) """log response bodies ``kwargs=log_response_body``""" self.LOG_LOGGER: logging.Logger = kwargs.get("log_logger", LOG) """logger to use as package root logger ``kwargs=log_logger``""" self.LOG_LEVEL_PACKAGE: Union[str, int] = kwargs.get("log_level_package", LOG_LEVEL_PACKAGE) """log level for entire package ``kwargs=log_level_package``""" self.LOG_LEVEL_HTTP: Union[str, int] = kwargs.get("log_level_http", LOG_LEVEL_HTTP) """log level for :obj:`axonius_api_client.http.Http` ``kwargs=log_level_http``""" self.LOG_LEVEL_AUTH: Union[str, int] = kwargs.get("log_level_auth", LOG_LEVEL_AUTH) """log level for :obj:`axonius_api_client.auth.models.Mixins` ``kwargs=log_level_auth``""" self.LOG_LEVEL_API: Union[str, int] = kwargs.get("log_level_api", LOG_LEVEL_API) """log level for :obj:`axonius_api_client.api.mixins.ModelMixins` ``kwargs=log_level_api``""" self.LOG_LEVEL_CONSOLE: Union[str, int] = kwargs.get("log_level_console", LOG_LEVEL_CONSOLE) """log level for logs sent to console ``kwargs=log_level_console``""" self.LOG_LEVEL_FILE: Union[str, int] = kwargs.get("log_level_file", LOG_LEVEL_FILE) """log level for logs sent to file ``kwargs=log_level_file``""" self.LOG_CONSOLE_FMT: str = kwargs.get("log_console_fmt", LOG_FMT_BRIEF) """logging format to use for logs sent to console ``kwargs=log_console_fmt``""" self.LOG_FILE_FMT: str = kwargs.get("log_file_fmt", LOG_FMT_VERBOSE) """logging format to use for logs sent to file ``kwargs=log_file_fmt``""" self.LOG_FILE_NAME: Union[str, pathlib.Path] = kwargs.get("log_file_name", LOG_FILE_NAME) """name of file to write logs to under :attr:`LOG_FILE_PATH` ``kwargs=log_file_name``""" self.LOG_FILE_PATH: Union[str, pathlib.Path] = kwargs.get("log_file_path", LOG_FILE_PATH) """path to write :attr:`LOG_FILE_NAME` to ``kwargs=log_file_path``""" self.LOG_FILE_MAX_MB: int = kwargs.get("log_file_max_mb", LOG_FILE_MAX_MB) """rollover file logs at this many MB ``kwargs=log_file_max_mb``""" self.LOG_FILE_MAX_FILES: int = kwargs.get("log_file_max_files", LOG_FILE_MAX_FILES) """number of rollover file logs to keep ``kwargs=log_file_max_files``""" self.WRAPERROR: bool = coerce_bool(kwargs.get("wraperror", True)) """wrap errors in human friendly way or show full traceback ``kwargs=wraperror``""" self.LOG: logging.Logger = get_obj_log(obj=self, level=self.LOG_LEVEL) """logger object to use""" set_log_level(obj=self.LOG_LOGGER, level=self.LOG_LEVEL_PACKAGE) self.STARTED: bool = False """track if :meth:`start` has been called""" self.HANDLER_FILE: logging.handlers.RotatingFileHandler = None """file logging handler""" self.HANDLER_CON: logging.StreamHandler = None """console logging handler""" if log_console: self.HANDLER_CON = add_stderr( obj=self.LOG_LOGGER, level=self.LOG_LEVEL_CONSOLE, fmt=self.LOG_CONSOLE_FMT ) if log_file: self.HANDLER_FILE = add_file( obj=self.LOG_LOGGER, level=self.LOG_LEVEL_FILE, file_path=self.LOG_FILE_PATH, file_name=self.LOG_FILE_NAME, max_mb=self.LOG_FILE_MAX_MB, max_files=self.LOG_FILE_MAX_FILES, fmt=self.LOG_FILE_FMT, ) headers = kwargs.get("headers") or {} self.HTTP_ARGS: dict = { "url": url, "https_proxy": proxy, "certpath": certpath, "certwarn": certwarn, "certverify": certverify, "cert_client_both": self.CERT_CLIENT_BOTH, "cert_client_cert": self.CERT_CLIENT_CERT, "cert_client_key": self.CERT_CLIENT_KEY, "log_level": self.LOG_LEVEL_HTTP, "log_request_attrs": self.LOG_REQUEST_ATTRS, "log_response_attrs": self.LOG_RESPONSE_ATTRS, "log_request_body": self.LOG_REQUEST_BODY, "log_response_body": self.LOG_RESPONSE_BODY, "save_history": self.SAVE_HISTORY, "connect_timeout": self.TIMEOUT_CONNECT, "response_timeout": self.TIMEOUT_RESPONSE, "headers": headers, } """arguments to use for creating :attr:`HTTP`""" self.AUTH_ARGS: dict = {"key": key, "secret": secret, "log_level": self.LOG_LEVEL_AUTH} """arguments to use for creating :attr:`AUTH`""" self.HTTP = Http(**self.HTTP_ARGS) """:obj:`axonius_api_client.http.Http` client to use for :attr:`AUTH`""" self.AUTH = ApiKey(http=self.HTTP, **self.AUTH_ARGS) """:obj:`axonius_api_client.auth.api_key.ApiKey` auth method to use for all API models""" self.API_ARGS: dict = {"auth": self.AUTH, "log_level": self.LOG_LEVEL_API} """arguments to use for all API models""" self.SIGNUP = Signup(**self.HTTP_ARGS) """Easy access to signup."""
[docs] def start(self): """Connect to and authenticate with Axonius.""" if not self.STARTED: sysinfo_dump = json_dump(sysinfo()) LOG.debug(f"SYSTEM INFO: {sysinfo_dump}") try: self.AUTH.login() except Exception as exc: if not self.WRAPERROR: raise pre = f"Unable to connect to {self.HTTP.url!r}" if isinstance(exc, requests.ConnectTimeout): timeout = self.HTTP.CONNECT_TIMEOUT msg = f"{pre}: connection timed out after {timeout} seconds" cnxexc = ConnectError(msg) elif isinstance(exc, requests.ConnectionError): reason = self._get_exc_reason(exc=exc) cnxexc = ConnectError(f"{pre}: {reason}") elif isinstance(exc, InvalidCredentials): cnxexc = ConnectError(f"{pre}: Invalid Credentials supplied") else: cnxexc = ConnectError(f"{pre}: {exc}") cnxexc.exc = exc raise cnxexc self.STARTED = True LOG.info(str(self))
@property def signup(self) -> Signup: """Work with signup endpoints.""" if not hasattr(self, "_signup"): self._signup = Signup(**self.HTTP_ARGS) return self._signup @property def users(self) -> Users: """Work with user assets.""" self.start() if not hasattr(self, "_users"): self._users = Users(**self.API_ARGS) return self._users @property def devices(self) -> Devices: """Work with device assets.""" self.start() if not hasattr(self, "_devices"): self._devices = Devices(**self.API_ARGS) return self._devices @property def adapters(self) -> Adapters: """Work with adapters and adapter connections.""" self.start() if not hasattr(self, "_adapters"): self._adapters = Adapters(**self.API_ARGS) return self._adapters @property def instances(self) -> Instances: """Work with instances.""" self.start() if not hasattr(self, "_instances"): self._instances = Instances(**self.API_ARGS) return self._instances @property def activity_logs(self) -> ActivityLogs: """Work with activity logs.""" self.start() if not hasattr(self, "_activity_logs"): self._activity_logs = ActivityLogs(**self.API_ARGS) return self._activity_logs @property def remote_support(self) -> RemoteSupport: """Work with configuring remote support.""" self.start() if not hasattr(self, "_remote_support"): self._remote_support = RemoteSupport(**self.API_ARGS) return self._remote_support @property def dashboard(self) -> Dashboard: """Work with dashboards and discovery cycles.""" self.start() if not hasattr(self, "_dashboard"): self._dashboard = Dashboard(**self.API_ARGS) return self._dashboard @property def enforcements(self) -> Enforcements: """Work with Enforcement Center.""" self.start() if not hasattr(self, "_enforcements"): self._enforcements = Enforcements(**self.API_ARGS) return self._enforcements @property def system_users(self) -> SystemUsers: """Work with system users.""" self.start() if not hasattr(self, "_system_users"): self._system_users = SystemUsers(**self.API_ARGS) return self._system_users @property def system_roles(self) -> SystemRoles: """Work with system roles.""" self.start() if not hasattr(self, "_system_roles"): self._system_roles = SystemRoles(**self.API_ARGS) return self._system_roles @property def meta(self) -> Meta: """Work with instance metadata.""" self.start() if not hasattr(self, "_meta"): self._meta = Meta(**self.API_ARGS) return self._meta @property def settings_ip(self) -> SettingsIdentityProviders: """Work with identity providers settings.""" self.start() if not hasattr(self, "_settings_ip"): self._settings_ip = SettingsIdentityProviders(**self.API_ARGS) return self._settings_ip @property def settings_global(self) -> SettingsGlobal: """Work with core system settings.""" self.start() if not hasattr(self, "_settings_global"): self._settings_global = SettingsGlobal(**self.API_ARGS) return self._settings_global @property def settings_gui(self) -> SettingsGui: """Work with gui system settings.""" self.start() if not hasattr(self, "_settings_gui"): self._settings_gui = SettingsGui(**self.API_ARGS) return self._settings_gui @property def settings_lifecycle(self) -> SettingsLifecycle: """Work with lifecycle system settings.""" self.start() if not hasattr(self, "_settings_lifecycle"): self._settings_lifecycle = SettingsLifecycle(**self.API_ARGS) return self._settings_lifecycle
[docs] def __str__(self) -> str: """Show object info.""" client = getattr(self, "HTTP", "") url = getattr(client, "URL", self.HTTP_ARGS["url"]) ax_env = get_env_ax() banner = ax_env.get("AX_BANNER") banner = f"[{banner}]" if banner else "" pkg_ver = f"API Client v{VERSION}" if self.STARTED: about = self.meta.about() version = about.get("Version", "") or "DEMO" version = version.replace("_", ".") built = about.get("Build Date", "") msg = [f"Connected to {url!r}", f"version {version}", f"(RELEASE DATE: {built})"] else: msg = [f"Not connected to {url!r}"] bits = [x for x in [*msg, pkg_ver, banner] if x] return " ".join(bits)
[docs] def __repr__(self) -> str: """Show object info.""" return self.__str__()
@property def openapi(self) -> OpenAPISpec: """Work with the OpenAPI specification file.""" self.start() if not hasattr(self, "_openapi"): self._openapi = OpenAPISpec(**self.API_ARGS) return self._openapi
[docs] @classmethod def _get_exc_reason(cls, exc: Exception) -> str: """Trim exceptions down to a more user friendly display. Uses :attr:`REASON_RES` to do regex substituions. """ reason = str(exc) for reason_re in cls.REASON_RES: if reason_re.search(reason): return reason_re.sub(r"\1", reason).rstrip("')") return reason
[docs] @staticmethod def jdump(obj, **kwargs): # pragma: no cover """JSON dump utility.""" print(json_reload(obj, **kwargs))
REASON_RES: List[str] = [ re.compile(r".*?object at.*?\>\: ([a-zA-Z0-9\]\[: ]+)"), re.compile(r".*?\] (.*) "), ] """patterns to look for in exceptions that we can pretty up for user display."""