Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def handle_api_config(self, http_context):
if os.getuid() != 0:
raise EndpointReturn(403)
if http_context.method == 'GET':
with authorize('core:config:read'):
self.context.worker.reload_master_config()
return aj.config.data
if http_context.method == 'POST':
with authorize('core:config:write'):
data = json.loads(http_context.body)
aj.config.data.update(data)
aj.config.save()
self.context.worker.reload_master_config()
return aj.config.data
@authorize('terminal:scripts')
@endpoint(api=True)
def handle_script(self, http_context):
data = http_context.json_body()
try:
p = subprocess.Popen(
['bash', '-c', data['script']],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
close_fds=True
)
o, e = p.communicate(data.get('input', None))
except subprocess.CalledProcessError as e:
raise EndpointError(e)
except OSError as e:
raise EndpointError(e)
return {
@authorize('services:manage')
@endpoint(api=True)
def handle_api_operate(self, http_context, manager_id=None, operation=None, service_id=None):
if operation not in ['start', 'stop', 'restart']:
return
try:
getattr(self.managers[manager_id], operation)(service_id)
except ServiceOperationError as e:
raise EndpointError(e)
@authorize('network:updown')
@endpoint(api=True)
def handle_api_downup(self, http_context, iface=None):
self.manager.down(iface)
self.manager.up(iface)
@authorize('filesystem:write')
@endpoint(api=True)
def handle_api_fs_chmod(self, http_context, path=None):
if not os.path.exists(path):
raise EndpointReturn(404)
data = json.loads(http_context.body)
try:
os.chmod(path, data['mode'])
except OSError as e:
raise EndpointError(e)
@authorize('network:updown')
@endpoint(api=True)
def handle_api_down(self, http_context, iface=None):
return self.manager.down(iface)
@authorize('power:manage')
@endpoint(api=True)
def handle_api_reboot(self, http_context):
self.manager.reboot()
@authorize('filesystem:write')
@endpoint(api=True)
def handle_api_fs_write(self, http_context, path=None):
try:
content = http_context.body
if http_context.query:
encoding = http_context.query.get('encoding', None)
if encoding:
content = content.decode('utf-8').encode(encoding)
with open(path, 'w') as f:
f.write(content)
except OSError as e:
raise EndpointError(e)
@authorize('filesystem:write')
@endpoint(api=True)
def handle_api_fs_finish_upload(self, http_context):
name = http_context.json_body()['name']
path = http_context.json_body()['path']
id = http_context.json_body()['id']
chunk_dir = '/tmp/upload-%s' % id
target = os.path.join(path, name.replace('/', ''))
with open(target, 'wb') as f:
for i in range(len(os.listdir(chunk_dir))):
f.write(open(os.path.join(chunk_dir, str(i + 1))).read())
shutil.rmtree(chunk_dir)
return target
@authorize('power:manage')
@endpoint(api=True)
def handle_api_poweroff(self, http_context):
self.manager.poweroff()