文章

背景 在K8s集群中，希望引入Falco进行统一的集群操作审计 这就需要先安装Falco，借助json插件与k8saudit插件创建一个Webhook客户端，专用于接收API Server的操作审计数据 当前已有falco:0.44.0镜像，以及一个Falco的custom_values.yaml： driver: enabled: false collectors: enabled: false controller: kind: deployment services: - name: k8saudit-webhook type: NodePort ports: - port: 9765 protocol: TCP falco: rules_files: - /etc/falco/k8s_audit_rules.yaml - /etc/falco/rules.d plugins: - name: k8saudit library_path: /usr/share/falco/plugins/libk8saudit.so init_config: "" open_params: "http://:9765/k8s-audit" - name: json library_path: /usr/share/falco/plugins/libjson.so init_config: "" load_plugins: - k8saudit - json json_output: true json_include_output_property: true json_include_tags_property: true 关于格式 values文件的格式可参考：https://raw.githubusercontent.com/falcosecurity/charts/refs/heads/master/charts/falco/values.yaml 在机器可以访问外网的情况下，执行helm install falco falcosecurity/falco -n kube-audit -f custom_values.yaml即可将Falco部署到集群 但我们的集群无法访问外网，首先拉取不到Chart，其次也拉取不到falcoctl以及两个插件，甚至连插件的index文件都无法获取 因此，需要考虑离线安装Falco 手动下载Chart 执行： wget https://github.com/falcosecurity/charts/releases/download/falco-9.0.0/falco-9.0.0.tgz 即可获取falco-9.0.0.tgzChart文件，如此一来，Chart的本地化解决 falcoctl离线安装 在Pod启动时，会尝试拉取falcoctl容器镜像，它会被自动配置为Falco Pod的一个Init Container，用于管理安全规则Rules、插件Plugins ...

依赖 SpringBoot：3.5.12 HttpClient5：5.6.1 发生场景 执行restTemplate.exchange(url, HttpMethod.GET, entity, NodeListResponse.class).getBody()时报错： java.lang.ClassNotFoundException: org.apache.hc.core5.io.IOFunction 修复 查看HttpCore5的版本： $ mvn dependency:tree [INFO] ... [INFO] \- org.apache.httpcomponents.client5:httpclient5:jar:5.6.1:compile [INFO] +- org.apache.httpcomponents.core5:httpcore5:jar:5.3.6:compile [INFO] +- org.apache.httpcomponents.core5:httpcore5-h2:jar:5.3.6:compile [INFO] \- org.slf4j:slf4j-api:jar:2.0.17:compile [INFO] ------------------------------------------------------------------------ [INFO] BUILD SUCCESS [INFO] ------------------------------------------------------------------------ [INFO] Total time: 10.139 s [INFO] Finished at: 2026-04-24T19:20:35+08:00 [INFO] ------------------------------------------------------------------------ 可以看到HttpCore5是由HttpClient5引入的，版本为5.3.6 查询资料后得知Core的5.3.6版本最好配套Client5的5.5.2版本 最终确定的版本： HttpCore5：5.3.6 HttpClient5：5.5.2

准备 我自己有一台接入米家的夜灯，型号是米家床头灯二代，官方已经给出了产品规格与各项控制参数 感谢Do1e开源的的mijiaAPI，这个仓库向调用者提供了傻瓜式的米家电器控制调用接口，只需要调用相应方法、传入对应于设备控制意图的参数，即可通过云端控制米家设备 AstrBot部署于阿里云服务器，已经接入微信的ClawBot。为了插件开发方便，也在本地部署了AstrBot 环境配置 首先在AstrBot的根目录下安装依赖，官方提供了uv依赖管理： cd AstrBot uv sync uv pip install mijiaAPI AstrBot为插件开发者提供了模板仓库，只需使用此模板，然后克隆到本地的AstrBot插件文件夹即可开始开发 cd AstrBot/data/plugins git clone <template-registry> 为了保证插件的正常运行，之后在插件开发调试中，Python环境都应使用外层目录的astrbot，而不应在插件目录新建环境 登录米家 mijiaAPI提供了便捷的登陆方式： import mijiaAPI from mijiaAPI api = mijiaAPI("auth.json") api.login() 执行login()时，程序会在终端打印登录二维码，使用登陆了米家账号的设备扫描后即可完成登录，此后程序不再阻塞，继续执行后面的脚本 不过，如果要将登录逻辑引入AstrBot机器人聊天环境存在一些问题： 登陆时的阻塞会导致机器人无法回复 登陆二维码在终端打印，需要某种方法将登陆手段以机器人回复的形式向用户呈现 在满足以上条件的基础上，还要维持原mijiaAPI中的轮询用户登录逻辑，并在用户扫码登录成功后使机器人回复“登陆成功”提示用户 关于阻塞问题的解决方案，我参考了已有的在AstrBot中调用mijiaAPI的仓库，由674537331开发，核心思路是创建一个独立的进程沙盒，在沙盒中运行login()从而避免主进程被阻塞 在超时时间内每隔0.1秒不断读取进程的输出，如果读取到了二维码URL，则立即向用户回复该URL，回复后不结束登录流程，而是继续循环读取输出，直到超时或读取到登陆成功或失败 在子进程中执行登录的脚本_login_worker.py： #!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 登录工作脚本，在子进程中执行登录流程 """ import sys import os import time from pathlib import Path import logging # 配置日志 logging.basicConfig(level=logging.INFO, encoding='utf-8') logging.getLogger("mijiaAPI").setLevel(logging.INFO) # 强制设置 stdout 编码为 UTF-8 if sys.stdout.encoding != 'utf-8': sys.stdout.reconfigure(encoding='utf-8') if sys.stderr.encoding != 'utf-8': sys.stderr.reconfigure(encoding='utf-8') from mijiaAPI import mijiaAPI def main(): if len(sys.argv) != 2: print("Usage: python _login_worker.py <auth_path>", flush=True) sys.exit(1) auth_path = Path(sys.argv[1]) print("[WORKER] 开始初始化认证环境。", flush=True) try: # 确保目录存在 auth_path.parent.mkdir(parents=True, exist_ok=True) api = mijiaAPI(auth_path) print("[WORKER] API 实例已创建，正在请求小米服务器...", flush=True) # 执行登录 result = api.login() print("[WORKER] 登录成功", flush=True) print("[WORKER_SUCCESS] 授权完毕。", flush=True) sys.exit(0) except Exception as e: print(f"[WORKER_ERROR] 登录流程失败: {type(e).__name__}: {e}", flush=True) sys.exit(1) if __name__ == "__main__": main() 主文件： ...

现状 我在集群中使用Keycloak + kubelgoin + Webhook的认证模式，具体流程如下： 在API-Server中将认证模式设置为Webhook模式，由Webhook手动校验OIDC 执行kubectl时，exec命令启动kubelogin，以device-code形式唤起Keycloak进行认证 认证完成后Keycloak给出token，由kubectl将id_token添加到GET请求的Authorization字段发送至API-Server API-Server向Webhook POST一个TokenReview，Webhook解析token、验证后根据规则组装成UserInfo发往后续准入环节等 现在项目中引入了Kite作为集群管理面板 Kite的底层是通过K8s-Client向API-Server发送HTTP请求执行相应操作，本质上与kubectl相似，且都需要读取预先配置的kubeconfig文件获取配置上下文，如集群地址、当前集群用户、用户证书等 不足 Kite本身不负责集群身份认证，而是将安全性转移至进入Kite面板的认证。因此Kite虽然原生支持OIDC，但只能支持进入Kite管理界面时的OIDC认证，而非集群内身份认证 也正因这个原因，Kite无法处理kubeconfig中配置了exec参数动态获取认证凭证的用户，在界面中将该用户上下文显示为不可用状态 为了解决这个问题，Kite创建了一个专用的Service Account，并使用其token通过集群认证（将token放入kubeconfig然后被Kite正常读取使用） 需求 我需要二次开发Kite，使Kite支持进行集群内OIDC device-code模式认证 用户只需要指定Issuer、client_secret等必要信息，即可提供Keycloak认证界面供用户进行认证 认证通过后，需要能够接收Keycloak响应的各种token 记录该token，禁用原kubeconfig配置 之后调用K8s-Client发送API请求时都在Authorization字段附加token 如果token过期，则在前端提示用户需要重新进行OIDC认证 如果请求被准入环节拦截，需要提示用户根据准入规定重新认证 / 增量认证 / 拒绝请求 准备 使用Postman测试不带Authorization的API请求响应，返回403： { "kind": "Status", "apiVersion": "v1", "metadata": {}, "status": "Failure", "message": "nodes is forbidden: User \"system:anonymous\" cannot list resource \"nodes\" in API group \"\" at the cluster scope", "reason": "Forbidden", "details": { "kind": "nodes" }, "code": 403 } 使用Postman附带Keycloak客户端ID、客户端Secret向$base_url发送认证请求： { "device_code": "vfpIiDEl8-M7euodTglJTkbUnKr99l7GGDNQK7If0EM", "user_code": "PSWY-TSRS", "verification_uri": "https://192.168.5.96:31443/realms/k8s-auth/device", "verification_uri_complete": "https://192.168.5.96:31443/realms/k8s-auth/device?user_code=PSWY-TSRS", "expires_in": 600, "interval": 5 } 认证完成后尝试拉取Token： ...

序列化手段有很多，之前的代码中都是硬编码 Serializer s = new JdkSerializer()，其中 JdkSerializer 是自定义的基于 JDK 的序列化器 现在可以给用户更多选择，如 JSON、Hessian、Kryo 额外三个内置序列化器，以及基于 SPI 的自定义序列化器 内置序列化器 首先分别实现 JSON、Hessian、Kryo 的序列化类，这一步与项目本身无关，这里不放代码了 接着，为了更有效地区分序列化器的名称，可以创建一个枚举类或者接口类存储内置序列化器的名称： public interface SerializerKeys { String JDK = "jdk"; String JSON = "json"; String KRYO = "kryo"; String HESSIAN = "hessian"; } 序列化器明显可以复用，所以创建一个单例的序列化器工厂： public class SerializerFactory { // 单例，存储名称到序列化器实例的映射 private static final Map<String, Serializer> KEY_SERIALIZER_MAP = new HashMap<String, Serializer>() {{ put(SerializerKeys.JDK, new JdkSerializer()); put(SerializerKeys.JSON, new JsonSerializer()); put(SerializerKeys.KRYO, new KryoSerializer()); put(SerializerKeys.HESSIAN, new HessianSerializer()); }}; // 默认序列化器 private static final Serializer DEFAULT_SERIALIZER = KEY_SERIALIZER_MAP.get("jdk"); // 获取实例 public static Serializer getInstance(String key) { return KEY_SERIALIZER_MAP.getOrDefault(key, DEFAULT_SERIALIZER); } } 最后，不要忘记修改默认 Config 配置： ...