4.3.2. Connections¶

Warning

This API is deprecated.

Axonius API v2 is now available, we recommend you move to API v2.

New features will no longer be added to this API. Axonius will only provide bug fixes.

Read about API v2

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

Add a connection: add()
Get all connections for an adapter: get_by_adapter()
Get a connection for an adapter by UUID: get_by_uuid()
Get a connection for an adapter by connection label: get_by_label()
Get a connection for an adapter by ID: get_by_id()
Update a connection for an adapter by ID: update_by_id()
Delete a connection for an adapter by ID: delete_by_id()
Test a connections parameters for an adapter without creating the connection: test()
Work with adapters axonius_api_client.api.adapters.adapters.Adapters

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, internal_axon_tenant_id=None, **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
internal_axon_tenant_id (typing.Optional[str]) – The ID of the Tenant that the connection is associated with
**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:

CnxTestError – When a connection test fails
ConfigRequired – When not enough arguments are supplied to test the connection

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, internal_axon_tenant_id=None)[source]¶

Pass.

Parameters:

connection (dict) –
instance_id (str) –
instance_name (str) –
adapter_name (str) –
connection_discovery (typing.Optional[dict]) –
is_instances_mode (bool) –
save_and_fetch (bool) –
active (bool) –
connection_label (typing.Optional[str]) –
tunnel_id (typing.Optional[str]) –
response_status_hook (typing.Optional[typing.Callable]) –
internal_axon_tenant_id (typing.Optional[str]) –

Return type:

axonius_api_client.api.json_api.adapters.cnx_create_response.CnxCreate

_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

_get_labels()[source]¶

Get all connection labels.

Returns:: dataclass model containing response
Return type:: CnxLabels

_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

LOG: logging.Logger = None¶: Logger for this object.

__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

auth: axonius_api_client.auth.model.AuthModel = None¶: Authentication model with bound Http object to use for requests.

http: axonius_api_client.http.Http = None¶: Http object to use for requests.

parent: axonius_api_client.api.mixins.Model = None¶: 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 (typing.List[str]) –
err (str) –
exc (typing.Type[Exception]) –
endpoint_regexes (typing.List[str]) –
with_schemas (bool) –
with_config (bool) –
with_cnxs (bool) –
skip_status (bool) –

response_regexes: typing.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: typing.Type[Exception]¶: Exception class to throw if response_regexes match.

endpoint_regexes: typing.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:

response_regexes (typing.List[str]) –
err (str) –
exc (typing.Type[Exception]) –
endpoint_regexes (typing.List[str]) –
with_schemas (bool) –
with_config (bool) –
with_cnxs (bool) –
skip_status (bool) –

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