基础篇：500 行代码实现一个最小 VPC

运行前提条件

• ​系统：Linux 系统（Ubuntu 20.04+ / Debian 11+ / CentOS 8+ 均可），Windows 用户用 WSL2 也可以
• ​权限：必须有 sudo /root 权限（操作内核网络模块需要）
• ​环境​：Python 3.8+，​不需要装任何第三方库，纯标准库就能跑
• 基础：懂一点 Linux 基础命令和 Python 语法就行，零基础跟着步骤也能跑通

一、学习目标

1. 理解 VPC 的核心本质：Linux 网络命名空间 + 虚拟网桥 + veth pair + 路由表
2. 亲手实现多 VPC 逻辑隔离（支持网段重叠）
3. 掌握子网划分和软件定义路由的原理
4. 验证 VPC 最核心的两个特性：同 VPC 互通、不同 VPC 隔离

二、核心原理

VPC 并不是什么黑科技，它只是 Linux 内核四个基础网络功能的组合：

• ​网络命名空间 (netns) ：实现多租户隔离，每个命名空间有独立的网络栈
• ​虚拟网桥 (bridge) ：模拟云平台的分布式虚拟交换机
• ​veth pair：虚拟网卡对，连接命名空间和网桥

三、代码分段解析

我们将从最基础的资源管理开始，一步步构建出完整的 VPC。

3.1 全局资源管理与自动清理

做什么：追踪所有创建的资源，在程序退出时自动清理，避免资源残留。

为什么这么做：如果程序异常崩溃，网络命名空间和网桥会残留在系统中，导致下次运行失败。

import subprocess
import ipaddress
from typing import List, Dict, Optional
import atexit

# 全局资源追踪字典和列表
ALL_VPCS: Dict[str, 'VPC'] = {}
ALL_NETNS: List[str] = []
ALL_BRIDGES: List[str] = []

def cleanup_all():
    """程序退出时自动清理所有资源"""
    print("\n" + "="*50)
    print("🔄 自动清理所有创建的资源...")
    
    # 清理所有网络命名空间
    for netns in ALL_NETNS:
        subprocess.run(f"ip netns del {netns} 2>/dev/null", shell=True, capture_output=True)
        subprocess.run(f"ip link del veth-{netns} 2>/dev/null", shell=True, capture_output=True)
    
    # 清理所有虚拟网桥及其附加的 iptables 规则
    for bridge in ALL_BRIDGES:
        subprocess.run(f"iptables -D FORWARD -i {bridge} -o {bridge} -j ACCEPT 2>/dev/null", shell=True, capture_output=True)
        subprocess.run(f"iptables -D FORWARD -i {bridge} -o br-+ -j DROP 2>/dev/null", shell=True, capture_output=True)
        subprocess.run(f"iptables -D FORWARD -o {bridge} -i br-+ -j DROP 2>/dev/null", shell=True, capture_output=True)
        subprocess.run(f"ip link del {bridge} 2>/dev/null", shell=True, capture_output=True)
    
    print("✅ 所有资源清理完成")

# 注册退出钩子，无论程序正常退出还是异常崩溃都会执行
atexit.register(cleanup_all)

​关键要点：

• ​atexit.register()是 Python 的标准库函数，用于注册程序退出时的回调函数
• 所有命令都加上了2>/dev/null，避免清理不存在的资源时报错

3.2 VPC 核心类实现

做什么：模拟云厂商的虚拟私有云，对应控制台的 "创建 VPC" 操作。

为什么这么做：每个 VPC 有自己独立的虚拟网桥和路由表，实现不同 VPC 之间的逻辑隔离。

class VPC:
    """
    模拟虚拟私有云(VPC)
    对应云厂商控制台：创建VPC
    """
    def __init__(self, name: str, cidr: str):
        self.name = name  # VPC名称
        self.cidr = ipaddress.IPv4Network(cidr)  # VPC网段
        self.subnets: List[Subnet] = []  # 该VPC下的所有子网
        self.route_table: Dict[str, str] = {}  # VPC路由表
        self.bridge_name = f"br-{self.name}"  # 虚拟网桥名称
        
        # 创建VPC对应的虚拟网桥
        self._create_bridge()
        # 将VPC加入全局资源追踪
        ALL_VPCS[name] = self
        ALL_BRIDGES.append(self.bridge_name)
        
        print(f"✅ 创建VPC [{self.name}]，网段：{self.cidr}")
        
    def _create_bridge(self):
        """创建VPC内部的虚拟网桥（对应分布式虚拟交换机DVS）"""
        # 先删除可能存在的旧网桥，避免重复创建报错
        subprocess.run(f"ip link del {self.bridge_name} 2>/dev/null", shell=True, capture_output=True)
        # 创建Linux bridge虚拟网桥
        subprocess.run(f"ip link add {self.bridge_name} type bridge", shell=True, check=True)
        # 启用网桥
        subprocess.run(f"ip link set {self.bridge_name} up", shell=True, check=True)
        
        # 如果安装了 Docker，FORWARD 链默认会被设置为 DROP，导致跨子网不通
        # 添加 iptables 规则：允许同 VPC 内部跨子网转发，同时隔离不同 VPC！
        # 注意：使用 -I 插入时，后执行的命令会在链的最上方！所以先 DROP，后 ACCEPT
        # 1. 隔离不同 VPC（丢弃跨网桥流量）
        subprocess.run(f"iptables -I FORWARD -i {self.bridge_name} -o br-+ -j DROP", shell=True, check=True)
        subprocess.run(f"iptables -I FORWARD -o {self.bridge_name} -i br-+ -j DROP", shell=True, check=True)
        # 2. 允许同 VPC 内部互通（这条会在最顶部优先匹配）
        subprocess.run(f"iptables -I FORWARD -i {self.bridge_name} -o {self.bridge_name} -j ACCEPT", shell=True, check=True)
        
    def add_subnet(self, name: str, subnet_cidr: str) -> 'Subnet':
        """
        在VPC内创建子网
        对应云厂商控制台：创建子网
        """
        # 验证子网是否属于VPC网段
        if not ipaddress.IPv4Network(subnet_cidr).subnet_of(self.cidr):
            raise ValueError(f"子网 {subnet_cidr} 不属于VPC网段 {self.cidr}")
            
        subnet = Subnet(name, subnet_cidr, self)
        self.subnets.append(subnet)
        # 添加默认路由：同VPC内流量直接通过网桥转发
        self.route_table[subnet_cidr] = self.bridge_name
        return subnet
        
    def add_route(self, dest_cidr: str, next_hop: str):
        """添加自定义路由规则"""
        self.route_table[dest_cidr] = next_hop
        print(f"✅ 添加路由：{dest_cidr} -> {next_hop}")

​关键要点：

• 每个 VPC 对应一个独立的 Linux bridge 虚拟网桥
• 虚拟网桥是 VPC 内部的 "虚拟交换机"，所有子网和虚拟机都连接到这个网桥上
• ​ipaddress模块用于处理 IP 地址和网段的计算，避免手动解析字符串

3.3 子网类实现

做什么：模拟 VPC 内的子网，对应控制台的 "创建子网" 操作。

为什么这么做：子网是 VPC 内的 IP 地址分段，用于组织和管理不同的资源（如 Web 层、应用层、数据库层）。

class Subnet:
    """
    模拟VPC子网
    对应云厂商控制台：子网管理
    """
    def __init__(self, name: str, cidr: str, vpc: VPC):
        self.name = name  # 子网名称
        self.cidr = ipaddress.IPv4Network(cidr)  # 子网网段
        self.vpc = vpc  # 所属VPC
        self.vms: List[VM] = []  # 该子网下的所有虚拟机
        self.gateway = str(self.cidr[1])  # 子网网关使用网段第一个可用IP
        
        # 配置子网网关
        self._configure_gateway()
        print(f"✅ 创建子网 {self.name} [{self.cidr}]，网关：{self.gateway}")
        
    def _configure_gateway(self):
        """配置子网网关（网桥IP）"""
        # 先删除可能存在的旧IP
        subprocess.run(
            f"ip addr del {self.gateway}/{self.cidr.prefixlen} dev {self.vpc.bridge_name} 2>/dev/null",
            shell=True, capture_output=True
        )
        # 将网关IP配置在VPC的虚拟网桥上
        subprocess.run(
            f"ip addr add {self.gateway}/{self.cidr.prefixlen} dev {self.vpc.bridge_name}",
            shell=True, check=True
        )
        
    def create_vm(self, name: str, ip: Optional[str] = None) -> 'VM':
        """
        在子网内创建虚拟机
        对应云厂商控制台：创建云主机并加入子网
        """
        if not ip:
            # 自动分配IP：从第2个可用IP开始（第1个是网关）
            ip = str(self.cidr[len(self.vms)+2])
        else:
            # 验证IP是否属于子网
            if not ipaddress.IPv4Address(ip) in self.cidr:
                raise ValueError(f"IP {ip} 不属于子网 {self.cidr}")
                
        vm = VM(name, ip, self)
        self.vms.append(vm)
        return vm

​关键要点：

• 子网网关是子网内所有虚拟机的默认路由
• 我们将网关 IP 配置在 VPC 的虚拟网桥上，这样网桥就成为了子网的网关
• 自动分配 IP 时，跳过第一个可用 IP（网关），从第二个开始分配

3.4 虚拟机类实现

做什么：模拟云主机实例，核心隔离机制是 Linux 网络命名空间。

为什么这么做：每个虚拟机有自己独立的网络命名空间，拥有独立的网络栈、IP 地址和路由表，实现了不同虚拟机之间的逻辑隔离。

class VM:
    """
    模拟云主机实例
    核心隔离机制：Linux网络命名空间
    """
    def __init__(self, name: str, ip: str, subnet: Subnet):
        self.name = name  # 虚拟机名称
        self.ip = ipaddress.IPv4Address(ip)  # 虚拟机IP
        self.subnet = subnet  # 所属子网
        self.vpc = subnet.vpc  # 所属VPC
        self.veth_name = f"veth-{self.name}"  # 虚拟网卡名称
        
        # 创建独立的网络命名空间
        self._create_netns()
        # 将虚拟机连接到VPC网桥
        self._connect_to_vpc()
        # 加入全局资源追踪
        ALL_NETNS.append(self.name)
        
        print(f"✅ 创建虚拟机 [{self.name}]，IP：{self.ip}")
        
    def _create_netns(self):
        """创建独立的网络命名空间（实现隔离的核心）"""
        # 先删除可能存在的旧命名空间
        subprocess.run(f"ip netns del {self.name} 2>/dev/null", shell=True, capture_output=True)
        # 创建新的网络命名空间
        subprocess.run(f"ip netns add {self.name}", shell=True, check=True)
        # 启用命名空间内的回环接口lo
        subprocess.run(f"ip netns exec {self.name} ip link set lo up", shell=True, check=True)
        
    def _connect_to_vpc(self):
        """用veth pair将虚拟机连接到VPC网桥"""
        # 删除可能存在的旧veth接口
        subprocess.run(f"ip link del {self.veth_name} 2>/dev/null", shell=True, capture_output=True)
        # 创建一对虚拟网卡veth pair
        # 一端叫veth-{name}，留在主机网络命名空间
        # 另一端叫eth0，放到虚拟机的网络命名空间
        subprocess.run(
            f"ip link add {self.veth_name} type veth peer name eth0 netns {self.name}",
            shell=True, check=True
        )
        # 将主机端的veth接口连接到VPC的虚拟网桥
        subprocess.run(
            f"ip link set {self.veth_name} master {self.vpc.bridge_name}",
            shell=True, check=True
        )
        # 启用主机端的veth接口
        subprocess.run(f"ip link set {self.veth_name} up", shell=True, check=True)
        
        # 配置虚拟机内的网卡IP
        subprocess.run(
            f"ip netns exec {self.name} ip addr add {self.ip}/{self.subnet.cidr.prefixlen} dev eth0",
            shell=True, check=True
        )
        # 启用虚拟机内的eth0接口
        subprocess.run(f"ip netns exec {self.name} ip link set eth0 up", shell=True, check=True)
        # 配置虚拟机的默认路由，指向子网网关
        subprocess.run(
            f"ip netns exec {self.name} ip route add default via {self.subnet.gateway}",
            shell=True, check=True
        )
        
    def ping(self, target_ip: str, count: int = 2, timeout: int = 1) -> bool:
        """测试到目标IP的连通性"""
        print(f"\n🔍 测试连通性：{self.name}({self.ip}) → {target_ip}")
        result = subprocess.run(
            f"ip netns exec {self.name} ping -c {count} -W {timeout} {target_ip}",
            shell=True, capture_output=True, text=True
        )
        
        if result.returncode == 0:
            print("✅ 连通成功！")
            return True
        else:
            print("❌ 连通失败！")
            return False
            
    def exec(self, command: str) -> str:
        """在虚拟机内执行任意命令"""
        result = subprocess.run(
            f"ip netns exec {self.name} {command}",
            shell=True, capture_output=True, text=True
        )
        return result.stdout

​关键要点：

• ​网络命名空间是实现隔离的核心：每个命名空间有自己独立的网络栈，IP 地址可以和其他命名空间重叠
• ​veth pair 是虚拟机的 "网线" ：一端在虚拟机的命名空间（eth0），一端在主机的命名空间（veth-{name}），连接到 VPC 的虚拟网桥
• 虚拟机的默认路由指向子网网关（VPC 网桥的 IP），这样所有出子网的流量都会经过网桥转发

3.5 交互式 CLI Shell

做什么：用 Python 自带的 cmd 模块构建一个类似网络设备路由器的交互式命令行界面，替代硬编码的演示逻辑。

为什么这么做：让读者可以在终端里自由、动态地创建网络资源并验证连通性，带来真实的沉浸感操作体验。

import cmd
import shlex

class VPCShell(cmd.Cmd):
    intro = '===========================================================\n' \
            '欢迎使用 VPC 交互式模拟器（基础篇）\n' \
            '【可用指令】:\n' \
            '  1. create_vpc <name> <cidr>                   - 创建 VPC (例: create_vpc vpc1 10.0.0.0/16)\n' \
            '  2. create_subnet <vpc_name> <name> <cidr>     - 创建子网 (例: create_subnet vpc1 sub1 10.0.1.0/24)\n' \
            '  3. create_vm <subnet_cidr> <vm_name> [ip]     - 创建虚拟机 (例: create_vm 10.0.1.0/24 vm1)\n' \
            '  4. ping <src_vm_name> <dst_vm_name_or_ip>     - 连通性测试 (例: ping vm1 vm2)\n' \
            '  5. show                                       - 查看网络拓扑\n' \
            '  6. exit                                       - 退出并清理资源\n' \
            '==========================================================='
    prompt = '(VPC-Sim) '

    def emptyline(self):
        """
        空行时只显示提示符，不做任何操作
        """
        pass

    def do_create_vpc(self, arg):
        """创建 VPC: create_vpc <name> <cidr>"""
        args = shlex.split(arg)
        if len(args) != 2:
            print("❌ 用法: create_vpc <name> <cidr>")
            return
        try:
            VPC(args[0], args[1])
        except Exception as e:
            print(f"❌ 创建失败: {e}")

    def do_create_subnet(self, arg):
        """创建子网: create_subnet <vpc_name> <subnet_name> <cidr>"""
        args = shlex.split(arg)
        if len(args) != 3:
            print("❌ 用法: create_subnet <vpc_name> <subnet_name> <cidr>")
            return
        vpc = ALL_VPCS.get(args[0])
        if not vpc:
            print(f"❌ VPC {args[0]} 不存在")
            return
        try:
            vpc.add_subnet(args[1], args[2])
        except Exception as e:
            print(f"❌ 创建失败: {e}")

    def do_create_vm(self, arg):
        """创建虚拟机: create_vm <subnet_cidr> <vm_name> [ip]"""
        args = shlex.split(arg)
        if len(args) < 2:
            print("❌ 用法: create_vm <subnet_cidr> <vm_name> [ip]")
            return
        # 查找子网
        target_subnet = None
        for vpc in ALL_VPCS.values():
            for subnet in vpc.subnets:
                if str(subnet.cidr) == args[0]:
                    target_subnet = subnet
                    break
        if not target_subnet:
            print(f"❌ 子网 {args[0]} 不存在")
            return
            
        ip = args[2] if len(args) > 2 else None
        try:
            target_subnet.create_vm(args[1], ip)
        except Exception as e:
            print(f"❌ 创建失败: {e}")

    def do_ping(self, arg):
        """Ping 测试: ping <vm_name> <target_ip>"""
        args = shlex.split(arg)
        if len(args) != 2:
            print("❌ 用法: ping <vm_name> <target_ip>")
            return
            
        # 查找虚拟机
        target_vm = None
        for vpc in ALL_VPCS.values():
            for subnet in vpc.subnets:
                for vm in subnet.vms:
                    if vm.name == args[0]:
                        target_vm = vm
                        break
        if not target_vm:
            print(f"❌ 虚拟机 {args[0]} 不存在")
            return
            
        target_vm.ping(args[1])
        
    def do_show(self, arg):
        """查看 VPC 拓扑: show"""
        print("\n=== 当前 VPC 拓扑 ===")
        if not ALL_VPCS:
            print("空")
        for vpc in ALL_VPCS.values():
            print(f"🌐 VPC: {vpc.name} ({vpc.cidr})")
            for subnet in vpc.subnets:
                print(f"  ├─ 📂 子网: {subnet.cidr} (网关: {subnet.gateway})")
                for vm in subnet.vms:
                    print(f"  │  └─ 💻 VM: {vm.name} (IP: {vm.ip})")
        print("===================\n")

    def do_exit(self, arg):
        """退出并清理资源: exit"""
        print("退出模拟器...")
        return True

​核心体验升级：
使用 cmd.Cmd 类可以非常方便地打造专属提示符 (VPC-Sim)，利用 shlex 解析输入的参数，这让我们的代码直接从“脚本”进化为了“工具”。

读者只需在启动后批量粘贴以下命令，即可一键构建复杂拓扑：

create_vpc vpc-1 10.1.0.0/16
create_vpc vpc-2 10.2.0.0/16
create_subnet vpc-1 subnet-1 10.1.1.0/24
create_subnet vpc-1 subnet-2 10.1.2.0/24
create_subnet vpc-2 subnet-3 10.2.1.0/24
create_vm 10.1.1.0/24 vm1 10.1.1.2
create_vm 10.1.1.0/24 vm2 10.1.1.3
create_vm 10.1.2.0/24 vm3 10.1.2.2
create_vm 10.2.1.0/24 vm4 10.2.1.2
show
ping vm1 vm2
ping vm1 vm3
ping vm1 vm4

四、完整可运行代码

import subprocess
import ipaddress
import cmd
import shlex
from typing import List, Dict, Optional
import atexit

# ------------------------------
# 全局资源管理与自动清理
# ------------------------------
ALL_VPCS: Dict[str, 'VPC'] = {}
ALL_NETNS: List[str] = []
ALL_BRIDGES: List[str] = []

def cleanup_all():
    """程序退出时自动清理所有资源，避免残留"""
    print("\n" + "="*50)
    print("🔄 自动清理所有创建的资源...")
    
    # 清理所有网络命名空间
    for netns in ALL_NETNS:
        subprocess.run(f"ip netns del {netns} 2>/dev/null", shell=True, capture_output=True)
        subprocess.run(f"ip link del veth-{netns} 2>/dev/null", shell=True, capture_output=True)
    
    # 清理所有网桥及其 iptables 规则
    for bridge in ALL_BRIDGES:
        subprocess.run(f"iptables -D FORWARD -i {bridge} -o {bridge} -j ACCEPT 2>/dev/null", shell=True, capture_output=True)
        subprocess.run(f"iptables -D FORWARD -i {bridge} -o br-+ -j DROP 2>/dev/null", shell=True, capture_output=True)
        subprocess.run(f"iptables -D FORWARD -o {bridge} -i br-+ -j DROP 2>/dev/null", shell=True, capture_output=True)
        subprocess.run(f"ip link del {bridge} 2>/dev/null", shell=True, capture_output=True)
    
    print("✅ 所有资源清理完成")

# 注册退出钩子，确保程序异常退出也能清理资源
atexit.register(cleanup_all)

# ------------------------------
# 核心VPC类
# ------------------------------
class VPC:
    """
    模拟虚拟私有云(VPC)
    对应云厂商控制台：创建VPC
    """
    def __init__(self, name: str, cidr: str):
        self.name = name
        self.cidr = ipaddress.IPv4Network(cidr)
        self.subnets: List[Subnet] = []
        self.route_table: Dict[str, str] = {}
        self.bridge_name = f"br-{self.name}"
        
        # 创建VPC对应的虚拟网桥
        self._create_bridge()
        # 将VPC加入全局资源追踪
        ALL_VPCS[name] = self
        ALL_BRIDGES.append(self.bridge_name)
        
        print(f"✅ 创建VPC [{self.name}]，网段：{self.cidr}")
        
    def _create_bridge(self):
        """创建VPC内部的虚拟网桥（对应分布式虚拟交换机DVS）"""
        # 先删除可能存在的旧网桥，避免重复创建报错
        subprocess.run(f"ip link del {self.bridge_name} 2>/dev/null", shell=True, capture_output=True)
        # 创建Linux bridge虚拟网桥
        subprocess.run(f"ip link add {self.bridge_name} type bridge", shell=True, check=True)
        # 启用网桥
        subprocess.run(f"ip link set {self.bridge_name} up", shell=True, check=True)
        
        # 如果安装了 Docker，FORWARD 链默认会被设置为 DROP，导致跨子网不通
        # 添加 iptables 规则：允许同 VPC 内部跨子网转发，同时隔离不同 VPC！
        # 注意：使用 -I 插入时，后执行的命令会在链的最上方！所以先 DROP，后 ACCEPT
        # 1. 隔离不同 VPC（丢弃跨网桥流量）
        subprocess.run(f"iptables -I FORWARD -i {self.bridge_name} -o br-+ -j DROP", shell=True, check=True)
        subprocess.run(f"iptables -I FORWARD -o {self.bridge_name} -i br-+ -j DROP", shell=True, check=True)
        # 2. 允许同 VPC 内部互通（这条会在最顶部优先匹配）
        subprocess.run(f"iptables -I FORWARD -i {self.bridge_name} -o {self.bridge_name} -j ACCEPT", shell=True, check=True)
        
    def add_subnet(self, name: str, subnet_cidr: str) -> 'Subnet':
        """
        在VPC内创建子网
        对应云厂商控制台：创建子网
        """
        if not ipaddress.IPv4Network(subnet_cidr).subnet_of(self.cidr):
            raise ValueError(f"子网 {subnet_cidr} 不属于VPC网段 {self.cidr}")
            
        subnet = Subnet(name, subnet_cidr, self)
        self.subnets.append(subnet)
        # 添加默认路由：同VPC内流量直接通过网桥转发
        self.route_table[subnet_cidr] = self.bridge_name
        return subnet
        
    def add_route(self, dest_cidr: str, next_hop: str):
        """添加自定义路由规则"""
        self.route_table[dest_cidr] = next_hop
        print(f"✅ 添加路由：{dest_cidr} -> {next_hop}")

# ------------------------------
# 子网类
# ------------------------------
class Subnet:
    """
    模拟VPC子网
    对应云厂商控制台：子网管理
    """
    def __init__(self, name: str, cidr: str, vpc: VPC):
        self.name = name
        self.cidr = ipaddress.IPv4Network(cidr)
        self.vpc = vpc
        self.vms: List[VM] = []
        self.gateway = str(self.cidr[1])  # 子网网关使用网段第一个可用IP
        
        self._configure_gateway()
        print(f"✅ 创建子网 {self.name} [{self.cidr}]，网关：{self.gateway}")
        
    def _configure_gateway(self):
        """配置子网网关（网桥IP）"""
        # 先删除可能存在的旧IP
        subprocess.run(
            f"ip addr del {self.gateway}/{self.cidr.prefixlen} dev {self.vpc.bridge_name} 2>/dev/null",
            shell=True, capture_output=True
        )
        subprocess.run(
            f"ip addr add {self.gateway}/{self.cidr.prefixlen} dev {self.vpc.bridge_name}",
            shell=True, check=True
        )
        
    def create_vm(self, name: str, ip: Optional[str] = None) -> 'VM':
        """
        在子网内创建虚拟机
        对应云厂商控制台：创建云主机并加入子网
        """
        if not ip:
            # 自动分配IP：从第2个可用IP开始（第1个是网关）
            ip = str(self.cidr[len(self.vms)+2])
        else:
            if not ipaddress.IPv4Address(ip) in self.cidr:
                raise ValueError(f"IP {ip} 不属于子网 {self.cidr}")
                
        vm = VM(name, ip, self)
        self.vms.append(vm)
        return vm

# ------------------------------
# 虚拟机类
# ------------------------------
class VM:
    """
    模拟云主机实例
    核心隔离机制：Linux网络命名空间
    """
    def __init__(self, name: str, ip: str, subnet: Subnet):
        self.name = name
        self.ip = ipaddress.IPv4Address(ip)
        self.subnet = subnet
        self.vpc = subnet.vpc
        self.veth_name = f"veth-{self.name}"
        
        self._create_netns()
        self._connect_to_vpc()
        ALL_NETNS.append(self.name)
        
        print(f"✅ 创建虚拟机 [{self.name}]，IP：{self.ip}")
        
    def _create_netns(self):
        """创建独立的网络命名空间（实现隔离的核心）"""
        # 先删除可能存在的旧命名空间
        subprocess.run(f"ip netns del {self.name} 2>/dev/null", shell=True, capture_output=True)
        subprocess.run(f"ip netns add {self.name}", shell=True, check=True)
        # 启用回环接口
        subprocess.run(f"ip netns exec {self.name} ip link set lo up", shell=True, check=True)
        
    def _connect_to_vpc(self):
        """用veth pair将虚拟机连接到VPC网桥"""
        # 删除可能存在的旧veth接口
        subprocess.run(f"ip link del {self.veth_name} 2>/dev/null", shell=True, capture_output=True)
        # 创建一对虚拟网卡
        subprocess.run(
            f"ip link add {self.veth_name} type veth peer name eth0 netns {self.name}",
            shell=True, check=True
        )
        # 将一端连接到VPC网桥
        subprocess.run(
            f"ip link set {self.veth_name} master {self.vpc.bridge_name}",
            shell=True, check=True
        )
        subprocess.run(f"ip link set {self.veth_name} up", shell=True, check=True)
        
        # 配置虚拟机内的网卡和默认路由
        subprocess.run(
            f"ip netns exec {self.name} ip addr add {self.ip}/{self.subnet.cidr.prefixlen} dev eth0",
            shell=True, check=True
        )
        subprocess.run(f"ip netns exec {self.name} ip link set eth0 up", shell=True, check=True)
        subprocess.run(
            f"ip netns exec {self.name} ip route add default via {self.subnet.gateway}",
            shell=True, check=True
        )
        
    def ping(self, target_ip: str, count: int = 2, timeout: int = 1) -> bool:
        """测试到目标IP的连通性"""
        print(f"\n🔍 测试连通性：{self.name}({self.ip}) → {target_ip}")
        result = subprocess.run(
            f"ip netns exec {self.name} ping -c {count} -W {timeout} {target_ip}",
            shell=True, capture_output=True, text=True
        )
        
        if result.returncode == 0:
            print("✅ 连通成功！")
            return True
        else:
            print("❌ 连通失败！")
            return False
            
    def exec(self, command: str) -> str:
        """在虚拟机内执行任意命令"""
        result = subprocess.run(
            f"ip netns exec {self.name} {command}",
            shell=True, capture_output=True, text=True
        )
        return result.stdout

# ------------------------------
# 交互式 CLI Shell
# ------------------------------
class VPCShell(cmd.Cmd):
    intro = '===========================================================\n' \
            '欢迎使用 VPC 交互式模拟器（基础篇）\n' \
            '【可用指令】:\n' \
            '  1. create_vpc <name> <cidr>                   - 创建 VPC (例: create_vpc vpc1 10.0.0.0/16)\n' \
            '  2. create_subnet <vpc_name> <name> <cidr>     - 创建子网 (例: create_subnet vpc1 sub1 10.0.1.0/24)\n' \
            '  3. create_vm <subnet_cidr> <vm_name> [ip]     - 创建虚拟机 (例: create_vm 10.0.1.0/24 vm1)\n' \
            '  4. ping <src_vm_name> <dst_vm_name_or_ip>     - 连通性测试 (例: ping vm1 vm2)\n' \
            '  5. show                                       - 查看网络拓扑\n' \
            '  6. exit                                       - 退出并清理资源\n' \
            '==========================================================='
    prompt = '(VPC-Sim) '

    def emptyline(self):
        """
        空行时只显示提示符，不做任何操作
        """
        pass

    def do_create_vpc(self, arg):
        """创建 VPC: create_vpc <name> <cidr>"""
        args = shlex.split(arg)
        if len(args) != 2:
            print("❌ 用法: create_vpc <name> <cidr>")
            return
        try:
            VPC(args[0], args[1])
        except Exception as e:
            print(f"❌ 创建失败: {e}")

    def do_create_subnet(self, arg):
        """创建子网: create_subnet <vpc_name> <subnet_name> <cidr>"""
        args = shlex.split(arg)
        if len(args) != 3:
            print("❌ 用法: create_subnet <vpc_name> <subnet_name> <cidr>")
            return
        vpc = ALL_VPCS.get(args[0])
        if not vpc:
            print(f"❌ VPC {args[0]} 不存在")
            return
        try:
            vpc.add_subnet(args[1], args[2])
        except Exception as e:
            print(f"❌ 创建失败: {e}")

    def do_create_vm(self, arg):
        """创建虚拟机: create_vm <subnet_cidr> <vm_name> [ip]"""
        args = shlex.split(arg)
        if len(args) < 2:
            print("❌ 用法: create_vm <subnet_cidr> <vm_name> [ip]")
            return
        
        target_subnet = None
        for vpc in ALL_VPCS.values():
            for subnet in vpc.subnets:
                if str(subnet.cidr) == args[0]:
                    target_subnet = subnet
                    break
        if not target_subnet:
            print(f"❌ 子网 {args[0]} 不存在")
            return
            
        ip = args[2] if len(args) > 2 else None
        try:
            target_subnet.create_vm(args[1], ip)
        except Exception as e:
            print(f"❌ 创建失败: {e}")

    def do_ping(self, arg):
        """Ping 测试: ping <src_vm_name> <dst_vm_name_or_ip>"""
        args = shlex.split(arg)
        if len(args) != 2:
            print("❌ 用法: ping <src_vm_name> <dst_vm_name_or_ip>")
            return
            
        target_vm = None
        target_ip = args[1]
        
        for vpc in ALL_VPCS.values():
            for subnet in vpc.subnets:
                for vm in subnet.vms:
                    if vm.name == args[0]:
                        target_vm = vm
                    if vm.name == args[1]:
                        target_ip = vm.ip
                        
        if not target_vm:
            print(f"❌ 源虚拟机 {args[0]} 不存在")
            return
            
        target_vm.ping(target_ip)
        
    def do_show(self, arg):
        """查看 VPC 拓扑: show"""
        print("\n=== 当前 VPC 拓扑 ===")
        if not ALL_VPCS:
            print("空")
        for vpc in ALL_VPCS.values():
            print(f"🌐 VPC: {vpc.name} ({vpc.cidr})")
            for subnet in vpc.subnets:
                print(f"  ├─ 📂 子网: {subnet.name} - {subnet.cidr} (网关: {subnet.gateway})")
                for vm in subnet.vms:
                    print(f"  │  └─ 💻 VM: {vm.name} (IP: {vm.ip})")
        print("===================\n")

    def do_exit(self, arg):
        """退出并清理资源: exit"""
        print("退出模拟器...")
        return True

if __name__ == "__main__":
    # 确保宿主机开启内核 IP 转发（VPC 路由器的关键）
    subprocess.run("sysctl -w net.ipv4.ip_forward=1", shell=True, capture_output=True)
    VPCShell().cmdloop()

五、运行方法

1. 将代码保存为vpc_part1_basic.py

2. 使用 sudo 权限执行：

sudo python3 vpc_part1_basic.py

3. 运行结果

root@ubuntu:~# python3 vpc_part1_basic.py 
===========================================================
欢迎使用 VPC 交互式模拟器（基础篇）
【可用指令】:
  1. create_vpc <name> <cidr>                   - 创建 VPC (例: create_vpc vpc1 10.0.0.0/16)
  2. create_subnet <vpc_name> <name> <cidr>     - 创建子网 (例: create_subnet vpc1 sub1 10.0.1.0/24)
  3. create_vm <subnet_cidr> <vm_name> [ip]     - 创建虚拟机 (例: create_vm 10.0.1.0/24 vm1)
  4. ping <src_vm_name> <dst_vm_name_or_ip>     - 连通性测试 (例: ping vm1 vm2)
  5. show                                       - 查看网络拓扑
  6. exit                                       - 退出并清理资源
===========================================================
(VPC-Sim) create_vpc vpc-1 10.1.0.0/16
✅ 创建VPC [vpc-1]，网段：10.1.0.0/16
(VPC-Sim) create_vpc vpc-2 10.2.0.0/16
✅ 创建VPC [vpc-2]，网段：10.2.0.0/16
(VPC-Sim) create_subnet vpc-1 subnet-1 10.1.1.0/24
✅ 创建子网 subnet-1 [10.1.1.0/24]，网关：10.1.1.1
(VPC-Sim) create_subnet vpc-1 subnet-2 10.1.2.0/24
✅ 创建子网 subnet-2 [10.1.2.0/24]，网关：10.1.2.1
(VPC-Sim) create_subnet vpc-2 subnet-3 10.2.1.0/24
✅ 创建子网 subnet-3 [10.2.1.0/24]，网关：10.2.1.1
(VPC-Sim) create_vm 10.1.1.0/24 vm1 10.1.1.2
✅ 创建虚拟机 [vm1]，IP：10.1.1.2
(VPC-Sim) create_vm 10.1.1.0/24 vm2 10.1.1.3
✅ 创建虚拟机 [vm2]，IP：10.1.1.3
(VPC-Sim) create_vm 10.1.2.0/24 vm3 10.1.2.2
✅ 创建虚拟机 [vm3]，IP：10.1.2.2
(VPC-Sim) create_vm 10.2.1.0/24 vm4 10.2.1.2
✅ 创建虚拟机 [vm4]，IP：10.2.1.2
(VPC-Sim) show

=== 当前 VPC 拓扑 ===
🌐 VPC: vpc-1 (10.1.0.0/16)
  ├─ 📂 子网: subnet-1 - 10.1.1.0/24 (网关: 10.1.1.1)
  │  └─ 💻 VM: vm1 (IP: 10.1.1.2)
  │  └─ 💻 VM: vm2 (IP: 10.1.1.3)
  ├─ 📂 子网: subnet-2 - 10.1.2.0/24 (网关: 10.1.2.1)
  │  └─ 💻 VM: vm3 (IP: 10.1.2.2)
🌐 VPC: vpc-2 (10.2.0.0/16)
  ├─ 📂 子网: subnet-3 - 10.2.1.0/24 (网关: 10.2.1.1)
  │  └─ 💻 VM: vm4 (IP: 10.2.1.2)
===================

(VPC-Sim) ping vm1 vm2

🔍 测试连通性：vm1(10.1.1.2) → 10.1.1.3
✅ 连通成功！
(VPC-Sim) ping vm1 vm3

🔍 测试连通性：vm1(10.1.1.2) → 10.1.2.2
✅ 连通成功！
(VPC-Sim) ping vm1 vm4

🔍 测试连通性：vm1(10.1.1.2) → 10.2.1.2
❌ 连通失败！

六、核心架构图

七、在线互动体验

看完了代码和架构图，现在用下面的交互式模拟器亲手验证 VPC 的核心特性吧！无需安装任何环境，直接在浏览器中体验：

🎮 动手试试：VPC 核心特性验证

点击下方按钮，直观感受 VPC 的隔离与互通

常见问题

1. 为什么需要 sudo 权限？

   操作 Linux 内核网络模块（创建命名空间、网桥等）需要 root 权限。

2. WSL2 环境下运行失败怎么办？

   确保 WSL2 版本为 0.58.0 以上，执行sudo sysctl -w net.ipv4.ip_forward=1开启 IP 转发。

3. 程序异常退出导致资源残留怎么办？

   执行代码中提示的手动清理命令，或重启系统。

下一篇预告

下一篇我们将给这个最小 VPC 添加​安全组和网络 ACL，实现云厂商标准的双重安全防护体系，让你的 VPC 网络更安全。

　　‍