SecurityClient

Client for the Kibana Security (roles and sessions) API.

Provides role management (create-or-update, get, query, bulk update and delete Kibana roles, including their Elasticsearch and Kibana privilege definitions) and user-session invalidation.

The Security API is not space-scoped: roles and sessions are global to the Kibana instance. Space-level access is granted through the kibana privilege entries of a role (each entry lists the spaces it applies to).

class kibana._sync.client.security.SecurityClient(client)

Bases: NamespaceClient

Client for the Kibana Security (roles and sessions) API.

Provides role management (create-or-update, get, query, bulk update and delete Kibana roles, including their Elasticsearch and Kibana privilege definitions) and user-session invalidation.

The Security API is not space-scoped: roles and sessions are global to the Kibana instance. Space-level access is granted through the kibana privilege entries of a role (each entry lists the spaces it applies to).

Example

>>> from kibana import Kibana
>>> client = Kibana("http://localhost:5601", api_key="...")
>>>
>>> # Create or update a role
>>> client.security.create_or_update_role(
...     name="my-role",
...     elasticsearch={
...         "cluster": ["monitor"],
...         "indices": [{"names": ["logs-*"], "privileges": ["read"]}],
...     },
...     kibana=[{"base": ["read"], "spaces": ["default"]}],
... )
>>>
>>> # Retrieve it
>>> role = client.security.get_role(name="my-role")
>>> print(role.body["name"])
my-role

Managing Roles

Create or update a role with create_or_update_role():

from kibana import Kibana

client = Kibana("http://localhost:5601", api_key="your_api_key")

# Create or update a role
client.security.create_or_update_role(
    name="my-role",
    elasticsearch={
        "cluster": ["monitor"],
        "indices": [{"names": ["logs-*"], "privileges": ["read"]}],
    },
    kibana=[{"base": ["read"], "spaces": ["default"]}],
)

# Retrieve it
role = client.security.get_role(name="my-role")
print(role.body["name"])

Listing and Querying Roles

# List all roles
roles = client.security.get_all_roles()
for role in roles.body:
    print(role["name"])

# Query roles with paging and sorting
results = client.security.query_roles(
    query="my-role",
    size=10,
    sort={"field": "name", "direction": "asc"},
)
for role in results.body["roles"]:
    print(role["name"])

Bulk Operations and Deletion

# Create or update several roles at once
client.security.bulk_create_or_update_roles(
    roles={
        "role-a": {
            "elasticsearch": {"cluster": ["monitor"]},
        },
        "role-b": {
            "kibana": [{"base": ["read"], "spaces": ["default"]}],
        },
    }
)

# Delete a role
client.security.delete_role(name="my-role")

Invalidating Sessions

Invalidate user sessions (requires superuser privileges):

# Invalidate sessions for a specific user of the basic realm
result = client.security.invalidate_sessions(
    match="query",
    query={
        "provider": {"type": "basic"},
        "username": "some-user",
    },
)
print(f"Invalidated {result.body['total']} sessions")

Warning

Calling invalidate_sessions() with match="all" logs out every user of the Kibana instance. Prefer match="query" with a narrow query to target specific sessions.

__init__(client)

Initialize the SecurityClient.

Parameters:

client (Kibana) – The parent Kibana client instance to delegate requests to.

Example

>>> security_client = SecurityClient(kibana_client)

get_all_roles(*, replace_deprecated_privileges=None)

Get all Kibana roles.

Retrieves every role, including reserved and system roles, with their Elasticsearch and Kibana privilege definitions.

Parameters:

replace_deprecated_privileges (bool | None) – If True and the response contains any privileges that are associated with deprecated features, they are omitted in favor of details about the appropriate replacement feature privileges.

Returns:

ObjectApiResponse containing a list of role objects. Each role has name, description, metadata, elasticsearch (cluster/indices/run_as privileges), kibana (base/feature privileges per space) and transient_metadata.

Raises:

• AuthenticationException – If authentication fails.

• AuthorizationException – If insufficient privileges to read roles.

• TransportError – If unable to connect to Kibana.

Return type:

ObjectApiResponse[Any]

Example

>>> roles = client.security.get_all_roles()
>>> for role in roles.body:
...     print(role["name"])
superuser
my-role

get_role(*, name, replace_deprecated_privileges=None)

Get a Kibana role by name.

Parameters:

• name (str) – The role name (required).

• replace_deprecated_privileges (bool | None) – If True and the response contains any privileges that are associated with deprecated features, they are omitted in favor of details about the appropriate replacement feature privileges.

Returns:

ObjectApiResponse containing the role object with name, description, metadata, elasticsearch and kibana privilege definitions.

Raises:

• ValueError – If name is empty.

• NotFoundError – If the role does not exist.

• AuthenticationException – If authentication fails.

• AuthorizationException – If insufficient privileges to read roles.

Return type:

ObjectApiResponse[Any]

Example

>>> role = client.security.get_role(name="my-role")
>>> print(role.body["elasticsearch"]["cluster"])
['monitor']

create_or_update_role(*, name, elasticsearch, description=None, kibana=None, metadata=None, create_only=None)

Create a new Kibana role or update an existing one.

Parameters:

• name (str) – The role name. Must be 1-1024 characters (required).

• elasticsearch (dict[str, Any]) –

  Elasticsearch privileges for the role (required). Supported keys:

  • cluster: list of cluster privilege names.

  • indices: list of index privilege objects, each with names and privileges (and optionally field_security, query, allow_restricted_indices).

  • remote_cluster: list of objects with clusters and privileges for remote clusters.

  • remote_indices: list of remote index privilege objects (like indices plus clusters).

  • run_as: list of usernames the role can impersonate.

• description (str | None) – Optional description for the role (max 2048 chars).

• kibana (list[dict[str, Any]] | None) – Kibana privileges for the role. A list of objects, each with base (base privileges such as ["all"] or ["read"]), feature (a mapping of feature ID to a list of feature privileges; mutually exclusive with a non-empty base) and spaces (the space IDs the entry applies to, or ["*"] for all spaces).

• metadata (dict[str, Any] | None) – Optional arbitrary metadata to store with the role.

• create_only (bool | None) – If True, the request fails if a role with the given name already exists (create-only semantics). Defaults to False on the server (create or update).

Returns:

ObjectApiResponse with an empty body (the server returns 204 No Content on success).

Raises:

• ValueError – If name is empty.

• BadRequestError – If the privilege definitions are invalid.

• ConflictError – If create_only=True and the role already exists.

• AuthenticationException – If authentication fails.

• AuthorizationException – If insufficient privileges to manage roles.

Return type:

ObjectApiResponse[Any]

Example

>>> client.security.create_or_update_role(
...     name="log-reader",
...     description="Read-only access to logs",
...     elasticsearch={
...         "cluster": [],
...         "indices": [
...             {"names": ["logs-*"], "privileges": ["read"]}
...         ],
...     },
...     kibana=[{"base": ["read"], "spaces": ["*"]}],
...     metadata={"version": 1},
... )

delete_role(*, name)

Delete a Kibana role by name.

Reserved roles cannot be deleted.

Parameters:

name (str) – The role name (required).

Returns:

ObjectApiResponse with an empty body (the server returns 204 No Content on success).

Raises:

• ValueError – If name is empty.

• NotFoundError – If the role does not exist.

• AuthenticationException – If authentication fails.

• AuthorizationException – If insufficient privileges to manage roles.

Return type:

ObjectApiResponse[Any]

Example

>>> client.security.delete_role(name="log-reader")

bulk_create_or_update_roles(*, roles)

Create new Kibana roles or update existing ones in bulk.

Parameters:

roles (dict[str, dict[str, Any]]) – A mapping of role name to role definition (required). Each definition accepts the same keys as create_or_update_role(): elasticsearch (required by the server), kibana, description and metadata.

Returns:

ObjectApiResponse summarizing the outcome per role, with keys such as created, updated and errors.

Raises:

• BadRequestError – If a role definition is invalid.

• AuthenticationException – If authentication fails.

• AuthorizationException – If insufficient privileges to manage roles.

Return type:

ObjectApiResponse[Any]

Example

>>> result = client.security.bulk_create_or_update_roles(
...     roles={
...         "role-a": {"elasticsearch": {"cluster": ["monitor"]}},
...         "role-b": {
...             "elasticsearch": {},
...             "kibana": [{"base": ["read"], "spaces": ["*"]}],
...         },
...     }
... )
>>> print(result.body.get("created"))
['role-a', 'role-b']

query_roles(*, query=None, from_=None, size=None, sort=None, filters=None)

Query Kibana roles with optional filters, paging and sorting.

Parameters:

• query (str | None) – Free-text query string used to match role names.

• from – Zero-based offset of the first role to return (paging).

• size (int | None) – Maximum number of roles to return (paging).

• sort (dict[str, Any] | None) – Sort definition with required keys field and direction ("asc" or "desc"), e.g. {"field": "name", "direction": "asc"}.

• filters (dict[str, Any] | None) – Additional filters. Supports showReservedRoles (bool) to include or exclude reserved roles.

Returns:

ObjectApiResponse containing the matching roles, typically with roles (the page of role objects), count and total.

Raises:

• BadRequestError – If the query payload is invalid.

• AuthenticationException – If authentication fails.

• AuthorizationException – If insufficient privileges to read roles.

Return type:

ObjectApiResponse[Any]

Example

>>> result = client.security.query_roles(
...     query="kbnpy",
...     from_=0,
...     size=10,
...     sort={"field": "name", "direction": "asc"},
...     filters={"showReservedRoles": False},
... )
>>> print(result.body["total"])
2

invalidate_sessions(*, match, query=None)

Invalidate user sessions that match a query.

To use this API, you must be a superuser.

Warning

Calling this with match="all" logs out every user of the Kibana instance. Prefer match="query" with a narrow query to target specific sessions.

Parameters:

• match (str) – How Kibana determines which sessions to invalidate (required). "all" invalidates every existing session; "query" invalidates only the sessions that match query.

• query (dict[str, Any] | None) – The query used to match sessions when match is "query". An object with a required provider ({"type": ..., "name": ...}, type required — e.g. basic, token, saml, oidc, kerberos or pki) and an optional username.

Returns:

ObjectApiResponse with the number of successfully invalidated sessions in total.

Raises:

• BadRequestError – If the payload is invalid (e.g. match is not "all" or "query").

• AuthenticationException – If authentication fails.

• AuthorizationException – If the calling user is not a superuser.

Return type:

ObjectApiResponse[Any]

Example

>>> result = client.security.invalidate_sessions(
...     match="query",
...     query={
...         "provider": {"type": "basic"},
...         "username": "some-user",
...     },
... )
>>> print(result.body["total"])
0

perform_request(method, path, *, params=None, headers=None, body=None)

Perform an HTTP request via the parent client with space context enhancement.

Parameters:

• method (str) – HTTP method (GET, POST, PUT, DELETE, etc.)

• path (str) – API endpoint path

• params (dict[str, Any] | None) – Query parameters

• headers (dict[str, str] | None) – Request headers

• body (dict[str, Any] | None) – Request body

Returns:

API response

Raises:

ApiError – If the API returns an error response (enhanced with space context)

Return type:

ObjectApiResponse[Any]

AsyncSecurityClient

Asynchronous version of the SecurityClient for use with async/await syntax.

class kibana._async.client.security.AsyncSecurityClient(client)

Bases: AsyncNamespaceClient

Async client for the Kibana Security (roles and sessions) API.

Provides role management (create-or-update, get, query, bulk update and delete Kibana roles, including their Elasticsearch and Kibana privilege definitions) and user-session invalidation.

The Security API is not space-scoped: roles and sessions are global to the Kibana instance. Space-level access is granted through the kibana privilege entries of a role (each entry lists the spaces it applies to).

Example

>>> from kibana import AsyncKibana
>>> client = AsyncKibana("http://localhost:5601", api_key="...")
>>>
>>> # Create or update a role
>>> await client.security.create_or_update_role(
...     name="my-role",
...     elasticsearch={
...         "cluster": ["monitor"],
...         "indices": [{"names": ["logs-*"], "privileges": ["read"]}],
...     },
...     kibana=[{"base": ["read"], "spaces": ["default"]}],
... )
>>>
>>> # Retrieve it
>>> role = await client.security.get_role(name="my-role")
>>> print(role.body["name"])
my-role

Usage

The AsyncSecurityClient provides the same methods as SecurityClient but all methods are async and must be awaited:

from kibana import AsyncKibana
import asyncio

async def main():
    async with AsyncKibana("http://localhost:5601") as client:
        # Create a role (async)
        await client.security.create_or_update_role(
            name="async-role",
            elasticsearch={"cluster": ["monitor"]},
        )

        # List roles (async)
        roles = await client.security.get_all_roles()

        # Delete the role (async)
        await client.security.delete_role(name="async-role")

asyncio.run(main())

__init__(client)

Initialize the AsyncSecurityClient.

Parameters:

client (AsyncKibana) – The parent AsyncKibana client instance to delegate requests to.

Example

>>> security_client = AsyncSecurityClient(kibana_client)

async get_all_roles(*, replace_deprecated_privileges=None)

Get all Kibana roles.

Retrieves every role, including reserved and system roles, with their Elasticsearch and Kibana privilege definitions.

Parameters:

replace_deprecated_privileges (bool | None) – If True and the response contains any privileges that are associated with deprecated features, they are omitted in favor of details about the appropriate replacement feature privileges.

Returns:

ObjectApiResponse containing a list of role objects. Each role has name, description, metadata, elasticsearch (cluster/indices/run_as privileges), kibana (base/feature privileges per space) and transient_metadata.

Raises:

• AuthenticationException – If authentication fails.

• AuthorizationException – If insufficient privileges to read roles.

• TransportError – If unable to connect to Kibana.

Return type:

ObjectApiResponse[Any]

Example

>>> roles = await client.security.get_all_roles()
>>> for role in roles.body:
...     print(role["name"])
superuser
my-role

async get_role(*, name, replace_deprecated_privileges=None)

Get a Kibana role by name.

Parameters:

• name (str) – The role name (required).

• replace_deprecated_privileges (bool | None) – If True and the response contains any privileges that are associated with deprecated features, they are omitted in favor of details about the appropriate replacement feature privileges.

Returns:

ObjectApiResponse containing the role object with name, description, metadata, elasticsearch and kibana privilege definitions.

Raises:

• ValueError – If name is empty.

• NotFoundError – If the role does not exist.

• AuthenticationException – If authentication fails.

• AuthorizationException – If insufficient privileges to read roles.

Return type:

ObjectApiResponse[Any]

Example

>>> role = await client.security.get_role(name="my-role")
>>> print(role.body["elasticsearch"]["cluster"])
['monitor']

async create_or_update_role(*, name, elasticsearch, description=None, kibana=None, metadata=None, create_only=None)

Create a new Kibana role or update an existing one.

Parameters:

• name (str) – The role name. Must be 1-1024 characters (required).

• elasticsearch (dict[str, Any]) –

  Elasticsearch privileges for the role (required). Supported keys:

  • cluster: list of cluster privilege names.

  • indices: list of index privilege objects, each with names and privileges (and optionally field_security, query, allow_restricted_indices).

  • remote_cluster: list of objects with clusters and privileges for remote clusters.

  • remote_indices: list of remote index privilege objects (like indices plus clusters).

  • run_as: list of usernames the role can impersonate.

• description (str | None) – Optional description for the role (max 2048 chars).

• kibana (list[dict[str, Any]] | None) – Kibana privileges for the role. A list of objects, each with base (base privileges such as ["all"] or ["read"]), feature (a mapping of feature ID to a list of feature privileges; mutually exclusive with a non-empty base) and spaces (the space IDs the entry applies to, or ["*"] for all spaces).

• metadata (dict[str, Any] | None) – Optional arbitrary metadata to store with the role.

• create_only (bool | None) – If True, the request fails if a role with the given name already exists (create-only semantics). Defaults to False on the server (create or update).

Returns:

ObjectApiResponse with an empty body (the server returns 204 No Content on success).

Raises:

• ValueError – If name is empty.

• BadRequestError – If the privilege definitions are invalid.

• ConflictError – If create_only=True and the role already exists.

• AuthenticationException – If authentication fails.

• AuthorizationException – If insufficient privileges to manage roles.

Return type:

ObjectApiResponse[Any]

Example

>>> await client.security.create_or_update_role(
...     name="log-reader",
...     description="Read-only access to logs",
...     elasticsearch={
...         "cluster": [],
...         "indices": [
...             {"names": ["logs-*"], "privileges": ["read"]}
...         ],
...     },
...     kibana=[{"base": ["read"], "spaces": ["*"]}],
...     metadata={"version": 1},
... )

async delete_role(*, name)

Delete a Kibana role by name.

Reserved roles cannot be deleted.

Parameters:

name (str) – The role name (required).

Returns:

ObjectApiResponse with an empty body (the server returns 204 No Content on success).

Raises:

• ValueError – If name is empty.

• NotFoundError – If the role does not exist.

• AuthenticationException – If authentication fails.

• AuthorizationException – If insufficient privileges to manage roles.

Return type:

ObjectApiResponse[Any]

Example

>>> await client.security.delete_role(name="log-reader")

async bulk_create_or_update_roles(*, roles)

Create new Kibana roles or update existing ones in bulk.

Parameters:

roles (dict[str, dict[str, Any]]) – A mapping of role name to role definition (required). Each definition accepts the same keys as create_or_update_role(): elasticsearch (required by the server), kibana, description and metadata.

Returns:

ObjectApiResponse summarizing the outcome per role, with keys such as created, updated and errors.

Raises:

• BadRequestError – If a role definition is invalid.

• AuthenticationException – If authentication fails.

• AuthorizationException – If insufficient privileges to manage roles.

Return type:

ObjectApiResponse[Any]

Example

>>> result = await client.security.bulk_create_or_update_roles(
...     roles={
...         "role-a": {"elasticsearch": {"cluster": ["monitor"]}},
...         "role-b": {
...             "elasticsearch": {},
...             "kibana": [{"base": ["read"], "spaces": ["*"]}],
...         },
...     }
... )
>>> print(result.body.get("created"))
['role-a', 'role-b']

async query_roles(*, query=None, from_=None, size=None, sort=None, filters=None)

Query Kibana roles with optional filters, paging and sorting.

Parameters:

• query (str | None) – Free-text query string used to match role names.

• from – Zero-based offset of the first role to return (paging).

• size (int | None) – Maximum number of roles to return (paging).

• sort (dict[str, Any] | None) – Sort definition with required keys field and direction ("asc" or "desc"), e.g. {"field": "name", "direction": "asc"}.

• filters (dict[str, Any] | None) – Additional filters. Supports showReservedRoles (bool) to include or exclude reserved roles.

Returns:

ObjectApiResponse containing the matching roles, typically with roles (the page of role objects), count and total.

Raises:

• BadRequestError – If the query payload is invalid.

• AuthenticationException – If authentication fails.

• AuthorizationException – If insufficient privileges to read roles.

Return type:

ObjectApiResponse[Any]

Example

>>> result = await client.security.query_roles(
...     query="kbnpy",
...     from_=0,
...     size=10,
...     sort={"field": "name", "direction": "asc"},
...     filters={"showReservedRoles": False},
... )
>>> print(result.body["total"])
2

async invalidate_sessions(*, match, query=None)

Invalidate user sessions that match a query.

To use this API, you must be a superuser.

Warning

Calling this with match="all" logs out every user of the Kibana instance. Prefer match="query" with a narrow query to target specific sessions.

Parameters:

• match (str) – How Kibana determines which sessions to invalidate (required). "all" invalidates every existing session; "query" invalidates only the sessions that match query.

• query (dict[str, Any] | None) – The query used to match sessions when match is "query". An object with a required provider ({"type": ..., "name": ...}, type required — e.g. basic, token, saml, oidc, kerberos or pki) and an optional username.

Returns:

ObjectApiResponse with the number of successfully invalidated sessions in total.

Raises:

• BadRequestError – If the payload is invalid (e.g. match is not "all" or "query").

• AuthenticationException – If authentication fails.

• AuthorizationException – If the calling user is not a superuser.

Return type:

ObjectApiResponse[Any]

Example

>>> result = await client.security.invalidate_sessions(
...     match="query",
...     query={
...         "provider": {"type": "basic"},
...         "username": "some-user",
...     },
... )
>>> print(result.body["total"])
0

async perform_request(method, path, *, params=None, headers=None, body=None)

Perform an async HTTP request via the parent client with space context enhancement.

Parameters:

• method (str) – HTTP method (GET, POST, PUT, DELETE, etc.)

• path (str) – API endpoint path

• params (dict[str, Any] | None) – Query parameters

• headers (dict[str, str] | None) – Request headers

• body (dict[str, Any] | None) – Request body

Returns:

API response

Raises:

ApiError – If the API returns an error response (enhanced with space context)

Return type:

ObjectApiResponse[Any]