# Copyright (C) 2023 Loren Eteval # # This file is part of Furious. # # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program. If not, see . from Furious.Utility.Constants import APP, PLATFORM, LogType from Furious.Utility.Utility import getAbsolutePath, eventLoopWait, isPythonw from PySide6 import QtCore import io import os import sys import uuid import time import ujson import logging import functools import threading import multiprocessing logger = logging.getLogger(__name__) # 300ms STDOUT_REDIRECT_THRESHOLD = 300 class Core: class ExitCode: ConfigurationError = 23 # Windows: 4294967295. Darwin, Linux: 255 (-1) ServerStartFailure = 4294967295 if PLATFORM == 'Windows' else 255 # Windows shutting down SystemShuttingDown = 0x40010004 def __init__(self, *args, exitCallback=None, **kwargs): self._process = None self._exitCallback = exitCallback self._msgQueue = multiprocessing.Queue() @QtCore.Slot() def appendCoreLog(): line = self.getLineNoWait() if line and not line.isspace(): APP().logViewerWidget.appendLog(LogType.Core, line) self._stdoutTimer = QtCore.QTimer() self._stdoutTimer.timeout.connect(appendCoreLog) @QtCore.Slot() def timeoutCallback(): self.checkAlive() self._daemonTimer = QtCore.QTimer() self._daemonTimer.timeout.connect(timeoutCallback) @staticmethod def name(): raise NotImplementedError @staticmethod def version(): raise NotImplementedError def registerExitCallback(self, exitCallback): self._exitCallback = exitCallback def isAlive(self): if isinstance(self._process, multiprocessing.Process): return self._process.is_alive() else: return False def checkAlive(self): if isinstance(self._process, multiprocessing.Process): if self._process.is_alive(): return True else: logger.error( f'{self.name()} stopped unexpectedly with exitcode {self._process.exitcode}' ) self._stdoutTimer.stop() self._daemonTimer.stop() if callable(self._exitCallback): self._exitCallback(self._process.exitcode) # Reset internal process self._process = None return False else: return False def start(self, *args, **kwargs): logger.info(f'{self.name()} {self.version()} started') waitCore = kwargs.pop('waitCore', True) waitTime = kwargs.pop('waitTime', 2000) self._process = multiprocessing.Process(**kwargs, daemon=True) self._process.start() if waitCore: # Wait for the core to start up completely eventLoopWait(waitTime) if self.checkAlive(): # Start core daemon self._stdoutTimer.start(STDOUT_REDIRECT_THRESHOLD) self._daemonTimer.start(2000) def stop(self): if self.isAlive(): self._stdoutTimer.stop() self._daemonTimer.stop() self._process.terminate() self._process.join() logger.info( f'{self.name()} terminated with exitcode {self._process.exitcode}' ) def getLineNoWait(self): try: return self._msgQueue.get_nowait() except Exception: # Any non-exit exceptions return '' class StdoutRedirectHelper: TemporaryDir = QtCore.QTemporaryDir() @staticmethod def launch(msgQueue, entrypoint, redirect): if not callable(entrypoint): return if ( not StdoutRedirectHelper.TemporaryDir.isValid() or not redirect # pythonw.exe or isPythonw() ): # Call entrypoint directly entrypoint() return temporaryFile = StdoutRedirectHelper.TemporaryDir.filePath(str(uuid.uuid4())) tmpFileStream = open(temporaryFile, 'w+b') stdoutFileno_ = sys.stdout.fileno() stderrFileno_ = sys.stderr.fileno() sys.stdout.close() sys.stderr.close() # Redirect os.dup2(tmpFileStream.fileno(), stdoutFileno_) os.dup2(tmpFileStream.fileno(), stderrFileno_) sys.stdout = tmpFileStream sys.stderr = tmpFileStream def produceMsg(): with open(temporaryFile, 'rb') as file: while True: for line in iter(file.readline, b''): if line and not line.isspace(): try: msgQueue.put_nowait(line.decode('utf-8', 'replace')) except Exception: # Any non-exit exceptions pass time.sleep(STDOUT_REDIRECT_THRESHOLD / 1000) msgThread = threading.Thread(target=produceMsg, daemon=True) msgThread.start() with tmpFileStream: entrypoint() def startXrayCore(json, msgQueue): try: import xray except ImportError: # Fake running process while True: pass else: if xray.__version__ <= '1.8.4': redirect = False else: redirect = True # Can be redirected if not isPythonw(): StdoutRedirectHelper.launch( msgQueue, lambda: xray.startFromJSON(json), redirect ) return if not redirect: xray.startFromJSON(json) return def xrayPythonwProduceMsg(): try: jsonObject = ujson.loads(json) except Exception: # Any non-exit exceptions xray.startFromJSON(json) return loggerPath = [] fileStream = [] for attr in ['access', 'error']: try: path = jsonObject['log'][attr] except Exception: # Any non-exit exceptions continue if path not in loggerPath: loggerPath.append(path) try: stream = open(path, 'rb') except Exception: # Any non-exit exceptions pass else: stream.seek(0, io.SEEK_END) fileStream.append(stream) def produceMsg(): while True: for file in fileStream: for line in iter(file.readline, b''): if line and not line.isspace(): try: msgQueue.put_nowait(line.decode('utf-8', 'replace')) except Exception: # Any non-exit exceptions pass time.sleep(STDOUT_REDIRECT_THRESHOLD / 1000) try: msgThread = threading.Thread(target=produceMsg, daemon=True) msgThread.start() xray.startFromJSON(json) finally: for stream in fileStream: stream.close() xrayPythonwProduceMsg() class XrayCore(Core): class ExitCode: ConfigurationError = 23 # Windows: 4294967295. Darwin, Linux: 255 (-1) ServerStartFailure = 4294967295 if PLATFORM == 'Windows' else 255 # Windows shutting down SystemShuttingDown = 0x40010004 def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) @staticmethod def name(): return 'Xray-core' @staticmethod @functools.lru_cache(None) def version(): try: import xray return xray.__version__ except Exception: # Any non-exit exceptions return '0.0.0' def start(self, json, **kwargs): super().start(target=startXrayCore, args=(json, self._msgQueue), **kwargs) def startHysteria1(json, rule, mmdb, msgQueue): try: import hysteria except ImportError: # Fake running process while True: pass else: if hysteria.__version__ <= '1.3.5': redirect = False else: redirect = True StdoutRedirectHelper.launch( msgQueue, lambda: hysteria.startFromJSON(json, rule, mmdb), redirect ) class Hysteria1(Core): class ExitCode: ConfigurationError = 23 RemoteNetworkError = 3 # Windows shutting down SystemShuttingDown = 0x40010004 def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) @staticmethod def rule(rulePath): path = getAbsolutePath(str(rulePath)) try: with open(path, 'rb') as file: data = file.read() logger.info(f'hysteria1 rule \'{path}\' load success') return data except Exception as ex: # Any non-exit exceptions logger.error(f'hysteria1 rule \'{path}\' load failed. {ex}') return '' @staticmethod def mmdb(mmdbPath): path = getAbsolutePath(str(mmdbPath)) try: with open(path, 'rb') as file: data = file.read() logger.info(f'hysteria1 mmdb \'{path}\' load success') return data except Exception as ex: # Any non-exit exceptions logger.error(f'hysteria1 mmdb \'{path}\' load failed. {ex}') return '' @staticmethod def name(): return 'Hysteria1' @staticmethod @functools.lru_cache(None) def version(): try: import hysteria return hysteria.__version__ except Exception: # Any non-exit exceptions return '0.0.0' def start(self, json, rule, mmdb, **kwargs): super().start( target=startHysteria1, args=(json, rule, mmdb, self._msgQueue), **kwargs ) def startHysteria2(json, msgQueue): try: import hysteria2 except ImportError: # Fake running process while True: pass else: if hysteria2.__version__ <= '2.0.0.1': redirect = False else: redirect = True StdoutRedirectHelper.launch( msgQueue, lambda: hysteria2.startFromJSON(json), redirect ) class Hysteria2(Core): class ExitCode: ConfigurationError = 23 # Windows: 4294967295. Darwin, Linux: 255 (-1) ServerStartFailure = 4294967295 if PLATFORM == 'Windows' else 255 # Windows shutting down SystemShuttingDown = 0x40010004 def __init__(self, *args, **kwargs): super().__init__(*args, **kwargs) @staticmethod def name(): return 'Hysteria2' @staticmethod @functools.lru_cache(None) def version(): try: import hysteria2 return hysteria2.__version__ except Exception: # Any non-exit exceptions return '0.0.0' def start(self, json, **kwargs): super().start(target=startHysteria2, args=(json, self._msgQueue), **kwargs)