diff --git a/src/cli/sentinel.py b/src/cli/sentinel.py index bd783639..e2debba5 100755 --- a/src/cli/sentinel.py +++ b/src/cli/sentinel.py @@ -21,6 +21,7 @@ from adapters import HTTPRequests from cli.v2ray import V2RayHandler from helpers import helpers +from helpers.windows_split_tunnel import stop_windows_split_tunnel import mospy import grpc @@ -42,7 +43,7 @@ class NodeTreeData(): NodeTypes = {} NodeHealth = {} NodeFormula = {} - + def __init__(self, node_tree): if not node_tree: self.NodeTree = Tree() @@ -394,7 +395,7 @@ def CreateNodeTreeStructure(self, data, **kwargs): continue return NodeTreeBase - + def return_denom(self, tokens): for ibc_coin in IBCTokens.IBCCOINS: for denom,ibc in ibc_coin.items(): @@ -402,7 +403,6 @@ def return_denom(self, tokens): tokens = tokens.replace(ibc, denom) return tokens - def parse_coin_deposit(self, tokens): UnitAmounts = [] tokenString = "" @@ -600,6 +600,7 @@ def disconnect(v2ray): if v2ray: try: if pltfrm == Arch.WINDOWS: + stop_windows_split_tunnel() V2Ray = V2RayHandler(v2ray_tun2routes_connect_bash + " down") chdir(MeileConfig.BASEBINDIR) rc = V2Ray.kill_daemon() @@ -615,6 +616,7 @@ def disconnect(v2ray): else: if pltfrm == Arch.WINDOWS: + stop_windows_split_tunnel() with open(path.join(MeileConfig.BASEBINDIR, 'disconnect.bat'), 'w') as DISBATFILE: DISBATFILE.write("%s /uninstalltunnelservice wg99\n" % MeileConfig.WIREGUARD_BIN) DISBATFILE.write("TASKKILL /F /IM WireGuard.exe\n") @@ -632,11 +634,9 @@ def disconnect(v2ray): else: CONFFILE = path.join(ConfParams.KEYRINGDIR, 'wg99.conf') wg_downCMD = ['pkexec', 'env', 'PATH=%s' % ConfParams.PATH, 'wg-quick', 'down', CONFFILE] - + proc1 = Popen(wg_downCMD) proc1.wait(timeout=30) - + proc_out,proc_err = proc1.communicate() return proc1.returncode, False - - \ No newline at end of file diff --git a/src/cli/wallet.py b/src/cli/wallet.py index 4072ccd5..4a83132d 100755 --- a/src/cli/wallet.py +++ b/src/cli/wallet.py @@ -21,6 +21,7 @@ from cli.v2ray import V2RayHandler, V2RayConfiguration, V2RayFragmentConfiguration from helpers.wireguard import WgKey from helpers.helpers import resolve_address +from helpers.windows_split_tunnel import activate_windows_split_tunnel, prepare_windows_split_tunnel import base64 import bcrypt @@ -58,12 +59,13 @@ class HandleWalletFunctions(): def __init__(self, **kwargs): super().__init__(**kwargs) - + CONFIG = MeileConfig.read_configuration(MeileConfig.CONFFILE) self.GRPC = CONFIG['network'].get('grpc', HTTParams.GRPC) self.sdk = None self.returncode = None self.connected = {"v2ray_pid" : None, "result" : False, "status" : None, "session_id" : None} + self.split_tunnel_session = None # Migrate existing wallet to v2 address = CONFIG['wallet'].get('address', None) @@ -74,6 +76,31 @@ def __init__(self, **kwargs): #if address and version < 200: # print("Migrating Wallet...") # self.__migrate_wallets() + + def prepare_split_tunnel(self, conndesc): + self.split_tunnel_session = None + if platform.system() != Arch.WINDOWS: + return + try: + self.split_tunnel_session = prepare_windows_split_tunnel(MeileConfig.CONFFILE) + if self.split_tunnel_session: + conndesc.write("Preparing split tunneling...\n") + conndesc.flush() + except Exception as e: + print("Split tunneling preparation failed:", str(e)) + + def start_split_tunnel(self, conndesc): + if not self.split_tunnel_session: + return + try: + added_routes = activate_windows_split_tunnel(self.split_tunnel_session) + conndesc.write("Split tunneling enabled") + if added_routes: + conndesc.write(f" ({len(added_routes)} route(s) added)") + conndesc.write(".\n") + conndesc.flush() + except Exception as e: + print("Split tunneling start failed:", str(e)) @staticmethod def decode_jwt_file(fpath: str, password: str) -> dict: @@ -1231,6 +1258,8 @@ def connect(self, "session_id" : session_id} print(self.connected) return + + self.prepare_split_tunnel(conndesc) if type == "WireGuard": iface = "wg99" @@ -1270,6 +1299,7 @@ def connect(self, "result": True, "status" : iface, "session_id" : session_id} + self.start_split_tunnel(conndesc) conndesc.write("Checking network connection...\n") conndesc.flush() sleep(1) @@ -1338,6 +1368,7 @@ def connect(self, "result": True, "status" : tuniface, "session_id" : session_id} + self.start_split_tunnel(conndesc) else: self.connected = {"v2ray_pid" : v2ray_handler.v2ray_pid, "result": False, @@ -1452,6 +1483,3 @@ def get_ip_address(self): with open(path.join(ConfParams.KEYRINGDIR, 'ip-api.json'), 'w') as f: f.write(json.dumps('{}')) return False - - - \ No newline at end of file diff --git a/src/conf/meile_config.py b/src/conf/meile_config.py index b876aab2..f31fcd85 100755 --- a/src/conf/meile_config.py +++ b/src/conf/meile_config.py @@ -115,6 +115,7 @@ def read_configuration(self, confpath): self.CONFIG.set('network', 'fragment', '1') self.CONFIG.set('network', 'dns', '1.1.1.1') self.CONFIG.set('network', 'ringsessions', '0') + self.CONFIG.set('network', 'splittunnel', '0') FILE = open(self.CONFFILE, 'w') self.CONFIG.write(FILE) FILE.close() @@ -143,10 +144,25 @@ def read_configuration(self, confpath): self.CONFIG.set('network', 'ringsessions', '1') if self.CONFIG.has_option('network', 'ringsessions'): self.CONFIG.set('network', 'ringsessions', '1') + if not self.CONFIG.has_option('network', 'splittunnel'): + self.CONFIG.set('network', 'splittunnel', '0') FILE = open(self.CONFFILE, 'w') self.CONFIG.write(FILE) FILE.close() + + if not self.CONFIG.has_section('split_tunnel'): + self.CONFIG.add_section('split_tunnel') + self.CONFIG.set('split_tunnel', 'apps', '[]') + FILE = open(self.CONFFILE, 'w') + self.CONFIG.write(FILE) + FILE.close() + else: + if not self.CONFIG.has_option('split_tunnel', 'apps'): + self.CONFIG.set('split_tunnel', 'apps', '[]') + FILE = open(self.CONFFILE, 'w') + self.CONFIG.write(FILE) + FILE.close() return self.CONFIG diff --git a/src/helpers/windows_split_tunnel.py b/src/helpers/windows_split_tunnel.py new file mode 100644 index 00000000..0277d55c --- /dev/null +++ b/src/helpers/windows_split_tunnel.py @@ -0,0 +1,500 @@ +import configparser +import ipaddress +import json +import os +import subprocess +import threading +from dataclasses import dataclass +from pathlib import Path +from time import sleep + + +CONFIG_SECTION = "split_tunnel" +CONFIG_APPS_KEY = "apps" +CONFIG_ENABLED_KEY = "splittunnel" + + +@dataclass(frozen=True) +class WindowsApp: + name: str + path: str + + +@dataclass(frozen=True) +class WindowsDefaultRoute: + interface_index: int + gateway: str + + +def _ensure_config(config): + if not config.has_section("network"): + config.add_section("network") + if not config.has_section(CONFIG_SECTION): + config.add_section(CONFIG_SECTION) + + +def _path_key(value): + return os.path.normpath(os.path.expandvars(str(value))).casefold() + + +def get_split_tunnel_apps(config): + if not config.has_section(CONFIG_SECTION): + return [] + + raw_apps = config.get(CONFIG_SECTION, CONFIG_APPS_KEY, fallback="[]") + try: + apps = json.loads(raw_apps) + except json.JSONDecodeError: + apps = [line for line in raw_apps.splitlines() if line.strip()] + + selected = [] + seen = set() + for app in apps: + app_path = str(app).strip() + if not app_path: + continue + key = _path_key(app_path) + if key in seen: + continue + selected.append(app_path) + seen.add(key) + return selected + + +def set_split_tunnel_apps(config, apps): + _ensure_config(config) + selected = [] + seen = set() + for app in apps: + app_path = str(app).strip() + if not app_path: + continue + key = _path_key(app_path) + if key in seen: + continue + selected.append(app_path) + seen.add(key) + config.set(CONFIG_SECTION, CONFIG_APPS_KEY, json.dumps(selected)) + + +def is_split_tunnel_enabled(config): + if not config.has_section("network"): + return False + return config.getboolean("network", CONFIG_ENABLED_KEY, fallback=False) + + +def set_split_tunnel_enabled(config, enabled): + _ensure_config(config) + config.set("network", CONFIG_ENABLED_KEY, "1" if enabled else "0") + + +def split_tunnel_summary(apps): + if not apps: + return "No apps selected" + names = [Path(app).stem for app in apps[:3]] + suffix = "" if len(apps) <= 3 else f" +{len(apps) - 3}" + return ", ".join(names) + suffix + + +class WindowsAppCatalog: + def __init__(self, search_roots=None, walker=os.walk, runner=None): + self.search_roots = search_roots + self.walker = walker + self.runner = runner or WindowsRouteManager._run + + def _default_search_roots(self): + roots = [] + for env_name in ("ProgramFiles", "ProgramFiles(x86)", "LocalAppData"): + env_path = os.environ.get(env_name) + if env_path: + roots.append(Path(env_path)) + + start_menu = os.environ.get("ProgramData") + if start_menu: + roots.append(Path(start_menu) / "Microsoft" / "Windows" / "Start Menu") + + user_profile = os.environ.get("USERPROFILE") + if user_profile: + roots.append( + Path(user_profile) + / "AppData" + / "Roaming" + / "Microsoft" + / "Windows" + / "Start Menu" + ) + return roots + + def list_apps(self, limit=250): + apps = [] + seen = set() + for app in self._registry_apps(): + self._append_app(apps, seen, app) + if len(apps) >= limit: + return sorted(apps, key=lambda app: app.name.casefold()) + + roots = self.search_roots + if roots is None: + if os.name != "nt": + return sorted(apps, key=lambda app: app.name.casefold()) + roots = self._default_search_roots() + + for root in roots: + root = Path(root) + if not root.exists(): + continue + for dirpath, _, filenames in self.walker(root): + for filename in sorted(filenames): + if not filename.lower().endswith(".exe"): + continue + app_path = str(Path(dirpath) / filename) + self._append_app( + apps, + seen, + WindowsApp(name=Path(filename).stem, path=app_path), + ) + if len(apps) >= limit: + return sorted(apps, key=lambda app: app.name.casefold()) + return sorted(apps, key=lambda app: app.name.casefold()) + + def _registry_apps(self): + if os.name != "nt" and self.search_roots is None: + return [] + command = [ + "powershell", + "-NoProfile", + "-ExecutionPolicy", + "Bypass", + "-Command", + ( + "$paths = @(" + "'HKLM:\\Software\\Microsoft\\Windows\\CurrentVersion\\Uninstall\\*'," + "'HKLM:\\Software\\WOW6432Node\\Microsoft\\Windows\\CurrentVersion\\Uninstall\\*'," + "'HKCU:\\Software\\Microsoft\\Windows\\CurrentVersion\\Uninstall\\*'" + "); " + "Get-ItemProperty $paths -ErrorAction SilentlyContinue | " + "Where-Object { $_.DisplayName } | " + "Select-Object DisplayName,DisplayIcon,InstallLocation | " + "ConvertTo-Json -Compress" + ), + ] + try: + result = self.runner(command) + except OSError: + return [] + if result.returncode != 0 or not result.stdout.strip(): + return [] + + try: + payload = json.loads(result.stdout) + except json.JSONDecodeError: + return [] + if isinstance(payload, dict): + payload = [payload] + + apps = [] + for item in payload: + name = str(item.get("DisplayName") or "").strip() + app_path = self._path_from_registry_item(item) + if name and app_path: + apps.append(WindowsApp(name=name, path=app_path)) + return apps + + @staticmethod + def _path_from_registry_item(item): + display_icon = str(item.get("DisplayIcon") or "").strip() + if not display_icon: + return None + app_path = os.path.expandvars(display_icon.strip('"')) + exe_index = app_path.lower().find(".exe") + if exe_index >= 0: + app_path = app_path[:exe_index + 4].strip().strip('"') + if app_path.lower().endswith(".exe"): + return app_path + return None + + @staticmethod + def _append_app(apps, seen, app): + key = _path_key(app.path) + if key in seen: + return + apps.append(app) + seen.add(key) + + +class WindowsRouteManager: + def __init__( + self, + runner=None, + process_provider=None, + connection_provider=None, + state_path=None, + ): + self.runner = runner or self._run + self.process_provider = process_provider or self._processes + self.connection_provider = connection_provider or self._connections + self.state_path = Path(state_path) if state_path else None + + @staticmethod + def _run(command, **kwargs): + return subprocess.run( + command, + check=False, + capture_output=True, + text=True, + **kwargs, + ) + + @staticmethod + def _processes(): + import psutil + + return psutil.process_iter(["pid", "name", "exe"]) + + @staticmethod + def _connections(): + import psutil + + return psutil.net_connections(kind="inet") + + def capture_default_route(self): + command = [ + "powershell", + "-NoProfile", + "-ExecutionPolicy", + "Bypass", + "-Command", + ( + "$route = Get-NetRoute -AddressFamily IPv4 " + "-DestinationPrefix '0.0.0.0/0' | " + "Where-Object { $_.NextHop -and $_.NextHop -ne '0.0.0.0' } | " + "Sort-Object RouteMetric,InterfaceMetric | " + "Select-Object -First 1 InterfaceIndex,NextHop; " + "$route | ConvertTo-Json -Compress" + ), + ] + result = self.runner(command) + if result.returncode != 0 or not result.stdout.strip(): + return None + + try: + payload = json.loads(result.stdout) + except json.JSONDecodeError: + return None + if isinstance(payload, list): + payload = payload[0] if payload else {} + interface_index = payload.get("InterfaceIndex") + gateway = payload.get("NextHop") + if interface_index is None or not gateway: + return None + return WindowsDefaultRoute(interface_index=int(interface_index), gateway=gateway) + + def collect_destinations(self, app_paths): + selected_paths = {_path_key(app) for app in app_paths} + selected_names = {Path(app).name.casefold() for app in app_paths} + selected_pids = set() + + for process in self.process_provider(): + info = process if isinstance(process, dict) else getattr(process, "info", {}) + pid = info.get("pid") + exe = info.get("exe") or "" + name = info.get("name") or Path(exe).name + if not pid: + continue + if _path_key(exe) in selected_paths or name.casefold() in selected_names: + selected_pids.add(pid) + + destinations = set() + for conn in self.connection_provider(): + if getattr(conn, "pid", None) not in selected_pids: + continue + remote_ip = self._remote_ip(conn) + if remote_ip and self._is_routeable_ipv4(remote_ip): + destinations.add(remote_ip) + return sorted(destinations) + + def apply_routes(self, app_paths, default_route=None): + default_route = default_route or self.capture_default_route() + if not default_route: + return [] + + state = self._load_state() + known_routes = { + ( + route.get("destination"), + route.get("gateway"), + route.get("interface_index"), + ) + for route in state.get("routes", []) + } + added = [] + routes = state.get("routes", []) + for destination in self.collect_destinations(app_paths): + route_key = ( + destination, + default_route.gateway, + default_route.interface_index, + ) + if route_key in known_routes: + continue + command = [ + "route", + "add", + destination, + "mask", + "255.255.255.255", + default_route.gateway, + "IF", + str(default_route.interface_index), + "METRIC", + "1", + ] + result = self.runner(command) + if result.returncode == 0: + added.append(destination) + routes.append( + { + "destination": destination, + "gateway": default_route.gateway, + "interface_index": default_route.interface_index, + } + ) + known_routes.add(route_key) + + if added: + state["routes"] = routes + self._save_state(state) + return added + + def remove_routes(self): + state = self._load_state() + for route in state.get("routes", []): + destination = route.get("destination") + gateway = route.get("gateway") + if not destination or not gateway: + continue + self.runner( + [ + "route", + "delete", + destination, + "mask", + "255.255.255.255", + gateway, + ] + ) + if self.state_path and self.state_path.exists(): + self.state_path.unlink() + + @staticmethod + def _remote_ip(connection): + remote = getattr(connection, "raddr", None) + if not remote: + return None + if hasattr(remote, "ip"): + return remote.ip + if isinstance(remote, (tuple, list)) and remote: + return remote[0] + return None + + @staticmethod + def _is_routeable_ipv4(value): + try: + address = ipaddress.ip_address(value) + except ValueError: + return False + return ( + address.version == 4 + and not address.is_loopback + and not address.is_link_local + and not address.is_multicast + and not address.is_unspecified + ) + + def _load_state(self): + if not self.state_path or not self.state_path.exists(): + return {"routes": []} + try: + with self.state_path.open("r", encoding="utf-8") as file: + return json.load(file) + except json.JSONDecodeError: + return {"routes": []} + + def _save_state(self, state): + if not self.state_path: + return + self.state_path.parent.mkdir(parents=True, exist_ok=True) + with self.state_path.open("w", encoding="utf-8") as file: + json.dump(state, file, indent=2) + + +class WindowsSplitTunnelSession: + def __init__(self, manager, app_paths, default_route, interval=5): + self.manager = manager + self.app_paths = app_paths + self.default_route = default_route + self.interval = interval + self._stop = threading.Event() + self._thread = None + + def start(self): + if not self.app_paths or not self.default_route: + return [] + + added = self.manager.apply_routes(self.app_paths, self.default_route) + self._thread = threading.Thread(target=self._loop, daemon=True) + self._thread.start() + return added + + def stop(self): + self._stop.set() + if self._thread and self._thread.is_alive(): + self._thread.join(timeout=1) + self.manager.remove_routes() + + def _loop(self): + while not self._stop.is_set(): + sleep(self.interval) + self.manager.apply_routes(self.app_paths, self.default_route) + + +_active_session = None + + +def activate_windows_split_tunnel(session): + global _active_session + if _active_session: + _active_session.stop() + _active_session = session + if not _active_session: + return [] + return _active_session.start() + + +def default_state_path(): + return Path.home() / ".meile-gui" / "split_tunnel_routes.json" + + +def prepare_windows_split_tunnel(config_path, route_manager=None): + config = configparser.ConfigParser() + config.read(config_path) + app_paths = get_split_tunnel_apps(config) + if not is_split_tunnel_enabled(config) or not app_paths: + return None + + manager = route_manager or WindowsRouteManager(state_path=default_state_path()) + default_route = manager.capture_default_route() + if not default_route: + return None + return WindowsSplitTunnelSession(manager, app_paths, default_route) + + +def stop_windows_split_tunnel(route_manager=None): + global _active_session + if _active_session: + _active_session.stop() + _active_session = None + return + + manager = route_manager or WindowsRouteManager(state_path=default_state_path()) + manager.remove_routes() diff --git a/src/kv/meile.kv b/src/kv/meile.kv index 0a6e13f7..d7bc8e9b 100755 --- a/src/kv/meile.kv +++ b/src/kv/meile.kv @@ -417,7 +417,7 @@ WindowManager: orientation: "vertical" padding: dp(20), dp(20), dp(20), dp(70) # extra bottom padding for ActionBar spacing: dp(20) - + # --- QR + Address block, centered horizontally --- MDBoxLayout: orientation: "vertical" @@ -781,13 +781,72 @@ WindowManager: size: dp(32), dp(32) pos_hint: {"center_y": 0.5} active: root.get_config('ringsessions') + + BoxLayout: + orientation: "horizontal" + spacing: dp(10) + padding: [20,0,0,20] + size_hint_y: None + height: dp(48) + pos_hint: {"center_x": 0.5, "top": 0.35} + canvas.before: + Color: + rgba: get_color_from_hex('#2a2a2a') + Rectangle: + pos: self.pos + size: self.size + + MDLabel: + font_name: "Roboto-Bold" + text: "Split Tunneling" + font_size: "20sp" + size_hint: None, None + width: sp(250) + pos_hint: {"center_y": 0.5} + halign: "left" + valign: "middle" + padding: [20,0,0,0] + + Check2: + id: split_tunnel + size_hint: None, None + size: dp(32), dp(32) + pos_hint: {"center_y": 0.5} + active: root.get_config('splittunnel') + + BoxLayout: + orientation: "horizontal" + spacing: dp(10) + padding: [20,0,0,20] + size_hint_y: None + height: dp(48) + pos_hint: {"center_x": 0.5, "top": 0.30} + + MDLabel: + font_name: "Roboto-Bold" + text: "Bypass Apps" + font_size: "20sp" + size_hint_x: None + width: sp(250) + padding: [20,0,0,0] + + MDDropDownItem: + id: split_tunnel_apps_drop_item + text: root.get_split_tunnel_summary() + on_release: root.open_split_tunnel_menu() + + MDFlatButton: + text: "CLEAR" + theme_text_color: "Custom" + text_color: get_color_from_hex(MeileColors.MEILE) + on_release: root.clear_split_tunnel_apps() - MDFlatButton: + MDFlatButton: text: "CANCEL" font_size: "18sp" pos_hint: { "center_x": .80, "top": 0.1} theme_text_color: "Custom" - text_color: get_color_from_hex(MeileColors.MEILE) + text_color: get_color_from_hex(MeileColors.MEILE) on_release: root.set_previous_screen() MDRaisedButton: @@ -3277,4 +3336,4 @@ WindowManager: Check: group: 'proto' on_active: root.select_share_type(self, self.active, "v2") - pos_hint: {"x": .5, "y" : .025} \ No newline at end of file + pos_hint: {"x": .5, "y" : .025} diff --git a/src/ui/screens.py b/src/ui/screens.py index c8218732..17379155 100755 --- a/src/ui/screens.py +++ b/src/ui/screens.py @@ -22,6 +22,12 @@ from helpers.aes import SecureSeed from helpers.v2ray import generate_v2ray_uri from helpers.update_checker import UpdateChecker, format_update_message +from helpers.windows_split_tunnel import ( + WindowsAppCatalog, + get_split_tunnel_apps, + set_split_tunnel_apps, + split_tunnel_summary, +) from ui.update_dialog import UpdateDialog from coin_api.get_price import GetPriceAPI @@ -2467,6 +2473,7 @@ def open_sentinel(self): class SettingsScreen(Screen): MeileConfig = MeileGuiConfig() SettingsNetworkMenu = ["grpc", "api", "mnapi", "cache", "dns", "resolver1", "resolver2", "resolver3"] + split_tunnel_apps = [] def __init__(self, **kwargs): super().__init__(**kwargs) @@ -2489,6 +2496,8 @@ def __init__(self, **kwargs): self.DNS = config['network'].get('dns', '1.1.1.1') self.CONFIGDNS = config['network'].get('dns', '1.1.1.1') self.RINGSESSIONS = config['network'].get('ringsessions', '0') + self.SPLITTUNNEL = config['network'].get('splittunnel', '0') + self.split_tunnel_apps = get_split_tunnel_apps(config) self.MeileConfig = MeileGuiConfig() @@ -2639,6 +2648,24 @@ def __init__(self, **kwargs): ) self.resolver3_menu.bind() + split_tunnel_items = [ + { + "viewclass": "IconListItem", + "icon": "application", + "text": app.name, + "height": dp(56), + "on_release": lambda x=app.path: self.add_split_tunnel_app(x), + } for app in WindowsAppCatalog().list_apps() + ] + self.split_tunnel_menu = MDDropdownMenu( + caller=self.ids.split_tunnel_apps_drop_item, + items=split_tunnel_items, + position="center", + width_mult=7, + ) + self.split_tunnel_menu.bind() + self.ids.split_tunnel_apps_drop_item.set_item(self.get_split_tunnel_summary()) + def get_config(self, what: str = "grpc"): config = self.MeileConfig.read_configuration(self.MeileConfig.CONFFILE) if what in self.SettingsNetworkMenu: @@ -2648,6 +2675,8 @@ def get_config(self, what: str = "grpc"): return bool(int(config['network'].get(what, "0"))) elif what == "ringsessions": return bool(int(config['network'].get(what, "0"))) + elif what == "splittunnel": + return bool(int(config['network'].get(what, "0"))) else: getattr(self.ids, f"{what}_drop_item").set_item(config['subscription'][what]) return config['subscription'][what] @@ -2670,6 +2699,29 @@ def set_item(self, text_item, what: str = "rpc"): setattr(self, what.upper(), text_item) getattr(self, f"{what.lower()}_menu").dismiss() + def get_split_tunnel_summary(self): + apps = getattr(self, "split_tunnel_apps", None) + if apps is None: + config = self.MeileConfig.read_configuration(self.MeileConfig.CONFFILE) + apps = get_split_tunnel_apps(config) + return split_tunnel_summary(apps) + + def open_split_tunnel_menu(self): + if not self.split_tunnel_menu.items: + toast(text="No Windows apps found", duration=3.5) + return + self.split_tunnel_menu.open() + + def add_split_tunnel_app(self, app_path): + if app_path not in self.split_tunnel_apps: + self.split_tunnel_apps.append(app_path) + self.ids.split_tunnel_apps_drop_item.set_item(self.get_split_tunnel_summary()) + self.split_tunnel_menu.dismiss() + + def clear_split_tunnel_apps(self): + self.split_tunnel_apps = [] + self.ids.split_tunnel_apps_drop_item.set_item(self.get_split_tunnel_summary()) + def build(self): return self.screen @@ -2694,6 +2746,10 @@ def SaveOptions(self): what = "ringsessions" config.set('network', what, '1' if self.ids.ring_sessions.active else '0') + + what = "splittunnel" + config.set('network', what, '1' if self.ids.split_tunnel.active else '0') + set_split_tunnel_apps(config, self.split_tunnel_apps) with open(self.MeileConfig.CONFFILE, 'w', encoding="utf-8") as f: config.write(f) diff --git a/tests/test_windows_split_tunnel.py b/tests/test_windows_split_tunnel.py new file mode 100644 index 00000000..32ddc636 --- /dev/null +++ b/tests/test_windows_split_tunnel.py @@ -0,0 +1,149 @@ +import configparser +import json +from types import SimpleNamespace + +from helpers.windows_split_tunnel import ( + WindowsAppCatalog, + WindowsDefaultRoute, + WindowsRouteManager, + get_split_tunnel_apps, + is_split_tunnel_enabled, + set_split_tunnel_apps, +) +from conf.meile_config import MeileGuiConfig + + +def test_split_tunnel_config_round_trip(): + config = configparser.ConfigParser() + config.add_section("network") + + set_split_tunnel_apps(config, ["C:\\Apps\\Browser\\browser.exe", "C:\\Tools\\chat.exe"]) + config.set("network", "splittunnel", "1") + + assert is_split_tunnel_enabled(config) + assert get_split_tunnel_apps(config) == [ + "C:\\Apps\\Browser\\browser.exe", + "C:\\Tools\\chat.exe", + ] + + +def test_windows_app_catalog_lists_unique_executables(tmp_path): + root = tmp_path / "Program Files" + app_dir = root / "Acme" + app_dir.mkdir(parents=True) + app = app_dir / "Acme.exe" + app.write_text("", encoding="utf-8") + duplicate = app_dir / "acme.EXE" + duplicate.write_text("", encoding="utf-8") + ignored = app_dir / "readme.txt" + ignored.write_text("", encoding="utf-8") + + apps = WindowsAppCatalog(search_roots=[root]).list_apps() + + assert [(item.name, item.path) for item in apps] == [("Acme", str(app))] + + +def test_windows_app_catalog_parses_registry_apps(): + def runner(command, **kwargs): + payload = [ + { + "DisplayName": "Browser", + "DisplayIcon": "C:\\Apps\\Browser\\browser.exe,0", + "InstallLocation": "C:\\Apps\\Browser", + }, + { + "DisplayName": "Broken", + "DisplayIcon": "", + "InstallLocation": "", + }, + ] + return SimpleNamespace(returncode=0, stdout=json.dumps(payload), stderr="") + + apps = WindowsAppCatalog(search_roots=[], runner=runner).list_apps() + + assert [(item.name, item.path) for item in apps] == [ + ("Browser", "C:\\Apps\\Browser\\browser.exe") + ] + + +def test_route_manager_adds_host_routes_for_selected_process_destinations(): + commands = [] + + def runner(command, **kwargs): + commands.append(command) + return SimpleNamespace(returncode=0, stdout="", stderr="") + + processes = [ + {"pid": 100, "name": "browser.exe", "exe": "C:\\Apps\\Browser\\browser.exe"}, + {"pid": 200, "name": "other.exe", "exe": "C:\\Apps\\Other\\other.exe"}, + ] + connections = [ + SimpleNamespace(pid=100, raddr=SimpleNamespace(ip="203.0.113.10")), + SimpleNamespace(pid=100, raddr=("198.51.100.25", 443)), + SimpleNamespace(pid=100, raddr=SimpleNamespace(ip="127.0.0.1")), + SimpleNamespace(pid=200, raddr=SimpleNamespace(ip="192.0.2.50")), + ] + manager = WindowsRouteManager( + runner=runner, + process_provider=lambda: processes, + connection_provider=lambda: connections, + ) + + added = manager.apply_routes( + ["C:\\Apps\\Browser\\browser.exe"], + WindowsDefaultRoute(interface_index=12, gateway="192.168.1.1"), + ) + + assert added == ["198.51.100.25", "203.0.113.10"] + assert commands == [ + [ + "route", + "add", + "198.51.100.25", + "mask", + "255.255.255.255", + "192.168.1.1", + "IF", + "12", + "METRIC", + "1", + ], + [ + "route", + "add", + "203.0.113.10", + "mask", + "255.255.255.255", + "192.168.1.1", + "IF", + "12", + "METRIC", + "1", + ], + ] + + +def test_default_route_is_parsed_from_powershell_json(): + def runner(command, **kwargs): + payload = {"InterfaceIndex": 7, "NextHop": "10.0.0.1"} + return SimpleNamespace(returncode=0, stdout=json.dumps(payload), stderr="") + + manager = WindowsRouteManager(runner=runner) + + assert manager.capture_default_route() == WindowsDefaultRoute( + interface_index=7, + gateway="10.0.0.1", + ) + + +def test_meile_config_adds_split_tunnel_defaults(tmp_path): + meile_config = MeileGuiConfig() + meile_config.BASEDIR = str(tmp_path) + meile_config.CONFFILE = str(tmp_path / "config.ini") + meile_config.IMGDIR = str(tmp_path / "img") + + config = meile_config.read_configuration(meile_config.CONFFILE) + + assert config.has_section("split_tunnel") + assert config["network"]["splittunnel"] == "0" + assert get_split_tunnel_apps(config) == []