有一个获取给定网络接口(interface)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'
有没有类似的方法返回那个接口(interface)的网络传输?我知道我可以阅读 /proc/net/dev
但我喜欢套接字方法。
最佳答案
轮询以太网接口(interface)统计信息的最佳方式是通过 SNMP...
看起来您正在使用 linux...如果是这样,请使用这些选项加载您的
snmpd
...在安装snmpd
之后,在您的 /etc/defaults/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 中设置您的监听接口(interface)。 conf
(如果不在loopback上)...假设您有一个功能正常的
snmpd
,此时,您可以轮询ifHCInBytes
和ifHCOutBytes
请参阅注释 2 用于您的相关接口(interface)使用此...
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)
结束注释:
SNMP v2c 使用明文身份验证。如果您担心安全性/有人嗅探您的流量,请更改您的社区并通过源 ip 地址限制对您的 linux 机器的查询。完美的世界是修改上面的
SNMP.py
以使用 SNMPv3(加密敏感数据);大多数人只是使用非公共(public)社区并通过源 IP 限制 snmp 查询。ifHCInOctets
和ifHCOutOctets
提供通过接口(interface)传输的字节数的瞬时值。如果您正在寻找数据传输速率,当然会涉及一些额外的数学。
关于python - 如何检查 python 网络接口(interface)上的数据传输?,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/7731411/