Add SSH jump host functionality and related configuration options

README.md

@@ -247,6 +247,8 @@ CA.
 [sync]
 device_statuses = ["active"]
 ignored_manufacturers = ["Example Manufacturer"]
+ignored_device_types = ["MX*"]
+ignored_name_patterns = ["*CORE", "TEST-*"]
 device_roles = [
   "Access Switch",
   "Core Router",
@@ -260,9 +262,26 @@ device_roles = [
   case-insensitively. An empty list includes every role.
 - `ignored_manufacturers` accepts manufacturer names, slugs, or display values,
   compared case-insensitively. An empty list excludes nothing.
-- Status filters are sent to the NetBox API. Role and manufacturer filters are
+- `ignored_device_types` contains case-insensitive glob patterns matched against
+  a device type's model, slug, and display value.
+- `ignored_name_patterns` contains case-insensitive glob patterns matched against
+  device names. `*` matches any text and `?` matches one character.
+- Status filters are sent to the NetBox API. Other inventory filters are
   applied before the cache is written.
 
+### SSH jump host
+
+```toml
+[ssh]
+jump_host = "jump-host"
+```
+
+The value may be a hostname, IP address, `user@host`, or an alias defined in
+`~/.ssh/config`. Highlight a device and press `J` to persistently enable or
+disable the jump host for it. Marked devices display `J` and are opened with
+OpenSSH ProxyJump (`ssh -J jump-host target`). SSH keys remain managed by the
+local OpenSSH client.
+
 ## First Run
 
 1. Set `url` and `api_token` in the private user `config.toml`.
@@ -289,6 +308,7 @@ required API requests and filters complete successfully.
 | `Enter` | Open a location or start SSH for a device |
 | `Ctrl+T` / `Space` | Select or unselect a device for a multi-session launch |
 | `Ctrl+U` | Clear all selected devices |
+| `J` | Enable or disable the configured jump host for a device |
 | `Esc` | Close search or return to the previous level |
 | `/` | Search all cached devices by name or primary IP |
 | `S` | Sync from NetBox |
@@ -422,6 +442,7 @@ The implementation is split by responsibility under `src/netbox_ssh`:
 - `model.py` builds and prunes the location tree.
 - `cache.py` validates and atomically writes cache version 2.
 - `manual.py` validates, stores, and merges persistent manual devices.
+- `jump_state.py` stores persistent per-device jump-host choices.
 - `tui.py` implements navigation, search, background sync, and SSH handoff.
 - `terminal.py` contains the optional multi-tab iTerm2 integration.
 

config.toml

@@ -14,8 +14,16 @@ device_statuses = ["active"]
 # Manufacturer names or slugs to exclude from synchronization.
 ignored_manufacturers = ["Example Manufacturer"]
 
+# Case-insensitive glob patterns (* and ? are supported).
+ignored_device_types = []
+ignored_name_patterns = []
+
 device_roles = [
   "Access Switch",
   "Core Router",
   "Edge Router",
 ]
+
+[ssh]
+# Hostname, IP, user@host, or an alias from ~/.ssh/config.
+jump_host = ""

src/netbox_ssh/cli.py

@@ -6,6 +6,7 @@
 from . import __version__
 from .cache import load_cache
 from .config import Config
+from .jump_state import load_jump_devices
 from .manual import load_manual_devices
 from .service import filter_device_roles
 from .tui import NetBoxSSHApp
@@ -35,10 +36,16 @@ def main(argv: list[str] | None = None) -> int:
         return 1
     try:
         manual_devices = load_manual_devices(config.manual_path)
+        jump_devices = load_jump_devices(
+            config.jump_state_path
+            or config.manual_path.with_name("jump-host-devices.json")
+        )
     except ValueError as error:
-        print(f"Manual inventory error: {error}", file=sys.stderr)
+        print(f"Local data error: {error}", file=sys.stderr)
         return 1
-    NetBoxSSHApp(config, load_cache(config.cache_path), manual_devices).run()
+    NetBoxSSHApp(
+        config, load_cache(config.cache_path), manual_devices, jump_devices
+    ).run()
     return 0
 
 

src/netbox_ssh/config.py

@@ -21,6 +21,10 @@ class Config:
     device_roles: tuple[str, ...]
     device_statuses: tuple[str, ...]
     ignored_manufacturers: tuple[str, ...]
+    ignored_device_types: tuple[str, ...] = ()
+    ignored_name_patterns: tuple[str, ...] = ()
+    jump_host: str | None = None
+    jump_state_path: Path | None = None
 
     @classmethod
     def from_env(cls) -> "Config":
@@ -47,6 +51,8 @@ def from_env(cls) -> "Config":
         file_config = _read_config(config_path)
         netbox = file_config.get("netbox", {})
         sync = file_config.get("sync", {})
+        ssh = file_config.get("ssh", {})
+        jump_host = _clean_ssh_value(ssh.get("jump_host"), "ssh.jump_host")
         return cls(
             # Zmienne powłoki celowo nadpisują ustawienia zapisane w TOML.
             netbox_url=_clean_url(os.environ.get("NETBOX_URL") or netbox.get("url")),
@@ -58,6 +64,7 @@ def from_env(cls) -> "Config":
             ),
             cache_path=cache_home / "devices.json",
             manual_path=data_home / "manual.json",
+            jump_state_path=data_home / "jump-host-devices.json",
             config_path=config_path,
             device_roles=tuple(str(role) for role in sync.get("device_roles", [])),
             device_statuses=tuple(
@@ -67,6 +74,13 @@ def from_env(cls) -> "Config":
                 str(manufacturer)
                 for manufacturer in sync.get("ignored_manufacturers", [])
             ),
+            ignored_device_types=tuple(
+                str(value) for value in sync.get("ignored_device_types", [])
+            ),
+            ignored_name_patterns=tuple(
+                str(value) for value in sync.get("ignored_name_patterns", [])
+            ),
+            jump_host=jump_host,
         )
 
     def validate_sync(self) -> None:
@@ -94,6 +108,17 @@ def _as_bool(value: str) -> bool:
     return value.lower() not in {"0", "false", "no", "off"}
 
 
+def _clean_ssh_value(value: Any, setting: str) -> str | None:
+    if value is None or not str(value).strip():
+        return None
+    result = str(value).strip()
+    if result.startswith("-") or any(character.isspace() for character in result):
+        raise ValueError(
+            f"{setting} must be a hostname, IP, or SSH alias without whitespace"
+        )
+    return result
+
+
 def _read_config(path: Path) -> dict[str, Any]:
     """Czyta opcjonalny TOML; brak pliku jest prawidłową konfiguracją domyślną."""
     try:

src/netbox_ssh/editor.py

@@ -21,9 +21,19 @@
 # Empty [] does not exclude any manufacturer.
 # Example: ignored_manufacturers = ["Cisco", "Juniper", "Arista"]
 ignored_manufacturers = []
+# Glob patterns matched case-insensitively against model, slug, and display.
+# Example: ignored_device_types = ["MX*", "ISR4451"]
+ignored_device_types = []
+# Glob patterns matched case-insensitively against device names.
+# Example: ignored_name_patterns = ["*CORE", "TEST-*"]
+ignored_name_patterns = []
 # Empty [] imports devices with every role.
 # Example: device_roles = ["Router", "Core Switch", "Distribution Switch"]
 device_roles = []
+
+[ssh]
+# Hostname, IP, user@host, or an alias from ~/.ssh/config.
+jump_host = ""
 """
 
 

src/netbox_ssh/jump_state.py

@@ -0,0 +1,40 @@
+from __future__ import annotations
+
+import json
+import os
+import tempfile
+from pathlib import Path
+
+
+def load_jump_devices(path: Path) -> set[str]:
+    """Loads stable device identifiers that should use the configured jump host."""
+    try:
+        data = json.loads(path.read_text(encoding="utf-8"))
+        if data.get("version") != 1 or not isinstance(data.get("devices"), list):
+            raise ValueError("Unsupported jump-host state format; expected version 1")
+        if not all(isinstance(value, str) and value for value in data["devices"]):
+            raise ValueError("Jump-host device identifiers must be non-empty strings")
+        return set(data["devices"])
+    except FileNotFoundError:
+        return set()
+    except (OSError, json.JSONDecodeError, AttributeError, TypeError) as error:
+        raise ValueError(f"Cannot read {path}: {error}") from error
+
+
+def save_jump_devices(path: Path, identifiers: set[str]) -> None:
+    """Atomically persists jump-host choices as private user data."""
+    payload = {"version": 1, "devices": sorted(identifiers)}
+    path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
+    os.chmod(path.parent, 0o700)
+    fd, temporary_name = tempfile.mkstemp(
+        prefix="jump-host-", suffix=".json", dir=path.parent
+    )
+    try:
+        with os.fdopen(fd, "w", encoding="utf-8") as handle:
+            json.dump(payload, handle, ensure_ascii=False, indent=2)
+            handle.write("\n")
+        os.replace(temporary_name, path)
+        os.chmod(path, 0o600)
+    finally:
+        if os.path.exists(temporary_name):
+            os.unlink(temporary_name)

src/netbox_ssh/manual.py

@@ -89,7 +89,16 @@ def merge_manual_devices(regions: list[Node], devices: list[ManualDevice]) -> li
             nodes = current.children
         assert current is not None
         current.devices.append(
-            Device(manual.name, manual.role, manual.target, source="manual")
+            Device(
+                manual.name,
+                manual.role,
+                manual.target,
+                source="manual",
+                identifier=(
+                    "manual:"
+                    + "/".join((*manual.location_path, manual.name)).casefold()
+                ),
+            )
         )
     _sort_tree(merged)
     return merged

src/netbox_ssh/model.py

@@ -12,6 +12,8 @@ class Device:
     role: str
     primary_ip: str | None = None
     source: str = "netbox"
+    identifier: str | None = None
+    use_jump_host: bool = False
 
     @property
     def ssh_target(self) -> str:
@@ -24,6 +26,7 @@ def to_dict(self) -> dict[str, Any]:
             "role": self.role,
             "primary_ip": self.primary_ip,
             "source": self.source,
+            "identifier": self.identifier,
         }
 
     @classmethod
@@ -33,6 +36,7 @@ def from_dict(cls, data: dict[str, Any]) -> "Device":
             data["role"],
             data.get("primary_ip"),
             data.get("source", "netbox"),
+            data.get("identifier"),
         )
 
 
@@ -109,7 +113,11 @@ def build_tree(
         ip = raw.get("primary_ip4") or raw.get("primary_ip6")
         if isinstance(ip, dict):
             ip = ip.get("address") or ip.get("display")
-        target_node.devices.append(Device(str(device_name), str(role_name), ip))
+        device_id = raw.get("id")
+        identifier = f"netbox:{device_id}" if device_id is not None else None
+        target_node.devices.append(
+            Device(str(device_name), str(role_name), ip, identifier=identifier)
+        )
 
     _prune_and_sort(roots)
     return roots

src/netbox_ssh/service.py

@@ -1,4 +1,4 @@
-from __future__ import annotations
+from fnmatch import fnmatchcase
 
 import httpx
 
@@ -17,6 +17,8 @@ def synchronize(config: Config) -> tuple[Cache, int]:
         regions, sites, devices = client.fetch_inventory(config.device_statuses)
     devices = filter_device_roles(devices, config.device_roles)
     devices = filter_ignored_manufacturers(devices, config.ignored_manufacturers)
+    devices = filter_ignored_device_types(devices, config.ignored_device_types)
+    devices = filter_ignored_name_patterns(devices, config.ignored_name_patterns)
     region_tree = build_tree(regions, sites, devices)
     return save_cache(config.cache_path, region_tree), len(devices)
 
@@ -57,6 +59,43 @@ def filter_ignored_manufacturers(
     return result
 
 
+def _matches_any(value: object, patterns: tuple[str, ...]) -> bool:
+    text = str(value or "").casefold()
+    return bool(text) and any(
+        fnmatchcase(text, pattern.casefold()) for pattern in patterns
+    )
+
+
+def filter_ignored_device_types(
+    devices: list[dict], ignored_patterns: tuple[str, ...]
+) -> list[dict]:
+    """Removes devices whose model, slug, or display matches a glob pattern."""
+    if not ignored_patterns:
+        return devices
+    result = []
+    for device in devices:
+        device_type = device.get("device_type") or {}
+        values = (device_type.get(field) for field in ("model", "slug", "display"))
+        if not any(_matches_any(value, ignored_patterns) for value in values):
+            result.append(device)
+    return result
+
+
+def filter_ignored_name_patterns(
+    devices: list[dict], ignored_patterns: tuple[str, ...]
+) -> list[dict]:
+    """Removes devices whose name (or display fallback) matches a glob pattern."""
+    if not ignored_patterns:
+        return devices
+    return [
+        device
+        for device in devices
+        if not _matches_any(
+            device.get("name") or device.get("display"), ignored_patterns
+        )
+    ]
+
+
 def describe_sync_error(error: Exception) -> str:
     """Zamienia techniczne wyjątki HTTP na komunikaty zrozumiałe w TUI."""
     if isinstance(error, ValueError):

src/netbox_ssh/terminal.py

@@ -32,21 +32,33 @@ def is_iterm2() -> bool:
     return platform.system() == "Darwin" and os.environ.get("TERM_PROGRAM") == "iTerm.app"
 
 
-def run_system_ssh(devices: Sequence[Device]) -> list[tuple[Device, int]]:
+def ssh_arguments(device: Device, jump_host: str | None = None) -> list[str]:
+    arguments = ["ssh"]
+    if device.use_jump_host:
+        if not jump_host:
+            raise ValueError("No SSH jump host is configured.")
+        arguments.extend(["-J", jump_host])
+    arguments.append(device.ssh_target)
+    return arguments
+
+
+def run_system_ssh(
+    devices: Sequence[Device], jump_host: str | None = None
+) -> list[tuple[Device, int]]:
     """Uruchamia systemowy OpenSSH, przenośnie także na Linuxie i WSL."""
     environment = os.environ.copy()
     environment.pop("NETBOX_API_TOKEN", None)
     environment.pop("NETBOX_URL", None)
     results = []
     for device in devices:
         result = subprocess.run(
-            ["ssh", device.ssh_target], check=False, env=environment
+            ssh_arguments(device, jump_host), check=False, env=environment
         )
         results.append((device, result.returncode))
     return results
 
 
-def open_iterm_tabs(devices: Sequence[Device]) -> None:
+def open_iterm_tabs(devices: Sequence[Device], jump_host: str | None = None) -> None:
     """Otwiera osobną kartę iTerm2 dla każdego urządzenia.
 
     Polecenie SSH jest cytowane jako pojedynczy argument powłoki, a sam
@@ -58,7 +70,7 @@ def open_iterm_tabs(devices: Sequence[Device]) -> None:
     if not is_iterm2():
         raise RuntimeError("Opening multiple sessions requires iTerm2 on macOS.")
 
-    commands = [f"ssh {shlex.quote(device.ssh_target)}" for device in devices]
+    commands = [shlex.join(ssh_arguments(device, jump_host)) for device in devices]
     environment = os.environ.copy()
     environment.pop("NETBOX_API_TOKEN", None)
     environment.pop("NETBOX_URL", None)