From 8527f21ffcb6bbe53efeafae888dbec311fe1a0d Mon Sep 17 00:00:00 2001 From: Wang Defa Date: Thu, 23 Jul 2026 10:18:39 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=9C=A8=20create=5Finstance.sh=20?= =?UTF-8?q?=E4=B8=AD=E6=B7=BB=E5=8A=A0=E7=A7=9F=E6=88=B7=E5=90=8D=E7=A7=B0?= =?UTF-8?q?=E8=8E=B7=E5=8F=96=E5=8A=9F=E8=83=BD=E5=B9=B6=E6=9B=B4=E6=96=B0?= =?UTF-8?q?=E6=A0=87=E7=AD=BE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- oci/create_instance.sh | 10 +- oci/sync_interfaces.py | 352 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 361 insertions(+), 1 deletion(-) create mode 100755 oci/sync_interfaces.py diff --git a/oci/create_instance.sh b/oci/create_instance.sh index 40b662d..04ac585 100644 --- a/oci/create_instance.sh +++ b/oci/create_instance.sh @@ -409,7 +409,15 @@ main() { fi local metadata="{\"user_data\":\"$(generate_user_data "$password")\"}" - local tags="{\"RootPassword\":\"$password\"}" + + log_info "正在获取租户名称..." + local tenant_name + tenant_name=$(oci iam tenancy get --tenancy-id "$ocid" \ + --query 'data.name' --raw-output 2>/dev/null) + check_return $? "获取租户名称失败" + log_info "租户名称: $tenant_name" + + local tags="{\"RootPassword\":\"$password\",\"TenantName\":\"$tenant_name\"}" # 获取可用域 log_info "正在获取可用域列表..." diff --git a/oci/sync_interfaces.py b/oci/sync_interfaces.py new file mode 100755 index 0000000..44d3059 --- /dev/null +++ b/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 "") + 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())