DHCP Service Bot
21e3d61871
New /scan page lists all active devices on the subnet (ARP scan): - 静态绑定: matches a host declaration - 动态分配: matches a dhcpd.leases entry - 未知 / 外部: nothing matches (manually configured or from other DHCP server) Click '转为静态绑定' to convert a discovered device to a static host binding. Also includes: - arp-scan wrapper in dhcp_ops.py (get_local_network, arp_scan, classify_devices) - /scan route with optional auto-refresh every 30s - README updated with arp-scan install step + new feature bullet
605 lines
22 KiB
Python
605 lines
22 KiB
Python
"""
|
||
dhcp_ops.py
|
||
===========
|
||
对 ISC DHCP 服务 (dhcpd) 的配置读写与租约解析封装。
|
||
- DHCPD_CONF /etc/dhcp/dhcpd.conf
|
||
- DHCP_LEASES /var/lib/dhcp/dhcpd.leases
|
||
- INTERFACES /etc/default/isc-dhcp-server
|
||
所有改动均经"写文件 -> 备份 -> dhcpd -t 语法检查 -> systemctl reload"流程。
|
||
"""
|
||
|
||
import ipaddress
|
||
import os
|
||
import re
|
||
import shutil
|
||
import subprocess
|
||
import time
|
||
from dataclasses import dataclass, field, asdict
|
||
from datetime import datetime
|
||
from typing import List, Optional, Dict, Tuple
|
||
|
||
DHCPD_CONF = "/etc/dhcp/dhcpd.conf"
|
||
DHCP_LEASES = "/var/lib/dhcp/dhcpd.leases"
|
||
INTERFACES_CONF = "/etc/default/isc-dhcp-server"
|
||
BACKUP_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "backups")
|
||
SERVICE_NAME = "isc-dhcp-server"
|
||
|
||
# 自定义配置区段标记 —— 我们维护的 subnet/host 块全部包含在 BEGIN/END 标记之间，
|
||
# 保留 dhcpd.conf 顶部原有的 option domain-name 等通用配置不动。
|
||
BEGIN_MARKER = "# >>> DHCP-WEB-MANAGER BEGIN >>>"
|
||
END_MARKER = "# <<< DHCP-WEB-MANAGER END <<<"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 数据类
|
||
# ---------------------------------------------------------------------------
|
||
|
||
@dataclass
|
||
class Subnet:
|
||
network: str # e.g. "192.168.1.0"
|
||
netmask: str # e.g. "255.255.255.0"
|
||
routers: str # e.g. "192.168.1.1"
|
||
dns: str # e.g. "8.8.8.8, 8.8.4.4"
|
||
range_start: str # e.g. "192.168.1.100"
|
||
range_end: str # e.g. "192.168.1.200"
|
||
default_lease: int = 600
|
||
max_lease: int = 7200
|
||
domain_name: str = ""
|
||
authoritative: bool = False
|
||
|
||
def validate(self) -> List[str]:
|
||
"""返回错误信息列表，空表示通过"""
|
||
errs = []
|
||
try:
|
||
net = ipaddress.ip_network(f"{self.network}/{self.netmask}", strict=False)
|
||
except ValueError as e:
|
||
errs.append(f"网段/掩码无效: {e}")
|
||
return errs
|
||
try:
|
||
start = ipaddress.ip_address(self.range_start)
|
||
end = ipaddress.ip_address(self.range_end)
|
||
except ValueError as e:
|
||
errs.append(f"地址池范围无效: {e}")
|
||
return errs
|
||
if start >= end:
|
||
errs.append("地址池起始必须小于结束")
|
||
if start not in net or end not in net:
|
||
errs.append("地址池必须在 subnet 范围内")
|
||
try:
|
||
routers_ip = ipaddress.ip_address(self.routers)
|
||
except ValueError:
|
||
errs.append(f"网关地址无效: {self.routers}")
|
||
else:
|
||
if routers_ip not in net:
|
||
errs.append(f"网关 {self.routers} 不在 subnet {self.network}/{self.netmask} 内")
|
||
for d in [x.strip() for x in self.dns.split(",") if x.strip()]:
|
||
try:
|
||
ipaddress.ip_address(d)
|
||
except ValueError:
|
||
errs.append(f"DNS 格式错误: {d}")
|
||
if self.default_lease <= 0 or self.max_lease <= 0:
|
||
errs.append("租期时间必须为正数")
|
||
if self.default_lease > self.max_lease:
|
||
errs.append("默认租期不能大于最大租期")
|
||
return errs
|
||
|
||
def to_conf(self) -> str:
|
||
"""生成 dhcpd.conf 子网块"""
|
||
lines = []
|
||
if self.authoritative:
|
||
lines.append("authoritative;")
|
||
lines.append(f"subnet {self.network} netmask {self.netmask} {{")
|
||
if self.domain_name:
|
||
lines.append(f" option domain-name \"{self.domain_name}\";")
|
||
lines.append(f" option routers {self.routers};")
|
||
dns_list = ", ".join(x.strip() for x in self.dns.split(",") if x.strip())
|
||
lines.append(f" option subnet-mask {self.netmask};")
|
||
lines.append(f" option domain-name-servers {dns_list};")
|
||
lines.append(f" default-lease-time {self.default_lease};")
|
||
lines.append(f" max-lease-time {self.max_lease};")
|
||
lines.append(f" range {self.range_start} {self.range_end};")
|
||
lines.append("}")
|
||
return "\n".join(lines)
|
||
|
||
|
||
@dataclass
|
||
class Host:
|
||
name: str # e.g. "printer-hall"
|
||
mac: str # e.g. "aa:bb:cc:dd:ee:ff"
|
||
ip: str # e.g. "192.168.1.50"
|
||
subnet_network: str = "" # 关联 subnet network（可选，不填则不强制匹配）
|
||
|
||
def validate(self) -> List[str]:
|
||
errs = []
|
||
if not re.match(r"^[A-Za-z0-9_\-\.]{1,63}$", self.name):
|
||
errs.append("主机名仅允许字母数字、下划线、连字符、点，长度 1-63")
|
||
if not re.match(r"^([0-9a-fA-F]{2}[:\-]){5}[0-9a-fA-F]{2}$", self.mac):
|
||
errs.append(f"MAC 格式错误: {self.mac} （应为 aa:bb:cc:dd:ee:ff）")
|
||
try:
|
||
ipaddress.ip_address(self.ip)
|
||
except ValueError:
|
||
errs.append(f"IP 无效: {self.ip}")
|
||
return errs
|
||
|
||
def to_conf(self) -> str:
|
||
return (
|
||
f"host {self.name} {{\n"
|
||
f" hardware ethernet {self.mac};\n"
|
||
f" fixed-address {self.ip};\n"
|
||
f"}}"
|
||
)
|
||
|
||
|
||
@dataclass
|
||
class Lease:
|
||
ip: str
|
||
starts: str # 原始文本
|
||
ends: str
|
||
tstp: str # 硬过期
|
||
mac: str
|
||
client_hostname: str = ""
|
||
state: str = "active" # active / abandoned / free
|
||
binding_state: str = "active"
|
||
next_event: str = ""
|
||
raw: str = ""
|
||
|
||
def expires_at_dt(self) -> Optional[datetime]:
|
||
"""解析 ends/tstp 时间为 datetime，无法解析返回 None"""
|
||
candidates = [self.ends, self.tstp]
|
||
for raw in candidates:
|
||
if not raw or raw == "never":
|
||
continue
|
||
raw = raw.strip()
|
||
# ISC 格式: 2024/07/09 14:23:01 或带时区
|
||
for fmt in ("%Y/%m/%d %H:%M:%S %Z", "%Y/%m/%d %H:%M:%S",
|
||
"%Y-%m-%d %H:%M:%S"):
|
||
try:
|
||
return datetime.strptime(raw, fmt)
|
||
except ValueError:
|
||
continue
|
||
return None
|
||
|
||
def remaining_seconds(self) -> Optional[int]:
|
||
dt = self.expires_at_dt()
|
||
if dt is None:
|
||
return None
|
||
return int((dt - datetime.now()).total_seconds())
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 配置文件读写
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def _backup_file(path: str) -> Optional[str]:
|
||
"""复制文件到 BACKUP_DIR，保留最近 20 份。返回备份路径"""
|
||
if not os.path.exists(path):
|
||
return None
|
||
os.makedirs(BACKUP_DIR, exist_ok=True)
|
||
ts = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||
dst = os.path.join(BACKUP_DIR, f"{os.path.basename(path)}.{ts}.bak")
|
||
shutil.copy2(path, dst)
|
||
# 清理老备份
|
||
backups = sorted(
|
||
[f for f in os.listdir(BACKUP_DIR) if f.startswith(os.path.basename(path) + ".")],
|
||
reverse=True,
|
||
)
|
||
for old in backups[20:]:
|
||
try:
|
||
os.remove(os.path.join(BACKUP_DIR, old))
|
||
except OSError:
|
||
pass
|
||
return dst
|
||
|
||
|
||
def read_conf_header() -> str:
|
||
"""读取 dhcpd.conf 中 BEGIN_MARKER 之前的部分（保留通用配置）"""
|
||
if not os.path.exists(DHCPD_CONF):
|
||
return ""
|
||
with open(DHCPD_CONF, "r") as f:
|
||
content = f.read()
|
||
idx = content.find(BEGIN_MARKER)
|
||
if idx == -1:
|
||
return content.rstrip() + "\n"
|
||
return content[:idx].rstrip() + "\n"
|
||
|
||
|
||
def read_managed_block() -> str:
|
||
"""读取 BEGIN/END 标记之间的内容（去掉标记本身）"""
|
||
if not os.path.exists(DHCPD_CONF):
|
||
return ""
|
||
with open(DHCPD_CONF, "r") as f:
|
||
content = f.read()
|
||
begin = content.find(BEGIN_MARKER)
|
||
end = content.find(END_MARKER)
|
||
if begin == -1 or end == -1:
|
||
return ""
|
||
return content[begin + len(BEGIN_MARKER):end].strip()
|
||
|
||
|
||
def write_managed_block(subnets: List[Subnet], hosts: List[Host]) -> Tuple[bool, str]:
|
||
"""
|
||
把 subnets + hosts 写入 dhcpd.conf，保留原有通用配置。
|
||
返回 (success, message)
|
||
"""
|
||
header = read_conf_header()
|
||
lines = [header.rstrip(), "", BEGIN_MARKER]
|
||
if subnets:
|
||
for s in subnets:
|
||
lines.append(s.to_conf())
|
||
lines.append("")
|
||
if hosts:
|
||
for h in hosts:
|
||
lines.append(h.to_conf())
|
||
lines.append("")
|
||
lines.append(END_MARKER)
|
||
lines.append("")
|
||
new_content = "\n".join(lines)
|
||
|
||
backup = _backup_file(DHCPD_CONF)
|
||
try:
|
||
with open(DHCPD_CONF, "w") as f:
|
||
f.write(new_content)
|
||
except PermissionError as e:
|
||
return False, f"写入失败（权限不足）: {e}"
|
||
msg = "已写入"
|
||
if backup:
|
||
msg += f"，备份至 {backup}"
|
||
return True, msg
|
||
|
||
|
||
def parse_managed_block(block: str) -> Tuple[List[Subnet], List[Host]]:
|
||
"""从 BEGIN/END 块文本中解析出 subnets 和 hosts 列表"""
|
||
subnets: List[Subnet] = []
|
||
hosts: List[Host] = []
|
||
|
||
# subnet 块
|
||
for m in re.finditer(
|
||
r"subnet\s+(\S+)\s+netmask\s+(\S+)\s*\{([^}]*)\}",
|
||
block,
|
||
re.DOTALL,
|
||
):
|
||
network, netmask, body = m.group(1), m.group(2), m.group(3)
|
||
d = {"network": network, "netmask": netmask}
|
||
for line in body.splitlines():
|
||
line = line.strip().rstrip(";").strip()
|
||
if line.startswith("option routers "):
|
||
parts = line.split()
|
||
if len(parts) >= 3:
|
||
d["routers"] = parts[2]
|
||
elif line.startswith("option domain-name-servers "):
|
||
parts = line.split(None, 2)
|
||
if len(parts) >= 3:
|
||
dns_parts = parts[2].split(",")
|
||
d["dns"] = ", ".join(p.strip() for p in dns_parts)
|
||
elif line.startswith("option domain-name ") and not line.startswith("option domain-name-servers"):
|
||
if "\"" in line:
|
||
d["domain_name"] = line.split("\"")[1]
|
||
elif line.startswith("range "):
|
||
_, start, end = line.split()
|
||
d["range_start"], d["range_end"] = start, end
|
||
elif line.startswith("default-lease-time"):
|
||
try:
|
||
d["default_lease"] = int(line.split()[1])
|
||
except (ValueError, IndexError):
|
||
pass
|
||
elif line.startswith("max-lease-time"):
|
||
try:
|
||
d["max_lease"] = int(line.split()[1])
|
||
except (ValueError, IndexError):
|
||
pass
|
||
# authoritative 在 subnet 外也可能 —— 我们生成时会放到 subnet 块外
|
||
defaults = {
|
||
"network": "", "netmask": "", "routers": "", "dns": "",
|
||
"range_start": "", "range_end": "", "default_lease": 600,
|
||
"max_lease": 7200, "domain_name": "", "authoritative": False,
|
||
}
|
||
subnets.append(Subnet(**{k: d.get(k, defaults[k]) for k in defaults}))
|
||
|
||
# authoritative
|
||
if "authoritative;" in block:
|
||
if subnets:
|
||
subnets[0].authoritative = True
|
||
|
||
# host 块
|
||
for m in re.finditer(
|
||
r"host\s+(\S+)\s*\{([^}]*)\}",
|
||
block,
|
||
re.DOTALL,
|
||
):
|
||
name = m.group(1)
|
||
body = m.group(2)
|
||
d = {"name": name}
|
||
for line in body.splitlines():
|
||
line = line.strip().rstrip(";").strip()
|
||
if line.startswith("hardware ethernet"):
|
||
parts = line.split()
|
||
if len(parts) >= 3:
|
||
d["mac"] = parts[2]
|
||
elif line.startswith("fixed-address"):
|
||
parts = line.split()
|
||
if len(parts) >= 2:
|
||
d["ip"] = parts[1]
|
||
if "mac" in d and "ip" in d:
|
||
hosts.append(Host(**d))
|
||
|
||
return subnets, hosts
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 租约文件解析
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def parse_leases() -> List[Lease]:
|
||
"""解析 dhcpd.leases，返回所有 lease（含重复的 binding state 变更）"""
|
||
if not os.path.exists(DHCP_LEASES):
|
||
return []
|
||
with open(DHCP_LEASES, "r") as f:
|
||
content = f.read()
|
||
leases: List[Lease] = []
|
||
for m in re.finditer(r"lease\s+(\S+)\s*\{([^}]*)\}", content, re.DOTALL):
|
||
ip = m.group(1)
|
||
body = m.group(2)
|
||
d = {"ip": ip}
|
||
for line in body.splitlines():
|
||
line = line.strip()
|
||
if not line or line.startswith("#"):
|
||
continue
|
||
key = line.split(None, 1)[0].rstrip(";")
|
||
val = line.split(None, 1)[1].rstrip(";").strip() if " " in line else ""
|
||
if key == "starts":
|
||
d["starts"] = val
|
||
elif key == "ends":
|
||
d["ends"] = val
|
||
elif key == "tstp":
|
||
d["tstp"] = val
|
||
elif key == "binding":
|
||
d["binding_state"] = val.split()[0] if val else "active"
|
||
d["state"] = val.split()[0] if val else "active"
|
||
elif key == "next":
|
||
d["next_event"] = val
|
||
elif key == "hardware":
|
||
# hardware ethernet aa:bb:cc:dd:ee:ff
|
||
parts = val.split()
|
||
if len(parts) == 2:
|
||
d["mac"] = parts[1]
|
||
elif key == "client-hostname":
|
||
d["client_hostname"] = val.strip('"')
|
||
d["raw"] = m.group(0)
|
||
leases.append(Lease(**{k: d.get(k, "") for k in
|
||
["ip", "starts", "ends", "tstp", "mac",
|
||
"client_hostname", "state", "binding_state",
|
||
"next_event", "raw"]}))
|
||
return leases
|
||
|
||
|
||
def active_leases() -> List[Lease]:
|
||
"""仅返回 binding state = active 且未过期的 lease"""
|
||
result = []
|
||
for lease in parse_leases():
|
||
if lease.binding_state != "active":
|
||
continue
|
||
rem = lease.remaining_seconds()
|
||
if rem is not None and rem <= 0:
|
||
continue
|
||
result.append(lease)
|
||
return result
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 服务控制 / 语法检查
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def check_syntax() -> Tuple[bool, str]:
|
||
"""运行 dhcpd -t 语法检查"""
|
||
try:
|
||
r = subprocess.run(
|
||
["dhcpd", "-t", "-cf", DHCPD_CONF],
|
||
capture_output=True, text=True, timeout=10,
|
||
)
|
||
out = (r.stdout + r.stderr).strip()
|
||
if r.returncode == 0:
|
||
return True, out or "语法 OK"
|
||
return False, out
|
||
except FileNotFoundError:
|
||
return False, "未找到 dhcpd 二进制"
|
||
except subprocess.TimeoutExpired:
|
||
return False, "dhcpd -t 超时"
|
||
|
||
|
||
def reload_service() -> Tuple[bool, str]:
|
||
"""重载 DHCP 服务（不丢连接）。
|
||
|
||
isc-dhcp-server.service 是 sysv-init 转 systemd 的，**不支持 reload**，
|
||
所以直接用 try-reload-or-restart —— 它会优先 reload，不支持就 restart。
|
||
失败时附带 journalctl 最近日志，便于排查。
|
||
"""
|
||
# isc-dhcp-server.service 是 sysv-init 转 systemd 的，**不支持 reload**，
|
||
# 且 systemctl try-reload-or-restart 对 failed unit 也会返回 0，
|
||
# 不能反映启动失败。策略：直接 restart + 主动 is-active 检查。
|
||
try:
|
||
r = subprocess.run(
|
||
["systemctl", "restart", SERVICE_NAME],
|
||
capture_output=True, text=True, timeout=20,
|
||
)
|
||
if r.returncode != 0:
|
||
out = (r.stdout + r.stderr).strip()
|
||
diag = _last_journal_lines(15)
|
||
return False, f"restart 命令失败: {out}\n\n--- journalctl -xeu {SERVICE_NAME} (最近) ---\n{diag}".strip()
|
||
time.sleep(1.5)
|
||
active = subprocess.run(
|
||
["systemctl", "is-active", "--quiet", SERVICE_NAME],
|
||
timeout=5,
|
||
).returncode == 0
|
||
if active:
|
||
return True, "服务已 restart"
|
||
diag = _last_journal_lines(15)
|
||
return False, f"restart 命令返回 0 但服务未进入 active 状态。\n\n--- journalctl -xeu {SERVICE_NAME} (最近) ---\n{diag}".strip()
|
||
except subprocess.TimeoutExpired:
|
||
return False, "systemctl 调用超时"
|
||
except FileNotFoundError as e:
|
||
return False, f"未找到 systemctl: {e}"
|
||
|
||
|
||
def restart_service() -> Tuple[bool, str]:
|
||
"""强制重启 DHCP 服务（与 reload_service 同语义，因本服务不支持 reload）"""
|
||
return reload_service()
|
||
|
||
|
||
def _last_journal_lines(n: int = 15) -> str:
|
||
"""获取 service 的最近日志（用于错误诊断）"""
|
||
try:
|
||
r = subprocess.run(
|
||
["journalctl", "-xeu", SERVICE_NAME, "-n", str(n), "--no-pager"],
|
||
capture_output=True, text=True, timeout=5,
|
||
)
|
||
return r.stdout.strip() or r.stderr.strip()
|
||
except Exception as e:
|
||
return f"(读取日志失败: {e})"
|
||
|
||
|
||
def get_interfaces() -> List[str]:
|
||
"""读取 INTERFACESv4 配置"""
|
||
if not os.path.exists(INTERFACES_CONF):
|
||
return []
|
||
v4 = []
|
||
for line in open(INTERFACES_CONF):
|
||
line = line.strip()
|
||
if line.startswith("INTERFACESv4="):
|
||
val = line.split("=", 1)[1].strip().strip('"')
|
||
v4 = [x for x in val.split() if x]
|
||
return v4
|
||
|
||
|
||
def save_interfaces(ifaces: List[str]) -> Tuple[bool, str]:
|
||
"""保存 INTERFACESv4 配置。空列表表示监听全部"""
|
||
if not os.path.exists(INTERFACES_CONF):
|
||
return False, f"文件不存在: {INTERFACES_CONF}"
|
||
backup = _backup_file(INTERFACES_CONF)
|
||
with open(INTERFACES_CONF, "r") as f:
|
||
lines = f.readlines()
|
||
new_val = " ".join(ifaces)
|
||
found = False
|
||
for i, line in enumerate(lines):
|
||
if line.strip().startswith("INTERFACESv4="):
|
||
lines[i] = f'INTERFACESv4="{new_val}"\n'
|
||
found = True
|
||
break
|
||
if not found:
|
||
lines.append(f'INTERFACESv4="{new_val}"\n')
|
||
with open(INTERFACES_CONF, "w") as f:
|
||
f.writelines(lines)
|
||
msg = "已保存接口配置"
|
||
if backup:
|
||
msg += f"，备份至 {backup}"
|
||
return True, msg
|
||
|
||
|
||
def get_service_status() -> Dict:
|
||
"""返回 dhcpd 服务状态摘要"""
|
||
info = {
|
||
"active": False,
|
||
"status_text": "",
|
||
"interfaces": get_interfaces(),
|
||
}
|
||
try:
|
||
r = subprocess.run(
|
||
["systemctl", "is-active", SERVICE_NAME],
|
||
capture_output=True, text=True, timeout=5,
|
||
)
|
||
info["active"] = r.stdout.strip() == "active"
|
||
r2 = subprocess.run(
|
||
["systemctl", "status", SERVICE_NAME, "--no-pager", "-n", "3"],
|
||
capture_output=True, text=True, timeout=5,
|
||
)
|
||
info["status_text"] = (r2.stdout or r2.stderr).strip().split("\n", 3)
|
||
info["status_text"] = "\n".join(info["status_text"][:5])
|
||
except (subprocess.TimeoutExpired, FileNotFoundError):
|
||
pass
|
||
return info
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 网络扫描（arp-scan）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
@dataclass
|
||
class ArpDevice:
|
||
ip: str
|
||
mac: str
|
||
vendor: str = ""
|
||
source: str = "" # 标记：static / dynamic / unknown
|
||
|
||
|
||
def get_local_network(iface: str = "enp0s31f6-ovs") -> Optional[str]:
|
||
"""从指定接口的 IPv4 地址 + 掩码计算出 /24 网络号"""
|
||
try:
|
||
out = subprocess.run(
|
||
["ip", "-4", "-o", "addr", "show", "dev", iface],
|
||
capture_output=True, text=True, timeout=5,
|
||
).stdout
|
||
# 形如: "4: enp0s31f6-ovs inet 10.168.1.209/24 brd 10.168.1.255 scope global enp0s31f6-ovs\"
|
||
import re as _re
|
||
m = _re.search(r"inet\s+(\d+\.\d+\.\d+\.\d+)/(\d+)", out)
|
||
if not m:
|
||
return None
|
||
ip, prefix = m.group(1), int(m.group(2))
|
||
if prefix != 24:
|
||
# 仅支持 /24 扫描；其他掩码返回 None 让 UI 提示用户
|
||
return None
|
||
return ".".join(ip.split(".")[:3]) + ".0/24"
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def arp_scan(iface: str = "enp0s31f6-ovs", network: Optional[str] = None,
|
||
timeout: int = 10) -> Tuple[bool, str, List[ArpDevice]]:
|
||
"""
|
||
扫描指定接口/网段，返回 (ok, msg, devices)。
|
||
- 用 arp-scan（系统二进制）
|
||
- 如果传 network=None，自动从 iface 推算
|
||
"""
|
||
if network is None:
|
||
network = get_local_network(iface) or "10.168.1.0/24"
|
||
|
||
cmd = ["arp-scan", f"--interface={iface}", "--numeric", "--quiet", "--retry=1",
|
||
"--timeout=" + str(timeout * 1000), network]
|
||
try:
|
||
r = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout + 5)
|
||
if r.returncode not in (0, 1): # arp-scan: 0=全部收到, 1=部分收到
|
||
return False, (r.stderr or r.stdout).strip(), []
|
||
except FileNotFoundError:
|
||
return False, "未安装 arp-scan（apt-get install arp-scan）", []
|
||
except subprocess.TimeoutExpired:
|
||
return False, "arp-scan 超时", []
|
||
|
||
devices: List[ArpDevice] = []
|
||
for line in r.stdout.splitlines():
|
||
line = line.strip()
|
||
if not line or line.startswith("Interface:") or "arp-scan" in line or "packets" in line:
|
||
continue
|
||
parts = line.split()
|
||
if len(parts) >= 2 and "." in parts[0] and ":" in parts[1]:
|
||
devices.append(ArpDevice(ip=parts[0], mac=parts[1].lower()))
|
||
msg = f"扫描完成，发现 {len(devices)} 个设备"
|
||
return True, msg, devices
|
||
|
||
|
||
def classify_devices(devices: List[ArpDevice], hosts, leases) -> List[ArpDevice]:
|
||
"""
|
||
给设备打标记：
|
||
- static: 在静态绑定（host 声明）里
|
||
- dynamic: 在 dhcpd.leases 里
|
||
- unknown: 都不在（手动配的或别处 DHCP 给的）
|
||
"""
|
||
static_macs = {h.mac.lower() for h in hosts}
|
||
static_ips = {h.ip for h in hosts}
|
||
lease_macs = {l.mac.lower() for l in leases if l.mac}
|
||
lease_ips = {l.ip for l in leases}
|
||
for d in devices:
|
||
if d.mac in static_macs or d.ip in static_ips:
|
||
d.source = "static"
|
||
elif d.mac in lease_macs or d.ip in lease_ips:
|
||
d.source = "dynamic"
|
||
else:
|
||
d.source = "unknown"
|
||
return devices |