我怎样才能检查在Python的networking接口上的数据传输?

有一种用于获取给定networking接口的IP的套接字方法:

import socket import fcntl import struct def get_ip_address(ifname): s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) return socket.inet_ntoa(fcntl.ioctl( s.fileno(), 0x8915, # SIOCGIFADDR struct.pack('256s', ifname[:15]) )[20:24]) 

其中返回以下内容:

 >>> get_ip_address('lo') '127.0.0.1' >>> get_ip_address('eth0') '38.113.228.130' 

有没有类似的方法来返回该接口的networking传输? 我知道我可以读取/proc/net/dev但我喜欢套接字方法。

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

  • 看起来你正在使用linux …如果是这样的话,用这些选项加载你的snmpd …在你的/ etc / defaults / snmpd中安装snmpd之后(确保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 请参阅注释1,并在/etc/snmp/snmpd.conf (如果不在回送中)设置您的监听接口…

  • 假设你有一个功能snmpd ,在这一点上,你可以轮询ifHCInBytesifHCOutBytes 请参阅注2你的接口使用这个…

poll_bytes.py

 from SNMP import v2Manager import time def 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, PIPE import re class 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. ifHCInOctetsifHCOutOctets提供通过接口传输的字节数的瞬时值。 如果你正在寻找数据传输速率,当然会有一些额外的数学参与。