Skip to content

Move index management privilege from Kibana to ElasticSearch #581

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
byewokko merged 4 commits into from
Jun 11, 2026
Merged

Move index management privilege from Kibana to ElasticSearch #581

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
48fcd5c
Move index management from KibanaUtils to ElasticSearchIntegration, r…
byewokko Jun 11, 2026
3e256b7
Merge branch 'main' into feature/kibana-index-management-resource
byewokko Jun 11, 2026
b6e36ce
Chore: Update CHANGELOG.md
byewokko Jun 11, 2026
0435b3d
Style: Flake8
byewokko Jun 11, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
## v26.15

### Pre-releases
- v26.15-alpha10
- v26.15-alpha9
- v26.15-alpha8
- v26.15-alpha7
Expand All @@ -15,6 +16,7 @@
- v26.15-alpha

### Fixes
- Move index management privilege from Kibana to ElasticSearch (#581, v26.15-alpha10)
- Remove password reset script from the Docker container (#578, v26.15-alpha7)
- Treat HTTP authentication scheme as case-insensitive (#575, v26.15-alpha5)
- Update webauthn package (#572, v26.15-alpha3)
Expand Down
221 changes: 111 additions & 110 deletions seacatauth/batman/elasticsearch.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,7 +51,9 @@ class ElasticSearchIntegration(asab.config.Configurable):
# IDs of Elasticsearch resources
"elasticsearch_superuser_resource_id": ResourceId.SUPERUSER, # Superuser access to the entire Elasticsearch cluster

"elasticsearch_monitoring_resource_id": "elasticsearch:monitoring"
"elasticsearch_monitoring_resource_id": "elasticsearch:monitoring",

"elasticsearch_index_management_resource_id": "elasticsearch:index:manage", # Manage indices, index templates and lifecycle policies
}


Expand Down Expand Up @@ -98,6 +100,7 @@ def __init__(self, batman_svc, config_section_name="batman:elasticsearch", confi
self.TenantIndices = re.split(r"\s+", self.Config.get("tenant_indices"))
self.ElasticsearchSuperuserResourceId = self.Config.get("elasticsearch_superuser_resource_id")
self.MonitoringResourceId = self.Config.get("elasticsearch_monitoring_resource_id")
self.IndexManagementResourceId = self.Config.get("elasticsearch_index_management_resource_id")

self.SeacatUserFlagRole = self.Config.get("seacat_user_flag")
self.IgnoreUsernames = self._prepare_ignored_usernames()
Expand Down Expand Up @@ -179,6 +182,7 @@ async def _on_init(self, event_name):
event_name: The name of the event that triggered this handler
"""
await self._initialize_resources()
await self._upsert_index_management_role()
# Ensure sync on startup even if housekeeping does not happen; prevent syncing twice
if not asab.Config.getboolean("housekeeping", "run_at_startup"):
await self.full_sync()
Expand Down Expand Up @@ -393,6 +397,105 @@ async def _upsert_role_for_index_access(self, tenant_id: str, privileges: str =
return role_name


async def _upsert_index_management_role(self):
"""
Create or update the Elasticsearch index management role.
This role grants privileges needed for Index Management features including
Index Lifecycle Management (ILM), index templates, and index operations.
"""
role_name = "index_management"
required_role_settings = {
"cluster": ["monitor", "manage_index_templates", "manage_ilm"],
"indices": [
{
"names": ["*"],
"privileges": ["view_index_metadata", "manage"],
}
],
}

async with self._with_elasticsearch_nodes(
lambda session: session.get("_security/role/{}".format(role_name))
) as resp:
if resp.status == 200:
role_data = (await resp.json()).get(role_name)
elif resp.status == 404:
role_data = None
else:
text = await resp.text()
L.error(
"Failed to get ElasticSearch role:\n{}".format(text[:1000]),
struct_data={"code": resp.status, "role": role_name}
)
return

# Check if required privileges are present in role settings
def privileges_match(existing, required):
if not existing:
return False
for key, value in required.items():
if key == "indices":
# Check if indices privileges match
if not existing.get(key):
return False
for req_index in value:
found = False
for existing_index in existing[key]:
if (
set(existing_index.get("names", [])) >= set(req_index.get("names", []))
and set(existing_index.get("privileges", [])) >= set(req_index.get("privileges", []))
):
found = True
break
if not found:
return False
else:
if set(existing.get(key, [])) < set(value):
return False
return True

if role_data and privileges_match(role_data, required_role_settings):
return

# Update role with required privileges
if not role_data:
role_data = {}
role = {
"cluster": list(set(role_data.get("cluster", []) + required_role_settings["cluster"])),
"indices": role_data.get("indices", []),
"metadata": role_data.get("metadata", {}),
}
# Add or update the indices settings
for req_index in required_role_settings["indices"]:
existing_idx = None
for idx in role["indices"]:
if set(idx.get("names", [])) >= set(req_index["names"]):
existing_idx = idx
break
if existing_idx:
existing_idx["privileges"] = list(set(existing_idx.get("privileges", [])) | set(req_index["privileges"]))
else:
role["indices"].append(req_index)

async with self._with_elasticsearch_nodes(
lambda session: session.put("_security/role/{}".format(role_name), json=role)
) as resp:
if not (200 <= resp.status < 300):
text = await resp.text()
L.error(
"Failed to create/update ElasticSearch role:\n{}".format(text[:1000]),
struct_data={"code": resp.status, "role": role_name}
)
return
result = await resp.json()

created = result.get("role", {}).get("created")
if created is True:
L.info("ElasticSearch index management role created.", struct_data={"role": role_name})
else:
L.info("ElasticSearch index management role updated.", struct_data={"role": role_name})


async def _initialize_resources(self):
"""
Create Seacat Auth resources that are mapped to ElasticSearch and Kibana roles
Expand All @@ -410,6 +513,10 @@ async def _initialize_resources(self):
"Grants access to Elasticsearch Stack monitoring via Elasticsearch role 'monitoring_user'.",
"global_only": True,
},
self.IndexManagementResourceId: {
"description": "Grants access to Elasticsearch Index Management features for managing indices and index templates.",
"global_only": True,
},
}
if self.Kibana.is_enabled():
resources.update(self.Kibana.get_kibana_resources())
Expand Down Expand Up @@ -497,6 +604,8 @@ async def sync_credentials(self, cred: dict):
# Seacat superuser is mapped to Elasticsearch "superuser" role
if self.ElasticsearchSuperuserResourceId in authorized_resources:
elk_roles.add("superuser")
if self.IndexManagementResourceId in authorized_resources:
elk_roles.add("index_management")
continue

if self.MonitoringResourceId in authorized_resources:
Expand Down Expand Up @@ -580,7 +689,6 @@ class KibanaUtils(asab.config.Configurable):
"kibana_admin_resource_id": "tools:kibana:admin", # Admin access to all of Kibana
"kibana_default_space_read_resource_id": "tools:kibana:space:default:read", # Read-only access to "Default" space
"kibana_default_space_all_resource_id": "tools:kibana:space:default:all", # Read-write access to "Default" space
"kibana_index_management_resource_id": "tools:kibana:manage_indices", # Manage all indices
}

def __init__(self, app, config_section_name="batman:elasticsearch", config=None):
Expand Down Expand Up @@ -617,7 +725,6 @@ def __init__(self, app, config_section_name="batman:elasticsearch", config=None)
self.AdminResourceId = self.Config.get("kibana_admin_resource_id")
self.DefaultSpaceReadResourceId = self.Config.get("kibana_default_space_read_resource_id")
self.DefaultSpaceAllResourceId = self.Config.get("kibana_default_space_all_resource_id")
self.IndexManagementResourceId = self.Config.get("kibana_index_management_resource_id")


def is_enabled(self):
Expand Down Expand Up @@ -652,10 +759,6 @@ def get_kibana_resources(self):
"description": "Read-write access to the Default space in Kibana",
"global_only": True,
},
self.IndexManagementResourceId: {
"description": "Access to Kibana Index Management features for managing indices and index templates",
"global_only": True,
},
}


Expand All @@ -675,8 +778,6 @@ def get_kibana_roles_by_authz(self, authz: dict):
roles.add(get_default_space_access_role_name("read"))
if self.DefaultSpaceAllResourceId in authorized_resources:
roles.add(get_default_space_access_role_name("all"))
if self.IndexManagementResourceId in authorized_resources:
roles.add("kibana_index_management")
continue
if self.ReadResourceId in authorized_resources:
roles.add(get_space_access_role_name(tenant_id, "read"))
Expand Down Expand Up @@ -962,117 +1063,17 @@ async def upsert_role_for_default_space_access(self, privileges: str = "read"):
await self._upsert_role_for_kibana_space(space_id, role_name, privileges)


async def _upsert_index_management_role(self):
"""
Create or update the Kibana index management role in ElasticSearch.
This role grants privileges needed for Kibana Index Management features.
"""
assert self.is_enabled()

role_name = "kibana_index_management"
required_elasticsearch_privileges = {
"cluster": ["monitor", "manage_index_templates", "manage_ilm"],
"indices": [
{
"names": ["*"],
"privileges": ["view_index_metadata", "manage"],
}
],
}

async with self._kibana_session() as session:
async with session.get("{}/api/security/role/{}".format(self.KibanaUrl, role_name)) as resp:
if resp.status == 200:
role_data = await resp.json()
elif resp.status == 404:
role_data = None
else:
text = await resp.text()
L.error("Failed to get ElasticSearch role:\n{}".format(text[:1000]), struct_data={
"role": role_name})
return

# Check if required privileges are present in role settings
def elasticsearch_privileges_match(existing_es, required_es):
if not existing_es:
return False
for key, value in required_es.items():
if key == "indices":
# Check if indices privileges match
if not existing_es.get(key):
return False
for req_index in value:
found = False
for existing_index in existing_es[key]:
if (
set(existing_index.get("names", [])) >= set(req_index.get("names", []))
and set(existing_index.get("privileges", [])) >= set(req_index.get("privileges", []))
):
found = True
break
if not found:
return False
else:
if set(existing_es.get(key, [])) < set(value):
return False
return True

existing_es = role_data.get("elasticsearch", {}) if role_data else None
if existing_es and elasticsearch_privileges_match(existing_es, required_elasticsearch_privileges):
return

# Update role with required privileges
if not role_data:
role_data = {}
existing_es = role_data.get("elasticsearch", {})
es_privileges = {
"cluster": list(set(existing_es.get("cluster", []) + required_elasticsearch_privileges["cluster"])),
"indices": existing_es.get("indices", []),
}
# Add or update the indices settings
for req_index in required_elasticsearch_privileges["indices"]:
existing_idx = None
for idx in es_privileges["indices"]:
if set(idx.get("names", [])) >= set(req_index["names"]):
existing_idx = idx
break
if existing_idx:
existing_idx["privileges"] = list(set(existing_idx.get("privileges", [])) | set(req_index["privileges"]))
else:
es_privileges["indices"].append(req_index)

role = {
"metadata": role_data.get("metadata", {}),
"elasticsearch": es_privileges,
}

async with self._kibana_session() as session:
async with session.put(
"{}/api/security/role/{}".format(self.KibanaUrl, role_name), json=role
) as resp:
if not (200 <= resp.status < 300):
text = await resp.text()
L.error("Failed to update role {!r} with index management privileges:\n{}".format(
role_name, text[:1000]))
return

L.info("Added index management privileges to Kibana role.", struct_data={"role": role_name})


async def sync_all_spaces_and_roles(self):
"""
Synchronize all Kibana spaces and roles.
Creates roles for Default space access, index management, and syncs all tenant spaces.
Creates roles for Default space access and syncs all tenant spaces.
"""
assert self.is_enabled()

# Sync roles for Default space access (global-only resources)
for privileges in {"read", "all"}:
await self.upsert_role_for_default_space_access(privileges)

# Sync index management role
await self._upsert_index_management_role()

async for tenant in self.TenantService.iterate():
await self.sync_space_and_roles(tenant)

Expand Down
Loading