Python で高性能ポートスキャナーを同時に作成する方法の例-Python チュートリアル-php.cn

ポートスキャナーについて
ポートスキャナーは、サーバーまたはホスト上で開いているポートを検出するために使用されるツールを指します。これは、コンピュータ管理者がセキュリティポリシーを確認するために、また攻撃者がターゲットホスト上で動作しているネットワークサービスを識別するためによく使用されます。

ポートスキャンの定義は、クライアントが対応するリクエストを一定範囲のサーバーポートに送信して、使用可能なポートを確認することです。これ自体は悪意のあるネットワーク活動ではありませんが、ネットワーク攻撃者がターゲットのホストサービスを検出し、サービスの既知の脆弱性を悪用するための重要な手段でもあります。ポートスキャンの主な目的は、依然としてリモートマシン上のサービスの可用性を確認することだけです。

複数のホストをスキャンして特定のポートを取得することを、ポートスイープ (ポートスイープ) と呼び、特定のサービスを取得します。たとえば、SQL サービスに基づくコンピュータワームは、多数のホスト上の同じポートをスイープして、ポート 1433 で TCP 接続を確立します。

Python 実装

ポートスキャナの原理は非常に単純で、接続できればポートが開いているとみなすだけです。

import socket 
def scan(port): 
  s = socket.socket() 
  if s.connect_ex(('localhost', port)) == 0: 
    print port, 'open' 
  s.close() 
if __name__ == '__main__': 
  map(scan,range(1,65536))

ログイン後にコピー

最も単純なポートスキャナーが登場しました。
待ってください。長時間応答がありません。これは、ソケットがブロックされており、各接続がタイムアウトになるまで長時間待機する必要があるためです。
私たちは自分たちでタイムアウトを追加しました。

s.settimeout(0.1)

ログイン後にコピー

もう一度実行すると、かなり速くなったように感じます。

マルチスレッドバージョン

import socket 
import threading 
def scan(port): 
  s = socket.socket() 
  s.settimeout(0.1) 
  if s.connect_ex(('localhost', port)) == 0: 
    print port, 'open' 
  s.close() 
 
if __name__ == '__main__': 
  threads = [threading.Thread(target=scan, args=(i,)) for i in xrange(1,65536)] 
  map(lambda x:x.start(),threads)

ログイン後にコピー

実行してみると、すごい速さです、速すぎてエラーがスローされます。 thread.error: 新しいスレッドを開始できません。
考えてみてください。このプロセスは 65535 個のスレッドをオープンしています。1 つはスレッドの最大数を超えていること、もう 1 つはソケットハンドルの最大数を超えていることです。 Linux では、ulimit を通じて変更できます。
最大制限を変更しない場合、エラーを報告せずにマルチスレッドを使用するにはどうすればよいでしょうか?
キューを追加し、プロデューサー/コンシューマーモードに切り替えて、固定スレッドを開きます。

マルチスレッド + キューバージョン

import socket 
import threading 
from Queue import Queue 
def scan(port): 
  s = socket.socket() 
  s.settimeout(0.1) 
  if s.connect_ex(('localhost', port)) == 0: 
    print port, 'open' 
  s.close() 
 
def worker(): 
  while not q.empty(): 
    port = q.get() 
    try: 
      scan(port) 
    finally: 
      q.task_done() 
 
if __name__ == '__main__': 
  q = Queue() 
  map(q.put,xrange(1,65535)) 
  threads = [threading.Thread(target=worker) for i in xrange(500)] 
  map(lambda x:x.start(),threads) 
  q.join()

ログイン後にコピー

ここで 500 個のスレッドを開き、キューからタスクを常にフェッチします。

マルチプロセッシング+キューバージョン
65535 プロセスを開くことはできませんね?または、生産者/消費者モデルを使用します

import multiprocessing 
def scan(port): 
  s = socket.socket() 
  s.settimeout(0.1) 
  if s.connect_ex(('localhost', port)) == 0: 
    print port, 'open' 
  s.close() 
 
def worker(q): 
  while not q.empty(): 
    port = q.get() 
    try: 
      scan(port) 
    finally: 
      q.task_done() 
 
if __name__ == '__main__': 
  q = multiprocessing.JoinableQueue() 
  map(q.put,xrange(1,65535)) 
  jobs = [multiprocessing.Process(target=worker, args=(q,)) for i in xrange(100)] 
  map(lambda x:x.start(),jobs)

ログイン後にコピー

キューはプロセスセーフキューであるため、ここではパラメーターとしてワーカーに渡されることに注意してください。そうでない場合はエラーが報告されます。
もう 1 つの便利な機能は JoinableQueue() です。これは名前が示すように join() が可能です。

gevent のスポーンバージョン

from gevent import monkey; monkey.patch_all(); 
import gevent 
import socket 
... 
if __name__ == '__main__': 
  threads = [gevent.spawn(scan, i) for i in xrange(1,65536)] 
  gevent.joinall(threads)

ログイン後にコピー

モンキーパッチはパッチを適用する前にインポートする必要があることに注意してください。そうしないと、たとえば、最初にスレッドをインポートしてからモンキーパッチをインポートすることはできません。

gevent のプールバージョン

from gevent import monkey; monkey.patch_all(); 
import socket 
from gevent.pool import Pool 
... 
if __name__ == '__main__': 
  pool = Pool(500) 
  pool.map(scan,xrange(1,65536)) 
  pool.join()

ログイン後にコピー

concurrent.futures バージョン

import socket 
from Queue import Queue 
from concurrent.futures import ThreadPoolExecutor 
... 
if __name__ == '__main__': 
  q = Queue() 
  map(q.put,xrange(1,65536)) 
  with ThreadPoolExecutor(max_workers=500) as executor: 
    for i in range(500): 
      executor.submit(worker,q)

ログイン後にコピー