Quantix
Platinum Üye
- Katılım
- 26 Tem 2025
- Mesajlar
- 202
- Beğeniler
- 32
Kod aşağıda açıklamaya gerek yok. Pipe bypass
import sysimport osimport timeimport threadingimport psutilimport win32pipeimport win32fileimport pywintypesimport win32conimport win32jobimport win32apiimport win32eventimport asyncioimport selectPIPE_NAME = r'\\.\pipe\QuantixVGCBypassPipe'VALORANT_EXECUTABLE = "VALORANT-Win64-Shipping.exe"RIOT_CLIENT_PATH = '"C:\\Riot Games\\Riot Client\\RiotClientServices.exe" --launch-product=valorant --launch-patchline=live'VGC_SERVICE = "vgc"valorant_running = Falsestopped_once = Falsecurrent_job = Nonepipe_handles = []monitor_thread = Nonemonitored_pids = set()monitored_lock = threading.Lock()monitoring_active = Falseshutdown_event = threading.Event()ASCII_ART = """ ____ ___ _ _ _ _______ ____ / __ \\__ \\| | | | | |/ /_ _|__ \\ | | | | ) | | | |__| | / | | ) | | | | |/ /| | | | |/ | | / / | |__| / /_| |_| | | | | / /_ \\____/____|\\___/ |_| |_| |____|"""def log_message(msg): print(f"[INFO] {msg}")def manage_service(action): os.system(f'sc {action} {VGC_SERVICE}') time.sleep(0.5) log_message(f"VGC service {action}")def override_vgc_pipe(): try: pipe = win32file.CreateFile( PIPE_NAME, win32con.GENERIC_READ | win32con.GENERIC_WRITE, 0, None, win32con.OPEN_EXISTING, 0, None) win32file.CloseHandle(pipe) log_message("VGC pipe overridden") except Exception: passasync def create_advanced_pipe(): global pipe_handles while not shutdown_event.is_set(): try: pipe = win32pipe.CreateNamedPipe( PIPE_NAME, win32con.PIPE_ACCESS_DUPLEX | win32con.FILE_FLAG_OVERLAPPED, win32con.PIPE_TYPE_MESSAGE | win32con.PIPE_READMODE_MESSAGE | win32con.PIPE_WAIT, win32con.PIPE_UNLIMITED_INSTANCES, 1048576, 1048576, 0, None) pipe_handles.append(pipe) overlapped = pywintypes.OVERLAPPED() overlapped.hEvent = win32event.CreateEvent(None, True, False, None) await asyncio.sleep(0) win32pipe.ConnectNamedPipe(pipe, overlapped) win32event.WaitForSingleObject(overlapped.hEvent, win32event.INFINITE) asyncio.create_task(handle_client_async(pipe)) log_message("New pipe instance created") except Exception as e: log_message(f"Pipe creation error: {e}") await asyncio.sleep(0.1)async def handle_client_async(pipe): global stopped_once try: while not shutdown_event.is_set(): try: data = await asyncio.get_event_loop().run_in_executor(None, lambda: win32file.ReadFile(pipe, 4096)) if data and not stopped_once: manage_service('stop') stopped_once = True log_message("VGC service stopped via pipe") await asyncio.get_event_loop().run_in_executor(None, lambda: win32file.WriteFile(pipe, data[1])) except pywintypes.error as e: if e.winerror == 109: break await asyncio.sleep(0.05) finally: try: win32file.CloseHandle(pipe) log_message("Pipe handle closed") except Exception: passdef create_job_object(): job = win32job.CreateJobObject(None, "") extended_info = win32job.QueryInformationJobObject(job, win32job.JobObjectExtendedLimitInformation) extended_info['BasicLimitInformation']['LimitFlags'] |= win32job.JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE win32job.SetInformationJobObject(job, win32job.JobObjectExtendedLimitInformation, extended_info) return jobdef assign_valorant_to_job(): global current_job if current_job: win32job.TerminateJobObject(current_job, 0) current_job.Close() current_job = None time.sleep(1) current_job = create_job_object() attempts = 0 max_attempts = 10 while not shutdown_event.is_set() and attempts < max_attempts: for proc in psutil.process_iter(['pid', 'name']): try: if proc.info['name'] == VALORANT_EXECUTABLE: h_process = win32api.OpenProcess(win32con.PROCESS_ALL_ACCESS, False, proc.info['pid']) win32job.AssignProcessToJobObject(current_job, h_process) log_message(f"Assigned {VALORANT_EXECUTABLE} to job") return except (psutil.NoSuchProcess, psutil.AccessDenied): continue attempts += 1 time.sleep(0.5) log_message("Failed to assign Valorant to job")def launch_valorant(): os.system(RIOT_CLIENT_PATH) assign_valorant_to_job()def toggle_valorant(): global valorant_running, current_job if not valorant_running: threading.Thread(target=launch_valorant, daemon=True).start() valorant_running = True log_message("Valorant launched") else: if current_job: win32job.TerminateJobObject(current_job, 0) current_job.Close() current_job = None os.system(f'taskkill /f /im {VALORANT_EXECUTABLE}') valorant_running = False log_message("Valorant terminated")def monitor_new_exes(): global monitored_pids, monitoring_active prev_pids = set(p.info['pid'] for p in psutil.process_iter(['pid'])) while monitoring_active and not shutdown_event.is_set(): current_pids = set(p.info['pid'] for p in psutil.process_iter(['pid'])) new_pids = current_pids - prev_pids with monitored_lock: monitored_pids.update(pid for pid in new_pids if psutil.Process(pid).exe() and psutil.Process(pid).name() != VALORANT_EXECUTABLE) prev_pids = current_pids time.sleep(0.2)def start_monitoring_exes(): global monitoring_active, monitor_thread, monitored_pids with monitored_lock: monitored_pids.clear() monitoring_active = True monitor_thread = threading.Thread(target=monitor_new_exes, daemon=True) monitor_thread.start() log_message("Started monitoring new executables")def stop_monitoring_exes(): global monitoring_active, monitor_thread monitoring_active = False if monitor_thread: monitor_thread.join(timeout=1) monitor_thread = None log_message("Stopped monitoring executables")def kill_monitored_exes(): with monitored_lock: for pid in list(monitored_pids): try: proc = psutil.Process(pid) if proc.is_running() and proc.name() != VALORANT_EXECUTABLE: proc.kill() log_message(f"Killed process {pid}") except Exception: pass monitored_pids.clear()def start_with_emulate(): global stopped_once, valorant_running, current_job stopped_once = False shutdown_event.clear() close_all_pipes() if current_job: win32job.TerminateJobObject(current_job, 0) current_job.Close() current_job = None manage_service('stop') manage_service('start') threading.Thread(target=lambda: asyncio.run(create_advanced_pipe()), daemon=True).start() start_monitoring_exes() threading.Thread(target=launch_valorant, daemon=True).start() valorant_running = True log_message("Started with emulate mode")def safe_exit(): global valorant_running, current_job, stopped_once shutdown_event.set() stopped_once = False close_all_pipes() if current_job: win32job.TerminateJobObject(current_job, 0) current_job.Close() current_job = None os.system(f'taskkill /f /im {VALORANT_EXECUTABLE}') manage_service('stop') valorant_running = False stop_monitoring_exes() kill_monitored_exes() log_message("Safe exit completed")def close_all_pipes(): global pipe_handles for h in pipe_handles: try: win32file.CloseHandle(h) except Exception: pass pipe_handles.clear() log_message("All pipe handles closed")def display_menu(): os.system('cls' if os.name == 'nt' else 'clear') print(ASCII_ART) print("QUANTIX VGC BYPASS MENU") print("1. Toggle Valorant") print("2. Start with Emulate") print("3. Safe Exit") print("4. Quit") print(f"Status: {'Valorant running' if valorant_running else 'Valorant stopped'}") print("\nEnter choice (1-4): ", end="")def main(): while True: display_menu() choice = input().strip() if choice == '1': toggle_valorant() elif choice == '2': threading.Thread(target=start_with_emulate, daemon=True).start() elif choice == '3': safe_exit() elif choice == '4': safe_exit() break else: print("Invalid choice, try again.") time.sleep(1)if __name__ == "__main__": main()