fix!: order privilege events by sequence number and merge snapshots by position #333
No files matched your search
@@ -23,10 +23,9 @@ import client "gitea.unbound.se/shiny/authz_client"
|
||||
handler := client.New(client.WithBaseURL("http://authz-service"))
|
||||
|
||||
// Check user privileges
|
||||
privileges := handler.Get(email, companyID)
|
||||
if privileges.Invoicing {
|
||||
// User has invoicing privileges
|
||||
}
|
||||
allowed := handler.IsAllowed(email, companyID, func(p client.CompanyPrivileges) bool {
|
||||
return p.Invoicing
|
||||
})
|
||||
```
|
||||
|
||||
### Privileges
|
||||
@@ -43,4 +42,10 @@ The `CompanyPrivileges` struct contains permission flags:
|
||||
|
||||
### Event Handling
|
||||
|
||||
Registers per-replica (transient) go-messaging-amqp consumers for privilege update events from the authz-service (`Setup()`), keeping the local privilege cache up-to-date. Each routing key gets its own queue, so events for one company arrive in any order — `Process` must never discard state it didn't itself record. `User.Added` therefore creates the company entry only when it is missing: a `Privilege.Added` processed ahead of it would otherwise be overwritten, and nothing re-reads the privilege until the next `Fetch()`. The converse doesn't hold yet — `setPrivileges` still creates an entry for a company it never saw a `User.Added` for, so a `Privilege.*` delivered after `User.Removed` resurrects an all-false membership. Don't combine `Setup()` with go-messaging-amqp's `WithReconnect`: a reconnect declares new per-replica queues, so revocations published during the outage are lost unless `Fetch()` runs again. Services exit on connection loss (`CloseListener`) and re-fetch on start.
|
||||
Registers per-replica (transient) go-messaging-amqp consumers for privilege events from authz-service (`Setup()`). Each routing key gets its own queue, so events arrive in any order. `Process` orders them by the `sequenceNo` authz-service's event store stamps on every event: all four events come from authz-service's `Company` aggregate, so the global sequence number orders them within a company. Each (email, company) keeps the sequence number of the last fact about membership and about each privilege; an event only overrides older facts. `User.Removed` stamps every privilege, so a late grant can't come back. An event without `sequenceNo` fails closed: additions are dropped, and removals apply and block every later event for those facts until the next snapshot. An event with a negative or implausibly large `sequenceNo` is dropped. Tests that seed the handler must set `SequenceNo`.
|
||||
|
||||
### Startup
|
||||
|
||||
Call `Fetch()` **after** `conn.Start` (consumers bound), never before: events published between the snapshot and the binding would otherwise be lost. authz-service only serves a snapshot whose read view has applied every stored event. It sends that position in the `X-Authz-Sequence` header and answers 503 while the read view lags (backlog, backfill, reset); `Fetch` retries 503 for about a minute. `Fetch` merges the snapshot as facts at that sequence number (newer event facts win, pairs missing from the snapshot are removed) and drops later events at or below it. A snapshot older than one already merged is ignored, so concurrent or repeated fetches are safe. A snapshot without the header merges at 0 and logs a warning (rollout window only).
|
||||
|
||||
Don't combine `Setup()` with go-messaging-amqp's `WithReconnect`: a reconnect declares new per-replica queues, so revocations published during the outage are lost unless `Fetch()` runs again. Services exit on connection loss (`CloseListener`) and re-fetch on start.
|
||||
@@ -3,16 +3,29 @@ package client
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"log/slog"
|
||||
"math"
|
||||
"net/http"
|
||||
"reflect"
|
||||
"strconv"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
goamqp "codeberg.org/messaging/go-messaging-amqp"
|
||||
spec "codeberg.org/messaging/messaging"
|
||||
)
|
||||
|
||||
// SequenceHeader carries the global sequence number of the authz-service read view the
|
||||
// /authz snapshot was read at.
|
||||
const SequenceHeader = "X-Authz-Sequence"
|
||||
|
||||
// pending marks a fact cleared by an event without a sequence number. No event can
|
||||
// override it; only a snapshot can.
|
||||
const pending = math.MaxInt
|
||||
|
||||
// CompanyPrivileges contains the privileges for a combination of email address and company id
|
||||
type CompanyPrivileges struct {
|
||||
Admin bool `json:"admin"`
|
||||
@@ -25,13 +38,84 @@ type CompanyPrivileges struct {
|
||||
Salary bool `json:"salary"`
|
||||
}
|
||||
|
||||
// field returns the flag for privilege, or nil for an unknown privilege.
|
||||
func (c *CompanyPrivileges) field(privilege Privilege) *bool {
|
||||
switch privilege {
|
||||
case PrivilegeAdmin:
|
||||
return &c.Admin
|
||||
case PrivilegeCompany:
|
||||
return &c.Company
|
||||
case PrivilegeConsumer:
|
||||
return &c.Consumer
|
||||
case PrivilegeTime:
|
||||
return &c.Time
|
||||
case PrivilegeInvoicing:
|
||||
return &c.Invoicing
|
||||
case PrivilegeAccounting:
|
||||
return &c.Accounting
|
||||
case PrivilegeSupplier:
|
||||
return &c.Supplier
|
||||
case PrivilegeSalary:
|
||||
return &c.Salary
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// companyState is what the handler knows about one email in one company. Every
|
||||
// privilege event for a company comes from authz-service's Company aggregate, so
|
||||
// its global sequence number orders the events for the pair. The four routing
|
||||
// keys arrive on separate queues in any order; a fact is only overwritten by a
|
||||
// fact with a higher sequence number.
|
||||
type companyState struct {
|
||||
privileges CompanyPrivileges
|
||||
// privilegeSeq is the sequence number of the last fact about each privilege.
|
||||
privilegeSeq map[Privilege]int
|
||||
// member is whether the user belongs to the company, as of memberSeq. A
|
||||
// User.Removed stamps every privilege with its sequence number, so a grant
|
||||
// older than the removal can't come back.
|
||||
member bool
|
||||
memberSeq int
|
||||
}
|
||||
|
||||
func newCompanyState() *companyState {
|
||||
return &companyState{privilegeSeq: map[Privilege]int{}}
|
||||
}
|
||||
|
||||
// clearPending turns facts cleared without a sequence number into unordered facts, so
|
||||
// the snapshot being merged replaces them.
|
||||
func (s *companyState) clearPending() {
|
||||
if s.memberSeq == pending {
|
||||
s.memberSeq = 0
|
||||
}
|
||||
for p, seq := range s.privilegeSeq {
|
||||
if seq == pending {
|
||||
s.privilegeSeq[p] = 0
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// maxSeq is the highest sequence number of any fact in the state.
|
||||
func (s *companyState) maxSeq() int {
|
||||
m := s.memberSeq
|
||||
for _, seq := range s.privilegeSeq {
|
||||
m = max(m, seq)
|
||||
}
|
||||
return m
|
||||
}
|
||||
|
||||
// PrivilegeHandler processes PrivilegeAdded-events and fetches the initial set of privileges from an authz-service
|
||||
type PrivilegeHandler struct {
|
||||
*sync.RWMutex
|
||||
client *http.Client
|
||||
baseURL string
|
||||
apiKey string
|
||||
privileges map[string]map[string]*CompanyPrivileges
|
||||
client *http.Client
|
||||
baseURL string
|
||||
apiKey string
|
||||
state map[string]map[string]*companyState
|
||||
// floor is the sequence number of the newest snapshot. Its effects are all in
|
||||
// the state, so an event at or below it is stale.
|
||||
floor int
|
||||
// retries is how many times Fetch retries a snapshot authz-service isn't ready to serve.
|
||||
retries int
|
||||
retryDelay time.Duration
|
||||
}
|
||||
|
||||
// OptsFunc is used to configure the PrivilegeHandler
|
||||
@@ -55,9 +139,11 @@ func WithAPIKey(key string) OptsFunc {
|
||||
func New(opts ...OptsFunc) *PrivilegeHandler {
|
||||
handler := &PrivilegeHandler{
|
||||
RWMutex: &sync.RWMutex{},
|
||||
client: &http.Client{},
|
||||
client: &http.Client{Timeout: 30 * time.Second},
|
||||
baseURL: "http://authz-service",
|
||||
privileges: map[string]map[string]*CompanyPrivileges{},
|
||||
state: map[string]map[string]*companyState{},
|
||||
retries: 60,
|
||||
retryDelay: time.Second,
|
||||
}
|
||||
for _, opt := range opts {
|
||||
opt(handler)
|
||||
@@ -65,8 +151,27 @@ func New(opts ...OptsFunc) *PrivilegeHandler {
|
||||
return handler
|
||||
}
|
||||
|
||||
// Fetch the initial set of privileges from an authz-service
|
||||
// Fetch a snapshot of all privileges from an authz-service and merge it into the state.
|
||||
//
|
||||
// Call it after the AMQP connection has started: authz-service only serves a snapshot
|
||||
// that includes every stored event, so together with the bound queues no event is
|
||||
// missed. While its read view is behind it answers 503, and Fetch retries. The snapshot
|
||||
// is applied as facts at its sequence number: a fact from an event newer than the
|
||||
// snapshot is kept, everything else is replaced, and a pair missing from the snapshot is
|
||||
// removed. A snapshot older than one already merged is ignored.
|
||||
func (h *PrivilegeHandler) Fetch() error {
|
||||
for attempt := 0; ; attempt++ {
|
||||
err := h.fetch()
|
||||
if !errors.Is(err, errNotReady) || attempt >= h.retries {
|
||||
return err
|
||||
}
|
||||
time.Sleep(h.retryDelay)
|
||||
}
|
||||
}
|
||||
|
||||
var errNotReady = errors.New("fetch privileges: authz-service read view not ready")
|
||||
|
||||
func (h *PrivilegeHandler) fetch() error {
|
||||
req, err := http.NewRequest(http.MethodGet, fmt.Sprintf("%s/authz", h.baseURL), nil)
|
||||
if err != nil {
|
||||
return err
|
||||
@@ -80,21 +185,83 @@ func (h *PrivilegeHandler) Fetch() error {
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer func() { _ = resp.Body.Close() }()
|
||||
|
||||
if resp.StatusCode == http.StatusServiceUnavailable {
|
||||
return errNotReady
|
||||
}
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
return fmt.Errorf("fetch privileges: unexpected status %s", resp.Status)
|
||||
}
|
||||
|
||||
buff, err := io.ReadAll(resp.Body)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
h.Lock()
|
||||
defer h.Unlock()
|
||||
err = json.Unmarshal(buff, &h.privileges)
|
||||
if err != nil {
|
||||
var snapshot map[string]map[string]CompanyPrivileges
|
||||
if err := json.Unmarshal(buff, &snapshot); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
seq := 0
|
||||
if v := resp.Header.Get(SequenceHeader); v != "" {
|
||||
seq, err = strconv.Atoi(v)
|
||||
if err != nil || seq < 0 {
|
||||
return fmt.Errorf("fetch privileges: invalid %s header %q", SequenceHeader, v)
|
||||
}
|
||||
} else {
|
||||
// ponytail: only during the rollout window, before authz-service sends the header.
|
||||
slog.Warn("authz snapshot has no sequence number; events older than it can revert it", "header", SequenceHeader)
|
||||
}
|
||||
|
||||
h.Lock()
|
||||
defer h.Unlock()
|
||||
h.merge(snapshot, seq)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (h *PrivilegeHandler) merge(snapshot map[string]map[string]CompanyPrivileges, seq int) {
|
||||
if seq < h.floor {
|
||||
// A newer snapshot is already merged; this one could only bring back removed state.
|
||||
slog.Warn("ignoring authz snapshot older than the one already merged", "seq", seq, "floor", h.floor)
|
||||
return
|
||||
}
|
||||
for email, companies := range h.state {
|
||||
for companyID, s := range companies {
|
||||
s.clearPending()
|
||||
if _, exists := snapshot[email][companyID]; exists {
|
||||
continue
|
||||
}
|
||||
if s.maxSeq() <= seq {
|
||||
// Nothing newer than the snapshot; the floor keeps stale events out.
|
||||
delete(companies, companyID)
|
||||
continue
|
||||
}
|
||||
removeUser(s, seq, false)
|
||||
}
|
||||
if len(companies) == 0 {
|
||||
delete(h.state, email)
|
||||
}
|
||||
}
|
||||
for email, companies := range snapshot {
|
||||
for companyID, privileges := range companies {
|
||||
s := h.company(email, companyID)
|
||||
if s.memberSeq <= seq {
|
||||
s.member = true
|
||||
s.memberSeq = seq
|
||||
}
|
||||
for _, p := range AllPrivilege {
|
||||
if s.privilegeSeq[p] <= seq {
|
||||
*s.privileges.field(p) = *privileges.field(p)
|
||||
s.privilegeSeq[p] = seq
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
h.floor = max(h.floor, seq)
|
||||
}
|
||||
|
||||
func (h *PrivilegeHandler) Setup() []goamqp.Setup {
|
||||
return []goamqp.Setup{
|
||||
goamqp.TransientEventStreamConsumer("User.Added", process[UserAdded](h)),
|
||||
@@ -116,71 +283,118 @@ func process[T privilegeEvent](h *PrivilegeHandler) spec.EventHandler[T] {
|
||||
}
|
||||
}
|
||||
|
||||
// Process privilege-related events and update the internal state
|
||||
// Process privilege-related events and update the internal state.
|
||||
//
|
||||
// Events are applied by sequence number, not by arrival order. An event without a
|
||||
// sequence number can't be ordered, so it fails closed: a removal is applied and an
|
||||
// addition is dropped.
|
||||
func (h *PrivilegeHandler) Process(msg any) error {
|
||||
h.Lock()
|
||||
defer h.Unlock()
|
||||
|
||||
switch ev := msg.(type) {
|
||||
case *UserAdded:
|
||||
// Keep the privileges already recorded for the company. Each routing key has
|
||||
// its own transient queue, so a Privilege.Added published after this event can
|
||||
// be processed before it; overwriting the entry here would drop that privilege
|
||||
// until the next Fetch, which only runs at start.
|
||||
priv, exists := h.privileges[ev.Email]
|
||||
if !exists {
|
||||
priv = map[string]*CompanyPrivileges{}
|
||||
h.privileges[ev.Email] = priv
|
||||
if h.stale(ev.SequenceNo, true, ev) {
|
||||
return nil
|
||||
}
|
||||
if _, exists := priv[ev.CompanyID]; !exists {
|
||||
priv[ev.CompanyID] = &CompanyPrivileges{}
|
||||
s := h.company(ev.Email, ev.CompanyID)
|
||||
if ev.SequenceNo > s.memberSeq {
|
||||
s.member = true
|
||||
s.memberSeq = ev.SequenceNo
|
||||
}
|
||||
return nil
|
||||
case *UserRemoved:
|
||||
if priv, exists := h.privileges[ev.Email]; exists {
|
||||
delete(priv, ev.CompanyID)
|
||||
if h.stale(ev.SequenceNo, false, ev) {
|
||||
return nil
|
||||
}
|
||||
removeUser(h.company(ev.Email, ev.CompanyID), ev.SequenceNo, ev.SequenceNo == 0)
|
||||
return nil
|
||||
case *PrivilegeAdded:
|
||||
h.setPrivileges(ev.Email, ev.CompanyID, ev.Privilege, true)
|
||||
if h.stale(ev.SequenceNo, true, ev) {
|
||||
return nil
|
||||
}
|
||||
h.setPrivilege(ev.Email, ev.CompanyID, ev.Privilege, ev.SequenceNo, true)
|
||||
return nil
|
||||
case *PrivilegeRemoved:
|
||||
h.setPrivileges(ev.Email, ev.CompanyID, ev.Privilege, false)
|
||||
if h.stale(ev.SequenceNo, false, ev) {
|
||||
return nil
|
||||
}
|
||||
h.setPrivilege(ev.Email, ev.CompanyID, ev.Privilege, ev.SequenceNo, false)
|
||||
return nil
|
||||
default:
|
||||
fmt.Printf("Got unexpected message type (%s): '%+v'\n", reflect.TypeOf(msg).String(), msg)
|
||||
slog.Error("unexpected privilege message type", "type", reflect.TypeOf(msg).String())
|
||||
return fmt.Errorf("unexpected event type: '%s'", reflect.TypeOf(msg))
|
||||
}
|
||||
}
|
||||
|
||||
func (h *PrivilegeHandler) setPrivileges(email, companyId string, privilege Privilege, set bool) {
|
||||
if priv, exists := h.privileges[email]; exists {
|
||||
if c, exists := priv[companyId]; exists {
|
||||
switch privilege {
|
||||
case PrivilegeAdmin:
|
||||
c.Admin = set
|
||||
case PrivilegeCompany:
|
||||
c.Company = set
|
||||
case PrivilegeConsumer:
|
||||
c.Consumer = set
|
||||
case PrivilegeTime:
|
||||
c.Time = set
|
||||
case PrivilegeInvoicing:
|
||||
c.Invoicing = set
|
||||
case PrivilegeAccounting:
|
||||
c.Accounting = set
|
||||
case PrivilegeSupplier:
|
||||
c.Supplier = set
|
||||
case PrivilegeSalary:
|
||||
c.Salary = set
|
||||
}
|
||||
} else {
|
||||
priv[companyId] = &CompanyPrivileges{}
|
||||
h.setPrivileges(email, companyId, privilege, set)
|
||||
// stale reports whether an event must be skipped: an addition without a sequence
|
||||
// number, or any event already covered by the latest snapshot.
|
||||
func (h *PrivilegeHandler) stale(seq int, addition bool, ev any) bool {
|
||||
if seq < 0 || seq >= pending {
|
||||
slog.Error("dropping privilege event with invalid sequence number", "type", reflect.TypeOf(ev).String(), "seq", seq)
|
||||
return true
|
||||
}
|
||||
if seq == 0 {
|
||||
if addition {
|
||||
slog.Warn("dropping privilege addition without sequence number", "type", reflect.TypeOf(ev).String())
|
||||
}
|
||||
} else {
|
||||
h.privileges[email] = map[string]*CompanyPrivileges{}
|
||||
h.setPrivileges(email, companyId, privilege, set)
|
||||
return addition
|
||||
}
|
||||
return seq <= h.floor
|
||||
}
|
||||
|
||||
func (h *PrivilegeHandler) company(email, companyID string) *companyState {
|
||||
companies, exists := h.state[email]
|
||||
if !exists {
|
||||
companies = map[string]*companyState{}
|
||||
h.state[email] = companies
|
||||
}
|
||||
s, exists := companies[companyID]
|
||||
if !exists {
|
||||
s = newCompanyState()
|
||||
companies[companyID] = s
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
// removeUser removes the user at seq, overriding every fact older than seq. An
|
||||
// unordered removal (all) overrides every fact and marks them pending, so no event can
|
||||
// bring them back before the next snapshot.
|
||||
func removeUser(s *companyState, seq int, all bool) {
|
||||
stamp := seq
|
||||
if all {
|
||||
stamp = pending
|
||||
}
|
||||
if all || s.memberSeq < seq {
|
||||
s.member = false
|
||||
s.memberSeq = max(s.memberSeq, stamp)
|
||||
}
|
||||
for _, p := range AllPrivilege {
|
||||
if all || s.privilegeSeq[p] < seq {
|
||||
*s.privileges.field(p) = false
|
||||
s.privilegeSeq[p] = max(s.privilegeSeq[p], stamp)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (h *PrivilegeHandler) setPrivilege(email, companyID string, privilege Privilege, seq int, set bool) {
|
||||
if !privilege.IsValid() {
|
||||
return
|
||||
}
|
||||
s := h.company(email, companyID)
|
||||
if seq != 0 && seq <= s.privilegeSeq[privilege] {
|
||||
return
|
||||
}
|
||||
*s.privileges.field(privilege) = set
|
||||
if seq == 0 {
|
||||
// Only an unordered removal gets here; keep it until the next snapshot.
|
||||
seq = pending
|
||||
}
|
||||
s.privilegeSeq[privilege] = max(s.privilegeSeq[privilege], seq)
|
||||
// authz-service's aggregate adds the user when a privilege is granted.
|
||||
if set && seq > s.memberSeq {
|
||||
s.member = true
|
||||
s.memberSeq = seq
|
||||
}
|
||||
}
|
||||
|
||||
@@ -188,12 +402,10 @@ func (h *PrivilegeHandler) setPrivileges(email, companyId string, privilege Priv
|
||||
func (h *PrivilegeHandler) CompaniesByUser(email string, predicate func(privileges CompanyPrivileges) bool) []string {
|
||||
h.RLock()
|
||||
defer h.RUnlock()
|
||||
var result []string
|
||||
if p, exists := h.privileges[email]; exists {
|
||||
for k, v := range p {
|
||||
if predicate(*v) {
|
||||
result = append(result, k)
|
||||
}
|
||||
result := []string{}
|
||||
for k, s := range h.state[email] {
|
||||
if s.member && predicate(s.privileges) {
|
||||
result = append(result, k)
|
||||
}
|
||||
}
|
||||
return result
|
||||
@@ -203,11 +415,8 @@ func (h *PrivilegeHandler) CompaniesByUser(email string, predicate func(privileg
|
||||
func (h *PrivilegeHandler) IsAllowed(email, companyID string, predicate func(privileges CompanyPrivileges) bool) bool {
|
||||
h.RLock()
|
||||
defer h.RUnlock()
|
||||
if p, exists := h.privileges[email]; exists {
|
||||
if v, exists := p[companyID]; exists {
|
||||
return predicate(*v)
|
||||
}
|
||||
if s, exists := h.state[email][companyID]; exists && s.member {
|
||||
return predicate(s.privileges)
|
||||
}
|
||||
|
||||
return false
|
||||
}
|
||||
+393
-61
@@ -3,11 +3,14 @@ package client
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"math"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"sort"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
goamqp "codeberg.org/messaging/go-messaging-amqp"
|
||||
spec "codeberg.org/messaging/messaging"
|
||||
@@ -27,9 +30,10 @@ func TestPrivilegeHandler_Process_PrivilegeRemoved(t *testing.T) {
|
||||
handler := New(WithBaseURL("base"))
|
||||
|
||||
err := handler.Process(&PrivilegeAdded{
|
||||
Email: "jim@example.org",
|
||||
CompanyID: "abc-123",
|
||||
Privilege: PrivilegeAdmin,
|
||||
SequenceNo: 1,
|
||||
Email: "jim@example.org",
|
||||
CompanyID: "abc-123",
|
||||
Privilege: PrivilegeAdmin,
|
||||
})
|
||||
assert.NoError(t, err)
|
||||
|
||||
@@ -40,9 +44,10 @@ func TestPrivilegeHandler_Process_PrivilegeRemoved(t *testing.T) {
|
||||
assert.Equal(t, []string{"abc-123"}, companies)
|
||||
|
||||
err = handler.Process(&PrivilegeRemoved{
|
||||
Email: "jim@example.org",
|
||||
CompanyID: "abc-123",
|
||||
Privilege: PrivilegeAdmin,
|
||||
SequenceNo: 2,
|
||||
Email: "jim@example.org",
|
||||
CompanyID: "abc-123",
|
||||
Privilege: PrivilegeAdmin,
|
||||
})
|
||||
assert.NoError(t, err)
|
||||
|
||||
@@ -57,14 +62,16 @@ func TestPrivilegeHandler_Process_UserAdded_And_UserRemoved(t *testing.T) {
|
||||
handler := New(WithBaseURL("base"))
|
||||
|
||||
err := handler.Process(&UserAdded{
|
||||
Email: "jim@example.org",
|
||||
CompanyID: "abc-123",
|
||||
SequenceNo: 1,
|
||||
Email: "jim@example.org",
|
||||
CompanyID: "abc-123",
|
||||
})
|
||||
assert.NoError(t, err)
|
||||
|
||||
err = handler.Process(&UserAdded{
|
||||
Email: "jim@example.org",
|
||||
CompanyID: "abc-456",
|
||||
SequenceNo: 2,
|
||||
Email: "jim@example.org",
|
||||
CompanyID: "abc-456",
|
||||
})
|
||||
assert.NoError(t, err)
|
||||
|
||||
@@ -75,14 +82,16 @@ func TestPrivilegeHandler_Process_UserAdded_And_UserRemoved(t *testing.T) {
|
||||
assert.Equal(t, []string{"abc-123", "abc-456"}, companies)
|
||||
|
||||
err = handler.Process(&UserRemoved{
|
||||
Email: "jim@example.org",
|
||||
CompanyID: "abc-123",
|
||||
SequenceNo: 3,
|
||||
Email: "jim@example.org",
|
||||
CompanyID: "abc-123",
|
||||
})
|
||||
assert.NoError(t, err)
|
||||
|
||||
err = handler.Process(&UserRemoved{
|
||||
Email: "jim@example.org",
|
||||
CompanyID: "abc-456",
|
||||
SequenceNo: 4,
|
||||
Email: "jim@example.org",
|
||||
CompanyID: "abc-456",
|
||||
})
|
||||
assert.NoError(t, err)
|
||||
|
||||
@@ -96,8 +105,8 @@ func TestPrivilegeHandler_Process_UserAdded_Keeps_Existing_Privileges(t *testing
|
||||
// Each routing key has its own transient queue, so Privilege.Added can be
|
||||
// processed before the User.Added published before it. The privilege must
|
||||
// survive either order: nothing re-reads it until the next Fetch at start.
|
||||
userAdded := &UserAdded{Email: "jim@example.org", CompanyID: "abc-123"}
|
||||
privilegeAdded := &PrivilegeAdded{Email: "jim@example.org", CompanyID: "abc-123", Privilege: PrivilegeAdmin}
|
||||
userAdded := &UserAdded{SequenceNo: 1, Email: "jim@example.org", CompanyID: "abc-123"}
|
||||
privilegeAdded := &PrivilegeAdded{SequenceNo: 2, Email: "jim@example.org", CompanyID: "abc-123", Privilege: PrivilegeAdmin}
|
||||
for name, order := range map[string][]any{
|
||||
"user added first": {userAdded, privilegeAdded},
|
||||
"privilege added first": {privilegeAdded, userAdded},
|
||||
@@ -118,14 +127,15 @@ func TestPrivilegeHandler_Process_UserAdded_Keeps_Existing_Privileges(t *testing
|
||||
func TestPrivilegeHandler_Process_UserAdded_After_UserRemoved_Starts_Empty(t *testing.T) {
|
||||
handler := New(WithBaseURL("base"))
|
||||
|
||||
assert.NoError(t, handler.Process(&UserAdded{Email: "jim@example.org", CompanyID: "abc-123"}))
|
||||
assert.NoError(t, handler.Process(&UserAdded{SequenceNo: 1, Email: "jim@example.org", CompanyID: "abc-123"}))
|
||||
assert.NoError(t, handler.Process(&PrivilegeAdded{
|
||||
Email: "jim@example.org",
|
||||
CompanyID: "abc-123",
|
||||
Privilege: PrivilegeAdmin,
|
||||
SequenceNo: 2,
|
||||
Email: "jim@example.org",
|
||||
CompanyID: "abc-123",
|
||||
Privilege: PrivilegeAdmin,
|
||||
}))
|
||||
assert.NoError(t, handler.Process(&UserRemoved{Email: "jim@example.org", CompanyID: "abc-123"}))
|
||||
assert.NoError(t, handler.Process(&UserAdded{Email: "jim@example.org", CompanyID: "abc-123"}))
|
||||
assert.NoError(t, handler.Process(&UserRemoved{SequenceNo: 3, Email: "jim@example.org", CompanyID: "abc-123"}))
|
||||
assert.NoError(t, handler.Process(&UserAdded{SequenceNo: 4, Email: "jim@example.org", CompanyID: "abc-123"}))
|
||||
|
||||
// Membership is back, without the privileges the removal took away.
|
||||
assert.Equal(t, []string{"abc-123"}, handler.CompaniesByUser("jim@example.org", func(CompanyPrivileges) bool {
|
||||
@@ -150,8 +160,9 @@ func TestPrivilegeHandler_GetCompanies_No_Companies_Found(t *testing.T) {
|
||||
handler := New(WithBaseURL("base"))
|
||||
|
||||
err := handler.Process(&UserAdded{
|
||||
Email: "jim@example.org",
|
||||
CompanyID: "abc-123",
|
||||
SequenceNo: 1,
|
||||
Email: "jim@example.org",
|
||||
CompanyID: "abc-123",
|
||||
})
|
||||
assert.NoError(t, err)
|
||||
|
||||
@@ -168,8 +179,9 @@ func TestPrivilegeHandler_GetCompanies_No_Companies_Found(t *testing.T) {
|
||||
assert.Equal(t, []string{"abc-123"}, companies)
|
||||
|
||||
err = handler.Process(&UserRemoved{
|
||||
Email: "jim@example.org",
|
||||
CompanyID: "abc-123",
|
||||
SequenceNo: 2,
|
||||
Email: "jim@example.org",
|
||||
CompanyID: "abc-123",
|
||||
})
|
||||
assert.NoError(t, err)
|
||||
|
||||
@@ -183,9 +195,10 @@ func TestPrivilegeHandler_GetCompanies_Company_With_Company_Access_Found(t *test
|
||||
handler := New(WithBaseURL("base"))
|
||||
|
||||
err := handler.Process(&PrivilegeAdded{
|
||||
Email: "jim@example.org",
|
||||
CompanyID: "abc-123",
|
||||
Privilege: PrivilegeCompany,
|
||||
SequenceNo: 1,
|
||||
Email: "jim@example.org",
|
||||
CompanyID: "abc-123",
|
||||
Privilege: PrivilegeCompany,
|
||||
})
|
||||
assert.NoError(t, err)
|
||||
|
||||
@@ -200,9 +213,10 @@ func TestPrivilegeHandler_GetCompanies_Company_With_Admin_Access_Found(t *testin
|
||||
handler := New(WithBaseURL("base"))
|
||||
|
||||
err := handler.Process(&PrivilegeAdded{
|
||||
Email: "jim@example.org",
|
||||
CompanyID: "abc-123",
|
||||
Privilege: PrivilegeConsumer,
|
||||
SequenceNo: 1,
|
||||
Email: "jim@example.org",
|
||||
CompanyID: "abc-123",
|
||||
Privilege: PrivilegeConsumer,
|
||||
})
|
||||
assert.NoError(t, err)
|
||||
|
||||
@@ -227,9 +241,10 @@ func TestPrivilegeHandler_IsAllowed_Return_True_If_Privilege_Exists(t *testing.T
|
||||
handler := New(WithBaseURL("base"))
|
||||
|
||||
_ = handler.Process(&PrivilegeAdded{
|
||||
Email: "jim@example.org",
|
||||
CompanyID: "abc-123",
|
||||
Privilege: PrivilegeTime,
|
||||
SequenceNo: 1,
|
||||
Email: "jim@example.org",
|
||||
CompanyID: "abc-123",
|
||||
Privilege: PrivilegeTime,
|
||||
})
|
||||
|
||||
result := handler.IsAllowed("jim@example.org", "abc-123", func(privileges CompanyPrivileges) bool {
|
||||
@@ -239,9 +254,10 @@ func TestPrivilegeHandler_IsAllowed_Return_True_If_Privilege_Exists(t *testing.T
|
||||
assert.True(t, result)
|
||||
|
||||
_ = handler.Process(&PrivilegeAdded{
|
||||
Email: "jim@example.org",
|
||||
CompanyID: "abc-123",
|
||||
Privilege: PrivilegeInvoicing,
|
||||
SequenceNo: 2,
|
||||
Email: "jim@example.org",
|
||||
CompanyID: "abc-123",
|
||||
Privilege: PrivilegeInvoicing,
|
||||
})
|
||||
|
||||
result = handler.IsAllowed("jim@example.org", "abc-123", func(privileges CompanyPrivileges) bool {
|
||||
@@ -251,9 +267,10 @@ func TestPrivilegeHandler_IsAllowed_Return_True_If_Privilege_Exists(t *testing.T
|
||||
assert.True(t, result)
|
||||
|
||||
_ = handler.Process(&PrivilegeAdded{
|
||||
Email: "jim@example.org",
|
||||
CompanyID: "abc-123",
|
||||
Privilege: PrivilegeAccounting,
|
||||
SequenceNo: 3,
|
||||
Email: "jim@example.org",
|
||||
CompanyID: "abc-123",
|
||||
Privilege: PrivilegeAccounting,
|
||||
})
|
||||
|
||||
result = handler.IsAllowed("jim@example.org", "abc-123", func(privileges CompanyPrivileges) bool {
|
||||
@@ -263,9 +280,10 @@ func TestPrivilegeHandler_IsAllowed_Return_True_If_Privilege_Exists(t *testing.T
|
||||
assert.True(t, result)
|
||||
|
||||
_ = handler.Process(&PrivilegeAdded{
|
||||
Email: "jim@example.org",
|
||||
CompanyID: "abc-123",
|
||||
Privilege: PrivilegeSupplier,
|
||||
SequenceNo: 4,
|
||||
Email: "jim@example.org",
|
||||
CompanyID: "abc-123",
|
||||
Privilege: PrivilegeSupplier,
|
||||
})
|
||||
|
||||
result = handler.IsAllowed("jim@example.org", "abc-123", func(privileges CompanyPrivileges) bool {
|
||||
@@ -275,9 +293,10 @@ func TestPrivilegeHandler_IsAllowed_Return_True_If_Privilege_Exists(t *testing.T
|
||||
assert.True(t, result)
|
||||
|
||||
_ = handler.Process(&PrivilegeAdded{
|
||||
Email: "jim@example.org",
|
||||
CompanyID: "abc-123",
|
||||
Privilege: PrivilegeSalary,
|
||||
SequenceNo: 5,
|
||||
Email: "jim@example.org",
|
||||
CompanyID: "abc-123",
|
||||
Privilege: PrivilegeSalary,
|
||||
})
|
||||
|
||||
result = handler.IsAllowed("jim@example.org", "abc-123", func(privileges CompanyPrivileges) bool {
|
||||
@@ -386,7 +405,7 @@ func TestPrivilegeHandler_Fetch_Valid(t *testing.T) {
|
||||
|
||||
err := handler.Fetch()
|
||||
assert.NoError(t, err)
|
||||
expectedPrivileges := map[string]map[string]*CompanyPrivileges{
|
||||
expectedPrivileges := map[string]map[string]CompanyPrivileges{
|
||||
"jim@example.org": {
|
||||
"00010203-0405-4607-8809-0a0b0c0d0e0f": {
|
||||
Admin: false,
|
||||
@@ -400,7 +419,7 @@ func TestPrivilegeHandler_Fetch_Valid(t *testing.T) {
|
||||
},
|
||||
},
|
||||
}
|
||||
assert.Equal(t, expectedPrivileges, handler.privileges)
|
||||
assert.Equal(t, expectedPrivileges, privilegesOf(handler))
|
||||
}
|
||||
|
||||
func TestPrivilegeHandler_Fetch_Concurrent_Fetches(t *testing.T) {
|
||||
@@ -450,7 +469,7 @@ func TestPrivilegeHandler_Fetch_Concurrent_Fetches(t *testing.T) {
|
||||
}
|
||||
|
||||
// Verify privileges were set correctly
|
||||
expectedPrivileges := map[string]map[string]*CompanyPrivileges{
|
||||
expectedPrivileges := map[string]map[string]CompanyPrivileges{
|
||||
"jim@example.org": {
|
||||
"00010203-0405-4607-8809-0a0b0c0d0e0f": {
|
||||
Admin: false,
|
||||
@@ -464,7 +483,7 @@ func TestPrivilegeHandler_Fetch_Concurrent_Fetches(t *testing.T) {
|
||||
},
|
||||
},
|
||||
}
|
||||
assert.Equal(t, expectedPrivileges, handler.privileges)
|
||||
assert.Equal(t, expectedPrivileges, privilegesOf(handler))
|
||||
}
|
||||
|
||||
func TestPrivilegeHandler_Concurrent_Fetch_And_Read(t *testing.T) {
|
||||
@@ -559,9 +578,10 @@ func TestPrivilegeHandler_Concurrent_Process_And_Read(t *testing.T) {
|
||||
go func(id string) {
|
||||
defer wg.Done()
|
||||
_ = handler.Process(&PrivilegeAdded{
|
||||
Email: "jim@example.org",
|
||||
CompanyID: id,
|
||||
Privilege: PrivilegeAdmin,
|
||||
SequenceNo: 1,
|
||||
Email: "jim@example.org",
|
||||
CompanyID: id,
|
||||
Privilege: PrivilegeAdmin,
|
||||
})
|
||||
}(companyID)
|
||||
}
|
||||
@@ -635,9 +655,10 @@ func TestPrivilegeHandler_Concurrent_Multiple_Operations(t *testing.T) {
|
||||
go func(idx int) {
|
||||
defer wg.Done()
|
||||
_ = handler.Process(&PrivilegeAdded{
|
||||
Email: "jane@example.org",
|
||||
CompanyID: fmt.Sprintf("company-%d", idx%5),
|
||||
Privilege: PrivilegeCompany,
|
||||
SequenceNo: 1,
|
||||
Email: "jane@example.org",
|
||||
CompanyID: fmt.Sprintf("company-%d", idx%5),
|
||||
Privilege: PrivilegeCompany,
|
||||
})
|
||||
}(i)
|
||||
}
|
||||
@@ -711,22 +732,333 @@ func TestPrivilegeHandler_process(t *testing.T) {
|
||||
admin := func(p CompanyPrivileges) bool { return p.Admin }
|
||||
|
||||
assert.NoError(t, process[UserAdded](handler)(ctx, spec.ConsumableEvent[UserAdded]{
|
||||
Payload: UserAdded{Email: "jim@example.org", CompanyID: "abc-123"},
|
||||
Payload: UserAdded{SequenceNo: 1, Email: "jim@example.org", CompanyID: "abc-123"},
|
||||
}))
|
||||
assert.False(t, handler.IsAllowed("jim@example.org", "abc-123", admin))
|
||||
|
||||
assert.NoError(t, process[PrivilegeAdded](handler)(ctx, spec.ConsumableEvent[PrivilegeAdded]{
|
||||
Payload: PrivilegeAdded{Email: "jim@example.org", CompanyID: "abc-123", Privilege: PrivilegeAdmin},
|
||||
Payload: PrivilegeAdded{SequenceNo: 2, Email: "jim@example.org", CompanyID: "abc-123", Privilege: PrivilegeAdmin},
|
||||
}))
|
||||
assert.True(t, handler.IsAllowed("jim@example.org", "abc-123", admin))
|
||||
|
||||
assert.NoError(t, process[PrivilegeRemoved](handler)(ctx, spec.ConsumableEvent[PrivilegeRemoved]{
|
||||
Payload: PrivilegeRemoved{Email: "jim@example.org", CompanyID: "abc-123", Privilege: PrivilegeAdmin},
|
||||
Payload: PrivilegeRemoved{SequenceNo: 3, Email: "jim@example.org", CompanyID: "abc-123", Privilege: PrivilegeAdmin},
|
||||
}))
|
||||
assert.False(t, handler.IsAllowed("jim@example.org", "abc-123", admin))
|
||||
|
||||
assert.NoError(t, process[UserRemoved](handler)(ctx, spec.ConsumableEvent[UserRemoved]{
|
||||
Payload: UserRemoved{Email: "jim@example.org", CompanyID: "abc-123"},
|
||||
Payload: UserRemoved{SequenceNo: 4, Email: "jim@example.org", CompanyID: "abc-123"},
|
||||
}))
|
||||
assert.Empty(t, handler.CompaniesByUser("jim@example.org", func(CompanyPrivileges) bool { return true }))
|
||||
}
|
||||
|
||||
// privilegesOf returns the privileges of every membership in the handler.
|
||||
func privilegesOf(h *PrivilegeHandler) map[string]map[string]CompanyPrivileges {
|
||||
h.RLock()
|
||||
defer h.RUnlock()
|
||||
result := map[string]map[string]CompanyPrivileges{}
|
||||
for email, companies := range h.state {
|
||||
for companyID, s := range companies {
|
||||
if !s.member {
|
||||
continue
|
||||
}
|
||||
if result[email] == nil {
|
||||
result[email] = map[string]CompanyPrivileges{}
|
||||
}
|
||||
result[email][companyID] = s.privileges
|
||||
}
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
func TestPrivilegeHandler_Process_Orders_By_SequenceNo(t *testing.T) {
|
||||
const email, company = "jim@example.org", "abc-123"
|
||||
userAdded := func(seq int) any { return &UserAdded{Email: email, CompanyID: company, SequenceNo: seq} }
|
||||
userRemoved := func(seq int) any { return &UserRemoved{Email: email, CompanyID: company, SequenceNo: seq} }
|
||||
added := func(seq int) any {
|
||||
return &PrivilegeAdded{Email: email, CompanyID: company, Privilege: PrivilegeAdmin, SequenceNo: seq}
|
||||
}
|
||||
removed := func(seq int) any {
|
||||
return &PrivilegeRemoved{Email: email, CompanyID: company, Privilege: PrivilegeAdmin, SequenceNo: seq}
|
||||
}
|
||||
tests := []struct {
|
||||
name string
|
||||
delivered []any
|
||||
member bool
|
||||
admin bool
|
||||
}{
|
||||
{name: "in order", delivered: []any{userAdded(1), added(2)}, member: true, admin: true},
|
||||
{name: "privilege before older user added keeps the grant", delivered: []any{added(2), userAdded(1)}, member: true, admin: true},
|
||||
{name: "late privilege added after user removed", delivered: []any{userAdded(1), userRemoved(3), added(2)}, member: false, admin: false},
|
||||
{name: "late privilege added after privilege removed", delivered: []any{userAdded(1), removed(3), added(2)}, member: true, admin: false},
|
||||
{name: "late user added does not revive privileges removed with the user", delivered: []any{userRemoved(3), added(2), userAdded(1)}, member: false, admin: false},
|
||||
{name: "re-added user starts without the removed privileges", delivered: []any{added(2), userRemoved(3), userAdded(4)}, member: true, admin: false},
|
||||
{name: "privilege granted after re-add", delivered: []any{userAdded(4), userRemoved(3), added(5), added(2)}, member: true, admin: true},
|
||||
{name: "late user removed does not remove a newer grant", delivered: []any{added(4), userRemoved(3)}, member: true, admin: true},
|
||||
{name: "late user removed does not remove a newer membership", delivered: []any{userAdded(4), userRemoved(3)}, member: true, admin: false},
|
||||
{name: "late privilege added before re-add stays revoked", delivered: []any{userAdded(1), userRemoved(3), added(2), userAdded(4)}, member: true, admin: false},
|
||||
{name: "privilege removed after user removed keeps the user out", delivered: []any{userRemoved(2), removed(3)}, member: false, admin: false},
|
||||
{name: "duplicate delivery is idempotent", delivered: []any{added(2), removed(3), added(2)}, member: true, admin: false},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
handler := New()
|
||||
for _, event := range tt.delivered {
|
||||
require.NoError(t, handler.Process(event))
|
||||
}
|
||||
all := func(CompanyPrivileges) bool { return true }
|
||||
assert.Equal(t, tt.member, handler.IsAllowed(email, company, all))
|
||||
assert.Equal(t, tt.admin, handler.IsAllowed(email, company, func(p CompanyPrivileges) bool { return p.Admin }))
|
||||
if tt.member {
|
||||
assert.Equal(t, []string{company}, handler.CompaniesByUser(email, all))
|
||||
} else {
|
||||
assert.Equal(t, []string{}, handler.CompaniesByUser(email, all))
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestPrivilegeHandler_Process_Without_SequenceNo_Fails_Closed(t *testing.T) {
|
||||
const email, company = "jim@example.org", "abc-123"
|
||||
admin := func(p CompanyPrivileges) bool { return p.Admin }
|
||||
handler := New()
|
||||
|
||||
require.NoError(t, handler.Process(&PrivilegeAdded{Email: email, CompanyID: company, Privilege: PrivilegeAdmin}))
|
||||
require.NoError(t, handler.Process(&UserAdded{Email: email, CompanyID: company}))
|
||||
assert.False(t, handler.IsAllowed(email, company, func(CompanyPrivileges) bool { return true }), "an unordered addition is dropped")
|
||||
require.NoError(t, handler.Process(&UserAdded{Email: email, CompanyID: company, SequenceNo: 1}))
|
||||
assert.False(t, handler.IsAllowed(email, company, admin), "the dropped grant doesn't appear with a later membership")
|
||||
|
||||
require.NoError(t, handler.Process(&PrivilegeAdded{Email: email, CompanyID: company, Privilege: PrivilegeAdmin, SequenceNo: 5}))
|
||||
require.NoError(t, handler.Process(&PrivilegeRemoved{Email: email, CompanyID: company, Privilege: PrivilegeAdmin}))
|
||||
assert.False(t, handler.IsAllowed(email, company, admin), "an unordered privilege removal applies")
|
||||
|
||||
require.NoError(t, handler.Process(&PrivilegeAdded{Email: email, CompanyID: company, Privilege: PrivilegeTime, SequenceNo: 6}))
|
||||
require.NoError(t, handler.Process(&UserRemoved{Email: email, CompanyID: company}))
|
||||
assert.False(t, handler.IsAllowed(email, company, func(CompanyPrivileges) bool { return true }), "an unordered user removal applies")
|
||||
}
|
||||
|
||||
// snapshotServer serves body as the /authz snapshot at seq; a negative seq sends no header.
|
||||
func snapshotServer(t *testing.T, seq int, body string) *httptest.Server {
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if seq >= 0 {
|
||||
w.Header().Set(SequenceHeader, fmt.Sprint(seq))
|
||||
}
|
||||
_, _ = w.Write([]byte(body))
|
||||
}))
|
||||
t.Cleanup(server.Close)
|
||||
return server
|
||||
}
|
||||
|
||||
func TestPrivilegeHandler_Fetch_Merges_Snapshot_By_SequenceNo(t *testing.T) {
|
||||
const email = "jim@example.org"
|
||||
admin := func(p CompanyPrivileges) bool { return p.Admin }
|
||||
all := func(CompanyPrivileges) bool { return true }
|
||||
snapshot := `{"jim@example.org": {"in-snapshot": {"admin": true}, "stale-in-snapshot": {"admin": true}}}`
|
||||
|
||||
handler := New(WithBaseURL(snapshotServer(t, 10, snapshot).URL))
|
||||
// Delivered while the snapshot was read.
|
||||
require.NoError(t, handler.Process(&PrivilegeRemoved{Email: email, CompanyID: "in-snapshot", Privilege: PrivilegeAdmin, SequenceNo: 11}))
|
||||
require.NoError(t, handler.Process(&PrivilegeAdded{Email: email, CompanyID: "stale-in-snapshot", Privilege: PrivilegeTime, SequenceNo: 9}))
|
||||
require.NoError(t, handler.Process(&PrivilegeAdded{Email: email, CompanyID: "new", Privilege: PrivilegeAdmin, SequenceNo: 12}))
|
||||
require.NoError(t, handler.Process(&PrivilegeAdded{Email: email, CompanyID: "revoked", Privilege: PrivilegeAdmin, SequenceNo: 8}))
|
||||
require.NoError(t, handler.Process(&UserAdded{Email: email, CompanyID: "revoked-later", SequenceNo: 7}))
|
||||
require.NoError(t, handler.Process(&PrivilegeAdded{Email: email, CompanyID: "revoked-later", Privilege: PrivilegeAdmin, SequenceNo: 13}))
|
||||
|
||||
require.NoError(t, handler.Fetch())
|
||||
|
||||
assert.True(t, handler.IsAllowed(email, "in-snapshot", all))
|
||||
assert.False(t, handler.IsAllowed(email, "in-snapshot", admin), "a removal newer than the snapshot wins")
|
||||
assert.False(t, handler.IsAllowed(email, "stale-in-snapshot", func(p CompanyPrivileges) bool { return p.Time }), "the snapshot replaces an older fact")
|
||||
assert.True(t, handler.IsAllowed(email, "new", admin), "a grant newer than the snapshot is kept")
|
||||
assert.False(t, handler.IsAllowed(email, "revoked", all), "a pair missing from the snapshot is removed")
|
||||
assert.True(t, handler.IsAllowed(email, "revoked-later", admin), "a grant newer than the snapshot survives the removal")
|
||||
|
||||
// Events the snapshot already covers are stale.
|
||||
require.NoError(t, handler.Process(&PrivilegeAdded{Email: email, CompanyID: "revoked", Privilege: PrivilegeAdmin, SequenceNo: 10}))
|
||||
assert.False(t, handler.IsAllowed(email, "revoked", all), "an event at the snapshot sequence is stale")
|
||||
require.NoError(t, handler.Process(&PrivilegeAdded{Email: email, CompanyID: "revoked", Privilege: PrivilegeAdmin, SequenceNo: 14}))
|
||||
assert.True(t, handler.IsAllowed(email, "revoked", admin))
|
||||
}
|
||||
|
||||
func TestPrivilegeHandler_Fetch_Replaces_Pairs_Missing_From_Snapshot(t *testing.T) {
|
||||
handler := New(WithBaseURL(snapshotServer(t, 10, `{"jim@example.org": {"kept": {"company": true}}}`).URL))
|
||||
require.NoError(t, handler.Process(&PrivilegeAdded{Email: "jim@example.org", CompanyID: "gone", Privilege: PrivilegeAdmin, SequenceNo: 3}))
|
||||
require.NoError(t, handler.Process(&PrivilegeAdded{Email: "jane@example.org", CompanyID: "gone", Privilege: PrivilegeAdmin, SequenceNo: 4}))
|
||||
|
||||
require.NoError(t, handler.Fetch())
|
||||
|
||||
assert.Equal(t, map[string]map[string]CompanyPrivileges{"jim@example.org": {"kept": {Company: true}}}, privilegesOf(handler))
|
||||
handler.RLock()
|
||||
defer handler.RUnlock()
|
||||
assert.NotContains(t, handler.state, "jane@example.org", "an email without companies is dropped")
|
||||
}
|
||||
|
||||
func TestPrivilegeHandler_Fetch_Without_SequenceHeader_Keeps_Sequenced_Facts(t *testing.T) {
|
||||
handler := New(WithBaseURL(snapshotServer(t, -1, `{"jim@example.org": {"abc-123": {"admin": true}}}`).URL))
|
||||
require.NoError(t, handler.Process(&PrivilegeRemoved{Email: "jim@example.org", CompanyID: "abc-123", Privilege: PrivilegeAdmin, SequenceNo: 3}))
|
||||
require.NoError(t, handler.Process(&PrivilegeAdded{Email: "jim@example.org", CompanyID: "other", Privilege: PrivilegeAdmin, SequenceNo: 4}))
|
||||
|
||||
require.NoError(t, handler.Fetch())
|
||||
|
||||
assert.Equal(t, map[string]map[string]CompanyPrivileges{
|
||||
"jim@example.org": {"abc-123": {}, "other": {Admin: true}},
|
||||
}, privilegesOf(handler))
|
||||
}
|
||||
|
||||
func TestPrivilegeHandler_Fetch_Rejects_Bad_Responses(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
status int
|
||||
header string
|
||||
err string
|
||||
}{
|
||||
{name: "status", status: http.StatusUnauthorized, err: "fetch privileges: unexpected status 401 Unauthorized"},
|
||||
{name: "invalid header", status: http.StatusOK, header: "abc", err: `fetch privileges: invalid X-Authz-Sequence header "abc"`},
|
||||
{name: "negative header", status: http.StatusOK, header: "-1", err: `fetch privileges: invalid X-Authz-Sequence header "-1"`},
|
||||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if tt.header != "" {
|
||||
w.Header().Set(SequenceHeader, tt.header)
|
||||
}
|
||||
w.WriteHeader(tt.status)
|
||||
_, _ = w.Write([]byte(`{"jim@example.org": {"abc-123": {"admin": true}}}`))
|
||||
}))
|
||||
defer server.Close()
|
||||
handler := New(WithBaseURL(server.URL))
|
||||
require.NoError(t, handler.Process(&PrivilegeAdded{Email: "jane@example.org", CompanyID: "abc-123", Privilege: PrivilegeTime, SequenceNo: 1}))
|
||||
|
||||
assert.EqualError(t, handler.Fetch(), tt.err)
|
||||
assert.Equal(t, map[string]map[string]CompanyPrivileges{"jane@example.org": {"abc-123": {Time: true}}}, privilegesOf(handler), "state is untouched")
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestPrivilegeHandler_Revocation_During_Fetch_Wins(t *testing.T) {
|
||||
// The consumers are bound before Fetch, so a revocation published after authz-service
|
||||
// read the snapshot is processed while the response is still on its way.
|
||||
handler := New()
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
// assert, not require: this runs on the server's goroutine.
|
||||
assert.NoError(t, handler.Process(&UserRemoved{Email: "jim@example.org", CompanyID: "abc-123", SequenceNo: 21}))
|
||||
w.Header().Set(SequenceHeader, "20")
|
||||
_, _ = w.Write([]byte(`{"jim@example.org": {"abc-123": {"admin": true}}}`))
|
||||
}))
|
||||
defer server.Close()
|
||||
handler.baseURL = server.URL
|
||||
|
||||
require.NoError(t, handler.Fetch())
|
||||
|
||||
assert.False(t, handler.IsAllowed("jim@example.org", "abc-123", func(CompanyPrivileges) bool { return true }))
|
||||
}
|
||||
|
||||
func TestPrivilegeHandler_Fetch_Ignores_Snapshot_Older_Than_Merged_One(t *testing.T) {
|
||||
const email = "jim@example.org"
|
||||
all := func(CompanyPrivileges) bool { return true }
|
||||
handler := New(WithBaseURL(snapshotServer(t, 12, `{}`).URL))
|
||||
require.NoError(t, handler.Process(&PrivilegeAdded{Email: email, CompanyID: "abc-123", Privilege: PrivilegeAdmin, SequenceNo: 5}))
|
||||
require.NoError(t, handler.Process(&UserRemoved{Email: email, CompanyID: "abc-123", SequenceNo: 11}))
|
||||
require.NoError(t, handler.Fetch())
|
||||
|
||||
// A concurrent Fetch read at seq 10, before the removal, merges last.
|
||||
handler.baseURL = snapshotServer(t, 10, `{"jim@example.org": {"abc-123": {"admin": true}}}`).URL
|
||||
require.NoError(t, handler.Fetch())
|
||||
assert.False(t, handler.IsAllowed(email, "abc-123", all))
|
||||
|
||||
// So does a snapshot without a sequence number.
|
||||
handler.baseURL = snapshotServer(t, -1, `{"jim@example.org": {"abc-123": {"admin": true}}}`).URL
|
||||
require.NoError(t, handler.Fetch())
|
||||
assert.False(t, handler.IsAllowed(email, "abc-123", all))
|
||||
}
|
||||
|
||||
func TestPrivilegeHandler_Unordered_Removal_Holds_Until_Next_Snapshot(t *testing.T) {
|
||||
const email, company = "jim@example.org", "abc-123"
|
||||
all := func(CompanyPrivileges) bool { return true }
|
||||
admin := func(p CompanyPrivileges) bool { return p.Admin }
|
||||
handler := New(WithBaseURL(snapshotServer(t, 30, `{"jim@example.org": {"abc-123": {"admin": true}, "def-456": {"time": true}}}`).URL))
|
||||
require.NoError(t, handler.Process(&UserAdded{Email: email, CompanyID: company, SequenceNo: 10}))
|
||||
require.NoError(t, handler.Process(&PrivilegeAdded{Email: email, CompanyID: company, Privilege: PrivilegeAdmin, SequenceNo: 11}))
|
||||
require.NoError(t, handler.Process(&PrivilegeAdded{Email: email, CompanyID: "def-456", Privilege: PrivilegeTime, SequenceNo: 12}))
|
||||
|
||||
require.NoError(t, handler.Process(&UserRemoved{Email: email, CompanyID: company}))
|
||||
require.NoError(t, handler.Process(&PrivilegeRemoved{Email: email, CompanyID: "def-456", Privilege: PrivilegeTime}))
|
||||
require.NoError(t, handler.Process(&PrivilegeAdded{Email: email, CompanyID: company, Privilege: PrivilegeSalary, SequenceNo: 15}))
|
||||
require.NoError(t, handler.Process(&PrivilegeAdded{Email: email, CompanyID: company, Privilege: PrivilegeAdmin, SequenceNo: 16}))
|
||||
require.NoError(t, handler.Process(&UserAdded{Email: email, CompanyID: company, SequenceNo: 17}))
|
||||
require.NoError(t, handler.Process(&PrivilegeAdded{Email: email, CompanyID: "def-456", Privilege: PrivilegeTime, SequenceNo: 18}))
|
||||
assert.False(t, handler.IsAllowed(email, company, all), "no event overrides an unordered user removal")
|
||||
assert.False(t, handler.IsAllowed(email, "def-456", func(p CompanyPrivileges) bool { return p.Time }), "no event overrides an unordered privilege removal")
|
||||
|
||||
require.NoError(t, handler.Fetch())
|
||||
assert.True(t, handler.IsAllowed(email, company, admin), "the next snapshot replaces the unordered removal")
|
||||
assert.True(t, handler.IsAllowed(email, "def-456", func(p CompanyPrivileges) bool { return p.Time }))
|
||||
}
|
||||
|
||||
func TestPrivilegeHandler_Fetch_Deletes_Unordered_Removal_Missing_From_Snapshot(t *testing.T) {
|
||||
handler := New(WithBaseURL(snapshotServer(t, 30, `{}`).URL))
|
||||
require.NoError(t, handler.Process(&UserRemoved{Email: "jim@example.org", CompanyID: "abc-123"}))
|
||||
|
||||
require.NoError(t, handler.Fetch())
|
||||
|
||||
handler.RLock()
|
||||
defer handler.RUnlock()
|
||||
assert.Empty(t, handler.state)
|
||||
}
|
||||
|
||||
func TestPrivilegeHandler_Process_Unknown_Privilege_Creates_No_State(t *testing.T) {
|
||||
handler := New()
|
||||
require.NoError(t, handler.Process(&PrivilegeRemoved{Email: "jim@example.org", CompanyID: "abc-123", Privilege: "UNKNOWN", SequenceNo: 1}))
|
||||
require.NoError(t, handler.Process(&PrivilegeAdded{Email: "jim@example.org", CompanyID: "abc-123", Privilege: "UNKNOWN", SequenceNo: 2}))
|
||||
|
||||
handler.RLock()
|
||||
defer handler.RUnlock()
|
||||
assert.Empty(t, handler.state)
|
||||
}
|
||||
|
||||
func TestPrivilegeHandler_Fetch_Retries_While_ReadView_Not_Ready(t *testing.T) {
|
||||
var calls atomic.Int32
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if calls.Add(1) < 3 {
|
||||
w.WriteHeader(http.StatusServiceUnavailable)
|
||||
return
|
||||
}
|
||||
w.Header().Set(SequenceHeader, "7")
|
||||
_, _ = w.Write([]byte(`{"jim@example.org": {"abc-123": {"admin": true}}}`))
|
||||
}))
|
||||
defer server.Close()
|
||||
handler := New(WithBaseURL(server.URL))
|
||||
handler.retryDelay = time.Millisecond
|
||||
|
||||
require.NoError(t, handler.Fetch())
|
||||
|
||||
assert.Equal(t, int32(3), calls.Load())
|
||||
assert.True(t, handler.IsAllowed("jim@example.org", "abc-123", func(p CompanyPrivileges) bool { return p.Admin }))
|
||||
}
|
||||
|
||||
func TestPrivilegeHandler_Fetch_Gives_Up_When_ReadView_Stays_Behind(t *testing.T) {
|
||||
var calls atomic.Int32
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
calls.Add(1)
|
||||
w.WriteHeader(http.StatusServiceUnavailable)
|
||||
}))
|
||||
defer server.Close()
|
||||
handler := New(WithBaseURL(server.URL))
|
||||
handler.retryDelay = time.Millisecond
|
||||
handler.retries = 2
|
||||
|
||||
assert.EqualError(t, handler.Fetch(), "fetch privileges: authz-service read view not ready")
|
||||
assert.Equal(t, int32(3), calls.Load())
|
||||
}
|
||||
|
||||
func TestPrivilegeHandler_Process_Drops_Invalid_SequenceNo(t *testing.T) {
|
||||
for _, seq := range []int{-1, math.MaxInt} {
|
||||
handler := New()
|
||||
require.NoError(t, handler.Process(&PrivilegeAdded{Email: "jim@example.org", CompanyID: "abc-123", Privilege: PrivilegeAdmin, SequenceNo: seq}))
|
||||
require.NoError(t, handler.Process(&UserRemoved{Email: "jim@example.org", CompanyID: "abc-123", SequenceNo: seq}))
|
||||
handler.RLock()
|
||||
assert.Empty(t, handler.state, "seq %d", seq)
|
||||
handler.RUnlock()
|
||||
}
|
||||
}
|
||||
@@ -1,15 +1,18 @@
|
||||
package client
|
||||
|
||||
// UserAdded is the event sent when a new user is added to a company
|
||||
// UserAdded is the event sent when a new user is added to a company.
|
||||
// SequenceNo is authz-service's global event sequence number; it orders the events.
|
||||
type UserAdded struct {
|
||||
Email string `json:"email"`
|
||||
CompanyID string `json:"companyId"`
|
||||
Email string `json:"email"`
|
||||
CompanyID string `json:"companyId"`
|
||||
SequenceNo int `json:"sequenceNo"`
|
||||
}
|
||||
|
||||
// UserRemoved is the event sent when a user is removed from a company
|
||||
type UserRemoved struct {
|
||||
Email string `json:"email"`
|
||||
CompanyID string `json:"companyId"`
|
||||
Email string `json:"email"`
|
||||
CompanyID string `json:"companyId"`
|
||||
SequenceNo int `json:"sequenceNo"`
|
||||
}
|
||||
|
||||
// Privilege is an enumeration of all available privileges
|
||||
@@ -51,14 +54,16 @@ func (e Privilege) String() string {
|
||||
|
||||
// PrivilegeAdded is the event sent when a new privilege is added
|
||||
type PrivilegeAdded struct {
|
||||
Email string `json:"email"`
|
||||
CompanyID string `json:"companyId"`
|
||||
Privilege Privilege `json:"privilege"`
|
||||
Email string `json:"email"`
|
||||
CompanyID string `json:"companyId"`
|
||||
Privilege Privilege `json:"privilege"`
|
||||
SequenceNo int `json:"sequenceNo"`
|
||||
}
|
||||
|
||||
// PrivilegeRemoved is the event sent when a privilege is removed
|
||||
type PrivilegeRemoved struct {
|
||||
Email string `json:"email"`
|
||||
CompanyID string `json:"companyId"`
|
||||
Privilege Privilege `json:"privilege"`
|
||||
Email string `json:"email"`
|
||||
CompanyID string `json:"companyId"`
|
||||
Privilege Privilege `json:"privilege"`
|
||||
SequenceNo int `json:"sequenceNo"`
|
||||
}
|
||||
Reference in new issue
Block a user