This repository has been archived by the owner on Jul 11, 2023. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 279
/
webhook.go
371 lines (314 loc) · 14.2 KB
/
webhook.go
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
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
package injector
import (
"context"
"crypto/tls"
"encoding/json"
"fmt"
"net/http"
"strings"
"time"
mapset "github.com/deckarep/golang-set"
"github.com/google/uuid"
"github.com/pkg/errors"
"k8s.io/api/admission/v1beta1"
admissionv1beta1 "k8s.io/api/admissionregistration/v1beta1"
corev1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/runtime/serializer"
"k8s.io/apimachinery/pkg/types"
"k8s.io/client-go/kubernetes"
admissionRegistrationTypes "k8s.io/client-go/kubernetes/typed/admissionregistration/v1beta1"
"github.com/openservicemesh/osm/pkg/catalog"
"github.com/openservicemesh/osm/pkg/certificate"
"github.com/openservicemesh/osm/pkg/configurator"
"github.com/openservicemesh/osm/pkg/constants"
k8s "github.com/openservicemesh/osm/pkg/kubernetes"
"github.com/openservicemesh/osm/pkg/webhook"
)
var (
codecs = serializer.NewCodecFactory(runtime.NewScheme())
deserializer = codecs.UniversalDeserializer()
)
const (
// MutatingWebhookName is the name of the mutating webhook used for sidecar injection
MutatingWebhookName = "osm-inject.k8s.io"
// webhookCreatePod is the HTTP path at which the webhook expects to receive pod creation events
webhookCreatePod = "/mutate-pod-creation"
// WebhookHealthPath is the HTTP path at which the health of the webhook can be queried
WebhookHealthPath = "/healthz"
// webhookTimeoutStr is the url variable name for timeout
webhookMutateTimeoutKey = "timeout"
contentTypeJSON = "application/json"
httpHeaderContentType = "Content-Type"
)
// NewMutatingWebhook starts a new web server handling requests from the injector MutatingWebhookConfiguration
func NewMutatingWebhook(config Config, kubeClient kubernetes.Interface, certManager certificate.Manager, meshCatalog catalog.MeshCataloger, kubeController k8s.Controller, meshName, osmNamespace, webhookConfigName string, stop <-chan struct{}, cfg configurator.Configurator) error {
// This is a certificate issued for the webhook handler
// This cert does not have to be related to the Envoy certs, but it does have to match
// the cert provisioned with the MutatingWebhookConfiguration
webhookHandlerCert, err := certManager.IssueCertificate(
certificate.CommonName(fmt.Sprintf("%s.%s.svc", constants.OSMControllerName, osmNamespace)),
constants.XDSCertificateValidityPeriod)
if err != nil {
return errors.Errorf("Error issuing certificate for the mutating webhook: %+v", err)
}
wh := mutatingWebhook{
config: config,
kubeClient: kubeClient,
certManager: certManager,
meshCatalog: meshCatalog,
kubeController: kubeController,
osmNamespace: osmNamespace,
cert: webhookHandlerCert,
configurator: cfg,
// Envoy sidecars should never be injected in these namespaces
nonInjectNamespaces: mapset.NewSetFromSlice([]interface{}{
metav1.NamespaceSystem,
metav1.NamespacePublic,
osmNamespace,
}),
}
// Start the MutatingWebhook web server
go wh.run(stop)
// Update the MutatingWebhookConfig with the OSM CA bundle
if err = updateMutatingWebhookCABundle(webhookHandlerCert, webhookConfigName, wh.kubeClient); err != nil {
return errors.Errorf("Error configuring MutatingWebhookConfiguration %s: %+v", webhookConfigName, err)
}
return nil
}
func (wh *mutatingWebhook) run(stop <-chan struct{}) {
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
mux := http.NewServeMux()
mux.HandleFunc(WebhookHealthPath, healthHandler)
// We know that the events arriving at this handler are CREATE POD only
// because of the specifics of MutatingWebhookConfiguration template in this repository.
mux.HandleFunc(webhookCreatePod, wh.podCreationHandler)
server := &http.Server{
Addr: fmt.Sprintf(":%d", wh.config.ListenPort),
Handler: mux,
}
log.Info().Msgf("Starting sidecar-injection webhook server on port: %v", wh.config.ListenPort)
go func() {
// Generate a key pair from your pem-encoded cert and key ([]byte).
cert, err := tls.X509KeyPair(wh.cert.GetCertificateChain(), wh.cert.GetPrivateKey())
if err != nil {
log.Error().Err(err).Msg("Error parsing webhook certificate")
return
}
// #nosec G402
server.TLSConfig = &tls.Config{
Certificates: []tls.Certificate{cert},
}
if err := server.ListenAndServeTLS("", ""); err != nil {
log.Error().Err(err).Msg("Sidecar injection webhook HTTP server failed to start")
return
}
}()
// Wait on exit signals
<-stop
// Stop the server
if err := server.Shutdown(ctx); err != nil {
log.Error().Err(err).Msg("Error shutting down sidecar-injection webhook HTTP server")
} else {
log.Info().Msg("Done shutting down sidecar-injection webhook HTTP server")
}
}
func healthHandler(w http.ResponseWriter, _ *http.Request) {
w.WriteHeader(http.StatusOK)
if _, err := w.Write([]byte("Health OK")); err != nil {
log.Error().Err(err).Msg("Error writing bytes for mutating webhook health check handler")
}
}
func (wh *mutatingWebhook) getAdmissionReqResp(proxyUUID uuid.UUID, admissionRequestBody []byte) (requestForNamespace string, admissionResp v1beta1.AdmissionReview) {
var admissionReq v1beta1.AdmissionReview
if _, _, err := deserializer.Decode(admissionRequestBody, nil, &admissionReq); err != nil {
log.Error().Err(err).Msg("Error decoding admission request body")
admissionResp.Response = webhook.AdmissionError(err)
} else {
admissionResp.Response = wh.mutate(admissionReq.Request, proxyUUID)
}
return admissionReq.Request.Namespace, admissionResp
}
// podCreationHandler is a MutatingWebhookConfiguration handler exclusive to POD CREATE events.
func (wh *mutatingWebhook) podCreationHandler(w http.ResponseWriter, req *http.Request) {
log.Trace().Msgf("Received mutating webhook request: Method=%v, URL=%v", req.Method, req.URL)
// Tracks the success of the current injector webhook request
success := false
// Read timeout from request url
reqTimeout, err := readTimeout(req)
if err != nil {
log.Error().Err(err).Msg("Could not read timeout from request url")
}
// Execute at return of this handler
defer webhookTimeTrack(time.Now(), reqTimeout, &success)
if contentType := req.Header.Get(httpHeaderContentType); contentType != contentTypeJSON {
err := errors.Errorf("Invalid content type %s; Expected %s", contentType, contentTypeJSON)
http.Error(w, err.Error(), http.StatusUnsupportedMediaType)
log.Error().Err(err).Msgf("Responded to admission request with HTTP %v", http.StatusUnsupportedMediaType)
return
}
admissionRequestBody, err := webhook.GetAdmissionRequestBody(w, req)
if err != nil {
// Error was already logged and written to the ResponseWriter
return
}
// Create the patches for the spec
// We use req.Namespace because pod.Namespace is "" at this point
// This string uniquely identifies the pod. Ideally this would be the pod.UID, but this is not available at this point.
proxyUUID := uuid.New()
requestForNamespace, admissionResp := wh.getAdmissionReqResp(proxyUUID, admissionRequestBody)
resp, err := json.Marshal(&admissionResp)
if err != nil {
http.Error(w, fmt.Sprintf("Error marshalling admission response: %s", err), http.StatusInternalServerError)
log.Error().Err(err).Msgf("Error marshalling admission response; Responded to admission request for pod with UUID %s in namespace %s with HTTP %v", proxyUUID, requestForNamespace, http.StatusInternalServerError)
return
}
if _, err := w.Write(resp); err != nil {
log.Error().Err(err).Msgf("Error writing admission response for pod with UUID %s in namespace %s", proxyUUID, requestForNamespace)
} else {
success = true // read by the deferred function
}
log.Trace().Msgf("Done responding to admission request for pod with UUID %s in namespace %s", proxyUUID, requestForNamespace)
}
func (wh *mutatingWebhook) mutate(req *v1beta1.AdmissionRequest, proxyUUID uuid.UUID) *v1beta1.AdmissionResponse {
if req == nil {
log.Error().Msg("nil admission Request")
return webhook.AdmissionError(errNilAdmissionRequest)
}
// Decode the Pod spec from the request
var pod corev1.Pod
if err := json.Unmarshal(req.Object.Raw, &pod); err != nil {
log.Error().Err(err).Msgf("Error unmarshaling request to pod with UUID %s in namespace %s", proxyUUID, req.Namespace)
return webhook.AdmissionError(err)
}
// Start building the response
resp := &v1beta1.AdmissionResponse{
Allowed: true,
UID: req.UID,
}
// Check if we must inject the sidecar
if inject, err := wh.mustInject(&pod, req.Namespace); err != nil {
log.Error().Err(err).Msgf("Error checking if sidecar must be injected for pod with UUID %s in namespace %s", proxyUUID, req.Namespace)
return webhook.AdmissionError(err)
} else if !inject {
log.Trace().Msgf("Skipping sidecar injection for pod with UUID %s in namespace %s", proxyUUID, req.Namespace)
return resp
}
patchBytes, err := wh.createPatch(&pod, req, proxyUUID)
if err != nil {
log.Error().Err(err).Msgf("Failed to create patch for pod with UUID %s in namespace %s", proxyUUID, req.Namespace)
return webhook.AdmissionError(err)
}
patchAdmissionResponse(resp, patchBytes)
log.Trace().Msgf("Done creating patch admission response for pod with UUID %s in namespace %s", proxyUUID, req.Namespace)
return resp
}
func (wh *mutatingWebhook) isNamespaceInjectable(namespace string) bool {
// Never inject pods in the OSM Controller namespace or kube-public or kube-system
isInjectableNS := !wh.nonInjectNamespaces.Contains(namespace)
// Ignore namespaces not joined in the mesh.
return isInjectableNS && wh.kubeController.IsMonitoredNamespace(namespace)
}
// mustInject determines whether the sidecar must be injected.
//
// The sidecar injection is performed when the namespace is labeled for monitoring and either of the following is true:
// 1. The pod is explicitly annotated with enabled/yes/true for sidecar injection, or
// 2. The namespace is annotated for sidecar injection and the pod is not explicitly annotated with disabled/no/false
//
// The function returns an error when it is unable to determine whether to perform sidecar injection.
func (wh *mutatingWebhook) mustInject(pod *corev1.Pod, namespace string) (bool, error) {
if !wh.isNamespaceInjectable(namespace) {
log.Warn().Msgf("Mutation request is for pod with UID %s; Injection in Namespace %s is not permitted", pod.ObjectMeta.UID, namespace)
return false, nil
}
// Check if the pod is annotated for injection
podInjectAnnotationExists, podInject, err := isAnnotatedForInjection(pod.Annotations, pod.Kind, fmt.Sprintf("%s/%s", pod.Namespace, pod.Name))
if err != nil {
log.Error().Err(err).Msg("Error determining if the pod is enabled for sidecar injection")
return false, err
}
// Check if the namespace is annotated for injection
ns := wh.kubeController.GetNamespace(namespace)
if ns == nil {
log.Error().Err(errNamespaceNotFound).Msgf("Error retrieving namespace %s", namespace)
return false, err
}
nsInjectAnnotationExists, nsInject, err := isAnnotatedForInjection(ns.Annotations, ns.Kind, ns.Name)
if err != nil {
log.Error().Err(err).Msgf("Error determining if namespace %s is enabled for sidecar injection", namespace)
return false, err
}
if podInjectAnnotationExists && podInject {
// Pod is explicitly annotated to enable sidecar injection
return true, nil
} else if nsInjectAnnotationExists && nsInject {
// Namespace is annotated to enable sidecar injection
if !podInjectAnnotationExists || podInject {
// If pod annotation doesn't exist or if an annotation exists to enable injection, enable it
return true, nil
}
}
// Conditions to inject the sidecar are not met
return false, nil
}
func isAnnotatedForInjection(annotations map[string]string, objectKind string, objectName string) (exists bool, enabled bool, err error) {
inject := strings.ToLower(annotations[constants.SidecarInjectionAnnotation])
log.Trace().Msgf("%s %s has sidecar injection annotation: '%s:%s'", objectKind, objectName, constants.SidecarInjectionAnnotation, inject)
if inject != "" {
exists = true
switch inject {
case "enabled", "yes", "true":
enabled = true
case "disabled", "no", "false":
enabled = false
default:
err = errors.Errorf("Invalid annotation value for key %q: %s", constants.SidecarInjectionAnnotation, inject)
}
}
return
}
func patchAdmissionResponse(resp *v1beta1.AdmissionResponse, patchBytes []byte) {
resp.Patch = patchBytes
pt := v1beta1.PatchTypeJSONPatch
resp.PatchType = &pt
}
// getPartialMutatingWebhookConfiguration returns only the portion of the MutatingWebhookConfiguration that needs to be updated.
func getPartialMutatingWebhookConfiguration(cert certificate.Certificater, webhookConfigName string) admissionv1beta1.MutatingWebhookConfiguration {
return admissionv1beta1.MutatingWebhookConfiguration{
ObjectMeta: metav1.ObjectMeta{
Name: webhookConfigName,
},
Webhooks: []admissionv1beta1.MutatingWebhook{
{
Name: MutatingWebhookName,
ClientConfig: admissionv1beta1.WebhookClientConfig{
CABundle: cert.GetCertificateChain(),
},
},
},
}
}
// updateMutatingWebhookCABundle updates the existing MutatingWebhookConfiguration with the CA this OSM instance runs with.
// It is necessary to perform this patch because the original MutatingWebhookConfig YAML does not contain the root certificate.
func updateMutatingWebhookCABundle(cert certificate.Certificater, webhookName string, clientSet kubernetes.Interface) error {
mwc := clientSet.AdmissionregistrationV1beta1().MutatingWebhookConfigurations()
if err := webhookExists(mwc, webhookName); err != nil {
log.Error().Err(err).Msgf("Error getting MutatingWebhookConfiguration %s; Will not update CA Bundle for webhook", webhookName)
}
patchJSON, err := json.Marshal(getPartialMutatingWebhookConfiguration(cert, webhookName))
if err != nil {
return err
}
if _, err = mwc.Patch(context.Background(), webhookName, types.StrategicMergePatchType, patchJSON, metav1.PatchOptions{}); err != nil {
log.Error().Err(err).Msgf("Error updating CA Bundle for MutatingWebhookConfiguration %s", webhookName)
return err
}
log.Info().Msgf("Finished updating CA Bundle for MutatingWebhookConfiguration %s", webhookName)
return nil
}
func webhookExists(mwc admissionRegistrationTypes.MutatingWebhookConfigurationInterface, webhookName string) error {
_, err := mwc.Get(context.Background(), webhookName, metav1.GetOptions{})
return err
}