如何在python中检查网络接口上的数据传输？

轮询以太网接口统计信息的最佳方法是通过SNMP …

• 看起来您正在使用linux …如果是，请在 / etc / defaults / snmpd中*

  snmpd

  安装后

  snmpd

  ，使用以下选项加载您的Linux （确保其中的行如下所示）： *

  SNMPDOPTS

SNMPDOPTS='-Lsd -Lf /dev/null -u snmp -I-smux,usmConf,iquery,dlmod,diskio,lmSensors,hr_network,snmpEngine,system_mib,at,interface,ifTable,ipAddressTable,ifXTable,ip,cpu,tcpTable,udpTable,ipSystemStatsTable,ip,snmp_mib,tcp,icmp,udp,proc,memory,snmpNotifyTable,inetNetToMediaTable,ipSystemStatsTable,disk-Lsd -p /var/run/snmpd.pid'

• 您可能还需要将ro社区更改为

  public

  See Note 1并设置您的侦听接口

  /etc/snmp/snmpd.conf

  （如果不在环回上）…

• 假设您现在具有功能

  snmpd

  ，则可以使用此方法轮询

  ifHCInBytes

  并

  ifHCOutBytes

  查看有问题的接口的注释2。

poll_bytes.py ：

from SNMP import v2Managerimport timedef poll_eth0(manager=None):    # NOTE: 2nd arg to get_index should be a valid ifName value    in_bytes = manager.get_index('ifHCInOctets', 'eth0')    out_bytes = manager.get_index('ifHCOutOctets', 'eth0')    return (time.time(), int(in_bytes), int(out_bytes))# Prep an SNMP manager object...mgr = v2Manager('localhost')mgr.index('ifName')stats = list()# Insert condition below, instead of True...while True:    stats.append(poll_eth0(mgr))    print poll_eth0(mgr)    time.sleep(5)

SNMP.py

from subprocess import Popen, PIPEimport reclass v2Manager(object):    def __init__(self, addr='127.0.0.1', community='public'):        self.addr      = addr        self.community = community        self._index = dict()    def bulkwalk(self, oid='ifName'):        cmd = 'snmpbulkwalk -v 2c -Osq -c %s %s %s' % (self.community, self.addr, oid)        po = Popen(cmd, shell=True, stdout=PIPE, executable='/bin/bash')        output = po.communicate()[0]        result = dict()        for line in re.split(r'r*n', output): if line.strip()=="":     continue idx, value = re.split(r's+', line, 1) idx = idx.replace(oid+".", '') result[idx] = value        return result    def bulkwalk_index(self, oid='ifOutOctets'):        result = dict()        if not (self._index==dict()): vals = self.bulkwalk(oid=oid) for key, val in vals.items():     idx = self._index.get(key, None)     if not (idx is None):         result[idx] = val     else:         raise ValueError, "Could not find '%s' in the index (%s)" % self.index        else: raise ValueError, "Call the index() method before calling bulkwalk_index()"        return result    def get_index(self, oid='ifOutOctets', index=''):        # This method is horribly inefficient... improvement left as exercise for the reader...        if index: return self.bulkwalk_index().get(index, "<unknown>")        else: raise ValueError, "Please include an index to get"    def index(self, oid='ifName'):        self._index = self.bulkwalk(oid=oid)

结束说明：

1. SNMP v2c使用明文身份验证。如果您担心安全性/有人在嗅探您的流量，请更改您的社区，并通过源IP地址将查询限制在您的linux计算机上。完美的世界是将

   SNMP.py

   上述内容修改为使用SNMPv3（用于加密敏感数据）；大多数人只是使用非公共社区，并通过源IP限制snmp查询。

2. ifHCInOctets

   并

   ifHCOutOctets

   提供通过接口传输的字节数的瞬时值。如果您正在寻找数据传输速率，那么当然会涉及一些额外的数学运算。