Add some conifguration and logging for the hub

registrar/approval.go

@@ -0,0 +1,91 @@
+package registrar
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"log"
+	"os"
+
+	"github.com/google/uuid"
+	svc "go.fixergrid.net/servicedemon/registrar/internal/services"
+)
+
+const APPROVAL_TOPIC = "net.fixergrid.events.app.approved"
+
+type ApprovalListener struct {
+	svc.ApprovalRequester
+	svc.PubSub
+	svc.AppRepo
+	log ilogger
+}
+
+type option func(r *ApprovalListener)
+
+func OptionLog(l logger) option {
+	return func(r *ApprovalListener) {
+		r.log = internalLog{l}
+	}
+}
+
+func NewApprovalListener(ps svc.PubSub, requester svc.ApprovalRequester, repo svc.AppRepo, options ...option) ApprovalListener {
+	out := ApprovalListener{
+		PubSub:            ps,
+		ApprovalRequester: requester,
+		AppRepo:           repo,
+		log:               log.New(os.Stderr, "servicedemon/registrar - ", log.LstdFlags|log.Lshortfile),
+	}
+
+	for _, opt := range options {
+		opt(&out)
+	}
+	return out
+}
+
+func (a ApprovalListener) Run(ctx context.Context) {
+	newAppEvents := a.Subscribe(REGISTRATION_TOPIC)
+	approvalEvents := a.Subscribe(APPROVAL_TOPIC)
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case event := <-newAppEvents:
+			msg, err := readMessage(event)
+			if err != nil {
+				a.log.Printf("failed to read message: got '%s': %v", event, err)
+			}
+			a.SendApprovalRequest(msg["name"])
+		case event := <-approvalEvents:
+			msg, err := readMessage(event)
+			if err != nil {
+				a.log.Printf("approvalevent: failed to read message: got '%s': %v", event, err)
+			}
+			id := uuid.MustParse(msg["id"])
+			a.HandleApprovedRequest(id)
+		}
+	}
+}
+
+func (a ApprovalListener) SendApprovalRequest(name string) {
+	a.log.Printf("sent approval message for [%s]", name)
+}
+
+func (a ApprovalListener) HandleApprovedRequest(id uuid.UUID) {
+	res, err := a.ApproveApp(id)
+	if err != nil {
+		a.log.Printf("ERROR: couldn't approve request: %v", err)
+		return
+	}
+
+	a.log.Printf("approved application [%s]", res.ID)
+}
+
+func readMessage(event string) (map[string]string, error) {
+	out := map[string]string{}
+
+	err := json.Unmarshal([]byte(event), &out)
+	if err != nil {
+		return nil, fmt.Errorf("readMessage: %w", err)
+	}
+	return out, nil
+}

registrar/endpoints.go

@@ -0,0 +1,114 @@
+package registrar
+
+import (
+	"encoding/json"
+	"fmt"
+	"log"
+	"net/http"
+	"os"
+
+	"github.com/google/uuid"
+	svc "go.fixergrid.net/servicedemon/registrar/internal/services"
+)
+
+const (
+	REGISTRATION_TOPIC = "net.fixergrid.events.app.admission"
+)
+
+var validApprover string = "Nautilus Admins"
+
+type Registrar struct {
+	svc.Publisher
+	repo svc.AppRepo
+	log  ilogger
+}
+
+func WithLogger(l logger) RegistrarOpt {
+	return func(r *Registrar) {
+		r.log = internalLog{l}
+	}
+}
+
+type RegistrarOpt func(*Registrar)
+
+func NewRegistrar(publisher svc.Publisher, repo svc.AppRepo, options ...RegistrarOpt) Registrar {
+	r := Registrar{
+		Publisher: publisher,
+		repo:      repo,
+		log:       log.New(os.Stderr, "servicedemon/registrar - ", log.LstdFlags|log.Lshortfile),
+	}
+	for _, opt := range options {
+		opt(&r)
+	}
+	return r
+}
+
+func (r Registrar) HandleRegistration(resp http.ResponseWriter, req *http.Request) {
+	cert := req.TLS.PeerCertificates[0]
+
+	name := cert.DNSNames[0]
+	app, err := r.repo.GetApp(name)
+	switch err {
+	case svc.ErrApplicationNotFound:
+		id := r.repo.StartAppRegistration(name)
+		event := map[string]string{
+			"id":   id.String(),
+			"name": name,
+			"type": "ApplicationRegistrationSubmitted",
+		}
+
+		r.sendEvent(REGISTRATION_TOPIC, event)
+
+		resp.WriteHeader(http.StatusCreated)
+		fmt.Fprintf(resp, `{"status": "pending_approval"}`)
+	case nil:
+		resp.WriteHeader(http.StatusOK)
+		fmt.Fprintf(resp, `{"status": "%s"}`, app.State)
+	default:
+		resp.WriteHeader(http.StatusInternalServerError)
+		fmt.Fprintf(resp, `{"errors": ["%s"]}`, err)
+	}
+}
+
+func (r Registrar) HandleApproval(resp http.ResponseWriter, req *http.Request) {
+	in := req.URL.Path
+	cert := req.TLS.PeerCertificates[0]
+	OUs := cert.Subject.OrganizationalUnit
+
+	var ou string
+	if len(OUs) == 1 {
+		ou = OUs[0]
+	} else {
+		fmt.Fprintf(resp, `{"errors": ["missing OU"]}`)
+		resp.WriteHeader(http.StatusBadRequest)
+		return
+	}
+
+	if ou != validApprover {
+		fmt.Fprintf(resp, `{"errors": ["approval forbidden"]}`)
+		resp.WriteHeader(http.StatusForbidden)
+		return
+	}
+
+	appID := uuid.MustParse(in)
+
+	event := map[string]string{
+		"id":       appID.String(),
+		"approver": ou,
+		"type":     "ApplicationRegistrationApproved",
+	}
+	r.sendEvent(APPROVAL_TOPIC, event)
+	resp.WriteHeader(http.StatusOK)
+	fmt.Fprintf(resp, `{"status": "registered"}`)
+}
+
+func (r Registrar) sendEvent(topic string, event map[string]string) error {
+	b, err := json.Marshal(event)
+	if err != nil {
+		return fmt.Errorf("registrar: sendEvent: failed to marshal: %w", err)
+	}
+
+	r.log.Printf("sending event: %s", event)
+	r.Publish(topic, string(b))
+	return nil
+}

registrar/endpoints_test.go

@@ -0,0 +1,139 @@
+package registrar
+
+import (
+	"crypto/tls"
+	"crypto/x509"
+	"encoding/json"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"go.fixergrid.net/servicedemon/pubsub"
+)
+
+func TestHandleRegisterPendingApproval(t *testing.T) {
+	resp := httptest.NewRecorder()
+
+	req := NewTestRequest(t, http.MethodPost, "http://example.com/v1/register", nil).
+		WithFakeTLSState("dev-app-1.delivery.engineering").
+		Create()
+
+	repo := NewRepo()
+	pubsub := pubsub.New()
+	subscriber := pubsub.Subscribe(REGISTRATION_TOPIC)
+
+	NewRegistrar(pubsub, repo).HandleRegistration(resp, req)
+
+	assert.Equal(t, http.StatusCreated, resp.Code)
+
+	body, err := io.ReadAll(resp.Body)
+	require.NoError(t, err)
+
+	assert.Equal(t, []byte(`{"status": "pending_approval"}`), body)
+
+	assert.Equal(t, 1, repo.PendingApprovalCount())
+
+	expectedEvent := map[string]string{
+		"name": "dev-app-1.delivery.engineering",
+		"type": "ApplicationRegistrationSubmitted",
+	}
+	assertMessage(t, expectedEvent, subscriber)
+}
+
+func TestHandleRegisterAlreadyRegistered(t *testing.T) {
+	resp := httptest.NewRecorder()
+
+	req := NewTestRequest(t, http.MethodPost, "http://example.com/v1/register", nil).
+		WithFakeTLSState("dev-app-1.delivery.engineering").
+		Create()
+
+	pubsub := pubsub.New()
+	repo := NewRepo()
+	appID := repo.StartAppRegistration("dev-app-1.delivery.engineering")
+	repo.ApproveApp(appID)
+
+	NewRegistrar(pubsub, repo).HandleRegistration(resp, req)
+
+	assert.Equal(t, http.StatusOK, resp.Code)
+
+	body, err := io.ReadAll(resp.Body)
+	require.NoError(t, err)
+
+	assert.Equal(t, []byte(`{"status": "registered"}`), body)
+}
+
+func TestHandleRegisterNotYetApproved(t *testing.T) {
+	resp := httptest.NewRecorder()
+
+	req := NewTestRequest(t, http.MethodPost, "http://example.com/v1/register", nil).
+		WithFakeTLSState("dev-app-1.delivery.engineering").
+		Create()
+
+	pubsub := pubsub.New()
+	repo := NewRepo()
+	repo.StartAppRegistration("dev-app-1.delivery.engineering")
+
+	NewRegistrar(pubsub, repo).HandleRegistration(resp, req)
+
+	assert.Equal(t, http.StatusOK, resp.Code)
+
+	body, err := io.ReadAll(resp.Body)
+	require.NoError(t, err)
+
+	assert.Equal(t, []byte(`{"status": "pending_approval"}`), body)
+}
+
+func assertMessage(t *testing.T, expected map[string]string, sender <-chan string) {
+	timer := time.After(2 * time.Second)
+
+	for {
+		select {
+		case msg := <-sender:
+			compareMessage(t, expected, msg)
+			return
+		case <-timer:
+			t.Fatal("failed waiting for message")
+		}
+	}
+}
+
+func compareMessage(t *testing.T, expected map[string]string, observed string) {
+	observedEvent := map[string]string{}
+
+	require.NoError(t, json.Unmarshal([]byte(observed), &observedEvent))
+
+	assert.NotEmpty(t, observedEvent["id"])
+	for k, v := range expected {
+		assert.Equal(t, v, observedEvent[k])
+	}
+}
+
+type TestRequest http.Request
+
+func NewTestRequest(t *testing.T, method string, url string, body io.Reader) TestRequest {
+	req, err := http.NewRequest(method, url, body)
+	require.NoError(t, err)
+	return TestRequest(*req)
+}
+func (tr TestRequest) WithFakeTLSState(dnsName string) TestRequest {
+	cert := x509.Certificate{
+		DNSNames: []string{
+			dnsName,
+		},
+	}
+
+	tr.TLS = &tls.ConnectionState{
+		PeerCertificates: []*x509.Certificate{&cert},
+	}
+
+	return tr
+}
+
+func (tr TestRequest) Create() *http.Request {
+	req := http.Request(tr)
+	return &req
+}

registrar/internal/persistence/fake_repo.go

@@ -0,0 +1,66 @@
+package persistence
+
+import (
+	"github.com/google/uuid"
+	svc "go.fixergrid.net/servicedemon/registrar/internal/services"
+)
+
+type FakeRepo struct {
+	apps []*svc.Application
+}
+
+func NewFakeRepo() *FakeRepo {
+	return &FakeRepo{
+		apps: make([]*svc.Application, 0),
+	}
+}
+func (repo FakeRepo) IsRegistered(name string) bool {
+	for _, app := range repo.apps {
+		if name == app.Name && app.State == "registered" {
+			return true
+		}
+	}
+	return false
+}
+
+func (repo FakeRepo) GetApp(name string) (svc.Application, error) {
+	for _, app := range repo.apps {
+		if name == app.Name {
+			return *app, nil
+		}
+	}
+
+	return svc.Application{}, svc.ErrApplicationNotFound
+}
+
+func (repo FakeRepo) PendingApprovalCount() int {
+	count := 0
+	for _, app := range repo.apps {
+		if app.State == "pending_approval" {
+			count += 1
+		}
+	}
+	return count
+}
+
+func (repo *FakeRepo) StartAppRegistration(name string) uuid.UUID {
+	newApp := &svc.Application{
+		ID:    uuid.New(),
+		Name:  name,
+		State: "pending_approval",
+	}
+	repo.apps = append(repo.apps, newApp)
+	return newApp.ID
+}
+
+func (repo *FakeRepo) ApproveApp(id uuid.UUID) (svc.Application, error) {
+	for _, application := range repo.apps {
+		if application.ID == id {
+			application.State = "registered"
+			return *application, nil
+		}
+	}
+
+	return svc.Application{}, svc.ErrApplicationNotFound
+
+}

registrar/internal/services/app_registration.go

@@ -0,0 +1,2 @@
+package services
+

registrar/internal/services/approvals.go

@@ -0,0 +1,5 @@
+package services
+
+type ApprovalRequester interface {
+	SendApprovalRequest(appName string) error
+}

registrar/internal/services/pubsub.go

@@ -0,0 +1,14 @@
+package services
+
+type Publisher interface {
+	Publish(topic string, message string)
+}
+
+type PubSub interface {
+	Publisher
+	Subscriber
+}
+
+type Subscriber interface {
+	Subscribe(topic string) <-chan string
+}

registrar/internal/services/repo.go

@@ -0,0 +1,27 @@
+package services
+
+import (
+	"errors"
+	"fmt"
+
+	"github.com/google/uuid"
+)
+
+var (
+	ErrAppRepo             = errors.New("AppRepo")
+	ErrApplicationNotFound = fmt.Errorf("%w: application not found", ErrAppRepo)
+)
+
+type Application struct {
+	ID    uuid.UUID
+	Name  string
+	State string
+}
+
+type AppRepo interface {
+	IsRegistered(name string) bool
+	PendingApprovalCount() int
+	GetApp(name string) (Application, error)
+	StartAppRegistration(name string) uuid.UUID
+	ApproveApp(id uuid.UUID) (Application, error)
+}

registrar/logging.go

@@ -0,0 +1,33 @@
+package registrar
+
+import "fmt"
+
+type logger interface {
+	Output(int, string) error
+}
+
+type ilogger interface {
+	logger
+	Print(...interface{})
+	Printf(string, ...interface{})
+	Println(...interface{})
+}
+
+type internalLog struct {
+	logger
+}
+
+// Println replicates the behaviour of the standard logger.
+func (t internalLog) Println(v ...interface{}) {
+	t.Output(2, fmt.Sprintln(v...))
+}
+
+// Printf replicates the behaviour of the standard logger.
+func (t internalLog) Printf(format string, v ...interface{}) {
+	t.Output(2, fmt.Sprintf(format, v...))
+}
+
+// Print replicates the behaviour of the standard logger.
+func (t internalLog) Print(v ...interface{}) {
+	t.Output(2, fmt.Sprint(v...))
+}

registrar/repo.go

@@ -0,0 +1,10 @@
+package registrar
+
+import (
+	persist "go.fixergrid.net/servicedemon/registrar/internal/persistence"
+	svc "go.fixergrid.net/servicedemon/registrar/internal/services"
+)
+
+func NewRepo() svc.AppRepo {
+	return persist.NewFakeRepo()
+}