Network interface traffic monitor based on vnstat, automatic disable network interface or shutdown

traffic-monitor.py

import subprocess
import json
import os
import sys

"""
通过 vnstat 统计当日&当月流量，达到第一限制时禁用网络
接口，达到第二限制则禁用网络接口+关机

p.s. 若开机时间不足最小开机时长，则达到第二限制时只会
禁用网络接口，不会关机（避免开机后立即被脚本关机）
"""

# ========== setting ===========
# 网卡名称
INTERFACE = "eth0"
# 日流量第一限制（单位：GB）
LIMIT_DAY_1 = 30
# 日流量第二限制（单位：GB）
LIMIT_DAY_2 = 50
# 月流量第一限制（单位：GB）
LIMIT_MONTH_1 = 100
# 月流量第二限制（单位：GB）
LIMIT_MONTH_2 = 200
# 最小开机时长（单位：s）
MIN_UPTIME = 180
# ========== setting ===========


# 系统开机时长获取
def get_uptime():
    result = subprocess.run(['cat', '/proc/uptime'],
                            stdout=subprocess.PIPE,
                            stderr=subprocess.PIPE)
    if result.returncode != 0:
        print(f"获取系统运行时间失败: {result.stderr.decode()}")
        sys.exit(1)
    uptime = float(result.stdout.decode().split()[0])
    return uptime


# vnstat 数据获取&解析，mode 参数支持 'day' 或 'month'
def get_vnstat(interface, mode):
    result = subprocess.run(
        ['vnstat', '-i', interface,
         '--json', mode[0], '1'],
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE)
    if result.returncode != 0:
        print(f"获取 vnstat 数据失败: {result.stderr.decode()}")
        sys.exit(1)
    res = {}
    data = result.stdout.decode()
    data = json.loads(data)
    data = data['interfaces'][0]
    res['rx'] = data['traffic'][mode][-1]['rx']
    res['tx'] = data['traffic'][mode][-1]['tx']
    res['tx_gb'] = res['tx'] / 1073741824

    year = data['updated']['date']['year']
    month = data['updated']['date']['month']
    day = data['updated']['date']['day']
    hour = data['updated']['time']['hour']
    minute = data['updated']['time']['minute']
    res['time'] = f"{year}.{month}.{day} {hour}:{minute}"

    return res


def main():
    # 获取系统开机时长
    uptime = get_uptime()
    print(f"开机时长: {uptime} s")

    # 获取流量数据并解析
    data_day = get_vnstat(INTERFACE, 'day')
    data_month = get_vnstat(INTERFACE, 'month')

    # 输出出入站流量
    print(f"日入站流量: {data_day['rx']} B")
    print(f"日出站流量: {data_day['tx']} B = {data_day['tx_gb']:.2f} GB")
    print(f"月入站流量: {data_month['rx']} B")
    print(f"月出站流量: {data_month['tx']} B = {data_month['tx_gb']:.2f} GB")

    # 输出统计时间
    print(f"统计时间: {data_day['time']}")

    # 流量限制检查
    if data_day['tx_gb'] >= LIMIT_DAY_2 \
            or data_month['tx_gb'] >= LIMIT_MONTH_2:
        print('达第二限制，正在禁用网络接口')
        os.system(f"sudo ip link set {INTERFACE} down")
        if uptime > MIN_UPTIME:
            print('达第二限制，正在关机')
            os.system('sudo shutdown -h now')
    elif data_day['tx_gb'] >= LIMIT_DAY_1 \
            or data_month['tx_gb'] >= LIMIT_MONTH_1:
        print('达第一限制，正在禁用网络接口')
        os.system(f"sudo ip link set {INTERFACE} down")
    else:
        print('当前流量未超限')


if __name__ == "__main__":
    main()

依赖

• python
• vnstat
• crontab(optional)

脚本基本逻辑

通过 vnstat 统计当日&当月流量，达到第一限制时禁用网络接口，达到第二限制则禁用网络接口+关机

p.s. 若开机时间不足最小开机时长，则达到第二限制时只会禁用网络接口，不会关机（避免开机后立即被脚本关机）

用途

在流量按量付费的机器上保命用，用最消极但有效的方案（叹气）防止被刷流量

需要注意的是，数据统计依赖 vnstat，所以数据实际上每 5 min 才会更新一次

请结合自己的网络带宽和该时间窗口评估是否能够接受其可能造成的损失

举例：在流量按量付费的阿里云 t6 机器上，最大带宽值为 80Mbps，流量费用最高为 0.8r/GB，则 5 min 的时间窗口可能造成的损失约为 $\frac{80 \times 300 \times 0.125}{1024} \times 0.8 \approx 2.34 \text{ 元}$