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"
	"fmt"
	"io"
	"net"
	"net/http"
	"net/url"
	"sync/atomic"
	"time"

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

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

	client          *http.Client
	streamingClient *http.Client
}

// 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: (&net.Dialer{
						Timeout: 10 * time.Second,
					}).DialContext,
				},
				useragent: useragent,
			},
		},
		streamingClient: &http.Client{
			Timeout: 15 * time.Minute,
			Transport: &uaTransport{
				T: &http.Transport{
					Proxy:               http.ProxyFromEnvironment,
					TLSHandshakeTimeout: 10 * time.Second,
					DialContext: (&net.Dialer{
						Timeout: 10 * time.Second,
					}).DialContext,
				},
				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
}

type InvalidCredentialsError struct{}

func (e InvalidCredentialsError) Error() string {
	return "invalid credentials"
}

type EnrollMeta struct {
	OrganizationID   string
	OrganizationName string
}

// 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.
// 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
// containing organization info.
func (c *Client) Enroll(ctx context.Context, logger logrus.FieldLogger, code string) ([]byte, []byte, *keys.Credentials, *EnrollMeta, error) {
	logger.WithFields(logrus.Fields{"server": c.dnServer}).Debug("Making enrollment request to API")

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

	hostEd25519PublicKeyPEM, err := newKeys.HostEd25519PublicKey.MarshalPEM()
	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
	jv, err := json.Marshal(message.EnrollRequest{
		Code:               code,
		NebulaPubkeyX25519: newKeys.NebulaX25519PublicKeyPEM,
		HostPubkeyEd25519:  hostEd25519PublicKeyPEM,
		NebulaPubkeyP256:   newKeys.NebulaP256PublicKeyPEM,
		HostPubkeyP256:     hostP256PublicKeyPEM,
		Timestamp:          time.Now(),
	})
	if err != nil {
		return nil, nil, nil, nil, err
	}

	enrollURL, err := url.JoinPath(c.dnServer, message.EnrollEndpoint)
	if err != nil {
		return nil, nil, nil, nil, err
	}

	req, err := http.NewRequestWithContext(ctx, "POST", enrollURL, bytes.NewBuffer(jv))
	if err != nil {
		return nil, nil, nil, nil, err
	}

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

	// Log the request ID returned from the server
	reqID := resp.Header.Get("X-Request-ID")
	l := logger.WithFields(logrus.Fields{"statusCode": resp.StatusCode, "reqID": reqID})
	if resp.StatusCode == http.StatusOK {
		l.Info("Enrollment request returned success code")
	} else {
		l.Error("Enrollment request returned error code")
	}

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

	if err := json.Unmarshal(b, &r); err != nil {
		return nil, nil, nil, 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.ToError(); err != nil {
		return nil, nil, nil, nil, &APIError{e: fmt.Errorf("unexpected error during enrollment: %v", err), ReqID: reqID}
	}

	meta := &EnrollMeta{
		OrganizationID:   r.Data.Organization.ID,
		OrganizationName: r.Data.Organization.Name,
	}

	// Determine the private keys to save based on the network curve type
	var privkeyPEM []byte
	var privkey keys.PrivateKey
	switch r.Data.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.Data.Network.Curve), ReqID: reqID}
	}

	trustedKeys, err := keys.TrustedKeysFromPEM(r.Data.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.Data.HostID,
		PrivateKey:  privkey,
		Counter:     r.Data.Counter,
		TrustedKeys: trustedKeys,
	}
	return r.Data.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 and new DNClient API credentials.
//
// 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, error) {
	// Rotate keys
	var nebulaPrivkeyPEM []byte     // ECDH
	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.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, 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, 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, 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, fmt.Errorf("failed to make API call to Defined Networking: %w", err)
	}
	resultWrapper := message.SignedResponseWrapper{}
	err = json.Unmarshal(resp, &resultWrapper)
	if err != nil {
		return nil, nil, nil, fmt.Errorf("failed to unmarshal signed response wrapper: %s", err)
	}

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

	// Consume the verified message
	result := message.DoUpdateResponse{}
	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,
	}

	return result.Config, nebulaPrivkeyPEM, newCreds, 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)
}

// 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 := url.JoinPath(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(InvalidCredentialsError{})
		default:
			var errors struct {
				Errors message.APIErrors
			}
			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.ToError())
			}
		}
	}()

	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 := url.JoinPath(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, InvalidCredentialsError{}
	default:
		var errors struct {
			Errors message.APIErrors
		}
		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.ToError()
	}
}

// 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
}