adelorenzo/kilo
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

cmd/kg/*: sub command peer validation webhook

Browse Source 
This commit adds a sub command `webhook` to Kilo.
It will start a https web server that answeres request from a Kubernetes
API server to validate updates and creations of Kilo peers.

It also updates the "Peer Validation" docs to enable users to
install the web hook server and generate the self signed certificates in
the cluster by only applying a manifest.

Signed-off-by: leonnicolas <leonloechner@gmx.de>

Apply suggestions from code review

Co-authored-by: Lucas Servén Marín <lserven@gmail.com>
This commit is contained in:
leonnicolas
2021-08-23 17:51:46 +02:00
parent 2b4487ba9a
commit 086b2e1ddd
19 changed files with 2989 additions and 90 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
cmd/kg/handlers.go
View File

@@ -1,4 +1,4 @@
// Copyright 2019 the Kilo authors
// Copyright 2021 the Kilo authors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.

126
cmd/kg/main.go
View File

@@ -1,4 +1,4 @@
// Copyright 2019 the Kilo authors
// Copyright 2021 the Kilo authors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
@@ -85,7 +85,10 @@ var cmd = &cobra.Command{
It runs on every node of a cluster,
setting up the public and private keys for the VPN
as well as the necessary rules to route packets between locations.`,
RunE: runRoot,
PreRunE: preRun,
RunE: runRoot,
SilenceUsage: true,
SilenceErrors: true,
}
var (
@@ -102,61 +105,47 @@ var (
iface string
listen string
local bool
logLevel string
master string
mtu uint
topologyLabel string
port uint
subnet string
resyncPeriod time.Duration
printVersion bool
printVersion bool
logLevel string
logger log.Logger
registry *prometheus.Registry
)
func init() {
cmd.PersistentFlags().StringVar(&backend, "backend", k8s.Backend, fmt.Sprintf("The backend for the mesh. Possible values: %s", availableBackends))
cmd.PersistentFlags().BoolVar(&cleanUpIface, "clean-up-interface", false, "Should Kilo delete its interface when it shuts down?")
cmd.PersistentFlags().BoolVar(&createIface, "create-interface", true, "Should kilo create an interface on startup?")
cmd.PersistentFlags().BoolVar(&cni, "cni", true, "Should Kilo manage the node's CNI configuration?")
cmd.PersistentFlags().StringVar(&cniPath, "cni-path", mesh.DefaultCNIPath, "Path to CNI config.")
cmd.PersistentFlags().StringVar(&compatibility, "compatibility", "", fmt.Sprintf("Should Kilo run in compatibility mode? Possible values: %s", availableCompatibilities))
cmd.PersistentFlags().StringVar(&encapsulate, "encapsulate", string(encapsulation.Always), fmt.Sprintf("When should Kilo encapsulate packets within a location? Possible values: %s", availableEncapsulations))
cmd.PersistentFlags().StringVar(&granularity, "mesh-granularity", string(mesh.LogicalGranularity), fmt.Sprintf("The granularity of the network mesh to create. Possible values: %s", availableGranularities))
cmd.PersistentFlags().StringVar(&kubeconfig, "kubeconfig", "", "Path to kubeconfig.")
cmd.PersistentFlags().StringVar(&hostname, "hostname", "", "Hostname of the node on which this process is running.")
cmd.PersistentFlags().StringVar(&iface, "interface", mesh.DefaultKiloInterface, "Name of the Kilo interface to use; if it does not exist, it will be created.")
cmd.PersistentFlags().StringVar(&listen, "listen", ":1107", "The address at which to listen for health and metrics.")
cmd.PersistentFlags().BoolVar(&local, "local", true, "Should Kilo manage routes within a location?")
cmd.PersistentFlags().StringVar(&logLevel, "log-level", logLevelInfo, fmt.Sprintf("Log level to use. Possible values: %s", availableLogLevels))
cmd.PersistentFlags().StringVar(&master, "master", "", "The address of the Kubernetes API server (overrides any value in kubeconfig).")
cmd.PersistentFlags().UintVar(&mtu, "mtu", wireguard.DefaultMTU, "The MTU of the WireGuard interface created by Kilo.")
cmd.PersistentFlags().StringVar(&topologyLabel, "topology-label", k8s.RegionLabelKey, "Kubernetes node label used to group nodes into logical locations.")
cmd.PersistentFlags().UintVar(&port, "port", mesh.DefaultKiloPort, "The port over which WireGuard peers should communicate.")
cmd.PersistentFlags().StringVar(&subnet, "subnet", mesh.DefaultKiloSubnet.String(), "CIDR from which to allocate addresses for WireGuard interfaces.")
cmd.PersistentFlags().DurationVar(&resyncPeriod, "resync-period", 30*time.Second, "How often should the Kilo controllers reconcile?")
cmd.Flags().StringVar(&backend, "backend", k8s.Backend, fmt.Sprintf("The backend for the mesh. Possible values: %s", availableBackends))
cmd.Flags().BoolVar(&cleanUpIface, "clean-up-interface", false, "Should Kilo delete its interface when it shuts down?")
cmd.Flags().BoolVar(&createIface, "create-interface", true, "Should kilo create an interface on startup?")
cmd.Flags().BoolVar(&cni, "cni", true, "Should Kilo manage the node's CNI configuration?")
cmd.Flags().StringVar(&cniPath, "cni-path", mesh.DefaultCNIPath, "Path to CNI config.")
cmd.Flags().StringVar(&compatibility, "compatibility", "", fmt.Sprintf("Should Kilo run in compatibility mode? Possible values: %s", availableCompatibilities))
cmd.Flags().StringVar(&encapsulate, "encapsulate", string(encapsulation.Always), fmt.Sprintf("When should Kilo encapsulate packets within a location? Possible values: %s", availableEncapsulations))
cmd.Flags().StringVar(&granularity, "mesh-granularity", string(mesh.LogicalGranularity), fmt.Sprintf("The granularity of the network mesh to create. Possible values: %s", availableGranularities))
cmd.Flags().StringVar(&kubeconfig, "kubeconfig", "", "Path to kubeconfig.")
cmd.Flags().StringVar(&hostname, "hostname", "", "Hostname of the node on which this process is running.")
cmd.Flags().StringVar(&iface, "interface", mesh.DefaultKiloInterface, "Name of the Kilo interface to use; if it does not exist, it will be created.")
cmd.Flags().StringVar(&listen, "listen", ":1107", "The address at which to listen for health and metrics.")
cmd.Flags().BoolVar(&local, "local", true, "Should Kilo manage routes within a location?")
cmd.Flags().StringVar(&master, "master", "", "The address of the Kubernetes API server (overrides any value in kubeconfig).")
cmd.Flags().UintVar(&mtu, "mtu", wireguard.DefaultMTU, "The MTU of the WireGuard interface created by Kilo.")
cmd.Flags().StringVar(&topologyLabel, "topology-label", k8s.RegionLabelKey, "Kubernetes node label used to group nodes into logical locations.")
cmd.Flags().UintVar(&port, "port", mesh.DefaultKiloPort, "The port over which WireGuard peers should communicate.")
cmd.Flags().StringVar(&subnet, "subnet", mesh.DefaultKiloSubnet.String(), "CIDR from which to allocate addresses for WireGuard interfaces.")
cmd.Flags().DurationVar(&resyncPeriod, "resync-period", 30*time.Second, "How often should the Kilo controllers reconcile?")
cmd.PersistentFlags().BoolVar(&printVersion, "version", false, "Print version and exit")
cmd.PersistentFlags().StringVar(&logLevel, "log-level", logLevelInfo, fmt.Sprintf("Log level to use. Possible values: %s", availableLogLevels))
}
// Main is the principal function for the binary, wrapped only by `main` for convenience.
func runRoot(_ *cobra.Command, _ []string) error {
if printVersion {
fmt.Println(version.Version)
return nil
}
_, s, err := net.ParseCIDR(subnet)
if err != nil {
return fmt.Errorf("failed to parse %q as CIDR: %v", subnet, err)
}
if hostname == "" {
var err error
hostname, err = os.Hostname()
if hostname == "" || err != nil {
return errors.New("failed to determine hostname")
}
}
logger := log.NewJSONLogger(log.NewSyncWriter(os.Stdout))
func preRun(_ *cobra.Command, _ []string) error {
logger = log.NewJSONLogger(log.NewSyncWriter(os.Stdout))
switch logLevel {
case logLevelAll:
logger = level.NewFilter(logger, level.AllowAll())
@@ -176,6 +165,35 @@ func runRoot(_ *cobra.Command, _ []string) error {
logger = log.With(logger, "ts", log.DefaultTimestampUTC)
logger = log.With(logger, "caller", log.DefaultCaller)
registry = prometheus.NewRegistry()
registry.MustRegister(
prometheus.NewGoCollector(),
prometheus.NewProcessCollector(prometheus.ProcessCollectorOpts{}),
)
return nil
}
// runRoot is the principal function for the binary.
func runRoot(_ *cobra.Command, _ []string) error {
if printVersion {
fmt.Println(version.Version)
return nil
}
_, s, err := net.ParseCIDR(subnet)
if err != nil {
return fmt.Errorf("failed to parse %q as CIDR: %v", subnet, err)
}
if hostname == "" {
var err error
hostname, err = os.Hostname()
if hostname == "" || err != nil {
return errors.New("failed to determine hostname")
}
}
e := encapsulation.Strategy(encapsulate)
switch e {
case encapsulation.Never:
@@ -221,20 +239,15 @@ func runRoot(_ *cobra.Command, _ []string) error {
return fmt.Errorf("failed to create Kilo mesh: %v", err)
}
r := prometheus.NewRegistry()
r.MustRegister(
prometheus.NewGoCollector(),
prometheus.NewProcessCollector(prometheus.ProcessCollectorOpts{}),
)
m.RegisterMetrics(r)
m.RegisterMetrics(registry)
var g run.Group
{
// Run the HTTP server.
mux := http.NewServeMux()
mux.HandleFunc("/health", healthHandler)
mux.Handle("/graph", &graphHandler{m, gr, hostname, s})
mux.Handle("/metrics", promhttp.HandlerFor(r, promhttp.HandlerOpts{}))
mux.Handle("/graph", &graphHandler{m, gr, &hostname, s})
mux.Handle("/metrics", promhttp.HandlerFor(registry, promhttp.HandlerOpts{}))
l, err := net.Listen("tcp", listen)
if err != nil {
return fmt.Errorf("failed to listen on %s: %v", listen, err)
@@ -286,7 +299,14 @@ func runRoot(_ *cobra.Command, _ []string) error {
return g.Run()
}
var versionCmd = &cobra.Command{
Use: "version",
Short: "Print the version and exit.",
Run: func(_ *cobra.Command, _ []string) { fmt.Println(version.Version) },
}
func main() {
cmd.AddCommand(webhookCmd, versionCmd)
if err := cmd.Execute(); err != nil {
fmt.Fprintf(os.Stderr, "%v\n", err)
os.Exit(1)

273
cmd/kg/webhook.go Normal file
View File

@@ -0,0 +1,273 @@
// Copyright 2021 the Kilo authors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package main
import (
"context"
"encoding/json"
"errors"
"fmt"
"io/ioutil"
"net/http"
"os"
"syscall"
"time"
"github.com/go-kit/kit/log/level"
"github.com/oklog/run"
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/client_golang/prometheus/promhttp"
"github.com/spf13/cobra"
v1 "k8s.io/api/admission/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/runtime/serializer"
kilo "github.com/squat/kilo/pkg/k8s/apis/kilo/v1alpha1"
"github.com/squat/kilo/pkg/version"
)
var webhookCmd = &cobra.Command{
Use: "webhook",
PreRunE: func(c *cobra.Command, a []string) error {
if c.HasParent() {
return c.Parent().PreRunE(c, a)
}
return nil
},
Short: "webhook starts a HTTPS server to validate updates and creations of Kilo peers.",
RunE: webhook,
}
var (
certPath string
keyPath string
metricsAddr string
listenAddr string
)
func init() {
webhookCmd.Flags().StringVar(&certPath, "cert-file", "", "The path to a certificate file")
webhookCmd.Flags().StringVar(&keyPath, "key-file", "", "The path to a key file")
webhookCmd.Flags().StringVar(&metricsAddr, "listen-metrics", ":1107", "The metrics server will be listening to that address")
webhookCmd.Flags().StringVar(&listenAddr, "listen", ":8443", "The webhook server will be listening to that address")
}
var deserializer = serializer.NewCodecFactory(runtime.NewScheme()).UniversalDeserializer()
var (
validationCounter = prometheus.NewCounterVec(
prometheus.CounterOpts{
Name: "admission_requests_total",
Help: "The number of received admission reviews requests",
},
[]string{"operation", "response"},
)
requestCounter = prometheus.NewCounterVec(
prometheus.CounterOpts{
Name: "http_requests_total",
Help: "The number of received http requests",
},
[]string{"handler", "method"},
)
errorCounter = prometheus.NewCounter(
prometheus.CounterOpts{
Name: "errors_total",
Help: "The total number of errors",
},
)
)
func validationHandler(w http.ResponseWriter, r *http.Request) {
level.Debug(logger).Log("msg", "handling request", "source", r.RemoteAddr)
body, err := ioutil.ReadAll(r.Body)
if err != nil {
errorCounter.Inc()
level.Error(logger).Log("err", "failed to parse body from incoming request", "source", r.RemoteAddr)
http.Error(w, err.Error(), http.StatusBadRequest)
return
}
var admissionReview v1.AdmissionReview
contentType := r.Header.Get("Content-Type")
if contentType != "application/json" {
errorCounter.Inc()
msg := fmt.Sprintf("received Content-Type=%s, expected application/json", contentType)
level.Error(logger).Log("err", msg)
http.Error(w, msg, http.StatusBadRequest)
return
}
response := v1.AdmissionReview{}
_, gvk, err := deserializer.Decode(body, nil, &admissionReview)
if err != nil {
errorCounter.Inc()
msg := fmt.Sprintf("Request could not be decoded: %v", err)
level.Error(logger).Log("err", msg)
http.Error(w, msg, http.StatusBadRequest)
return
}
if *gvk != v1.SchemeGroupVersion.WithKind("AdmissionReview") {
errorCounter.Inc()
msg := "only API v1 is supported"
level.Error(logger).Log("err", msg)
http.Error(w, msg, http.StatusBadRequest)
return
}
response.SetGroupVersionKind(*gvk)
response.Response = &v1.AdmissionResponse{
UID: admissionReview.Request.UID,
}
rawExtension := admissionReview.Request.Object
var peer kilo.Peer
if err := json.Unmarshal(rawExtension.Raw, &peer); err != nil {
errorCounter.Inc()
msg := fmt.Sprintf("could not unmarshal extension to peer spec: %v:", err)
level.Error(logger).Log("err", msg)
http.Error(w, msg, http.StatusBadRequest)
return
}
if err := peer.Validate(); err == nil {
level.Debug(logger).Log("msg", "got valid peer spec", "spec", peer.Spec, "name", peer.ObjectMeta.Name)
validationCounter.With(prometheus.Labels{"operation": string(admissionReview.Request.Operation), "response": "allowed"}).Inc()
response.Response.Allowed = true
} else {
level.Debug(logger).Log("msg", "got invalid peer spec", "spec", peer.Spec, "name", peer.ObjectMeta.Name)
validationCounter.With(prometheus.Labels{"operation": string(admissionReview.Request.Operation), "response": "denied"}).Inc()
response.Response.Result = &metav1.Status{
Message: err.Error(),
}
}
res, err := json.Marshal(response)
if err != nil {
errorCounter.Inc()
msg := fmt.Sprintf("failed to marshal response: %v", err)
level.Error(logger).Log("err", msg)
http.Error(w, msg, http.StatusInternalServerError)
return
}
w.Header().Set("Content-Type", "application/json")
if _, err := w.Write(res); err != nil {
level.Error(logger).Log("err", err, "msg", "failed to write response")
}
}
func metricsMiddleWare(path string, next func(http.ResponseWriter, *http.Request)) func(http.ResponseWriter, *http.Request) {
return func(w http.ResponseWriter, r *http.Request) {
requestCounter.With(prometheus.Labels{"method": r.Method, "handler": path}).Inc()
next(w, r)
}
}
func webhook(_ *cobra.Command, _ []string) error {
if printVersion {
fmt.Println(version.Version)
os.Exit(0)
}
registry.MustRegister(
errorCounter,
validationCounter,
requestCounter,
)
ctx, cancel := context.WithCancel(context.Background())
defer func() {
cancel()
}()
var g run.Group
g.Add(run.SignalHandler(ctx, syscall.SIGINT, syscall.SIGTERM))
{
mm := http.NewServeMux()
mm.Handle("/metrics", promhttp.HandlerFor(registry, promhttp.HandlerOpts{}))
msrv := &http.Server{
Addr: metricsAddr,
Handler: mm,
}
g.Add(
func() error {
level.Info(logger).Log("msg", "starting metrics server", "address", msrv.Addr)
err := msrv.ListenAndServe()
level.Info(logger).Log("msg", "metrics server exited", "err", err)
return err
},
func(err error) {
var serr run.SignalError
if ok := errors.As(err, &serr); ok {
level.Info(logger).Log("msg", "received signal", "signal", serr.Signal.String(), "err", err.Error())
} else {
level.Error(logger).Log("msg", "received error", "err", err.Error())
}
level.Info(logger).Log("msg", "shutting down metrics server gracefully")
ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
defer func() {
cancel()
}()
if err := msrv.Shutdown(ctx); err != nil {
level.Error(logger).Log("msg", "failed to shut down metrics server gracefully", "err", err.Error())
msrv.Close()
}
},
)
}
{
mux := http.NewServeMux()
mux.HandleFunc("/validate", metricsMiddleWare("/validate", validationHandler))
srv := &http.Server{
Addr: listenAddr,
Handler: mux,
}
g.Add(
func() error {
level.Info(logger).Log("msg", "starting webhook server", "address", srv.Addr)
err := srv.ListenAndServeTLS(certPath, keyPath)
level.Info(logger).Log("msg", "webhook server exited", "err", err)
return err
},
func(err error) {
var serr run.SignalError
if ok := errors.As(err, &serr); ok {
level.Info(logger).Log("msg", "received signal", "signal", serr.Signal.String(), "err", err.Error())
} else {
level.Error(logger).Log("msg", "received error", "err", err.Error())
}
level.Info(logger).Log("msg", "shutting down webhook server gracefully")
ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
defer func() {
cancel()
}()
if err := srv.Shutdown(ctx); err != nil {
level.Error(logger).Log("msg", "failed to shut down webhook server gracefully", "err", err.Error())
srv.Close()
}
},
)
}
err := g.Run()
var serr run.SignalError
if ok := errors.As(err, &serr); ok {
return nil
}
return err
}