From 7405b51e534497fe96f13fcda4c0bbb2cfaddfe0 Mon Sep 17 00:00:00 2001 From: Levalicious Date: Tue, 13 Oct 2026 23:51:44 -0400 Subject: [PATCH] lb: serve Gateways (Gateway API v1) from the node holding their address Gateways of a GatewayClass whose controllerName is reweave.levk.dev/gateway are served like LoadBalancer Services: a Gateway names one IPv4 address in spec.addresses, and the node holding that address serves all of it and writes its status. - TCP listeners forward the oldest TCPRoute attached to them in the kernel, through the same DNAT and reply steering as LoadBalancer Services, so the backend sees the client's own address. - HTTP and HTTPS listeners are served by the agent itself on the holder: TLS terminated with the listener's certificate (chosen by SNI, swapped in place when its Secret changes, HTTP/2 offered), requests routed by the attached HTTPRoutes in the order the spec prescribes (hostname, exact path, longest prefix, method, headers, query, age, name, rule), and passed to a ready endpoint of the backend with X-Forwarded-For built from this hop only. RequestRedirect, URLRewrite and both header modifiers are implemented; other filters and regular expressions are refused (Accepted False, UnsupportedValue). Bodies are streamed, never limited. Unresolved backends answer 500, backends without ready endpoints 503, requests nothing matches 404. Those ports are let in ahead of the host firewall (REWEAVE-LB-IN). Listeners are checked for protocol, route kinds, certificates (Secrets in the Gateway's namespace or granted by a ReferenceGrant, and only in the namespaces the agent may read: --gateway-namespaces, default gateway-system), conflicts on a port, and ports LoadBalancer Services already forward. Routes attach by parentRef (sectionName, port), allowedRoutes (namespaces Same, All or Selector; kinds) and hostname intersection, with the spec's conditions and reasons. One writer per status: the holder writes its Gateway's (the first node by name when nobody holds the address, so failures are reported once), the first node by name writes GatewayClass and route statuses, keeping other controllers' route parents. Writes are conditional on the resourceVersion they were computed from, and a condition keeps its transition time while its status stands, so an unchanged status is never rewritten. The types are plain structs for the subset implemented (the upstream module needs a newer client-go than the fork's); the CRDs are installed apart from this manifest, which grants the agent read access to the Gateway API kinds, patch on their status, and Secrets in gateway-system only. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01UHXzGrsxfkGk8LmwrBbHNZ --- lb/controller.go | 169 +++ lb/datapath.go | 4 + lb/gateway.go | 1009 +++++++++++++++++ lb/gateway_api.go | 415 +++++++ lb/gateway_proxy.go | 542 +++++++++ lb/gateway_proxy_test.go | 314 +++++ lb/gateway_test.go | 576 ++++++++++ lb/model.go | 10 + prog/weave-kube/weave-daemonset-k8s-1.11.yaml | 58 + prog/weave-npc/main.go | 29 +- 10 files changed, 3117 insertions(+), 9 deletions(-) create mode 100644 lb/gateway.go create mode 100644 lb/gateway_api.go create mode 100644 lb/gateway_proxy.go create mode 100644 lb/gateway_proxy_test.go create mode 100644 lb/gateway_test.go diff --git a/lb/controller.go b/lb/controller.go index bfcdf05f..2db86057 100644 --- a/lb/controller.go +++ b/lb/controller.go @@ -53,6 +53,15 @@ type Controller struct { // lastEgress is what was last applied, reused while the EgressService caches are not synced (never wipe // members' rules on a partial view) lastEgress EgressDesired + + // Gateways (nil gateway client: not served). Secrets are read only in gwNamespaces. + gateway rest.Interface + gwNamespaces []string + proxy *Proxy + classes, gateways, httpRoutes, tcpRoutes, grants, namespaces cache.Store + secrets []cache.Store + gatewaySynced []cache.InformerSynced // empty while the Gateway API is absent + unbound map[string]string // listeners the proxy could not bind, last pass } // Resync is how often the node re-reads its interfaces and re-applies everything even without API events. @@ -93,6 +102,86 @@ func (c *Controller) egressAPI() (bool, error) { return false, err } +// WithGateway has the controller serve Gateways of the classes whose controllerName is GatewayControllerName, +// reading their certificates from Secrets in these namespaces only. +func (c *Controller) WithGateway(client rest.Interface, namespaces []string) *Controller { + c.gateway, c.gwNamespaces, c.proxy, c.unbound = client, namespaces, NewProxy(), map[string]string{} + return c +} + +// gatewayAPI is which Gateway API resources are served (nil: none, the CRDs are not installed). +func (c *Controller) gatewayAPI() (map[string]bool, error) { + l, err := c.client.Discovery().ServerResourcesForGroupVersion(GatewayGroupVersion.String()) + if apierrors.IsNotFound(err) { + return nil, nil + } + if err != nil { + return nil, err + } + served := map[string]bool{} + for _, r := range l.APIResources { + served[r.Name] = true + } + return served, nil +} + +func (c *Controller) startGateway(ctx context.Context, served map[string]bool) { + run := func(store *cache.Store, resource string, obj runtime.Object) { + if !served[resource] { + return // an older bundle without this kind: nothing of it to serve + } + var ctl cache.Controller + *store, ctl = newInformer(c, cache.NewListWatchFromClient(c.gateway, resource, "", fields.Everything()), obj) + go ctl.Run(ctx.Done()) + c.gatewaySynced = append(c.gatewaySynced, ctl.HasSynced) + } + run(&c.classes, GatewayClasses, &GatewayClass{}) + run(&c.gateways, Gateways, &Gateway{}) + run(&c.httpRoutes, HTTPRoutes, &HTTPRoute{}) + run(&c.tcpRoutes, TCPRoutes, &TCPRoute{}) + run(&c.grants, ReferenceGrants, &ReferenceGrant{}) + var ctl cache.Controller + c.namespaces, ctl = newInformer(c, cache.NewListWatchFromClient(c.client.CoreV1().RESTClient(), "namespaces", "", fields.Everything()), &coreapi.Namespace{}) + go ctl.Run(ctx.Done()) + c.gatewaySynced = append(c.gatewaySynced, ctl.HasSynced) + for _, ns := range c.gwNamespaces { + var store cache.Store + store, ctl = newInformer(c, cache.NewListWatchFromClient(c.client.CoreV1().RESTClient(), "secrets", ns, fields.Everything()), &coreapi.Secret{}) + go ctl.Run(ctx.Done()) + c.secrets = append(c.secrets, store) + c.gatewaySynced = append(c.gatewaySynced, ctl.HasSynced) + } + common.Log.Infof("lb: serving Gateways of controller %s (certificates from %v)", GatewayControllerName, c.gwNamespaces) +} + +// gatewayReady: the Gateway view can be trusted (the API is served and every informer has synced). +func (c *Controller) gatewayReady() bool { + if c.gateway == nil || len(c.gatewaySynced) == 0 { + return false + } + for _, s := range c.gatewaySynced { + if !s() { + return false + } + } + return true +} + +// tryStartGateway starts the Gateway informers once the API is served. +func (c *Controller) tryStartGateway(ctx context.Context) { + if c.gateway == nil || len(c.gatewaySynced) != 0 { + return + } + served, err := c.gatewayAPI() + if err != nil { + common.Log.Warnf("lb: cannot tell whether the Gateway API is served: %v", err) + return + } + if served[Gateways] { + c.startGateway(ctx, served) + } +} + func (c *Controller) startEgress(ctx context.Context) { var ctl cache.Controller c.egresses, ctl = newInformer(c, cache.NewListWatchFromClient(c.egress, EgressServices, "", fields.Everything()), &EgressService{}) @@ -130,6 +219,10 @@ func (c *Controller) Run(ctx context.Context) { case <-time.After(5 * time.Second): } } + c.tryStartGateway(ctx) + if c.proxy != nil { + defer c.proxy.Close() + } go c.serveAdmission(ctx) if !cache.WaitForCacheSync(ctx.Done(), c.synced...) { return @@ -143,6 +236,9 @@ func (c *Controller) Run(ctx context.Context) { case <-ctx.Done(): return case <-tick.C: + c.mu.Lock() + c.tryStartGateway(ctx) + c.mu.Unlock() if c.egressSynced == nil { // the API may have been installed since if present, err := c.egressAPI(); err == nil && present { @@ -208,6 +304,14 @@ func (c *Controller) sync(ctx context.Context) error { slices = append(slices, o.(*discovery.EndpointSlice)) } d := Compute(c.self, nodes, svcs, slices) + var g *GatewayDesired + if c.gatewayReady() { + gd := ComputeGateways(c.self, nodes, c.gatewayInput(svcs, slices)) + d = MergeGateways(d, gd, c.self, nodes) + g = &gd + } else if c.gateway != nil && len(c.gatewaySynced) != 0 { + problems = append(problems, "Gateway view not synced yet: Gateways not served") + } // rewrite once after start what an earlier run may have written without ipMode for _, f := range d.Forwards { if s := c.service(f.Namespace, f.Name); s != nil && c.written[s.UID] != f.Address.String() && !hasStatusFor(d, f) { @@ -236,6 +340,18 @@ func (c *Controller) sync(ctx context.Context) error { return fmt.Errorf("applying datapath: %v", err) } c.lastEgress = e + if g != nil { + // the ports are let in by now; bind what this node serves, close what it no longer does + unbound := map[string]string{} + for k, err := range c.proxy.Apply(g.Serve) { + unbound[k] = err.Error() + } + if !sameJSON(unbound, c.unbound) { + c.unbound = unbound + c.poke() // the statuses say so on the next pass + } + c.writeGatewayStatuses(ctx, *g) + } for _, st := range d.Status { if err := c.writeStatus(ctx, st); err != nil { return err @@ -458,3 +574,56 @@ func (c *Controller) writeStatus(ctx context.Context, st Status) error { } return nil } + +// gatewayInput is the Gateway view the policy reads. +func (c *Controller) gatewayInput(svcs []*coreapi.Service, slices []*discovery.EndpointSlice) GatewayInput { + in := GatewayInput{Services: svcs, Slices: slices, SecretNamespaces: map[string]bool{}, Unbound: c.unbound, Now: metav1.Now()} + list := func(s cache.Store, each func(interface{})) { + if s != nil { + for _, o := range s.List() { + each(o) + } + } + } + list(c.classes, func(o interface{}) { in.Classes = append(in.Classes, o.(*GatewayClass)) }) + list(c.gateways, func(o interface{}) { in.Gateways = append(in.Gateways, o.(*Gateway)) }) + list(c.httpRoutes, func(o interface{}) { in.HTTPRoutes = append(in.HTTPRoutes, o.(*HTTPRoute)) }) + list(c.tcpRoutes, func(o interface{}) { in.TCPRoutes = append(in.TCPRoutes, o.(*TCPRoute)) }) + list(c.grants, func(o interface{}) { in.Grants = append(in.Grants, o.(*ReferenceGrant)) }) + list(c.namespaces, func(o interface{}) { in.Namespaces = append(in.Namespaces, o.(*coreapi.Namespace)) }) + for _, s := range c.secrets { + list(s, func(o interface{}) { in.Secrets = append(in.Secrets, o.(*coreapi.Secret)) }) + } + for _, ns := range c.gwNamespaces { + in.SecretNamespaces[ns] = true + } + return in +} + +// writeGatewayStatuses writes what the policy says this node writes. Each write is conditional on the version it +// was computed from: one built from a stale view fails, and the next pass recomputes it. +func (c *Controller) writeGatewayStatuses(ctx context.Context, g GatewayDesired) { + for _, w := range g.Classes { + if err := patchStatus(ctx, c.gateway, GatewayClasses, "", w.Name, w.ResourceVersion, w.Status); err != nil { + common.Log.Warnf("lb: status of GatewayClass %s: %v", w.Name, err) + continue + } + common.Log.Infof("lb: GatewayClass %s accepted", w.Name) + } + for _, w := range g.Gateways { + if err := patchStatus(ctx, c.gateway, Gateways, w.Namespace, w.Name, w.ResourceVersion, w.Status); err != nil { + common.Log.Warnf("lb: status of Gateway %s/%s: %v", w.Namespace, w.Name, err) + continue + } + var conds []string + for _, cd := range w.Status.Conditions { + conds = append(conds, fmt.Sprintf("%s=%s (%s)", cd.Type, cd.Status, cd.Reason)) + } + common.Log.Infof("lb: Gateway %s/%s: %v", w.Namespace, w.Name, conds) + } + for _, w := range g.Routes { + if err := patchStatus(ctx, c.gateway, w.Resource, w.Namespace, w.Name, w.ResourceVersion, w.Status); err != nil { + common.Log.Warnf("lb: status of %s %s/%s: %v", w.Resource, w.Namespace, w.Name, err) + } + } +} diff --git a/lb/datapath.go b/lb/datapath.go index 91e19541..5973a36e 100644 --- a/lb/datapath.go +++ b/lb/datapath.go @@ -117,6 +117,10 @@ func Render(d Desired, e EgressDesired, self string, podNet *net.IPNet, stale [] for _, c := range []string{FwdChain, InChain} { fmt.Fprintf(&filter, ":%s - [0:0]\n", c) } + for _, a := range d.Accept { + // served by the agent itself (a Gateway's HTTP or HTTPS listener): in, whatever the host firewall says + fmt.Fprintf(&filter, "-A %s -d %s/32 -p tcp -m tcp --dport %d -m comment --comment %q -j ACCEPT\n", InChain, a.Address, a.Port, a.Comment) + } renderEgress(e, self, podNet, &body, &mangle, &filter) nat.Write(body.Bytes()) diff --git a/lb/gateway.go b/lb/gateway.go new file mode 100644 index 00000000..b047af5b --- /dev/null +++ b/lb/gateway.go @@ -0,0 +1,1009 @@ +package lb + +import ( + "crypto/tls" + "encoding/json" + "fmt" + "net" + "sort" + "strings" + "time" + + coreapi "k8s.io/api/core/v1" + discovery "k8s.io/api/discovery/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/labels" +) + +// Gateways of a class this agent implements are served like LoadBalancer Services: a Gateway names one address a +// node holds (spec.addresses), and that node, the holder, serves all of it and writes its status. TCP listeners +// are forwarded in the kernel (the DNAT of LoadBalancer Services: the backend sees the client's own address); +// HTTP and HTTPS listeners are served by the agent itself on the holder (TLS terminated with the listener's +// certificate, requests routed by their HTTPRoutes). ComputeGateways is the whole policy as a pure function of +// what the API shows. + +const ( + ProtocolHTTP = "HTTP" + ProtocolHTTPS = "HTTPS" + ProtocolTCP = "TCP" + + kindGateway = "Gateway" + kindHTTPRoute = "HTTPRoute" + kindTCPRoute = "TCPRoute" + kindService = "Service" + kindSecret = "Secret" + + condAccepted = "Accepted" + condProgrammed = "Programmed" + condResolvedRefs = "ResolvedRefs" + condConflicted = "Conflicted" + + filterRequestHeaderModifier = "RequestHeaderModifier" + filterResponseHeaderModifier = "ResponseHeaderModifier" + filterRequestRedirect = "RequestRedirect" + filterURLRewrite = "URLRewrite" + + pathExact = "Exact" + pathPrefix = "PathPrefix" +) + +// GatewayInput is what the API shows that the gateway policy reads. +type GatewayInput struct { + Classes []*GatewayClass + Gateways []*Gateway + HTTPRoutes []*HTTPRoute + TCPRoutes []*TCPRoute + Grants []*ReferenceGrant + Namespaces []*coreapi.Namespace + Services []*coreapi.Service + Slices []*discovery.EndpointSlice + // Secrets the agent can read, and the namespaces they come from: a certificate elsewhere cannot be served + Secrets []*coreapi.Secret + SecretNamespaces map[string]bool + // Unbound are the listeners ("namespace/gateway/listener") this node could not bind on its last try, and why + Unbound map[string]string + Now metav1.Time +} + +// GatewayDesired is what one node does for Gateways, and the statuses it writes. +type GatewayDesired struct { + Forwards []Forward // this node's TCP listeners, forwarded like LoadBalancer Services + Holding []string // nodes forwarding some Gateway's TCP listener (other nodes steer replies back to them) + Serve []ServeListener // this node's HTTP and HTTPS listeners + Classes []ClassStatusWrite + Gateways []GatewayStatusWrite + Routes []RouteStatusWrite + Problems []string +} + +// ServeListener is an HTTP or HTTPS listener this node serves. +type ServeListener struct { + Gateway string // namespace/name + Name string + Address net.IP + Port int32 + HTTPS bool + Hostname string // "" takes any host + Cert *coreapi.Secret // HTTPS: the listener's certificate + Routes []ServeRoute +} + +// ServeRoute is an HTTPRoute attached to a listener. +type ServeRoute struct { + Key string // namespace/name + Created time.Time + // Hostnames the route answers on this listener (its own, narrowed to the listener's); none: the listener's + Hostnames []string + Rules []ServeRule +} + +// ServeRule is one rule of an attached route. +type ServeRule struct { + Matches []HTTPRouteMatch // defaulted: a match without a path is a PathPrefix match on "/" + Filters []HTTPRouteFilter + Backends []ServeBackend + // Broken: some backend could not be resolved; requests the rule takes are answered 500, as the spec says + Broken bool +} + +// ServeBackend is a Service port a rule sends requests to, with its ready endpoints. +type ServeBackend struct { + Key string // namespace/name:port + Weight int32 + Endpoints []Endpoint + Filters []HTTPRouteFilter +} + +type ClassStatusWrite struct { + Name, ResourceVersion string + Status GatewayClassStatus +} + +type GatewayStatusWrite struct { + Namespace, Name, ResourceVersion string + Status GatewayStatus +} + +type RouteStatusWrite struct { + Resource, Namespace, Name, ResourceVersion string + Status RouteStatus +} + +// listener is a listener as the policy judges it. +type listener struct { + spec Listener + kinds []string // route kinds it takes + usable bool // accepted, references resolved, port free: it can be served unless it conflicts + unbound bool // served, but this node could not bind it on its last try + conds []metav1.Condition + conflicted bool + cert *coreapi.Secret + attached int32 + tcpRoute *TCPRoute // the TCPRoute a TCP listener forwards (the oldest of those attached) + http []ServeRoute +} + +type gatewayView struct { + gw *Gateway + address net.IP + addrErr string + holder *Node + holderErr string + listeners []*listener +} + +// ComputeGateways decides what node self does for Gateways, given every node and what the API shows. taken are +// the address:protocol:port triples LoadBalancer Services already forward (a listener cannot share one). +func ComputeGateways(self string, nodes []Node, in GatewayInput) GatewayDesired { + var d GatewayDesired + sorted := append([]Node(nil), nodes...) + sort.Slice(sorted, func(i, j int) bool { return sorted[i].Name < sorted[j].Name }) + first := "" + if len(sorted) > 0 { + first = sorted[0].Name + } + holdersOf := func(ip net.IP) []Node { + var hs []Node + for _, n := range sorted { + if n.Holds(ip) { + hs = append(hs, n) + } + } + return hs + } + + ours := map[string]*GatewayClass{} + classes := append([]*GatewayClass(nil), in.Classes...) + sort.Slice(classes, func(i, j int) bool { return classes[i].Name < classes[j].Name }) + for _, c := range classes { + if c.Spec.ControllerName != GatewayControllerName { + continue + } + ours[c.Name] = c + want := GatewayClassStatus{Conditions: []metav1.Condition{ + condition(c.Status.Conditions, condAccepted, true, "Accepted", "served by the reweave agent on the node holding each Gateway's address", c.Generation, in.Now), + }} + if self == first && !sameJSON(want, c.Status) { + d.Classes = append(d.Classes, ClassStatusWrite{Name: c.Name, ResourceVersion: c.ResourceVersion, Status: want}) + } + } + + nsLabels := map[string]labels.Set{} + for _, n := range in.Namespaces { + nsLabels[n.Name] = labels.Set(n.Labels) + } + secrets := map[string]*coreapi.Secret{} + for _, s := range in.Secrets { + secrets[s.Namespace+"/"+s.Name] = s + } + services := map[string]*coreapi.Service{} + taken := map[string]string{} + for _, s := range in.Services { + services[s.Namespace+"/"+s.Name] = s + if !eligible(s) { + continue + } + if ip := net.ParseIP(s.Spec.LoadBalancerIP).To4(); ip != nil { + for _, p := range s.Spec.Ports { + proto := p.Protocol + if proto == "" { + proto = coreapi.ProtocolTCP + } + taken[fmt.Sprintf("%s:%s:%d", ip, proto, p.Port)] = key(s) + } + } + } + + // the Gateways of our classes, judged listener by listener + var views []*gatewayView + byKey := map[string]*gatewayView{} + gws := append([]*Gateway(nil), in.Gateways...) + sort.Slice(gws, func(i, j int) bool { return objKey(&gws[i].ObjectMeta) < objKey(&gws[j].ObjectMeta) }) + for _, gw := range gws { + if ours[gw.Spec.GatewayClassName] == nil { + continue + } + v := &gatewayView{gw: gw} + v.address, v.addrErr = gatewayAddress(gw) + if v.address != nil { + switch hs := holdersOf(v.address); len(hs) { + case 0: + v.holderErr = fmt.Sprintf("no node holds %s", v.address) + case 1: + v.holder = &hs[0] + default: + names := make([]string, len(hs)) + for i, h := range hs { + names[i] = h.Name + } + v.holderErr = fmt.Sprintf("%s is held by several nodes (%s); serving it from none", v.address, strings.Join(names, ", ")) + } + } + v.listeners = judgeListeners(gw, v.address, in, secrets, taken) + views = append(views, v) + byKey[objKey(&gw.ObjectMeta)] = v + if v.addrErr != "" { + d.Problems = append(d.Problems, fmt.Sprintf("gateway %s: %s", objKey(&gw.ObjectMeta), v.addrErr)) + } else if v.holderErr != "" { + d.Problems = append(d.Problems, fmt.Sprintf("gateway %s: %s", objKey(&gw.ObjectMeta), v.holderErr)) + } + } + + // routes, oldest first: that order decides which TCPRoute a TCP listener forwards, and HTTP precedence ties + type routeRef struct { + resource, kind string + meta *metav1.ObjectMeta + parents []ParentReference + current RouteStatus + http *HTTPRoute + tcp *TCPRoute + } + var routes []routeRef + for _, r := range in.HTTPRoutes { + routes = append(routes, routeRef{resource: HTTPRoutes, kind: kindHTTPRoute, meta: &r.ObjectMeta, parents: r.Spec.ParentRefs, current: r.Status, http: r}) + } + for _, r := range in.TCPRoutes { + routes = append(routes, routeRef{resource: TCPRoutes, kind: kindTCPRoute, meta: &r.ObjectMeta, parents: r.Spec.ParentRefs, current: r.Status, tcp: r}) + } + sort.Slice(routes, func(i, j int) bool { + a, b := routes[i], routes[j] + if !a.meta.CreationTimestamp.Equal(&b.meta.CreationTimestamp) { + return a.meta.CreationTimestamp.Before(&b.meta.CreationTimestamp) + } + if a.kind != b.kind { + return a.kind < b.kind + } + return objKey(a.meta) < objKey(b.meta) + }) + for _, r := range routes { + var ourParents []RouteParentStatus + for _, pr := range r.parents { + if group(pr.Group, GatewayGroup) != GatewayGroup || kind(pr.Kind, kindGateway) != kindGateway { + continue + } + v := byKey[ns(pr.Namespace, r.meta.Namespace)+"/"+pr.Name] + if v == nil { + continue // not one of ours (another implementation's, or none's) + } + var accepted metav1.Condition + var resolved metav1.Condition + if r.http != nil { + accepted, resolved = attachHTTP(r.http, pr, v, in, nsLabels, services) + } else { + accepted, resolved = attachTCP(r.tcp, pr, v, in, nsLabels, services) + } + ourParents = append(ourParents, RouteParentStatus{ + ParentRef: pr, + ControllerName: GatewayControllerName, + Conditions: []metav1.Condition{ + condition(currentParent(r.current, pr), condAccepted, accepted.Status == metav1.ConditionTrue, accepted.Reason, accepted.Message, r.meta.Generation, in.Now), + condition(currentParent(r.current, pr), condResolvedRefs, resolved.Status == metav1.ConditionTrue, resolved.Reason, resolved.Message, r.meta.Generation, in.Now), + }, + }) + } + // our entries are rewritten whole; other controllers' entries are kept as they are + var parents []RouteParentStatus + hadOurs := false + for _, p := range r.current.Parents { + if p.ControllerName == GatewayControllerName { + hadOurs = true + continue + } + parents = append(parents, p) + } + parents = append(parents, ourParents...) + if len(ourParents) == 0 && !hadOurs { + continue + } + want := RouteStatus{Parents: parents} + if want.Parents == nil { + want.Parents = []RouteParentStatus{} + } + if self == first && !sameJSON(want, r.current) { + d.Routes = append(d.Routes, RouteStatusWrite{Resource: r.resource, Namespace: r.meta.Namespace, Name: r.meta.Name, ResourceVersion: r.meta.ResourceVersion, Status: want}) + } + } + + // what this node serves, and every Gateway's status from its writer + holding := map[string]bool{} + for _, v := range views { + gw := v.gw + served := v.holder != nil + if served { + for _, l := range v.listeners { + if l.usable && !l.conflicted && l.spec.Protocol == ProtocolTCP && l.tcpRoute != nil { + holding[v.holder.Name] = true + } + } + } + if served && v.holder.Name == self { + for _, l := range v.listeners { + if !l.usable || l.conflicted { + continue + } + switch l.spec.Protocol { + case ProtocolTCP: + if l.tcpRoute == nil { + continue + } + d.Forwards = append(d.Forwards, Forward{ + Namespace: gw.Namespace, Name: gw.Name + "/" + l.spec.Name, Address: v.address, + Ports: []Port{{Protocol: coreapi.ProtocolTCP, Port: l.spec.Port, Endpoints: tcpEndpoints(l.tcpRoute, in, services)}}, + }) + case ProtocolHTTP, ProtocolHTTPS: + d.Serve = append(d.Serve, ServeListener{ + Gateway: objKey(&gw.ObjectMeta), Name: l.spec.Name, Address: v.address, Port: l.spec.Port, + HTTPS: l.spec.Protocol == ProtocolHTTPS, Hostname: str(l.spec.Hostname), Cert: l.cert, Routes: l.http, + }) + } + } + } + writer := first + if v.holder != nil { + writer = v.holder.Name + } + if writer != self { + continue + } + want := gatewayStatus(v, in.Now) + if !sameJSON(want, gw.Status) { + d.Gateways = append(d.Gateways, GatewayStatusWrite{Namespace: gw.Namespace, Name: gw.Name, ResourceVersion: gw.ResourceVersion, Status: want}) + } + } + for n := range holding { + d.Holding = append(d.Holding, n) + } + sort.Strings(d.Holding) + return d +} + +// gatewayAddress is the one IPv4 address a Gateway names, or why it names none this agent can serve. +func gatewayAddress(gw *Gateway) (net.IP, string) { + if len(gw.Spec.Addresses) != 1 { + return nil, fmt.Sprintf("spec.addresses must name exactly one IPv4 address a node holds (has %d)", len(gw.Spec.Addresses)) + } + a := gw.Spec.Addresses[0] + if t := str(a.Type); t != "" && t != "IPAddress" { + return nil, fmt.Sprintf("address type %s is not supported (IPAddress only)", t) + } + ip := net.ParseIP(a.Value).To4() + if ip == nil { + return nil, fmt.Sprintf("%q is not an IPv4 address", a.Value) + } + return ip, "" +} + +// judgeListeners checks each listener on its own (protocol, route kinds, certificate) and against the others on +// its port (conflicts), and against LoadBalancer Services forwarding the same address and port. +func judgeListeners(gw *Gateway, addr net.IP, in GatewayInput, secrets map[string]*coreapi.Secret, taken map[string]string) []*listener { + gen := gw.Generation + var out []*listener + for _, spec := range gw.Spec.Listeners { + l := &listener{spec: spec, usable: true} + cur := currentListener(gw, spec.Name) + resolved := condition(cur, condResolvedRefs, true, "ResolvedRefs", "", gen, in.Now) + accepted := condition(cur, condAccepted, true, "Accepted", "", gen, in.Now) + var supported []string + switch spec.Protocol { + case ProtocolHTTP, ProtocolHTTPS: + supported = []string{kindHTTPRoute} + case ProtocolTCP: + supported = []string{kindTCPRoute} + default: + l.usable = false + accepted = condition(cur, condAccepted, false, "UnsupportedProtocol", fmt.Sprintf("protocol %s is not supported (HTTP, HTTPS and TCP are)", spec.Protocol), gen, in.Now) + } + l.kinds = supported + if spec.AllowedRoutes != nil && len(spec.AllowedRoutes.Kinds) > 0 { + l.kinds = nil + var bad []string + for _, k := range spec.AllowedRoutes.Kinds { + if group(k.Group, GatewayGroup) == GatewayGroup && contains(supported, k.Kind) { + l.kinds = append(l.kinds, k.Kind) + } else { + bad = append(bad, k.Kind) + } + } + if len(bad) > 0 { + resolved = condition(cur, condResolvedRefs, false, "InvalidRouteKinds", fmt.Sprintf("route kinds %s cannot attach to a %s listener", strings.Join(bad, ", "), spec.Protocol), gen, in.Now) + } + } + if spec.Protocol == ProtocolHTTPS && l.usable { + cert, reason, msg := listenerCert(gw, spec, in, secrets) + if cert == nil { + // without its certificate the listener cannot be served (not programmed); an unsupported TLS mode + // is not accepted at all, an unresolved certificate is a reference problem + l.usable = false + if reason == "UnsupportedValue" { + accepted = condition(cur, condAccepted, false, reason, msg, gen, in.Now) + } else { + resolved = condition(cur, condResolvedRefs, false, reason, msg, gen, in.Now) + } + } + l.cert = cert + } + if addr != nil && l.usable { + if svc, ok := taken[fmt.Sprintf("%s:TCP:%d", addr, spec.Port)]; ok { + l.usable = false + accepted = condition(cur, condAccepted, false, "PortUnavailable", fmt.Sprintf("LoadBalancer Service %s already forwards %s:%d", svc, addr, spec.Port), gen, in.Now) + } else if why, ok := in.Unbound[objKey(&gw.ObjectMeta)+"/"+spec.Name]; ok { + // still served (each pass tries again), but not reported as working + l.unbound = true + accepted = condition(cur, condAccepted, false, "PortUnavailable", "binding "+addr.String()+": "+why, gen, in.Now) + } + } + l.conds = []metav1.Condition{accepted, resolved} + out = append(out, l) + } + // listeners sharing a port must agree on the protocol, and HTTP(S) ones must differ in hostname + byPort := map[int32][]*listener{} + for _, l := range out { + byPort[l.spec.Port] = append(byPort[l.spec.Port], l) + } + for _, ls := range byPort { + protos := map[string]bool{} + for _, l := range ls { + protos[l.spec.Protocol] = true + } + for _, l := range ls { + reason := "" + if len(protos) > 1 { + reason = "ProtocolConflict" + } else { + for _, o := range ls { + if o != l && (l.spec.Protocol == ProtocolTCP || str(o.spec.Hostname) == str(l.spec.Hostname)) { + reason = "HostnameConflict" + } + } + } + if reason != "" { + l.conflicted = true + } + cur := currentListener(gw, l.spec.Name) + if reason != "" { + l.conds = append(l.conds, condition(cur, condConflicted, true, reason, fmt.Sprintf("port %d is shared with listeners that cannot be told apart", l.spec.Port), gen, in.Now)) + } else { + l.conds = append(l.conds, condition(cur, condConflicted, false, "NoConflicts", "", gen, in.Now)) + } + } + } + return out +} + +// listenerCert resolves an HTTPS listener's certificate (the first certificateRef: one certificate per listener). +func listenerCert(gw *Gateway, l Listener, in GatewayInput, secrets map[string]*coreapi.Secret) (*coreapi.Secret, string, string) { + if l.TLS == nil || len(l.TLS.CertificateRefs) == 0 { + return nil, "InvalidCertificateRef", "an HTTPS listener needs tls.certificateRefs" + } + if m := str(l.TLS.Mode); m != "" && m != "Terminate" { + return nil, "UnsupportedValue", fmt.Sprintf("tls mode %s is not supported on HTTPS (Terminate is)", m) + } + ref := l.TLS.CertificateRefs[0] + if group(ref.Group, "") != "" || kind(ref.Kind, kindSecret) != kindSecret { + return nil, "InvalidCertificateRef", fmt.Sprintf("certificateRef %s/%s is not a Secret", group(ref.Group, ""), kind(ref.Kind, kindSecret)) + } + target := ns(ref.Namespace, gw.Namespace) + if target != gw.Namespace && !granted(in.Grants, GatewayGroup, kindGateway, gw.Namespace, "", kindSecret, target, ref.Name) { + return nil, "RefNotPermitted", fmt.Sprintf("no ReferenceGrant in %s lets Gateways in %s use Secret %s", target, gw.Namespace, ref.Name) + } + if !in.SecretNamespaces[target] { + return nil, "RefNotPermitted", fmt.Sprintf("this agent reads Secrets only in its gateway namespaces, not in %s", target) + } + s := secrets[target+"/"+ref.Name] + if s == nil { + return nil, "InvalidCertificateRef", fmt.Sprintf("Secret %s/%s not found", target, ref.Name) + } + if _, err := tls.X509KeyPair(s.Data[coreapi.TLSCertKey], s.Data[coreapi.TLSPrivateKeyKey]); err != nil { + return nil, "InvalidCertificateRef", fmt.Sprintf("Secret %s/%s holds no usable certificate and key: %v", target, ref.Name, err) + } + return s, "", "" +} + +// listenersFor are the listeners a parentRef names (all, or the one its sectionName and port pick). +func listenersFor(v *gatewayView, pr ParentReference) []*listener { + var out []*listener + for _, l := range v.listeners { + if pr.SectionName != nil && *pr.SectionName != l.spec.Name { + continue + } + if pr.Port != nil && *pr.Port != l.spec.Port { + continue + } + out = append(out, l) + } + return out +} + +// allows reports whether a listener takes routes of this kind from this namespace. +func allows(l *listener, gw *Gateway, routeKind, routeNS string, nsLabels map[string]labels.Set) bool { + if !l.usable || l.conflicted || !contains(l.kinds, routeKind) { + return false + } + from := "Same" + var sel *metav1.LabelSelector + if l.spec.AllowedRoutes != nil && l.spec.AllowedRoutes.Namespaces != nil { + from = str(l.spec.AllowedRoutes.Namespaces.From) + if from == "" { + from = "Same" + } + sel = l.spec.AllowedRoutes.Namespaces.Selector + } + switch from { + case "All": + return true + case "Same": + return routeNS == gw.Namespace + case "Selector": + if sel == nil { + return false + } + s, err := metav1.LabelSelectorAsSelector(sel) + return err == nil && s.Matches(nsLabels[routeNS]) + } + return false +} + +func attachHTTP(r *HTTPRoute, pr ParentReference, v *gatewayView, in GatewayInput, nsLabels map[string]labels.Set, services map[string]*coreapi.Service) (metav1.Condition, metav1.Condition) { + ls := listenersFor(v, pr) + if len(ls) == 0 { + return cond(false, "NoMatchingParent", "no listener matches the parentRef's sectionName and port"), cond(true, "ResolvedRefs", "") + } + if msg := unsupportedHTTP(r); msg != "" { + return cond(false, "UnsupportedValue", msg), cond(true, "ResolvedRefs", "") + } + rules, resolved := serveRules(r, in, services) + allowed, hostOK := false, false + for _, l := range ls { + if !allows(l, v.gw, kindHTTPRoute, r.Namespace, nsLabels) { + continue + } + allowed = true + hosts, ok := narrowHostnames(str(l.spec.Hostname), r.Spec.Hostnames) + if !ok { + continue + } + hostOK = true + l.attached++ + l.http = append(l.http, ServeRoute{Key: objKey(&r.ObjectMeta), Created: r.CreationTimestamp.Time, Hostnames: hosts, Rules: rules}) + } + switch { + case !allowed: + return cond(false, "NotAllowedByListeners", "no listener the parentRef names takes HTTPRoutes from this namespace"), resolved + case !hostOK: + return cond(false, "NoMatchingListenerHostname", "the route's hostnames intersect no listener's"), resolved + } + return cond(true, "Accepted", ""), resolved +} + +func attachTCP(r *TCPRoute, pr ParentReference, v *gatewayView, in GatewayInput, nsLabels map[string]labels.Set, services map[string]*coreapi.Service) (metav1.Condition, metav1.Condition) { + ls := listenersFor(v, pr) + if len(ls) == 0 { + return cond(false, "NoMatchingParent", "no listener matches the parentRef's sectionName and port"), cond(true, "ResolvedRefs", "") + } + resolved := cond(true, "ResolvedRefs", "") + for _, rule := range r.Spec.Rules { + for _, b := range rule.BackendRefs { + if _, why, reason := resolveBackend(b, kindTCPRoute, r.Namespace, in, services); why != "" { + resolved = cond(false, reason, why) + } + } + } + allowed, attached := false, false + busy := "" + for _, l := range ls { + if !allows(l, v.gw, kindTCPRoute, r.Namespace, nsLabels) { + continue + } + allowed = true + if l.tcpRoute != nil { + // a TCP listener forwards one route: the oldest attached keeps it + busy = objKey(&l.tcpRoute.ObjectMeta) + continue + } + l.tcpRoute = r + l.attached++ + attached = true + } + switch { + case !allowed: + return cond(false, "NotAllowedByListeners", "no listener the parentRef names takes TCPRoutes from this namespace"), resolved + case !attached: + return cond(false, "NotAllowedByListeners", fmt.Sprintf("the listener already forwards TCPRoute %s", busy)), resolved + } + return cond(true, "Accepted", ""), resolved +} + +// unsupportedHTTP names what in an HTTPRoute this agent does not implement ("" if nothing). +func unsupportedHTTP(r *HTTPRoute) string { + okFilter := func(fs []HTTPRouteFilter) string { + for _, f := range fs { + switch f.Type { + case filterRequestHeaderModifier, filterResponseHeaderModifier, filterRequestRedirect, filterURLRewrite: + default: + return fmt.Sprintf("filter %s is not supported", f.Type) + } + } + return "" + } + for _, rule := range r.Spec.Rules { + for _, m := range rule.Matches { + if m.Path != nil && m.Path.Type != nil && *m.Path.Type != pathExact && *m.Path.Type != pathPrefix { + return fmt.Sprintf("path match type %s is not supported", *m.Path.Type) + } + for _, h := range m.Headers { + if t := str(h.Type); t != "" && t != "Exact" { + return fmt.Sprintf("header match type %s is not supported", t) + } + } + for _, q := range m.QueryParams { + if t := str(q.Type); t != "" && t != "Exact" { + return fmt.Sprintf("query parameter match type %s is not supported", t) + } + } + } + if msg := okFilter(rule.Filters); msg != "" { + return msg + } + for _, b := range rule.BackendRefs { + if msg := okFilter(b.Filters); msg != "" { + return msg + } + } + } + return "" +} + +// serveRules resolves an HTTPRoute's rules into what the proxy runs, and whether all its backends resolved. +func serveRules(r *HTTPRoute, in GatewayInput, services map[string]*coreapi.Service) ([]ServeRule, metav1.Condition) { + resolved := cond(true, "ResolvedRefs", "") + var rules []ServeRule + for _, rule := range r.Spec.Rules { + sr := ServeRule{Filters: rule.Filters} + for _, m := range rule.Matches { + if m.Path == nil { + m.Path = &HTTPPathMatch{} + } + if m.Path.Type == nil { + t := pathPrefix + m.Path.Type = &t + } + if m.Path.Value == nil { + v := "/" + m.Path.Value = &v + } + sr.Matches = append(sr.Matches, m) + } + if len(sr.Matches) == 0 { + t, v := pathPrefix, "/" + sr.Matches = []HTTPRouteMatch{{Path: &HTTPPathMatch{Type: &t, Value: &v}}} + } + for _, b := range rule.BackendRefs { + eps, why, reason := resolveBackend(b.BackendRef, kindHTTPRoute, r.Namespace, in, services) + if why != "" { + resolved = cond(false, reason, why) + sr.Broken = true + continue + } + w := int32(1) + if b.Weight != nil { + w = *b.Weight + } + sr.Backends = append(sr.Backends, ServeBackend{ + Key: fmt.Sprintf("%s/%s:%d", ns(b.Namespace, r.Namespace), b.Name, *b.Port), Weight: w, Endpoints: eps, Filters: b.Filters, + }) + } + rules = append(rules, sr) + } + return rules, resolved +} + +// resolveBackend finds a backendRef's ready endpoints, or why it cannot be used and the reason to report. +func resolveBackend(b BackendRef, routeKind, routeNS string, in GatewayInput, services map[string]*coreapi.Service) ([]Endpoint, string, string) { + if group(b.Group, "") != "" || kind(b.Kind, kindService) != kindService { + return nil, fmt.Sprintf("backend %s %s is not a Service", kind(b.Kind, kindService), b.Name), "InvalidKind" + } + target := ns(b.Namespace, routeNS) + if target != routeNS && !granted(in.Grants, GatewayGroup, routeKind, routeNS, "", kindService, target, b.Name) { + return nil, fmt.Sprintf("no ReferenceGrant in %s lets %ss in %s use Service %s", target, routeKind, routeNS, b.Name), "RefNotPermitted" + } + s := services[target+"/"+b.Name] + if s == nil { + return nil, fmt.Sprintf("Service %s/%s not found", target, b.Name), "BackendNotFound" + } + if b.Port == nil { + return nil, fmt.Sprintf("backend Service %s/%s needs a port", target, b.Name), "BackendNotFound" + } + for _, p := range ports(s, in.Slices) { + if p.Port == *b.Port && p.Protocol == coreapi.ProtocolTCP { + return p.Endpoints, "", "" + } + } + return nil, fmt.Sprintf("Service %s/%s has no TCP port %d", target, b.Name, *b.Port), "BackendNotFound" +} + +// tcpEndpoints is the union of a TCPRoute's resolved backends' endpoints (the kernel forward does not weigh them). +func tcpEndpoints(r *TCPRoute, in GatewayInput, services map[string]*coreapi.Service) []Endpoint { + seen := map[string]bool{} + var out []Endpoint + for _, rule := range r.Spec.Rules { + for _, b := range rule.BackendRefs { + if b.Weight != nil && *b.Weight == 0 { + continue + } + eps, why, _ := resolveBackend(b, kindTCPRoute, r.Namespace, in, services) + if why != "" { + continue + } + for _, e := range eps { + if !seen[e.String()] { + seen[e.String()] = true + out = append(out, e) + } + } + } + } + sort.Slice(out, func(i, j int) bool { return out[i].String() < out[j].String() }) + return out +} + +// granted reports whether a ReferenceGrant in the target namespace lets (fromGroup, fromKind) objects in fromNS +// refer to the (toGroup, toKind) object named name there. +func granted(grants []*ReferenceGrant, fromGroup, fromKind, fromNS, toGroup, toKind, toNS, name string) bool { + for _, g := range grants { + if g.Namespace != toNS { + continue + } + from := false + for _, f := range g.Spec.From { + if f.Group == fromGroup && f.Kind == fromKind && f.Namespace == fromNS { + from = true + } + } + if !from { + continue + } + for _, t := range g.Spec.To { + if t.Group == toGroup && t.Kind == toKind && (t.Name == nil || *t.Name == name) { + return true + } + } + } + return false +} + +// narrowHostnames: the route's hostnames that a listener with this hostname takes (narrowed to it), and whether +// the route attaches at all. A route without hostnames takes the listener's (an empty result: any host). +func narrowHostnames(listener string, route []string) ([]string, bool) { + if len(route) == 0 { + if listener == "" { + return nil, true + } + return []string{listener}, true + } + if listener == "" { + return append([]string(nil), route...), true + } + var out []string + for _, h := range route { + switch { + case hostMatches(listener, h): + out = append(out, h) // the route's is as narrow or narrower + case hostMatches(h, listener): + out = append(out, listener) // the listener's is narrower + } + } + return out, len(out) > 0 +} + +// hostMatches reports whether a hostname pattern (exact, or "*." for any non-empty prefix) covers name, which may +// itself be a wildcard (then the pattern must cover its whole suffix). +func hostMatches(pattern, name string) bool { + pattern, name = strings.ToLower(pattern), strings.ToLower(name) + if !strings.HasPrefix(pattern, "*.") { + return pattern == name + } + suffix := pattern[1:] // ".example.com" + if strings.HasPrefix(name, "*.") { + return strings.HasSuffix(name[1:], suffix) && len(name[1:]) >= len(suffix) + } + return strings.HasSuffix(name, suffix) && len(name) > len(suffix) +} + +// gatewayStatus is what a Gateway's status should say. +func gatewayStatus(v *gatewayView, now metav1.Time) GatewayStatus { + gw := v.gw + gen := gw.Generation + var st GatewayStatus + invalid := 0 + for _, l := range v.listeners { + if !l.usable || l.conflicted || l.unbound { + invalid++ + } + } + switch { + case v.addrErr != "": + st.Conditions = append(st.Conditions, condition(gw.Status.Conditions, condAccepted, false, "UnsupportedAddress", v.addrErr, gen, now)) + case invalid > 0: + st.Conditions = append(st.Conditions, condition(gw.Status.Conditions, condAccepted, true, "ListenersNotValid", fmt.Sprintf("%d of %d listeners are not valid", invalid, len(v.listeners)), gen, now)) + default: + st.Conditions = append(st.Conditions, condition(gw.Status.Conditions, condAccepted, true, "Accepted", "", gen, now)) + } + programmed := v.addrErr == "" && v.holder != nil + switch { + case v.addrErr != "": + st.Conditions = append(st.Conditions, condition(gw.Status.Conditions, condProgrammed, false, "Invalid", v.addrErr, gen, now)) + case v.holder == nil: + st.Conditions = append(st.Conditions, condition(gw.Status.Conditions, condProgrammed, false, "AddressNotUsable", v.holderErr, gen, now)) + default: + st.Conditions = append(st.Conditions, condition(gw.Status.Conditions, condProgrammed, true, "Programmed", "served by "+v.holder.Name, gen, now)) + ipType := "IPAddress" + st.Addresses = []GatewayAddress{{Type: &ipType, Value: v.address.String()}} + } + for _, l := range v.listeners { + cur := currentListener(gw, l.spec.Name) + ls := ListenerStatus{Name: l.spec.Name, AttachedRoutes: l.attached, SupportedKinds: []RouteGroupKind{}} + for _, k := range l.kinds { + g := GatewayGroup + ls.SupportedKinds = append(ls.SupportedKinds, RouteGroupKind{Group: &g, Kind: k}) + } + ls.Conditions = append(ls.Conditions, l.conds...) + switch { + case !programmed: + ls.Conditions = append(ls.Conditions, condition(cur, condProgrammed, false, "Pending", "the Gateway is not served", gen, now)) + case !l.usable || l.conflicted: + ls.Conditions = append(ls.Conditions, condition(cur, condProgrammed, false, "Invalid", "the listener is not valid", gen, now)) + case l.unbound: + ls.Conditions = append(ls.Conditions, condition(cur, condProgrammed, false, "Invalid", "its port could not be bound", gen, now)) + default: + ls.Conditions = append(ls.Conditions, condition(cur, condProgrammed, true, "Programmed", "", gen, now)) + } + st.Listeners = append(st.Listeners, ls) + } + return st +} + +func currentListener(gw *Gateway, name string) []metav1.Condition { + for _, l := range gw.Status.Listeners { + if l.Name == name { + return l.Conditions + } + } + return nil +} + +func currentParent(st RouteStatus, pr ParentReference) []metav1.Condition { + for _, p := range st.Parents { + if p.ControllerName == GatewayControllerName && sameJSON(p.ParentRef, pr) { + return p.Conditions + } + } + return nil +} + +// condition builds a condition, keeping the current one's transition time while its status stays the same (so an +// unchanged status reads back equal, and nothing is rewritten). +func condition(current []metav1.Condition, typ string, ok bool, reason, message string, gen int64, now metav1.Time) metav1.Condition { + status := metav1.ConditionFalse + if ok { + status = metav1.ConditionTrue + } + c := metav1.Condition{Type: typ, Status: status, Reason: reason, Message: message, ObservedGeneration: gen, LastTransitionTime: now} + for _, o := range current { + if o.Type == typ && o.Status == status { + c.LastTransitionTime = o.LastTransitionTime + } + } + return c +} + +// cond is a condition's verdict before it is dated (attach functions decide; the caller dates). +func cond(ok bool, reason, message string) metav1.Condition { + s := metav1.ConditionFalse + if ok { + s = metav1.ConditionTrue + } + return metav1.Condition{Status: s, Reason: reason, Message: message} +} + +func sameJSON(a, b interface{}) bool { + x, err1 := json.Marshal(a) + y, err2 := json.Marshal(b) + return err1 == nil && err2 == nil && string(x) == string(y) +} + +func objKey(m *metav1.ObjectMeta) string { return m.Namespace + "/" + m.Name } + +func str(p *string) string { + if p == nil { + return "" + } + return *p +} + +func group(p *string, def string) string { + if p == nil { + return def + } + return *p +} + +func kind(p *string, def string) string { + if p == nil || *p == "" { + return def + } + return *p +} + +func ns(p *string, def string) string { + if p == nil || *p == "" { + return def + } + return *p +} + +func contains(xs []string, x string) bool { + for _, y := range xs { + if y == x { + return true + } + } + return false +} + +// MergeGateways folds what a node does for Gateways into what it does for LoadBalancer Services: one set of +// forwards, one set of holders whose replies are steered back (marks over both), and the ports it serves itself. +func MergeGateways(d Desired, g GatewayDesired, self string, nodes []Node) Desired { + d.Forwards = append(d.Forwards, g.Forwards...) + holding := map[string]bool{} + for _, h := range d.Holders { + holding[h.Node.Name] = true + } + for _, n := range g.Holding { + holding[n] = true + } + sorted := append([]Node(nil), nodes...) + sort.Slice(sorted, func(i, j int) bool { return sorted[i].Name < sorted[j].Name }) + var others []Node + for _, n := range sorted { + if !holding[n.Name] || n.Name == self { + continue + } + if n.WeaveIP == nil { + d.Problems = append(d.Problems, fmt.Sprintf("node %s holds an address but has not published its weave bridge; replies to its forwards cannot reach it", n.Name)) + continue + } + others = append(others, n) + } + d.Holders = assignMarks(others) + seen := map[string]bool{} + for _, l := range g.Serve { + k := fmt.Sprintf("%s:%d", l.Address, l.Port) + if seen[k] { + continue + } + seen[k] = true + d.Accept = append(d.Accept, Accept{Address: l.Address, Port: l.Port, Comment: "gateway " + l.Gateway + " " + l.Name}) + } + d.Problems = append(d.Problems, g.Problems...) + return d +} diff --git a/lb/gateway_api.go b/lb/gateway_api.go new file mode 100644 index 00000000..38983517 --- /dev/null +++ b/lb/gateway_api.go @@ -0,0 +1,415 @@ +package lb + +import ( + "context" + "encoding/json" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/runtime/serializer" + "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/rest" +) + +// The Gateway API (gateway.networking.k8s.io/v1, as the v1.6.3 standard channel serves it): the subset this agent +// implements, as plain types. The upstream Go module needs a newer client-go than the fork's; the JSON is the +// contract, and fields this agent does not implement are simply not decoded. + +const ( + GatewayGroup = "gateway.networking.k8s.io" + GatewayVersion = "v1" + // GatewayControllerName is the spec.controllerName of the GatewayClasses this agent implements. + GatewayControllerName = "reweave.levk.dev/gateway" + + GatewayClasses = "gatewayclasses" + Gateways = "gateways" + HTTPRoutes = "httproutes" + TCPRoutes = "tcproutes" + ReferenceGrants = "referencegrants" +) + +// GatewayGroupVersion of the Gateway API. +var GatewayGroupVersion = schema.GroupVersion{Group: GatewayGroup, Version: GatewayVersion} + +type GatewayClass struct { + metav1.TypeMeta `json:",inline"` + metav1.ObjectMeta `json:"metadata,omitempty"` + Spec GatewayClassSpec `json:"spec"` + Status GatewayClassStatus `json:"status,omitempty"` +} + +type GatewayClassSpec struct { + ControllerName string `json:"controllerName"` +} + +type GatewayClassStatus struct { + Conditions []metav1.Condition `json:"conditions,omitempty"` +} + +type Gateway struct { + metav1.TypeMeta `json:",inline"` + metav1.ObjectMeta `json:"metadata,omitempty"` + Spec GatewaySpec `json:"spec"` + Status GatewayStatus `json:"status,omitempty"` +} + +type GatewaySpec struct { + GatewayClassName string `json:"gatewayClassName"` + Listeners []Listener `json:"listeners"` + Addresses []GatewayAddress `json:"addresses,omitempty"` +} + +// GatewayAddress is a spec or status address; Type defaults to IPAddress. +type GatewayAddress struct { + Type *string `json:"type,omitempty"` + Value string `json:"value"` +} + +type Listener struct { + Name string `json:"name"` + Hostname *string `json:"hostname,omitempty"` + Port int32 `json:"port"` + Protocol string `json:"protocol"` + TLS *ListenerTLSConfig `json:"tls,omitempty"` + AllowedRoutes *AllowedRoutes `json:"allowedRoutes,omitempty"` +} + +type ListenerTLSConfig struct { + Mode *string `json:"mode,omitempty"` + CertificateRefs []SecretObjectReference `json:"certificateRefs,omitempty"` +} + +type SecretObjectReference struct { + Group *string `json:"group,omitempty"` + Kind *string `json:"kind,omitempty"` + Name string `json:"name"` + Namespace *string `json:"namespace,omitempty"` +} + +type AllowedRoutes struct { + Namespaces *RouteNamespaces `json:"namespaces,omitempty"` + Kinds []RouteGroupKind `json:"kinds,omitempty"` +} + +type RouteNamespaces struct { + From *string `json:"from,omitempty"` + Selector *metav1.LabelSelector `json:"selector,omitempty"` +} + +type RouteGroupKind struct { + Group *string `json:"group,omitempty"` + Kind string `json:"kind"` +} + +type GatewayStatus struct { + Addresses []GatewayAddress `json:"addresses,omitempty"` + Conditions []metav1.Condition `json:"conditions,omitempty"` + Listeners []ListenerStatus `json:"listeners,omitempty"` +} + +type ListenerStatus struct { + Name string `json:"name"` + SupportedKinds []RouteGroupKind `json:"supportedKinds"` + AttachedRoutes int32 `json:"attachedRoutes"` + Conditions []metav1.Condition `json:"conditions"` +} + +type ParentReference struct { + Group *string `json:"group,omitempty"` + Kind *string `json:"kind,omitempty"` + Namespace *string `json:"namespace,omitempty"` + Name string `json:"name"` + SectionName *string `json:"sectionName,omitempty"` + Port *int32 `json:"port,omitempty"` +} + +type BackendRef struct { + Group *string `json:"group,omitempty"` + Kind *string `json:"kind,omitempty"` + Name string `json:"name"` + Namespace *string `json:"namespace,omitempty"` + Port *int32 `json:"port,omitempty"` + Weight *int32 `json:"weight,omitempty"` +} + +type RouteStatus struct { + Parents []RouteParentStatus `json:"parents"` +} + +type RouteParentStatus struct { + ParentRef ParentReference `json:"parentRef"` + ControllerName string `json:"controllerName"` + Conditions []metav1.Condition `json:"conditions,omitempty"` +} + +type HTTPRoute struct { + metav1.TypeMeta `json:",inline"` + metav1.ObjectMeta `json:"metadata,omitempty"` + Spec HTTPRouteSpec `json:"spec"` + Status RouteStatus `json:"status,omitempty"` +} + +type HTTPRouteSpec struct { + ParentRefs []ParentReference `json:"parentRefs,omitempty"` + Hostnames []string `json:"hostnames,omitempty"` + Rules []HTTPRouteRule `json:"rules,omitempty"` +} + +type HTTPRouteRule struct { + Matches []HTTPRouteMatch `json:"matches,omitempty"` + Filters []HTTPRouteFilter `json:"filters,omitempty"` + BackendRefs []HTTPBackendRef `json:"backendRefs,omitempty"` +} + +type HTTPRouteMatch struct { + Path *HTTPPathMatch `json:"path,omitempty"` + Headers []HTTPHeaderMatch `json:"headers,omitempty"` + QueryParams []HTTPQueryParamMatch `json:"queryParams,omitempty"` + Method *string `json:"method,omitempty"` +} + +type HTTPPathMatch struct { + Type *string `json:"type,omitempty"` + Value *string `json:"value,omitempty"` +} + +type HTTPHeaderMatch struct { + Type *string `json:"type,omitempty"` + Name string `json:"name"` + Value string `json:"value"` +} + +type HTTPQueryParamMatch struct { + Type *string `json:"type,omitempty"` + Name string `json:"name"` + Value string `json:"value"` +} + +// HTTPRouteFilter: the filters this agent implements are decoded; for the others only their presence is known +// (through Type), which is enough to refuse them. +type HTTPRouteFilter struct { + Type string `json:"type"` + RequestHeaderModifier *HTTPHeaderFilter `json:"requestHeaderModifier,omitempty"` + ResponseHeaderModifier *HTTPHeaderFilter `json:"responseHeaderModifier,omitempty"` + RequestRedirect *HTTPRequestRedirectFilter `json:"requestRedirect,omitempty"` + URLRewrite *HTTPURLRewriteFilter `json:"urlRewrite,omitempty"` +} + +type HTTPHeaderFilter struct { + Set []HTTPHeader `json:"set,omitempty"` + Add []HTTPHeader `json:"add,omitempty"` + Remove []string `json:"remove,omitempty"` +} + +type HTTPHeader struct { + Name string `json:"name"` + Value string `json:"value"` +} + +type HTTPRequestRedirectFilter struct { + Scheme *string `json:"scheme,omitempty"` + Hostname *string `json:"hostname,omitempty"` + Path *HTTPPathModifier `json:"path,omitempty"` + Port *int32 `json:"port,omitempty"` + StatusCode *int `json:"statusCode,omitempty"` +} + +type HTTPURLRewriteFilter struct { + Hostname *string `json:"hostname,omitempty"` + Path *HTTPPathModifier `json:"path,omitempty"` +} + +type HTTPPathModifier struct { + Type string `json:"type"` + ReplaceFullPath *string `json:"replaceFullPath,omitempty"` + ReplacePrefixMatch *string `json:"replacePrefixMatch,omitempty"` +} + +type HTTPBackendRef struct { + BackendRef `json:",inline"` + Filters []HTTPRouteFilter `json:"filters,omitempty"` +} + +type TCPRoute struct { + metav1.TypeMeta `json:",inline"` + metav1.ObjectMeta `json:"metadata,omitempty"` + Spec TCPRouteSpec `json:"spec"` + Status RouteStatus `json:"status,omitempty"` +} + +type TCPRouteSpec struct { + ParentRefs []ParentReference `json:"parentRefs,omitempty"` + Rules []TCPRouteRule `json:"rules,omitempty"` +} + +type TCPRouteRule struct { + BackendRefs []BackendRef `json:"backendRefs,omitempty"` +} + +type ReferenceGrant struct { + metav1.TypeMeta `json:",inline"` + metav1.ObjectMeta `json:"metadata,omitempty"` + Spec ReferenceGrantSpec `json:"spec"` +} + +type ReferenceGrantSpec struct { + From []ReferenceGrantFrom `json:"from"` + To []ReferenceGrantTo `json:"to"` +} + +type ReferenceGrantFrom struct { + Group string `json:"group"` + Kind string `json:"kind"` + Namespace string `json:"namespace"` +} + +type ReferenceGrantTo struct { + Group string `json:"group"` + Kind string `json:"kind"` + Name *string `json:"name,omitempty"` +} + +type GatewayClassList struct { + metav1.TypeMeta `json:",inline"` + metav1.ListMeta `json:"metadata,omitempty"` + Items []GatewayClass `json:"items"` +} + +type GatewayList struct { + metav1.TypeMeta `json:",inline"` + metav1.ListMeta `json:"metadata,omitempty"` + Items []Gateway `json:"items"` +} + +type HTTPRouteList struct { + metav1.TypeMeta `json:",inline"` + metav1.ListMeta `json:"metadata,omitempty"` + Items []HTTPRoute `json:"items"` +} + +type TCPRouteList struct { + metav1.TypeMeta `json:",inline"` + metav1.ListMeta `json:"metadata,omitempty"` + Items []TCPRoute `json:"items"` +} + +type ReferenceGrantList struct { + metav1.TypeMeta `json:",inline"` + metav1.ListMeta `json:"metadata,omitempty"` + Items []ReferenceGrant `json:"items"` +} + +// jsonCopy deep-copies through JSON: these types are only what the JSON says, so nothing is lost. +func jsonCopy(in, out interface{}) { + b, err := json.Marshal(in) + if err != nil { + panic(err) + } + if err := json.Unmarshal(b, out); err != nil { + panic(err) + } +} + +func (o *GatewayClass) DeepCopyObject() runtime.Object { + out := &GatewayClass{} + jsonCopy(o, out) + return out +} + +func (o *Gateway) DeepCopyObject() runtime.Object { + out := &Gateway{} + jsonCopy(o, out) + return out +} + +func (o *HTTPRoute) DeepCopyObject() runtime.Object { + out := &HTTPRoute{} + jsonCopy(o, out) + return out +} + +func (o *TCPRoute) DeepCopyObject() runtime.Object { + out := &TCPRoute{} + jsonCopy(o, out) + return out +} + +func (o *ReferenceGrant) DeepCopyObject() runtime.Object { + out := &ReferenceGrant{} + jsonCopy(o, out) + return out +} + +func (o *GatewayClassList) DeepCopyObject() runtime.Object { + out := &GatewayClassList{} + jsonCopy(o, out) + return out +} + +func (o *GatewayList) DeepCopyObject() runtime.Object { + out := &GatewayList{} + jsonCopy(o, out) + return out +} + +func (o *HTTPRouteList) DeepCopyObject() runtime.Object { + out := &HTTPRouteList{} + jsonCopy(o, out) + return out +} + +func (o *TCPRouteList) DeepCopyObject() runtime.Object { + out := &TCPRouteList{} + jsonCopy(o, out) + return out +} + +func (o *ReferenceGrantList) DeepCopyObject() runtime.Object { + out := &ReferenceGrantList{} + jsonCopy(o, out) + return out +} + +func addGatewayTypes(s *runtime.Scheme) error { + s.AddKnownTypes(GatewayGroupVersion, + &GatewayClass{}, &GatewayClassList{}, &Gateway{}, &GatewayList{}, &HTTPRoute{}, &HTTPRouteList{}, + &TCPRoute{}, &TCPRouteList{}, &ReferenceGrant{}, &ReferenceGrantList{}) + metav1.AddToGroupVersion(s, GatewayGroupVersion) + return nil +} + +// NewGatewayClient is a REST client for the Gateway API kinds this agent reads and writes the status of. +func NewGatewayClient(base *rest.Config) (*rest.RESTClient, error) { + scheme := runtime.NewScheme() + if err := addGatewayTypes(scheme); err != nil { + return nil, err + } + cfg := *base + cfg.GroupVersion = &GatewayGroupVersion + cfg.APIPath = "/apis" + cfg.ContentType = runtime.ContentTypeJSON + cfg.NegotiatedSerializer = serializer.NewCodecFactory(scheme).WithoutConversion() + if cfg.UserAgent == "" { + cfg.UserAgent = rest.DefaultKubernetesUserAgent() + } + return rest.RESTClientFor(&cfg) +} + +// patchStatus merge-patches an object's status. resourceVersion, when given, makes it conditional: a status built +// from a stale read fails instead of overwriting what someone else wrote since. +func patchStatus(ctx context.Context, c rest.Interface, resource, namespace, name, resourceVersion string, status interface{}) error { + body := map[string]interface{}{"status": status} + if resourceVersion != "" { + body["metadata"] = map[string]interface{}{"resourceVersion": resourceVersion} + } + patch, err := json.Marshal(body) + if err != nil { + return err + } + req := c.Patch(types.MergePatchType).Resource(resource).Name(name).SubResource("status").Body(patch) + if namespace != "" { + req = req.Namespace(namespace) + } + return req.Do(ctx).Error() +} diff --git a/lb/gateway_proxy.go b/lb/gateway_proxy.go new file mode 100644 index 00000000..d21889eb --- /dev/null +++ b/lb/gateway_proxy.go @@ -0,0 +1,542 @@ +package lb + +import ( + "context" + "crypto/tls" + "fmt" + "log" + "math/rand" + "net" + "net/http" + "net/http/httputil" + "net/url" + "sort" + "strconv" + "strings" + "sync" + "sync/atomic" + "time" + + coreapi "k8s.io/api/core/v1" + + "github.com/rajch/weave/common" +) + +// Proxy serves the HTTP and HTTPS listeners of the Gateways this node holds. TLS is terminated with the +// listener's certificate, chosen by SNI and swapped in place when its Secret changes (no restart, no rebind). +// Requests are routed by the attached HTTPRoutes in the order the spec prescribes and passed to a ready endpoint +// of the chosen backend, with the client's own address in X-Forwarded-For. Bodies are streamed, never limited. +type Proxy struct { + mu sync.Mutex + servers map[string]*portServer // by "address:port" + // listen binds an address (swappable in tests) + listen func(addr string) (net.Listener, error) + transport http.RoundTripper + certs map[string]*tls.Certificate // parsed certificates, by Secret uid@resourceVersion + rr sync.Map // backend key -> *uint64, for round robin over its endpoints +} + +func NewProxy() *Proxy { + return &Proxy{ + servers: map[string]*portServer{}, + certs: map[string]*tls.Certificate{}, + listen: func(addr string) (net.Listener, error) { return net.Listen("tcp", addr) }, + transport: &http.Transport{ + DialContext: (&net.Dialer{Timeout: 10 * time.Second, KeepAlive: 30 * time.Second}).DialContext, + MaxIdleConns: 256, + MaxIdleConnsPerHost: 32, + IdleConnTimeout: 90 * time.Second, + ExpectContinueTimeout: time.Second, + }, + } +} + +// portServer is one bound address:port and what it currently serves. +type portServer struct { + addr string + https bool + ln net.Listener + srv *http.Server + cfg atomic.Pointer[portConfig] +} + +// portConfig is everything one address:port serves: its listeners, most specific hostname first. +type portConfig struct { + listeners []*liveListener +} + +type liveListener struct { + spec ServeListener + cert *tls.Certificate +} + +// Apply serves exactly these listeners: ports no longer listed are closed, new ones bound, and every served +// port's routes and certificates replaced at once. It returns, by "namespace/gateway/listener", the listeners +// it could not bind. +func (p *Proxy) Apply(listeners []ServeListener) map[string]error { + p.mu.Lock() + defer p.mu.Unlock() + failed := map[string]error{} + byAddr := map[string][]ServeListener{} + for _, l := range listeners { + a := net.JoinHostPort(l.Address.String(), strconv.Itoa(int(l.Port))) + byAddr[a] = append(byAddr[a], l) + } + for a, s := range p.servers { + if group, ok := byAddr[a]; !ok || group[0].HTTPS != s.https { + s.close() + delete(p.servers, a) + } + } + used := map[string]bool{} + for a, group := range byAddr { + cfg := p.compile(group, used) + s := p.servers[a] + if s == nil { + var err error + if s, err = p.start(a, group[0].HTTPS); err != nil { + for _, l := range group { + failed[l.Gateway+"/"+l.Name] = err + } + continue + } + p.servers[a] = s + } + s.cfg.Store(cfg) + } + // forget certificates no listener uses any more + for k := range p.certs { + if !used[k] { + delete(p.certs, k) + } + } + return failed +} + +// Close stops serving everything. +func (p *Proxy) Close() { + p.Apply(nil) +} + +func (p *Proxy) compile(group []ServeListener, used map[string]bool) *portConfig { + cfg := &portConfig{} + for _, l := range group { + ll := &liveListener{spec: l} + if l.HTTPS && l.Cert != nil { + ll.cert = p.cert(l.Cert) + used[certKey(l.Cert)] = true + } + cfg.listeners = append(cfg.listeners, ll) + } + // most specific first: exact hostnames, then wildcards (longest first), then the catch-all + sort.SliceStable(cfg.listeners, func(i, j int) bool { + return hostRank(cfg.listeners[i].spec.Hostname) > hostRank(cfg.listeners[j].spec.Hostname) + }) + return cfg +} + +func certKey(s *coreapi.Secret) string { + return string(s.UID) + "@" + s.ResourceVersion + "/" + s.Namespace + "/" + s.Name +} + +// cert parses a Secret's certificate once per version of the Secret. +func (p *Proxy) cert(s *coreapi.Secret) *tls.Certificate { + k := certKey(s) + if c, ok := p.certs[k]; ok { + return c + } + c, err := tls.X509KeyPair(s.Data[coreapi.TLSCertKey], s.Data[coreapi.TLSPrivateKeyKey]) + if err != nil { + common.Log.Warnf("lb: gateway certificate %s/%s: %v", s.Namespace, s.Name, err) + return nil + } + p.certs[k] = &c + return &c +} + +// hostRank orders listener hostnames: exact (by length) above wildcards (by length) above none. +func hostRank(h string) int { + switch { + case h == "": + return 0 + case strings.HasPrefix(h, "*."): + return 1000 + len(h) + } + return 100000 + len(h) +} + +func (p *Proxy) start(addr string, https bool) (*portServer, error) { + ln, err := p.listen(addr) + if err != nil { + return nil, err + } + s := &portServer{addr: addr, https: https, ln: ln} + s.cfg.Store(&portConfig{}) + s.srv = &http.Server{ + Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { p.serve(s, w, r) }), + ReadHeaderTimeout: 30 * time.Second, + IdleTimeout: 2 * time.Minute, + // handshakes from scanners and the like would flood the log + ErrorLog: log.New(debugWriter{}, "lb: gateway "+addr+": ", 0), + } + if https { + s.srv.TLSConfig = &tls.Config{MinVersion: tls.VersionTLS12, GetCertificate: s.certificate} + go func() { + if err := s.srv.ServeTLS(ln, "", ""); err != nil && err != http.ErrServerClosed { + common.Log.Errorf("lb: gateway %s: %v", addr, err) + } + }() + } else { + go func() { + if err := s.srv.Serve(ln); err != nil && err != http.ErrServerClosed { + common.Log.Errorf("lb: gateway %s: %v", addr, err) + } + }() + } + common.Log.Infof("lb: gateway serving %s (%s)", addr, map[bool]string{true: "HTTPS", false: "HTTP"}[https]) + return s, nil +} + +func (s *portServer) close() { + srv := s.srv + go func() { + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + _ = srv.Shutdown(ctx) + _ = srv.Close() + }() + common.Log.Infof("lb: gateway no longer serving %s", s.addr) +} + +type debugWriter struct{} + +func (debugWriter) Write(b []byte) (int, error) { + common.Log.Debugf("%s", strings.TrimRight(string(b), "\n")) + return len(b), nil +} + +// listenerFor picks the listener a name (SNI or Host) reaches on a port, most specific hostname first. +func (c *portConfig) listenerFor(name string) *liveListener { + name = strings.ToLower(name) + for _, l := range c.listeners { + if l.spec.Hostname == "" || hostMatches(l.spec.Hostname, name) { + return l + } + } + return nil +} + +// certificate answers a TLS handshake with the certificate of the listener its SNI names. +func (s *portServer) certificate(hello *tls.ClientHelloInfo) (*tls.Certificate, error) { + cfg := s.cfg.Load() + l := cfg.listenerFor(hello.ServerName) + if l == nil || l.cert == nil { + return nil, fmt.Errorf("no listener for %q", hello.ServerName) + } + return l.cert, nil +} + +// candidate is one match of one rule that a request satisfies, with what decides its precedence. +type candidate struct { + route *ServeRoute + rule *ServeRule + ruleIdx int + // the matched hostname: characters if not a wildcard, and characters + hostExact, hostLen int + exact bool + prefixLen int + method bool + headers, query int + prefix string // the path prefix matched (for ReplacePrefixMatch) +} + +func (p *Proxy) serve(s *portServer, w http.ResponseWriter, r *http.Request) { + cfg := s.cfg.Load() + host := hostOnly(r.Host) + name := host + if r.TLS != nil && r.TLS.ServerName != "" { + name = r.TLS.ServerName + } + l := cfg.listenerFor(name) + if l == nil { + http.Error(w, "no listener for this host", http.StatusNotFound) + return + } + c := pick(l, host, r) + if c == nil { + http.Error(w, "no route", http.StatusNotFound) + return + } + for _, f := range c.rule.Filters { + if f.Type == filterRequestRedirect && f.RequestRedirect != nil { + to, code := redirect(r, f.RequestRedirect, l.spec.Port, c.prefix) + http.Redirect(w, r, to, code) + return + } + } + b := chooseBackend(c.rule.Backends) + if b == nil { + // no resolved backend to send it to: the spec's answer for unresolved backends + http.Error(w, "no usable backend", http.StatusInternalServerError) + return + } + if len(b.Endpoints) == 0 { + http.Error(w, "no ready endpoint", http.StatusServiceUnavailable) + return + } + n, _ := p.rr.LoadOrStore(b.Key, new(uint64)) + ep := b.Endpoints[int(atomic.AddUint64(n.(*uint64), 1)-1)%len(b.Endpoints)] + filters := append(append([]HTTPRouteFilter(nil), c.rule.Filters...), b.Filters...) + rp := &httputil.ReverseProxy{ + Transport: p.transport, + FlushInterval: -1, // stream as it comes (git, server-sent events) + Rewrite: func(pr *httputil.ProxyRequest) { + pr.SetURL(&url.URL{Scheme: "http", Host: ep.String()}) + pr.Out.Host = pr.In.Host // forwarded unmodified unless a filter says otherwise + // SetXForwarded replaces whatever the client sent: only this hop's view is trusted + pr.SetXForwarded() + if ip, _, err := net.SplitHostPort(pr.In.RemoteAddr); err == nil { + pr.Out.Header.Set("X-Real-IP", ip) + } + for _, f := range filters { + switch { + case f.Type == filterRequestHeaderModifier && f.RequestHeaderModifier != nil: + modifyHeaders(pr.Out.Header, f.RequestHeaderModifier) + case f.Type == filterURLRewrite && f.URLRewrite != nil: + if f.URLRewrite.Hostname != nil { + pr.Out.Host = *f.URLRewrite.Hostname + } + if f.URLRewrite.Path != nil { + pr.Out.URL.Path = rewritePath(pr.In.URL.Path, f.URLRewrite.Path, c.prefix) + pr.Out.URL.RawPath = "" + } + } + } + }, + ModifyResponse: func(resp *http.Response) error { + for _, f := range filters { + if f.Type == filterResponseHeaderModifier && f.ResponseHeaderModifier != nil { + modifyHeaders(resp.Header, f.ResponseHeaderModifier) + } + } + return nil + }, + ErrorHandler: func(w http.ResponseWriter, r *http.Request, err error) { + common.Log.Debugf("lb: gateway %s -> %s: %v", s.addr, ep, err) + w.WriteHeader(http.StatusBadGateway) + }, + } + rp.ServeHTTP(w, r) +} + +// pick is the match that takes the request, by the spec's precedence: the most specific matching hostname, then +// an exact path, the longest prefix, a method, the most header matches, the most query matches, then the oldest +// route, its namespace/name, and the first rule. +func pick(l *liveListener, host string, r *http.Request) *candidate { + var best *candidate + for ri := range l.spec.Routes { + route := &l.spec.Routes[ri] + hostExact, hostLen, ok := routeHost(route, l.spec.Hostname, host) + if !ok { + continue + } + for i := range route.Rules { + rule := &route.Rules[i] + for _, m := range rule.Matches { + c, ok := matches(m, r) + if !ok { + continue + } + c.route, c.rule, c.ruleIdx, c.hostExact, c.hostLen = route, rule, i, hostExact, hostLen + if best == nil || better(c, *best) { + cc := c + best = &cc + } + } + } + } + return best +} + +func better(a, b candidate) bool { + switch { + case a.hostExact != b.hostExact: + return a.hostExact > b.hostExact + case a.hostLen != b.hostLen: + return a.hostLen > b.hostLen + case a.exact != b.exact: + return a.exact + case a.prefixLen != b.prefixLen: + return a.prefixLen > b.prefixLen + case a.method != b.method: + return a.method + case a.headers != b.headers: + return a.headers > b.headers + case a.query != b.query: + return a.query > b.query + case !a.route.Created.Equal(b.route.Created): + return a.route.Created.Before(b.route.Created) + case a.route.Key != b.route.Key: + return a.route.Key < b.route.Key + } + return a.ruleIdx < b.ruleIdx +} + +// routeHost reports whether a route answers this host on its listener, and how specifically. +func routeHost(route *ServeRoute, listenerHost, host string) (int, int, bool) { + names := route.Hostnames + if len(names) == 0 { + if listenerHost == "" { + return 0, 0, true + } + names = []string{listenerHost} + } + exact, length, ok := 0, 0, false + for _, h := range names { + if !hostMatches(h, host) { + continue + } + ok = true + e := 0 + if !strings.HasPrefix(h, "*.") { + e = len(h) + } + if e > exact || (e == exact && len(h) > length) { + exact, length = e, len(h) + } + } + return exact, length, ok +} + +func matches(m HTTPRouteMatch, r *http.Request) (candidate, bool) { + var c candidate + path, value := r.URL.Path, str(m.Path.Value) + switch str(m.Path.Type) { + case pathExact: + if path != value { + return c, false + } + c.exact = true + c.prefixLen = len(value) + default: + trimmed := strings.TrimSuffix(value, "/") + if trimmed != "" && path != trimmed && !strings.HasPrefix(path, trimmed+"/") { + return c, false + } + c.prefixLen = len(value) + c.prefix = value + } + if m.Method != nil { + if r.Method != *m.Method { + return c, false + } + c.method = true + } + for _, h := range m.Headers { + if r.Header.Get(h.Name) != h.Value { + return c, false + } + } + c.headers = len(m.Headers) + q := r.URL.Query() + for _, qp := range m.QueryParams { + if q.Get(qp.Name) != qp.Value { + return c, false + } + } + c.query = len(m.QueryParams) + return c, true +} + +// chooseBackend picks a backend by weight (nil when there is none to pick). +func chooseBackend(bs []ServeBackend) *ServeBackend { + total := int32(0) + for _, b := range bs { + if b.Weight > 0 { + total += b.Weight + } + } + if total == 0 { + return nil + } + n := rand.Int31n(total) + for i := range bs { + if bs[i].Weight <= 0 { + continue + } + if n < bs[i].Weight { + return &bs[i] + } + n -= bs[i].Weight + } + return nil +} + +// redirect is where a RequestRedirect filter sends a request, and with which status. +func redirect(r *http.Request, f *HTTPRequestRedirectFilter, listenerPort int32, prefix string) (string, int) { + scheme := "http" + if r.TLS != nil { + scheme = "https" + } + port := int(listenerPort) + if f.Scheme != nil { + // a new scheme brings its well-known port, unless a port is given + scheme = *f.Scheme + port = map[string]int{"http": 80, "https": 443}[scheme] + } + if f.Port != nil { + port = int(*f.Port) + } + host := hostOnly(r.Host) + if f.Hostname != nil { + host = *f.Hostname + } + if !(scheme == "http" && port == 80) && !(scheme == "https" && port == 443) && port != 0 { + host = net.JoinHostPort(host, strconv.Itoa(port)) + } + path := r.URL.Path + if f.Path != nil { + path = rewritePath(path, f.Path, prefix) + } + code := http.StatusFound + if f.StatusCode != nil { + code = *f.StatusCode + } + return (&url.URL{Scheme: scheme, Host: host, Path: path, RawQuery: r.URL.RawQuery}).String(), code +} + +// rewritePath applies a path modifier: the whole path, or the matched prefix replaced. +func rewritePath(path string, m *HTTPPathModifier, prefix string) string { + switch m.Type { + case "ReplaceFullPath": + return str(m.ReplaceFullPath) + case "ReplacePrefixMatch": + rest := strings.TrimPrefix(path, strings.TrimSuffix(prefix, "/")) + out := strings.TrimSuffix(str(m.ReplacePrefixMatch), "/") + rest + if !strings.HasPrefix(out, "/") { + out = "/" + out + } + return out + } + return path +} + +func modifyHeaders(h http.Header, f *HTTPHeaderFilter) { + for _, s := range f.Set { + h.Set(s.Name, s.Value) + } + for _, a := range f.Add { + h.Add(a.Name, a.Value) + } + for _, name := range f.Remove { + h.Del(name) + } +} + +// hostOnly is a Host header without its port, lowercased (the spec ignores the port when matching). +func hostOnly(h string) string { + if host, _, err := net.SplitHostPort(h); err == nil { + h = host + } + return strings.ToLower(strings.TrimSuffix(h, ".")) +} diff --git a/lb/gateway_proxy_test.go b/lb/gateway_proxy_test.go new file mode 100644 index 00000000..20d63796 --- /dev/null +++ b/lb/gateway_proxy_test.go @@ -0,0 +1,314 @@ +package lb + +import ( + "bytes" + "crypto/tls" + "crypto/x509" + "fmt" + "io" + "net" + "net/http" + "net/http/httptest" + "strconv" + "strings" + "sync" + "testing" + "time" + + coreapi "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/types" +) + +// testProxy binds every requested address on loopback and remembers where. +type testProxy struct { + *Proxy + mu sync.Mutex + actual map[string]string +} + +func newTestProxy(t *testing.T) *testProxy { + tp := &testProxy{Proxy: NewProxy(), actual: map[string]string{}} + tp.listen = func(addr string) (net.Listener, error) { + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err == nil { + tp.mu.Lock() + tp.actual[addr] = ln.Addr().String() + tp.mu.Unlock() + } + return ln, err + } + t.Cleanup(tp.Close) + return tp +} + +func (tp *testProxy) at(port int32) string { + tp.mu.Lock() + defer tp.mu.Unlock() + return tp.actual[net.JoinHostPort("192.0.2.1", strconv.Itoa(int(port)))] +} + +// named is a backend that answers with its name and the request it saw. +func named(t *testing.T, name string) (*httptest.Server, Endpoint) { + s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("X-Backend", name) + n, _ := io.Copy(io.Discard, r.Body) + fmt.Fprintf(w, "%s path=%s host=%s xff=%s proto=%s xhost=%s realip=%s added=%s removed=%s body=%d", + name, r.URL.Path, r.Host, r.Header.Get("X-Forwarded-For"), r.Header.Get("X-Forwarded-Proto"), + r.Header.Get("X-Forwarded-Host"), r.Header.Get("X-Real-IP"), strings.Join(r.Header.Values("X-Added"), ","), + r.Header.Get("X-Remove-Me"), n) + })) + t.Cleanup(s.Close) + _, port, _ := net.SplitHostPort(s.Listener.Addr().String()) + p, _ := strconv.Atoi(port) + return s, Endpoint{IP: net.ParseIP("127.0.0.1").To4(), Port: int32(p)} +} + +func backend(name string, ep Endpoint) []ServeBackend { + return []ServeBackend{{Key: name, Weight: 1, Endpoints: []Endpoint{ep}}} +} + +func prefix(v string) HTTPRouteMatch { + return HTTPRouteMatch{Path: &HTTPPathMatch{Type: ptr(pathPrefix), Value: ptr(v)}} +} + +func exact(v string) HTTPRouteMatch { + return HTTPRouteMatch{Path: &HTTPPathMatch{Type: ptr(pathExact), Value: ptr(v)}} +} + +func get(t *testing.T, url, host string, hdr map[string]string) (int, string, http.Header) { + t.Helper() + req, _ := http.NewRequest("GET", url, nil) + if host != "" { + req.Host = host + } + for k, v := range hdr { + req.Header.Set(k, v) + } + c := &http.Client{CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }} + resp, err := c.Do(req) + if err != nil { + t.Fatalf("GET %s: %v", url, err) + } + defer resp.Body.Close() + b, _ := io.ReadAll(resp.Body) + return resp.StatusCode, string(b), resp.Header +} + +func TestProxyRoutesByPrecedence(t *testing.T) { + tp := newTestProxy(t) + _, a := named(t, "A") + _, b := named(t, "B") + _, c := named(t, "C") + _, d := named(t, "D") + _, e := named(t, "E") + t0 := time.Date(2026, 10, 14, 1, 0, 0, 0, time.UTC) + routes := []ServeRoute{ + {Key: "ns/first", Created: t0, Rules: []ServeRule{ + {Matches: []HTTPRouteMatch{exact("/exact")}, Backends: backend("A", a)}, + {Matches: []HTTPRouteMatch{prefix("/api")}, Backends: backend("B", b)}, + {Matches: []HTTPRouteMatch{prefix("/")}, Backends: backend("C", c)}, + }}, + {Key: "ns/second", Created: t0.Add(time.Minute), Rules: []ServeRule{ + {Matches: []HTTPRouteMatch{prefix("/api/v2")}, Backends: backend("D", d)}, + {Matches: []HTTPRouteMatch{{Path: &HTTPPathMatch{Type: ptr(pathPrefix), Value: ptr("/api")}, + Headers: []HTTPHeaderMatch{{Name: "X-Ver", Value: "2"}}}}, Backends: backend("E", e)}, + }}, + } + if f := tp.Apply([]ServeListener{{Gateway: "ns/gw", Name: "http", Address: net.ParseIP("192.0.2.1"), Port: 8080, Routes: routes}}); len(f) != 0 { + t.Fatal(f) + } + base := "http://" + tp.at(8080) + for path, want := range map[string]string{ + "/exact": "A", "/exactly": "C", "/api/x": "B", "/api": "B", "/api/v2/y": "D", "/apix": "C", "/": "C", + } { + if _, body, _ := get(t, base+path, "example.com", nil); !strings.HasPrefix(body, want+" ") { + t.Errorf("%s went to %q, want %s", path, body, want) + } + } + // same prefix length: the match with more headers takes it, though its route is younger + if _, body, _ := get(t, base+"/api/x", "example.com", map[string]string{"X-Ver": "2"}); !strings.HasPrefix(body, "E ") { + t.Errorf("header match lost: %q", body) + } +} + +func TestProxyHostPrecedence(t *testing.T) { + tp := newTestProxy(t) + _, w := named(t, "W") + _, x := named(t, "X") + routes := []ServeRoute{ + {Key: "ns/wild", Hostnames: []string{"*.example.com"}, Rules: []ServeRule{{Matches: []HTTPRouteMatch{prefix("/")}, Backends: backend("W", w)}}}, + {Key: "ns/exact", Hostnames: []string{"a.example.com"}, Rules: []ServeRule{{Matches: []HTTPRouteMatch{prefix("/")}, Backends: backend("X", x)}}}, + } + tp.Apply([]ServeListener{{Gateway: "ns/gw", Name: "http", Address: net.ParseIP("192.0.2.1"), Port: 8081, Routes: routes}}) + base := "http://" + tp.at(8081) + for host, want := range map[string]string{"a.example.com": "X", "A.Example.com:8081": "X", "b.example.com": "W"} { + if _, body, _ := get(t, base+"/", host, nil); !strings.HasPrefix(body, want+" ") { + t.Errorf("host %s went to %q, want %s", host, body, want) + } + } + if code, _, _ := get(t, base+"/", "example.org", nil); code != http.StatusNotFound { + t.Errorf("a host no route names: %d, want 404", code) + } +} + +func certSecret(t *testing.T, name, rv string) *coreapi.Secret { + s := tlsSecret(t, "gateway-system", strings.ReplaceAll(name, ".", "-"), name) + s.UID = types.UID("uid-" + name) + s.ResourceVersion = rv + return s +} + +func dialTLS(t *testing.T, addr, sni string) (*x509.Certificate, error) { + conn, err := tls.Dial("tcp", addr, &tls.Config{ServerName: sni, InsecureSkipVerify: true}) + if err != nil { + return nil, err + } + defer conn.Close() + return conn.ConnectionState().PeerCertificates[0], nil +} + +func TestProxyTLSBySNIAndSwapsCertificates(t *testing.T) { + tp := newTestProxy(t) + _, a := named(t, "A") + _, b := named(t, "B") + listeners := func(certA *coreapi.Secret) []ServeListener { + return []ServeListener{ + {Gateway: "ns/gw", Name: "a", Address: net.ParseIP("192.0.2.1"), Port: 8443, HTTPS: true, Hostname: "a.test", Cert: certA, + Routes: []ServeRoute{{Key: "ns/a", Rules: []ServeRule{{Matches: []HTTPRouteMatch{prefix("/")}, Backends: backend("A", a)}}}}}, + {Gateway: "ns/gw", Name: "b", Address: net.ParseIP("192.0.2.1"), Port: 8443, HTTPS: true, Hostname: "b.test", Cert: certSecret(t, "b.test", "1"), + Routes: []ServeRoute{{Key: "ns/b", Rules: []ServeRule{{Matches: []HTTPRouteMatch{prefix("/")}, Backends: backend("B", b)}}}}}, + } + } + first := certSecret(t, "a.test", "1") + if f := tp.Apply(listeners(first)); len(f) != 0 { + t.Fatal(f) + } + addr := tp.at(8443) + ca, err := dialTLS(t, addr, "a.test") + if err != nil || ca.Subject.CommonName != "a.test" { + t.Fatalf("SNI a.test: %v %v", ca, err) + } + cb, err := dialTLS(t, addr, "b.test") + if err != nil || cb.Subject.CommonName != "b.test" { + t.Fatalf("SNI b.test: %v %v", cb, err) + } + if _, err := dialTLS(t, addr, "c.test"); err == nil { + t.Fatal("a name no listener has must not get a certificate") + } + // requests go by listener, over HTTP/2 when the client offers it + client := &http.Client{Transport: &http.Transport{TLSClientConfig: &tls.Config{InsecureSkipVerify: true, ServerName: "b.test"}, ForceAttemptHTTP2: true}} + req, _ := http.NewRequest("GET", "https://"+addr+"/", nil) + req.Host = "b.test" + resp, err := client.Do(req) + if err != nil { + t.Fatal(err) + } + body, _ := io.ReadAll(resp.Body) + resp.Body.Close() + if !strings.HasPrefix(string(body), "B ") || !strings.Contains(string(body), "proto=https") || resp.ProtoMajor != 2 { + t.Fatalf("over HTTPS to b.test: %s %q", resp.Proto, body) + } + // a renewed Secret: new handshakes get the new certificate on the same, still-bound port + renewed := certSecret(t, "a.test", "2") + tp.Apply(listeners(renewed)) + if tp.at(8443) != addr { + t.Fatal("the port was rebound for a certificate change") + } + cn, err := dialTLS(t, addr, "a.test") + if err != nil || cn.SerialNumber.Cmp(ca.SerialNumber) == 0 { + t.Fatalf("still the old certificate after the swap: %v", err) + } +} + +func TestProxyRedirectsAndForwards(t *testing.T) { + tp := newTestProxy(t) + _, a := named(t, "A") + routes := []ServeRoute{{Key: "ns/r", Rules: []ServeRule{ + {Matches: []HTTPRouteMatch{prefix("/old")}, Filters: []HTTPRouteFilter{{Type: filterRequestRedirect, + RequestRedirect: &HTTPRequestRedirectFilter{Scheme: ptr("https"), StatusCode: ptr(301)}}}}, + {Matches: []HTTPRouteMatch{prefix("/moved")}, Filters: []HTTPRouteFilter{{Type: filterRequestRedirect, + RequestRedirect: &HTTPRequestRedirectFilter{Path: &HTTPPathModifier{Type: "ReplacePrefixMatch", ReplacePrefixMatch: ptr("/new")}}}}}, + {Matches: []HTTPRouteMatch{prefix("/v1")}, Backends: backend("A", a), Filters: []HTTPRouteFilter{ + {Type: filterRequestHeaderModifier, RequestHeaderModifier: &HTTPHeaderFilter{ + Set: []HTTPHeader{{Name: "X-Set", Value: "s"}}, Add: []HTTPHeader{{Name: "X-Added", Value: "two"}}, Remove: []string{"X-Remove-Me"}}}, + {Type: filterResponseHeaderModifier, ResponseHeaderModifier: &HTTPHeaderFilter{Set: []HTTPHeader{{Name: "X-Resp", Value: "r"}}}}, + {Type: filterURLRewrite, URLRewrite: &HTTPURLRewriteFilter{Path: &HTTPPathModifier{Type: "ReplacePrefixMatch", ReplacePrefixMatch: ptr("/v2")}}}, + }}, + {Matches: []HTTPRouteMatch{prefix("/")}, Backends: backend("A", a)}, + }}} + tp.Apply([]ServeListener{{Gateway: "ns/gw", Name: "http", Address: net.ParseIP("192.0.2.1"), Port: 80, Routes: routes}}) + base := "http://" + tp.at(80) + code, _, h := get(t, base+"/old/page?q=1", "git.levk.dev", nil) + if code != http.StatusMovedPermanently || h.Get("Location") != "https://git.levk.dev/old/page?q=1" { + t.Fatalf("scheme redirect: %d %q", code, h.Get("Location")) + } + code, _, h = get(t, base+"/moved/x", "git.levk.dev", nil) + if code != http.StatusFound || h.Get("Location") != "http://git.levk.dev/new/x" { + t.Fatalf("prefix redirect: %d %q", code, h.Get("Location")) + } + // the client's own claims about who it is are not passed on; the backend sees this hop's view + _, body, _ := get(t, base+"/", "git.levk.dev", map[string]string{"X-Forwarded-For": "6.6.6.6", "X-Real-IP": "6.6.6.6"}) + if !strings.Contains(body, "host=git.levk.dev") || !strings.Contains(body, "xff=127.0.0.1 ") || + !strings.Contains(body, "proto=http ") || !strings.Contains(body, "xhost=git.levk.dev") || !strings.Contains(body, "realip=127.0.0.1") { + t.Fatalf("forwarded headers: %q", body) + } + _, body, h = get(t, base+"/v1/thing", "git.levk.dev", map[string]string{"X-Added": "one", "X-Remove-Me": "x"}) + if !strings.Contains(body, "path=/v2/thing") || !strings.Contains(body, "added=one,two") || !strings.Contains(body, "removed= ") || h.Get("X-Resp") != "r" { + t.Fatalf("filters: %q %v", body, h) + } +} + +func TestProxyAnswersWhatItCannotRoute(t *testing.T) { + tp := newTestProxy(t) + routes := []ServeRoute{{Key: "ns/r", Rules: []ServeRule{ + {Matches: []HTTPRouteMatch{prefix("/broken")}, Broken: true}, + {Matches: []HTTPRouteMatch{prefix("/empty")}, Backends: []ServeBackend{{Key: "ns/svc:80", Weight: 1}}}, + }}} + tp.Apply([]ServeListener{{Gateway: "ns/gw", Name: "http", Address: net.ParseIP("192.0.2.1"), Port: 8082, Routes: routes}}) + base := "http://" + tp.at(8082) + for path, want := range map[string]int{"/broken": 500, "/empty": 503, "/nothing": 404} { + if code, _, _ := get(t, base+path, "x", nil); code != want { + t.Errorf("%s: %d, want %d", path, code, want) + } + } +} + +func TestProxyStreamsLargeBodies(t *testing.T) { + tp := newTestProxy(t) + _, a := named(t, "A") + tp.Apply([]ServeListener{{Gateway: "ns/gw", Name: "http", Address: net.ParseIP("192.0.2.1"), Port: 8083, + Routes: []ServeRoute{{Key: "ns/r", Rules: []ServeRule{{Matches: []HTTPRouteMatch{prefix("/")}, Backends: backend("A", a)}}}}}}) + const size = 32 << 20 + resp, err := http.Post("http://"+tp.at(8083)+"/upload", "application/octet-stream", bytes.NewReader(make([]byte, size))) + if err != nil { + t.Fatal(err) + } + body, _ := io.ReadAll(resp.Body) + resp.Body.Close() + if !strings.Contains(string(body), fmt.Sprintf("body=%d", size)) { + t.Fatalf("upload: %q", body) + } +} + +func TestProxyBindFailureAndClose(t *testing.T) { + tp := newTestProxy(t) + tp.listen = func(addr string) (net.Listener, error) { return nil, fmt.Errorf("address in use") } + f := tp.Apply([]ServeListener{{Gateway: "ns/gw", Name: "http", Address: net.ParseIP("192.0.2.1"), Port: 8084}}) + if f["ns/gw/http"] == nil { + t.Fatalf("bind failure not reported: %v", f) + } + tp2 := newTestProxy(t) + tp2.Apply([]ServeListener{{Gateway: "ns/gw", Name: "http", Address: net.ParseIP("192.0.2.1"), Port: 8085}}) + addr := tp2.at(8085) + tp2.Apply(nil) + deadline := time.Now().Add(5 * time.Second) + for time.Now().Before(deadline) { + c, err := net.DialTimeout("tcp", addr, 200*time.Millisecond) + if err != nil { + return + } + c.Close() + time.Sleep(50 * time.Millisecond) + } + t.Fatal("a port no longer served still accepts connections") +} diff --git a/lb/gateway_test.go b/lb/gateway_test.go new file mode 100644 index 00000000..e1176ef1 --- /dev/null +++ b/lb/gateway_test.go @@ -0,0 +1,576 @@ +package lb + +import ( + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/x509" + "crypto/x509/pkix" + "encoding/pem" + "math/big" + "net" + "strings" + "testing" + "time" + + coreapi "k8s.io/api/core/v1" + discovery "k8s.io/api/discovery/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" +) + +var gwNow = metav1.NewTime(time.Date(2026, 10, 14, 4, 0, 0, 0, time.UTC)) + +// selfSigned is a certificate and key for the given names, PEM-encoded. +func selfSigned(t *testing.T, names ...string) ([]byte, []byte) { + t.Helper() + key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + if err != nil { + t.Fatal(err) + } + tmpl := &x509.Certificate{ + SerialNumber: big.NewInt(time.Now().UnixNano()), Subject: pkix.Name{CommonName: names[0]}, DNSNames: names, + NotBefore: time.Now().Add(-time.Hour), NotAfter: time.Now().Add(24 * time.Hour), + KeyUsage: x509.KeyUsageDigitalSignature, ExtKeyUsage: []x509.ExtKeyUsage{x509.ExtKeyUsageServerAuth}, + } + der, err := x509.CreateCertificate(rand.Reader, tmpl, tmpl, &key.PublicKey, key) + if err != nil { + t.Fatal(err) + } + kb, _ := x509.MarshalECPrivateKey(key) + return pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der}), pem.EncodeToMemory(&pem.Block{Type: "EC PRIVATE KEY", Bytes: kb}) +} + +func tlsSecret(t *testing.T, namespace, name string, names ...string) *coreapi.Secret { + crt, key := selfSigned(t, names...) + return &coreapi.Secret{ObjectMeta: metav1.ObjectMeta{Namespace: namespace, Name: name}, Type: coreapi.SecretTypeTLS, + Data: map[string][]byte{coreapi.TLSCertKey: crt, coreapi.TLSPrivateKeyKey: key}} +} + +func gwNodes() []Node { + return []Node{ + {Name: "cp", Addresses: []net.IP{net.ParseIP("86.107.168.24").To4()}, WeaveIP: net.ParseIP("10.32.0.1").To4(), WeaveMAC: mustMAC("02:00:00:00:00:01")}, + {Name: "w15", Addresses: []net.IP{net.ParseIP("86.107.168.15").To4()}, WeaveIP: net.ParseIP("10.44.0.0").To4(), WeaveMAC: mustMAC("02:00:00:00:00:02")}, + } +} + +func mustMAC(s string) net.HardwareAddr { + m, _ := net.ParseMAC(s) + return m +} + +func ourClass() *GatewayClass { + return &GatewayClass{ObjectMeta: metav1.ObjectMeta{Name: "reweave", Generation: 1}, Spec: GatewayClassSpec{ControllerName: GatewayControllerName}} +} + +// giteaGateway is the shape Gitea's documentation prescribes: HTTPS with a certificate, plain HTTP, and SSH over TCP. +func giteaGateway() *Gateway { + return &Gateway{ + ObjectMeta: metav1.ObjectMeta{Namespace: "gateway-system", Name: "shared", Generation: 3}, + Spec: GatewaySpec{ + GatewayClassName: "reweave", + Addresses: []GatewayAddress{{Value: "86.107.168.15"}}, + Listeners: []Listener{ + {Name: "https-gitea", Port: 443, Protocol: ProtocolHTTPS, Hostname: ptr("git.levk.dev"), + TLS: &ListenerTLSConfig{CertificateRefs: []SecretObjectReference{{Name: "git-levk-dev"}}}, + AllowedRoutes: &AllowedRoutes{Namespaces: &RouteNamespaces{From: ptr("Selector"), Selector: &metav1.LabelSelector{MatchLabels: map[string]string{"kubernetes.io/metadata.name": "gitea"}}}}}, + {Name: "http", Port: 80, Protocol: ProtocolHTTP, Hostname: ptr("git.levk.dev"), + AllowedRoutes: &AllowedRoutes{Namespaces: &RouteNamespaces{From: ptr("All")}}}, + {Name: "ssh", Port: 22, Protocol: ProtocolTCP, + AllowedRoutes: &AllowedRoutes{Namespaces: &RouteNamespaces{From: ptr("All")}}}, + }, + }, + } +} + +func giteaService() *coreapi.Service { + return &coreapi.Service{ObjectMeta: metav1.ObjectMeta{Namespace: "gitea", Name: "gitea"}, Spec: coreapi.ServiceSpec{ + Ports: []coreapi.ServicePort{{Name: "http", Port: 3000}, {Name: "ssh", Port: 22}}, + }} +} + +func giteaSlice() *discovery.EndpointSlice { + return &discovery.EndpointSlice{ + ObjectMeta: metav1.ObjectMeta{Namespace: "gitea", Name: "gitea-abc", Labels: map[string]string{serviceNameLabel: "gitea"}}, + AddressType: discovery.AddressTypeIPv4, + Ports: []discovery.EndpointPort{{Name: ptr("http"), Port: ptr(int32(3000))}, {Name: ptr("ssh"), Port: ptr(int32(22))}}, + Endpoints: []discovery.Endpoint{{Addresses: []string{"10.44.0.12"}, Conditions: discovery.EndpointConditions{Ready: ptr(true)}}}, + } +} + +func giteaRoutes() ([]*HTTPRoute, []*TCPRoute) { + t0 := metav1.NewTime(time.Date(2026, 10, 14, 1, 0, 0, 0, time.UTC)) + https := &HTTPRoute{ + ObjectMeta: metav1.ObjectMeta{Namespace: "gitea", Name: "gitea", Generation: 1, CreationTimestamp: t0}, + Spec: HTTPRouteSpec{ + ParentRefs: []ParentReference{{Namespace: ptr("gateway-system"), Name: "shared", SectionName: ptr("https-gitea")}}, + Hostnames: []string{"git.levk.dev"}, + Rules: []HTTPRouteRule{{BackendRefs: []HTTPBackendRef{{BackendRef: BackendRef{Name: "gitea", Port: ptr(int32(3000))}}}}}, + }, + } + redirect := &HTTPRoute{ + ObjectMeta: metav1.ObjectMeta{Namespace: "gitea", Name: "gitea-redirect", Generation: 1, CreationTimestamp: t0}, + Spec: HTTPRouteSpec{ + ParentRefs: []ParentReference{{Namespace: ptr("gateway-system"), Name: "shared", SectionName: ptr("http")}}, + Rules: []HTTPRouteRule{{Filters: []HTTPRouteFilter{{Type: filterRequestRedirect, + RequestRedirect: &HTTPRequestRedirectFilter{Scheme: ptr("https"), StatusCode: ptr(301)}}}}}, + }, + } + ssh := &TCPRoute{ + ObjectMeta: metav1.ObjectMeta{Namespace: "gitea", Name: "gitea-ssh", Generation: 1, CreationTimestamp: t0}, + Spec: TCPRouteSpec{ + ParentRefs: []ParentReference{{Namespace: ptr("gateway-system"), Name: "shared", SectionName: ptr("ssh")}}, + Rules: []TCPRouteRule{{BackendRefs: []BackendRef{{Name: "gitea", Port: ptr(int32(22))}}}}, + }, + } + return []*HTTPRoute{https, redirect}, []*TCPRoute{ssh} +} + +func giteaInput(t *testing.T) GatewayInput { + hr, tr := giteaRoutes() + return GatewayInput{ + Classes: []*GatewayClass{ourClass()}, Gateways: []*Gateway{giteaGateway()}, HTTPRoutes: hr, TCPRoutes: tr, + Namespaces: []*coreapi.Namespace{ + {ObjectMeta: metav1.ObjectMeta{Name: "gitea", Labels: map[string]string{"kubernetes.io/metadata.name": "gitea"}}}, + {ObjectMeta: metav1.ObjectMeta{Name: "gateway-system", Labels: map[string]string{"kubernetes.io/metadata.name": "gateway-system"}}}, + }, + Services: []*coreapi.Service{giteaService()}, Slices: []*discovery.EndpointSlice{giteaSlice()}, + Secrets: []*coreapi.Secret{tlsSecret(t, "gateway-system", "git-levk-dev", "git.levk.dev")}, + SecretNamespaces: map[string]bool{"gateway-system": true}, + Now: gwNow, + } +} + +func condOf(cs []metav1.Condition, typ string) metav1.Condition { + for _, c := range cs { + if c.Type == typ { + return c + } + } + return metav1.Condition{} +} + +func TestGatewayServedByItsHolder(t *testing.T) { + in := giteaInput(t) + d := ComputeGateways("w15", gwNodes(), in) + if len(d.Problems) != 0 { + t.Fatalf("problems: %v", d.Problems) + } + // HTTPS and HTTP served by the agent; SSH forwarded in the kernel to Gitea's pod with the client address intact + if len(d.Serve) != 2 { + t.Fatalf("want the https and http listeners served, got %+v", d.Serve) + } + https := d.Serve[0] + if !https.HTTPS || https.Port != 443 || https.Hostname != "git.levk.dev" || https.Cert == nil || len(https.Routes) != 1 { + t.Fatalf("https listener: %+v", https) + } + be := https.Routes[0].Rules[0].Backends + if len(be) != 1 || be[0].Key != "gitea/gitea:3000" || len(be[0].Endpoints) != 1 || be[0].Endpoints[0].String() != "10.44.0.12:3000" { + t.Fatalf("https backends: %+v", be) + } + if len(d.Forwards) != 1 || d.Forwards[0].Ports[0].Port != 22 || d.Forwards[0].Ports[0].Endpoints[0].String() != "10.44.0.12:22" { + t.Fatalf("ssh forward: %+v", d.Forwards) + } + if len(d.Holding) != 1 || d.Holding[0] != "w15" { + t.Fatalf("holding: %v", d.Holding) + } + // the holder writes the Gateway's status: accepted, programmed, its address, each listener with one route + if len(d.Gateways) != 1 { + t.Fatalf("gateway status writes: %+v", d.Gateways) + } + st := d.Gateways[0].Status + if c := condOf(st.Conditions, condProgrammed); c.Status != metav1.ConditionTrue || c.ObservedGeneration != 3 { + t.Fatalf("programmed: %+v", c) + } + if len(st.Addresses) != 1 || st.Addresses[0].Value != "86.107.168.15" { + t.Fatalf("addresses: %+v", st.Addresses) + } + for _, l := range st.Listeners { + if l.AttachedRoutes != 1 || condOf(l.Conditions, condProgrammed).Status != metav1.ConditionTrue { + t.Fatalf("listener %s: %+v", l.Name, l) + } + } + // route statuses come from the first node by name, not the holder + if len(d.Routes) != 0 || len(d.Classes) != 0 { + t.Fatalf("w15 wrote route or class statuses: %+v %+v", d.Routes, d.Classes) + } +} + +func TestGatewayElsewhere(t *testing.T) { + in := giteaInput(t) + d := ComputeGateways("cp", gwNodes(), in) + if len(d.Serve) != 0 || len(d.Forwards) != 0 || len(d.Gateways) != 0 { + t.Fatalf("cp serves or writes what w15 holds: %+v %+v %+v", d.Serve, d.Forwards, d.Gateways) + } + // but steers replies of w15's forwards back to it, and writes class and route statuses (first by name) + if len(d.Holding) != 1 || d.Holding[0] != "w15" { + t.Fatalf("holding: %v", d.Holding) + } + if len(d.Classes) != 1 || condOf(d.Classes[0].Status.Conditions, condAccepted).Status != metav1.ConditionTrue { + t.Fatalf("class status: %+v", d.Classes) + } + if len(d.Routes) != 3 { + t.Fatalf("want 3 route statuses, got %+v", d.Routes) + } + for _, r := range d.Routes { + if len(r.Status.Parents) != 1 || condOf(r.Status.Parents[0].Conditions, condAccepted).Status != metav1.ConditionTrue || + condOf(r.Status.Parents[0].Conditions, condResolvedRefs).Status != metav1.ConditionTrue { + t.Fatalf("route %s/%s status: %+v", r.Namespace, r.Name, r.Status) + } + } +} + +func TestGatewayStatusIsWrittenOnce(t *testing.T) { + in := giteaInput(t) + d := ComputeGateways("w15", gwNodes(), in) + in.Gateways[0].Status = d.Gateways[0].Status + in.Now = metav1.NewTime(gwNow.Add(time.Hour)) // later passes see the status they wrote + if again := ComputeGateways("w15", gwNodes(), in); len(again.Gateways) != 0 { + t.Fatalf("unchanged status rewritten: %+v", again.Gateways) + } + c := ComputeGateways("cp", gwNodes(), in) + for _, r := range c.Routes { + for _, hr := range in.HTTPRoutes { + if hr.Name == r.Name { + hr.Status = r.Status + } + } + for _, tr := range in.TCPRoutes { + if tr.Name == r.Name { + tr.Status = r.Status + } + } + } + in.Classes[0].Status = c.Classes[0].Status + if again := ComputeGateways("cp", gwNodes(), in); len(again.Routes) != 0 || len(again.Classes) != 0 { + t.Fatalf("unchanged route or class status rewritten: %+v %+v", again.Routes, again.Classes) + } +} + +func TestGatewayFailsClosedWithoutHolder(t *testing.T) { + in := giteaInput(t) + in.Gateways[0].Spec.Addresses[0].Value = "198.51.100.7" + for _, self := range []string{"cp", "w15"} { + d := ComputeGateways(self, gwNodes(), in) + if len(d.Serve) != 0 || len(d.Forwards) != 0 || len(d.Holding) != 0 { + t.Fatalf("%s serves an address nobody holds: %+v", self, d) + } + } + d := ComputeGateways("cp", gwNodes(), in) + if len(d.Gateways) != 1 { + t.Fatalf("the first node by name must write the failure: %+v", d.Gateways) + } + c := condOf(d.Gateways[0].Status.Conditions, condProgrammed) + if c.Status != metav1.ConditionFalse || c.Reason != "AddressNotUsable" || len(d.Gateways[0].Status.Addresses) != 0 { + t.Fatalf("status: %+v", d.Gateways[0].Status) + } +} + +func TestGatewayAddressMustBeOneIPv4(t *testing.T) { + in := giteaInput(t) + in.Gateways[0].Spec.Addresses = nil + d := ComputeGateways("cp", gwNodes(), in) + if c := condOf(d.Gateways[0].Status.Conditions, condAccepted); c.Status != metav1.ConditionFalse || c.Reason != "UnsupportedAddress" { + t.Fatalf("accepted: %+v", c) + } +} + +func TestListenerCertificateReferences(t *testing.T) { + cases := map[string]struct { + mutate func(in *GatewayInput) + reason string + }{ + "missing Secret": {func(in *GatewayInput) { in.Secrets = nil }, "InvalidCertificateRef"}, + "other namespace without grant": {func(in *GatewayInput) { + in.Gateways[0].Spec.Listeners[0].TLS.CertificateRefs[0].Namespace = ptr("gitea") + in.SecretNamespaces["gitea"] = true + }, "RefNotPermitted"}, + "namespace the agent cannot read": {func(in *GatewayInput) { + in.SecretNamespaces = map[string]bool{} + }, "RefNotPermitted"}, + "not a certificate": {func(in *GatewayInput) { + in.Secrets[0].Data[coreapi.TLSCertKey] = []byte("nope") + }, "InvalidCertificateRef"}, + } + for name, c := range cases { + in := giteaInput(t) + c.mutate(&in) + d := ComputeGateways("w15", gwNodes(), in) + var https ListenerStatus + for _, l := range d.Gateways[0].Status.Listeners { + if l.Name == "https-gitea" { + https = l + } + } + if r := condOf(https.Conditions, condResolvedRefs); r.Status != metav1.ConditionFalse || r.Reason != c.reason { + t.Errorf("%s: resolvedRefs %+v", name, r) + } + if p := condOf(https.Conditions, condProgrammed); p.Status != metav1.ConditionFalse { + t.Errorf("%s: an HTTPS listener without its certificate must not be programmed", name) + } + for _, s := range d.Serve { + if s.HTTPS { + t.Errorf("%s: HTTPS listener served without a certificate", name) + } + } + if a := condOf(d.Gateways[0].Status.Conditions, condAccepted); a.Reason != "ListenersNotValid" || a.Status != metav1.ConditionTrue { + t.Errorf("%s: gateway accepted %+v", name, a) + } + } +} + +func TestCrossNamespaceCertificateWithGrant(t *testing.T) { + in := giteaInput(t) + in.Gateways[0].Spec.Listeners[0].TLS.CertificateRefs[0].Namespace = ptr("gitea") + in.Secrets = []*coreapi.Secret{tlsSecret(t, "gitea", "git-levk-dev", "git.levk.dev")} + in.SecretNamespaces = map[string]bool{"gitea": true} + in.Grants = []*ReferenceGrant{{ObjectMeta: metav1.ObjectMeta{Namespace: "gitea", Name: "gw"}, Spec: ReferenceGrantSpec{ + From: []ReferenceGrantFrom{{Group: GatewayGroup, Kind: kindGateway, Namespace: "gateway-system"}}, + To: []ReferenceGrantTo{{Group: "", Kind: kindSecret}}, + }}} + d := ComputeGateways("w15", gwNodes(), in) + if len(d.Serve) != 2 || d.Serve[0].Cert == nil { + t.Fatalf("a granted certificate in another namespace must be served: %+v", d.Serve) + } +} + +func TestListenerConflicts(t *testing.T) { + in := giteaInput(t) + l := in.Gateways[0].Spec.Listeners + in.Gateways[0].Spec.Listeners = append(l, + Listener{Name: "ssh2", Port: 22, Protocol: ProtocolTCP}, // two TCP listeners on 22 + Listener{Name: "http-on-443", Port: 443, Protocol: ProtocolHTTP}, // HTTP and HTTPS on 443 + ) + d := ComputeGateways("w15", gwNodes(), in) + want := map[string]string{"ssh": "HostnameConflict", "ssh2": "HostnameConflict", "https-gitea": "ProtocolConflict", "http-on-443": "ProtocolConflict", "http": ""} + for _, ls := range d.Gateways[0].Status.Listeners { + c := condOf(ls.Conditions, condConflicted) + if want[ls.Name] == "" { + if c.Status != metav1.ConditionFalse { + t.Errorf("%s: conflicted %+v", ls.Name, c) + } + continue + } + if c.Status != metav1.ConditionTrue || c.Reason != want[ls.Name] { + t.Errorf("%s: want conflicted %s, got %+v", ls.Name, want[ls.Name], c) + } + } + if len(d.Forwards) != 0 { + t.Fatalf("conflicted TCP listeners forwarded: %+v", d.Forwards) + } + if len(d.Serve) != 1 || d.Serve[0].Name != "http" { + t.Fatalf("only the plain http listener on 80 is unconflicted: %+v", d.Serve) + } +} + +func TestListenerPortTakenByLoadBalancerService(t *testing.T) { + in := giteaInput(t) + class := Class + in.Services = append(in.Services, &coreapi.Service{ObjectMeta: metav1.ObjectMeta{Namespace: "mail", Name: "smtp"}, Spec: coreapi.ServiceSpec{ + Type: coreapi.ServiceTypeLoadBalancer, LoadBalancerClass: &class, LoadBalancerIP: "86.107.168.15", + Ports: []coreapi.ServicePort{{Port: 22}}, + }}) + d := ComputeGateways("w15", gwNodes(), in) + for _, ls := range d.Gateways[0].Status.Listeners { + if ls.Name == "ssh" { + if c := condOf(ls.Conditions, condAccepted); c.Status != metav1.ConditionFalse || c.Reason != "PortUnavailable" { + t.Fatalf("ssh accepted: %+v", c) + } + } + } + if len(d.Forwards) != 0 { + t.Fatalf("a listener on a port a Service forwards must not be forwarded: %+v", d.Forwards) + } +} + +func TestRouteAttachment(t *testing.T) { + cases := map[string]struct { + mutate func(in *GatewayInput) + route string + reason string + }{ + "unknown section": {func(in *GatewayInput) { in.HTTPRoutes[0].Spec.ParentRefs[0].SectionName = ptr("nope") }, "gitea", "NoMatchingParent"}, + "namespace not selected": {func(in *GatewayInput) { in.Namespaces[0].Labels = nil }, "gitea", "NotAllowedByListeners"}, + "hostname outside the listener's": {func(in *GatewayInput) { in.HTTPRoutes[0].Spec.Hostnames = []string{"git.example.net"} }, "gitea", "NoMatchingListenerHostname"}, + "unsupported filter": {func(in *GatewayInput) { + in.HTTPRoutes[0].Spec.Rules[0].Filters = []HTTPRouteFilter{{Type: "RequestMirror"}} + }, "gitea", "UnsupportedValue"}, + "regular expression path": {func(in *GatewayInput) { + in.HTTPRoutes[0].Spec.Rules[0].Matches = []HTTPRouteMatch{{Path: &HTTPPathMatch{Type: ptr("RegularExpression"), Value: ptr("/.*")}}} + }, "gitea", "UnsupportedValue"}, + "wrong port": {func(in *GatewayInput) { in.TCPRoutes[0].Spec.ParentRefs[0].Port = ptr(int32(2222)) }, "gitea-ssh", "NoMatchingParent"}, + } + for name, c := range cases { + in := giteaInput(t) + c.mutate(&in) + d := ComputeGateways("cp", gwNodes(), in) + found := false + for _, r := range d.Routes { + if r.Name != c.route { + continue + } + found = true + a := condOf(r.Status.Parents[0].Conditions, condAccepted) + if a.Status != metav1.ConditionFalse || a.Reason != c.reason { + t.Errorf("%s: accepted %+v", name, a) + } + } + if !found { + t.Errorf("%s: no status for route %s", name, c.route) + } + h := ComputeGateways("w15", gwNodes(), in) + if c.route == "gitea" { + for _, s := range h.Serve { + if s.Name == "https-gitea" && len(s.Routes) != 0 { + t.Errorf("%s: a route that is not accepted was attached", name) + } + } + } + } +} + +func TestBackendNotFoundBreaksTheRule(t *testing.T) { + in := giteaInput(t) + in.HTTPRoutes[0].Spec.Rules[0].BackendRefs[0].Name = "nope" + d := ComputeGateways("cp", gwNodes(), in) + for _, r := range d.Routes { + if r.Name == "gitea" { + if c := condOf(r.Status.Parents[0].Conditions, condResolvedRefs); c.Status != metav1.ConditionFalse || c.Reason != "BackendNotFound" { + t.Fatalf("resolvedRefs: %+v", c) + } + if c := condOf(r.Status.Parents[0].Conditions, condAccepted); c.Status != metav1.ConditionTrue { + t.Fatalf("a route with an unresolved backend is still accepted: %+v", c) + } + } + } + h := ComputeGateways("w15", gwNodes(), in) + for _, s := range h.Serve { + if s.Name == "https-gitea" && (len(s.Routes) != 1 || !s.Routes[0].Rules[0].Broken) { + t.Fatalf("the rule must be served broken (500): %+v", s.Routes) + } + } +} + +func TestTCPListenerForwardsTheOldestRoute(t *testing.T) { + in := giteaInput(t) + later := in.TCPRoutes[0].DeepCopyObject().(*TCPRoute) + later.Name = "other-ssh" + later.CreationTimestamp = metav1.NewTime(later.CreationTimestamp.Add(time.Minute)) + later.Spec.Rules[0].BackendRefs[0].Name = "other" + in.TCPRoutes = append(in.TCPRoutes, later) + d := ComputeGateways("cp", gwNodes(), in) + for _, r := range d.Routes { + a := condOf(r.Status.Parents[0].Conditions, condAccepted) + switch r.Name { + case "gitea-ssh": + if a.Status != metav1.ConditionTrue { + t.Fatalf("oldest route: %+v", a) + } + case "other-ssh": + if a.Status != metav1.ConditionFalse || !strings.Contains(a.Message, "gitea/gitea-ssh") { + t.Fatalf("second route: %+v", a) + } + } + } +} + +func TestOtherControllersParentsKept(t *testing.T) { + in := giteaInput(t) + theirs := RouteParentStatus{ParentRef: ParentReference{Name: "elsewhere"}, ControllerName: "example.com/other", + Conditions: []metav1.Condition{{Type: condAccepted, Status: metav1.ConditionTrue, Reason: "Accepted", LastTransitionTime: gwNow}}} + in.HTTPRoutes[0].Status.Parents = []RouteParentStatus{theirs} + d := ComputeGateways("cp", gwNodes(), in) + for _, r := range d.Routes { + if r.Name == "gitea" { + if len(r.Status.Parents) != 2 || r.Status.Parents[0].ControllerName != "example.com/other" { + t.Fatalf("another controller's entry must be kept: %+v", r.Status.Parents) + } + } + } +} + +func TestOtherClassesIgnored(t *testing.T) { + in := giteaInput(t) + in.Classes[0].Spec.ControllerName = "example.com/other" + for _, self := range []string{"cp", "w15"} { + d := ComputeGateways(self, gwNodes(), in) + if len(d.Serve)+len(d.Forwards)+len(d.Gateways)+len(d.Routes)+len(d.Classes) != 0 { + t.Fatalf("%s acted on another controller's class: %+v", self, d) + } + } +} + +func TestNarrowHostnames(t *testing.T) { + cases := []struct { + listener string + route []string + want []string + ok bool + }{ + {"", nil, nil, true}, + {"git.levk.dev", nil, []string{"git.levk.dev"}, true}, + {"", []string{"a.example.com"}, []string{"a.example.com"}, true}, + {"test.example.com", []string{"*.example.com"}, []string{"test.example.com"}, true}, + {"*.example.com", []string{"test.example.com", "test.example.net", "foo.test.example.com"}, []string{"test.example.com", "foo.test.example.com"}, true}, + {"*.example.com", []string{"example.com"}, nil, false}, + {"*.example.com", []string{"*.a.example.com"}, []string{"*.a.example.com"}, true}, + {"*.a.example.com", []string{"*.example.com"}, []string{"*.a.example.com"}, true}, + {"git.levk.dev", []string{"git.example.net"}, nil, false}, + } + for _, c := range cases { + got, ok := narrowHostnames(c.listener, c.route) + if ok != c.ok || strings.Join(got, ",") != strings.Join(c.want, ",") { + t.Errorf("listener %q route %v: got %v %v, want %v %v", c.listener, c.route, got, ok, c.want, c.ok) + } + } +} + +func TestMergeGateways(t *testing.T) { + in := giteaInput(t) + nodes := gwNodes() + // on the holder: Gitea's SSH forwarded, its HTTP(S) ports let in + d := MergeGateways(Compute("w15", nodes, nil, nil), ComputeGateways("w15", nodes, in), "w15", nodes) + if len(d.Forwards) != 1 || len(d.Accept) != 2 || len(d.Holders) != 0 { + t.Fatalf("holder: forwards %+v accept %+v holders %+v", d.Forwards, d.Accept, d.Holders) + } + r := Render(d, EgressDesired{}, "w15", &net.IPNet{IP: net.ParseIP("10.32.0.0"), Mask: net.CIDRMask(12, 32)}, nil) + for _, want := range []string{ + "-A REWEAVE-LB-IN -d 86.107.168.15/32 -p tcp -m tcp --dport 443", + "-A REWEAVE-LB-IN -d 86.107.168.15/32 -p tcp -m tcp --dport 80", + } { + if !strings.Contains(r.Filter, want) { + t.Errorf("filter lacks %q:\n%s", want, r.Filter) + } + } + if !strings.Contains(r.Nat, "-A REWEAVE-LB -d 86.107.168.15/32 -p tcp -m tcp --dport 22") || !strings.Contains(r.Nat, "--to-destination 10.44.0.12:22") { + t.Errorf("nat lacks the SSH forward:\n%s", r.Nat) + } + // elsewhere: w15 is a holder whose replies come back to it + e := MergeGateways(Compute("cp", nodes, nil, nil), ComputeGateways("cp", nodes, in), "cp", nodes) + if len(e.Forwards) != 0 || len(e.Accept) != 0 || len(e.Holders) != 1 || e.Holders[0].Node.Name != "w15" { + t.Fatalf("cp: forwards %+v accept %+v holders %+v", e.Forwards, e.Accept, e.Holders) + } +} + +func TestUnboundListenerReported(t *testing.T) { + in := giteaInput(t) + in.Unbound = map[string]string{"gateway-system/shared/https-gitea": "address already in use"} + d := ComputeGateways("w15", gwNodes(), in) + for _, l := range d.Gateways[0].Status.Listeners { + if l.Name != "https-gitea" { + continue + } + if a := condOf(l.Conditions, condAccepted); a.Status != metav1.ConditionFalse || a.Reason != "PortUnavailable" { + t.Fatalf("accepted: %+v", a) + } + if p := condOf(l.Conditions, condProgrammed); p.Status != metav1.ConditionFalse { + t.Fatalf("programmed: %+v", p) + } + } + served := false + for _, s := range d.Serve { + served = served || s.Name == "https-gitea" + } + if !served { + t.Fatal("an unbound listener is still tried on every pass") + } +} diff --git a/lb/model.go b/lb/model.go index df208711..c25c6fe0 100644 --- a/lb/model.go +++ b/lb/model.go @@ -136,9 +136,19 @@ type Desired struct { Forwards []Forward Holders []Holder Status []Status + // Accept are the address:ports this node serves itself (Gateways' HTTP and HTTPS listeners), let in ahead of + // the host firewall + Accept []Accept Problems []string // what an operator should hear about (logged and evented by the controller) } +// Accept is a TCP address:port this node serves itself. +type Accept struct { + Address net.IP + Port int32 + Comment string +} + // eligible reports whether the Service asks for this agent. func eligible(s *coreapi.Service) bool { return s.Spec.Type == coreapi.ServiceTypeLoadBalancer && s.Spec.LoadBalancerClass != nil && diff --git a/prog/weave-kube/weave-daemonset-k8s-1.11.yaml b/prog/weave-kube/weave-daemonset-k8s-1.11.yaml index e1de7041..84faf1ce 100644 --- a/prog/weave-kube/weave-daemonset-k8s-1.11.yaml +++ b/prog/weave-kube/weave-daemonset-k8s-1.11.yaml @@ -154,6 +154,28 @@ items: - egressservices/status verbs: - patch + # Gateways of controller reweave.levk.dev/gateway (the Gateway API CRDs are installed separately) + - apiGroups: + - gateway.networking.k8s.io + resources: + - gatewayclasses + - gateways + - httproutes + - tcproutes + - referencegrants + verbs: + - get + - list + - watch + - apiGroups: + - gateway.networking.k8s.io + resources: + - gatewayclasses/status + - gateways/status + - httproutes/status + - tcproutes/status + verbs: + - patch - apiVersion: rbac.authorization.k8s.io/v1 kind: ClusterRoleBinding metadata: @@ -206,6 +228,42 @@ items: - kind: ServiceAccount name: weave-net namespace: kube-system + # Gateways' certificates: the agent reads Secrets in gateway-system only (weave-npc --gateway-namespaces) + - apiVersion: v1 + kind: Namespace + metadata: + name: gateway-system + - apiVersion: rbac.authorization.k8s.io/v1 + kind: Role + metadata: + name: weave-net-gateway-certificates + namespace: gateway-system + labels: + name: weave-net + rules: + - apiGroups: + - '' + resources: + - secrets + verbs: + - get + - list + - watch + - apiVersion: rbac.authorization.k8s.io/v1 + kind: RoleBinding + metadata: + name: weave-net-gateway-certificates + namespace: gateway-system + labels: + name: weave-net + roleRef: + kind: Role + name: weave-net-gateway-certificates + apiGroup: rbac.authorization.k8s.io + subjects: + - kind: ServiceAccount + name: weave-net + namespace: kube-system - apiVersion: apps/v1 kind: DaemonSet metadata: diff --git a/prog/weave-npc/main.go b/prog/weave-npc/main.go index a7b2b369..48a0bd1f 100644 --- a/prog/weave-npc/main.go +++ b/prog/weave-npc/main.go @@ -6,6 +6,7 @@ import ( "os" "os/signal" goruntime "runtime" + "strings" "syscall" "github.com/coreos/go-iptables/iptables" @@ -30,14 +31,15 @@ import ( ) var ( - version = "unreleased" - metricsAddr string - logLevel string - allowMcast bool - nodeName string - maxList int - bridgePortName string - serveLB bool + version = "unreleased" + metricsAddr string + logLevel string + allowMcast bool + nodeName string + maxList int + bridgePortName string + serveLB bool + gatewayNamespaces string ) func handleError(err error) { @@ -344,7 +346,14 @@ func root(cmd *cobra.Command, args []string) { // EgressServices: those Services' pods leave the cluster from that address (or not at all) egress, err := lb.NewEgressClient(config) handleFatal(err) - go lb.NewController(nodeName, client, egress, lb.NewDatapath(ipt)).Run(context.Background()) + ctl := lb.NewController(nodeName, client, egress, lb.NewDatapath(ipt)) + if gatewayNamespaces != "" { + // and Gateways of class controller lb.GatewayControllerName, from the node holding their address + gw, err := lb.NewGatewayClient(config) + handleFatal(err) + ctl = ctl.WithGateway(gw, strings.Split(gatewayNamespaces, ",")) + } + go ctl.Run(context.Background()) } signals := make(chan os.Signal, 1) @@ -365,6 +374,8 @@ func main() { rootCmd.PersistentFlags().IntVar(&maxList, "max-list-size", 1024, "maximum size of ipset list (for namespaces)") rootCmd.PersistentFlags().StringVar(&bridgePortName, "bridge-port-name", "vethwe-bridge", "name of the brige port on which packets are received and sent") rootCmd.PersistentFlags().BoolVar(&serveLB, "lb", true, "serve LoadBalancer Services of class "+lb.Class+" from addresses nodes hold") + rootCmd.PersistentFlags().StringVar(&gatewayNamespaces, "gateway-namespaces", "gateway-system", + "serve Gateways of controller "+lb.GatewayControllerName+", with certificates from Secrets in these namespaces (comma-separated; empty: no Gateways)") handleFatal(rootCmd.Execute()) }