Why
我们访问 K8S 的 ApiServer 服务,由于为了保证安全性,证书中签发的域名仅包括 kubernetes
和初始的有限 IP 列表,当 ApiServer 服务的 Master 节点需要被替换时,就无法使用新的节点 IP 访问了。解决的方案就是将 kubernetes
域名和新的 IP 临时绑定,骗过证书校验。
How?
废话不多说,直接看代码
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45
| import contextlib import importlib import threading from typing import Callable, Dict, Optional
from urllib3.util import connection
class CustomLocalDnsResolver(threading.local): """支持线程级自定义 Dns 记录 """
def __init__(self, dns_map: Optional[dict]=None): self.dns_map = dns_map or {}
def get_patch_create_connection_with_dns(dns_resolver) -> Callable: """simply get patched create_connection"""
_orig_create_connection = getattr(importlib.import_module('urllib3.util.connection'), 'create_connection')
def patched_create_connection(address, *args, **kwargs): """在 urllib3's create_connection 流程前解析 address""" domain, port = address host = dns_resolver.dns_map.get(domain, domain) return _orig_create_connection((host, port), *args, **kwargs)
return patched_create_connection
_local_dns = CustomLocalDnsResolver()
connection.create_connection = get_patch_create_connection_with_dns(_local_dns)
@contextlib.contextmanager def update_local_dns_once(dns_map: Dict): """一次性修改线程 dns 解析""" _local_dns.dns_map = dns_map yield _local_dns.dns_map = {}
with update_local_dns_once({"kubernetes": "192.168.1.1"}):
|
Author:
Blues Yu
Permalink:
https://emergencyexit.xyz/update-dns-in-python.html
License:
Copyright (c) 2025 CC-BY-NC-4.0 LICENSE