en2zmax / patch-ups5-for-uc-flora.py

import json import logging import traceback import time import os import jwt import asyncio from urllib.parse import unquote, urlencode from datetime import datetime import aiohttp from aiohttp import web from app.tools import tools from app.auth import key, alg, checkRight, userlog, hasRights from app.db import DB from app.logger import getProxyLogger proxy_logger = getProxyLogger() try: config = json.load(open('app/conf/proxy.json', 'r')) except Exception as error: proxy_logger.error(f"Ошибка загрузки конфигурации прокси: {error}") proxy_logger.error(traceback.format_exc()) raise kibana = tools.params['servers']['res']['cluster'][0] kibana_config = { 'url': kibana, 'username': tools.params['servers']['res'].get('auth', {}).get('username', ''), 'password': tools.params['servers']['res'].get('auth', {}).get('password', '') } class ProxyHandler: def __init__(self): self.sessions = {} async def close(self): """Закрытие всех сессий""" for session_dict in self.sessions.values(): for session in session_dict.values(): await session.close() def get_session(self, service, method, path): t = config.get(service,{}).get('cache',{}).get(method,{}).get(path) if t: connect_timeout = t read_timeout = t else: connect_timeout=tools.params['servers'][service]['timers']['connect'] read_timeout=tools.params['servers'][service]['timers']['read'] session_key = f"{connect_timeout}_{read_timeout}" if service not in self.sessions: self.sessions[service] = {} if session_key not in self.sessions[service]: timeout = aiohttp.ClientTimeout( total=None, connect=connect_timeout, sock_read=read_timeout ) self.sessions[service][session_key] = aiohttp.ClientSession(timeout=timeout, auto_decompress=False) return self.sessions[service][session_key] async def response_postprocessing(self, response_data, dataset, request_path, request_params=None, request_method='GET', raw_response=None): """Постобработка ответа - фильтрация по dataset""" if dataset is None: dataset = "0" def new_value(value, dataset): if dataset == "0": if "~" in value: return None else: if f"~{dataset}" not in value: return None value = "~".join(value.split('~')[:-1]) return value # Логирование классификации if '/smc/classify' in request_path: corpus_name = datetime.today().strftime('%Y-%m-%d') + '.csv' t = datetime.now().strftime("%H:%M:%S") try: model = unquote(request_path.split('/')[-1].split('?')[0]) os.makedirs(f'data/smc/logs/{model}', exist_ok=True) with open(f'data/smc/logs/{model}/{corpus_name}', 'a') as f: query_text = request_params.get('text', '') if request_params else '' response_text = raw_response if raw_response else json.dumps(response_data) f.write(f"{t}\t{query_text}\t{response_text}\n") except Exception as e: proxy_logger.error(f"[SMC ERROR] Ошибка при логировании классификации: {e}") if not isinstance(response_data, dict): return response_data if response_data.get('error'): return response_data # Фильтрация результатов по dataset for service, params in config.items(): if not request_path.startswith(f"/{service}"): continue for method in params.get("filters", {}): if method != request_method: continue paths = params['filters'][method] for path, filters in paths.items(): if not request_path.startswith(f"/{service}{path}"): continue for cnf in filters: # Фильтрация списков if cnf['type'] == 'list': if cnf['name'] in response_data and isinstance(response_data[cnf['name']], list): filtered_list = [] for item in response_data[cnf['name']]: value = new_value(item, dataset) if value is not None: filtered_list.append(value) response_data[cnf['name']] = filtered_list # Фильтрация ключей elif cnf['type'] == 'key': keys_to_process = list(response_data.keys()) for resp_key in keys_to_process: if resp_key in cnf.get('exclude', []): continue value = new_value(resp_key, dataset) if value is None: del response_data[resp_key] elif resp_key != value: response_data[value] = response_data[resp_key] del response_data[resp_key] return response_data async def modify_request_data(self, request, service, path, method, dataset, user_id): """Модификация данных запроса - добавление dataset""" # Без авторизации или с дефолтным dataset не модифицируем if user_id is None or dataset == "0" or dataset is None or dataset == "None": return request dataset_config = config.get(service, {}).get('datasets', {}).get(method, {}).get(path) if not dataset_config: return request for param in dataset_config: if param['type'] == 'path': # Модификация пути path_parts = request['full_path'].split('?') url_parts = path_parts[0].split('/') if not param.get('splitter'): if url_parts[param['position']] in param.get('exclude', []): return request if f"~{dataset}" not in url_parts[param['position']]: url_parts[param['position']] = f"{url_parts[param['position']]}~{dataset}" else: split_parts = url_parts[param['position']].split(param['splitter']) for i in range(len(split_parts)): if split_parts[i] not in param.get('exclude', []): if f"~{dataset}" not in split_parts[i]: split_parts[i] = f"{split_parts[i].strip()}~{dataset}" url_parts[param['position']] = param['splitter'].join(split_parts) path_parts[0] = '/'.join(url_parts) request['full_path'] = '?'.join(path_parts) elif param['type'] == 'param': # Модификация параметра param_name = param['name'] if 'params' in request and param_name in request['params']: value = request['params'][param_name] if value not in param.get('exclude', []) and f"~{dataset}" not in value: request['params'][param_name] = f"{value}~{dataset}" elif 'data' in request and isinstance(request['data'], dict) and param_name in request['data']: value = request['data'][param_name] if value not in param.get('exclude', []) and f"~{dataset}" not in value: request['data'][param_name] = f"{value}~{dataset}" elif param['type'] == 'insert': # Вставка нового параметра param_name = param['name'] if method == 'GET' and 'params' in request: request['params'][param_name] = dataset elif 'data' in request and isinstance(request['data'], dict): # Если data это dict - добавляем в data request['data'][param_name] = dataset elif 'data' in request and isinstance(request['data'], bytes): # Если data это urlencoded/multipart - добавляем в квери if 'params' not in request: request['params'] = {} request['params'][param_name] = dataset return request async def change_request(self, request, service, servertype): """Специальная обработка запросов для разных сервисов""" path = request['full_path'] method = request['method'] # Обрабатываем data - может быть dict или bytes data = request.get('data', {}) if isinstance(data, bytes): # Если это сырые байты, пытаемся распарсить return 0, request, servertype # SMC/SEE: Проверка на модели-черновики if '/see/entities' in path or '/smc/classify' in path: model = path.split('/')[-1].split('?')[0] if len(model) > 4 and model[-4:] == '-new': servertype = 'trainer' # SMC: Замена корпуса на группы elif '/smc/groups' in path and method == 'POST': if isinstance(data, dict) and 'corpus' in data: corpus = data['corpus'] corpus_file = f"data/db/marks/{corpus}.json" if not os.path.isfile(corpus_file): return 1, f"CorpusNotFound: {corpus}", servertype try: grouplist = json.load(open(corpus_file, 'r'))['groups'] groups = dict() for item in grouplist: if not item[1] in groups: groups[item[1]] = list() groups[item[1]].append(list(item[0])) data['groups'] = json.dumps(groups) del data['corpus'] request['data'] = data except Exception: return 1, "GroupsLoadFailed", servertype # SMC: Обучение/тестирование elif '/smc/train' in path or '/smc/test' in path: if isinstance(data, dict) and 'corpus' in data: corpus = data['corpus'] is_marks = data.get('isMarks', '0') corpus_path = f"data/db/marks/{corpus}.json" if not os.path.isfile(corpus_path): return 1, f"CorpusNotFound: {corpus}", servertype try: corpus_name = json.load(open(corpus_path, 'r'))['name'] except Exception: return 1, f"CorpusCorrupted: {corpus}", servertype if str(is_marks) != '1': csv, _ = tools.loadCorpusToCSV(corpusFile=corpus_path) else: csv = open(corpus_path, 'r').read() data['corpusName'] = corpus_name data['corpusData'] = csv del data['corpus'] request['data'] = data if not path.endswith('-new'): request['full_path'] = path + '-new' # SEE: Обучение elif '/see/train' in path: if isinstance(data, dict) and 'corpus' in data: corpus = data['corpus'] corpus_path = f"data/db/marks/{corpus}.json" if not os.path.isfile(corpus_path): return 1, f"CorpusNotFound: {corpus}", servertype try: corpus_name = json.load(open(corpus_path, 'r'))['name'] except Exception: return 1, f"CorpusCorrupted: {corpus}", servertype csv, _ = tools.loadCorpusToCSV(corpusFile=corpus_path) data['corpusName'] = corpus_name data['corpusData'] = csv del data['corpus'] request['data'] = data if not path.endswith('-new'): request['full_path'] = path + '-new' # SMC: Автоматическая разметка elif '/smc/automark/start' in path: if isinstance(data, dict): corpus = data.get('corpus') marks = data.get('marks') if corpus and marks: try: corpus_data = json.load(open(f"data/db/marks/{corpus}.json", "r"))['data'] except Exception: return 1, f"CorpusCorrupted: {corpus}", servertype try: marks_data = json.load(open(f"data/db/marks/{marks}.json", "r"))['data'] except Exception: return 1, f"CorpusCorrupted: {marks}", servertype data['corpus'] = json.dumps(corpus_data) data['marks'] = json.dumps(marks_data) request['data'] = data return 0, request, servertype def parse_multipart_form(self, content, content_type): """Парсинг multipart/form-data""" if not content_type or 'multipart/form-data' not in content_type: return None try: boundary = content_type.split('boundary=')[-1].strip() if boundary.startswith('"') and boundary.endswith('"'): boundary = boundary[1:-1] parts = content.split(f'--{boundary}'.encode()) form_data = {} for part in parts: if not part or part == b'--\r\n' or part == b'--': continue # Разделяем заголовки и данные if b'\r\n\r\n' in part: headers_part, data_part = part.split(b'\r\n\r\n', 1) # Ищем имя поля name_match = None for line in headers_part.split(b'\r\n'): if b'name="' in line: name_start = line.find(b'name="') + 6 name_end = line.find(b'"', name_start) if name_end > name_start: name_match = line[name_start:name_end].decode('utf-8', errors='ignore') break if name_match: # Убираем trailing \r\n data_part = data_part.rstrip(b'\r\n') form_data[name_match] = data_part.decode('utf-8', errors='ignore') return form_data except Exception as e: proxy_logger.error(f"[MULTIPART] Parse error: {e}") return None def parse_urlencoded_form(self, content): """Парсинг application/x-www-form-urlencoded""" try: from urllib.parse import parse_qs decoded = content.decode('utf-8', errors='ignore') parsed = parse_qs(decoded, keep_blank_values=True) # Преобразуем списки в единичные значения return {k: v[0] if len(v) == 1 else v for k, v in parsed.items()} except Exception as e: proxy_logger.error(f"[URLENCODED] Parse error: {e}") return None async def proxy_request(self, service, path, method, headers, params, data, dataset, user_id): """Основная функция проксирования""" proxy_logger.info(f"[PROXY] {method} /{service}{path}") # Определяем тип сервера и путь из конфигурации servertype = None matched_path = None for conf_method, paths in config[service].get('requests', {}).items(): if conf_method != method: continue for conf_path, conf_servertype in paths.items(): if path.startswith(conf_path): servertype = conf_servertype matched_path = conf_path break if servertype: break if not servertype: return {'error': 1, 'message': 'Route not found'}, 404 # Статические файлы if "/" in servertype: if not os.path.isfile(servertype): return {'error': 1, 'message': 'File not found'}, 404 with open(servertype, 'rb') as f: return f.read(), 200, {'Content-Type': 'text/javascript; charset=utf-8'} # Фейковый запрос if servertype == 'fake': return b'', 200, {'Content-Type': 'application/octet-stream', 'Content-Length': '50'} request_data = { 'method': method, 'path': path, 'full_path': f"/{service}{path}", 'headers': dict(headers), 'params': params or {}, 'data': data or {} } # Специальная обработка error, result, servertype = await self.change_request(request_data, service, servertype) if error: return {'error': 1, 'message': result}, 405 request_data = result # Добавление dataset request_data = await self.modify_request_data( request_data, service, matched_path, method, dataset, user_id ) # Определение серверов для запроса if servertype in ['find', 'all']: servers = list(tools.params['servers'][service]['cluster']) if not isinstance(servers, list): servers = list([servers]) else: servers = list(servers) # Специальная обработка для архива if '/archive/dialog' in path: dialog_id = path.split('/')[-1] servers.insert(0, (tools.params['servers']['res']['trainer'], f'/res/dialog/{dialog_id}')) elif '/archive/list' in path: res_path = '/res/archive/list' if '?' in path: res_path += '?' + path.split('?')[-1] servers.insert(0, (tools.params['servers']['res']['trainer'], res_path)) else: servers = None # Выполнение запроса if servertype == 'find': return await self.proxy_find(service, request_data, servers, dataset) elif servertype == 'all': return await self.proxy_all(service, request_data, servers, dataset) else: return await self.proxy_single(service, request_data, servertype, dataset) async def proxy_single(self, service, request_data, servertype, dataset): """Проксирование на один сервер (cluster/trainer) с повторными попытками при ошибках""" if servertype == 'trainer': max_attempts = 1 else: # cluster servers = tools.params['servers'][service].get('cluster', []) max_attempts = len(servers) if isinstance(servers, list) else 1 method = request_data['method'] session = self.get_session(service=service, method=method, path=request_data['path']) for attempt in range(max_attempts): server_url = tools.getProductionServer(servicetype=service, servertype=servertype) url = server_url + request_data['full_path'] headers = {k: v for k, v in request_data['headers'].items() if k.lower() not in ['content-length']} proxy_logger.info(f"[SINGLE][{service.upper()}] Attempt {attempt + 1}/{max_attempts} -> {server_url}") try: if method == 'GET': resp = await session.get(url, params=request_data['params'], headers=headers) # это streaming ответ content_type = resp.headers.get('Content-Type', '') if 'text/event-stream' in content_type: proxy_logger.info(f"[STREAMING][{service.upper()}] {server_url}") # Возвращаем маркер streaming и response объект return ('stream', resp), resp.status # Обычный ответ - читаем полностью async with resp: content = await resp.read() status = resp.status elif method == 'POST': if isinstance(request_data['data'], dict): headers['Content-Type'] = 'application/json' resp = await session.post(url, json=request_data['data'], headers=headers, params=request_data['params']) else: resp = await session.post(url, data=request_data['data'], headers=headers, params=request_data['params']) # Проверка на streaming ответ content_type = resp.headers.get('Content-Type', '') if 'text/event-stream' in content_type: proxy_logger.info(f"[STREAMING][{service.upper()}] {server_url}") return ('stream', resp), resp.status # Обычный ответ - читаем полностью async with resp: content = await resp.read() status = resp.status elif method == 'DELETE': if isinstance(request_data['data'], dict): headers['Content-Type'] = 'application/json' async with session.delete(url, json=request_data['data'], headers=headers, params=request_data['params']) as resp: content = await resp.read() status = resp.status else: async with session.delete(url, data=request_data['data'], headers=headers, params=request_data['params']) as resp: content = await resp.read() status = resp.status elif method == 'PUT': if isinstance(request_data['data'], dict): headers['Content-Type'] = 'application/json' async with session.put(url, json=request_data['data'], headers=headers, params=request_data['params']) as resp: content = await resp.read() status = resp.status else: async with session.put(url, data=request_data['data'], headers=headers, params=request_data['params']) as resp: content = await resp.read() status = resp.status else: return {'error': 1, 'message': 'Method not supported'}, 405 if status >= 400: try: error_text = content.decode('utf-8') proxy_logger.error(f"[RESPONSE ERROR][{service.upper()}] ERROR {status} {server_url}: {error_text}") except: proxy_logger.error(f"[RESPONSE ERROR][{service.upper()}] ERROR {status} {server_url}: (binary data, {len(content)} bytes)") else: proxy_logger.info(f"[RESPONSE][{service.upper()}] {status} {server_url}") try: response_data = json.loads(content.decode('utf-8')) response_data = await self.response_postprocessing( response_data, dataset, request_data['full_path'], request_data.get('params'), request_data['method'] ) # Сохраним результат в кэше, если это требуется id = self.get_cache_id(service=service,method=method,path=request_data['path'],dataset=dataset) if id: cont = json.dumps(response_data) error, result = DB.get(table='cache',id=id,md5clean=False) if error: logging.info(DB.create(table='cache',data={'id': id, 'content': cont})) else: if DB.md5(cont) != result['md5']: DB.modify(table='cache',data={'id': id, 'content': cont}) return response_data, status except Exception: return content, status except (asyncio.TimeoutError, Exception) as e: error_type = "TIMEOUT" if isinstance(e, asyncio.TimeoutError) else "ERROR" proxy_logger.error(f"[{error_type}][{service.upper()}] {server_url}: {str(e)}") tools.setFailedServer(server_url, service) # Если таймаут, то пробуем вернуть из кэша if error_type == "TIMEOUT": id = self.get_cache_id(service=service,method=method,path=request_data['path'],dataset=dataset) if id: error, result = DB.get(table='cache',id=id,md5clean=False) if not error: return json.loads(result['content']), 200 # Если это последняя попытка - возвращаем ошибку if attempt == max_attempts - 1: if isinstance(e, asyncio.TimeoutError): return {'error': 1, 'message': 'Request timeout'}, 504 return {'error': 1, 'message': str(e)}, 500 proxy_logger.info("[RETRY] Trying next server...") continue return {'error': 1, 'message': 'No servers available'}, 503 def get_cache_id(self,service,method,path,dataset=None): if path in config.get(service,{}).get('cache',{}).get(method,{}): cache_key = f"{method}:/{service}{path}:{dataset if dataset else '0'}" return DB.md5(cache_key) else: return False async def proxy_find(self, service, request_data, servers, dataset): """Поиск на первом доступном сервере""" method = request_data['method'] session = self.get_session(service=service, method=method, path=request_data['path']) for server_info in servers: if isinstance(server_info, tuple): server_url, custom_path = server_info url = server_url + custom_path else: server_url = server_info url = server_url + request_data['full_path'] proxy_logger.info(f"[FIND][{service.upper()}] Trying {server_url}") try: method = request_data['method'] headers = {k: v for k, v in request_data['headers'].items() if k.lower() not in ['content-length']} if method == 'GET': async with session.get(url, params=request_data['params'], headers=headers) as resp: if resp.status == 200: content = await resp.read() try: response_data = json.loads(content.decode('utf-8')) response_data = await self.response_postprocessing( response_data, dataset, request_data['full_path'], request_data.get('params'), request_data['method'] ) proxy_logger.info(f"[FIND] Success on {server_url}") return response_data, 200 except Exception: return content, 200 else: proxy_logger.info(f"[FIND] {resp.status} from {server_url}, trying next") continue else: # Для POST/DELETE/PUT аналогично async with session.request(method, url, data=request_data.get('data'), headers=headers) as resp: if resp.status == 200: content = await resp.read() try: response_data = json.loads(content.decode('utf-8')) response_data = await self.response_postprocessing( response_data, dataset, request_data['full_path'], request_data.get('params'), request_data['method'] ) return response_data, 200 except Exception: return content, 200 else: continue except Exception as e: proxy_logger.error(f"[FIND] Error on {server_url}: {str(e)}") tools.setFailedServer(server_url, service) continue return {'error': 1, 'message': 'No servers available'}, 503 async def proxy_all(self, service, request_data, servers, dataset): """Сбор данных со всех серверов и объединение результатов""" method = request_data['method'] session = self.get_session(service=service, method=method, path=request_data['path']) tasks = [] for server_info in servers: if isinstance(server_info, tuple): server_url, custom_path = server_info url = server_url + custom_path else: server_url = server_info url = server_url + request_data['full_path'] tasks.append(self._fetch_from_server(session, request_data, url, server_url, service)) responses = await asyncio.gather(*tasks, return_exceptions=True) # Объединение результатов combined_result = None for resp_data in responses: if isinstance(resp_data, Exception): continue if resp_data is None: continue if combined_result is None: combined_result = [] if isinstance(resp_data, list) else {} if isinstance(combined_result, list) and isinstance(resp_data, list): combined_result.extend(resp_data) elif isinstance(combined_result, dict) and isinstance(resp_data, dict): combined_result.update(resp_data) if combined_result is None: return {'error': 1, 'message': 'No data collected'}, 503 # Уникализация и сортировка if isinstance(combined_result, list) and combined_result: if 'uuid' in str(combined_result): seen_uuids = set() unique_result = [] for item in combined_result: if isinstance(item, dict) and 'uuid' in item: if item['uuid'] not in seen_uuids: seen_uuids.add(item['uuid']) unique_result.append(item) else: unique_result.append(item) combined_result = unique_result if combined_result and isinstance(combined_result[0], dict) and 'datetime' in combined_result[0]: combined_result = sorted(combined_result, key=lambda x: x.get('datetime', '')) combined_result = tools.add_utc_suffix(combined_result) combined_result = await self.response_postprocessing( combined_result, dataset, request_data['full_path'], request_data.get('params'), request_data['method'] ) return combined_result, 200 async def _fetch_from_server(self, session, request_data, url, server_url, service): """Вспомогательная функция для получения данных с одного сервера""" try: method = request_data['method'] headers = {k: v for k, v in request_data['headers'].items() if k.lower() not in ['content-length']} async with session.request( method, url, params=request_data.get('params') if method == 'GET' else None, data=request_data.get('data') if method != 'GET' else None, headers=headers ) as resp: if resp.status == 200: content = await resp.read() return json.loads(content.decode('utf-8')) return None except Exception as e: proxy_logger.error(f"[ALL] Error on {server_url}: {str(e)}") tools.setFailedServer(server_url, service) return None proxy_handler = ProxyHandler() async def handle_proxy(request): """Основной обработчик для всех запросов""" path = request.path method = request.method proxy_logger.info(f"[INCOMING] {method} {path}") if path == "/": return await handle_internal_api(request) # Проверка на внутренние запросы for internal_path in config.get('internals', {}).get(method, []): if path.startswith(internal_path): proxy_logger.info(f"[INTERNAL] {method} {path} -> Flask") return await handle_internal_api(request) # Определение сервиса service = None service_path = None for serv in config.keys(): if serv == 'internals': continue if path.startswith(f"/{serv}"): service = serv service_path = path[len(f"/{serv}"):] break # Проверка на Kibana if not service: return await handle_kibana(request) # Аутентификация token = request.headers.get('x-access-token', '') error, message, user_id = checkRight(token, path, method) if error: client_ip = request.remote ses_trainer_ip = tools.params['servers']['ses']['trainer'].split('://')[1].split(':')[0] # Пропускаем без токена если запрос от локального IP или SES trainer (отсекаем только внешние) if client_ip == '127.0.0.1' or client_ip == ses_trainer_ip: user_id = None else: proxy_logger.warning(f"[AUTH FAILED] {client_ip}, path: {path}, error: {message}") return web.json_response({'error': 1, 'message': message}, status=error) # Получение dataset пользователя dataset = None if user_id: error, user_data = DB.get('users', user_id) if error: return web.json_response({'error': 1, 'message': user_data}, status=405) dataset = user_data.get('dataset', '0') # Проверим, существует ли этот набор error, d = DB.get(table='datasets', id=dataset) if error: return web.json_response({'error': 1, 'message': 'DatasetNotFound'}, status=405) # Чтение тела запроса один раз body = None if request.can_read_body and method in ['POST', 'PUT', 'DELETE']: body = await request.read() # Парсинг данных params = dict(request.query) if request.query else None data = None if body: content_type = request.headers.get('Content-Type', '') if 'application/json' in content_type: try: data = json.loads(body.decode('utf-8')) except Exception: data = body elif 'application/x-www-form-urlencoded' in content_type: data = body elif 'multipart/form-data' in content_type: data = body else: data = body # Логирование пользователя try: request_data = {} if params: request_data.update(params) if data and isinstance(data, dict): request_data.update(data) userlog(user_id, method, path, request_data) except Exception as e: proxy_logger.error(f"[USERLOG] Failed: {e}") # Проксирование try: result = await proxy_handler.proxy_request( service, service_path, method, request.headers, params, data, dataset, user_id ) if len(result) == 2: response_data, status = result headers_out = {} else: response_data, status, headers_out = result # Формирование ответа if isinstance(response_data, tuple) and len(response_data) > 0 and response_data[0] == 'stream': # Streaming ответ _, upstream_resp = response_data response = web.StreamResponse( status=status, headers={ 'Content-Type': 'text/event-stream', 'Cache-Control': 'no-cache', 'Transfer-Encoding': 'chunked' } ) await response.prepare(request) try: async for chunk in upstream_resp.content.iter_any(): if chunk: await response.write(chunk) except Exception as e: proxy_logger.error(f"[STREAM ERROR] {e}") finally: await upstream_resp.release() await response.write_eof() return response elif isinstance(response_data, (dict, list)): return web.json_response(response_data, status=status, headers=headers_out) else: return web.Response( body=response_data, status=status, headers=headers_out or {'Content-Type': 'application/octet-stream'} ) except Exception as e: proxy_logger.error(f"[ERROR] {traceback.format_exc()}") return web.json_response({'error': 1, 'message': str(e)}, status=500) async def handle_kibana(request): """Обработка запросов к Kibana""" token_cookie = request.cookies.get('accessToken', '') if not token_cookie: proxy_logger.warning("[KIBANA] No accessToken cookie found") return web.json_response({'error': 1, 'message': 'access denied'}, status=403) try: data = jwt.decode(token_cookie, key, algorithms=[alg]) if (data['type'] == 'access' and time.time() - data['created'] < tools.params['auth']['accessLifeTime'] and hasRights(data['user_id'], '/res/dashboards', 'GET')): # Проксирование на Kibana import base64 auth_string = f"{kibana_config['username']}:{kibana_config['password']}" basic_auth = base64.b64encode(auth_string.encode('utf-8')).decode('utf-8') headers = dict(request.headers) headers['Authorization'] = f'Basic {basic_auth}' headers = {k: v for k, v in headers.items() if k.lower() not in ['host', 'cookie']} async with aiohttp.ClientSession(auto_decompress=False) as session: url = kibana_config['url'] + request.path_qs async with session.request( request.method, url, data=await request.read() if request.can_read_body else None, headers=headers ) as resp: content = await resp.read() response_headers = {} for header_key, header_value in resp.headers.items(): if header_key.lower() not in ['connection', 'transfer-encoding', 'x-frame-options']: response_headers[header_key] = header_value response_headers.update({ 'Access-Control-Allow-Origin': '*', 'Access-Control-Allow-Credentials': 'true', 'Access-Control-Allow-Headers': 'x-access-token', 'x-frame-options': 'SAMEORIGIN' }) return web.Response(body=content, status=resp.status, headers=response_headers) else: proxy_logger.warning("[KIBANA] Access denied: token validation failed") return web.json_response({'error': 1, 'message': 'access denied'}, status=403) except Exception as e: proxy_logger.error(f"[KIBANA] Error: {e}") return web.json_response({'error': 1, 'message': 'access denied'}, status=403) async def handle_internal_api(request): """Проксирование на внутренний Flask""" try: headers = {k: v for k, v in request.headers.items() if k.lower() not in ['host']} async with aiohttp.ClientSession() as session: url = f"http://127.0.0.1:6282{request.path_qs}" async with session.request( request.method, url, data=await request.read() if request.can_read_body else None, headers=headers, params=request.query if request.query else None ) as resp: content = await resp.read() response_headers = dict(resp.headers) return web.Response( body=content, status=resp.status, headers=response_headers ) except Exception as e: proxy_logger.error(f"[INTERNAL API] Error: {str(e)}") return web.json_response({'error': 1, 'message': str(e)}, status=500) async def start_proxy_server(): max_body_size = tools.params.get('proxy', {}).get('maxBodySize', 209715200) # Default 200MB app = web.Application(client_max_size=max_body_size) app.router.add_route('*', '/{tail:.*}', handle_proxy) runner = web.AppRunner(app) await runner.setup() site = web.TCPSite(runner, '0.0.0.0', 6182) await site.start() proxy_logger.info("[START] Proxy server started on 0.0.0.0:6182") try: await asyncio.Future() finally: await proxy_handler.close() await runner.cleanup() def run_proxy(): asyncio.run(start_proxy_server())