spb/internetpressure

1// Package client talks to the ingestion API (GET /config, POST /batch, GET /agent/latest) with signed requests2// and keeps the clock-offset estimate derived from server_time in every response.3package client45import (6	"bytes"7	"context"8	"encoding/json"9	"errors"10	"fmt"11	"io"12	"net/http"13	"net/url"14	"strings"15	"time"1617	"internetpressure.io/probe-agent/internal/protocol"18	"internetpressure.io/probe-agent/internal/signer"19)2021// Timeouts.22const (23	ConfigTimeout = 15 * time.Second24	BatchTimeout  = 10 * time.Second25	LatestTimeout = 15 * time.Second26	maxRespBody   = 4 << 2027)2829// HTTPError is returned for non-2xx responses.30type HTTPError struct {31	Status int32	Body   string33}3435func (e *HTTPError) Error() string {36	b := e.Body37	if len(b) > 200 {38		b = b[:200] + "…"39	}40	return fmt.Sprintf("http %d: %s", e.Status, strings.TrimSpace(b))41}4243// IsSkew reports whether a 401 body mentions a clock skew (server hint → resync and retry once).44func (e *HTTPError) IsSkew() bool {45	return e.Status == http.StatusUnauthorized && strings.Contains(strings.ToLower(e.Body), "skew")46}4748// Retryable reports whether the failure warrants spooling (network error or 5xx).49func Retryable(err error) bool {50	var he *HTTPError51	if errors.As(err, &he) {52		return he.Status >= 500 || he.Status == http.StatusTooManyRequests53	}54	return err != nil55}5657// Client is a signed HTTP client for the ingestion API.58type Client struct {59	base    *url.URL60	signer  *signer.Signer61	ua      string62	version string63	http    *http.Client64	Clock   *Clock65}6667// New builds a client for baseURL (e.g. https://www.internetpressure.io/ingest/v1).68func New(baseURL string, s *signer.Signer, version string) (*Client, error) {69	u, err := url.Parse(strings.TrimRight(baseURL, "/"))70	if err != nil {71		return nil, err72	}73	return &Client{74		base:    u,75		signer:  s,76		version: version,77		ua:      UserAgent(version),78		http: &http.Client{Transport: &http.Transport{79			Proxy:               nil,80			MaxIdleConns:        2,81			IdleConnTimeout:     90 * time.Second,82			ForceAttemptHTTP2:   true,83			TLSHandshakeTimeout: 10 * time.Second,84		}},85		Clock: NewClock(),86	}, nil87}8889// UserAgent returns the protocol User-Agent string.90func UserAgent(version string) string {91	return "InternetPressureProbe/" + version + " (+https://www.internetpressure.io/probes)"92}9394// endpoint joins base path + suffix.95func (c *Client) endpoint(suffix string) *url.URL {96	u := *c.base97	u.Path = strings.TrimRight(u.Path, "/") + suffix98	return &u99}100101// do sends a signed request. body is the exact bytes to send (nil for GET). The timestamp header is the local102// clock corrected by the estimated offset so a drifting probe still passes the ±300 s window.103func (c *Client) do(ctx context.Context, method string, u *url.URL, body []byte, gzip bool, timeout time.Duration) ([]byte, time.Duration, error) {104	ctx, cancel := context.WithTimeout(ctx, timeout)105	defer cancel()106	var rd io.Reader107	if body != nil {108		rd = bytes.NewReader(body)109	}110	req, err := http.NewRequestWithContext(ctx, method, u.String(), rd)111	if err != nil {112		return nil, 0, err113	}114	req.Header.Set("User-Agent", c.ua)115	req.Header.Set("Accept", "application/json")116	if body != nil {117		req.Header.Set("Content-Type", "application/json")118		if gzip {119			req.Header.Set("Content-Encoding", "gzip")120		}121	}122	c.signer.Apply(req, c.Clock.ServerNow().Unix(), body)123124	t0 := time.Now()125	resp, err := c.http.Do(req)126	if err != nil {127		return nil, 0, err128	}129	defer resp.Body.Close()130	data, err := io.ReadAll(io.LimitReader(resp.Body, maxRespBody))131	rtt := time.Since(t0)132	if err != nil {133		return nil, rtt, err134	}135	if resp.StatusCode/100 != 2 {136		he := &HTTPError{Status: resp.StatusCode, Body: string(data)}137		// On an auth skew the server's Date header is the best resync source we have.138		if he.IsSkew() {139			if d, perr := http.ParseTime(resp.Header.Get("Date")); perr == nil {140				c.Clock.Set(t0.Add(rtt/2), d)141			}142		}143		return data, rtt, he144	}145	// Feed the clock from server_time when present (all JSON responses carry it), else from Date.146	var st struct {147		ServerTime string `json:"server_time"`148	}149	if json.Unmarshal(data, &st) == nil && st.ServerTime != "" {150		if t, perr := protocol.ParseTime(st.ServerTime); perr == nil {151			c.Clock.Observe(t0, rtt, t)152		}153	}154	return data, rtt, nil155}156157// GetConfig fetches the probe assignment.158func (c *Client) GetConfig(ctx context.Context) (*protocol.RemoteConfig, error) {159	data, _, err := c.do(ctx, http.MethodGet, c.endpoint("/config"), nil, false, ConfigTimeout)160	if err != nil {161		return nil, err162	}163	var cfg protocol.RemoteConfig164	if err := json.Unmarshal(data, &cfg); err != nil {165		return nil, fmt.Errorf("config: bad json: %w", err)166	}167	return &cfg, nil168}169170// PostBatch sends an already gzip-compressed batch body.171func (c *Client) PostBatch(ctx context.Context, gz []byte, timeout time.Duration) (*protocol.BatchResponse, error) {172	if timeout <= 0 {173		timeout = BatchTimeout174	}175	data, _, err := c.do(ctx, http.MethodPost, c.endpoint("/batch"), gz, true, timeout)176	if err != nil {177		return nil, err178	}179	var resp protocol.BatchResponse180	if len(data) > 0 {181		if err := json.Unmarshal(data, &resp); err != nil {182			return nil, fmt.Errorf("batch: bad json: %w", err)183		}184	}185	return &resp, nil186}187188// AgentLatest fetches the newest release descriptor.189func (c *Client) AgentLatest(ctx context.Context) (*protocol.AgentLatest, error) {190	data, _, err := c.do(ctx, http.MethodGet, c.endpoint("/agent/latest"), nil, false, LatestTimeout)191	if err != nil {192		return nil, err193	}194	var al protocol.AgentLatest195	if err := json.Unmarshal(data, &al); err != nil {196		return nil, fmt.Errorf("agent/latest: bad json: %w", err)197	}198	return &al, nil199}200201// Download fetches an arbitrary URL (release asset) into w with the probe User-Agent. Not signed.202func (c *Client) Download(ctx context.Context, rawURL string, w io.Writer, limit int64) (int64, error) {203	req, err := http.NewRequestWithContext(ctx, http.MethodGet, rawURL, nil)204	if err != nil {205		return 0, err206	}207	req.Header.Set("User-Agent", c.ua)208	resp, err := c.http.Do(req)209	if err != nil {210		return 0, err211	}212	defer resp.Body.Close()213	if resp.StatusCode/100 != 2 {214		return 0, &HTTPError{Status: resp.StatusCode}215	}216	return io.Copy(w, io.LimitReader(resp.Body, limit))217}218