import os
import time
import subprocess
import threading
import psutil
import shutil
import ctypes
from collections import deque
from pathlib import Path
from ctypes import wintypes
PIPE_NAME = r'\\.\pipe\933823D3-C77B-4BAE-89D7-A92B567236BC'
STATUS_PANEL_MAX_LOGS = 10
VALORANT_EXE_NAME = 'VALORANT-Win64-Shipping.exe'
RIOT_CLIENT_EXE_NAME = 'RiotClientServices.exe'
VANGUARD_SERVICE_NAME = 'vgc'
valorant_running = False
stopped_once_on_pipe_data = False
current_job_object = None
console_status_logs = deque()
pipe_listener_threads = []
active_pipe_handles = []
console_mutex = threading.Lock()
pipe_mutex = threading.Lock()
monitor_process_thread = None
monitored_pids = []
monitored_pids_mutex = threading.Lock()
is_monitoring_active = False
global_shutdown_event = False
def log_output(message, display_on_status_panel=True):
global console_status_logs
if message == console_status_logs[-1] if console_status_logs else "":
return
if display_on_status_panel:
with console_mutex:
if len(console_status_logs) >= STATUS_PANEL_MAX_LOGS:
console_status_logs.popleft()
console_status_logs.append(message)
print(message)
def log_status_item(status_text, is_done=False, color_done="\033[92m", color_fail="\033[91m", highlight_text="", highlight_color="\033[91m"):
prefix = f"{color_done}(DONE) \033[0m" if is_done else ""
formatted_text = status_text
if highlight_text:
formatted_text = formatted_text.replace(highlight_text, f"{highlight_color}{highlight_text}\033[0m")
message = f"{prefix}{formatted_text}"
log_output(message)
def run_system_command_silent(command):
subprocess.run(command, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, shell=True)
def path_exists(path):
return Path(path).exists()
def get_riot_client_path():
possible_paths = [
f"C:\\Program Files\\Riot Games\\Riot Client\\{RIOT_CLIENT_EXE_NAME}",
f"C:\\Program Files (x86)\\Riot Games\\Riot Client\\{RIOT_CLIENT_EXE_NAME}",
f"C:\\Riot Games\\Riot Client\\{RIOT_CLIENT_EXE_NAME}"
]
for p in possible_paths:
if path_exists(p):
return p
return ""
def find_process_by_name(name):
for proc in psutil.process_iter(['pid', 'name']):
if proc.info['name'].lower() == name.lower():
return proc.info['pid']
return None
def terminate_process(process_name):
log_status_item(f"VANGUARD EMULATE {process_name}")
pid = find_process_by_name(process_name)
if pid:
try:
p = psutil.Process(pid)
p.terminate()
p.wait()
log_status_item(f"VANGUARD EMULATE {process_name}", True)
except psutil.NoSuchProcess:
pass
def handle_pipe_client(pipe_handle):
while not global_shutdown_event:
try:
data = pipe_handle.read(4096)
if data:
if not stopped_once_on_pipe_data:
stopped_once_on_pipe_data = True
log_status_item("VANGUARD EMULATE")
pipe_handle.write(data)
except IOError:
break
time.sleep(0.1)
pipe_handle.close()
def start_named_pipe_listener():
log_status_item("VANGUARD EMULATE")
while not global_shutdown_event:
try:
pipe_server_handle = open(PIPE_NAME, 'r+b', buffering=0)
threading.Thread(target=handle_pipe_client, args=(pipe_server_handle,)).start()
with pipe_mutex:
active_pipe_handles.append(pipe_server_handle)
except OSError:
time.sleep(1)
def close_all_active_pipes():
with pipe_mutex:
for handle in active_pipe_handles:
handle.close()
active_pipe_handles.clear()
log_status_item("EMULATE.", True)
def create_and_configure_job_object():
log_status_item("VANGUARD EMULATE")
# Python doesn't have direct job object support like Windows API
return None
def assign_valorant_to_job_object():
log_status_item("VANGUARD EMULATE")
global current_job_object
if current_job_object:
current_job_object = None
time.sleep(2)
current_job_object = create_and_configure_job_object()
if not current_job_object:
log_status_item("ERROR", False, "\033[92m", "\033[91m")
return
max_attempts = 30
attempt_count = 0
found_valorant = False
while not found_valorant and attempt_count < max_attempts and not global_shutdown_event:
pid = find_process_by_name(VALORANT_EXE_NAME)
if pid:
try:
p = psutil.Process(pid)
p.suspend()
found_valorant = True
except psutil.NoSuchProcess:
pass
if not found_valorant:
attempt_count += 1
time.sleep(1)
if not found_valorant:
log_status_item("ERROR", False, "\033[92m", "\033[91m")
def launch_valorant_client():
log_status_item("VANGUARD EMULATE")
riot_client_path = get_riot_client_path()
if not riot_client_path:
log_status_item("ERROR", False, "\033[92m", "\033[91m")
return
run_system_command_silent(f'"{riot_client_path}" --launch-product=valorant --launch-patchline=live')
log_status_item("VANGUARD EMULATE")
assign_valorant_to_job_object()
log_status_item("VANGUARD EMULATE", True)
def monitor_new_processes():
while is_monitoring_active and not global_shutdown_event:
pid = find_process_by_name(VALORANT_EXE_NAME)
if pid:
with monitored_pids_mutex:
if pid not in monitored_pids:
monitored_pids.append(pid)
time.sleep(0.5)
def start_process_monitoring():
with monitored_pids_mutex:
monitored_pids.clear()
global is_monitoring_active
is_monitoring_active = True
global monitor_process_thread
monitor_process_thread = threading.Thread(target=monitor_new_processes)
monitor_process_thread.start()
log_status_item("VANGUARD EMULATE", True)
def stop_process_monitoring():
global is_monitoring_active
is_monitoring_active = False
if monitor_process_thread.is_alive():
monitor_process_thread.join()
log_status_item("VANGUARD EMULATE", True)
def kill_monitored_executables():
with monitored_pids_mutex:
for pid in monitored_pids:
try:
p = psutil.Process(pid)
p.terminate()
p.wait()
except psutil.NoSuchProcess:
pass
monitored_pids.clear()
log_status_item("VANGUARD EMULATE", True)
def safe_exit_procedure():
log_status_item("VANGUARD EMULATE")
global stopped_once_on_pipe_data
stopped_once_on_pipe_data = False
for t in pipe_listener_threads:
t.join()
pipe_listener_threads.clear()
close_all_active_pipes()
if current_job_object:
current_job_object = None
log_status_item("VANGUARD EMULATE", True)
terminate_process(VALORANT_EXE_NAME)
terminate_process(RIOT_CLIENT_EXE_NAME)
run_system_command_silent(f'sc stop {VANGUARD_SERVICE_NAME}')
stop_process_monitoring()
kill_monitored_executables()
global valorant_running
valorant_running = False
log_status_item("VANGUARD EMULATE", True)
def display_main_menu():
with console_mutex:
os.system("cls" if os.name == "nt" else "clear")
print(r'''
$$$$$$$\
$$ __$$\
$$ | $$ |$$\ $$\ $$$$$$$$\ $$$$$$$$\ $$\ $$\
$$$$$$$ |\$$\ $$ |\____$$ |\____$$ |$$ | $$ |
$$ __$$< \$$$$ / $$$$ _/ $$$$ _/ $$ | $$ |
$$ | $$ | $$ $$< $$ _/ $$ _/ $$ | $$ |
$$ | $$ |$$ /\$$\ $$$$$$$$\ $$$$$$$$\ \$$$$$$$ |
\__| \__|\__/ \__|\________|\________| \____$$ |
$$\ $$ |
\$$$$$$ |
\______/
''')
print("\n1 = Start Valorant\n2 = Vanguard Bypass\n3 = Stop Vanguard\n4 = Match Ended/Restart Game\n\nSelection: ", end='')
def main_loop():
while True:
display_main_menu()
choice = input().strip()
if choice == "1":
log_status_item("VANGUARD EMULATE")
launch_valorant_client()
elif choice == "2":
stopped_once_on_pipe_data = False
global_shutdown_event = False
threading.Thread(target=start_named_pipe_listener).start()
start_process_monitoring()
launch_valorant_client()
global valorant_running
valorant_running = True
log_status_item("VANGUARD EMULATE", True)
elif choice == "3":
log_status_item("Stopping Vanguard Service")
run_system_command_silent(f'sc stop {VANGUARD_SERVICE_NAME}')
log_status_item("Vanguard Stopped", True)
elif choice == "4":
safe_exit_procedure()
elif choice.lower() == "exit":
safe_exit_procedure()
break
else:
log_status_item(f"Invalid Selection: {choice}", False, "\033[92m", "\033[91m")
time.sleep(1.5)
if __name__ == "__main__":
os.system("title Xy9bLmp3NzRwQd34E2f324Hıp34fh34Fkd21NdfaKy9bLmp3NzRwQd34E2f324Hıp34fh34Fkd21NdfaD21N24af53hJ41ga1")
main_loop()
