feat: 在 create_instance.sh 中添加租户名称获取功能并更新标签

oci/sync_interfaces.py

@@ -0,0 +1,352 @@
+#!/usr/bin/env python3
+"""Generate persistent ifupdown configuration for OCI secondary VNICs."""
+
+from __future__ import annotations
+
+import argparse
+import datetime as dt
+import ipaddress
+import json
+import os
+import re
+import shutil
+import subprocess
+import sys
+import tempfile
+import urllib.request
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+
+METADATA_URL = "http://169.254.169.254/opc/v2/vnics/"
+BEGIN_MARKER = "# BEGIN OCI SECONDARY VNICS - managed by oci-sync-interfaces.py"
+END_MARKER = "# END OCI SECONDARY VNICS - managed by oci-sync-interfaces.py"
+TOP_LEVEL = re.compile(r"^(auto|allow-hotplug|iface|mapping|source|source-directory)\b")
+IFACE_NAME = re.compile(r"^[A-Za-z0-9_.:-]+$")
+
+
+@dataclass(frozen=True)
+class Vnic:
+    interface: str
+    ipv4: tuple[ipaddress.IPv4Interface, ...]
+    ipv4_networks: tuple[ipaddress.IPv4Network, ...]
+    gateway4: ipaddress.IPv4Address
+    ipv6: tuple[ipaddress.IPv6Interface, ...]
+    gateway6: ipaddress.IPv6Address | None
+    table: int
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--interfaces", default="/etc/network/interfaces")
+    parser.add_argument("--metadata-url", default=METADATA_URL)
+    parser.add_argument("--metadata-file", type=Path)
+    parser.add_argument("--sys-class-net", type=Path, default=Path("/sys/class/net"))
+    parser.add_argument("--primary-interface")
+    parser.add_argument("--dry-run", action="store_true")
+    return parser.parse_args()
+
+
+def fetch_metadata(url: str) -> list[dict[str, Any]]:
+    request = urllib.request.Request(url, headers={"Authorization": "Bearer Oracle"})
+    with urllib.request.urlopen(request, timeout=5) as response:
+        payload = json.load(response)
+    if not isinstance(payload, list) or not payload:
+        raise ValueError("OCI 元数据未返回 VNIC 列表")
+    return payload
+
+
+def load_metadata(args: argparse.Namespace) -> list[dict[str, Any]]:
+    if not args.metadata_file:
+        return fetch_metadata(args.metadata_url)
+    payload = json.loads(args.metadata_file.read_text(encoding="utf-8"))
+    if not isinstance(payload, list) or not payload:
+        raise ValueError("元数据文件不是非空 VNIC 列表")
+    return payload
+
+
+def local_interfaces(root: Path) -> dict[str, str]:
+    result: dict[str, str] = {}
+    for address_file in root.glob("*/address"):
+        mac = address_file.read_text(encoding="ascii").strip().lower()
+        result[mac] = address_file.parent.name
+    return result
+
+
+def match_vnics(
+    metadata: list[dict[str, Any]], interfaces: dict[str, str]
+) -> list[tuple[dict[str, Any], str]]:
+    matched: list[tuple[dict[str, Any], str]] = []
+    missing: list[str] = []
+    for item in metadata:
+        mac = str(item.get("macAddr", "")).lower()
+        if not mac or mac not in interfaces:
+            missing.append(mac or "<missing macAddr>")
+            continue
+        matched.append((item, interfaces[mac]))
+    if missing:
+        raise RuntimeError("以下 OCI VNIC 尚未出现在系统中: " + ", ".join(missing))
+    return matched
+
+
+def default_interface() -> str | None:
+    command = ["ip", "-4", "route", "show", "default"]
+    result = subprocess.run(command, check=False, capture_output=True, text=True)
+    for line in result.stdout.splitlines():
+        match = re.search(r"\bdev\s+(\S+)", line)
+        if match:
+            return match.group(1)
+    return None
+
+
+def choose_primary(
+    pairs: list[tuple[dict[str, Any], str]], requested: str | None
+) -> str:
+    names = {name for _, name in pairs}
+    candidate = requested or default_interface()
+    if candidate in names:
+        return str(candidate)
+    if requested:
+        raise ValueError(f"指定的主接口不在 OCI 元数据中: {requested}")
+    return pairs[0][1]
+
+
+def listed_networks(item: dict[str, Any], family: int) -> list[str]:
+    if family == 4:
+        plural, singular = "subnetCidrBlocks", "subnetCidrBlock"
+    else:
+        plural, singular = "ipv6SubnetCidrBlocks", "ipv6SubnetCidrBlock"
+    values = list(item.get(plural) or [])
+    if item.get(singular) and item[singular] not in values:
+        values.append(item[singular])
+    return values
+
+
+def address_interface(address: str, networks: list[str]) -> ipaddress._BaseInterface:
+    ip = ipaddress.ip_address(address)
+    parsed = [ipaddress.ip_network(value) for value in networks]
+    for network in parsed:
+        if ip in network:
+            return ipaddress.ip_interface(f"{ip}/{network.prefixlen}")
+    raise ValueError(f"地址 {ip} 不属于元数据中的任何子网")
+
+
+def parse_ipv4(item: dict[str, Any]) -> tuple[ipaddress.IPv4Interface, ...]:
+    addresses = [item.get("privateIp"), *(item.get("secondaryPrivateIps") or [])]
+    networks = listed_networks(item, 4)
+    parsed = [address_interface(str(value), networks) for value in addresses if value]
+    if not parsed or not all(isinstance(value, ipaddress.IPv4Interface) for value in parsed):
+        raise ValueError("VNIC 缺少有效的主 IPv4 地址")
+    return tuple(parsed)  # type: ignore[return-value]
+
+
+def parse_ipv6(item: dict[str, Any]) -> tuple[ipaddress.IPv6Interface, ...]:
+    networks = listed_networks(item, 6)
+    parsed = [address_interface(str(value), networks) for value in item.get("ipv6Addresses") or []]
+    if not all(isinstance(value, ipaddress.IPv6Interface) for value in parsed):
+        raise ValueError("VNIC 包含无效的 IPv6 地址")
+    return tuple(parsed)  # type: ignore[return-value]
+
+
+def make_vnic(item: dict[str, Any], interface: str, table: int) -> Vnic:
+    if not IFACE_NAME.fullmatch(interface):
+        raise ValueError(f"非法接口名: {interface}")
+    ipv4 = parse_ipv4(item)
+    networks = tuple(dict.fromkeys(value.network for value in ipv4))
+    gateway4 = ipaddress.IPv4Address(item["virtualRouterIp"])
+    gateway6_value = item.get("ipv6VirtualRouterIp")
+    gateway6 = ipaddress.IPv6Address(gateway6_value) if gateway6_value else None
+    return Vnic(interface, ipv4, networks, gateway4, parse_ipv6(item), gateway6, table)
+
+
+def render_ipv4(vnic: Vnic) -> list[str]:
+    name, table = vnic.interface, vnic.table
+    lines: list[str] = []
+    for address in vnic.ipv4:
+        lines.append(f"    up ip -4 addr replace {address} dev {name} noprefixroute")
+    for network in vnic.ipv4_networks:
+        source = next(value.ip for value in vnic.ipv4 if value.ip in network)
+        lines.append(f"    up ip -4 route replace {network} dev {name} src {source} table {table}")
+    lines.append(f"    up ip -4 route replace default via {vnic.gateway4} dev {name} table {table}")
+    for index, address in enumerate(vnic.ipv4):
+        priority = 10000 + table * 10 + index
+        lines.extend(render_rule(4, str(address.ip), table, priority))
+    return lines
+
+
+def render_rule(family: int, address: str, table: int, priority: int) -> list[str]:
+    ip = "ip -4" if family == 4 else "ip -6"
+    prefix = 32 if family == 4 else 128
+    selector = f"from {address}/{prefix} table {table} priority {priority}"
+    return [
+        f"    up {ip} rule del {selector} 2>/dev/null || true",
+        f"    up {ip} rule add {selector}",
+        f"    down {ip} rule del {selector} 2>/dev/null || true",
+    ]
+
+
+def render_ipv6(vnic: Vnic) -> list[str]:
+    if not vnic.ipv6:
+        return []
+    if vnic.gateway6 is None:
+        raise ValueError(f"{vnic.interface} 有 IPv6 地址但没有 IPv6 网关")
+    name, table = vnic.interface, vnic.table
+    lines: list[str] = []
+    for address in vnic.ipv6:
+        lines.append(f"    up ip -6 addr replace {address.ip}/128 dev {name}")
+    lines.append(f"    up ip -6 route replace default via {vnic.gateway6} dev {name} table {table}")
+    for index, address in enumerate(vnic.ipv6):
+        priority = 10000 + table * 10 + index
+        lines.extend(render_rule(6, str(address.ip), table, priority))
+    return lines
+
+
+def render_cleanup(vnic: Vnic) -> list[str]:
+    name, table = vnic.interface, vnic.table
+    lines = [
+        f"    down ip -4 route flush table {table} 2>/dev/null || true",
+        f"    down ip -6 route flush table {table} 2>/dev/null || true",
+    ]
+    for address in vnic.ipv4:
+        lines.append(f"    down ip -4 addr del {address} dev {name} 2>/dev/null || true")
+    for address in vnic.ipv6:
+        lines.append(f"    down ip -6 addr del {address.ip}/128 dev {name} 2>/dev/null || true")
+    return lines
+
+
+def render_vnic(vnic: Vnic) -> str:
+    name = vnic.interface
+    lines = [
+        f"auto {name}",
+        f"iface {name} inet manual",
+        f"    pre-up sysctl -qw net.ipv4.conf.{name}.rp_filter=2",
+    ]
+    if vnic.ipv6:
+        lines.append(f"    pre-up sysctl -qw net.ipv6.conf.{name}.accept_ra=0")
+    lines.append(f"    pre-up ip link set dev {name} up")
+    lines.extend(render_ipv4(vnic))
+    lines.extend(render_ipv6(vnic))
+    lines.extend(render_cleanup(vnic))
+    return "\n".join(lines)
+
+
+def remove_managed_block(text: str) -> str:
+    begin_count, end_count = text.count(BEGIN_MARKER), text.count(END_MARKER)
+    if begin_count != end_count or begin_count > 1:
+        raise ValueError("托管块标记缺失或重复，请先人工检查 interfaces 文件")
+    if not begin_count:
+        return text
+    pattern = re.compile(
+        rf"(?ms)^{re.escape(BEGIN_MARKER)}\n.*?^{re.escape(END_MARKER)}\n?"
+    )
+    return pattern.sub("", text)
+
+
+def filter_auto_line(line: str, targets: set[str]) -> str:
+    newline = "\n" if line.endswith("\n") else ""
+    words = line.strip().split()
+    remaining = [word for word in words[1:] if word not in targets]
+    return f"{words[0]} {' '.join(remaining)}{newline}" if remaining else ""
+
+
+def remove_iface_stanzas(text: str, targets: set[str]) -> str:
+    lines, output, index = text.splitlines(keepends=True), [], 0
+    while index < len(lines):
+        line = lines[index]
+        auto = re.match(r"^(auto|allow-hotplug)\s+", line)
+        iface = re.match(r"^iface\s+(\S+)\s+", line)
+        if auto:
+            output.append(filter_auto_line(line, targets))
+            index += 1
+            continue
+        if not iface or iface.group(1) not in targets:
+            output.append(line)
+            index += 1
+            continue
+        index += 1
+        while index < len(lines) and not TOP_LEVEL.match(lines[index]):
+            index += 1
+    return "".join(output)
+
+
+def compose_config(original: str, vnics: list[Vnic]) -> str:
+    base = remove_managed_block(original)
+    base = remove_iface_stanzas(base, {vnic.interface for vnic in vnics}).rstrip()
+    body = "\n\n".join(render_vnic(vnic) for vnic in vnics)
+    managed = f"{BEGIN_MARKER}\n# 请勿手工编辑此区域；重新运行脚本会覆盖它。"
+    if body:
+        managed += f"\n\n{body}"
+    return f"{base}\n\n{managed}\n{END_MARKER}\n"
+
+
+def validate_config(content: str, target: Path) -> Path:
+    descriptor, name = tempfile.mkstemp(prefix=".interfaces.", dir=target.parent, text=True)
+    temporary = Path(name)
+    with os.fdopen(descriptor, "w", encoding="utf-8") as stream:
+        stream.write(content)
+    result = subprocess.run(
+        ["ifquery", "--interfaces", str(temporary), "--list", "--all"],
+        check=False,
+        capture_output=True,
+        text=True,
+    )
+    if result.returncode:
+        temporary.unlink(missing_ok=True)
+        raise RuntimeError("ifquery 校验失败:\n" + result.stderr.strip())
+    return temporary
+
+
+def install_config(content: str, target: Path) -> Path | None:
+    current = target.read_text(encoding="utf-8")
+    if current == content:
+        return None
+    temporary = validate_config(content, target)
+    stamp = dt.datetime.now().strftime("%Y%m%d-%H%M%S-%f")
+    backup = target.with_name(f"{target.name}.bak.{stamp}")
+    shutil.copy2(target, backup)
+    os.chmod(temporary, target.stat().st_mode & 0o777)
+    os.replace(temporary, target)
+    return backup
+
+
+def build_vnics(args: argparse.Namespace) -> tuple[str, list[Vnic]]:
+    metadata = load_metadata(args)
+    pairs = match_vnics(metadata, local_interfaces(args.sys_class_net))
+    primary = choose_primary(pairs, args.primary_interface)
+    secondary = sorted(
+        ((item, name) for item, name in pairs if name != primary),
+        key=lambda pair: pair[1],
+    )
+    vnics = [make_vnic(item, name, 100 + index) for index, (item, name) in enumerate(secondary)]
+    return primary, vnics
+
+
+def run(args: argparse.Namespace) -> int:
+    target = Path(args.interfaces)
+    if not target.is_file():
+        raise FileNotFoundError(target)
+    primary, vnics = build_vnics(args)
+    content = compose_config(target.read_text(encoding="utf-8"), vnics)
+    if args.dry_run:
+        print(content, end="")
+        return 0
+    if os.geteuid() != 0:
+        raise PermissionError("修改 interfaces 文件需要 root 权限")
+    backup = install_config(content, target)
+    status = "配置未变化" if backup is None else f"已更新，备份: {backup}"
+    print(f"主接口: {primary}; 辅助 VNIC: {len(vnics)}; {status}")
+    print("未重载网络；请在控制台可用的情况下安排重启或逐个 ifup。")
+    return 0
+
+
+def main() -> int:
+    try:
+        return run(parse_args())
+    except (OSError, ValueError, RuntimeError) as error:
+        print(f"错误: {error}", file=sys.stderr)
+        return 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())