| 1 | import json |
| 2 | import urllib.request |
| 3 | from urllib.parse import urlparse |
| 4 | |
| 5 | |
| 6 | _DEFAULT_PORTS = { |
| 7 | "http": 80, |
| 8 | "https": 443, |
| 9 | "ws": 80, |
| 10 | "wss": 443, |
| 11 | } |
| 12 | |
| 13 | |
| 14 | def origin_from_url(value): |
| 15 | """Normalize a URL or Origin header to scheme://host[:port].""" |
| 16 | if not isinstance(value, str) or not value.strip(): |
| 17 | return None |
| 18 | parsed = urlparse(value.strip()) |
| 19 | if not parsed.scheme or not parsed.hostname: |
| 20 | return None |
| 21 | |
| 22 | scheme = parsed.scheme.lower() |
| 23 | host = parsed.hostname.lower() |
| 24 | try: |
| 25 | port = parsed.port |
| 26 | except ValueError: |
| 27 | return None |
| 28 | |
| 29 | origin = f"{scheme}://{host}" |
| 30 | if port and port != _DEFAULT_PORTS.get(scheme): |
| 31 | origin += f":{port}" |
| 32 | return origin |
| 33 | |
| 34 | |
| 35 | def origin_key(value): |
| 36 | """Return a comparable same-origin tuple including default ports.""" |
| 37 | origin = origin_from_url(value) |
| 38 | if not origin: |
| 39 | return None |
| 40 | parsed = urlparse(origin) |
| 41 | try: |
| 42 | port = parsed.port or _DEFAULT_PORTS.get(parsed.scheme) |
| 43 | except ValueError: |
| 44 | return None |
| 45 | if not parsed.scheme or not parsed.hostname or port is None: |
| 46 | return None |
| 47 | return parsed.scheme, parsed.hostname.lower(), int(port) |
| 48 | |
| 49 | |
| 50 | def get_active_tunnel_origins(): |
| 51 | """Return normalized origins for currently active Remote Control URLs.""" |
| 52 | origins = [] |
| 53 | |
| 54 | try: |
| 55 | from helpers.tunnel_manager import TunnelManager |
| 56 | |
| 57 | tunnel_url = TunnelManager.get_instance().get_tunnel_url() |
| 58 | _append_origin(origins, tunnel_url) |
| 59 | except Exception: |
| 60 | pass |
| 61 | |
| 62 | try: |
| 63 | _append_origin(origins, _get_tunnel_service_url()) |
| 64 | except Exception: |
| 65 | pass |
| 66 | |
| 67 | return origins |
| 68 | |
| 69 | |
| 70 | def _append_origin(origins, url): |
| 71 | origin = origin_from_url(url) |
| 72 | if origin and origin not in origins: |
| 73 | origins.append(origin) |
| 74 | |
| 75 | |
| 76 | def _get_tunnel_service_url(): |
| 77 | try: |
| 78 | from helpers import dotenv, runtime |
| 79 | |
| 80 | should_query_service = bool( |
| 81 | runtime.is_dockerized() |
| 82 | or runtime.get_arg("tunnel_api_port") |
| 83 | or dotenv.get_dotenv_value("TUNNEL_API_PORT") |
| 84 | ) |
| 85 | if not should_query_service: |
| 86 | return None |
| 87 | |
| 88 | port = runtime.get_tunnel_api_port() |
| 89 | except Exception: |
| 90 | return None |
| 91 | |
| 92 | body = json.dumps({"action": "get"}).encode("utf-8") |
| 93 | request = urllib.request.Request( |
| 94 | f"http://localhost:{port}/", |
| 95 | data=body, |
| 96 | headers={"Content-Type": "application/json"}, |
| 97 | method="POST", |
| 98 | ) |
| 99 | try: |
| 100 | with urllib.request.urlopen(request, timeout=0.35) as response: |
| 101 | payload = json.loads(response.read().decode("utf-8", errors="replace")) |
| 102 | except Exception: |
| 103 | return None |
| 104 | |
| 105 | if isinstance(payload, dict) and payload.get("success"): |
| 106 | return payload.get("tunnel_url") |
| 107 | return None |