client.go

// Package dnapi handles communication with the Defined Networking cloud API server.
package dnapi

import (
	"bytes"
	"context"
	"crypto/ecdsa"
	"crypto/ed25519"
	"crypto/rand"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"net"
	"net/http"
	"net/url"
	"sync/atomic"
	"time"

	"github.com/DefinedNet/dnapi/keys"
	"github.com/DefinedNet/dnapi/message"
)

// Client communicates with the API server.
type Client struct {
	dnServer string

	client          *http.Client
	streamingClient *http.Client
}

// fallbackAddrs holds the pinned addresses behind api-backup.defined.net.
// They are dialed only when the system resolver cannot resolve the host, so a
// client whose DNS points into a Nebula network that is down (expired OIDC,
// lost network access) can still reach the API to recover. TLS still
// verifies against the hostname, so a fallback address cannot impersonate
// the API.
var fallbackAddrs = map[string][]string{
	"api.defined.net": {"209.46.53.8", "209.46.53.81", "209.46.53.153"},
}

// dialWithFallback returns a DialContext that retries DNS failures for hosts
// in fallback against their pinned addresses.
func dialWithFallback(fallback map[string][]string) func(ctx context.Context, network, addr string) (net.Conn, error) {
	d := &net.Dialer{Timeout: 10 * time.Second}
	return func(ctx context.Context, network, addr string) (net.Conn, error) {
		conn, err := d.DialContext(ctx, network, addr)
		host, port, splitErr := net.SplitHostPort(addr)
		var dnsErr *net.DNSError
		if err == nil || splitErr != nil || !errors.As(err, &dnsErr) {
			return conn, err
		}
		for _, ip := range fallback[host] {
			conn, fbErr := d.DialContext(ctx, network, net.JoinHostPort(ip, port))
			if fbErr == nil {
				return conn, nil
			}
			err = fmt.Errorf("%w; fallback %s: %w", err, ip, fbErr)
		}
		return nil, err
	}
}

// NewClient returns new Client configured with the given useragent.
// It also supports reading Proxy information from the environment.
func NewClient(useragent string, dnServer string) *Client {
	return &Client{
		client: &http.Client{
			Timeout: 2 * time.Minute,
			Transport: &uaTransport{
				T: &http.Transport{
					Proxy:               http.ProxyFromEnvironment,
					TLSHandshakeTimeout: 10 * time.Second,
					DialContext:         dialWithFallback(fallbackAddrs),
				},
				useragent: useragent,
			},
		},
		streamingClient: &http.Client{
			Timeout: 15 * time.Minute,
			Transport: &uaTransport{
				T: &http.Transport{
					Proxy:               http.ProxyFromEnvironment,
					TLSHandshakeTimeout: 10 * time.Second,
					DialContext:         dialWithFallback(fallbackAddrs),
				},
				useragent: useragent,
			},
		},
		dnServer: dnServer,
	}
}

// APIError wraps an error and contains the RequestID from the X-Request-ID
// header of an API response. ReqID defaults to empty string  if the header is
// not in the response.
type APIError struct {
	e     error
	ReqID string
}

func (e *APIError) Error() string {
	return e.e.Error()
}

func (e *APIError) Unwrap() error {
	return e.e
}

var ErrInvalidCredentials = fmt.Errorf("invalid credentials")
var ErrInvalidCode = fmt.Errorf("invalid enrollment code")
var ErrHostBlocked = fmt.Errorf("host is blocked")

type ConfigMeta struct {
	Org          ConfigOrg
	Network      ConfigNetwork
	Host         ConfigHost
	EndpointOIDC *ConfigEndpointOIDC
}

type ConfigOrg struct {
	ID   string
	Name string
}

type ConfigNetwork struct {
	ID   string
	Name string
}

type ConfigHost struct {
	ID          string
	Name        string
	IPAddresses []string
}

type ConfigEndpointOIDC struct {
	Email     string
	ExpiresAt *time.Time
}

// mergeIPAddresses returns the plural field if populated, otherwise wraps the singular value.
func mergeIPAddresses(plural []string, singular string) []string {
	if len(plural) > 0 {
		return plural
	}
	if singular != "" {
		return []string{singular}
	}
	return nil
}

// Enroll issues an enrollment request against the REST API using the given enrollment code, passing along a locally
// generated DH X25519 public key to be signed by the CA, and an Ed 25519 public key for future API call authentication.
// The hostname is the enrolling machine's OS hostname, sent as informational metadata; it may be empty.
// On success it returns the Nebula config generated by the server, a Nebula private key PEM to be inserted into the
// config (see api.InsertConfigPrivateKey), credentials to be used in DNClient API requests, and a meta object.
func (c *Client) Enroll(ctx context.Context, logger *slog.Logger, code string, hostname string) ([]byte, []byte, *keys.Credentials, *ConfigMeta, error) {
	logger.Debug("Making enrollment request to API", "server", c.dnServer)

	// Generate newKeys for the enrollment request
	newKeys, err := keys.New()
	if err != nil {
		return nil, nil, nil, nil, err
	}

	hostP256PublicKeyPEM, err := newKeys.HostP256PublicKey.MarshalPEM()
	if err != nil {
		return nil, nil, nil, nil, err
	}

	// Make a request to the API with the enrollment code. P256 keys are always
	// generated; the 25519 keys are omitted under FIPS (see keys.New), so only
	// include them when they were generated. The DN API selects the key set to
	// use based on the enrolled network's curve.
	payload := message.EnrollRequest{
		Code:             code,
		NebulaPubkeyP256: newKeys.NebulaP256PublicKeyPEM,
		HostPubkeyP256:   hostP256PublicKeyPEM,
		Timestamp:        time.Now(),
		Hostname:         hostname,
	}

	if newKeys.HostEd25519PublicKey != nil {
		hostEd25519PublicKeyPEM, err := newKeys.HostEd25519PublicKey.MarshalPEM()
		if err != nil {
			return nil, nil, nil, nil, err
		}
		payload.NebulaPubkeyX25519 = newKeys.NebulaX25519PublicKeyPEM
		payload.HostPubkeyEd25519 = hostEd25519PublicKeyPEM
	}

	reqID, r, err := callAPI[message.EnrollResponseData](ctx, c, "POST", message.EnrollEndpoint, payload)
	l := logger.With("reqID", reqID)
	if err != nil {
		var apiErrors message.APIErrors
		if errors.As(err, &apiErrors) && len(apiErrors) == 1 {
			// Check for *only* an "invalid code" error returned by the API
			if err := apiErrors[0]; err.Path == "code" && err.Code == "ERR_INVALID_VALUE" {
				l.Warn("Enrollment request failed for invalid code")
				return nil, nil, nil, nil, &APIError{e: ErrInvalidCode, ReqID: reqID}
			}

			// Check for *only* a blocked host error returned by the API
			if err := apiErrors[0]; err.Path == "" && err.Code == "ERR_HOST_BLOCKED" {
				l.Warn("Enrollment request failed for blocked host")
				return nil, nil, nil, nil, &APIError{e: ErrHostBlocked, ReqID: reqID}
			}
		}

		l.Error("Enrollment request failed with unexpected error", "error", err)
		return nil, nil, nil, nil, &APIError{e: fmt.Errorf("unexpected error during enrollment: %w", err), ReqID: reqID}
	}
	l.Info("Enrollment request succeeded")

	meta := &ConfigMeta{
		Org: ConfigOrg{
			ID:   r.Organization.ID,
			Name: r.Organization.Name,
		},
		Network: ConfigNetwork{
			ID:   r.Network.ID,
			Name: r.Network.Name,
		},
		Host: ConfigHost{
			ID:          r.HostID,
			Name:        r.Host.Name,
			IPAddresses: mergeIPAddresses(r.Host.IPAddresses, r.Host.IPAddress),
		},
	}

	if r.EndpointOIDCMeta != nil {
		meta.EndpointOIDC = &ConfigEndpointOIDC{
			Email:     r.EndpointOIDCMeta.Email,
			ExpiresAt: r.EndpointOIDCMeta.ExpiresAt,
		}
	}

	// Determine the private keys to save based on the network curve type
	var privkeyPEM []byte
	var privkey keys.PrivateKey
	switch r.Network.Curve {
	case message.NetworkCurve25519:
		privkeyPEM = newKeys.NebulaX25519PrivateKeyPEM
		privkey = newKeys.HostEd25519PrivateKey
	case message.NetworkCurveP256:
		privkeyPEM = newKeys.NebulaP256PrivateKeyPEM
		privkey = newKeys.HostP256PrivateKey
	default:
		return nil, nil, nil, nil, &APIError{e: fmt.Errorf("unsupported curve type: %s", r.Network.Curve), ReqID: reqID}
	}

	trustedKeys, err := keys.TrustedKeysFromPEM(r.TrustedKeys)
	if err != nil {
		return nil, nil, nil, nil, &APIError{e: fmt.Errorf("failed to load trusted keys from bundle: %s", err), ReqID: reqID}
	}

	creds := &keys.Credentials{
		HostID:      r.HostID,
		PrivateKey:  privkey,
		Counter:     r.Counter,
		TrustedKeys: trustedKeys,
	}
	return r.Config, privkeyPEM, creds, meta, nil
}

// CheckForUpdate sends a signed message to the DNClient API to learn if there is a new configuration available.
func (c *Client) CheckForUpdate(ctx context.Context, creds keys.Credentials) (bool, error) {
	respBody, err := c.postDNClient(ctx, message.CheckForUpdate, nil, creds.HostID, creds.Counter, creds.PrivateKey)
	if err != nil {
		return false, fmt.Errorf("failed to post message to dnclient api: %w", err)
	}
	result := message.CheckForUpdateResponseWrapper{}
	err = json.Unmarshal(respBody, &result)
	if err != nil {
		return false, fmt.Errorf("failed to interpret API response: %s", err)
	}
	return result.Data.UpdateAvailable, nil
}

// LongPollWait sends a signed message to a DNClient API endpoint that will block, returning only
// if there is an action the client should take before the timeout (config updates, debug commands)
func (c *Client) LongPollWait(ctx context.Context, creds keys.Credentials, supportedActions []string) (*message.LongPollWaitResponse, error) {
	value, err := json.Marshal(message.LongPollWaitRequest{
		SupportedActions: supportedActions,
	})
	if err != nil {
		return nil, fmt.Errorf("failed to marshal DNClient message: %s", err)
	}

	respBody, err := c.postDNClient(ctx, message.LongPollWait, value, creds.HostID, creds.Counter, creds.PrivateKey)
	if err != nil {
		return nil, fmt.Errorf("failed to post message to dnclient api: %w", err)
	}
	result := message.LongPollWaitResponseWrapper{}
	err = json.Unmarshal(respBody, &result)
	if err != nil {
		return nil, fmt.Errorf("failed to interpret API response: %s", err)
	}
	return &result.Data, nil
}

// DoUpdate sends a signed message to the DNClient API to fetch the new configuration update. During this call new keys
// are generated both for Nebula and DNClient API communication. If the API response is successful, the new configuration
// is returned along with the new Nebula private key PEM, new DNClient API credentials, and a meta object.
//
// See dnapi.InsertConfigPrivateKey for how to insert the new Nebula private key into the configuration.
func (c *Client) DoUpdate(ctx context.Context, creds keys.Credentials) ([]byte, []byte, *keys.Credentials, *ConfigMeta, error) {
	// Rotate keys
	var nebulaPrivkeyPEM []byte     // ECDH
	var hostPrivkey keys.PrivateKey // ECDSA

	newKeys, err := keys.New()
	if err != nil {
		return nil, nil, nil, nil, fmt.Errorf("failed to generate new keys: %s", err)
	}

	msg := message.DoUpdateRequest{
		Nonce: nonce(),
	}

	// Set the correct keypair based on the current private key type
	switch creds.PrivateKey.Unwrap().(type) {
	case ed25519.PrivateKey:
		hostPubkeyPEM, err := newKeys.HostEd25519PublicKey.MarshalPEM()
		if err != nil {
			return nil, nil, nil, nil, fmt.Errorf("failed to marshal Ed25519 public key: %s", err)
		}
		hostPrivkey = newKeys.HostEd25519PrivateKey
		nebulaPrivkeyPEM = newKeys.NebulaX25519PrivateKeyPEM
		msg.HostPubkeyEd25519 = hostPubkeyPEM
		msg.NebulaPubkeyX25519 = newKeys.NebulaX25519PublicKeyPEM
	case *ecdsa.PrivateKey:
		hostPubkeyPEM, err := newKeys.HostP256PublicKey.MarshalPEM()
		if err != nil {
			return nil, nil, nil, nil, fmt.Errorf("failed to marshal P256 public key: %s", err)
		}
		hostPrivkey = newKeys.HostP256PrivateKey
		nebulaPrivkeyPEM = newKeys.NebulaP256PrivateKeyPEM
		msg.HostPubkeyP256 = hostPubkeyPEM
		msg.NebulaPubkeyP256 = newKeys.NebulaP256PublicKeyPEM
	}

	blob, err := json.Marshal(msg)
	if err != nil {
		return nil, nil, nil, nil, fmt.Errorf("failed to marshal DNClient message: %s", err)
	}

	// Make API call
	resp, err := c.postDNClient(ctx, message.DoUpdate, blob, creds.HostID, creds.Counter, creds.PrivateKey)
	if err != nil {
		return nil, nil, nil, nil, fmt.Errorf("failed to make API call to Defined Networking: %w", err)
	}

	// Verify the signature
	resultWrapper, err := verifySignature(resp, creds)
	if err != nil {
		return nil, nil, nil, nil, err
	}

	// Consume the verified message
	result := message.DoUpdateResponse{}
	err = json.Unmarshal(resultWrapper.Data.Message, &result)
	if err != nil {
		return nil, nil, nil, nil, fmt.Errorf("failed to unmarshal response (%s): %s", resultWrapper.Data.Message, err)
	}

	// Verify the nonce
	if !bytes.Equal(result.Nonce, msg.Nonce) {
		return nil, nil, nil, nil, fmt.Errorf("nonce mismatch between request (%s) and response (%s)", msg.Nonce, result.Nonce)
	}

	// Verify the counter
	if result.Counter <= creds.Counter {
		return nil, nil, nil, nil, fmt.Errorf("counter in request (%d) should be less than counter in response (%d)", creds.Counter, result.Counter)
	}

	trustedKeys, err := keys.TrustedKeysFromPEM(result.TrustedKeys)
	if err != nil {
		return nil, nil, nil, nil, fmt.Errorf("failed to load trusted keys from bundle: %s", err)
	}

	newCreds := &keys.Credentials{
		HostID:      creds.HostID,
		Counter:     result.Counter,
		PrivateKey:  hostPrivkey,
		TrustedKeys: trustedKeys,
	}

	meta := &ConfigMeta{
		Org: ConfigOrg{
			ID:   result.Organization.ID,
			Name: result.Organization.Name,
		},
		Network: ConfigNetwork{
			ID:   result.Network.ID,
			Name: result.Network.Name,
		},
		Host: ConfigHost{
			ID:          result.Host.ID,
			Name:        result.Host.Name,
			IPAddresses: mergeIPAddresses(result.Host.IPAddresses, result.Host.IPAddress),
		},
	}

	if result.EndpointOIDCMeta != nil {
		meta.EndpointOIDC = &ConfigEndpointOIDC{
			Email:     result.EndpointOIDCMeta.Email,
			ExpiresAt: result.EndpointOIDCMeta.ExpiresAt,
		}
	}

	return result.Config, nebulaPrivkeyPEM, newCreds, meta, nil
}

// DoConfigUpdate sends a signed message to the DNClient API to fetch the new configuration update. During this call new keys
// are generated for DNClient API communication. If the API response is successful, the new configuration
// is returned along with the new DNClient API credentials and a meta object.
//
// See dnapi.InsertConfigPrivateKey and dnapi.InsertConfigCert for how to insert the old Nebula cert/private key into the configuration.
func (c *Client) DoConfigUpdate(ctx context.Context, creds keys.Credentials) ([]byte, *keys.Credentials, *ConfigMeta, error) {
	// Rotate key
	var hostPrivkey keys.PrivateKey // ECDSA

	newKeys, err := keys.New()
	if err != nil {
		return nil, nil, nil, fmt.Errorf("failed to generate new keys: %s", err)
	}

	msg := message.DoConfigUpdateRequest{
		Nonce: nonce(),
	}

	// Set the correct keypair based on the current private key type
	switch creds.PrivateKey.Unwrap().(type) {
	case ed25519.PrivateKey:
		hostPubkeyPEM, err := newKeys.HostEd25519PublicKey.MarshalPEM()
		if err != nil {
			return nil, nil, nil, fmt.Errorf("failed to marshal Ed25519 public key: %s", err)
		}
		hostPrivkey = newKeys.HostEd25519PrivateKey
		msg.HostPubkeyEd25519 = hostPubkeyPEM
	case *ecdsa.PrivateKey:
		hostPubkeyPEM, err := newKeys.HostP256PublicKey.MarshalPEM()
		if err != nil {
			return nil, nil, nil, fmt.Errorf("failed to marshal P256 public key: %s", err)
		}
		hostPrivkey = newKeys.HostP256PrivateKey
		msg.HostPubkeyP256 = hostPubkeyPEM
	}

	blob, err := json.Marshal(msg)
	if err != nil {
		return nil, nil, nil, fmt.Errorf("failed to marshal DNClient message: %s", err)
	}

	// Make API call
	resp, err := c.postDNClient(ctx, message.DoConfigUpdate, blob, creds.HostID, creds.Counter, creds.PrivateKey)
	if err != nil {
		return nil, nil, nil, fmt.Errorf("failed to make API call to Defined Networking: %w", err)
	}

	// Verify the signature
	resultWrapper, err := verifySignature(resp, creds)
	if err != nil {
		return nil, nil, nil, err
	}

	// Consume the verified message
	result := message.DoConfigUpdateResponse{}
	err = json.Unmarshal(resultWrapper.Data.Message, &result)
	if err != nil {
		return nil, nil, nil, fmt.Errorf("failed to unmarshal response (%s): %s", resultWrapper.Data.Message, err)
	}

	// Verify the nonce
	if !bytes.Equal(result.Nonce, msg.Nonce) {
		return nil, nil, nil, fmt.Errorf("nonce mismatch between request (%s) and response (%s)", msg.Nonce, result.Nonce)
	}

	// Verify the counter
	if result.Counter <= creds.Counter {
		return nil, nil, nil, fmt.Errorf("counter in request (%d) should be less than counter in response (%d)", creds.Counter, result.Counter)
	}

	trustedKeys, err := keys.TrustedKeysFromPEM(result.TrustedKeys)
	if err != nil {
		return nil, nil, nil, fmt.Errorf("failed to load trusted keys from bundle: %s", err)
	}

	newCreds := &keys.Credentials{
		HostID:      creds.HostID,
		Counter:     result.Counter,
		PrivateKey:  hostPrivkey,
		TrustedKeys: trustedKeys,
	}

	meta := &ConfigMeta{
		Org: ConfigOrg{
			ID:   result.Organization.ID,
			Name: result.Organization.Name,
		},
		Network: ConfigNetwork{
			ID:   result.Network.ID,
			Name: result.Network.Name,
		},
		Host: ConfigHost{
			ID:          result.Host.ID,
			Name:        result.Host.Name,
			IPAddresses: mergeIPAddresses(result.Host.IPAddresses, result.Host.IPAddress),
		},
	}

	if result.EndpointOIDCMeta != nil {
		meta.EndpointOIDC = &ConfigEndpointOIDC{
			Email:     result.EndpointOIDCMeta.Email,
			ExpiresAt: result.EndpointOIDCMeta.ExpiresAt,
		}
	}

	return result.Config, newCreds, meta, nil
}

// verifySignature is a helper function that takes in an API call repsonse message and
// ensures it is signed by a trusted key. It returns the JSON unmarshalled response section
// if the message is valid JSON and the signature is trusted, otherwise it returns an error.
func verifySignature(resp []byte, creds keys.Credentials) (message.SignedResponseWrapper, error) {
	resultWrapper := message.SignedResponseWrapper{}
	err := json.Unmarshal(resp, &resultWrapper)
	if err != nil {
		return message.SignedResponseWrapper{}, fmt.Errorf("failed to unmarshal signed response wrapper: %s", err)
	}

	valid := false
	for _, caPubkey := range creds.TrustedKeys {
		if caPubkey.Verify(resultWrapper.Data.Message, resultWrapper.Data.Signature) {
			valid = true
			break
		}
	}
	if !valid {
		return message.SignedResponseWrapper{}, fmt.Errorf("failed to verify signed API result")
	}
	return resultWrapper, nil
}

func (c *Client) CommandResponse(ctx context.Context, creds keys.Credentials, responseToken string, response any) error {
	value, err := json.Marshal(message.CommandResponseRequest{
		ResponseToken: responseToken,
		Response:      response,
	})
	if err != nil {
		return fmt.Errorf("failed to marshal DNClient message: %s", err)
	}

	_, err = c.postDNClient(ctx, message.CommandResponse, value, creds.HostID, creds.Counter, creds.PrivateKey)
	return err
}

func (c *Client) StreamCommandResponse(ctx context.Context, creds keys.Credentials, responseToken string) (*StreamController, error) {
	value, err := json.Marshal(message.CommandResponseRequest{
		ResponseToken: responseToken,
	})
	if err != nil {
		return nil, fmt.Errorf("failed to marshal DNClient message: %s", err)
	}

	return c.streamingPostDNClient(ctx, message.CommandResponse, value, creds.HostID, creds.Counter, creds.PrivateKey)
}

func (c *Client) Reauthenticate(ctx context.Context, creds keys.Credentials) (*message.ReauthenticateResponse, error) {
	value, err := json.Marshal(message.ReauthenticateRequest{})
	if err != nil {
		return nil, fmt.Errorf("failed to marshal DNClient message: %s", err)
	}

	resp, err := c.postDNClient(ctx, message.Reauthenticate, value, creds.HostID, creds.Counter, creds.PrivateKey)
	if err != nil {
		return nil, err
	}

	resultWrapper, err := verifySignature(resp, creds)
	if err != nil {
		return nil, err
	}

	var response message.ReauthenticateResponse
	if err := json.Unmarshal(resultWrapper.Data.Message, &response); err != nil {
		return nil, fmt.Errorf("failed to unmarshal DNClient response: %s", err)
	}

	return &response, nil
}

// streamingPostDNClient wraps and signs the given dnclientRequestWrapper message, and makes a streaming API call.
// On success, it returns a StreamController to interact with the request. On error, the error is returned.
func (c *Client) streamingPostDNClient(ctx context.Context, reqType string, value []byte, hostID string, counter uint, privkey keys.PrivateKey) (*StreamController, error) {
	pr, pw := io.Pipe()

	postBody, err := SignRequestV1(reqType, value, hostID, counter, privkey)
	if err != nil {
		return nil, err
	}
	pbb := bytes.NewBuffer(postBody)

	endpointV1URL, err := urlPath(c.dnServer, message.EndpointV1)
	if err != nil {
		return nil, err
	}
	req, err := http.NewRequestWithContext(ctx, "POST", endpointV1URL, io.MultiReader(pbb, pr))
	if err != nil {
		return nil, err
	}

	done := make(chan struct{})
	sc := &StreamController{w: pw, done: done}

	go func() {
		defer close(done)

		resp, err := c.streamingClient.Do(req)
		if err != nil {
			sc.err.Store(fmt.Errorf("failed to call dnclient endpoint: %w", err))
			return
		}
		defer resp.Body.Close()

		respBody, err := io.ReadAll(resp.Body)
		if err != nil {
			sc.err.Store(fmt.Errorf("failed to read the response body: %s", err))
		}

		switch resp.StatusCode {
		case http.StatusOK:
			sc.respBytes = respBody
		case http.StatusUnauthorized:
			sc.err.Store(ErrInvalidCredentials)
		default:
			var errors struct {
				Errors message.APIResponseErrors
			}
			if err := json.Unmarshal(respBody, &errors); err != nil {
				sc.err.Store(fmt.Errorf("dnclient endpoint returned bad status code '%d', body: %s", resp.StatusCode, respBody))
			} else {
				sc.err.Store(errors.Errors.Err())
			}
		}
	}()

	return sc, nil
}

// postDNClient wraps and signs the given dnclientRequestWrapper message, and makes the API call.
// On success, it returns the response message body. On error, the error is returned.
func (c *Client) postDNClient(ctx context.Context, reqType string, value []byte, hostID string, counter uint, privkey keys.PrivateKey) ([]byte, error) {
	postBody, err := SignRequestV1(reqType, value, hostID, counter, privkey)
	if err != nil {
		return nil, err
	}

	endpointV1URL, err := urlPath(c.dnServer, message.EndpointV1)
	if err != nil {
		return nil, err
	}
	req, err := http.NewRequestWithContext(ctx, "POST", endpointV1URL, bytes.NewReader(postBody))
	if err != nil {
		return nil, err
	}
	resp, err := c.client.Do(req)
	if err != nil {
		return nil, fmt.Errorf("failed to call dnclient endpoint: %w", err)
	}
	defer resp.Body.Close()

	respBody, err := io.ReadAll(resp.Body)
	if err != nil {
		return nil, fmt.Errorf("failed to read the response body: %s", err)
	}

	switch resp.StatusCode {
	case http.StatusOK:
		return respBody, nil
	case http.StatusUnauthorized:
		return nil, ErrInvalidCredentials
	default:
		var errors struct {
			Errors message.APIResponseErrors
		}
		if err := json.Unmarshal(respBody, &errors); err != nil {
			return nil, fmt.Errorf("dnclient endpoint returned bad status code '%d', body: %s", resp.StatusCode, respBody)
		}
		return nil, errors.Errors.Err()
	}
}

// callAPI returns the request ID, requested response data, and any error if applicable.
func callAPI[T any](ctx context.Context, c *Client, method string, endpoint string, payload any) (string, *T, error) {
	dest, err := urlPath(c.dnServer, endpoint)
	if err != nil {
		return "", nil, err
	}

	var br io.Reader
	if payload != nil {
		b, err := json.Marshal(payload)
		if err != nil {
			return "", nil, fmt.Errorf("failed to marshal payload: %s", err)
		}
		br = bytes.NewReader(b)
	}

	req, err := http.NewRequestWithContext(ctx, method, dest, br)
	if err != nil {
		return "", nil, err
	}

	resp, err := c.client.Do(req)
	if err != nil {
		return "", nil, err
	}
	defer resp.Body.Close()

	reqID := resp.Header.Get("X-Request-ID")

	r := message.APIResponse[T]{}
	b, err := io.ReadAll(resp.Body)
	if err != nil {
		return reqID, nil, &APIError{e: fmt.Errorf("error reading response body: %s", err), ReqID: reqID}
	}

	if err := json.Unmarshal(b, &r); err != nil {
		return reqID, nil, &APIError{e: fmt.Errorf("error decoding JSON response: %s\nbody: %s", err, b), ReqID: reqID}
	}

	// Check for any errors returned by the API
	if err := r.Errors.Err(); err != nil {
		return reqID, nil, &APIError{e: err, ReqID: reqID}
	}

	// If we didn't detect an error in the response, but received a 4XX or 5XX status code, return error
	if resp.StatusCode >= 400 {
		return reqID, nil, &APIError{e: fmt.Errorf("received HTTP %d from API without error details\nbody: %s", resp.StatusCode, b), ReqID: reqID}
	}

	return reqID, &r.Data, nil
}

// StreamController is used for interacting with streaming requests to the API.
//
// When a streaming request is started in a background goroutine, a StreamController is returned to the caller to allow
// writing to the request body. The request will be sent when the caller closes the StreamController. The response body
// can be read by calling ResponseBytes, which will block until the response is received.
type StreamController struct {
	w         *io.PipeWriter
	respBytes []byte
	err       atomic.Value
	done      chan struct{}
}

// Err returns any error that occurred during the streaming request. If the request was successful, Err will return nil.
// Err should be called after Close to ensure the request has completed.
func (sc *StreamController) Err() error {
	err := sc.err.Load()
	if err == nil {
		return nil
	}
	return err.(error)
}

// Write implements the io.Writer interface for StreamController. It writes to the request body. If the StreamController
// has already encountered an error, it will be returned and nothing will be written.
func (sc *StreamController) Write(p []byte) (int, error) {
	if sc.Err() != nil {
		return 0, sc.Err()
	}

	n, err := sc.w.Write(p)
	if err != nil {
		sc.err.Store(err)
	}
	return n, err
}

// Close closes the StreamController, signaling that the request body is complete and the response can be read.
func (sc *StreamController) Close() error {
	err := sc.w.Close()
	<-sc.done
	return err
}

// ResponseBytes blocks until the response is received, then returns the response body. If an error occurred during the
// request, ResponseBytes will return the error.
func (sc *StreamController) ResponseBytes() ([]byte, error) {
	<-sc.done
	if sc.Err() != nil {
		return nil, sc.Err()
	}
	return sc.respBytes, nil
}

// uaTransport wraps an http.RoundTripper and sets the User-Agent header on all requests.
type uaTransport struct {
	useragent string
	T         http.RoundTripper
}

func (t *uaTransport) RoundTrip(req *http.Request) (*http.Response, error) {
	req.Header.Set("User-Agent", t.useragent)
	return t.T.RoundTrip(req)
}

func nonce() []byte {
	nonce := make([]byte, 16)
	if _, err := io.ReadFull(rand.Reader, nonce); err != nil {
		panic(err)
	}
	return nonce
}

func (c *Client) EndpointPreAuth(ctx context.Context) (*message.PreAuthData, error) {
	_, d, err := callAPI[message.PreAuthData](ctx, c, "POST", message.PreAuthEndpoint, nil)
	return d, err
}

func (c *Client) EndpointAuthPoll(ctx context.Context, pollCode string) (*message.EndpointAuthPollData, error) {
	pollURL := fmt.Sprintf("%s?pollToken=%s", message.AuthPollEndpoint, url.QueryEscape(pollCode))
	_, d, err := callAPI[message.EndpointAuthPollData](ctx, c, "GET", pollURL, nil)
	return d, err
}

func (c *Client) Downloads(ctx context.Context) (*message.DownloadsData, error) {
	_, d, err := callAPI[message.DownloadsData](ctx, c, "GET", message.DownloadsEndpoint, nil)
	return d, err
}

func urlPath(base, path string) (string, error) {
	baseURL, err := url.Parse(base)
	if err != nil {
		return "", fmt.Errorf("invalid base: %s", err)
	}

	pathURL, err := url.Parse(path)
	if err != nil {
		return "", fmt.Errorf("invalid path: %s", err)
	}

	finalURL := baseURL.ResolveReference(pathURL)
	return finalURL.String(), nil
}