1package app
2
3import (
4 "context"
5 "encoding/json"
6 "os"
7 "path/filepath"
8 "testing"
9
10 "github.com/bnema/vev/internal/adapters/brokerconfig"
11 "github.com/bnema/vev/internal/adapters/brokeripc"
12 "github.com/bnema/vev/internal/adapters/brokerstore"
13 "github.com/bnema/vev/internal/adapters/clock"
14 "github.com/bnema/vev/internal/adapters/daemonmux"
15 "github.com/bnema/vev/internal/adapters/ipc"
16 "github.com/bnema/vev/internal/domain"
17 "github.com/bnema/vev/internal/ports"
18 "github.com/bnema/vev/internal/usecase/broker"
19 "github.com/bnema/vev/pkg/safedir"
20 "github.com/stretchr/testify/require"
21)
22
23type routeTestHosts struct {
24 record ports.BrokerHostRecord
25 found bool
26}
27
28func (h *routeTestHosts) LookupHost(ctx context.Context, endpoint string) (ports.BrokerHostRecord, bool, error) {
29 return h.record, h.found && endpoint == h.record.Registration.Endpoint, ctx.Err()
30}
31
32func TestBrokerRoutesDurableFirstAddAndRestart(t *testing.T) {
33 ctx := context.Background()
34 dir := filepath.Join(t.TempDir(), "state")
35 store, err := brokerstore.Open(brokerstore.Options{Dir: dir})
36 require.NoError(t, err)
37 defer func() { require.NoError(t, store.Close()) }()
38 routes := newBrokerRoutes(nil)
39 probe := &broker.RemoteProbe{Epoch: 1, Routes: routes}
40 registry, err := broker.NewRegistryWithConfig(1, store, probe, clock.New(), nil, broker.RegistryConfig{MembershipMode: broker.MembershipMutable})
41 require.NoError(t, err)
42 routes.Hosts = registry
43 policy := remoteBrokerPolicy(hostTransportSSH)
44 reg, err := registry.AddHost(ctx, "user@example.test", policy)
45 require.NoError(t, err)
46 request := ports.BrokerOpenStreamRequest{Endpoint: reg.Endpoint, Registration: reg, Policy: policy, StartMode: ports.BrokerDaemonExistingOnly}
47 target, err := routes.ResolveDialTarget(ctx, request)
48 require.NoError(t, err)
49 require.False(t, target.ExpectedIdentity.Bound)
50 spec, err := routes.AuthorizeRemoteDial(ctx, target)
51 require.NoError(t, err)
52 route, err := brokerconfig.RouteFromSpec(spec)
53 require.NoError(t, err)
54 require.Equal(t, brokerconfig.RouteSSHStdio, route.Kind())
55 require.Equal(t, []string{"vev", brokerMuxStdioCommand, "--production"}, route.Argv())
56 identity, err := registry.BindAuthenticatedIdentity(ctx, ports.BrokerIdentityBindingRequest{Fence: target.Fence, Policy: policy, Identity: "daemon-test"})
57 require.NoError(t, err)
58 _, err = routes.AuthorizeRemoteDial(ctx, target)
59 require.Error(t, err, "an unbound target cannot bypass a newly committed identity")
60 target, err = routes.ResolveDialTarget(ctx, request)
61 require.NoError(t, err)
62 require.Equal(t, identity, target.ExpectedIdentity.Identity)
63 require.True(t, target.ExpectedIdentity.Bound)
64 cancelled, cancel := context.WithCancel(ctx)
65 cancel()
66 registry.Run(cancelled) // flush advisory writes before closing durable ownership
67 require.NoError(t, store.Close())
68 store, err = brokerstore.Open(brokerstore.Options{Dir: dir})
69 require.NoError(t, err)
70 restarted, err := broker.NewRegistryWithConfig(2, store, probe, clock.New(), nil, broker.RegistryConfig{MembershipMode: broker.MembershipMutable})
71 require.NoError(t, err)
72 routes.Hosts = restarted
73 recovered, err := routes.ResolveDialTarget(ctx, request)
74 require.NoError(t, err)
75 require.Equal(t, target, recovered)
76 restarted.Run(cancelled)
77}
78
79func TestCanonicalBrokerRoutesSelectProductionHelpers(t *testing.T) {
80 for _, transport := range []string{"ssh-stdio", "ssh-quic"} {
81 t.Run(transport, func(t *testing.T) {
82 spec, err := ports.BrokerRouteForTransport(transport, "user@example.test")
83 require.NoError(t, err)
84 cmd, err := parseArgs(spec.Argv[1:])
85 require.NoError(t, err)
86 require.Equal(t, ports.BrokerDaemonExistingOnly, cmd.brokerMux.startMode)
87 for _, args := range [][]string{{"--production", "--production"}, {"--production", "--offline-root", "/tmp/root"}, {"--offline-root", "/tmp/root", "--production"}} {
88 _, err = parseBrokerMuxArgs(spec.Argv[1], cmd.kind, args)
89 require.Error(t, err)
90 }
91 })
92 }
93}
94
95// The real production composition starts with no remote import. The first IPC
96// add must succeed and the same running registry must retain it across restart.
97func TestProductionBrokerFirstHostAdd(t *testing.T) {
98 t.Setenv("XDG_RUNTIME_DIR", shortTempDir(t, "vevroute"))
99 t.Setenv("XDG_STATE_HOME", t.TempDir())
100 t.Setenv("XDG_CONFIG_HOME", t.TempDir())
101 require.NoError(t, safedir.EnsurePrivate(filepath.Dir(productionBrokerConfigPath())))
102 raw, err := json.Marshal(map[string]any{"marker": brokerconfig.Marker, "registrations": []any{}})
103 require.NoError(t, err)
104 require.NoError(t, os.WriteFile(productionBrokerConfigPath(), raw, 0600))
105 var registered domain.RemoteRegistration
106 for i := 0; i < 2; i++ {
107 deps, ready := testBrokerServeDeps(newSandboxClock())
108 ctx, cancel := context.WithCancel(context.Background())
109 done := runSandbox(ctx, brokerServeOptions{}, deps)
110 socket := awaitSandboxReady(t, ready, done)
111 service, err := brokeripc.NewConnector(socket, brokeripc.Config{}).Connect(ctx)
112 require.NoError(t, err)
113 reg, err := service.AddHost(ctx, "user@127.0.0.1:1", remoteBrokerPolicy(hostTransportSSH))
114 require.NoError(t, err)
115 if i == 0 {
116 registered = reg
117 } else {
118 require.Equal(t, registered, reg)
119 }
120 require.NoError(t, service.Close())
121 awaitSandboxCancel(t, cancel, done)
122 }
123}
124
125func TestBrokerDynamicConnectorDialsDurableRouteWithoutConfigEntry(t *testing.T) {
126 policy := brokerTestPolicy()
127 fixture := newBrokerMuxFixture(t, policy)
128 reg := brokerTestRegistration()
129 hosts := &routeTestHosts{found: true, record: ports.BrokerHostRecord{
130 Registration: reg, Policy: policy, Route: ports.BrokerRouteSpec{Kind: ports.BrokerRouteUnix, Path: fixture.route},
131 }}
132 routes := newBrokerRoutes(nil)
133 routes.Hosts = hosts
134 target, err := routes.ResolveDialTarget(context.Background(), ports.BrokerOpenStreamRequest{
135 Endpoint: reg.Endpoint, Registration: reg, Policy: policy, StartMode: ports.BrokerDaemonExistingOnly,
136 })
137 require.NoError(t, err)
138 connector, err := brokerDynamicMuxConnector(nil, routes, nil)
139 require.NoError(t, err)
140 physical, err := connector.Connect(context.Background(), target)
141 require.NoError(t, err)
142 require.Equal(t, ports.BrokerDaemonIdentity(brokerTestIdentity), physical.Identity())
143 require.NoError(t, physical.Close())
144 hosts.found = false
145 _, err = connector.Connect(context.Background(), target)
146 require.Error(t, err)
147}
148
149func TestProductionMuxHelperExistingOnlyNeverSpawns(t *testing.T) {
150 t.Setenv("XDG_RUNTIME_DIR", shortTempDir(t, "vevhelper"))
151 previous := localDaemonStarter
152 defer func() { localDaemonStarter = previous }()
153 dials, spawns := 0, 0
154 localDaemonStarter = brokerDaemonStarter{
155 dial: func(_ context.Context, path string) (daemonmux.RawFramedTransport, error) {
156 dials++
157 require.Equal(t, daemonmux.SocketPath(ipc.SocketDir()), path)
158 return nil, os.ErrNotExist
159 },
160 spawn: func() error { spawns++; return nil },
161 }
162 _, err := dialBrokerMuxHelper(context.Background(), brokerMuxOptions{startMode: ports.BrokerDaemonExistingOnly})
163 require.ErrorIs(t, err, os.ErrNotExist)
164 require.Equal(t, 1, dials)
165 require.Zero(t, spawns)
166}
167
168func TestBrokerRoutesLocalRemainsConfigured(t *testing.T) {
169 config, binding := testLocalSandboxConfigForCarriage(t, filepath.Join(t.TempDir(), "daemon.sock"))
170 routes := newBrokerRoutes(config) // no remote authority at all
171 request := ports.BrokerOpenStreamRequest{Local: true, Policy: binding.Policy, StartMode: ports.BrokerDaemonExistingOnly}
172 target, err := routes.ResolveDialTarget(context.Background(), request)
173 require.NoError(t, err)
174 require.Equal(t, binding.Route.Address(), target.Address)
175 require.Equal(t, binding.Identity, target.ExpectedIdentity.Identity)
176 request.Policy.Trust = "caller-selected"
177 _, err = routes.ResolveDialTarget(context.Background(), request)
178 require.Error(t, err)
179}