carlos / platform Public

Download this file

1	package host
2
3	import (
4	"context"
5	"github.com/carlosframework/platform/internal/hostbox"
6	"github.com/carlosframework/platform/internal/provision"
7	"net"
8	"os/exec"
9	"testing"
10	"time"
11
12	"github.com/carlosframework/platform/internal/activator"
13	"github.com/carlosframework/platform/internal/hostclaim"
14	"github.com/carlosframework/platform/internal/hostguard"
15	"github.com/carlosframework/platform/internal/registry"
16	)
17
18	func TestPlan14RecoveryInventorySourceGap(t *testing.T) {
19	for _, attack := range []string{"none", "record moved", "claim moved", "fenced", "peer"} {
20	t.Run(attack, func(t *testing.T) {
21	f, o, h := plan12Handover(t, hostclaim.KindRestoring, "stop")
22	holder := o.Claim.Holder
23	authority := f.guard
24	authority.Root = hostguard.NewRootAuthority(&rollbackDisk{dir: t.TempDir()})
25	in := hostguard.RootInstallation{Host: o.Host, Incarnation: holder.Incarnation, Stamp: holder.Stamp, Migration: 1, InventoryRevision: 1, Account: holder.Account, App: holder.App, Record: holder.Key, Box: "box-1", Enrollment: "enrollment-1"}
26	if err := authority.EstablishRoot(t.Context(), in, func(*hostclaim.Claim) error { return nil }); err != nil {
27	t.Fatal(err)
28	}
29	h.Guard.FenceView = authority.Root.View
30	plan14PutRecord(t, f.claims, provision.Record{Schema: 1, Host: o.Host, Account: holder.Account, App: holder.App, Channel: "stable", CreatedAt: f.now})
31	actor := hostbox.Actor{Box: "box-1", Enrollment: "enrollment-1", Revision: 7, Binding: "7"}
32	root := ClaimRootHTTP{Stop: ClaimStopHTTP{Coordinator: ClaimStopCoordinator{Guard: authority, Evidence: f.claims}, Authenticate: func(context.Context, net.Conn) (StopPeer, error) { return f.peer, nil }}, InventoryActor: func(context.Context) (hostbox.Actor, error) { return actor, nil }, Roots: map[string]string{"data": f.guard.DataDir}, Reconcile: hostbox.ReconcileAdapter{Inventory: h.Inventory, Registry: h.Registry, Box: actor.Box, Enrollment: actor.Enrollment}}
33	// Inventory reads use the same authenticated Unix endpoint as production.
34	journals, err := hostbox.NewJournalService(f.journal)
35	if err != nil {
36	t.Fatal(err)
37	}
38	journals.Inventory = h.Inventory
39	root.Journals = journals
40	socket := startRetirementRecoveryTransport(t, root)
41	_, h.Inventory = hostbox.RemoteJournals(rootJournalClient(socket))
42	if attack == "peer" {
43	root.Stop.Authenticate = func(context.Context, net.Conn) (StopPeer, error) { return StopPeer{}, hostbox.ErrChanged }
44	socket = startRetirementRecoveryTransport(t, root)
45	}
46	h.RecoveryRebind = func(ctx context.Context, q recoveryRebindRequest) error {
47	err := recoveryRebindClient(socket)(ctx, q)
48	if err != nil {
49	return err
50	}
51	if attack == "fenced" {
52	_, err = root.Stop.Coordinator.Guard.BeginFence(ctx, hostguard.FenceTarget{Host: q.Next.Host, Incarnation: q.Next.Incarnation, Stamp: q.Next.Stamp, Session: q.Next.Session, Op: q.Next.Op, Phase: q.Next.Phase, Box: q.Next.Box, Enrollment: q.Next.Enrollment, DirectoryRevision: q.Next.DirectoryRevision, ClaimETag: q.Next.ETag, BootID: "boot-1", ControllerID: "controller-1", FenceID: "new-fence"})
53	if err != nil {
54	return err
55	}
56	}
57	if attack == "record moved" {
58	rec := provision.Record{Schema: 1, Host: o.Host, Account: holder.Account, App: holder.App, Channel: "stable", CreatedAt: f.now, RecordSeq: 2}
59	plan14PutRecord(t, f.claims, rec)
60	}
61	if attack == "claim moved" {
62	_, err = hostclaim.Update(ctx, f.claims, f.read(), func(c *hostclaim.Claim) error { c.Op.Journal["moved"] = "yes"; return nil })
63	}
64	return err
65	}
66	before, err := h.Registry.Get(o.Host)
67	if err != nil {
68	t.Fatal(err)
69	}
70	c := f.controller("boot-1", "controller-1", 4101, "start-1")
71	c.wake.Inputs = plan9D2Inputs{recoveryInputs{f}}
72	c.wake.Operations = &plan9D2Attempts{}
73	c.wake.ProbeReady = func(context.Context, *exec.Cmd, registry.Route, ClaimLaunchInputs, string) error { return nil }
74	wireClaimAdmission(c.wake, "box-1", func(context.Context, hostclaim.Observed, hostclaim.Observed) error {
75	t.Fatal("wrong handover")
76	return nil
77	}, h, nil, nil)
78	if err = c.wake.stopPredecessor(t.Context(), o); err != nil {
79	t.Fatal(err)
80	}
81	live, cancel := context.WithCancel(t.Context())
82	defer cancel()
83	done := make(chan struct{})
84	go func() { <-live.Done(); close(done) }()
85	c.wake.mu.Lock()
86	c.wake.pending[o.Host] = &claimWakeAttempt{observed: o, held: recoveryHeld{}, ctx: live, cancel: cancel, done: done}
87	c.wake.mu.Unlock()
88	ctx, stop := context.WithTimeout(t.Context(), 3*time.Second)
89	defer stop()
90	_, err = c.runtime.Admit(ctx, registry.Route{Host: o.Host}, o, activator.StartPolicy{Probe: true})
91	if attack != "none" {
92	if err == nil {
93	t.Fatal("stale or unauthenticated rebind accepted")
94	}
95	after, e := h.Registry.Get(o.Host)
96	if e != nil || after.ClaimOwnership != before.ClaimOwnership {
97	t.Fatal("stale result rebound rows", e)
98	}
99	return
100	}
101	if err != nil {
102	t.Fatal("root recovery admission failed", err)
103	}
104	current := f.read()
105	if current.Claim.Holder.Stamp <= holder.Stamp {
106	t.Fatal("fixture did not reach the new claim stamp", err)
107	}
108	after, e := h.Registry.Get(o.Host)
109	if e != nil || after.ClaimOwnership == before.ClaimOwnership {
110	t.Fatal("rows not rebound after root commit", e)
111	}
112	view, e := authority.Root.View(t.Context(), o.Host, hostguard.Captured{Incarnation: holder.Incarnation, Stamp: holder.Stamp})
113	if e != nil || view.Installation.Stamp != current.Claim.Holder.Stamp {
114	t.Fatal("root installation not rebound", view, e)
115	}
116	t.Logf("fresh claim stamp %d, retained root stamp %d; recovery succeeds: %v", current.Claim.Holder.Stamp, view.Installation.Stamp, err)
117	})
118	}
119	}
120