Blog

Python pty 모듈로 대화형 CLI 자동화하기

subprocess로 ssh에 비밀번호를 흘려넣으면 실패한다. 프로그램이 stdin이 아니라 제어 터미널에서 비밀번호를 읽기 때문이다. 같은 이유로 파이프에 물린 자식은 출력 버퍼링이 달라져 진행 상황이 실시간으로 안 보이고, bash -i는 프롬프트조차 띄우지 않는다. 표준 라이브러리 pty 모듈은 의사 터미널(pseudo-terminal)을 만들어 자식에게 붙여주는 방식으로 이 문제들을 한꺼번에 푼다. 이 글에서는 pty.fork()로 expect 스타일 자동화 래퍼를 직접 만들고, 터미널 크기 전달과 에코 제어까지 정리한다.

import subprocess

p = subprocess.run(["ssh", "-o", "PubkeyAuthentication=no",
                    "-o", "PreferredAuthentications=password",
                    "audit_demo@192.168.0.46", "id -un"],
                   input="demopass123\n", capture_output=True, text=True)
print("rc :", p.returncode)
print("err:", p.stderr.strip().splitlines()[0])
$ python3 demo_pipe_ssh.py
rc : 255
err: ssh_askpass: exec(/usr/bin/ssh-askpass): No such file or directory

파이프와 pty의 차이

자식 쪽에서 무엇이 달라지는지부터 확인한다. 아래 프로그램은 stdin/stdout이 터미널인지만 출력한다.

import sys
print("stdout isatty:", sys.stdout.isatty())
print("stdin  isatty:", sys.stdin.isatty())

pty.fork()는 fork()와 비슷하지만 자식에게 새 의사 터미널을 제어 터미널로 붙이고, 부모에게는 마스터 쪽 파일 디스크립터를 돌려준다.

import os, pty, sys

pid, fd = pty.fork()
if pid == 0:
    os.execvp(sys.executable, [sys.executable, "child.py"])

out = b""
while True:
    try:
        chunk = os.read(fd, 1024)
    except OSError:          # 자식이 끝나면 EIO
        break
    if not chunk:
        break
    out += chunk
os.waitpid(pid, 0)
sys.stdout.write(out.decode())
$ python3 -c "import subprocess,sys; print(subprocess.run([sys.executable,'child.py'],capture_output=True,text=True).stdout, end='')"
stdout isatty: False
stdin  isatty: False

$ python3 run_pty.py
stdout isatty: True
stdin  isatty: True

버퍼링이 바뀐다

isatty 결과가 바뀌면 libc(그리고 파이썬)의 출력 버퍼링 정책도 따라 바뀐다. 터미널이면 줄 단위, 파이프면 블록 단위로 모았다가 내보낸다.

import time
print("working")        # 개행은 있지만 flush는 안 함
time.sleep(2)
print("done")
import os, pty, subprocess, sys, time

def first_output_delay(use_pty: bool) -> float:
    t0 = time.monotonic()
    if use_pty:
        pid, fd = pty.fork()
        if pid == 0:
            os.execvp(sys.executable, [sys.executable, "slow.py"])
        data = os.read(fd, 1024)
        delay = time.monotonic() - t0
        os.waitpid(pid, 0)
    else:
        p = subprocess.Popen([sys.executable, "slow.py"], stdout=subprocess.PIPE)
        data = p.stdout.read(1024)
        delay = time.monotonic() - t0
        p.wait()
    return delay, data

for use_pty in (False, True):
    d, data = first_output_delay(use_pty)
    print(f"pty={use_pty!s:5} 첫 출력까지 {d:.2f}s  data={data!r}")
$ python3 buf_test.py
pty=False 첫 출력까지 2.01s  data=b'working\ndone\n'
pty=True  첫 출력까지 0.01s  data=b'working\r\n'

파이프 쪽은 프로세스가 끝나는 2초 뒤에야 두 줄이 한꺼번에 나온다. 진행 상황을 실시간으로 받아야 하는 자동화라면 이것만으로도 pty를 쓸 이유가 된다. 덤으로 pty 출력은 개행이 \r\n으로 바뀌는데(ONLCR), 출력을 파싱할 때 자주 걸리는 부분이다.

expect 스타일 래퍼 만들기

프롬프트를 기다렸다 응답하는 패턴을 클래스로 감싼다. 외부 패키지 없이 pty·select·re만 쓴다.

import os, pty, re, select, time


class PtyProcess:
    """의사 터미널에 붙여 자식 프로세스를 띄우고 프롬프트를 기다렸다 응답한다."""

    def __init__(self, argv, env=None):
        self.pid, self.fd = pty.fork()
        if self.pid == 0:                       # 자식: pty가 제어 터미널이 된다
            os.execvpe(argv[0], argv, env or os.environ)
        self.buf = ""

    def expect(self, pattern, timeout=10.0):
        """pattern이 나올 때까지 읽고, 매치 지점까지의 출력을 반환한다."""
        rx = re.compile(pattern)
        deadline = time.monotonic() + timeout
        while True:
            m = rx.search(self.buf)
            if m:
                out, self.buf = self.buf[: m.end()], self.buf[m.end():]
                return out
            remain = deadline - time.monotonic()
            if remain <= 0:
                raise TimeoutError(f"{pattern!r} 못 찾음, 버퍼={self.buf!r}")
            if not select.select([self.fd], [], [], remain)[0]:
                continue
            try:
                data = os.read(self.fd, 4096)
            except OSError:                     # 자식 종료 → EIO
                raise EOFError(f"자식 종료, 버퍼={self.buf!r}")
            self.buf += data.decode(errors="replace")

    def read_all(self, timeout=10.0):
        """자식이 끝날 때까지 남은 출력을 모두 모아서 반환한다."""
        deadline = time.monotonic() + timeout
        while select.select([self.fd], [], [], max(0, deadline - time.monotonic()))[0]:
            try:
                data = os.read(self.fd, 4096)
            except OSError:                     # EIO = 자식 종료
                break
            if not data:
                break
            self.buf += data.decode(errors="replace")
        out, self.buf = self.buf, ""
        return out

    def sendline(self, line):
        os.write(self.fd, line.encode() + b"\n")

    def wait(self):
        """자식이 끝날 때까지 기다렸다 종료 코드를 반환한다."""
        _, status = os.waitpid(self.pid, 0)   # 먼저 회수해야 SIGHUP으로 안 죽는다
        os.close(self.fd)
        return os.waitstatus_to_exitcode(status)

pty에서는 자식이 끝났을 때 마스터 읽기가 0바이트가 아니라 OSError(EIO)로 실패한다. 파이프와 다른 이 부분을 EOFError로 바꿔 던지게 해두면 호출부가 단순해진다.

대화형 셸 자동화

bash -i는 터미널이 없으면 프롬프트를 띄우지 않는다. pty를 붙이면 사람이 치는 것과 똑같이 동작한다.

from ptyproc import PtyProcess

env = {"PS1": "READY> ", "TERM": "dumb", "PATH": "/usr/bin:/bin"}
sh = PtyProcess(["bash", "--norc", "--noprofile", "-i"], env)

sh.expect(r"READY> ")                 # 첫 프롬프트 대기
sh.sendline("echo $-")                # 셸 옵션 플래그 확인
print("flags   :", sh.expect(r"READY> ").strip().splitlines()[-2])

sh.sendline("tty")
print("tty     :", sh.expect(r"READY> ").strip().splitlines()[-2])

sh.sendline("exit")
print("exitcode:", sh.wait())
$ python3 demo_bash.py
flags   : himBHs
tty     : /dev/pts/2
exitcode: 0

$-에 i가 들어있으니 실제로 대화형 모드로 떴다. 종료 코드를 제대로 받으려면 waitpid()를 마스터 fd close()보다 먼저 호출해야 하고, 순서를 바꾸면 자식이 SIGHUP으로 죽어 결과가 -1이 된다.

ssh 비밀번호 자동 입력

맨 앞에서 실패했던 시나리오를 같은 조건으로 다시 돌린다.

from ptyproc import PtyProcess

HOST, USER, PW = "192.168.0.46", "audit_demo", "demopass123"

ssh = PtyProcess(["ssh", "-o", "PubkeyAuthentication=no",
                  "-o", "PreferredAuthentications=password",
                  f"{USER}@{HOST}", "id -un; hostname"])
ssh.expect(r"[Pp]assword: ")
ssh.sendline(PW)
print("원격 출력:", repr(ssh.read_all()))
print("종료 코드:", ssh.wait())
$ python3 demo_ssh.py
원격 출력: '\r\naudit_demo\r\nnoble\r\n'
종료 코드: 0

운영 환경이라면 키 인증을 쓰는 게 맞다. 이 방식은 비밀번호 인증만 열려 있는 장비나 시리얼 콘솔·부트로더처럼 대안이 없는 대상에 쓴다.

터미널 크기 전달하기

새로 만든 pty는 크기가 0×0이라 top·less·vim 같은 전체 화면 프로그램이 깨진다. TIOCSWINSZ ioctl로 크기를 넣어줘야 한다.

import fcntl, struct, termios
from ptyproc import PtyProcess

def set_winsize(fd, rows, cols):
    fcntl.ioctl(fd, termios.TIOCSWINSZ, struct.pack("HHHH", rows, cols, 0, 0))

def ask_size(set_size):
    env = {"PS1": "READY> ", "TERM": "xterm", "PATH": "/usr/bin:/bin", "COLUMNS": "", "LINES": ""}
    sh = PtyProcess(["bash", "--norc", "--noprofile", "-i"], env)
    if set_size:
        set_winsize(sh.fd, 40, 120)
    sh.expect(r"READY> ")
    sh.sendline("stty size; tput cols")
    out = sh.expect(r"READY> ").strip().splitlines()
    sh.sendline("exit"); sh.wait()
    return out[-3:-1]

print("winsize 미설정:", ask_size(False))
print("winsize 설정  :", ask_size(True))
$ python3 demo_winsize.py
winsize 미설정: ['0 0', '80']
winsize 설정  : ['40 120', '120']

크기를 안 넣으면 stty size가 0 0을 반환하고 tput cols는 terminfo 기본값 80으로 떨어진다. 창 크기가 바뀔 때 같은 ioctl을 다시 호출하면 커널이 자식에게 SIGWINCH를 보낸다.

에코 끄기

pty는 기본적으로 입력을 되돌려 보낸다(ECHO). 비밀번호를 보내면 그 값이 그대로 읽기 스트림에 섞여 들어온다.

token = input("Token: ")
print("received", len(token), "chars")
import sys, termios
from ptyproc import PtyProcess

def run(disable_echo):
    p = PtyProcess([sys.executable, "ask_plain.py"])
    if disable_echo:
        attrs = termios.tcgetattr(p.fd)
        attrs[3] &= ~termios.ECHO            # lflag에서 ECHO 비트 제거
        termios.tcsetattr(p.fd, termios.TCSANOW, attrs)
    p.expect(r"Token: ")
    p.sendline("s3cr3t-token")
    out = p.expect(r"chars")
    p.wait()
    return out

print("ECHO 켜짐:", repr(run(False)))
print("ECHO 꺼짐:", repr(run(True)))
$ python3 demo_echo.py
ECHO 켜짐: 's3cr3t-token\r\nreceived 12 chars'
ECHO 꺼짐: 'received 12 chars'

termios.tcgetattr()가 돌려주는 리스트의 네 번째 항목이 lflag이고, 여기서 ECHO 비트만 지우면 된다. getpass처럼 프로그램이 스스로 에코를 끄는 경우는 이 처리가 필요 없다.

어떤 API를 쓸까

API쓰임출력 수집
pty.spawn(argv)내 터미널을 그대로 자식에게 넘김화면으로 통과, master_read 콜백으로 가로채기
pty.fork()자식에 새 pty를 붙이고 마스터 fd 확보직접 os.read()
pty.openpty()pty 쌍만 만들고 실행은 subprocess에 맡김슬레이브 fd를 stdin/stdout으로 전달
$ python3 -c "import pty; print('status:', pty.spawn(['bash','-c','tty; echo \$-']))"
/dev/pts/2
hBc
status: 0

pty.spawn()은 호출한 쪽 터미널을 물려주므로 가장 간단하지만, 위 hBc에서 보듯 자식은 대화형(i)이 아니다. 프롬프트를 기다렸다 응답하는 자동화에는 pty.fork() 쪽이 맞다.

주의사항

  • 보낸 입력이 출력으로 되돌아온다. ECHO를 끄지 않으면 expect()가 자기가 보낸 문자열에 먼저 매치될 수 있다. 패턴은 되도록 명령의 결과 쪽에 맞춘다.
  • 마스터를 계속 읽어야 한다. 커널 pty 버퍼가 차면 자식이 write에서 블록된 채 멈춘다. 자식 종료를 기다리기 전에 출력을 먼저 비워야 데드락이 없다.
  • 비밀번호를 소스에 박지 말 것. 위 예제는 재현을 위해 상수로 뒀지만, 실제로는 환경변수나 시크릿 저장소에서 읽는다. expect()가 예외를 낼 때 버퍼를 그대로 로그에 남기지 않도록 하는 것도 함께 봐야 한다.
  • pty 개수에는 한계가 있다. /proc/sys/kernel/pty/max가 상한이고, wait()에서 fd를 안 닫으면 누수된다.
  • 기능이 더 필요하면 pexpect를 쓴다. 여기서 만든 래퍼는 원리를 보기 위한 최소 구현이고, 패턴 목록·로깅·SIGWINCH 처리까지 필요하면 검증된 라이브러리가 낫다.

마무리

pty를 붙인다는 것은 자식 프로세스에게 “너는 사람이 앉아 있는 터미널에 연결돼 있다”고 알려주는 일이다. 그 한 가지가 바뀌면 isatty 분기, 버퍼링, 비밀번호 입력, 프롬프트, 화면 크기가 전부 따라 바뀐다. 자동화가 이유 없이 멈추거나 출력이 안 보일 때 subprocess 대신 pty를 의심해볼 만하다.

참고

Tags: , , , , , , ,

답글 남기기