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

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