OPEN_RPA_PUBLIC
/
ORPA-pyOpenRPA
2
0
Fork 1
Code Issues 2 Pull Requests Packages Projects Releases Wiki Activity

orc update methods (add what will be deprecated in 1.3.2)

Browse Source
merge-requests/1/merge
Ivan Maslov 2 years ago
parent fec135e16f
commit cce7f1ebe3
1 changed files with 97 additions and 32 deletions
Show Stats Download Patch File Download Diff File

129
Sources/pyOpenRPA/Orchestrator/__Orchestrator__.py
Unescape View File

@ -814,7 +814,7 @@ def OrchestratorSessionRestore(inGSettings=None):
os.remove("_SessionLast_GSettings.pickle") # remove the temp file
def UACKeyListCheck(inRequest, inRoleKeyList) -> bool:
"""L+,W+: Проверить права доступа для пользователя запроса по списку ключей до права.
"""L+,W+: [ПРЕКРАЩЕНИЕ ПОДДЕРЖКИ В 1.3.2, см. WebUserUACCheck] Проверить права доступа для пользователя запроса по списку ключей до права.
.. code-block:: python
@ -829,7 +829,7 @@ def UACKeyListCheck(inRequest, inRoleKeyList) -> bool:
return inRequest.UACClientCheck(inRoleKeyList=inRoleKeyList)
def UACUserDictGet(inRequest) -> dict:
"""L+,W+: Вернуть UAC (User Access Control) словарь доступов для пользователя, который отправил запрос. Пустой словарь - супердоступ (доступ ко всему)
"""L+,W+: [ПРЕКРАЩЕНИЕ ПОДДЕРЖКИ В 1.3.2, см. WebUserUACHierarchyGet] Вернуть UAC (User Access Control) словарь доступов для пользователя, который отправил запрос. Пустой словарь - супердоступ (доступ ко всему)
:param inRequest: Экземпляр request (from http.server import BaseHTTPRequestHandler)
:return: Словарь доступов пользователя. Пустой словарь - супердоступ (доступ ко всему)
@ -1026,8 +1026,18 @@ def WebCPUpdate(inCPKeyStr, inHTMLRenderDef=None, inJSONGeneratorDef=None, inJSI
if inJSInitGeneratorDef is not None: lCPManager.mBackwardCompatibilityJSDef = inJSInitGeneratorDef
def WebAuditMessageCreate(inRequest=None, inOperationCodeStr="-", inMessageStr="-"):
"""L+,W+: Создание сообщения ИТ аудита с такими сведениями как (Домен, IP, логин и тд.). Данная функция особенно актуальна в том случае, если требуется реализовать дополнительные меры контроля ИТ системы.
def WebRequestHostGet(inRequest) -> str:
"""L+,W+: Получить наименование хоста, с которого поступил запрос
:param inRequest: Экземпляр fastapi.Request, по умолчанию None
:type inRequest: fastapi.Request, опционально
:return: Наименование хоста
:rtype: str
"""
return inRequest.client.host
def WebAuditMessageCreate(inAuthTokenStr:str = None, inHostStr:str=None, inOperationCodeStr:str="-", inMessageStr:str="-"):
"""L+,W+: [ИЗМЕНЕНИЕ В 1.3.1] Создание сообщения ИТ аудита с такими сведениями как (Домен, IP, логин и тд.). Данная функция особенно актуальна в том случае, если требуется реализовать дополнительные меры контроля ИТ системы.
.. code-block:: python
@ -1042,25 +1052,25 @@ def WebAuditMessageCreate(inRequest=None, inOperationCodeStr="-", inMessageStr="
# Логгирование сообщения
lLogger.info(lWebAuditMessageStr)
:param inRequest: Экземпляр HTTP request. Опционален, если сообщение фиксируется из под потока, который был инициирован запросом пользователя
:param inAuthTokenStr: Токен авторизации пользователя / бота, по умолчанию None (не установлен)
:type inAuthTokenStr: str, опционально
:param inHostStr: IP адрес хоста пользователя / бота, по умолчанию None (не установлен)
:type inHostStr: str, опционально
:param inOperationCodeStr: Код операции, который принят в компании в соответствии с регламентными процедурами
:param inMessageStr: Дополнительное сообщение, которое необходимо отправить в сообщение об ИТ аудите
:return: Формат сообщения: "WebAudit :: DOMAIN\\USER@101.121.123.12 :: operation code :: message"
"""
try:
if inRequest is None: inRequest = WebRequestGet()
lClientIPStr = inRequest.client_address[0]
lUserDict = WebUserInfoGet(inRequest=inRequest)
lDomainUpperStr = lUserDict["DomainUpperStr"]
lUserLoginStr = lUserDict["UserNameUpperStr"]
lResultStr = f"ВебАудит :: {lDomainUpperStr}\\\\{lUserLoginStr}@{lClientIPStr} :: {inOperationCodeStr} :: {inMessageStr}"
lDomainUpperStr = WebUserDomainGet(inAuthTokenStr=inAuthTokenStr).upper()
lUserLoginStr = WebUserLoginGet(inAuthTokenStr=inAuthTokenStr).upper()
lResultStr = f"ВебАудит :: {lDomainUpperStr}\\\\{lUserLoginStr}@{inHostStr} :: {inOperationCodeStr} :: {inMessageStr}"
except Exception as e:
print(str(e)) # Has no logger - must be dead alg branch
lResultStr = inMessageStr
return lResultStr
def WebRequestParseBodyBytes(inRequest=None):
"""L+,W+: Извлечь данные в байт виде из тела (body) HTTP запроса.
"""L+,W+: [ПРЕКРАЩЕНИЕ ПОДДЕРЖКИ В 1.3.1, см. FASTAPI] Извлечь данные в байт виде из тела (body) HTTP запроса.
:param inRequest: Экземпляр HTTP request. Опционален, если сообщение фиксируется из под потока, который был инициирован запросом пользователя
:return: Строка байт b'' или None (если тело запроса было пустым)
@ -1073,7 +1083,7 @@ def WebRequestParseBodyBytes(inRequest=None):
return lBodyBytes
def WebRequestParseBodyStr(inRequest=None):
"""L+,W+: Извлечь данные в виде строки из тела (body) HTTP запроса.
"""L+,W+: [ПРЕКРАЩЕНИЕ ПОДДЕРЖКИ В 1.3.1, см. FASTAPI] Извлечь данные в виде строки из тела (body) HTTP запроса.
:param inRequest: Экземпляр HTTP request. Опционален, если сообщение фиксируется из под потока, который был инициирован запросом пользователя
:return: Текстовая строка '' или None (если тело запроса было пустым)
@ -1082,7 +1092,7 @@ def WebRequestParseBodyStr(inRequest=None):
return WebRequestParseBodyBytes(inRequest=inRequest).decode('utf-8')
def WebRequestParseBodyJSON(inRequest=None):
"""L+,W+: Извлечь из тела (body) запроса HTTP JSON данные и преобразовать в Dict / List структуры языка Python.
"""L+,W+: [ПРЕКРАЩЕНИЕ ПОДДЕРЖКИ В 1.3.1, см. FASTAPI] Извлечь из тела (body) запроса HTTP JSON данные и преобразовать в Dict / List структуры языка Python.
:param inRequest: Экземпляр HTTP request. Опционален, если сообщение фиксируется из под потока, который был инициирован запросом пользователя
:return: Словарь (dict), список (list) или None (если тело запроса было пустым)
@ -1091,7 +1101,7 @@ def WebRequestParseBodyJSON(inRequest=None):
return json.loads(WebRequestParseBodyStr(inRequest=inRequest))
def WebRequestParsePath(inRequest=None):
"""L+,W+: Извлечь декодированный URL путь из HTTP запроса пользователя в формате строки.
"""L+,W+: [ПРЕКРАЩЕНИЕ ПОДДЕРЖКИ В 1.3.1, см. FASTAPI] Извлечь декодированный URL путь из HTTP запроса пользователя в формате строки.
:param inRequest: Экземпляр HTTP request. Опционален, если сообщение фиксируется из под потока, который был инициирован запросом пользователя
:return: str, пример: /pyOpenRPA/Debugging/DefHelper
@ -1100,7 +1110,7 @@ def WebRequestParsePath(inRequest=None):
return urllib.parse.unquote(inRequest.path)
def WebRequestParseFile(inRequest=None):
"""L+,W+: Извлечь файл (наименование + содержимое в виде строки байт b'') из HTTP запроса пользователя.
"""L+,W+: [ПРЕКРАЩЕНИЕ ПОДДЕРЖКИ В 1.3.1, см. FASTAPI] Извлечь файл (наименование + содержимое в виде строки байт b'') из HTTP запроса пользователя.
:param inRequest: Экземпляр HTTP request. Опционален, если сообщение фиксируется из под потока, который был инициирован запросом пользователя
:return: Кортеж формата (FileNameStr, FileBodyBytes) or (None, None), если файл не был обнаружен
@ -1126,7 +1136,7 @@ def WebRequestParseFile(inRequest=None):
return lResultTurple
def WebRequestResponseSend(inResponeStr, inRequest=None, inContentTypeStr: str = None, inHeadersDict: dict = None):
"""L+,W+: Установить ответ на HTTP запрос пользователя.
"""L+,W+: [ПРЕКРАЩЕНИЕ ПОДДЕРЖКИ В 1.3.1, см. FASTAPI] Установить ответ на HTTP запрос пользователя.
:param inResponeStr: Тело ответа (строка)
:param inRequest: Экземпляр HTTP request. Опционален, если сообщение фиксируется из под потока, который был инициирован запросом пользователя
@ -1142,14 +1152,39 @@ def WebRequestResponseSend(inResponeStr, inRequest=None, inContentTypeStr: str =
def WebRequestGet():
"""L+,W+: Вернуть экземпляр HTTP запроса, если функция вызвана в потоке, который был порожден для отработки HTTP запроса пользователя.
"""L+,W+: [ПРЕКРАЩЕНИЕ ПОДДЕРЖКИ В 1.3.2] Вернуть экземпляр HTTP запроса, если функция вызвана в потоке, который был порожден для отработки HTTP запроса пользователя.
"""
lCurrentThread = threading.current_thread()
if hasattr(lCurrentThread, "request"):
return lCurrentThread.request
def WebUserLoginGet(inAuthTokenStr: str=None) -> str:
"""L+,W+: Получить логин авторизованного пользователя. Если авторизация не производилась - вернуть None
:param inAuthTokenStr: Токен авторизации пользователя / бота, по умолчанию None (не установлен)
:type inAuthTokenStr: str, опционально
:return: Логин пользователя
:rtype: str
"""
if inAuthTokenStr is None: return None
inGS = GSettingsGet() # Get the global settings
return inGS.get("ServerDict", {}).get("AccessUsers", {}).get("AuthTokensDict", {}).get(inAuthTokenStr, {}).get("User", None)
def WebUserDomainGet(inAuthTokenStr: str=None) -> str:
"""L+,W+: Получить домен авторизованного пользователя. Если авторизация не производилась - вернуть None
:param inAuthTokenStr: Токен авторизации пользователя / бота, по умолчанию None (не установлен)
:type inAuthTokenStr: str, опционально
:return: Домен пользователя
:rtype: str
"""
if inAuthTokenStr is None: return None
inGS = GSettingsGet() # Get the global settings
return inGS.get("ServerDict", {}).get("AccessUsers", {}).get("AuthTokensDict", {}).get(inAuthTokenStr, {}).get("Domain", None)
def WebUserInfoGet(inRequest=None):
"""L+,W+: Информация о пользователе, который отправил HTTP запрос.
"""L+,W+: [ПРЕКРАЩЕНИЕ ПОДДЕРЖКИ В 1.3.2, см. WebUserLoginGet, WebUserDomainGet] Информация о пользователе, который отправил HTTP запрос.
:param inRequest: Экземпляр HTTP request. Опционален, если сообщение фиксируется из под потока, который был инициирован запросом пользователя
:return: Сведения в формате {"DomainUpperStr": "PYOPENRPA", "UserNameUpperStr": "IVAN.MASLOV"}
@ -1159,29 +1194,59 @@ def WebUserInfoGet(inRequest=None):
lUserUpperStr = inRequest.OpenRPA["User"].upper()
return {"DomainUpperStr": lDomainUpperStr, "UserNameUpperStr": lUserUpperStr}
def WebUserIsSuperToken(inRequest = None, inGSettings = None):
"""L+,W+: Проверить, авторизован ли HTTP запрос с помощью супер токена (токен, который не истекает).
def WebUserIsSuperToken(inAuthTokenStr: str=None):
"""L+,W+: [ИЗМЕНЕНИЕ В 1.3.1] Проверить, авторизован ли HTTP запрос с помощью супер токена (токен, который не истекает).
:param inRequest: Экземпляр HTTP request. Опционален, если сообщение фиксируется из под потока, который был инициирован запросом пользователя
:param inGSettings: Глобальный словарь настроек Оркестратора (синглтон)
:return: True - является супертокеном; False - не является супертокеном
:param inAuthTokenStr: Токен авторизации пользователя / бота, по умолчанию None (не установлен)
:type inAuthTokenStr: str, опционально
:return: True - является супертокеном; False - не является супертокеном; None - авторизация не производилась
"""
if inRequest is None: inRequest = WebRequestGet()
inGSettings = GSettingsGet(inGSettings=inGSettings) # Set the global settings
if inAuthTokenStr is None: return None
inGSettings = GSettingsGet(inGSettings=inGSettings) # Get the global settings
lIsSuperTokenBool = False
# Get Flag is supertoken (True|False)
lIsSuperTokenBool = inGSettings.get("ServerDict", {}).get("AccessUsers", {}).get("AuthTokensDict", {}).get(inRequest.OpenRPA["AuthToken"], {}).get("FlagDoNotExpire", False)
lIsSuperTokenBool = inGSettings.get("ServerDict", {}).get("AccessUsers", {}).get("AuthTokensDict", {}).get(inAuthTokenStr, {}).get("FlagDoNotExpire", False)
return lIsSuperTokenBool
def WebUserUACHierarchyGet(inRequest = None):
"""L+,W+: Вернуть словарь доступа UAC в отношении пользователя, который выполнил HTTP запрос inRequest
def WebUserUACHierarchyGet(inAuthTokenStr: str=None) -> dict:
"""L+,W+: [ИЗМЕНЕНИЕ В 1.3.1] Вернуть словарь доступа UAC в отношении пользователя, который выполнил HTTP запрос inRequest
:param inRequest: Экземпляр HTTP request. Опционален, если сообщение фиксируется из под потока, который был инициирован запросом пользователя
:param inAuthTokenStr: Токен авторизации пользователя / бота, по умолчанию None (не установлен)
:type inAuthTokenStr: str, опционально
:return: UAC словарь доступа или {}, что означает полный доступ
"""
if inRequest is None: inRequest = WebRequestGet()
return inRequest.UserRoleHierarchyGet()
lDomainUpperStr = WebUserDomainGet(inAuthTokenStr=inAuthTokenStr).upper()
lUserUpperStr = WebUserLoginGet(inAuthTokenStr=inAuthTokenStr).upper()
if lUserUpperStr is None: return {}
else: return GSettingsGet().get("ServerDict", {}).get("AccessUsers", {}).get("RuleDomainUserDict", {}).get((lDomainUpperStr, lUserUpperStr), {}).get("RoleHierarchyAllowedDict", {})
def WebUserUACCheck(inAuthTokenStr:str=None, inKeyList:list=None) -> bool:
"""L+,W+: Проверить UAC доступ списка ключей для пользователя
:param inAuthTokenStr: Токен авторизации пользователя / бота, по умолчанию None (не установлен)
:type inAuthTokenStr: str, опционально
:return: True - доступ имеется, False - доступа нет
:rtype: bool
"""
if inAuthTokenStr is None: return True # Если авторизации не происходило - супердоступ
lResult = True # Init flag
lRoleHierarchyDict = WebUserUACHierarchyGet(inAuthTokenStr=inAuthTokenStr) # get the Hierarchy
# Try to get value from key list
lKeyValue = lRoleHierarchyDict # Init the base
for lItem in inKeyList:
if type(lKeyValue) is dict:
if lItem in lKeyValue: # Has key
lKeyValue = lKeyValue[lItem] # Get the value and go to the next loop iteration
else: # Else branch - true or false
if len(lKeyValue)>0: # False - if Dict has some elements
lResult = False # Set the False Flag
else:
lResult = True # Set the True flag
break # Stop the loop
else: # Has element with no detalization - return True
lResult = True # Set the flag
break # Close the loop
return lResult # Return the result
## GSettings defs

Write Preview
Loading…
Cancel
Save