pyssh-client/tests/test_bastion.py

"""ProxyJump / Bastion : resolution (app.bastion) pour les 3 sources (session,
OpenSSH, manuel), et etablissement du canal direct-tcpip par SSHConnection via
un bastion entierement mocke (aucun reseau reel)."""
import pathlib
import sys

sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent))

import tempfile
from pathlib import Path
from unittest.mock import MagicMock, patch

from PySide6.QtWidgets import QApplication

from app.bastion import BastionResolutionError, BastionTarget, resolve_bastion_target
from app.session_manager import (
    AUTH_AUTO,
    AUTH_KEY,
    BASTION_MANUAL,
    BASTION_NONE,
    BASTION_OPENSSH,
    BASTION_SESSION,
    BastionConfig,
    Session,
    SessionStore,
)

app = QApplication.instance() or QApplication([])

tmp = Path(tempfile.mkdtemp())

print("=== 1. Aucun bastion -> None ===")
assert resolve_bastion_target(BastionConfig(mode=BASTION_NONE)) is None
assert resolve_bastion_target(None) is None
print("  OK")

print()
print("=== 2. Bastion manuel ===")
manual = BastionConfig(
    mode=BASTION_MANUAL, host="bastion.example.com", port=2222,
    username="jump", auth_type=AUTH_KEY, key_path="/tmp/key",
)
target = resolve_bastion_target(manual)
print("  target:", target)
assert target == BastionTarget("bastion.example.com", 2222, "jump", AUTH_KEY, "/tmp/key")
print("  OK")

print()
print("=== 3. Bastion manuel incomplet -> erreur claire ===")
try:
    resolve_bastion_target(BastionConfig(mode=BASTION_MANUAL, host=""))
    ok = False
except BastionResolutionError:
    ok = True
assert ok
print("  OK")

print()
print("=== 4. Bastion = session PySSH existante ===")
store = SessionStore(path=tmp / "sessions.json")
bastion_session = Session(name="bastion-srv", host="10.0.0.1", port=22, username="root", auth_type=AUTH_AUTO)
store.add(bastion_session)
target = resolve_bastion_target(BastionConfig(mode=BASTION_SESSION, session_id=bastion_session.id), store)
print("  target:", target)
assert target.host == "10.0.0.1" and target.username == "root"
print("  OK")

print()
print("=== 5. Bastion = session supprimee -> erreur claire ===")
try:
    resolve_bastion_target(BastionConfig(mode=BASTION_SESSION, session_id="inconnu"), store)
    ok = False
except BastionResolutionError as exc:
    ok = True
    print("  message:", exc)
assert ok
print("  OK")

print()
print("=== 6. Bastion sans registre de sessions fourni -> erreur claire (pas de crash) ===")
try:
    resolve_bastion_target(BastionConfig(mode=BASTION_SESSION, session_id="x"), None)
    ok = False
except BastionResolutionError:
    ok = True
assert ok
print("  OK")

print()
print("=== 7. Bastion = Host OpenSSH ===")
config_path = tmp / "sshconfig"
config_path.write_text(
    "Host bastion\n    HostName bastion.example.com\n    User jump\n    IdentityFile ~/.ssh/jumpkey\n",
    encoding="utf-8",
)
with patch("app.openssh.config.default_config_path", return_value=config_path):
    target = resolve_bastion_target(BastionConfig(mode=BASTION_OPENSSH, openssh_alias="bastion"))
print("  target:", target)
assert target.host == "bastion.example.com"
assert target.username == "jump"
assert target.auth_type == AUTH_KEY
print("  OK")

print()
print("=== 8. Bastion = alias OpenSSH introuvable -> erreur claire ===")
with patch("app.openssh.config.default_config_path", return_value=config_path):
    try:
        resolve_bastion_target(BastionConfig(mode=BASTION_OPENSSH, openssh_alias="n-existe-pas"))
        ok = False
    except BastionResolutionError:
        ok = True
assert ok
print("  OK")

print()
print("=== 9. SSHConnection : ouverture du canal direct-tcpip via un bastion mocke ===")
from app.ssh_connection import SSHConnection

bastion_target = BastionTarget(host="bastion.example.com", port=22, username="jump", auth_type=AUTH_AUTO)
conn = SSHConnection(host="10.20.20.15", port=22, username="root", auth_type=AUTH_AUTO, bastion_target=bastion_target)

fake_channel = MagicMock(name="direct-tcpip-channel")
fake_transport = MagicMock()
fake_transport.open_channel.return_value = fake_channel

with patch("app.ssh_connection.paramiko.SSHClient") as FakeSSHClient:
    fake_bastion_client = MagicMock()
    fake_bastion_client.get_transport.return_value = fake_transport
    FakeSSHClient.return_value = fake_bastion_client

    channel = conn._open_bastion_channel()

hostname_used = fake_bastion_client.connect.call_args.kwargs.get("hostname")
print("  connect() vers le bastion, hostname =", hostname_used)
assert hostname_used == "bastion.example.com"
assert fake_transport.open_channel.call_args.args[0] == "direct-tcpip"
assert fake_transport.open_channel.call_args.args[1] == ("10.20.20.15", 22)
assert channel is fake_channel
assert conn._bastion_client is fake_bastion_client
print("  OK")

print()
print("=== 10. Cible connectee avec sock=<canal du bastion> (pas de socket TCP direct) ===")
conn2 = SSHConnection(host="10.20.20.15", port=22, username="root", auth_type=AUTH_AUTO, bastion_target=bastion_target)
with patch("app.ssh_connection.paramiko.SSHClient") as FakeSSHClient:
    fake_bastion_client2 = MagicMock()
    fake_target_client2 = MagicMock()
    fake_transport2 = MagicMock()
    fake_transport2.open_channel.return_value = fake_channel
    fake_bastion_client2.get_transport.return_value = fake_transport2
    # comme dans run() : le 1er SSHClient() cree est celui de la cible (self.client,
    # assigne avant l'appel a _open_bastion_channel), le 2e celui du bastion (cree a
    # l'interieur de _open_bastion_channel)
    FakeSSHClient.side_effect = [fake_target_client2, fake_bastion_client2]

    conn2.client = conn2._new_client()
    sock = conn2._open_bastion_channel()
    conn2._connect_host(
        conn2.client, conn2.host, conn2.port, conn2.username, conn2.auth_type,
        conn2.key_path, conn2.password, conn2.passphrase, sock=sock,
    )

print("  cible connect() sock=canal du bastion :", fake_target_client2.connect.call_args.kwargs.get("sock") is fake_channel)
assert fake_target_client2.connect.call_args.kwargs.get("sock") is fake_channel
print("  OK")

print()
print("=== 11. Echec de connexion au bastion -> message clair, jamais de fuite d'exception brute ===")
conn3 = SSHConnection(host="10.20.20.15", port=22, username="root", auth_type=AUTH_AUTO, bastion_target=bastion_target)
with patch("app.ssh_connection.paramiko.SSHClient") as FakeSSHClient:
    fake_bastion_client3 = MagicMock()
    fake_bastion_client3.connect.side_effect = OSError("Connexion refusee")
    FakeSSHClient.return_value = fake_bastion_client3
    try:
        conn3._open_bastion_channel()
        ok = False
    except Exception as exc:  # noqa: BLE001
        ok = True
        message = str(exc)
        print("  message:", message)
        assert "Bastion" in message
assert ok
print("  OK")

print()
print("BASTION / PROXYJUMP : TOUS LES TESTS OK")