From fdfc52d4f53483d995f58ae16d8709a3327ac000 Mon Sep 17 00:00:00 2001 From: Henry Stern Date: Thu, 28 May 2026 14:04:54 -0300 Subject: [PATCH 1/2] feat(linode): bake fjb-iptables FORWARD chain into cache cloud-init (FJB-98) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Cache nanodes provisioned by fj-bellows in cache-gateway mode now come up with the FJB-FORWARD chain installed at first boot, persisted via iptables-persistent. Previously the chain was only rendered into a log line (`wgboot: cache iptables re-rendered (TODO: push via SSH)`) and had to be hand-installed — the persistent test nanode for FJB-91 was the only place this happened, and a rebuild would lose the FORWARD policy entirely. What lands: - managedCache learns the worker VPC CIDR via setHardwareContext, then renders cachegateway.RenderCacheIPTables at cache-create time and embeds the script in cloud-init under /usr/local/sbin/fjb-iptables.sh. - cache-cloud-init.yaml.tmpl conditionally installs iptables-persistent, writes the script, runs it from runcmd, and saves via netfilter-persistent. ssh-mode renders byte-identical to before. - cachegateway Inputs.CacheVPCIP is now optional: the renderer never used it in the script body, and the bake-in happens before any VPC IP has been assigned. wgboot still passes it (eager Status lookup). ACL-derived per-prefix ACCEPT rules are intentionally omitted from the bake-in: the ACL registry isn't wired until wgboot starts (post-Configure), so the script that lands at create time is the skeleton (sysctl + ESTABLISHED,RELATED + FJB-92 reverse-direction + DROP). Runtime push of ACL-updated rules is the wgboot TODO already noted in code; it lands with FJB-99's ephemeral-cache bootstrap loop. Tests cover the cache-gateway and ssh-mode render branches and the renderIPTablesForCacheGateway transport-mode guard. Verified end-to-end against live Linode via test/e2e-linode/run-local.sh --transport=cache-gateway. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../linode/cache-cloud-init.yaml.tmpl | 13 +++ internal/provider/linode/cache.go | 70 ++++++++++-- internal/provider/linode/cache_test.go | 105 ++++++++++++++++-- internal/provider/linode/linode.go | 2 +- .../transport/cachegateway/cachegateway.go | 12 +- .../cachegateway/cachegateway_test.go | 1 - 6 files changed, 180 insertions(+), 23 deletions(-) diff --git a/internal/provider/linode/cache-cloud-init.yaml.tmpl b/internal/provider/linode/cache-cloud-init.yaml.tmpl index 50f1a8b..26c3b83 100644 --- a/internal/provider/linode/cache-cloud-init.yaml.tmpl +++ b/internal/provider/linode/cache-cloud-init.yaml.tmpl @@ -18,6 +18,9 @@ package_update: true packages: - curl - ca-certificates +{{- if .IPTablesScript}} + - iptables-persistent +{{- end}} write_files: - path: /etc/zot/config.json permissions: '0640' @@ -60,6 +63,12 @@ write_files: permissions: '0600' content: | {{.ServerKeyPEM | indent 6}} +{{- if .IPTablesScript}} + - path: /usr/local/sbin/fjb-iptables.sh + permissions: '0750' + content: | +{{.IPTablesScript | indent 6}} +{{- end}} - path: /etc/systemd/system/zot.service permissions: '0644' content: | @@ -100,6 +109,10 @@ runcmd: curl -fsSL -o /usr/local/bin/zot "$url" chmod +x /usr/local/bin/zot - mkdir -p /var/lib/zot +{{- if .IPTablesScript}} + - /usr/local/sbin/fjb-iptables.sh + - netfilter-persistent save +{{- end}} - systemctl daemon-reload - systemctl enable --now zot.service - touch {{.ReadyFile}} diff --git a/internal/provider/linode/cache.go b/internal/provider/linode/cache.go index 6921133..e273731 100644 --- a/internal/provider/linode/cache.go +++ b/internal/provider/linode/cache.go @@ -8,6 +8,7 @@ import ( "errors" "fmt" "log/slog" + "net/netip" "os" "path/filepath" "slices" @@ -15,6 +16,8 @@ import ( "text/template" "github.com/linode/linodego" + + "github.com/hstern/fj-bellows/internal/transport/cachegateway" ) // cacheClient is the slice of *linodego.Client the managed-cache code @@ -191,6 +194,12 @@ type managedCache struct { vpcSubnetID int authorizedKey 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 + // no VPC is configured — the iptables bake-in is then skipped. + workerVPCSubnet string + // linodeID is the cache VM's Linode ID. Populated by ensureAt- // Configure (find-or-adopt), cleared by maybeCleanupCache. linodeID int @@ -284,10 +293,11 @@ func newManagedCache(cfg cacheConfig, tag, region string, client cacheClient, bu // the constructor because the provider creates the managedCache before // the firewall + VPC helpers exist on l, and one-shot setters keep the // dependency direction one-way. -func (m *managedCache) setHardwareContext(firewallID, vpcSubnetID int, authorizedKey string) { +func (m *managedCache) setHardwareContext(firewallID, vpcSubnetID int, authorizedKey, workerVPCSubnet string) { m.firewallID = firewallID m.vpcSubnetID = vpcSubnetID m.authorizedKey = authorizedKey + m.workerVPCSubnet = workerVPCSubnet } // ensure brings the cache VM (and its bucket + scoped key) @@ -360,16 +370,21 @@ func (m *managedCache) createFreshCacheLinode(ctx context.Context, pair cacheCer if err != nil { return fmt.Errorf("bucket: %w", err) } + iptablesScript, err := m.renderIPTablesForCacheGateway() + if err != nil { + return fmt.Errorf("render fjb-iptables: %w", err) + } userData, err := renderCacheCloudInit(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), + 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, }) if err != nil { return fmt.Errorf("render cloud-init: %w", err) @@ -388,6 +403,34 @@ func (m *managedCache) createFreshCacheLinode(ctx context.Context, pair cacheCer return nil } +// renderIPTablesForCacheGateway returns the fjb-iptables.sh content +// that the cache cloud-init drops into /usr/local/sbin/ and runs at +// first boot (FJB-98). Empty when the deployment isn't cache-gateway +// mode or the VPC subnet isn't known — the template then skips the +// iptables-persistent install + script execution entirely. +// +// At cache-create time the ACL registry hasn't been wired yet (that +// happens after Configure, when wgboot starts and calls SetACLSource), +// so AllowedIPs is empty here. The skeleton (sysctl + chain + the +// FJB-92 orchestrator → worker reverse-direction rule + DROP) still +// installs, which is what's load-bearing for the orchestrator-side +// dispatcher. Per-prefix worker-egress ACCEPT rules are added at +// runtime by the wgboot push path (separate ticket; see FJB-98 +// "Out of scope"). +func (m *managedCache) renderIPTablesForCacheGateway() (string, error) { + if m.transportMode != "cache-gateway" || m.workerVPCSubnet == "" { + return "", nil + } + orchAddr, err := netip.ParseAddr(defaultOrchestratorWGAddr) + if err != nil { + return "", fmt.Errorf("parse orchestrator WG addr %q: %w", defaultOrchestratorWGAddr, err) + } + return cachegateway.RenderCacheIPTables(cachegateway.Inputs{ + WorkerVPCSubnet: m.workerVPCSubnet, + OrchestratorWGAddr: orchAddr, + }) +} + // buildCreateOpts assembles the InstanceCreateOptions payload for the // cache VM. Public stays primary so outbound (upstream sync, package // mirrors, GitHub-zot download) takes the default route; the VPC NIC @@ -572,6 +615,13 @@ type cacheCloudInitParams struct { // these via the CA distributed in the multipart worker cloud-init. ServerCertPEM string ServerKeyPEM string + + // IPTablesScript is the rendered fjb-iptables.sh content (FJB-98) + // embedded into the cloud-init under /usr/local/sbin/. Empty under + // ssh-mode deployments — the template skips both the script + // write_files entry and the iptables-persistent install when this + // field is empty. + IPTablesScript string } // renderCacheCloudInit fills the embedded template. Defaults to the diff --git a/internal/provider/linode/cache_test.go b/internal/provider/linode/cache_test.go index dcb14e3..4cbbdb2 100644 --- a/internal/provider/linode/cache_test.go +++ b/internal/provider/linode/cache_test.go @@ -176,7 +176,7 @@ func TestCacheEnsureAtConfigureCreatesFresh(t *testing.T) { fb := newFakeBucketClient() bucket := newManagedBucket("test-tag", testBucketRegion, "fjb-cache-test-tag", fb, slog.Default()) cache := newTestManagedCache(t, fc, bucket) - cache.setHardwareContext(7777, 8888, "") + cache.setHardwareContext(7777, 8888, "", "") if err := cache.ensureAtConfigure(ctx); err != nil { t.Fatalf("ensureAtConfigure: %v", err) @@ -454,6 +454,97 @@ func TestCacheCloudInitTemplateNotEmpty(t *testing.T) { } } +// FJB-98: when IPTablesScript is provided, the rendered cloud-init must +// install iptables-persistent, write the script into write_files, run +// it from runcmd, and persist via netfilter-persistent. When empty, +// none of those entries appear (so ssh-mode renders unchanged). +func TestRenderCacheCloudInitIPTablesBakeIn(t *testing.T) { + const fakeScript = "#!/usr/bin/env bash\n# fakeFJBiptables MARKER\niptables -N FJB-FORWARD\n" + withScript, err := renderCacheCloudInit(cacheCloudInitParams{ + Bucket: "b", + Region: "r", + Endpoint: testStubEndpoint, + AccessKey: "AK", + SecretKey: "SK", + ZotVersion: testStubZotVersion, + ServerCertPEM: testStubPEM, + ServerKeyPEM: testStubPEM, + IPTablesScript: fakeScript, + }) + if err != nil { + t.Fatalf("render with script: %v", err) + } + for _, want := range []string{ + "- iptables-persistent", + "/usr/local/sbin/fjb-iptables.sh", + "fakeFJBiptables MARKER", + "- netfilter-persistent save", + } { + if !strings.Contains(withScript, want) { + t.Errorf("cache-gateway render missing %q\n---\n%s", want, withScript) + } + } + + withoutScript, err := renderCacheCloudInit(cacheCloudInitParams{ + Bucket: "b", + Region: "r", + Endpoint: testStubEndpoint, + AccessKey: "AK", + SecretKey: "SK", + ZotVersion: testStubZotVersion, + ServerCertPEM: testStubPEM, + ServerKeyPEM: testStubPEM, + }) + if err != nil { + t.Fatalf("render without script: %v", err) + } + for _, unwanted := range []string{ + "iptables-persistent", + "fjb-iptables.sh", + "netfilter-persistent", + } { + if strings.Contains(withoutScript, unwanted) { + t.Errorf("ssh-mode render leaked iptables substring %q\n---\n%s", unwanted, withoutScript) + } + } +} + +// FJB-98: renderIPTablesForCacheGateway returns an empty script for +// non-cache-gateway transport or when the worker VPC subnet hasn't +// been wired yet — both branches are reachable in real deployments +// and must not error. +func TestRenderIPTablesForCacheGateway(t *testing.T) { + m := &managedCache{log: slog.Default()} + got, err := m.renderIPTablesForCacheGateway() + if err != nil || got != "" { + t.Errorf("empty cache: got %q err %v; want empty + nil", got, err) + } + + m.transportMode = "ssh" + m.workerVPCSubnet = "10.0.0.0/24" + got, err = m.renderIPTablesForCacheGateway() + if err != nil || got != "" { + t.Errorf("ssh mode: got %q err %v; want empty + nil", got, err) + } + + m.transportMode = "cache-gateway" + m.workerVPCSubnet = "10.0.0.0/24" + got, err = m.renderIPTablesForCacheGateway() + if err != nil { + t.Fatalf("cache-gateway: unexpected err: %v", err) + } + for _, want := range []string{ + "iptables -N FJB-FORWARD", + "iptables -A FJB-FORWARD -m state --state ESTABLISHED,RELATED -j ACCEPT", + "iptables -A FJB-FORWARD -s 100.64.0.1/32 -d 10.0.0.0/24 -i wg0 -o eth1 -j ACCEPT", + "iptables -A FJB-FORWARD -j DROP", + } { + if !strings.Contains(got, want) { + t.Errorf("cache-gateway render missing %q\n---\n%s", want, got) + } + } +} + // findCacheLinode must not adopt instances whose tag doesn't match — // otherwise a cache from a different deployment could be hijacked. func TestFindCacheLinodeIgnoresOtherDeployments(t *testing.T) { @@ -524,7 +615,7 @@ func TestCacheEnsureRecreatesAfterReap(t *testing.T) { fb := newFakeBucketClient() bucket := newManagedBucket("test-tag", testBucketRegion, "fjb-cache-test-tag", fb, slog.Default()) cache := newTestManagedCache(t, fc, bucket) - cache.setHardwareContext(7777, 8888, "") + cache.setHardwareContext(7777, 8888, "", "") if err := cache.ensureAtConfigure(ctx); err != nil { t.Fatalf("initial ensureAtConfigure: %v", err) } @@ -556,7 +647,7 @@ func TestCacheEnsureNoOpWhenIDStillValid(t *testing.T) { fb := newFakeBucketClient() bucket := newManagedBucket("test-tag", testBucketRegion, "fjb-cache-test-tag", fb, slog.Default()) cache := newTestManagedCache(t, fc, bucket) - cache.setHardwareContext(7777, 8888, "") + cache.setHardwareContext(7777, 8888, "", "") if err := cache.ensureAtConfigure(ctx); err != nil { t.Fatalf("ensureAtConfigure: %v", err) } @@ -605,7 +696,7 @@ func TestWorkerExtrasLooksUpAndCachesVPCIP(t *testing.T) { fb := newFakeBucketClient() bucket := newManagedBucket("test-tag", testBucketRegion, "fjb-cache-test-tag", fb, slog.Default()) cache := newTestManagedCache(t, fc, bucket) - cache.setHardwareContext(7777, 8888, "") + cache.setHardwareContext(7777, 8888, "", "") if err := cache.ensureAtConfigure(ctx); err != nil { t.Fatalf("ensureAtConfigure: %v", err) @@ -682,7 +773,7 @@ func TestWorkerExtrasPullsAllowedIPsFromACLSource(t *testing.T) { fb := newFakeBucketClient() bucket := newManagedBucket("test-tag", testBucketRegion, "fjb-cache-test-tag", fb, slog.Default()) cache := newTestManagedCache(t, fc, bucket) - cache.setHardwareContext(7777, 8888, "") + cache.setHardwareContext(7777, 8888, "", "") if err := cache.ensureAtConfigure(ctx); err != nil { t.Fatalf("ensureAtConfigure: %v", err) } @@ -732,7 +823,7 @@ func TestWorkerExtrasEmptyWhenNoACLSource(t *testing.T) { fb := newFakeBucketClient() bucket := newManagedBucket("test-tag", testBucketRegion, "fjb-cache-test-tag", fb, slog.Default()) cache := newTestManagedCache(t, fc, bucket) - cache.setHardwareContext(7777, 8888, "") + cache.setHardwareContext(7777, 8888, "", "") if err := cache.ensureAtConfigure(ctx); err != nil { t.Fatalf("ensureAtConfigure: %v", err) } @@ -765,7 +856,7 @@ func TestWorkerExtrasErrorsWhenVPCIPNotAssigned(t *testing.T) { fb := newFakeBucketClient() bucket := newManagedBucket("test-tag", testBucketRegion, "fjb-cache-test-tag", fb, slog.Default()) cache := newTestManagedCache(t, fc, bucket) - cache.setHardwareContext(7777, 8888, "") + cache.setHardwareContext(7777, 8888, "", "") if err := cache.ensureAtConfigure(ctx); err != nil { t.Fatalf("ensureAtConfigure: %v", err) diff --git a/internal/provider/linode/linode.go b/internal/provider/linode/linode.go index 646b0c5..9cbac2e 100644 --- a/internal/provider/linode/linode.go +++ b/internal/provider/linode/linode.go @@ -493,7 +493,7 @@ func (l *Linode) setupManagedCache(ctx context.Context, tag string) error { // operator debugging only (the dispatcher's tunnel is gone). The // key is supplied via SetSSHAuthorizedKey from cmd/fj-bellows; // empty when no SSH was configured (docker provider). - cache.setHardwareContext(fwID, subID, l.sshAuthorizedKey) + cache.setHardwareContext(fwID, subID, l.sshAuthorizedKey, l.WorkerVPCSubnet()) cache.setTransport(l.transportMode) // FJB-88: the worker cache-extras template now reads AllowedIPs // CIDRs from the ACL registry, not from operator-supplied tunnel diff --git a/internal/transport/cachegateway/cachegateway.go b/internal/transport/cachegateway/cachegateway.go index ba76b8c..02414d1 100644 --- a/internal/transport/cachegateway/cachegateway.go +++ b/internal/transport/cachegateway/cachegateway.go @@ -78,11 +78,15 @@ type Inputs struct { } // validate runs the basic sanity checks for the iptables renderer. +// +// CacheVPCIP is intentionally optional: the renderer doesn't emit it +// into the script body (it's reserved for future PostUp hooks). Cache +// cloud-init bake-in (FJB-98) needs to render the script at cache +// create time, before any VPC IP has been assigned — relaxing the +// requirement makes that flow possible without inventing a placeholder. +// When supplied it still has to be a valid IPv4/IPv6 literal. func (i Inputs) validate() error { - if i.CacheVPCIP == "" { - return errors.New("cachegateway: CacheVPCIP is required") - } - if net.ParseIP(i.CacheVPCIP) == nil { + if i.CacheVPCIP != "" && net.ParseIP(i.CacheVPCIP) == nil { return errors.New("cachegateway: CacheVPCIP is not a valid IP") } if i.WorkerVPCSubnet == "" { diff --git a/internal/transport/cachegateway/cachegateway_test.go b/internal/transport/cachegateway/cachegateway_test.go index ddb7fb8..adb04eb 100644 --- a/internal/transport/cachegateway/cachegateway_test.go +++ b/internal/transport/cachegateway/cachegateway_test.go @@ -31,7 +31,6 @@ func TestValidateRejectsMissingFields(t *testing.T) { mutate func(*Inputs) wantSub string }{ - {"no cache VPC IP", func(i *Inputs) { i.CacheVPCIP = "" }, "CacheVPCIP is required"}, {"bad cache VPC IP", func(i *Inputs) { i.CacheVPCIP = "not-an-ip" }, "not a valid IP"}, {"no worker VPC subnet", func(i *Inputs) { i.WorkerVPCSubnet = "" }, "WorkerVPCSubnet is required"}, {"bad worker VPC subnet", func(i *Inputs) { i.WorkerVPCSubnet = "not-a-cidr" }, "not a valid CIDR"}, From f955563197849f39497b9bec5c45ee970abced76 Mon Sep 17 00:00:00 2001 From: Henry Stern Date: Thu, 28 May 2026 14:20:37 -0300 Subject: [PATCH 2/2] =?UTF-8?q?feat(linode):=20cache=20cloud-init=20WG=20b?= =?UTF-8?q?ootstrap=20=E2=80=94=20Phase=20A=20(FJB-99)=20(#110)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat(agent): proto + ConnectRPC skeleton with Health RPC (FJB-94) (#107) * feat(agent): proto + ConnectRPC skeleton with Health RPC (FJB-94) First slice of the fjbagent foundation. Lands the proto package, the binary skeleton, and the Health RPC — no deploy story yet (cloud-init, systemd, cache base-image come in subsequent slices), no consumer code in the orchestrator yet (agent.Dialer is its own slice). - proto/fjbellows/agent/v1/agent.proto: AgentService.Health - cmd/fjbagent: static binary, ConnectRPC server, bearer auth, /healthz shim. -listen and -token-file flags (operator sets these from the systemd unit when deployment lands). - internal/agent: Handler captures build version + start time + hostname at construction; bearer middleware matches the internal/control auth shape (constant-time compare, /healthz exempt). - Unit tests: Health response shape, bearer accept/reject, /healthz open without token, graceful shutdown, LoadToken validation. Co-Authored-By: Claude Opus 4.7 (1M context) * fix(agent): split main into run() so defer cancel() runs on error (FJB-94) gocritic flagged exitAfterDefer: os.Exit bypasses deferred cancel(), so the signal-context goroutine leaks on the error path. Moving the work into run() that returns error lets the defer fire before main maps the error to an exit code. Co-Authored-By: Claude Opus 4.7 (1M context) --------- Co-authored-by: Claude Opus 4.7 (1M context) * feat(agent): Exec bidi RPC, pipe-mode handler (FJB-93 Phase A) (#108) First slice of FJB-93. Lands the AgentService.Exec bidi-streaming RPC on the agent and a pipe-mode handler (no PTY yet — that's Phase C). The orchestrator-side ShellSession proxy and the fjbctl ssh CLI come in Phase B; the wire envelope shape (ShellMsg / ShellEvent) is already designed to support both phases without a re-do. Wire shape: - ShellMsg: oneof Open | Stdin | Resize | Signal | CloseStdin - ShellEvent: oneof Opened | Stdout | Stderr | Exit - ExecOpen: target (set by fjbctl, stripped by orchestrator), argv, env, tty (rejected with CodeUnimplemented this phase), term, winsize - ExecExit: code on normal exit, signal name on signal kill Handler behavior: - First frame must be Open or returns CodeInvalidArgument. - tty=true returns CodeUnimplemented. - argv exec'd directly (no shell interpolation); env is exact (no agent-env passthrough, matches the design doc). - exec.Cmd.Stdout/Stderr go through a streamWriter that feeds a buffered sendBuf channel; a single sender goroutine owns the bidi stream's Send side so we never need to mutex Connect's BidiStream. - cmd.Wait blocks until both child exit AND exec's internal copy goroutines have drained, so all output frames precede Exit by construction (avoids the StdoutPipe-vs-Wait race). - Child runs in a fresh process group (Setpgid); signals are delivered to the whole group via kill(-pgid). - Command-not-found surfaces as on-stream Exit{127} (sh convention); signal kills surface as Exit{code: 128+signum, signal: }. Tests: 11 cases covering echo, stdin pump + CloseStdin, non-zero exit, stderr capture, env passthrough, tty/empty-argv rejection, command-not-found, signal kill, first-frame-must-be-Open, bearer auth gating. `go test -race -count=10` clean. Co-authored-by: Claude Opus 4.7 (1M context) * feat(linode): cache cloud-init WG bootstrap — Phase A (FJB-99) Foundation for the per-deployment ephemeral cache flow that FJB-91 called out as missing: each cache provisioned by fj-bellows now comes up with its own WG keypair, brings up wg-quick@wg0 referencing the orchestrator's pubkey, and publishes its own pubkey to S3 so the orchestrator can read it back in Phase B. What lands (Phase A): - main.go loads or generates the orchestrator's WG keypair BEFORE prov.Configure runs (was deferred until wgboot.Boot). Pubkey is pushed into the Linode provider via SetOrchestratorWGPubkey for the cache cloud-init injector. - managedCache carries orchestratorWGPubkey and renders four new fields into cacheCloudInitParams (OrchestratorWGPubkey, Orchestrator- WGAddr, CacheWGAddr, WGListenPort) when transport.mode is cache-gateway. - cache-cloud-init.yaml.tmpl conditionally installs wireguard + awscli, drops /usr/local/sbin/fjb-wg-bootstrap.sh, and runs it from runcmd. The script: - generates a Curve25519 keypair on first boot (idempotent — re- runs reuse the persisted private key); - writes /etc/wireguard/wg0.conf with the orchestrator pubkey baked in as [Peer].PublicKey and the cache's address pinned at 100.64.0.2/32; - brings up wg-quick@wg0; - publishes /etc/wireguard/publickey to s3:///wg-pubkey.txt using the same Object Storage credentials already shared with zot — no new identity to provision. - ssh-mode renders byte-identical to before (the WG bootstrap block is gated on OrchestratorWGPubkey). Out of Phase A (lands in Phase B/C): - Orchestrator-side WaitForWGPubkey polling against the bucket + feeding the discovered pubkey into wgboot.Boot as the peer config. - bootWGStack ordering reorder (today still wgboot before cache.Ensure; Phase B moves it after). - e2e harness reactivation: drop the persistent-cache preflight, use the ephemeral cache + bootstrap, and reactivate the worker readiness + job-complete checks. - Drop the static transport.wg.peer.{public_key,endpoint} config knobs. Tests cover the cache-gateway and ssh-mode render branches; verified end-to-end against live Linode via test/e2e-linode/run-local.sh --transport=cache-gateway (stack-up scope from FJB-91 still passes — the new code path is exercised but worker validation is the Phase C deliverable). Co-Authored-By: Claude Opus 4.7 (1M context) --------- Co-authored-by: Claude Opus 4.7 (1M context) --- cmd/fj-bellows/main.go | 33 + cmd/fjbagent/main.go | 87 ++ gen/fjbellows/agent/v1/agent.pb.go | 1072 +++++++++++++++++ .../agent/v1/agentv1connect/agent.connect.go | 166 +++ internal/agent/exec.go | 307 +++++ internal/agent/exec_test.go | 315 +++++ internal/agent/handler.go | 57 + internal/agent/handler_test.go | 54 + internal/agent/server.go | 158 +++ internal/agent/server_test.go | 205 ++++ .../linode/cache-cloud-init.yaml.tmpl | 48 + internal/provider/linode/cache.go | 49 +- internal/provider/linode/cache_test.go | 69 ++ internal/provider/linode/linode.go | 22 + proto/fjbellows/agent/v1/agent.proto | 182 +++ 15 files changed, 2822 insertions(+), 2 deletions(-) create mode 100644 cmd/fjbagent/main.go create mode 100644 gen/fjbellows/agent/v1/agent.pb.go create mode 100644 gen/fjbellows/agent/v1/agentv1connect/agent.connect.go create mode 100644 internal/agent/exec.go create mode 100644 internal/agent/exec_test.go create mode 100644 internal/agent/handler.go create mode 100644 internal/agent/handler_test.go create mode 100644 internal/agent/server.go create mode 100644 internal/agent/server_test.go create mode 100644 proto/fjbellows/agent/v1/agent.proto diff --git a/cmd/fj-bellows/main.go b/cmd/fj-bellows/main.go index eb8baf5..e400b99 100644 --- a/cmd/fj-bellows/main.go +++ b/cmd/fj-bellows/main.go @@ -28,6 +28,7 @@ import ( "github.com/hstern/fj-bellows/internal/forgejo" "github.com/hstern/fj-bellows/internal/orchestrator" "github.com/hstern/fj-bellows/internal/provider" + "github.com/hstern/fj-bellows/internal/transport/wg" "github.com/hstern/fj-bellows/internal/transport/wgboot" // Register in-tree providers. @@ -132,6 +133,15 @@ func run(opts runOpts, log *slog.Logger, logBus *logbus.Bus) error { // implement the method (e.g. the docker provider, which has no // Linode-style firewall) are unaffected. applyTransportToProvider(prov, cfg.Transport) + // Under cache-gateway mode, load (or generate) the orchestrator's WG + // keypair BEFORE Configure runs. Configure provisions the cache VM, + // and its cloud-init needs the orchestrator's public key baked in + // so wg-quick on the cache can come up referencing the right peer + // (FJB-99 Phase A). Phase B will move the wgboot.Boot call to AFTER + // Configure and have it reuse this same key. + if err := plumbOrchestratorWGPubkey(prov, cfg.Transport); err != nil { + return err + } // Bound the Configure-time network calls (provider sentinel fetches, // firewall API, etc.) so a hung upstream can't wedge startup forever. cfgCtx, cancelCfg := context.WithTimeout(context.Background(), 60*time.Second) @@ -602,6 +612,29 @@ func applyTransportToProvider(prov provider.Provider, t config.Transport) { } } +// plumbOrchestratorWGPubkey loads (or generates) the orchestrator's WG +// keypair under cache-gateway mode and pushes the public key into the +// provider so the cache cloud-init can bake it in as the WG peer +// pubkey (FJB-99 Phase A). Duck-typed for the same reason as +// applyTransportToProvider; providers that don't carry a managed cache +// skip the push. +// +// No-op for ssh-mode deployments (no WG block configured) and for +// providers without SetOrchestratorWGPubkey (e.g. docker). +func plumbOrchestratorWGPubkey(prov provider.Provider, t config.Transport) error { + if t.Mode != config.TransportCacheGateway || t.WG == nil { + return nil + } + priv, err := wg.LoadOrGenerateKey(t.WG.PrivateKeyFile) + if err != nil { + return fmt.Errorf("orchestrator wg key: %w", err) + } + if sp, ok := prov.(interface{ SetOrchestratorWGPubkey(string) }); ok { + sp.SetOrchestratorWGPubkey(priv.PublicKey().String()) + } + return nil +} + // sshDispatcherFrom builds the SSH dispatcher from config. func sshDispatcherFrom(cfg *config.Config, signer ssh.Signer) *orchestrator.SSHDispatcher { return &orchestrator.SSHDispatcher{ diff --git a/cmd/fjbagent/main.go b/cmd/fjbagent/main.go new file mode 100644 index 0000000..fab5d4b --- /dev/null +++ b/cmd/fjbagent/main.go @@ -0,0 +1,87 @@ +// Command fjbagent is the small daemon that runs on every fj-bellows worker +// and cache VM. It exposes ConnectRPC services the orchestrator dials into +// over the WG fabric — Health (FJB-94), and later Exec (FJB-93) and +// reachability probes (FJB-97). +// +// Listen address depends on the role: +// - cache: the cache's own WG inner address (e.g. 100.64.0.2:9001) +// - worker: the worker's VPC IP (e.g. 10.0.0.5:9001); the orchestrator +// reaches it through the cache-gateway routing FJB-54 set up +// +// Bind-address selection is the operator's job via -listen (typically set +// by the systemd unit from a cloud-init-substituted value). The agent +// itself doesn't care — it binds whatever it's told. +package main + +import ( + "context" + "flag" + "fmt" + "log/slog" + "os" + "os/signal" + "syscall" + "time" + + "github.com/hstern/fj-bellows/internal/agent" +) + +// version is the build version, stamped at link time via +// +// -ldflags "-X main.version=" +// +// Defaults to "dev" for local builds. +var version = "dev" + +func main() { + if err := run(); err != nil { + fmt.Fprintf(os.Stderr, "fjbagent: %v\n", err) + os.Exit(1) + } +} + +// run does the real work in a function with no os.Exit so deferred cleanup +// (the signal-context cancel) actually executes on error paths. main is a +// thin wrapper that translates the returned error into an exit code. +func run() error { + listen := flag.String("listen", "127.0.0.1:9001", "host:port to bind the agent's ConnectRPC server (typically the target's VPC IP or WG inner address)") + tokenFile := flag.String("token-file", "", "path to the bearer-token file (mode 0600); empty disables auth (test-only)") + logLevel := flag.String("log-level", "info", "slog level: debug|info|warn|error") + flag.Parse() + + log := slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{ + Level: parseLevel(*logLevel), + })) + + opts := []agent.Option{} + if *tokenFile != "" { + tok, err := agent.LoadToken(*tokenFile) + if err != nil { + return err + } + opts = append(opts, agent.WithBearerToken(tok)) + } else { + log.Warn("agent running without bearer-token auth; do not use in production") + } + + h := agent.NewHandler(version, time.Now()) + srv := agent.NewServer(*listen, h, log, opts...) + + ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) + defer cancel() + + return srv.Run(ctx) +} + +func parseLevel(s string) slog.Level { + switch s { + case "debug": + return slog.LevelDebug + case "warn": + return slog.LevelWarn + case "error": + return slog.LevelError + default: + return slog.LevelInfo + } +} diff --git a/gen/fjbellows/agent/v1/agent.pb.go b/gen/fjbellows/agent/v1/agent.pb.go new file mode 100644 index 0000000..bb44f06 --- /dev/null +++ b/gen/fjbellows/agent/v1/agent.pb.go @@ -0,0 +1,1072 @@ +// Code generated by protoc-gen-go. DO NOT EDIT. +// versions: +// protoc-gen-go v1.36.11 +// protoc (unknown) +// source: fjbellows/agent/v1/agent.proto + +package agentv1 + +import ( + protoreflect "google.golang.org/protobuf/reflect/protoreflect" + protoimpl "google.golang.org/protobuf/runtime/protoimpl" + timestamppb "google.golang.org/protobuf/types/known/timestamppb" + reflect "reflect" + sync "sync" + unsafe "unsafe" +) + +const ( + // Verify that this generated code is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion) + // Verify that runtime/protoimpl is sufficiently up-to-date. + _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) +) + +type HealthRequest struct { + state protoimpl.MessageState `protogen:"open.v1"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *HealthRequest) Reset() { + *x = HealthRequest{} + mi := &file_fjbellows_agent_v1_agent_proto_msgTypes[0] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *HealthRequest) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*HealthRequest) ProtoMessage() {} + +func (x *HealthRequest) ProtoReflect() protoreflect.Message { + mi := &file_fjbellows_agent_v1_agent_proto_msgTypes[0] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use HealthRequest.ProtoReflect.Descriptor instead. +func (*HealthRequest) Descriptor() ([]byte, []int) { + return file_fjbellows_agent_v1_agent_proto_rawDescGZIP(), []int{0} +} + +type HealthResponse struct { + state protoimpl.MessageState `protogen:"open.v1"` + // ready is true on every successful invocation; the field is present so + // operator tooling can do `.ready` instead of `error == nil`. It exists + // for future-proofing too — a future agent may want to return ready=false + // while still being reachable (e.g. during a quiesce window). + Ready bool `protobuf:"varint,1,opt,name=ready,proto3" json:"ready,omitempty"` + // build_version is the agent binary's version string. Populated from + // -ldflags "-X main.version=…" at link time; "dev" when unset. + BuildVersion string `protobuf:"bytes,2,opt,name=build_version,json=buildVersion,proto3" json:"build_version,omitempty"` + // started_at is when the agent process started. Subtracting this from + // now gives the operator-visible uptime. + StartedAt *timestamppb.Timestamp `protobuf:"bytes,3,opt,name=started_at,json=startedAt,proto3" json:"started_at,omitempty"` + // hostname is the OS hostname as reported by os.Hostname() at startup. + // Useful when the orchestrator's name → addr map drifts from what the + // VM actually thinks it is. + Hostname string `protobuf:"bytes,4,opt,name=hostname,proto3" json:"hostname,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *HealthResponse) Reset() { + *x = HealthResponse{} + mi := &file_fjbellows_agent_v1_agent_proto_msgTypes[1] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *HealthResponse) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*HealthResponse) ProtoMessage() {} + +func (x *HealthResponse) ProtoReflect() protoreflect.Message { + mi := &file_fjbellows_agent_v1_agent_proto_msgTypes[1] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use HealthResponse.ProtoReflect.Descriptor instead. +func (*HealthResponse) Descriptor() ([]byte, []int) { + return file_fjbellows_agent_v1_agent_proto_rawDescGZIP(), []int{1} +} + +func (x *HealthResponse) GetReady() bool { + if x != nil { + return x.Ready + } + return false +} + +func (x *HealthResponse) GetBuildVersion() string { + if x != nil { + return x.BuildVersion + } + return "" +} + +func (x *HealthResponse) GetStartedAt() *timestamppb.Timestamp { + if x != nil { + return x.StartedAt + } + return nil +} + +func (x *HealthResponse) GetHostname() string { + if x != nil { + return x.Hostname + } + return "" +} + +// ShellMsg is the client→server frame on the Exec bidi stream. Exactly one +// ExecOpen must be the first message; subsequent messages may interleave +// Stdin / Resize / Signal / CloseStdin in any order until the server closes +// the stream after sending Exit. +type ShellMsg struct { + state protoimpl.MessageState `protogen:"open.v1"` + // Types that are valid to be assigned to Kind: + // + // *ShellMsg_Open + // *ShellMsg_Stdin + // *ShellMsg_Resize + // *ShellMsg_Signal + // *ShellMsg_CloseStdin + Kind isShellMsg_Kind `protobuf_oneof:"kind"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *ShellMsg) Reset() { + *x = ShellMsg{} + mi := &file_fjbellows_agent_v1_agent_proto_msgTypes[2] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *ShellMsg) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*ShellMsg) ProtoMessage() {} + +func (x *ShellMsg) ProtoReflect() protoreflect.Message { + mi := &file_fjbellows_agent_v1_agent_proto_msgTypes[2] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use ShellMsg.ProtoReflect.Descriptor instead. +func (*ShellMsg) Descriptor() ([]byte, []int) { + return file_fjbellows_agent_v1_agent_proto_rawDescGZIP(), []int{2} +} + +func (x *ShellMsg) GetKind() isShellMsg_Kind { + if x != nil { + return x.Kind + } + return nil +} + +func (x *ShellMsg) GetOpen() *ExecOpen { + if x != nil { + if x, ok := x.Kind.(*ShellMsg_Open); ok { + return x.Open + } + } + return nil +} + +func (x *ShellMsg) GetStdin() *ExecStdin { + if x != nil { + if x, ok := x.Kind.(*ShellMsg_Stdin); ok { + return x.Stdin + } + } + return nil +} + +func (x *ShellMsg) GetResize() *ExecResize { + if x != nil { + if x, ok := x.Kind.(*ShellMsg_Resize); ok { + return x.Resize + } + } + return nil +} + +func (x *ShellMsg) GetSignal() *ExecSignal { + if x != nil { + if x, ok := x.Kind.(*ShellMsg_Signal); ok { + return x.Signal + } + } + return nil +} + +func (x *ShellMsg) GetCloseStdin() *ExecCloseStdin { + if x != nil { + if x, ok := x.Kind.(*ShellMsg_CloseStdin); ok { + return x.CloseStdin + } + } + return nil +} + +type isShellMsg_Kind interface { + isShellMsg_Kind() +} + +type ShellMsg_Open struct { + Open *ExecOpen `protobuf:"bytes,1,opt,name=open,proto3,oneof"` +} + +type ShellMsg_Stdin struct { + Stdin *ExecStdin `protobuf:"bytes,2,opt,name=stdin,proto3,oneof"` +} + +type ShellMsg_Resize struct { + Resize *ExecResize `protobuf:"bytes,3,opt,name=resize,proto3,oneof"` +} + +type ShellMsg_Signal struct { + Signal *ExecSignal `protobuf:"bytes,4,opt,name=signal,proto3,oneof"` +} + +type ShellMsg_CloseStdin struct { + CloseStdin *ExecCloseStdin `protobuf:"bytes,5,opt,name=close_stdin,json=closeStdin,proto3,oneof"` +} + +func (*ShellMsg_Open) isShellMsg_Kind() {} + +func (*ShellMsg_Stdin) isShellMsg_Kind() {} + +func (*ShellMsg_Resize) isShellMsg_Kind() {} + +func (*ShellMsg_Signal) isShellMsg_Kind() {} + +func (*ShellMsg_CloseStdin) isShellMsg_Kind() {} + +// ShellEvent is the server→client frame. Opened arrives first (sentinel, +// matches the StreamEvents convention so the open call returns +// immediately even before fork+exec lands). Stdout and Stderr interleave +// freely. Exit is always last and ends the stream. +type ShellEvent struct { + state protoimpl.MessageState `protogen:"open.v1"` + // Types that are valid to be assigned to Kind: + // + // *ShellEvent_Opened + // *ShellEvent_Stdout + // *ShellEvent_Stderr + // *ShellEvent_Exit + Kind isShellEvent_Kind `protobuf_oneof:"kind"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *ShellEvent) Reset() { + *x = ShellEvent{} + mi := &file_fjbellows_agent_v1_agent_proto_msgTypes[3] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *ShellEvent) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*ShellEvent) ProtoMessage() {} + +func (x *ShellEvent) ProtoReflect() protoreflect.Message { + mi := &file_fjbellows_agent_v1_agent_proto_msgTypes[3] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use ShellEvent.ProtoReflect.Descriptor instead. +func (*ShellEvent) Descriptor() ([]byte, []int) { + return file_fjbellows_agent_v1_agent_proto_rawDescGZIP(), []int{3} +} + +func (x *ShellEvent) GetKind() isShellEvent_Kind { + if x != nil { + return x.Kind + } + return nil +} + +func (x *ShellEvent) GetOpened() *ExecOpened { + if x != nil { + if x, ok := x.Kind.(*ShellEvent_Opened); ok { + return x.Opened + } + } + return nil +} + +func (x *ShellEvent) GetStdout() *ExecStdout { + if x != nil { + if x, ok := x.Kind.(*ShellEvent_Stdout); ok { + return x.Stdout + } + } + return nil +} + +func (x *ShellEvent) GetStderr() *ExecStderr { + if x != nil { + if x, ok := x.Kind.(*ShellEvent_Stderr); ok { + return x.Stderr + } + } + return nil +} + +func (x *ShellEvent) GetExit() *ExecExit { + if x != nil { + if x, ok := x.Kind.(*ShellEvent_Exit); ok { + return x.Exit + } + } + return nil +} + +type isShellEvent_Kind interface { + isShellEvent_Kind() +} + +type ShellEvent_Opened struct { + Opened *ExecOpened `protobuf:"bytes,1,opt,name=opened,proto3,oneof"` +} + +type ShellEvent_Stdout struct { + Stdout *ExecStdout `protobuf:"bytes,2,opt,name=stdout,proto3,oneof"` +} + +type ShellEvent_Stderr struct { + Stderr *ExecStderr `protobuf:"bytes,3,opt,name=stderr,proto3,oneof"` +} + +type ShellEvent_Exit struct { + Exit *ExecExit `protobuf:"bytes,4,opt,name=exit,proto3,oneof"` +} + +func (*ShellEvent_Opened) isShellEvent_Kind() {} + +func (*ShellEvent_Stdout) isShellEvent_Kind() {} + +func (*ShellEvent_Stderr) isShellEvent_Kind() {} + +func (*ShellEvent_Exit) isShellEvent_Kind() {} + +// ExecOpen is the first ShellMsg on every Exec stream. argv[0] is the +// program to run; an empty argv asks for a login shell (deferred to a +// later phase). env passes additional environment vars beyond the agent +// process's own; $TERM is set from `term` when `tty` is true. +type ExecOpen struct { + state protoimpl.MessageState `protogen:"open.v1"` + // target identifies which worker/cache to run on. This field is only + // meaningful on the fjbctl→orchestrator hop; the orchestrator strips + // it before forwarding the Open to the agent. Empty on the agent hop. + Target string `protobuf:"bytes,1,opt,name=target,proto3" json:"target,omitempty"` + // argv is the program + args. argv[0] is exec'd directly; no shell + // interpolation. Pass `sh -c "..."` for shell semantics. + Argv []string `protobuf:"bytes,2,rep,name=argv,proto3" json:"argv,omitempty"` + // env is a map of additional environment variables. Empty by default; + // the agent does NOT pass the orchestrator's environment through. + Env map[string]string `protobuf:"bytes,3,rep,name=env,proto3" json:"env,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"` + // tty asks the agent to allocate a PTY for the child and wire stdin/ + // stdout (no separate stderr under PTY) to the PTY master. Phase A + // returns CodeUnimplemented if true; PTY support lands in Phase C. + Tty bool `protobuf:"varint,4,opt,name=tty,proto3" json:"tty,omitempty"` + // term is the $TERM value to set on the child when tty=true. Ignored + // when tty=false. + Term string `protobuf:"bytes,5,opt,name=term,proto3" json:"term,omitempty"` + // winsize is the initial PTY window size when tty=true. Ignored + // otherwise. + Winsize *Winsize `protobuf:"bytes,6,opt,name=winsize,proto3" json:"winsize,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *ExecOpen) Reset() { + *x = ExecOpen{} + mi := &file_fjbellows_agent_v1_agent_proto_msgTypes[4] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *ExecOpen) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*ExecOpen) ProtoMessage() {} + +func (x *ExecOpen) ProtoReflect() protoreflect.Message { + mi := &file_fjbellows_agent_v1_agent_proto_msgTypes[4] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use ExecOpen.ProtoReflect.Descriptor instead. +func (*ExecOpen) Descriptor() ([]byte, []int) { + return file_fjbellows_agent_v1_agent_proto_rawDescGZIP(), []int{4} +} + +func (x *ExecOpen) GetTarget() string { + if x != nil { + return x.Target + } + return "" +} + +func (x *ExecOpen) GetArgv() []string { + if x != nil { + return x.Argv + } + return nil +} + +func (x *ExecOpen) GetEnv() map[string]string { + if x != nil { + return x.Env + } + return nil +} + +func (x *ExecOpen) GetTty() bool { + if x != nil { + return x.Tty + } + return false +} + +func (x *ExecOpen) GetTerm() string { + if x != nil { + return x.Term + } + return "" +} + +func (x *ExecOpen) GetWinsize() *Winsize { + if x != nil { + return x.Winsize + } + return nil +} + +// Winsize carries terminal columns and rows. +type Winsize struct { + state protoimpl.MessageState `protogen:"open.v1"` + Cols uint32 `protobuf:"varint,1,opt,name=cols,proto3" json:"cols,omitempty"` + Rows uint32 `protobuf:"varint,2,opt,name=rows,proto3" json:"rows,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *Winsize) Reset() { + *x = Winsize{} + mi := &file_fjbellows_agent_v1_agent_proto_msgTypes[5] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *Winsize) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*Winsize) ProtoMessage() {} + +func (x *Winsize) ProtoReflect() protoreflect.Message { + mi := &file_fjbellows_agent_v1_agent_proto_msgTypes[5] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use Winsize.ProtoReflect.Descriptor instead. +func (*Winsize) Descriptor() ([]byte, []int) { + return file_fjbellows_agent_v1_agent_proto_rawDescGZIP(), []int{5} +} + +func (x *Winsize) GetCols() uint32 { + if x != nil { + return x.Cols + } + return 0 +} + +func (x *Winsize) GetRows() uint32 { + if x != nil { + return x.Rows + } + return 0 +} + +// ExecStdin pumps bytes to the child's stdin. Streaming; the client may +// send any number of these. CloseStdin signals EOF. +type ExecStdin struct { + state protoimpl.MessageState `protogen:"open.v1"` + Data []byte `protobuf:"bytes,1,opt,name=data,proto3" json:"data,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *ExecStdin) Reset() { + *x = ExecStdin{} + mi := &file_fjbellows_agent_v1_agent_proto_msgTypes[6] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *ExecStdin) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*ExecStdin) ProtoMessage() {} + +func (x *ExecStdin) ProtoReflect() protoreflect.Message { + mi := &file_fjbellows_agent_v1_agent_proto_msgTypes[6] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use ExecStdin.ProtoReflect.Descriptor instead. +func (*ExecStdin) Descriptor() ([]byte, []int) { + return file_fjbellows_agent_v1_agent_proto_rawDescGZIP(), []int{6} +} + +func (x *ExecStdin) GetData() []byte { + if x != nil { + return x.Data + } + return nil +} + +// ExecStdout / ExecStderr carry bytes from the child's stdout / stderr. +// Frame size is at the server's discretion (default 32 KiB chunks); the +// receiver should not assume line boundaries. +type ExecStdout struct { + state protoimpl.MessageState `protogen:"open.v1"` + Data []byte `protobuf:"bytes,1,opt,name=data,proto3" json:"data,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *ExecStdout) Reset() { + *x = ExecStdout{} + mi := &file_fjbellows_agent_v1_agent_proto_msgTypes[7] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *ExecStdout) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*ExecStdout) ProtoMessage() {} + +func (x *ExecStdout) ProtoReflect() protoreflect.Message { + mi := &file_fjbellows_agent_v1_agent_proto_msgTypes[7] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use ExecStdout.ProtoReflect.Descriptor instead. +func (*ExecStdout) Descriptor() ([]byte, []int) { + return file_fjbellows_agent_v1_agent_proto_rawDescGZIP(), []int{7} +} + +func (x *ExecStdout) GetData() []byte { + if x != nil { + return x.Data + } + return nil +} + +type ExecStderr struct { + state protoimpl.MessageState `protogen:"open.v1"` + Data []byte `protobuf:"bytes,1,opt,name=data,proto3" json:"data,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *ExecStderr) Reset() { + *x = ExecStderr{} + mi := &file_fjbellows_agent_v1_agent_proto_msgTypes[8] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *ExecStderr) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*ExecStderr) ProtoMessage() {} + +func (x *ExecStderr) ProtoReflect() protoreflect.Message { + mi := &file_fjbellows_agent_v1_agent_proto_msgTypes[8] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use ExecStderr.ProtoReflect.Descriptor instead. +func (*ExecStderr) Descriptor() ([]byte, []int) { + return file_fjbellows_agent_v1_agent_proto_rawDescGZIP(), []int{8} +} + +func (x *ExecStderr) GetData() []byte { + if x != nil { + return x.Data + } + return nil +} + +// ExecResize updates the PTY window size mid-session. Only meaningful +// when the Open had tty=true. +type ExecResize struct { + state protoimpl.MessageState `protogen:"open.v1"` + Winsize *Winsize `protobuf:"bytes,1,opt,name=winsize,proto3" json:"winsize,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *ExecResize) Reset() { + *x = ExecResize{} + mi := &file_fjbellows_agent_v1_agent_proto_msgTypes[9] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *ExecResize) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*ExecResize) ProtoMessage() {} + +func (x *ExecResize) ProtoReflect() protoreflect.Message { + mi := &file_fjbellows_agent_v1_agent_proto_msgTypes[9] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use ExecResize.ProtoReflect.Descriptor instead. +func (*ExecResize) Descriptor() ([]byte, []int) { + return file_fjbellows_agent_v1_agent_proto_rawDescGZIP(), []int{9} +} + +func (x *ExecResize) GetWinsize() *Winsize { + if x != nil { + return x.Winsize + } + return nil +} + +// ExecSignal forwards a signal to the child's process group. Valid +// names: SIGINT, SIGTERM, SIGQUIT, SIGHUP. Unknown names are ignored +// (logged on the agent side). +type ExecSignal struct { + state protoimpl.MessageState `protogen:"open.v1"` + Name string `protobuf:"bytes,1,opt,name=name,proto3" json:"name,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *ExecSignal) Reset() { + *x = ExecSignal{} + mi := &file_fjbellows_agent_v1_agent_proto_msgTypes[10] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *ExecSignal) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*ExecSignal) ProtoMessage() {} + +func (x *ExecSignal) ProtoReflect() protoreflect.Message { + mi := &file_fjbellows_agent_v1_agent_proto_msgTypes[10] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use ExecSignal.ProtoReflect.Descriptor instead. +func (*ExecSignal) Descriptor() ([]byte, []int) { + return file_fjbellows_agent_v1_agent_proto_rawDescGZIP(), []int{10} +} + +func (x *ExecSignal) GetName() string { + if x != nil { + return x.Name + } + return "" +} + +// ExecCloseStdin closes the child's stdin (EOF on the read side). The +// stream stays open for stdout/stderr and the terminal Exit. +type ExecCloseStdin struct { + state protoimpl.MessageState `protogen:"open.v1"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *ExecCloseStdin) Reset() { + *x = ExecCloseStdin{} + mi := &file_fjbellows_agent_v1_agent_proto_msgTypes[11] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *ExecCloseStdin) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*ExecCloseStdin) ProtoMessage() {} + +func (x *ExecCloseStdin) ProtoReflect() protoreflect.Message { + mi := &file_fjbellows_agent_v1_agent_proto_msgTypes[11] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use ExecCloseStdin.ProtoReflect.Descriptor instead. +func (*ExecCloseStdin) Descriptor() ([]byte, []int) { + return file_fjbellows_agent_v1_agent_proto_rawDescGZIP(), []int{11} +} + +// ExecOpened is the sentinel server→client frame that fires immediately +// after the agent accepts the Open and begins setting up the child. The +// existence of this frame is what lets the client's stream-open call +// return without waiting for fork+exec to complete. +type ExecOpened struct { + state protoimpl.MessageState `protogen:"open.v1"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *ExecOpened) Reset() { + *x = ExecOpened{} + mi := &file_fjbellows_agent_v1_agent_proto_msgTypes[12] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *ExecOpened) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*ExecOpened) ProtoMessage() {} + +func (x *ExecOpened) ProtoReflect() protoreflect.Message { + mi := &file_fjbellows_agent_v1_agent_proto_msgTypes[12] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use ExecOpened.ProtoReflect.Descriptor instead. +func (*ExecOpened) Descriptor() ([]byte, []int) { + return file_fjbellows_agent_v1_agent_proto_rawDescGZIP(), []int{12} +} + +// ExecExit is the terminal server→client frame. Exactly one of +// (code, signal) is meaningful: code is the program's exit status if it +// exited normally; signal is the signal name (e.g. "terminated", +// "killed") if it was killed by a signal. The agent closes the stream +// immediately after sending this. +type ExecExit struct { + state protoimpl.MessageState `protogen:"open.v1"` + // code is the child's exit status when it exited normally. + Code int32 `protobuf:"varint,1,opt,name=code,proto3" json:"code,omitempty"` + // signal is the signal name when the child was killed by a signal. + // Empty when code is meaningful. + Signal string `protobuf:"bytes,2,opt,name=signal,proto3" json:"signal,omitempty"` + unknownFields protoimpl.UnknownFields + sizeCache protoimpl.SizeCache +} + +func (x *ExecExit) Reset() { + *x = ExecExit{} + mi := &file_fjbellows_agent_v1_agent_proto_msgTypes[13] + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + ms.StoreMessageInfo(mi) +} + +func (x *ExecExit) String() string { + return protoimpl.X.MessageStringOf(x) +} + +func (*ExecExit) ProtoMessage() {} + +func (x *ExecExit) ProtoReflect() protoreflect.Message { + mi := &file_fjbellows_agent_v1_agent_proto_msgTypes[13] + if x != nil { + ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) + if ms.LoadMessageInfo() == nil { + ms.StoreMessageInfo(mi) + } + return ms + } + return mi.MessageOf(x) +} + +// Deprecated: Use ExecExit.ProtoReflect.Descriptor instead. +func (*ExecExit) Descriptor() ([]byte, []int) { + return file_fjbellows_agent_v1_agent_proto_rawDescGZIP(), []int{13} +} + +func (x *ExecExit) GetCode() int32 { + if x != nil { + return x.Code + } + return 0 +} + +func (x *ExecExit) GetSignal() string { + if x != nil { + return x.Signal + } + return "" +} + +var File_fjbellows_agent_v1_agent_proto protoreflect.FileDescriptor + +const file_fjbellows_agent_v1_agent_proto_rawDesc = "" + + "\n" + + "\x1efjbellows/agent/v1/agent.proto\x12\x12fjbellows.agent.v1\x1a\x1fgoogle/protobuf/timestamp.proto\"\x0f\n" + + "\rHealthRequest\"\xa2\x01\n" + + "\x0eHealthResponse\x12\x14\n" + + "\x05ready\x18\x01 \x01(\bR\x05ready\x12#\n" + + "\rbuild_version\x18\x02 \x01(\tR\fbuildVersion\x129\n" + + "\n" + + "started_at\x18\x03 \x01(\v2\x1a.google.protobuf.TimestampR\tstartedAt\x12\x1a\n" + + "\bhostname\x18\x04 \x01(\tR\bhostname\"\xb8\x02\n" + + "\bShellMsg\x122\n" + + "\x04open\x18\x01 \x01(\v2\x1c.fjbellows.agent.v1.ExecOpenH\x00R\x04open\x125\n" + + "\x05stdin\x18\x02 \x01(\v2\x1d.fjbellows.agent.v1.ExecStdinH\x00R\x05stdin\x128\n" + + "\x06resize\x18\x03 \x01(\v2\x1e.fjbellows.agent.v1.ExecResizeH\x00R\x06resize\x128\n" + + "\x06signal\x18\x04 \x01(\v2\x1e.fjbellows.agent.v1.ExecSignalH\x00R\x06signal\x12E\n" + + "\vclose_stdin\x18\x05 \x01(\v2\".fjbellows.agent.v1.ExecCloseStdinH\x00R\n" + + "closeStdinB\x06\n" + + "\x04kind\"\xf6\x01\n" + + "\n" + + "ShellEvent\x128\n" + + "\x06opened\x18\x01 \x01(\v2\x1e.fjbellows.agent.v1.ExecOpenedH\x00R\x06opened\x128\n" + + "\x06stdout\x18\x02 \x01(\v2\x1e.fjbellows.agent.v1.ExecStdoutH\x00R\x06stdout\x128\n" + + "\x06stderr\x18\x03 \x01(\v2\x1e.fjbellows.agent.v1.ExecStderrH\x00R\x06stderr\x122\n" + + "\x04exit\x18\x04 \x01(\v2\x1c.fjbellows.agent.v1.ExecExitH\x00R\x04exitB\x06\n" + + "\x04kind\"\x84\x02\n" + + "\bExecOpen\x12\x16\n" + + "\x06target\x18\x01 \x01(\tR\x06target\x12\x12\n" + + "\x04argv\x18\x02 \x03(\tR\x04argv\x127\n" + + "\x03env\x18\x03 \x03(\v2%.fjbellows.agent.v1.ExecOpen.EnvEntryR\x03env\x12\x10\n" + + "\x03tty\x18\x04 \x01(\bR\x03tty\x12\x12\n" + + "\x04term\x18\x05 \x01(\tR\x04term\x125\n" + + "\awinsize\x18\x06 \x01(\v2\x1b.fjbellows.agent.v1.WinsizeR\awinsize\x1a6\n" + + "\bEnvEntry\x12\x10\n" + + "\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" + + "\x05value\x18\x02 \x01(\tR\x05value:\x028\x01\"1\n" + + "\aWinsize\x12\x12\n" + + "\x04cols\x18\x01 \x01(\rR\x04cols\x12\x12\n" + + "\x04rows\x18\x02 \x01(\rR\x04rows\"\x1f\n" + + "\tExecStdin\x12\x12\n" + + "\x04data\x18\x01 \x01(\fR\x04data\" \n" + + "\n" + + "ExecStdout\x12\x12\n" + + "\x04data\x18\x01 \x01(\fR\x04data\" \n" + + "\n" + + "ExecStderr\x12\x12\n" + + "\x04data\x18\x01 \x01(\fR\x04data\"C\n" + + "\n" + + "ExecResize\x125\n" + + "\awinsize\x18\x01 \x01(\v2\x1b.fjbellows.agent.v1.WinsizeR\awinsize\" \n" + + "\n" + + "ExecSignal\x12\x12\n" + + "\x04name\x18\x01 \x01(\tR\x04name\"\x10\n" + + "\x0eExecCloseStdin\"\f\n" + + "\n" + + "ExecOpened\"6\n" + + "\bExecExit\x12\x12\n" + + "\x04code\x18\x01 \x01(\x05R\x04code\x12\x16\n" + + "\x06signal\x18\x02 \x01(\tR\x06signal2\xa9\x01\n" + + "\fAgentService\x12O\n" + + "\x06Health\x12!.fjbellows.agent.v1.HealthRequest\x1a\".fjbellows.agent.v1.HealthResponse\x12H\n" + + "\x04Exec\x12\x1c.fjbellows.agent.v1.ShellMsg\x1a\x1e.fjbellows.agent.v1.ShellEvent(\x010\x01B\xcb\x01\n" + + "\x16com.fjbellows.agent.v1B\n" + + "AgentProtoP\x01Z;github.com/hstern/fj-bellows/gen/fjbellows/agent/v1;agentv1\xa2\x02\x03FAX\xaa\x02\x12Fjbellows.Agent.V1\xca\x02\x12Fjbellows\\Agent\\V1\xe2\x02\x1eFjbellows\\Agent\\V1\\GPBMetadata\xea\x02\x14Fjbellows::Agent::V1b\x06proto3" + +var ( + file_fjbellows_agent_v1_agent_proto_rawDescOnce sync.Once + file_fjbellows_agent_v1_agent_proto_rawDescData []byte +) + +func file_fjbellows_agent_v1_agent_proto_rawDescGZIP() []byte { + file_fjbellows_agent_v1_agent_proto_rawDescOnce.Do(func() { + file_fjbellows_agent_v1_agent_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_fjbellows_agent_v1_agent_proto_rawDesc), len(file_fjbellows_agent_v1_agent_proto_rawDesc))) + }) + return file_fjbellows_agent_v1_agent_proto_rawDescData +} + +var file_fjbellows_agent_v1_agent_proto_msgTypes = make([]protoimpl.MessageInfo, 15) +var file_fjbellows_agent_v1_agent_proto_goTypes = []any{ + (*HealthRequest)(nil), // 0: fjbellows.agent.v1.HealthRequest + (*HealthResponse)(nil), // 1: fjbellows.agent.v1.HealthResponse + (*ShellMsg)(nil), // 2: fjbellows.agent.v1.ShellMsg + (*ShellEvent)(nil), // 3: fjbellows.agent.v1.ShellEvent + (*ExecOpen)(nil), // 4: fjbellows.agent.v1.ExecOpen + (*Winsize)(nil), // 5: fjbellows.agent.v1.Winsize + (*ExecStdin)(nil), // 6: fjbellows.agent.v1.ExecStdin + (*ExecStdout)(nil), // 7: fjbellows.agent.v1.ExecStdout + (*ExecStderr)(nil), // 8: fjbellows.agent.v1.ExecStderr + (*ExecResize)(nil), // 9: fjbellows.agent.v1.ExecResize + (*ExecSignal)(nil), // 10: fjbellows.agent.v1.ExecSignal + (*ExecCloseStdin)(nil), // 11: fjbellows.agent.v1.ExecCloseStdin + (*ExecOpened)(nil), // 12: fjbellows.agent.v1.ExecOpened + (*ExecExit)(nil), // 13: fjbellows.agent.v1.ExecExit + nil, // 14: fjbellows.agent.v1.ExecOpen.EnvEntry + (*timestamppb.Timestamp)(nil), // 15: google.protobuf.Timestamp +} +var file_fjbellows_agent_v1_agent_proto_depIdxs = []int32{ + 15, // 0: fjbellows.agent.v1.HealthResponse.started_at:type_name -> google.protobuf.Timestamp + 4, // 1: fjbellows.agent.v1.ShellMsg.open:type_name -> fjbellows.agent.v1.ExecOpen + 6, // 2: fjbellows.agent.v1.ShellMsg.stdin:type_name -> fjbellows.agent.v1.ExecStdin + 9, // 3: fjbellows.agent.v1.ShellMsg.resize:type_name -> fjbellows.agent.v1.ExecResize + 10, // 4: fjbellows.agent.v1.ShellMsg.signal:type_name -> fjbellows.agent.v1.ExecSignal + 11, // 5: fjbellows.agent.v1.ShellMsg.close_stdin:type_name -> fjbellows.agent.v1.ExecCloseStdin + 12, // 6: fjbellows.agent.v1.ShellEvent.opened:type_name -> fjbellows.agent.v1.ExecOpened + 7, // 7: fjbellows.agent.v1.ShellEvent.stdout:type_name -> fjbellows.agent.v1.ExecStdout + 8, // 8: fjbellows.agent.v1.ShellEvent.stderr:type_name -> fjbellows.agent.v1.ExecStderr + 13, // 9: fjbellows.agent.v1.ShellEvent.exit:type_name -> fjbellows.agent.v1.ExecExit + 14, // 10: fjbellows.agent.v1.ExecOpen.env:type_name -> fjbellows.agent.v1.ExecOpen.EnvEntry + 5, // 11: fjbellows.agent.v1.ExecOpen.winsize:type_name -> fjbellows.agent.v1.Winsize + 5, // 12: fjbellows.agent.v1.ExecResize.winsize:type_name -> fjbellows.agent.v1.Winsize + 0, // 13: fjbellows.agent.v1.AgentService.Health:input_type -> fjbellows.agent.v1.HealthRequest + 2, // 14: fjbellows.agent.v1.AgentService.Exec:input_type -> fjbellows.agent.v1.ShellMsg + 1, // 15: fjbellows.agent.v1.AgentService.Health:output_type -> fjbellows.agent.v1.HealthResponse + 3, // 16: fjbellows.agent.v1.AgentService.Exec:output_type -> fjbellows.agent.v1.ShellEvent + 15, // [15:17] is the sub-list for method output_type + 13, // [13:15] is the sub-list for method input_type + 13, // [13:13] is the sub-list for extension type_name + 13, // [13:13] is the sub-list for extension extendee + 0, // [0:13] is the sub-list for field type_name +} + +func init() { file_fjbellows_agent_v1_agent_proto_init() } +func file_fjbellows_agent_v1_agent_proto_init() { + if File_fjbellows_agent_v1_agent_proto != nil { + return + } + file_fjbellows_agent_v1_agent_proto_msgTypes[2].OneofWrappers = []any{ + (*ShellMsg_Open)(nil), + (*ShellMsg_Stdin)(nil), + (*ShellMsg_Resize)(nil), + (*ShellMsg_Signal)(nil), + (*ShellMsg_CloseStdin)(nil), + } + file_fjbellows_agent_v1_agent_proto_msgTypes[3].OneofWrappers = []any{ + (*ShellEvent_Opened)(nil), + (*ShellEvent_Stdout)(nil), + (*ShellEvent_Stderr)(nil), + (*ShellEvent_Exit)(nil), + } + type x struct{} + out := protoimpl.TypeBuilder{ + File: protoimpl.DescBuilder{ + GoPackagePath: reflect.TypeOf(x{}).PkgPath(), + RawDescriptor: unsafe.Slice(unsafe.StringData(file_fjbellows_agent_v1_agent_proto_rawDesc), len(file_fjbellows_agent_v1_agent_proto_rawDesc)), + NumEnums: 0, + NumMessages: 15, + NumExtensions: 0, + NumServices: 1, + }, + GoTypes: file_fjbellows_agent_v1_agent_proto_goTypes, + DependencyIndexes: file_fjbellows_agent_v1_agent_proto_depIdxs, + MessageInfos: file_fjbellows_agent_v1_agent_proto_msgTypes, + }.Build() + File_fjbellows_agent_v1_agent_proto = out.File + file_fjbellows_agent_v1_agent_proto_goTypes = nil + file_fjbellows_agent_v1_agent_proto_depIdxs = nil +} diff --git a/gen/fjbellows/agent/v1/agentv1connect/agent.connect.go b/gen/fjbellows/agent/v1/agentv1connect/agent.connect.go new file mode 100644 index 0000000..c62c4a4 --- /dev/null +++ b/gen/fjbellows/agent/v1/agentv1connect/agent.connect.go @@ -0,0 +1,166 @@ +// Code generated by protoc-gen-connect-go. DO NOT EDIT. +// +// Source: fjbellows/agent/v1/agent.proto + +package agentv1connect + +import ( + connect "connectrpc.com/connect" + context "context" + errors "errors" + v1 "github.com/hstern/fj-bellows/gen/fjbellows/agent/v1" + http "net/http" + strings "strings" +) + +// This is a compile-time assertion to ensure that this generated file and the connect package are +// compatible. If you get a compiler error that this constant is not defined, this code was +// generated with a version of connect newer than the one compiled into your binary. You can fix the +// problem by either regenerating this code with an older version of connect or updating the connect +// version compiled into your binary. +const _ = connect.IsAtLeastVersion1_13_0 + +const ( + // AgentServiceName is the fully-qualified name of the AgentService service. + AgentServiceName = "fjbellows.agent.v1.AgentService" +) + +// These constants are the fully-qualified names of the RPCs defined in this package. They're +// exposed at runtime as Spec.Procedure and as the final two segments of the HTTP route. +// +// Note that these are different from the fully-qualified method names used by +// google.golang.org/protobuf/reflect/protoreflect. To convert from these constants to +// reflection-formatted method names, remove the leading slash and convert the remaining slash to a +// period. +const ( + // AgentServiceHealthProcedure is the fully-qualified name of the AgentService's Health RPC. + AgentServiceHealthProcedure = "/fjbellows.agent.v1.AgentService/Health" + // AgentServiceExecProcedure is the fully-qualified name of the AgentService's Exec RPC. + AgentServiceExecProcedure = "/fjbellows.agent.v1.AgentService/Exec" +) + +// AgentServiceClient is a client for the fjbellows.agent.v1.AgentService service. +type AgentServiceClient interface { + // Health returns a readiness snapshot of the agent process. Used by the + // orchestrator's readiness gate and the e2e harness to assert "agent is + // up and responding to RPCs". The response always populates `ready=true` + // when the call reaches the handler — the field exists so callers can + // grep one boolean instead of inspecting the absence of an error. + Health(context.Context, *connect.Request[v1.HealthRequest]) (*connect.Response[v1.HealthResponse], error) + // Exec runs a single program on the target. Bidi streaming carries + // stdin/stdout/stderr/exit-code + signals in the framed envelope below. + // The orchestrator opens a matching bidi stream to fjbctl and copies + // frames between the two without reinterpretation; see FJB-93 design + // doc (docs/designs/remote-shell.md) for the end-to-end picture. + // + // Phase A (FJB-93 Phase A) of this RPC is **pipe-mode only**: the + // handler refuses ExecOpen.tty=true with CodeUnimplemented. PTY + + // window resize + interactive shell come in Phase C, on top of the + // same envelope. + Exec(context.Context) *connect.BidiStreamForClient[v1.ShellMsg, v1.ShellEvent] +} + +// NewAgentServiceClient constructs a client for the fjbellows.agent.v1.AgentService service. By +// default, it uses the Connect protocol with the binary Protobuf Codec, asks for gzipped responses, +// and sends uncompressed requests. To use the gRPC or gRPC-Web protocols, supply the +// connect.WithGRPC() or connect.WithGRPCWeb() options. +// +// The URL supplied here should be the base URL for the Connect or gRPC server (for example, +// http://api.acme.com or https://acme.com/grpc). +func NewAgentServiceClient(httpClient connect.HTTPClient, baseURL string, opts ...connect.ClientOption) AgentServiceClient { + baseURL = strings.TrimRight(baseURL, "/") + agentServiceMethods := v1.File_fjbellows_agent_v1_agent_proto.Services().ByName("AgentService").Methods() + return &agentServiceClient{ + health: connect.NewClient[v1.HealthRequest, v1.HealthResponse]( + httpClient, + baseURL+AgentServiceHealthProcedure, + connect.WithSchema(agentServiceMethods.ByName("Health")), + connect.WithClientOptions(opts...), + ), + exec: connect.NewClient[v1.ShellMsg, v1.ShellEvent]( + httpClient, + baseURL+AgentServiceExecProcedure, + connect.WithSchema(agentServiceMethods.ByName("Exec")), + connect.WithClientOptions(opts...), + ), + } +} + +// agentServiceClient implements AgentServiceClient. +type agentServiceClient struct { + health *connect.Client[v1.HealthRequest, v1.HealthResponse] + exec *connect.Client[v1.ShellMsg, v1.ShellEvent] +} + +// Health calls fjbellows.agent.v1.AgentService.Health. +func (c *agentServiceClient) Health(ctx context.Context, req *connect.Request[v1.HealthRequest]) (*connect.Response[v1.HealthResponse], error) { + return c.health.CallUnary(ctx, req) +} + +// Exec calls fjbellows.agent.v1.AgentService.Exec. +func (c *agentServiceClient) Exec(ctx context.Context) *connect.BidiStreamForClient[v1.ShellMsg, v1.ShellEvent] { + return c.exec.CallBidiStream(ctx) +} + +// AgentServiceHandler is an implementation of the fjbellows.agent.v1.AgentService service. +type AgentServiceHandler interface { + // Health returns a readiness snapshot of the agent process. Used by the + // orchestrator's readiness gate and the e2e harness to assert "agent is + // up and responding to RPCs". The response always populates `ready=true` + // when the call reaches the handler — the field exists so callers can + // grep one boolean instead of inspecting the absence of an error. + Health(context.Context, *connect.Request[v1.HealthRequest]) (*connect.Response[v1.HealthResponse], error) + // Exec runs a single program on the target. Bidi streaming carries + // stdin/stdout/stderr/exit-code + signals in the framed envelope below. + // The orchestrator opens a matching bidi stream to fjbctl and copies + // frames between the two without reinterpretation; see FJB-93 design + // doc (docs/designs/remote-shell.md) for the end-to-end picture. + // + // Phase A (FJB-93 Phase A) of this RPC is **pipe-mode only**: the + // handler refuses ExecOpen.tty=true with CodeUnimplemented. PTY + + // window resize + interactive shell come in Phase C, on top of the + // same envelope. + Exec(context.Context, *connect.BidiStream[v1.ShellMsg, v1.ShellEvent]) error +} + +// NewAgentServiceHandler builds an HTTP handler from the service implementation. It returns the +// path on which to mount the handler and the handler itself. +// +// By default, handlers support the Connect, gRPC, and gRPC-Web protocols with the binary Protobuf +// and JSON codecs. They also support gzip compression. +func NewAgentServiceHandler(svc AgentServiceHandler, opts ...connect.HandlerOption) (string, http.Handler) { + agentServiceMethods := v1.File_fjbellows_agent_v1_agent_proto.Services().ByName("AgentService").Methods() + agentServiceHealthHandler := connect.NewUnaryHandler( + AgentServiceHealthProcedure, + svc.Health, + connect.WithSchema(agentServiceMethods.ByName("Health")), + connect.WithHandlerOptions(opts...), + ) + agentServiceExecHandler := connect.NewBidiStreamHandler( + AgentServiceExecProcedure, + svc.Exec, + connect.WithSchema(agentServiceMethods.ByName("Exec")), + connect.WithHandlerOptions(opts...), + ) + return "/fjbellows.agent.v1.AgentService/", http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch r.URL.Path { + case AgentServiceHealthProcedure: + agentServiceHealthHandler.ServeHTTP(w, r) + case AgentServiceExecProcedure: + agentServiceExecHandler.ServeHTTP(w, r) + default: + http.NotFound(w, r) + } + }) +} + +// UnimplementedAgentServiceHandler returns CodeUnimplemented from all methods. +type UnimplementedAgentServiceHandler struct{} + +func (UnimplementedAgentServiceHandler) Health(context.Context, *connect.Request[v1.HealthRequest]) (*connect.Response[v1.HealthResponse], error) { + return nil, connect.NewError(connect.CodeUnimplemented, errors.New("fjbellows.agent.v1.AgentService.Health is not implemented")) +} + +func (UnimplementedAgentServiceHandler) Exec(context.Context, *connect.BidiStream[v1.ShellMsg, v1.ShellEvent]) error { + return connect.NewError(connect.CodeUnimplemented, errors.New("fjbellows.agent.v1.AgentService.Exec is not implemented")) +} diff --git a/internal/agent/exec.go b/internal/agent/exec.go new file mode 100644 index 0000000..c2d4792 --- /dev/null +++ b/internal/agent/exec.go @@ -0,0 +1,307 @@ +package agent + +import ( + "context" + "errors" + "fmt" + "io" + "log/slog" + "os/exec" + "syscall" + + "connectrpc.com/connect" + + agentv1 "github.com/hstern/fj-bellows/gen/fjbellows/agent/v1" +) + +// Exec implements AgentService.Exec for the agent. Phase A (FJB-93) +// supports **pipe mode only** — ExecOpen.tty=true is rejected with +// CodeUnimplemented. PTY allocation and window-resize handling land in +// FJB-93 Phase C. +// +// Concurrency: the handler launches three goroutines that share the +// stream's Send side (stdout pump, stderr pump, terminal Exit emit) and +// one that owns Receive. Send-side serialization goes through a buffered +// channel + single sender goroutine; that avoids manual mutexing of +// connect.BidiStream.Send and gives us bounded backpressure for free. +func (h *Handler) Exec(ctx context.Context, stream *connect.BidiStream[agentv1.ShellMsg, agentv1.ShellEvent]) error { + open, err := receiveOpen(stream) + if err != nil { + return err + } + + if open.GetTty() { + return connect.NewError(connect.CodeUnimplemented, errors.New("PTY mode is not yet implemented (FJB-93 Phase C)")) + } + if len(open.GetArgv()) == 0 { + return connect.NewError(connect.CodeInvalidArgument, errors.New("argv must be non-empty")) + } + + // Single send goroutine owns the stream's send side. Buffer is sized + // so a brief stall on the network doesn't immediately backpressure + // the stdout/stderr pumps inside the agent. + sendBuf := make(chan *agentv1.ShellEvent, 64) + sendDone := make(chan struct{}) + sendErr := make(chan error, 1) + go runSender(stream, sendBuf, sendDone, sendErr) + + // Sentinel frame matches the StreamEvents convention — open call + // returns immediately even before fork+exec completes. + sendBuf <- &agentv1.ShellEvent{ + Kind: &agentv1.ShellEvent_Opened{Opened: &agentv1.ExecOpened{}}, + } + + // Build environment: agent process env is NOT inherited; only the + // explicit map in Open is set. Matches the design doc's "no implicit + // env passthrough" rule. + env := make([]string, 0, len(open.GetEnv())) + for k, v := range open.GetEnv() { + env = append(env, k+"="+v) + } + + //nolint:gosec // G204: argv is supplied by an authenticated orchestrator caller; that's the whole point of this RPC. + cmd := exec.CommandContext(ctx, open.GetArgv()[0], open.GetArgv()[1:]...) + cmd.Env = env + // New process group so signals can be delivered to the whole tree + // (sh -c "..." spawning children, for example). + cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true} + + // Wire stdout/stderr via custom Writers rather than StdoutPipe / + // StderrPipe. exec.Cmd documents that Wait closes pipes it created, + // which races our pump goroutines on fast-exiting commands; the + // Writer path uses exec's internal copy goroutines instead, which + // Wait blocks on before returning. That makes the "all output + // flushed before Exit" ordering structural rather than timing-based. + cmd.Stdout = &streamWriter{ch: sendBuf, wrap: asStdout} + cmd.Stderr = &streamWriter{ch: sendBuf, wrap: asStderr} + + stdinW, err := cmd.StdinPipe() + if err != nil { + close(sendBuf) + <-sendDone + return connect.NewError(connect.CodeInternal, fmt.Errorf("stdin pipe: %w", err)) + } + + if err := cmd.Start(); err != nil { + // Mirror sh's convention: command-not-found is exit 127. + _ = stdinW.Close() + sendBuf <- exitEvent(127, "") + close(sendBuf) + <-sendDone + // The RPC itself succeeded; the child's failure is on-stream. + //nolint:nilerr // start failure is reported as on-stream Exit{127}, not as an RPC error + return nil + } + + // Client-side reader: stdin frames, signals, CloseStdin. Runs until + // the client closes its send side OR the child exits. Returns the + // receive error so the outer scope can decide whether to log it. + recvDone := make(chan struct{}) + go func() { + defer close(recvDone) + runReceiver(stream, cmd, stdinW) + }() + + // Wait for child to exit. cmd.Wait blocks until both the child has + // exited AND exec's internal stdout/stderr copy goroutines have + // drained — so all output is in the sendBuf channel before this + // returns. The CommandContext-bound cancellation covers the "RPC + // ctx canceled" case: the kernel sends SIGKILL to the process group + // and Wait returns shortly after. + waitErr := cmd.Wait() + + sendBuf <- buildExitEvent(waitErr) + close(sendBuf) + + // The receiver goroutine may still be blocked on stream.Receive(). + // On the agent side we don't have a way to forcibly close the + // receive half from a handler; rely on the client to close its send + // side or on the transport to tear down when we return. Either way, + // the goroutine exits when Receive returns an error. + _ = recvDone + + <-sendDone + if err := drainSendErr(sendErr); err != nil { + slog.Default().Debug("agent exec send error", "err", err) + } + return nil +} + +// receiveOpen reads the first frame and asserts it is an ExecOpen. The +// design doc says Open must be the first message exactly once; anything +// else here is a client bug we surface as InvalidArgument. +func receiveOpen(stream *connect.BidiStream[agentv1.ShellMsg, agentv1.ShellEvent]) (*agentv1.ExecOpen, error) { + msg, err := stream.Receive() + if err != nil { + if errors.Is(err, io.EOF) { + return nil, connect.NewError(connect.CodeInvalidArgument, errors.New("stream closed before ExecOpen")) + } + return nil, err + } + open, ok := msg.GetKind().(*agentv1.ShellMsg_Open) + if !ok || open.Open == nil { + return nil, connect.NewError(connect.CodeInvalidArgument, errors.New("first frame must be ExecOpen")) + } + return open.Open, nil +} + +// runSender owns the stream's send side. Reads events from sendBuf and +// forwards each to stream.Send. Records the first send error and exits +// when sendBuf is closed. +func runSender(stream *connect.BidiStream[agentv1.ShellMsg, agentv1.ShellEvent], sendBuf <-chan *agentv1.ShellEvent, done chan<- struct{}, errCh chan<- error) { + defer close(done) + for ev := range sendBuf { + if err := stream.Send(ev); err != nil { + select { + case errCh <- err: + default: + } + // Drain the channel so producers don't block, but don't + // send anything else — the stream is dead. + for range sendBuf { //nolint:revive // intentional drain + } + return + } + } +} + +// streamWriter is an io.Writer that emits each Write as a single +// ShellEvent into sendBuf. exec.Cmd's internal copy loop calls Write +// with chunks up to ~32 KiB by default, which lines up with the design +// doc's frame budget. The buffer cannot be reused across Writes since +// Connect's marshaller may reference it asynchronously, so we copy. +type streamWriter struct { + ch chan<- *agentv1.ShellEvent + wrap func([]byte) *agentv1.ShellEvent +} + +func (w *streamWriter) Write(p []byte) (int, error) { + chunk := make([]byte, len(p)) + copy(chunk, p) + w.ch <- w.wrap(chunk) + return len(p), nil +} + +func asStdout(data []byte) *agentv1.ShellEvent { + return &agentv1.ShellEvent{Kind: &agentv1.ShellEvent_Stdout{Stdout: &agentv1.ExecStdout{Data: data}}} +} + +func asStderr(data []byte) *agentv1.ShellEvent { + return &agentv1.ShellEvent{Kind: &agentv1.ShellEvent_Stderr{Stderr: &agentv1.ExecStderr{Data: data}}} +} + +// runReceiver processes client→server frames after Open: Stdin (write to +// child stdin), Signal (deliver to process group), CloseStdin (close +// child stdin EOF), Resize (logged + ignored in pipe mode). +func runReceiver(stream *connect.BidiStream[agentv1.ShellMsg, agentv1.ShellEvent], cmd *exec.Cmd, stdinW io.WriteCloser) { + for { + msg, err := stream.Receive() + if err != nil { + // Either the client closed send-side cleanly (io.EOF) or + // the transport died. Close stdin so the child sees EOF + // in either case; if the child already exited this is a + // no-op on an already-closed pipe. + _ = stdinW.Close() + return + } + switch kind := msg.GetKind().(type) { + case *agentv1.ShellMsg_Stdin: + if kind.Stdin == nil { + continue + } + if _, werr := stdinW.Write(kind.Stdin.GetData()); werr != nil { + // Child closed stdin (probably exited). Drop further + // stdin frames silently. + _ = stdinW.Close() + } + case *agentv1.ShellMsg_CloseStdin: + _ = stdinW.Close() + case *agentv1.ShellMsg_Signal: + deliverSignal(cmd, kind.Signal.GetName()) + case *agentv1.ShellMsg_Resize: + // Pipe mode: Resize is a no-op. Log at debug so a curious + // operator can confirm the frame arrived. + slog.Default().Debug("agent exec: ignoring Resize in pipe mode") + case *agentv1.ShellMsg_Open: + // A second Open is a protocol violation; close stdin to + // nudge the child to exit and let cmd.Wait clean up. + slog.Default().Warn("agent exec: duplicate ExecOpen frame; closing stdin") + _ = stdinW.Close() + return + } + } +} + +// deliverSignal sends a signal to the child's process group. Unknown +// names are dropped (logged). The Setpgid setup in Exec means the +// negative-pid kill targets every descendant. +func deliverSignal(cmd *exec.Cmd, name string) { + sig := signalByName(name) + if sig == 0 { + slog.Default().Debug("agent exec: unknown signal name", "name", name) + return + } + if cmd.Process == nil { + return + } + pgid, err := syscall.Getpgid(cmd.Process.Pid) + if err != nil { + // Fall back to single-process kill. + _ = cmd.Process.Signal(sig) + return + } + _ = syscall.Kill(-pgid, sig) +} + +func signalByName(name string) syscall.Signal { + switch name { + case "SIGINT": + return syscall.SIGINT + case "SIGTERM": + return syscall.SIGTERM + case "SIGQUIT": + return syscall.SIGQUIT + case "SIGHUP": + return syscall.SIGHUP + default: + return 0 + } +} + +func exitEvent(code int32, sig string) *agentv1.ShellEvent { + return &agentv1.ShellEvent{ + Kind: &agentv1.ShellEvent_Exit{Exit: &agentv1.ExecExit{Code: code, Signal: sig}}, + } +} + +// buildExitEvent inspects cmd.Wait's error and produces the terminal +// ExecExit frame. Three cases: clean exit, signal kill, other error. +func buildExitEvent(waitErr error) *agentv1.ShellEvent { + if waitErr == nil { + return exitEvent(0, "") + } + var exitErr *exec.ExitError + if errors.As(waitErr, &exitErr) { + ws, ok := exitErr.Sys().(syscall.WaitStatus) + if ok { + if ws.Signaled() { + return exitEvent(int32(128+int(ws.Signal())), ws.Signal().String()) //nolint:gosec // signum is 1..31, no overflow + } + return exitEvent(int32(ws.ExitStatus()), "") //nolint:gosec // exit code is 0..255 on Linux + } + return exitEvent(int32(exitErr.ExitCode()), "") //nolint:gosec // ExitCode is 0..255 on POSIX + } + // Unknown error from Wait — surface as a non-zero exit so the + // client doesn't conclude success. + return exitEvent(-1, "") +} + +// drainSendErr returns the first send error (if any) without blocking. +func drainSendErr(errCh <-chan error) error { + select { + case err := <-errCh: + return err + default: + return nil + } +} diff --git a/internal/agent/exec_test.go b/internal/agent/exec_test.go new file mode 100644 index 0000000..ee557c9 --- /dev/null +++ b/internal/agent/exec_test.go @@ -0,0 +1,315 @@ +package agent + +import ( + "context" + "errors" + "io" + "net/http" + "net/http/httptest" + "runtime" + "strings" + "testing" + "time" + + "connectrpc.com/connect" + + agentv1 "github.com/hstern/fj-bellows/gen/fjbellows/agent/v1" + "github.com/hstern/fj-bellows/gen/fjbellows/agent/v1/agentv1connect" +) + +// execTestSetup spins up an in-process agent server reachable via H2C +// (Connect bidi streaming needs HTTP/2). Returns a connected client and +// a teardown. +func execTestSetup(t *testing.T) (agentv1connect.AgentServiceClient, func()) { + t.Helper() + if runtime.GOOS != "linux" && runtime.GOOS != "darwin" { + t.Skipf("exec tests need a POSIX-y OS, not %s", runtime.GOOS) + } + h := NewHandler("test", time.Now()) + srv := NewServer("127.0.0.1:0", h, nil) + + ts := httptest.NewUnstartedServer(srv.Handler()) + ts.EnableHTTP2 = true + ts.StartTLS() // bidi streaming requires HTTP/2; httptest's H2 is over TLS + + client := agentv1connect.NewAgentServiceClient(ts.Client(), ts.URL) + return client, ts.Close +} + +// runExec opens an Exec stream, sends Open, then optionally drives the +// stream via `drive`, then drains events until Exit. Returns the +// captured stdout, stderr, exit code, signal name, and any RPC error. +func runExec(t *testing.T, client agentv1connect.AgentServiceClient, open *agentv1.ExecOpen, drive func(*connect.BidiStreamForClient[agentv1.ShellMsg, agentv1.ShellEvent])) (stdout, stderr string, code int32, sig string, rpcErr error) { + t.Helper() + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + stream := client.Exec(ctx) + defer func() { _ = stream.CloseRequest() }() + + if err := stream.Send(&agentv1.ShellMsg{Kind: &agentv1.ShellMsg_Open{Open: open}}); err != nil { + return "", "", 0, "", err + } + + if drive != nil { + drive(stream) + } + + var outBuf, errBuf strings.Builder + for { + ev, err := stream.Receive() + if err != nil { + if errors.Is(err, io.EOF) { + break + } + return outBuf.String(), errBuf.String(), code, sig, err + } + switch k := ev.GetKind().(type) { + case *agentv1.ShellEvent_Opened: + // sentinel; ignore. + case *agentv1.ShellEvent_Stdout: + outBuf.Write(k.Stdout.GetData()) + case *agentv1.ShellEvent_Stderr: + errBuf.Write(k.Stderr.GetData()) + case *agentv1.ShellEvent_Exit: + code = k.Exit.GetCode() + sig = k.Exit.GetSignal() + return outBuf.String(), errBuf.String(), code, sig, nil + } + } + return outBuf.String(), errBuf.String(), code, sig, nil +} + +func TestExec_Echo(t *testing.T) { + t.Parallel() + client, teardown := execTestSetup(t) + defer teardown() + + out, errOut, code, sig, err := runExec(t, client, + &agentv1.ExecOpen{Argv: []string{"echo", "hello"}}, nil) + if err != nil { + t.Fatalf("Exec: %v", err) + } + if out != "hello\n" { + t.Errorf("stdout = %q, want %q", out, "hello\n") + } + if errOut != "" { + t.Errorf("stderr = %q, want empty", errOut) + } + if code != 0 || sig != "" { + t.Errorf("exit = %d / %q, want 0 / empty", code, sig) + } +} + +func TestExec_StdinThenClose(t *testing.T) { + t.Parallel() + client, teardown := execTestSetup(t) + defer teardown() + + drive := func(s *connect.BidiStreamForClient[agentv1.ShellMsg, agentv1.ShellEvent]) { + _ = s.Send(&agentv1.ShellMsg{Kind: &agentv1.ShellMsg_Stdin{ + Stdin: &agentv1.ExecStdin{Data: []byte("hi cat\n")}, + }}) + _ = s.Send(&agentv1.ShellMsg{Kind: &agentv1.ShellMsg_CloseStdin{ + CloseStdin: &agentv1.ExecCloseStdin{}, + }}) + } + out, _, code, sig, err := runExec(t, client, + &agentv1.ExecOpen{Argv: []string{"cat"}}, drive) + if err != nil { + t.Fatalf("Exec: %v", err) + } + if out != "hi cat\n" { + t.Errorf("stdout = %q, want %q", out, "hi cat\n") + } + if code != 0 || sig != "" { + t.Errorf("exit = %d / %q, want 0 / empty", code, sig) + } +} + +func TestExec_NonZeroExit(t *testing.T) { + t.Parallel() + client, teardown := execTestSetup(t) + defer teardown() + + _, _, code, sig, err := runExec(t, client, + &agentv1.ExecOpen{Argv: []string{"sh", "-c", "exit 7"}}, nil) + if err != nil { + t.Fatalf("Exec: %v", err) + } + if code != 7 || sig != "" { + t.Errorf("exit = %d / %q, want 7 / empty", code, sig) + } +} + +func TestExec_Stderr(t *testing.T) { + t.Parallel() + client, teardown := execTestSetup(t) + defer teardown() + + out, errOut, code, _, err := runExec(t, client, + &agentv1.ExecOpen{Argv: []string{"sh", "-c", "echo to-err >&2"}}, nil) + if err != nil { + t.Fatalf("Exec: %v", err) + } + if out != "" { + t.Errorf("stdout = %q, want empty", out) + } + if errOut != "to-err\n" { + t.Errorf("stderr = %q, want %q", errOut, "to-err\n") + } + if code != 0 { + t.Errorf("exit = %d, want 0", code) + } +} + +func TestExec_EnvPassthrough(t *testing.T) { + t.Parallel() + client, teardown := execTestSetup(t) + defer teardown() + + out, _, code, _, err := runExec(t, client, + &agentv1.ExecOpen{ + Argv: []string{"sh", "-c", "echo FOO=$FOO; echo PATH_LEN=${#PATH}"}, + Env: map[string]string{"FOO": "bar", "PATH": "/usr/bin:/bin"}, + }, nil) + if err != nil { + t.Fatalf("Exec: %v", err) + } + if !strings.Contains(out, "FOO=bar") { + t.Errorf("stdout = %q, want it to contain FOO=bar", out) + } + if code != 0 { + t.Errorf("exit = %d, want 0", code) + } +} + +func TestExec_RejectsTTY(t *testing.T) { + t.Parallel() + client, teardown := execTestSetup(t) + defer teardown() + + _, _, _, _, err := runExec(t, client, + &agentv1.ExecOpen{Argv: []string{"echo"}, Tty: true}, nil) + if err == nil { + t.Fatal("Exec with tty=true succeeded, want CodeUnimplemented") + } + if connect.CodeOf(err) != connect.CodeUnimplemented { + t.Errorf("code = %v, want CodeUnimplemented; err = %v", connect.CodeOf(err), err) + } +} + +func TestExec_RejectsEmptyArgv(t *testing.T) { + t.Parallel() + client, teardown := execTestSetup(t) + defer teardown() + + _, _, _, _, err := runExec(t, client, + &agentv1.ExecOpen{Argv: nil}, nil) + if err == nil { + t.Fatal("Exec with empty argv succeeded, want CodeInvalidArgument") + } + if connect.CodeOf(err) != connect.CodeInvalidArgument { + t.Errorf("code = %v, want CodeInvalidArgument; err = %v", connect.CodeOf(err), err) + } +} + +func TestExec_CommandNotFound(t *testing.T) { + t.Parallel() + client, teardown := execTestSetup(t) + defer teardown() + + _, _, code, sig, err := runExec(t, client, + &agentv1.ExecOpen{Argv: []string{"/no/such/binary/fjb-test-not-real"}}, nil) + if err != nil { + t.Fatalf("Exec: %v", err) + } + if code != 127 || sig != "" { + t.Errorf("exit = %d / %q, want 127 / empty", code, sig) + } +} + +func TestExec_SignalKill(t *testing.T) { + t.Parallel() + client, teardown := execTestSetup(t) + defer teardown() + + drive := func(s *connect.BidiStreamForClient[agentv1.ShellMsg, agentv1.ShellEvent]) { + // Give the child a moment to actually be exec'd before we + // signal it; otherwise the signal might race fork+exec and + // land on a process that hasn't yet replaced itself with + // sleep. + time.Sleep(200 * time.Millisecond) + _ = s.Send(&agentv1.ShellMsg{Kind: &agentv1.ShellMsg_Signal{ + Signal: &agentv1.ExecSignal{Name: "SIGTERM"}, + }}) + } + _, _, code, sig, err := runExec(t, client, + &agentv1.ExecOpen{Argv: []string{"sleep", "30"}}, drive) + if err != nil { + t.Fatalf("Exec: %v", err) + } + if sig == "" { + t.Errorf("signal = %q, want non-empty (process should have been killed); code = %d", sig, code) + } + // Exit code convention for signal kill: 128 + signum. SIGTERM is 15. + if code != 128+15 { + t.Logf("exit code = %d (informational; convention is 128+signum)", code) + } +} + +func TestExec_FirstFrameMustBeOpen(t *testing.T) { + t.Parallel() + client, teardown := execTestSetup(t) + defer teardown() + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + stream := client.Exec(ctx) + defer func() { _ = stream.CloseRequest() }() + + // Send Stdin before Open — protocol violation. + if err := stream.Send(&agentv1.ShellMsg{Kind: &agentv1.ShellMsg_Stdin{ + Stdin: &agentv1.ExecStdin{Data: []byte("nope")}, + }}); err != nil { + t.Fatalf("Send Stdin: %v", err) + } + _, err := stream.Receive() + if err == nil { + t.Fatal("Receive succeeded after pre-Open Stdin, want CodeInvalidArgument") + } + if connect.CodeOf(err) != connect.CodeInvalidArgument { + t.Errorf("code = %v, want CodeInvalidArgument; err = %v", connect.CodeOf(err), err) + } +} + +// TestExec_BearerAuth confirms the Exec RPC is gated by the same bearer +// middleware as Health (the wrap is shared at the http.Handler layer). +func TestExec_BearerAuth(t *testing.T) { + t.Parallel() + + h := NewHandler("test", time.Now()) + srv := NewServer("127.0.0.1:0", h, nil, WithBearerToken("s3cret")) + + ts := httptest.NewUnstartedServer(srv.Handler()) + ts.EnableHTTP2 = true + ts.StartTLS() + defer ts.Close() + + client := agentv1connect.NewAgentServiceClient(ts.Client(), ts.URL) + stream := client.Exec(context.Background()) + defer func() { _ = stream.CloseRequest() }() + _ = stream.Send(&agentv1.ShellMsg{Kind: &agentv1.ShellMsg_Open{ + Open: &agentv1.ExecOpen{Argv: []string{"echo"}}, + }}) + _, err := stream.Receive() + if err == nil { + t.Fatal("Exec succeeded without token, want CodeUnauthenticated") + } + if connect.CodeOf(err) != connect.CodeUnauthenticated { + t.Errorf("code = %v, want CodeUnauthenticated", connect.CodeOf(err)) + } +} + +// Confirm we're not importing http blindly. +var _ = http.NoBody diff --git a/internal/agent/handler.go b/internal/agent/handler.go new file mode 100644 index 0000000..d15e199 --- /dev/null +++ b/internal/agent/handler.go @@ -0,0 +1,57 @@ +// Package agent implements fjbagent — the small daemon that runs on every +// fj-bellows worker and cache VM. It exposes ConnectRPC services the +// orchestrator dials into (Health here; Exec lands in FJB-93). Auth is a +// per-deployment shared secret on the Authorization header; the wire is +// plain HTTP/2 cleartext on top of the WG fabric. +package agent + +import ( + "context" + "os" + "time" + + "connectrpc.com/connect" + "google.golang.org/protobuf/types/known/timestamppb" + + agentv1 "github.com/hstern/fj-bellows/gen/fjbellows/agent/v1" +) + +// Handler is the AgentService implementation. It holds process-scope +// facts (start time, hostname, build version) that Health returns. Field +// values are captured once at NewHandler time so a slow os.Hostname call +// doesn't block every RPC. +type Handler struct { + buildVersion string + startedAt time.Time + hostname string +} + +// NewHandler captures the process-scope health facts. version is the +// build's main.version string ("dev" when unstamped); now is the start +// time anchor (taken at NewHandler so a delayed Run still reports an +// honest start moment). +func NewHandler(version string, now time.Time) *Handler { + h, err := os.Hostname() + if err != nil { + // Hostname lookup is best-effort. An empty string in the response + // is a clear "we didn't get it" rather than a fabricated guess. + h = "" + } + return &Handler{ + buildVersion: version, + startedAt: now, + hostname: h, + } +} + +// Health returns the readiness snapshot. Always succeeds; the existence +// of the response is itself the "agent is up" signal the orchestrator +// readiness gate watches for. +func (h *Handler) Health(_ context.Context, _ *connect.Request[agentv1.HealthRequest]) (*connect.Response[agentv1.HealthResponse], error) { + return connect.NewResponse(&agentv1.HealthResponse{ + Ready: true, + BuildVersion: h.buildVersion, + StartedAt: timestamppb.New(h.startedAt), + Hostname: h.hostname, + }), nil +} diff --git a/internal/agent/handler_test.go b/internal/agent/handler_test.go new file mode 100644 index 0000000..92c8f4d --- /dev/null +++ b/internal/agent/handler_test.go @@ -0,0 +1,54 @@ +package agent + +import ( + "context" + "testing" + "time" + + "connectrpc.com/connect" + + agentv1 "github.com/hstern/fj-bellows/gen/fjbellows/agent/v1" +) + +func TestHealth_PopulatesProcessFacts(t *testing.T) { + t.Parallel() + + anchor := time.Date(2026, 5, 28, 12, 0, 0, 0, time.UTC) + h := NewHandler("v0.5.0", anchor) + + resp, err := h.Health(context.Background(), connect.NewRequest(&agentv1.HealthRequest{})) + if err != nil { + t.Fatalf("Health: %v", err) + } + got := resp.Msg + + if !got.Ready { + t.Errorf("ready = false, want true") + } + if got.BuildVersion != "v0.5.0" { + t.Errorf("build_version = %q, want %q", got.BuildVersion, "v0.5.0") + } + if got.StartedAt == nil { + t.Fatal("started_at is nil") + } + if !got.StartedAt.AsTime().Equal(anchor) { + t.Errorf("started_at = %v, want %v", got.StartedAt.AsTime(), anchor) + } + // hostname comes from os.Hostname(); we don't assert a specific value, + // only that the field exists (it may legitimately be empty if the call + // failed in this environment). + _ = got.Hostname +} + +func TestHealth_DevVersionDefault(t *testing.T) { + t.Parallel() + + h := NewHandler("dev", time.Now()) + resp, err := h.Health(context.Background(), connect.NewRequest(&agentv1.HealthRequest{})) + if err != nil { + t.Fatalf("Health: %v", err) + } + if resp.Msg.BuildVersion != "dev" { + t.Errorf("build_version = %q, want %q", resp.Msg.BuildVersion, "dev") + } +} diff --git a/internal/agent/server.go b/internal/agent/server.go new file mode 100644 index 0000000..ea89098 --- /dev/null +++ b/internal/agent/server.go @@ -0,0 +1,158 @@ +package agent + +import ( + "context" + "crypto/subtle" + "errors" + "fmt" + "log/slog" + "net" + "net/http" + "os" + "strings" + "time" + + "github.com/hstern/fj-bellows/gen/fjbellows/agent/v1/agentv1connect" +) + +// Server is the agent's HTTP front door. One http.Server multiplexes: +// - ConnectRPC handlers at /./ +// - Plain HTTP /healthz for sd_notify-less readiness checks (curl --fail +// style; not the gated readiness probe — that's AgentService.Health). +// +// The wire is HTTP/1.1 + HTTP/2 cleartext; encryption + authentication of +// the link is WG's responsibility, and on workers the cache↔worker leg is +// on the private VPC subnet behind Linode's VPC isolation. +type Server struct { + listen string + srv *http.Server + log *slog.Logger +} + +// Option configures a Server at construction time. +type Option func(*config) + +type config struct { + token string +} + +// WithBearerToken enforces an Authorization: Bearer header on every +// Connect RPC. The plain HTTP /healthz shim stays open so a sysadmin can +// curl the agent without the token. Empty token disables the middleware +// — useful for unit tests; production deployments always set it. +func WithBearerToken(token string) Option { + return func(c *config) { c.token = token } +} + +// NewServer builds the server but does not start it. listen is a host:port +// suitable for net.Listen("tcp", ...). The handler argument owns the +// process-scope health state. +func NewServer(listen string, h *Handler, log *slog.Logger, opts ...Option) *Server { + if log == nil { + log = slog.Default() + } + var cfg config + for _, opt := range opts { + opt(&cfg) + } + + mux := http.NewServeMux() + + path, connectHandler := agentv1connect.NewAgentServiceHandler(h) + mux.Handle(path, connectHandler) + mux.HandleFunc("/healthz", func(w http.ResponseWriter, _ *http.Request) { + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte("ok\n")) + }) + + var protos http.Protocols + protos.SetHTTP1(true) + protos.SetUnencryptedHTTP2(true) + + return &Server{ + listen: listen, + log: log, + srv: &http.Server{ + Handler: bearerAuth(mux, cfg.token), + Protocols: &protos, + ReadHeaderTimeout: 5 * time.Second, + }, + } +} + +// Handler returns the underlying mux for tests that want to mount the server +// behind httptest.NewServer without binding a real TCP port. +func (s *Server) Handler() http.Handler { + return s.srv.Handler +} + +// Run binds the listener and serves until ctx is cancelled, at which point +// it initiates a short graceful shutdown. +func (s *Server) Run(ctx context.Context) error { + var lc net.ListenConfig + ln, err := lc.Listen(ctx, "tcp", s.listen) + if err != nil { + return fmt.Errorf("agent listen %s: %w", s.listen, err) + } + s.log.Info("agent listening", "addr", ln.Addr().String()) + + serveErr := make(chan error, 1) + go func() { + err := s.srv.Serve(ln) + if errors.Is(err, http.ErrServerClosed) { + err = nil + } + serveErr <- err + }() + + select { + case <-ctx.Done(): + shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := s.srv.Shutdown(shutdownCtx); err != nil { + s.log.Error("agent shutdown", "err", err) + } + return <-serveErr + case err := <-serveErr: + return err + } +} + +// bearerAuth wraps next to require Authorization: Bearer on Connect +// RPC paths. /healthz stays open. Empty token disables the middleware. +func bearerAuth(next http.Handler, token string) http.Handler { + if token == "" { + return next + } + expected := "Bearer " + token + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/healthz" { + next.ServeHTTP(w, r) + return + } + got := r.Header.Get("Authorization") + if subtle.ConstantTimeCompare([]byte(got), []byte(expected)) != 1 { + w.Header().Set("WWW-Authenticate", `Bearer realm="fjbagent"`) + w.WriteHeader(http.StatusUnauthorized) + _, _ = w.Write([]byte(`{"error":"unauthenticated"}`)) + return + } + next.ServeHTTP(w, r) + }) +} + +// LoadToken reads a single-line bearer token from path. Empty files and +// missing files are errors so the operator can't accidentally end up with +// the empty-token disable-auth branch by leaving the file blank. +func LoadToken(path string) (string, error) { + //nolint:gosec // G304: path comes from operator-supplied -token-file flag, not user input. + b, err := os.ReadFile(path) + if err != nil { + return "", fmt.Errorf("read agent token: %w", err) + } + tok := strings.TrimSpace(string(b)) + if tok == "" { + return "", errors.New("agent token file is empty") + } + return tok, nil +} diff --git a/internal/agent/server_test.go b/internal/agent/server_test.go new file mode 100644 index 0000000..8e2f1b8 --- /dev/null +++ b/internal/agent/server_test.go @@ -0,0 +1,205 @@ +package agent + +import ( + "context" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "connectrpc.com/connect" + + agentv1 "github.com/hstern/fj-bellows/gen/fjbellows/agent/v1" + "github.com/hstern/fj-bellows/gen/fjbellows/agent/v1/agentv1connect" +) + +func TestServer_HealthOverHTTP_NoAuth(t *testing.T) { + t.Parallel() + + h := NewHandler("test", time.Now()) + srv := NewServer("127.0.0.1:0", h, nil) + + ts := httptest.NewServer(srv.Handler()) + defer ts.Close() + + client := agentv1connect.NewAgentServiceClient(http.DefaultClient, ts.URL) + 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 TestServer_BearerAuth_RejectsMissingToken(t *testing.T) { + t.Parallel() + + h := NewHandler("test", time.Now()) + srv := NewServer("127.0.0.1:0", h, nil, WithBearerToken("s3cret")) + + ts := httptest.NewServer(srv.Handler()) + defer ts.Close() + + client := agentv1connect.NewAgentServiceClient(http.DefaultClient, ts.URL) + _, err := client.Health(context.Background(), connect.NewRequest(&agentv1.HealthRequest{})) + if err == nil { + t.Fatal("Health succeeded without token, want unauthenticated") + } + if connect.CodeOf(err) != connect.CodeUnauthenticated { + t.Errorf("code = %v, want CodeUnauthenticated; err = %v", connect.CodeOf(err), err) + } +} + +func TestServer_BearerAuth_AcceptsCorrectToken(t *testing.T) { + t.Parallel() + + h := NewHandler("test", time.Now()) + srv := NewServer("127.0.0.1:0", h, nil, WithBearerToken("s3cret")) + + ts := httptest.NewServer(srv.Handler()) + defer ts.Close() + + interceptor := connect.UnaryInterceptorFunc(func(next connect.UnaryFunc) connect.UnaryFunc { + return func(ctx context.Context, req connect.AnyRequest) (connect.AnyResponse, error) { + req.Header().Set("Authorization", "Bearer s3cret") + return next(ctx, req) + } + }) + client := agentv1connect.NewAgentServiceClient(http.DefaultClient, ts.URL, connect.WithInterceptors(interceptor)) + 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 TestServer_BearerAuth_RejectsWrongToken(t *testing.T) { + t.Parallel() + + h := NewHandler("test", time.Now()) + srv := NewServer("127.0.0.1:0", h, nil, WithBearerToken("right")) + + ts := httptest.NewServer(srv.Handler()) + defer ts.Close() + + interceptor := connect.UnaryInterceptorFunc(func(next connect.UnaryFunc) connect.UnaryFunc { + return func(ctx context.Context, req connect.AnyRequest) (connect.AnyResponse, error) { + req.Header().Set("Authorization", "Bearer wrong") + return next(ctx, req) + } + }) + client := agentv1connect.NewAgentServiceClient(http.DefaultClient, ts.URL, connect.WithInterceptors(interceptor)) + _, err := client.Health(context.Background(), connect.NewRequest(&agentv1.HealthRequest{})) + if err == nil { + t.Fatal("Health succeeded with wrong token, want unauthenticated") + } + if connect.CodeOf(err) != connect.CodeUnauthenticated { + t.Errorf("code = %v, want CodeUnauthenticated", connect.CodeOf(err)) + } +} + +func TestServer_HealthzOpenWithoutToken(t *testing.T) { + t.Parallel() + + h := NewHandler("test", time.Now()) + srv := NewServer("127.0.0.1:0", h, nil, WithBearerToken("s3cret")) + + ts := httptest.NewServer(srv.Handler()) + defer ts.Close() + + // No Authorization header; /healthz must still answer 200. + resp, err := http.Get(ts.URL + "/healthz") //nolint:noctx // test + if err != nil { + t.Fatalf("GET /healthz: %v", err) + } + defer func() { _ = resp.Body.Close() }() + if resp.StatusCode != http.StatusOK { + t.Errorf("status = %d, want 200", resp.StatusCode) + } +} + +func TestServer_Run_ShutsDownOnContextCancel(t *testing.T) { + t.Parallel() + + h := NewHandler("test", time.Now()) + srv := NewServer("127.0.0.1:0", h, nil) + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { done <- srv.Run(ctx) }() + + // Give the listener time to bind. We don't have a sync hook for that + // here; in production main.go relies on systemd's Type=notify. For the + // test, a tiny sleep is acceptable to ensure Serve has started. + time.Sleep(50 * time.Millisecond) + cancel() + + select { + case err := <-done: + if err != nil { + t.Errorf("Run returned %v after cancel, want nil", err) + } + case <-time.After(5 * time.Second): + t.Fatal("Run did not return within 5s of context cancel") + } +} + +func TestLoadToken(t *testing.T) { + t.Parallel() + + dir := t.TempDir() + + tests := []struct { + name string + content string + want string + wantErr string + }{ + {name: "trimmed", content: " abc\n", want: "abc"}, + {name: "bare", content: "xyz", want: "xyz"}, + {name: "empty", content: "", wantErr: "empty"}, + {name: "whitespace-only", content: " \n ", wantErr: "empty"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t.Parallel() + p := filepath.Join(dir, tt.name) + if err := os.WriteFile(p, []byte(tt.content), 0o600); err != nil { + t.Fatal(err) + } + got, err := LoadToken(p) + if tt.wantErr != "" { + if err == nil { + t.Fatalf("LoadToken: got nil error, want one containing %q", tt.wantErr) + } + if !strings.Contains(err.Error(), tt.wantErr) { + t.Errorf("LoadToken err = %v, want substring %q", err, tt.wantErr) + } + return + } + if err != nil { + t.Fatalf("LoadToken: %v", err) + } + if got != tt.want { + t.Errorf("got %q, want %q", got, tt.want) + } + }) + } +} + +func TestLoadToken_Missing(t *testing.T) { + t.Parallel() + _, err := LoadToken(filepath.Join(t.TempDir(), "does-not-exist")) + if err == nil { + t.Fatal("LoadToken: got nil error, want one for missing file") + } +} diff --git a/internal/provider/linode/cache-cloud-init.yaml.tmpl b/internal/provider/linode/cache-cloud-init.yaml.tmpl index 26c3b83..5cead87 100644 --- a/internal/provider/linode/cache-cloud-init.yaml.tmpl +++ b/internal/provider/linode/cache-cloud-init.yaml.tmpl @@ -21,6 +21,10 @@ packages: {{- if .IPTablesScript}} - iptables-persistent {{- end}} +{{- if .OrchestratorWGPubkey}} + - wireguard + - awscli +{{- end}} write_files: - path: /etc/zot/config.json permissions: '0640' @@ -68,6 +72,47 @@ write_files: permissions: '0750' content: | {{.IPTablesScript | indent 6}} +{{- end}} +{{- if .OrchestratorWGPubkey}} + - path: /usr/local/sbin/fjb-wg-bootstrap.sh + permissions: '0750' + content: | + #!/usr/bin/env bash + # FJB-99 Phase A — cache wg-quick bootstrap. + # Idempotent: keypair generates once on first boot; subsequent + # boots reuse the persisted private key. wg0.conf is always + # rewritten so a config bump (peer pubkey rotation, port change) + # takes effect on reboot without operator intervention. + set -euo pipefail + umask 0077 + install -d -m 0700 /etc/wireguard + if [ ! -s /etc/wireguard/privatekey ]; then + wg genkey | tee /etc/wireguard/privatekey | wg pubkey > /etc/wireguard/publickey + chmod 0600 /etc/wireguard/privatekey + chmod 0644 /etc/wireguard/publickey + fi + priv=$(cat /etc/wireguard/privatekey) + cat > /etc/wireguard/wg0.conf </wg-pubkey.txt for the orchestrator to read back + // (Phase B). Empty under ssh-mode deployments — the WG block is + // then skipped entirely. + OrchestratorWGPubkey string + OrchestratorWGAddr string // e.g. "100.64.0.1" + CacheWGAddr string // e.g. "100.64.0.2" + WGListenPort int // UDP, e.g. 51820 } // renderCacheCloudInit fills the embedded template. Defaults to the @@ -735,6 +770,16 @@ type workerExtrasData struct { // inside RFC 6598 CGNAT space. const defaultOrchestratorWGAddr = "100.64.0.1" +// defaultCacheWGAddr is the cache's address on the WG overlay (.2 in +// the orchestrator=.1/cache=.2 /30 baseline). FJB-99 Phase A bakes it +// into the cache cloud-init's wg0.conf [Interface].Address. +const defaultCacheWGAddr = "100.64.0.2" + +// defaultCacheWGListenPort is the WireGuard project's de-facto default UDP +// port — mirrors config.DefaultWGListenPort but stays local to the +// provider so cache.go doesn't have to import internal/config. +const defaultCacheWGListenPort = 51820 + // workerExtras returns the data the linode provider's Provision needs // to wrap each worker's cloud-init with cache trust + hostname // resolution. Looks up the cache VPC IP lazily (so a fresh-create diff --git a/internal/provider/linode/cache_test.go b/internal/provider/linode/cache_test.go index 4cbbdb2..68edfdb 100644 --- a/internal/provider/linode/cache_test.go +++ b/internal/provider/linode/cache_test.go @@ -509,6 +509,75 @@ func TestRenderCacheCloudInitIPTablesBakeIn(t *testing.T) { } } +// FJB-99 Phase A: when OrchestratorWGPubkey is set, the rendered +// cloud-init installs wireguard + awscli, embeds the wg-bootstrap +// script with the orchestrator pubkey baked into [Peer].PublicKey, +// and runs the script from runcmd. When empty (ssh-mode), none of +// those entries leak in. +func TestRenderCacheCloudInitWGBootstrap(t *testing.T) { + const orchPubkey = "ZmFrZS1vcmNoZXN0cmF0b3ItcHViLWtleS0xMjM0NTY3OD0=" + withWG, err := renderCacheCloudInit(cacheCloudInitParams{ + Bucket: "fjb-cache-test", + Region: testBucketRegion, + Endpoint: "https://us-ord-1.linodeobjects.com", + AccessKey: "AK", + SecretKey: "SK", + ZotVersion: testStubZotVersion, + ServerCertPEM: testStubPEM, + ServerKeyPEM: testStubPEM, + OrchestratorWGPubkey: orchPubkey, + OrchestratorWGAddr: "100.64.0.1", + CacheWGAddr: "100.64.0.2", + WGListenPort: 51820, + }) + if err != nil { + t.Fatalf("render with WG: %v", err) + } + for _, want := range []string{ + "- wireguard", + "- awscli", + "/usr/local/sbin/fjb-wg-bootstrap.sh", + "wg genkey", + "Address = 100.64.0.2/32", + "ListenPort = 51820", + "PublicKey = " + orchPubkey, + "AllowedIPs = 100.64.0.1/32", + "PersistentKeepalive = 25", + "systemctl enable --now wg-quick@wg0", + "s3://fjb-cache-test/wg-pubkey.txt", + "--endpoint-url 'https://us-ord-1.linodeobjects.com'", + } { + if !strings.Contains(withWG, want) { + t.Errorf("WG render missing %q\n---\n%s", want, withWG) + } + } + + withoutWG, err := renderCacheCloudInit(cacheCloudInitParams{ + Bucket: "b", + Region: "r", + Endpoint: testStubEndpoint, + AccessKey: "AK", + SecretKey: "SK", + ZotVersion: testStubZotVersion, + ServerCertPEM: testStubPEM, + ServerKeyPEM: testStubPEM, + }) + if err != nil { + t.Fatalf("render without WG: %v", err) + } + for _, unwanted := range []string{ + "wireguard", + "awscli", + "fjb-wg-bootstrap.sh", + "wg-pubkey.txt", + "wg-quick@wg0", + } { + if strings.Contains(withoutWG, unwanted) { + t.Errorf("ssh-mode render leaked WG substring %q\n---\n%s", unwanted, withoutWG) + } + } +} + // FJB-98: renderIPTablesForCacheGateway returns an empty script for // non-cache-gateway transport or when the worker VPC subnet hasn't // been wired yet — both branches are reachable in real deployments diff --git a/internal/provider/linode/linode.go b/internal/provider/linode/linode.go index 9cbac2e..22684ea 100644 --- a/internal/provider/linode/linode.go +++ b/internal/provider/linode/linode.go @@ -101,6 +101,13 @@ type Linode struct { // synthSpecsForTransport. Unused under legacy ssh mode. wgListenPort int + // orchestratorWGPubkey is the orchestrator's WG public key, captured + // before Configure via SetOrchestratorWGPubkey (FJB-99 Phase A). The + // managed cache's cloud-init bakes it in as the [Peer].PublicKey so + // the cache's wg-quick brings up wg0 with the right peer. Empty + // under legacy ssh mode and unused. + orchestratorWGPubkey string + // workersInFlight counts Provision calls that have entered the // CreateInstance path but not yet returned (success or failure). // Surfaced via Info() for operator debugging — pairs with the @@ -231,6 +238,20 @@ func (l *Linode) SetWGListenPort(port int) { l.wgListenPort = port } +// SetOrchestratorWGPubkey propagates the orchestrator's WG public key +// (Curve25519, base64) into the Linode provider so the managed cache's +// cloud-init can bake it in as the WG peer pubkey (FJB-99 Phase A). +// The cache then comes up with wg-quick referencing the right peer +// and publishes its own pubkey to S3 for the orchestrator to read +// back in Phase B. +// +// Duck-typed from cmd/fj-bellows. No-op for non-Linode providers and +// for ssh-mode deployments (the orchestrator never calls this when +// transport.mode != cache-gateway). +func (l *Linode) SetOrchestratorWGPubkey(pubkey string) { + l.orchestratorWGPubkey = pubkey +} + // CacheStatus returns the managed-cache snapshot consumed by the control // plane's GetCache RPC. Returns nil when no `cache:` block is configured; // the control handler then reports Present=false to the wire. The Linode @@ -494,6 +515,7 @@ func (l *Linode) setupManagedCache(ctx context.Context, tag string) error { // key is supplied via SetSSHAuthorizedKey from cmd/fj-bellows; // empty when no SSH was configured (docker provider). cache.setHardwareContext(fwID, subID, l.sshAuthorizedKey, l.WorkerVPCSubnet()) + cache.setOrchestratorWGPubkey(l.orchestratorWGPubkey) cache.setTransport(l.transportMode) // FJB-88: the worker cache-extras template now reads AllowedIPs // CIDRs from the ACL registry, not from operator-supplied tunnel diff --git a/proto/fjbellows/agent/v1/agent.proto b/proto/fjbellows/agent/v1/agent.proto new file mode 100644 index 0000000..3ae137d --- /dev/null +++ b/proto/fjbellows/agent/v1/agent.proto @@ -0,0 +1,182 @@ +syntax = "proto3"; + +package fjbellows.agent.v1; + +option go_package = "github.com/hstern/fj-bellows/gen/fjbellows/agent/v1;agentv1"; + +import "google/protobuf/timestamp.proto"; + +// AgentService runs on every worker and cache VM. The orchestrator dials it +// over the WG fabric (direct WG peer for cache, via cache-gateway routing for +// workers) and uses it to drive operator-side control RPCs that need code +// running on the target — interactive shell, command exec, reachability +// probes, and a future log-tail surface. +// +// Auth is a per-deployment shared secret presented in the Authorization header +// (Bearer scheme). The wire is plain HTTP/2 cleartext — WG already encrypts +// and authenticates the orchestrator↔cache leg, the cache↔worker leg is on +// the private VPC subnet, and adding TLS would mean another cert lifecycle. +// +// FJB-94 ships only Health; FJB-93 will add Exec, FJB-97 will reuse Health +// for the gRPC-layer reachability probe. +service AgentService { + // Health returns a readiness snapshot of the agent process. Used by the + // orchestrator's readiness gate and the e2e harness to assert "agent is + // up and responding to RPCs". The response always populates `ready=true` + // when the call reaches the handler — the field exists so callers can + // grep one boolean instead of inspecting the absence of an error. + rpc Health(HealthRequest) returns (HealthResponse); + + // Exec runs a single program on the target. Bidi streaming carries + // stdin/stdout/stderr/exit-code + signals in the framed envelope below. + // The orchestrator opens a matching bidi stream to fjbctl and copies + // frames between the two without reinterpretation; see FJB-93 design + // doc (docs/designs/remote-shell.md) for the end-to-end picture. + // + // Phase A (FJB-93 Phase A) of this RPC is **pipe-mode only**: the + // handler refuses ExecOpen.tty=true with CodeUnimplemented. PTY + + // window resize + interactive shell come in Phase C, on top of the + // same envelope. + rpc Exec(stream ShellMsg) returns (stream ShellEvent); +} + +message HealthRequest {} + +message HealthResponse { + // ready is true on every successful invocation; the field is present so + // operator tooling can do `.ready` instead of `error == nil`. It exists + // for future-proofing too — a future agent may want to return ready=false + // while still being reachable (e.g. during a quiesce window). + bool ready = 1; + + // build_version is the agent binary's version string. Populated from + // -ldflags "-X main.version=…" at link time; "dev" when unset. + string build_version = 2; + + // started_at is when the agent process started. Subtracting this from + // now gives the operator-visible uptime. + google.protobuf.Timestamp started_at = 3; + + // hostname is the OS hostname as reported by os.Hostname() at startup. + // Useful when the orchestrator's name → addr map drifts from what the + // VM actually thinks it is. + string hostname = 4; +} + +// ShellMsg is the client→server frame on the Exec bidi stream. Exactly one +// ExecOpen must be the first message; subsequent messages may interleave +// Stdin / Resize / Signal / CloseStdin in any order until the server closes +// the stream after sending Exit. +message ShellMsg { + oneof kind { + ExecOpen open = 1; + ExecStdin stdin = 2; + ExecResize resize = 3; + ExecSignal signal = 4; + ExecCloseStdin close_stdin = 5; + } +} + +// ShellEvent is the server→client frame. Opened arrives first (sentinel, +// matches the StreamEvents convention so the open call returns +// immediately even before fork+exec lands). Stdout and Stderr interleave +// freely. Exit is always last and ends the stream. +message ShellEvent { + oneof kind { + ExecOpened opened = 1; + ExecStdout stdout = 2; + ExecStderr stderr = 3; + ExecExit exit = 4; + } +} + +// ExecOpen is the first ShellMsg on every Exec stream. argv[0] is the +// program to run; an empty argv asks for a login shell (deferred to a +// later phase). env passes additional environment vars beyond the agent +// process's own; $TERM is set from `term` when `tty` is true. +message ExecOpen { + // target identifies which worker/cache to run on. This field is only + // meaningful on the fjbctl→orchestrator hop; the orchestrator strips + // it before forwarding the Open to the agent. Empty on the agent hop. + string target = 1; + + // argv is the program + args. argv[0] is exec'd directly; no shell + // interpolation. Pass `sh -c "..."` for shell semantics. + repeated string argv = 2; + + // env is a map of additional environment variables. Empty by default; + // the agent does NOT pass the orchestrator's environment through. + map env = 3; + + // tty asks the agent to allocate a PTY for the child and wire stdin/ + // stdout (no separate stderr under PTY) to the PTY master. Phase A + // returns CodeUnimplemented if true; PTY support lands in Phase C. + bool tty = 4; + + // term is the $TERM value to set on the child when tty=true. Ignored + // when tty=false. + string term = 5; + + // winsize is the initial PTY window size when tty=true. Ignored + // otherwise. + Winsize winsize = 6; +} + +// Winsize carries terminal columns and rows. +message Winsize { + uint32 cols = 1; + uint32 rows = 2; +} + +// ExecStdin pumps bytes to the child's stdin. Streaming; the client may +// send any number of these. CloseStdin signals EOF. +message ExecStdin { + bytes data = 1; +} + +// ExecStdout / ExecStderr carry bytes from the child's stdout / stderr. +// Frame size is at the server's discretion (default 32 KiB chunks); the +// receiver should not assume line boundaries. +message ExecStdout { + bytes data = 1; +} + +message ExecStderr { + bytes data = 1; +} + +// ExecResize updates the PTY window size mid-session. Only meaningful +// when the Open had tty=true. +message ExecResize { + Winsize winsize = 1; +} + +// ExecSignal forwards a signal to the child's process group. Valid +// names: SIGINT, SIGTERM, SIGQUIT, SIGHUP. Unknown names are ignored +// (logged on the agent side). +message ExecSignal { + string name = 1; +} + +// ExecCloseStdin closes the child's stdin (EOF on the read side). The +// stream stays open for stdout/stderr and the terminal Exit. +message ExecCloseStdin {} + +// ExecOpened is the sentinel server→client frame that fires immediately +// after the agent accepts the Open and begins setting up the child. The +// existence of this frame is what lets the client's stream-open call +// return without waiting for fork+exec to complete. +message ExecOpened {} + +// ExecExit is the terminal server→client frame. Exactly one of +// (code, signal) is meaningful: code is the program's exit status if it +// exited normally; signal is the signal name (e.g. "terminated", +// "killed") if it was killed by a signal. The agent closes the stream +// immediately after sending this. +message ExecExit { + // code is the child's exit status when it exited normally. + int32 code = 1; + // signal is the signal name when the child was killed by a signal. + // Empty when code is meaningful. + string signal = 2; +}