diff --git a/manager-solution/manager/manager.py b/manager-solution/manager/manager.py index e9d26e3..1548e7b 100644 --- a/manager-solution/manager/manager.py +++ b/manager-solution/manager/manager.py @@ -13,7 +13,6 @@ import socket import threading import time from typing import Any, Dict, Iterable, Optional -from urllib.parse import urlsplit import docker @@ -28,8 +27,6 @@ STARTUP_STABILIZATION_SECONDS = float(os.getenv("STARTUP_STABILIZATION_SECONDS", TCP_CONNECT_TIMEOUT_SECONDS = float(os.getenv("TCP_CONNECT_TIMEOUT_SECONDS", "1.5")) INITIAL_REQUEST_TIMEOUT_SECONDS = float(os.getenv("INITIAL_REQUEST_TIMEOUT_SECONDS", "3")) MAX_REQUEST_HEADER_BYTES = int(os.getenv("MAX_REQUEST_HEADER_BYTES", "65536")) -GATEWAY_ROUTE_TTL_SECONDS = int(os.getenv("GATEWAY_ROUTE_TTL_SECONDS", "3600")) -GATEWAY_ROUTE_COOKIE_NAME = "fis_user_route" DEFAULTS: Dict[str, Any] = { "container_prefix": "fis-", @@ -46,7 +43,6 @@ DEFAULTS: Dict[str, Any] = { _docker_client = None user_activity: Dict[str, float] = {} users_by_id: Dict[str, Dict[str, Any]] = {} -gateway_client_routes: Dict[str, Dict[str, Any]] = {} user_container_locks: Dict[str, threading.Lock] = {} user_container_locks_guard = threading.Lock() @@ -222,45 +218,53 @@ def build_http_error(status_code: int, reason: str, message: str) -> bytes: return "\r\n".join(headers).encode("utf-8") + body -def build_http_redirect(location: str, extra_headers: Optional[Iterable[str]] = None) -> bytes: +def build_http_redirect(location: str) -> bytes: headers = [ "HTTP/1.1 302 Found", f"Location: {location}", "Content-Length: 0", "Connection: close", + "", + "", ] - headers.extend(extra_headers or []) - headers.extend(["", ""]) return "\r\n".join(headers).encode("utf-8") -def rewrite_http_request_target(data: bytes, target: str) -> bytes: - header_end = data.find(b"\r\n") - if header_end == -1: - return data - - request_line = data[:header_end].decode("iso-8859-1", errors="replace") - parts = request_line.split(" ", 2) - if len(parts) != 3: - return data - - method, _, version = parts - rewritten_line = f"{method} {target} {version}".encode("iso-8859-1") - return rewritten_line + data[header_end:] - - -def parse_http_request(data: bytes): - if not data: +def build_user_host(host: str, user_id: str) -> Optional[str]: + if not host: return None + hostname, separator, port = host.partition(":") + hostname = hostname.strip().lower() + if not hostname: + return None + + if hostname == "localhost": + target_host = f"{user_id}.localhost" + else: + labels = [label for label in hostname.split(".") if label] + if not labels: + return None + if labels[0] == user_id: + target_host = hostname + else: + target_host = ".".join([user_id, *labels]) + + return f"{target_host}:{port}" if separator else target_host + + +def extract_user_from_http_request(data: bytes, known_users: Dict[str, Dict[str, Any]]): + if not data: + return None, data, None + header_end = data.find(b"\r\n\r\n") if header_end == -1: - return None + return None, data, None header_block = data[:header_end].decode("iso-8859-1", errors="replace") lines = header_block.split("\r\n") if not lines or len(lines[0].split(" ")) < 3: - return None + return None, data, None method, target, version = lines[0].split(" ", 2) headers = {} @@ -270,98 +274,29 @@ def parse_http_request(data: bytes): key, value = line.split(":", 1) headers[key.strip().lower()] = value.strip() - return { - "method": method, - "target": target, - "version": version, - "headers": headers, - } - - -def extract_cookie_user(headers: Dict[str, str], known_users: Dict[str, Dict[str, Any]]) -> Optional[Dict[str, Any]]: - cookie_header = headers.get("cookie", "") - if not cookie_header: - return None - - for part in cookie_header.split(";"): - key, separator, value = part.strip().partition("=") - if separator and key == GATEWAY_ROUTE_COOKIE_NAME: - user_id = value.strip() - return known_users.get(user_id) - return None - - -def is_websocket_request(headers: Dict[str, str]) -> bool: - return headers.get("upgrade", "").lower() == "websocket" - - -def extract_request_target(data: bytes) -> Optional[str]: - parsed = parse_http_request(data) - if not parsed: - return None - return parsed["target"] - - -def remember_gateway_route(client_ip: str, user_id: str) -> None: - if not client_ip: - return - gateway_client_routes[client_ip] = {"user_id": user_id, "updated_at": time.time()} - - -def get_remembered_gateway_user(client_ip: str) -> Optional[Dict[str, Any]]: - if not client_ip: - return None - - route = gateway_client_routes.get(client_ip) - if not route: - return None - - if time.time() - route["updated_at"] > GATEWAY_ROUTE_TTL_SECONDS: - gateway_client_routes.pop(client_ip, None) - return None - - user_id = route["user_id"] - return users_by_id.get(user_id) - - -def extract_user_from_http_request(data: bytes, known_users: Dict[str, Dict[str, Any]]): - parsed = parse_http_request(data) - if not parsed: - return None, data, None, False - - target = parsed["target"] - headers = parsed["headers"] - host = headers.get("host", "").split(":", 1)[0].strip().lower() if host: labels = [label for label in host.split(".") if label] if labels: candidate = labels[0] if candidate in known_users and candidate not in {"localhost", "127", "0"}: - return known_users[candidate], data, None, False + return known_users[candidate], data, None - parsed_target = urlsplit(target) - if parsed_target.scheme and parsed_target.netloc: - path = parsed_target.path or "/" - query = parsed_target.query - separator = "?" if query else "" - else: - path, separator, query = target.partition("?") + path, separator, query = target.partition("?") segments = [segment for segment in path.split("/") if segment] if segments: candidate = segments[0] if candidate in known_users: - rewritten_path = "/" + "/".join(segments[1:]) if len(segments) > 1 else "/" - rewritten_target = f"{rewritten_path}?{query}" if separator else rewritten_path - rewritten_data = rewrite_http_request_target(data, rewritten_target) - return known_users[candidate], rewritten_data, None, True + user_host = build_user_host(headers.get("host", ""), candidate) + if user_host: + redirect_path = "/" + "/".join(segments[1:]) if len(segments) > 1 else "/" + if separator: + redirect_path = f"{redirect_path}?{query}" + redirect_target = f"http://{user_host}{redirect_path}" + return None, data, redirect_target - cookie_user = extract_cookie_user(headers, known_users) - if cookie_user: - return cookie_user, data, None, False - - return None, data, None, False + return None, data, None def wait_until_ready(container_name: str) -> bool: @@ -535,7 +470,6 @@ async def handle_connection(reader, writer, user: Dict[str, Any], initial_data: async def handle_gateway_connection(reader, writer): peer = writer.get_extra_info("peername") - client_ip = peer[0] if isinstance(peer, tuple) and peer else None try: initial_data = await read_initial_http_request(reader) except asyncio.TimeoutError: @@ -545,11 +479,7 @@ async def handle_gateway_connection(reader, writer): await writer.wait_closed() return - parsed_request = parse_http_request(initial_data) - user, rewritten_data, redirect_target, selected_via_path = extract_user_from_http_request(initial_data, users_by_id) - if user: - remember_gateway_route(client_ip, user["id"]) - + user, rewritten_data, redirect_target = extract_user_from_http_request(initial_data, users_by_id) if redirect_target: writer.write(build_http_redirect(redirect_target)) await writer.drain() @@ -557,29 +487,6 @@ async def handle_gateway_connection(reader, writer): await writer.wait_closed() return - if user and selected_via_path and parsed_request and not is_websocket_request(parsed_request["headers"]): - normalized_target = extract_request_target(rewritten_data) or "/" - writer.write( - build_http_redirect( - normalized_target, - extra_headers=[ - f"Set-Cookie: {GATEWAY_ROUTE_COOKIE_NAME}={user['id']}; Path=/; HttpOnly; SameSite=Lax" - ], - ) - ) - await writer.drain() - writer.close() - await writer.wait_closed() - log(f"[gateway] Switched {client_ip} to user {user['id']} via path route") - return - - if not user: - remembered_user = get_remembered_gateway_user(client_ip) - if remembered_user: - user = remembered_user - rewritten_data = initial_data - log(f"[gateway] Reused remembered route for {client_ip} -> user {user['id']}") - if not user: writer.write( build_http_error( @@ -603,10 +510,6 @@ async def idle_checker(): await asyncio.sleep(IDLE_CHECK_INTERVAL) now = time.time() - for client_ip, route in list(gateway_client_routes.items()): - if now - route["updated_at"] > GATEWAY_ROUTE_TTL_SECONDS: - gateway_client_routes.pop(client_ip, None) - for user_id, last_active in list(user_activity.items()): if now - last_active <= INACTIVITY_SECONDS: continue diff --git a/manager-solution/manager/test_manager.py b/manager-solution/manager/test_manager.py index b7a9f22..71f3353 100644 --- a/manager-solution/manager/test_manager.py +++ b/manager-solution/manager/test_manager.py @@ -192,14 +192,13 @@ class ManagerConfigTests(unittest.TestCase): ) users = {"user2": {"id": "user2"}} - user, rewritten, redirect_target, selected_via_path = self.manager.extract_user_from_http_request(request, users) + user, rewritten, redirect_target = self.manager.extract_user_from_http_request(request, users) self.assertEqual(user["id"], "user2") self.assertEqual(rewritten, request) self.assertIsNone(redirect_target) - self.assertFalse(selected_via_path) - def test_extract_user_from_path_rewrites_request_for_direct_proxying(self): + def test_extract_user_from_path_returns_redirect_target(self): request = ( b"GET /user1/websockify?token=abc HTTP/1.1\r\n" b"Host: localhost:8080\r\n" @@ -208,112 +207,11 @@ class ManagerConfigTests(unittest.TestCase): ) users = {"user1": {"id": "user1"}} - user, rewritten, redirect_target, selected_via_path = self.manager.extract_user_from_http_request(request, users) - - self.assertEqual(user["id"], "user1") - self.assertEqual( - rewritten, - ( - b"GET /websockify?token=abc HTTP/1.1\r\n" - b"Host: localhost:8080\r\n" - b"Upgrade: websocket\r\n" - b"\r\n" - ), - ) - self.assertIsNone(redirect_target) - self.assertTrue(selected_via_path) - - def test_extract_user_from_bare_user_path_rewrites_to_root(self): - request = ( - b"GET /user1 HTTP/1.1\r\n" - b"Host: 10.200.200.113:8080\r\n" - b"Connection: keep-alive\r\n" - b"\r\n" - ) - users = {"user1": {"id": "user1"}} - - user, rewritten, redirect_target, selected_via_path = self.manager.extract_user_from_http_request(request, users) - - self.assertEqual(user["id"], "user1") - self.assertEqual( - rewritten, - ( - b"GET / HTTP/1.1\r\n" - b"Host: 10.200.200.113:8080\r\n" - b"Connection: keep-alive\r\n" - b"\r\n" - ), - ) - self.assertIsNone(redirect_target) - self.assertTrue(selected_via_path) - - def test_extract_user_from_absolute_form_request_rewrites_to_root(self): - request = ( - b"GET http://10.200.200.113:8080/user1 HTTP/1.1\r\n" - b"Host: 10.200.200.113:8080\r\n" - b"Connection: keep-alive\r\n" - b"\r\n" - ) - users = {"user1": {"id": "user1"}} - - user, rewritten, redirect_target, selected_via_path = self.manager.extract_user_from_http_request(request, users) - - self.assertEqual(user["id"], "user1") - self.assertEqual( - rewritten, - ( - b"GET / HTTP/1.1\r\n" - b"Host: 10.200.200.113:8080\r\n" - b"Connection: keep-alive\r\n" - b"\r\n" - ), - ) - self.assertIsNone(redirect_target) - self.assertTrue(selected_via_path) - - def test_extract_user_from_cookie_route(self): - request = ( - b"GET / HTTP/1.1\r\n" - b"Host: 10.200.200.113:8080\r\n" - b"Cookie: fis_user_route=user2\r\n" - b"Connection: keep-alive\r\n" - b"\r\n" - ) - users = {"user2": {"id": "user2"}} - - user, rewritten, redirect_target, selected_via_path = self.manager.extract_user_from_http_request(request, users) - - self.assertEqual(user["id"], "user2") - self.assertEqual(rewritten, request) - self.assertIsNone(redirect_target) - self.assertFalse(selected_via_path) - - def test_remembered_gateway_route_returns_user_within_ttl(self): - self.manager.users_by_id = {"user1": {"id": "user1"}} - self.manager.gateway_client_routes.clear() - self.manager.time = types.SimpleNamespace(time=lambda: 100.0) - self.manager.GATEWAY_ROUTE_TTL_SECONDS = 60 - - self.manager.remember_gateway_route("10.200.200.136", "user1") - - user = self.manager.get_remembered_gateway_user("10.200.200.136") - - self.assertEqual(user["id"], "user1") - - def test_remembered_gateway_route_expires_after_ttl(self): - now = {"value": 100.0} - self.manager.users_by_id = {"user1": {"id": "user1"}} - self.manager.gateway_client_routes.clear() - self.manager.time = types.SimpleNamespace(time=lambda: now["value"]) - self.manager.GATEWAY_ROUTE_TTL_SECONDS = 60 - - self.manager.remember_gateway_route("10.200.200.136", "user1") - now["value"] = 161.0 - - user = self.manager.get_remembered_gateway_user("10.200.200.136") + user, rewritten, redirect_target = self.manager.extract_user_from_http_request(request, users) self.assertIsNone(user) - self.assertEqual(self.manager.gateway_client_routes, {}) + self.assertEqual(rewritten, request) + self.assertEqual(redirect_target, "http://user1.localhost:8080/websockify?token=abc") def test_wait_until_ready_applies_stabilization_delay(self): sleep_calls = []