diff --git a/cmd/ecs-agent/agent.go b/cmd/ecs-agent/agent.go index 676becb62..a7c63b569 100644 --- a/cmd/ecs-agent/agent.go +++ b/cmd/ecs-agent/agent.go @@ -40,6 +40,9 @@ type Agent struct { // netns builds awsvpc task network namespaces (nil-safe; bridge/host skip it). netns *taskNetns + // cred serves task IAM role credentials at 169.254.170.2 (nil in unit tests). + cred *credEndpoint + closers []func() error } @@ -103,6 +106,8 @@ func New(cfg config) (*Agent, error) { } a := newAgent(cfg, id, cp, puller, runner, resolver) + a.cred = newCredEndpoint(creds, cfg.Region, cfg.GatewayURL, cfg.GatewayCA, + cfg.CredEndpointIP, cfg.CredEndpointPort, execNetRunner{}) if puller != nil { a.closers = append(a.closers, puller.Close) } @@ -118,6 +123,12 @@ func (a *Agent) Run(ctx context.Context) error { slog.Info("ecs-agent: registered", "cluster", a.id.ClusterName, "instance", a.id.InstanceID) } + if a.cred != nil { + if err := a.cred.Start(); err != nil { + slog.Error("ecs-agent: credential endpoint start failed", "err", err) + } + } + go a.hb.Run(ctx) go a.pollAssignments(ctx) @@ -128,6 +139,9 @@ func (a *Agent) Run(ctx context.Context) error { // Stop closes the runtime. Safe to call once. func (a *Agent) Stop() error { var firstErr error + if a.cred != nil { + _ = a.cred.Stop() + } for _, c := range a.closers { if err := c(); err != nil && firstErr == nil { firstErr = err diff --git a/cmd/ecs-agent/config.go b/cmd/ecs-agent/config.go index 143dfa798..c919354e5 100644 --- a/cmd/ecs-agent/config.go +++ b/cmd/ecs-agent/config.go @@ -3,6 +3,7 @@ package main import ( "bufio" "os" + "strconv" "strings" "time" ) @@ -32,6 +33,8 @@ type config struct { ContainerdSocket string Heartbeat time.Duration PollInterval time.Duration + CredEndpointIP string + CredEndpointPort int } // loadConfig reads the cloud-init env file then lets real env vars override. @@ -78,6 +81,16 @@ func loadConfig(envFile string) config { cfg.PollInterval = d } } + cfg.CredEndpointIP = get("ECS_CRED_ENDPOINT_IP") + if cfg.CredEndpointIP == "" { + cfg.CredEndpointIP = defaultCredEndpointIP + } + cfg.CredEndpointPort = defaultCredEndpointPort + if v := get("ECS_CRED_ENDPOINT_PORT"); v != "" { + if p, err := strconv.Atoi(v); err == nil && p > 0 { + cfg.CredEndpointPort = p + } + } return cfg } diff --git a/cmd/ecs-agent/credendpoint.go b/cmd/ecs-agent/credendpoint.go new file mode 100644 index 000000000..b242af8da --- /dev/null +++ b/cmd/ecs-agent/credendpoint.go @@ -0,0 +1,300 @@ +package main + +import ( + "context" + "encoding/json" + "fmt" + "log/slog" + "net" + "net/http" + "strconv" + "strings" + "sync" + "time" + + "github.com/mulgadc/spinifex/cmd/ecs-agent/credentials" + "github.com/mulgadc/spinifex/internal/ecrauth" + "github.com/mulgadc/spinifex/internal/stsauth" +) + +const ( + // defaultCredEndpointIP is the AWS-reserved link-local address tasks reach for + // container credentials; defaultCredEndpointPort is the standard HTTP port. + defaultCredEndpointIP = "169.254.170.2" + defaultCredEndpointPort = 80 + // credRefreshMargin refreshes a cached credential set this long before expiry + // so a container never starts a request with a token about to lapse. + credRefreshMargin = 5 * time.Minute + // credDummyIface is the dummy interface the endpoint IP is added to in the host + // netns so the agent can bind it. + credDummyIface = "ecs-cred" +) + +// credResponse is the ECS task-credentials JSON the container's AWS SDK expects +// at AWS_CONTAINER_CREDENTIALS_RELATIVE_URI. +type credResponse struct { + RoleArn string `json:"RoleArn"` + AccessKeyID string `json:"AccessKeyId"` + SecretAccessKey string `json:"SecretAccessKey"` + Token string `json:"Token"` + Expiration string `json:"Expiration"` +} + +// credEntry caches one task's assumed-role credentials alongside the role it was +// assumed from. +type credEntry struct { + roleARN string + creds stsauth.Credentials +} + +// credEndpoint is the host-netns HTTP server that hands a task its IAM role +// credentials. It maps credID -> taskRoleARN (registered as tasks start) and +// mints/caches credentials on demand by assuming the role over the gateway with +// the agent's instance credentials. +type credEndpoint struct { + provider credentials.CredentialsProvider + region string + gatewayURL string + caPath string + ip string + port int + run netCmdRunner + + // assume mints fresh credentials for a role; defaults to AssumeRole over the + // gateway and is overridable in tests. + assume func(ctx context.Context, roleARN, sessionName string) (stsauth.Credentials, error) + + mu sync.Mutex + roles map[string]string // credID -> taskRoleARN + cache map[string]credEntry // credID -> cached creds + httpClient *http.Client + + srv *http.Server + ln net.Listener +} + +// newCredEndpoint builds the endpoint. An empty ip/port falls back to the +// AWS-reserved 169.254.170.2:80; tests pass loopback + an ephemeral port. run +// adds the link-local address to the dummy interface (real or fake). +func newCredEndpoint(provider credentials.CredentialsProvider, region, gatewayURL, caPath, ip string, port int, run netCmdRunner) *credEndpoint { + if ip == "" { + ip = defaultCredEndpointIP + } + if port == 0 { + port = defaultCredEndpointPort + } + c := &credEndpoint{ + provider: provider, + region: region, + gatewayURL: gatewayURL, + caPath: caPath, + ip: ip, + port: port, + run: run, + roles: map[string]string{}, + cache: map[string]credEntry{}, + } + c.assume = c.assumeOverGateway + return c +} + +// Register associates a credID with the task role it serves; Deregister removes +// it and drops any cached credentials. Both are no-ops for an empty credID. +func (c *credEndpoint) Register(credID, roleARN string) { + if credID == "" || roleARN == "" { + return + } + c.mu.Lock() + c.roles[credID] = roleARN + c.mu.Unlock() +} + +func (c *credEndpoint) Deregister(credID string) { + if credID == "" { + return + } + c.mu.Lock() + delete(c.roles, credID) + delete(c.cache, credID) + c.mu.Unlock() +} + +// relativeURI is the value injected into a container's +// AWS_CONTAINER_CREDENTIALS_RELATIVE_URI for the given credID. +func credRelativeURI(credID string) string { return "/v2/credentials/" + credID } + +// ServeHTTP serves GET /v2/credentials/{credID}. Unknown paths/credIDs return +// 404; non-GET returns 405; an AssumeRole failure returns 502. +func (c *credEndpoint) ServeHTTP(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + credID, ok := strings.CutPrefix(r.URL.Path, "/v2/credentials/") + if !ok || credID == "" || strings.Contains(credID, "/") { + http.Error(w, "not found", http.StatusNotFound) + return + } + creds, roleARN, err := c.fetch(r.Context(), credID) + if err != nil { + if err == errUnknownCredID { + http.Error(w, "not found", http.StatusNotFound) + return + } + slog.Error("ecs-agent: assume task role failed", "credId", credID, "err", err) + http.Error(w, "credentials unavailable", http.StatusBadGateway) + return + } + w.Header().Set("Content-Type", "application/json") + if err := json.NewEncoder(w).Encode(credResponse{ + RoleArn: roleARN, + AccessKeyID: creds.AccessKeyID, + SecretAccessKey: creds.SecretAccessKey, + Token: creds.SessionToken, + Expiration: creds.Expiration.UTC().Format(time.RFC3339), + }); err != nil { + slog.Warn("ecs-agent: encode credential response", "credId", credID, "err", err) + } +} + +// errUnknownCredID signals a credID with no registered task role. +var errUnknownCredID = fmt.Errorf("unknown credId") + +// fetch returns valid credentials for credID, reusing the cache until within the +// refresh margin and otherwise assuming the task role over the gateway. +func (c *credEndpoint) fetch(ctx context.Context, credID string) (stsauth.Credentials, string, error) { + c.mu.Lock() + roleARN, known := c.roles[credID] + if !known { + c.mu.Unlock() + return stsauth.Credentials{}, "", errUnknownCredID + } + if e, ok := c.cache[credID]; ok && e.roleARN == roleARN && credValid(e.creds, credRefreshMargin) { + c.mu.Unlock() + return e.creds, roleARN, nil + } + c.mu.Unlock() + + creds, err := c.assume(ctx, roleARN, sessionName(credID)) + if err != nil { + return stsauth.Credentials{}, "", err + } + + c.mu.Lock() + c.cache[credID] = credEntry{roleARN: roleARN, creds: creds} + c.mu.Unlock() + return creds, roleARN, nil +} + +// assumeOverGateway mints credentials for roleARN by SigV4-signing AssumeRole +// against the gateway with the agent's instance credentials. +func (c *credEndpoint) assumeOverGateway(ctx context.Context, roleARN, sessionName string) (stsauth.Credentials, error) { + instCreds, err := c.provider.Retrieve(ctx) + if err != nil { + return stsauth.Credentials{}, fmt.Errorf("instance credentials: %w", err) + } + client, err := c.client() + if err != nil { + return stsauth.Credentials{}, err + } + return stsauth.AssumeRole(c.region, c.gatewayURL, client, + instCreds.AccessKeyID, instCreds.SecretAccessKey, instCreds.SessionToken, + roleARN, sessionName) +} + +// client builds the pinned-CA gateway HTTP client on first use and caches it. +func (c *credEndpoint) client() (*http.Client, error) { + c.mu.Lock() + defer c.mu.Unlock() + if c.httpClient != nil { + return c.httpClient, nil + } + hc, err := ecrauth.GatewayHTTPClient(c.caPath) + if err != nil { + return nil, err + } + c.httpClient = hc + return hc, nil +} + +// Start adds the endpoint IP to a dummy interface in the host netns and serves +// HTTP on it. A bind or address failure is returned so the caller can log it +// without aborting register/heartbeat. +func (c *credEndpoint) Start() error { + if err := c.addEndpointAddr(); err != nil { + return err + } + ln, err := net.Listen("tcp", net.JoinHostPort(c.ip, strconv.Itoa(c.port))) + if err != nil { + return fmt.Errorf("listen %s:%d: %w", c.ip, c.port, err) + } + c.ln = ln + c.srv = &http.Server{Handler: c, ReadHeaderTimeout: 5 * time.Second} + go func() { + if err := c.srv.Serve(ln); err != nil && err != http.ErrServerClosed { + slog.Error("ecs-agent: credential endpoint serve", "err", err) + } + }() + slog.Info("ecs-agent: credential endpoint listening", "addr", ln.Addr().String()) + return nil +} + +// Stop shuts the server down and removes the dummy interface. Safe on a nil/ +// never-started endpoint. +func (c *credEndpoint) Stop() error { + if c == nil { + return nil + } + if c.srv != nil { + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + _ = c.srv.Shutdown(ctx) + } + c.delEndpointAddr() + return nil +} + +// addEndpointAddr brings up the dummy interface holding the endpoint IP. Skipped +// for loopback (tests bind 127.0.0.1 directly, no interface plumbing needed). +func (c *credEndpoint) addEndpointAddr() error { + if c.run == nil || c.ip == "127.0.0.1" { + return nil + } + _, _ = c.run.Run("ip", "link", "add", credDummyIface, "type", "dummy") + if _, err := c.run.Run("ip", "addr", "add", c.ip+"/32", "dev", credDummyIface); err != nil && + !strings.Contains(err.Error(), "exists") { + return fmt.Errorf("add endpoint addr %s: %w", c.ip, err) + } + if _, err := c.run.Run("ip", "link", "set", credDummyIface, "up"); err != nil { + return fmt.Errorf("bring up %s: %w", credDummyIface, err) + } + return nil +} + +func (c *credEndpoint) delEndpointAddr() { + if c.run == nil || c.ip == "127.0.0.1" { + return + } + _, _ = c.run.Run("ip", "link", "del", credDummyIface) +} + +// credValid reports whether creds are usable now with margin to spare. +func credValid(c stsauth.Credentials, margin time.Duration) bool { + if c.AccessKeyID == "" || c.SecretAccessKey == "" { + return false + } + if c.Expiration.IsZero() { + return true + } + return time.Now().Add(margin).Before(c.Expiration) +} + +// sessionName derives an AssumeRole RoleSessionName from credID, satisfying the +// STS [\w+=,.@-]{2,64} constraint. +func sessionName(credID string) string { + name := "ecs-" + credID + if len(name) > 64 { + name = name[:64] + } + return name +} diff --git a/cmd/ecs-agent/credendpoint_test.go b/cmd/ecs-agent/credendpoint_test.go new file mode 100644 index 000000000..c62a350a6 --- /dev/null +++ b/cmd/ecs-agent/credendpoint_test.go @@ -0,0 +1,158 @@ +package main + +import ( + "context" + "encoding/json" + "net/http" + "net/http/httptest" + "sync/atomic" + "testing" + "time" + + "github.com/mulgadc/spinifex/internal/stsauth" +) + +// newTestCredEndpoint builds an endpoint whose assume seam returns canned creds, +// counting calls so cache behaviour is observable. run is nil (no iface plumbing). +func newTestCredEndpoint(creds stsauth.Credentials) (*credEndpoint, *int32) { + c := newCredEndpoint(nil, "us-east-1", "https://gw", "", "127.0.0.1", 0, nil) + var calls int32 + c.assume = func(ctx context.Context, roleARN, sessionName string) (stsauth.Credentials, error) { + atomic.AddInt32(&calls, 1) + return creds, nil + } + return c, &calls +} + +func TestCredEndpoint_ServesRegisteredCredID(t *testing.T) { + exp := time.Now().Add(time.Hour).UTC().Truncate(time.Second) + c, _ := newTestCredEndpoint(stsauth.Credentials{ + AccessKeyID: "AKIA", SecretAccessKey: "secret", SessionToken: "tok", Expiration: exp, + }) + c.Register("cred-1", "arn:aws:iam::111122223333:role/task") + + rr := httptest.NewRecorder() + c.ServeHTTP(rr, httptest.NewRequest(http.MethodGet, "/v2/credentials/cred-1", nil)) + if rr.Code != http.StatusOK { + t.Fatalf("status = %d, want 200; body=%s", rr.Code, rr.Body.String()) + } + var got credResponse + if err := json.Unmarshal(rr.Body.Bytes(), &got); err != nil { + t.Fatalf("decode: %v", err) + } + if got.AccessKeyID != "AKIA" || got.SecretAccessKey != "secret" || got.Token != "tok" { + t.Errorf("bad creds in response: %+v", got) + } + if got.RoleArn != "arn:aws:iam::111122223333:role/task" { + t.Errorf("RoleArn = %q", got.RoleArn) + } + if got.Expiration != exp.Format(time.RFC3339) { + t.Errorf("Expiration = %q, want %q", got.Expiration, exp.Format(time.RFC3339)) + } +} + +func TestCredEndpoint_UnknownCredIDIs404(t *testing.T) { + c, calls := newTestCredEndpoint(stsauth.Credentials{AccessKeyID: "x", SecretAccessKey: "y"}) + rr := httptest.NewRecorder() + c.ServeHTTP(rr, httptest.NewRequest(http.MethodGet, "/v2/credentials/nope", nil)) + if rr.Code != http.StatusNotFound { + t.Fatalf("status = %d, want 404", rr.Code) + } + if *calls != 0 { + t.Errorf("assume called %d times for unknown credID, want 0", *calls) + } +} + +func TestCredEndpoint_WrongMethodIs405(t *testing.T) { + c, _ := newTestCredEndpoint(stsauth.Credentials{AccessKeyID: "x", SecretAccessKey: "y"}) + c.Register("cred-1", "arn:role") + rr := httptest.NewRecorder() + c.ServeHTTP(rr, httptest.NewRequest(http.MethodPost, "/v2/credentials/cred-1", nil)) + if rr.Code != http.StatusMethodNotAllowed { + t.Fatalf("status = %d, want 405", rr.Code) + } +} + +func TestCredEndpoint_CachesUntilExpiry(t *testing.T) { + c, calls := newTestCredEndpoint(stsauth.Credentials{ + AccessKeyID: "AKIA", SecretAccessKey: "secret", Expiration: time.Now().Add(time.Hour), + }) + c.Register("cred-1", "arn:role") + for hit := range 3 { + rr := httptest.NewRecorder() + c.ServeHTTP(rr, httptest.NewRequest(http.MethodGet, "/v2/credentials/cred-1", nil)) + if rr.Code != http.StatusOK { + t.Fatalf("hit %d: status %d", hit, rr.Code) + } + } + if *calls != 1 { + t.Errorf("assume called %d times, want 1 (cached)", *calls) + } +} + +func TestCredEndpoint_RefreshesNearExpiry(t *testing.T) { + // Expiry inside the refresh margin -> every hit re-assumes. + c, calls := newTestCredEndpoint(stsauth.Credentials{ + AccessKeyID: "AKIA", SecretAccessKey: "secret", Expiration: time.Now().Add(time.Minute), + }) + c.Register("cred-1", "arn:role") + for range 2 { + rr := httptest.NewRecorder() + c.ServeHTTP(rr, httptest.NewRequest(http.MethodGet, "/v2/credentials/cred-1", nil)) + } + if *calls != 2 { + t.Errorf("assume called %d times, want 2 (expiry within margin)", *calls) + } +} + +func TestCredEndpoint_DeregisterDropsCredID(t *testing.T) { + c, _ := newTestCredEndpoint(stsauth.Credentials{AccessKeyID: "x", SecretAccessKey: "y"}) + c.Register("cred-1", "arn:role") + c.Deregister("cred-1") + rr := httptest.NewRecorder() + c.ServeHTTP(rr, httptest.NewRequest(http.MethodGet, "/v2/credentials/cred-1", nil)) + if rr.Code != http.StatusNotFound { + t.Fatalf("status = %d, want 404 after deregister", rr.Code) + } +} + +func TestCredEndpoint_AssumeFailureIs502(t *testing.T) { + c, _ := newTestCredEndpoint(stsauth.Credentials{}) + c.assume = func(ctx context.Context, roleARN, sessionName string) (stsauth.Credentials, error) { + return stsauth.Credentials{}, context.DeadlineExceeded + } + c.Register("cred-1", "arn:role") + rr := httptest.NewRecorder() + c.ServeHTTP(rr, httptest.NewRequest(http.MethodGet, "/v2/credentials/cred-1", nil)) + if rr.Code != http.StatusBadGateway { + t.Fatalf("status = %d, want 502", rr.Code) + } +} + +func TestCredEndpoint_StartServesOverLoopback(t *testing.T) { + c, _ := newTestCredEndpoint(stsauth.Credentials{ + AccessKeyID: "AKIA", SecretAccessKey: "secret", Expiration: time.Now().Add(time.Hour), + }) + c.Register("cred-1", "arn:role") + c.port = 0 // bind an ephemeral port instead of privileged 80 + if err := c.Start(); err != nil { + t.Fatalf("start: %v", err) + } + defer c.Stop() + + url := "http://" + c.ln.Addr().String() + credRelativeURI("cred-1") + resp, err := http.Get(url) + if err != nil { + t.Fatalf("get: %v", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + t.Fatalf("status = %d, want 200", resp.StatusCode) + } +} + +func TestCredRelativeURI(t *testing.T) { + if got := credRelativeURI("abc"); got != "/v2/credentials/abc" { + t.Errorf("credRelativeURI = %q", got) + } +} diff --git a/cmd/ecs-agent/credinject_test.go b/cmd/ecs-agent/credinject_test.go new file mode 100644 index 000000000..b69dc1867 --- /dev/null +++ b/cmd/ecs-agent/credinject_test.go @@ -0,0 +1,55 @@ +package main + +import ( + "testing" + + "github.com/mulgadc/spinifex/spinifex/handlers/ecs/bus" +) + +func TestTaskCredID(t *testing.T) { + cases := []struct { + name string + as bus.Assign + want string + }{ + {"no role", bus.Assign{TaskID: "t1"}, ""}, + {"role, no credID -> taskID", bus.Assign{TaskID: "t1", TaskRoleARN: "arn:role"}, "t1"}, + {"role + credID", bus.Assign{TaskID: "t1", CredID: "c9", TaskRoleARN: "arn:role"}, "c9"}, + } + for _, tc := range cases { + if got := taskCredID(&tc.as); got != tc.want { + t.Errorf("%s: taskCredID = %q, want %q", tc.name, got, tc.want) + } + } +} + +func TestWithCredEnv(t *testing.T) { + base := map[string]string{"FOO": "bar"} + + // No credID -> env returned unchanged (same map). + if got := withCredEnv(base, ""); got["FOO"] != "bar" || got["AWS_CONTAINER_CREDENTIALS_RELATIVE_URI"] != "" { + t.Errorf("blank credID mutated env: %v", got) + } + + got := withCredEnv(base, "cred-1") + if got["AWS_CONTAINER_CREDENTIALS_RELATIVE_URI"] != "/v2/credentials/cred-1" { + t.Errorf("missing relative-uri: %v", got) + } + if got["FOO"] != "bar" { + t.Errorf("dropped existing env: %v", got) + } + // Original map must be untouched (copy semantics). + if _, ok := base["AWS_CONTAINER_CREDENTIALS_RELATIVE_URI"]; ok { + t.Errorf("source env mutated: %v", base) + } +} + +func TestSessionName(t *testing.T) { + if got := sessionName("abc"); got != "ecs-abc" { + t.Errorf("sessionName = %q", got) + } + long := sessionName(string(make([]byte, 100))) + if len(long) != 64 { + t.Errorf("sessionName not truncated to 64: len=%d", len(long)) + } +} diff --git a/cmd/ecs-agent/netns.go b/cmd/ecs-agent/netns.go index 17c0cadf4..2097f8c9b 100644 --- a/cmd/ecs-agent/netns.go +++ b/cmd/ecs-agent/netns.go @@ -2,6 +2,7 @@ package main import ( "fmt" + "hash/crc32" "os/exec" "strings" "time" @@ -11,6 +12,15 @@ import ( // path containerd joins via the OCI network namespace path. const netnsRunDir = "/var/run/netns" +// Credentials path constants: each task netns gets a /30 veth into the host netns +// from 169.254.172.0/22 and a host-routed /32 to the credential endpoint, keeping +// the endpoint off the task ENI's VPC route. credTaskVeth is the task-side name +// (unique per netns); the host side is per-task (credVethName). +const ( + credTaskVeth = "ecsc0" + credEndpointCIDR = defaultCredEndpointIP + "/32" +) + // netCmdRunner executes a host networking command. Abstracted so unit tests can // assert the ip/netns sequence without touching the kernel. type netCmdRunner interface { @@ -56,11 +66,22 @@ func (n *taskNetns) Setup(taskID, mac string) (string, error) { if _, err := n.run.Run("ip", "netns", "add", name); err != nil { return "", fmt.Errorf("netns add %s: %w", name, err) } + hostVeth := credVethName(taskID) + hostIP, taskIP := credSubnet(taskID) steps := [][]string{ {"ip", "link", "set", iface, "netns", name}, {"ip", "-n", name, "link", "set", "lo", "up"}, {"ip", "-n", name, "link", "set", iface, "up"}, {"ip", "netns", "exec", name, "udhcpc", "-i", iface, "-q", "-n"}, + // Credentials path: a veth into the host netns plus a host-routed /32 to the + // credential endpoint, so the container reaches 169.254.170.2 without it + // leaking onto the task ENI's VPC route. + {"ip", "link", "add", hostVeth, "type", "veth", "peer", "name", credTaskVeth, "netns", name}, + {"ip", "addr", "add", hostIP + "/30", "dev", hostVeth}, + {"ip", "link", "set", hostVeth, "up"}, + {"ip", "-n", name, "addr", "add", taskIP + "/30", "dev", credTaskVeth}, + {"ip", "-n", name, "link", "set", credTaskVeth, "up"}, + {"ip", "-n", name, "route", "add", credEndpointCIDR, "via", hostIP}, } for _, s := range steps { if _, err := n.run.Run(s[0], s[1:]...); err != nil { @@ -71,10 +92,31 @@ func (n *taskNetns) Setup(taskID, mac string) (string, error) { return netnsPathFor(taskID), nil } +// credVethName is the host-side veth name for a task's credential path. Derived +// from the taskID and kept within the 15-char interface-name limit. +func credVethName(taskID string) string { + return fmt.Sprintf("ecsv%08x", crc32.ChecksumIEEE([]byte(taskID))) +} + +// credSubnet derives a unique /30 in 169.254.172.0/22 for a task's credential +// veth, returning the host-side and task-side addresses. The host side is the +// container's gateway to the credential endpoint. +func credSubnet(taskID string) (hostIP, taskIP string) { + slot := crc32.ChecksumIEEE([]byte(taskID)) % 256 // 256 /30s in the /22 + third := 172 + (slot*4)/256 + base := (slot * 4) % 256 + hostIP = fmt.Sprintf("169.254.%d.%d", third, base+1) + taskIP = fmt.Sprintf("169.254.%d.%d", third, base+2) + return hostIP, taskIP +} + // Teardown deletes the task netns (releasing the moved NIC). Idempotent enough // for the stop path: a missing netns is reported but not fatal to the caller. func (n *taskNetns) Teardown(taskID string) error { name := netnsName(taskID) + // Deleting the netns destroys the task-side veth and its host peer; remove the + // host side explicitly too so a leaked half from a partial setup is reaped. + _, _ = n.run.Run("ip", "link", "del", credVethName(taskID)) if _, err := n.run.Run("ip", "netns", "del", name); err != nil { return fmt.Errorf("netns del %s: %w", name, err) } diff --git a/cmd/ecs-agent/netns_test.go b/cmd/ecs-agent/netns_test.go index cb4a25019..4e77ab215 100644 --- a/cmd/ecs-agent/netns_test.go +++ b/cmd/ecs-agent/netns_test.go @@ -83,12 +83,18 @@ func TestNetns_SetupSequence(t *testing.T) { if path != netnsPathFor("t-001") { t.Fatalf("want %q, got %q", netnsPathFor("t-001"), path) } + hostIP, taskIP := credSubnet("t-001") + hostVeth := credVethName("t-001") want := []string{ "ip netns add ecs-t-001", "ip link set eth1 netns ecs-t-001", "ip -n ecs-t-001 link set lo up", "ip -n ecs-t-001 link set eth1 up", "ip netns exec ecs-t-001 udhcpc -i eth1 -q -n", + "ip link add " + hostVeth + " type veth peer name ecsc0 netns ecs-t-001", + "ip addr add " + hostIP + "/30 dev " + hostVeth, + "ip -n ecs-t-001 addr add " + taskIP + "/30 dev ecsc0", + "ip -n ecs-t-001 route add 169.254.170.2/32 via " + hostIP, } for _, w := range want { if !f.sawAny(w) { @@ -97,6 +103,23 @@ func TestNetns_SetupSequence(t *testing.T) { } } +func TestCredSubnet_UniquePer30(t *testing.T) { + hostA, taskA := credSubnet("task-a") + hostB, taskB := credSubnet("task-b") + if hostA == hostB { + t.Errorf("distinct tasks share a host IP: %s", hostA) + } + for _, ip := range []string{hostA, taskA, hostB, taskB} { + if !strings.HasPrefix(ip, "169.254.17") { + t.Errorf("cred IP %s outside 169.254.172.0/22", ip) + } + } + // host and task ends of one task sit in the same /30 (differ in last octet). + if hostA[:strings.LastIndexByte(hostA, '.')] != taskA[:strings.LastIndexByte(taskA, '.')] { + t.Errorf("host %s and task %s not in same /30", hostA, taskA) + } +} + func TestNetns_SetupNICNotFound(t *testing.T) { f := &fakeNetRunner{linkOut: linkWithENI} n := newTestNetns(f) diff --git a/cmd/ecs-agent/taskrun.go b/cmd/ecs-agent/taskrun.go index 10e845d67..174d4b98e 100644 --- a/cmd/ecs-agent/taskrun.go +++ b/cmd/ecs-agent/taskrun.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "log/slog" + "maps" "time" ctrruntime "github.com/mulgadc/spinifex/cmd/ecs-agent/runtime" @@ -62,6 +63,8 @@ func (a *Agent) runTask(ctx context.Context, as *bus.Assign) { return } + credID := a.registerTaskCreds(as) + statuses := make([]bus.ContainerStatus, 0, len(as.Containers)) for _, c := range as.Containers { if _, err := a.puller.Pull(ctx, ctrruntime.PullSpec{Ref: c.Image}, a.resolver); err != nil { @@ -75,7 +78,7 @@ func (a *Agent) runTask(ctx context.Context, as *bus.Assign) { spec := ctrruntime.RunSpec{ Image: c.Image, Command: c.Command, - Env: c.Environment, + Env: withCredEnv(c.Environment, credID), Labels: taskLabels(as, c.Name), NetnsPath: netnsPath, } @@ -118,8 +121,13 @@ func (a *Agent) setupTaskNetns(as *bus.Assign) (string, error) { return a.netns.Setup(as.TaskID, as.ENIMacAddress) } -// teardownTaskNetns removes the awsvpc task netns; no-op for bridge/host tasks. +// teardownTaskNetns releases the task's IAM credential registration, then removes +// the awsvpc task netns (the latter is a no-op for bridge/host tasks). It runs at +// every task-stop path so credentials never outlive the task that owns them. func (a *Agent) teardownTaskNetns(as *bus.Assign) { + if a.cred != nil { + a.cred.Deregister(taskCredID(as)) + } if as.ENIMacAddress == "" || a.netns == nil { return } @@ -128,6 +136,43 @@ func (a *Agent) teardownTaskNetns(as *bus.Assign) { } } +// registerTaskCreds registers the task's credID -> role mapping with the +// credential endpoint and returns the credID (empty when the task has no role). +func (a *Agent) registerTaskCreds(as *bus.Assign) string { + credID := taskCredID(as) + if credID == "" || a.cred == nil { + return credID + } + a.cred.Register(credID, as.TaskRoleARN) + return credID +} + +// taskCredID is the credential ID a task's containers fetch credentials under: +// the scheduler-assigned CredID, falling back to the taskID. Empty when the task +// carries no IAM role. +func taskCredID(as *bus.Assign) string { + if as.TaskRoleARN == "" { + return "" + } + if as.CredID != "" { + return as.CredID + } + return as.TaskID +} + +// withCredEnv returns env with AWS_CONTAINER_CREDENTIALS_RELATIVE_URI set for +// credID, copying the map so the assign's container env is left untouched. A +// blank credID (no task role) returns env unchanged. +func withCredEnv(env map[string]string, credID string) map[string]string { + if credID == "" { + return env + } + out := make(map[string]string, len(env)+1) + maps.Copy(out, env) + out["AWS_CONTAINER_CREDENTIALS_RELATIVE_URI"] = credRelativeURI(credID) + return out +} + // reportTaskState reports a task transition through the gateway's // SubmitTaskStateChange action. func (a *Agent) reportTaskState(as *bus.Assign, status, reason string, containers []bus.ContainerStatus) { diff --git a/internal/stsauth/stsauth.go b/internal/stsauth/stsauth.go new file mode 100644 index 000000000..008a93e42 --- /dev/null +++ b/internal/stsauth/stsauth.go @@ -0,0 +1,70 @@ +// Package stsauth mints task-role credentials by SigV4-signing sts:AssumeRole +// against the Spinifex AWS gateway. It mirrors internal/ecrauth: the in-guest +// ecs-agent already holds instance-role credentials and signs gateway calls with +// them, so it assumes a task's IAM role the same way rather than reaching STS +// over NATS (the agent has no NATS connection). +package stsauth + +import ( + "fmt" + "net/http" + "time" + + "github.com/aws/aws-sdk-go/aws" + awscreds "github.com/aws/aws-sdk-go/aws/credentials" + "github.com/aws/aws-sdk-go/aws/session" + "github.com/aws/aws-sdk-go/service/sts" +) + +// defaultRegion is used when the caller supplies no region; SigV4 still needs a +// scope and the gateway does not enforce a specific value. +const defaultRegion = "us-east-1" + +// Credentials is the decoded result of AssumeRole: a SigV4 credential set bound +// to the assumed role, with its expiry. +type Credentials struct { + AccessKeyID string + SecretAccessKey string + SessionToken string + Expiration time.Time +} + +// AssumeRole builds an aws-sdk STS client pointed at the gateway and calls +// AssumeRole with the supplied static (instance-role) credentials, returning the +// assumed-role credential set. httpClient must trust the gateway CA. +func AssumeRole(region, gatewayURL string, httpClient *http.Client, akid, secret, sessionToken, roleARN, sessionName string) (Credentials, error) { + if region == "" { + region = defaultRegion + } + if roleARN == "" || sessionName == "" { + return Credentials{}, fmt.Errorf("stsauth: roleARN and sessionName are required") + } + sess, err := session.NewSession(&aws.Config{ + Region: aws.String(region), + Endpoint: aws.String(gatewayURL), + Credentials: awscreds.NewStaticCredentials(akid, secret, sessionToken), + HTTPClient: httpClient, + DisableSSL: aws.Bool(false), + }) + if err != nil { + return Credentials{}, fmt.Errorf("new session: %w", err) + } + + out, err := sts.New(sess).AssumeRole(&sts.AssumeRoleInput{ + RoleArn: aws.String(roleARN), + RoleSessionName: aws.String(sessionName), + }) + if err != nil { + return Credentials{}, fmt.Errorf("AssumeRole: %w", err) + } + if out.Credentials == nil { + return Credentials{}, fmt.Errorf("AssumeRole returned no credentials") + } + c := out.Credentials + return Credentials{ + AccessKeyID: aws.StringValue(c.AccessKeyId), + SecretAccessKey: aws.StringValue(c.SecretAccessKey), + SessionToken: aws.StringValue(c.SessionToken), + Expiration: aws.TimeValue(c.Expiration), + }, nil +} diff --git a/internal/stsauth/stsauth_test.go b/internal/stsauth/stsauth_test.go new file mode 100644 index 000000000..9bbf6406b --- /dev/null +++ b/internal/stsauth/stsauth_test.go @@ -0,0 +1,63 @@ +package stsauth + +import ( + "io" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" +) + +// assumeRoleXML is a minimal AWS AssumeRole response the aws-sdk STS client +// decodes into Credentials. +const assumeRoleXML = ` + + + ASIATESTKEY + testsecret + testtoken + 2031-01-02T03:04:05Z + + +` + +func TestAssumeRole_DecodesCredentials(t *testing.T) { + var gotBody string + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + b, _ := io.ReadAll(r.Body) + gotBody = string(b) + w.Header().Set("Content-Type", "text/xml") + _, _ = w.Write([]byte(assumeRoleXML)) + })) + defer srv.Close() + + creds, err := AssumeRole("us-east-1", srv.URL, srv.Client(), + "AKIAINSTANCE", "instancesecret", "", "arn:aws:iam::111122223333:role/task", "ecs-cred-1") + if err != nil { + t.Fatalf("AssumeRole: %v", err) + } + if creds.AccessKeyID != "ASIATESTKEY" || creds.SecretAccessKey != "testsecret" || creds.SessionToken != "testtoken" { + t.Errorf("bad creds: %+v", creds) + } + want, _ := time.Parse(time.RFC3339, "2031-01-02T03:04:05Z") + if !creds.Expiration.Equal(want) { + t.Errorf("Expiration = %v, want %v", creds.Expiration, want) + } + // The request must carry the AssumeRole action and the role ARN we asked for. + if !strings.Contains(gotBody, "Action=AssumeRole") { + t.Errorf("request not an AssumeRole call: %s", gotBody) + } + if !strings.Contains(gotBody, "role%2Ftask") && !strings.Contains(gotBody, "role/task") { + t.Errorf("request missing role ARN: %s", gotBody) + } +} + +func TestAssumeRole_RequiresRoleAndSession(t *testing.T) { + if _, err := AssumeRole("us-east-1", "https://gw", http.DefaultClient, "a", "b", "", "", "sess"); err == nil { + t.Error("want error for empty roleARN") + } + if _, err := AssumeRole("us-east-1", "https://gw", http.DefaultClient, "a", "b", "", "arn:role", ""); err == nil { + t.Error("want error for empty sessionName") + } +} diff --git a/spinifex/handlers/ecs/bus/messages.go b/spinifex/handlers/ecs/bus/messages.go index 6f153c467..6835bc7d6 100644 --- a/spinifex/handlers/ecs/bus/messages.go +++ b/spinifex/handlers/ecs/bus/messages.go @@ -86,8 +86,9 @@ type Assign struct { ENIMacAddress string `json:"eniMac,omitempty"` ENIPrivateIP string `json:"eniPrivateIp,omitempty"` ENISubnetID string `json:"eniSubnetId,omitempty"` - // CredID + TaskRoleARN are placeholders for the Sprint 4g IAM credential - // endpoint; the agent ignores them in 4e. + // TaskRoleARN is the task's IAM role (from the taskdef); when set, the agent + // serves its credentials at 169.254.170.2 under CredID, defaulting CredID to + // the taskID when the scheduler leaves it empty. CredID string `json:"credId,omitempty"` TaskRoleARN string `json:"taskRoleArn,omitempty"` AssignedAt time.Time `json:"assignedAt"` diff --git a/spinifex/handlers/ecs/records.go b/spinifex/handlers/ecs/records.go index a43cf974c..04fbb018c 100644 --- a/spinifex/handlers/ecs/records.go +++ b/spinifex/handlers/ecs/records.go @@ -101,6 +101,7 @@ type TaskDefRecord struct { NetworkMode string `json:"networkMode,omitempty"` CPU string `json:"cpu,omitempty"` Memory string `json:"memory,omitempty"` + TaskRoleArn string `json:"taskRoleArn,omitempty"` Containers []ContainerDef `json:"containers"` Status string `json:"status"` RegisteredAt time.Time `json:"registeredAt"` diff --git a/spinifex/handlers/ecs/service.go b/spinifex/handlers/ecs/service.go index 18db20cc1..8a8a9ed14 100644 --- a/spinifex/handlers/ecs/service.go +++ b/spinifex/handlers/ecs/service.go @@ -263,6 +263,7 @@ func (s *Service) RegisterTaskDefinition(input *ecs.RegisterTaskDefinitionInput, NetworkMode: aws.StringValue(input.NetworkMode), CPU: aws.StringValue(input.Cpu), Memory: aws.StringValue(input.Memory), + TaskRoleArn: aws.StringValue(input.TaskRoleArn), Status: TaskDefStatusActive, RegisteredAt: time.Now().UTC(), Containers: containerDefsFromAWS(input.ContainerDefinitions), @@ -396,6 +397,9 @@ func (r *TaskDefRecord) toAWS() *ecs.TaskDefinition { if r.Memory != "" { td.Memory = aws.String(r.Memory) } + if r.TaskRoleArn != "" { + td.TaskRoleArn = aws.String(r.TaskRoleArn) + } for _, c := range r.Containers { td.ContainerDefinitions = append(td.ContainerDefinitions, c.toAWS()) } diff --git a/spinifex/handlers/ecs/service_test.go b/spinifex/handlers/ecs/service_test.go index 917b77b5d..334bac00f 100644 --- a/spinifex/handlers/ecs/service_test.go +++ b/spinifex/handlers/ecs/service_test.go @@ -155,6 +155,38 @@ func TestService_RunTask_PlacesAndAssigns(t *testing.T) { assert.Equal(t, as.TaskID, containerInstanceShortID(aws.StringValue(out.Tasks[0].TaskArn))) } +func TestService_RunTask_AssignCarriesTaskRole(t *testing.T) { + svc, _ := newTestService(t) + _, err := svc.CreateCluster(&ecs.CreateClusterInput{ClusterName: aws.String("web")}, testAccountID) + require.NoError(t, err) + roleARN := "arn:aws:iam::123456789012:role/task-app" + _, err = svc.RegisterTaskDefinition(&ecs.RegisterTaskDefinitionInput{ + Family: aws.String("app"), + TaskRoleArn: aws.String(roleARN), + ContainerDefinitions: []*ecs.ContainerDefinition{{ + Name: aws.String("app"), Image: aws.String("registry/app:1"), + Cpu: aws.Int64(128), Memory: aws.Int64(256), Essential: aws.Bool(true), + }}, + }, testAccountID) + require.NoError(t, err) + registerInstance(t, svc, "web", "i-1", 1024, 2048) + + // Round-trips through DescribeTaskDefinition. + d, err := svc.DescribeTaskDefinition(&ecs.DescribeTaskDefinitionInput{TaskDefinition: aws.String("app")}, testAccountID) + require.NoError(t, err) + assert.Equal(t, roleARN, aws.StringValue(d.TaskDefinition.TaskRoleArn)) + + _, err = svc.RunTask(&ecs.RunTaskInput{ + Cluster: aws.String("web"), TaskDefinition: aws.String("app"), Count: aws.Int64(1), + }, testAccountID) + require.NoError(t, err) + + poll, err := svc.PollAssignments(&PollAssignmentsInput{Cluster: "web", ContainerInstance: "i-1"}, testAccountID) + require.NoError(t, err) + require.Len(t, poll.Assignments, 1) + assert.Equal(t, roleARN, poll.Assignments[0].TaskRoleARN) +} + // PollAssignments is at-least-once: re-poll without ack re-delivers the assign; // acking it (then STOPPED) drains the inbox so it is never re-delivered. func TestService_PollAssignments_AckAndReclaim(t *testing.T) { diff --git a/spinifex/handlers/ecs/tasks.go b/spinifex/handlers/ecs/tasks.go index 174f39750..9ab0131f2 100644 --- a/spinifex/handlers/ecs/tasks.go +++ b/spinifex/handlers/ecs/tasks.go @@ -199,6 +199,7 @@ func (s *Service) publishAssign(kv nats.KeyValue, accountID, cluster, instanceID TaskARN: rec.ARN, TaskDefFamily: td.Family, TaskDefRevision: td.Revision, + TaskRoleARN: td.TaskRoleArn, ENIID: rec.ENIID, ENIMacAddress: rec.ENIMacAddress, ENIPrivateIP: rec.ENIPrivateIP,