|
| 1 | +package images |
| 2 | + |
| 3 | +import ( |
| 4 | + "bufio" |
| 5 | + "context" |
| 6 | + "encoding/base64" |
| 7 | + "errors" |
| 8 | + "fmt" |
| 9 | + "io" |
| 10 | + "net" |
| 11 | + "net/http" |
| 12 | + "net/url" |
| 13 | + "strings" |
| 14 | + "time" |
| 15 | +) |
| 16 | + |
| 17 | +// errBlockedAddress marks a registry request refused by the network guard. |
| 18 | +var errBlockedAddress = errors.New("registry address not allowed") |
| 19 | + |
| 20 | +// maxRegistryResponseBytes bounds every response body read from a registry |
| 21 | +// for a config lookup: manifests, indexes, token responses and the config |
| 22 | +// blob. Real ones are kilobytes; this stops a hostile or broken registry from |
| 23 | +// streaming an unbounded body into memory. |
| 24 | +const maxRegistryResponseBytes = 4 << 20 |
| 25 | + |
| 26 | +type registryHostKey struct{} |
| 27 | + |
| 28 | +// namedEndpoint is the registry a lookup is for, as the Pod named it. port is |
| 29 | +// empty when the reference had none (the scheme's default port). |
| 30 | +type namedEndpoint struct{ host, port string } |
| 31 | + |
| 32 | +// withRegistryHost records the registry a lookup is for. The guard lets that |
| 33 | +// endpoint, and only that endpoint, sit on a private or loopback address: |
| 34 | +// internal registries are normal, but a redirect or token realm pointing |
| 35 | +// somewhere else internal is not. |
| 36 | +func withRegistryHost(ctx context.Context, registry string) context.Context { |
| 37 | + host, port := splitHostPort(registry) |
| 38 | + return context.WithValue(ctx, registryHostKey{}, namedEndpoint{host: host, port: port}) |
| 39 | +} |
| 40 | + |
| 41 | +// isNamed reports whether host:port is the endpoint the lookup was for. |
| 42 | +func isNamed(ctx context.Context, host, port string) bool { |
| 43 | + n, _ := ctx.Value(registryHostKey{}).(namedEndpoint) |
| 44 | + if n.host == "" || !strings.EqualFold(n.host, host) { |
| 45 | + return false |
| 46 | + } |
| 47 | + if n.port == "" { |
| 48 | + return port == "443" || port == "80" |
| 49 | + } |
| 50 | + return n.port == port |
| 51 | +} |
| 52 | + |
| 53 | +// guardedTransport is the transport runtime-version lookups use. Image |
| 54 | +// references come from Pod specs, which anyone who can create a Pod |
| 55 | +// controls, and Radar's network position differs from a node's. Every |
| 56 | +// request, including redirects and token endpoints, is resolved here and |
| 57 | +// sent only to an address that passed the policy: |
| 58 | +// - link-local (cloud metadata), unspecified and multicast: never; |
| 59 | +// - loopback: never in-cluster (it is Radar's own Pod), and locally only |
| 60 | +// for the registry endpoint the Pod named (a local dev registry); |
| 61 | +// - private ranges: only for the registry endpoint (host and port) the Pod |
| 62 | +// named. |
| 63 | +// |
| 64 | +// Direct connections dial the checked address. Through an HTTP proxy, HTTPS |
| 65 | +// requests are tunneled with a CONNECT to the checked address, so the proxy |
| 66 | +// can't resolve the name somewhere else. A proxied request that can't be |
| 67 | +// pinned that way (plain HTTP, a name that doesn't resolve here, a non-HTTP |
| 68 | +// proxy) is refused in-cluster and left to the proxy locally, where the proxy |
| 69 | +// is the user's own. |
| 70 | +type guardedTransport struct { |
| 71 | + requireTLS bool |
| 72 | + inCluster bool |
| 73 | + proxy func(*http.Request) (*url.URL, error) |
| 74 | + resolver *net.Resolver |
| 75 | + dialer *net.Dialer |
| 76 | + connectTimeout time.Duration |
| 77 | + // pinned never uses a proxy on its own; its dialer tunnels through one |
| 78 | + // when the request carries a pinned proxy route. |
| 79 | + pinned *http.Transport |
| 80 | + // unpinned lets the proxy resolve and dial: local mode only. |
| 81 | + unpinned *http.Transport |
| 82 | +} |
| 83 | + |
| 84 | +// forLookup returns a transport with its own connection pools, for one |
| 85 | +// lookup. Pools aren't shared across lookups: a connection one lookup opened |
| 86 | +// to the registry it named (maybe on a private address) must never carry |
| 87 | +// another lookup's request, even after that name resolves elsewhere. |
| 88 | +func (g *guardedTransport) forLookup() *guardedTransport { |
| 89 | + return newGuardedTransportWith(g.inCluster, g.proxy, g.resolver) |
| 90 | +} |
| 91 | + |
| 92 | +func (g *guardedTransport) closeIdle() { |
| 93 | + g.pinned.CloseIdleConnections() |
| 94 | + g.unpinned.CloseIdleConnections() |
| 95 | +} |
| 96 | + |
| 97 | +func newGuardedTransport(inCluster bool) *guardedTransport { |
| 98 | + return newGuardedTransportWith(inCluster, http.ProxyFromEnvironment, net.DefaultResolver) |
| 99 | +} |
| 100 | + |
| 101 | +func newGuardedTransportWith(inCluster bool, proxy func(*http.Request) (*url.URL, error), resolver *net.Resolver) *guardedTransport { |
| 102 | + g := &guardedTransport{ |
| 103 | + inCluster: inCluster, |
| 104 | + proxy: proxy, |
| 105 | + resolver: resolver, |
| 106 | + dialer: &net.Dialer{Timeout: 10 * time.Second, KeepAlive: 30 * time.Second}, |
| 107 | + // The CONNECT handshake: request and response headers only. |
| 108 | + connectTimeout: 10 * time.Second, |
| 109 | + } |
| 110 | + newTransport := func() *http.Transport { |
| 111 | + return &http.Transport{ |
| 112 | + ForceAttemptHTTP2: true, |
| 113 | + MaxIdleConns: 10, |
| 114 | + IdleConnTimeout: 30 * time.Second, |
| 115 | + TLSHandshakeTimeout: 10 * time.Second, |
| 116 | + ResponseHeaderTimeout: 15 * time.Second, |
| 117 | + ExpectContinueTimeout: time.Second, |
| 118 | + } |
| 119 | + } |
| 120 | + g.pinned = newTransport() |
| 121 | + g.pinned.DialContext = g.dialContext |
| 122 | + g.unpinned = newTransport() |
| 123 | + g.unpinned.Proxy = func(r *http.Request) (*url.URL, error) { return g.proxy(r) } |
| 124 | + g.unpinned.DialContext = g.dialer.DialContext |
| 125 | + return g |
| 126 | +} |
| 127 | + |
| 128 | +type proxyRouteKey struct{} |
| 129 | + |
| 130 | +func (g *guardedTransport) RoundTrip(req *http.Request) (*http.Response, error) { |
| 131 | + if g.requireTLS && req.URL.Scheme != "https" { |
| 132 | + return nil, fmt.Errorf("%w: registry credentials require HTTPS", errBlockedAddress) |
| 133 | + } |
| 134 | + ctx := req.Context() |
| 135 | + host, port := req.URL.Hostname(), req.URL.Port() |
| 136 | + if port == "" { |
| 137 | + port = map[string]string{"https": "443", "http": "80"}[req.URL.Scheme] |
| 138 | + } |
| 139 | + proxyURL, err := g.proxy(req) |
| 140 | + if err != nil { |
| 141 | + proxyURL = nil |
| 142 | + } |
| 143 | + |
| 144 | + // Checked per request, not only when dialing: redirects and token |
| 145 | + // endpoints are requests too. |
| 146 | + ips, lookupErr := g.lookup(ctx, host) |
| 147 | + if lookupErr != nil { |
| 148 | + if proxyURL != nil && !g.inCluster { |
| 149 | + return g.send(g.unpinned, req) |
| 150 | + } |
| 151 | + if proxyURL != nil { |
| 152 | + return nil, fmt.Errorf("%w: %s can't be checked before the proxy resolves it", errBlockedAddress, host) |
| 153 | + } |
| 154 | + return nil, lookupErr |
| 155 | + } |
| 156 | + named := isNamed(ctx, host, port) |
| 157 | + for _, ip := range ips { |
| 158 | + if !registryIPAllowed(ip, g.inCluster, named) { |
| 159 | + return nil, fmt.Errorf("%w: %s", errBlockedAddress, host) |
| 160 | + } |
| 161 | + } |
| 162 | + if proxyURL == nil { |
| 163 | + return g.send(g.pinned, req) |
| 164 | + } |
| 165 | + if req.URL.Scheme == "https" && proxyURL.Scheme == "http" { |
| 166 | + return g.send(g.pinned, req.WithContext(context.WithValue(ctx, proxyRouteKey{}, proxyURL))) |
| 167 | + } |
| 168 | + if !g.inCluster { |
| 169 | + return g.send(g.unpinned, req) |
| 170 | + } |
| 171 | + return nil, fmt.Errorf("%w: %s can't be pinned through the proxy", errBlockedAddress, host) |
| 172 | +} |
| 173 | + |
| 174 | +func (g *guardedTransport) send(rt *http.Transport, req *http.Request) (*http.Response, error) { |
| 175 | + resp, err := rt.RoundTrip(req) |
| 176 | + if err != nil || resp.Body == nil { |
| 177 | + return resp, err |
| 178 | + } |
| 179 | + resp.Body = &cappedBody{ReadCloser: resp.Body, left: maxRegistryResponseBytes} |
| 180 | + return resp, nil |
| 181 | +} |
| 182 | + |
| 183 | +// dialContext connects to the first allowed address of addr, directly or |
| 184 | +// through a CONNECT tunnel when the request carries a proxy route. It dials |
| 185 | +// the address that was checked, not the name, so a second DNS answer can't |
| 186 | +// swap in a forbidden address. |
| 187 | +func (g *guardedTransport) dialContext(ctx context.Context, network, addr string) (net.Conn, error) { |
| 188 | + host, port, err := net.SplitHostPort(addr) |
| 189 | + if err != nil { |
| 190 | + return nil, err |
| 191 | + } |
| 192 | + ips, err := g.lookup(ctx, host) |
| 193 | + if err != nil { |
| 194 | + return nil, err |
| 195 | + } |
| 196 | + named := isNamed(ctx, host, port) |
| 197 | + proxyURL, _ := ctx.Value(proxyRouteKey{}).(*url.URL) |
| 198 | + lastErr := fmt.Errorf("%w: %s", errBlockedAddress, host) |
| 199 | + for _, ip := range ips { |
| 200 | + if !registryIPAllowed(ip, g.inCluster, named) { |
| 201 | + continue |
| 202 | + } |
| 203 | + target := net.JoinHostPort(ip.String(), port) |
| 204 | + var conn net.Conn |
| 205 | + if proxyURL != nil { |
| 206 | + conn, err = g.connectVia(ctx, proxyURL, target) |
| 207 | + } else { |
| 208 | + conn, err = g.dialer.DialContext(ctx, network, target) |
| 209 | + } |
| 210 | + if err == nil { |
| 211 | + return conn, nil |
| 212 | + } |
| 213 | + lastErr = err |
| 214 | + } |
| 215 | + return nil, lastErr |
| 216 | +} |
| 217 | + |
| 218 | +// connectVia opens a CONNECT tunnel to target (an IP:port) through an HTTP |
| 219 | +// proxy. The proxy itself is configuration, not tenant input, so it isn't |
| 220 | +// checked against the address policy. |
| 221 | +func (g *guardedTransport) connectVia(ctx context.Context, proxyURL *url.URL, target string) (net.Conn, error) { |
| 222 | + proxyAddr := proxyURL.Host |
| 223 | + if proxyURL.Port() == "" { |
| 224 | + proxyAddr = net.JoinHostPort(proxyURL.Hostname(), "80") |
| 225 | + } |
| 226 | + conn, err := g.dialer.DialContext(ctx, "tcp", proxyAddr) |
| 227 | + if err != nil { |
| 228 | + return nil, err |
| 229 | + } |
| 230 | + // http.Transport dials without the request's deadline, so the handshake |
| 231 | + // gets its own: a proxy that accepts and never answers must not hold the |
| 232 | + // connection (and this goroutine) forever. |
| 233 | + _ = conn.SetDeadline(time.Now().Add(g.connectTimeout)) |
| 234 | + req := "CONNECT " + target + " HTTP/1.1\r\nHost: " + target + "\r\n" |
| 235 | + if u := proxyURL.User; u != nil { |
| 236 | + pass, _ := u.Password() |
| 237 | + req += "Proxy-Authorization: Basic " + base64.StdEncoding.EncodeToString([]byte(u.Username()+":"+pass)) + "\r\n" |
| 238 | + } |
| 239 | + if _, err := io.WriteString(conn, req+"\r\n"); err != nil { |
| 240 | + _ = conn.Close() |
| 241 | + return nil, err |
| 242 | + } |
| 243 | + br := bufio.NewReader(conn) |
| 244 | + // The body of a CONNECT response is never read: after a 200 the bytes |
| 245 | + // belong to the tunnel, whatever framing headers the proxy sent. |
| 246 | + resp, err := http.ReadResponse(br, &http.Request{Method: http.MethodConnect}) |
| 247 | + if err != nil { |
| 248 | + _ = conn.Close() |
| 249 | + return nil, err |
| 250 | + } |
| 251 | + if resp.StatusCode != http.StatusOK { |
| 252 | + _ = conn.Close() |
| 253 | + return nil, fmt.Errorf("proxy refused the tunnel: %s", resp.Status) |
| 254 | + } |
| 255 | + _ = conn.SetDeadline(time.Time{}) |
| 256 | + if br.Buffered() > 0 { |
| 257 | + return &bufferedConn{Conn: conn, r: br}, nil |
| 258 | + } |
| 259 | + return conn, nil |
| 260 | +} |
| 261 | + |
| 262 | +type bufferedConn struct { |
| 263 | + net.Conn |
| 264 | + r *bufio.Reader |
| 265 | +} |
| 266 | + |
| 267 | +func (c *bufferedConn) Read(p []byte) (int, error) { return c.r.Read(p) } |
| 268 | + |
| 269 | +func (g *guardedTransport) lookup(ctx context.Context, host string) ([]net.IP, error) { |
| 270 | + if ip := net.ParseIP(host); ip != nil { |
| 271 | + return []net.IP{ip}, nil |
| 272 | + } |
| 273 | + addrs, err := g.resolver.LookupIPAddr(ctx, host) |
| 274 | + if err != nil { |
| 275 | + return nil, err |
| 276 | + } |
| 277 | + ips := make([]net.IP, 0, len(addrs)) |
| 278 | + for _, a := range addrs { |
| 279 | + ips = append(ips, a.IP) |
| 280 | + } |
| 281 | + return ips, nil |
| 282 | +} |
| 283 | + |
| 284 | +// registryIPAllowed is the address policy; see guardedTransport. |
| 285 | +func registryIPAllowed(ip net.IP, inCluster, namedRegistry bool) bool { |
| 286 | + switch { |
| 287 | + case ip == nil, ip.IsLinkLocalUnicast(), ip.IsLinkLocalMulticast(), ip.IsUnspecified(), ip.IsMulticast(): |
| 288 | + return false |
| 289 | + case ip.IsLoopback(): |
| 290 | + return !inCluster && namedRegistry |
| 291 | + case ip.IsPrivate(), isSharedAddressSpace(ip): |
| 292 | + return namedRegistry |
| 293 | + } |
| 294 | + return true |
| 295 | +} |
| 296 | + |
| 297 | +// isSharedAddressSpace reports RFC 6598 carrier-grade NAT space (100.64/10), |
| 298 | +// which net.IP.IsPrivate leaves out and some clusters use for Pods. |
| 299 | +func isSharedAddressSpace(ip net.IP) bool { |
| 300 | + v4 := ip.To4() |
| 301 | + return v4 != nil && v4[0] == 100 && v4[1]&0xc0 == 64 |
| 302 | +} |
| 303 | + |
| 304 | +// splitHostPort splits "host", "host:port", "[v6]" or "[v6]:port". |
| 305 | +func splitHostPort(hostport string) (host, port string) { |
| 306 | + if h, p, err := net.SplitHostPort(hostport); err == nil { |
| 307 | + return h, p |
| 308 | + } |
| 309 | + return strings.TrimSuffix(strings.TrimPrefix(hostport, "["), "]"), "" |
| 310 | +} |
| 311 | + |
| 312 | +type cappedBody struct { |
| 313 | + io.ReadCloser |
| 314 | + left int64 |
| 315 | +} |
| 316 | + |
| 317 | +func (b *cappedBody) Read(p []byte) (int, error) { |
| 318 | + if b.left <= 0 { |
| 319 | + // At the cap: a body that ends exactly here is fine. |
| 320 | + var one [1]byte |
| 321 | + if n, err := b.ReadCloser.Read(one[:]); n == 0 && err == io.EOF { |
| 322 | + return 0, io.EOF |
| 323 | + } |
| 324 | + return 0, fmt.Errorf("registry response exceeds %d bytes", maxRegistryResponseBytes) |
| 325 | + } |
| 326 | + if int64(len(p)) > b.left { |
| 327 | + p = p[:b.left] |
| 328 | + } |
| 329 | + n, err := b.ReadCloser.Read(p) |
| 330 | + b.left -= int64(n) |
| 331 | + return n, err |
| 332 | +} |
| 333 | + |
| 334 | +// isBlocked reports whether err came from the network guard. |
| 335 | +func isBlocked(err error) bool { |
| 336 | + return errors.Is(err, errBlockedAddress) |
| 337 | +} |
0 commit comments