# 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 ipaddress
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:
                loop = asyncio.get_running_loop()
                ip_info = await loop.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
            url = None
            link_local_only = True
            for i in range(len(ip_info)):
                if ip_info[i][0] == socket.AF_INET:
                    link_local_only = False
                    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:
                    address = ip_info[i][-1][0].split('%', 1)[0]
                    if ipaddress.ip_address(address).is_link_local:
                        continue
                    link_local_only = False
                    url = 'https://[{0}]/'.format(ip_info[i][-1][0])
            if link_local_only and ip_info:
                # A link-local manager address is not usable by the browser;
                # use the address reported by the controller instead.
                v6addr = None
                v4addr = None
                try:
                    netpath = '/nodes/{}/configuration/management_controller/' \
                              'net_interfaces/management'.format(nodename)
                    nethdlr = pluginapi.handle_path(netpath, 'retrieve', cfgmgr,
                                                    {})
                    async for datum in pluginapi.iterate_responses(nethdlr):
                        if isinstance(datum, confluent.messages.NetworkConfiguration):
                            v4addr = datum.kvpairs.get('ipv4_address', {}).get('value')
                            v6addr = datum.kvpairs.get('static_v6_addresses', {}).get('value')
                except Exception:
                    pass
                if v4addr:
                    url = 'https://{0}/'.format(str(v4addr).split('/', 1)[0])
                else:
                    if isinstance(v6addr, str):
                        v6addr = v6addr.split(',', 1)[0]
                    elif isinstance(v6addr, list):
                        v6addr = v6addr[0] if v6addr else None
                    if v6addr:
                        v6addr = v6addr.split('/', 1)[0]
                        url = 'https://[{0}]/'.format(v6addr)
            if not url:
                rsp = await make_response('text/plain', 404)
                await rsp.write(b'No usable management controller address found')
                return rsp
            rsp = await make_response('text/plain', 302, headers={'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 = await asyncio.get_running_loop().getaddrinfo(
                    bind_host, bind_port, type=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))
