Communicate with a KVM/QEMU-VM via serial console

readme.org

How to use serial_comm.py

Connect to a console

URI = "qemu+ssh://user@192.168.178.180/system?keyfile=/home/user/.ssh/id_rsa"
domain = "d10c"
user = "root"
pw = "pwd"

sc = SerialComm(URI, domain, user, pw)
sc.connect()
print(sc.conn)
sc.disconnect(logout=True, close=True)

Execute a command

URI = "qemu+ssh://user01@192.168.178.180/system?keyfile=/home/user01/.ssh/id_rsa"
domain = "d10c"
user = "root"
pw = "pwd"

sc = SerialComm(URI, domain, user, pw)
sc.connect()
username = sc.send_cmd("whoami", waitfor=["\r\n\w+\r\n"])
print(username)
sc.disconnect(logout=True, close=True)

serial_comm.py

import pexpect

class CustomCmd:

    def __init__(self, cmd, waitfor):
        self.cmd = cmd
        self.waitfor = waitfor


class SerialComm:
    split_chars = "\r\n"
    std_prompt = [".*~.*"]

    def __init__(self, uri, domain, user, pw):
        self.uri = uri
        self.domain = domain
        self._user = user
        self._password = pw

        self.conn = None
        self._connected = False
        self._logged_in = False

    def _login(self):
        try:
            self.conn.sendline(f"{self._user}")
            self.conn.expect(".*assword:.*")
            self.conn.sendline(f"{self._password}")
            self.conn.expect(self.std_prompt)
            self._logged_in = True
        except pexpect.ExceptionPexpect as e:
            raise(e)

    def disconnect(self, logout=False, close=False):
        if logout:
            self.conn.sendline("exit\r")
            self._logged_in = False

        self.conn.sendline("^]")
        self._connected = False

        if close:
            self.conn.close()
            self.conn = None

    def connect(self):
        if not self.conn:
            self.conn = pexpect.spawn(f"virsh -c {self.uri} console {self.domain}")
        self.conn.expect(".*]\)")
        self.conn.sendline("\n")

        try:
            i = self.conn.expect([".*login: ", ".*~.*"])
            self._connected = True
            if i == 0:
                self._login()
        except pexpect.ExceptionPexpect as e:
            raise(e)

    def send_cmd(self, cmd, waitfor=[".*~.*"]):
        # Check states
        if not self._connected:
            self.connect()
        if not self._logged_in:
            self._login()

        try:
            self.conn.sendline(f"{cmd}\r")
            self.conn.expect(waitfor)
            output = self.conn.after
            return [l for l in output.decode("utf-8").split(self.split_chars) if l != "" ]
        except pexpect.ExceptionPexpect as e:
            return None