in helper-scripts/export_kibana_config.py [0:0]
def fetch_kibana_object(self, obj_type):
"""
Get saved object from Kibana API
"""
try:
print(f"# Fetching kibana objects: {obj_type}")
response = requests.post(
KIBANA_OBJECTS_EXPORT_URL,
json={"type": obj_type},
verify=False,
auth=(self.kibana_user, self.kibana_password),
headers={"kbn-xsrf": "true"},
)
if response.status_code != 200:
print(
f"!!! Error fetching kibana object {obj_type}: HTTP status code {response.status_code}"
)
else:
raw_data = response.text.encode("utf-8")
items = ndjson.loads(raw_data)
if obj_type != "index-pattern":
to_export = []
for ip in items: # pylint: disable=invalid-name
if "attributes" in ip.keys() and "title" in ip["attributes"]:
if re.match(
REDELK_OBJ_FILTER,
ip["attributes"]["title"],
re.IGNORECASE,
):
ip.pop("updated_at", None)
ip["version"] = "1"
to_export.append(ip)
export_file = os.path.join(
self.export_path,
f"{EXPORT_FILES_PREFIX_KIBANA}{obj_type}.ndjson",
)
print(f"\tExporting {obj_type}: {export_file}")
with open(export_file, "w", encoding="utf-8") as obj_file:
ndjson.dump(to_export, obj_file)
else:
for ip in items: # pylint: disable=invalid-name
if "attributes" in ip.keys() and "title" in ip["attributes"]:
if re.match(
INDEX_PATTERNS_FILTER,
ip["attributes"]["title"],
re.IGNORECASE,
):
# print('%s: %s' % (obj_type,ip['attributes']['title']))
index_pattern_name = (
ip["attributes"]["title"][:-2]
if ip["attributes"]["title"].endswith("-*")
else ip["attributes"]["title"]
)
ip.pop("updated_at", None)
ip["version"] = "1"
export_file = os.path.join(
self.export_path,
f"{EXPORT_FILES_PREFIX_KIBANA}{obj_type}_{index_pattern_name}.ndjson",
)
print(f"\tExporting {obj_type}: {export_file}")
with open(
export_file, "w", encoding="utf-8"
) as obj_file:
ndjson.dump([ip], obj_file)
except Exception as error: # pylint: disable=broad-except
print(f"!!! Error fetching kibana object {obj_type}: {error}")