internal/server/api/events_test.go

Ref: main v0.0.5   Size: 3.2 KiB   History

package api

import (
	"bufio"
	"context"
	"encoding/json"
	"net/http"
	"strings"
	"testing"
	"time"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
)

// mintTicket mints a one-time stream ticket via the admin API.
func mintTicket(t *testing.T, ts string, token string) string {
	t.Helper()
	resp := do(t, "POST", ts+"/api/v1/stream-tickets", token, nil)
	require.Equal(t, 201, resp.StatusCode)
	var out map[string]string
	require.NoError(t, json.NewDecoder(resp.Body).Decode(&out))
	require.NotEmpty(t, out["ticket"])
	return out["ticket"]
}

// readFirstSSEEvent connects to the events URL and returns true when a
// "event: state" frame arrives before the deadline.
func readFirstSSEEvent(t *testing.T, url string) (int, bool) {
	t.Helper()
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	req, _ := http.NewRequestWithContext(ctx, "GET", url, nil)
	resp, err := http.DefaultClient.Do(req)
	require.NoError(t, err)
	defer resp.Body.Close()
	if resp.StatusCode != 200 {
		return resp.StatusCode, false
	}
	sc := bufio.NewScanner(resp.Body)
	for sc.Scan() {
		if strings.HasPrefix(sc.Text(), "event: state") {
			return 200, true
		}
	}
	return 200, false
}

// TestStreamTicketFlow pins the SSE auth model: long-lived credentials never
// ride in a URL. A one-time short-TTL ticket is minted over an authenticated
// POST; the events stream consumes it; replay fails; the old ?token= path is
// gone.
func TestStreamTicketFlow(t *testing.T) {
	ts, _, _ := testServer(t)

	t.Run("mint requires a credential", func(t *testing.T) {
		resp := do(t, "POST", ts.URL+"/api/v1/stream-tickets", "wrong", nil)
		assert.Equal(t, 401, resp.StatusCode)
	})

	tick := mintTicket(t, ts.URL, testPAT)

	t.Run("valid ticket streams", func(t *testing.T) {
		code, gotState := readFirstSSEEvent(t, ts.URL+"/api/v1/events?ticket="+tick)
		assert.Equal(t, 200, code)
		assert.True(t, gotState, "first state frame must arrive")
	})

	t.Run("replay rejected", func(t *testing.T) {
		code, _ := readFirstSSEEvent(t, ts.URL+"/api/v1/events?ticket="+tick)
		assert.Equal(t, 401, code, "a consumed ticket must be rejected")
	})

	t.Run("garbage ticket rejected", func(t *testing.T) {
		code, _ := readFirstSSEEvent(t, ts.URL+"/api/v1/events?ticket=nope")
		assert.Equal(t, 401, code)
	})

	t.Run("PAT in query string is rejected", func(t *testing.T) {
		code, _ := readFirstSSEEvent(t, ts.URL+"/api/v1/events?token="+testPAT)
		assert.Equal(t, 401, code, "a PAT in the URL must not authenticate the stream; only tickets do")
	})
}

// TestStreamTicketExpiry pins the TTL with an injected clock, and that consume
// returns the tenant a ticket was minted for.
func TestStreamTicketExpiry(t *testing.T) {
	now := time.Unix(1_750_000_000, 0)
	tk := newTicketStore(func() time.Time { return now })
	tick := tk.mint("acme")
	now = now.Add(streamTicketTTL + time.Second)
	_, ok := tk.consume(tick)
	assert.False(t, ok, "expired ticket must not be consumable")

	fresh := tk.mint("acme")
	tenant, ok := tk.consume(fresh)
	assert.True(t, ok)
	assert.Equal(t, "acme", tenant, "consume returns the minting tenant")
	_, ok = tk.consume(fresh)
	assert.False(t, ok, "one-time: second consume fails")
}