Example of a possible implementation of pipe() for fabric: "a remote (SSH) Popen". It is just a proof-of-concept! (shame)

add-to-fabric-operations.py

import socket
import types

def new_readline(self, n=-1):
    buf = []
    bytes_read = True
    while bytes_read:
        try:
            bytes_read = self._old_read(1)
            buf.append(bytes_read)
            if bytes_read == '\n':
                break
        except socket.timeout:
            bytes_read = ''
    return ''.join(buf)

def new_read(self, n=-1):
    buf = []
    bytes_read = True
    while bytes_read:
        try:
            bytes_read = self._old_read(1)
            buf.append(bytes_read)
        except socket.timeout:
            bytes_read = ''
    return ''.join(buf)

class Pipe(object):
    def __init__(self, command, channel, stdin, stdout, stderr):
        self.command = command
        self.channel = channel
        self.stdin = stdin
        self.stdout = stdout
        self.stderr = stderr
        self.stdin.write(command + '\n')
        self.stdin._old_read = self.stdin.read
        self.stdout._old_read = self.stdout.read
        self.stderr._old_read = self.stderr.read
        self.stdin.read = types.MethodType(new_read, self.stdin)
        self.stdout.read = types.MethodType(new_read, self.stdout)
        self.stderr.read = types.MethodType(new_read, self.stderr)

        self.stdin._old_readline = self.stdin.readline
        self.stdout._old_readline = self.stdout.readline
        self.stderr._old_readline = self.stderr.readline
        self.stdin.readline = types.MethodType(new_readline, self.stdin)
        self.stdout.readline = types.MethodType(new_readline, self.stdout)
        self.stderr.readline = types.MethodType(new_readline, self.stderr)


    def close(self):
        self.stdin.close()
        self.stdout.close()
        self.stderr.close()
        self.channel.close()

def pipe(command):
    channel = default_channel()
    channel.setblocking(0)
    channel.get_pty()
    channel.invoke_shell()
    stdin = channel.makefile('wb', -1)
    stdout = channel.makefile('rb', -1)
    stderr = channel.makefile_stderr('rb', -1)

    new_pipe = Pipe(command, channel, stdin, stdout, stderr)
    return new_pipe

expected-output.txt

$ fab -H testing@localhost test
[testing@localhost] Executing task 'test'
[testing@localhost] Passphrase for private key: 
Welcome to Ubuntu 11.04 (GNU/Linux 2.6.38-11-generic x86_64)

 * Documentation:  https://help.ubuntu.com/

86 packages can be updated.
63 updates are security updates.

New release 'oneiric' available.
Run 'do-release-upgrade' to upgrade to it.

Last login: Mon Nov  7 15:48:16 2011 from localhost
passwd

DEBUG: "testing@ideas:~$ passwd"
DEBUG: "Changing password for testing."
DEBUG: "(current) UNIX password:"
DEBUG: "Enter new UNIX password:"
DEBUG: "Retype new UNIX password:"
DEBUG: "passwd: password updated successfully"

Done.
Disconnecting from testing@localhost... done.

fabfile.py

from fabric.api import task
from fabric.operations import pipe
import time


@task
def test():
    new_password = 'qwe123qwe456'
    old_password = 'This15C00l!'
    p = pipe('passwd')
    time.sleep(0.1)
    print p.stdout.read()
    running = True
    while running:
        line = p.stdout.readline()
        if not line:
            time.sleep(0.1)
            line = p.stdout.read()
        if line and line.strip():
            print 'DEBUG: "%s"' % line.strip()
            if '(current) UNIX password:' in line:
                p.stdin.write(old_password + '\n')
            elif 'passwd: Authentication token manipulation error' in line:
                running = False
                p.close()
                raise ValueError('Invalid old password')
            elif 'Enter new UNIX password:' in line or \
                 'Retype new UNIX password:' in line:
                p.stdin.write(new_password + '\n')
            elif 'You must choose a longer password' in line or \
                 'Bad: ' in line:
                running = False
                p.close()
                raise ValueError('Dummy password')
            elif 'passwd: password updated successfully' in line:
                running = False
        time.sleep(0.1)
    p.close()