Skip to content

SSH API

SSH connection management for remote OpenWRT devices.

SSHConnection

SSHConnection

Manages SSH connections to OpenWRT devices.

Source code in src/wrtkit/ssh.py 
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
class SSHConnection:
    """Manages SSH connections to OpenWRT devices."""

    def __init__(
        self,
        host: str,
        port: int = 22,
        username: str = "root",
        password: Optional[str] = None,
        key_filename: Optional[str] = None,
        timeout: int = 30,
    ):
        """
        Initialize SSH connection parameters.

        Args:
            host: The hostname or IP address of the OpenWRT device
            port: SSH port (default: 22)
            username: SSH username (default: root)
            password: SSH password
            key_filename: Path to SSH private key file
            timeout: Connection timeout in seconds
        """
        self.host = host
        self.port = port
        self.username = username
        self.password = password
        self.key_filename = key_filename
        self.timeout = timeout
        self._client: Optional[paramiko.SSHClient] = None

    def connect(self) -> None:
        """Establish SSH connection to the device."""
        if self._client is not None:
            return

        self._client = paramiko.SSHClient()
        self._client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

        try:
            self._client.connect(
                hostname=self.host,
                port=self.port,
                username=self.username,
                password=self.password,
                key_filename=self.key_filename,
                timeout=self.timeout,
            )
        except Exception as e:
            self._client = None
            raise ConnectionError(f"Failed to connect to {self.host}: {e}")

    def disconnect(self) -> None:
        """Close the SSH connection."""
        if self._client is not None:
            self._client.close()
            self._client = None

    def execute(self, command: str) -> Tuple[str, str, int]:
        """
        Execute a command on the remote device.

        Args:
            command: The command to execute

        Returns:
            Tuple of (stdout, stderr, exit_code)
        """
        if self._client is None:
            self.connect()

        stdin, stdout, stderr = self._client.exec_command(command)
        exit_code = stdout.channel.recv_exit_status()

        return (
            stdout.read().decode("utf-8"),
            stderr.read().decode("utf-8"),
            exit_code,
        )

    def execute_uci_command(self, command: str) -> Tuple[str, str, int]:
        """
        Execute a UCI command on the remote device.

        Args:
            command: The UCI command to execute

        Returns:
            Tuple of (stdout, stderr, exit_code)
        """
        return self.execute(command)

    def get_uci_config(self, package: str) -> str:
        """
        Retrieve the current UCI configuration for a package.

        Args:
            package: The UCI package name (e.g., 'network', 'wireless')

        Returns:
            The UCI configuration as a string
        """
        stdout, stderr, exit_code = self.execute(f"uci export {package}")
        if exit_code != 0:
            raise RuntimeError(f"Failed to get UCI config for {package}: {stderr}")
        return stdout

    def commit_changes(self, packages: Optional[List[str]] = None) -> None:
        """
        Commit UCI changes.

        Args:
            packages: List of packages to commit. If None, commits all changes.
        """
        if packages:
            for package in packages:
                stdout, stderr, exit_code = self.execute(f"uci commit {package}")
                if exit_code != 0:
                    raise RuntimeError(f"Failed to commit {package}: {stderr}")
        else:
            stdout, stderr, exit_code = self.execute("uci commit")
            if exit_code != 0:
                raise RuntimeError(f"Failed to commit changes: {stderr}")

    def reload_config(
        self,
        reload_dhcp: bool = True,
        changed_packages: Optional[Set[str]] = None,
    ) -> List[str]:
        """
        Reload network configuration and wireless settings.

        Only restarts services related to packages that actually changed.

        Args:
            reload_dhcp: If True, also restart dnsmasq to apply DHCP changes
                (only when dhcp package changed or changed_packages is None)
            changed_packages: Set of package names that have changes.
                If None, restarts all services (legacy behavior).
                If empty set, restarts nothing.

        Returns:
            List of commands that were executed.
        """
        # If changed_packages is empty set, nothing to restart
        if changed_packages is not None and len(changed_packages) == 0:
            return []

        commands: List[str] = []

        # Determine which services need restart based on changed packages
        if changed_packages is None:
            # Legacy behavior: restart all
            commands.append("/etc/init.d/network restart")
            commands.append("wifi reload")
            if reload_dhcp:
                commands.append("/etc/init.d/dnsmasq restart")
        else:
            # Package-aware restart
            # network and sqm changes require network restart
            if "network" in changed_packages or "sqm" in changed_packages:
                commands.append("/etc/init.d/network restart")

            # wireless changes require wifi reload
            if "wireless" in changed_packages:
                commands.append("wifi reload")

            # dhcp changes require dnsmasq restart
            if "dhcp" in changed_packages and reload_dhcp:
                commands.append("/etc/init.d/dnsmasq restart")

            # firewall changes require firewall reload
            if "firewall" in changed_packages:
                commands.append("/etc/init.d/firewall reload")

        # Execute commands (deduplicated by using list, no duplicates added above)
        for cmd in commands:
            stdout, stderr, exit_code = self.execute(cmd)
            if exit_code != 0:
                print(f"Warning: {cmd} returned non-zero exit code: {stderr}")
            time.sleep(1)

        return commands

    def __enter__(self) -> "SSHConnection":
        """Context manager entry."""
        self.connect()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb) -> None:
        """Context manager exit."""
        self.disconnect()

__init__(host, port=22, username='root', password=None, key_filename=None, timeout=30)

Initialize SSH connection parameters.

Parameters:

Name Type Description Default
host str

The hostname or IP address of the OpenWRT device

 required
port int

SSH port (default: 22)

 22
username str

SSH username (default: root)

 'root'
password Optional[str]

SSH password

 None
key_filename Optional[str]

Path to SSH private key file

 None
timeout int

Connection timeout in seconds

 30
Source code in src/wrtkit/ssh.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
def __init__(
    self,
    host: str,
    port: int = 22,
    username: str = "root",
    password: Optional[str] = None,
    key_filename: Optional[str] = None,
    timeout: int = 30,
):
    """
    Initialize SSH connection parameters.

    Args:
        host: The hostname or IP address of the OpenWRT device
        port: SSH port (default: 22)
        username: SSH username (default: root)
        password: SSH password
        key_filename: Path to SSH private key file
        timeout: Connection timeout in seconds
    """
    self.host = host
    self.port = port
    self.username = username
    self.password = password
    self.key_filename = key_filename
    self.timeout = timeout
    self._client: Optional[paramiko.SSHClient] = None

connect()

Establish SSH connection to the device.

Source code in src/wrtkit/ssh.py 
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
def connect(self) -> None:
    """Establish SSH connection to the device."""
    if self._client is not None:
        return

    self._client = paramiko.SSHClient()
    self._client.set_missing_host_key_policy(paramiko.AutoAddPolicy())

    try:
        self._client.connect(
            hostname=self.host,
            port=self.port,
            username=self.username,
            password=self.password,
            key_filename=self.key_filename,
            timeout=self.timeout,
        )
    except Exception as e:
        self._client = None
        raise ConnectionError(f"Failed to connect to {self.host}: {e}")

disconnect()

Close the SSH connection.

Source code in src/wrtkit/ssh.py 
60
61
62
63
64
def disconnect(self) -> None:
    """Close the SSH connection."""
    if self._client is not None:
        self._client.close()
        self._client = None

execute(command)

Execute a command on the remote device.

Parameters:

Name Type Description Default
command str

The command to execute

 required

Returns:

Type Description
Tuple[str, str, int]

Tuple of (stdout, stderr, exit_code)
Source code in src/wrtkit/ssh.py 
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
def execute(self, command: str) -> Tuple[str, str, int]:
    """
    Execute a command on the remote device.

    Args:
        command: The command to execute

    Returns:
        Tuple of (stdout, stderr, exit_code)
    """
    if self._client is None:
        self.connect()

    stdin, stdout, stderr = self._client.exec_command(command)
    exit_code = stdout.channel.recv_exit_status()

    return (
        stdout.read().decode("utf-8"),
        stderr.read().decode("utf-8"),
        exit_code,
    )

execute_uci_command(command)

Execute a UCI command on the remote device.

Parameters:

Name Type Description Default
command str

The UCI command to execute

 required

Returns:

Type Description
Tuple[str, str, int]

Tuple of (stdout, stderr, exit_code)
Source code in src/wrtkit/ssh.py 
88
89
90
91
92
93
94
95
96
97
98
def execute_uci_command(self, command: str) -> Tuple[str, str, int]:
    """
    Execute a UCI command on the remote device.

    Args:
        command: The UCI command to execute

    Returns:
        Tuple of (stdout, stderr, exit_code)
    """
    return self.execute(command)

get_uci_config(package)

Retrieve the current UCI configuration for a package.

Parameters:

Name Type Description Default
package str

The UCI package name (e.g., 'network', 'wireless')

 required

Returns:

Type Description
str

The UCI configuration as a string
Source code in src/wrtkit/ssh.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def get_uci_config(self, package: str) -> str:
    """
    Retrieve the current UCI configuration for a package.

    Args:
        package: The UCI package name (e.g., 'network', 'wireless')

    Returns:
        The UCI configuration as a string
    """
    stdout, stderr, exit_code = self.execute(f"uci export {package}")
    if exit_code != 0:
        raise RuntimeError(f"Failed to get UCI config for {package}: {stderr}")
    return stdout

commit_changes(packages=None)

Commit UCI changes.

Parameters:

Name Type Description Default
packages Optional[List[str]]

List of packages to commit. If None, commits all changes.

 None
Source code in src/wrtkit/ssh.py 
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
def commit_changes(self, packages: Optional[List[str]] = None) -> None:
    """
    Commit UCI changes.

    Args:
        packages: List of packages to commit. If None, commits all changes.
    """
    if packages:
        for package in packages:
            stdout, stderr, exit_code = self.execute(f"uci commit {package}")
            if exit_code != 0:
                raise RuntimeError(f"Failed to commit {package}: {stderr}")
    else:
        stdout, stderr, exit_code = self.execute("uci commit")
        if exit_code != 0:
            raise RuntimeError(f"Failed to commit changes: {stderr}")

reload_config(reload_dhcp=True, changed_packages=None)

Reload network configuration and wireless settings.

Only restarts services related to packages that actually changed.

Parameters:

Name Type Description Default
reload_dhcp bool

If True, also restart dnsmasq to apply DHCP changes (only when dhcp package changed or changed_packages is None)

 True
changed_packages Optional[Set[str]]

Set of package names that have changes. If None, restarts all services (legacy behavior). If empty set, restarts nothing.

 None

Returns:

Type Description
List[str]

List of commands that were executed.
Source code in src/wrtkit/ssh.py 
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
def reload_config(
    self,
    reload_dhcp: bool = True,
    changed_packages: Optional[Set[str]] = None,
) -> List[str]:
    """
    Reload network configuration and wireless settings.

    Only restarts services related to packages that actually changed.

    Args:
        reload_dhcp: If True, also restart dnsmasq to apply DHCP changes
            (only when dhcp package changed or changed_packages is None)
        changed_packages: Set of package names that have changes.
            If None, restarts all services (legacy behavior).
            If empty set, restarts nothing.

    Returns:
        List of commands that were executed.
    """
    # If changed_packages is empty set, nothing to restart
    if changed_packages is not None and len(changed_packages) == 0:
        return []

    commands: List[str] = []

    # Determine which services need restart based on changed packages
    if changed_packages is None:
        # Legacy behavior: restart all
        commands.append("/etc/init.d/network restart")
        commands.append("wifi reload")
        if reload_dhcp:
            commands.append("/etc/init.d/dnsmasq restart")
    else:
        # Package-aware restart
        # network and sqm changes require network restart
        if "network" in changed_packages or "sqm" in changed_packages:
            commands.append("/etc/init.d/network restart")

        # wireless changes require wifi reload
        if "wireless" in changed_packages:
            commands.append("wifi reload")

        # dhcp changes require dnsmasq restart
        if "dhcp" in changed_packages and reload_dhcp:
            commands.append("/etc/init.d/dnsmasq restart")

        # firewall changes require firewall reload
        if "firewall" in changed_packages:
            commands.append("/etc/init.d/firewall reload")

    # Execute commands (deduplicated by using list, no duplicates added above)
    for cmd in commands:
        stdout, stderr, exit_code = self.execute(cmd)
        if exit_code != 0:
            print(f"Warning: {cmd} returned non-zero exit code: {stderr}")
        time.sleep(1)

    return commands

Usage Example

from wrtkit import SSHConnection

# Create connection
ssh = SSHConnection(
    host="192.168.1.1",
    username="root",
    password="password"
)

# Connect
ssh.connect()

# Execute command
stdout, stderr, exit_code = ssh.execute("uci show network")

# Get UCI config
config = ssh.get_uci_config("network")

# Commit and reload
ssh.commit_changes()
ssh.reload_config()

# Disconnect
ssh.disconnect()

# Or use context manager
with SSHConnection(host="192.168.1.1", username="root", password="pass") as ssh:
    config = ssh.get_uci_config("network")

See Also