-
Notifications
You must be signed in to change notification settings - Fork 57
/
export_dashboards.py
165 lines (131 loc) · 5.58 KB
/
export_dashboards.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
#!/usr/bin/env python3
import argparse
import base64
import json
import os
import re
import requests
from pathlib import Path
from urllib3.exceptions import InsecureRequestWarning
# Suppress the InsecureRequestWarning (We are using a self-signed cert)
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
ALL = 'all'
class Api:
def __init__(self, args):
self.ids = None
self.basic_auth = self.get_basic_auth(args.user, args.password)
self.root_url = f'https://{args.host}:{args.port}'
def export_dashboards(self):
self.set_ids()
self.export_selected_dashboard(self.select_dashboard())
@staticmethod
def get_basic_auth(username, password):
return base64.b64encode(f"{username}:{password}".encode()).decode()
def get_ids(self):
url = f'{self.root_url}/api/kibana/management/saved_objects/_find?perPage=500&page=1&fields=id&type=dashboard&sortField=updated_at&sortOrder=desc'
try:
response = requests.get(url, headers={'Authorization': f'Basic {self.basic_auth}'}, verify=False)
if response.status_code == 200:
data = response.json()
ids = {item['id']: item['meta']['title'] for item in data.get('saved_objects', [])}
return ids
else:
print(f"HTTP request failed with status code: {response.status_code}")
print(response.text)
return {}
except Exception as e:
print(f"An error occurred: {str(e)}")
return {}
def set_ids(self, ids=None):
if ids is None:
ids = self.get_ids()
self.ids = ids
def select_dashboard(self):
print("Please select a dashboard ID:")
item = 1
choices = {}
# Iterate through ids and display them with corresponding numbers
for this_id, title in self.ids.items():
print(item, this_id, title)
choices[item] = this_id
item += 1
if item == 1:
print("I could not find any dashboards")
return
choices[item] = ALL
print(item, "Select all dashboards")
# Ask the user to select a number
while True:
try:
choice = int(input("Select a number: "))
if choice in choices:
selected_id = choices[choice]
if selected_id == ALL:
return ALL # Return 'all' if the user selects all dashboards
else:
return selected_id # Return the selected dashboard ID
else:
print("Invalid choice. Please select a valid number.")
except ValueError:
print("Invalid input. Please enter a number.")
def export_selected_dashboard(self, selected_dashboard):
if selected_dashboard == ALL:
print("You selected to export all dashboards")
self.dump_all_dashboards()
else:
print(f"You selected dashboard ID: {selected_dashboard}")
self.dump_dashboard(selected_dashboard)
def dump_dashboard(self, selected_id):
print(f"Dumping dashboard: {selected_id}: {self.ids[selected_id]}...")
# Dumping dashboard: e5f203f0-6182-11ee-b035-d5f231e90733: User Security
dashboard_json = self.get_dashboard_json(selected_id)
if dashboard_json is not None:
script_dir = os.path.dirname(os.path.abspath(__file__))
export_path = Path(script_dir) / 'exported'
os.makedirs(export_path, exist_ok=True)
filename = re.sub(r"\W+", "_", self.ids[selected_id].lower()) + ".dumped.ndjson"
print(f"Writing to file {filename}")
export_path = export_path / filename
Api.write_to_file(export_path, dashboard_json)
return
print("There was a problem dumping the dashboard")
def dump_all_dashboards(self):
for this_id in self.ids:
self.dump_dashboard(this_id)
def get_dashboard_json(self, selected_id):
url = f'{self.root_url}/api/saved_objects/_export'
data = {
"objects": [{"id": selected_id, "type": "dashboard"}],
"includeReferencesDeep": True
}
headers = {
"kbn-xsrf": "true",
'Authorization': f'Basic {self.basic_auth}'
}
try:
response = requests.post(url, headers=headers, json=data, verify=False)
if response.status_code == 200:
return response.text
else:
print(f"HTTP request failed with status code: {response.status_code}")
print(response.text)
return None
except Exception as e:
print(f"An error occurred: {str(e)}")
return None
@staticmethod
def write_to_file(filename, content):
with open(filename, 'wb') as file:
file.write(content.encode('utf-8'))
def main():
# Define command-line arguments with defaults
parser = argparse.ArgumentParser(description='Retrieve IDs from Elasticsearch')
parser.add_argument('-u', '--user', required=True, help='Elasticsearch username')
parser.add_argument('-p', '--password', required=True, help='Elasticsearch password')
parser.add_argument('--host', default='localhost', help='Elasticsearch host (default: localhost)')
parser.add_argument('--port', default='443', help='Elasticsearch port (default: 443)')
args = parser.parse_args()
api = Api(args)
api.export_dashboards()
if __name__ == '__main__':
main()