☰
Kubernetes Python 客户端 V1IPBlock 模型全解:NetworkPolicy 的 CIDR 白名单/黑名单字段、序列化与实战用法
2026/10/10 2:01:16 网站建设 项目流程
  • 后端
  • 云原生
  • 容器编排

【免费下载链接】python

Official Python client library for kubernetes

项目地址:https://gitcode.com/gh_mirrors/python1/python
点击查看免费下载

V1IPBlock 是官方 Kubernetes Python 客户端(含 asyncio 变体)中用于表达NetworkPolicy 规则里 IP 地址段(CIDR)放行与排除的核心数据模型:cidr声明允许访问的网段,except声明在cidr范围内仍需排除的网段。本文以 doc/source/kubernetes.aio.client.models.v1_ip_block.rst 对应的异步(aio)模型为主线,逐字段拆解其属性约束、Pydantic 校验行为、JSON/字典双向序列化接口,并结合V1NetworkPolicyPeer给出构造 NetworkPolicy 的完整可运行示例。读完本文,你将掌握V1IPBlock的全部公开 API、输入别名规则与序列化细节,能够准确地在 asyncio 程序中创建和解析基于 IP 网段的网络策略。

一、模型定位:NetworkPolicy 中的 IP 网段描述

在 Kubernetes 中,NetworkPolicy 通过podSelector圈定策略作用的 Pod 集合,再通过 ingress/egress 规则控制进出流量。V1IPBlock正是这些规则中描述"放行哪一个 IP 网段"的类型。其官方文档语义(见 V1IPBlock.md)为:

IPBlock describes a particular CIDR (Ex. "192.168.1.0/24","2001:db8::/64") that is allowed to the pods matched by a NetworkPolicySpec's podSelector. The except entry describes CIDRs that should not be included within this rule.

即:cidr是允许访问(或允许被访问)的地址块,except列出该地址块内需要排除的子网。两个字段均为 CIDR 字符串,同时支持 IPv4(如192.168.1.0/24)与 IPv6(如2001:db8::/64)。

在 OpenAPI 规格文件 scripts/swagger.json 的v1.IPBlock定义中(第 15018–15037 行),可以确认该模型的结构:

"v1.IPBlock": { "description": "IPBlock describes a particular CIDR ... ", "properties": { "cidr": { "description": "...", "type": "string" }, "except": { "description": "... Except values will be rejected if they are outside the cidr range", "items": { "type": "string" }, "type": "array", "x-kubernetes-list-type": "atomic" } }, "required": ["cidr"], "type": "object" }

要点:

  • cidr是必填字段;
  • except是字符串数组(切片),可选,且标注了x-kubernetes-list-type: atomic,表示该列表在服务端按整体原子替换处理;
  • 若except中的 CIDR 超出了cidr的地址范围,请求会被拒绝。

二、字段详解与属性约束

V1IPBlock只包含两个属性,完整参数表如下(字段名、类型与说明均以 v1_ip_block.py 为准):

属性名(Python)线协议字段(JSON)类型必填说明
cidrcidrStrictStr(str)是表示 IPBlock 的 CIDR 字符串,合法示例"192.168.1.0/24"、"2001:db8::/64"
var_exceptexceptOptional[List[StrictStr]]否不应包含在 IPBlock 内的 CIDR 切片,合法示例同上;超出cidr范围的取值会被拒绝,默认None

源码中的字段声明如下(v1_ip_block.py):

cidr: StrictStr = Field( description='cidr is a string representing the IPBlock ' 'Valid examples are "192.168.1.0/24" or "2001:db8::/64"' ) var_except: Optional[List[StrictStr]] = Field( default=None, validation_alias=AliasChoices("except", "_except"), serialization_alias="except", description="except is a slice of CIDRs that should not be included " "within an IPBlock ... Except values will be rejected if " "they are outside the cidr range", alias="_except", )

三个值得注意的细节:

  1. 字段名避开 Python 关键字:except是 Python 保留字,无法作为属性名,因此内部字段被命名为var_except。但通过validation_alias=AliasChoices("except", "_except")与serialization_alias="except",对外 JSON 始终使用except键。
  2. cidr使用StrictStr:Pydantic 严格字符串校验,传入非字符串类型不会被宽松转换,从而保证序列化给 API Server 的 CIDR 始终是字符串。
  3. 模型级校验配置(v1_ip_block.py):
    model_config = ConfigDict( validate_by_name=True, # 允许按 Python 字段名传参 validate_by_alias=True, # 也允许按别名(except/_except)传参 validate_assignment=True, # 赋值时同样触发校验 extra="forbid", # 禁止未知字段 protected_namespaces=(), )

    extra="forbid"意味着传入任何既非cidr、也非except/var_except的键都会直接报错,有助于在编写策略代码时尽早发现拼写错误。

三、输入别名预处理:except与_except的兼容逻辑

由于历史原因,该模型同时接受except与_except两个线协议键。__preprocess_input_names类方法(v1_ip_block.py)在模型校验前完成归一化:

if "except" in obj and "_except" in obj: raise ValueError("%s received both %r and %r" % (cls.__name__, "except", "_except")) if "except" not in obj and "_except" in obj: obj["except"] = obj["_except"] obj.pop("_except", None)

规则总结:

  • 同时给出except与_except会抛出ValueError;
  • 只给_except时自动映射为except;
  • 最终以except作为唯一输入键进入模型校验。

这意味着从from_dict反序列化时,两种键写法都能被接受,但应避免二者同时出现。该处理同时服务于从 JSON 反序列化与从 Python dict 构造两条路径。

四、序列化与反序列化 API 全览

V1IPBlock继承pydantic.BaseModel,并额外实现了 OpenAPI Generator 生成模型标准的一组方法(v1_ip_block.py):

方法作用
to_str()返回pprint.pformat(self.to_dict())的可读字符串表示,__repr__亦复用之
to_json()返回按别名(cidr、except)序列化的 JSON 字符串
from_json(json_str)从 JSON 字符串创建V1IPBlock实例
to_dict(serialize=False)返回字典:serialize=False时键为cidr/_except(公开名),serialize=True时键为cidr/except(线协议名)
from_dict(obj)从字典创建实例,内部先走__preprocess_input_names归一化,再调用model_validate
__eq__/__ne__基于to_dict()的结果比较两个模型是否相等

其中to_dict的两种输出模式对应生成器的"legacy / modern"兼容设计:默认返回 Python 公开名(_except),需要交给 API 客户端时可用serialize=True得到线协议名(except)。from_json是from_dict(json.loads(json_str))的便捷封装。

一个完整的构造与往返示例(可直接复制运行):

from kubernetes.aio.client.models.v1_ip_block import V1IPBlock # 方式一:按字段名直接构造(except 是关键字,需用 var_except) block = V1IPBlock( cidr="192.168.1.0/24", var_except=["192.168.1.10/32", "192.168.1.11/32"], ) # 方式二:按线协议键通过 dict 构造 block2 = V1IPBlock.from_dict({ "cidr": "2001:db8::/64", "except": ["2001:db8::1/128"], # 或 "_except": [...] }) print(block) # to_str() -> pprint 输出 print(block.to_json()) # {"cidr": "192.168.1.0/24", "except": [...]} d = block.to_dict(serialize=True) # {'cidr': ..., 'except': [...]} restored = V1IPBlock.from_dict(d) assert block == restored # 同时传入 except 与 _except 会抛 ValueError # V1IPBlock.from_dict({"cidr": "10.0.0.0/8", "except": [], "_except": []})

五、在 NetworkPolicy 中的实际使用

V1IPBlock本身并不单独出现在策略 YAML 顶层,而是作为peer(对端)的一个可选项被引用。查看 v1_network_policy_peer.py 可见:

class V1NetworkPolicyPeer(BaseModel): ip_block: Optional[V1IPBlock] = Field( default=None, validation_alias=AliasChoices("ipBlock", "ip_block"), serialization_alias="ipBlock", ) namespace_selector: Optional[V1LabelSelector] = ... pod_selector: Optional[V1LabelSelector] = ...

OpenAPI 对v1.NetworkPolicyPeer的约束(scripts/swagger.json)明确指出:

If this field [ipBlock] is set then neither of the other fields can be.

也就是说,V1NetworkPolicyPeer中ipBlock、namespaceSelector、podSelector三选一,ipBlock一旦设置,其余两个选择器字段必须为空。V1NetworkPolicyPeer随后被V1NetworkPolicyIngressRule(from列表)与V1NetworkPolicyEgressRule(to列表)引用,形成完整的 ingress/egress 规则。

下面是一个在 asyncio 客户端中构造"允许从192.168.1.0/24访问,但排除其中两个管理地址"的 NetworkPolicy 示例:

import asyncio from kubernetes import config from kubernetes.aio import client as aio_client from kubernetes.aio.client.models import ( V1IPBlock, V1NetworkPolicy, V1NetworkPolicyPeer, V1NetworkPolicyIngressRule, V1NetworkPolicySpec, ) async def create_ip_policy(): await config.load_kube_config() # 或用 load_incluster_config() async with aio_client.ApiClient() as api_client: api = aio_client.NetworkingV1Api(api_client) peer = V1NetworkPolicyPeer( ip_block=V1IPBlock( cidr="192.168.1.0/24", var_except=["192.168.1.10/32", "192.168.1.11/32"], ) ) policy = V1NetworkPolicy( api_version="networking.k8s.io/v1", kind="NetworkPolicy", metadata={"name": "allow-app-cidr", "namespace": "default"}, spec=V1NetworkPolicySpec( pod_selector={"matchLabels": {"app": "my-app"}}, ingress=[V1NetworkPolicyIngressRule(from_=[peer])], policy_types=["Ingress"], ), ) await api.create_namespaced_network_policy( namespace="default", body=policy ) # 读取回查,确认服务端接受的 IPBlock got = await api.read_namespaced_network_policy( name="allow-app-cidr", namespace="default" ) print(got.spec.ingress[0].from_[0].ip_block.to_dict(serialize=True)) asyncio.run(create_ip_policy())

等价于以下 Kubernetes YAML 语义:

apiVersion: networking.k8s.io/v1 kind: NetworkPolicy metadata: name: allow-app-cidr namespace: default spec: podSelector: matchLabels: app: my-app policyTypes: ["Ingress"] ingress: - from: - ipBlock: cidr: 192.168.1.0/24 except: - 192.168.1.10/32 - 192.168.1.11/32

六、同步与异步客户端的模型一致性

仓库同时维护两个版本的V1IPBlock:

  • 异步(asyncio):kubernetes/aio/client/models/v1_ip_block.py
  • 同步:kubernetes/client/models/v1_ip_block.py

两者内容完全一致(均基于 OpenAPI release-1.37 规格生成),字段、别名、校验逻辑与方法签名完全相同。差异仅体现在上层 API 客户端的使用方式:同步客户端直接调用kubernetes.client.NetworkingV1Api,而 asyncio 客户端使用kubernetes.aio.client.NetworkingV1Api并配合async with与await。因此本文所有关于V1IPBlock的字段与序列化结论,对两种模式同样适用;选择哪个包,取决于你的应用是否基于事件循环。

七、常见陷阱与最佳实践

  1. except必须落在cidr范围内:这是 Kubernetes API Server 的硬性校验。例如cidr: 10.0.0.0/8搭配except: ["172.16.0.0/16"]会被拒绝,应在本地编码时就校验(可借助ipaddress模块):
    import ipaddress cidr = ipaddress.ip_network("192.168.1.0/24") for exc in ["192.168.1.10/32"]: assert ipaddress.ip_network(exc).subnet_of(cidr), f"{exc} 超出 {cidr}"
  2. 同时支持 IPv4 与 IPv6:cidr与except类型均为普通字符串,客户端不做 CIDR 语法校验(校验由 API Server 完成),但示例写法应遵循标准 CIDR 格式,如2001:db8::/64。
  3. 不要同时传except与_except:__preprocess_input_names会直接抛ValueError。推荐统一使用except键(或 Python 侧构造时用var_except=)。
  4. extra="forbid"会拒绝未知字段:构造时传入cidr、except以外的键(如笔误的cidrs)会立即报错,属于客户端提供的第一道防线。
  5. 优先使用from_dict处理来自 API 的响应:读取 NetworkPolicy 后拿到的响应 dict 可通过V1IPBlock.from_dict(...)/from_json(...)还原为类型化对象,便于后续比较(==基于to_dict()逐字段比较)。

八、相关资源

  • 文档入口:doc/source/kubernetes.aio.client.models.v1_ip_block.rst
  • 异步模型实现:kubernetes/aio/client/models/v1_ip_block.py
  • 同步模型实现:kubernetes/client/models/v1_ip_block.py
  • 模型速查文档:kubernetes/aio/docs/V1IPBlock.md
  • 引用方V1NetworkPolicyPeer:kubernetes/client/models/v1_network_policy_peer.py
  • OpenAPI 原始定义(v1.IPBlock):scripts/swagger.json
  • 后端
  • 云原生
  • 容器编排

【免费下载链接】python

Official Python client library for kubernetes

项目地址:https://gitcode.com/gh_mirrors/python1/python
点击查看免费下载

相关推荐

上一篇:革命性API安全审计:APIKit智能扫描引擎深度解析
下一篇:跨平台资源下载终极指南:res-downloader让你的数字收藏变得简单高效

创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询