forked from getredash/redash
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathscript.py
More file actions
87 lines (66 loc) · 2.28 KB
/
script.py
File metadata and controls
87 lines (66 loc) · 2.28 KB
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
import os
import subprocess
import sys
from redash.query_runner import *
def query_to_script_path(path, query):
if path != "*":
script = os.path.join(path, query.split(" ")[0])
if not os.path.exists(script):
raise IOError("Script '{}' not found in script directory".format(query))
return os.path.join(path, query).split(" ")
return query
def run_script(script, shell):
output = subprocess.check_output(script, shell=shell)
if output is None:
return None, "Error reading output"
output = output.strip()
if not output:
return None, "Empty output from script"
return output, None
class Script(BaseQueryRunner):
@classmethod
def annotate_query(cls):
return False
@classmethod
def enabled(cls):
return "check_output" in subprocess.__dict__
@classmethod
def configuration_schema(cls):
return {
'type': 'object',
'properties': {
'path': {
'type': 'string',
'title': 'Scripts path'
},
'shell': {
'type': 'boolean',
'title': 'Execute command through the shell'
}
},
'required': ['path']
}
@classmethod
def type(cls):
return "insecure_script"
def __init__(self, configuration):
super(Script, self).__init__(configuration)
# If path is * allow any execution path
if self.configuration["path"] == "*":
return
# Poor man's protection against running scripts from outside the scripts directory
if self.configuration["path"].find("../") > -1:
raise ValueError("Scripts can only be run from the configured scripts directory")
def test_connection(self):
pass
def run_query(self, query, user):
try:
script = query_to_script_path(self.configuration["path"], query)
return run_script(script, self.configuration['shell'])
except IOError as e:
return None, e.message
except subprocess.CalledProcessError as e:
return None, str(e)
except KeyboardInterrupt:
return None, "Query cancelled by user."
register(Script)