旧代码拾遗:如何在 Python 代码中修改 DNS 解析
date
Feb 18, 2022
slug
update-dns-in-python.html
status
Published
tags
python
tech
summary
旧代码拾遗
type
Post
Why
我们访问 K8S 的 ApiServer 服务,由于为了保证安全性,证书中签发的域名仅包括
kubernetes
和初始的有限 IP 列表,当 ApiServer 服务的 Master 节点需要被替换时,就无法使用新的节点 IP 访问了。解决的方案就是将 kubernetes
域名和新的 IP 临时绑定,骗过证书校验。How?
废话不多说,直接看代码
# -*- coding: utf-8 -*-
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):
# 线程保存各自 dns_map,但是访问入口均为 dns_map
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
# 当 _local_dns.dns_map 为空,对正常流程无影响
host = dns_resolver.dns_map.get(domain, domain)
return _orig_create_connection((host, port), *args, **kwargs)
return patched_create_connection
_local_dns = CustomLocalDnsResolver()
# patch 全局 create_connection
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"}):
# 可以在该 context 中请求外部系统