7 Commits
Author SHA1 Message Date
argoyle 6a708965e4 chore(renovate): group OpenTelemetry, otelsetup and logging updates (#343)
Unbound Release / Check Preconditions (push) Successful in 21s
Unbound Release / Create Tag (push) Skipped
authz_client / test (push) Skipped
authz_client / vulnerabilities (push) Skipped
pre-commit / pre-commit (push) Skipped
Unbound Release / Generate Changelog and Handle PR (push) Successful in 38s
Unbound Release / Create Release (push) Successful in 36s
Release / release (push) Successful in 4m48s
2026-10-05 10:29:00 +00:00
renovate bb450e5d8c chore(deps): update pre-commit hook golangci/golangci-lint to v2.14.0 (#341)
Unbound Release / Check Preconditions (push) Successful in 27s
Unbound Release / Create Tag (push) Skipped
Unbound Release / Create Release (push) Successful in 28s
authz_client / test (push) Skipped
authz_client / vulnerabilities (push) Skipped
pre-commit / pre-commit (push) Skipped
Unbound Release / Generate Changelog and Handle PR (push) Successful in 52s
Release / release (push) Successful in 21m31s
2026-09-27 12:09:00 +00:00
renovate 206d2ab14b chore(deps): update pre-commit hook alessandrojcm/commitlint-pre-commit-hook to v9.27.0 (#339)
Unbound Release / Check Preconditions (push) Successful in 26s
Unbound Release / Create Tag (push) Skipped
Unbound Release / Generate Changelog and Handle PR (push) Successful in 43s
authz_client / test (push) Skipped
authz_client / vulnerabilities (push) Skipped
pre-commit / pre-commit (push) Skipped
Unbound Release / Create Release (push) Successful in 28s
Release / release (push) Successful in 11m17s
2026-09-23 19:19:20 +00:00
renovate 78b4730687 fix(deps): update module codeberg.org/messaging/go-messaging-amqp to v0.0.6 (#337)
Unbound Release / Check Preconditions (push) Successful in 26s
Unbound Release / Create Tag (push) Skipped
Unbound Release / Generate Changelog and Handle PR (push) Successful in 42s
Release / release (push) Successful in 6m44s
authz_client / test (push) Skipped
authz_client / vulnerabilities (push) Skipped
pre-commit / pre-commit (push) Skipped
Unbound Release / Create Release (push) Successful in 28s
2026-09-20 11:08:10 +00:00
argoyle 3941ff4950 docs: deploy authz-service first when the /authz contract changes (#335)
Unbound Release / Check Preconditions (push) Successful in 26s
Unbound Release / Create Tag (push) Skipped
authz_client / test (push) Skipped
authz_client / vulnerabilities (push) Skipped
pre-commit / pre-commit (push) Skipped
Unbound Release / Generate Changelog and Handle PR (push) Successful in 50s
Unbound Release / Create Release (push) Successful in 29s
Release / release (push) Successful in 5m38s
2026-09-17 06:15:33 +00:00
releaser 09766ebe8b chore(release): prepare for v0.7.0 (#334)
Unbound Release / Check Preconditions (push) Successful in 27s
Unbound Release / Create Tag (push) Skipped
Unbound Release / Generate Changelog and Handle PR (push) Successful in 29s
authz_client / test (push) Skipped
authz_client / vulnerabilities (push) Skipped
Unbound Release / Create Release (push) Successful in 30s
Release / release (push) Successful in 1m15s
pre-commit / pre-commit (push) Successful in 3m11s
## [0.7.0] - 2026-09-16

### 🐛 Bug Fixes

- [**breaking**] Order privilege events by sequence number and merge snapshots by position (#333)

<!-- generated by git-cliff -->

---

**Note:** Please use **Squash Merge** when merging this PR.

Reviewed-on: https://gitea.unbound.se/shiny/authz_client/pulls/334
Co-authored-by: Unbound Releaser <releaser@unbound.se>
2026-09-16 18:40:41 +00:00
argoyle e0d1ce3b31 fix!: order privilege events by sequence number and merge snapshots by position (#333)
Unbound Release / Check Preconditions (push) Successful in 27s
Unbound Release / Create Tag (push) Skipped
Unbound Release / Create Release (push) Successful in 27s
authz_client / test (push) Successful in 1m10s
authz_client / vulnerabilities (push) Successful in 53s
Unbound Release / Generate Changelog and Handle PR (push) Successful in 41s
Release / release (push) Successful in 1m10s
pre-commit / pre-commit (push) Successful in 3m12s
## Why
The privilege cache could keep a grant authz-service had revoked:
- **Unordered keys:** each routing key has its own transient queue, so a late `Privilege.Added`/`User.Added` resurrected a revoked grant.
- **Startup gap:** services fetched `/authz` before binding their queues, so revocations published in between were lost until restart.

Design: ADR-0015 (docs PR, Proposed).

## What
- `Process` orders events by authz-service's global `sequenceNo` per (email, company). All four events come from the Company aggregate, so seq order equals commit order. An event only overrides older facts, and `User.Removed` stamps every privilege.
- Events without a sequence number fail closed: additions are dropped, and removals hold until the next snapshot. Negative or huge sequence numbers are dropped.
- `Fetch` checks the status and retries 503 (60×1 s, 30 s HTTP timeout). It reads `X-Authz-Sequence`, merges the snapshot as facts at that position, and raises a floor; snapshots older than the floor are ignored. A missing header merges at 0 with a warning (rollout window only).
- `CompaniesByUser` returns `[]`, and unknown privileges create no state. CLAUDE.md is rewritten.

**BREAKING:** `Process` without `SequenceNo` no longer grants, so service tests must set it. Services must call `Fetch()` after `conn.Start`.

## Verification
- `go test -race`: 98.3% coverage, including table-driven reorderings, snapshot-merge cases and a revocation-during-Fetch race test.
- 26 mutants on the ordering, merge and retry checks: all killed (each compiled and produced `--- FAIL`).
- prek passes.

**Expert review:** two rounds. Round 1: Security, Go Backend, Event Sourcing and Database experts reviewed both diffs. Round 2: Security and Event Sourcing re-reviewed the fixes. A final Event Sourcing review covered the committed-events wrapper. Fixed from the reviews: older snapshots merged after newer ones (floor check); a lagging read view serving snapshots that miss revocations (503 plus catch-up); a reset's TRUNCATE emptying a REPEATABLE READ snapshot (LOCK TABLE privileges, verified on PostgreSQL); seq-0 removals undone by older additions (pending stamp); late-committing events skipped by read view backfills (CommittedEventStore with LOCK TABLE events IN SHARE MODE, verified on PostgreSQL 18); an unbounded lock wait (lock_timeout plus retries); catch-up firing on ordinary lag (5 s stall, 10 s cooldown, 10 min deadline); plus smaller items (unknown privileges, invalid seqs, `require` in a goroutine, wrapped errors, `[]` not nil). **Deliberately deferred (tracked in Ambix):** stored-but-unpublished revocations (user decision: authz-service outbox); the readview library's commit-order gap for other services (upstream); removing the missing-header fallback and alerting on /authz 503s; moving Fetch after conn.Start in the 13 consumers (separate bump PRs).

🤖 Generated with [Claude Code](https://claude.com/claude-code)

https://claude.ai/code/session_01DVGsVQ8AMFR4NZoxyCoEqS
Reviewed-on: https://gitea.unbound.se/shiny/authz_client/pulls/333
2026-09-16 18:36:07 +00:00
10 changed files with 715 additions and 148 deletions

No files matched your search

+2 -2
View File
@@ -11,7 +11,7 @@ repos:
- --allow-multiple-documents - --allow-multiple-documents
- id: check-added-large-files - id: check-added-large-files
- repo: https://github.com/alessandrojcm/commitlint-pre-commit-hook - repo: https://github.com/alessandrojcm/commitlint-pre-commit-hook
rev: v9.26.0 rev: v9.27.0
hooks: hooks:
- id: commitlint - id: commitlint
stages: [ commit-msg ] stages: [ commit-msg ]
@@ -30,7 +30,7 @@ repos:
- id: go-test - id: go-test
- id: gofumpt - id: gofumpt
- repo: https://github.com/golangci/golangci-lint - repo: https://github.com/golangci/golangci-lint
rev: v2.13.2 rev: v2.14.0
hooks: hooks:
- id: golangci-lint-full - id: golangci-lint-full
- repo: https://github.com/gitleaks/gitleaks - repo: https://github.com/gitleaks/gitleaks
+1 -1
View File
@@ -1,3 +1,3 @@
{ {
"version": "v0.6.1" "version": "v0.7.0"
} }
+6
View File
@@ -2,6 +2,12 @@
All notable changes to this project will be documented in this file. All notable changes to this project will be documented in this file.
## [0.7.0] - 2026-09-16
### 🐛 Bug Fixes
- [**breaking**] Order privilege events by sequence number and merge snapshots by position (#333)
## [0.6.1] - 2026-09-16 ## [0.6.1] - 2026-09-16
### 🐛 Bug Fixes ### 🐛 Bug Fixes
+10 -5
View File
@@ -23,10 +23,9 @@ import client "gitea.unbound.se/shiny/authz_client"
handler := client.New(client.WithBaseURL("http://authz-service")) handler := client.New(client.WithBaseURL("http://authz-service"))
// Check user privileges // Check user privileges
privileges := handler.Get(email, companyID) allowed := handler.IsAllowed(email, companyID, func(p client.CompanyPrivileges) bool {
if privileges.Invoicing { return p.Invoicing
// User has invoicing privileges })
}
``` ```
### Privileges ### Privileges
@@ -43,4 +42,10 @@ The `CompanyPrivileges` struct contains permission flags:
### Event Handling ### 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). When the `/authz` contract changes, deploy authz-service before the services that use this library.
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.
+274 -65
View File
@@ -3,16 +3,29 @@ package client
import ( import (
"context" "context"
"encoding/json" "encoding/json"
"errors"
"fmt" "fmt"
"io" "io"
"log/slog"
"math"
"net/http" "net/http"
"reflect" "reflect"
"strconv"
"sync" "sync"
"time"
goamqp "codeberg.org/messaging/go-messaging-amqp" goamqp "codeberg.org/messaging/go-messaging-amqp"
spec "codeberg.org/messaging/messaging" 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 // CompanyPrivileges contains the privileges for a combination of email address and company id
type CompanyPrivileges struct { type CompanyPrivileges struct {
Admin bool `json:"admin"` Admin bool `json:"admin"`
@@ -25,13 +38,84 @@ type CompanyPrivileges struct {
Salary bool `json:"salary"` 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 // PrivilegeHandler processes PrivilegeAdded-events and fetches the initial set of privileges from an authz-service
type PrivilegeHandler struct { type PrivilegeHandler struct {
*sync.RWMutex *sync.RWMutex
client *http.Client client *http.Client
baseURL string baseURL string
apiKey string apiKey string
privileges map[string]map[string]*CompanyPrivileges 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 // OptsFunc is used to configure the PrivilegeHandler
@@ -55,9 +139,11 @@ func WithAPIKey(key string) OptsFunc {
func New(opts ...OptsFunc) *PrivilegeHandler { func New(opts ...OptsFunc) *PrivilegeHandler {
handler := &PrivilegeHandler{ handler := &PrivilegeHandler{
RWMutex: &sync.RWMutex{}, RWMutex: &sync.RWMutex{},
client: &http.Client{}, client: &http.Client{Timeout: 30 * time.Second},
baseURL: "http://authz-service", 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 { for _, opt := range opts {
opt(handler) opt(handler)
@@ -65,8 +151,27 @@ func New(opts ...OptsFunc) *PrivilegeHandler {
return handler 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 { 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) req, err := http.NewRequest(http.MethodGet, fmt.Sprintf("%s/authz", h.baseURL), nil)
if err != nil { if err != nil {
return err return err
@@ -80,21 +185,83 @@ func (h *PrivilegeHandler) Fetch() error {
if err != nil { if err != nil {
return err 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) buff, err := io.ReadAll(resp.Body)
if err != nil { if err != nil {
return err return err
} }
h.Lock() var snapshot map[string]map[string]CompanyPrivileges
defer h.Unlock() if err := json.Unmarshal(buff, &snapshot); err != nil {
err = json.Unmarshal(buff, &h.privileges)
if err != nil {
return err 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 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 { func (h *PrivilegeHandler) Setup() []goamqp.Setup {
return []goamqp.Setup{ return []goamqp.Setup{
goamqp.TransientEventStreamConsumer("User.Added", process[UserAdded](h)), 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 { func (h *PrivilegeHandler) Process(msg any) error {
h.Lock() h.Lock()
defer h.Unlock() defer h.Unlock()
switch ev := msg.(type) { switch ev := msg.(type) {
case *UserAdded: case *UserAdded:
// Keep the privileges already recorded for the company. Each routing key has if h.stale(ev.SequenceNo, true, ev) {
// its own transient queue, so a Privilege.Added published after this event can return nil
// 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 _, exists := priv[ev.CompanyID]; !exists { s := h.company(ev.Email, ev.CompanyID)
priv[ev.CompanyID] = &CompanyPrivileges{} if ev.SequenceNo > s.memberSeq {
s.member = true
s.memberSeq = ev.SequenceNo
} }
return nil return nil
case *UserRemoved: case *UserRemoved:
if priv, exists := h.privileges[ev.Email]; exists { if h.stale(ev.SequenceNo, false, ev) {
delete(priv, ev.CompanyID) return nil
} }
removeUser(h.company(ev.Email, ev.CompanyID), ev.SequenceNo, ev.SequenceNo == 0)
return nil return nil
case *PrivilegeAdded: 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 return nil
case *PrivilegeRemoved: 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 return nil
default: 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)) return fmt.Errorf("unexpected event type: '%s'", reflect.TypeOf(msg))
} }
} }
func (h *PrivilegeHandler) setPrivileges(email, companyId string, privilege Privilege, set bool) { // stale reports whether an event must be skipped: an addition without a sequence
if priv, exists := h.privileges[email]; exists { // number, or any event already covered by the latest snapshot.
if c, exists := priv[companyId]; exists { func (h *PrivilegeHandler) stale(seq int, addition bool, ev any) bool {
switch privilege { if seq < 0 || seq >= pending {
case PrivilegeAdmin: slog.Error("dropping privilege event with invalid sequence number", "type", reflect.TypeOf(ev).String(), "seq", seq)
c.Admin = set return true
case PrivilegeCompany: }
c.Company = set if seq == 0 {
case PrivilegeConsumer: if addition {
c.Consumer = set slog.Warn("dropping privilege addition without sequence number", "type", reflect.TypeOf(ev).String())
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)
} }
} else { return addition
h.privileges[email] = map[string]*CompanyPrivileges{} }
h.setPrivileges(email, companyId, privilege, set) 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 { func (h *PrivilegeHandler) CompaniesByUser(email string, predicate func(privileges CompanyPrivileges) bool) []string {
h.RLock() h.RLock()
defer h.RUnlock() defer h.RUnlock()
var result []string result := []string{}
if p, exists := h.privileges[email]; exists { for k, s := range h.state[email] {
for k, v := range p { if s.member && predicate(s.privileges) {
if predicate(*v) { result = append(result, k)
result = append(result, k)
}
} }
} }
return result 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 { func (h *PrivilegeHandler) IsAllowed(email, companyID string, predicate func(privileges CompanyPrivileges) bool) bool {
h.RLock() h.RLock()
defer h.RUnlock() defer h.RUnlock()
if p, exists := h.privileges[email]; exists { if s, exists := h.state[email][companyID]; exists && s.member {
if v, exists := p[companyID]; exists { return predicate(s.privileges)
return predicate(*v)
}
} }
return false return false
} }
+393 -61
View File
@@ -3,11 +3,14 @@ package client
import ( import (
"context" "context"
"fmt" "fmt"
"math"
"net/http" "net/http"
"net/http/httptest" "net/http/httptest"
"sort" "sort"
"sync" "sync"
"sync/atomic"
"testing" "testing"
"time"
goamqp "codeberg.org/messaging/go-messaging-amqp" goamqp "codeberg.org/messaging/go-messaging-amqp"
spec "codeberg.org/messaging/messaging" spec "codeberg.org/messaging/messaging"
@@ -27,9 +30,10 @@ func TestPrivilegeHandler_Process_PrivilegeRemoved(t *testing.T) {
handler := New(WithBaseURL("base")) handler := New(WithBaseURL("base"))
err := handler.Process(&PrivilegeAdded{ err := handler.Process(&PrivilegeAdded{
Email: "jim@example.org", SequenceNo: 1,
CompanyID: "abc-123", Email: "jim@example.org",
Privilege: PrivilegeAdmin, CompanyID: "abc-123",
Privilege: PrivilegeAdmin,
}) })
assert.NoError(t, err) assert.NoError(t, err)
@@ -40,9 +44,10 @@ func TestPrivilegeHandler_Process_PrivilegeRemoved(t *testing.T) {
assert.Equal(t, []string{"abc-123"}, companies) assert.Equal(t, []string{"abc-123"}, companies)
err = handler.Process(&PrivilegeRemoved{ err = handler.Process(&PrivilegeRemoved{
Email: "jim@example.org", SequenceNo: 2,
CompanyID: "abc-123", Email: "jim@example.org",
Privilege: PrivilegeAdmin, CompanyID: "abc-123",
Privilege: PrivilegeAdmin,
}) })
assert.NoError(t, err) assert.NoError(t, err)
@@ -57,14 +62,16 @@ func TestPrivilegeHandler_Process_UserAdded_And_UserRemoved(t *testing.T) {
handler := New(WithBaseURL("base")) handler := New(WithBaseURL("base"))
err := handler.Process(&UserAdded{ err := handler.Process(&UserAdded{
Email: "jim@example.org", SequenceNo: 1,
CompanyID: "abc-123", Email: "jim@example.org",
CompanyID: "abc-123",
}) })
assert.NoError(t, err) assert.NoError(t, err)
err = handler.Process(&UserAdded{ err = handler.Process(&UserAdded{
Email: "jim@example.org", SequenceNo: 2,
CompanyID: "abc-456", Email: "jim@example.org",
CompanyID: "abc-456",
}) })
assert.NoError(t, err) 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) assert.Equal(t, []string{"abc-123", "abc-456"}, companies)
err = handler.Process(&UserRemoved{ err = handler.Process(&UserRemoved{
Email: "jim@example.org", SequenceNo: 3,
CompanyID: "abc-123", Email: "jim@example.org",
CompanyID: "abc-123",
}) })
assert.NoError(t, err) assert.NoError(t, err)
err = handler.Process(&UserRemoved{ err = handler.Process(&UserRemoved{
Email: "jim@example.org", SequenceNo: 4,
CompanyID: "abc-456", Email: "jim@example.org",
CompanyID: "abc-456",
}) })
assert.NoError(t, err) 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 // Each routing key has its own transient queue, so Privilege.Added can be
// processed before the User.Added published before it. The privilege must // processed before the User.Added published before it. The privilege must
// survive either order: nothing re-reads it until the next Fetch at start. // survive either order: nothing re-reads it until the next Fetch at start.
userAdded := &UserAdded{Email: "jim@example.org", CompanyID: "abc-123"} userAdded := &UserAdded{SequenceNo: 1, Email: "jim@example.org", CompanyID: "abc-123"}
privilegeAdded := &PrivilegeAdded{Email: "jim@example.org", CompanyID: "abc-123", Privilege: PrivilegeAdmin} privilegeAdded := &PrivilegeAdded{SequenceNo: 2, Email: "jim@example.org", CompanyID: "abc-123", Privilege: PrivilegeAdmin}
for name, order := range map[string][]any{ for name, order := range map[string][]any{
"user added first": {userAdded, privilegeAdded}, "user added first": {userAdded, privilegeAdded},
"privilege added first": {privilegeAdded, userAdded}, "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) { func TestPrivilegeHandler_Process_UserAdded_After_UserRemoved_Starts_Empty(t *testing.T) {
handler := New(WithBaseURL("base")) 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{ assert.NoError(t, handler.Process(&PrivilegeAdded{
Email: "jim@example.org", SequenceNo: 2,
CompanyID: "abc-123", Email: "jim@example.org",
Privilege: PrivilegeAdmin, CompanyID: "abc-123",
Privilege: PrivilegeAdmin,
})) }))
assert.NoError(t, handler.Process(&UserRemoved{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{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. // Membership is back, without the privileges the removal took away.
assert.Equal(t, []string{"abc-123"}, handler.CompaniesByUser("jim@example.org", func(CompanyPrivileges) bool { 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")) handler := New(WithBaseURL("base"))
err := handler.Process(&UserAdded{ err := handler.Process(&UserAdded{
Email: "jim@example.org", SequenceNo: 1,
CompanyID: "abc-123", Email: "jim@example.org",
CompanyID: "abc-123",
}) })
assert.NoError(t, err) assert.NoError(t, err)
@@ -168,8 +179,9 @@ func TestPrivilegeHandler_GetCompanies_No_Companies_Found(t *testing.T) {
assert.Equal(t, []string{"abc-123"}, companies) assert.Equal(t, []string{"abc-123"}, companies)
err = handler.Process(&UserRemoved{ err = handler.Process(&UserRemoved{
Email: "jim@example.org", SequenceNo: 2,
CompanyID: "abc-123", Email: "jim@example.org",
CompanyID: "abc-123",
}) })
assert.NoError(t, err) assert.NoError(t, err)
@@ -183,9 +195,10 @@ func TestPrivilegeHandler_GetCompanies_Company_With_Company_Access_Found(t *test
handler := New(WithBaseURL("base")) handler := New(WithBaseURL("base"))
err := handler.Process(&PrivilegeAdded{ err := handler.Process(&PrivilegeAdded{
Email: "jim@example.org", SequenceNo: 1,
CompanyID: "abc-123", Email: "jim@example.org",
Privilege: PrivilegeCompany, CompanyID: "abc-123",
Privilege: PrivilegeCompany,
}) })
assert.NoError(t, err) assert.NoError(t, err)
@@ -200,9 +213,10 @@ func TestPrivilegeHandler_GetCompanies_Company_With_Admin_Access_Found(t *testin
handler := New(WithBaseURL("base")) handler := New(WithBaseURL("base"))
err := handler.Process(&PrivilegeAdded{ err := handler.Process(&PrivilegeAdded{
Email: "jim@example.org", SequenceNo: 1,
CompanyID: "abc-123", Email: "jim@example.org",
Privilege: PrivilegeConsumer, CompanyID: "abc-123",
Privilege: PrivilegeConsumer,
}) })
assert.NoError(t, err) assert.NoError(t, err)
@@ -227,9 +241,10 @@ func TestPrivilegeHandler_IsAllowed_Return_True_If_Privilege_Exists(t *testing.T
handler := New(WithBaseURL("base")) handler := New(WithBaseURL("base"))
_ = handler.Process(&PrivilegeAdded{ _ = handler.Process(&PrivilegeAdded{
Email: "jim@example.org", SequenceNo: 1,
CompanyID: "abc-123", Email: "jim@example.org",
Privilege: PrivilegeTime, CompanyID: "abc-123",
Privilege: PrivilegeTime,
}) })
result := handler.IsAllowed("jim@example.org", "abc-123", func(privileges CompanyPrivileges) bool { 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) assert.True(t, result)
_ = handler.Process(&PrivilegeAdded{ _ = handler.Process(&PrivilegeAdded{
Email: "jim@example.org", SequenceNo: 2,
CompanyID: "abc-123", Email: "jim@example.org",
Privilege: PrivilegeInvoicing, CompanyID: "abc-123",
Privilege: PrivilegeInvoicing,
}) })
result = handler.IsAllowed("jim@example.org", "abc-123", func(privileges CompanyPrivileges) bool { 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) assert.True(t, result)
_ = handler.Process(&PrivilegeAdded{ _ = handler.Process(&PrivilegeAdded{
Email: "jim@example.org", SequenceNo: 3,
CompanyID: "abc-123", Email: "jim@example.org",
Privilege: PrivilegeAccounting, CompanyID: "abc-123",
Privilege: PrivilegeAccounting,
}) })
result = handler.IsAllowed("jim@example.org", "abc-123", func(privileges CompanyPrivileges) bool { 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) assert.True(t, result)
_ = handler.Process(&PrivilegeAdded{ _ = handler.Process(&PrivilegeAdded{
Email: "jim@example.org", SequenceNo: 4,
CompanyID: "abc-123", Email: "jim@example.org",
Privilege: PrivilegeSupplier, CompanyID: "abc-123",
Privilege: PrivilegeSupplier,
}) })
result = handler.IsAllowed("jim@example.org", "abc-123", func(privileges CompanyPrivileges) bool { 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) assert.True(t, result)
_ = handler.Process(&PrivilegeAdded{ _ = handler.Process(&PrivilegeAdded{
Email: "jim@example.org", SequenceNo: 5,
CompanyID: "abc-123", Email: "jim@example.org",
Privilege: PrivilegeSalary, CompanyID: "abc-123",
Privilege: PrivilegeSalary,
}) })
result = handler.IsAllowed("jim@example.org", "abc-123", func(privileges CompanyPrivileges) bool { 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() err := handler.Fetch()
assert.NoError(t, err) assert.NoError(t, err)
expectedPrivileges := map[string]map[string]*CompanyPrivileges{ expectedPrivileges := map[string]map[string]CompanyPrivileges{
"jim@example.org": { "jim@example.org": {
"00010203-0405-4607-8809-0a0b0c0d0e0f": { "00010203-0405-4607-8809-0a0b0c0d0e0f": {
Admin: false, 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) { 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 // Verify privileges were set correctly
expectedPrivileges := map[string]map[string]*CompanyPrivileges{ expectedPrivileges := map[string]map[string]CompanyPrivileges{
"jim@example.org": { "jim@example.org": {
"00010203-0405-4607-8809-0a0b0c0d0e0f": { "00010203-0405-4607-8809-0a0b0c0d0e0f": {
Admin: false, 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) { 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) { go func(id string) {
defer wg.Done() defer wg.Done()
_ = handler.Process(&PrivilegeAdded{ _ = handler.Process(&PrivilegeAdded{
Email: "jim@example.org", SequenceNo: 1,
CompanyID: id, Email: "jim@example.org",
Privilege: PrivilegeAdmin, CompanyID: id,
Privilege: PrivilegeAdmin,
}) })
}(companyID) }(companyID)
} }
@@ -635,9 +655,10 @@ func TestPrivilegeHandler_Concurrent_Multiple_Operations(t *testing.T) {
go func(idx int) { go func(idx int) {
defer wg.Done() defer wg.Done()
_ = handler.Process(&PrivilegeAdded{ _ = handler.Process(&PrivilegeAdded{
Email: "jane@example.org", SequenceNo: 1,
CompanyID: fmt.Sprintf("company-%d", idx%5), Email: "jane@example.org",
Privilege: PrivilegeCompany, CompanyID: fmt.Sprintf("company-%d", idx%5),
Privilege: PrivilegeCompany,
}) })
}(i) }(i)
} }
@@ -711,22 +732,333 @@ func TestPrivilegeHandler_process(t *testing.T) {
admin := func(p CompanyPrivileges) bool { return p.Admin } admin := func(p CompanyPrivileges) bool { return p.Admin }
assert.NoError(t, process[UserAdded](handler)(ctx, spec.ConsumableEvent[UserAdded]{ 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.False(t, handler.IsAllowed("jim@example.org", "abc-123", admin))
assert.NoError(t, process[PrivilegeAdded](handler)(ctx, spec.ConsumableEvent[PrivilegeAdded]{ 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.True(t, handler.IsAllowed("jim@example.org", "abc-123", admin))
assert.NoError(t, process[PrivilegeRemoved](handler)(ctx, spec.ConsumableEvent[PrivilegeRemoved]{ 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.False(t, handler.IsAllowed("jim@example.org", "abc-123", admin))
assert.NoError(t, process[UserRemoved](handler)(ctx, spec.ConsumableEvent[UserRemoved]{ 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 })) 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()
}
}
+16 -11
View File
@@ -1,15 +1,18 @@
package client 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 { type UserAdded struct {
Email string `json:"email"` Email string `json:"email"`
CompanyID string `json:"companyId"` CompanyID string `json:"companyId"`
SequenceNo int `json:"sequenceNo"`
} }
// UserRemoved is the event sent when a user is removed from a company // UserRemoved is the event sent when a user is removed from a company
type UserRemoved struct { type UserRemoved struct {
Email string `json:"email"` Email string `json:"email"`
CompanyID string `json:"companyId"` CompanyID string `json:"companyId"`
SequenceNo int `json:"sequenceNo"`
} }
// Privilege is an enumeration of all available privileges // 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 // PrivilegeAdded is the event sent when a new privilege is added
type PrivilegeAdded struct { type PrivilegeAdded struct {
Email string `json:"email"` Email string `json:"email"`
CompanyID string `json:"companyId"` CompanyID string `json:"companyId"`
Privilege Privilege `json:"privilege"` Privilege Privilege `json:"privilege"`
SequenceNo int `json:"sequenceNo"`
} }
// PrivilegeRemoved is the event sent when a privilege is removed // PrivilegeRemoved is the event sent when a privilege is removed
type PrivilegeRemoved struct { type PrivilegeRemoved struct {
Email string `json:"email"` Email string `json:"email"`
CompanyID string `json:"companyId"` CompanyID string `json:"companyId"`
Privilege Privilege `json:"privilege"` Privilege Privilege `json:"privilege"`
SequenceNo int `json:"sequenceNo"`
} }
+1 -1
View File
@@ -3,7 +3,7 @@ module gitea.unbound.se/shiny/authz_client
go 1.26.2 go 1.26.2
require ( require (
codeberg.org/messaging/go-messaging-amqp v0.0.5 codeberg.org/messaging/go-messaging-amqp v0.0.6
codeberg.org/messaging/messaging v0.0.5 codeberg.org/messaging/messaging v0.0.5
github.com/stretchr/testify v1.12.1 github.com/stretchr/testify v1.12.1
) )
+2 -2
View File
@@ -1,5 +1,5 @@
codeberg.org/messaging/go-messaging-amqp v0.0.5 h1:zzcJ+FVWgBPKyTkq9RXF11yvnhs8IN5BALBC+mps1zA= codeberg.org/messaging/go-messaging-amqp v0.0.6 h1:TAPcIspjjg44MNXwbcscW+ol1kIX49lnX2N3dsh0e0s=
codeberg.org/messaging/go-messaging-amqp v0.0.5/go.mod h1:6bSCIkKH0V/oI7vaIq8ttINKYwov9c8ckZrwjfvp7kc= codeberg.org/messaging/go-messaging-amqp v0.0.6/go.mod h1:R4Hg1w964P7OP7vQ9AsXdd4KjgN/YlQ/9+XzXKpEVp4=
codeberg.org/messaging/messaging v0.0.5 h1:/ueH90F4RNPUeeJIwdG9oVhDW7+XjCzwOYq8xc0kfQk= codeberg.org/messaging/messaging v0.0.5 h1:/ueH90F4RNPUeeJIwdG9oVhDW7+XjCzwOYq8xc0kfQk=
codeberg.org/messaging/messaging v0.0.5/go.mod h1:xyWLUcfaVzcN6GWk9uaQsxPVUC2Vm2S89mYZGdiWTWg= codeberg.org/messaging/messaging v0.0.5/go.mod h1:xyWLUcfaVzcN6GWk9uaQsxPVUC2Vm2S89mYZGdiWTWg=
codeberg.org/messaging/messaging/tck v0.0.3 h1:a4Nr7ytFqEJloTOyVeAtMNgO9Fig1ZUirjW4FVlK0lc= codeberg.org/messaging/messaging/tck v0.0.3 h1:a4Nr7ytFqEJloTOyVeAtMNgO9Fig1ZUirjW4FVlK0lc=
+10
View File
@@ -2,5 +2,15 @@
"$schema": "https://docs.renovatebot.com/renovate-schema.json", "$schema": "https://docs.renovatebot.com/renovate-schema.json",
"extends": [ "extends": [
"config:recommended" "config:recommended"
],
"packageRules": [
{
"groupName": "OpenTelemetry",
"matchPackageNames": [
"go.opentelemetry.io/**",
"gitea.unbound.se/shiny/otelsetup",
"gitea.unbound.se/shiny/logging"
]
}
] ]
} }