Clever Geek Handbook

Filter the command that needs terminal in Python subprocess module

I am developing a robot that takes commands from the network (XMPP) and uses a subprocess module in Python to execute them and send commands back. It is essentially a non-interactive XMPP-based shell based on SSH.

The robot only executes commands from authenticated trusted sources, so any shell ( shell=True

) commands are allowed .

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

For example:

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

If each of the above commands is received, the robot gets stuck and the terminal from which the robot is launched is broken.

Although the robot only receives commands from authenticated trusted sources, human error. How can I get the robot to filter out those commands that might break? I know there is os.isatty

, but how can I use it? Is there a way to detect these "bad" commands and refuse to execute them?

TL; DR

Let's say there are two kinds of commands:

  • Commands like ls

    : no need to run tty.
  • Commands like vim

    : requires tty; aborts the subprocess if there is no tty.

How can I tell the command is ls

-like or is vim

-like and refuses to run the command if it is vim

-like?

+3


source to share


4 answers


What you expect is a function that takes a command as input and returns meaningful output when you run the command.

Since the command is arbitrary, the requirement for a tty is just one of many bad cases (the other involves starting an infinite loop), your function should only concern the run period, in other words, the command is "bad" or not, should be determined if it ends in for a limited time or not, and since it subprocess

is asynchronous in nature, you can simply run the command and process it in a higher vision.

Demo code to reproduce, you can change the value cmd

to see how it works 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'


There are three points of note in the code above:

  • We use Popen

    instead check_output

    to run a command, as opposed to check_output

    one that will wait for the process to complete, Popen

    returns immediately, so we can take further action to control the process.

  • We will implement a timer to check the status of the process, if it runs for too long, we killed it manually because we think that the process is meaningless if it cannot finish within a limited time. This way your original problem will be solved since it vim

    will never end and it will definitely get killed as the "unmeaningful" command.

  • After the timer helps us filter out bad commands, we can get the stdout and stderr of the command by calling a method on the communicate

    object Popen

    and then selecting it to determine what to return to the user.

Conclusion

Modeling

tty is not required, we have to start the subprocess asynchronously and then run a timer on it to determine if it should be killed or not, for those that ended normally its safe and easy to get the result.

+3


source


Well, SSH is already a tool that will allow users to remotely execute commands and be authenticated at the same time . The authentication part is extremely tricky, remember that building the software you describe is a little dangerous from a security standpoint.

There is no way to tell if a process needs a tty or not. And there is no method os.isatty

, because if you execute the subprocesses that are needed, that does not mean that they are. :)



Overall, it would probably be safer from a security standpoint, as well as a solution to this problem, if you considered the command whitelist. You can choose this whitelist to avoid things that the tty needs, because I don't think you can get around this easily.

+1


source


Thanks a lot for @JF Help Sebastia (see comments on this), I found a solution (workaround?) For my case.

The reason it vim

interrupts the terminal rather ls

than not is because vim

a tty is required. As Sebastia says, we can feed vim with pty using pty.openpty()

. By issuing pty gurantees, the command does not interrupt the terminal and we can add timout

to automatically kill such processes. Here's a (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 that we have to take care of it, not stdout or stderr.

0


source


You can link to my answer at: fooobar.com/questions/3871 / ... , which use a pseudo-terminal to make stdout no-blocking, and use select on the stdin / stdout descriptor.

I can just change command

var to 'vim'

. And the script works 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)
0


source






More articles:


All Articles