Start implementing AddClient (not tested)

Author: sechmann

apiserver/api/api.go

@@ -18,6 +18,12 @@ type Peer struct {
 	IP        string
 }
 
+type ClientRegistrationRequest struct {
+	Username  string
+	PublicKey string
+	Serial    string
+}
+
 // TODO(jhrv): do actual filtering of the clients.
 // TODO(jhrv): keep cache of gateway access group members to remove AAD runtime dependency
 // gatewayConfig returns the clients for the gateway that has the group membership required
@@ -62,25 +68,41 @@ func (a *api) updateHealth(w http.ResponseWriter, r *http.Request) {
 	if err := json.NewDecoder(r.Body).Decode(&healthUpdates); err != nil {
 		defer r.Body.Close()
 
-		w.WriteHeader(http.StatusBadRequest)
-		w.Write([]byte(fmt.Sprintf("error during JSON unmarshal: %s\n", err)))
+		respondf(w, http.StatusBadRequest, "error during JSON unmarshal: %s\n", err)
 		return
 	}
 
 	// Abort status update if it contains incomplete entries
 	// is_healthy and serial is required
 	for _, s := range healthUpdates {
 		if s.Healthy == nil || len(s.Serial) == 0 {
-			w.WriteHeader(http.StatusBadRequest)
-			w.Write([]byte("missing required field\n"))
+			respondf(w, http.StatusBadRequest, "missing required field\n")
 			return
 		}
 	}
 
 	if err := a.db.UpdateClientStatus(healthUpdates); err != nil {
-		w.WriteHeader(http.StatusInternalServerError)
 		log.Error(err)
-		w.Write([]byte("unable to persist client statuses\n"))
+		respondf(w, http.StatusInternalServerError,"unable to persist client statuses\n")
 		return
 	}
 }
+
+func (a *api) registerClient(w http.ResponseWriter, r *http.Request) {
+	var reg ClientRegistrationRequest
+	if err := json.NewDecoder(r.Body).Decode(&reg); err != nil {
+		respondf(w, http.StatusBadRequest, "error during JSON unmarshal: %s\n", err)
+	}
+
+	if err := a.db.AddClient(reg.Username, reg.PublicKey, reg.Serial); err != nil {
+		respondf(w, http.StatusInternalServerError, "unable to add new peer: %s\n", err)
+	}
+}
+
+func respondf(w http.ResponseWriter, statusCode int, format string, args... interface{}) {
+	w.WriteHeader(statusCode)
+
+	if _, wErr := w.Write([]byte(fmt.Sprintf(format, args...))); wErr != nil {
+		log.Errorf("unable to write client response: %v", wErr)
+	}
+}

apiserver/api/router.go

@@ -16,6 +16,7 @@ func New(cfg Config) chi.Router {
 	r.Get("/gateways/{gateway}", api.gatewayConfig)
 	r.Get("/clients", api.clients)
 	r.Put("/clients/health", api.updateHealth)
+	r.Put("/clients/register", api.registerClient)
 
 	return r
 }

apiserver/database/database.go

@@ -3,14 +3,22 @@ package database
 import (
 	"context"
 	"fmt"
+	"github.com/jackc/pgx/v4"
+	"sync"
 	"time"
 
 	"github.com/jackc/pgx/v4/pgxpool"
 )
 
+const (
+	ControlPlaneCidr = "10.255.240.0/21"
+	DataPlaneCidr    = "10.255.248.0/21"
+)
+
 type APIServerDB interface {
 	ReadClients() ([]Client, error)
 	UpdateClientStatus([]Client) error
+	AddClient(username, publicKey, serial string) error
 }
 
 type Client struct {
@@ -101,3 +109,68 @@ func (d *database) UpdateClientStatus(clients []Client) error {
 
 	return tx.Commit(ctx)
 }
+
+var mux sync.Mutex
+
+func (d *database) AddClient(username, publicKey, serial string) error {
+	mux.Lock()
+	defer mux.Unlock()
+
+	ctx := context.Background()
+
+	tx, err := d.conn.Begin(ctx)
+	if err != nil {
+		return fmt.Errorf("start transaction: %w", err)
+	}
+
+	defer tx.Rollback(ctx)
+
+	ips, err := ips(tx, ctx)
+	if err != nil {
+		return fmt.Errorf("fetch ips: %w", err)
+	}
+
+	ip, err := FindAvailableIP(ControlPlaneCidr, ips)
+	if err != nil {
+		return fmt.Errorf("finding available ip: %w", err)
+	}
+
+	statement := `
+WITH
+  client_key AS
+    (INSERT INTO client (serial, healthy) VALUES ($1, false) RETURNING id),
+  peer_control_key AS
+    (INSERT INTO peer (public_key, ip, type) VALUES ($2, $3, 'control') RETURNING id)
+INSERT
+  INTO client_peer(client_id, peer_id)
+  (
+    SELECT client_key.id, peer_control_key.id
+    FROM client_key, peer_control_key
+  );
+`
+	_, err = tx.Exec(ctx, statement, serial, publicKey, ip)
+
+	if err != nil {
+		return fmt.Errorf("inserting new client: %w", err)
+	}
+
+	return nil
+}
+
+func ips(tx pgx.Tx, ctx context.Context) (ips []string, err error) {
+	rows, err := tx.Query(ctx, "SELECT ip FROM peer;")
+	if err != nil {
+		return nil, fmt.Errorf("get peers: %w", err)
+	}
+
+	if rows.Err() != nil {
+		return nil, fmt.Errorf("get peers: %w", err)
+	}
+
+	err = rows.Scan(ips)
+	if err != nil {
+		return nil, fmt.Errorf("scan peers: %w", err)
+	}
+
+	return
+}