[SUPPORT] - Cannot get output from APC devices

[desultory]

Are you using paramiko as a client or server?

Client

What feature(s) aren't working right?

Something else

What version(s) of paramiko are you using?

3.4.0

What version(s) of Python are you using?

Python 3.11.8

What operating system and version are you using?

Gentoo 23.0

If you're connecting as a client, which SSH server are you connecting to?

APC

If you're using paramiko as part of another tool, which tool/version?

No response

What are you trying to do with paramiko?

I'm trying to use paramiko to connect to some APC devices.

How are you trying to do it, and what's happening instead?

I want to read the information displayed when logged in, then act on that. Currently I can't get any output:

#!/usr/bin/python3


from zenlib.util import get_kwargs
from zenlib.logging import loggify


@loggify
class APCClient:
    def __init__(self, host, username='apc', password='apc'):
        self.host = host
        self.username = username
        self.password = password
        self.connect()

    def connect(self):
        from paramiko.client import SSHClient
        from paramiko.client import AutoAddPolicy
        self.client = SSHClient()
        self.client.set_missing_host_key_policy(AutoAddPolicy())
        self.client.connect(self.host, username=self.username, password=self.password, allow_agent=False, look_for_keys=False)
        self.logger.info("Connected to %s as %s", self.host, self.username)
        self.logger.debug(self.client.get_transport().get_banner())

    def exec(self, command):
        self.logger.debug("Executing command: %s", command)
        stdin, stdout, stderr = self.client.exec_command(command)
        stdout = stdout.readlines()
        stderr = stderr.readlines()
        self.logger.debug("Command output: %s", stdout)
        self.logger.debug("Command error: %s", stderr)
        return stdout


def main():
    args = [{'flags': ['-u', '--username'], 'action': 'store', 'help': 'Username for the device', 'default': 'apc'},
            {'flags': {'-p', '--password'}, 'action': 'store', 'help': 'Password for the device', 'default': 'apc'},
            {'flags': ['host'], 'action': 'store', 'help': 'Host to connect to'}]
    kwargs = get_kwargs(package='apc', description="APC device configurator", arguments=args)

    client = APCClient(**kwargs)
    client.exec('?')


if __name__ == '__main__':
    main()

DEBUG    | apc.APCClient                              | None
DEBUG    | apc.APCClient                              | Executing command: ?
DEBUG    | apc.APCClient                              | Command output: []
DEBUG    | apc.APCClient                              | Command error: []

Anything else?

No response

[desultory]

I was able to get it working by using the following code:

#!/usr/bin/python3

from time import sleep
from zenlib.util import get_kwargs
from zenlib.logging import loggify


@loggify
class APCClient:
    def __init__(self, host, username='apc', password='apc'):
        self.host = host
        self.username = username
        self.password = password
        self.connect()

    def connect(self):
        from paramiko.client import SSHClient
        from paramiko.client import AutoAddPolicy

        self.client = SSHClient()
        self.client.set_missing_host_key_policy(AutoAddPolicy())
        self.client.connect(self.host, username=self.username, password=self.password, allow_agent=False, look_for_keys=False)
        self.logger.info("Connected to %s as %s", self.host, self.username)
        self.shell = self.client.invoke_shell()
        self.wait_for_response()
        self.process_banner(self.shell.recv(4096).decode('utf-8'))

    def wait_for_response(self):
        while not self.shell.exit_status_ready():
            if self.shell.recv_ready():
                break
            sleep(1)

    def process_banner(self, banner):
        """ Process the banner returned by the device. """
        from re import search, MULTILINE
        self.logger.debug("Processing banner: %s", banner)

        self.stat_string = search(r'Stat : (.*)$', banner, MULTILINE).group(1)
        self.name = search(r'Name\s+: (.*?)\s', banner, MULTILINE).group(1)

        if search(r'SNMPv3\s+: Enabled', banner, MULTILINE):
            self.logger.debug("SNMPv3 is enabled.")
            self.snmpv3 = True

        self.snmpv3 = False if search(r'SNMPv3\s+: Disabled', banner, MULTILINE) else True
        self.corrupted = False if 'A-' not in self.stat_string else True

        self.enable_snmpv3()

    def enable_snmpv3(self):
        if self.snmpv3:
            self.logger.debug("SNMPv3 is already enabled.")
            return

        self.exec('snmpv3 -S enable')
        self.logger.info("SNMPv3 enabled.")
        self.snmpv3 = True

    def exec(self, command):
        self.logger.debug("Executing command: %s", command)
        if not command.endswith('\r\n'):
            command += '\r\n'
        self.shell.sendall(command)
        self.wait_for_response()
        output = self.shell.recv(4096).decode('utf-8')[len(command):]
        if not output.split(': ')[1][:7] == 'Success':
            raise Exception("Command failed: %s", output)
        return output


def main():
    args = [{'flags': ['-u', '--username'], 'action': 'store', 'help': 'Username for the device', 'default': 'apc'},
            {'flags': {'-p', '--password'}, 'action': 'store', 'help': 'Password for the device', 'default': 'apc'},
            {'flags': ['host'], 'action': 'store', 'help': 'Host to connect to'}]
    kwargs = get_kwargs(package='node_configurator', description="APC device configurator", arguments=args)

    APCClient(**kwargs)


if __name__ == '__main__':
    main()

I'm not sure if I need the shell mode, I'd prefer not to use it. I wasn't even able to read the banner without it. Is that wait method acceptable?