5

プログラムの流れは次のとおりです。

  1. Paramiko ライブラリを使用して Linux マシン上の OpenSSH サーバーに接続する
  2. X11 セッションを開く
  3. xterm 実行可能ファイルを実行する
  4. ターミナルに実行可能ファイル名を入力して実行することにより、他のプログラム (Firefox など) を実行します。

次のコードを使用して開いているターミナルで実行可能ファイルを実行する方法を誰かが説明し、サンプル ソース コード ( source )を提供していただければ幸いです。

import select
import sys
import paramiko
import Xlib.support.connect as xlib_connect
import os
import socket
import subprocess



# run xming
XmingProc = subprocess.Popen("C:/Program Files (x86)/Xming/Xming.exe :0 -clipboard -multiwindow")
ssh_client = paramiko.SSHClient()
ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh_client.connect(SSHServerIP, SSHServerPort, username=user, password=pwd)
transport = ssh_client.get_transport()
channelOppositeEdges = {}

local_x11_display = xlib_connect.get_display(os.environ['DISPLAY'])
inputSockets = []

def x11_handler(channel, (src_addr, src_port)):
    local_x11_socket = xlib_connect.get_socket(*local_x11_display[:3])
    inputSockets.append(local_x11_socket)
    inputSockets.append(channel)
    channelOppositeEdges[local_x11_socket.fileno()] = channel
    channelOppositeEdges[channel.fileno()] = local_x11_socket
    transport._queue_incoming_channel(channel)

session = transport.open_session()
inputSockets.append(session)
session.request_x11(handler = x11_handler)
session.exec_command('xterm')
transport.accept()

while not session.exit_status_ready():
    readable, writable, exceptional = select.select(inputSockets,[],[])
    if len(transport.server_accepts) > 0:
        transport.accept()
    for sock in readable:
        if sock is session:
            while session.recv_ready():
                sys.stdout.write(session.recv(4096))
            while session.recv_stderr_ready():
                sys.stderr.write(session.recv_stderr(4096))   
        else: 
            try:
                data = sock.recv(4096)
                counterPartSocket  = channelOppositeEdges[sock.fileno()]
                counterPartSocket.sendall(data)
            except socket.error:
                inputSockets.remove(sock)
                inputSockets.remove(counterPartSocket)
                del channelOppositeEdges[sock.fileno()]
                del channelOppositeEdges[counterPartSocket.fileno()]
                sock.close()
                counterPartSocket.close()

print 'Exit status:', session.recv_exit_status()
while session.recv_ready():
    sys.stdout.write(session.recv(4096))
while session.recv_stderr_ready():
    sys.stdout.write(session.recv_stderr(4096))
session.close()
XmingProc.terminate()
XmingProc.wait()

xterm を実行しているスレッドが子スレッドの終了を待っている間に、プログラムを子スレッドで実行することを考えていました。

4

1 に答える 1

2

まあ、これはちょっとしたハックですが、ねえ。

リモート エンドでできることは次のとおりです。xterm 内で を実行netcatし、ポートで着信データをリッスンし、取得したものをパイプしますbash。xterm direclty に入力するのとまったく同じではありませんが、bash に直接入力するのとほぼ同じくらい良いので、目標に少しでも近づくことができれば幸いです。本当に xterm を直接操作したい場合は、こちらをお読みください。

例えば：

ターミナル1：

% nc -l 3333 | bash

ターミナル 2 (echo hiここに入力):

% nc localhost 3333
echo hi

hiこれで、最初のターミナルからポップアウトが表示されるはずです。で試してみてくださいxterm&。それは私のために働いた。

これを Python で自動化する方法を次に示します。time.sleepばかげたsを使用するのではなく、準備が整ったときにサーバーがクライアントに通知できるようにするコードを追加したい場合があります。

import select
import sys
import paramiko
import Xlib.support.connect as xlib_connect
import os
import socket
import subprocess

# for connecting to netcat running remotely
from multiprocessing import Process
import time

# data
import getpass
SSHServerPort=22
SSHServerIP = "localhost"
# get username/password interactively, or use some other method..
user = getpass.getuser()
pwd = getpass.getpass("enter pw for '" + user + "': ")
NETCAT_PORT = 3333
FIREFOX_CMD="/path/to/firefox &"
#FIREFOX_CMD="xclock&"#or this :)

def run_stuff_in_xterm():
    time.sleep(5)
    s = socket.socket(socket.AF_INET6 if ":" in SSHServerIP else socket.AF_INET, socket.SOCK_STREAM)
    s.connect((SSHServerIP, NETCAT_PORT))
    s.send("echo \"Hello there! Are you watching?\"\n")
    s.send(FIREFOX_CMD + "\n")
    time.sleep(30)
    s.send("echo bye bye\n")
    time.sleep(2)
    s.close()

# run xming
XmingProc = subprocess.Popen("C:/Program Files (x86)/Xming/Xming.exe :0 -clipboard -multiwindow")
ssh_client = paramiko.SSHClient()
ssh_client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
ssh_client.connect(SSHServerIP, SSHServerPort, username=user, password=pwd)
transport = ssh_client.get_transport()
channelOppositeEdges = {}

local_x11_display = xlib_connect.get_display(os.environ['DISPLAY'])
inputSockets = []

def x11_handler(channel, (src_addr, src_port)):
    local_x11_socket = xlib_connect.get_socket(*local_x11_display[:3])
    inputSockets.append(local_x11_socket)
    inputSockets.append(channel)
    channelOppositeEdges[local_x11_socket.fileno()] = channel
    channelOppositeEdges[channel.fileno()] = local_x11_socket
    transport._queue_incoming_channel(channel)

session = transport.open_session()
inputSockets.append(session)
session.request_x11(handler = x11_handler)
session.exec_command("xterm -e \"nc -l 0.0.0.0 %d | /bin/bash\"" % NETCAT_PORT)
p = Process(target=run_stuff_in_xterm)
transport.accept()
p.start()

while not session.exit_status_ready():
    readable, writable, exceptional = select.select(inputSockets,[],[])
    if len(transport.server_accepts) > 0:
        transport.accept()
    for sock in readable:
        if sock is session:
            while session.recv_ready():
                sys.stdout.write(session.recv(4096))
            while session.recv_stderr_ready():
                sys.stderr.write(session.recv_stderr(4096))   
        else: 
            try:
                data = sock.recv(4096)
                counterPartSocket  = channelOppositeEdges[sock.fileno()]
                counterPartSocket.sendall(data)
            except socket.error:
                inputSockets.remove(sock)
                inputSockets.remove(counterPartSocket)
                del channelOppositeEdges[sock.fileno()]
                del channelOppositeEdges[counterPartSocket.fileno()]
                sock.close()
                counterPartSocket.close()

p.join()
print 'Exit status:', session.recv_exit_status()
while session.recv_ready():
    sys.stdout.write(session.recv(4096))
while session.recv_stderr_ready():
    sys.stdout.write(session.recv_stderr(4096))
session.close()
XmingProc.terminate()
XmingProc.wait()

XmingProcこれを Mac でテストしたので、ビットをコメントアウトして(and )/Applications/Firefox.app/Contents/MacOS/firefoxとして使用しました。FIREFOX_CMDxclock

適切なタイミングでポートに接続すると、リモートサーバーで任意のコードを実行できるため、上記は正確には安全な設定ではありませんが、とにかくテスト目的でこれを使用することを計画しているようです. セキュリティを向上させたい場合は、netcat を127.0.0.1ではなく0.0.0.0にバインドし、ssh トンネルをセットアップして (ssh -L3333:localhost:3333 username@remote-host.comポート 3333 でローカルに受信したすべてのトラフィックを remote-host.com:3333 にトンネリングするために実行します)、Python を に接続させ("localhost", 3333)ます。

これをブラウザの自動化のためにセレンと組み合わせることができます：

このページの指示に従います。つまり、Selenium スタンドアロン サーバー jar ファイルをダウンロードし、/path/to/some/place(サーバー上に) 配置し、pip install -U selenium(再びサーバー上に) 配置します。

次に、次のコードを in に挿入selenium-example.pyし/path/to/some/placeます。

#!/usr/bin/env python
from selenium import webdriver
from selenium.common.exceptions import NoSuchElementException
from selenium.webdriver.common.keys import Keys
import time

browser = webdriver.Firefox() # Get local session of firefox
browser.get("http://www.yahoo.com") # Load page
assert "Yahoo" in browser.title
elem = browser.find_element_by_name("p") # Find the query box
elem.send_keys("seleniumhq" + Keys.RETURN)
time.sleep(0.2) # Let the page load, will be added to the API
try:
    browser.find_element_by_xpath("//a[contains(@href,'http://docs.seleniumhq.org')]")
except NoSuchElementException:
    assert 0, "can't find seleniumhq"
browser.close()

Firefox コマンドを次のように変更します。

FIREFOX_CMD="cd /path/to/some/place && python selenium-example.py"

そして、firefox が Yahoo 検索を行うのを見てください。を増やすこともできますtime.sleep。

さらに多くのプログラムを実行したい場合は、firefox を実行する前または実行した後に、次のようにすることができます。

# start up xclock, wait for some time to pass, kill it.
s.send("xclock&\n")
time.sleep(1)
s.send("XCLOCK_PID=$!\n")  # stash away the process id (into a bash variable)
time.sleep(30)
s.send("echo \"killing $XCLOCK_PID\"\n")
s.send("kill $XCLOCK_PID\n\n")
time.sleep(5)

一般的な X11 アプリケーション制御を実行したい場合は、ライブラリは異なりますが、同様の「ドライバー アプリケーション」を作成する必要があると思います。より一般的なアプローチを見つけるために、「x11 send {mouse|keyboard} events」を検索することをお勧めします。それはこれらの 質問を引き起こしますが、もっとたくさんあると確信しています。

リモート エンドが即座に応答しない場合は、Wireshark でネットワーク トラフィックをスニッフィングし、TCP がデータを 1 行ずつ送信するのではなく、データをバッチ処理しているかどうかを確認することをお勧めします (\nここではこれが役立つようですが、私は推測します)。保証はありません）。この場合、運が悪いかもしれませんが、不可能なことは何もありません。とはいえ、そこまで行かなくてもいいと思います ;-)

もう 1 つ注意: CLI プログラムの STDIN/STDOUT と通信する必要がある場合は、expect スクリプトを参照することをお勧めします (たとえば、pexpectを使用するか、単純なケースでは subprocess.Popen.communicate]( http:/ /docs.python.org/2/library/subprocess.html#subprocess.Popen.communicate )))。

于 2013-10-14T22:11:16.370 に答える