diff --git a/cmd/fj-bellows/main.go b/cmd/fj-bellows/main.go index f533081..e70a7c0 100644 --- a/cmd/fj-bellows/main.go +++ b/cmd/fj-bellows/main.go @@ -846,7 +846,12 @@ func buildOrchestratorConfig(cfg *config.Config, opts runOpts, buildVersion, aut if err != nil { return orchestrator.Config{}, err } - _ = prov // reserved for Phase C: provider-aware SetFJBAgent wiring lives in its own helper there. + // Phase C: push the resolved knobs into the provider via duck-typing + // so the managed cache renders the same install block as workers. + // Providers that don't implement SetFJBAgent (docker) are unaffected. + if a, ok := prov.(interface{ SetFJBAgent(string, string) }); ok { + a.SetFJBAgent(fjbAgentURL, fjbAgentToken) + } return orchestrator.Config{ Tag: cfg.Tag, MaxScale: cfg.Scale.Max, diff --git a/internal/agent/client.go b/internal/agent/client.go new file mode 100644 index 0000000..5b55b66 --- /dev/null +++ b/internal/agent/client.go @@ -0,0 +1,118 @@ +package agent + +import ( + "context" + "crypto/tls" + "net" + "net/http" + "time" + + "connectrpc.com/connect" + "golang.org/x/net/http2" + + "github.com/hstern/fj-bellows/gen/fjbellows/agent/v1/agentv1connect" +) + +// DialContextFunc is the network-dial primitive a Client uses to open +// TCP connections. The orchestrator passes wg.Tunnel.DialContext so +// agent traffic rides the WireGuard netstack; tests pass an httptest- +// or bufconn-backed dialer. Signature matches net.Dialer.DialContext. +type DialContextFunc func(ctx context.Context, network, address string) (net.Conn, error) + +// ClientOptions configures NewClient. +type ClientOptions struct { + // Addr is the agent's host:port. For workers, the worker's VPC IP + + // the agent listen port; for cache, the cache's WG inner address + + // the agent listen port. + Addr string + + // Token is the per-deployment bearer secret presented on every RPC. + // Empty disables the Authorization header — useful in tests. + Token string + + // DialContext opens TCP connections. Required. + DialContext DialContextFunc + + // RequestTimeout caps how long a single RPC may take. Defaults to + // 30s when zero — long enough for a slow Health call on a busy + // worker, short enough that callers don't sit forever on a dead + // agent. Bidi-streaming RPCs (Exec) should use a per-stream + // context instead. + RequestTimeout time.Duration +} + +// NewClient builds a ConnectRPC AgentServiceClient that dials through +// the supplied DialContext and presents the bearer token. The wire is +// HTTP/2 cleartext (h2c) — the orchestrator↔agent leg is already +// authenticated and encrypted by WG. +func NewClient(opts ClientOptions) agentv1connect.AgentServiceClient { + if opts.RequestTimeout == 0 { + opts.RequestTimeout = 30 * time.Second + } + + httpClient := &http.Client{ + Transport: &http2.Transport{ + // AllowHTTP + DialTLSContext-on-http2 over a plain TCP + // dialer is how Connect speaks h2c. We pretend tls is + // configured but the DialTLS just opens a TCP conn. + AllowHTTP: true, + DialTLSContext: func(ctx context.Context, network, addr string, _ *tls.Config) (net.Conn, error) { + return opts.DialContext(ctx, network, addr) + }, + }, + Timeout: opts.RequestTimeout, + } + + connectOpts := []connect.ClientOption{} + if opts.Token != "" { + connectOpts = append(connectOpts, connect.WithInterceptors(bearerInterceptor(opts.Token))) + } + + return agentv1connect.NewAgentServiceClient(httpClient, "http://"+opts.Addr, connectOpts...) +} + +// bearerInterceptor attaches `Authorization: Bearer ` to every +// outgoing request and stream. +func bearerInterceptor(token string) connect.Interceptor { + auth := "Bearer " + token + return interceptorFunc{ + unary: func(next connect.UnaryFunc) connect.UnaryFunc { + return func(ctx context.Context, req connect.AnyRequest) (connect.AnyResponse, error) { + req.Header().Set("Authorization", auth) + return next(ctx, req) + } + }, + streamClient: func(next connect.StreamingClientFunc) connect.StreamingClientFunc { + return func(ctx context.Context, spec connect.Spec) connect.StreamingClientConn { + conn := next(ctx, spec) + conn.RequestHeader().Set("Authorization", auth) + return conn + } + }, + } +} + +// interceptorFunc adapts plain funcs into connect.Interceptor without +// implementing the full type for every middleware. +type interceptorFunc struct { + unary func(connect.UnaryFunc) connect.UnaryFunc + streamClient func(connect.StreamingClientFunc) connect.StreamingClientFunc +} + +func (f interceptorFunc) WrapUnary(next connect.UnaryFunc) connect.UnaryFunc { + if f.unary == nil { + return next + } + return f.unary(next) +} + +func (f interceptorFunc) WrapStreamingClient(next connect.StreamingClientFunc) connect.StreamingClientFunc { + if f.streamClient == nil { + return next + } + return f.streamClient(next) +} + +func (f interceptorFunc) WrapStreamingHandler(next connect.StreamingHandlerFunc) connect.StreamingHandlerFunc { + return next +} diff --git a/internal/agent/client_test.go b/internal/agent/client_test.go new file mode 100644 index 0000000..3566867 --- /dev/null +++ b/internal/agent/client_test.go @@ -0,0 +1,202 @@ +package agent + +import ( + "context" + "errors" + "net" + "net/http" + "strings" + "sync" + "testing" + "time" + + "connectrpc.com/connect" + + agentv1 "github.com/hstern/fj-bellows/gen/fjbellows/agent/v1" +) + +// newTestServerForClient starts an in-process agent server bound to +// 127.0.0.1:0 with HTTP/2 cleartext (h2c) so bidi-streaming RPCs work +// without TLS — same shape as production where the wire rides WG. The +// returned DialContext bypasses normal DNS and connects directly to +// the bound listener, mirroring how wg.Tunnel.DialContext will route in +// production. +func newTestServerForClient(t *testing.T, token string) (addr string, dial DialContextFunc, teardown func()) { + t.Helper() + h := NewHandler("test", time.Now()) + var opts []Option + if token != "" { + opts = append(opts, WithBearerToken(token)) + } + srv := NewServer("127.0.0.1:0", h, nil, opts...) + + var lc net.ListenConfig + ln, err := lc.Listen(context.Background(), "tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + var protos http.Protocols + protos.SetHTTP1(true) + protos.SetUnencryptedHTTP2(true) + hsrv := &http.Server{ + Handler: srv.Handler(), + Protocols: &protos, + ReadHeaderTimeout: 5 * time.Second, + } + serveErr := make(chan error, 1) + go func() { + err := hsrv.Serve(ln) + if errors.Is(err, http.ErrServerClosed) { + err = nil + } + serveErr <- err + }() + + host := ln.Addr().String() + dial = func(ctx context.Context, _, _ string) (net.Conn, error) { + var d net.Dialer + return d.DialContext(ctx, "tcp", host) + } + + var once sync.Once + teardown = func() { + once.Do(func() { + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + _ = hsrv.Shutdown(ctx) + <-serveErr + }) + } + return host, dial, teardown +} + +func TestClient_Health_NoAuth(t *testing.T) { + t.Parallel() + host, dial, teardown := newTestServerForClient(t, "") + defer teardown() + + client := NewClient(ClientOptions{ + Addr: host, + DialContext: dial, + }) + resp, err := client.Health(context.Background(), connect.NewRequest(&agentv1.HealthRequest{})) + if err != nil { + t.Fatalf("Health: %v", err) + } + if !resp.Msg.Ready { + t.Errorf("ready = false, want true") + } + if resp.Msg.BuildVersion != "test" { + t.Errorf("build_version = %q, want %q", resp.Msg.BuildVersion, "test") + } +} + +func TestClient_Health_BearerToken_OK(t *testing.T) { + t.Parallel() + host, dial, teardown := newTestServerForClient(t, "s3cret") + defer teardown() + + client := NewClient(ClientOptions{ + Addr: host, + Token: "s3cret", + DialContext: dial, + }) + resp, err := client.Health(context.Background(), connect.NewRequest(&agentv1.HealthRequest{})) + if err != nil { + t.Fatalf("Health: %v", err) + } + if !resp.Msg.Ready { + t.Errorf("ready = false, want true") + } +} + +func TestClient_Health_WrongToken_Rejected(t *testing.T) { + t.Parallel() + host, dial, teardown := newTestServerForClient(t, "right") + defer teardown() + + client := NewClient(ClientOptions{ + Addr: host, + Token: "wrong", + DialContext: dial, + }) + _, err := client.Health(context.Background(), connect.NewRequest(&agentv1.HealthRequest{})) + if err == nil { + t.Fatal("Health succeeded with wrong token, want CodeUnauthenticated") + } + if connect.CodeOf(err) != connect.CodeUnauthenticated { + t.Errorf("code = %v, want CodeUnauthenticated", connect.CodeOf(err)) + } +} + +// Confirm the bearer interceptor sets the header even on stream RPCs +// (Exec is bidi; the orchestrator side will use this client for it). +func TestClient_Exec_BearerToken_OK(t *testing.T) { + t.Parallel() + host, dial, teardown := newTestServerForClient(t, "s3cret") + defer teardown() + + client := NewClient(ClientOptions{ + Addr: host, + Token: "s3cret", + DialContext: dial, + }) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + stream := client.Exec(ctx) + defer func() { _ = stream.CloseRequest() }() + + if err := stream.Send(&agentv1.ShellMsg{Kind: &agentv1.ShellMsg_Open{ + Open: &agentv1.ExecOpen{Argv: []string{execTestArgvBin, "via-client-helper"}}, + }}); err != nil { + t.Fatalf("Send Open: %v", err) + } + var got strings.Builder + for { + ev, err := stream.Receive() + if err != nil { + break + } + switch k := ev.GetKind().(type) { + case *agentv1.ShellEvent_Stdout: + got.Write(k.Stdout.GetData()) + case *agentv1.ShellEvent_Exit: + if k.Exit.GetCode() != 0 { + t.Errorf("exit = %d, want 0", k.Exit.GetCode()) + } + if want := "via-client-helper\n"; got.String() != want { + t.Errorf("stdout = %q, want %q", got.String(), want) + } + return + } + } +} + +// Sanity: confirm we route via DialContext, not via the default dialer. +// We pass a DialContext that errors and assert the call fails with that +// error. +func TestClient_UsesDialContext(t *testing.T) { + t.Parallel() + sentinel := dialSentinelError("dial sentinel reached") + client := NewClient(ClientOptions{ + Addr: "host-that-should-not-be-resolved-by-default-dialer:9001", + DialContext: func(_ context.Context, _, _ string) (net.Conn, error) { + return nil, sentinel + }, + RequestTimeout: 2 * time.Second, + }) + _, err := client.Health(context.Background(), connect.NewRequest(&agentv1.HealthRequest{})) + if err == nil { + t.Fatal("Health succeeded; want sentinel error from DialContext") + } + if !strings.Contains(err.Error(), "dial sentinel reached") { + t.Errorf("expected sentinel in error, got %v", err) + } +} + +type dialSentinelError string + +func (e dialSentinelError) Error() string { return string(e) } + +// Sanity: the helper does not import http blindly. +var _ = http.NoBody diff --git a/internal/agent/exec_test.go b/internal/agent/exec_test.go index ee557c9..bf3183a 100644 --- a/internal/agent/exec_test.go +++ b/internal/agent/exec_test.go @@ -17,6 +17,10 @@ import ( "github.com/hstern/fj-bellows/gen/fjbellows/agent/v1/agentv1connect" ) +// execTestArgvBin is the binary used across exec tests. Extracted so +// goconst doesn't complain about repeated string literals. +const execTestArgvBin = "echo" + // execTestSetup spins up an in-process agent server reachable via H2C // (Connect bidi streaming needs HTTP/2). Returns a connected client and // a teardown. @@ -86,7 +90,7 @@ func TestExec_Echo(t *testing.T) { defer teardown() out, errOut, code, sig, err := runExec(t, client, - &agentv1.ExecOpen{Argv: []string{"echo", "hello"}}, nil) + &agentv1.ExecOpen{Argv: []string{execTestArgvBin, "hello"}}, nil) if err != nil { t.Fatalf("Exec: %v", err) } @@ -190,7 +194,7 @@ func TestExec_RejectsTTY(t *testing.T) { defer teardown() _, _, _, _, err := runExec(t, client, - &agentv1.ExecOpen{Argv: []string{"echo"}, Tty: true}, nil) + &agentv1.ExecOpen{Argv: []string{execTestArgvBin}, Tty: true}, nil) if err == nil { t.Fatal("Exec with tty=true succeeded, want CodeUnimplemented") } @@ -300,7 +304,7 @@ func TestExec_BearerAuth(t *testing.T) { stream := client.Exec(context.Background()) defer func() { _ = stream.CloseRequest() }() _ = stream.Send(&agentv1.ShellMsg{Kind: &agentv1.ShellMsg_Open{ - Open: &agentv1.ExecOpen{Argv: []string{"echo"}}, + Open: &agentv1.ExecOpen{Argv: []string{execTestArgvBin}}, }}) _, err := stream.Receive() if err == nil { diff --git a/internal/bootstrap/bootstrap.go b/internal/bootstrap/bootstrap.go index de98695..7f8508f 100644 --- a/internal/bootstrap/bootstrap.go +++ b/internal/bootstrap/bootstrap.go @@ -18,6 +18,12 @@ var cloudInitTemplate string //go:embed fjbagent.service var fjbagentServiceUnit string +// FJBAgentServiceUnit returns the embedded fjbagent systemd unit so +// other cloud-init renderers (the linode cache, etc.) can write the +// same unit without duplicating the file. Returning a copy keeps the +// embedded variable read-only to callers. +func FJBAgentServiceUnit() string { return fjbagentServiceUnit } + // DefaultReadyFile is touched by cloud-init once the worker is provisioned. // The orchestrator polls for it over SSH to decide a node is ready. const DefaultReadyFile = "/run/fj-bellows-ready" diff --git a/internal/provider/linode/cache-cloud-init.yaml.tmpl b/internal/provider/linode/cache-cloud-init.yaml.tmpl index 06b535b..240208e 100644 --- a/internal/provider/linode/cache-cloud-init.yaml.tmpl +++ b/internal/provider/linode/cache-cloud-init.yaml.tmpl @@ -25,7 +25,38 @@ packages: - wireguard - awscli {{- end}} +{{- if .FJBAgentDownloadURL}} +# fjbagent (FJB-94 Phase C, cache role). Same shape as the worker +# install — unprivileged `fjb` system user, /etc/fjbagent/auth.token, +# systemd Type=notify unit. The orchestrator dials this listener over +# WG (the cache is a direct WG peer); 0.0.0.0:9001 is fine because the +# cache firewall drops public tcp/9001 and WG-tunneled traffic doesn't +# traverse the public stack. +groups: + - fjb +users: + - name: fjb + system: true + no_create_home: true + shell: /usr/sbin/nologin + primary_group: fjb +{{- end}} write_files: +{{- if .FJBAgentDownloadURL}} + - path: /etc/fjbagent/auth.token + permissions: '0600' + owner: 'fjb:fjb' + content: | + {{.FJBAgentToken}} + - path: /etc/default/fjbagent + permissions: '0644' + content: | + FJBAGENT_LISTEN=0.0.0.0:{{.FJBAgentListenPort}} + - path: /etc/systemd/system/fjbagent.service + permissions: '0644' + content: | +{{.FJBAgentServiceUnit | indent 6}} +{{- end}} - path: /etc/zot/config.json permissions: '0640' content: | @@ -166,4 +197,18 @@ runcmd: {{- end}} - systemctl daemon-reload - systemctl enable --now zot.service +{{- if .FJBAgentDownloadURL}} + - | + set -eu + arch="$(uname -m)" + case "$arch" in + x86_64) rarch=amd64 ;; + aarch64|arm64) rarch=arm64 ;; + *) echo "unsupported arch: $arch" >&2; exit 1 ;; + esac + url="{{.FJBAgentDownloadURL}}" + curl -fsSL -o /usr/local/bin/fjbagent "$url" + chmod +x /usr/local/bin/fjbagent + - systemctl enable --now fjbagent +{{- end}} - touch {{.ReadyFile}} diff --git a/internal/provider/linode/cache.go b/internal/provider/linode/cache.go index b9d9b6a..e51e214 100644 --- a/internal/provider/linode/cache.go +++ b/internal/provider/linode/cache.go @@ -20,6 +20,7 @@ import ( "github.com/linode/linodego" + "github.com/hstern/fj-bellows/internal/bootstrap" "github.com/hstern/fj-bellows/internal/transport/cachegateway" ) @@ -197,6 +198,13 @@ type managedCache struct { vpcSubnetID int authorizedKey string + // fjbAgentDownloadURL + fjbAgentToken drive the cache's fjbagent + // install (FJB-94 Phase C). Empty disables agent install on the + // cache, matching the worker-side opt-in. Wired by SetFJBAgent at + // provider configure time. + fjbAgentDownloadURL string + fjbAgentToken string + // workerVPCSubnet is the CIDR (e.g. "10.0.0.0/24") of the VPC subnet // workers attach to. Used at cache-create time to bake the // fjb-iptables FORWARD chain into cloud-init (FJB-98). Empty when @@ -311,6 +319,14 @@ func (m *managedCache) setHardwareContext(firewallID, vpcSubnetID int, authorize m.workerVPCSubnet = workerVPCSubnet } +// setFJBAgent wires the fjbagent install knobs that the cache cloud-init +// renderer expects (FJB-94 Phase C). Both empty disables the install; +// both non-empty enables it. Called by Linode.SetFJBAgent. +func (m *managedCache) setFJBAgent(downloadURL, token string) { + m.fjbAgentDownloadURL = downloadURL + m.fjbAgentToken = token +} + // setOrchestratorWGPubkey supplies the orchestrator's WG public key // (FJB-99 Phase A). Called by the Linode provider after the daemon has // loaded its own keypair from disk; the value lands in cloud-init at @@ -394,16 +410,19 @@ func (m *managedCache) createFreshCacheLinode(ctx context.Context, pair cacheCer return fmt.Errorf("render fjb-iptables: %w", err) } params := cacheCloudInitParams{ - Bucket: creds.Bucket, - Region: creds.Region, - Endpoint: creds.Endpoint, - AccessKey: creds.AccessKey, - SecretKey: creds.SecretKey, - ZotVersion: m.cfg.resolvedZotVersion(), - ReadyFile: defaultCacheReadyFile, - ServerCertPEM: string(pair.ServerCertPEM), - ServerKeyPEM: string(pair.ServerKeyPEM), - IPTablesScript: iptablesScript, + Bucket: creds.Bucket, + Region: creds.Region, + Endpoint: creds.Endpoint, + AccessKey: creds.AccessKey, + SecretKey: creds.SecretKey, + ZotVersion: m.cfg.resolvedZotVersion(), + ReadyFile: defaultCacheReadyFile, + ServerCertPEM: string(pair.ServerCertPEM), + ServerKeyPEM: string(pair.ServerKeyPEM), + IPTablesScript: iptablesScript, + FJBAgentDownloadURL: m.fjbAgentDownloadURL, + FJBAgentToken: m.fjbAgentToken, + FJBAgentServiceUnit: bootstrap.FJBAgentServiceUnit(), } if m.transportMode == "cache-gateway" && m.orchestratorWGPubkey != "" { params.OrchestratorWGPubkey = m.orchestratorWGPubkey @@ -777,6 +796,17 @@ type cacheCloudInitParams struct { OrchestratorWGAddr string // e.g. "100.64.0.1" CacheWGAddr string // e.g. "100.64.0.2" WGListenPort int // UDP, e.g. 51820 + + // FJBAgentDownloadURL + FJBAgentToken + FJBAgentServiceUnit drive + // the same fjbagent install on the cache as on workers (FJB-94 + // Phase C). All-or-nothing: when FJBAgentDownloadURL is non-empty, + // FJBAgentToken and FJBAgentServiceUnit must also be populated. + // FJBAgentListenPort defaults to bootstrap.DefaultAgentListenPort + // in the renderer when zero. + FJBAgentDownloadURL string + FJBAgentToken string + FJBAgentServiceUnit string + FJBAgentListenPort int } // renderCacheCloudInit fills the embedded template. Defaults to the @@ -789,6 +819,17 @@ func renderCacheCloudInit(p cacheCloudInitParams) (string, error) { if p.ReadyFile == "" { p.ReadyFile = defaultCacheReadyFile } + if p.FJBAgentDownloadURL != "" { + if p.FJBAgentToken == "" { + return "", errors.New("cache cloud-init: FJBAgentToken required when FJBAgentDownloadURL is set") + } + if p.FJBAgentServiceUnit == "" { + return "", errors.New("cache cloud-init: FJBAgentServiceUnit required when FJBAgentDownloadURL is set") + } + if p.FJBAgentListenPort == 0 { + p.FJBAgentListenPort = bootstrap.DefaultAgentListenPort + } + } tmpl, err := template.New("cache").Funcs(template.FuncMap{ "b64enc": func(s string) string { return base64.StdEncoding.EncodeToString([]byte(s)) }, "indent": func(spaces int, s string) string { diff --git a/internal/provider/linode/cache_test.go b/internal/provider/linode/cache_test.go index 7c91309..445691b 100644 --- a/internal/provider/linode/cache_test.go +++ b/internal/provider/linode/cache_test.go @@ -1020,3 +1020,81 @@ func TestWorkerExtrasErrorsWhenVPCIPNotAssigned(t *testing.T) { t.Errorf("error should mention VPC IP, got: %v", err) } } + +func TestRenderCacheCloudInit_WithoutFJBAgent_NoAgentArtifacts(t *testing.T) { + out, err := renderCacheCloudInit(cacheCloudInitParams{ //nolint:gosec // G101 false-positive on PEM-shaped test stubs + Bucket: "b", + Region: testBucketRegion, + Endpoint: testStubEndpoint, + AccessKey: "AK", + SecretKey: "SK", + ZotVersion: defaultZotVersion, + ServerCertPEM: "-----BEGIN CERTIFICATE-----\nX\n-----END CERTIFICATE-----\n", + ServerKeyPEM: "-----BEGIN EC PRIVATE KEY-----\nX\n-----END EC PRIVATE KEY-----\n", + }) + if err != nil { + t.Fatalf("render: %v", err) + } + for _, needle := range []string{"fjbagent.service", "/etc/fjbagent/auth.token", "name: fjb"} { + if strings.Contains(out, needle) { + t.Errorf("agent artifact %q leaked into agent-disabled cache render", needle) + } + } +} + +func TestRenderCacheCloudInit_WithFJBAgent_ProducesAllArtifacts(t *testing.T) { + out, err := renderCacheCloudInit(cacheCloudInitParams{ //nolint:gosec // G101 false-positive on PEM-shaped test stubs + Bucket: "b", + Region: testBucketRegion, + Endpoint: testStubEndpoint, + AccessKey: "AK", + SecretKey: "SK", + ZotVersion: defaultZotVersion, + ServerCertPEM: "-----BEGIN CERTIFICATE-----\nX\n-----END CERTIFICATE-----\n", + ServerKeyPEM: "-----BEGIN EC PRIVATE KEY-----\nX\n-----END EC PRIVATE KEY-----\n", + FJBAgentDownloadURL: "https://example.com/fjbagent-linux-$rarch", + FJBAgentToken: "cafebabedead", + FJBAgentServiceUnit: "[Unit]\nDescription=fj-bellows agent\n", + }) + if err != nil { + t.Fatalf("render: %v", err) + } + wants := []string{ + "name: fjb", + "primary_group: fjb", + "/etc/fjbagent/auth.token", + "cafebabedead", + "FJBAGENT_LISTEN=0.0.0.0:9001", + "/etc/systemd/system/fjbagent.service", + "[Unit]", + "https://example.com/fjbagent-linux-$rarch", + "systemctl enable --now fjbagent", + } + for _, needle := range wants { + if !strings.Contains(out, needle) { + t.Errorf("expected %q in cache cloud-init:\n%s", needle, out) + } + } +} + +func TestRenderCacheCloudInit_FJBAgent_RequiresToken(t *testing.T) { + _, err := renderCacheCloudInit(cacheCloudInitParams{ //nolint:gosec // G101 false-positive on PEM-shaped test stubs + Bucket: "b", + Region: testBucketRegion, + Endpoint: testStubEndpoint, + AccessKey: "AK", + SecretKey: "SK", + ZotVersion: defaultZotVersion, + ServerCertPEM: "-----BEGIN CERTIFICATE-----\nX\n-----END CERTIFICATE-----\n", + ServerKeyPEM: "-----BEGIN EC PRIVATE KEY-----\nX\n-----END EC PRIVATE KEY-----\n", + FJBAgentDownloadURL: "https://example.com/fjbagent-linux-$rarch", + FJBAgentServiceUnit: "[Unit]\n", + // Token intentionally unset + }) + if err == nil { + t.Fatal("expected error when FJBAgentDownloadURL set without token") + } + if !strings.Contains(err.Error(), "FJBAgentToken") { + t.Errorf("error did not mention token: %v", err) + } +} diff --git a/internal/provider/linode/linode.go b/internal/provider/linode/linode.go index de12299..33b435e 100644 --- a/internal/provider/linode/linode.go +++ b/internal/provider/linode/linode.go @@ -196,6 +196,17 @@ func (l *Linode) SetTransportMode(mode string) { l.transportMode = mode } +// SetFJBAgent wires the per-deployment fjbagent install knobs into the +// managed cache (FJB-94 Phase C). Empty pair disables the install on +// the cache. Duck-typed from cmd/fj-bellows so non-linode providers +// (docker) are unaffected. No-op when the cache isn't configured. +func (l *Linode) SetFJBAgent(downloadURL, token string) { + if l.cache == nil { + return + } + l.cache.setFJBAgent(downloadURL, token) +} + // SetACLSource wires the orchestrator's ACL snapshot adapter into the // Linode provider's managed cache (FJB-88 / FJB-90). The provider // reads it on every Provision so the worker cloud-init's `ip route