4.3.2. Connections

API for working with adapter connections.

class axonius_api_client.api.adapters.cnx.Cnx(parent)[source]

Bases: ChildMixins

API model for working with adapter connections.

Examples

Create a client using axonius_api_client.connect.Connect

Notes

All methods use the Core instance by default, but you can work with another instance by passing the name of the instance to adapter_node.

Supplying unknown keys/values for configurations will throw an error showing the valid keys/values.

Parameters

parent (axonius_api_client.api.mixins.Model) –

get_by_adapter(adapter_name, adapter_node=None, tunnel=None)[source]

Get all connections of an adapter on a node.

Examples

First, create a client using axonius_api_client.connect.Connect.

Get all connections for an adapter on the Core instance

>>> cnxs = client.adapters.cnx.get_by_adapter(adapter_name="active_directory")
Parameters
  • adapter_name (str) – name of adapter

  • adapter_node (Optional[str], optional) – name of node running adapter

  • tunnel (Optional[str], optional) – name or ID of tunnel

Returns

connection metadata for adapter

Return type

List[dict]

get_by_uuid(cnx_uuid, adapter_name, adapter_node=None, tunnel=None, **kwargs)[source]

Get a connection for an adapter on a node by UUID.

Examples

First, create a client using axonius_api_client.connect.Connect.

Get a single connection by UUID

>>> cnx = client.adapters.cnx.get_by_uuid(
...     cnx_id='5f76735be4557d5cba94237f', adapter_name='aws'
... )

Notes

UUID of connections change when a connection configuration is updated, the more persistent way to get a connection is get_by_id().

Parameters
  • cnx_uuid (str) – UUID to search for

  • adapter_name (str) – name of adapter

  • adapter_node (typing.Optional[str]) – name of node running adapter

  • tunnel (Optional[str], optional) – name or ID of tunnel

  • **kwargs – passed to get_by_key()

Raises

NotFoundError – when no connections found with supplied uuid

Returns

connection metadata for uuid on adapter

Return type

dict

get_by_label(value, adapter_name, adapter_node=None, tunnel=None)[source]

Get a connection for an adapter on a node using a specific connection identifier key.

Examples

First, create a client using axonius_api_client.connect.Connect.

Get a single connection by connection label

>>> cnx = client.adapters.cnx.get_by_label(
...     value='test label', adapter_name='active_directory'
... )
Parameters
  • value (str) – value that connection_label must match for a connection

  • adapter_name (str) – name of adapter

  • adapter_node (Optional[str], optional) – name of node running adapter

  • tunnel (Optional[str], optional) – name or ID of tunnel

Raises

NotFoundError – when no connections found with supplied connection label

Returns

connection metadata for label on adapter

Return type

dict

get_by_id(cnx_id, adapter_name, adapter_node=None, tunnel=None, **kwargs)[source]

Get a connection for an adapter on a node by ID.

Examples

First, create a client using axonius_api_client.connect.Connect.

Get a single connection by ID

>>> cnx = client.adapters.cnx.get_by_id(
...     cnx_id='192.168.1.10', adapter_name='active_directory'
... )

Notes

ID is constructed from some variance of connection keys, usually “domain” (i.e. for active_directory TestDomain.test)

Parameters
  • cnx_id (str) – connection ID to get

  • adapter_name (str) – name of adapter

  • adapter_node (Optional[str], optional) – name of node running

  • tunnel (Optional[str], optional) – name or ID of tunnel

  • **kwargs – passed to get_by_key()

Raises

NotFoundError – when no connections found with supplied id

Returns

connection metadata for id on adapter

Return type

dict

update_by_id(cnx_id, adapter_name, adapter_node=None, tunnel=None, **kwargs)[source]

Update a connection for an adapter on a node by ID.

Examples

First, create a client using axonius_api_client.connect.Connect.

Change the connection label for a connection

>>> cnx = client.adapters.cnx.update_by_id(
...     cnx_id='TestDomain.test',
...     adapter_name='active_directory',
...     connection_label="new label",
... )
Parameters
  • cnx_id (str) – connection ID to update

  • adapter_name (str) – name of adapter

  • adapter_node (typing.Optional[str]) – name of node running adapter

  • tunnel (Optional[str], optional) – name or ID of tunnel

  • **kwargs – passed to update_cnx()

Returns

updated connection metadata

Return type

dict

delete_by_id(cnx_id, adapter_name, adapter_node=None, tunnel=None, delete_entities=False)[source]

Delete a connection for an adapter on a node by connection ID.

Examples

First, create a client using axonius_api_client.connect.Connect.

Delete a connection by ID

>>> cnx = client.adapters.cnx.delete_by_id(
...     cnx_id='192.168.1.10', adapter_name='active_directory'
... )
Parameters
  • cnx_id (str) – connection ID to delete

  • adapter_name (str) – name of adapter

  • adapter_node (typing.Optional[str]) – name of node running adapter

  • tunnel (Optional[str], optional) – name or ID of tunnel

  • delete_entities (bool) – delete all assets fetched by this connection

Return type

str

test_by_id(**kwargs)[source]

Test a connection for an adapter on a node by ID.

Examples

First, create a client using axonius_api_client.connect.Connect.

Test the reachability of a connection by ID

>>> cnx = client.adapters.cnx.test_by_id(
...     cnx_id='192.168.1.10', adapter_name='active_directory'
... )
Parameters

**kwargs – passed to get_by_id()

Return type

dict

add(adapter_name, adapter_node=None, save_and_fetch=True, active=True, tunnel=None, connection_label=None, kwargs_config=None, new_config=None, parse_config=True, **kwargs)[source]

Add a connection to an adapter on a node.

Examples

First, create a client using axonius_api_client.connect.Connect.

Establish a connection dictionary

>>> config = dict(
...     dc_name="192.168.1.10",
...     user="svc_user",
...     password="test",
...     do_not_fetch_users=False,
...     fetch_disabled_users=False,
...     fetch_disabled_devices=False,
...     is_ad_gc=False,
... )

Add a connection for an adapter to the Core instance

>>> cnx = client.adapters.cnx.add(adapter_name="active_directory", **config)
Parameters
  • adapter_name (str) – name of adapter

  • adapter_node (typing.Optional[str]) – name of node running adapter

  • save_and_fetch (bool) – perform a fetch when saving, or just save without fetching

  • active (bool) – set the connection as active after creating

  • connection_label (typing.Optional[str]) – label to assign to connection

  • tunnel (typing.Union[axonius_api_client.api.json_api.instances.Tunnel, str, None]) – tunnel ID or name to use for new connection

  • kwargs_config (typing.Optional[dict]) – connection args that conflict with this methods signature

  • new_config (typing.Optional[dict]) – connection args that conflict with this methods signature

  • parse_config (bool) – perform api client side parsing of connection args

  • **kwargs – configuration of new connection

Raises

CnxAddError – when an error happens while adding the connection

Return type

dict

test(adapter_name, adapter_node=None, tunnel=None, kwargs_config=None, new_config=None, parse_config=True, **kwargs)[source]

Test a connection to an adapter on a node.

Examples

First, create a client using axonius_api_client.connect.Connect.

Test the reachability of a connection without creating the connection

>>> config = dict(dc_name="192.168.1.10", user="svc_user", password="test")
>>> cnx = client.adapters.cnx.test(adapter_name="active_directory", **config)

Notes

This can be used to test the configuration of a connection before creating the connection. Usually you need just whatever configuration keys are related to hostname/domain/ip address to test a connection.

Parameters
  • adapter_name (str) – name of adapter

  • adapter_node (Optional[str], optional) – name of node running adapter

  • tunnel (typing.Union[axonius_api_client.api.json_api.instances.Tunnel, str, None]) – tunnel ID or name to use for testing connection

  • kwargs_config (Optional[dict], optional) – configuration of connection to test

  • new_config (Optional[dict], optional) – configuration of connection to test

  • parse_config (bool, optional) – parse configuration for unknowns, requireds, etc

  • **kwargs – configuration of connection to test

Raises
Return type

dict

test_cnx(cnx_test, **kwargs)[source]

Test a connection for an adapter on a node.

Parameters

cnx_test (dict) – connection fetched previously

Return type

dict

update_cnx(cnx_update, save_and_fetch=True, active=None, new_node=None, new_tunnel=None, kwargs_config=None, new_config=None, parse_config=True, **kwargs)[source]

Update a connection for an adapter on a node.

Parameters
  • cnx_update (dict) – connection fetched previously

  • save_and_fetch (bool) – perform a fetch when saving, or just save without fetching

  • active (typing.Optional[bool]) – set the connection as active or inactive after updating

  • new_node (typing.Optional[str]) – move connection to a new node

  • new_tunnel (typing.Union[axonius_api_client.api.json_api.instances.Tunnel, str, None]) – move connection to a new tunnel ID or name

  • kwargs_config (typing.Optional[dict]) – connection args that conflict with this methods signature

  • new_config (typing.Optional[dict]) – connection args that conflict with this methods signature

  • parse_config (bool) – perform api client side parsing of connection args

  • **kwargs – configuration of connection to update

Raises

CnxUpdateError – When an error occurs while updating the connection

Returns

updated connection metadata

Return type

dict

delete_cnx(cnx_delete, delete_entities=False)[source]

Delete a connection for an adapter on a node.

Parameters
  • cnx_delete (dict) – connection fetched previously

  • delete_entities (bool, optional) – delete all assets fetched by this connection

Returns

dataclass model containing response

Return type

CnxDelete

set_cnx_active(cnx, value, save_and_fetch=False)[source]

Set a connection to active.

Parameters
  • cnx (dict) – Previously fetched connection object

  • value (bool) – Set cnx as active or inactive

  • save_and_fetch (bool, optional) – When saving, perform a fetch as well

Returns

updated cnx object

Return type

dict

set_cnx_label(cnx, value, save_and_fetch=False)[source]

Set a connections label.

Parameters
  • cnx (dict) – Previously fetched connection object

  • value (bool) – label to apply to cnx

  • save_and_fetch (bool, optional) – When saving, perform a fetch as well

Returns

updated cnx object

Return type

dict

build_config(cnx_schemas, new_config, source, adapter_name, adapter_node, old_config=None)[source]

Build and parse a configuration for a connection.

Parameters
  • cnx_schemas (typing.List[dict]) – configuration schemas for connection

  • new_config (dict) – new configuration for connection

  • source (str) – description of what called this method

  • adapter_name (str) – name of adapter

  • adapter_node (str) – name of node running adapter

  • old_config (typing.Optional[dict]) – configuration that is being updated

Return type

dict

get_sane_defaults(adapter_name)[source]

Get the API client defined sane defaults for a specific adapter.

Parameters

adapter_name (str) – name of adapter

Return type

dict

cb_file_upload(value, schema, callbacks, source)[source]

Config parsing callback to upload a file for a connection.

Parameters
  • value (typing.Union[str, pathlib.Path, dict]) – file to upload

  • schema (dict) – connection configuration schema of type “file”

  • callbacks (dict) – callbacks supplied

  • source (str) – description of what called this method

Raises

ConfigInvalidValue – When value is a path that does not exist, or a dictionary that does not have ‘uuid’ and ‘filename’ keys’, or if value is not a file

Return type

dict

_add(connection, instance_id, instance_name, adapter_name, connection_discovery=None, is_instances_mode=False, save_and_fetch=True, active=True, connection_label=None, tunnel_id=None, response_status_hook=None)[source]

Pass.

Parameters
Return type

axonius_api_client.api.json_api.adapters.CnxModifyResponse

_test(adapter_name, instance, connection, tunnel_id=None, response_status_hook=None)[source]

Direct API method to add a connection to an adapter.

Parameters
  • adapter_name (str) – raw name of the adapter i.e. aws_adapter

  • instance (str) – id of node running adapter

  • connection (dict) – configuration to test

  • tunnel_id (Optional[str], optional) – tunnel ID to use

  • response_status_hook (Optional[Callable], optional) – callable to use when checking status codes of response

Returns

containing response

Return type

dict

_get(adapter_name)[source]

Get all connections for a given adapter.

Parameters

adapter_name (str) – name of adapter

Returns

dataclass model containing response

Return type

Cnxs

_delete(uuid, adapter_name, instance_id, instance_name, delete_entities=False, is_instances_mode=False, response_status_hook=None)[source]

Direct API method to delete a connection from an adapter.

Parameters
  • uuid (str) – uuid of connection

  • adapter_name (str) – name of adapter

  • instance_id (str) – UUID of instance to delete connection from

  • instance_name (str) – NAME of instance to delete connection from

  • delete_entities (bool, optional) – delete all assets fetched by this connection

  • is_instances_mode (bool, optional) – instance_id & instance_name is NOT the Core instance

  • response_status_hook (Optional[Callable], optional) – callable to use when checking status codes of response

Returns

dataclass model containing response

Return type

CnxDelete

_update(uuid, connection, instance_id, instance_name, adapter_name, instance_prev=None, instance_prev_name=None, tunnel_id=None, connection_discovery=None, is_instances_mode=False, save_and_fetch=True, active=True, connection_label=None, response_status_hook=None)[source]

Pass.

Parameters
  • uuid (str) – uuid of connection

  • connection (dict) – new configuration to apply

  • instance_id (str) – UUID of instance to move connection to

  • instance_name (str) – NAME of instance to move connection to

  • adapter_name (str) – name of adapter

  • tunnel_id (Optional[str], optional) – tunnel ID to use

  • instance_prev (Optional[str], optional) – UUID of instance to move connection from

  • instance_prev_name (Optional[str], optional) – NAME of instance to move connection from

  • connection_discovery (Optional[dict], optional) – connection specific discovery settings to update

  • is_instances_mode (bool, optional) – instance_id & instance_name is NOT the Core instance

  • save_and_fetch (bool, optional) – save and fetch the connection, or just save it

  • active (bool, optional) – set connection as active

  • connection_label (Optional[str], optional) – label to set on connection

  • response_status_hook (Optional[Callable], optional) – callable to use when checking status codes of response

Returns

dataclass model containing response

Return type

CnxModifyResponse

get_response_status_hook(cnx)[source]

Check if the result of updating a connection shows that the connection is gone.

Parameters

cnx (dict) – Connection metadata to use in response hook

Returns

response status hook

Return type

Callable

__init__(parent)

Mixins model for API child objects.

Parameters

parent (axonius_api_client.api.mixins.Model) – parent API model of this child

__repr__()

Show info for this model object.

Return type

str

__str__()

Show info for this model object.

Return type

str

_init(parent)

Post init method for subclasses to use for extra setup.

Parameters

parent (axonius_api_client.api.mixins.Model) – parent API model of this child

class axonius_api_client.api.adapters.cnx.ErrorMap(response_regexes, err, exc, endpoint_regexes=<factory>, with_schemas=True, with_config=True, with_cnxs=True, skip_status=False)[source]

Bases: object

Mapping container for use in Cnx.get_response_status_hook().

Parameters
response_regexes: List[str]

List of regexes that if they match response body, will throw exc

err: str

Error string to include in exception if response_regexes match.

exc: Type[Exception]

Exception class to throw if response_regexes match.

endpoint_regexes: List[str]

List of regexes that if they match request URL, will check response_regexes.

with_schemas: bool = True

Include connection configuration schemas in error output.

with_config: bool = True

Include connection configuration in error output.

with_cnxs: bool = True

Include all connections for adapter in question in error output.

__init__(response_regexes, err, exc, endpoint_regexes=<factory>, with_schemas=True, with_config=True, with_cnxs=True, skip_status=False)
Parameters
skip_status: bool = False

Skip the rest of the built in status checks.

__post_init__()[source]

Post init dataclass setup.

is_response_error(response)[source]

Check if response matches endpoint_regexes and response_regexes.

Parameters

response (requests.Response) – response object to check

Returns

if response matches this error map

Return type

bool

handle_exc(response, cnx, apiobj)[source]

Handle raising an exception if is_response_error().

Parameters
  • response (requests.Response) – response object to check

  • cnx (dict) – connection metadata

  • apiobj (Cnx) – API model to use to fetch connections

Returns

return skip_status

Return type

bool

Raises

Exception – of type exc

axonius_api_client.api.adapters.cnx.get_cnx_strs(cnx, with_config=False, with_schemas=False)[source]

Build a string with details about a connection.

Parameters
  • cnx (dict) – connection to build string for

  • with_config (bool, optional) – include connection config in output

  • with_schemas (bool, optional) – include connection config schemas in output

Returns

connection details

Return type

List[str]

axonius_api_client.api.adapters.cnx.cnx_from_adapter(adapter)[source]

Build a base dict for a connection with details from an adapter metadata dict.

Parameters

adapter (dict) – adapter to get details from

Returns

base dict for connection

Return type

dict