Compare commits

..

6 Commits

Author SHA1 Message Date
joshsiv-mulga cc2a0b8871 test(iam): account for UpdateUser as named by bare name
UpdateUser addresses its user by bare name, so the cross-tenant ARN table
cannot target another account; list it with the other byName IAM actions.
Add a focused test that a rename stays within the caller's account.
2026-11-02 17:07:36 +10:00
joshsiv-mulga 473042ce47 test(iam): check the resolved NoSuchEntity code in UpdateUser tests
IAM errors now carry a message, so err.Error() no longer equals the bare
code. Assert the resolved AWS error code instead.
2026-11-02 16:55:12 +10:00
joshsiv-mulga 5fcd86cb0e fix(lambda): match IAM error codes, not messages, in code-store provisioning
IAM errors now carry a message, so comparing err.Error() with the bare
code no longer matched EntityAlreadyExists or NoSuchEntity. Provisioning
then failed on every start after the first and could not adopt its own
role. Resolve the AWS error code instead.
2026-11-02 16:54:29 +10:00
joshsiv-mulga 0fd267263e fix(lambda): render the stored record in the CreateFunction response
CodeStored now returns the record the owner wrote, and CreateFunction
renders it instead of stamping its own time, so CreatedAt, LastModified
and RevisionId match an immediate GetFunction.
2026-11-02 16:49:45 +10:00
joshsiv-mulga a440ca9dba fix(lambda): return and accept only the default LoggingConfig
Every function configuration now carries LoggingConfig with LogFormat Text
and LogGroup /aws/lambda/<name>. CreateFunction accepts the default when
supplied and refuses a JSON format, another log group or a log level.

Adds an integration test that drives a signed CreateFunction and
GetFunction through the gateway, the NATS record owner and the code store,
then downloads Code.Location and checks the bytes. The STS trust check
moves to stsdomain.RoleTrusts so the test uses the production evaluation.
2026-11-02 16:49:45 +10:00
joshsiv-mulga 8bc6b3f47a feat(lambda): add unregistered CreateFunction and GetFunction
Function records live in a daemon-owned per-account KV bucket, keyed by
Region and name, claimed with one conditional create that carries the
function ID and a Pending state. A claim whose code copy never completes
expires, so a later create can take the name over; the original gateway
then fails to mark its code stored. Code copies are never deleted here.

CreateFunction checks the ZIP is readable, refuses an execution role that
does not trust lambda.amazonaws.com (using the STS trust evaluation), and
encrypts environment values in the gateway with the master key, bound to
the account and function ID. GetFunction returns Code.Location as a
10-minute presigned URL for the off-host Predastore address, signed by a
session that outlasts it; with no off-host address it fails instead of
naming an internal one. Lambda uses the REST-JSON error envelope.

The handlers are built in the gateway but not routed: no public Lambda API
is registered until a runtime image exists.
2026-11-02 16:49:45 +10:00
30 changed files with 3002 additions and 30 deletions
+34 -2
View File
@@ -39,7 +39,9 @@ func TestUpdateUser_RenameKeepsRelationships(t *testing.T) {
_, err = svc.GetUser(testAccountID, &iam.GetUserInput{UserName: aws.String("bob")})
require.Error(t, err)
assert.Equal(t, awserrors.ErrorIAMNoSuchEntity, err.Error())
code, ok := awserrors.ResolveErrorCode(err)
require.True(t, ok, "no AWS code on %v", err)
assert.Equal(t, awserrors.ErrorIAMNoSuchEntity, code)
got, err := svc.GetUser(testAccountID, &iam.GetUserInput{UserName: aws.String("robert")})
require.NoError(t, err)
@@ -164,8 +166,38 @@ func TestUpdateUser_RenameRetriesAfterRacingWrite(t *testing.T) {
assert.Equal(t, []string{"arn:aws:iam::" + testAccountID + ":policy/raced"}, user.AttachedPolicies,
"the racing write survives the rename")
_, err = svc.getUser(t.Context(), testAccountID, "frank")
assert.Equal(t, awserrors.ErrorIAMNoSuchEntity, err.Error())
code, ok := awserrors.ResolveErrorCode(err)
require.True(t, ok, "no AWS code on %v", err)
assert.Equal(t, awserrors.ErrorIAMNoSuchEntity, code)
ak, err := svc.LookupAccessKey(*key.AccessKey.AccessKeyId)
require.NoError(t, err)
assert.Equal(t, "franklin", ak.UserName)
}
// UpdateUser names its user by bare name, so the caller's account is the only
// scope: a rename must never reach a same-named user in another account.
func TestUpdateUser_ScopedToCallerAccount(t *testing.T) {
t.Parallel()
svc := setupTestIAMService(t)
const other = "999988887777"
mine := createTestUser(t, svc, "bob")
theirs, err := svc.CreateUser(other, &iam.CreateUserInput{UserName: aws.String("bob")})
require.NoError(t, err)
_, err = svc.CreateUser(other, &iam.CreateUserInput{UserName: aws.String("carol")})
require.NoError(t, err)
_, err = svc.UpdateUser(testAccountID, &iam.UpdateUserInput{UserName: aws.String("bob"), NewUserName: aws.String("robert")})
require.NoError(t, err)
got, err := svc.GetUser(other, &iam.GetUserInput{UserName: aws.String("bob")})
require.NoError(t, err)
assert.Equal(t, *theirs.User.UserId, *got.User.UserId)
assert.NotEqual(t, *mine.UserId, *got.User.UserId)
_, err = svc.UpdateUser(testAccountID, &iam.UpdateUserInput{UserName: aws.String("carol"), NewUserName: aws.String("caroline")})
require.Error(t, err)
code, ok := awserrors.ResolveErrorCode(err)
require.True(t, ok, "no AWS code on %v", err)
assert.Equal(t, awserrors.ErrorIAMNoSuchEntity, code)
_, err = svc.GetUser(other, &iam.GetUserInput{UserName: aws.String("carol")})
require.NoError(t, err)
}
+168
View File
@@ -0,0 +1,168 @@
package awsapi
import (
"context"
"crypto/rand"
"encoding/base64"
"encoding/hex"
"fmt"
"log/slog"
"net/http"
"strconv"
"time"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/lambda"
"github.com/mulgadc/spinifex/spinifex/domains/lambda/codestore"
"github.com/mulgadc/spinifex/spinifex/domains/lambda/function"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/arn"
awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
)
const (
rollbackDeadline = 10 * time.Second
lastModifiedFmt = "2006-01-02T15:04:05.000-0700"
)
func randomID() (string, error) {
b := make([]byte, 16)
if _, err := rand.Read(b); err != nil {
return "", err
}
return hex.EncodeToString(b), nil
}
func serviceError(msg string) error {
return awserrors.Errorf(awserrors.ErrorServiceException, "%s", msg)
}
// createFunction implements CreateFunction. The name is claimed first, the code
// copy written second, and the claim completed last, so a failure at any point
// leaves either a free name or a claim that expires on its own.
func (d Deps) createFunction(ctx context.Context, c *call) (reply, error) {
p, err := validateCreate(c.Create)
if err != nil {
return reply{}, err
}
functionID, err := randomID()
if err != nil {
return reply{}, serviceError("could not generate a function identifier")
}
revisionID, err := randomID()
if err != nil {
return reply{}, serviceError("could not generate a revision identifier")
}
var envCipher string
if len(p.Env) > 0 {
if d.AEAD == nil {
return reply{}, serviceError("environment encryption is unavailable")
}
if envCipher, err = function.SealEnv(d.AEAD, c.AccountID, functionID, p.Env); err != nil {
slog.ErrorContext(ctx, "Lambda: seal environment failed", "function_id", functionID, "err", err)
return reply{}, serviceError("could not protect the function environment")
}
}
rec := function.Record{
FormatVersion: function.FormatVersion,
FunctionID: functionID,
AccountID: c.AccountID,
Region: c.Region,
FunctionName: p.Name,
Role: p.Role,
Runtime: supportedRuntime,
Handler: p.Handler,
Description: p.Description,
Timeout: p.TimeoutSec,
MemorySize: p.MemorySize,
EnvCipher: envCipher,
State: function.StatePending,
Code: function.Code{RevisionID: revisionID, State: function.CodeStoring},
}
createRev, err := d.Meta.Create(ctx, c.AccountID, c.Region, rec)
if err != nil {
return reply{}, err
}
ref := codestore.Ref{AccountID: c.AccountID, FunctionID: functionID, RevisionID: revisionID}
store, err := d.Codes(ctx)
if err != nil {
d.rollback(ctx, c, p.Name, createRev, ref, "code store unavailable", err)
return reply{}, serviceError("the function code could not be stored")
}
sha, err := store.Put(ctx, ref, p.Code)
if err != nil {
d.rollback(ctx, c, p.Name, createRev, ref, "code write failed", err)
return reply{}, serviceError("the function code could not be stored")
}
stored, storedRev, err := d.Meta.CodeStored(ctx, c.AccountID, c.Region, p.Name, createRev, sha, int64(len(p.Code)))
if err != nil {
d.rollback(ctx, c, p.Name, createRev, ref, "code-stored was not recorded", err)
return reply{}, serviceError("the function could not be completed")
}
cfg, err := configuration(stored, storedRev, p.Env)
if err != nil {
return reply{}, serviceError("could not build the function configuration")
}
return reply{Status: http.StatusCreated, Body: cfg}, nil
}
// rollback makes one compare-and-swap attempt to free the name. It cannot undo
// a create that completed, because the revision it holds has moved on. The code
// object, if one landed, is left in place and recorded as a cleanup obligation.
func (d Deps) rollback(ctx context.Context, c *call, name string, createRev uint64, ref codestore.Ref, why string, cause error) {
slog.ErrorContext(ctx, "Lambda: create did not complete", "reason", why, "function_name", name,
"function_id", ref.FunctionID, "err", cause)
rctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), rollbackDeadline)
defer cancel()
if err := d.Meta.RollbackCreate(rctx, c.AccountID, c.Region, name, createRev); err != nil {
slog.WarnContext(ctx, "Lambda: rollback of the name claim failed; it expires on its own",
"function_name", name, "function_id", ref.FunctionID, "err", err)
}
slog.WarnContext(ctx, "Lambda: code object cleanup is owed",
"account_id", ref.AccountID, "function_id", ref.FunctionID, "revision_id", ref.RevisionID)
}
// configuration renders the record as a FunctionConfiguration.
func configuration(rec function.Record, revision uint64, env map[string]string) (*lambda.FunctionConfiguration, error) {
cfg := &lambda.FunctionConfiguration{
FunctionName: aws.String(rec.FunctionName),
FunctionArn: aws.String(arn.FormatLambdaFunction(rec.Region, rec.AccountID, rec.FunctionName)),
Runtime: aws.String(rec.Runtime),
Role: aws.String(rec.Role),
Handler: aws.String(rec.Handler),
Description: aws.String(rec.Description),
Timeout: aws.Int64(rec.Timeout),
MemorySize: aws.Int64(rec.MemorySize),
LastModified: aws.String(rec.LastModified.UTC().Format(lastModifiedFmt)),
Version: aws.String("$LATEST"),
State: aws.String(rec.State),
PackageType: aws.String(lambda.PackageTypeZip),
Architectures: aws.StringSlice([]string{lambda.ArchitectureX8664}),
RevisionId: aws.String(strconv.FormatUint(revision, 10)),
LastUpdateStatus: aws.String(lambda.LastUpdateStatusSuccessful),
EphemeralStorage: &lambda.EphemeralStorage{Size: aws.Int64(512)},
LoggingConfig: &lambda.LoggingConfig{LogFormat: aws.String(lambda.LogFormatText), LogGroup: aws.String(logGroup(rec.FunctionName))},
}
if rec.StateReason != "" {
cfg.StateReason = aws.String(rec.StateReason)
cfg.StateReasonCode = aws.String(rec.StateReasonCode)
}
if rec.Code.State == function.CodeStored {
raw, err := hex.DecodeString(rec.Code.SHA256)
if err != nil {
return nil, fmt.Errorf("stored digest is not hex: %w", err)
}
cfg.CodeSha256 = aws.String(base64.StdEncoding.EncodeToString(raw))
cfg.CodeSize = aws.Int64(rec.Code.Size)
}
if len(env) > 0 {
vars := make(map[string]*string, len(env))
for k, v := range env {
vars[k] = aws.String(v)
}
cfg.Environment = &lambda.EnvironmentResponse{Variables: vars}
}
return cfg, nil
}
+82
View File
@@ -0,0 +1,82 @@
package awsapi
import (
"context"
"errors"
"log/slog"
"net/http"
"time"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/lambda"
"github.com/mulgadc/spinifex/spinifex/domains/lambda/codestore"
"github.com/mulgadc/spinifex/spinifex/domains/lambda/function"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/arn"
awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
)
// codeURLTTL is how long the presigned code location stays valid.
const codeURLTTL = 10 * time.Minute
func notFound(region, accountID, name string) error {
return awserrors.Errorf(awserrors.ErrorResourceNotFoundException, "Function not found: %s",
arn.FormatLambdaFunction(region, accountID, name))
}
// getFunction implements GetFunction for the $LATEST version.
func (d Deps) getFunction(ctx context.Context, c *call) (reply, error) {
name, q, err := resolveFunctionName(c.Region, c.AccountID, c.Params[0])
if err != nil {
return reply{}, err
}
if query := c.Query.Get("Qualifier"); query != "" {
if q != "" && q != query {
return reply{}, invalid("The derived qualifier from the function name does not match the specified qualifier.")
}
q = query
}
if q != "" && q != "$LATEST" {
return reply{}, notFound(c.Region, c.AccountID, name+":"+q)
}
rec, revision, err := d.Meta.Get(ctx, c.AccountID, c.Region, name)
if errors.Is(err, function.ErrNotFound) {
return reply{}, notFound(c.Region, c.AccountID, name)
}
if err != nil {
return reply{}, err
}
var env map[string]string
if rec.EnvCipher != "" {
if d.AEAD == nil {
return reply{}, serviceError("environment decryption is unavailable")
}
if env, err = function.OpenEnv(d.AEAD, rec.AccountID, rec.FunctionID, rec.EnvCipher); err != nil {
slog.ErrorContext(ctx, "Lambda: open environment failed", "function_id", rec.FunctionID, "err", err)
return reply{}, serviceError("could not read the function environment")
}
}
cfg, err := configuration(rec, revision, env)
if err != nil {
return reply{}, serviceError("could not build the function configuration")
}
out := &lambda.GetFunctionOutput{
Configuration: cfg,
Code: &lambda.FunctionCodeLocation{RepositoryType: aws.String("S3")},
}
if rec.Code.State == function.CodeStored {
store, err := d.Codes(ctx)
if err != nil {
slog.ErrorContext(ctx, "Lambda: code store unavailable for GetFunction", "function_id", rec.FunctionID, "err", err)
return reply{}, serviceError("the function code location is unavailable")
}
ref := codestore.Ref{AccountID: rec.AccountID, FunctionID: rec.FunctionID, RevisionID: rec.Code.RevisionID}
location, err := store.PresignGet(ctx, ref, rec.Code.SHA256, codeURLTTL)
if err != nil {
slog.ErrorContext(ctx, "Lambda: presign code location failed", "function_id", rec.FunctionID, "err", err)
return reply{}, serviceError("the function code location is unavailable")
}
out.Code.Location = aws.String(location)
}
return reply{Status: http.StatusOK, Body: out}, nil
}
@@ -0,0 +1,669 @@
package awsapi_test
import (
"archive/zip"
"bytes"
"context"
"crypto/sha256"
"encoding/base64"
"encoding/hex"
"encoding/json"
"errors"
"net/http"
"net/http/httptest"
"runtime"
"strings"
"sync"
"testing"
"time"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/lambda"
"github.com/mulgadc/bluebottle/pkg/auth"
"github.com/mulgadc/bluebottle/pkg/masterkey"
"github.com/mulgadc/spinifex/internal/testkit"
"github.com/mulgadc/spinifex/internal/testkit/gwtest"
iamdomain "github.com/mulgadc/spinifex/spinifex/domains/iam"
"github.com/mulgadc/spinifex/spinifex/domains/lambda/awsapi"
"github.com/mulgadc/spinifex/spinifex/domains/lambda/codestore"
"github.com/mulgadc/spinifex/spinifex/domains/lambda/function"
awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/passrole"
"github.com/nats-io/nats.go/jetstream"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
const (
roleARN = "arn:aws:iam::" + gwtest.AccountID + ":role/lambda-exec"
foreignRoleARN = "arn:aws:iam::999999999999:role/lambda-exec"
)
// localClient reaches a real MetaService without NATS.
type localClient struct {
svc *function.MetaService
mu sync.Mutex
failCodeStored bool
passOwnerClock bool
rollbacks int
}
var _ function.Client = (*localClient)(nil)
func (c *localClient) Create(ctx context.Context, account, region string, rec function.Record) (uint64, error) {
r, err := c.svc.Create(ctx, &function.CreateRequest{Region: region, Record: rec}, account)
if err != nil {
return 0, err
}
return r.Revision, nil
}
func (c *localClient) Get(ctx context.Context, account, region, name string) (function.Record, uint64, error) {
r, err := c.svc.Get(ctx, &function.GetRequest{Region: region, FunctionName: name}, account)
if err != nil {
return function.Record{}, 0, err
}
if !r.Found {
return function.Record{}, 0, function.ErrNotFound
}
return r.Record, r.Revision, nil
}
func (c *localClient) CodeStored(ctx context.Context, account, region, name string, rev uint64, sha string, size int64) (function.Record, uint64, error) {
c.mu.Lock()
fail, passClock := c.failCodeStored, c.passOwnerClock
c.mu.Unlock()
if fail {
return function.Record{}, 0, errors.New("nats: timeout")
}
r, err := c.svc.CodeStored(ctx, &function.CodeStoredRequest{Region: region, FunctionName: name, Revision: rev, SHA256: sha, Size: size}, account)
if err != nil {
return function.Record{}, 0, err
}
// Let the wall clock pass the owner's millisecond, so a timestamp the
// gateway reads for itself would differ from the stored one.
for passClock && !time.Now().Truncate(time.Millisecond).After(r.Record.LastModified.Truncate(time.Millisecond)) {
runtime.Gosched()
}
return r.Record, r.Revision, nil
}
func (c *localClient) RollbackCreate(ctx context.Context, account, region, name string, rev uint64) error {
c.mu.Lock()
c.rollbacks++
c.mu.Unlock()
_, err := c.svc.RollbackCreate(ctx, &function.RollbackCreateRequest{Region: region, FunctionName: name, Revision: rev}, account)
return err
}
// memStore is an in-memory code store; objects is keyed by Ref.
type memStore struct {
mu sync.Mutex
objects map[codestore.Ref][]byte
putErr error
urlErr error
}
var _ awsapi.CodeStore = (*memStore)(nil)
func (m *memStore) Put(_ context.Context, ref codestore.Ref, code []byte) (string, error) {
m.mu.Lock()
defer m.mu.Unlock()
if m.putErr != nil {
return "", m.putErr
}
m.objects[ref] = append([]byte(nil), code...)
sum := sha256.Sum256(code)
return hex.EncodeToString(sum[:]), nil
}
func (m *memStore) PresignGet(_ context.Context, ref codestore.Ref, sha string, ttl time.Duration) (string, error) {
if m.urlErr != nil {
return "", m.urlErr
}
return "https://code.test/" + ref.FunctionID + "/" + sha + "?ttl=" + ttl.String(), nil
}
func (m *memStore) count() int {
m.mu.Lock()
defer m.mu.Unlock()
return len(m.objects)
}
type roleSet map[string]bool
func (r roleSet) ResolveRole(accountID, arn string) (passrole.Role, error) {
acct, name, err := auth.ParseRoleARN(arn)
if err != nil {
return passrole.Role{}, passrole.ErrMalformed
}
if acct != accountID {
return passrole.Role{}, passrole.ErrForeignAccount
}
if !r[arn] {
return passrole.Role{}, passrole.ErrNotFound
}
return passrole.Role{ARN: arn, Name: name}, nil
}
func newHarness(t *testing.T, statements ...iamdomain.Statement) (*gwtest.Harness, *localClient, *memStore, func(*testing.T) jetstream.KeyValue, *masterkey.Key) {
t.Helper()
_, _, js := testutil.StartTestJetStream(t)
key, err := masterkey.New(make([]byte, 32))
require.NoError(t, err)
meta := &localClient{svc: function.NewMetaService(js)}
store := &memStore{objects: map[codestore.Ref][]byte{}}
kv := func(t *testing.T) jetstream.KeyValue {
t.Helper()
b, err := js.KeyValue(t.Context(), function.KVAccountBucket(gwtest.AccountID))
require.NoError(t, err)
return b
}
return gwtest.Scoped(statements...), meta, store, kv, key
}
type env struct {
h *gwtest.Harness
meta *localClient
store *memStore
kv func(*testing.T) jetstream.KeyValue
deps awsapi.Deps
}
func newEnv(t *testing.T, statements ...iamdomain.Statement) *env {
t.Helper()
if len(statements) == 0 {
statements = []iamdomain.Statement{gwtest.Statement("Allow", "*", "*")}
}
h, meta, store, kv, key := newHarness(t, statements...)
e := &env{h: h, meta: meta, store: store, kv: kv}
e.deps = awsapi.Deps{
Meta: meta,
Codes: func(context.Context) (awsapi.CodeStore, error) {
return store, nil
},
Roles: roleSet{roleARN: true},
Trusts: func(string, string) (bool, error) { return true, nil },
AEAD: key.AEAD,
}
return e
}
func (e *env) do(method, path string, body any) (*httptest.ResponseRecorder, error) {
var raw []byte
if body != nil {
raw, _ = json.Marshal(body)
}
rec := httptest.NewRecorder()
req := httptest.NewRequest(method, path, strings.NewReader(string(raw)))
err := e.h.Dispatch(awsapi.NewRegistration(e.deps), awsapi.ServiceName, rec, req)
return rec, err
}
var testZip = buildZip("hello")
func buildZip(content string) []byte {
var buf bytes.Buffer
w := zip.NewWriter(&buf)
f, err := w.Create("bootstrap")
if err == nil {
_, err = f.Write([]byte(content))
}
if err == nil {
err = w.Close()
}
if err != nil {
panic(err)
}
return buf.Bytes()
}
func mustZip(t *testing.T, content string) []byte {
t.Helper()
return buildZip(content)
}
func createBody(name string, mutate func(map[string]any)) map[string]any {
b := map[string]any{
"FunctionName": name,
"Runtime": "provided.al2023",
"Role": roleARN,
"Handler": "bootstrap",
"Code": map[string]any{"ZipFile": testZip},
}
if mutate != nil {
mutate(b)
}
return b
}
func requireCode(t *testing.T, err error, code string) {
t.Helper()
require.Error(t, err)
got, ok := awserrors.ResolveErrorCode(err)
require.True(t, ok, "no AWS code on %v", err)
assert.Equal(t, code, got)
}
func TestCreateFunction_Success(t *testing.T) {
e := newEnv(t)
code := testZip
rec, err := e.do(http.MethodPost, "/2015-03-31/functions", createBody("hello", func(b map[string]any) {
b["Environment"] = map[string]any{"Variables": map[string]string{"DB_PASSWORD": "hunter2"}}
b["Description"] = "demo"
b["Timeout"] = 30
b["MemorySize"] = 256
}))
require.NoError(t, err)
require.Equal(t, http.StatusCreated, rec.Code)
var cfg lambda.FunctionConfiguration
require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &cfg))
sum := sha256.Sum256(code)
assert.Equal(t, base64.StdEncoding.EncodeToString(sum[:]), *cfg.CodeSha256, "AWS reports base64 of the raw digest")
assert.Equal(t, int64(len(code)), *cfg.CodeSize)
assert.Equal(t, "Pending", *cfg.State)
assert.Equal(t, "Creating", *cfg.StateReasonCode)
assert.Equal(t, "arn:aws:lambda:"+gwtest.Region+":"+gwtest.AccountID+":function:hello", *cfg.FunctionArn)
assert.Equal(t, "$LATEST", *cfg.Version)
assert.Equal(t, "Zip", *cfg.PackageType)
assert.Equal(t, int64(30), *cfg.Timeout)
assert.Equal(t, int64(256), *cfg.MemorySize)
assert.Equal(t, "hunter2", *cfg.Environment.Variables["DB_PASSWORD"])
assert.Regexp(t, `^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}\.\d{3}\+0000$`, *cfg.LastModified)
require.Equal(t, 1, e.store.count())
for _, stored := range e.store.objects {
assert.Equal(t, code, stored, "the code copy is retrievable")
}
entry, err := e.kv(t).Get(t.Context(), function.KVFunctionKey(gwtest.Region, "hello"))
require.NoError(t, err)
assert.NotContains(t, string(entry.Value()), "hunter2", "the environment is never stored in plaintext")
assert.Contains(t, string(entry.Value()), "envCipher")
}
func TestCreateFunction_RendersTheStoredTimestamps(t *testing.T) {
e := newEnv(t)
e.meta.passOwnerClock = true
rec, err := e.do(http.MethodPost, "/2015-03-31/functions", createBody("hello", nil))
require.NoError(t, err)
var created lambda.FunctionConfiguration
require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &created))
rec, err = e.do(http.MethodGet, "/2015-03-31/functions/hello", nil)
require.NoError(t, err)
var got lambda.GetFunctionOutput
require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &got))
assert.Equal(t, *got.Configuration.LastModified, *created.LastModified)
assert.Equal(t, *got.Configuration.RevisionId, *created.RevisionId)
}
func TestCreateFunction_DuplicateNameLeavesFirstUntouched(t *testing.T) {
e := newEnv(t)
_, err := e.do(http.MethodPost, "/2015-03-31/functions", createBody("hello", nil))
require.NoError(t, err)
first, _, err := e.meta.Get(t.Context(), gwtest.AccountID, gwtest.Region, "hello")
require.NoError(t, err)
_, err = e.do(http.MethodPost, "/2015-03-31/functions", createBody("hello", func(b map[string]any) {
b["Code"] = map[string]any{"ZipFile": mustZip(t, "different")}
}))
requireCode(t, err, awserrors.ErrorResourceConflictException)
assert.Contains(t, err.Error(), "Function already exist: hello")
assert.Equal(t, 1, e.store.count(), "the loser wrote no code")
after, _, err := e.meta.Get(t.Context(), gwtest.AccountID, gwtest.Region, "hello")
require.NoError(t, err)
assert.Equal(t, first, after)
}
func TestCreateFunction_CodePutFailureFreesTheName(t *testing.T) {
e := newEnv(t)
e.store.putErr = errors.New("predastore down")
_, err := e.do(http.MethodPost, "/2015-03-31/functions", createBody("hello", nil))
requireCode(t, err, awserrors.ErrorServiceException)
assert.NotContains(t, err.Error(), "predastore down")
assert.Equal(t, 1, e.meta.rollbacks)
_, _, err = e.meta.Get(t.Context(), gwtest.AccountID, gwtest.Region, "hello")
require.ErrorIs(t, err, function.ErrNotFound)
e.store.putErr = nil
rec, err := e.do(http.MethodPost, "/2015-03-31/functions", createBody("hello", nil))
require.NoError(t, err)
assert.Equal(t, http.StatusCreated, rec.Code)
}
func TestCreateFunction_CodeStoredFailureRollsBackOnce(t *testing.T) {
e := newEnv(t)
e.meta.failCodeStored = true
_, err := e.do(http.MethodPost, "/2015-03-31/functions", createBody("hello", nil))
requireCode(t, err, awserrors.ErrorServiceException)
assert.Equal(t, 1, e.meta.rollbacks)
assert.Equal(t, 1, e.store.count(), "the code object is left in place, never deleted here")
_, _, err = e.meta.Get(t.Context(), gwtest.AccountID, gwtest.Region, "hello")
require.ErrorIs(t, err, function.ErrNotFound, "the still-storing claim was withdrawn")
}
func TestCreateFunction_RefusesUnsupportedInput(t *testing.T) {
cases := map[string]func(map[string]any){
"Code.S3Bucket": func(b map[string]any) { b["Code"] = map[string]any{"S3Bucket": "b", "S3Key": "k"} },
"Code.S3Key": func(b map[string]any) { b["Code"] = map[string]any{"ZipFile": []byte("x"), "S3Key": "k"} },
"Code.S3ObjectVersion": func(b map[string]any) { b["Code"] = map[string]any{"ZipFile": []byte("x"), "S3ObjectVersion": "v"} },
"Code.ImageUri": func(b map[string]any) { b["Code"] = map[string]any{"ImageUri": "x"} },
"Code.SourceKMSKeyArn": func(b map[string]any) { b["Code"] = map[string]any{"ZipFile": []byte("x"), "SourceKMSKeyArn": "k"} },
"Layers": func(b map[string]any) { b["Layers"] = []string{"arn:aws:lambda:r:1:layer:l:1"} },
"VpcConfig": func(b map[string]any) { b["VpcConfig"] = map[string]any{"SubnetIds": []string{"subnet-1"}} },
"KMSKeyArn": func(b map[string]any) { b["KMSKeyArn"] = "arn:aws:kms:r:1:key/k" },
"DeadLetterConfig": func(b map[string]any) { b["DeadLetterConfig"] = map[string]any{"TargetArn": "arn:aws:sqs:r:1:q"} },
"TracingConfig": func(b map[string]any) { b["TracingConfig"] = map[string]any{"Mode": "Active"} },
"FileSystemConfigs": func(b map[string]any) {
b["FileSystemConfigs"] = []map[string]any{{"Arn": "a", "LocalMountPath": "/mnt/x"}}
},
"ImageConfig": func(b map[string]any) { b["ImageConfig"] = map[string]any{"Command": []string{"x"}} },
"Architectures": func(b map[string]any) { b["Architectures"] = []string{"arm64"} },
"Publish": func(b map[string]any) { b["Publish"] = true },
"SnapStart": func(b map[string]any) { b["SnapStart"] = map[string]any{"ApplyOn": "PublishedVersions"} },
"EphemeralStorage": func(b map[string]any) { b["EphemeralStorage"] = map[string]any{"Size": 1024} },
"CodeSigningConfigArn": func(b map[string]any) { b["CodeSigningConfigArn"] = "arn:aws:lambda:r:1:code-signing-config:c" },
"LoggingConfig": func(b map[string]any) { b["LoggingConfig"] = map[string]any{"LogFormat": "JSON"} },
"Tags": func(b map[string]any) { b["Tags"] = map[string]string{"a": "b"} },
"TenancyConfig": func(b map[string]any) { b["TenancyConfig"] = map[string]any{"TenantIsolationMode": "PER_TENANT"} },
}
for field, mutate := range cases {
t.Run(field, func(t *testing.T) {
e := newEnv(t)
_, err := e.do(http.MethodPost, "/2015-03-31/functions", createBody("hello", mutate))
requireCode(t, err, awserrors.ErrorInvalidParameterValueException)
assert.Zero(t, e.store.count())
_, _, getErr := e.meta.Get(t.Context(), gwtest.AccountID, gwtest.Region, "hello")
require.ErrorIs(t, getErr, function.ErrNotFound, "a refused request claims nothing")
})
}
}
func TestCreateFunction_AcceptsDefaultsEquivalentToAbsent(t *testing.T) {
e := newEnv(t)
rec, err := e.do(http.MethodPost, "/2015-03-31/functions", createBody("hello", func(b map[string]any) {
b["Architectures"] = []string{"x86_64"}
b["Publish"] = false
b["EphemeralStorage"] = map[string]any{"Size": 512}
b["TracingConfig"] = map[string]any{"Mode": "PassThrough"}
b["PackageType"] = "Zip"
b["Layers"] = []string{}
b["Tags"] = map[string]string{}
b["LoggingConfig"] = map[string]any{"LogFormat": "Text", "LogGroup": "/aws/lambda/hello"}
}))
require.NoError(t, err)
assert.Equal(t, http.StatusCreated, rec.Code)
}
func TestCreateFunction_RefusesNonDefaultLoggingConfig(t *testing.T) {
for _, cfg := range []map[string]any{
{"LogGroup": "/custom/group"},
{"LogFormat": "Text", "LogGroup": "/aws/lambda/other"},
{"LogFormat": "Text", "ApplicationLogLevel": "DEBUG"},
{"SystemLogLevel": "WARN"},
} {
e := newEnv(t)
_, err := e.do(http.MethodPost, "/2015-03-31/functions", createBody("hello", func(b map[string]any) { b["LoggingConfig"] = cfg }))
requireCode(t, err, awserrors.ErrorInvalidParameterValueException)
_, _, getErr := e.meta.Get(t.Context(), gwtest.AccountID, gwtest.Region, "hello")
require.ErrorIs(t, getErr, function.ErrNotFound, "%v claims nothing", cfg)
}
}
func TestFunctionResponses_CarryTheDefaultLoggingConfig(t *testing.T) {
e := newEnv(t)
rec, err := e.do(http.MethodPost, "/2015-03-31/functions", createBody("hello", nil))
require.NoError(t, err)
var created lambda.FunctionConfiguration
require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &created))
want := &lambda.LoggingConfig{LogFormat: aws.String("Text"), LogGroup: aws.String("/aws/lambda/hello")}
assert.Equal(t, want, created.LoggingConfig)
rec, err = e.do(http.MethodGet, "/2015-03-31/functions/hello", nil)
require.NoError(t, err)
var got lambda.GetFunctionOutput
require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &got))
assert.Equal(t, want, got.Configuration.LoggingConfig)
}
func TestCreateFunction_ParameterValidation(t *testing.T) {
cases := map[string]func(map[string]any){
"bad name": func(b map[string]any) { b["FunctionName"] = "has space" },
"arn name": func(b map[string]any) { b["FunctionName"] = "arn:aws:lambda:r:1:function:x" },
"long name": func(b map[string]any) { b["FunctionName"] = strings.Repeat("a", 65) },
"runtime": func(b map[string]any) { b["Runtime"] = "python3.12" },
"no runtime": func(b map[string]any) { delete(b, "Runtime") },
"package type": func(b map[string]any) { b["PackageType"] = "Image" },
"no code": func(b map[string]any) { delete(b, "Code") },
"empty zip": func(b map[string]any) { b["Code"] = map[string]any{"ZipFile": []byte{}} },
"no handler": func(b map[string]any) { delete(b, "Handler") },
"no role": func(b map[string]any) { delete(b, "Role") },
"timeout low": func(b map[string]any) { b["Timeout"] = 0 },
"timeout high": func(b map[string]any) { b["Timeout"] = 901 },
"memory low": func(b map[string]any) { b["MemorySize"] = 64 },
"memory high": func(b map[string]any) { b["MemorySize"] = 10241 },
"cross account": func(b map[string]any) { b["Role"] = foreignRoleARN },
"missing role": func(b map[string]any) { b["Role"] = "arn:aws:iam::" + gwtest.AccountID + ":role/ghost" },
"not a role arn": func(b map[string]any) { b["Role"] = "arn:aws:iam::" + gwtest.AccountID + ":user/alice" },
"malformed body": func(b map[string]any) { b["Timeout"] = "soon" },
"empty env name": func(b map[string]any) { b["Environment"] = map[string]any{"Variables": map[string]string{"": "x"}} },
"description len": func(b map[string]any) { b["Description"] = strings.Repeat("d", 257) },
}
for name, mutate := range cases {
t.Run(name, func(t *testing.T) {
e := newEnv(t)
_, err := e.do(http.MethodPost, "/2015-03-31/functions", createBody("hello", mutate))
requireCode(t, err, awserrors.ErrorInvalidParameterValueException)
assert.Zero(t, e.store.count())
})
}
}
func TestCreateFunction_RoleErrorsDoNotLeakAcrossAccounts(t *testing.T) {
e := newEnv(t)
_, err := e.do(http.MethodPost, "/2015-03-31/functions", createBody("hello", func(b map[string]any) { b["Role"] = foreignRoleARN }))
requireCode(t, err, awserrors.ErrorInvalidParameterValueException)
assert.Contains(t, err.Error(), "cannot be assumed by Lambda")
}
func TestCreateFunction_RefusesBytesThatAreNotAZip(t *testing.T) {
e := newEnv(t)
_, err := e.do(http.MethodPost, "/2015-03-31/functions", createBody("hello", func(b map[string]any) {
b["Code"] = map[string]any{"ZipFile": []byte("PK\x03\x04 not really a zip")}
}))
requireCode(t, err, awserrors.ErrorInvalidParameterValueException)
assert.Contains(t, err.Error(), "Could not unzip uploaded file")
_, _, err = e.meta.Get(t.Context(), gwtest.AccountID, gwtest.Region, "hello")
require.ErrorIs(t, err, function.ErrNotFound)
assert.Empty(t, e.store.objects)
}
func TestCreateFunction_RoleLambdaCannotAssumeRecordsNothing(t *testing.T) {
e := newEnv(t)
e.deps.Trusts = func(string, string) (bool, error) { return false, nil }
_, err := e.do(http.MethodPost, "/2015-03-31/functions", createBody("hello", nil))
requireCode(t, err, awserrors.ErrorInvalidParameterValueException)
assert.Contains(t, err.Error(), "cannot be assumed by Lambda")
_, _, err = e.meta.Get(t.Context(), gwtest.AccountID, gwtest.Region, "hello")
require.ErrorIs(t, err, function.ErrNotFound)
assert.Empty(t, e.store.objects)
}
func TestCreateFunction_MissingRoleCheckFailsClosed(t *testing.T) {
for name, mutate := range map[string]func(*awsapi.Deps){
"no resolver": func(d *awsapi.Deps) { d.Roles = nil },
"no trust checker": func(d *awsapi.Deps) { d.Trusts = nil },
} {
t.Run(name, func(t *testing.T) {
e := newEnv(t)
mutate(&e.deps)
_, err := e.do(http.MethodPost, "/2015-03-31/functions", createBody("hello", nil))
requireCode(t, err, awserrors.ErrorServiceException)
_, _, err = e.meta.Get(t.Context(), gwtest.AccountID, gwtest.Region, "hello")
require.ErrorIs(t, err, function.ErrNotFound)
})
}
}
func TestGetFunction_DecryptsEnvAndLocatesStoredCode(t *testing.T) {
e := newEnv(t)
_, err := e.do(http.MethodPost, "/2015-03-31/functions", createBody("hello", func(b map[string]any) {
b["Environment"] = map[string]any{"Variables": map[string]string{"MODE": "prod"}}
}))
require.NoError(t, err)
rec, err := e.do(http.MethodGet, "/2015-03-31/functions/hello", nil)
require.NoError(t, err)
require.Equal(t, http.StatusOK, rec.Code)
var out lambda.GetFunctionOutput
require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &out))
assert.Equal(t, "prod", *out.Configuration.Environment.Variables["MODE"])
assert.Equal(t, "S3", *out.Code.RepositoryType)
require.NotNil(t, out.Code.Location)
assert.Contains(t, *out.Code.Location, "https://code.test/")
assert.Contains(t, *out.Code.Location, "ttl=10m0s")
rec, err = e.do(http.MethodGet, "/2015-03-31/functions/hello?Qualifier=%24LATEST", nil)
require.NoError(t, err)
assert.Equal(t, http.StatusOK, rec.Code)
_, err = e.do(http.MethodGet, "/2015-03-31/functions/hello?Qualifier=3", nil)
requireCode(t, err, awserrors.ErrorResourceNotFoundException)
}
func TestGetFunction_AcceptsSameAccountARNForms(t *testing.T) {
e := newEnv(t)
_, err := e.do(http.MethodPost, "/2015-03-31/functions", createBody("hello", nil))
require.NoError(t, err)
full := "arn:aws:lambda:" + gwtest.Region + ":" + gwtest.AccountID + ":function:hello"
esc := func(s string) string { return strings.ReplaceAll(strings.ReplaceAll(s, ":", "%3A"), "$", "%24") }
for _, name := range []string{full, full + ":$LATEST", gwtest.AccountID + ":function:hello", "hello:$LATEST"} {
t.Run(name, func(t *testing.T) {
rec, err := e.do(http.MethodGet, "/2015-03-31/functions/"+esc(name), nil)
require.NoError(t, err)
require.Equal(t, http.StatusOK, rec.Code)
var out lambda.GetFunctionOutput
require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &out))
assert.Equal(t, full, *out.Configuration.FunctionArn)
})
}
notFound := map[string]string{
"other account": "arn:aws:lambda:" + gwtest.Region + ":999999999999:function:hello",
"other region": "arn:aws:lambda:us-west-2:" + gwtest.AccountID + ":function:hello",
"partial foreign": "999999999999:function:hello",
"version": full + ":3",
}
for label, name := range notFound {
t.Run(label, func(t *testing.T) {
_, err := e.do(http.MethodGet, "/2015-03-31/functions/"+esc(name), nil)
requireCode(t, err, awserrors.ErrorResourceNotFoundException)
})
}
_, err = e.do(http.MethodGet, "/2015-03-31/functions/"+esc(full+":$LATEST")+"?Qualifier=3", nil)
requireCode(t, err, awserrors.ErrorInvalidParameterValueException)
_, err = e.do(http.MethodGet, "/2015-03-31/functions/"+esc("arn:aws:s3:::bucket"), nil)
requireCode(t, err, awserrors.ErrorInvalidParameterValueException)
}
func TestGetFunction_NoLocationUntilCodeIsStored(t *testing.T) {
e := newEnv(t)
_, err := e.meta.Create(t.Context(), gwtest.AccountID, gwtest.Region, function.Record{
FunctionID: "f1", FunctionName: "pending", Role: roleARN, Runtime: "provided.al2023", Handler: "h",
Timeout: 3, MemorySize: 128, Code: function.Code{RevisionID: "r1"},
})
require.NoError(t, err)
rec, err := e.do(http.MethodGet, "/2015-03-31/functions/pending", nil)
require.NoError(t, err)
var out lambda.GetFunctionOutput
require.NoError(t, json.Unmarshal(rec.Body.Bytes(), &out))
assert.Nil(t, out.Code.Location)
assert.Nil(t, out.Configuration.CodeSha256)
assert.Equal(t, "Pending", *out.Configuration.State)
}
func TestGetFunction_FailsWithoutAnOffHostCodeURL(t *testing.T) {
e := newEnv(t)
_, err := e.do(http.MethodPost, "/2015-03-31/functions", createBody("hello", nil))
require.NoError(t, err)
e.store.urlErr = codestore.ErrNoPresignEndpoint
rec, err := e.do(http.MethodGet, "/2015-03-31/functions/hello", nil)
requireCode(t, err, awserrors.ErrorServiceException)
assert.NotContains(t, rec.Body.String(), "Location")
}
func TestGetFunction_NotFoundNamesTheARN(t *testing.T) {
e := newEnv(t)
_, err := e.do(http.MethodGet, "/2015-03-31/functions/ghost", nil)
requireCode(t, err, awserrors.ErrorResourceNotFoundException)
assert.Contains(t, err.Error(), "Function not found: arn:aws:lambda:"+gwtest.Region+":"+gwtest.AccountID+":function:ghost")
}
func TestGetFunction_FunctionsAreRegional(t *testing.T) {
e := newEnv(t)
_, err := e.do(http.MethodPost, "/2015-03-31/functions", createBody("hello", nil))
require.NoError(t, err)
e.h.Region = "us-west-2"
_, err = e.do(http.MethodGet, "/2015-03-31/functions/hello", nil)
requireCode(t, err, awserrors.ErrorResourceNotFoundException)
}
func TestAuthorization(t *testing.T) {
create := func(e *env) error {
_, err := e.do(http.MethodPost, "/2015-03-31/functions", createBody("hello", nil))
return err
}
get := func(e *env) error {
_, err := e.do(http.MethodGet, "/2015-03-31/functions/hello", nil)
return err
}
t.Run("create needs lambda:CreateFunction", func(t *testing.T) {
e := newEnv(t, gwtest.Statement("Allow", "lambda:GetFunction", "*"), gwtest.Statement("Allow", "iam:PassRole", "*"))
gwtest.AssertDenied(t, create(e))
assert.Zero(t, e.store.count())
})
t.Run("create needs iam:PassRole", func(t *testing.T) {
e := newEnv(t, gwtest.Statement("Allow", "lambda:*", "*"))
gwtest.AssertDenied(t, create(e))
assert.Zero(t, e.store.count())
})
t.Run("create is scoped to the function ARN", func(t *testing.T) {
arn := "arn:aws:lambda:" + gwtest.Region + ":" + gwtest.AccountID + ":function:hello"
e := newEnv(t,
gwtest.Statement("Allow", "lambda:*", "*"),
gwtest.Statement("Allow", "iam:PassRole", "*"),
gwtest.Statement("Deny", "lambda:CreateFunction", arn))
gwtest.AssertDenied(t, create(e))
})
t.Run("create passes with both grants", func(t *testing.T) {
e := newEnv(t, gwtest.Statement("Allow", "lambda:CreateFunction", "*"), gwtest.Statement("Allow", "iam:PassRole", "*"))
require.NoError(t, create(e))
})
t.Run("get needs lambda:GetFunction", func(t *testing.T) {
e := newEnv(t, gwtest.Statement("Allow", "lambda:CreateFunction", "*"))
gwtest.AssertDenied(t, get(e))
})
t.Run("a denied caller does not learn a role is absent", func(t *testing.T) {
e := newEnv(t, gwtest.Statement("Allow", "lambda:*", "*"))
_, err := e.do(http.MethodPost, "/2015-03-31/functions", createBody("hello", func(b map[string]any) {
b["Role"] = "arn:aws:iam::" + gwtest.AccountID + ":role/ghost"
}))
gwtest.AssertDenied(t, err)
})
}
func TestRegistration_ResolvesActionsAndInventory(t *testing.T) {
reg := awsapi.NewRegistration(awsapi.Deps{})
assert.Equal(t, []string{"CreateFunction", "GetFunction"}, reg.Inventory.Registered)
assert.Equal(t, "CreateFunction", reg.ResolveAction(httptest.NewRequest(http.MethodPost, "/2015-03-31/functions", nil)))
assert.Equal(t, "GetFunction", reg.ResolveAction(httptest.NewRequest(http.MethodGet, "/2015-03-31/functions/x", nil)))
assert.Empty(t, reg.ResolveAction(httptest.NewRequest(http.MethodDelete, "/2015-03-31/functions/x", nil)))
}
@@ -0,0 +1,196 @@
// Package awsapi serves the Lambda REST-JSON control plane (API version
// 2015-03-31) for the actions this build supports.
package awsapi
import (
"context"
"crypto/cipher"
"errors"
"io"
"log/slog"
"net/http"
"time"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/private/protocol/json/jsonutil"
"github.com/mulgadc/spinifex/spinifex/domains/lambda/codestore"
"github.com/mulgadc/spinifex/spinifex/domains/lambda/function"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/arn"
awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/passrole"
"github.com/mulgadc/spinifex/spinifex/ingress/aws/dispatch"
"github.com/mulgadc/spinifex/spinifex/ingress/aws/envelope"
)
// ServiceName is the SigV4 credential-scope service Lambda clients sign for.
const ServiceName = "lambda"
// servicePrincipal is the service a function execution role is passed to.
const servicePrincipal = "lambda.amazonaws.com"
// CodeStore is what the handlers need of the Regional code store.
type CodeStore interface {
Put(ctx context.Context, ref codestore.Ref, code []byte) (string, error)
PresignGet(ctx context.Context, ref codestore.Ref, sha256Hex string, ttl time.Duration) (string, error)
}
var _ CodeStore = (*codestore.Store)(nil)
// CodeStoreFunc opens the code store on first use; it fails while the store is
// not ready.
type CodeStoreFunc func(ctx context.Context) (CodeStore, error)
// Deps are the composed capabilities the Lambda control plane dispatches to.
type Deps struct {
Meta function.Client
Codes CodeStoreFunc
Roles passrole.Resolver
// Trusts reports whether the role's trust policy lets Lambda assume it for
// the account; false is a refusal, an error an outage.
Trusts func(accountID, roleARN string) (bool, error)
// AEAD seals function environment variables; it is the master key's cipher.
AEAD cipher.AEAD
}
// NewRegistration returns the Lambda REST-JSON registration.
func NewRegistration(deps Deps) dispatch.Registration {
return dispatch.Registration{
Service: ServiceName,
ResolveAction: resolveAction,
Dispatch: deps.dispatch,
Errors: dispatch.ErrorEnvelopeRESTJSON,
Inventory: dispatch.Inventory{Registered: actionNames()},
}
}
func resolveAction(r *http.Request) string {
action, _, _, _ := router.Lookup(r.Method, r.URL.EscapedPath())
return action
}
func readBody(r *http.Request) ([]byte, error) {
body, err := io.ReadAll(io.LimitReader(r.Body, maxBodyBytes+1))
if err != nil {
return nil, errors.New(awserrors.ErrorInvalidParameterValueException)
}
if len(body) > maxBodyBytes {
return nil, errors.New(awserrors.ErrorRequestEntityTooLarge)
}
return body, nil
}
// functionARN names the function the request is about, or "*" when the request
// does not yet name a legal one, so a malformed request stays the handler's
// validation fault rather than becoming an authorization failure.
func functionARN(region, accountID, name string) string {
if region == "" || accountID == "" || !nameRE.MatchString(name) {
return "*"
}
return arn.FormatLambdaFunction(region, accountID, name)
}
// serve resolves the route, authorizes the request, then runs the action.
func (d Deps) serve(w http.ResponseWriter, inv dispatch.Invocation) error {
r := inv.Request
action, params, handler, ok := router.Lookup(r.Method, r.URL.EscapedPath())
if !ok {
return errors.New(awserrors.ErrorInvalidAction)
}
if inv.AccountID == "" {
slog.ErrorContext(r.Context(), "Lambda dispatch: no account ID in auth context")
return errors.New(awserrors.ErrorInternalError)
}
body, err := readBody(r)
if err != nil {
return err
}
c := &call{AccountID: inv.AccountID, Region: inv.Region, Params: params, Query: r.URL.Query(), Body: body}
// The region is the gateway's own, never the caller-supplied credential
// scope, so a caller cannot sign for another region to slide out from
// under a Deny.
resource := "*"
var decodeErr error
switch action {
case "CreateFunction":
c.Create, decodeErr = decodeCreate(body)
if decodeErr == nil {
resource = functionARN(inv.Region, inv.AccountID, aws.StringValue(c.Create.FunctionName))
}
case "GetFunction":
if name, _, err := resolveFunctionName(inv.Region, inv.AccountID, params[0]); err == nil {
resource = functionARN(inv.Region, inv.AccountID, name)
}
}
if err := inv.Authorize(ServiceName, action, []string{resource}, nil); err != nil {
return err
}
if decodeErr != nil {
return decodeErr
}
if action == "CreateFunction" {
if err := d.checkPassRole(inv, aws.StringValue(c.Create.Role)); err != nil {
return err
}
}
if d.Meta == nil || d.Codes == nil {
return errors.New(awserrors.ErrorServiceException)
}
out, err := handler(r.Context(), d, c)
if err != nil {
return err
}
writeJSON(w, out)
return nil
}
func (d Deps) dispatch(w http.ResponseWriter, inv dispatch.Invocation) error {
return d.serve(w, inv)
}
// checkPassRole enforces iam:PassRole on the execution role and that its trust
// policy admits Lambda. Only the exact stored ARN resolves, and a missing
// resolver fails closed.
func (d Deps) checkPassRole(inv dispatch.Invocation, roleARN string) error {
if roleARN == "" {
return nil
}
if d.Roles == nil || d.Trusts == nil {
return errors.New(awserrors.ErrorServiceException)
}
role, err := passrole.Check(d.Roles, passrole.AuthorizeFunc(inv.Authorize), inv.AccountID, roleARN, servicePrincipal)
switch {
case err == nil:
trusted, terr := d.Trusts(inv.AccountID, role.ARN)
if terr != nil {
return terr
}
if !trusted {
return invalid("The role defined for the function cannot be assumed by Lambda.")
}
return nil
case errors.Is(err, passrole.ErrMalformed), errors.Is(err, passrole.ErrForeignAccount):
return invalid("The role defined for the function cannot be assumed by Lambda.")
case errors.Is(err, passrole.ErrNotFound):
return invalid("The role defined for the function cannot be assumed by Lambda.")
default:
return err
}
}
func writeJSON(w http.ResponseWriter, out reply) {
body, err := jsonutil.BuildJSON(out.Body)
if err != nil {
slog.Error("Failed to marshal Lambda response", "err", err)
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusInternalServerError)
_, _ = w.Write(envelope.JSONBody(awserrors.ErrorServiceException, "failed to marshal response"))
return
}
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(out.Status)
if _, err := w.Write(body); err != nil {
slog.Error("Failed to write Lambda response", "err", err)
}
}
+56
View File
@@ -0,0 +1,56 @@
package awsapi
import (
"context"
"net/url"
"slices"
"github.com/aws/aws-sdk-go/service/lambda"
"github.com/mulgadc/spinifex/spinifex/ingress/aws/rest"
)
// call is one authorized request as a handler sees it.
type call struct {
AccountID string
Region string
Params []string
Query url.Values
Body []byte
// Create is the decoded CreateFunction body, decoded once ahead of the
// authorization gate and the PassRole check that both read it.
Create *lambda.CreateFunctionInput
}
// reply is a handler's success: the HTTP status and the SDK shape to encode.
type reply struct {
Status int
Body any
}
type route = rest.Route[routeHandler]
type routeHandler func(ctx context.Context, d Deps, c *call) (reply, error)
// routes is the dispatch table. Lambda API version 2015-03-31.
var routes = []route{
{Method: "POST", Pattern: "/2015-03-31/functions", Action: "CreateFunction",
Handler: func(ctx context.Context, d Deps, c *call) (reply, error) {
return d.createFunction(ctx, c)
}},
{Method: "GET", Pattern: "/2015-03-31/functions/{FunctionName}", Action: "GetFunction",
Handler: func(ctx context.Context, d Deps, c *call) (reply, error) {
return d.getFunction(ctx, c)
}},
}
// actionNames returns the distinct actions in routes in stable order.
func actionNames() []string {
names := make([]string, 0, len(routes))
for _, r := range routes {
names = append(names, r.Action)
}
slices.Sort(names)
return slices.Compact(names)
}
var router = rest.NewRouter("lambda", routes)
+259
View File
@@ -0,0 +1,259 @@
package awsapi
import (
"archive/zip"
"bytes"
"encoding/json"
"reflect"
"regexp"
"strings"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/lambda"
"github.com/mulgadc/spinifex/spinifex/domains/lambda/codestore"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/arn"
awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
)
const (
supportedRuntime = "provided.al2023"
defaultTimeout = 3
maxTimeout = 900
defaultMemory = 128
maxMemory = 10240
maxDescription = 256
maxEnvironmentLen = 4096
// maxBodyBytes bounds a CreateFunction body: the base64 of the largest
// code copy plus room for the rest of the document.
maxBodyBytes = (codestore.MaxCodeBytes/3+1)*4 + 1<<20
)
var nameRE = regexp.MustCompile(`^[a-zA-Z0-9-_]{1,64}$`)
func invalid(format string, args ...any) error {
return awserrors.Errorf(awserrors.ErrorInvalidParameterValueException, format, args...)
}
func unsupported(field string) error {
return invalid("%s is not supported", field)
}
// decodeCreate reads a CreateFunction body strictly: a member this build does
// not model is refused rather than silently dropped.
func decodeCreate(body []byte) (*lambda.CreateFunctionInput, error) {
in := new(lambda.CreateFunctionInput)
dec := json.NewDecoder(bytes.NewReader(body))
dec.DisallowUnknownFields()
if err := dec.Decode(in); err != nil {
return nil, invalid("the request body is not a valid CreateFunction document: %v", err)
}
return in, nil
}
// createParams is a CreateFunction request after validation, defaults applied.
type createParams struct {
Name string
Role string
Handler string
Description string
TimeoutSec int64
MemorySize int64
Code []byte
Env map[string]string
}
// isSet reports whether a modelled optional member carries anything: a non-nil
// pointer to a non-zero value, a non-empty list or map, or a struct with any
// such member.
func isSet(v any) bool {
return valueSet(reflect.ValueOf(v))
}
func valueSet(v reflect.Value) bool {
switch v.Kind() {
case reflect.Pointer, reflect.Interface:
return !v.IsNil() && valueSet(v.Elem())
case reflect.Slice, reflect.Map:
return v.Len() > 0
case reflect.Struct:
for i := range v.NumField() {
if v.Type().Field(i).IsExported() && valueSet(v.Field(i)) {
return true
}
}
return false
case reflect.Invalid:
return false
default:
return !v.IsZero()
}
}
// refuseUnsupported names the first supplied member this build cannot honour.
func refuseUnsupported(in *lambda.CreateFunctionInput) error {
checks := []struct {
field string
set bool
}{
{"Code.S3Bucket", in.Code != nil && isSet(in.Code.S3Bucket)},
{"Code.S3Key", in.Code != nil && isSet(in.Code.S3Key)},
{"Code.S3ObjectVersion", in.Code != nil && isSet(in.Code.S3ObjectVersion)},
{"Code.ImageUri", in.Code != nil && isSet(in.Code.ImageUri)},
{"Layers", isSet(in.Layers)},
{"VpcConfig", isSet(in.VpcConfig)},
{"KMSKeyArn", isSet(in.KMSKeyArn)},
{"DeadLetterConfig", isSet(in.DeadLetterConfig)},
{"TracingConfig", in.TracingConfig != nil && aws.StringValue(in.TracingConfig.Mode) == lambda.TracingModeActive},
{"FileSystemConfigs", isSet(in.FileSystemConfigs)},
{"ImageConfig", isSet(in.ImageConfig)},
{"Architectures", isSet(in.Architectures) && (len(in.Architectures) != 1 || aws.StringValue(in.Architectures[0]) != lambda.ArchitectureX8664)},
{"Publish", aws.BoolValue(in.Publish)},
{"SnapStart", in.SnapStart != nil && aws.StringValue(in.SnapStart.ApplyOn) != "" && aws.StringValue(in.SnapStart.ApplyOn) != lambda.SnapStartApplyOnNone},
{"EphemeralStorage", in.EphemeralStorage != nil && in.EphemeralStorage.Size != nil && aws.Int64Value(in.EphemeralStorage.Size) != 512},
{"CodeSigningConfigArn", isSet(in.CodeSigningConfigArn)},
{"Tags", isSet(in.Tags)},
}
for _, c := range checks {
if c.set {
return unsupported(c.field)
}
}
return nil
}
// validateCreate checks a decoded CreateFunction and applies the defaults.
// resolveFunctionName accepts the GetFunction forms AWS does: a name, a partial
// ARN (account:function:name) or a full ARN, each with an optional qualifier.
// Another account or Region is reported not found under the ARN it named.
func resolveFunctionName(region, accountID, raw string) (name, qualifier string, err error) {
parts := strings.Split(raw, ":")
qualified := false
switch {
case len(parts) <= 2:
name = parts[0]
if len(parts) == 2 {
qualifier, qualified = parts[1], true
}
case len(parts) <= 4 && parts[1] == "function":
if parts[0] != accountID {
return "", "", awserrors.Errorf(awserrors.ErrorResourceNotFoundException, "Function not found: %s",
arn.FormatLambdaFunction(region, parts[0], strings.Join(parts[2:], ":")))
}
name = parts[2]
if len(parts) == 4 {
qualifier, qualified = parts[3], true
}
case len(parts) >= 7 && len(parts) <= 8 && parts[0] == "arn" && parts[2] == "lambda" && parts[5] == "function":
if parts[1] != "aws" || parts[3] != region || parts[4] != accountID {
return "", "", awserrors.Errorf(awserrors.ErrorResourceNotFoundException, "Function not found: %s", raw)
}
name = parts[6]
if len(parts) == 8 {
qualifier, qualified = parts[7], true
}
default:
return "", "", invalidFunctionName()
}
if !nameRE.MatchString(name) || (qualified && qualifier == "") {
return "", "", invalidFunctionName()
}
return name, qualifier, nil
}
func invalidFunctionName() error {
return invalid("1 validation error detected: Value at 'functionName' failed to satisfy constraint: Member must satisfy regular expression pattern: (arn:(aws[a-zA-Z-]*)?:lambda:)?([a-z]{2}(-gov)?-[a-z]+-\\d{1}:)?(\\d{12}:)?(function:)?([a-zA-Z0-9-_]+)(:(\\$LATEST|[a-zA-Z0-9-_]+))?")
}
// logGroup is the log group AWS names for a function by default.
func logGroup(name string) string { return "/aws/lambda/" + name }
// defaultLogging reports whether c is absent or spells AWS's default exactly;
// without a logs capability any other value would be inert.
func defaultLogging(c *lambda.LoggingConfig, name string) bool {
if c == nil {
return true
}
format, group := aws.StringValue(c.LogFormat), aws.StringValue(c.LogGroup)
return (format == "" || format == lambda.LogFormatText) && (group == "" || group == logGroup(name)) &&
c.ApplicationLogLevel == nil && c.SystemLogLevel == nil
}
func validateCreate(in *lambda.CreateFunctionInput) (*createParams, error) {
if err := refuseUnsupported(in); err != nil {
return nil, err
}
name := aws.StringValue(in.FunctionName)
if strings.HasPrefix(name, "arn:") || !nameRE.MatchString(name) {
return nil, invalid("1 validation error detected: Value at 'functionName' failed to satisfy constraint: Member must satisfy regular expression pattern: [a-zA-Z0-9-_]+ and have length 1 to 64")
}
if !defaultLogging(in.LoggingConfig, name) {
return nil, unsupported("A LoggingConfig other than LogFormat Text with log group " + logGroup(name))
}
if aws.StringValue(in.Runtime) != supportedRuntime {
return nil, invalid("Runtime must be %s", supportedRuntime)
}
if pt := aws.StringValue(in.PackageType); pt != "" && pt != lambda.PackageTypeZip {
return nil, invalid("PackageType must be %s", lambda.PackageTypeZip)
}
if in.Code == nil || len(in.Code.ZipFile) == 0 {
return nil, invalid("Code.ZipFile is required")
}
if len(in.Code.ZipFile) > codestore.MaxCodeBytes {
return nil, awserrors.Errorf(awserrors.ErrorRequestEntityTooLarge, "Code.ZipFile exceeds %d bytes", codestore.MaxCodeBytes)
}
if _, err := zip.NewReader(bytes.NewReader(in.Code.ZipFile), int64(len(in.Code.ZipFile))); err != nil {
return nil, invalid("Could not unzip uploaded file. Please check your file, then try to upload again.")
}
handler := aws.StringValue(in.Handler)
if handler == "" {
return nil, invalid("Handler is required")
}
role := aws.StringValue(in.Role)
if role == "" {
return nil, invalid("Role is required")
}
limit, memory := int64(defaultTimeout), int64(defaultMemory)
if in.Timeout != nil {
limit = *in.Timeout
}
if limit < 1 || limit > maxTimeout {
return nil, invalid("Timeout must be between 1 and %d", maxTimeout)
}
if in.MemorySize != nil {
memory = *in.MemorySize
}
if memory < 128 || memory > maxMemory {
return nil, invalid("MemorySize must be between 128 and %d", maxMemory)
}
desc := aws.StringValue(in.Description)
if len(desc) > maxDescription {
return nil, invalid("Description must be at most %d characters", maxDescription)
}
env, err := envVars(in.Environment)
if err != nil {
return nil, err
}
return &createParams{
Name: name, Role: role, Handler: handler, Description: desc,
TimeoutSec: limit, MemorySize: memory, Code: in.Code.ZipFile, Env: env,
}, nil
}
func envVars(e *lambda.Environment) (map[string]string, error) {
if e == nil || len(e.Variables) == 0 {
return nil, nil
}
vars := make(map[string]string, len(e.Variables))
size := 0
for k, v := range e.Variables {
if k == "" {
return nil, invalid("Environment variable names must not be empty")
}
vars[k] = aws.StringValue(v)
size += len(k) + len(vars[k])
}
if size > maxEnvironmentLen {
return nil, invalid("Environment variables exceed %d bytes", maxEnvironmentLen)
}
return vars, nil
}
@@ -34,9 +34,15 @@ type sessionProvider struct {
value credentials.Value
expires time.Time
lastError time.Time
// minRemaining makes the next Retrieve re-assume unless the session has at
// least this long left; it is cleared by that Retrieve.
minRemaining time.Duration
}
var _ credentials.Provider = (*sessionProvider)(nil)
var (
_ credentials.Provider = (*sessionProvider)(nil)
_ credentials.Expirer = (*sessionProvider)(nil)
)
func newSessionProvider(assume AssumeFunc, now func() time.Time) *sessionProvider {
return &sessionProvider{assume: assume, now: now}
@@ -48,7 +54,13 @@ func (p *sessionProvider) Retrieve() (credentials.Value, error) {
now := p.now()
valid := p.value.AccessKeyID != "" && now.Before(p.expires)
if valid && (now.Before(p.expires.Add(-refreshMargin)) || now.Before(p.lastError.Add(retryInterval))) {
need := p.minRemaining
p.minRemaining = 0
if need > 0 {
if valid && p.expires.Sub(now) >= need {
return p.value, nil
}
} else if valid && (now.Before(p.expires.Add(-refreshMargin)) || now.Before(p.lastError.Add(retryInterval))) {
return p.value, nil
}
@@ -84,3 +96,18 @@ func (p *sessionProvider) IsExpired() bool {
defer p.mu.Unlock()
return p.value.AccessKeyID == "" || !p.now().Before(p.expires.Add(-refreshMargin))
}
// ExpiresAt is when the current session stops working.
func (p *sessionProvider) ExpiresAt() time.Time {
p.mu.Lock()
defer p.mu.Unlock()
return p.expires
}
// requireRemaining makes the next Retrieve draw a new session unless the
// current one outlasts d.
func (p *sessionProvider) requireRemaining(d time.Duration) {
p.mu.Lock()
defer p.mu.Unlock()
p.minRemaining = d
}
@@ -0,0 +1,158 @@
package codestore
import (
"errors"
"net/url"
"strings"
"testing"
"time"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/aws/credentials"
"github.com/aws/aws-sdk-go/aws/session"
"github.com/aws/aws-sdk-go/service/s3"
ststypes "github.com/aws/aws-sdk-go/service/sts"
"github.com/mulgadc/spinifex/spinifex/providers/objectstore"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func TestStore_PresignGet(t *testing.T) {
sess, err := session.NewSession(&aws.Config{
Endpoint: aws.String("https://predastore.test:8443"),
Region: aws.String("ap-southeast-2"),
Credentials: credentials.NewStaticCredentials("AKIATEST", "secret", "session-token"),
S3ForcePathStyle: aws.Bool(true),
})
require.NoError(t, err)
s := newStore(testRegion, objectstore.NewMemoryObjectStore())
s.presign = presignWith(s3.New(sess))
sum := "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
raw, err := s.PresignGet(t.Context(), testRef, sum, 10*time.Minute)
require.NoError(t, err)
u, err := url.Parse(raw)
require.NoError(t, err)
assert.Equal(t, "predastore.test:8443", u.Host)
assert.Equal(t, "/"+BucketName(testRegion)+"/"+copyKey(sum), u.Path)
q := u.Query()
assert.Equal(t, "600", q.Get("X-Amz-Expires"))
assert.Equal(t, "session-token", q.Get("X-Amz-Security-Token"))
assert.NotEmpty(t, q.Get("X-Amz-Signature"))
}
func TestStore_PresignGetRefusals(t *testing.T) {
s := newStore(testRegion, objectstore.NewMemoryObjectStore())
sum := "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
_, err := s.PresignGet(t.Context(), testRef, sum, time.Minute)
require.Error(t, err, "a store without a signer cannot presign")
s.presign = func(*s3.GetObjectInput, time.Duration) (string, error) { return "unused", nil }
_, err = s.PresignGet(t.Context(), testRef, "not-a-digest", time.Minute)
require.ErrorIs(t, err, ErrInvalidRef)
}
func presignStoreOn(t *testing.T, c *clock, a *fakeAssume) *Store {
t.Helper()
provider := newSessionProvider(a.assume, c.now)
creds := credentials.NewCredentials(provider)
sess, err := session.NewSession(&aws.Config{
Endpoint: aws.String("https://predastore.test:8443"), Region: aws.String("ap-southeast-2"),
Credentials: creds, S3ForcePathStyle: aws.Bool(true),
})
require.NoError(t, err)
s := newStore(testRegion, objectstore.NewMemoryObjectStore())
s.presign = presignWith(s3.New(sess))
s.creds, s.provider, s.now = creds, provider, c.now
return s
}
func signedKeyID(t *testing.T, raw string) string {
t.Helper()
u, err := url.Parse(raw)
require.NoError(t, err)
return strings.SplitN(u.Query().Get("X-Amz-Credential"), "/", 2)[0]
}
func TestStore_PresignGetRefreshesAShortSession(t *testing.T) {
sum := "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
c := &clock{t: time.Unix(1_000_000, 0)}
a := &fakeAssume{expires: c.t.Add(7 * time.Minute)}
s := presignStoreOn(t, c, a)
_, err := s.creds.Get()
require.NoError(t, err)
first := "ASIAB"
require.Equal(t, 1, a.calls)
a.expires = c.t.Add(time.Hour)
raw, err := s.PresignGet(t.Context(), testRef, sum, 10*time.Minute)
require.NoError(t, err)
assert.Equal(t, 2, a.calls, "a 7 minute session is re-assumed before a 10 minute signature")
assert.NotEqual(t, first, signedKeyID(t, raw))
assert.Equal(t, "ASIAC", signedKeyID(t, raw), "the URL is signed with the new session")
_, err = s.PresignGet(t.Context(), testRef, sum, 10*time.Minute)
require.NoError(t, err)
assert.Equal(t, 2, a.calls, "a long session is reused")
}
func TestStore_PresignGetRefusesWhenSessionStaysShort(t *testing.T) {
sum := "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
c := &clock{t: time.Unix(1_000_000, 0)}
a := &fakeAssume{expires: c.t.Add(7 * time.Minute)}
s := presignStoreOn(t, c, a)
_, err := s.PresignGet(t.Context(), testRef, sum, 10*time.Minute)
require.Error(t, err, "a re-assume that is still short signs nothing")
assert.Equal(t, 2, a.calls)
a.err = errors.New("AccessDenied")
_, err = s.PresignGet(t.Context(), testRef, sum, 10*time.Minute)
require.Error(t, err, "a failed re-assume does not fall back to the short session")
}
// Open signs download URLs for the off-host endpoint while its own reads and
// writes keep the local one, so a client outside the node can fetch the copy.
func TestOpen_PresignsForTheOffHostEndpoint(t *testing.T) {
assume := func() (*ststypes.AssumeRoleOutput, error) {
return &ststypes.AssumeRoleOutput{Credentials: &ststypes.Credentials{
AccessKeyId: aws.String("ASIATEST"),
SecretAccessKey: aws.String("secret"),
SessionToken: aws.String("session-token"),
Expiration: aws.Time(time.Now().Add(time.Hour)),
}}, nil
}
s, err := Open(OpenConfig{
Region: testRegion,
Endpoint: "https://127.0.0.1:8443",
SigningRegion: "ap-southeast-2",
Assume: assume,
PresignEndpoint: "https://192.0.2.15:8443",
})
require.NoError(t, err)
raw, err := s.PresignGet(t.Context(), testRef, strings.Repeat("a", 64), 10*time.Minute)
require.NoError(t, err)
u, err := url.Parse(raw)
require.NoError(t, err)
assert.Equal(t, "192.0.2.15:8443", u.Host)
}
func TestOpen_NoOffHostEndpointRefusesDownloadURLs(t *testing.T) {
assume := func() (*ststypes.AssumeRoleOutput, error) {
return &ststypes.AssumeRoleOutput{Credentials: &ststypes.Credentials{
AccessKeyId: aws.String("ASIATEST"),
SecretAccessKey: aws.String("secret"),
SessionToken: aws.String("session-token"),
Expiration: aws.Time(time.Now().Add(time.Hour)),
}}, nil
}
s, err := Open(OpenConfig{Region: testRegion, Endpoint: "https://127.0.0.1:8443", SigningRegion: "ap-southeast-2", Assume: assume})
require.NoError(t, err)
raw, err := s.PresignGet(t.Context(), testRef, strings.Repeat("a", 64), 10*time.Minute)
require.ErrorIs(t, err, ErrNoPresignEndpoint)
assert.Empty(t, raw)
}
+89 -2
View File
@@ -31,6 +31,8 @@ var (
ErrCodeTooLarge = errors.New("lambda code copy too large")
// ErrInvalidRef means a reference segment is empty or unsafe as a key segment.
ErrInvalidRef = errors.New("invalid lambda code reference")
// ErrNoPresignEndpoint means no off-host Predastore address is configured.
ErrNoPresignEndpoint = errors.New("lambda code store: no off-host endpoint for download URLs")
)
var (
@@ -71,6 +73,27 @@ func (r Ref) key(sha256Hex string) (string, error) {
type Store struct {
bucket string
objects objectstore.ObjectStore
presign presignFunc
// creds and provider are the scoped session behind presign; nil on stores
// built without one.
creds *credentials.Credentials
provider *sessionProvider
now func() time.Time
}
// presignMargin is slack kept between a signed URL's end and its session's.
const presignMargin = 30 * time.Second
// presignFunc signs a GET for the input; nil on stores that cannot presign.
type presignFunc func(in *s3.GetObjectInput, ttl time.Duration) (string, error)
// presignWith signs GETs with the session credentials behind client.
func presignWith(client *s3.S3) presignFunc {
return func(in *s3.GetObjectInput, ttl time.Duration) (string, error) {
req, _ := client.GetObjectRequest(in)
return req.Presign(ttl)
}
}
// OpenConfig is what the composition layer supplies to open the scoped store.
@@ -81,6 +104,9 @@ type OpenConfig struct {
Assume AssumeFunc
HTTPClient *http.Client
now func() time.Time
// PresignEndpoint is the Predastore URL off-host clients reach; download
// URLs are signed for it. Empty disables download URLs.
PresignEndpoint string
}
// Open builds the store on a refreshing code-store session and draws the first
@@ -97,7 +123,8 @@ func Open(cfg OpenConfig) (*Store, error) {
if client == nil {
client = &http.Client{Timeout: 120 * time.Second}
}
creds := credentials.NewCredentials(newSessionProvider(cfg.Assume, now))
provider := newSessionProvider(cfg.Assume, now)
creds := credentials.NewCredentials(provider)
if _, err := creds.Get(); err != nil {
return nil, err
}
@@ -111,7 +138,14 @@ func Open(cfg OpenConfig) (*Store, error) {
if err != nil {
return nil, fmt.Errorf("lambda code store session: %w", err)
}
return newStore(cfg.Region, objectstore.NewS3ObjectStore(s3.New(sess))), nil
s3Client := s3.New(sess)
store := newStore(cfg.Region, objectstore.NewS3ObjectStore(s3Client))
// Without an off-host address a download URL would name an internal one.
if cfg.PresignEndpoint != "" {
store.presign = presignWith(s3.New(sess, &aws.Config{Endpoint: aws.String(cfg.PresignEndpoint)}))
}
store.creds, store.provider, store.now = creds, provider, now
return store, nil
}
func newStore(region string, objects objectstore.ObjectStore) *Store {
@@ -169,6 +203,59 @@ func (s *Store) Get(ctx context.Context, ref Ref, wantSHA256 string) ([]byte, er
return code, nil
}
// ensureSession guarantees the signing session outlasts ttl, drawing a new one
// if it does not, and fails rather than sign a URL that would outlive it.
func (s *Store) ensureSession(ttl time.Duration) error {
if s.creds == nil || s.provider == nil {
return nil
}
need := ttl + presignMargin
if _, err := s.creds.Get(); err != nil {
return fmt.Errorf("lambda code store session: %w", err)
}
for attempt := range 2 {
exp, err := s.creds.ExpiresAt()
if err != nil {
return fmt.Errorf("lambda code store session expiry: %w", err)
}
if !exp.Before(s.now().Add(need)) {
return nil
}
if attempt == 1 {
break
}
s.provider.requireRemaining(need)
s.creds.Expire()
if _, err := s.creds.Get(); err != nil {
return fmt.Errorf("lambda code store session: %w", err)
}
}
return errors.New("lambda code store: session expires before the presigned URL would")
}
// PresignGet returns a time-limited GET URL for the copy with the recorded
// digest, signed by a session that outlasts the URL.
func (s *Store) PresignGet(ctx context.Context, ref Ref, sha256Hex string, ttl time.Duration) (string, error) {
if s.presign == nil {
return "", ErrNoPresignEndpoint
}
key, err := ref.key(sha256Hex)
if err != nil {
return "", err
}
if err := ctx.Err(); err != nil {
return "", err
}
if err := s.ensureSession(ttl); err != nil {
return "", err
}
url, err := s.presign(&s3.GetObjectInput{Bucket: aws.String(s.bucket), Key: aws.String(key)}, ttl)
if err != nil {
return "", fmt.Errorf("presign lambda code copy: %w", err)
}
return url, nil
}
// Delete removes the copy with the recorded digest; an absent copy is already
// deleted.
func (s *Store) Delete(ctx context.Context, ref Ref, sha256Hex string) error {
@@ -0,0 +1,67 @@
package function
import (
"context"
"time"
natsmsg "github.com/mulgadc/spinifex/spinifex/foundation/messaging/nats"
"github.com/nats-io/nats.go"
)
const metaRequestTimeout = 30 * time.Second
// Client is the gateway's view of the daemon-owned function records.
type Client interface {
// Create binds the name and returns the KV revision.
Create(ctx context.Context, accountID, region string, rec Record) (uint64, error)
// Get returns the record and revision, or ErrNotFound.
Get(ctx context.Context, accountID, region, name string) (Record, uint64, error)
// CodeStored marks the code stored and returns the stored record and revision.
CodeStored(ctx context.Context, accountID, region, name string, revision uint64, sha256Hex string, size int64) (Record, uint64, error)
// RollbackCreate frees the name while the record is still storing at revision.
RollbackCreate(ctx context.Context, accountID, region, name string, revision uint64) error
}
// NATSClient is the Client over NATS request/reply to the daemon.
type NATSClient struct {
conn *nats.Conn
}
var _ Client = (*NATSClient)(nil)
// NewNATSClient builds a NATS-backed Client.
func NewNATSClient(conn *nats.Conn) *NATSClient { return &NATSClient{conn: conn} }
func (c *NATSClient) Create(ctx context.Context, accountID, region string, rec Record) (uint64, error) {
resp, err := natsmsg.NATSRequest[CreateResponse](ctx, c.conn, SubjectCreate, &CreateRequest{Region: region, Record: rec}, metaRequestTimeout, accountID)
if err != nil {
return 0, err
}
return resp.Revision, nil
}
func (c *NATSClient) Get(ctx context.Context, accountID, region, name string) (Record, uint64, error) {
resp, err := natsmsg.NATSRequest[GetResponse](ctx, c.conn, SubjectGet, &GetRequest{Region: region, FunctionName: name}, metaRequestTimeout, accountID)
if err != nil {
return Record{}, 0, err
}
if !resp.Found {
return Record{}, 0, ErrNotFound
}
return resp.Record, resp.Revision, nil
}
func (c *NATSClient) CodeStored(ctx context.Context, accountID, region, name string, revision uint64, sha256Hex string, size int64) (Record, uint64, error) {
resp, err := natsmsg.NATSRequest[CodeStoredResponse](ctx, c.conn, SubjectCodeStored,
&CodeStoredRequest{Region: region, FunctionName: name, Revision: revision, SHA256: sha256Hex, Size: size}, metaRequestTimeout, accountID)
if err != nil {
return Record{}, 0, err
}
return resp.Record, resp.Revision, nil
}
func (c *NATSClient) RollbackCreate(ctx context.Context, accountID, region, name string, revision uint64) error {
_, err := natsmsg.NATSRequest[RollbackCreateResponse](ctx, c.conn, SubjectRollbackCreate,
&RollbackCreateRequest{Region: region, FunctionName: name, Revision: revision}, metaRequestTimeout, accountID)
return err
}
@@ -0,0 +1,56 @@
package function
import (
"crypto/cipher"
"crypto/rand"
"encoding/base64"
"encoding/json"
"errors"
"fmt"
)
func envAAD(accountID, functionID string) []byte {
return []byte("lambda-env/v1|" + accountID + "|" + functionID)
}
// SealEnv encrypts the variables as base64(nonce||ciphertext), bound to the
// account and function. No variables yield "".
func SealEnv(aead cipher.AEAD, accountID, functionID string, vars map[string]string) (string, error) {
if len(vars) == 0 {
return "", nil
}
plain, err := json.Marshal(vars)
if err != nil {
return "", fmt.Errorf("encode environment: %w", err)
}
nonce := make([]byte, aead.NonceSize())
if _, err := rand.Read(nonce); err != nil {
return "", fmt.Errorf("environment nonce: %w", err)
}
sealed := aead.Seal(nonce, nonce, plain, envAAD(accountID, functionID))
return base64.StdEncoding.EncodeToString(sealed), nil
}
// OpenEnv reverses SealEnv. A blob sealed for another account or function fails.
func OpenEnv(aead cipher.AEAD, accountID, functionID, sealed string) (map[string]string, error) {
if sealed == "" {
return nil, nil
}
raw, err := base64.StdEncoding.DecodeString(sealed)
if err != nil {
return nil, fmt.Errorf("decode environment: %w", err)
}
n := aead.NonceSize()
if len(raw) < n+aead.Overhead() {
return nil, errors.New("environment blob too short")
}
plain, err := aead.Open(nil, raw[:n], raw[n:], envAAD(accountID, functionID))
if err != nil {
return nil, fmt.Errorf("open environment: %w", err)
}
var vars map[string]string
if err := json.Unmarshal(plain, &vars); err != nil {
return nil, fmt.Errorf("decode environment: %w", err)
}
return vars, nil
}
@@ -0,0 +1,62 @@
package function
import (
"testing"
"github.com/mulgadc/bluebottle/pkg/masterkey"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
func testAEAD(t *testing.T) *masterkey.Key {
t.Helper()
k, err := masterkey.New(make([]byte, 32))
require.NoError(t, err)
return k
}
func TestEnv_RoundTripAndNoPlaintext(t *testing.T) {
k := testAEAD(t)
vars := map[string]string{"DB_PASSWORD": "hunter2", "MODE": "prod"}
sealed, err := SealEnv(k.AEAD, "000000000001", "fn1", vars)
require.NoError(t, err)
assert.NotContains(t, sealed, "hunter2")
got, err := OpenEnv(k.AEAD, "000000000001", "fn1", sealed)
require.NoError(t, err)
assert.Equal(t, vars, got)
again, err := SealEnv(k.AEAD, "000000000001", "fn1", vars)
require.NoError(t, err)
assert.NotEqual(t, sealed, again, "each seal draws a fresh nonce")
}
func TestEnv_BoundToAccountAndFunction(t *testing.T) {
k := testAEAD(t)
sealed, err := SealEnv(k.AEAD, "000000000001", "fn1", map[string]string{"A": "b"})
require.NoError(t, err)
_, err = OpenEnv(k.AEAD, "000000000001", "fn2", sealed)
require.Error(t, err)
_, err = OpenEnv(k.AEAD, "000000000002", "fn1", sealed)
require.Error(t, err)
_, err = OpenEnv(k.AEAD, "000000000001", "fn1", "not base64 !!")
require.Error(t, err)
_, err = OpenEnv(k.AEAD, "000000000001", "fn1", "AAAA")
require.Error(t, err)
}
func TestEnv_EmptyIsEmpty(t *testing.T) {
k := testAEAD(t)
sealed, err := SealEnv(k.AEAD, "a", "f", nil)
require.NoError(t, err)
assert.Empty(t, sealed)
sealed, err = SealEnv(k.AEAD, "a", "f", map[string]string{})
require.NoError(t, err)
assert.Empty(t, sealed)
got, err := OpenEnv(k.AEAD, "a", "f", "")
require.NoError(t, err)
assert.Empty(t, got)
}
+259
View File
@@ -0,0 +1,259 @@
package function
import (
"context"
"encoding/json"
"errors"
"fmt"
"log/slog"
"regexp"
"sync"
"time"
awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/foundation/state/kvutil"
"github.com/nats-io/nats.go/jetstream"
)
// ClaimTTL is how long a record still storing its code holds its name. It must
// exceed the gateway request deadline plus the longest permitted code write.
const ClaimTTL = 5 * time.Minute
// ErrNotFound means no record is bound to the name.
var ErrNotFound = errors.New("lambda function not found")
var (
nameRE = regexp.MustCompile(`^[a-zA-Z0-9-_]{1,64}$`)
regionRE = regexp.MustCompile(`^[a-z0-9-]{1,32}$`)
digestRE = regexp.MustCompile(`^[0-9a-f]{64}$`)
)
// ValidName reports whether name is a legal function name.
func ValidName(name string) bool { return nameRE.MatchString(name) }
func badRegion() error {
return awserrors.Errorf(awserrors.ErrorInvalidParameterValueException, "region is invalid")
}
// MetaService is the daemon-side owner of function records. Every update is a
// compare-and-swap on the revision the caller read or was handed.
type MetaService struct {
js jetstream.JetStream
now func() time.Time
mu sync.Mutex
buckets map[string]jetstream.KeyValue
}
// NewMetaService builds the service over a JetStream handle.
func NewMetaService(js jetstream.JetStream) *MetaService {
return &MetaService{js: js, now: time.Now, buckets: make(map[string]jetstream.KeyValue)}
}
func (s *MetaService) bucket(ctx context.Context, accountID string) (jetstream.KeyValue, error) {
s.mu.Lock()
defer s.mu.Unlock()
if kv, ok := s.buckets[accountID]; ok {
return kv, nil
}
name := KVAccountBucket(accountID)
kv, err := kvutil.GetOrCreateBucket(ctx, s.js, name, KVBucketAccountHistory)
if err != nil {
return nil, fmt.Errorf("lambda: open account bucket %s: %w", name, err)
}
s.buckets[accountID] = kv
return kv, nil
}
func conflict(format string, args ...any) error {
return awserrors.Errorf(awserrors.ErrorResourceConflictException, format, args...)
}
func decode(data []byte) (Record, error) {
var rec Record
if err := json.Unmarshal(data, &rec); err != nil {
return Record{}, awserrors.Errorf(awserrors.ErrorServiceException, "function record is unreadable")
}
if rec.FormatVersion != FormatVersion {
return Record{}, awserrors.Errorf(awserrors.ErrorServiceException,
"function record has unsupported format version %d", rec.FormatVersion)
}
return rec, nil
}
// Create binds the name by writing the record with kv.Create. The record starts
// Pending with its code storing and a claim that expires after ClaimTTL. A name
// held only by an expired, still-storing claim is taken over once.
//
// Orphaned code objects from rolled-back or taken-over claims are not cleaned up yet.
func (s *MetaService) Create(ctx context.Context, req *CreateRequest, accountID string) (*CreateResponse, error) {
rec := req.Record
if !regionRE.MatchString(req.Region) {
return nil, badRegion()
}
if !ValidName(rec.FunctionName) || rec.FunctionID == "" || rec.Code.RevisionID == "" || accountID == "" {
return nil, awserrors.Errorf(awserrors.ErrorInvalidParameterValueException, "function record is incomplete")
}
kv, err := s.bucket(ctx, accountID)
if err != nil {
return nil, err
}
now := s.now().UTC()
rec.FormatVersion = FormatVersion
rec.AccountID, rec.Region = accountID, req.Region
rec.State = StatePending
rec.StateReason, rec.StateReasonCode = "", ""
rec.Code = Code{RevisionID: rec.Code.RevisionID, State: CodeStoring}
rec.ClaimExpiresAt = now.Add(ClaimTTL)
rec.CreatedAt, rec.LastModified = now, now
data, err := json.Marshal(rec)
if err != nil {
return nil, err
}
key := KVFunctionKey(req.Region, rec.FunctionName)
rev, err := kv.Create(ctx, key, data)
if err == nil {
return &CreateResponse{Revision: rev}, nil
}
if !errors.Is(err, jetstream.ErrKeyExists) {
return nil, fmt.Errorf("lambda: create function record: %w", err)
}
exists := conflict("Function already exist: %s", rec.FunctionName)
entry, err := kv.Get(ctx, key)
if err != nil {
return nil, exists
}
old, err := decode(entry.Value())
if err != nil || old.Code.State != CodeStoring || !s.now().After(old.ClaimExpiresAt) {
return nil, exists
}
if err := kv.Delete(ctx, key, jetstream.LastRevision(entry.Revision())); err != nil {
return nil, exists
}
slog.WarnContext(ctx, "Lambda: took over an expired function name claim",
"function_name", rec.FunctionName, "old_function_id", old.FunctionID)
rev, err = kv.Create(ctx, key, data)
if err != nil {
if errors.Is(err, jetstream.ErrKeyExists) {
return nil, exists
}
return nil, fmt.Errorf("lambda: create function record: %w", err)
}
return &CreateResponse{Revision: rev}, nil
}
// Get returns the record, or Found false when the name is unbound.
func (s *MetaService) Get(ctx context.Context, req *GetRequest, accountID string) (*GetResponse, error) {
if !regionRE.MatchString(req.Region) {
return nil, badRegion()
}
kv, err := s.bucket(ctx, accountID)
if err != nil {
return nil, err
}
entry, err := kv.Get(ctx, KVFunctionKey(req.Region, req.FunctionName))
if errors.Is(err, jetstream.ErrKeyNotFound) {
return &GetResponse{}, nil
}
if err != nil {
return nil, fmt.Errorf("lambda: get function record: %w", err)
}
rec, err := decode(entry.Value())
if err != nil {
return nil, err
}
return &GetResponse{Found: true, Record: rec, Revision: entry.Revision()}, nil
}
// readStoring loads the record and checks it is the one the caller created: the
// revision matches and the code is still storing.
func (s *MetaService) readStoring(ctx context.Context, kv jetstream.KeyValue, region, name string, revision uint64) (Record, error) {
entry, err := kv.Get(ctx, KVFunctionKey(region, name))
if errors.Is(err, jetstream.ErrKeyNotFound) {
return Record{}, ErrNotFound
}
if err != nil {
return Record{}, fmt.Errorf("lambda: get function record: %w", err)
}
if entry.Revision() != revision {
return Record{}, conflict("Function was modified by another request: %s", name)
}
rec, err := decode(entry.Value())
if err != nil {
return Record{}, err
}
if rec.Code.State != CodeStoring {
return Record{}, conflict("Function code is not being stored: %s", name)
}
return rec, nil
}
// CodeStored marks the code stored. It is refused, leaving the record unchanged,
// when the revision moved, the code is not storing, or the claim has expired.
func (s *MetaService) CodeStored(ctx context.Context, req *CodeStoredRequest, accountID string) (*CodeStoredResponse, error) {
if !regionRE.MatchString(req.Region) {
return nil, badRegion()
}
if !digestRE.MatchString(req.SHA256) || req.Size < 0 {
return nil, awserrors.Errorf(awserrors.ErrorInvalidParameterValueException, "code digest is invalid")
}
kv, err := s.bucket(ctx, accountID)
if err != nil {
return nil, err
}
rec, err := s.readStoring(ctx, kv, req.Region, req.FunctionName, req.Revision)
if errors.Is(err, ErrNotFound) {
return nil, awserrors.Errorf(awserrors.ErrorResourceNotFoundException, "Function not found: %s", req.FunctionName)
}
if err != nil {
return nil, err
}
now := s.now()
if now.After(rec.ClaimExpiresAt) {
return nil, conflict("Function creation claim expired: %s", req.FunctionName)
}
rec.Code.State, rec.Code.SHA256, rec.Code.Size = CodeStored, req.SHA256, req.Size
rec.State, rec.StateReasonCode, rec.StateReason = StatePending, ReasonCodeCreating, ReasonCreating
rec.LastModified = now.UTC()
data, err := json.Marshal(rec)
if err != nil {
return nil, err
}
rev, err := kv.Update(ctx, KVFunctionKey(req.Region, req.FunctionName), data, req.Revision)
if err != nil {
if errors.Is(err, jetstream.ErrKeyExists) {
return nil, conflict("Function was modified by another request: %s", req.FunctionName)
}
return nil, fmt.Errorf("lambda: update function record: %w", err)
}
return &CodeStoredResponse{Record: rec, Revision: rev}, nil
}
// RollbackCreate frees the name of a create whose own code write failed. It
// deletes only a record still storing at exactly the revision create returned;
// an already-absent name is success.
func (s *MetaService) RollbackCreate(ctx context.Context, req *RollbackCreateRequest, accountID string) (*RollbackCreateResponse, error) {
if !regionRE.MatchString(req.Region) {
return nil, badRegion()
}
kv, err := s.bucket(ctx, accountID)
if err != nil {
return nil, err
}
_, err = s.readStoring(ctx, kv, req.Region, req.FunctionName, req.Revision)
if errors.Is(err, ErrNotFound) {
return &RollbackCreateResponse{}, nil
}
if err != nil {
return nil, err
}
if err := kv.Delete(ctx, KVFunctionKey(req.Region, req.FunctionName), jetstream.LastRevision(req.Revision)); err != nil {
if errors.Is(err, jetstream.ErrKeyExists) {
return nil, conflict("Function was modified by another request: %s", req.FunctionName)
}
return nil, fmt.Errorf("lambda: delete function record: %w", err)
}
return &RollbackCreateResponse{}, nil
}
@@ -0,0 +1,263 @@
package function
import (
"encoding/json"
"testing"
"time"
"github.com/mulgadc/spinifex/internal/testkit"
awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
const (
testAccount = "000000000001"
testRegion = "ap-southeast-2"
testDigest = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
)
type clock struct{ t time.Time }
func (c *clock) now() time.Time { return c.t }
func newTestService(t *testing.T) (*MetaService, *clock) {
t.Helper()
_, nc, _ := testutil.StartTestJetStream(t)
svc := NewMetaService(testutil.NewJetStream(t, nc))
c := &clock{t: time.Date(2026, 11, 2, 10, 0, 0, 0, time.UTC)}
svc.now = c.now
return svc, c
}
func createReq(name, functionID string) *CreateRequest {
return &CreateRequest{Region: testRegion, Record: Record{
FunctionID: functionID, FunctionName: name, Role: "arn:aws:iam::000000000001:role/r",
Runtime: "provided.al2023", Handler: "bootstrap", Timeout: 3, MemorySize: 128,
Code: Code{RevisionID: "rev-" + functionID},
}}
}
func mustCreate(t *testing.T, svc *MetaService, name, functionID string) uint64 {
t.Helper()
resp, err := svc.Create(t.Context(), createReq(name, functionID), testAccount)
require.NoError(t, err)
return resp.Revision
}
func mustGet(t *testing.T, svc *MetaService, name string) *GetResponse {
t.Helper()
resp, err := svc.Get(t.Context(), &GetRequest{Region: testRegion, FunctionName: name}, testAccount)
require.NoError(t, err)
return resp
}
func requireCode(t *testing.T, err error, code string) {
t.Helper()
require.Error(t, err)
got, ok := awserrors.ResolveErrorCode(err)
require.True(t, ok, "no AWS code on %v", err)
assert.Equal(t, code, got)
}
func TestMeta_CreateOwnsStateAndRejectsDuplicate(t *testing.T) {
svc, c := newTestService(t)
req := createReq("hello", "f1")
req.Record.State = StateActive
req.Record.Code.State = CodeStored
req.Record.AccountID = "someone-else"
resp, err := svc.Create(t.Context(), req, testAccount)
require.NoError(t, err)
got := mustGet(t, svc, "hello")
require.True(t, got.Found)
assert.Equal(t, resp.Revision, got.Revision)
assert.Equal(t, StatePending, got.Record.State, "the daemon owns state")
assert.Equal(t, CodeStoring, got.Record.Code.State)
assert.Equal(t, testAccount, got.Record.AccountID, "the account comes from the request header")
assert.Equal(t, FormatVersion, got.Record.FormatVersion)
assert.Equal(t, c.t.Add(ClaimTTL), got.Record.ClaimExpiresAt)
_, err = svc.Create(t.Context(), createReq("hello", "f2"), testAccount)
requireCode(t, err, awserrors.ErrorResourceConflictException)
assert.Equal(t, "f1", mustGet(t, svc, "hello").Record.FunctionID, "the first record is untouched")
other, err := svc.Create(t.Context(), createReq("hello", "f3"), "000000000002")
require.NoError(t, err, "names are per account")
assert.NotZero(t, other.Revision)
}
func TestMeta_CreateRefusesIncompleteRecords(t *testing.T) {
svc, _ := newTestService(t)
for name, mutate := range map[string]func(*Record){
"bad name": func(r *Record) { r.FunctionName = "has space" },
"no id": func(r *Record) { r.FunctionID = "" },
"no revision": func(r *Record) { r.Code.RevisionID = "" },
} {
t.Run(name, func(t *testing.T) {
req := createReq("ok", "f1")
mutate(&req.Record)
_, err := svc.Create(t.Context(), req, testAccount)
requireCode(t, err, awserrors.ErrorInvalidParameterValueException)
})
}
}
func TestMeta_CodeStoredUsesCAS(t *testing.T) {
svc, _ := newTestService(t)
rev := mustCreate(t, svc, "hello", "f1")
_, err := svc.CodeStored(t.Context(), &CodeStoredRequest{Region: testRegion, FunctionName: "hello", Revision: rev + 5, SHA256: testDigest, Size: 9}, testAccount)
requireCode(t, err, awserrors.ErrorResourceConflictException)
assert.Equal(t, CodeStoring, mustGet(t, svc, "hello").Record.Code.State, "a stale revision changes nothing")
_, err = svc.CodeStored(t.Context(), &CodeStoredRequest{Region: testRegion, FunctionName: "hello", Revision: rev, SHA256: "xyz", Size: 9}, testAccount)
requireCode(t, err, awserrors.ErrorInvalidParameterValueException)
resp, err := svc.CodeStored(t.Context(), &CodeStoredRequest{Region: testRegion, FunctionName: "hello", Revision: rev, SHA256: testDigest, Size: 9}, testAccount)
require.NoError(t, err)
got := mustGet(t, svc, "hello")
assert.Equal(t, resp.Revision, got.Revision)
assert.Equal(t, CodeStored, got.Record.Code.State)
assert.Equal(t, testDigest, got.Record.Code.SHA256)
assert.Equal(t, int64(9), got.Record.Code.Size)
assert.Equal(t, StatePending, got.Record.State)
assert.Equal(t, ReasonCodeCreating, got.Record.StateReasonCode)
assert.Equal(t, ReasonCreating, got.Record.StateReason)
_, err = svc.CodeStored(t.Context(), &CodeStoredRequest{Region: testRegion, FunctionName: "hello", Revision: resp.Revision, SHA256: testDigest, Size: 9}, testAccount)
requireCode(t, err, awserrors.ErrorResourceConflictException)
_, err = svc.CodeStored(t.Context(), &CodeStoredRequest{Region: testRegion, FunctionName: "ghost", Revision: 1, SHA256: testDigest}, testAccount)
requireCode(t, err, awserrors.ErrorResourceNotFoundException)
}
func TestMeta_RollbackDeletesOnlyStoringAtTheRightRevision(t *testing.T) {
svc, _ := newTestService(t)
rev := mustCreate(t, svc, "hello", "f1")
_, err := svc.RollbackCreate(t.Context(), &RollbackCreateRequest{Region: testRegion, FunctionName: "hello", Revision: rev + 1}, testAccount)
requireCode(t, err, awserrors.ErrorResourceConflictException)
require.True(t, mustGet(t, svc, "hello").Found)
_, err = svc.RollbackCreate(t.Context(), &RollbackCreateRequest{Region: testRegion, FunctionName: "hello", Revision: rev}, testAccount)
require.NoError(t, err)
assert.False(t, mustGet(t, svc, "hello").Found)
_, err = svc.RollbackCreate(t.Context(), &RollbackCreateRequest{Region: testRegion, FunctionName: "hello", Revision: rev}, testAccount)
require.NoError(t, err, "rolling back an absent name is success")
mustCreate(t, svc, "hello", "f2")
assert.Equal(t, "f2", mustGet(t, svc, "hello").Record.FunctionID, "the name is free again")
rev = mustGet(t, svc, "hello").Revision
stored, err := svc.CodeStored(t.Context(), &CodeStoredRequest{Region: testRegion, FunctionName: "hello", Revision: rev, SHA256: testDigest, Size: 1}, testAccount)
require.NoError(t, err)
_, err = svc.RollbackCreate(t.Context(), &RollbackCreateRequest{Region: testRegion, FunctionName: "hello", Revision: stored.Revision}, testAccount)
requireCode(t, err, awserrors.ErrorResourceConflictException)
assert.True(t, mustGet(t, svc, "hello").Found, "a stored record is never rolled back")
}
func TestMeta_GetMissingAndUnknownFormat(t *testing.T) {
svc, _ := newTestService(t)
assert.False(t, mustGet(t, svc, "ghost").Found)
kv, err := svc.bucket(t.Context(), testAccount)
require.NoError(t, err)
data, err := json.Marshal(Record{FormatVersion: 99, FunctionName: "future"})
require.NoError(t, err)
_, err = kv.Create(t.Context(), KVFunctionKey(testRegion, "future"), data)
require.NoError(t, err)
_, err = svc.Get(t.Context(), &GetRequest{Region: testRegion, FunctionName: "future"}, testAccount)
requireCode(t, err, awserrors.ErrorServiceException)
_, err = svc.Create(t.Context(), createReq("future", "f1"), testAccount)
requireCode(t, err, awserrors.ErrorResourceConflictException)
_, err = svc.RollbackCreate(t.Context(), &RollbackCreateRequest{Region: testRegion, FunctionName: "future", Revision: 1}, testAccount)
requireCode(t, err, awserrors.ErrorServiceException)
}
func TestMeta_ExpiredStoringClaimIsTakenOver(t *testing.T) {
svc, c := newTestService(t)
oldRev := mustCreate(t, svc, "hello", "old")
c.t = c.t.Add(ClaimTTL - time.Second)
_, err := svc.Create(t.Context(), createReq("hello", "new"), testAccount)
requireCode(t, err, awserrors.ErrorResourceConflictException)
assert.Equal(t, "old", mustGet(t, svc, "hello").Record.FunctionID, "an unexpired claim holds the name")
c.t = c.t.Add(2 * time.Second)
newRev := mustCreate(t, svc, "hello", "new")
got := mustGet(t, svc, "hello")
assert.Equal(t, "new", got.Record.FunctionID)
assert.Equal(t, "rev-new", got.Record.Code.RevisionID)
assert.Equal(t, newRev, got.Revision)
assert.Equal(t, c.t.Add(ClaimTTL), got.Record.ClaimExpiresAt)
_, err = svc.CodeStored(t.Context(), &CodeStoredRequest{Region: testRegion, FunctionName: "hello", Revision: oldRev, SHA256: testDigest, Size: 1}, testAccount)
requireCode(t, err, awserrors.ErrorResourceConflictException)
_, err = svc.RollbackCreate(t.Context(), &RollbackCreateRequest{Region: testRegion, FunctionName: "hello", Revision: oldRev}, testAccount)
requireCode(t, err, awserrors.ErrorResourceConflictException)
got = mustGet(t, svc, "hello")
assert.Equal(t, "new", got.Record.FunctionID, "the stale gateway cannot resurrect or free the old claim")
assert.Equal(t, CodeStoring, got.Record.Code.State)
}
func TestMeta_StoredRecordIsNeverTakenOver(t *testing.T) {
svc, c := newTestService(t)
rev := mustCreate(t, svc, "hello", "f1")
_, err := svc.CodeStored(t.Context(), &CodeStoredRequest{Region: testRegion, FunctionName: "hello", Revision: rev, SHA256: testDigest, Size: 1}, testAccount)
require.NoError(t, err)
c.t = c.t.Add(10 * ClaimTTL)
_, err = svc.Create(t.Context(), createReq("hello", "f2"), testAccount)
requireCode(t, err, awserrors.ErrorResourceConflictException)
assert.Equal(t, "f1", mustGet(t, svc, "hello").Record.FunctionID)
}
func TestMeta_CodeStoredAfterExpiryIsRefused(t *testing.T) {
svc, c := newTestService(t)
rev := mustCreate(t, svc, "hello", "f1")
c.t = c.t.Add(ClaimTTL + time.Second)
_, err := svc.CodeStored(t.Context(), &CodeStoredRequest{Region: testRegion, FunctionName: "hello", Revision: rev, SHA256: testDigest, Size: 1}, testAccount)
requireCode(t, err, awserrors.ErrorResourceConflictException)
got := mustGet(t, svc, "hello")
assert.Equal(t, CodeStoring, got.Record.Code.State, "a refused update leaves the record unchanged")
assert.Equal(t, rev, got.Revision)
}
func TestMeta_RegionsAreIndependent(t *testing.T) {
svc, _ := newTestService(t)
sydney := createReq("hello", "syd")
oregon := createReq("hello", "pdx")
oregon.Region = "us-west-2"
_, err := svc.Create(t.Context(), sydney, testAccount)
require.NoError(t, err)
_, err = svc.Create(t.Context(), oregon, testAccount)
require.NoError(t, err, "the same name in another Region is a different function")
got, err := svc.Get(t.Context(), &GetRequest{Region: "us-west-2", FunctionName: "hello"}, testAccount)
require.NoError(t, err)
assert.Equal(t, "pdx", got.Record.FunctionID)
assert.Equal(t, "us-west-2", got.Record.Region)
assert.Equal(t, "syd", mustGet(t, svc, "hello").Record.FunctionID)
assert.Equal(t, testRegion, mustGet(t, svc, "hello").Record.Region)
_, err = svc.Create(t.Context(), createReq("hello", "again"), testAccount)
requireCode(t, err, awserrors.ErrorResourceConflictException)
}
func TestMeta_RefusesBadRegion(t *testing.T) {
svc, _ := newTestService(t)
for _, region := range []string{"", "AP-SOUTH", "a.b", "a b", "x/y"} {
req := createReq("hello", "f1")
req.Region = region
_, err := svc.Create(t.Context(), req, testAccount)
requireCode(t, err, awserrors.ErrorInvalidParameterValueException)
_, err = svc.Get(t.Context(), &GetRequest{Region: region, FunctionName: "hello"}, testAccount)
requireCode(t, err, awserrors.ErrorInvalidParameterValueException)
_, err = svc.CodeStored(t.Context(), &CodeStoredRequest{Region: region, FunctionName: "hello", Revision: 1, SHA256: testDigest}, testAccount)
requireCode(t, err, awserrors.ErrorInvalidParameterValueException)
_, err = svc.RollbackCreate(t.Context(), &RollbackCreateRequest{Region: region, FunctionName: "hello", Revision: 1}, testAccount)
requireCode(t, err, awserrors.ErrorInvalidParameterValueException)
}
}
+131
View File
@@ -0,0 +1,131 @@
// Package function holds the Lambda function record and the daemon-owned
// metadata service that creates and updates it.
package function
import "time"
// FormatVersion is the only record layout this build reads or writes.
const FormatVersion = 1
// Function states and the code copy states.
const (
StatePending = "Pending"
StateActive = "Active"
StateFailed = "Failed"
CodeStoring = "storing"
CodeStored = "stored"
CodeFailed = "failed"
)
// Reason reported while a function waits for a runtime image to exist.
const (
ReasonCodeCreating = "Creating"
ReasonCreating = "The function is waiting for its runtime image."
)
// Code describes the stored copy of a function's code.
type Code struct {
RevisionID string `json:"revisionId"`
SHA256 string `json:"sha256,omitempty"`
Size int64 `json:"size,omitempty"`
State string `json:"state"`
}
// Record is one function. Its KV key both binds the name and holds it, so name
// reservation and the record are a single atomic write.
type Record struct {
FormatVersion int `json:"formatVersion"`
FunctionID string `json:"functionId"`
AccountID string `json:"accountId"`
Region string `json:"region"`
FunctionName string `json:"functionName"`
Role string `json:"role"`
Runtime string `json:"runtime"`
Handler string `json:"handler"`
Description string `json:"description,omitempty"`
Timeout int64 `json:"timeout"`
MemorySize int64 `json:"memorySize"`
EnvCipher string `json:"envCipher,omitempty"`
State string `json:"state"`
StateReason string `json:"stateReason,omitempty"`
StateReasonCode string `json:"stateReasonCode,omitempty"`
Code Code `json:"code"`
// ClaimExpiresAt bounds how long a record still storing its code holds the name.
ClaimExpiresAt time.Time `json:"claimExpiresAt"`
CreatedAt time.Time `json:"createdAt"`
LastModified time.Time `json:"lastModified"`
}
// KV bucket and key layout. The bucket is per account and owned by the daemon.
const (
KVBucketAccountPrefix = "lambda-account-"
KVBucketAccountHistory = 1
kvFunctionKeyPrefix = "fn."
)
// KVAccountBucket returns the per-account JetStream KV bucket name.
func KVAccountBucket(accountID string) string { return KVBucketAccountPrefix + accountID }
// KVFunctionKey returns the key that binds a function name in a Region and
// holds its record. JetStream is shared across Regions, so the Region is part
// of the key.
func KVFunctionKey(region, name string) string { return kvFunctionKeyPrefix + region + "." + name }
// NATS subjects served by the daemon's metadata service.
const (
SubjectCreate = "lambda.function.create"
SubjectGet = "lambda.function.get"
SubjectCodeStored = "lambda.function.code-stored"
SubjectRollbackCreate = "lambda.function.rollback-create"
)
// CreateRequest carries the record the gateway built; the daemon owns the
// state, claim and timestamps and overwrites whatever the request holds there.
type CreateRequest struct {
Region string `json:"region"`
Record Record `json:"record"`
}
// CreateResponse returns the KV revision the later steps must present.
type CreateResponse struct {
Revision uint64 `json:"revision"`
}
// GetRequest names the function.
type GetRequest struct {
Region string `json:"region"`
FunctionName string `json:"functionName"`
}
// GetResponse holds the record when Found.
type GetResponse struct {
Found bool `json:"found"`
Record Record `json:"record"`
Revision uint64 `json:"revision"`
}
// CodeStoredRequest records that the code copy landed.
type CodeStoredRequest struct {
Region string `json:"region"`
FunctionName string `json:"functionName"`
Revision uint64 `json:"revision"`
SHA256 string `json:"sha256"`
Size int64 `json:"size"`
}
// CodeStoredResponse returns the record as stored and its revision.
type CodeStoredResponse struct {
Record Record `json:"record"`
Revision uint64 `json:"revision"`
}
// RollbackCreateRequest withdraws a create whose own code write failed.
type RollbackCreateRequest struct {
Region string `json:"region"`
FunctionName string `json:"functionName"`
Revision uint64 `json:"revision"`
}
// RollbackCreateResponse is empty: success means the name is free.
type RollbackCreateResponse struct{}
+9 -6
View File
@@ -254,12 +254,6 @@ const assumeRoleConditionKeySourceAccount = "aws:SourceAccount"
// Deny statements, second pass scans all Allows. A single-pass loop would skip a later
// Deny and silently grant access.
func evalTrustPolicy(docJSON, callerARN string, principalSources []string) error {
doc, err := iamdomain.ValidateTrustPolicyDocument(docJSON)
if err != nil {
// Docs are validated at write time; reaching here implies on-disk corruption.
return fmt.Errorf("stored trust policy invalid: %w", err)
}
// sourceAccount is the caller ARN's own account — available whenever the
// caller is an authenticated HTTPS SigV4 identity, and not caller-asserted.
// It is empty on the IMDS path, where principalSource is synthesised and
@@ -268,6 +262,15 @@ func evalTrustPolicy(docJSON, callerARN string, principalSources []string) error
if caller, ok := parsePrincipalARN(callerARN); ok {
sourceAccount = caller.account
}
return evalTrustPolicyFor(docJSON, callerARN, sourceAccount, principalSources)
}
func evalTrustPolicyFor(docJSON, callerARN, sourceAccount string, principalSources []string) error {
doc, err := iamdomain.ValidateTrustPolicyDocument(docJSON)
if err != nil {
// Docs are validated at write time; reaching here implies on-disk corruption.
return fmt.Errorf("stored trust policy invalid: %w", err)
}
for _, stmt := range doc.Statement {
if stmt.Effect != iamdomain.PolicyEffectDeny {
+37
View File
@@ -0,0 +1,37 @@
package sts
import (
"errors"
"github.com/aws/aws-sdk-go/aws"
awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
)
// ServiceTrusts reports whether trustDoc lets service assume the role on behalf
// of sourceAccount, with the same deny-first evaluation AssumeRole uses. A nil
// error means trusted; a refusal is AccessDenied.
func ServiceTrusts(trustDoc, service, sourceAccount string) error {
return evalTrustPolicyFor(trustDoc, "", sourceAccount, []string{service})
}
// RoleTrusts evaluates a role's stored trust policy for service acting for an
// account: false is a refusal (or no such role), an error an outage.
func RoleTrusts(svc RoleGetter, service string) func(accountID, roleARN string) (bool, error) {
return func(accountID, roleARN string) (bool, error) {
_, role, err := ResolveRoleByARN(svc, roleARN)
if errors.Is(err, ErrRoleUnresolved) {
return false, nil
}
if err != nil {
return false, err
}
err = ServiceTrusts(aws.StringValue(role.AssumeRolePolicyDocument), service, accountID)
if err == nil {
return true, nil
}
if code, ok := awserrors.ResolveErrorCode(err); ok && code == awserrors.ErrorAccessDenied {
return false, nil
}
return false, err
}
}
+8
View File
@@ -0,0 +1,8 @@
package arn
import "fmt"
// FormatLambdaFunction builds arn:aws:lambda:<region>:<account>:function:<name>.
func FormatLambdaFunction(region, accountID, name string) string {
return fmt.Sprintf("arn:aws:lambda:%s:%s:function:%s", region, accountID, name)
}
@@ -620,6 +620,9 @@ var (
ErrorMissingRequiredParameterException = "MissingRequiredParameterException"
ErrorSerializationException = "SerializationException"
ErrorUnknownOperationException = "UnknownOperationException"
// Lambda wire codes that no other service shares.
ErrorResourceConflictException = "ResourceConflictException"
ErrorServiceException = "ServiceException"
)
// ValidErrorCode returns the error code if it exists in ErrorLookup,
@@ -1452,6 +1455,8 @@ var ErrorLookup = map[string]ErrorMessage{
ErrorMissingRequiredParameterException: {HTTPCode: 400, Message: "The request is missing a required parameter."},
ErrorSerializationException: {HTTPCode: 400, Message: "The request body could not be deserialized."},
ErrorUnknownOperationException: {HTTPCode: 400, Message: "The requested operation is not recognized."},
ErrorResourceConflictException: {HTTPCode: 409, Message: "The resource already exists or is in a state that does not allow this operation."},
ErrorServiceException: {HTTPCode: 500, Message: "The Lambda service encountered an internal error."},
// Route 53.
ErrorNoSuchHostedZone: {HTTPCode: 404, Message: "No hosted zone exists with the ID that you specified."},
@@ -0,0 +1,34 @@
package pipeline
import (
"bytes"
"net/http"
"net/http/httptest"
"testing"
"time"
"github.com/mulgadc/spinifex/internal/testkit/gwfix"
lambdaawsapi "github.com/mulgadc/spinifex/spinifex/domains/lambda/awsapi"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
// A refused CreateFunction leaves the real ingress in the REST-JSON shape the
// Lambda SDK decodes: the code in X-Amzn-Errortype, Type and message in the body.
func TestLambdaCreateFunctionErrorUsesRESTJSONEnvelope(t *testing.T) {
_, app := dispatchSeamApp(t, allowAllECRDocs(), lambdaawsapi.NewRegistration(lambdaawsapi.Deps{}))
body := []byte(`{"FunctionName":"hello","TenancyConfig":{}}`)
req := httptest.NewRequest(http.MethodPost, "/2015-03-31/functions", bytes.NewReader(body))
req.Host = "localhost:9999"
gwfix.SignRequestAs(t, req, body, gwfix.AccessKey, gwfix.SecretKey, lambdaawsapi.ServiceName, time.Now().UTC())
resp := doRequest(app, req)
assert.Equal(t, http.StatusBadRequest, resp.StatusCode)
assert.Equal(t, "application/json", resp.Header.Get("Content-Type"))
assert.Equal(t, "InvalidParameterValueException", resp.Header.Get("X-Amzn-Errortype"))
assert.NotEmpty(t, resp.Header.Get("X-Amzn-Requestid"))
env := decodeJSONError(t, resp)
require.NotEmpty(t, env.Message)
assert.Empty(t, env.Type, "REST-JSON carries the code in the header, not __type")
}
+14
View File
@@ -67,6 +67,7 @@ import (
ekscluster "github.com/mulgadc/spinifex/spinifex/domains/eks/cluster"
elbv2domain "github.com/mulgadc/spinifex/spinifex/domains/elbv2"
iamdomain "github.com/mulgadc/spinifex/spinifex/domains/iam"
lambdafunction "github.com/mulgadc/spinifex/spinifex/domains/lambda/function"
"github.com/mulgadc/spinifex/spinifex/domains/network/external"
"github.com/mulgadc/spinifex/spinifex/domains/network/external/dhcp"
"github.com/mulgadc/spinifex/spinifex/domains/network/external/exonet"
@@ -186,6 +187,7 @@ type Daemon struct {
ochreAppliance *ochrevector.Appliance
ochreBackupService *ochrevector.BackupService
ecrMetaService *ecr.MetaServiceImpl
lambdaMetaService *lambdafunction.MetaService
routeTableService *ec2routetable.RouteTableServiceImpl
natGatewayService *ec2natgw.NatGatewayServiceImpl
externalIPAM *ec2vpc.ExternalIPAM
@@ -1308,6 +1310,17 @@ func (d *Daemon) subscribeAll() error {
)
}
// Lambda gateway → daemon subscriptions. The daemon owns the per-account
// JetStream KV function records; code bytes never traverse these subjects.
if d.lambdaMetaService != nil {
subs = append(subs,
natsSub{lambdafunction.SubjectCreate, handleNATSRequest(d.node, d.lambdaMetaService.Create), "spinifex-workers"},
natsSub{lambdafunction.SubjectGet, handleNATSRequest(d.node, d.lambdaMetaService.Get), "spinifex-workers"},
natsSub{lambdafunction.SubjectCodeStored, handleNATSRequest(d.node, d.lambdaMetaService.CodeStored), "spinifex-workers"},
natsSub{lambdafunction.SubjectRollbackCreate, handleNATSRequest(d.node, d.lambdaMetaService.RollbackCreate), "spinifex-workers"},
)
}
// EIP handlers are always registered: without external IPAM d.eipService is
// the disabled stub, so reads return empty lists and mutations a clean
// UnsupportedOperation instead of a NATS timeout.
@@ -2408,6 +2421,7 @@ func (d *Daemon) startCluster() error {
slog.Warn("ECR metadata service disabled: JetStream unavailable", "err", jsErr)
} else {
d.ecrMetaService = ecr.NewKVMetaService(js)
d.lambdaMetaService = lambdafunction.NewMetaService(js)
}
if err := d.eksService.SpawnRegisteredReconcilers(); err != nil {
+27 -7
View File
@@ -41,6 +41,8 @@ import (
elbv2awsapi "github.com/mulgadc/spinifex/spinifex/domains/elbv2/awsapi"
iamdomain "github.com/mulgadc/spinifex/spinifex/domains/iam"
iamawsapi "github.com/mulgadc/spinifex/spinifex/domains/iam/awsapi"
lambdaawsapi "github.com/mulgadc/spinifex/spinifex/domains/lambda/awsapi"
lambdafunction "github.com/mulgadc/spinifex/spinifex/domains/lambda/function"
"github.com/mulgadc/spinifex/spinifex/domains/network/reconcile"
"github.com/mulgadc/spinifex/spinifex/domains/ochre"
ochreawsapi "github.com/mulgadc/spinifex/spinifex/domains/ochre/awsapi"
@@ -368,17 +370,35 @@ func launchService(config *config.ClusterConfig) error {
// Lambda code copies live in a Regional system bucket reached through a
// scoped role session. Nothing is provisioned until the store is first used.
lambdaCodeStore, err := newLambdaCodeStore(codeStoreWiring{
Region: nodeConfig.Region,
Node: config.Node,
Endpoint: netaddr.DialTarget(nodeConfig.Predastore.Host),
SigningRegion: nodeConfig.Predastore.Region,
RootAccessKey: nodeConfig.Predastore.AccessKey,
RootSecretKey: nodeConfig.Predastore.SecretKey,
Region: nodeConfig.Region,
Node: config.Node,
Endpoint: netaddr.DialTarget(nodeConfig.Predastore.Host),
PresignEndpoint: lambdaPresignEndpoint(nodeConfig.Predastore.Host, nodeConfig.AdvertiseIP),
SigningRegion: nodeConfig.Predastore.Region,
RootAccessKey: nodeConfig.Predastore.AccessKey,
RootSecretKey: nodeConfig.Predastore.SecretKey,
}, iamService, stsService)
if err != nil {
return err
}
_ = lambdaCodeStore // no Lambda handler is registered to use it yet
lambdaKey, err := masterkey.New(masterKey)
if err != nil {
return fmt.Errorf("lambda environment key: %w", err)
}
// Built but not routed: Lambda stays out of the public router until its runtime image exists.
_ = lambdaawsapi.NewRegistration(lambdaawsapi.Deps{
Meta: lambdafunction.NewNATSClient(natsConn),
Codes: func(ctx context.Context) (lambdaawsapi.CodeStore, error) {
store, err := lambdaCodeStore.Store(ctx)
if err != nil {
return nil, err
}
return store, nil
},
Roles: iamdomain.NewRoleResolver(iamService),
Trusts: lambdaRoleTrusts(iamService),
AEAD: lambdaKey.AEAD,
})
// ECR auth bridge: load (or first-run create) the ES256 signing key from the
// cluster-replicated awsgw-keys KV bucket, then build the token issuer
@@ -4,6 +4,7 @@ import (
"context"
"errors"
"fmt"
"net"
"net/http"
"time"
@@ -49,7 +50,7 @@ func (r *codeStoreRole) Create(trust string, tags map[string]string) error {
in.Tags = append(in.Tags, &iam.Tag{Key: aws.String(k), Value: aws.String(v)})
}
_, err := r.iam.CreateRole(identifiers.GlobalAccountID, in)
if err != nil && err.Error() == awserrors.ErrorIAMEntityAlreadyExists {
if code, _ := awserrors.ResolveErrorCode(err); code == awserrors.ErrorIAMEntityAlreadyExists {
return codestore.ErrExists
}
return err
@@ -59,7 +60,7 @@ func (r *codeStoreRole) Describe() (*codestore.RoleState, error) {
acct, name := identifiers.GlobalAccountID, aws.String(r.name)
got, err := r.iam.GetRole(acct, &iam.GetRoleInput{RoleName: name})
if err != nil {
if err.Error() == awserrors.ErrorIAMNoSuchEntity {
if code, _ := awserrors.ResolveErrorCode(err); code == awserrors.ErrorIAMNoSuchEntity {
return nil, codestore.ErrNotFound
}
return nil, err
@@ -162,6 +163,8 @@ type codeStoreWiring struct {
SigningRegion string
RootAccessKey string
RootSecretKey string
// PresignEndpoint is the off-host Predastore URL download links name.
PresignEndpoint string
}
// newLambdaCodeStore composes the code store. The root client and root keys stay
@@ -186,11 +189,29 @@ func newLambdaCodeStore(w codeStoreWiring, roles codeStoreRoleIAM, assumer codeS
}
open := func() (*codestore.Store, error) {
return codestore.Open(codestore.OpenConfig{
Region: w.Region,
Endpoint: w.Endpoint,
SigningRegion: w.SigningRegion,
Assume: func() (*sts.AssumeRoleOutput, error) { return assumer.AssumeCodeStoreRole(w.Region, sessionName) },
Region: w.Region,
Endpoint: w.Endpoint,
PresignEndpoint: w.PresignEndpoint,
SigningRegion: w.SigningRegion,
Assume: func() (*sts.AssumeRoleOutput, error) { return assumer.AssumeCodeStoreRole(w.Region, sessionName) },
})
}
return codestore.NewManager(provision, open), nil
}
// lambdaPresignEndpoint is the Predastore address an off-host client dials.
// Config loading turns a wildcard listener into loopback, so loopback and
// wildcard hosts use the advertise IP; empty when no off-host address is known.
func lambdaPresignEndpoint(predastoreHost, advertiseIP string) string {
host, port, err := net.SplitHostPort(predastoreHost)
if err != nil {
return ""
}
if ip := net.ParseIP(host); host == "" || host == "localhost" || (ip != nil && (ip.IsLoopback() || ip.IsUnspecified())) {
host = advertiseIP
}
if host == "" {
return ""
}
return "https://" + net.JoinHostPort(host, port)
}
@@ -19,6 +19,7 @@ import (
iamdomain "github.com/mulgadc/spinifex/spinifex/domains/iam"
"github.com/mulgadc/spinifex/spinifex/domains/lambda/codestore"
stsdomain "github.com/mulgadc/spinifex/spinifex/domains/sts"
awserrors "github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
@@ -180,7 +181,8 @@ func TestNewLambdaCodeStore_ProvisionsOnlyOnFirstUse(t *testing.T) {
roleName := aws.String(stsdomain.CodeStoreRoleName(codeStoreTestRegion))
_, err = iamSvc.GetRole(identifiers.GlobalAccountID, &iam.GetRoleInput{RoleName: roleName})
require.EqualError(t, err, "NoSuchEntity", "no role before first use")
code, _ := awserrors.ResolveErrorCode(err)
require.Equal(t, awserrors.ErrorIAMNoSuchEntity, code, "no role before first use")
assert.Empty(t, s3Calls, "no bucket request before first use")
_, err = mgr.Store(t.Context())
@@ -0,0 +1,24 @@
package awsgw
import (
"testing"
"github.com/stretchr/testify/assert"
)
func TestLambdaPresignEndpoint(t *testing.T) {
cases := []struct{ name, host, advertise, want string }{
{"wildcard uses the advertise IP", "0.0.0.0:8443", "192.168.0.15", "https://192.168.0.15:8443"},
{"loaded wildcard (loopback) uses the advertise IP", "127.0.0.1:8443", "192.168.0.15", "https://192.168.0.15:8443"},
{"localhost uses the advertise IP", "localhost:8443", "192.168.0.15", "https://192.168.0.15:8443"},
{"concrete listen host is kept", "10.22.7.1:8443", "192.168.0.15", "https://10.22.7.1:8443"},
{"wildcard with no advertise IP is unknown", "0.0.0.0:8443", "", ""},
{"loopback with no advertise IP is unknown", "127.0.0.1:8443", "", ""},
{"no port is unknown", "192.168.0.15", "192.168.0.15", ""},
}
for _, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
assert.Equal(t, tc.want, lambdaPresignEndpoint(tc.host, tc.advertise))
})
}
}
@@ -0,0 +1,13 @@
package awsgw
import (
stsdomain "github.com/mulgadc/spinifex/spinifex/domains/sts"
)
const lambdaServicePrincipal = "lambda.amazonaws.com"
// lambdaRoleTrusts evaluates an execution role's stored trust policy for
// lambda.amazonaws.com acting for the function's account.
func lambdaRoleTrusts(svc stsdomain.RoleGetter) func(accountID, roleARN string) (bool, error) {
return stsdomain.RoleTrusts(svc, lambdaServicePrincipal)
}
@@ -0,0 +1,50 @@
package awsgw
import (
"testing"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/service/iam"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/identifiers"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
// The trust decision is the stored role's real trust policy evaluated for
// lambda.amazonaws.com, so only a role Lambda could assume is accepted.
func TestLambdaRoleTrusts_EvaluatesTheStoredTrustPolicy(t *testing.T) {
iamSvc, _ := newCodeStoreIAM(t)
acct := identifiers.GlobalAccountID
trusts := lambdaRoleTrusts(iamSvc)
allowLambda := `{"Effect":"Allow","Principal":{"Service":"lambda.amazonaws.com"},"Action":"sts:AssumeRole"}`
cases := []struct {
name, statements string
want bool
}{
{"trusts lambda", allowLambda, true},
{"trusts only ec2", `{"Effect":"Allow","Principal":{"Service":"ec2.amazonaws.com"},"Action":"sts:AssumeRole"}`, false},
{"explicit deny wins", allowLambda + `,{"Effect":"Deny","Principal":{"Service":"lambda.amazonaws.com"},"Action":"sts:AssumeRole"}`, false},
{"source account matches", `{"Effect":"Allow","Principal":{"Service":"lambda.amazonaws.com"},"Action":"sts:AssumeRole","Condition":{"StringEquals":{"aws:SourceAccount":"` + acct + `"}}}`, true},
{"source account differs", `{"Effect":"Allow","Principal":{"Service":"lambda.amazonaws.com"},"Action":"sts:AssumeRole","Condition":{"StringEquals":{"aws:SourceAccount":"123456789012"}}}`, false},
}
for i, tc := range cases {
t.Run(tc.name, func(t *testing.T) {
name := "exec-" + string(rune('a'+i))
out, err := iamSvc.CreateRole(acct, &iam.CreateRoleInput{
RoleName: aws.String(name),
AssumeRolePolicyDocument: aws.String(`{"Version":"2012-10-17","Statement":[` + tc.statements + `]}`),
})
require.NoError(t, err)
got, err := trusts(acct, aws.StringValue(out.Role.Arn))
require.NoError(t, err)
assert.Equal(t, tc.want, got)
})
}
t.Run("missing role is not trusted", func(t *testing.T) {
got, err := trusts(acct, "arn:aws:iam::"+acct+":role/absent")
require.NoError(t, err)
assert.False(t, got)
})
}
+4 -4
View File
@@ -3,11 +3,11 @@
package integration
import (
"github.com/mulgadc/spinifex/internal/testkit/gwfix"
"bytes"
"encoding/json"
"encoding/xml"
"fmt"
"github.com/mulgadc/spinifex/internal/testkit/gwfix"
"io"
"net/http"
"net/url"
@@ -21,12 +21,12 @@ import (
"github.com/aws/aws-sdk-go/service/iam"
acmawsapi "github.com/mulgadc/spinifex/spinifex/domains/acm/awsapi"
awsapi "github.com/mulgadc/spinifex/spinifex/domains/ecr/awsapi"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/policy"
ecsawsapi "github.com/mulgadc/spinifex/spinifex/domains/ecs/awsapi"
eksawsapi "github.com/mulgadc/spinifex/spinifex/domains/eks/awsapi"
elbv2awsapi "github.com/mulgadc/spinifex/spinifex/domains/elbv2/awsapi"
iamawsapi "github.com/mulgadc/spinifex/spinifex/domains/iam/awsapi"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/errors"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/policy"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)
@@ -523,7 +523,7 @@ func iamUncased() map[string]string {
"PutGroupPolicy", "PutRolePolicy", "PutUserPolicy", "RemoveRoleFromInstanceProfile",
"RemoveUserFromGroup", "TagInstanceProfile", "TagRole", "TagUser",
"UntagInstanceProfile", "UntagRole", "UntagUser", "UpdateAccessKey",
"UpdateAssumeRolePolicy", "UpdateRole", "UpdateRoleDescription",
"UpdateAssumeRolePolicy", "UpdateRole", "UpdateRoleDescription", "UpdateUser",
} {
uncased[action] = byName
}
+169
View File
@@ -0,0 +1,169 @@
//go:build integration
package integration
import (
"archive/zip"
"bytes"
"context"
"crypto/rand"
"crypto/sha256"
"encoding/base64"
"errors"
"io"
"net/http"
"testing"
"time"
"github.com/aws/aws-sdk-go/aws"
"github.com/aws/aws-sdk-go/aws/awserr"
"github.com/aws/aws-sdk-go/aws/credentials"
"github.com/aws/aws-sdk-go/aws/session"
"github.com/aws/aws-sdk-go/service/iam"
"github.com/aws/aws-sdk-go/service/lambda"
"github.com/aws/aws-sdk-go/service/s3"
ststypes "github.com/aws/aws-sdk-go/service/sts"
"github.com/mulgadc/bluebottle/pkg/masterkey"
"github.com/mulgadc/spinifex/internal/testkit/gwfix"
iamdomain "github.com/mulgadc/spinifex/spinifex/domains/iam"
lambdaawsapi "github.com/mulgadc/spinifex/spinifex/domains/lambda/awsapi"
"github.com/mulgadc/spinifex/spinifex/domains/lambda/codestore"
lambdafunction "github.com/mulgadc/spinifex/spinifex/domains/lambda/function"
stsdomain "github.com/mulgadc/spinifex/spinifex/domains/sts"
"github.com/mulgadc/spinifex/spinifex/foundation/aws/passrole"
awsdispatch "github.com/mulgadc/spinifex/spinifex/ingress/aws/dispatch"
testpredastore "github.com/mulgadc/spinifex/tests/fixtures/predastore"
"github.com/nats-io/nats.go"
"github.com/nats-io/nats.go/jetstream"
"github.com/stretchr/testify/require"
)
const lambdaTrustPolicy = `{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Principal":{"Service":"lambda.amazonaws.com"},"Action":"sts:AssumeRole"}]}`
// The gateway's NATS connection and IAM service exist only once
// StartGatewayWithServices has built them, after the registration list.
type lateMeta struct{ lambdafunction.Client }
type lateRoles struct{ passrole.Resolver }
// startLambdaDaemonLite subscribes the production MetaService to the subjects
// the gateway's Lambda client dials, on JetStream KV like the live daemon.
func startLambdaDaemonLite(t *testing.T, nc *nats.Conn) {
t.Helper()
js, err := jetstream.New(nc)
require.NoError(t, err)
svc := lambdafunction.NewMetaService(js)
sub(t, nc, lambdafunction.SubjectCreate, func(m *nats.Msg) { dispatch(m, svc.Create) })
sub(t, nc, lambdafunction.SubjectGet, func(m *nats.Msg) { dispatch(m, svc.Get) })
sub(t, nc, lambdafunction.SubjectCodeStored, func(m *nats.Msg) { dispatch(m, svc.CodeStored) })
sub(t, nc, lambdafunction.SubjectRollbackCreate, func(m *nats.Msg) { dispatch(m, svc.RollbackCreate) })
}
// TestLambda_SignedCreateAndGetDownloadsTheStoredCode drives CreateFunction and
// GetFunction through signed ingress, the NATS-owned function record and the
// real code store on the Predastore fixture, then downloads Code.Location.
// The fixture authenticates only its service key, so that key stands in for
// the scoped code-store session; provisioning and session scoping are not here.
func TestLambda_SignedCreateAndGetDownloadsTheStoredCode(t *testing.T) {
fixture := testpredastore.Start(t)
bucket := codestore.BucketName(testRegion)
if _, err := fixtureS3(t, fixture).CreateBucket(&s3.CreateBucketInput{Bucket: aws.String(bucket)}); err != nil {
var ae awserr.Error
require.True(t, errors.As(err, &ae) && ae.Code() == s3.ErrCodeBucketAlreadyOwnedByYou, "create %s: %v", bucket, err)
}
store, err := codestore.Open(codestore.OpenConfig{
Region: testRegion,
Endpoint: "https://" + fixture.Host,
PresignEndpoint: "https://" + fixture.Host,
SigningRegion: fixture.Region,
Assume: func() (*ststypes.AssumeRoleOutput, error) {
return &ststypes.AssumeRoleOutput{Credentials: &ststypes.Credentials{
AccessKeyId: aws.String(fixture.AccessKey),
SecretAccessKey: aws.String(fixture.SecretKey),
Expiration: aws.Time(time.Now().Add(time.Hour)),
}}, nil
},
})
require.NoError(t, err)
keyBytes := make([]byte, 32)
_, err = rand.Read(keyBytes)
require.NoError(t, err)
envKey, err := masterkey.New(keyBytes)
require.NoError(t, err)
meta, roles := &lateMeta{}, &lateRoles{}
var trusts func(accountID, roleARN string) (bool, error)
gw := StartGatewayWithServices(t, []awsdispatch.Registration{lambdaawsapi.NewRegistration(lambdaawsapi.Deps{
Meta: meta,
Codes: func(context.Context) (lambdaawsapi.CodeStore, error) { return store, nil },
Roles: roles,
Trusts: func(accountID, roleARN string) (bool, error) { return trusts(accountID, roleARN) },
AEAD: envKey.AEAD,
})})
iamSvc, ok := gwfix.DepsOf(gw.Config).IAM.(*iamdomain.IAMServiceImpl)
require.True(t, ok, "harness IAM service type")
meta.Client = lambdafunction.NewNATSClient(gw.NATSConn)
roles.Resolver = iamdomain.NewRoleResolver(iamSvc)
trusts = stsdomain.RoleTrusts(iamSvc, "lambda.amazonaws.com")
startLambdaDaemonLite(t, gw.NATSConn)
role, err := gw.IAMClient(t).CreateRole(&iam.CreateRoleInput{
RoleName: aws.String("lambda-e2e-exec"), AssumeRolePolicyDocument: aws.String(lambdaTrustPolicy),
})
require.NoError(t, err)
code := lambdaTestZip(t)
sum := sha256.Sum256(code)
cli := lambda.New(gw.session(t))
created, err := cli.CreateFunction(&lambda.CreateFunctionInput{
FunctionName: aws.String("e2e-fn"),
Runtime: aws.String("provided.al2023"),
Handler: aws.String("bootstrap"),
Role: role.Role.Arn,
Code: &lambda.FunctionCode{ZipFile: code},
Environment: &lambda.Environment{Variables: map[string]*string{"MODE": aws.String("e2e")}},
})
require.NoError(t, err)
require.Equal(t, base64.StdEncoding.EncodeToString(sum[:]), aws.StringValue(created.CodeSha256))
got, err := cli.GetFunction(&lambda.GetFunctionInput{FunctionName: aws.String("e2e-fn")})
require.NoError(t, err)
require.Equal(t, "e2e", aws.StringValue(got.Configuration.Environment.Variables["MODE"]))
require.NotNil(t, got.Code.Location)
// SSL_CERT_FILE (set by the fixture) makes the default client trust it.
resp, err := http.Get(aws.StringValue(got.Code.Location)) //nolint:gosec,noctx // URL is the presigned location under test
require.NoError(t, err)
defer func() { _ = resp.Body.Close() }()
require.Equal(t, http.StatusOK, resp.StatusCode)
downloaded, err := io.ReadAll(resp.Body)
require.NoError(t, err)
require.Equal(t, code, downloaded)
}
// fixtureS3 signs as the fixture's service key; the fixture's certificate is in
// the process trust store via SSL_CERT_FILE.
func fixtureS3(t *testing.T, fixture *testpredastore.Fixture) *s3.S3 {
t.Helper()
sess, err := session.NewSession(&aws.Config{
Region: aws.String(fixture.Region),
Endpoint: aws.String("https://" + fixture.Host),
Credentials: credentials.NewStaticCredentials(fixture.AccessKey, fixture.SecretKey, ""),
S3ForcePathStyle: aws.Bool(true),
})
require.NoError(t, err)
return s3.New(sess)
}
func lambdaTestZip(t *testing.T) []byte {
t.Helper()
var buf bytes.Buffer
zw := zip.NewWriter(&buf)
w, err := zw.Create("bootstrap")
require.NoError(t, err)
_, err = w.Write([]byte("#!/bin/sh\necho e2e\n"))
require.NoError(t, err)
require.NoError(t, zw.Close())
return buf.Bytes()
}