diff --git a/src/autossh_mgr/check.py b/src/autossh_mgr/check.py
new file mode 100644
index 0000000..03d9c7d
--- /dev/null
+++ b/src/autossh_mgr/check.py
@@ -0,0 +1,24 @@
+import os
+import subprocess
+from autossh_mgr.config import TunnelConfig
+
+
+def build_ssh_check_cmd(tunnel: TunnelConfig) -> list[str]:
+    return [
+        "ssh",
+        "-o", "BatchMode=yes",
+        "-o", "ConnectTimeout=5",
+        "-i", os.path.expanduser(tunnel.identity_file),
+        "-p", str(tunnel.port),
+        f"{tunnel.user}@{tunnel.host}",
+        "true",
+    ]
+
+
+def check_connectivity(tunnel: TunnelConfig) -> tuple[bool, str]:
+    result = subprocess.run(
+        build_ssh_check_cmd(tunnel),
+        capture_output=True,
+        text=True,
+    )
+    return result.returncode == 0, result.stderr.strip()
diff --git a/tests/unit/test_check.py b/tests/unit/test_check.py
new file mode 100644
index 0000000..0491496
--- /dev/null
+++ b/tests/unit/test_check.py
@@ -0,0 +1,45 @@
+import pytest
+import subprocess
+from unittest.mock import patch, MagicMock
+from autossh_mgr.check import check_connectivity, build_ssh_check_cmd
+from autossh_mgr.config import TunnelConfig
+
+
+@pytest.fixture
+def tunnel():
+    return TunnelConfig(
+        name="web", host="relay.example.com", user="deploy",
+        local_port=8080, remote_port=18080,
+    )
+
+
+def test_build_ssh_check_cmd(tunnel):
+    cmd = build_ssh_check_cmd(tunnel)
+    assert cmd[0] == "ssh"
+    assert "-o" in cmd and "BatchMode=yes" in cmd
+    assert "ConnectTimeout=5" in " ".join(cmd)
+    assert f"{tunnel.user}@{tunnel.host}" in cmd
+    assert "true" in cmd
+    assert "-p" in cmd and str(tunnel.port) in cmd
+
+
+def test_check_connectivity_success(tunnel):
+    mock_result = MagicMock()
+    mock_result.returncode = 0
+    mock_result.stderr = ""
+    with patch("autossh_mgr.check.subprocess.run", return_value=mock_result) as mock_run:
+        success, msg = check_connectivity(tunnel)
+    assert success is True
+    assert msg == ""
+    called_cmd = mock_run.call_args[0][0]
+    assert called_cmd == build_ssh_check_cmd(tunnel)
+
+
+def test_check_connectivity_failure(tunnel):
+    mock_result = MagicMock()
+    mock_result.returncode = 255
+    mock_result.stderr = "Connection refused"
+    with patch("autossh_mgr.check.subprocess.run", return_value=mock_result):
+        success, msg = check_connectivity(tunnel)
+    assert success is False
+    assert "Connection refused" in msg