diff --git a/.env.template b/.env.template index 1b55f8b5b..909d0ae77 100644 --- a/.env.template +++ b/.env.template @@ -25,13 +25,28 @@ MONGO_URI=mongodb://localhost:27017/ # Database name (default: von_db) VON_DB_NAME=von_db -# Local fallback URI (used if primary connection fails) +# Local fallback is disabled by default so a remote primary cannot silently +# downgrade to an old workstation database. Set this to 1 only as an explicit +# local-development choice. MONGO_LOCAL_URI=mongodb://127.0.0.1:27017/?directConnection=true -MONGO_ALLOW_LOCAL_FALLBACK=1 - -# Hosted/cloud recommendation: disable local fallback so Atlas failures do not -# silently fall back to localhost. -# MONGO_ALLOW_LOCAL_FALLBACK=0 +MONGO_ALLOW_LOCAL_FALLBACK=0 +# MONGO_ALLOW_LOCAL_FALLBACK=1 + +# Optional remote recovery paths after the primary URI. For a pure DNS/SRV +# failure Von tries the lower-latency direct-host Atlas URI first. For TLS, +# TCP, health-check transport, or shared-retry operation failures it tries the +# supervised SSH forwards first. The explicit local-development fallback, if +# enabled, remains last. +# +# Supervised SSH forwards on loopback. Von derives the credential-bearing +# Mongo URIs in memory and selects only the writable replica member. +# MONGO_SSH_TUNNEL_FALLBACK_ENDPOINTS=127.0.0.1:27018,127.0.0.1:27019,127.0.0.1:27020 +# MONGO_SSH_TUNNEL_EXPECTED_REPLICA_SET=atlas-example-shard-0 +# A direct-host Atlas URI, useful when only SRV lookup is failing. +# MONGO_DNS_FALLBACK_URI=mongodb://YOUR_DIRECT_ATLAS_URI +# +# How long to keep a working fallback before probing the primary route again. +# VON_MONGO_FALLBACK_STICKY_SECONDS=300 # # Optional hosted Mongo startup hardening: # VON_MONGO_STRICT_STARTUP=1 diff --git a/README.md b/README.md index ebc38c83f..b1dc03902 100644 --- a/README.md +++ b/README.md @@ -111,10 +111,18 @@ MongoDB will be automatically installed during the setup process (via `setup_py. MONGO_PROJECT=Your Project Name ``` - For hosted/cloud development, also set `MONGO_ALLOW_LOCAL_FALLBACK=0` so - Atlas failures do not silently fall back to localhost. Once connectivity is - stable, prefer `VON_MONGO_STRICT_STARTUP=1` and - `VON_MONGO_STARTUP_PROBE=1`. + Keep the default `MONGO_ALLOW_LOCAL_FALLBACK=0` so Atlas failures do not + silently fall back to an old localhost database. After installing a + separately supervised SSH tunnel, expose its loopback listeners through + `MONGO_SSH_TUNNEL_FALLBACK_ENDPOINTS`; Von derives its Mongo credentials in + memory, discovers the writable forwarded replica member, and probes the + direct primary again on later database use after a bounded recovery + window. Once connectivity is stable, prefer `VON_MONGO_STRICT_STARTUP=1` + and `VON_MONGO_STARTUP_PROBE=1`. + + On macOS, the persistent per-user launchd tunnel and its install, status, + and uninstall commands are documented in + [`docs/engineering/environment_minimums.md`](docs/engineering/environment_minimums.md#macos-supervised-atlas-ssh-fallback). Von decides local versus remote MongoDB from the effective `MONGO_URI` host and logs either `[Von Database] Connecting to LOCAL MongoDB ...` or diff --git a/docs/engineering/environment_minimums.md b/docs/engineering/environment_minimums.md index 9cc6f2d63..3a194b2fb 100644 --- a/docs/engineering/environment_minimums.md +++ b/docs/engineering/environment_minimums.md @@ -11,6 +11,7 @@ Ollama path, Von can start without any explicit environment variables because: - `MONGO_URI` defaults to `mongodb://localhost:27017/` - `VON_DB_NAME` defaults to `von_db` +- fallback to a different local Mongo instance is disabled by default - the active LLM provider falls back to Ollama when no explicit remote provider is configured @@ -81,12 +82,104 @@ Why these matter: - `MONGO_ALLOW_LOCAL_FALLBACK=0` prevents Atlas failures from silently falling back to localhost +- `MONGO_SSH_TUNNEL_FALLBACK_ENDPOINTS=` enables a + secret-free, supervised SSH transport fallback. Von derives credentials from + the primary URI in memory, validates the expected replica set when configured, + and selects only the writable forwarded member. The SSH process itself must + be supervised separately; on macOS use + `scripts/install_macos_mongo_ssh_tunnel.py`. +- `MONGO_SSH_TUNNEL_EXPECTED_REPLICA_SET=` pins the forwarded members + when the set name cannot be derived from an existing URI - `MONGO_DNS_FALLBACK_URI=` can still be used as a non-local recovery path when SRV/DNS resolution is flaky +- `VON_MONGO_FALLBACK_STICKY_SECONDS=300` controls how long a working fallback + is reused before Von probes the direct primary route again - `VON_SKIP_BROWSER_LAUNCH=1` avoids remote hosts trying to open a local browser - `FLASK_SECRET_KEY` is required for sane hosted session handling and is mandatory when strict hosted OAuth startup is enabled +### macOS supervised Atlas SSH fallback + +Use a per-user launchd agent to own a persistent tunnel independently of the +Von process. It starts within that user's GUI login domain, restarts after SSH +or network failures, and is not a pre-login system daemon. Supply one forward +for every Atlas replica-set member so Von can select the current writable +primary after an election. The installer writes no Mongo credentials to the +LaunchAgent; it accepts SSH transport details only. + +```sh +.venv/bin/python scripts/install_macos_mongo_ssh_tunnel.py install \ + --ssh-destination operator@relay.example \ + --identity-file /absolute/path/to/private-key \ + --known-hosts-file /absolute/path/to/known_hosts \ + --forward 27018:atlas-member-00.example:27017 \ + --forward 27019:atlas-member-01.example:27017 \ + --forward 27020:atlas-member-02.example:27017 +``` + +The SSH identity must be owner-only, the relay must already have a pinned host +key in the selected known-hosts file, and the local ports must not be owned by +another process. The generated service uses strict host-key checking and +isolates itself from ambient SSH configuration. Then configure Von and restart +it, because these values are loaded when the backend module starts: + +```dotenv +MONGO_ALLOW_LOCAL_FALLBACK=0 +MONGO_SSH_TUNNEL_FALLBACK_ENDPOINTS=127.0.0.1:27018,127.0.0.1:27019,127.0.0.1:27020 +MONGO_SSH_TUNNEL_EXPECTED_REPLICA_SET=atlas-example-shard-0 +``` + +Explicitly pinning the actual replica-set name is strongly recommended, +especially for an SRV primary URI that does not carry a `replicaSet` option. +Check launchd and listener ownership without repeating the forward list: + +```sh +.venv/bin/python scripts/install_macos_mongo_ssh_tunnel.py status +./run.sh restart -NoBrowser -HealthTimeoutSec 180 +./run.sh status +``` + +Installer status proves only that the launchd-managed SSH PID owns each +configured loopback listener; it does not probe Mongo. While the primary route +is healthy, `./run.sh status` should continue to report `Mongo: Atlas`. A +functional fallback test must report `Mongo: Atlas via SSH tunnel (...)` and +complete a Mongo ping or canonical read through that route. + +The status command exits successfully when the query itself completed, even +when the service is absent or degraded. For a healthy listener layer, require +`configuration_status="configured"`, `plist_valid=true`, +`label_matches=true`, `installed=true`, `loaded=true`, +`listeners_checked=true`, and `listener_coverage_complete=true`. +`mongo_route_probed=false` is expected until a separate Mongo operation is +performed. + +Before removal, delete or disable +`MONGO_SSH_TUNNEL_FALLBACK_ENDPOINTS`, restart Von, and verify that its live +Mongo route no longer depends on the tunnel. Removal is then explicit: + +```sh +.venv/bin/python scripts/install_macos_mongo_ssh_tunnel.py uninstall \ + --confirm-uninstall +.venv/bin/python scripts/install_macos_mongo_ssh_tunnel.py status +``` + +Uninstall removes the LaunchAgent plist, but retains the SSH key, +known-hosts file, and `~/Library/Logs/Von` tunnel logs. + +Mongo still validates the remote certificate chain. Because a forwarded Atlas +member is contacted at a loopback hostname, only hostname matching is relaxed +for URIs derived from syntactically validated loopback endpoints. The +authenticated SSH tunnel, strict relay host-key check, writable-primary check, +and expected replica-set check when configured bound that exception. Do not +expose these ports on a non-loopback interface. + +Fallback selection is failure-sensitive: a pure DNS/SRV failure tries the +direct-host Atlas URI first; TLS, TCP, periodic health-check transport failures, +and operation failures handled by the shared retry path try the SSH route +first. An explicitly enabled local-development Mongo fallback is always last. +A working remote fallback is reused for a bounded window and Von then probes +the direct primary again. + For stricter hosted Mongo startup, also set: ```powershell diff --git a/run.sh b/run.sh index 09a60f2c8..c2ceca5c5 100755 --- a/run.sh +++ b/run.sh @@ -1487,6 +1487,7 @@ start_server() { echo "To fix: export VON_DB_NAME='von_db'" >&2 return 1 fi + log_mongo_ssh_tunnel_status local existing existing="$(get_pid || true)" @@ -1936,16 +1937,24 @@ try: except Exception: sys.exit(1) using_fallback=data.get("using_fallback") +fallback_kind=data.get("fallback_kind") atlas=data.get("atlas_detected") host=data.get("effective_host") if using_fallback: - label="Mongo: local fallback" + if fallback_kind == "ssh_tunnel": + label="Mongo: Atlas via SSH tunnel" + elif fallback_kind == "dns": + label="Mongo: Atlas direct-host fallback" + elif fallback_kind == "local": + label="Mongo: local fallback" + else: + label="Mongo: fallback" elif atlas: label="Mongo: Atlas" else: label="Mongo: "+(host or "unknown") -if host and label.startswith("Mongo: local fallback"): - label=f"Mongo: local fallback ({host})" +if host and using_fallback: + label=f"{label} ({host})" elif host and label.startswith("Mongo: Atlas"): label=f"Mongo: Atlas ({host})" elif host and label.startswith("Mongo: "): @@ -1960,6 +1969,41 @@ print(label)' 2>/dev/null)" log "$line" } +log_mongo_ssh_tunnel_status() { + local endpoints="${MONGO_SSH_TUNNEL_FALLBACK_ENDPOINTS:-}" + if [ -z "$endpoints" ]; then + return 0 + fi + if ! command -v lsof >/dev/null 2>&1; then + log "Mongo SSH tunnel: configured; listener status unavailable" + return 0 + fi + local ready=0 + local total=0 + local value="" + local port="" + local values=() + IFS=',' read -r -a values <<<"$endpoints" + for value in "${values[@]}"; do + port="${value##*:}" + port="${port//[[:space:]]/}" + if ! [[ "$port" =~ ^[0-9]+$ ]] || [ "$port" -lt 1 ] || [ "$port" -gt 65535 ]; then + continue + fi + total=$((total + 1)) + if lsof -nP -iTCP:"$port" -sTCP:LISTEN -t >/dev/null 2>&1; then + ready=$((ready + 1)) + fi + done + if [ "$total" -eq 0 ]; then + log "Mongo SSH tunnel: configured; listener status unavailable" + elif [ "$ready" -eq "$total" ]; then + log "Mongo SSH tunnel: listeners present (${ready}/${total}); Mongo route not probed" + else + log "Mongo SSH tunnel: listener coverage incomplete (${ready}/${total}); inspect live Mongo route" + fi +} + parse_scan_counts() { local output="$1" local py @@ -3215,6 +3259,7 @@ run_relation_alias_cleanup() { status_server() { # Match run.ps1: if PID file stale, try to sync from current listener. sync_pidfile_to_listener + log_mongo_ssh_tunnel_status local run_maintenance=0 if [ "$STATUS_RUN_MAINTENANCE" -eq 1 ] || is_truthy "${VON_STATUS_RUN_MAINTENANCE:-}"; then run_maintenance=1 diff --git a/scripts/install_macos_mongo_ssh_tunnel.py b/scripts/install_macos_mongo_ssh_tunnel.py new file mode 100755 index 000000000..a84eeebea --- /dev/null +++ b/scripts/install_macos_mongo_ssh_tunnel.py @@ -0,0 +1,600 @@ +#!/usr/bin/env python3 +"""Install or inspect a launchd-supervised loopback SSH tunnel for MongoDB. + +The generated LaunchAgent contains SSH transport configuration only. MongoDB +credentials remain in Von's existing ``MONGO_URI`` or ``MONGO_URI_FILE`` and +must never be supplied to this command. +""" + +from __future__ import annotations + +import argparse +import json +import os +import plistlib +import re +import socket +import subprocess +import tempfile +import time +from collections.abc import Sequence +from dataclasses import dataclass +from pathlib import Path + +DEFAULT_LABEL = "org.strongailab.von.mongo-atlas-tunnel" +_SSH_DESTINATION_RE = re.compile( + r"^[A-Za-z0-9_][A-Za-z0-9._-]*@" + r"(?:[A-Za-z0-9_][A-Za-z0-9._-]*|\[[0-9A-Fa-f:]+\])$" +) +_REMOTE_HOST_RE = re.compile(r"^(?:[A-Za-z0-9._-]+|\[[0-9A-Fa-f:]+\])$") + + +@dataclass(frozen=True) +class Forward: + local_port: int + remote_host: str + remote_port: int + + @property + def ssh_argument(self) -> str: + return f"127.0.0.1:{self.local_port}:" f"{self.remote_host}:{self.remote_port}" + + +def _tcp_port(value: str) -> int: + try: + port = int(value) + except ValueError as exc: + raise argparse.ArgumentTypeError("port must be an integer") from exc + if not 1 <= port <= 65535: + raise argparse.ArgumentTypeError("port must be between 1 and 65535") + return port + + +def _forward(value: str) -> Forward: + parts = value.rsplit(":", 2) + if len(parts) != 3: + raise argparse.ArgumentTypeError( + "forward must be LOCAL_PORT:REMOTE_HOST:REMOTE_PORT" + ) + local_port = _tcp_port(parts[0]) + remote_host = parts[1].strip() + remote_port = _tcp_port(parts[2]) + if not _REMOTE_HOST_RE.fullmatch(remote_host): + raise argparse.ArgumentTypeError("forward has an invalid remote host") + return Forward(local_port, remote_host, remote_port) + + +def _absolute_existing_file( + value: str, + *, + description: str, + allow_public_read: bool = False, +) -> Path: + path = Path(value).expanduser() + if not path.is_absolute(): + raise ValueError(f"{description} must be an absolute path") + if not path.is_file(): + raise ValueError(f"{description} must be an existing regular file") + forbidden_mode = 0o022 if allow_public_read else 0o077 + if path.stat().st_mode & forbidden_mode: + raise ValueError(f"{description} has unsafe group or other permissions") + return path + + +def _validate_unique_local_ports(forwards: Sequence[Forward]) -> None: + ports = [forward.local_port for forward in forwards] + if not ports: + raise ValueError("at least one forward is required") + if len(ports) != len(set(ports)): + raise ValueError("each forward must use a distinct local port") + + +def _validate_label(label: str) -> None: + if not re.fullmatch(r"[A-Za-z0-9._-]+", label): + raise ValueError( + "launchd label may contain only letters, numbers, dot, dash and underscore" + ) + + +def build_program_arguments( + *, + ssh_destination: str, + identity_file: Path, + known_hosts_file: Path, + forwards: Sequence[Forward], +) -> list[str]: + if not _SSH_DESTINATION_RE.fullmatch(ssh_destination): + raise ValueError("SSH destination must be a simple USER@HOST value") + _validate_unique_local_ports(forwards) + arguments = [ + "/usr/bin/ssh", + "-F", + "/dev/null", + "-N", + "-T", + "-o", + "BatchMode=yes", + "-o", + "IdentitiesOnly=yes", + "-o", + "StrictHostKeyChecking=yes", + "-o", + f"UserKnownHostsFile={known_hosts_file}", + "-o", + "ExitOnForwardFailure=yes", + "-o", + "ServerAliveInterval=15", + "-o", + "ServerAliveCountMax=3", + "-o", + "TCPKeepAlive=yes", + "-o", + "ControlMaster=no", + "-i", + str(identity_file), + ] + for forward in forwards: + arguments.extend(["-L", forward.ssh_argument]) + arguments.append(ssh_destination) + return arguments + + +def build_launch_agent( + *, + label: str, + program_arguments: Sequence[str], + stdout_path: Path, + stderr_path: Path, +) -> dict[str, object]: + return { + "Label": label, + "ProgramArguments": list(program_arguments), + "RunAtLoad": True, + "KeepAlive": True, + "ThrottleInterval": 10, + "ProcessType": "Background", + "Umask": 0o077, + "StandardOutPath": str(stdout_path), + "StandardErrorPath": str(stderr_path), + } + + +def _service_target(label: str) -> str: + return f"gui/{os.getuid()}/{label}" + + +def _run_launchctl( + arguments: Sequence[str], + *, + check: bool, +) -> subprocess.CompletedProcess[str]: + return subprocess.run( + ["/bin/launchctl", *arguments], + check=check, + capture_output=True, + text=True, + timeout=15, + ) + + +def _service_is_loaded(label: str) -> bool: + result = _run_launchctl(["print", _service_target(label)], check=False) + return result.returncode == 0 + + +def _service_pid(label: str) -> int | None: + result = _run_launchctl(["print", _service_target(label)], check=False) + if result.returncode != 0: + return None + match = re.search(r"\bpid = (\d+)", result.stdout) + return int(match.group(1)) if match else None + + +def _listener_pids(port: int) -> set[int]: + result = subprocess.run( + [ + "/usr/sbin/lsof", + "-nP", + "-t", + f"-iTCP:{port}", + "-sTCP:LISTEN", + ], + check=False, + capture_output=True, + text=True, + timeout=3, + ) + if result.returncode not in {0, 1}: + return set() + return { + int(line) + for line in result.stdout.splitlines() + if line.strip().isdigit() + } + + +def _ports_are_available(forwards: Sequence[Forward]) -> bool: + probes: list[socket.socket] = [] + try: + for forward in forwards: + probe = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + probe.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + probe.bind(("127.0.0.1", forward.local_port)) + probes.append(probe) + return True + except OSError: + return False + finally: + for probe in probes: + probe.close() + + +def _wait_for_listeners( + label: str, + forwards: Sequence[Forward], + *, + timeout_seconds: float = 15.0, +) -> bool: + deadline = time.monotonic() + timeout_seconds + while time.monotonic() < deadline: + pid = _service_pid(label) + if pid is not None and all( + pid in _listener_pids(forward.local_port) for forward in forwards + ): + return True + time.sleep(0.2) + return False + + +def _write_private_bytes(path: Path, data: bytes) -> None: + path.parent.mkdir(mode=0o700, parents=True, exist_ok=True) + with tempfile.NamedTemporaryFile( + mode="wb", + dir=path.parent, + prefix=f".{path.name}.", + delete=False, + ) as handle: + temporary_path = Path(handle.name) + handle.write(data) + handle.flush() + os.fsync(handle.fileno()) + try: + temporary_path.chmod(0o600) + os.replace(temporary_path, path) + finally: + temporary_path.unlink(missing_ok=True) + + +def _write_plist(path: Path, payload: dict[str, object]) -> None: + data = plistlib.dumps(payload, fmt=plistlib.FMT_XML, sort_keys=False) + _write_private_bytes(path, data) + + +def _prepare_log_directory(path: Path) -> None: + """Create a private log directory without mutating an existing path.""" + + if path.is_symlink(): + raise ValueError("log directory must not be a symbolic link") + if path.exists(): + if not path.is_dir(): + raise ValueError("log directory path must be a directory") + stat_result = path.stat() + if stat_result.st_uid != os.getuid(): + raise ValueError("existing log directory must be owned by the current user") + if stat_result.st_mode & 0o077 or stat_result.st_mode & 0o700 != 0o700: + raise ValueError( + "existing log directory must have owner-only rwx permissions" + ) + return + if not path.parent.is_dir(): + raise ValueError("log directory parent must already exist") + path.mkdir(mode=0o700) + + +def _bootstrap_service(label: str, plist_path: Path) -> None: + del label + _run_launchctl( + ["bootstrap", f"gui/{os.getuid()}", str(plist_path)], + check=True, + ) + + +def _restore_previous_install( + *, + label: str, + plist_path: Path, + previous_plist: bytes | None, + was_loaded: bool, +) -> None: + rollback_errors: list[str] = [] + try: + if _service_is_loaded(label): + _run_launchctl(["bootout", _service_target(label)], check=True) + except (OSError, RuntimeError, subprocess.SubprocessError) as exc: + rollback_errors.append(f"could not boot out failed replacement: {exc}") + try: + if previous_plist is None: + plist_path.unlink(missing_ok=True) + else: + _write_private_bytes(plist_path, previous_plist) + except (OSError, RuntimeError, subprocess.SubprocessError) as exc: + rollback_errors.append(f"could not restore previous plist: {exc}") + if was_loaded and previous_plist is not None: + try: + _bootstrap_service(label, plist_path) + except (OSError, RuntimeError, subprocess.SubprocessError) as exc: + rollback_errors.append(f"could not reload previous service: {exc}") + if rollback_errors: + raise RuntimeError("; ".join(rollback_errors)) + + +def _status_payload( + *, + label: str, + plist_path: Path, + forwards: Sequence[Forward], +) -> dict[str, object]: + loaded = _service_is_loaded(label) + pid = _service_pid(label) if loaded else None + configuration_status, plist_valid, label_matches = _installed_plist_status( + plist_path, label + ) + listeners = { + str(forward.local_port): bool( + pid is not None and pid in _listener_pids(forward.local_port) + ) + for forward in forwards + } + return { + "label": label, + "installed": plist_path.is_file(), + "loaded": loaded, + "plist_path": str(plist_path), + "configuration_status": configuration_status, + "plist_valid": plist_valid, + "label_matches": label_matches, + "service_pid": pid, + "listeners_checked": bool(forwards), + "listener_coverage_complete": bool(listeners) and all(listeners.values()), + "listeners": listeners, + "mongo_route_probed": False, + } + + +def _installed_plist_status( + plist_path: Path, + expected_label: str, +) -> tuple[str, bool | None, bool | None]: + if not plist_path.exists(): + return ("absent", None, None) + try: + with plist_path.open("rb") as handle: + payload = plistlib.load(handle) + except (OSError, plistlib.InvalidFileException, ValueError, TypeError): + return ("invalid", False, None) + if not isinstance(payload, dict) or not isinstance(payload.get("Label"), str): + return ("invalid", False, None) + label_matches = payload["Label"] == expected_label + return ( + "configured" if label_matches else "foreign_label", + True, + label_matches, + ) + + +def _forwards_from_plist(plist_path: Path) -> list[Forward]: + try: + with plist_path.open("rb") as handle: + payload = plistlib.load(handle) + except (OSError, plistlib.InvalidFileException, ValueError, TypeError): + return [] + if not isinstance(payload, dict): + return [] + arguments = payload.get("ProgramArguments") + if not isinstance(arguments, list): + return [] + forwards: list[Forward] = [] + for index, argument in enumerate(arguments[:-1]): + if argument != "-L" or not isinstance(arguments[index + 1], str): + continue + specification = arguments[index + 1] + prefix = "127.0.0.1:" + if not specification.startswith(prefix): + continue + try: + local_port, remote = specification[len(prefix) :].split(":", 1) + remote_host, remote_port = remote.rsplit(":", 1) + forwards.append( + _forward(f"{local_port}:{remote_host}:{remote_port}") + ) + except (ValueError, argparse.ArgumentTypeError): + return [] + return forwards + + +def _validate_installed_plist_label(plist_path: Path, expected_label: str) -> None: + """Refuse to replace or remove a file that is not this LaunchAgent.""" + + try: + with plist_path.open("rb") as handle: + payload = plistlib.load(handle) + except (OSError, plistlib.InvalidFileException, ValueError, TypeError) as exc: + raise ValueError( + f"refusing to modify invalid LaunchAgent plist: {plist_path}" + ) from exc + if not isinstance(payload, dict) or not isinstance(payload.get("Label"), str): + raise TypeError( + f"refusing to modify invalid LaunchAgent plist: {plist_path}" + ) + if payload["Label"] != expected_label: + raise ValueError( + "refusing to modify a LaunchAgent plist owned by a different label" + ) + + +def _build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("action", choices=("install", "status", "uninstall")) + parser.add_argument("--label", default=DEFAULT_LABEL) + parser.add_argument("--ssh-destination") + parser.add_argument("--identity-file") + parser.add_argument( + "--known-hosts-file", + default=str(Path.home() / ".ssh" / "known_hosts"), + ) + parser.add_argument( + "--forward", + action="append", + type=_forward, + default=[], + metavar="LOCAL_PORT:REMOTE_HOST:REMOTE_PORT", + ) + parser.add_argument("--plist-path") + parser.add_argument("--log-directory") + parser.add_argument("--confirm-uninstall", action="store_true") + return parser + + +def _paths_from_args(args: argparse.Namespace) -> tuple[Path, Path]: + plist_path = ( + Path(args.plist_path).expanduser() + if args.plist_path + else Path.home() / "Library" / "LaunchAgents" / f"{args.label}.plist" + ) + log_directory = ( + Path(args.log_directory).expanduser() + if args.log_directory + else Path.home() / "Library" / "Logs" / "Von" + ) + if not plist_path.is_absolute() or not log_directory.is_absolute(): + raise ValueError("plist and log paths must be absolute") + return plist_path, log_directory + + +def run(argv: Sequence[str] | None = None) -> int: + args = _build_parser().parse_args(argv) + _validate_label(args.label) + plist_path, log_directory = _paths_from_args(args) + forwards: list[Forward] = list(args.forward) + + if args.action == "status": + if not forwards: + forwards = _forwards_from_plist(plist_path) + if forwards: + _validate_unique_local_ports(forwards) + print( + json.dumps( + _status_payload( + label=args.label, + plist_path=plist_path, + forwards=forwards, + ), + sort_keys=True, + ) + ) + return 0 + + if args.action == "uninstall": + if not args.confirm_uninstall: + raise ValueError("uninstall requires --confirm-uninstall") + if plist_path.exists(): + _validate_installed_plist_label(plist_path, args.label) + loaded = _service_is_loaded(args.label) + if loaded and not plist_path.exists(): + raise RuntimeError( + "refusing to boot out a loaded service without its matching plist" + ) + if loaded: + _run_launchctl(["bootout", _service_target(args.label)], check=True) + plist_path.unlink(missing_ok=True) + print( + json.dumps( + {"label": args.label, "installed": False, "loaded": False}, + sort_keys=True, + ) + ) + return 0 + + if not args.ssh_destination or not args.identity_file: + raise ValueError("install requires --ssh-destination and --identity-file") + _validate_unique_local_ports(forwards) + identity_file = _absolute_existing_file( + args.identity_file, + description="identity file", + ) + known_hosts_file = _absolute_existing_file( + args.known_hosts_file, + description="known-hosts file", + allow_public_read=True, + ) + program_arguments = build_program_arguments( + ssh_destination=args.ssh_destination, + identity_file=identity_file, + known_hosts_file=known_hosts_file, + forwards=forwards, + ) + _prepare_log_directory(log_directory) + payload = build_launch_agent( + label=args.label, + program_arguments=program_arguments, + stdout_path=log_directory / "mongo-atlas-tunnel.stdout.log", + stderr_path=log_directory / "mongo-atlas-tunnel.stderr.log", + ) + + was_loaded = _service_is_loaded(args.label) + previous_plist = plist_path.read_bytes() if plist_path.is_file() else None + if previous_plist is not None: + _validate_installed_plist_label(plist_path, args.label) + if was_loaded and previous_plist is None: + raise RuntimeError( + "refusing to replace a loaded service whose plist cannot be recovered" + ) + try: + if was_loaded: + _run_launchctl(["bootout", _service_target(args.label)], check=True) + if not _ports_are_available(forwards): + raise RuntimeError( + "one or more loopback ports are already owned; no process was killed" + ) + _write_plist(plist_path, payload) + _bootstrap_service(args.label, plist_path) + if not _wait_for_listeners(args.label, forwards): + raise RuntimeError( + "launchd loaded the tunnel but not all loopback listeners became ready" + ) + except (OSError, RuntimeError, subprocess.SubprocessError) as exc: + try: + _restore_previous_install( + label=args.label, + plist_path=plist_path, + previous_plist=previous_plist, + was_loaded=was_loaded, + ) + except (OSError, RuntimeError, subprocess.SubprocessError) as rollback_exc: + raise RuntimeError( + f"{exc}; rollback also failed: {rollback_exc}" + ) from exc + raise + print( + json.dumps( + _status_payload( + label=args.label, + plist_path=plist_path, + forwards=forwards, + ), + sort_keys=True, + ) + ) + return 0 + + +def main() -> int: + try: + return run() + except (OSError, ValueError, RuntimeError, subprocess.SubprocessError) as exc: + print(json.dumps({"success": False, "error": str(exc)}, sort_keys=True)) + return 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/setup_py.ps1 b/setup_py.ps1 index c8a310424..2099153d8 100644 --- a/setup_py.ps1 +++ b/setup_py.ps1 @@ -516,7 +516,7 @@ function New-EnvFile { MONGO_URI=mongodb://localhost:27017/ VON_DB_NAME=von_db MONGO_LOCAL_URI=mongodb://127.0.0.1:27017/?directConnection=true -MONGO_ALLOW_LOCAL_FALLBACK=1 +MONGO_ALLOW_LOCAL_FALLBACK=0 OLLAMA_HOSTS_LIST=127.0.0.1 "@ $envContent | Out-File -FilePath ".env" -Encoding UTF8 diff --git a/setup_py.sh b/setup_py.sh index 2a170dd75..1cbcff2ff 100755 --- a/setup_py.sh +++ b/setup_py.sh @@ -348,7 +348,7 @@ ensure_env_file() { MONGO_URI=mongodb://localhost:27017/ VON_DB_NAME=von_db MONGO_LOCAL_URI=mongodb://127.0.0.1:27017/?directConnection=true -MONGO_ALLOW_LOCAL_FALLBACK=1 +MONGO_ALLOW_LOCAL_FALLBACK=0 OLLAMA_HOSTS_LIST=127.0.0.1 EOF echo -e "${GREEN}✓ Created minimal .env file${NC}" diff --git a/src/backend/db/connection_manager.py b/src/backend/db/connection_manager.py index 8f43f213b..b9a0851c9 100644 --- a/src/backend/db/connection_manager.py +++ b/src/backend/db/connection_manager.py @@ -6,6 +6,7 @@ from pymongo.errors import ConnectionFailure from . import mongo_client as _mc +from .mongo_error_classification import is_mongo_transport_error from .mongo_uri_redaction import build_safe_mongo_connection_location logger = logging.getLogger(__name__) @@ -40,13 +41,17 @@ def get_db(): def health_summary() -> Dict[str, Any]: db = get_db() connected = False + route_degraded = False + health_error_type = None if db is not None: try: # Use lightweight ping; ignore actual response object db.command("ping") # type: ignore[call-arg] connected = True - except Exception: + except Exception as exc: connected = False + route_degraded = is_mongo_transport_error(exc) + health_error_type = type(exc).__name__ uptime = None if connected and _state["last_connected_at"]: uptime = _now() - _state["last_connected_at"] @@ -78,19 +83,26 @@ def health_summary() -> Dict[str, Any]: effective_uri = str(eff_uri_raw) if eff_uri_raw is not None else None except Exception: effective_uri = None + try: + fallback_policy = getattr(_mc, "get_mongo_fallback_policy_state", lambda: {})() + except Exception: + fallback_policy = {} + fallback_kind = ( + fallback_policy.get("active_fallback_kind") + if isinstance(fallback_policy, dict) + else None + ) connection_location = build_safe_mongo_connection_location( effective_uri, using_fallback=using_fallback, + fallback_kind=fallback_kind if isinstance(fallback_kind, str) else None, + fallback_target_uri=getattr(_mc, "MONGO_URI", None), ) effective_uri_sanitized = connection_location.get("sanitized_uri") - try: - fallback_policy = getattr( - _mc, "get_mongo_fallback_policy_state", lambda: {} - )() - except Exception: - fallback_policy = {} return { "connected": connected, + "route_degraded": route_degraded, + "health_error_type": health_error_type, "using_fallback": using_fallback, # Backwards-compatible key. This is intentionally sanitized only. "effective_uri": effective_uri_sanitized, @@ -109,7 +121,12 @@ def _exponential_backoff(base_ms: int, attempt: int, max_ms: int) -> float: def attempt_reconnect( - force: bool = False, max_attempts: int = 5, base_ms: int = 1000, max_ms: int = 30000 + force: bool = False, + max_attempts: int = 5, + base_ms: int = 1000, + max_ms: int = 30000, + route_degraded: bool = False, + degradation_reason: str | None = None, ) -> Dict[str, Any]: with _lock: # If already healthy and not forced, short circuit @@ -125,6 +142,13 @@ def attempt_reconnect( _state["reconnect_attempts_total"] += 1 _state["last_attempt_started_at"] = _now() try: + if route_degraded: + prefer_alternate = getattr( + _mc, "prefer_alternate_mongo_route", lambda _reason: None + ) + prefer_alternate( + degradation_reason or "transient operation transport failure" + ) _mc.invalidate_connection() db = _mc.get_db() if db is None: @@ -143,6 +167,9 @@ def attempt_reconnect( } except Exception as e: # Broad catch; we classify as failure last_err = str(e) + if is_mongo_transport_error(e): + route_degraded = True + degradation_reason = f"reconnect ping {type(e).__name__}" with _lock: _state["reconnect_failure_total"] += 1 _state["consecutive_failures"] += 1 @@ -169,6 +196,12 @@ def _monitor_loop(interval_sec: int, base_ms: int, max_ms: int, max_attempts: in max_attempts=max_attempts, base_ms=base_ms, max_ms=max_ms, + route_degraded=bool(h.get("route_degraded")), + degradation_reason=( + f"monitor ping {h.get('health_error_type')}" + if h.get("route_degraded") + else None + ), ) except Exception as exc: with _lock: diff --git a/src/backend/db/mongo_client.py b/src/backend/db/mongo_client.py index 5775a7a42..ee8e05096 100644 --- a/src/backend/db/mongo_client.py +++ b/src/backend/db/mongo_client.py @@ -1,16 +1,20 @@ +import datetime # Added for type hinting and __main__ example +import ipaddress +import logging import os import sys -import logging import threading import time from collections.abc import Mapping +from dataclasses import dataclass from pathlib import Path -from pymongo import MongoClient, ASCENDING, DESCENDING, monitoring +from urllib.parse import parse_qsl, urlencode, urlsplit + +from pymongo import ASCENDING, DESCENDING, MongoClient, monitoring from pymongo.collection import Collection from pymongo.database import Database from pymongo.errors import ConnectionFailure, OperationFailure -import datetime # Added for type hinting and __main__ example -from ..utils.runtime_env import get_env_bool, load_secret_from_env_or_file + from ..services.mongo_observability_service import ( build_mongo_command_shape, extract_n_returned_from_reply, @@ -18,6 +22,8 @@ record_mongo_command_observation, record_mongo_operation, ) +from ..utils.runtime_env import get_env_bool, load_secret_from_env_or_file +from .mongo_error_classification import is_mongo_transport_error logger = logging.getLogger(__name__) @@ -350,6 +356,66 @@ def _get_nonempty_env_value(name: str) -> str | None: return value if value else None +def _normalise_loopback_tunnel_endpoint(value: str) -> str: + """Validate and normalise a loopback host:port tunnel endpoint. + + Tunnel-derived Mongo URIs deliberately relax TLS hostname matching because + the client connects to loopback while the certificate names the remote + Atlas member. Restricting these endpoints to loopback keeps that exception + inside the authenticated SSH transport. + """ + + candidate = value.strip() + parsed = urlsplit(f"//{candidate}") + try: + host = parsed.hostname + port = parsed.port + except ValueError as exc: + raise ValueError("endpoint must contain a valid TCP port") from exc + if ( + not host + or port is None + or not 1 <= port <= 65535 + or parsed.username is not None + or parsed.password is not None + or parsed.path + or parsed.query + or parsed.fragment + ): + raise ValueError("endpoint must be a loopback host and TCP port") + is_loopback = host.lower() == "localhost" + if not is_loopback: + try: + is_loopback = ipaddress.ip_address(host).is_loopback + except ValueError: + is_loopback = False + if not is_loopback: + raise ValueError("endpoint must resolve syntactically to loopback") + normalised_host = f"[{host}]" if ":" in host else host + return f"{normalised_host}:{port}" + + +def _parse_ssh_tunnel_fallback_endpoints(raw: str | None) -> tuple[str, ...]: + if not raw: + return () + endpoints: list[str] = [] + for index, value in enumerate(raw.split(","), start=1): + if not value.strip(): + continue + try: + endpoint = _normalise_loopback_tunnel_endpoint(value) + except ValueError as exc: + logger.warning( + "[mongo_fallback] Ignoring invalid SSH tunnel endpoint #%s: %s", + index, + exc, + ) + continue + if endpoint not in endpoints: + endpoints.append(endpoint) + return tuple(endpoints) + + MONGO_URI = ( load_secret_from_env_or_file("MONGO_URI", "MONGO_URI_FILE") or _get_nonempty_env_value("MONGO_URI") @@ -361,7 +427,22 @@ def _get_nonempty_env_value(name: str) -> str | None: or "mongodb://127.0.0.1:27017/?directConnection=true" ) MONGO_DNS_FALLBACK_URI = _get_nonempty_env_value("MONGO_DNS_FALLBACK_URI") -MONGO_ALLOW_LOCAL_FALLBACK = get_env_bool("MONGO_ALLOW_LOCAL_FALLBACK", True) +MONGO_ALLOW_LOCAL_FALLBACK = get_env_bool("MONGO_ALLOW_LOCAL_FALLBACK", False) +MONGO_SSH_TUNNEL_FALLBACK_ENDPOINTS = _parse_ssh_tunnel_fallback_endpoints( + _get_nonempty_env_value("MONGO_SSH_TUNNEL_FALLBACK_ENDPOINTS") +) +MONGO_SSH_TUNNEL_EXPECTED_REPLICA_SET = _get_nonempty_env_value( + "MONGO_SSH_TUNNEL_EXPECTED_REPLICA_SET" +) + + +@dataclass(frozen=True) +class _MongoFallbackTarget: + uri: str + kind: str + label: str + require_writable_primary: bool = False + expected_replica_set: str | None = None def _get_positive_int_env(name: str, default: int) -> int: @@ -401,6 +482,15 @@ def _mongo_auto_recovery_ping_timeout_ms() -> int: def _mongo_dns_fallback_sticky_seconds() -> float: + """Return the fallback retry window, retaining the legacy setting name.""" + + generic_value = _get_nonempty_env_value("VON_MONGO_FALLBACK_STICKY_SECONDS") + if generic_value is not None: + try: + parsed = float(generic_value) + except ValueError: + parsed = 300.0 + return parsed if parsed > 0 else 300.0 return _get_positive_float_env("VON_MONGO_DNS_FALLBACK_STICKY_SECONDS", 300.0) @@ -416,23 +506,153 @@ def _mongo_socket_timeout_ms() -> int: return _get_positive_int_env("MONGO_SOCKET_TIMEOUT_MS", 5000) +def _build_ssh_tunnel_mongo_uri(primary_uri: str, endpoint: str) -> str: + """Derive a direct loopback Mongo URI without decoding URI credentials. + + The SSH process owns transport and authentication to the relay host. Mongo + credentials remain sourced from ``MONGO_URI`` and are only carried in this + in-memory derived URI. Certificate-chain validation stays enabled, while + hostname validation is relaxed solely for the validated loopback endpoint. + """ + + endpoint = _normalise_loopback_tunnel_endpoint(endpoint) + scheme_separator = primary_uri.find("://") + if scheme_separator < 0: + raise ValueError("primary Mongo URI has no recognised scheme") + scheme = primary_uri[:scheme_separator].lower() + if scheme not in {"mongodb", "mongodb+srv"}: + raise ValueError("primary Mongo URI must use mongodb or mongodb+srv") + + remainder = primary_uri[scheme_separator + 3 :] + tail_positions = [ + position + for position in (remainder.find("/"), remainder.find("?")) + if position >= 0 + ] + tail_start = min(tail_positions) if tail_positions else len(remainder) + authority = remainder[:tail_start] + tail = remainder[tail_start:] + userinfo_prefix = f"{authority.rsplit('@', 1)[0]}@" if "@" in authority else "" + + if tail.startswith("?"): + path = "/" + raw_query = tail[1:] + else: + path, separator, raw_query = tail.partition("?") + if not separator: + raw_query = "" + path = path or "/" + + controlled_options = { + "directconnection", + "loadbalanced", + "maxstalenessseconds", + "readpreference", + "readpreferencetags", + "replicaset", + "ssl", + "srvmaxhosts", + "srvservicename", + "tls", + "tlsallowinvalidcertificates", + "tlsallowinvalidhostnames", + "tlsinsecure", + } + options = [ + (key, value) + for key, value in parse_qsl(raw_query, keep_blank_values=True) + if key.lower() not in controlled_options + ] + if not any(key.lower() == "authsource" for key, _ in options): + options.append(("authSource", "admin")) + options.extend( + [ + ("directConnection", "true"), + ("tls", "true"), + ("tlsAllowInvalidCertificates", "false"), + ("tlsAllowInvalidHostnames", "true"), + ] + ) + return f"mongodb://{userinfo_prefix}{endpoint}{path}?{urlencode(options)}" + + +def _configured_replica_set_name() -> str | None: + """Read the expected replica-set identity from existing Mongo URI options.""" + + if MONGO_SSH_TUNNEL_EXPECTED_REPLICA_SET: + return MONGO_SSH_TUNNEL_EXPECTED_REPLICA_SET + for uri in (MONGO_DNS_FALLBACK_URI, MONGO_URI): + if not uri or "?" not in uri: + continue + raw_query = uri.split("?", 1)[1] + for key, value in parse_qsl(raw_query, keep_blank_values=True): + if key.lower() == "replicaset" and value.strip(): + return value.strip() + return None + + +def _connection_fallback_targets( + *, + prefer_dns: bool, +) -> list[_MongoFallbackTarget]: + """Return configured recovery candidates in decreasing preference order.""" + + tunnel_targets: list[_MongoFallbackTarget] = [] + expected_replica_set = _configured_replica_set_name() + for endpoint in MONGO_SSH_TUNNEL_FALLBACK_ENDPOINTS: + try: + uri = _build_ssh_tunnel_mongo_uri(MONGO_URI, endpoint) + except ValueError as exc: + logger.warning( + "[mongo_fallback] Cannot derive SSH tunnel Mongo URI: %s", exc + ) + break + tunnel_targets.append( + _MongoFallbackTarget( + uri=uri, + kind="ssh_tunnel", + label="SSH tunnel", + require_writable_primary=True, + expected_replica_set=expected_replica_set, + ) + ) + dns_target = ( + _MongoFallbackTarget( + uri=MONGO_DNS_FALLBACK_URI, + kind="dns", + label="direct-host Atlas", + ) + if MONGO_DNS_FALLBACK_URI + else None + ) + targets: list[_MongoFallbackTarget] = [] + if prefer_dns and dns_target is not None: + targets.append(dns_target) + targets.extend(tunnel_targets) + if not prefer_dns and dns_target is not None: + targets.append(dns_target) + if MONGO_ALLOW_LOCAL_FALLBACK: + targets.append( + _MongoFallbackTarget( + uri=MONGO_LOCAL_URI, + kind="local", + label="local", + ) + ) + return targets + + def _resolve_connection_fallback_target( *, prefer_dns: bool, ) -> tuple[str, str] | None: - """Return the best configured fallback URI for Mongo recovery. - - Hosted guidance uses ``MONGO_ALLOW_LOCAL_FALLBACK=0`` to block a silent - downgrade to localhost. That flag should not suppress a separately - configured direct-host Atlas fallback URI because that remains a - non-local recovery path. - """ + """Compatibility wrapper returning the first configured fallback.""" - if prefer_dns and MONGO_DNS_FALLBACK_URI: - return (MONGO_DNS_FALLBACK_URI, "dns fallback") - if MONGO_ALLOW_LOCAL_FALLBACK: - return (MONGO_LOCAL_URI, "local fallback") - return None + targets = _connection_fallback_targets(prefer_dns=prefer_dns) + if not targets: + return None + target = targets[0] + return (target.uri, f"{target.kind} fallback") def _is_running_under_pytest() -> bool: @@ -534,12 +754,14 @@ def assert_destructive_db_operation_allowed(operation: str) -> None: _mongo_client_real: MongoClient | None = None _mongo_client_mock = None -# Track whether we are using a fallback URI (local) rather than the primary MONGO_URI +# Track whether we are using a fallback URI rather than the primary MONGO_URI. _using_fallback_real = False +_active_fallback_kind_real: str | None = None _effective_uri_real = MONGO_URI # The URI actually used to create the real client _last_auto_recovery_check_at = 0.0 _dns_fallback_preferred_until_monotonic = 0.0 _dns_fallback_preference_reason: str | None = None +_preferred_fallback_kind: str | None = None _COLLECTION_INDEXES_LOCK = threading.Lock() _COLLECTION_INDEXES_READY: set[tuple[int, str, str]] = set() @@ -580,28 +802,78 @@ def _dns_fallback_is_preferred() -> bool: return bool( MONGO_DNS_FALLBACK_URI and _dns_fallback_preferred_until_monotonic > time.monotonic() + and _preferred_fallback_kind in {None, "dns"} ) -def _mark_dns_fallback_preferred(reason: str) -> None: - """Prefer the configured direct-host Atlas fallback for a bounded window.""" +def _fallback_is_preferred() -> bool: + return bool( + _preferred_fallback_kind + and _dns_fallback_preferred_until_monotonic > time.monotonic() + ) + + +def _mark_fallback_preferred(kind: str, reason: str) -> None: + """Prefer a successful fallback kind for a bounded recovery window.""" global _dns_fallback_preferred_until_monotonic global _dns_fallback_preference_reason + global _preferred_fallback_kind sticky_seconds = _mongo_dns_fallback_sticky_seconds() - if not MONGO_DNS_FALLBACK_URI or sticky_seconds <= 0: + if sticky_seconds <= 0: _dns_fallback_preferred_until_monotonic = 0.0 _dns_fallback_preference_reason = None + _preferred_fallback_kind = None return _dns_fallback_preferred_until_monotonic = time.monotonic() + sticky_seconds _dns_fallback_preference_reason = reason + _preferred_fallback_kind = kind + + +def _mark_dns_fallback_preferred(reason: str) -> None: + """Prefer the configured direct-host Atlas fallback for a bounded window.""" + + if MONGO_DNS_FALLBACK_URI: + _mark_fallback_preferred("dns", reason) def _clear_dns_fallback_preference() -> None: global _dns_fallback_preferred_until_monotonic global _dns_fallback_preference_reason + global _preferred_fallback_kind _dns_fallback_preferred_until_monotonic = 0.0 _dns_fallback_preference_reason = None + _preferred_fallback_kind = None + + +def prefer_alternate_mongo_route(reason: str) -> str | None: + """Prefer a different configured route after an operation transport failure. + + A route may still answer ``hello`` while real operations are timing out or + being reset. This bounded hint lets the normal reconnect path try the next + remote transport rather than repeatedly selecting the superficially + healthy route. + """ + + current_kind = _active_fallback_kind_real if _using_fallback_real else None + seen_kinds: set[str] = set() + for target in _connection_fallback_targets(prefer_dns=False): + if target.kind in seen_kinds or target.kind == current_kind: + continue + seen_kinds.add(target.kind) + _mark_fallback_preferred(target.kind, reason) + logger.warning( + "[mongo_recovery] Marked current Mongo route degraded; next reconnect will try %s.", + target.label, + ) + return target.kind + if current_kind is not None: + _clear_dns_fallback_preference() + logger.warning( + "[mongo_recovery] No alternate fallback route is configured; " + "next reconnect will probe the direct primary." + ) + return None def _try_connect_uri( @@ -609,12 +881,28 @@ def _try_connect_uri( *, fallback_label: str | None = None, server_selection_timeout_ms: int | None = None, + require_writable_primary: bool = False, + expected_replica_set: str | None = None, ) -> MongoClient: client = _new_mongo_client( uri, server_selection_timeout_ms=server_selection_timeout_ms, ) - client.admin.command("ismaster") + try: + hello = client.admin.command("hello") + if require_writable_primary and not bool( + hello.get("isWritablePrimary", hello.get("ismaster", False)) + ): + raise ConnectionFailure( + "SSH tunnel endpoint reached a replica member that is not writable" + ) + if expected_replica_set and hello.get("setName") != expected_replica_set: + raise ConnectionFailure( + "SSH tunnel endpoint reached an unexpected Mongo replica set" + ) + except Exception: + client.close() + raise if ( fallback_label == "dns fallback" and MONGO_DNS_FALLBACK_URI @@ -626,31 +914,114 @@ def _try_connect_uri( return client -def _try_preferred_dns_fallback_first() -> bool: - """Connect to the direct-host Atlas fallback while the sticky window is active.""" +def _try_preferred_fallback_first() -> bool: + """Reconnect through the recent fallback kind during its recovery window.""" global _mongo_client_real, _effective_uri_real, _using_fallback_real - if not _dns_fallback_is_preferred() or not MONGO_DNS_FALLBACK_URI: + global _active_fallback_kind_real, _preferred_fallback_kind + + preferred_kind = _preferred_fallback_kind + # Preserve compatibility with an already-established legacy DNS sticky + # window created before the fallback kind was recorded. + if preferred_kind is None and _dns_fallback_is_preferred(): + preferred_kind = "dns" + _preferred_fallback_kind = preferred_kind + if not preferred_kind or ( + _dns_fallback_preferred_until_monotonic <= time.monotonic() + ): return False - try: + + targets = [ + target + for target in _connection_fallback_targets(prefer_dns=True) + if target.kind == preferred_kind + ] + for target in targets: + try: + logger.warning( + "[mongo_fallback] Preferring recent %s route during fallback recovery window.", + target.label, + ) + _mongo_client_real = _try_connect_uri( + target.uri, + fallback_label=f"{target.kind} fallback", + server_selection_timeout_ms=3000, + require_writable_primary=target.require_writable_primary, + expected_replica_set=target.expected_replica_set, + ) + _effective_uri_real = target.uri + _using_fallback_real = True + _active_fallback_kind_real = target.kind + return True + except Exception as exc: + logger.warning( + "[mongo_fallback] Preferred %s connection failed: %s", + target.label, + exc, + ) + _mongo_client_real = None + + _clear_dns_fallback_preference() + return False + + +def _try_preferred_dns_fallback_first() -> bool: + """Compatibility alias for the general preferred-fallback recovery path.""" + + return _try_preferred_fallback_first() + + +def _try_configured_fallbacks(*, reason: str, prefer_dns: bool) -> bool: + """Try each configured fallback, selecting a writable tunneled member.""" + + global _mongo_client_real, _effective_uri_real, _using_fallback_real + global _active_fallback_kind_real, _last_auto_recovery_check_at + + for target in _connection_fallback_targets(prefer_dns=prefer_dns): logger.warning( - "[mongo_fallback] Preferring configured direct-host Mongo URI during DNS fallback recovery window." + "[mongo_fallback] Attempting %s MongoDB fallback due to %s.", + target.label, + reason, ) - _mongo_client_real = _try_connect_uri( - MONGO_DNS_FALLBACK_URI, - fallback_label="dns fallback", - server_selection_timeout_ms=3000, - ) - _effective_uri_real = MONGO_DNS_FALLBACK_URI + try: + client = _try_connect_uri( + target.uri, + fallback_label=f"{target.kind} fallback", + server_selection_timeout_ms=3000, + require_writable_primary=target.require_writable_primary, + expected_replica_set=target.expected_replica_set, + ) + except Exception as exc: + logger.warning( + "[mongo_fallback] %s connection failed: %s", + target.label, + exc, + ) + continue + + _mongo_client_real = client + _effective_uri_real = target.uri _using_fallback_real = True - return True - except Exception as exc: + _active_fallback_kind_real = target.kind + _last_auto_recovery_check_at = time.monotonic() + _mark_fallback_preferred(target.kind, f"primary {reason}") logger.warning( - "[mongo_fallback] Preferred DNS fallback connection failed: %s", exc + "[mongo_fallback] Using %s Mongo URI instead of primary.", + target.label, ) - _mongo_client_real = None - _clear_dns_fallback_preference() - return False + if _debug_mongo_enabled(): + logger.debug( + "[MongoConnect] %s fallback host: %s", + target.kind, + _host_display_from_uri(target.uri), + ) + return True + + _mongo_client_real = None + _effective_uri_real = MONGO_URI + _using_fallback_real = False + _active_fallback_kind_real = None + return False def _should_run_auto_recovery_check() -> bool: @@ -669,26 +1040,73 @@ def _should_run_auto_recovery_check() -> bool: def _invalidate_real_client_for_recovery(reason: str) -> None: """Drop the active client reference so the next get_db() call rebuilds it.""" global _mongo_client_real, _effective_uri_real, _using_fallback_real + global _active_fallback_kind_real logger.warning("[mongo_recovery] Invalidating Mongo client: %s", reason) _mongo_client_real = None - if _dns_fallback_is_preferred() and MONGO_DNS_FALLBACK_URI: - _effective_uri_real = MONGO_DNS_FALLBACK_URI + if _fallback_is_preferred() and _active_fallback_kind_real: _using_fallback_real = True else: _effective_uri_real = MONGO_URI _using_fallback_real = False + _active_fallback_kind_real = None def _ensure_active_client_is_healthy() -> None: - global _mongo_client_real + global _mongo_client_real, _effective_uri_real, _using_fallback_real + global _active_fallback_kind_real if _mongo_client_real is None: return if not _should_run_auto_recovery_check(): return timeout_ms = _mongo_auto_recovery_ping_timeout_ms() + + if _using_fallback_real and not _fallback_is_preferred(): + fallback_kind = _active_fallback_kind_real + try: + primary_client = _try_connect_uri( + MONGO_URI, + server_selection_timeout_ms=3000, + ) + except Exception as exc: + if fallback_kind: + _mark_fallback_preferred( + fallback_kind, + f"primary recovery probe failed: {type(exc).__name__}", + ) + else: + _mongo_client_real = primary_client + _effective_uri_real = MONGO_URI + _using_fallback_real = False + _active_fallback_kind_real = None + logger.info( + "[mongo_recovery] Direct primary Mongo route recovered; leaving fallback." + ) + return + try: - _mongo_client_real.admin.command("ping", maxTimeMS=timeout_ms) + tunnel_member_reselection_required = False + if _active_fallback_kind_real == "ssh_tunnel": + hello = _mongo_client_real.admin.command("hello", maxTimeMS=timeout_ms) + if not bool(hello.get("isWritablePrimary", hello.get("ismaster", False))): + tunnel_member_reselection_required = True + raise ConnectionFailure( + "active SSH tunnel member is no longer writable" + ) + expected_replica_set = _configured_replica_set_name() + if expected_replica_set and hello.get("setName") != expected_replica_set: + raise ConnectionFailure( + "active SSH tunnel member has unexpected replica-set identity" + ) + else: + _mongo_client_real.admin.command("ping", maxTimeMS=timeout_ms) except Exception as exc: + if ( + not tunnel_member_reselection_required + and is_mongo_transport_error(exc) + ): + prefer_alternate_mongo_route( + f"periodic health check {type(exc).__name__}" + ) _invalidate_real_client_for_recovery(f"periodic ping failed: {exc}") @@ -699,6 +1117,7 @@ def get_db() -> Database | None: """ global _mongo_client_real, _mongo_client_mock global _using_fallback_real, _effective_uri_real, _last_auto_recovery_check_at + global _active_fallback_kind_real db_name = get_configured_database_name() assert_safe_database_name_for_pytest(db_name) @@ -725,6 +1144,7 @@ def get_db() -> Database | None: _mongo_client_real = _try_connect_uri(MONGO_URI) _effective_uri_real = MONGO_URI _using_fallback_real = False + _active_fallback_kind_real = None _last_auto_recovery_check_at = time.monotonic() if _debug_mongo_enabled(): try: @@ -756,6 +1176,10 @@ def get_db() -> Database | None: is_ssl_error = ( "SSL" in str(e) or "10054" in str(e) or "handshake" in str(e).lower() ) + is_dns_error = any( + marker in str(e).lower() + for marker in ("dns", "resolution", "name does not exist", "srv") + ) # Debug logging for fallback logic if _debug_mongo_enabled(): @@ -771,110 +1195,39 @@ def get_db() -> Database | None: MONGO_URI.startswith("mongodb+srv://"), ) - # Prefer a dedicated direct-host Atlas fallback when configured; - # only use localhost fallback when it is explicitly allowed. - fallback_target = None - if MONGO_URI.strip("\"'").startswith("mongodb+srv://") or is_ssl_error: - fallback_target = _resolve_connection_fallback_target(prefer_dns=True) - - if fallback_target is not None: - try: - fallback_uri, fallback_label = fallback_target - logger.warning( - "[mongo_fallback] Attempting %s MongoDB fallback due to connection failure.", - fallback_label, - ) - _mongo_client_real = _new_mongo_client( - fallback_uri, server_selection_timeout_ms=3000 - ) - _mongo_client_real.admin.command("ismaster") - if ( - fallback_label == "dns fallback" - and MONGO_DNS_FALLBACK_URI - and fallback_uri == MONGO_DNS_FALLBACK_URI - ): - _mark_dns_fallback_preferred("primary connection failure") - _effective_uri_real = fallback_uri - _using_fallback_real = True - _last_auto_recovery_check_at = time.monotonic() - try: - logger.warning( - "[mongo_fallback] Using %s Mongo URI instead of primary (connection failure).", - fallback_label, - ) - except Exception: - pass - if _debug_mongo_enabled(): - try: - logger.debug( - "[MongoConnect] %s host: %s", - fallback_label, - _host_display_from_uri(fallback_uri), - ) - except Exception: - pass - except Exception as fe: - logger.warning("%s connection failed: %s", fallback_label, fe) - _mongo_client_real = None - return None - else: + should_try_fallback = ( + bool(MONGO_SSH_TUNNEL_FALLBACK_ENDPOINTS) + or bool(MONGO_DNS_FALLBACK_URI) + or MONGO_ALLOW_LOCAL_FALLBACK + or MONGO_URI.strip("\"'").startswith("mongodb+srv://") + or is_ssl_error + ) + if not should_try_fallback or not _try_configured_fallbacks( + reason="connection failure", + prefer_dns=is_dns_error and not is_ssl_error, + ): return None except Exception as e: logger.warning( "An unexpected error occurred during MongoDB client initialisation: %s", e, ) - # Attempt direct-host DNS fallback first; only use localhost when - # it is explicitly allowed. - fallback_target = None - if ( + is_dns_error = ( "resolution" in str(e).lower() or "dns" in str(e).lower() - or MONGO_URI.startswith("mongodb+srv://") + or "name does not exist" in str(e).lower() + or "srv" in str(e).lower() + ) + should_try_fallback = ( + bool(MONGO_SSH_TUNNEL_FALLBACK_ENDPOINTS) + or bool(MONGO_DNS_FALLBACK_URI) + or MONGO_ALLOW_LOCAL_FALLBACK + or is_dns_error + ) + if not should_try_fallback or not _try_configured_fallbacks( + reason=("DNS/SRV error" if is_dns_error else "initialisation failure"), + prefer_dns=is_dns_error, ): - fallback_target = _resolve_connection_fallback_target(prefer_dns=True) - - if fallback_target is not None: - try: - fallback_uri, fallback_label = fallback_target - logger.warning( - "[mongo_fallback] Attempting %s MongoDB fallback due to DNS/SRV error.", - fallback_label, - ) - _mongo_client_real = _new_mongo_client( - fallback_uri, server_selection_timeout_ms=3000 - ) - _mongo_client_real.admin.command("ismaster") - if ( - fallback_label == "dns fallback" - and MONGO_DNS_FALLBACK_URI - and fallback_uri == MONGO_DNS_FALLBACK_URI - ): - _mark_dns_fallback_preferred("primary DNS/SRV error") - _effective_uri_real = fallback_uri - _using_fallback_real = True - _last_auto_recovery_check_at = time.monotonic() - try: - logger.warning( - "[mongo_fallback] Using %s Mongo URI instead of primary (DNS/SRV error).", - fallback_label, - ) - except Exception: - pass - if _debug_mongo_enabled(): - try: - logger.debug( - "[MongoConnect] %s host: %s", - fallback_label, - _host_display_from_uri(fallback_uri), - ) - except Exception: - pass - except Exception as fe: - logger.warning("Local fallback connection failed: %s", fe) - _mongo_client_real = None - return None - else: _mongo_client_real = None # Ensure client is None on failure return None @@ -888,10 +1241,15 @@ def get_db() -> Database | None: def print_connection_info(): """Log helpful connection info on startup.""" - global _effective_uri_real + global _effective_uri_real, _active_fallback_kind_real uri_to_use = _effective_uri_real if _effective_uri_real else MONGO_URI host = _host_display_from_uri(uri_to_use) - if "localhost" in host or "127.0.0.1" in host: + if _active_fallback_kind_real == "ssh_tunnel": + logger.info( + "[Von Database] Connecting to REMOTE MongoDB through SSH tunnel at %s", + host, + ) + elif "localhost" in host or "127.0.0.1" in host: logger.info("[Von Database] Connecting to LOCAL MongoDB at %s", host) else: logger.info("[Von Database] Connecting to REMOTE MongoDB at %s", host) @@ -903,7 +1261,7 @@ def get_effective_mongo_uri() -> str: def is_using_fallback_uri() -> bool: - """Return True if a local fallback URI is being used instead of the primary MONGO_URI.""" + """Return True when any fallback URI is used instead of primary MONGO_URI.""" return _using_fallback_real @@ -914,22 +1272,48 @@ def get_mongo_fallback_policy_state() -> dict[str, object]: 0.0, _dns_fallback_preferred_until_monotonic - time.monotonic() ) return { + "active_fallback_kind": _active_fallback_kind_real, + "preferred_fallback_kind": _preferred_fallback_kind, + "fallback_sticky": bool(remaining_seconds > 0), + "fallback_sticky_seconds_remaining": round(remaining_seconds, 3), + "fallback_sticky_seconds_configured": _mongo_dns_fallback_sticky_seconds(), + "fallback_preference_reason": _dns_fallback_preference_reason, + "ssh_tunnel_fallback_configured": bool(MONGO_SSH_TUNNEL_FALLBACK_ENDPOINTS), + "ssh_tunnel_endpoint_count": len(MONGO_SSH_TUNNEL_FALLBACK_ENDPOINTS), + "ssh_tunnel_expected_replica_set_configured": bool( + _configured_replica_set_name() + ), "dns_fallback_configured": bool(MONGO_DNS_FALLBACK_URI), - "dns_fallback_sticky": bool(remaining_seconds > 0), - "dns_fallback_sticky_seconds_remaining": round(remaining_seconds, 3), + "dns_fallback_sticky": bool( + remaining_seconds > 0 and _preferred_fallback_kind in {None, "dns"} + ), + "dns_fallback_sticky_seconds_remaining": ( + round(remaining_seconds, 3) + if _preferred_fallback_kind in {None, "dns"} + else 0.0 + ), "dns_fallback_sticky_seconds_configured": _mongo_dns_fallback_sticky_seconds(), - "dns_fallback_preference_reason": _dns_fallback_preference_reason, + "dns_fallback_preference_reason": ( + _dns_fallback_preference_reason + if _preferred_fallback_kind in {None, "dns"} + else None + ), + "local_fallback_allowed": MONGO_ALLOW_LOCAL_FALLBACK, } def close_connection(): """Closes the MongoDB connection.""" global _mongo_client_real, _mongo_client_mock, _last_auto_recovery_check_at + global _using_fallback_real, _effective_uri_real, _active_fallback_kind_real if _mongo_client_real: _mongo_client_real.close() _mongo_client_real = None # mongomock doesn't require close(), but clear ref for correctness. _mongo_client_mock = None + _using_fallback_real = False + _effective_uri_real = MONGO_URI + _active_fallback_kind_real = None _last_auto_recovery_check_at = 0.0 _clear_dns_fallback_preference() with _COLLECTION_INDEXES_LOCK: @@ -947,8 +1331,12 @@ def invalidate_connection(): simply drop our reference and let GC clean up. """ global _mongo_client_real, _mongo_client_mock, _last_auto_recovery_check_at + global _using_fallback_real, _effective_uri_real, _active_fallback_kind_real _mongo_client_real = None _mongo_client_mock = None + _using_fallback_real = False + _effective_uri_real = MONGO_URI + _active_fallback_kind_real = None _last_auto_recovery_check_at = 0.0 with _COLLECTION_INDEXES_LOCK: _COLLECTION_INDEXES_READY.clear() diff --git a/src/backend/db/mongo_error_classification.py b/src/backend/db/mongo_error_classification.py new file mode 100644 index 000000000..cbf3f4fdb --- /dev/null +++ b/src/backend/db/mongo_error_classification.py @@ -0,0 +1,61 @@ +"""Shared Mongo error classification for retry and route-recovery decisions.""" + +from __future__ import annotations + +from pymongo.errors import ( + AutoReconnect, + ConnectionFailure, + NetworkTimeout, + NotPrimaryError, + PyMongoError, + ServerSelectionTimeoutError, +) + +_TRANSIENT_MONGO_ERROR_TYPES = ( + NetworkTimeout, + ServerSelectionTimeoutError, + AutoReconnect, + ConnectionFailure, + PyMongoError, +) +_MONGO_TRANSPORT_ERROR_TYPES = ( + NetworkTimeout, + ServerSelectionTimeoutError, + AutoReconnect, + ConnectionFailure, +) +_TRANSIENT_MONGO_ERROR_MARKERS = ( + "timed out", + "no primary", + "replicasetnoprimary", + "connection pool paused", + "server selection timeout", + "networktimeout", + "temporarily unavailable", +) + + +def is_transient_mongo_error(exc: Exception | str | None) -> bool: + """Return True when ``exc`` looks like a retryable Mongo failure.""" + + if exc is None: + return False + if isinstance(exc, _TRANSIENT_MONGO_ERROR_TYPES): + return True + message = str(exc).lower() + return any(marker in message for marker in _TRANSIENT_MONGO_ERROR_MARKERS) + + +def is_mongo_transport_error(exc: Exception | str | None) -> bool: + """Return True when trying another configured Mongo route is justified.""" + + if exc is None: + return False + # A normal replica election needs a topology refresh/reconnect on the + # current route, not abandonment of an otherwise healthy transport. + if isinstance(exc, NotPrimaryError): + return False + if isinstance(exc, _MONGO_TRANSPORT_ERROR_TYPES): + return True + message = str(exc).lower() + return any(marker in message for marker in _TRANSIENT_MONGO_ERROR_MARKERS) diff --git a/src/backend/db/mongo_uri_redaction.py b/src/backend/db/mongo_uri_redaction.py index 8c009975d..bcc1cc60d 100644 --- a/src/backend/db/mongo_uri_redaction.py +++ b/src/backend/db/mongo_uri_redaction.py @@ -110,6 +110,8 @@ def build_safe_mongo_connection_location( uri: object, *, using_fallback: bool | None = None, + fallback_kind: str | None = None, + fallback_target_uri: object = None, ) -> dict[str, Any]: text = _coerce_uri(uri) scheme = None @@ -123,9 +125,29 @@ def build_safe_mongo_connection_location( if sanitized_uri is None: hosts = [] - classification = classify_mongo_connection_location( + endpoint_classification = classify_mongo_connection_location( sanitized_uri, scheme=scheme, hosts=hosts ) + upstream_sanitized_uri = None + upstream_classification = None + if fallback_kind == "ssh_tunnel": + upstream_sanitized_uri = sanitize_mongo_uri_for_display(fallback_target_uri) + upstream_classification = classify_mongo_connection_location( + upstream_sanitized_uri + ) + if fallback_kind == "ssh_tunnel": + classification = ( + upstream_classification + if upstream_classification not in {None, "unknown"} + else "remote" + ) + else: + classification = endpoint_classification + logical_sanitized_uri = ( + upstream_sanitized_uri + if fallback_kind == "ssh_tunnel" and upstream_sanitized_uri + else sanitized_uri + ) return { "available": sanitized_uri is not None, "scheme": scheme, @@ -133,8 +155,14 @@ def build_safe_mongo_connection_location( "hosts": hosts, "host_count": len(hosts), "sanitized_uri": sanitized_uri, + "logical_sanitized_uri": logical_sanitized_uri, "classification": classification, + "endpoint_classification": endpoint_classification, + "upstream_classification": upstream_classification, + "upstream_sanitized_uri": upstream_sanitized_uri, + "transport": "ssh_tunnel" if fallback_kind == "ssh_tunnel" else "direct", "is_atlas": classification == "atlas", "is_local": classification == "local", "using_fallback": using_fallback, + "fallback_kind": fallback_kind, } diff --git a/src/backend/db/transient_errors.py b/src/backend/db/transient_errors.py index c59571609..bb63cc9d2 100644 --- a/src/backend/db/transient_errors.py +++ b/src/backend/db/transient_errors.py @@ -8,35 +8,14 @@ from collections.abc import Callable from typing import TypeVar -from pymongo.errors import ( - AutoReconnect, - ConnectionFailure, - NetworkTimeout, - PyMongoError, - ServerSelectionTimeoutError, -) - from .connection_manager import attempt_reconnect +from .mongo_error_classification import ( + is_mongo_transport_error, + is_transient_mongo_error, +) T = TypeVar("T") -_TRANSIENT_MONGO_ERROR_TYPES = ( - NetworkTimeout, - ServerSelectionTimeoutError, - AutoReconnect, - ConnectionFailure, - PyMongoError, -) -_TRANSIENT_MONGO_ERROR_MARKERS = ( - "timed out", - "no primary", - "replicasetnoprimary", - "connection pool paused", - "server selection timeout", - "networktimeout", - "temporarily unavailable", -) - def _positive_int_env(name: str, default: int) -> int: raw = os.getenv(name) @@ -60,17 +39,6 @@ def _positive_float_env(name: str, default: float) -> float: return value if value > 0 else default -def is_transient_mongo_error(exc: Exception | str | None) -> bool: - """Return True when ``exc`` looks like a retryable Mongo transport failure.""" - - if exc is None: - return False - if isinstance(exc, _TRANSIENT_MONGO_ERROR_TYPES): - return True - message = str(exc).lower() - return any(marker in message for marker in _TRANSIENT_MONGO_ERROR_MARKERS) - - def run_with_transient_mongo_retry( operation: Callable[[], T], *, @@ -131,7 +99,12 @@ def run_with_transient_mongo_retry( delay_seconds, ) try: - attempt_reconnect(force=True, max_attempts=1) + attempt_reconnect( + force=True, + max_attempts=1, + route_degraded=is_mongo_transport_error(exc), + degradation_reason=f"transient operation {type(exc).__name__}", + ) except Exception: if logger_obj is not None: logger_obj.debug( diff --git a/src/backend/integrations/internal_mcp/catalogue.py b/src/backend/integrations/internal_mcp/catalogue.py index 398c607e5..9f9b2e56c 100644 --- a/src/backend/integrations/internal_mcp/catalogue.py +++ b/src/backend/integrations/internal_mcp/catalogue.py @@ -22613,21 +22613,29 @@ def _mongo_cost_guardrails_report(**kwargs): if denial := _internal_mcp_operator_control_plane_denial("Mongo diagnostics"): return denial - from ...db.mongo_client import get_effective_mongo_uri, is_using_fallback_uri - from ...db.mongo_uri_redaction import ( - classify_mongo_connection_location, - sanitize_mongo_uri_for_display, + from ...db.mongo_client import ( + MONGO_URI, + get_effective_mongo_uri, + get_mongo_fallback_policy_state, + is_using_fallback_uri, ) + from ...db.mongo_uri_redaction import build_safe_mongo_connection_location from ...services.mongo_observability_service import ( build_mongo_cost_guardrail_report, ) try: - sanitized_uri = sanitize_mongo_uri_for_display(get_effective_mongo_uri()) - return build_mongo_cost_guardrail_report( - mongo_classification=classify_mongo_connection_location(sanitized_uri), - sanitized_uri=sanitized_uri, + fallback_policy = get_mongo_fallback_policy_state() + connection_location = build_safe_mongo_connection_location( + get_effective_mongo_uri(), using_fallback=is_using_fallback_uri(), + fallback_kind=fallback_policy.get("active_fallback_kind"), + fallback_target_uri=MONGO_URI, + ) + return build_mongo_cost_guardrail_report( + mongo_classification=connection_location["classification"], + sanitized_uri=connection_location["logical_sanitized_uri"], + using_fallback=connection_location["using_fallback"], local_development=kwargs.get("local_development"), reset=bool(kwargs.get("reset")), ) diff --git a/src/backend/server/routes/settings_routes.py b/src/backend/server/routes/settings_routes.py index 86a9bcbb4..0b4bda82e 100644 --- a/src/backend/server/routes/settings_routes.py +++ b/src/backend/server/routes/settings_routes.py @@ -106,10 +106,12 @@ DATABASE_NAME, get_db, get_effective_mongo_uri, + get_mongo_fallback_policy_state, is_using_fallback_uri, assert_destructive_db_operation_allowed, ) from ...db.mongo_uri_redaction import ( + build_safe_mongo_connection_location, classify_mongo_connection_location, sanitize_mongo_uri_for_display, ) @@ -674,10 +676,22 @@ def get_db_location_info(): error_message = str(e) ping_ok = False - classification = _classify_mongo_sanitized_uri(sanitized_uri) + # get_db()/ping may have switched between the primary and a fallback. + # Snapshot route identity only after that recovery opportunity. + effective_uri = get_effective_mongo_uri() + sanitized_uri = _sanitize_mongo_uri_for_display(effective_uri) using_fallback = is_using_fallback_uri() + fallback_policy = get_mongo_fallback_policy_state() + fallback_kind = fallback_policy.get("active_fallback_kind") + connection_location = build_safe_mongo_connection_location( + effective_uri, + using_fallback=using_fallback, + fallback_kind=fallback_kind, + fallback_target_uri=MONGO_URI, + ) + classification = connection_location["classification"] public_ip = None - if using_fallback: + if using_fallback and fallback_kind != "ssh_tunnel": # Try to discover outward-facing IP (best-effort, short timeout). Avoid blocking failures. try: import urllib.request @@ -698,7 +712,9 @@ def get_db_location_info(): "ping_ok": ping_ok, "error": error_message, "classification": classification, + "connection_location": connection_location, "using_fallback": using_fallback, + "fallback_kind": fallback_kind, "primary_uri_sanitized": ( _sanitize_mongo_uri_for_display(MONGO_URI) if using_fallback else None ), @@ -724,11 +740,18 @@ def get_db_guardrails(): try: effective_uri = get_effective_mongo_uri() sanitized_uri = _sanitize_mongo_uri_for_display(effective_uri) - classification = _classify_mongo_sanitized_uri(sanitized_uri) + fallback_policy = get_mongo_fallback_policy_state() + connection_location = build_safe_mongo_connection_location( + effective_uri, + using_fallback=is_using_fallback_uri(), + fallback_kind=fallback_policy.get("active_fallback_kind"), + fallback_target_uri=MONGO_URI, + ) + classification = connection_location["classification"] reset = request.args.get("reset") in {"1", "true", "yes", "on"} report = build_mongo_cost_guardrail_report( mongo_classification=classification, - sanitized_uri=sanitized_uri, + sanitized_uri=connection_location["logical_sanitized_uri"], using_fallback=is_using_fallback_uri(), reset=reset, ) diff --git a/src/backend/server/utils_flask.py b/src/backend/server/utils_flask.py index acad73901..930eed152 100644 --- a/src/backend/server/utils_flask.py +++ b/src/backend/server/utils_flask.py @@ -2822,15 +2822,20 @@ def _build_health_runtime_authority_projection(app: Flask) -> dict[str, object]: try: from ..db.mongo_client import ( + MONGO_URI, get_configured_database_name, get_effective_mongo_uri, + get_mongo_fallback_policy_state, is_using_fallback_uri, ) from ..db.mongo_uri_redaction import build_safe_mongo_connection_location + fallback_policy = get_mongo_fallback_policy_state() location = build_safe_mongo_connection_location( get_effective_mongo_uri(), using_fallback=is_using_fallback_uri(), + fallback_kind=fallback_policy.get("active_fallback_kind"), + fallback_target_uri=MONGO_URI, ) encoded_location = json.dumps( location, @@ -3652,8 +3657,10 @@ def _build_diagnostics_response(app: Flask): import hashlib from ..db.mongo_client import ( # type: ignore + MONGO_URI, get_configured_database_name, get_effective_mongo_uri, + get_mongo_fallback_policy_state, is_using_fallback_uri, ) from ..db.mongo_uri_redaction import ( @@ -3663,9 +3670,12 @@ def _build_diagnostics_response(app: Flask): effective_uri = get_effective_mongo_uri() using_fallback = is_using_fallback_uri() + fallback_policy = get_mongo_fallback_policy_state() connection_location = build_safe_mongo_connection_location( effective_uri, using_fallback=using_fallback, + fallback_kind=fallback_policy.get("active_fallback_kind"), + fallback_target_uri=MONGO_URI, ) mongo_diag = { "effective_mongo_uri": sanitize_mongo_uri_for_display(effective_uri), @@ -3902,15 +3912,25 @@ def _build_db_status_response(): using_fallback = None atlas_detected = None host_only = None + fallback_kind = None try: - from ..db.mongo_client import get_effective_mongo_uri, is_using_fallback_uri # type: ignore + from ..db.mongo_client import ( # type: ignore + MONGO_URI, + get_effective_mongo_uri, + get_mongo_fallback_policy_state, + is_using_fallback_uri, + ) from ..db.mongo_uri_redaction import build_safe_mongo_connection_location effective_uri = get_effective_mongo_uri() using_fallback = is_using_fallback_uri() + fallback_policy = get_mongo_fallback_policy_state() + fallback_kind = fallback_policy.get("active_fallback_kind") connection_location = build_safe_mongo_connection_location( effective_uri, using_fallback=using_fallback, + fallback_kind=(fallback_kind if isinstance(fallback_kind, str) else None), + fallback_target_uri=MONGO_URI, ) host_only = connection_location.get("host") atlas_detected = connection_location.get("is_atlas") @@ -3918,6 +3938,7 @@ def _build_db_status_response(): pass return jsonify( using_fallback=using_fallback, + fallback_kind=fallback_kind, atlas_detected=atlas_detected, effective_host=host_only, timestamp=datetime.now(_tz.utc).isoformat().replace("+00:00", "Z"), diff --git a/src/backend/services/model_registry_service.py b/src/backend/services/model_registry_service.py index 6b5862db9..939709be9 100644 --- a/src/backend/services/model_registry_service.py +++ b/src/backend/services/model_registry_service.py @@ -178,18 +178,23 @@ def _model_registry_authority_fingerprint() -> str: namespace = str(os.getenv("VON_DEFAULT_NAMESPACE") or "").strip() try: from ..db.mongo_client import ( + MONGO_URI, get_configured_database_name, get_effective_mongo_uri, + get_mongo_fallback_policy_state, is_using_fallback_uri, ) from ..db.mongo_uri_redaction import build_safe_mongo_connection_location effective_mongo_uri = get_effective_mongo_uri() + fallback_policy = get_mongo_fallback_policy_state() authority = { "database_name": get_configured_database_name(), "mongo_location": build_safe_mongo_connection_location( effective_mongo_uri, using_fallback=is_using_fallback_uri(), + fallback_kind=fallback_policy.get("active_fallback_kind"), + fallback_target_uri=MONGO_URI, ), "mongo_principal_fingerprint": _mongo_principal_fingerprint( effective_mongo_uri diff --git a/src/backend/services/mongo_startup_config.py b/src/backend/services/mongo_startup_config.py index 769f67eb1..bb05f8cd1 100644 --- a/src/backend/services/mongo_startup_config.py +++ b/src/backend/services/mongo_startup_config.py @@ -133,7 +133,7 @@ def collect_mongo_startup_errors() -> list[str]: + ", ".join(invalid_hosts) ) - if get_env_bool("MONGO_ALLOW_LOCAL_FALLBACK", True): + if get_env_bool("MONGO_ALLOW_LOCAL_FALLBACK", False): errors.append( "MONGO_ALLOW_LOCAL_FALLBACK must be disabled when VON_MONGO_STRICT_STARTUP is enabled." ) diff --git a/tests/backend/test_mongo_dns_fallback_recovery.py b/tests/backend/test_mongo_dns_fallback_recovery.py index 0103d156e..21568dacd 100644 --- a/tests/backend/test_mongo_dns_fallback_recovery.py +++ b/tests/backend/test_mongo_dns_fallback_recovery.py @@ -2,9 +2,17 @@ import time +import pytest + from src.backend.db import mongo_client as mc +@pytest.fixture(autouse=True) +def _disable_machine_specific_tunnel_fallback(monkeypatch) -> None: + monkeypatch.setattr(mc, "MONGO_SSH_TUNNEL_FALLBACK_ENDPOINTS", ()) + monkeypatch.setattr(mc, "MONGO_SSH_TUNNEL_EXPECTED_REPLICA_SET", None) + + class _FakeAdmin: def __init__(self) -> None: self.commands: list[tuple[str, dict]] = [] diff --git a/tests/backend/test_mongo_observability_service.py b/tests/backend/test_mongo_observability_service.py index aed36d342..e36881285 100644 --- a/tests/backend/test_mongo_observability_service.py +++ b/tests/backend/test_mongo_observability_service.py @@ -673,6 +673,47 @@ def test_internal_mcp_gateway_exposes_mongo_cost_guardrails_report(monkeypatch): assert payload["runtime_posture"]["sanitized_uri"] == "mongodb+srv://example.invalid" +def test_internal_mcp_guardrails_classify_atlas_through_ssh_tunnel(monkeypatch): + from src.backend.db import mongo_client + + monkeypatch.setenv("VON_MONGO_OPERATION_AUDIT_ENABLED", "1") + monkeypatch.setattr( + mongo_client, + "MONGO_URI", + "mongodb+srv://user:secret@cluster.mongodb.net/von_db", + ) + monkeypatch.setattr( + mongo_client, + "get_effective_mongo_uri", + lambda: "mongodb://user:secret@127.0.0.1:27019/?directConnection=true", + ) + monkeypatch.setattr(mongo_client, "is_using_fallback_uri", lambda: True) + monkeypatch.setattr( + mongo_client, + "get_mongo_fallback_policy_state", + lambda: {"active_fallback_kind": "ssh_tunnel"}, + ) + gateway = InternalMCPGateway( + catalogue=build_default_catalogue(), + transport=InternalMCPTransport(), + enabled=True, + trusted_actor_payload_fallback=True, + ) + + payload = gateway.invoke( + "mongo_cost_guardrails_report", + {"local_development": True}, + ).payload + + assert payload["runtime_posture"]["remote_mongo"] is True + assert payload["runtime_posture"]["mongo_classification"] == "atlas" + assert ( + payload["runtime_posture"]["sanitized_uri"] + == "mongodb+srv://cluster.mongodb.net" + ) + assert "user:secret" not in str(payload) + + def test_settings_db_guardrails_route_returns_redacted_report(monkeypatch): from src.backend.server.routes import settings_routes diff --git a/tests/backend/test_mongo_recovery_behaviour.py b/tests/backend/test_mongo_recovery_behaviour.py index 1e7ae27cb..aeb5a1d35 100644 --- a/tests/backend/test_mongo_recovery_behaviour.py +++ b/tests/backend/test_mongo_recovery_behaviour.py @@ -3,10 +3,16 @@ import time import pytest -from pymongo.errors import ConnectionFailure +from pymongo.errors import ( + ConnectionFailure, + NetworkTimeout, + NotPrimaryError, + PyMongoError, +) from src.backend.db import connection_manager as conn_mgr from src.backend.db import mongo_client as mc +from src.backend.db import transient_errors @pytest.fixture(autouse=True) @@ -18,21 +24,33 @@ def _restore_connection_manager_state(): class _FakeAdmin: - def __init__(self, fail_ping: bool = False) -> None: + def __init__( + self, + fail_ping: bool = False, + hello: dict[str, object] | None = None, + ) -> None: self.fail_ping = fail_ping + self.hello = hello or {"ok": 1} self.commands: list[tuple[str, dict]] = [] def command(self, name: str, **kwargs): self.commands.append((name, dict(kwargs))) if name == "ping" and self.fail_ping: raise ConnectionFailure("simulated ping failure") + if name == "hello": + return dict(self.hello) return {"ok": 1} class _FakeClient: - def __init__(self, label: str, fail_ping: bool = False) -> None: + def __init__( + self, + label: str, + fail_ping: bool = False, + hello: dict[str, object] | None = None, + ) -> None: self.label = label - self.admin = _FakeAdmin(fail_ping=fail_ping) + self.admin = _FakeAdmin(fail_ping=fail_ping, hello=hello) def __getitem__(self, db_name: str): return {"label": self.label, "db_name": db_name} @@ -53,6 +71,8 @@ def _fake_mongo_client(uri: str, **kwargs): monkeypatch.setattr(mc, "is_mock_db_enabled", lambda: False) monkeypatch.setattr(mc, "assert_safe_database_name_for_pytest", lambda _: None) monkeypatch.setattr(mc, "MONGO_ALLOW_LOCAL_FALLBACK", False) + monkeypatch.setattr(mc, "MONGO_SSH_TUNNEL_FALLBACK_ENDPOINTS", ()) + monkeypatch.setattr(mc, "MONGO_DNS_FALLBACK_URI", None) monkeypatch.setattr(mc, "MONGO_URI", "mongodb://cluster0.mongodb.net/von_db?tls=true") monkeypatch.setattr(mc, "_mongo_client_real", existing) monkeypatch.setattr(mc, "_mongo_client_mock", None) @@ -70,6 +90,52 @@ def _fake_mongo_client(uri: str, **kwargs): assert any(name == "ping" for name, _ in existing.admin.commands) +def test_direct_ping_failure_selects_ssh_even_when_direct_hello_would_succeed( + monkeypatch: pytest.MonkeyPatch, +) -> None: + existing = _FakeClient(label="direct", fail_ping=True) + tunneled = _FakeClient( + label="tunneled", + hello={"ok": 1, "setName": "atlas-set", "isWritablePrimary": True}, + ) + primary_uri = "mongodb+srv://user:pass@cluster.example/" + created: list[str] = [] + + def _fake_new_client(uri: str, **_kwargs): + created.append(uri) + if uri == primary_uri: + raise AssertionError("degraded direct route must not be retried first") + return tunneled + + monkeypatch.setattr(mc, "_new_mongo_client", _fake_new_client) + monkeypatch.setattr(mc, "is_mock_db_enabled", lambda: False) + monkeypatch.setattr(mc, "assert_safe_database_name_for_pytest", lambda _: None) + monkeypatch.setattr(mc, "MONGO_URI", primary_uri) + monkeypatch.setattr(mc, "MONGO_DNS_FALLBACK_URI", None) + monkeypatch.setattr( + mc, "MONGO_SSH_TUNNEL_FALLBACK_ENDPOINTS", ("127.0.0.1:27018",) + ) + monkeypatch.setattr(mc, "MONGO_SSH_TUNNEL_EXPECTED_REPLICA_SET", "atlas-set") + monkeypatch.setattr(mc, "MONGO_ALLOW_LOCAL_FALLBACK", False) + monkeypatch.setattr(mc, "_mongo_client_real", existing) + monkeypatch.setattr(mc, "_mongo_client_mock", None) + monkeypatch.setattr(mc, "_using_fallback_real", False) + monkeypatch.setattr(mc, "_active_fallback_kind_real", None) + monkeypatch.setattr(mc, "_preferred_fallback_kind", None) + monkeypatch.setattr(mc, "_dns_fallback_preferred_until_monotonic", 0.0) + monkeypatch.setattr(mc, "_last_auto_recovery_check_at", 0.0) + monkeypatch.setenv("VON_DB_NAME", "test_von_db") + monkeypatch.setenv("VON_MONGO_AUTO_RECOVERY", "1") + monkeypatch.setenv("VON_MONGO_AUTO_RECOVERY_CHECK_INTERVAL_SECONDS", "0.001") + + db = mc.get_db() + + assert db is not None + assert db["label"] == "tunneled" + assert created and "127.0.0.1:27018" in created[0] + assert mc.get_mongo_fallback_policy_state()["active_fallback_kind"] == "ssh_tunnel" + + def test_get_db_skips_health_ping_when_interval_not_elapsed( monkeypatch: pytest.MonkeyPatch, ) -> None: @@ -95,6 +161,176 @@ def _unexpected_constructor(*args, **kwargs): assert all(name != "ping" for name, _ in healthy.admin.commands) +def test_direct_route_degradation_prefers_ssh_tunnel(monkeypatch) -> None: + monkeypatch.setattr(mc, "_using_fallback_real", False) + monkeypatch.setattr(mc, "_active_fallback_kind_real", None) + monkeypatch.setattr( + mc, "MONGO_URI", "mongodb+srv://user:pass@cluster.example/" + ) + monkeypatch.setattr( + mc, "MONGO_SSH_TUNNEL_FALLBACK_ENDPOINTS", ("127.0.0.1:27018",) + ) + monkeypatch.setattr( + mc, + "MONGO_DNS_FALLBACK_URI", + "mongodb://direct.example:27017/?tls=true", + ) + monkeypatch.setattr(mc, "MONGO_ALLOW_LOCAL_FALLBACK", False) + + selected = mc.prefer_alternate_mongo_route("test transport degradation") + + assert selected == "ssh_tunnel" + assert mc.get_mongo_fallback_policy_state()["preferred_fallback_kind"] == ( + "ssh_tunnel" + ) + + +def test_dns_route_degradation_rotates_to_ssh_tunnel(monkeypatch) -> None: + monkeypatch.setattr(mc, "_using_fallback_real", True) + monkeypatch.setattr(mc, "_active_fallback_kind_real", "dns") + monkeypatch.setattr( + mc, "MONGO_URI", "mongodb+srv://user:pass@cluster.example/" + ) + monkeypatch.setattr( + mc, "MONGO_SSH_TUNNEL_FALLBACK_ENDPOINTS", ("127.0.0.1:27018",) + ) + monkeypatch.setattr( + mc, + "MONGO_DNS_FALLBACK_URI", + "mongodb://direct.example:27017/?tls=true", + ) + monkeypatch.setattr(mc, "MONGO_ALLOW_LOCAL_FALLBACK", False) + + assert ( + mc.prefer_alternate_mongo_route("test DNS route degradation") + == "ssh_tunnel" + ) + + +def test_transient_transport_retry_marks_route_degraded(monkeypatch) -> None: + calls: list[dict] = [] + attempts = 0 + + def _operation() -> str: + nonlocal attempts + attempts += 1 + if attempts == 1: + raise NetworkTimeout("simulated operation timeout") + return "ok" + + monkeypatch.setattr( + transient_errors, + "attempt_reconnect", + lambda **kwargs: calls.append(dict(kwargs)), + ) + monkeypatch.setattr(transient_errors.time, "sleep", lambda _delay: None) + + result = transient_errors.run_with_transient_mongo_retry( + _operation, + operation_name="test-operation", + max_attempts=2, + ) + + assert result == "ok" + assert calls[0]["route_degraded"] is True + assert calls[0]["degradation_reason"] == "transient operation NetworkTimeout" + + +def test_non_transport_pymongo_retry_does_not_rotate_route(monkeypatch) -> None: + calls: list[dict] = [] + attempts = 0 + + def _operation() -> str: + nonlocal attempts + attempts += 1 + if attempts == 1: + raise PyMongoError("semantic command failure") + return "ok" + + monkeypatch.setattr( + transient_errors, + "attempt_reconnect", + lambda **kwargs: calls.append(dict(kwargs)), + ) + monkeypatch.setattr(transient_errors.time, "sleep", lambda _delay: None) + + assert ( + transient_errors.run_with_transient_mongo_retry( + _operation, + operation_name="test-operation", + max_attempts=2, + ) + == "ok" + ) + assert calls[0]["route_degraded"] is False + + +def test_replica_election_retry_does_not_rotate_route(monkeypatch) -> None: + calls: list[dict] = [] + attempts = 0 + + def _operation() -> str: + nonlocal attempts + attempts += 1 + if attempts == 1: + raise NotPrimaryError("simulated replica election") + return "ok" + + monkeypatch.setattr( + transient_errors, + "attempt_reconnect", + lambda **kwargs: calls.append(dict(kwargs)), + ) + monkeypatch.setattr(transient_errors.time, "sleep", lambda _delay: None) + + assert ( + transient_errors.run_with_transient_mongo_retry( + _operation, + operation_name="test-operation", + max_attempts=2, + ) + == "ok" + ) + assert calls[0]["route_degraded"] is False + + +def test_route_degradation_without_alternate_keeps_primary(monkeypatch) -> None: + monkeypatch.setattr(mc, "_using_fallback_real", False) + monkeypatch.setattr(mc, "_active_fallback_kind_real", None) + monkeypatch.setattr(mc, "MONGO_SSH_TUNNEL_FALLBACK_ENDPOINTS", ()) + monkeypatch.setattr(mc, "MONGO_DNS_FALLBACK_URI", None) + monkeypatch.setattr(mc, "MONGO_ALLOW_LOCAL_FALLBACK", False) + + assert mc.prefer_alternate_mongo_route("test") is None + + +@pytest.mark.parametrize("active_kind", ("dns", "ssh_tunnel")) +def test_route_degradation_without_alternate_reprobes_primary( + monkeypatch, + active_kind: str, +) -> None: + monkeypatch.setattr(mc, "_using_fallback_real", True) + monkeypatch.setattr(mc, "_active_fallback_kind_real", active_kind) + monkeypatch.setattr( + mc, + "MONGO_SSH_TUNNEL_FALLBACK_ENDPOINTS", + ("127.0.0.1:27018",) if active_kind == "ssh_tunnel" else (), + ) + monkeypatch.setattr( + mc, + "MONGO_DNS_FALLBACK_URI", + "mongodb://direct.example:27017/" if active_kind == "dns" else None, + ) + monkeypatch.setattr(mc, "MONGO_ALLOW_LOCAL_FALLBACK", False) + monkeypatch.setattr(mc, "_preferred_fallback_kind", active_kind) + monkeypatch.setattr( + mc, "_dns_fallback_preferred_until_monotonic", time.monotonic() + 60 + ) + + assert mc.prefer_alternate_mongo_route("test") is None + assert mc.get_mongo_fallback_policy_state()["preferred_fallback_kind"] is None + + def test_monitor_loop_survives_health_summary_exceptions( monkeypatch: pytest.MonkeyPatch, ) -> None: @@ -106,7 +342,11 @@ def _fake_health_summary(): if health_calls["count"] == 1: raise RuntimeError("simulated monitor error") if health_calls["count"] == 2: - return {"connected": False} + return { + "connected": False, + "route_degraded": True, + "health_error_type": "NetworkTimeout", + } return {"connected": True} def _fake_attempt_reconnect(**kwargs): @@ -128,4 +368,8 @@ def _fake_sleep(_seconds: float) -> None: conn_mgr._monitor_loop(interval_sec=1, base_ms=100, max_ms=1000, max_attempts=2) assert reconnect_calls, "Expected reconnect attempt after transient monitor error." + assert reconnect_calls[0]["route_degraded"] is True + assert reconnect_calls[0]["degradation_reason"] == ( + "monitor ping NetworkTimeout" + ) assert "monitor_loop_error" in str(conn_mgr._state.get("last_error")) diff --git a/tests/backend/test_mongo_ssh_tunnel_fallback.py b/tests/backend/test_mongo_ssh_tunnel_fallback.py new file mode 100644 index 000000000..b8651ad1e --- /dev/null +++ b/tests/backend/test_mongo_ssh_tunnel_fallback.py @@ -0,0 +1,343 @@ +from __future__ import annotations + +import logging +import time +from urllib.parse import parse_qsl, urlsplit + +from pymongo.errors import ConnectionFailure + +from src.backend.db import mongo_client as mc + + +class _FakeAdmin: + def __init__(self, hello: dict[str, object]) -> None: + self.hello = hello + self.commands: list[tuple[str, dict[str, object]]] = [] + + def command(self, name: str, **kwargs): + self.commands.append((name, dict(kwargs))) + if name == "hello": + return dict(self.hello) + return {"ok": 1} + + +class _FakeClient: + def __init__(self, label: str, hello: dict[str, object]) -> None: + self.label = label + self.admin = _FakeAdmin(hello) + self.closed = False + + def __getitem__(self, db_name: str): + return {"label": self.label, "db_name": db_name} + + def close(self) -> None: + self.closed = True + + +def _set_real_db_test_state(monkeypatch) -> None: + monkeypatch.setattr(mc, "is_mock_db_enabled", lambda: False) + monkeypatch.setattr(mc, "assert_safe_database_name_for_pytest", lambda _: None) + monkeypatch.setattr(mc, "MONGO_ALLOW_LOCAL_FALLBACK", False) + monkeypatch.setattr(mc, "_mongo_client_real", None) + monkeypatch.setattr(mc, "_mongo_client_mock", None) + monkeypatch.setattr(mc, "_using_fallback_real", False) + monkeypatch.setattr(mc, "_active_fallback_kind_real", None) + monkeypatch.setattr(mc, "_preferred_fallback_kind", None) + monkeypatch.setattr(mc, "_dns_fallback_preferred_until_monotonic", 0.0) + monkeypatch.setattr(mc, "_dns_fallback_preference_reason", None) + monkeypatch.setattr(mc, "_last_auto_recovery_check_at", 0.0) + monkeypatch.setenv("VON_DB_NAME", "test_von_db") + + +def test_tunnel_uri_preserves_encoded_userinfo_and_controls_tls() -> None: + primary = ( + "mongodb+srv://user%40example:p%2Fass%3Fword@cluster.example/" + "?retryWrites=true&w=majority&replicaSet=old&ssl=false" + "&tlsInsecure=true&tlsAllowInvalidCertificates=true" + "&readPreference=secondary&readPreferenceTags=region%3Anz" + "&maxStalenessSeconds=120" + ) + + derived = mc._build_ssh_tunnel_mongo_uri(primary, "127.0.0.1:27018") + + assert derived.startswith( + "mongodb://user%40example:p%2Fass%3Fword@127.0.0.1:27018/" + ) + options = dict(parse_qsl(urlsplit(derived).query, keep_blank_values=True)) + assert options["retryWrites"] == "true" + assert options["w"] == "majority" + assert options["authSource"] == "admin" + assert options["directConnection"] == "true" + assert options["tls"] == "true" + assert options["tlsAllowInvalidCertificates"] == "false" + assert options["tlsAllowInvalidHostnames"] == "true" + assert "replicaSet" not in options + assert "ssl" not in options + assert "tlsInsecure" not in options + assert "readPreference" not in options + assert "readPreferenceTags" not in options + assert "maxStalenessSeconds" not in options + + +def test_tunnel_endpoints_are_loopback_only_and_rejections_do_not_leak( + caplog, +) -> None: + secret = "do-not-log-this" + with caplog.at_level(logging.WARNING): + endpoints = mc._parse_ssh_tunnel_fallback_endpoints( + f"127.0.0.1:27018,remote.example:27019,{secret}@127.0.0.1:27020," + "[::1]:27021,127.0.0.1:0" + ) + + assert endpoints == ("127.0.0.1:27018", "[::1]:27021") + assert secret not in caplog.text + + +def test_network_failure_selects_writable_tunnel_member_before_direct_host( + monkeypatch, +) -> None: + _set_real_db_test_state(monkeypatch) + primary_uri = "mongodb+srv://user:pass@cluster.example/" + direct_uri = "mongodb://direct.example:27017/?tls=true" + secondary = _FakeClient( + "secondary", + {"ok": 1, "setName": "atlas-set", "isWritablePrimary": False}, + ) + writable = _FakeClient( + "writable", + {"ok": 1, "setName": "atlas-set", "isWritablePrimary": True}, + ) + created: list[str] = [] + + def _fake_new_client(uri: str, **_kwargs): + created.append(uri) + if uri == primary_uri: + raise ConnectionFailure("TLS handshake failed") + if "127.0.0.1:27018" in uri: + return secondary + if "127.0.0.1:27019" in uri: + return writable + raise AssertionError(f"Unexpected URI: {mc._host_display_from_uri(uri)}") + + monkeypatch.setattr(mc, "_new_mongo_client", _fake_new_client) + monkeypatch.setattr(mc, "MONGO_URI", primary_uri) + monkeypatch.setattr(mc, "MONGO_DNS_FALLBACK_URI", direct_uri) + monkeypatch.setattr( + mc, + "MONGO_SSH_TUNNEL_FALLBACK_ENDPOINTS", + ("127.0.0.1:27018", "127.0.0.1:27019", "127.0.0.1:27020"), + ) + monkeypatch.setattr(mc, "MONGO_SSH_TUNNEL_EXPECTED_REPLICA_SET", "atlas-set") + + db = mc.get_db() + + assert db is not None + assert db["label"] == "writable" + assert secondary.closed is True + assert created[0] == primary_uri + assert "127.0.0.1:27018" in created[1] + assert "127.0.0.1:27019" in created[2] + assert direct_uri not in created + state = mc.get_mongo_fallback_policy_state() + assert state["active_fallback_kind"] == "ssh_tunnel" + assert state["preferred_fallback_kind"] == "ssh_tunnel" + assert state["local_fallback_allowed"] is False + + +def test_fallback_order_is_cost_sensitive_to_failure_kind(monkeypatch) -> None: + monkeypatch.setattr( + mc, "MONGO_URI", "mongodb+srv://user:pass@cluster.example/" + ) + monkeypatch.setattr( + mc, + "MONGO_DNS_FALLBACK_URI", + "mongodb://direct.example:27017/?tls=true", + ) + monkeypatch.setattr( + mc, + "MONGO_SSH_TUNNEL_FALLBACK_ENDPOINTS", + ("127.0.0.1:27018", "127.0.0.1:27019"), + ) + monkeypatch.setattr(mc, "MONGO_ALLOW_LOCAL_FALLBACK", False) + + dns_failure_order = [ + target.kind for target in mc._connection_fallback_targets(prefer_dns=True) + ] + route_failure_order = [ + target.kind for target in mc._connection_fallback_targets(prefer_dns=False) + ] + + assert dns_failure_order == ["dns", "ssh_tunnel", "ssh_tunnel"] + assert route_failure_order == ["ssh_tunnel", "ssh_tunnel", "dns"] + + +def test_unwritable_or_foreign_tunnels_fall_through_without_local_mongo( + monkeypatch, +) -> None: + _set_real_db_test_state(monkeypatch) + primary_uri = "mongodb+srv://user:pass@cluster.example/" + direct_uri = "mongodb://direct.example:27017/?tls=true" + wrong_set = _FakeClient( + "wrong-set", + {"ok": 1, "setName": "foreign-set", "isWritablePrimary": True}, + ) + secondary = _FakeClient( + "secondary", + {"ok": 1, "setName": "atlas-set", "isWritablePrimary": False}, + ) + direct = _FakeClient("direct", {"ok": 1, "isWritablePrimary": True}) + created: list[str] = [] + + def _fake_new_client(uri: str, **_kwargs): + created.append(uri) + if uri == primary_uri: + raise ConnectionFailure("connection reset by peer") + if "127.0.0.1:27018" in uri: + return wrong_set + if "127.0.0.1:27019" in uri: + return secondary + if uri == direct_uri: + return direct + raise AssertionError(f"Unexpected URI: {mc._host_display_from_uri(uri)}") + + monkeypatch.setattr(mc, "_new_mongo_client", _fake_new_client) + monkeypatch.setattr(mc, "MONGO_URI", primary_uri) + monkeypatch.setattr(mc, "MONGO_DNS_FALLBACK_URI", direct_uri) + monkeypatch.setattr( + mc, + "MONGO_SSH_TUNNEL_FALLBACK_ENDPOINTS", + ("127.0.0.1:27018", "127.0.0.1:27019"), + ) + monkeypatch.setattr(mc, "MONGO_SSH_TUNNEL_EXPECTED_REPLICA_SET", "atlas-set") + monkeypatch.setattr( + mc, + "MONGO_LOCAL_URI", + "mongodb://127.0.0.1:27017/?directConnection=true", + ) + + db = mc.get_db() + + assert db is not None + assert db["label"] == "direct" + assert wrong_set.closed is True + assert secondary.closed is True + assert created[-1] == direct_uri + assert all("127.0.0.1:27017" not in uri for uri in created) + assert mc.get_mongo_fallback_policy_state()["active_fallback_kind"] == "dns" + + +def test_all_tunnel_failures_never_downgrade_to_disabled_local_mongo( + monkeypatch, +) -> None: + _set_real_db_test_state(monkeypatch) + primary_uri = "mongodb://remote.example:27017/?tls=true" + created: list[str] = [] + + def _fake_new_client(uri: str, **_kwargs): + created.append(uri) + raise ConnectionFailure("simulated route failure") + + monkeypatch.setattr(mc, "_new_mongo_client", _fake_new_client) + monkeypatch.setattr(mc, "MONGO_URI", primary_uri) + monkeypatch.setattr(mc, "MONGO_DNS_FALLBACK_URI", None) + monkeypatch.setattr( + mc, + "MONGO_SSH_TUNNEL_FALLBACK_ENDPOINTS", + ("127.0.0.1:27018", "127.0.0.1:27019"), + ) + monkeypatch.setattr( + mc, + "MONGO_LOCAL_URI", + "mongodb://127.0.0.1:27017/?directConnection=true", + ) + + assert mc.get_db() is None + assert len(created) == 3 + assert all("127.0.0.1:27017" not in uri for uri in created) + assert mc.is_using_fallback_uri() is False + + +def test_active_tunnel_reselects_after_replica_election(monkeypatch) -> None: + _set_real_db_test_state(monkeypatch) + old_primary = _FakeClient( + "old-primary", + {"ok": 1, "setName": "atlas-set", "isWritablePrimary": False}, + ) + secondary = _FakeClient( + "secondary", + {"ok": 1, "setName": "atlas-set", "isWritablePrimary": False}, + ) + new_primary = _FakeClient( + "new-primary", + {"ok": 1, "setName": "atlas-set", "isWritablePrimary": True}, + ) + endpoint_clients = iter((secondary, new_primary)) + + def _fake_new_client(uri: str, **_kwargs): + if "127.0.0.1:" in uri: + return next(endpoint_clients) + raise AssertionError("Direct primary must not be probed inside sticky window") + + monkeypatch.setattr(mc, "_new_mongo_client", _fake_new_client) + monkeypatch.setattr(mc, "MONGO_URI", "mongodb+srv://user:pass@cluster.example/") + monkeypatch.setattr(mc, "MONGO_DNS_FALLBACK_URI", None) + monkeypatch.setattr( + mc, + "MONGO_SSH_TUNNEL_FALLBACK_ENDPOINTS", + ("127.0.0.1:27018", "127.0.0.1:27019"), + ) + monkeypatch.setattr(mc, "MONGO_SSH_TUNNEL_EXPECTED_REPLICA_SET", "atlas-set") + monkeypatch.setattr(mc, "_mongo_client_real", old_primary) + monkeypatch.setattr(mc, "_using_fallback_real", True) + monkeypatch.setattr(mc, "_active_fallback_kind_real", "ssh_tunnel") + monkeypatch.setattr(mc, "_preferred_fallback_kind", "ssh_tunnel") + monkeypatch.setattr( + mc, "_dns_fallback_preferred_until_monotonic", time.monotonic() + 60 + ) + monkeypatch.setenv("VON_MONGO_AUTO_RECOVERY_CHECK_INTERVAL_SECONDS", "0.001") + + db = mc.get_db() + + assert db is not None + assert db["label"] == "new-primary" + assert secondary.closed is True + assert mc.get_mongo_fallback_policy_state()["active_fallback_kind"] == "ssh_tunnel" + + +def test_healthy_tunnel_returns_to_direct_primary_after_sticky_window( + monkeypatch, +) -> None: + _set_real_db_test_state(monkeypatch) + tunnel_client = _FakeClient( + "tunnel", + {"ok": 1, "setName": "atlas-set", "isWritablePrimary": True}, + ) + primary_client = _FakeClient("direct", {"ok": 1, "isWritablePrimary": True}) + primary_uri = "mongodb+srv://user:pass@cluster.example/" + + monkeypatch.setattr( + mc, + "_new_mongo_client", + lambda uri, **_kwargs: ( + primary_client + if uri == primary_uri + else (_ for _ in ()).throw(AssertionError("Unexpected fallback probe")) + ), + ) + monkeypatch.setattr(mc, "MONGO_URI", primary_uri) + monkeypatch.setattr(mc, "MONGO_DNS_FALLBACK_URI", None) + monkeypatch.setattr(mc, "MONGO_SSH_TUNNEL_FALLBACK_ENDPOINTS", ()) + monkeypatch.setattr(mc, "_mongo_client_real", tunnel_client) + monkeypatch.setattr(mc, "_using_fallback_real", True) + monkeypatch.setattr(mc, "_active_fallback_kind_real", "ssh_tunnel") + monkeypatch.setattr(mc, "_preferred_fallback_kind", "ssh_tunnel") + monkeypatch.setattr(mc, "_dns_fallback_preferred_until_monotonic", 0.0) + monkeypatch.setenv("VON_MONGO_AUTO_RECOVERY_CHECK_INTERVAL_SECONDS", "0.001") + + db = mc.get_db() + + assert db is not None + assert db["label"] == "direct" + assert tunnel_client.closed is False + assert mc.is_using_fallback_uri() is False + assert mc.get_mongo_fallback_policy_state()["active_fallback_kind"] is None diff --git a/tests/backend/test_mongo_uri_redaction.py b/tests/backend/test_mongo_uri_redaction.py index d5e91e88e..188e87eb7 100644 --- a/tests/backend/test_mongo_uri_redaction.py +++ b/tests/backend/test_mongo_uri_redaction.py @@ -11,6 +11,7 @@ sanitize_mongo_uri_for_display, ) from src.backend.server import utils_flask +from src.backend.server.routes import settings_routes from src.backend.server.routes.settings_routes import ( _classify_mongo_sanitized_uri, _sanitize_mongo_uri_for_display, @@ -65,6 +66,24 @@ def test_safe_mongo_location_classifies_without_retaining_raw_uri() -> None: _assert_no_secret_fragments(payload) +def test_safe_mongo_location_identifies_atlas_through_ssh_tunnel() -> None: + payload = build_safe_mongo_connection_location( + "mongodb://appuser:s3cr3t-pass@127.0.0.1:27018/" "?directConnection=true", + using_fallback=True, + fallback_kind="ssh_tunnel", + fallback_target_uri=FAKE_SECRET_URI, + ) + + assert payload["classification"] == "atlas" + assert payload["endpoint_classification"] == "local" + assert payload["upstream_classification"] == "atlas" + assert payload["transport"] == "ssh_tunnel" + assert payload["is_atlas"] is True + assert payload["is_local"] is False + assert payload["fallback_kind"] == "ssh_tunnel" + _assert_no_secret_fragments(payload) + + def test_safe_mongo_location_fails_closed_for_malformed_credential_uri() -> None: malformed_uri = ( "mongodb+srv://appuser:s3cr3t/pass@cluster0.example.mongodb.net/" @@ -94,6 +113,11 @@ def test_health_summary_exposes_only_sanitized_effective_uri(monkeypatch) -> Non lambda: FAKE_SECRET_URI, ) monkeypatch.setattr(conn_mgr._mc, "is_using_fallback_uri", lambda: False) + monkeypatch.setattr( + conn_mgr._mc, + "get_mongo_fallback_policy_state", + lambda: {"active_fallback_kind": None}, + ) payload = conn_mgr.health_summary() @@ -108,6 +132,11 @@ def test_admin_diag_mongo_payload_redacts_effective_uri(monkeypatch) -> None: app = Flask(__name__) monkeypatch.setattr(mongo_client, "get_effective_mongo_uri", lambda: FAKE_SECRET_URI) monkeypatch.setattr(mongo_client, "is_using_fallback_uri", lambda: False) + monkeypatch.setattr( + mongo_client, + "get_mongo_fallback_policy_state", + lambda: {"active_fallback_kind": None}, + ) with app.test_request_context("/diag"): response = utils_flask._build_diagnostics_response(app) @@ -122,6 +151,11 @@ def test_system_db_status_uses_safe_effective_host(monkeypatch) -> None: app = Flask(__name__) monkeypatch.setattr(mongo_client, "get_effective_mongo_uri", lambda: FAKE_SECRET_URI) monkeypatch.setattr(mongo_client, "is_using_fallback_uri", lambda: False) + monkeypatch.setattr( + mongo_client, + "get_mongo_fallback_policy_state", + lambda: {"active_fallback_kind": None}, + ) with app.test_request_context("/api/system/db_status"): response = utils_flask._build_db_status_response() @@ -129,4 +163,56 @@ def test_system_db_status_uses_safe_effective_host(monkeypatch) -> None: payload = response.get_json() assert payload["effective_host"] == "cluster0.example.mongodb.net" assert payload["atlas_detected"] is True + assert payload["fallback_kind"] is None + _assert_no_secret_fragments(payload) + + +def test_db_info_snapshots_route_after_ping_recovery(monkeypatch) -> None: + app = Flask(__name__) + state = { + "uri": FAKE_SECRET_URI, + "using_fallback": False, + "kind": None, + } + + def _get_db_after_route_switch(): + state.update( + { + "uri": ( + "mongodb://appuser:s3cr3t-pass@127.0.0.1:27019/" + "?directConnection=true" + ), + "using_fallback": True, + "kind": "ssh_tunnel", + } + ) + return _FakeDb() + + monkeypatch.setattr(settings_routes, "_read_cached_db_info", lambda **_kwargs: None) + monkeypatch.setattr(settings_routes, "_write_cached_db_info", lambda _payload: None) + monkeypatch.setattr(settings_routes, "get_db", _get_db_after_route_switch) + monkeypatch.setattr( + settings_routes, "get_effective_mongo_uri", lambda: state["uri"] + ) + monkeypatch.setattr( + settings_routes, "is_using_fallback_uri", lambda: state["using_fallback"] + ) + monkeypatch.setattr( + settings_routes, + "get_mongo_fallback_policy_state", + lambda: {"active_fallback_kind": state["kind"]}, + ) + monkeypatch.setattr(settings_routes, "MONGO_URI", FAKE_SECRET_URI) + monkeypatch.setattr(settings_routes, "DATABASE_NAME", "test_database") + + with app.test_request_context("/api/settings/db/info?nocache=1"): + response, status = settings_routes.get_db_location_info() + + payload = response.get_json() + assert status == 200 + assert payload["sanitized_uri"] == "mongodb://127.0.0.1:27019" + assert payload["using_fallback"] is True + assert payload["fallback_kind"] == "ssh_tunnel" + assert payload["classification"] == "atlas" + assert payload["connection_location"]["upstream_sanitized_uri"] == SAFE_URI _assert_no_secret_fragments(payload) diff --git a/tests/test_get_safe_env.py b/tests/test_get_safe_env.py index fd03aaca4..360aafa3f 100644 --- a/tests/test_get_safe_env.py +++ b/tests/test_get_safe_env.py @@ -20,7 +20,21 @@ def test_get_safe_env_value_redacts_mongo_password(tmp_path: Path): value = get_safe_env_value(env_path, "MONGO_URI") assert value assert "supersecret" not in value - assert "[REDACTED]" in value + assert "user" not in value + assert value == "mongodb+srv://cluster.mongodb.net" + + +def test_get_safe_env_value_redacts_standard_mongo_uri(tmp_path: Path): + env_path = tmp_path / ".env" + env_path.write_text( + "MONGO_DNS_FALLBACK_URI=" + "mongodb://user:supersecret@host.mongodb.net:27017/?tls=true\n", + encoding="utf-8", + ) + + value = get_safe_env_value(env_path, "MONGO_DNS_FALLBACK_URI") + + assert value == "mongodb://host.mongodb.net:27017" def test_get_safe_env_value_returns_plain_value_for_non_secret(tmp_path: Path): diff --git a/tests/test_install_macos_mongo_ssh_tunnel.py b/tests/test_install_macos_mongo_ssh_tunnel.py new file mode 100644 index 000000000..3c27d997e --- /dev/null +++ b/tests/test_install_macos_mongo_ssh_tunnel.py @@ -0,0 +1,637 @@ +from __future__ import annotations + +import json +import plistlib +from pathlib import Path + +import pytest + +from scripts import install_macos_mongo_ssh_tunnel as installer + + +def _forwards() -> list[installer.Forward]: + return [ + installer.Forward( + 27018, + "ac-example-shard-00-00.mongodb.net", + 27017, + ), + installer.Forward( + 27019, + "ac-example-shard-00-01.mongodb.net", + 27017, + ), + installer.Forward( + 27020, + "ac-example-shard-00-02.mongodb.net", + 27017, + ), + ] + + +def test_program_arguments_are_direct_fixed_and_secret_free(tmp_path: Path) -> None: + identity = tmp_path / "identity" + known_hosts = tmp_path / "known_hosts" + + arguments = installer.build_program_arguments( + ssh_destination="operator@relay.example", + identity_file=identity, + known_hosts_file=known_hosts, + forwards=_forwards(), + ) + + assert arguments[0] == "/usr/bin/ssh" + assert arguments[-1] == "operator@relay.example" + assert arguments[1:3] == ["-F", "/dev/null"] + assert "BatchMode=yes" in arguments + assert "StrictHostKeyChecking=yes" in arguments + assert "ExitOnForwardFailure=yes" in arguments + assert "ControlMaster=no" in arguments + assert arguments.count("-L") == 3 + assert all( + "127.0.0.1:" in arguments[index + 1] + for index, value in enumerate(arguments) + if value == "-L" + ) + rendered = "\n".join(arguments) + assert "mongodb://" not in rendered + assert "password" not in rendered.lower() + + +def test_launch_agent_is_supervised_and_owner_only_by_design(tmp_path: Path) -> None: + payload = installer.build_launch_agent( + label=installer.DEFAULT_LABEL, + program_arguments=["/usr/bin/ssh", "-N", "operator@relay.example"], + stdout_path=tmp_path / "stdout.log", + stderr_path=tmp_path / "stderr.log", + ) + + assert payload["RunAtLoad"] is True + assert payload["KeepAlive"] is True + assert payload["ThrottleInterval"] == 10 + assert payload["Umask"] == 0o077 + assert payload["ProgramArguments"][0] == "/usr/bin/ssh" + + +@pytest.mark.parametrize( + "value", + ( + "0:remote.example:27017", + "27018:bad host:27017", + "27018:remote.example:70000", + "27018:remote.example", + ), +) +def test_invalid_forwards_are_rejected(value: str) -> None: + with pytest.raises(installer.argparse.ArgumentTypeError): + installer._forward(value) + + +def test_install_writes_private_plist_and_bootstraps_launchd( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + identity = tmp_path / "identity" + known_hosts = tmp_path / "known_hosts" + identity.write_text("test key material\n", encoding="utf-8") + known_hosts.write_text("relay.example test-host-key\n", encoding="utf-8") + identity.chmod(0o600) + known_hosts.chmod(0o600) + plist_path = tmp_path / "LaunchAgents" / "tunnel.plist" + log_directory = tmp_path / "logs" + launchctl_calls: list[tuple[list[str], bool]] = [] + + monkeypatch.setattr(installer, "_service_is_loaded", lambda _label: False) + monkeypatch.setattr(installer, "_ports_are_available", lambda _items: True) + monkeypatch.setattr( + installer, "_wait_for_listeners", lambda _label, _items: True + ) + + def _fake_launchctl(arguments, *, check): + launchctl_calls.append((list(arguments), check)) + + monkeypatch.setattr(installer, "_run_launchctl", _fake_launchctl) + monkeypatch.setattr( + installer, + "_status_payload", + lambda **_kwargs: {"installed": True, "loaded": True}, + ) + + result = installer.run( + [ + "install", + "--ssh-destination", + "operator@relay.example", + "--identity-file", + str(identity), + "--known-hosts-file", + str(known_hosts), + "--plist-path", + str(plist_path), + "--log-directory", + str(log_directory), + *[ + item + for forward in _forwards() + for item in ( + "--forward", + f"{forward.local_port}:{forward.remote_host}:{forward.remote_port}", + ) + ], + ] + ) + + assert result == 0 + assert plist_path.stat().st_mode & 0o777 == 0o600 + with plist_path.open("rb") as handle: + payload = plistlib.load(handle) + assert payload["ProgramArguments"].count("-L") == 3 + assert launchctl_calls == [ + ( + [ + "bootstrap", + f"gui/{installer.os.getuid()}", + str(plist_path), + ], + True, + ) + ] + + +def test_known_hosts_may_be_world_readable_but_not_world_writable( + tmp_path: Path, +) -> None: + known_hosts = tmp_path / "known_hosts" + known_hosts.write_text("relay.example test-host-key\n", encoding="utf-8") + known_hosts.chmod(0o644) + + assert ( + installer._absolute_existing_file( + str(known_hosts), + description="known-hosts file", + allow_public_read=True, + ) + == known_hosts + ) + + known_hosts.chmod(0o646) + with pytest.raises(ValueError, match="unsafe group or other permissions"): + installer._absolute_existing_file( + str(known_hosts), + description="known-hosts file", + allow_public_read=True, + ) + + +def test_install_does_not_replace_a_foreign_port_owner( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + identity = tmp_path / "identity" + known_hosts = tmp_path / "known_hosts" + identity.write_text("test\n", encoding="utf-8") + known_hosts.write_text("test\n", encoding="utf-8") + identity.chmod(0o600) + known_hosts.chmod(0o600) + plist_path = tmp_path / "tunnel.plist" + + monkeypatch.setattr(installer, "_service_is_loaded", lambda _label: False) + monkeypatch.setattr(installer, "_ports_are_available", lambda _items: False) + + with pytest.raises(RuntimeError, match="no process was killed"): + installer.run( + [ + "install", + "--ssh-destination", + "operator@relay.example", + "--identity-file", + str(identity), + "--known-hosts-file", + str(known_hosts), + "--plist-path", + str(plist_path), + "--log-directory", + str(tmp_path / "logs"), + "--forward", + "27018:remote.example:27017", + ] + ) + + assert not plist_path.exists() + + +def test_install_refuses_unsafe_existing_log_directory_without_changing_mode( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + identity = tmp_path / "identity" + known_hosts = tmp_path / "known_hosts" + identity.write_text("test\n", encoding="utf-8") + known_hosts.write_text("test\n", encoding="utf-8") + identity.chmod(0o600) + known_hosts.chmod(0o600) + log_directory = tmp_path / "existing-logs" + log_directory.mkdir(mode=0o755) + log_directory.chmod(0o755) + monkeypatch.setattr(installer, "_service_is_loaded", lambda _label: False) + + with pytest.raises(ValueError, match="owner-only rwx"): + installer.run( + [ + "install", + "--ssh-destination", + "operator@relay.example", + "--identity-file", + str(identity), + "--known-hosts-file", + str(known_hosts), + "--plist-path", + str(tmp_path / "tunnel.plist"), + "--log-directory", + str(log_directory), + "--forward", + "27018:remote.example:27017", + ] + ) + + assert log_directory.stat().st_mode & 0o777 == 0o755 + + +def test_install_refuses_symlinked_log_directory( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + identity = tmp_path / "identity" + known_hosts = tmp_path / "known_hosts" + identity.write_text("test\n", encoding="utf-8") + known_hosts.write_text("test\n", encoding="utf-8") + identity.chmod(0o600) + known_hosts.chmod(0o600) + real_logs = tmp_path / "real-logs" + real_logs.mkdir(mode=0o700) + log_directory = tmp_path / "linked-logs" + log_directory.symlink_to(real_logs, target_is_directory=True) + monkeypatch.setattr(installer, "_service_is_loaded", lambda _label: False) + + with pytest.raises(ValueError, match="symbolic link"): + installer.run( + [ + "install", + "--ssh-destination", + "operator@relay.example", + "--identity-file", + str(identity), + "--known-hosts-file", + str(known_hosts), + "--plist-path", + str(tmp_path / "tunnel.plist"), + "--log-directory", + str(log_directory), + "--forward", + "27018:remote.example:27017", + ] + ) + + +def test_status_infers_forwards_and_checks_launchd_pid_ownership( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], +) -> None: + plist_path = tmp_path / "tunnel.plist" + plist_path.write_bytes( + plistlib.dumps( + { + "Label": installer.DEFAULT_LABEL, + "ProgramArguments": [ + "/usr/bin/ssh", + "-L", + "127.0.0.1:27018:remote.example:27017", + "operator@relay.example", + ], + } + ) + ) + monkeypatch.setattr(installer, "_service_is_loaded", lambda _label: True) + monkeypatch.setattr(installer, "_service_pid", lambda _label: 1234) + monkeypatch.setattr(installer, "_listener_pids", lambda _port: {1234}) + + assert ( + installer.run( + [ + "status", + "--plist-path", + str(plist_path), + "--log-directory", + str(tmp_path / "logs"), + ] + ) + == 0 + ) + + result = json.loads(capsys.readouterr().out) + assert result["configuration_status"] == "configured" + assert result["plist_valid"] is True + assert result["label_matches"] is True + assert result["listeners_checked"] is True + assert result["listener_coverage_complete"] is True + assert result["listeners"] == {"27018": True} + assert result["mongo_route_probed"] is False + + +@pytest.mark.parametrize( + ("contents", "expected_status", "expected_valid", "expected_label_matches"), + ( + (b"not a plist", "invalid", False, None), + ( + plistlib.dumps( + { + "Label": "example.unrelated.service", + "ProgramArguments": ["/usr/bin/true"], + } + ), + "foreign_label", + True, + False, + ), + ), +) +def test_status_types_invalid_and_foreign_plists( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + capsys: pytest.CaptureFixture[str], + contents: bytes, + expected_status: str, + expected_valid: bool, + expected_label_matches: bool | None, +) -> None: + plist_path = tmp_path / "tunnel.plist" + plist_path.write_bytes(contents) + monkeypatch.setattr(installer, "_service_is_loaded", lambda _label: False) + + assert ( + installer.run( + [ + "status", + "--plist-path", + str(plist_path), + "--log-directory", + str(tmp_path / "logs"), + ] + ) + == 0 + ) + + result = json.loads(capsys.readouterr().out) + assert result["installed"] is True + assert result["configuration_status"] == expected_status + assert result["plist_valid"] is expected_valid + assert result["label_matches"] is expected_label_matches + assert result["listeners_checked"] is False + assert result["listener_coverage_complete"] is False + + +def test_status_and_uninstall_do_not_require_forward_arguments( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + plist_path = tmp_path / "tunnel.plist" + plist_path.write_bytes( + plistlib.dumps( + { + "Label": installer.DEFAULT_LABEL, + "ProgramArguments": ["/usr/bin/ssh", "-N", "operator@relay.example"], + } + ) + ) + monkeypatch.setattr(installer, "_service_is_loaded", lambda _label: False) + + assert ( + installer.run( + [ + "status", + "--plist-path", + str(plist_path), + "--log-directory", + str(tmp_path / "logs"), + ] + ) + == 0 + ) + assert ( + installer.run( + [ + "uninstall", + "--confirm-uninstall", + "--plist-path", + str(plist_path), + "--log-directory", + str(tmp_path / "logs"), + ] + ) + == 0 + ) + assert not plist_path.exists() + + +def test_uninstall_refuses_loaded_service_without_matching_plist( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + launchctl_calls: list[list[str]] = [] + monkeypatch.setattr(installer, "_service_is_loaded", lambda _label: True) + monkeypatch.setattr( + installer, + "_run_launchctl", + lambda arguments, *, check: launchctl_calls.append(list(arguments)), + ) + + with pytest.raises(RuntimeError, match="without its matching plist"): + installer.run( + [ + "uninstall", + "--confirm-uninstall", + "--plist-path", + str(tmp_path / "missing.plist"), + "--log-directory", + str(tmp_path / "logs"), + ] + ) + + assert launchctl_calls == [] + + +@pytest.mark.parametrize("action", ("install", "uninstall")) +def test_mutating_actions_refuse_plist_owned_by_another_label( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + action: str, +) -> None: + identity = tmp_path / "identity" + known_hosts = tmp_path / "known_hosts" + identity.write_text("test\n", encoding="utf-8") + known_hosts.write_text("test\n", encoding="utf-8") + identity.chmod(0o600) + known_hosts.chmod(0o600) + plist_path = tmp_path / "other.plist" + plist_path.write_bytes( + plistlib.dumps( + { + "Label": "example.unrelated.service", + "ProgramArguments": ["/usr/bin/true"], + } + ) + ) + original = plist_path.read_bytes() + monkeypatch.setattr(installer, "_service_is_loaded", lambda _label: False) + + arguments = [ + action, + "--plist-path", + str(plist_path), + "--log-directory", + str(tmp_path / "logs"), + ] + if action == "install": + arguments.extend( + [ + "--ssh-destination", + "operator@relay.example", + "--identity-file", + str(identity), + "--known-hosts-file", + str(known_hosts), + "--forward", + "27018:remote.example:27017", + ] + ) + else: + arguments.append("--confirm-uninstall") + + with pytest.raises(ValueError, match="different label"): + installer.run(arguments) + + assert plist_path.read_bytes() == original + + +def test_failed_reinstall_restores_and_reloads_previous_service( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + identity = tmp_path / "identity" + known_hosts = tmp_path / "known_hosts" + identity.write_text("test\n", encoding="utf-8") + known_hosts.write_text("test\n", encoding="utf-8") + identity.chmod(0o600) + known_hosts.chmod(0o600) + plist_path = tmp_path / "tunnel.plist" + previous_plist = plistlib.dumps( + { + "Label": installer.DEFAULT_LABEL, + "ProgramArguments": ["/usr/bin/ssh", "-N", "old@relay.example"], + } + ) + plist_path.write_bytes(previous_plist) + service_loaded_calls = 0 + bootstrap_calls = 0 + + def _fake_service_is_loaded(_label: str) -> bool: + nonlocal service_loaded_calls + service_loaded_calls += 1 + return service_loaded_calls == 1 + + def _fake_launchctl(arguments, *, check): + nonlocal bootstrap_calls + assert check is True + if arguments[0] == "bootstrap": + bootstrap_calls += 1 + if bootstrap_calls == 1: + raise RuntimeError("simulated replacement bootstrap failure") + + monkeypatch.setattr(installer, "_service_is_loaded", _fake_service_is_loaded) + monkeypatch.setattr(installer, "_ports_are_available", lambda _items: True) + monkeypatch.setattr(installer, "_run_launchctl", _fake_launchctl) + + with pytest.raises(RuntimeError, match="simulated replacement"): + installer.run( + [ + "install", + "--ssh-destination", + "operator@relay.example", + "--identity-file", + str(identity), + "--known-hosts-file", + str(known_hosts), + "--plist-path", + str(plist_path), + "--log-directory", + str(tmp_path / "logs"), + "--forward", + "27018:remote.example:27017", + ] + ) + + assert plist_path.read_bytes() == previous_plist + assert plist_path.stat().st_mode & 0o777 == 0o600 + assert bootstrap_calls == 2 + + +def test_listener_timeout_rolls_back_loaded_previous_service( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + identity = tmp_path / "identity" + known_hosts = tmp_path / "known_hosts" + identity.write_text("test\n", encoding="utf-8") + known_hosts.write_text("test\n", encoding="utf-8") + identity.chmod(0o600) + known_hosts.chmod(0o600) + plist_path = tmp_path / "tunnel.plist" + previous_plist = plistlib.dumps( + { + "Label": installer.DEFAULT_LABEL, + "ProgramArguments": ["/usr/bin/ssh", "-N", "old@relay.example"], + } + ) + plist_path.write_bytes(previous_plist) + service_loaded_calls = 0 + launchctl_calls: list[str] = [] + + def _fake_service_is_loaded(_label: str) -> bool: + nonlocal service_loaded_calls + service_loaded_calls += 1 + return True + + def _fake_launchctl(arguments, *, check): + assert check is True + launchctl_calls.append(arguments[0]) + + monkeypatch.setattr(installer, "_service_is_loaded", _fake_service_is_loaded) + monkeypatch.setattr(installer, "_ports_are_available", lambda _items: True) + monkeypatch.setattr( + installer, "_wait_for_listeners", lambda _label, _items: False + ) + monkeypatch.setattr(installer, "_run_launchctl", _fake_launchctl) + + with pytest.raises(RuntimeError, match="listeners became ready"): + installer.run( + [ + "install", + "--ssh-destination", + "operator@relay.example", + "--identity-file", + str(identity), + "--known-hosts-file", + str(known_hosts), + "--plist-path", + str(plist_path), + "--log-directory", + str(tmp_path / "logs"), + "--forward", + "27018:remote.example:27017", + ] + ) + + assert plist_path.read_bytes() == previous_plist + assert plist_path.stat().st_mode & 0o777 == 0o600 + assert launchctl_calls == ["bootout", "bootstrap", "bootout", "bootstrap"] diff --git a/tests/test_run_sh_start_behaviour.py b/tests/test_run_sh_start_behaviour.py index aae3b9b49..8f591987e 100644 --- a/tests/test_run_sh_start_behaviour.py +++ b/tests/test_run_sh_start_behaviour.py @@ -36,6 +36,28 @@ def _run_bash_probe(script: str) -> dict[str, object]: return json.loads(result.stdout) +def _run_bash_text_probe(script: str) -> str: + bash = shutil.which("bash") + if not bash: + pytest.skip("bash is required for run.sh launcher-path tests") + + wrapped = f""" +set -euo pipefail +cd {shlex_quote(str(REPO_ROOT))} +. ./run.sh help -NoBackupMigrate >/dev/null +{script} +""".strip() + result = subprocess.run( + [bash, "-c", wrapped], + cwd=REPO_ROOT, + text=True, + capture_output=True, + timeout=20, + check=True, + ) + return result.stdout + + def shlex_quote(value: str) -> str: return "'" + value.replace("'", "'\"'\"'") + "'" @@ -53,6 +75,47 @@ def test_run_sh_default_health_timeout_covers_slow_startup() -> None: assert payload == {"health_timeout_seconds": 180} +def test_run_sh_labels_ssh_tunnel_fallback_as_remote_atlas() -> None: + output = _run_bash_text_probe(r""" +curl() { + printf '%s' '{"using_fallback":true,"fallback_kind":"ssh_tunnel","atlas_detected":true,"effective_host":"127.0.0.1:27019"}' +} +log_mongo_status +""") + + assert "Mongo: Atlas via SSH tunnel (127.0.0.1:27019)" in output + assert "Mongo: local fallback" not in output + + +def test_run_sh_reports_listener_presence_without_claiming_mongo_readiness() -> None: + output = _run_bash_text_probe(r""" +MONGO_SSH_TUNNEL_FALLBACK_ENDPOINTS="127.0.0.1:27018,127.0.0.1:27019" +lsof() { + return 0 +} +log_mongo_ssh_tunnel_status +""") + + assert "listeners present (2/2); Mongo route not probed" in output + assert "ready" not in output + + +def test_run_sh_does_not_claim_direct_atlas_for_incomplete_listeners() -> None: + output = _run_bash_text_probe(r""" +MONGO_SSH_TUNNEL_FALLBACK_ENDPOINTS="127.0.0.1:27018,127.0.0.1:27019" +lsof() { + case "$*" in + *27018*) return 0 ;; + *) return 1 ;; + esac +} +log_mongo_ssh_tunnel_status +""") + + assert "listener coverage incomplete (1/2); inspect live Mongo route" in output + assert "direct Atlas remains primary" not in output + + def _prepare_backup_launcher( tmp_path: Path, *, @@ -947,6 +1010,7 @@ def test_run_sh_workflow_purity_check_respects_daily_success_ttl() -> None: start_rag_worker_bg() { :; } start_concept_index_worker_bg() { :; } log_von_version() { :; } + log_mongo_ssh_tunnel_status() { :; } PORT=5124 AGENT_TEST_INSTANCE=0 diff --git a/utilities/get_safe_env.py b/utilities/get_safe_env.py index 114bf4f41..2be926831 100644 --- a/utilities/get_safe_env.py +++ b/utilities/get_safe_env.py @@ -18,16 +18,25 @@ import argparse import re +import sys from pathlib import Path from typing import Dict, Optional +_REPO_ROOT = Path(__file__).resolve().parents[1] +if str(_REPO_ROOT) not in sys.path: + sys.path.insert(0, str(_REPO_ROOT)) + +from src.backend.db.mongo_uri_redaction import sanitize_mongo_uri_for_display SAFE_KEYS = { # Database config "VON_DB_NAME", "MONGO_PROJECT", "MONGO_ALLOW_LOCAL_FALLBACK", + "MONGO_DNS_FALLBACK_URI", "MONGO_LOCAL_URI", + "MONGO_SSH_TUNNEL_EXPECTED_REPLICA_SET", + "MONGO_SSH_TUNNEL_FALLBACK_ENDPOINTS", "MONGO_URI", # Non-secret feature toggles "VON_INTERNAL_MCP_ENABLE", @@ -87,9 +96,8 @@ def _redact_value(key: str, value: str) -> str: return "[REDACTED]" # Redact credentials in Mongo-style URIs. - if key in {"MONGO_URI", "MONGO_LOCAL_URI"}: - # mongodb+srv://user:pass@host/... -> mongodb+srv://user:[REDACTED]@host/... - return re.sub(r"(mongodb\+srv://[^:]+:)([^@]+)(@)", r"\1[REDACTED]\3", value) + if key in {"MONGO_URI", "MONGO_LOCAL_URI", "MONGO_DNS_FALLBACK_URI"}: + return sanitize_mongo_uri_for_display(value) or "[REDACTED]" # Heuristic: long opaque values are probably secrets. if len(value) >= 40 and re.fullmatch(r"[A-Za-z0-9_\-\.]+", value):