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

import base64
import uuid
import datetime
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__

# ИНИЦИАЛИЗАЦИЯ 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")):
    lResult={"Domain": "", "User": ""}
    #print("IdentifyAuthorize")
    ######################################
    #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 mOpenRPA
    ######################################
    #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 mOpenRPA
            #inRequest.OpenRPASetCookie = {}
            #New engine of server
            #inRequest.OpenRPAResponseDict["SetCookies"]["AuthToken"] = lAuthToken
        else:
            raise HTTPException(status_code=401, detail="here is the details", headers={})
    ######################################
    else:
        raise HTTPException(status_code=401, detail="here is the details", headers={'Content-type':'text/html', 'WWW-Authenticate':'Basic'})

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"]
    )

def BackwardCompatibility(inRequest:Request, inResponse:Response, inBodyStr:str = Body(""), inAuthDict = None):
    lHTTPRequest = ServerBC.HTTPRequestOld(inRequest=inRequest, inResponse=inResponse, inAuthDict=inAuthDict)
    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(""),
    inAuthDict:dict=Depends(IdentifyAuthorize)): # Old from v1.3.1 (updated to FastAPI)
    return BackwardCompatibility(inRequest = inRequest, inResponse = inResponse, inBodyStr = inBodyStr, inAuthDict=inAuthDict)
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, inAuthDict=None)
def BackwardCompatibityBeginWrapperAuth(inBeginTokenStr, inRequest:Request, inResponse:Response, inBodyStr:str = Body(""),
    inAuthDict:dict=Depends(IdentifyAuthorize)): # Old from v1.3.1 (updated to FastAPI)
    return BackwardCompatibility(inRequest = inRequest, inResponse = inResponse, inBodyStr = inBodyStr, inAuthDict=inAuthDict)
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, inAuthDict=None)



    
from . import ServerSettings
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"]
                    )
                
    uvicorn.run('pyOpenRPA.Orchestrator.Server:app', host='0.0.0.0', port=1024)
    #if lL: lL.info(f"Сервер инициализирован успешно (с поддержкой SSL):: Наименование: {self.name}, Слушает URL: {lAddressStr}, Слушает порт: {lPortInt}, Путь к файлу сертификата (.pem): {lCertFilePathStr}")
    #if lL: lL.info(f"Сервер инициализирован успешно (без поддержки SSL):: Наименование: {self.name}, Слушает URL: {lAddressStr}, Слушает порт: {lPortInt}")