92

我如何知道某个端口是否在 linux ubuntu 上打开/关闭,而不是远程系统,使用 python?如何在 python 中列出这些开放端口?

  • Netstat:有没有办法将 netstat 输出与 python 集成?
4

13 回答 13

160

您可以使用套接字模块简单地检查端口是否打开。

它看起来像这样。

import socket
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
result = sock.connect_ex(('127.0.0.1',80))
if result == 0:
   print "Port is open"
else:
   print "Port is not open"
sock.close()
于 2013-10-05T09:38:24.547 回答
110

如果你想在更一般的上下文中使用它,你应该确保你打开的套接字也被关闭。所以检查应该更像这样:

import socket
from contextlib import closing
   
def check_socket(host, port):
    with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as sock:
        if sock.connect_ex((host, port)) == 0:
            print("Port is open")
        else:
            print("Port is not open")
于 2016-02-12T18:46:41.297 回答
44

对我来说,如果端口未打开,上面的示例将挂起。第 4 行显示了使用settimeout来防止挂起

import socket

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(2)                                      #2 Second Timeout
result = sock.connect_ex(('127.0.0.1',80))
if result == 0:
  print 'port OPEN'
else:
  print 'port CLOSED, connect_ex returned: '+str(result)
于 2016-07-17T03:24:39.333 回答
22

如果你只关心本地机器,你可以依赖 psutil 包。您可以:

  1. 检查特定 pid 使用的所有端口:

    proc = psutil.Process(pid)
    print proc.connections()
    
  2. 检查本地机器上使用的所有端口:

    print psutil.net_connections()
    

它也适用于 Windows。

https://github.com/giampaolo/psutil

于 2015-03-26T19:39:04.367 回答
5

这是一个快速的多线程端口扫描器:

from time import sleep
import socket, ipaddress, threading

max_threads = 50
final = {}
def check_port(ip, port):
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) # TCP
        #sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) # UDP
        socket.setdefaulttimeout(2.0) # seconds (float)
        result = sock.connect_ex((ip,port))
        if result == 0:
            # print ("Port is open")
            final[ip] = "OPEN"
        else:
            # print ("Port is closed/filtered")
            final[ip] = "CLOSED"
        sock.close()
    except:
        pass
port = 80
for ip in ipaddress.IPv4Network('192.168.1.0/24'): 
    threading.Thread(target=check_port, args=[str(ip), port]).start()
    #sleep(0.1)

    # limit the number of threads.
    while threading.active_count() > max_threads :
        sleep(1)

print(final)

现场演示

于 2019-04-13T07:17:37.927 回答
4

如果您在探测 TCP 端口并打算监听它时,最好实际调用 listen。使用tring 连接的方法不会“看到”已建立连接的客户端端口,因为没有人监听它。但是这些端口不能用来监听它。

import socket


def check_port(port, rais=True):
    """ True -- it's possible to listen on this port for TCP/IPv4 or TCP/IPv6
    connections. False -- otherwise.
    """
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.bind(('127.0.0.1', port))
        sock.listen(5)
        sock.close()
        sock = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
        sock.bind(('::1', port))
        sock.listen(5)
        sock.close()
    except socket.error as e:
        return False
        if rais:
            raise RuntimeError(
                "The server is already running on port {0}".format(port))
    return True
于 2017-04-05T18:01:58.770 回答
4

同意萨钦。只是一个改进,用connect_ex代替connect,这样可以避免try except

>>> def port_check(ip_port):
...     s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
...     s.settimeout(1)
...     return s.connect_ex(ip_port) == 0
... 
>>> port_check(loc)
True
>>> port_check(loc_x)
False
>>> loc
('10.3.157.24', 6443)
>>> 
于 2020-09-14T15:26:30.140 回答
3

Netstat 工具只是解析一些 /proc 文件,如 /proc/net/tcp 并将其与其他文件内容结合起来。是的,它是高度特定于平台的,但对于仅限 Linux 的解决方案,您可以坚持使用它。Linux 内核文档详细描述了这些文件,因此您可以在那里找到如何阅读它们。

另请注意,您的问题过于含糊,因为“端口”也可能表示串行端口(/dev/ttyS* 和类似物)、并行端口等;我已经重用了另一个答案的理解,这是网络端口,但我会要求您更准确地提出您的问题。

于 2013-12-22T07:53:59.710 回答
3

我在这篇文章中找到了多种解决方案。但是有些解决方案存在挂起问题或在未打开端口的情况下需要很长时间。以下解决方案对我有用:

import socket 

def port_check(HOST):
   s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
   s.settimeout(2) #Timeout in case of port not open
   try:
      s.connect((HOST, 22)) #Port ,Here 22 is port 
      return True
   except:
      return False

port_check("127.0.1.1")
于 2020-08-10T11:22:05.650 回答
2

请检查迈克尔的答案并投票。这是检查开放端口的正确方法。如果您正在开发服务或守护程序,则 Netstat 和其他工具没有任何用处。例如,我正在为工业网络创建 modbus TCP 服务器和客户端服务。服务可以监听任何端口,但问题是该端口是否开放?该程序将在不同的地方使用,我无法手动检查它们,所以这就是我所做的:

from contextlib import closing
import socket
class example:
    def __init__():

       self.machine_ip = socket.gethostbyname(socket.gethostname())
       self.ready:bool = self.check_socket()

    def check_socket(self)->bool:
        result:bool = True
        with closing(socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as sock:
        modbus_tcp_port:int = 502
        if not sock.connect_ex((self.machine_ip, modbus_tcp_port)) == 0:
            result = False
        return result
于 2018-11-07T01:00:47.943 回答
2

基于psutilJoe 提到的解决方案(仅适用于检查本地端口):

import psutil
1111 in [i.laddr.port for i in psutil.net_connections()]

True如果当前使用端口 1111,则返回。

psutil不是 python stdlib 的一部分,所以你需要pip install psutil先。它还需要 python 标头可用,所以你需要类似的东西python-devel

于 2019-11-16T15:37:49.560 回答
2

刚刚在 mrjandro 的解决方案中添加了一个快速破解方法,以摆脱简单的连接错误/超时。

您可以调整更改 max_error_count 变量值的阈值并添加任何类型的通知。

import socket

max_error_count = 10

def increase_error_count():
    # Quick hack to handle false Port not open errors 
    with open('ErrorCount.log') as f:
        for line in f:
            error_count = line
    error_count = int(error_count)
    print "Error counter: " + str(error_count)
    file = open('ErrorCount.log', 'w')
    file.write(str(error_count + 1))
    file.close()
    if error_count == max_error_count:
        # Send email, pushover, slack or do any other fancy stuff
        print "Sending out notification"
        # Reset error counter so it won't flood you with notifications
        file = open('ErrorCount.log', 'w')
        file.write('0')
        file.close()

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(2) 
result = sock.connect_ex(('127.0.0.1',80))
if result == 0:
        print "Port is open"
else:
        print "Port is not open"
        increase_error_count()

在这里你可以找到一个 Python 3 兼容版本(只是固定的打印语法):

import socket

max_error_count = 10

def increase_error_count():
    # Quick hack to handle false Port not open errors
    with open('ErrorCount.log') as f:
        for line in f:
            error_count = line
    error_count = int(error_count)
    print ("Error counter: " + str(error_count))
    file = open('ErrorCount.log', 'w')
    file.write(str(error_count + 1))
    file.close()
    if error_count == max_error_count:
        # Send email, pushover, slack or do any other fancy stuff
        print ("Sending out notification")
        # Reset error counter so it won't flood you with notifications
        file = open('ErrorCount.log', 'w')
        file.write('0')
        file.close()

sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(2) 
result = sock.connect_ex(('127.0.0.1',80))
if result == 0:
        print ("Port is open")
else:
        print ("Port is not open")
        increase_error_count()
于 2015-09-17T09:58:27.530 回答
0

This will find a random port in the given range:

import socket
import random 
from typing import Tuple

def find_listening_port(
    port_range:Tuple[int,int]=None, 
    host='', 
    socket_type='tcp', 
    default:int=None
) -> int:
    """Find an available listening port
    
    Arguments:
        port_range: Optional tuple of ports to randomly search, ``[min_port, max_port]``
            If omitted, then randomly search between ``[6000, 65534]``
        host: Host interface to search, if omitted then bind to all interfaces
        socket_type: The socket type, this should be ``tcp`` or ``udp``
        default: The port to try first before randomly searching the port range

    Returns:
        Available port for listening
    """
    def _test_port(host, port, socket_protocol):
        with socket.socket(socket.AF_INET, socket_protocol) as sock:
            try:
                sock.bind((host, port))
                if socket_type == 'tcp':
                    sock.listen(1)
                return port
            except:
                pass 

        return -1

    if port_range is None:
        port_range = (6000,65534)

    if socket_type == 'tcp':
        socket_protocol = socket.SOCK_STREAM
    elif socket_type == 'udp':
        socket_protocol = socket.SOCK_DGRAM
    else:
        raise Exception('Invalid socket_type argument, must be: tcp or udp')

    searched_ports = []
    if default is not None:
        port = _test_port(host, default, socket_protocol)
        if port != -1:
            return port
        searched_ports.append(default)

    for _ in range(100):
        port = random.randint(port_range[0], port_range[1])
        if port in searched_ports:
            continue 

        port = _test_port(host, port, socket_protocol)
        if port != -1:
            return port

        searched_ports.append(port)

    raise Exception(f'Failed to find {socket_type} listening port for host={host}')

Example usage:

# Find a TCP port,
# first check if port 80 is availble
port = find_listening_port(
    port_range=(4000, 60000),
    host='',
    socket_type='tcp',
    default=80
)
print(f'Available TCP port: {port}')
于 2022-02-22T04:23:05.210 回答