amadan
The drain: credential outside the home, and an honest instance state
Two findings drove this, one from review and one from reading the code it had to fit into. Review (Astra) was decisive about where the credential lives. The platform now HAS a machine credential for instance creation — carlos svcrole's 'instance' role — so section 7's premise that none existed is obsolete and self-serve is possible. But a credential held by the web process is reachable from the web process, and an in-process queue constrains nobody who has stolen it: they call the console directly and skip the queue. So the queue stays in the home and the credential lives in this package, in a process with no listening socket. Signup is self-serve with no human in it, and the conflation section 7 feared stays impossible by construction. Reading internal/operator/directory.go:269 corrected the other half. Provisioning an instance is not one row: the operator path mints the consumer token, signs a roster provisioning record and writes roster_consumers beside team_instances, and that signed binding is what ruling 23's roster verifies against. Nothing in directorydata can produce it — it needs the directory's signing key and an authorized operation. So MarkProvisioned now writes 'pending', never 'live'. A row it writes is a claim about infrastructure, not about standing, and Choices already declines to offer a member a pending instance. Writing 'live' would have put a team's app in front of people while the record the rest of the system reads was still missing. Run is one pass: release what expired, then build what is waiting. A request that fails is counted and skipped, never fatal to the others — it stays queued for the next pass, and its TTL still expires it if it never succeeds, so a permanently broken request cannot hold its label for ever. A failed recording leaves the request queued too, which is why Provision is required to be idempotent: the drain can crash between building and recording, and the retry has to converge rather than build twice. A drain with no provisioner refuses rather than reporting a successful pass with nothing done, because those two look identical in a log.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
4 files changed,
+307
−5
internal/drain/drain.go+114 −0internal/drain/drain_test.go+146 −0internal/home/directorydata/provision.go+17 −5internal/home/directorydata/provision_test.go+30 −0
diff --git a/internal/drain/drain.go b/internal/drain/drain.go| new file mode 100644 |
| index 0000000..9a2ae74 |
| --- /dev/null |
| +++ b/internal/drain/drain.go |
| @@ -0,0 +1,114 @@ |
| +// Package drain turns queued provisioning requests into infrastructure. |
| +// |
| +// WHY THIS IS NOT IN THE HOME PROCESS. §7's load-bearing safety fact is that |
| +// the home holds no provisioning credential, so an RCE on an internet-facing |
| +// sign-in surface cannot reach instance creation. The platform has since |
| +// shipped service principals (carlos svcrole), which means a machine |
| +// credential for instance creation now exists where §7 said none did — but |
| +// a credential held by the web process is still reachable from the web |
| +// process, and an in-process queue constrains nobody who has stolen it: they |
| +// call the console directly and skip the queue entirely. |
| +// |
| +// So the queue stays in the home and the credential lives out here, in a |
| +// process with no listening socket. That keeps signup self-serve with no |
| +// human in the loop while leaving the dangerous conflation impossible by |
| +// construction rather than unlikely by policy, which is what §7 was |
| +// protecting. |
| +package drain |
| + |
| +import ( |
| + "context" |
| + "errors" |
| + "log/slog" |
| + "time" |
| + |
| + "amadan.net/oficina/home/internal/home/directorydata" |
| +) |
| + |
| +// Queue is the directory's side: what was asked for, and where to record |
| +// that it was built. Narrow on purpose — the drain never reads a person, a |
| +// membership or a sealed name. |
| +type Queue interface { |
| + Queued(ctx context.Context, now time.Time, limit int) ([]directorydata.Request, error) |
| + MarkProvisioned(ctx context.Context, id int64, origin, consumerID string, now time.Time) error |
| + Sweep(ctx context.Context, now time.Time) (int, int, error) |
| +} |
| + |
| +// Provisioner is the credential's side: the only thing in this design that |
| +// can create infrastructure. Implementations hold the service credential; |
| +// nothing else does. |
| +type Provisioner interface { |
| + // Provision creates the instance for one (label, app) and returns the |
| + // origin it serves and the consumer id recorded against it. It must be |
| + // safe to call twice for the same pair: the drain can crash between |
| + // provisioning and recording, and the retry has to converge rather than |
| + // build a second instance. |
| + Provision(ctx context.Context, label, app string) (origin, consumerID string, err error) |
| +} |
| + |
| +// Result reports one pass, for a caller that logs or alerts on it. |
| +type Result struct { |
| + Swept, Released, Provisioned, Failed int |
| +} |
| + |
| +// ErrNoProvisioner is returned rather than silently doing nothing, because a |
| +// drain that runs with no credential and reports success is indistinguishable |
| +// from a drain that has nothing to do. |
| +var ErrNoProvisioner = errors.New("drain: no provisioner") |
| + |
| +// Run makes one pass: release what expired, then build what is waiting. |
| +// |
| +// A request that fails to provision is counted and skipped, never retried in |
| +// the same pass and never fatal to the others. It stays queued, so the next |
| +// pass tries it again, and its TTL still expires it if it never succeeds — |
| +// which is what stops a permanently broken request holding its label for |
| +// ever. |
| +func Run(ctx context.Context, q Queue, p Provisioner, now time.Time, limit int, log *slog.Logger) (Result, error) { |
| + var out Result |
| + if q == nil || limit <= 0 { |
| + return out, errors.New("drain: invalid configuration") |
| + } |
| + if p == nil { |
| + return out, ErrNoProvisioner |
| + } |
| + if log == nil { |
| + log = slog.New(slog.DiscardHandler) |
| + } |
| + |
| + swept, released, err := q.Sweep(ctx, now) |
| + if err != nil { |
| + // A failed sweep is reported but does not stop the pass: releasing |
| + // expired labels and building new instances are independent jobs, |
| + // and a queue that cannot sweep can still provision. |
| + log.Error("drain: sweep", "err", err) |
| + } |
| + out.Swept, out.Released = swept, released |
| + |
| + requests, err := q.Queued(ctx, now, limit) |
| + if err != nil { |
| + return out, err |
| + } |
| + for _, req := range requests { |
| + if ctx.Err() != nil { |
| + return out, ctx.Err() |
| + } |
| + origin, consumer, err := p.Provision(ctx, req.Label, req.App) |
| + if err != nil { |
| + out.Failed++ |
| + // The label is deliberately absent from this line: a drain log |
| + // is not a place to accumulate who asked for what. |
| + log.Error("drain: provision", "app", req.App, "request", req.ID, "err", err) |
| + continue |
| + } |
| + if err := q.MarkProvisioned(ctx, req.ID, origin, consumer, now); err != nil { |
| + out.Failed++ |
| + // Infrastructure exists that the directory does not know about. |
| + // Provision must be idempotent for exactly this case: the next |
| + // pass calls it again for the same pair and records the result. |
| + log.Error("drain: record", "app", req.App, "request", req.ID, "err", err) |
| + continue |
| + } |
| + out.Provisioned++ |
| + } |
| + return out, nil |
| +} |
diff --git a/internal/drain/drain_test.go b/internal/drain/drain_test.go| new file mode 100644 |
| index 0000000..a05a66d |
| --- /dev/null |
| +++ b/internal/drain/drain_test.go |
| @@ -0,0 +1,146 @@ |
| +package drain |
| + |
| +import ( |
| + "context" |
| + "errors" |
| + "testing" |
| + "time" |
| + |
| + "amadan.net/oficina/home/internal/home/directorydata" |
| +) |
| + |
| +type fakeQueue struct { |
| + queued []directorydata.Request |
| + recorded map[int64]string |
| + sweepErr error |
| + queuedErr error |
| + recordErr map[int64]error |
| + swept int |
| + released int |
| +} |
| + |
| +func (f *fakeQueue) Queued(context.Context, time.Time, int) ([]directorydata.Request, error) { |
| + return f.queued, f.queuedErr |
| +} |
| +func (f *fakeQueue) MarkProvisioned(_ context.Context, id int64, origin, consumer string, _ time.Time) error { |
| + if err := f.recordErr[id]; err != nil { |
| + return err |
| + } |
| + if f.recorded == nil { |
| + f.recorded = map[int64]string{} |
| + } |
| + f.recorded[id] = origin + "|" + consumer |
| + return nil |
| +} |
| +func (f *fakeQueue) Sweep(context.Context, time.Time) (int, int, error) { |
| + return f.swept, f.released, f.sweepErr |
| +} |
| + |
| +type fakeProvisioner struct { |
| + calls []string |
| + failFor map[string]error |
| +} |
| + |
| +func (f *fakeProvisioner) Provision(_ context.Context, label, app string) (string, string, error) { |
| + f.calls = append(f.calls, label+"/"+app) |
| + if err := f.failFor[label+"/"+app]; err != nil { |
| + return "", "", err |
| + } |
| + return "https://" + label + "." + app + ".oficina.works", "consumer-" + label + "-" + app, nil |
| +} |
| + |
| +func requests() []directorydata.Request { |
| + return []directorydata.Request{ |
| + {ID: 1, Label: "acme", App: "docs"}, |
| + {ID: 2, Label: "acme", App: "sheets"}, |
| + {ID: 3, Label: "beta", App: "docs"}, |
| + } |
| +} |
| + |
| +func TestRunProvisionsEveryQueuedRequest(t *testing.T) { |
| + q := &fakeQueue{queued: requests(), swept: 2, released: 1} |
| + p := &fakeProvisioner{} |
| + got, err := Run(context.Background(), q, p, time.Now(), 10, nil) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if got.Provisioned != 3 || got.Failed != 0 || got.Swept != 2 || got.Released != 1 { |
| + t.Fatalf("%+v", got) |
| + } |
| + if len(q.recorded) != 3 || q.recorded[1] != "https://acme.docs.oficina.works|consumer-acme-docs" { |
| + t.Fatalf("recorded %v", q.recorded) |
| + } |
| +} |
| + |
| +// One broken request must not deny every other team its instance. It stays |
| +// queued, so the next pass retries it, and its TTL still expires it if it |
| +// never succeeds — which is what stops it holding its label for ever. |
| +func TestOneFailedProvisionDoesNotStopTheOthers(t *testing.T) { |
| + q := &fakeQueue{queued: requests()} |
| + p := &fakeProvisioner{failFor: map[string]error{"acme/sheets": errors.New("console refused")}} |
| + got, err := Run(context.Background(), q, p, time.Now(), 10, nil) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if got.Provisioned != 2 || got.Failed != 1 { |
| + t.Fatalf("%+v", got) |
| + } |
| + if _, recorded := q.recorded[2]; recorded { |
| + t.Fatal("a failed provision was recorded as done") |
| + } |
| + if len(q.recorded) != 2 { |
| + t.Fatalf("recorded %v", q.recorded) |
| + } |
| +} |
| + |
| +// If recording fails, infrastructure exists that the directory does not know |
| +// about. The request must stay queued so the next pass can converge, which |
| +// is exactly why Provision is required to be idempotent. |
| +func TestAFailedRecordingLeavesTheRequestQueued(t *testing.T) { |
| + q := &fakeQueue{queued: requests(), recordErr: map[int64]error{1: errors.New("db gone")}} |
| + p := &fakeProvisioner{} |
| + got, err := Run(context.Background(), q, p, time.Now(), 10, nil) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if got.Provisioned != 2 || got.Failed != 1 { |
| + t.Fatalf("%+v", got) |
| + } |
| + if _, recorded := q.recorded[1]; recorded { |
| + t.Fatal("request 1 recorded despite a failing write") |
| + } |
| +} |
| + |
| +// A sweep failure is reported but must not cancel provisioning: releasing |
| +// expired labels and building new instances are independent jobs. |
| +func TestASweepFailureStillProvisions(t *testing.T) { |
| + q := &fakeQueue{queued: requests(), sweepErr: errors.New("locked")} |
| + got, err := Run(context.Background(), q, &fakeProvisioner{}, time.Now(), 10, nil) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if got.Provisioned != 3 { |
| + t.Fatalf("%+v", got) |
| + } |
| +} |
| + |
| +// A drain with no credential that reported success would be indistinguishable |
| +// from one with nothing to do — and would look healthy while provisioning |
| +// silently stopped. |
| +func TestRunRefusesWithoutAProvisioner(t *testing.T) { |
| + if _, err := Run(context.Background(), &fakeQueue{}, nil, time.Now(), 10, nil); !errors.Is(err, ErrNoProvisioner) { |
| + t.Fatalf("want ErrNoProvisioner, got %v", err) |
| + } |
| +} |
| + |
| +func TestRunStopsOnACancelledContext(t *testing.T) { |
| + ctx, cancel := context.WithCancel(context.Background()) |
| + cancel() |
| + q := &fakeQueue{queued: requests()} |
| + if _, err := Run(ctx, q, &fakeProvisioner{}, time.Now(), 10, nil); !errors.Is(err, context.Canceled) { |
| + t.Fatalf("want context.Canceled, got %v", err) |
| + } |
| + if len(q.recorded) != 0 { |
| + t.Fatal("work happened after cancellation") |
| + } |
| +} |
diff --git a/internal/home/directorydata/provision.go b/internal/home/directorydata/provision.go| index fbecf2f..5ce7387 100644 |
| --- a/internal/home/directorydata/provision.go |
| +++ b/internal/home/directorydata/provision.go |
| @@ -200,10 +200,22 @@ func (s *Store) Queued(ctx context.Context, now time.Time, limit int) ([]Request |
| return out, nil |
| } |
| -// MarkProvisioned records the instance the drain created and closes the |
| -// request, in one transaction. The team_instances row is what |
| -// <team>.<app>.oficina.works resolves against, so it and the state change |
| -// commit together or neither does. |
| +// MarkProvisioned records that the drain built the INFRASTRUCTURE for a |
| +// request and closes it. It writes the instance as 'pending', never 'live', |
| +// and the distinction is the whole point. |
| +// |
| +// A live instance is not just a row here. internal/operator's provision |
| +// command (directory.go:269) mints the consumer token, signs a roster |
| +// provisioning record and writes roster_consumers alongside team_instances, |
| +// and that signed binding is what the roster — ruling 23's membership |
| +// distribution — verifies against. Nothing in this file can produce it: it |
| +// needs the directory's signing key and an authorized operator operation. |
| +// |
| +// So a row written here is a claim about infrastructure, not about standing. |
| +// 'pending' is what Choices (store.go:141) declines to offer a member, which |
| +// is the correct behaviour for an instance whose roster binding does not |
| +// exist yet. Writing 'live' here would put a team's app in front of people |
| +// while the record the rest of the system reads was still missing. |
| func (s *Store) MarkProvisioned(ctx context.Context, id int64, origin, consumerID string, now time.Time) error { |
| if id <= 0 || !ValidOrigin(origin) || !validConsumer(consumerID) { |
| return ErrInvalid |
| @@ -233,7 +245,7 @@ func (s *Store) MarkProvisioned(ctx context.Context, id int64, origin, consumerI |
| return ErrInvalid |
| } |
| if _, err := tx.ExecContext(ctx, |
| - `INSERT INTO team_instances (team_id, app, origin, consumer_id, state) VALUES (?, ?, ?, ?, 'live')`, |
| + `INSERT INTO team_instances (team_id, app, origin, consumer_id, state) VALUES (?, ?, ?, ?, 'pending')`, |
| teamID, app, origin, consumerID); err != nil { |
| return ErrUnavailable |
| } |
diff --git a/internal/home/directorydata/provision_test.go b/internal/home/directorydata/provision_test.go| index 13f0d58..924263b 100644 |
| --- a/internal/home/directorydata/provision_test.go |
| +++ b/internal/home/directorydata/provision_test.go |
| @@ -58,6 +58,36 @@ func TestCreateTeamIsAtomic(t *testing.T) { |
| mustCreate(t, s, personA, "acme", []string{"docs"}, now) |
| } |
| +// An instance the drain built is 'pending', not 'live': the signed roster |
| +// binding that makes it live is the operator path's to write, and Choices |
| +// must not offer a member an app whose binding does not exist yet. |
| +func TestMarkProvisionedRecordsPendingNotLive(t *testing.T) { |
| + s, d, _ := fixture(t) |
| + now := time.Unix(1_700_000_000, 0).UTC() |
| + mustCreate(t, s, personA, "acme", []string{"docs"}, now) |
| + got, err := s.Queued(context.Background(), now, 10) |
| + if err != nil || len(got) != 1 { |
| + t.Fatal(err) |
| + } |
| + if err := s.MarkProvisioned(context.Background(), got[0].ID, "https://acme.docs.oficina.works", "consumer-1", now); err != nil { |
| + t.Fatal(err) |
| + } |
| + var state string |
| + if err := d.Writer().QueryRow(`SELECT state FROM team_instances WHERE app='docs'`).Scan(&state); err != nil { |
| + t.Fatal(err) |
| + } |
| + if state != "pending" { |
| + t.Fatalf("state %q: the drain must not mark an instance live without its roster binding", state) |
| + } |
| + choices, err := s.Choices(context.Background(), personA, "docs") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if len(choices) != 0 { |
| + t.Fatal("a pending instance was offered to a member") |
| + } |
| +} |
| + |
| func TestQueuedAndMarkProvisioned(t *testing.T) { |
| s, _, _ := fixture(t) |
| now := time.Unix(1_700_000_000, 0).UTC() |