ORPA-pyOpenRPA/Sources/pyOpenRPA/Orchestrator/Server.py

#     inRequest.OpenRPA = {}
#     inRequest.OpenRPA["AuthToken"] = None
#     inRequest.OpenRPA["Domain"] = None
#     inRequest.OpenRPA["User"] = None

# lResponseDict = {"Headers": {}, "SetCookies": {}, "Body": b"", "StatusCode": None}
# self.OpenRPAResponseDict = lResponseDict

#from http.client import HTTPException
import threading

from pyOpenRPA.Tools import CrossOS


from http import cookies
from . import ServerBC

# объявление import
from fastapi import FastAPI, Form, Request, HTTPException, Depends, Header, Response, Body
from fastapi.responses import PlainTextResponse, HTMLResponse, FileResponse
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from pydantic import BaseModel
import uvicorn
import io
from starlette.responses import StreamingResponse
from typing import Union
from pyOpenRPA import __version__

import base64
import uuid
import datetime

# ИНИЦИАЛИЗАЦИЯ FASTAPI!
app = FastAPI(
        title = "pyOpenRPA (ORPA) Orchestrator",
        description = "Сервер оркестратора pyOpenRPA Orchestrator",
        version = __version__,
        openapi_url="/orpa/fastapi/openapi.json", 
        docs_url = "/orpa/fastapi/docs",
        redoc_url = "/orpa/fastapi/redoc",
        swagger_ui_oauth2_redirect_url = "/orpa/fastapi/docs/oauth2-redirect",
    )      

def IdentifyAuthorize(inRequest:Request, inResponse:Response,
    inCookiesStr: Union[str, None] = Header(default=None,alias="Cookie"), 
    inAuthorizationStr: Union[str, None] = Header(default="",alias="Authorization")):
    if __Orchestrator__.GSettingsGet().get("ServerDict", {}).get("AccessUsers", {}).get("FlagCredentialsAsk", False)==True:
        lResult={"Domain": "", "User": ""}
        ######################################
        #Way 1 - try to find AuthToken
        lCookies = cookies.SimpleCookie(inCookiesStr) # inRequest.headers.get("Cookie", "")
        __Orchestrator__.GSettingsGet()
        lHeaderAuthorization = inAuthorizationStr.split(" ")
        if "AuthToken" in lCookies:
            lCookieAuthToken = lCookies.get("AuthToken", "").value
            if lCookieAuthToken:
                #Find AuthToken in GlobalDict
                if lCookieAuthToken in __Orchestrator__.GSettingsGet().get("ServerDict", {}).get("AccessUsers", {}).get("AuthTokensDict", {}):
                    #Auth Token Has Been Founded
                    lResult["Domain"] = __Orchestrator__.GSettingsGet()["ServerDict"]["AccessUsers"]["AuthTokensDict"][lCookieAuthToken]["Domain"]
                    lResult["User"] = __Orchestrator__.GSettingsGet()["ServerDict"]["AccessUsers"]["AuthTokensDict"][lCookieAuthToken]["User"]
                    #Set auth token
                    mOpenRPA={}
                    mOpenRPA["AuthToken"] = lCookieAuthToken
                    mOpenRPA["Domain"] = lResult["Domain"]
                    mOpenRPA["User"] = lResult["User"]
                    mOpenRPA["IsSuperToken"] = __Orchestrator__.GSettingsGet().get("ServerDict", {}).get("AccessUsers", {}).get("AuthTokensDict", {}).get(mOpenRPA["AuthToken"], {}).get("FlagDoNotExpire", False)
                    return lCookieAuthToken
        ######################################
        #Way 2 - try to logon
        if len(lHeaderAuthorization) == 2:
            llHeaderAuthorizationDecodedUserPasswordList = base64.b64decode(lHeaderAuthorization[1]).decode("utf-8").split(
                ":")
            lUser = llHeaderAuthorizationDecodedUserPasswordList[0]
            lPassword = llHeaderAuthorizationDecodedUserPasswordList[1]
            lDomain = ""
            if "\\" in lUser:
                lDomain = lUser.split("\\")[0]
                lUser = lUser.split("\\")[1]
            lLogonBool = __Orchestrator__.OSCredentialsVerify(inUserStr=lUser, inPasswordStr=lPassword, inDomainStr=lDomain)
            #Check result
            if lLogonBool:
                lResult["Domain"] = lDomain
                lResult["User"] = lUser
                #Create token
                lAuthToken=str(uuid.uuid1())
                __Orchestrator__.GSettingsGet()["ServerDict"]["AccessUsers"]["AuthTokensDict"][lAuthToken] = {}
                __Orchestrator__.GSettingsGet()["ServerDict"]["AccessUsers"]["AuthTokensDict"][lAuthToken]["Domain"] = lResult["Domain"]
                __Orchestrator__.GSettingsGet()["ServerDict"]["AccessUsers"]["AuthTokensDict"][lAuthToken]["User"] = lResult["User"]
                __Orchestrator__.GSettingsGet()["ServerDict"]["AccessUsers"]["AuthTokensDict"][lAuthToken]["FlagDoNotExpire"] = False
                __Orchestrator__.GSettingsGet()["ServerDict"]["AccessUsers"]["AuthTokensDict"][lAuthToken]["TokenDatetime"] = datetime.datetime.now()
                #Set-cookie
                inResponse.set_cookie(key="AuthToken",value=lAuthToken)
                mOpenRPA={}
                mOpenRPA["AuthToken"] = lAuthToken
                mOpenRPA["Domain"] = lResult["Domain"]
                mOpenRPA["User"] = lResult["User"]
                mOpenRPA["IsSuperToken"] = __Orchestrator__.GSettingsGet().get("ServerDict", {}).get("AccessUsers", {}).get("AuthTokensDict", {}).get(mOpenRPA["AuthToken"], {}).get("FlagDoNotExpire", False)
                return lAuthToken
                #inRequest.OpenRPASetCookie = {}
                #New engine of server
                #inRequest.OpenRPAResponseDict["SetCookies"]["AuthToken"] = lAuthToken
            else:
                raise HTTPException(status_code=401, detail="Попытка авторизации не прошла успешно (неверная пара логин / пароль)", headers={})
        ######################################
        else:
            raise HTTPException(status_code=401, detail="Попытка авторизации не прошла успешно (неполная пара логин / пароль)", headers={'Content-type':'text/html', 'WWW-Authenticate':'Basic'})
    else: return None # Credentials are not required - return none



lRouteList =[]
for lItem in app.router.routes:
    lRouteList.append(lItem)
app.router.routes=[]
for lItem in lRouteList:
    app.add_api_route(
        path=lItem.path,
        endpoint=lItem.endpoint,
        methods=["GET"],
        dependencies=[Depends(IdentifyAuthorize)],
        tags=["FastAPI"]
    )


from . import ServerSettings  

def BackwardCompatibility(inRequest:Request, inResponse:Response, inBodyStr:str = Body(""), inAuthTokenStr = None):
    lHTTPRequest = ServerBC.HTTPRequestOld(inRequest=inRequest, inResponse=inResponse, inAuthTokenStr=inAuthTokenStr)
    lHTTPRequest.path = inRequest.url.path
    lHTTPRequest.body = inBodyStr
    lHTTPRequest.client_address = [inRequest.client.host]
    threading.current_thread().request = lHTTPRequest
    lResult = lHTTPRequest.do_GET(inBodyStr=inBodyStr)
    if lResult is None:
        lResult = lHTTPRequest.do_POST(inBodyStr=inBodyStr)
    if lHTTPRequest.OpenRPAResponseDict["Headers"]["Content-type"] != None:
        return StreamingResponse(io.BytesIO(lResult), media_type=lHTTPRequest.OpenRPAResponseDict["Headers"]["Content-type"])

#WRAPPERS!
def BackwardCompatibityWrapperAuth(inRequest:Request, inResponse:Response, inBodyStr:str = Body(""),
    inAuthTokenStr:str=Depends(ServerSettings.IdentifyAuthorize)): # Old from v1.3.1 (updated to FastAPI)
    return BackwardCompatibility(inRequest = inRequest, inResponse = inResponse, inBodyStr = inBodyStr, inAuthTokenStr=inAuthTokenStr)
def BackwardCompatibityWrapperNoAuth(inRequest:Request, inResponse:Response, inBodyStr:str = Body("")): # Old from v1.3.1 (updated to FastAPI)
    return BackwardCompatibility(inRequest = inRequest, inResponse = inResponse, inBodyStr = inBodyStr, inAuthTokenStr=None)
def BackwardCompatibityBeginWrapperAuth(inBeginTokenStr, inRequest:Request, inResponse:Response, inBodyStr:str = Body(""),
    inAuthTokenStr:str=Depends(ServerSettings.IdentifyAuthorize)): # Old from v1.3.1 (updated to FastAPI)
    return BackwardCompatibility(inRequest = inRequest, inResponse = inResponse, inBodyStr = inBodyStr, inAuthTokenStr=inAuthTokenStr)
def BackwardCompatibityBeginWrapperNoAuth(inBeginTokenStr, inRequest:Request, inResponse:Response, inBodyStr:str = Body("")): # Old from v1.3.1 (updated to FastAPI)
    return BackwardCompatibility(inRequest = inRequest, inResponse = inResponse, inBodyStr = inBodyStr, inAuthTokenStr=None)



from . import __Orchestrator__
import mimetypes
mimetypes.add_type("font/woff2",".woff2")
mimetypes.add_type("text/javascript",".js")
from typing import Union

def InitFastAPI():
    global app
    lL = __Orchestrator__.OrchestratorLoggerGet()
    __Orchestrator__.GSettingsGet()["ServerDict"]["ServerThread"] = app
    ServerSettings.SettingsUpdate()
    for lConnectItemDict in __Orchestrator__.GSettingsGet()["ServerDict"]["URLList"]:
        if "ResponseFolderPath" in lConnectItemDict:
            app.mount(lConnectItemDict["URL"], 
        StaticFiles(directory=CrossOS.PathStr(lConnectItemDict["ResponseFolderPath"])), 
        name=lConnectItemDict["URL"].replace('/',"_"))
        else:
            if lConnectItemDict.get("MatchType") in ["EqualCase", "Equal","EqualNoParam"]:
                if lConnectItemDict.get("UACBool",True):
                    app.add_api_route(
                        path=lConnectItemDict["URL"],
                        endpoint=BackwardCompatibityWrapperAuth,
                        response_class=PlainTextResponse,
                        methods=[lConnectItemDict["Method"]],
                        tags=["BackwardCompatibility"]
                    )
                else:
                    app.add_api_route(
                        path=lConnectItemDict["URL"],
                        endpoint=BackwardCompatibityWrapperNoAuth,
                        response_class=PlainTextResponse,
                        methods=[lConnectItemDict["Method"]],
                        tags=["BackwardCompatibility"]
                    )
            elif lConnectItemDict.get("MatchType") in ["BeginWith", "Contains"]:
                lURLStr = lConnectItemDict["URL"]
                if lURLStr[-1]!="/": lURLStr+="/"
                lURLStr+="{inBeginTokenStr}"
                if lConnectItemDict.get("UACBool",True):
                    app.add_api_route(
                        path=lURLStr,
                        endpoint=BackwardCompatibityBeginWrapperAuth,
                        response_class=PlainTextResponse,
                        methods=[lConnectItemDict["Method"]],
                        tags=["BackwardCompatibility"]
                    )
                else:
                    app.add_api_route(
                        path=lURLStr,
                        endpoint=BackwardCompatibityBeginWrapperNoAuth,
                        response_class=PlainTextResponse,
                        methods=[lConnectItemDict["Method"]],
                        tags=["BackwardCompatibility"]
                    )
                
def InitUvicorn(inHostStr=None, inPortInt=None, inSSLCertPathStr=None, inSSLKeyPathStr=None, inSSLPasswordStr=None):
    if inHostStr is None: inHostStr="0.0.0.0"
    if inPortInt is None: inPortInt=1024
    if inSSLCertPathStr != None: inSSLCertPathStr=CrossOS.PathStr(inSSLCertPathStr)
    if inSSLKeyPathStr != None: inSSLKeyPathStr=CrossOS.PathStr(inSSLKeyPathStr)
    global app
    lL = __Orchestrator__.OrchestratorLoggerGet()
    #uvicorn.run('pyOpenRPA.Orchestrator.Server:app', host='0.0.0.0', port=1024)
    uvicorn.run(app, host=inHostStr, port=inPortInt,ssl_keyfile=inSSLKeyPathStr,ssl_certfile=inSSLCertPathStr,ssl_keyfile_password=inSSLPasswordStr)
    if lL and inSSLKeyPathStr != None: lL.info(f"Сервер инициализирован успешно (с поддержкой SSL):: Слушает URL: {inHostStr}, Слушает порт: {inPortInt}, Путь к файлу сертификата (.pem, base64): {inSSLCertPathStr}")
    if lL and inSSLKeyPathStr == None: lL.info(f"Сервер инициализирован успешно (без поддержки SSL):: Слушает URL: {inHostStr}, Слушает порт: {inPortInt}")