mirror of
https://github.com/xcat2/confluent.git
synced 2026-09-05 20:47:57 +00:00
Markus Hilger
8a3fce85c0
Every one of these raises NameError if its code path is reached: - nodeapply: run_automation accumulated into an exitcode that only existed in run(), so any automation error crashed instead of being reported. It now keeps and returns its own, tracked separately from the exit code of the ssh commands: the early exit after the spawn loop tests that one, and folding automation failures into it would exit with children already running and their pipes abandoned. Both are reported at the real exits. - nodeconsole: redraw() reads firstnodename, which was local to do_screenshot(); promote it to a module global like the other drawing state. - nodedeploy: the redeploy path appended to a lockednodes list that did not exist yet. The block that follows re-reads the same lock state and acts on it, so drop the dead duplicate. - samples/nodeattrib_from_switch.py, misc/filterpasswd: missing import sys. - xcc3: fixuuid was never imported. xcc imports xcc3, so take a local copy the way the smm handler does instead of creating an import cycle. - httpapi: the async session call still passed the WSGI-era env and an extra argument to handle_async(), which has taken only querydict since the aiohttp port. Calling it correctly exposed that handle_async() registers an AsyncSession before raising on the discontinued long poll path, so every request to it would leak a session that is never reaped. It now only creates one when there is a websocket handler to yield it to. - messages: the InputFirmwareUpdate.filename property checked self.filebynode[node] with no node in scope. __init__ already validates every expanded path and nodefile() rechecks per node, so drop the checks. - pam: drop the python2 branches referencing unicode and raw_input. The server has been python3 only since the asyncio port. - cooltera: the sensor-name listing referenced a nonexistent sensors dict. The available sensors depend on the model, which is only known after reading the device, so list them from the same status data the readings use. - deltapdu, eatonpdu, geist: the not-implemented response in update() used node outside the loop, unlike retrieve() in the same files and unlike raritan/enlogic. - confluentdbgcli: stray self. on a module-level socket connect.
1267 lines
54 KiB
Python
1267 lines
54 KiB
Python
# vim: tabstop=4 shiftwidth=4 softtabstop=4
|
|
|
|
# Copyright 2014 IBM Corporation
|
|
# Copyright 2015-2019 Lenovo
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
# you may not use this file except in compliance with the License.
|
|
# You may obtain a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
# See the License for the specific language governing permissions and
|
|
# limitations under the License.
|
|
# This SCGI server provides a http wrap to confluent api
|
|
# It additionally manages httprequest console sessions
|
|
import base64
|
|
import shutil
|
|
|
|
import aiohttp
|
|
try:
|
|
import Cookie
|
|
except ModuleNotFoundError:
|
|
import http.cookies as Cookie
|
|
try:
|
|
import confluent.webauthn as webauthn
|
|
except ImportError:
|
|
webauthn = None
|
|
import asyncio
|
|
from aiohttp import web, WSMsgType
|
|
import confluent.auth as auth
|
|
import confluent.config.attributes as attribs
|
|
import confluent.config.configmanager as configmanager
|
|
import confluent.consoleserver as consoleserver
|
|
import confluent.discovery.core as disco
|
|
import confluent.discovery.protocols.pxe as pxe
|
|
import confluent.forwarder as forwarder
|
|
import confluent.exceptions as exc
|
|
import confluent.log as log
|
|
import confluent.messages
|
|
import confluent.core as pluginapi
|
|
import confluent.asynchttp
|
|
import confluent.selfservice as selfservice
|
|
import confluent.shellserver as shellserver
|
|
import confluent.asynctlvdata as tlvdata
|
|
import confluent.util as util
|
|
import copy
|
|
import json
|
|
import os
|
|
import socket
|
|
import sys
|
|
import traceback
|
|
import time
|
|
try:
|
|
import urlparse
|
|
except ModuleNotFoundError:
|
|
import urllib.parse as urlparse
|
|
|
|
|
|
_cleaner = None
|
|
|
|
auditlog = None
|
|
tracelog = None
|
|
consolesessions = {}
|
|
confluent.asynchttp.set_console_sessions(consolesessions)
|
|
httpsessions = {}
|
|
opmap = {
|
|
'POST': 'create',
|
|
'GET': 'retrieve',
|
|
'PUT': 'update',
|
|
'DELETE': 'delete',
|
|
}
|
|
|
|
|
|
def get_user_for_session(sessionid, sessiontok):
|
|
if not isinstance(sessionid, str):
|
|
sessionid = sessionid.decode()
|
|
if not isinstance(sessiontok, str):
|
|
sessiontok = sessiontok.decode()
|
|
if not sessiontok or not sessionid:
|
|
raise Exception("invalid session id or token")
|
|
if sessiontok != httpsessions.get(sessionid, {}).get('csrftoken', None):
|
|
raise Exception("Invalid csrf token for session")
|
|
user = httpsessions[sessionid]['name']
|
|
if not isinstance(user, str):
|
|
user = user.decode()
|
|
return user
|
|
|
|
def group_creation_resources():
|
|
yield confluent.messages.Attributes(
|
|
kv={'name': None}, desc="Name of the group").html() + '<br>'
|
|
yield confluent.messages.ListAttributes(kv={'nodes': []},
|
|
desc='Nodes to add to the group'
|
|
).html() + '<br>\n'
|
|
for attr in sorted(attribs.node):
|
|
if attr == 'groups':
|
|
continue
|
|
if attr.startswith('secret.') or attr.startswith('custom.nodesecret.'):
|
|
yield confluent.messages.CryptedAttributes(
|
|
kv={attr: None},
|
|
desc=attribs.node[attr]['description']).html() + '<br>\n'
|
|
elif ('type' in attribs.node[attr] and
|
|
list == attribs.node[attr]['type']):
|
|
yield confluent.messages.ListAttributes(
|
|
kv={attr: []},
|
|
desc=attribs.node[attr]['description']).html() + '<br>\n'
|
|
else:
|
|
yield confluent.messages.Attributes(
|
|
kv={attr: None},
|
|
desc=attribs.node[attr]['description']).html() + '<br>\n'
|
|
|
|
|
|
def node_creation_resources():
|
|
yield confluent.messages.Attributes(
|
|
kv={'name': None}, desc="Name of the node").html() + '<br>'
|
|
for attr in sorted(attribs.node):
|
|
if attr.startswith('secret.') or attr.startswith('custom.nodesecret.'):
|
|
yield confluent.messages.CryptedAttributes(
|
|
kv={attr: None},
|
|
desc=attribs.node[attr]['description']).html() + '<br>\n'
|
|
elif ('type' in attribs.node[attr] and
|
|
list == attribs.node[attr]['type']):
|
|
yield confluent.messages.ListAttributes(
|
|
kv={attr: []},
|
|
desc=attribs.node[attr]['description']).html() + '<br>\n'
|
|
else:
|
|
yield confluent.messages.Attributes(
|
|
kv={attr: None},
|
|
desc=attribs.node[attr]['description']).html() + '<br>\n'
|
|
|
|
|
|
def user_creation_resources():
|
|
credential = {
|
|
'uid': {
|
|
'description': (''),
|
|
},
|
|
'username': {
|
|
'description': (''),
|
|
},
|
|
'password': {
|
|
'description': (''),
|
|
},
|
|
'privilege_level': {
|
|
'description': (''),
|
|
},
|
|
}
|
|
for attr in sorted(credential):
|
|
if attr == "password":
|
|
yield confluent.messages.CryptedAttributes(
|
|
kv={attr: None},
|
|
desc=credential[attr]['description']).html() + '<br>\n'
|
|
else:
|
|
yield confluent.messages.Attributes(
|
|
kv={attr: None},
|
|
desc=credential[attr]['description']).html() + '<br>\n'
|
|
|
|
|
|
create_resource_functions = {
|
|
'nodes': node_creation_resources,
|
|
'groups': group_creation_resources,
|
|
'users': user_creation_resources,
|
|
}
|
|
|
|
|
|
async def _sessioncleaner():
|
|
while True:
|
|
currtime = time.time()
|
|
targsessions = []
|
|
for session in httpsessions:
|
|
if httpsessions[session]['expiry'] < currtime:
|
|
targsessions.append(session)
|
|
for session in targsessions:
|
|
forwarder.close_session(session)
|
|
del httpsessions[session]
|
|
targsessions = []
|
|
for session in consolesessions:
|
|
if consolesessions[session]['expiry'] < currtime:
|
|
targsessions.append(session)
|
|
for session in targsessions:
|
|
del consolesessions[session]
|
|
await asyncio.sleep(10)
|
|
|
|
|
|
def _get_query_dict(req, reqbody, reqtype):
|
|
qdict = {}
|
|
qdict.update(req.rel_url.query)
|
|
if reqbody is not None:
|
|
if "application/x-www-form-urlencoded" in reqtype:
|
|
if not isinstance(reqbody, str):
|
|
reqbody = reqbody.decode('utf8')
|
|
pbody = urlparse.parse_qs(reqbody, True)
|
|
for ky in pbody:
|
|
if len(pbody[ky]) > 1: # e.g. REST explorer
|
|
na = [i for i in pbody[ky] if i != '']
|
|
qdict[ky] = na
|
|
else:
|
|
qdict[ky] = pbody[ky][0]
|
|
elif 'application/json' in reqtype:
|
|
if not isinstance(reqbody, str):
|
|
reqbody = reqbody.decode('utf8')
|
|
pbody = json.loads(reqbody)
|
|
for key in pbody:
|
|
qdict[key] = pbody[key]
|
|
if 'restexplorerhonorkey' in qdict:
|
|
nqdict = {}
|
|
for key in qdict:
|
|
if key == 'restexplorerop':
|
|
nqdict[key] = qdict['restexplorerop']
|
|
continue
|
|
if key in qdict['restexplorerhonorkey']:
|
|
nqdict[key] = qdict[key]
|
|
qdict = nqdict
|
|
return qdict
|
|
|
|
def _should_skip_authlog(req):
|
|
thepath = req.rel_url.path
|
|
if ('/console/session' in thepath or
|
|
'/shell/sessions/' in thepath):
|
|
# we should only log starting of a console
|
|
return True
|
|
if '/sessions/current/async' in thepath:
|
|
# this is effectively invisible
|
|
return True
|
|
if '/sessions/current/webauthn/registered_credentials' in thepath:
|
|
return True
|
|
if (req.method == 'GET' and
|
|
('/sensors/' in thepath or
|
|
'/health/' in thepath or
|
|
'/power/state' in thepath or
|
|
'/nodes/' == thepath or
|
|
'/sessions/current/info' == thepath or
|
|
(thepath.startswith('/noderange/') and
|
|
thepath.endswith('/nodes/')))):
|
|
# these are pretty innocuous, and noisy to log.
|
|
return True
|
|
return False
|
|
|
|
|
|
def _csrf_exempt(path):
|
|
# first a get of info to get CSRF key, also '/forward/web' to enable
|
|
# the popup ability to just forward
|
|
return path == '/sessions/current/info' or path.endswith('/forward/web')
|
|
|
|
|
|
def _csrf_valid(req, session):
|
|
# This could be simplified into a statement, but this is more readable
|
|
# to have it broken out
|
|
if req.method == 'GET' and _csrf_exempt(req.rel_url.path):
|
|
# Provide a web client a safe hook to request the CSRF token
|
|
# This means that we consider GET of /sessions/current/info to be
|
|
# a safe thing to inflict via CSRF, since CORS should prevent
|
|
# hypothetical attacker from reading the data and it has no
|
|
# side effects to speak of
|
|
return True
|
|
if 'csrftoken' not in session:
|
|
# The client has not (yet) requested CSRF protection
|
|
# so we return true
|
|
if 'Confluentauthtoken' in req.headers:
|
|
# The client has requested CSRF countermeasures,
|
|
# oblige the request and apply a new token to the
|
|
# session
|
|
session['csrftoken'] = util.randomstring(32)
|
|
elif 'Referer' in req.headers:
|
|
print('refererrrrrrrr')
|
|
# If there is a referrer, make sure it stays consistent
|
|
# across the session. A change in referer is a bad thing
|
|
try:
|
|
referer = req.headers['Referer'].split('/')[2]
|
|
except IndexError:
|
|
return False
|
|
if 'validreferer' not in session:
|
|
session['validreferer'] = referer
|
|
elif session['validreferer'] != referer:
|
|
return False
|
|
return True
|
|
# The session has CSRF protection enabled, only mark valid if
|
|
# the client has provided an auth token and that token matches the
|
|
# value protecting the session
|
|
return ('Confluentauthtoken' in req.headers and
|
|
req.headers['ConfluentAuthToken'] == session['csrftoken'])
|
|
|
|
|
|
async def _authorize_request(req, operation, reqbody):
|
|
"""Grant/Deny access based on data from request
|
|
|
|
"""
|
|
authdata = None
|
|
name = ''
|
|
sessionid = None
|
|
sessid = None
|
|
cookie = Cookie.SimpleCookie()
|
|
element = req.rel_url.path
|
|
if element.startswith('/sessions/current/'):
|
|
if (element.startswith('/sessions/current/webauthn/registered_credentials/')
|
|
or element.startswith('/sessions/current/webauthn/validate/')):
|
|
name = element.rsplit('/')[-1]
|
|
baseelement = element.rsplit(name, 1)[0]
|
|
if baseelement not in ['/sessions/current/webauthn/registered_credentials/',
|
|
'/sessions/current/webauthn/validate/']:
|
|
raise exc.InvalidArgumentException('Invalid username for passkey operation')
|
|
authdata = auth.authorize(name, element=element, operation=operation)
|
|
else:
|
|
element = None
|
|
if not authdata:
|
|
if 'ConfluentSession' in req.headers:
|
|
sessionid = req.headers['ConfluentSession']
|
|
sessid = sessionid
|
|
elif 'confluentsessionid' in req.cookies:
|
|
sessionid = req.cookies['confluentsessionid']
|
|
sessid = sessionid
|
|
if sessionid:
|
|
if sessionid in httpsessions:
|
|
if _csrf_valid(req, httpsessions[sessionid]):
|
|
if req.rel_url.path == '/sessions/current/logout':
|
|
targets = []
|
|
for mythread in httpsessions[sessionid]['inflight']:
|
|
targets.append(mythread)
|
|
for mythread in targets:
|
|
print(repr(mythread))
|
|
forwarder.close_session(sessionid)
|
|
del httpsessions[sessionid]
|
|
return ('logout',)
|
|
httpsessions[sessionid]['expiry'] = time.time() + 90
|
|
name = httpsessions[sessionid]['name']
|
|
authdata = auth.authorize(
|
|
name, element=element, operation=operation,
|
|
skipuserobj=httpsessions[sessionid]['skipuserobject'])
|
|
|
|
if (not authdata) and 'Authorization' in req.headers:
|
|
if req.rel_url.path == '/sessions/current/logout':
|
|
if 'Referer' in req.headers:
|
|
# note that this doesn't actually do harm
|
|
# otherwise, but this way do not give appearance
|
|
# of something having a side effect if it has the smell
|
|
# of a CSRF
|
|
return {'code': 401}
|
|
return ('logout',)
|
|
if req.headers['Authorization'].startswith('MultiBasic '):
|
|
name, passphrase = base64.b64decode(
|
|
req.headers['Authorization'].replace('MultiBasic ', '')).split(b':', 1)
|
|
passphrase = json.loads(passphrase)
|
|
else:
|
|
name, passphrase = base64.b64decode(
|
|
req.headers['Authorization'].replace('Basic ', '')).split(b':', 1)
|
|
try:
|
|
authdata = await auth.check_user_passphrase(name, passphrase, operation=operation, element=element)
|
|
except Exception as e:
|
|
if hasattr(e, 'prompts'):
|
|
return {'code': 403, 'prompts': e.prompts}
|
|
raise
|
|
if authdata is False:
|
|
return {'code': 403}
|
|
elif not authdata:
|
|
return {'code': 401}
|
|
sessid = _establish_http_session(req, authdata, name, cookie)
|
|
if authdata and element and element.startswith('/sessions/current/webauthn/validate/'):
|
|
if not webauthn:
|
|
raise exc.NotFoundException('WebAuthn support not available')
|
|
rsp = await webauthn.handle_api_request(element, req, authdata[2], authdata[1], reqbody, None)
|
|
if rsp['verified']:
|
|
if rsp.get('username', None):
|
|
name = rsp['username']
|
|
authdata = (authdata[0], authdata[1], name, authdata[3], authdata[4])
|
|
sessid = _establish_http_session(req, authdata, name, cookie)
|
|
else:
|
|
return {'code': 403}
|
|
skiplog = _should_skip_authlog(req)
|
|
if authdata:
|
|
auditmsg = {
|
|
'user': util.stringify(name),
|
|
'operation': operation,
|
|
'target': req.rel_url.path,
|
|
}
|
|
authinfo = {'code': 200,
|
|
'cookie': cookie,
|
|
'cfgmgr': authdata[1],
|
|
'username': authdata[2],
|
|
'userdata': authdata[0]}
|
|
if authdata[3] is not None:
|
|
auditmsg['tenant'] = authdata[3]
|
|
authinfo['tenant'] = authdata[3]
|
|
auditmsg['user'] = util.stringify(authdata[2])
|
|
if sessid is not None:
|
|
authinfo['sessionid'] = sessid
|
|
if 'csrftoken' in httpsessions[sessid]:
|
|
authinfo['authtoken'] = httpsessions[sessid]['csrftoken']
|
|
httpsessions[sessid]['cfgmgr'] = authdata[1]
|
|
if not skiplog:
|
|
auditlog.log(auditmsg)
|
|
return authinfo
|
|
elif authdata is None:
|
|
return {'code': 401}
|
|
else:
|
|
return {'code': 403}
|
|
|
|
def _establish_http_session(req, authdata, name, cookie):
|
|
sessid = util.randomstring(32)
|
|
while sessid in httpsessions:
|
|
sessid = util.randomstring(32)
|
|
httpsessions[sessid] = {'name': name, 'expiry': time.time() + 90,
|
|
'skipuserobject': authdata[4],
|
|
'inflight': set([])}
|
|
if 'ConfluentAuthToken' in req.headers:
|
|
httpsessions[sessid]['csrftoken'] = util.randomstring(32)
|
|
cookie['confluentsessionid'] = util.stringify(sessid)
|
|
cookie['confluentsessionid']['secure'] = 1
|
|
cookie['confluentsessionid']['httponly'] = 1
|
|
cookie['confluentsessionid']['path'] = '/'
|
|
return sessid
|
|
|
|
|
|
def _pick_mimetype(req):
|
|
"""Detect the http indicated mime to send back.
|
|
|
|
Note that as it gets into the ACCEPT header honoring, it only looks for
|
|
application/json and else gives up and assumes html. This is because
|
|
browsers are very chaotic about ACCEPT HEADER. It is assumed that
|
|
XMLHttpRequest.setRequestHeader will be used by clever javascript
|
|
if the '.json' scheme doesn't cut it.
|
|
"""
|
|
if req.rel_url.path.endswith('.json'):
|
|
return 'application/json; charset=utf-8', '.json'
|
|
elif req.rel_url.path.endswith('.html'):
|
|
return 'text/html', '.html'
|
|
elif 'Accept' in req.headers and 'application/json' in req.headers['Accept']:
|
|
return 'application/json; charset=utf-8', ''
|
|
else:
|
|
return 'text/html', ''
|
|
|
|
|
|
def _assign_consessionid(consolesession):
|
|
sessid = util.randomstring(32)
|
|
while sessid in consolesessions:
|
|
sessid = util.randomstring(32)
|
|
consolesessions[sessid] = {'session': consolesession,
|
|
'expiry': time.time() + 60}
|
|
return sessid
|
|
|
|
|
|
def websockify_data(data):
|
|
if isinstance(data, dict):
|
|
data = json.dumps(data)
|
|
data = u'!' + data
|
|
else:
|
|
try:
|
|
data = data.decode('utf8')
|
|
except UnicodeDecodeError:
|
|
data = data.decode('cp437')
|
|
except AttributeError: # already str
|
|
pass
|
|
data = u' ' + data
|
|
return data
|
|
|
|
def datacallback_bound(clientsessid, rsp):
|
|
async def datacallback(data):
|
|
data = websockify_data(data)
|
|
try:
|
|
await rsp.send_str(u'${0}$'.format(clientsessid) + data)
|
|
except aiohttp.client_exceptions.ClientConnectionResetError:
|
|
raise exc.Disconnect("Client disconnected")
|
|
return datacallback
|
|
|
|
async def wsock_handler(req):
|
|
|
|
rsp = web.WebSocketResponse(
|
|
heartbeat=25.0,
|
|
protocols=('confluent.console', 'confluent.asyncweb'))
|
|
await rsp.prepare(req)
|
|
|
|
|
|
sessid = await rsp.receive()
|
|
if not sessid:
|
|
return
|
|
sessid = sessid.data
|
|
sessid = sessid.replace('ConfluentSessionId:', '')
|
|
sessid = sessid[:-1]
|
|
currsess = httpsessions.get(sessid, None)
|
|
if not currsess:
|
|
return
|
|
authtoken = await rsp.receive()
|
|
authtoken = authtoken.data
|
|
authtoken = authtoken.replace('ConfluentAuthToken:', '')
|
|
authtoken = authtoken[:-1]
|
|
if currsess['csrftoken'] != authtoken:
|
|
return
|
|
httpsessions[sessid]['inflight'].add(rsp)
|
|
name = httpsessions[sessid]['name']
|
|
authdata = auth.authorize(name, req.rel_url.path, operation='start')
|
|
if not authdata:
|
|
return
|
|
cfgmgr = httpsessions[sessid]['cfgmgr']
|
|
username = httpsessions[sessid]['name']
|
|
if req.rel_url.path == '/sessions/current/async':
|
|
myconsoles = {}
|
|
async def asyncwscallback(rspm):
|
|
rspm = json.dumps(rspm.raw())
|
|
await rsp.send_str(u'!' + rspm)
|
|
currsess['inflight'].add(rsp)
|
|
asess = None
|
|
try:
|
|
for asess in confluent.asynchttp.handle_async(
|
|
{}, asyncwscallback):
|
|
await rsp.send_str(u' ASYNCID: {0}'.format(asess.asyncid))
|
|
clientmsg = True
|
|
while clientmsg:
|
|
clientmsg = await rsp.receive()
|
|
if clientmsg.type == WSMsgType.CLOSE:
|
|
break
|
|
elif clientmsg.type != WSMsgType.TEXT:
|
|
continue
|
|
clientmsg = clientmsg.data
|
|
if clientmsg:
|
|
if clientmsg[0] == '?':
|
|
await rsp.send_str('?')
|
|
elif clientmsg[0] == '$':
|
|
targid, data = clientmsg[1:].split('$', 1)
|
|
if data[0] == ' ':
|
|
await myconsoles[targid].write(data[1:])
|
|
elif clientmsg[0] == '!':
|
|
msg = json.loads(clientmsg[1:])
|
|
action = msg.get('operation', None)
|
|
if not action:
|
|
action = msg.get('action', None)
|
|
targ = msg.get('target', None)
|
|
if targ:
|
|
authdata = auth.authorize(name, targ, operation=action)
|
|
if not authdata:
|
|
continue
|
|
if action == 'start':
|
|
if '/console/session' in targ or '/shell/sessions' in targ:
|
|
width = msg['width']
|
|
height = msg['height']
|
|
clientsessid = '{0}'.format(msg['sessid'])
|
|
skipreplay = msg.get('skipreplay', False)
|
|
delimit = None
|
|
if '/console/session' in targ:
|
|
delimit = '/console/session'
|
|
shellsession = False
|
|
else:
|
|
delimit = '/shell/sessions'
|
|
shellsession = True
|
|
nodesess = targ.split(delimit, 1)
|
|
node = nodesess[0]
|
|
sessidx = None
|
|
if len(nodesess) == 2 and len(nodesess[1]) > 1:
|
|
sessidx = nodesess[1][1:]
|
|
node = node.rsplit('/', 1)[-1]
|
|
auditmsg = {'operation': 'start', 'target': targ,
|
|
'user': util.stringify(username)}
|
|
auditlog.log(auditmsg)
|
|
datacallback = datacallback_bound(clientsessid, rsp)
|
|
if shellsession:
|
|
consession = await shellserver.ShellSession.create(
|
|
node=node, configmanager=cfgmgr,
|
|
username=username, skipreplay=skipreplay,
|
|
datacallback=datacallback,
|
|
width=width, height=height, sessionid=sessidx)
|
|
else:
|
|
consession = await consoleserver.ConsoleSession.create(
|
|
node=node, configmanager=cfgmgr,
|
|
username=username, skipreplay=skipreplay,
|
|
datacallback=datacallback,
|
|
width=width, height=height)
|
|
myconsoles[clientsessid] = consession
|
|
elif action == 'resize':
|
|
clientsessid = '{0}'.format(msg['sessid'])
|
|
myconsoles[clientsessid].resize(
|
|
width=msg['width'], height=msg['height'])
|
|
if action == 'break':
|
|
clientsessid = '{0}'.format(msg['sessid'])
|
|
await myconsoles[clientsessid].send_break()
|
|
elif action == 'stop':
|
|
sessid = '{0}'.format(msg.get('sessid', None))
|
|
if sessid in myconsoles:
|
|
await myconsoles[sessid].destroy()
|
|
del myconsoles[sessid]
|
|
else:
|
|
print(repr(clientmsg))
|
|
finally:
|
|
for cons in myconsoles:
|
|
await myconsoles[cons].destroy()
|
|
if asess:
|
|
await asess.destroy()
|
|
return rsp
|
|
path = req.rel_url.path
|
|
if '/console/session' in path or '/shell/sessions/' in path:
|
|
async def datacallback(data):
|
|
await rsp.send_str(websockify_data(data))
|
|
geom = await rsp.receive()
|
|
if geom.type != WSMsgType.TEXT:
|
|
return rsp
|
|
geom = geom.data[1:]
|
|
geom = json.loads(geom)
|
|
width = geom['width']
|
|
height = geom['height']
|
|
skipreplay = geom.get('skipreplay', False)
|
|
#hard bake JSON into this path, do not support other incarnations
|
|
if '/console/session' in path:
|
|
prefix, _, _ = path.partition('/console/session')
|
|
shellsession = False
|
|
elif '/shell/sessions/' in path:
|
|
prefix, _, _ = path.partition('/shell/sessions')
|
|
shellsession = True
|
|
_, _, nodename = prefix.rpartition('/')
|
|
|
|
try:
|
|
if shellsession:
|
|
consession = await shellserver.ShellSession.create(
|
|
node=nodename, configmanager=cfgmgr,
|
|
username=username, skipreplay=skipreplay,
|
|
datacallback=datacallback, width=width, height=height
|
|
)
|
|
else:
|
|
consession = await consoleserver.ConsoleSession.create(
|
|
node=nodename, configmanager=cfgmgr,
|
|
username=username, skipreplay=skipreplay,
|
|
datacallback=datacallback, width=width, height=height
|
|
)
|
|
except exc.NotFoundException:
|
|
return
|
|
clientmsg = await rsp.receive()
|
|
try:
|
|
while clientmsg.type == WSMsgType.TEXT:
|
|
clientmsg = clientmsg.data
|
|
if clientmsg[0] == ' ':
|
|
await consession.write(clientmsg[1:])
|
|
elif clientmsg[0] == '!':
|
|
cmd = json.loads(clientmsg[1:])
|
|
action = cmd.get('action', None)
|
|
if action == 'break':
|
|
await consession.send_break()
|
|
elif action == 'resize':
|
|
consession.resize(
|
|
width=cmd['width'], height=cmd['height'])
|
|
elif clientmsg[0] == '?':
|
|
await rsp.send_str(u'?')
|
|
clientmsg = await rsp.receive()
|
|
finally:
|
|
await consession.destroy()
|
|
return rsp
|
|
|
|
|
|
async def resourcehandler(request):
|
|
# start_response is akin to doing headers
|
|
# and calling 'prepare() on a 'StreamResponse'
|
|
# any 'yield' needs to become a write to the streamresponse
|
|
#TODO:asyncmerge: Replace /confluent-api with '' in path
|
|
# Needs testing for confluent header names with golang clients
|
|
async def make_response(mimetype='application/json', status=200, reason=None, headers=None, cookies=None, body=None):
|
|
rspheaders = {
|
|
'Cache-Control': 'no-store',
|
|
'Pragma': 'no-cache',
|
|
'X-Content-Type-Options': 'nosniff',
|
|
'Content-Security-Policy': "default-src 'self'",
|
|
'X-XSS-Protection': '1; mode=block',
|
|
'X-Frame-Options': 'deny',
|
|
'Strict-Transport-Security': 'max-age=86400',
|
|
'X-Permitted-Cross-Domain-Policies': 'none',
|
|
}
|
|
if headers:
|
|
rspheaders.update(headers)
|
|
rsp = web.StreamResponse(status=status, reason=reason, headers=rspheaders)
|
|
if cookies:
|
|
rsp.cookies.update(cookies)
|
|
rsp.content_type = mimetype
|
|
await rsp.prepare(request)
|
|
if body:
|
|
if isinstance(body, str):
|
|
body = body.encode('utf8')
|
|
await rsp.write(body)
|
|
return rsp
|
|
try:
|
|
if 'Sec-WebSocket-Version' in request.headers:
|
|
return await wsock_handler(request)
|
|
else:
|
|
return await resourcehandler_backend(request, make_response)
|
|
except Exception as e:
|
|
tracelog.log(traceback.format_exc(), ltype=log.DataTypes.event,
|
|
event=log.Events.stacktrace)
|
|
#start_response('500 - ' + str(e), [])
|
|
rsp = web.StreamResponse(status=500, reason=str(e))
|
|
await rsp.prepare(request)
|
|
await rsp.write(str(e).encode('utf8'))
|
|
return rsp
|
|
|
|
|
|
async def resourcehandler_backend(req, make_response):
|
|
"""Function to handle new wsgi requests
|
|
"""
|
|
mimetype, extension = _pick_mimetype(req)
|
|
|
|
reqbody = None
|
|
reqtype = None
|
|
reqpath = req.rel_url.path
|
|
if reqpath.startswith('/self/'):
|
|
return await selfservice.handle_request(req, make_response, mimetype)
|
|
if reqpath == '/httpapi_initialized':
|
|
if (len(configmanager.ConfigManager(None).list_usergroups()) > 0
|
|
or len(configmanager.ConfigManager(None).list_users()) > 0):
|
|
return await make_response(mimetype, 200, "OK")
|
|
return await make_response(mimetype, 500, "No authorized users")
|
|
if reqpath.startswith('/boot/'):
|
|
request = reqpath.split('/')
|
|
if not request[0]:
|
|
request = request[1:]
|
|
if request[1] == 'su': # shorturl
|
|
#TODO:asyncmerge: update with aiohttp behavior
|
|
targurl, can302, relurl, bootfilename = pxe.shorturls.get(request[2], (None, None, None, None))
|
|
if not targurl:
|
|
rsp = await make_response(mimetype, 404, 'Not Found')
|
|
await rsp.write(b'')
|
|
return rsp
|
|
if can302: # Maximum transparency helps iPXE and whatever else know the most
|
|
headers = {'Location': targurl}
|
|
rsp = await make_response(mimetype, 302, 'Found', headers=headers)
|
|
await rsp.write(b'')
|
|
return rsp
|
|
else: # The user agent is too dumb, check headers for server side redirects
|
|
delegatemethod = req.headers.get('X-Delegate-Method', None)
|
|
if delegatemethod == 'accel':
|
|
headers = {'Content-Type': 'application/octet-stream'}
|
|
headers['X-Accel-Redirect'] = relurl
|
|
rsp = await make_response(mimetype, 200, 'OK', headers=headers)
|
|
await rsp.write(b'')
|
|
return rsp
|
|
else:
|
|
rsp = await make_response(mimetype, 502, 'Bad Gateway')
|
|
await rsp.write(b'URL shortening for a limited client without proxy advertised accel support')
|
|
log.log({'error': f'Profile name exceeded DHCP limits, and reverse proxy capabilities not detected, switch to the nginx configuration or shorten the profile name: {relurl}'})
|
|
return rsp
|
|
if len(request) != 4:
|
|
return await make_response(mimetype, 400, 'Bad Request')
|
|
if request[1] == 'by-mac':
|
|
mac = request[2].replace('-', ':')
|
|
nodename = disco.get_node_by_uuid_or_mac(mac)
|
|
elif request[1] == 'by-uuid':
|
|
uuid = request[2]
|
|
nodename = disco.get_node_by_uuid_or_mac(uuid)
|
|
elif request[1] == 'by-node':
|
|
nodename = request[2]
|
|
bootfile = request[3]
|
|
cfg = configmanager.ConfigManager(None)
|
|
nodec = cfg.get_node_attributes(nodename, 'deployment.pendingprofile')
|
|
pprofile = nodec.get(nodename, {}).get('deployment.pendingprofile', {}).get('value', None)
|
|
if not pprofile:
|
|
return await make_response(mimetype, 404, 'Not Found')
|
|
redir = '/confluent-public/os/{0}/boot.{1}'.format(pprofile, bootfile)
|
|
rsp = await make_response(mimetype, 302, 'Found', {'Location': redir})
|
|
return rsp
|
|
if req.content_length and '/staging' not in reqpath:
|
|
reqbody = await req.read()
|
|
reqtype = req.content_type
|
|
operation = opmap.get(req.method, None)
|
|
if not operation:
|
|
rsp = await make_response(mimetype, 400, 'Bad Request')
|
|
await rsp.write(b'Unsupported method')
|
|
return rsp
|
|
querydict = _get_query_dict(req, reqbody, reqtype)
|
|
if operation != 'retrieve' and 'restexplorerop' in querydict:
|
|
operation = querydict['restexplorerop']
|
|
del querydict['restexplorerop']
|
|
try:
|
|
authorized = await _authorize_request(req, operation, reqbody)
|
|
except exc.NotFoundException as e:
|
|
rsp = await make_response(mimetype, 404, 'Not Found')
|
|
await rsp.write(e.get_error_body().encode('utf8'))
|
|
return rsp
|
|
if 'logout' in authorized:
|
|
rsp = await make_response("application/json", 200, 'Successful logout')
|
|
await rsp.write(b'{"result": "200 - Successful logout"}')
|
|
return rsp
|
|
if 'SuppressAuthHeader' in req.headers or 'ConfluentAuthToken' in req.headers:
|
|
badauth = {'Content-type': 'text/plain'}
|
|
else:
|
|
badauth = {'Content-type': 'text/plain',
|
|
'WWW-Authenticate': 'Basic realm="confluent"'}
|
|
if authorized['code'] == 401:
|
|
rsp = await make_response('text/plain', 401, 'Authentication Required', badauth)
|
|
await rsp.write(b'authentication required')
|
|
return rsp
|
|
if authorized['code'] == 403:
|
|
rsp = await make_response('application/json', 403, 'Forbidden', badauth)
|
|
response = {'result': 'Forbidden'}
|
|
if 'prompts' in authorized:
|
|
response['prompts'] = []
|
|
for prompt in authorized['prompts']:
|
|
if not isinstance(prompt, str):
|
|
prompt = prompt.decode('utf8')
|
|
response['prompts'].append(prompt)
|
|
await rsp.write(json.dumps(response).encode('utf8'))
|
|
return rsp
|
|
if authorized['code'] != 200:
|
|
raise Exception("Unrecognized code from auth engine")
|
|
cookies = authorized.get('cookie', None)
|
|
cfgmgr = authorized['cfgmgr']
|
|
if (operation == 'create') and reqpath == '/sessions/current/async':
|
|
pagecontent = ""
|
|
try:
|
|
async for rsp in _assemble_json(
|
|
confluent.asynchttp.handle_async(querydict)):
|
|
pagecontent += rsp
|
|
rsp = await make_response(mimetype, 200, cookies=cookies)
|
|
if not isinstance(pagecontent, bytes):
|
|
pagecontent = pagecontent.encode('utf-8')
|
|
await rsp.write(pagecontent)
|
|
return rsp
|
|
except exc.ConfluentException as e:
|
|
if e.apierrorcode == 500:
|
|
# raise generics to trigger the tracelog
|
|
raise
|
|
rsp = await make_response(mimetype, e.apierrorcode, e.apierrorstr)
|
|
await rsp.write(e.get_error_body())
|
|
return rsp
|
|
elif (reqpath.endswith('/forward/web') and
|
|
reqpath.startswith('/nodes/')):
|
|
prefix, _, _ = reqpath.partition('/forward/web')
|
|
#_, _, nodename = prefix.rpartition('/')
|
|
default = False
|
|
if 'default' in reqpath:
|
|
default = True
|
|
_,_,nodename,_ = prefix.split('/')
|
|
else:
|
|
_, _, nodename = prefix.rpartition('/')
|
|
hm = cfgmgr.get_node_attributes(nodename, 'hardwaremanagement.manager')
|
|
targip = hm.get(nodename, {}).get(
|
|
'hardwaremanagement.manager', {}).get('value', None)
|
|
if not targip:
|
|
rsp = await make_response('text/plain', 404)
|
|
await rsp.write(b'No hardwaremanagement.manager defined for node')
|
|
return rsp
|
|
targip = targip.split('/', 1)[0]
|
|
if default:
|
|
try:
|
|
ip_info = socket.getaddrinfo(targip, 0, 0, socket.SOCK_STREAM)
|
|
except socket.gaierror:
|
|
rsp = await make_response('text/plain', 404)
|
|
await rsp.write(b'hardwaremanagement.manager definition could not be resolved')
|
|
return rsp
|
|
# this is just to future proof just in case the indexes of the address family change in future
|
|
for i in range(len(ip_info)):
|
|
if ip_info[i][0] == socket.AF_INET:
|
|
url = 'https://{0}/'.format(ip_info[i][-1][0])
|
|
rsp = await make_response('text/plain', 302, headers={'Location': url})
|
|
await rsp.write(b'Our princess is in another castle!')
|
|
return rsp
|
|
elif ip_info[i][0] == socket.AF_INET6:
|
|
url = 'https://[{0}]/'.format(ip_info[i][-1][0])
|
|
if url.startswith('https://[fe80'):
|
|
rsp = await make_response('text/plain', 405)
|
|
await rsp.write(b'link local ipv6 address cannot be used in browser')
|
|
return rsp
|
|
rsp = await make_response('text/plain', 302, {'Location': url})
|
|
await rsp.write(b'Our princess is in another castle!')
|
|
return rsp
|
|
funport = forwarder.get_port(targip, req.headers.get('X-Forwarded-For', None),
|
|
authorized['sessionid'])
|
|
host = req.headers.get('X-Forwarded-Host', None)
|
|
if ']' in host:
|
|
host = host.split(']')[0] + ']'
|
|
elif ':' in host:
|
|
host = host.rsplit(':', 1)[0]
|
|
url = 'https://{0}:{1}/'.format(host, funport)
|
|
rsp = await make_response('text/plain', 302, headers={'Location': url})
|
|
await rsp.write(b'Our princess is in another castle!')
|
|
return rsp
|
|
elif (operation == 'create' and ('/console/session' in reqpath or
|
|
'/shell/sessions/' in reqpath)) and (reqpath.startswith(('/nodes/', '/noderange/'))):
|
|
if '/console/session' in reqpath:
|
|
prefix, _, _ = reqpath.partition('/console/session')
|
|
shellsession = False
|
|
elif '/shell/sessions/' in reqpath:
|
|
prefix, _, _ = reqpath.partition('/shell/sessions')
|
|
shellsession = True
|
|
_, _, nodename = prefix.rpartition('/')
|
|
if 'session' not in querydict.keys() or not querydict['session']:
|
|
auditmsg = {
|
|
'operation': 'start',
|
|
'target': reqpath,
|
|
'user': util.stringify(authorized['username']),
|
|
}
|
|
if 'tenant' in authorized:
|
|
auditmsg['tenant'] = authorized['tenant']
|
|
auditlog.log(auditmsg)
|
|
# Request for new session
|
|
skipreplay = False
|
|
if 'skipreplay' in querydict and querydict['skipreplay']:
|
|
skipreplay = True
|
|
width = querydict.get('width', 80)
|
|
height = querydict.get('height', 24)
|
|
datacallback = None
|
|
asynchdl = None
|
|
try:
|
|
if shellsession:
|
|
consession = await shellserver.ShellSession.create(
|
|
node=nodename, configmanager=cfgmgr,
|
|
username=authorized['username'], skipreplay=skipreplay,
|
|
datacallback=datacallback, width=width, height=height
|
|
)
|
|
else:
|
|
consession = await consoleserver.ConsoleSession.create(
|
|
node=nodename, configmanager=cfgmgr,
|
|
username=authorized['username'], skipreplay=skipreplay,
|
|
datacallback=datacallback, width=width, height=height
|
|
)
|
|
except exc.NotFoundException:
|
|
rsp = await make_response('text/plain', 404)
|
|
await rsp.write(b"404 - Request Path not recognized")
|
|
return rsp
|
|
if not consession:
|
|
rsp = await make_response('', 500)
|
|
return rsp
|
|
sessid = _assign_consessionid(consession)
|
|
if asynchdl:
|
|
asynchdl.add_console_session(sessid)
|
|
rsp = await make_response('application/json', 200, cookies=cookies)
|
|
await rsp.write(b'{"session":"%s","data":""}' % sessid.encode('utf8'))
|
|
return rsp
|
|
elif 'bytes' in querydict.keys(): # not keycodes...
|
|
myinput = querydict['bytes']
|
|
sessid = querydict['session']
|
|
if sessid not in consolesessions:
|
|
rsp = await make_response('text/plain', 400, 'Expired Session')
|
|
return rsp
|
|
consolesessions[sessid]['expiry'] = time.time() + 90
|
|
await consolesessions[sessid]['session'].write(myinput)
|
|
rsp = await make_response('application/json', 200)
|
|
await rsp.write(json.dumps({'session': querydict['session']}))
|
|
return rsp # client has requests to send or receive, not both...
|
|
elif 'closesession' in querydict:
|
|
await consolesessions[querydict['session']]['session'].destroy()
|
|
del consolesessions[querydict['session']]
|
|
rsp = await make_response('application/json', 200)
|
|
await rsp.write(b'{"sessionclosed": true}')
|
|
return rsp
|
|
elif 'action' in querydict:
|
|
if querydict['action'] == 'break':
|
|
await consolesessions[querydict['session']]['session'].send_break()
|
|
elif querydict['action'] == 'resize':
|
|
consolesessions[querydict['session']]['session'].resize(
|
|
width=querydict['width'], height=querydict['height'])
|
|
elif querydict['action'] == 'reopen':
|
|
consolesessions[querydict['session']]['session'].reopen()
|
|
else:
|
|
rsp = await make_response('text/plain', 400, 'Bad Request')
|
|
await rsp.write(b'Unrecognized action ' + querydict['action'])
|
|
return rsp
|
|
rsp = await make_response('application/json', 200)
|
|
await rsp.write(json.dumps({'session': querydict['session']}))
|
|
return rsp
|
|
else: # no keys, but a session, means it's hooking to receive data
|
|
raise Exception("long polling console sessions are discontinued")
|
|
elif (operation == 'create' and ('/firmware/updates/active' in reqpath)) and (reqpath.startswith(('/nodes/', '/noderange/'))):
|
|
if 'application/json' in reqtype:
|
|
if not isinstance(reqbody, str):
|
|
reqbody = reqbody.decode('utf8')
|
|
pbody = json.loads(reqbody)
|
|
args = pbody['args']
|
|
file_directory = '/var/lib/confluent/client_assets/{}'.format(args.split('/')[-1])
|
|
filepath = '{0}/{1}'.format(file_directory, os.listdir(file_directory)[0]) # TODO find a way to validate that the file is found and its the expected one
|
|
args_dict = {'filename': filepath}
|
|
nodeurls = {}
|
|
hdlr = pluginapi.handle_path(reqpath, operation, cfgmgr, args_dict)
|
|
async for res in pluginapi.iterate_responses(hdlr):
|
|
if isinstance(res, confluent.messages.CreatedResource):
|
|
watchurl = res.kvpairs['created']
|
|
currnode = watchurl.split('/')[1]
|
|
nodeurls[currnode] = '/' + watchurl
|
|
|
|
rsp = await make_response(mimetype, 200, 'OK')
|
|
await rsp.write(json.dumps({'data': nodeurls}).encode('utf8'))
|
|
return rsp
|
|
elif (operation == 'create' and (reqpath.startswith('/staging'))):
|
|
url = reqpath
|
|
args_dict = {}
|
|
content_length = int(req.content_length)
|
|
if content_length > 0 and (len(url.split('/')) > 2):
|
|
# check if the user and the url defined user are the same
|
|
if authorized['username'] == url.split('/')[2]:
|
|
args_dict.update({'filedata':req.content, 'content_length': content_length})
|
|
hdlr = pluginapi.handle_path(url, operation, cfgmgr, args_dict)
|
|
async for resp in pluginapi.iterate_responses(hdlr):
|
|
if isinstance(resp, confluent.messages.FileUploadProgress):
|
|
if resp.kvpairs['progress']['value'] == 100:
|
|
progress = resp.kvpairs['progress']['value']
|
|
rsp = await make_response(mimetype, 200, 'OK')
|
|
await rsp.write(json.dumps({'data': 'done'}).encode('utf8'))
|
|
return rsp
|
|
else:
|
|
rsp = await make_response(mimetype, 401, 'Unauthorized')
|
|
await rsp.write(json.dumps({'data': 'You do not have permission to write to file'}).encode('utf8'))
|
|
return rsp
|
|
elif len(url.split('/')) == 2:
|
|
reqbody = await req.read()
|
|
reqtype = req.content_type
|
|
if not isinstance(reqbody, str):
|
|
reqbody = reqbody.decode('utf8')
|
|
pbody = json.loads(reqbody)
|
|
args = pbody['args']
|
|
args_dict.update({'filename': args, 'user': authorized['username']})
|
|
try:
|
|
args_dict.update({'bank': pbody['bank']})
|
|
except KeyError:
|
|
pass
|
|
hdlr = pluginapi.handle_path(url, operation, cfgmgr, args_dict)
|
|
async for res in pluginapi.iterate_responses(hdlr):
|
|
if isinstance(res, confluent.messages.CreatedResource):
|
|
stageurl = res.kvpairs['created']
|
|
rsp = await make_response(mimetype, 200, 'OK')
|
|
await rsp.write(json.dumps({'data': stageurl}).encode('utf8'))
|
|
return rsp
|
|
else:
|
|
# normal request
|
|
url = reqpath
|
|
url = url.replace('.json', '')
|
|
url = url.replace('.html', '')
|
|
if url == '/sessions/current/info' or url.startswith('/sessions/current/webauthn/validate/'):
|
|
rsp = await make_response('application/json', 200, cookies=cookies)
|
|
sessinfo = {'username': authorized['username']}
|
|
if 'authtoken' in authorized:
|
|
sessinfo['authtoken'] = authorized['authtoken']
|
|
if 'sessionid' in authorized:
|
|
sessinfo['sessionid'] = authorized['sessionid']
|
|
tlvdata.unicode_dictvalues(sessinfo)
|
|
await rsp.write(json.dumps(sessinfo).encode('utf8'))
|
|
return rsp
|
|
elif url.startswith('/sessions/current/webauthn/'):
|
|
if not webauthn:
|
|
rsp = await make_response('text/plain', 501, 'Not Implemented')
|
|
return rsp
|
|
try:
|
|
wauthbody = await webauthn.handle_api_request(url, req, authorized['username'], cfgmgr, reqbody, authorized)
|
|
except Exception as e:
|
|
rsp = await make_response('text/plain', 401, 'Unauthorized')
|
|
await rsp.write(json.dumps({'error': 'Unable to complete passkey authentication'}).encode('utf8'))
|
|
return rsp
|
|
return await make_response(body=wauthbody)
|
|
resource = '.' + url[url.rindex('/'):]
|
|
lquerydict = copy.deepcopy(querydict)
|
|
try:
|
|
hdlr = pluginapi.handle_path(url, operation,
|
|
cfgmgr, querydict)
|
|
if 'ConfluentAsyncId' in req.headers:
|
|
await confluent.asynchttp.run_handler(hdlr, req)
|
|
rsp = await make_response('text/plain', 202, cookies=cookies)
|
|
await rsp.write(b'Request queued')
|
|
return rsp
|
|
pagecontent = ""
|
|
if mimetype == 'text/html':
|
|
for datum in _assemble_html(hdlr, resource, lquerydict, url,
|
|
extension):
|
|
pagecontent += datum
|
|
else:
|
|
async for datum in _assemble_json(hdlr, resource, url, extension):
|
|
pagecontent += datum
|
|
rsp = await make_response(mimetype, 200, cookies=cookies)
|
|
if not isinstance(pagecontent, bytes):
|
|
pagecontent = pagecontent.encode('utf-8')
|
|
await rsp.write(pagecontent)
|
|
return rsp
|
|
except exc.ConfluentException as e:
|
|
if ((not isinstance(e, exc.LockedCredentials)) and
|
|
e.apierrorcode == 500):
|
|
# raise generics to trigger the tracelog
|
|
raise
|
|
rsp = await make_response(mimetype, e.apierrorcode, e.apierrorstr,
|
|
cookies=cookies)
|
|
await rsp.write(e.get_error_body().encode('utf8'))
|
|
return rsp
|
|
|
|
def _assemble_html(responses, resource, querydict, url, extension):
|
|
yield '<html><body>HTML API Explorer is discontinued, notify developers if you want this back</body></html>'
|
|
return
|
|
yield '<html><head><meta charset="UTF-8"><title>' \
|
|
'Confluent REST Explorer: ' + url + '</title></head>' \
|
|
'<body><form action="' + \
|
|
resource + '" method="post">'
|
|
if querydict:
|
|
yield 'Response to input data:<br>' + \
|
|
json.dumps(querydict, separators=(',', ': '),
|
|
indent=4, sort_keys=True) + '<hr>'
|
|
yield 'Only fields that have their boxes checked will have their ' \
|
|
'respective values honored by the confluent server.<hr>' \
|
|
'<input type="hidden" name="restexplorerhonorkey" value="">' + \
|
|
'<a rel="self" href="{0}{1}">{0}{1}</a><br>'.format(
|
|
resource, extension)
|
|
if url == '/':
|
|
iscollection = True
|
|
elif resource[-1] == '/':
|
|
iscollection = True
|
|
yield '<a rel="collection" href="../{0}">../{0}</a><br>'.format(
|
|
extension)
|
|
else:
|
|
iscollection = False
|
|
yield '<a rel="collection" href="./{0}">./{0}</a><br>'.format(
|
|
extension)
|
|
pendingrsp = []
|
|
for rsp in responses:
|
|
if isinstance(rsp, confluent.messages.LinkRelation):
|
|
yield rsp.html(extension) + "<br>"
|
|
else:
|
|
pendingrsp.append(rsp)
|
|
for rsp in pendingrsp:
|
|
yield rsp.html() + "<br>"
|
|
if iscollection:
|
|
# localpath = url[:-2] (why was this here??)
|
|
try:
|
|
if url == '/users/':
|
|
return
|
|
firstpass = True
|
|
module = url.split('/')
|
|
if not module:
|
|
return
|
|
for y in create_resource_functions[module[-2]]():
|
|
if firstpass:
|
|
yield "<hr>Define new resource in %s:<BR>" % module[-2]
|
|
firstpass = False
|
|
yield y
|
|
yield ('<input value="create" name="restexplorerop" type="submit">'
|
|
'</form></body></html>')
|
|
except KeyError:
|
|
pass
|
|
else:
|
|
yield ('<input value="update" name="restexplorerop" type="submit">'
|
|
'</form></body></html>')
|
|
|
|
|
|
async def _assemble_json(responses, resource=None, url=None, extension=None):
|
|
#NOTE(jbjohnso) I'm considering giving up on yielding bit by bit
|
|
#in json case over http. Notably, duplicate key values from plugin
|
|
#overwrite, but we'd want to preserve them into an array instead.
|
|
#the downside is that http would just always blurt it ll out at
|
|
#once and hold on to all the data in memory
|
|
links = {}
|
|
if resource is not None:
|
|
links['self'] = {"href": resource + extension}
|
|
if url == '/':
|
|
pass
|
|
elif resource[-1] == '/':
|
|
links['collection'] = {"href": "../" + extension}
|
|
else:
|
|
links['collection'] = {"href": "./" + extension}
|
|
rspdata = {}
|
|
async for rsp in pluginapi.iterate_responses(responses):
|
|
if isinstance(rsp, confluent.messages.LinkRelation):
|
|
haldata = rsp.raw()
|
|
for hk in haldata:
|
|
if 'href' in haldata[hk]:
|
|
if isinstance(haldata[hk]['href'], int):
|
|
haldata[hk]['href'] = str(haldata[hk]['href'])
|
|
haldata[hk]['href'] += extension
|
|
if hk in links:
|
|
if isinstance(links[hk], list):
|
|
links[hk].append(haldata[hk])
|
|
else:
|
|
links[hk] = [links[hk], haldata[hk]]
|
|
elif hk == 'item':
|
|
links[hk] = [haldata[hk],]
|
|
else:
|
|
links[hk] = haldata[hk]
|
|
else:
|
|
rsp = rsp.raw()
|
|
for dk in rsp:
|
|
if dk in rspdata:
|
|
if isinstance(rspdata[dk], list):
|
|
if isinstance(rsp[dk], list):
|
|
rspdata[dk].extend(rsp[dk])
|
|
else:
|
|
rspdata[dk].append(rsp[dk])
|
|
else:
|
|
rspdata[dk] = [rspdata[dk], rsp[dk]]
|
|
else:
|
|
if dk == 'databynode' or dk == 'asyncresponse':
|
|
# a quirk, databynode suggests noderange
|
|
# multi response. This should *always* be a list,
|
|
# even if it will be length 1
|
|
rspdata[dk] = [rsp[dk]]
|
|
else:
|
|
rspdata[dk] = rsp[dk]
|
|
rspdata["_links"] = links
|
|
tlvdata.unicode_dictvalues(rspdata)
|
|
yield util.stringify(json.dumps(
|
|
rspdata, sort_keys=True, indent=4, ensure_ascii=False).encode('utf-8'))
|
|
|
|
|
|
async def serve(bind_host, bind_port, bind_group, bind_perms):
|
|
# TODO(jbjohnso): move to unix socket and explore
|
|
# either making apache deal with it
|
|
# or just supporting nginx or lighthttpd
|
|
# for now, http port access
|
|
# todo remains unix domain socket for even http
|
|
sock = None
|
|
if not bind_perms:
|
|
bind_perms = 0o666
|
|
while not sock:
|
|
try:
|
|
bind_arg = None
|
|
if '/' in bind_host:
|
|
try:
|
|
os.remove(bind_host)
|
|
except OSError:
|
|
pass
|
|
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
|
oldumask = os.umask(0o777 - bind_perms)
|
|
sock.bind(bind_host)
|
|
os.chmod(bind_host, bind_perms)
|
|
if bind_group:
|
|
shutil.chown(bind_host, group=bind_group)
|
|
os.umask(oldumask)
|
|
bind_arg = bind_host
|
|
else:
|
|
bindinfo = socket.getaddrinfo(
|
|
bind_host, bind_port, 0, socket.SOCK_STREAM)
|
|
if bindinfo[0][0] == socket.AF_INET:
|
|
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
elif bindinfo[0][0] == socket.AF_INET6:
|
|
sock = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
|
|
bind_arg = bindinfo[0][4]
|
|
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
|
sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
|
|
sock.settimeout(0)
|
|
sock.bind(bind_arg)
|
|
sock.listen(128)
|
|
except socket.error as e:
|
|
if e.errno != 98:
|
|
raise
|
|
sys.stderr.write(
|
|
'Failed to open HTTP due to busy port, trying again in'
|
|
' a second\n')
|
|
await asyncio.sleep(1)
|
|
# TCP_FASTOPEN
|
|
try:
|
|
sock.setsockopt(socket.SOL_TCP, 23, 5)
|
|
except Exception:
|
|
pass # we gave it our best shot there
|
|
app = web.Application()
|
|
app.router.add_route("*", "/{path_info:.*}", resourcehandler)
|
|
runner = web.AppRunner(app)
|
|
await runner.setup()
|
|
site = web.SockSite(runner, sock)
|
|
await site.start()
|
|
|
|
|
|
|
|
|
|
|
|
class HttpApi(object):
|
|
def __init__(self, bind_host=None, bind_port=None, bind_group=None, bind_perms=None):
|
|
self.server = None
|
|
self.bind_host = bind_host or '127.0.0.1'
|
|
self.bind_port = bind_port or 4005
|
|
self.bind_group = bind_group
|
|
self.bind_perms = bind_perms
|
|
|
|
def start(self):
|
|
global _cleaner
|
|
global auditlog
|
|
global tracelog
|
|
if _cleaner is None:
|
|
_cleaner = asyncio.get_running_loop().create_task(
|
|
_sessioncleaner())
|
|
tracelog = log.Logger('trace')
|
|
auditlog = log.Logger('audit')
|
|
self.server = asyncio.get_running_loop().create_task(
|
|
serve(self.bind_host, self.bind_port, self.bind_group, self.bind_perms))
|