Filter out command that needs a terminal in Python subprocess module

Question

I am developing a robot that accepts commands from network (XMPP) and uses subprocess module in Python to execute them and sends back the output of commands. Essentially it is an SSH-like XMPP-based non-interactive shell.

The robot only executes commands from authenticated trusted sources, so arbitrary shell commands are allowed (shell=True).

However, when I accidentally send some command that needs a tty, the robot is stuck.

For example:

subprocess.check_output(['vim'], shell=False)
subprocess.check_output('vim', shell=True)

Should each of the above commands is received, the robot is stuck, and the terminal from which the robot is run, is broken.

Though the robot only receives commands from authenticated trusted sources, human errs. How could I make the robot filter out those commands that will break itself? I know there is os.isatty but how could I utilize it? Is there a way to detect those "bad" commands and refuse to execute them?

TL;DR:

Say, there are two kinds of commands:

• Commands like ls: does not need a tty to run.
• Commands like vim: needs a tty; breaks subprocess if no tty is given.

How could I tell a command is ls-like or is vim-like and refuses to run the command if it is vim-like?

Answer 1

What you expect is a function that receives command as input, and returns meaningful output by running the command.

Since the command is arbitrary, requirement for tty is just one of many bad cases may happen (other includes running a infinite loop), your function should only concern about its running period, in other words, a command is “bad” or not should be determined by if it ends in a limited time or not, and since subprocess is asynchronous by nature, you can just run the command and handle it in a higher vision.

Demo code to play, you can change the cmd value to see how it performs differently:

#!/usr/bin/env python
# coding: utf-8

import time
import subprocess
from subprocess import PIPE


#cmd = ['ls']
#cmd = ['sleep', '3']
cmd = ['vim', '-u', '/dev/null']

print 'call cmd'
p = subprocess.Popen(cmd, shell=True,
                     stdin=PIPE, stderr=PIPE, stdout=PIPE)
print 'called', p

time_limit = 2
timer = 0
time_gap = 0.2

ended = False
while True:
    time.sleep(time_gap)

    returncode = p.poll()
    print 'process status', returncode

    timer += time_gap
    if timer >= time_limit:
        print 'timeout, kill process'
        p.kill()
        break

    if returncode is not None:
        ended = True
        break

if ended:
    print 'process ended by', returncode

    print 'read'
    out, err = p.communicate()
    print 'out', repr(out)
    print 'error', repr(err)
else:
    print 'process failed'

Three points are notable in the above code:

1. We use Popen instead of check_output to run the command, unlike check_output which will wait for the process to end, Popen returns immediately, thus we can do further things to control the process.

2. We implement a timer to check for the process's status, if it runs for too long, we killed it manually because we think a process is not meaningful if it could not end in a limited time. In this way your original problem will be solved, as vim will never end and it will definitely being killed as an “unmeaningful” command.

3. After the timer helps us filter out bad commands, we can get stdout and stderr of the command by calling communicate method of the Popen object, after that its your choice to determine what to return to the user.

Conclusion

tty simulation is not needed, we should run the subprocess asynchronously, then control it by a timer to determine whether it should be killed or not, for those ended normally, its safe and easy to get the output.

Answer 2

Well, SSH is already a tool that will allow users to remotely execute commands and be authenticated at the same time. The authentication piece is extremely tricky, please be aware that building the software you're describing is a bit risky from a security perspective.

There isn't a way to determine whether a process is going to need a tty or not. And there's no os.isatty method because if you ran a sub-processes that needed one wouldn't mean that there was one. :)

In general, it would probably be safer from a security perspective and also a solution to this problem if you were to consider a white list of commands. You could choose that white list to avoid things that would need a tty, because I don't think you'll easily get around this.

Answer 3

Thanks a lot for @J.F. Sebastia's help (see comments under the question), I've found a solution (workaround?) for my case.

The reason why vim breaks terminal while ls does not, is that vim needs a tty. As Sebastia says, we can feed vim with a pty using pty.openpty(). Feeding a pty gurantees the command will not break terminal, and we can add a timout to auto-kill such processes. Here is (dirty) working example:

#!/usr/bin/env python3

import pty
from subprocess import STDOUT, check_output, TimeoutExpired

master_fd, slave_fd = pty.openpty()


try:
    output1 = check_output(['ls', '/'], stdin=slave_fd, stderr=STDOUT, universal_newlines=True, timeout=3)
    print(output1)
except TimeoutExpired:
    print('Timed out')

try:
    output2 = check_output(['vim'], stdin=slave_fd, stderr=STDOUT, universal_newlines=True, timeout=3)
    print(output2)
except TimeoutExpired:
    print('Timed out')

Note it is stdin that we need to take care of, not stdout or stderr.

Answer 4

You can refer to my answer in: https://stackoverflow.com/a/43012138/3555925, which use pseudo-terminal to make stdout no-blocking, and use select in handle stdin/stdout.

I can just modify the command var to 'vim'. And the script is working fine.

#!/usr/bin/env python
# -*- coding: utf-8 -*-

import os
import sys
import select
import termios
import tty
import pty
from subprocess import Popen

command = 'vim'

# save original tty setting then set it to raw mode
old_tty = termios.tcgetattr(sys.stdin)
tty.setraw(sys.stdin.fileno())

# open pseudo-terminal to interact with subprocess
master_fd, slave_fd = pty.openpty()

# use os.setsid() process the leader of a new session, or bash job control will not be enabled
p = Popen(command,
          preexec_fn=os.setsid,
          stdin=slave_fd,
          stdout=slave_fd,
          stderr=slave_fd,
          universal_newlines=True)

while p.poll() is None:
    r, w, e = select.select([sys.stdin, master_fd], [], [])
    if sys.stdin in r:
        d = os.read(sys.stdin.fileno(), 10240)
        os.write(master_fd, d)
    elif master_fd in r:
        o = os.read(master_fd, 10240)
        if o:
            os.write(sys.stdout.fileno(), o)

# restore tty settings back
termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_tty)