BACKGROUND admission controllers的特点:
- 可定制性:准入功能可针对不同的场景进行调整。
- 可预防性:审计则是为了检测问题,而准入控制器可以预防问题发生
- 可扩展性:在kubernetes自有的验证机制外,增加了另外的防线,弥补了RBAC仅能对资源提供安全保证。
文章图片
图:Kubernetes API 请求的请求处理步骤图 Source:https://kubernetes.io/blog/2019/03/21/a-guide-to-kubernetes-admission-controllers/ 这里找到一个大佬博客画的图,通过两张图可以很清晰的了解到admission webhook流程,与官方给出的不一样的地方在于,这里清楚地定位了kubernetes admission webhook 处于准入控制中,RBAC之后,push 之前。
文章图片
图:Kubernetes API 请求的请求处理步骤图(详细) Source:https://www.armosec.io/blog/kubernetes-admission-controller/ 两种控制器有什么区别?
根据官方提供的说法是
Mutating controllers may modify related objects to the requests they admit; validating controllers may not从结构图中也可以看出,
validating
是在持久化之前,而 Mutating
是在结构验证前,根据这些特性我们可以使用 Mutating
修改这个资源对象内容(如增加验证的信息),在 validating
中验证是否合法。composition of admission controllers
kubernetes中的 admission controllers 由两部分组成:
- 内置在APIServer中的准入控制器 build-in list
- 特殊的控制器;也是内置在APIServer中,但提供一些自定义的功能
- MutatingAdmission
- ValidatingAdmission
admission webhook
由于准入控制器是内置在
kube-apiserver
中的,这种情况下就限制了admission controller的可扩展性。在这种背景下,kubernetes提供了一种可扩展的准入控制器 extensible admission controllers
,这种行为叫做动态准入控制 Dynamic Admission Control
,而提供这个功能的就是 admission webhook
。admission webhook
通俗来讲就是 HTTP 回调,通过定义一个http server,接收准入请求并处理。用户可以通过kubernetes提供的两种类型的 admission webhook
,validating admission webhook 和 mutating admission webhook。来完成自定义的准入策略的处理。webhook 就是
注:从上面的流程图也可以看出,admission webhook 也是有顺序的。首先调用mutating webhook,然后会调用validating webhook。如何使用准入控制器 使用条件:kubernetes v1.16 使用
admissionregistration.k8s.io/v1
;kubernetes v1.9 使用 admissionregistration.k8s.io/v1beta1
。如何在集群中开启准入控制器? :查看kube-apiserver 的启动参数
--enable-admission-plugins
;通过该参数来配置要启动的准入控制器,如 --enable-admission-plugins=NodeRestriction
多个准入控制器以 ,
分割,顺序无关紧要。 反之使用 --disable-admission-plugins
参数可以关闭相应的准入控制器(Refer to apiserver opts)。通过
kubectl
命令可以看到,当前kubernetes集群所支持准入控制器的版本$ kubectl api-versions | grep admissionregistration.k8s.io/v1
admissionregistration.k8s.io/v1
admissionregistration.k8s.io/v1beta1
webhook工作原理 通过上面的学习,已经了解到了两种webhook的工作原理如下所示:
mutating webhook,会在持久化前拦截在 MutatingWebhookConfiguration 中定义的规则匹配的请求。MutatingAdmissionWebhook 通过向 mutating webhook 服务器发送准入请求来执行验证。那么接下来将从源码中看这个在这个工作流程中,究竟做了些什么?
validaing webhook,会在持久化前拦截在ValidatingWebhookConfiguration
中定义的规则匹配的请求。ValidatingAdmissionWebhook 通过将准入请求发送到 validating webhook server来执行验证。
资源类型
对于 1.9 版本之后,也就是
v1
版本 ,admission 被定义在 k8s.io\api\admissionregistration\v1\types.go ,大同小异,因为本地只有1.18集群,所以以这个讲解。对于
Validating Webhook
来讲实现主要都在webhook中type ValidatingWebhookConfiguration struct {
// 每个api必须包含下列的metadata,这个是kubernetes规范,可以在注释中的url看到相关文档
metav1.TypeMeta `json:",inline"`
metav1.ObjectMeta `json:"metadata,omitempty" protobuf:"bytes,1,opt,name=metadata"`
// Webhooks在这里被表示为[]ValidatingWebhook,表示我们可以注册多个
// +optional
// +patchMergeKey=name
// +patchStrategy=merge
Webhooks []ValidatingWebhook `json:"webhooks,omitempty" patchStrategy:"merge" patchMergeKey:"name" protobuf:"bytes,2,rep,name=Webhooks"`
}
webhook,则是对这种类型的webhook提供的操作、资源等。对于这部分不做过多的注释了,因为这里本身为kubernetes API资源,官网有很详细的例子与说明。这里更多字段的意思的可以参考官方 doc
type ValidatingWebhook struct {
//admission webhook的名词,Required
Name string `json:"name" protobuf:"bytes,1,opt,name=name"` // ClientConfig 定义了与webhook通讯的方式 Required
ClientConfig WebhookClientConfig `json:"clientConfig" protobuf:"bytes,2,opt,name=clientConfig"` // rule表示了webhook对于哪些资源及子资源的操作进行关注
Rules []RuleWithOperations `json:"rules,omitempty" protobuf:"bytes,3,rep,name=rules"` // FailurePolicy 对于无法识别的value将如何处理,allowed/Ignore optional
FailurePolicy *FailurePolicyType `json:"failurePolicy,omitempty" protobuf:"bytes,4,opt,name=failurePolicy,casttype=FailurePolicyType"` // matchPolicy 定义了如何使用“rules”列表来匹配传入的请求。
MatchPolicy *MatchPolicyType `json:"matchPolicy,omitempty" protobuf:"bytes,9,opt,name=matchPolicy,casttype=MatchPolicyType"`
NamespaceSelector *metav1.LabelSelector `json:"namespaceSelector,omitempty" protobuf:"bytes,5,opt,name=namespaceSelector"`
SideEffects *SideEffectClass `json:"sideEffects" protobuf:"bytes,6,opt,name=sideEffects,casttype=SideEffectClass"`
AdmissionReviewVersions []string `json:"admissionReviewVersions" protobuf:"bytes,8,rep,name=admissionReviewVersions"`
}
到这里了解了一个webhook资源的定义,那么这个如何使用呢?通过
Find Usages
找到一个 k8s.io/apiserver/pkg/admission/plugin/webhook/accessors.go 在使用它。这里没有注释,但在结构上可以看出,包含客户端与一系列选择器组成type mutatingWebhookAccessor struct {
*v1.MutatingWebhook
uidstring
configurationName string initObjectSelector sync.Once
objectSelectorlabels.Selector
objectSelectorErrerror initNamespaceSelector sync.Once
namespaceSelectorlabels.Selector
namespaceSelectorErrerror initClient sync.Once
client*rest.RESTClient
clientErrerror
}
accessor
因为包含了整个webhookconfig定义的一些动作(这里个人这么觉得)。accessor.go
下面 有一个 GetRESTClient
方法 ,通过这里可以看出,这里做的就是使用根据 accessor
构造一个客户端。func (m *mutatingWebhookAccessor) GetRESTClient(clientManager *webhookutil.ClientManager) (*rest.RESTClient, error) {
m.initClient.Do(func() {
m.client, m.clientErr = clientManager.HookClient(hookClientConfigForWebhook(m))
})
return m.client, m.clientErr
}
到这步骤已经没必要往下看了,因已经知道这里是请求webhook前的步骤了,下面就是何时请求了。
k8s.io\apiserver\pkg\admission\plugin\webhook\validating\dispatcher.go 下面有两个方法,Dispatch去请求我们自己定义的webhook
func (d *validatingDispatcher) Dispatch(ctx context.Context, attr admission.Attributes, o admission.ObjectInterfaces, hooks []webhook.WebhookAccessor) error {
var relevantHooks []*generic.WebhookInvocation
// Construct all the versions we need to call our webhooks
versionedAttrs := map[schema.GroupVersionKind]*generic.VersionedAttributes{}
for _, hook := range hooks {
invocation, statusError := d.plugin.ShouldCallHook(hook, attr, o)
if statusError != nil {
return statusError
}
if invocation == nil {
continue
}
relevantHooks = append(relevantHooks, invocation)
// If we already have this version, continue
if _, ok := versionedAttrs[invocation.Kind];
ok {
continue
}
versionedAttr, err := generic.NewVersionedAttributes(attr, invocation.Kind, o)
if err != nil {
return apierrors.NewInternalError(err)
}
versionedAttrs[invocation.Kind] = versionedAttr
} if len(relevantHooks) == 0 {
// no matching hooks
return nil
} // Check if the request has already timed out before spawning remote calls
select {
case <-ctx.Done():
// parent context is canceled or timed out, no point in continuing
return apierrors.NewTimeoutError("request did not complete within requested timeout", 0)
default:
} wg := sync.WaitGroup{}
errCh := make(chan error, len(relevantHooks))
wg.Add(len(relevantHooks))
// 循环所有相关的注册的hook
for i := range relevantHooks {
go func(invocation *generic.WebhookInvocation) {
defer wg.Done()
// invacation 中有一个 Accessor,Accessor注册了一个相关的webhookconfig
// 也就是我们 kubectl -f 注册进来的那个webhook的相关配置
hook, ok := invocation.Webhook.GetValidatingWebhook()
if !ok {
utilruntime.HandleError(fmt.Errorf("validating webhook dispatch requires v1.ValidatingWebhook, but got %T", hook))
return
}
versionedAttr := versionedAttrs[invocation.Kind]
t := time.Now()
// 调用了callHook去请求我们自定义的webhook
err := d.callHook(ctx, hook, invocation, versionedAttr)
ignoreClientCallFailures := hook.FailurePolicy != nil && *hook.FailurePolicy == v1.Ignore
rejected := false
if err != nil {
switch err := err.(type) {
case *webhookutil.ErrCallingWebhook:
if !ignoreClientCallFailures {
rejected = true
admissionmetrics.Metrics.ObserveWebhookRejection(hook.Name, "validating", string(versionedAttr.Attributes.GetOperation()), admissionmetrics.WebhookRejectionCallingWebhookError, 0)
}
case *webhookutil.ErrWebhookRejection:
rejected = true
admissionmetrics.Metrics.ObserveWebhookRejection(hook.Name, "validating", string(versionedAttr.Attributes.GetOperation()), admissionmetrics.WebhookRejectionNoError, int(err.Status.ErrStatus.Code))
default:
rejected = true
admissionmetrics.Metrics.ObserveWebhookRejection(hook.Name, "validating", string(versionedAttr.Attributes.GetOperation()), admissionmetrics.WebhookRejectionAPIServerInternalError, 0)
}
}
admissionmetrics.Metrics.ObserveWebhook(time.Since(t), rejected, versionedAttr.Attributes, "validating", hook.Name)
if err == nil {
return
}if callErr, ok := err.(*webhookutil.ErrCallingWebhook);
ok {
if ignoreClientCallFailures {
klog.Warningf("Failed calling webhook, failing open %v: %v", hook.Name, callErr)
utilruntime.HandleError(callErr)
return
}klog.Warningf("Failed calling webhook, failing closed %v: %v", hook.Name, err)
errCh <- apierrors.NewInternalError(err)
return
}if rejectionErr, ok := err.(*webhookutil.ErrWebhookRejection);
ok {
err = rejectionErr.Status
}
klog.Warningf("rejected by webhook %q: %#v", hook.Name, err)
errCh <- err
}(relevantHooks[i])
}
wg.Wait()
close(errCh) var errs []error
for e := range errCh {
errs = append(errs, e)
}
if len(errs) == 0 {
return nil
}
if len(errs) > 1 {
for i := 1;
i < len(errs);
i++ {
// TODO: merge status errors;
until then, just return the first one.
utilruntime.HandleError(errs[i])
}
}
return errs[0]
}
callHook 可以理解为真正去请求我们自定义的webhook服务的动作
func (d *validatingDispatcher) callHook(ctx context.Context, h *v1.ValidatingWebhook, invocation *generic.WebhookInvocation, attr *generic.VersionedAttributes) error {
if attr.Attributes.IsDryRun() {
if h.SideEffects == nil {
return &webhookutil.ErrCallingWebhook{WebhookName: h.Name, Reason: fmt.Errorf("Webhook SideEffects is nil")}
}
if !(*h.SideEffects == v1.SideEffectClassNone || *h.SideEffects == v1.SideEffectClassNoneOnDryRun) {
return webhookerrors.NewDryRunUnsupportedErr(h.Name)
}
}uid, request, response, err := webhookrequest.CreateAdmissionObjects(attr, invocation)
if err != nil {
return &webhookutil.ErrCallingWebhook{WebhookName: h.Name, Reason: err}
}
// 发生请求,可以看到,这里从上面的讲到的地方获取了一个客户端
client, err := invocation.Webhook.GetRESTClient(d.cm)
if err != nil {
return &webhookutil.ErrCallingWebhook{WebhookName: h.Name, Reason: err}
}
trace := utiltrace.New("Call validating webhook",
utiltrace.Field{"configuration", invocation.Webhook.GetConfigurationName()},
utiltrace.Field{"webhook", h.Name},
utiltrace.Field{"resource", attr.GetResource()},
utiltrace.Field{"subresource", attr.GetSubresource()},
utiltrace.Field{"operation", attr.GetOperation()},
utiltrace.Field{"UID", uid})
defer trace.LogIfLong(500 * time.Millisecond)// 这里设置超时,超时时长就是在yaml资源清单中设置的那个值
if h.TimeoutSeconds != nil {
var cancel context.CancelFunc
ctx, cancel = context.WithTimeout(ctx, time.Duration(*h.TimeoutSeconds)*time.Second)
defer cancel()
}
// 直接用post请求我们自己定义的webhook接口
r := client.Post().Body(request)// if the context has a deadline, set it as a parameter to inform the backend
if deadline, hasDeadline := ctx.Deadline();
hasDeadline {
// compute the timeout
if timeout := time.Until(deadline);
timeout > 0 {
// if it's not an even number of seconds, round up to the nearest second
if truncated := timeout.Truncate(time.Second);
truncated != timeout {
timeout = truncated + time.Second
}
// set the timeout
r.Timeout(timeout)
}
}if err := r.Do(ctx).Into(response);
err != nil {
return &webhookutil.ErrCallingWebhook{WebhookName: h.Name, Reason: err}
}
trace.Step("Request completed")result, err := webhookrequest.VerifyAdmissionResponse(uid, false, response)
if err != nil {
return &webhookutil.ErrCallingWebhook{WebhookName: h.Name, Reason: err}
}for k, v := range result.AuditAnnotations {
key := h.Name + "/" + k
if err := attr.Attributes.AddAnnotation(key, v);
err != nil {
klog.Warningf("Failed to set admission audit annotation %s to %s for validating webhook %s: %v", key, v, h.Name, err)
}
}
if result.Allowed {
return nil
}
return &webhookutil.ErrWebhookRejection{Status: webhookerrors.ToStatusErr(h.Name, result.Result)}
}
走到这里基本上对
admission webhook
有了大致的了解,可以知道这个操作是由 apiserver 完成的。下面就实际操作下自定义一个webhook。这里还有两个概念,就是请求参数 AdmissionRequest 和相应参数 AdmissionResponse,这些可以在
callHook
中看到,这两个参数被定义在 k8s.io\api\admission\v1\types.go ;这两个参数也就是我们在自定义 webhook 时需要处理接收到的body的结构,以及我们响应内容数据结构。如何编写一个自定义的admission webhook 通过上面的学习了解到了,自定义的webhook就是做为kubernetes提供给用户两种admission controller来验证自定义业务的一个中间件 admission webhook。本质上他是一个HTTP Server,用户可以使用任何语言来完成这部分功能。当然,如果涉及到需要对kubernetes集群资源操作的话,还是建议使用kubernetes官方提供了SDK的编程语言来完成自定义的webhook。
那么完成一个自定义admission webhook需要两个步骤:
- 将相关的webhook config注册给kubernetes,也就是让kubernetes知道你的webhook
- 准备一个http server来处理 apiserver发过来验证的信息
注:这里使用go net/http包,本身不区分方法处理HTTP的何种请求,如果用其他框架实现的,如django,需要指定对应方法需要为POST向kubernetes注册webhook对象
kubernetes提供的两种类型可自定义的准入控制器,和其他资源一样,可以利用资源清单,动态配置那些资源要被adminssion webhook处理。 kubernetes将这种形式抽象为两种资源:
- ValidatingWebhookConfiguration
- MutatingWebhookConfiguration
apiVersion: admissionregistration.k8s.io/v1
kind: ValidatingWebhookConfiguration
metadata:
name: "pod-policy.example.com"
webhooks:
- name: "pod-policy.example.com"
rules:
- apiGroups:[""] # 拦截资源的Group "" 表示 core。"*" 表示所有。
apiVersions: ["v1"] # 拦截资源的版本
operations:["CREATE"] # 什么请求下拦截
resources:["pods"]# 拦截什么资源
scope:"Namespaced" # 生效的范围,cluster还是namespace "*"表示没有范围限制。
clientConfig: # 我们部署的webhook服务,
service: # service是在cluster-in模式下
namespace: "example-namespace"
name: "example-service"
port: 443 # 服务的端口
path: "/validate" # path是对应用于验证的接口
# caBundle是提供给 admission webhook CA证书
caBundle: "Ci0tLS0tQk......tLS0K"
admissionReviewVersions: ["v1", "v1beta1"]
sideEffects: None
timeoutSeconds: 5 # 1-30s直接,表示请求api的超时时间
MutatingAdmission
apiVersion: admissionregistration.k8s.io/v1
kind: ValidatingWebhookConfiguration
metadata:
name: "valipod-policy.example.com"
webhooks:
- name: "valipod-policy.example.com"
rules:
- apiGroups:["apps"] # 拦截资源的Group "" 表示 core。"*" 表示所有。
apiVersions: ["v1"] # 拦截资源的版本
operations:["CREATE"] # 什么请求下拦截
resources:["deployments"]# 拦截什么资源
scope:"Namespaced" # 生效的范围,cluster还是namespace "*"表示没有范围限制。
clientConfig: # 我们部署的webhook服务,
url: "https://10.0.0.1:81/validate" # 这里是外部模式
#service: # service是在cluster-in模式下
#namespace: "default"
#name: "admission-webhook"
#port: 81 # 服务的端口
#path: "/mutate" # path是对应用于验证的接口
# caBundle是提供给 admission webhook CA证书
caBundle: "Ci0tLS0tQk......tLS0K"
admissionReviewVersions: ["v1"]
sideEffects: None
timeoutSeconds: 5 # 1-30s直接,表示请求api的超时时间
注:对于webhook,也可以引入外部的服务,并非必须部署到集群内部对于外部服务来讲,需要
clientConfig
中的 service
, 更换为 url
;
通过 url
参数可以将一个外部的服务引入apiVersion: admissionregistration.k8s.io/v1
kind: MutatingWebhookConfiguration
...
webhooks:
- name: my-webhook.example.com
clientConfig:
url: "https://my-webhook.example.com:9443/my-webhook-path"
...
注:这里的url规则必须准守下列形式:更多的配置可以参考kubernetes官方提供的 doc
scheme://host:port/path
- 使用了url 时,这里不应填写集群内的服务
scheme
必须是 https,不能为http,这就意味着,引入外部时也需要- 配置时使用了,
?xx=xx
的参数也是不被允许的(官方说法是这样的,通过源码学习了解到因为会发送特定的请求体,所以无需管参数)
准备一个webhook
让我们编写我们的 webhook server。将创建两个钩子,
/mutate
与 /validate
;/mutate
将在创建deployment资源时,基于版本,给资源加上注释webhook.example.com/allow: true
/validate
将对/mutate
增加的allow:true
那么则继续,否则拒绝。
package mainimport (
"context"
"crypto/tls"
"encoding/json"
"fmt"
"io/ioutil"
"net/http"
"os"
"os/signal"
"strings"
"syscall" v1admission "k8s.io/api/admission/v1"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/runtime/serializer" appv1 "k8s.io/api/apps/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/klog"
)type patch struct {
Opstring`json:"op"`
Pathstring`json:"path"`
Value map[string]string `json:"value"`
}func serve(w http.ResponseWriter, r *http.Request) { var body []byte
if data, err := ioutil.ReadAll(r.Body);
err == nil {
body = data
}
klog.Infof(fmt.Sprintf("receive request: %v....", string(body)[:130]))
if len(body) == 0 {
klog.Error(fmt.Sprintf("admission request body is empty"))
http.Error(w, fmt.Errorf("admission request body is empty").Error(), http.StatusBadRequest)
return
}
var admission v1admission.AdmissionReview
codefc := serializer.NewCodecFactory(runtime.NewScheme())
decoder := codefc.UniversalDeserializer()
_, _, err := decoder.Decode(body, nil, &admission) if err != nil {
msg := fmt.Sprintf("Request could not be decoded: %v", err)
klog.Error(msg)
http.Error(w, msg, http.StatusBadRequest)
return
} if admission.Request == nil {
klog.Error(fmt.Sprintf("admission review can't be used: Request field is nil"))
http.Error(w, fmt.Errorf("admission review can't be used: Request field is nil").Error(), http.StatusBadRequest)
return
} switch strings.Split(r.RequestURI, "?")[0] {
case "/mutate":
req := admission.Request
var admissionResp v1admission.AdmissionReview
admissionResp.APIVersion = admission.APIVersion
admissionResp.Kind = admission.Kind
klog.Infof("AdmissionReview for Kind=%v, Namespace=%v Name=%v UID=%v Operation=%v",
req.Kind.Kind, req.Namespace, req.Name, req.UID, req.Operation)
switch req.Kind.Kind {
case "Deployment":
var (
respstr []byte
errerror
deployappv1.Deployment
)
if err = json.Unmarshal(req.Object.Raw, &deploy);
err != nil {
respStructure := v1admission.AdmissionResponse{Result: &metav1.Status{
Message: fmt.Sprintf("could not unmarshal resouces review request: %v", err),
Code:http.StatusInternalServerError,
}}
klog.Error(fmt.Sprintf("could not unmarshal resouces review request: %v", err))
if respstr, err = json.Marshal(respStructure);
err != nil {
klog.Error(fmt.Errorf("could not unmarshal resouces review response: %v", err))
http.Error(w, fmt.Errorf("could not unmarshal resouces review response: %v", err).Error(), http.StatusInternalServerError)
return
}
http.Error(w, string(respstr), http.StatusBadRequest)
return
}current_annotations := deploy.GetAnnotations()
pl := []patch{}
for k, v := range current_annotations {
pl = append(pl, patch{
Op:"add",
Path: "/metadata/annotations",
Value: map[string]string{
k: v,
},
})
}
pl = append(pl, patch{
Op:"add",
Path: "/metadata/annotations",
Value: map[string]string{
deploy.Name + "/Allow": "true",
},
})annotationbyte, err := json.Marshal(pl)if err != nil {
http.Error(w, err.Error(), http.StatusInternalServerError)
return
}
respStructure := &v1admission.AdmissionResponse{
UID:req.UID,
Allowed: true,
Patch:annotationbyte,
PatchType: func() *v1admission.PatchType {
t := v1admission.PatchTypeJSONPatch
return &t
}(),
Result: &metav1.Status{
Message: fmt.Sprintf("could not unmarshal resouces review request: %v", err),
Code:http.StatusOK,
},
}
admissionResp.Response = respStructureklog.Infof("sending response: %s....", admissionResp.Response.String()[:130])
respByte, err := json.Marshal(admissionResp)
if err != nil {
klog.Errorf("Can't encode response messages: %v", err)
http.Error(w, err.Error(), http.StatusInternalServerError)
}
klog.Infof("prepare to write response...")
w.Header().Set("Content-Type", "application/json")
if _, err := w.Write(respByte);
err != nil {
klog.Errorf("Can't write response: %v", err)
http.Error(w, fmt.Sprintf("could not write response: %v", err), http.StatusInternalServerError)
}default:
klog.Error(fmt.Sprintf("unsupport resouces review request type"))
http.Error(w, "unsupport resouces review request type", http.StatusBadRequest)
} case "/validate":
req := admission.Request
var admissionResp v1admission.AdmissionReview
admissionResp.APIVersion = admission.APIVersion
admissionResp.Kind = admission.Kind
klog.Infof("AdmissionReview for Kind=%v, Namespace=%v Name=%v UID=%v Operation=%v",
req.Kind.Kind, req.Namespace, req.Name, req.UID, req.Operation)
var (
deployappv1.Deployment
respstr []byte
)
switch req.Kind.Kind {
case "Deployment":
if err = json.Unmarshal(req.Object.Raw, &deploy);
err != nil {
respStructure := v1admission.AdmissionResponse{Result: &metav1.Status{
Message: fmt.Sprintf("could not unmarshal resouces review request: %v", err),
Code:http.StatusInternalServerError,
}}
klog.Error(fmt.Sprintf("could not unmarshal resouces review request: %v", err))
if respstr, err = json.Marshal(respStructure);
err != nil {
klog.Error(fmt.Errorf("could not unmarshal resouces review response: %v", err))
http.Error(w, fmt.Errorf("could not unmarshal resouces review response: %v", err).Error(), http.StatusInternalServerError)
return
}
http.Error(w, string(respstr), http.StatusBadRequest)
return
}
}
al := deploy.GetAnnotations()
respStructure := v1admission.AdmissionResponse{
UID: req.UID,
}
if al[fmt.Sprintf("%s/Allow", deploy.Name)] == "true" {
respStructure.Allowed = true
respStructure.Result = &metav1.Status{
Code: http.StatusOK,
}
} else {
respStructure.Allowed = false
respStructure.Result = &metav1.Status{
Code: http.StatusForbidden,
Reason: func() metav1.StatusReason {
return metav1.StatusReasonForbidden
}(),
Message: fmt.Sprintf("the resource %s couldn't to allow entry.", deploy.Kind),
}
}admissionResp.Response = &respStructureklog.Infof("sending response: %s....", admissionResp.Response.String()[:130])
respByte, err := json.Marshal(admissionResp)
if err != nil {
klog.Errorf("Can't encode response messages: %v", err)
http.Error(w, err.Error(), http.StatusInternalServerError)
}
klog.Infof("prepare to write response...")
w.Header().Set("Content-Type", "application/json")
if _, err := w.Write(respByte);
err != nil {
klog.Errorf("Can't write response: %v", err)
http.Error(w, fmt.Sprintf("could not write response: %v", err), http.StatusInternalServerError)
}
}
}func main() {
var (
cert, key string
) if cert = os.Getenv("TLS_CERT");
len(cert) == 0 {
cert = "./tls/tls.crt"
} if key = os.Getenv("TLS_KEY");
len(key) == 0 {
key = "./tls/tls.key"
} ca, err := tls.LoadX509KeyPair(cert, key)
if err != nil {
klog.Error(err.Error())
return
} server := &http.Server{
Addr: ":81",
TLSConfig: &tls.Config{
Certificates: []tls.Certificate{
ca,
},
},
} httpserver := http.NewServeMux() httpserver.HandleFunc("/validate", serve)
httpserver.HandleFunc("/mutate", serve)
httpserver.HandleFunc("/ping", func(w http.ResponseWriter, r *http.Request) {
klog.Info(fmt.Sprintf("%s %s", r.RequestURI, "pong"))
fmt.Fprint(w, "pong")
})
server.Handler = httpserver go func() {
if err := server.ListenAndServeTLS("", "");
err != nil {
klog.Errorf("Failed to listen and serve webhook server: %v", err)
}
}() klog.Info("starting serve.")
signalChan := make(chan os.Signal, 1)
signal.Notify(signalChan, syscall.SIGINT, syscall.SIGTERM)
<-signalChan klog.Infof("Got shut signal, shutting...")
if err := server.Shutdown(context.Background());
err != nil {
klog.Errorf("HTTP server Shutdown: %v", err)
}
}
对应的Dockerfile
FROM golang:alpine AS builder
MAINTAINER cylon
WORKDIR /admission
COPY ./ /admission
ENV GOPROXY https://goproxy.cn,direct
RUN \
sed -i 's/dl-cdn.alpinelinux.org/mirrors.ustc.edu.cn/g' /etc/apk/repositories && \
apk add upx&& \
GOOS=linux GOARCH=amd64 CGO_ENABLED=0 go build -ldflags "-s -w" -o webhook main.go && \
upx -1 webhook && \
chmod +x webhookFROM alpine AS runner
WORKDIR /go/admission
COPY --from=builder /admission/webhook .
VOLUME ["/admission"]
集群内部部署所需的资源清单
apiVersion: v1
kind: Service
metadata:
name: admission-webhook
labels:
app: admission-webhook
spec:
ports:
- port: 81
targetPort: 81
selector:
app: simple-webhook
---
apiVersion: apps/v1
kind: Deployment
metadata:
labels:
app: simple-webhook
name: simple-webhook
spec:
replicas: 1
selector:
matchLabels:
app: simple-webhook
template:
metadata:
labels:
app: simple-webhook
spec:
containers:
- image: cylonchau/simple-webhook:v0.0.2
imagePullPolicy: IfNotPresent
name: webhook
command: ["./webhook"]
env:
- name: "TLS_CERT"
value: "./tls/tls.crt"
- name: "TLS_KEY"
value: "./tls/tls.key"
- name: NS_NAME
valueFrom:
fieldRef:
apiVersion: v1
fieldPath: metadata.namespace
ports:
- containerPort: 81
volumeMounts:
- name: tlsdir
mountPath: /go/admission/tls
readOnly: true
volumes:
- name: tlsdir
secret:
secretName: webhook
---
apiVersion: admissionregistration.k8s.io/v1
kind: MutatingWebhookConfiguration
metadata:
name: "pod-policy.example.com"
webhooks:
- name: "pod-policy.example.com"
rules:
- apiGroups:["apps"] # 拦截资源的Group "" 表示 core。"*" 表示所有。
apiVersions: ["v1"] # 拦截资源的版本
operations:["CREATE"] # 什么请求下拦截
resources:["deployments"]# 拦截什么资源
scope:"Namespaced" # 生效的范围,cluster还是namespace "*"表示没有范围限制。
clientConfig: # 我们部署的webhook服务,
url: "https://10.0.0.1:81/mutate"
#service: # service是在cluster-in模式下
#namespace: "default"
#name: "admission-webhook"
#port: 81 # 服务的端口
#path: "/mutate" # path是对应用于验证的接口
# caBundle是提供给 admission webhook CA证书
caBundle: Put you CA (base64 encode) in here
admissionReviewVersions: ["v1"]
sideEffects: None
timeoutSeconds: 5 # 1-30s直接,表示请求api的超时时间
---
apiVersion: admissionregistration.k8s.io/v1
kind: ValidatingWebhookConfiguration
metadata:
name: "valipod-policy.example.com"
webhooks:
- name: "valipod-policy.example.com"
rules:
- apiGroups:["apps"] # 拦截资源的Group "" 表示 core。"*" 表示所有。
apiVersions: ["v1"] # 拦截资源的版本
operations:["CREATE"] # 什么请求下拦截
resources:["deployments"]# 拦截什么资源
scope:"Namespaced" # 生效的范围,cluster还是namespace "*"表示没有范围限制。
clientConfig: # 我们部署的webhook服务,
#service: # service是在cluster-in模式下
#namespace: "default"
#name: "admission-webhook"
#port: 81 # 服务的端口
#path: "/mutate" # path是对应用于验证的接口
# caBundle是提供给 admission webhook CA证书
caBundle: Put you CA (base64 encode) in here
admissionReviewVersions: ["v1"]
sideEffects: None
timeoutSeconds: 5 # 1-30s直接,表示请求api的超时时间
这里需要主义的问题 证书问题
如果需要
cluster-in
,那么则需要对对应webhookconfig资源配置 service
;如果使用的是外部部署,则需要配置对应访问地址,如:“https://xxxx:port/method”这两种方式的证书均需要对应的
subjectAltName
,cluster-in
模式 需要对应service名称,如,至少包含serviceName.NS.svc
这一个域名。下面就是证书类问题的错误
Failed calling webhook, failing closed pod-policy.example.com: failed calling webhook "pod-policy.example.com": Post https://admission-webhook.default.svc:81/mutate?timeout=5s: x509: certificate signed by unknown authority (possibly because of "crypto/rsa: verification error" while trying to verify candidate authority certificate "admission-webhook-ca")
相应信息问题
上面我们了解到的APIServer是去发出
v1admission.AdmissionReview
也就是 Request 和 Response类型的,所以,为了更清晰的表示出问题所在,需要对响应格式中的 Reason
与 Message
配置,这也就是我们在客户端看到的报错信息。&metav1.Status{
Code: http.StatusForbidden,
Reason: func() metav1.StatusReason {
return metav1.StatusReasonForbidden
}(),
Message: fmt.Sprintf("the resource %s couldn't to allow entry.", deploy.Kind),
}
通过上面的设置用户可以看到下列错误
$ kubectl apply -f nginx.yaml
Error from server (Forbidden): error when creating "nginx.yaml": admission webhook "valipod-policy.example.com" denied the request: the resource Deployment couldn't to allow entry.
注:必须的参数还包含,UID,allowed,这两个是必须的,上面阐述的只是对用户友好的提示信息下面的报错就是对相应格式设置错误
Error from server (InternalError): error when creating "nginx.yaml": Internal error occurred: failed calling webhook "pod-policy.example.com": the server rejected our request for an unknown reason
相应信息版本问题
相应信息也需要指定一个版本,这个与请求来的结构中拿即可
admissionResp.APIVersion = admission.APIVersion
admissionResp.Kind = admission.Kind
下面是没有为对应相应信息配置对应KV的值出现的报错
Error from server (InternalError): error when creating "nginx.yaml": Internal error occurred: failed calling webhook "pod-policy.example.com": expected webhook response of admission.k8s.io/v1, Kind=AdmissionReview, got /, Kind=
关于patch
kubernetes中patch使用的是特定的规范,如
jsonpatch
kubernetes当前唯一支持的下面就是字符串类型设置为布尔型产生的报错patchType
是JSONPatch
。 有关更多详细信息,请参见 JSON patch
对于jsonpatch
是一个固定的类型,在go中必须定义其结构体
{ "op": "add", // 做什么操作 "path": "/spec/replicas", // 操作的路径 "value": 3 // 对应添加的key value }
Error from server (InternalError): error when creating "nginx.yaml": Internal error occurred: v1.Deployment.ObjectMeta: v1.ObjectMeta.Annotations: ReadString: expects " or n, but found t, error found in #10 byte of ...|t/Allow":true},"crea|..., bigger context ...|tadata":{"annotations":{"nginx-deployment/Allow":true},"creationTimestamp":null,"managedFields":[{"m|..
准备证书
Ubuntu
touch ./demoCAindex.txt
touch ./demoCA/serial
touch ./demoCA/crlnumber
echo 01 > ./demoCA/serial
mkdir ./demoCA/newcertsopenssl genrsa -out cakey.pem 2048openssl req -new \
-x509 \
-key cakey.pem \
-out cacert.pem \
-days 3650 \
-subj "/CN=admission webhook ca"openssl genrsa -out tls.key 2048openssl req -new \
-key tls.key \
-subj "/CN=admission webhook client" \
-reqexts webhook \
-config <(cat /etc/ssl/openssl.cnf \
<(printf "[webhook]\nsubjectAltName=DNS: admission-webhook, DNS: admission-webhook.default.svc, DNS: admission-webhook.default.svc.cluster.local, IP:10.0.0.1,IP:10.0.0.4")) \
-out tls.csrsed -i 's/= match/= optional/g' /etc/ssl/openssl.cnfopenssl ca \
-in tls.csr \
-cert cacert.pem \
-keyfile cakey.pem \
-out tls.crt \
-days 300 \
-extensions webhook \
-extfile <(cat /etc/ssl/openssl.cnf \
<(printf "[webhook]\nsubjectAltName=DNS: admission-webhook, DNS: admission-webhook.default.svc, DNS: admission-webhook.default.svc.cluster.local, IP:10.0.0.1,IP:10.0.0.4"))
CentOS
touch /etc/pki/CA/index.txt
touch /etc/pki/CA/serial # 下一个要颁发的编号 16进制
touch /etc/pki/CA/crlnumber
echo 01 > /etc/pki/CA/serialopenssl req -new \
-x509 \
-key cakey.pem \
-out cacert.pem \
-days 3650 \
-subj "/CN=admission webhook ca"openssl genrsa -out tls.key 2048openssl req -new \
-key tls.key \
-subj "/CN=admission webhook client" \
-reqexts webhook \
-config <(cat /etc/pki/tls/openssl.cnf \
<(printf "[webhook]\nsubjectAltName=DNS: admission-webhook, DNS: admission-webhook.default.svc, DNS: admission-webhook.default.svc.cluster.local, IP:10.0.0.1,IP:10.0.0.4")) \
-out tls.csrsed -i 's/= match/= optional/g' /etc/ssl/openssl.cnfopenssl ca \
-in tls.csr \
-cert cacert.pem \
-keyfile cakey.pem \
-out tls.crt \
-days 300 \
-extensions webhook \
-extfile <(cat /etc/pki/tls/openssl.cnf \
<(printf "[webhook]\nsubjectAltName=DNS: admission-webhook, DNS: admission-webhook.default.svc, DNS: admission-webhook.default.svc.cluster.local, IP:10.0.0.1,IP:10.0.0.4"))
通过部署测试结果
可以看到我们自己注入的 annotation
nginx-deployment/Allow: true
,在该示例中,仅为演示过程,而不是真的策略,实际环境中可以根据情况进行定制自己的策略。结果可以看出,当在
mutating
中不通过,即缺少对应的 annotation 标签 , 则 validating
会不允许准入$ kubectl describe deploy nginx-deployment
Name:nginx-deployment
Namespace:default
CreationTimestamp:Mon, 11 Jul 2022 20:25:16 +0800
Labels:
Annotations:deployment.kubernetes.io/revision: 1
nginx-deployment/Allow: true
Selector:app=nginx
Replicas:1 desired | 1 updated | 1 total | 1 available | 0 unavailable
StrategyType:RollingUpdate
MinReadySeconds:0
RollingUpdateStrategy:25% max unavailable, 25% max surge
Pod Template:
Labels:app=nginx
Containers:
nginx:
Image:nginx:1.14.2
Reference
extensible admission controllers
K8S client-go Patch example
admission controllers response
【#|深入解析Kubernetes admission webhooks】a guide to kubernetes admission controllers
推荐阅读
- Docker|开发者,Docker是什么能干什么()
- 一图看懂 | 腾讯云原生安全“3+1”重保防护体系
- 中间件|蚂蚁金服开源了这一款研发框架!还是增强版的
- Docker|【五、docker镜像详解】
- docker|【云原生】五、Docker 网络管理
- spring|Kubernetes、Spring Cloud和微服务
- Nginx|Nginx+Kubernetes+Spring系列等微服务+PDF
- 程序人生|微服务的构建环境比较--Spring Cloud和Kubernetes
- K8S系列Kubernetes简介