Skip to content

netops.core — Core Modules

Connection management, device inventory, and credential vault.


netops.core.connection

Unified device connection manager. Handles SSH, SSH2, and Telnet connections with a single interface. Uses Netmiko under the hood for vendor-aware CLI interaction.

connection

Unified device connection manager.

Handles SSH, SSH2, and Telnet connections with a single interface. Uses Netmiko under the hood for vendor-aware CLI interaction.

Classes
Transport

Bases: Enum

Supported connection transports for device communication.

AuthMethod

Bases: Enum

Authentication mechanisms accepted when connecting to a device.

JumpHostParams dataclass
JumpHostParams(host: str, username: str | None = None, password: str | None = None, auth_method: AuthMethod = AuthMethod.PASSWORD, port: int = 22, key_file: str | None = None, key_passphrase: str | None = None, timeout: int = 30)

Connection parameters for an SSH jump box / bastion used purely as network transport.

The jump host is never sent netops-toolkit CLI commands - it is only used to open an SSH direct-tcpip channel that Netmiko's ConnectHandler treats as a pre-established socket (via the sock= kwarg) when talking to the real target device. See docs/guides/jump-host-tunnel.md.

ConnectionParams dataclass
ConnectionParams(host: str, username: str | None = None, password: str | None = None, transport: Transport = Transport.SSH, auth_method: AuthMethod = AuthMethod.PASSWORD, port: int | None = None, key_file: str | None = None, device_type: str = 'autodetect', timeout: int = 30, enable_password: str | None = None, jump_host: JumpHostParams | None = None, extras: dict = dict())

Everything needed to connect to a device.

Attributes
effective_port property
effective_port: int

Return the resolved TCP port (explicit override, or 23 for Telnet, 22 otherwise).

DeviceConnection
DeviceConnection(params: ConnectionParams)

Unified connection to a network device.

Usage: params = ConnectionParams(host="10.0.0.1", username="admin", password="secret") with DeviceConnection(params) as conn: output = conn.send("show version") config = conn.send("show running-config")

Initialise the connection manager with the given connection parameters.

Methods:
__enter__
__enter__() -> DeviceConnection

Connect on entering the context manager block.

__exit__
__exit__(exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None) -> None

Disconnect when leaving the context manager block.

connect
connect() -> None

Establish connection using configured transport.

disconnect
disconnect() -> None

Close the connection.

Each resource is released in its own finally block so that an exception from Netmiko's disconnect() (or from closing one resource) cannot prevent the others from being cleaned up.

send
send(command: str, expect_string: str | None = None) -> str

Send a command and return output.

send_config
send_config(commands: list[str]) -> str

Send configuration commands.

Functions:
resolve_jump_host_params
resolve_jump_host_params(jump_host: str | None, jump_port: int = 22, jump_username: str | None = None, jump_password: str | None = None, jump_key_file: str | None = None, jump_key_passphrase: str | None = None, vault: object | None = None) -> JumpHostParams | None

Build :class:JumpHostParams from inventory-declared jump-host fields.

Returns None if jump_host is not set (the common, non-tunneled case). Jump-box credentials are resolved through the same :class:~netops.core.vault.CredentialVault lookup used for device credentials (env vars -> device entry -> group -> default), keyed by the jump host's own hostname so operators can store bastion creds once and reuse them across every device that tunnels through it. If vault is an unlocked CredentialVault instance and no explicit username/key is given, its credentials for the jump host are used.

jump_host_from_inventory
jump_host_from_inventory(device: object, vault: object | None = None) -> JumpHostParams | None

Translate a :class:netops.core.inventory.Device's bastion fields.

Keeping this mapping at the connection boundary means every inventory-driven command uses exactly the same tunnel semantics rather than independently rebuilding a partial set of jump-host parameters.


netops.core.inventory

Device inventory management. Simple YAML/JSON inventory that maps to Ansible inventory format.

CLI usage:

python -m netops.core.inventory export --format ansible --output ansible_inventory.yaml
python -m netops.core.inventory export --format ansible-json --output ansible_inventory.json

inventory

Device inventory management.

Simple YAML/JSON inventory that maps to Ansible inventory format.

CLI usage::

python -m netops.core.inventory export --format ansible --output ansible_inventory.yaml
python -m netops.core.inventory export --format ansible-json --output ansible_inventory.json
Classes
Device dataclass
Device(hostname: str, host: str, vendor: str, transport: str = 'ssh', port: int | None = None, username: str | None = None, password: str | None = None, enable_password: str | None = None, key_file: str | None = None, groups: list[str] = list(), tags: dict[str, str] = dict(), site: str | None = None, role: str | None = None, jump_host: str | None = None, jump_port: int = 22, jump_username: str | None = None, jump_password: str | None = None, jump_key_file: str | None = None, jump_key_passphrase: str | None = None)

A network device in the inventory.

Methods:
to_dict
to_dict() -> dict

Return a dict representation of the device, omitting None fields.

Inventory
Inventory()

Device inventory with group support.

Supports: - YAML/JSON file loading - Group-based filtering - Tag-based filtering - Export to Ansible inventory format

Initialise an empty inventory with no devices, groups, or defaults.

Methods:
add
add(device: Device) -> None

Add device to the inventory, registering it under all its groups.

get
get(hostname: str) -> Device | None

Look up a device by hostname; returns None if not found.

filter
filter(group: str | None = None, vendor: str | None = None, role: str | None = None, site: str | None = None, tag: tuple | None = None) -> list[Device]

Filter devices by criteria.

from_file classmethod
from_file(path: str | Path) -> Inventory

Load inventory from YAML or JSON file.

to_ansible
to_ansible() -> dict

Export as Ansible inventory format (JSON-compatible dict).

The returned structure follows the Ansible JSON inventory spec: https://docs.ansible.com/ansible/latest/dev_guide/developing_inventory.html

to_ansible_yaml
to_ansible_yaml() -> str

Export as Ansible inventory in YAML format.

to_ansible_json
to_ansible_json() -> str

Export as Ansible inventory in JSON format.

to_file
to_file(path: str | Path, format: str = 'yaml') -> None

Save inventory to file.

Functions:
main
main() -> None

CLI entry point: python -m netops.core.inventory export ...


netops.core.vault

Credential vault — encrypted storage for device credentials.

Stores per-device, per-group, and default credentials in an AES-256-GCM encrypted YAML file. The encryption key is derived from a master password using PBKDF2-HMAC-SHA256.

Lookup order for CredentialVault.get_credentials: 1. Environment variables (NETOPS_CRED_<HOSTNAME>_USER / _PASS / _ENABLE) 2. Device-specific entry 3. First matching group entry 4. Default entry

CLI usage:

python -m netops.core.vault init [--vault VAULT_FILE]
python -m netops.core.vault set --device HOSTNAME --user USER [--vault VAULT_FILE]
python -m netops.core.vault set --group  GROUP   --user USER [--vault VAULT_FILE]
python -m netops.core.vault set --default        --user USER [--vault VAULT_FILE]
python -m netops.core.vault get --device HOSTNAME            [--vault VAULT_FILE]
python -m netops.core.vault delete --device HOSTNAME         [--vault VAULT_FILE]

The global --vault FILE option must appear before the vault subcommand, for example python -m netops.core.vault --vault secrets.yaml set --default --user admin.

netops.core.bastion

The active-bastion CLI is exposed through the dispatcher:

netops bastion connect --host BASTION --username USER [--password-stdin]
netops bastion status
netops bastion disconnect

See Active Bastion Routing for the operational model and protocol limitations.

vault

Credential vault — encrypted storage for device credentials.

Stores per-device, per-group, and default credentials in an AES-256-GCM encrypted YAML file. The encryption key is derived from a master password using PBKDF2-HMAC-SHA256.

Lookup order for :meth:CredentialVault.get_credentials:

  1. Environment variables (NETOPS_CRED_<HOSTNAME>_USER / _PASS / _ENABLE)
  2. Device-specific entry
  3. First matching group entry
  4. Default entry

Environment variable names are normalised: hyphens and dots in the hostname are replaced with underscores and the whole name is upper-cased, e.g. core-rtr-01NETOPS_CRED_CORE_RTR_01_USER.

CLI usage::

python -m netops.core.vault init [--vault VAULT_FILE]
python -m netops.core.vault set --device HOSTNAME --user USER [--vault VAULT_FILE]
python -m netops.core.vault set --group  GROUP   --user USER [--vault VAULT_FILE]
python -m netops.core.vault set --default        --user USER [--vault VAULT_FILE]
python -m netops.core.vault get --device HOSTNAME            [--vault VAULT_FILE]
python -m netops.core.vault delete --device HOSTNAME         [--vault VAULT_FILE]
python -m netops.core.vault delete --group  GROUP            [--vault VAULT_FILE]
python -m netops.core.vault delete --default                 [--vault VAULT_FILE]

The master password may be provided via the NETOPS_VAULT_PASSWORD environment variable to avoid interactive prompts (useful in CI pipelines).

Classes
CredentialVault
CredentialVault(vault_path: str | Path | None = None)

Encrypted credential store backed by a YAML file.

Parameters:

Name Type Description Default
vault_path str | Path | None

Path to the vault file (will be created by :meth:init).

None

Initialise with an optional vault file path (defaults to ~/.netops/vault.yaml).

Methods:
init
init(password: str) -> None

Create a new, empty vault protected by password.

Raises :class:FileExistsError if the vault already exists.

unlock
unlock(password: str) -> None

Decrypt and load the vault. Must be called before any read/write operation.

save
save(password: str) -> None

Re-derive the key from password, then encrypt and persist the vault.

Call this after :meth:unlock to persist any changes made in memory.

set_device
set_device(hostname: str, username: str, password: str, enable_password: str | None = None) -> None

Store credentials for a specific device hostname.

set_group
set_group(group: str, username: str, password: str, enable_password: str | None = None) -> None

Store credentials for all devices in group.

set_default
set_default(username: str, password: str, enable_password: str | None = None) -> None

Store fallback credentials used when no device or group entry matches.

delete_device
delete_device(hostname: str) -> bool

Remove the device entry for hostname.

Returns:

Type Description
bool

True if the entry existed and was removed.

delete_group
delete_group(group: str) -> bool

Remove the group entry for group.

Returns:

Type Description
bool

True if the entry existed and was removed.

delete_default
delete_default() -> bool

Clear the default credentials entry.

Returns:

Type Description
bool

True if the entry existed and was removed.

get_credentials
get_credentials(hostname: str, groups: list[str] | None = None) -> dict | None

Return a credentials dict for hostname, or None if nothing matches.

Lookup priority:

  1. Environment variables (NETOPS_CRED_<HOSTNAME>_USER, _PASS, _ENABLE)
  2. Device-specific vault entry
  3. First matching group vault entry
  4. Default vault entry

Returns:

Type Description
dict or None

Credentials dict with username and password keys (and optionally enable_password), or None if nothing matches.

Functions:
main
main(argv: list[str] | None = None) -> int

CLI entry point for the credential vault management tool.