images / images (push) Successful in 7m29s
The proxy sets Forwarded, X-Forwarded-For/-Host/-Proto and X-Real-IP from the connection, but applied a route's request header filters after that, so whoever could write a route to a backend could tell it any client address (a backend that trusts the gateway's X-Forwarded-For believes it). An HTTPRoute whose request header filter sets, adds or removes one of them (names compared case-insensitively) is now not accepted (UnsupportedValue), like the other filters we do not serve. The proxy also sets them after the filters, clearing them first: SetXForwarded appends to an X-Forwarded-For already there. The proxy test's backend now prints every X-Forwarded-For value; it printed only the first, which hid an appended one. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01UHXzGrsxfkGk8LmwrBbHNZ
1050 lines
36 KiB
Go
1050 lines
36 KiB
Go
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"
|
|
)
|
|
|
|
// forwardingHeaders are the request headers the proxy sets itself, from the connection: who the client is and what it
|
|
// asked for, over what. No route filter may change them, or whoever can write a route to a backend could tell that
|
|
// backend any client address.
|
|
var forwardingHeaders = []string{"Forwarded", "X-Forwarded-For", "X-Forwarded-Host", "X-Forwarded-Proto", "X-Real-IP"}
|
|
|
|
// forwardingHeader is the forwarding header a name denotes, if it denotes one (header names are case-insensitive).
|
|
func forwardingHeader(name string) (string, bool) {
|
|
for _, h := range forwardingHeaders {
|
|
if strings.EqualFold(name, h) {
|
|
return h, true
|
|
}
|
|
}
|
|
return "", false
|
|
}
|
|
|
|
// touchesForwarding is why a request header filter cannot be served: it sets, adds or removes a forwarding header.
|
|
func touchesForwarding(f *HTTPHeaderFilter) string {
|
|
if f == nil {
|
|
return ""
|
|
}
|
|
var names []string
|
|
for _, h := range f.Set {
|
|
names = append(names, h.Name)
|
|
}
|
|
for _, h := range f.Add {
|
|
names = append(names, h.Name)
|
|
}
|
|
names = append(names, f.Remove...)
|
|
for _, n := range names {
|
|
if h, ok := forwardingHeader(n); ok {
|
|
return fmt.Sprintf("a request header filter may not change %s: the gateway sets it from the connection", h)
|
|
}
|
|
}
|
|
return ""
|
|
}
|
|
|
|
// 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:
|
|
if msg := touchesForwarding(f.RequestHeaderModifier); msg != "" {
|
|
return msg
|
|
}
|
|
case 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
|
|
}
|