Compare commits

..

2 Commits

9 changed files with 367 additions and 17 deletions
+2
View File
@@ -578,6 +578,8 @@ jobs:
echo \"::group::suite ${SUITE}\"; \
date -u +%Y-%m-%dT%H:%M:%SZ > '${ARTIFACT_DIR}/test-${SUITE}.start'; \
GITHUB_ACTIONS=true FORCE_COLOR=1 SPINIFEX_E2E=1 SPINIFEX_MODE=single \
SPINIFEX_NODES='${{ steps.vm.outputs.wan_ip }}' SPINIFEX_SSH_USER=tf-user \
SPINIFEX_SSH_KEY=\$HOME/.ssh/tf-user-ap-southeast-2 \
ARTIFACT_DIR='${ARTIFACT_DIR}' \
./${SUITE}.test -test.v -test.timeout=${TMO} \
2>&1 | tee '${ARTIFACT_DIR}/test-${SUITE}.log'; \
+65
View File
@@ -0,0 +1,65 @@
//go:build lambdatestlink
package main
import (
"crypto/x509"
"errors"
"fmt"
"net"
"net/url"
"os"
"path/filepath"
)
const (
sdkEndpointEntry = "test-sdk-endpoint"
sdkCAEntry = "test-sdk-ca"
sdkBundlePath = "/run/lambda-sdk/ca-bundle.pem"
systemBundlePath = "/etc/ssl/certs/ca-certificates.crt"
)
type sdkConfig struct {
endpoint string
ca []byte
}
// readSDKConfig reads optional public settings before the boot config is sealed.
func readSDKConfig(dir string) (*sdkConfig, error) {
endpoint, endpointErr := os.ReadFile(filepath.Join(dir, sdkEndpointEntry, "raw"))
ca, caErr := os.ReadFile(filepath.Join(dir, sdkCAEntry, "raw"))
if errors.Is(endpointErr, os.ErrNotExist) && errors.Is(caErr, os.ErrNotExist) {
return nil, nil
}
if endpointErr != nil || caErr != nil {
return nil, errors.New("incomplete SDK gateway settings")
}
u, err := url.Parse(string(endpoint))
if err != nil || u.Scheme != "https" || u.User != nil || u.Path != "" || u.RawQuery != "" || u.Fragment != "" ||
net.ParseIP(u.Hostname()) == nil || u.Port() == "" {
return nil, errors.New("invalid SDK gateway endpoint")
}
if len(ca) > 64<<10 || !x509.NewCertPool().AppendCertsFromPEM(ca) {
return nil, errors.New("invalid SDK gateway CA")
}
return &sdkConfig{endpoint: string(endpoint), ca: ca}, nil
}
// writeBundle combines the system roots with the cluster's public root.
func (c *sdkConfig) writeBundle(path, systemPath string) error {
system, err := os.ReadFile(systemPath)
if err != nil && !errors.Is(err, os.ErrNotExist) {
return fmt.Errorf("read system CA bundle: %w", err)
}
if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
return fmt.Errorf("create SDK CA directory: %w", err)
}
if err := os.Chmod(filepath.Dir(path), 0o755); err != nil {
return fmt.Errorf("make SDK CA directory readable: %w", err)
}
bundle := append(append(system, '\n'), c.ca...)
if err := os.WriteFile(path, bundle, 0o644); err != nil {
return fmt.Errorf("write SDK CA bundle: %w", err)
}
return os.Chmod(path, 0o644)
}
@@ -0,0 +1,106 @@
//go:build lambdatestlink
//test:in-package exercises the unexported boot configuration reader
package main
import (
"crypto/ecdsa"
"crypto/elliptic"
"crypto/rand"
"crypto/tls"
"crypto/x509"
"crypto/x509/pkix"
"encoding/pem"
"math/big"
"net/http"
"net/http/httptest"
"os"
"path/filepath"
"testing"
"time"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func TestSDKConfig_OptionalAndInvalid(t *testing.T) {
server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusOK) }))
defer server.Close()
ca := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: server.Certificate().Raw})
for _, tc := range []struct {
name string
endpoint []byte
ca []byte
wantErr bool
want bool
}{
{name: "absent"},
{name: "valid", endpoint: []byte(server.URL), ca: ca, want: true},
{name: "missing ca", endpoint: []byte(server.URL), wantErr: true},
{name: "missing endpoint", ca: ca, wantErr: true},
{name: "invalid endpoint", endpoint: []byte("http://127.0.0.1:1234"), ca: ca, wantErr: true},
{name: "invalid ca", endpoint: []byte(server.URL), ca: []byte("not a certificate"), wantErr: true},
} {
t.Run(tc.name, func(t *testing.T) {
dir := t.TempDir()
for name, value := range map[string][]byte{sdkEndpointEntry: tc.endpoint, sdkCAEntry: tc.ca} {
if value != nil {
require.NoError(t, os.MkdirAll(filepath.Join(dir, name), 0o700))
require.NoError(t, os.WriteFile(filepath.Join(dir, name, "raw"), value, 0o600))
}
}
cfg, err := readSDKConfig(dir)
if tc.wantErr {
require.Error(t, err)
require.Nil(t, cfg)
return
}
require.NoError(t, err)
if !tc.want {
require.Nil(t, cfg)
return
}
bundle := filepath.Join(t.TempDir(), "sdk", "ca.pem")
require.NoError(t, cfg.writeBundle(bundle, filepath.Join(t.TempDir(), "absent")))
data, err := os.ReadFile(bundle)
require.NoError(t, err)
roots := x509.NewCertPool()
require.True(t, roots.AppendCertsFromPEM(data))
client := &http.Client{Transport: &http.Transport{TLSClientConfig: &tls.Config{RootCAs: roots}}}
resp, err := client.Get(cfg.endpoint)
require.NoError(t, err)
defer resp.Body.Close()
assert.Equal(t, http.StatusOK, resp.StatusCode)
})
}
}
func TestSDKConfig_BundleKeepsSystemRoots(t *testing.T) {
cluster := httptest.NewTLSServer(nil)
defer cluster.Close()
clusterCA := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: cluster.Certificate().Raw})
key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
require.NoError(t, err)
template := &x509.Certificate{SerialNumber: big.NewInt(1), Subject: pkix.Name{CommonName: "test system root"},
NotBefore: time.Now().Add(-time.Minute), NotAfter: time.Now().Add(time.Hour),
IsCA: true, BasicConstraintsValid: true, KeyUsage: x509.KeyUsageCertSign}
der, err := x509.CreateCertificate(rand.Reader, template, template, &key.PublicKey, key)
require.NoError(t, err)
systemCA := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der})
dir := t.TempDir()
systemPath, bundlePath := filepath.Join(dir, "system.pem"), filepath.Join(dir, "sdk", "ca.pem")
require.NoError(t, os.WriteFile(systemPath, systemCA, 0o600))
require.NoError(t, (&sdkConfig{ca: clusterCA}).writeBundle(bundlePath, systemPath))
bundle, err := os.ReadFile(bundlePath)
require.NoError(t, err)
roots := x509.NewCertPool()
require.True(t, roots.AppendCertsFromPEM(bundle))
client := &http.Client{Transport: &http.Transport{TLSClientConfig: &tls.Config{RootCAs: roots}}}
resp, err := client.Get(cluster.URL)
require.NoError(t, err)
require.NoError(t, resp.Body.Close())
systemCert, err := x509.ParseCertificate(der)
require.NoError(t, err)
_, err = systemCert.Verify(x509.VerifyOptions{Roots: roots})
require.NoError(t, err)
}
+29 -8
View File
@@ -12,6 +12,7 @@ import (
"net/http"
"os"
"os/exec"
"strings"
"time"
"github.com/mulgadc/spinifex/spinifex/agents/lambda/agent"
@@ -31,13 +32,17 @@ type testlink struct {
pkg taskPackage
runtimeAddr string
startBootstrap func(ctx context.Context, env []string) (*exec.Cmd, error)
caBundlePath string
systemCAPath string
}
func start(ctx context.Context) int {
t := testlink{
startup: defaultStartup(),
pkg: devTaskPackage(),
runtimeAddr: runtimeAddr,
startup: defaultStartup(),
pkg: devTaskPackage(),
runtimeAddr: runtimeAddr,
caBundlePath: sdkBundlePath,
systemCAPath: systemBundlePath,
startBootstrap: func(ctx context.Context, env []string) (*exec.Cmd, error) {
return agent.StartBootstrap(ctx, taskRoot+"/bootstrap", env, os.Stdout, os.Stderr)
},
@@ -53,6 +58,7 @@ func (t testlink) run(parent context.Context) (int, error) {
// The package values are fw_cfg entries, so read them before connect seals it.
spec, specErr := readPackageSpec(t.fwCfgDir)
grant, grantErr := readGrant(t.fwCfgDir)
sdk, sdkErr := readSDKConfig(t.fwCfgDir)
conn, scope, code, err := t.connect(parent)
if err != nil {
return code, errors.Join(err, specErr, grantErr)
@@ -85,6 +91,15 @@ func (t testlink) run(parent context.Context) (int, error) {
}
cred := *first.Credential
first.Credential = nil
if sdkErr != nil {
slog.Warn("lambda-agent: SDK gateway settings unavailable", "err", sdkErr)
}
if sdk != nil {
if err := sdk.writeBundle(t.caBundlePath, t.systemCAPath); err != nil {
slog.Warn("lambda-agent: SDK gateway settings unavailable", "err", err)
sdk = nil
}
}
if err := t.pkg.install(spec); err != nil {
return exitRefused, fmt.Errorf("install package: %w", err)
}
@@ -100,7 +115,8 @@ func (t testlink) run(parent context.Context) (int, error) {
ctx, cancel := context.WithCancelCause(parent)
defer cancel(nil)
cmd, err := t.startBootstrap(ctx, bootstrapEnv(ln.Addr().String(), spec.handler, cred))
region, _, _ := strings.Cut(string(scope), "/")
cmd, err := t.startBootstrap(ctx, bootstrapEnv(ln.Addr().String(), spec.handler, cred, region, sdk, t.caBundlePath))
if err != nil {
return exitLink, fmt.Errorf("start bootstrap: %w", err)
}
@@ -124,10 +140,9 @@ func (t testlink) run(parent context.Context) (int, error) {
return exitClosed, nil
}
// bootstrapEnv is the bootstrap's whole environment: the runtime variables and
// the execution role's three credential variables.
func bootstrapEnv(runtimeAPI, handler string, cred hostlink.Credential) []string {
return []string{
// bootstrapEnv supplies the runtime, execution-role and SDK settings.
func bootstrapEnv(runtimeAPI, handler string, cred hostlink.Credential, region string, sdk *sdkConfig, caPath string) []string {
env := []string{
"AWS_LAMBDA_RUNTIME_API=" + runtimeAPI,
"_HANDLER=" + handler,
"LAMBDA_TASK_ROOT=" + taskRoot,
@@ -135,7 +150,13 @@ func bootstrapEnv(runtimeAPI, handler string, cred hostlink.Credential) []string
"AWS_ACCESS_KEY_ID=" + cred.AccessKeyID,
"AWS_SECRET_ACCESS_KEY=" + cred.SecretAccessKey,
"AWS_SESSION_TOKEN=" + cred.SessionToken,
"AWS_REGION=" + region,
"AWS_DEFAULT_REGION=" + region,
}
if sdk != nil {
env = append(env, "AWS_ENDPOINT_URL="+sdk.endpoint, "AWS_CA_BUNDLE="+caPath)
}
return env
}
// shutdownGrace is how long past the invocation deadline the agent waits for
+42 -1
View File
@@ -7,8 +7,10 @@ import (
"context"
"crypto/sha256"
"encoding/hex"
"encoding/pem"
"io"
"net/http"
"net/http/httptest"
"os"
"os/exec"
"path/filepath"
@@ -70,6 +72,7 @@ type bootstrapStarts struct {
sealed bool
// handlerOut receives the _HANDLER the bootstrap process itself saw.
handlerOut string
caBundle string
// env is the environment the agent gave the bootstrap.
env []string
}
@@ -94,6 +97,7 @@ func (b *bootstrapStarts) testlink(t *testing.T, entries map[string][]byte) test
b.started = make(chan struct{}, 1)
b.task = t.TempDir()
b.handlerOut = filepath.Join(t.TempDir(), "handler")
b.caBundle = filepath.Join(t.TempDir(), "sdk", "ca-bundle.pem")
return testlink{
startup: g.startup(),
pkg: taskPackage{
@@ -102,7 +106,9 @@ func (b *bootstrapStarts) testlink(t *testing.T, entries map[string][]byte) test
mountDir: func(string) error { return nil },
sealDir: func(string) error { b.sealed = true; return nil },
},
runtimeAddr: "127.0.0.1:0",
runtimeAddr: "127.0.0.1:0",
caBundlePath: b.caBundle,
systemCAPath: filepath.Join(t.TempDir(), "system-ca.pem"),
startBootstrap: func(ctx context.Context, env []string) (*exec.Cmd, error) {
b.n++
b.env = env
@@ -176,10 +182,45 @@ func TestTestlinkServesOneInvocationThenStops(t *testing.T) {
"AWS_ACCESS_KEY_ID=" + testCredential.AccessKeyID,
"AWS_SECRET_ACCESS_KEY=" + testCredential.SecretAccessKey,
"AWS_SESSION_TOKEN=" + testCredential.SessionToken,
"AWS_REGION=ap-southeast-2",
"AWS_DEFAULT_REGION=ap-southeast-2",
}, b.env, "the bootstrap gets exactly the runtime and credential variables")
assert.NotEmpty(t, envValue(b.env, "AWS_LAMBDA_RUNTIME_API"))
}
func TestTestlinkProvidesSDKSettingsToBootstrap(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
defer cancel()
server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusOK) }))
defer server.Close()
h := newHost(t)
h.accept(ctx)
entries := h.entries()
entries[sdkEndpointEntry] = []byte(server.URL)
entries[sdkCAEntry] = pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: server.Certificate().Raw})
var b bootstrapStarts
done := runTestlinkAsync(ctx, b.testlink(t, entries))
conn := h.guest(t)
defer conn.Close()
hs := hostlink.NewTestStream(conn)
req := hostlink.Request{Credential: &testCredential, GrantID: "g-1", AttemptID: "a-1", RequestID: "r-1", DeadlineMs: time.Now().Add(5 * time.Second).UnixMilli(), Payload: []byte(`{}`)}
require.NoError(t, hs.HostSend(req))
_, err := hs.HostResult(ctx)
require.NoError(t, err)
res := wait(t, done, "agent to stop")
require.NoError(t, res.err)
assert.Equal(t, server.URL, envValue(b.env, "AWS_ENDPOINT_URL"))
assert.Equal(t, b.caBundle, envValue(b.env, "AWS_CA_BUNDLE"))
assert.Equal(t, "ap-southeast-2", envValue(b.env, "AWS_REGION"))
assert.Equal(t, "ap-southeast-2", envValue(b.env, "AWS_DEFAULT_REGION"))
ca, err := os.ReadFile(b.caBundle)
require.NoError(t, err)
assert.Contains(t, string(ca), string(entries[sdkCAEntry]))
info, err := os.Stat(b.caBundle)
require.NoError(t, err)
assert.Equal(t, os.FileMode(0o644), info.Mode().Perm())
}
var testCredential = hostlink.Credential{AccessKeyID: "ASIAGUESTKEY", SecretAccessKey: "guest-secret", SessionToken: "guest-token"}
func envValue(env []string, name string) string {
@@ -83,6 +83,9 @@ type Daemon struct {
MTU int
WorkDir string
HTTP *http.Client
// SDKEndpoint and SDKCA are public gateway connection settings for the guest.
SDKEndpoint string
SDKCA []byte
// Image is the prepared runtime image every launch boots; the daemon
// prepares it once before serving and keeps it open for its whole life.
Image *Image
@@ -582,6 +585,10 @@ func (d *Daemon) prepare(req Request, code []byte, dir, tap, mac, ip string) (*v
entries["test-package-sha256"] = []byte(req.CodeSHA256)
entries["test-grant"] = grantEntry(req.Grant)
entries["test-handler"] = []byte(req.Handler)
if d.SDKEndpoint != "" && len(d.SDKCA) != 0 {
entries["test-sdk-endpoint"] = []byte(d.SDKEndpoint)
entries["test-sdk-ca"] = d.SDKCA
}
netcfgBytes := fmt.Appendf(nil, "NIC0_MAC=%s\nNIC0_CIDR=%s/24\nNIC0_DEFAULT=0\n", mac, ip)
if d.Firecracker != "" {
archive := make(map[string][]byte, len(entries)+1)
@@ -368,6 +368,36 @@ func TestPrepare_ScopeAndGrantEntryComeFromTheGrant(t *testing.T) {
require.Equal(t, "ap-southeast-2/123456789012/fn1/env-x/3", strings.TrimSpace(string(expect)))
}
func TestPrepare_DeliversSDKSettingsForEachVMM(t *testing.T) {
for _, firecracker := range []bool{false, true} {
t.Run(fmt.Sprintf("firecracker=%t", firecracker), func(t *testing.T) {
req := baseRequest("https://example.invalid/code", []byte("zip"))
req.Grant = execution.Grant{GrantID: "g-7", AttemptID: req.AttemptID, EnvironmentID: "env-x", Generation: 3, Node: testNode}
d := &Daemon{HostIP: "127.0.0.1", SDKEndpoint: "https://10.15.8.1:9999", SDKCA: []byte("public-ca")}
if firecracker {
d.Firecracker = "/test/firecracker"
}
dir := filepath.Join(t.TempDir(), "attempt")
_, _, ln, err := d.prepare(req, []byte("zip"), dir, "tap0", "02:00:00:00:00:01", "10.0.0.2")
require.NoError(t, err)
defer ln.Close()
endpointPath, caPath := filepath.Join(dir, "fwcfg-test-sdk-endpoint"), filepath.Join(dir, "fwcfg-test-sdk-ca")
if firecracker {
root := filepath.Join(t.TempDir(), "config")
extract(t, filepath.Join(dir, configArchive), root)
entryDir := filepath.Join(root, "by_name", lambdav1.FwCfgPrefix)
endpointPath, caPath = filepath.Join(entryDir, "test-sdk-endpoint", "raw"), filepath.Join(entryDir, "test-sdk-ca", "raw")
}
endpoint, err := os.ReadFile(endpointPath)
require.NoError(t, err)
ca, err := os.ReadFile(caPath)
require.NoError(t, err)
require.Equal(t, d.SDKEndpoint, string(endpoint))
require.Equal(t, d.SDKCA, ca)
})
}
}
// elfHeader is a bare 64-bit little-endian shared-object header for machine.
func elfHeader(machine elf.Machine) []byte {
h := make([]byte, 64)
@@ -7,6 +7,7 @@ import (
"crypto/tls"
"crypto/x509"
"encoding/hex"
"encoding/pem"
"errors"
"fmt"
"io"
@@ -82,15 +83,21 @@ func (d *Daemon) startLambdaTestLink() {
return
}
httpClient, trustErr := d.lambdaTestLinkHTTP()
sdkEndpoint, sdkCA, sdkErr := d.lambdaTestLinkSDKConfig()
if sdkErr != nil {
slog.Warn("Lambda test link: SDK gateway settings unavailable", "err", sdkErr)
}
imageDir := filepath.Join(microvmDir, "lambda-testlink")
tl := &testlink.Daemon{
Node: d.node,
Grants: grants,
Taps: taps,
IPs: d.mgmtIPAllocator,
Bridge: d.mgmtBridgeName(),
HostIP: d.mgmtBridgeIP,
WorkDir: workDir,
Node: d.node,
Grants: grants,
Taps: taps,
IPs: d.mgmtIPAllocator,
Bridge: d.mgmtBridgeName(),
HostIP: d.mgmtBridgeIP,
WorkDir: workDir,
SDKEndpoint: sdkEndpoint,
SDKCA: sdkCA,
Initializer: filepath.Join(imageDir, "hardkill.so"),
InitializerSHA256: lambdaHardKillSHA256,
@@ -135,7 +142,6 @@ func (d *Daemon) startLambdaTestLink() {
// missed, by a crash or an earlier refusal, waits for the next pass.
const lambdaActivationInterval = 30 * time.Second
// lambdaTestLinkReady runs the serving checks and, only once they pass,
// prepares the pinned image into tl; a refusal leaves tl.Image nil.
func lambdaTestLinkReady(tl *testlink.Daemon, imageDir, imagePin string, trustErr error) error {
@@ -201,4 +207,32 @@ func (d *Daemon) lambdaTestLinkHTTP() (*http.Client, error) {
}}, nil
}
// lambdaTestLinkSDKConfig returns public settings for the guest's AWS SDK.
func (d *Daemon) lambdaTestLinkSDKConfig() (string, []byte, error) {
if d.mgmtBridgeIP == "" || d.config.NATS.CACert == "" {
return "", nil, errors.New("management address or cluster CA not configured")
}
ca, err := os.ReadFile(d.config.NATS.CACert)
if err != nil {
return "", nil, errors.New("cluster CA unreadable")
}
if len(ca) > 64<<10 || !x509.NewCertPool().AppendCertsFromPEM(ca) {
return "", nil, errors.New("cluster CA invalid or too large")
}
var public []byte
for rest := ca; len(rest) != 0; {
block, next := pem.Decode(rest)
if block == nil {
break
}
rest = next
if block.Type == "CERTIFICATE" {
if _, err := x509.ParseCertificate(block.Bytes); err == nil {
public = append(public, pem.EncodeToMemory(block)...)
}
}
}
return d.resolveSystemGatewayBaseURL(), public, nil
}
var _ testlink.Taps = (*host.OVSPlumber)(nil)
@@ -0,0 +1,44 @@
//go:build lambdatestlink
//test:in-package exercises daemon-owned gateway configuration
package daemon
import (
"encoding/pem"
"net/http/httptest"
"os"
"path/filepath"
"testing"
"github.com/mulgadc/spinifex/spinifex/bootstrap/config"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func TestLambdaSDKConfigUsesManagementGatewayAndClusterCA(t *testing.T) {
server := httptest.NewTLSServer(nil)
defer server.Close()
ca := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: server.Certificate().Raw})
path := filepath.Join(t.TempDir(), "ca.pem")
require.NoError(t, os.WriteFile(path, ca, 0o600))
d := &Daemon{mgmtBridgeIP: "10.15.8.1", config: &config.Config{
NATS: config.NATSConfig{CACert: path}, AWSGW: config.AWSGWConfig{Host: "0.0.0.0:9999"},
}}
endpoint, gotCA, err := d.lambdaTestLinkSDKConfig()
require.NoError(t, err)
assert.Equal(t, "https://10.15.8.1:9999", endpoint)
assert.Equal(t, ca, gotCA)
withKey := append(append([]byte{}, ca...), pem.EncodeToMemory(&pem.Block{Type: "PRIVATE KEY", Bytes: []byte("private")})...)
withKey = append(withKey, pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: []byte("not a certificate")})...)
require.NoError(t, os.WriteFile(path, withKey, 0o600))
_, gotCA, err = d.lambdaTestLinkSDKConfig()
require.NoError(t, err)
assert.Equal(t, ca, gotCA, "only public certificates reach the guest")
require.NoError(t, os.WriteFile(path, []byte("bad CA"), 0o600))
_, _, err = d.lambdaTestLinkSDKConfig()
require.Error(t, err)
_, err = d.lambdaTestLinkHTTP()
require.Error(t, err, "configured invalid download trust must still refuse serving")
}