seaweedfs/weed/s3api/lance/handlers_test.go

package lance

import (
	"encoding/json"
	"net/http"
	"net/http/httptest"
	"strings"
	"testing"
	"time"

	"github.com/gorilla/mux"
	"github.com/seaweedfs/seaweedfs/weed/s3api/s3err"
	"github.com/seaweedfs/seaweedfs/weed/s3api/s3tables"
	"github.com/seaweedfs/seaweedfs/weed/s3api/s3tables/s3tablestest"
)

// openAuthenticator stands in for a gateway with no IAM configured, which is the
// mode the namespace falls open in.
type openAuthenticator struct{}

func (openAuthenticator) AuthenticateRequest(*http.Request) (string, interface{}, s3err.ErrorCode) {
	return s3tables.DefaultAccountID, nil, s3err.ErrNone
}

func (openAuthenticator) DefaultAllow() bool { return true }

func (openAuthenticator) AuthEnabled() bool { return false }

// signedAuthenticator admits the same caller on a gateway that checks
// credentials.
type signedAuthenticator struct{ openAuthenticator }

func (signedAuthenticator) AuthEnabled() bool { return true }

type testHarness struct {
	router *mux.Router
	filer  *s3tablestest.MemFiler
	admin  *s3tables.Manager
	server *Server
}

func newTestHarness(t *testing.T) *testHarness {
	t.Helper()
	filer := s3tablestest.Start(t)
	server := NewServer(s3tables.NewManagerClient(filer.Client), openAuthenticator{})
	server.SetS3Endpoint("http://127.0.0.1:8333")
	server.SetS3Region(s3tables.DefaultRegion)
	router := mux.NewRouter().SkipClean(true)
	server.RegisterRoutes(router)

	// Table buckets are created by an operator, not by a namespace client, so
	// the harness seeds them the way the shell and admin console would.
	admin := s3tables.NewManager()
	admin.SetTrusted(true)
	return &testHarness{router: router, filer: filer, admin: admin, server: server}
}

// createBucket seeds a table bucket the Lance namespace can then be pointed at.
// An empty format strips the declaration afterwards, which is the only way to
// get the shape a bucket made before formats existed has: one that still
// accepts either format.
func (h *testHarness) createBucket(t *testing.T, name, format string) {
	t.Helper()
	var resp s3tables.CreateTableBucketResponse
	declared := format
	if declared == "" {
		declared = s3tables.FormatIceberg
	}
	err := h.admin.Execute(t.Context(), s3tables.NewManagerClient(h.filer.Client), "CreateTableBucket",
		&s3tables.CreateTableBucketRequest{Name: name, Format: declared}, &resp, "")
	if err != nil {
		t.Fatalf("create table bucket %s: %v", name, err)
	}
	if format == "" {
		h.undeclareBucket(t, name)
	}
}

// undeclareBucket removes a bucket's format from its metadata, ageing it back to
// what the filer holds for a bucket created before the field existed.
func (h *testHarness) undeclareBucket(t *testing.T, name string) {
	t.Helper()
	entry := h.filer.Get(s3tables.TablesPath, name)
	if entry == nil {
		t.Fatalf("table bucket %s is not in the filer", name)
	}
	var metadata map[string]any
	if err := json.Unmarshal(entry.Extended[s3tables.ExtendedKeyMetadata], &metadata); err != nil {
		t.Fatalf("read bucket metadata: %v", err)
	}
	delete(metadata, "format")
	updated, err := json.Marshal(metadata)
	if err != nil {
		t.Fatalf("write bucket metadata: %v", err)
	}
	extended := map[string][]byte{}
	for key, value := range entry.Extended {
		extended[key] = value
	}
	extended[s3tables.ExtendedKeyMetadata] = updated
	h.filer.Put(s3tables.TablesPath, name, extended)
}

func (h *testHarness) bucketARN(t *testing.T, name string) string {
	t.Helper()
	arn, err := s3tables.BuildBucketARN(s3tables.DefaultRegion, s3tables.DefaultAccountID, name)
	if err != nil {
		t.Fatalf("build arn: %v", err)
	}
	return arn
}

func (h *testHarness) do(t *testing.T, method, target, body string) *httptest.ResponseRecorder {
	t.Helper()
	var reader *strings.Reader
	if body == "" {
		reader = strings.NewReader("")
	} else {
		reader = strings.NewReader(body)
	}
	req := httptest.NewRequest(method, target, reader)
	recorder := httptest.NewRecorder()
	h.router.ServeHTTP(recorder, req)
	return recorder
}

func (h *testHarness) mustDo(t *testing.T, method, target, body string, want int) *httptest.ResponseRecorder {
	t.Helper()
	recorder := h.do(t, method, target, body)
	if recorder.Code != want {
		t.Fatalf("%s %s = %d (%s), want %d", method, target, recorder.Code, recorder.Body.String(), want)
	}
	return recorder
}

func decode[T any](t *testing.T, recorder *httptest.ResponseRecorder) T {
	t.Helper()
	var out T
	if err := json.Unmarshal(recorder.Body.Bytes(), &out); err != nil {
		t.Fatalf("decode %s: %v", recorder.Body.String(), err)
	}
	return out
}

// While the S3 port takes unsigned requests the catalog says so, and a client
// with no credentials of its own still reaches the data. Once auth is on it
// has to sign again.
func TestStorageOptionsSkipSignatureOnlyWhileS3AuthIsOff(t *testing.T) {
	h := newTestHarness(t)
	h.createBucket(t, "open", s3tables.FormatLance)
	h.mustDo(t, http.MethodPost, "/v1/namespace/open$ns/create", `{}`, http.StatusOK)

	declared := decode[DeclareTableResponse](t,
		h.mustDo(t, http.MethodPost, "/v1/table/open$ns$t/declare", `{}`, http.StatusOK))
	if declared.StorageOptions["aws_skip_signature"] != "true" {
		t.Fatalf("auth off: storage options = %v, want aws_skip_signature", declared.StorageOptions)
	}

	h.server.authenticator = signedAuthenticator{}
	described := decode[DescribeTableResponse](t,
		h.mustDo(t, http.MethodPost, "/v1/table/open$ns$t/describe", `{}`, http.StatusOK))
	if _, ok := described.StorageOptions["aws_skip_signature"]; ok {
		t.Fatalf("auth on: storage options = %v, want no aws_skip_signature", described.StorageOptions)
	}
}

// The lifecycle a Lance client drives: create the namespace, declare the table
// before any data exists, resolve it to a location, then deregister and bring it
// back by registering the same location.
func TestTableLifecycle(t *testing.T) {
	h := newTestHarness(t)

	h.createBucket(t, "analytics", s3tables.FormatLance)
	h.mustDo(t, http.MethodPost, "/v1/namespace/analytics$sales/create", `{}`, http.StatusOK)

	declared := decode[DeclareTableResponse](t,
		h.mustDo(t, http.MethodPost, "/v1/table/analytics$sales$orders/declare", `{}`, http.StatusOK))
	if declared.Location != "s3://analytics/sales/orders" {
		t.Fatalf("declared location = %q", declared.Location)
	}
	// The endpoint is plaintext, so a client that does not get allow_http fails
	// with what looks like a credential error.
	if declared.StorageOptions["aws_endpoint"] != "http://127.0.0.1:8333" ||
		declared.StorageOptions["allow_http"] != "true" {
		t.Fatalf("storage options = %v", declared.StorageOptions)
	}

	described := decode[DescribeTableResponse](t,
		h.mustDo(t, http.MethodPost, "/v1/table/analytics$sales$orders/describe?check_declared=true", `{}`, http.StatusOK))
	if described.Location != declared.Location {
		t.Fatalf("described location = %q, want %q", described.Location, declared.Location)
	}
	if described.IsOnlyDeclared == nil || !*described.IsOnlyDeclared {
		t.Fatalf("is_only_declared = %v, want true for a table with no data", described.IsOnlyDeclared)
	}

	listed := decode[ListTablesResponse](t,
		h.mustDo(t, http.MethodGet, "/v1/namespace/analytics$sales/table/list", "", http.StatusOK))
	if len(listed.Tables) != 1 || listed.Tables[0] != "analytics$sales$orders" {
		t.Fatalf("listed tables = %v, want the full identifier", listed.Tables)
	}

	h.mustDo(t, http.MethodPost, "/v1/table/analytics$sales$orders/exists", `{}`, http.StatusOK)

	h.mustDo(t, http.MethodPost, "/v1/table/analytics$sales$orders/deregister", `{}`, http.StatusOK)
	h.mustDo(t, http.MethodPost, "/v1/table/analytics$sales$orders/exists", `{}`, http.StatusNotFound)
	afterDrop := decode[ListTablesResponse](t,
		h.mustDo(t, http.MethodGet, "/v1/namespace/analytics$sales/table/list", "", http.StatusOK))
	if len(afterDrop.Tables) != 0 {
		t.Fatalf("deregistered table still listed: %v", afterDrop.Tables)
	}
	// Deregistering preserves the data. The catalog entry is the dataset
	// directory, so dropping it would take the dataset with it.
	if h.filer.Get(s3tables.GetNamespacePath("analytics", "sales"), "orders") == nil {
		t.Fatal("deregister deleted the dataset directory")
	}

	h.mustDo(t, http.MethodPost, "/v1/table/analytics$sales$orders/register",
		`{"location":"s3://analytics/sales/orders"}`, http.StatusOK)
	h.mustDo(t, http.MethodPost, "/v1/table/analytics$sales$orders/exists", `{}`, http.StatusOK)

	// Dropping is the operation that does remove the data.
	h.mustDo(t, http.MethodPost, "/v1/table/analytics$sales$orders/drop", `{}`, http.StatusOK)
	if h.filer.Get(s3tables.GetNamespacePath("analytics", "sales"), "orders") != nil {
		t.Fatal("drop left the dataset directory behind")
	}
}

// Repointing a registered name at another dataset must not take the dataset it
// used to name with it.
func TestRegisterOverwriteKeepsTheOldDataset(t *testing.T) {
	h := newTestHarness(t)
	h.createBucket(t, "analytics", s3tables.FormatLance)
	h.mustDo(t, http.MethodPost, "/v1/namespace/analytics$sales/create", `{}`, http.StatusOK)
	h.mustDo(t, http.MethodPost, "/v1/table/analytics$sales$orders/declare", `{}`, http.StatusOK)
	h.mustDo(t, http.MethodPost, "/v1/table/analytics$sales$archive/declare", `{}`, http.StatusOK)

	h.mustDo(t, http.MethodPost, "/v1/table/analytics$sales$orders/register",
		`{"location":"s3://analytics/sales/archive","mode":"Overwrite"}`, http.StatusOK)

	described := decode[DescribeTableResponse](t,
		h.mustDo(t, http.MethodPost, "/v1/table/analytics$sales$orders/describe", `{}`, http.StatusOK))
	if described.Location != "s3://analytics/sales/archive" {
		t.Fatalf("location after repointing = %q", described.Location)
	}
	if h.filer.Get(s3tables.GetNamespacePath("analytics", "sales"), "orders") == nil {
		t.Fatal("repointing deleted the dataset the name used to hold")
	}
}

// A required list field answers empty rather than null, which a generated
// client may decode differently.
func TestListAllTablesIsNeverNull(t *testing.T) {
	h := newTestHarness(t)
	h.createBucket(t, "analytics", s3tables.FormatLance)

	body := h.mustDo(t, http.MethodGet, "/v1/table", "", http.StatusOK).Body.String()
	if strings.Contains(body, `"tables":null`) {
		t.Fatalf("ListAllTables returned null for a required field: %s", body)
	}
}

func TestNamespaceListing(t *testing.T) {
	h := newTestHarness(t)
	h.createBucket(t, "analytics", s3tables.FormatLance)
	h.mustDo(t, http.MethodPost, "/v1/namespace/analytics$sales/create", `{}`, http.StatusOK)
	h.mustDo(t, http.MethodPost, "/v1/namespace/analytics$finance/create", `{}`, http.StatusOK)

	roots := decode[ListNamespacesResponse](t,
		h.mustDo(t, http.MethodGet, "/v1/namespace/$/list", "", http.StatusOK))
	if len(roots.Namespaces) != 1 || roots.Namespaces[0] != "analytics" {
		t.Fatalf("root listing = %v, want the table buckets", roots.Namespaces)
	}

	children := decode[ListNamespacesResponse](t,
		h.mustDo(t, http.MethodGet, "/v1/namespace/analytics/list", "", http.StatusOK))
	if len(children.Namespaces) != 2 {
		t.Fatalf("bucket listing = %v, want two namespaces", children.Namespaces)
	}

	h.mustDo(t, http.MethodPost, "/v1/namespace/analytics$sales/exists", `{}`, http.StatusOK)
	h.mustDo(t, http.MethodPost, "/v1/namespace/analytics$nope/exists", `{}`, http.StatusNotFound)

	// Fail mode reports a missing namespace as 400 on this operation, which is
	// what the spec asks for; Skip reports success.
	h.mustDo(t, http.MethodPost, "/v1/namespace/analytics$nope/drop", `{}`, http.StatusBadRequest)
	h.mustDo(t, http.MethodPost, "/v1/namespace/analytics$nope/drop", `{"mode":"Skip"}`, http.StatusNoContent)
}

// Storage flattens a namespace's parts, so creating a$b without a would leave
// an intermediate that listing derives from the name and describe denies
// exists. The spec asks for NamespaceNotFound, which keeps them consistent.
func TestCreateNamespaceRequiresItsParent(t *testing.T) {
	h := newTestHarness(t)
	h.createBucket(t, "analytics", s3tables.FormatLance)

	recorder := h.mustDo(t, http.MethodPost, "/v1/namespace/analytics$missing$child/create", `{}`, http.StatusNotFound)
	if got := decode[errorResponse](t, recorder); got.Code != codeNamespaceNotFound {
		t.Fatalf("error code = %d, want %d", got.Code, codeNamespaceNotFound)
	}

	// With the parent in place the child is fine, and both are then listed and
	// describable.
	h.mustDo(t, http.MethodPost, "/v1/namespace/analytics$parent/create", `{}`, http.StatusOK)
	h.mustDo(t, http.MethodPost, "/v1/namespace/analytics$parent$child/create", `{}`, http.StatusOK)

	listed := decode[ListNamespacesResponse](t,
		h.mustDo(t, http.MethodGet, "/v1/namespace/analytics/list", "", http.StatusOK))
	for _, name := range listed.Namespaces {
		path := "/v1/namespace/analytics$" + name
		h.mustDo(t, http.MethodPost, path+"/exists", `{}`, http.StatusOK)
	}
}

func TestCreateNamespaceModes(t *testing.T) {
	h := newTestHarness(t)
	h.createBucket(t, "analytics", s3tables.FormatLance)
	h.mustDo(t, http.MethodPost, "/v1/namespace/analytics$sales/create", `{}`, http.StatusOK)

	h.mustDo(t, http.MethodPost, "/v1/namespace/analytics$sales/create", `{}`, http.StatusConflict)
	h.mustDo(t, http.MethodPost, "/v1/namespace/analytics$sales/create", `{"mode":"ExistOk"}`, http.StatusOK)
	// snake_case is the other spelling the spec accepts.
	h.mustDo(t, http.MethodPost, "/v1/namespace/analytics$sales/create", `{"mode":"exist_ok"}`, http.StatusOK)

	h.mustDo(t, http.MethodPost, "/v1/namespace/$/create", `{}`, http.StatusBadRequest)
}

// A Lance client must never resolve an Iceberg table's location, or it would
// write a dataset over a table another engine owns.
func TestIcebergTablesAreInvisible(t *testing.T) {
	h := newTestHarness(t)
	// Undeclared, because a bucket that declares one format cannot hold the
	// other - and mixing is exactly what this test needs to prove is hidden.
	h.createBucket(t, "analytics", "")
	h.mustDo(t, http.MethodPost, "/v1/namespace/analytics$sales/create", `{}`, http.StatusOK)
	h.mustDo(t, http.MethodPost, "/v1/table/analytics$sales$vectors/declare", `{}`, http.StatusOK)

	manager := h.admin
	var created s3tables.CreateTableResponse
	err := manager.Execute(t.Context(), s3tables.NewManagerClient(h.filer.Client), "CreateTable", &s3tables.CreateTableRequest{
		TableBucketARN:   h.bucketARN(t, "analytics"),
		Namespace:        []string{"sales"},
		Name:             "ledger",
		Format:           s3tables.FormatIceberg,
		MetadataLocation: "s3://analytics/sales/ledger/metadata/v1.metadata.json",
	}, &created, "")
	if err != nil {
		t.Fatalf("create iceberg table: %v", err)
	}

	listed := decode[ListTablesResponse](t,
		h.mustDo(t, http.MethodGet, "/v1/namespace/analytics$sales/table/list", "", http.StatusOK))
	if len(listed.Tables) != 1 || listed.Tables[0] != "analytics$sales$vectors" {
		t.Fatalf("listing = %v, want only the lance table", listed.Tables)
	}

	h.mustDo(t, http.MethodPost, "/v1/table/analytics$sales$ledger/describe", `{}`, http.StatusNotFound)
	h.mustDo(t, http.MethodPost, "/v1/table/analytics$sales$ledger/exists", `{}`, http.StatusNotFound)

	// Declaring over the Iceberg table must not quietly succeed and hand the
	// Lance client a directory another format owns.
	recorder := h.mustDo(t, http.MethodPost, "/v1/table/analytics$sales$ledger/declare", `{}`, http.StatusConflict)
	if got := decode[errorResponse](t, recorder); got.Code != codeTableAlreadyExists {
		t.Fatalf("error code = %d, want %d", got.Code, codeTableAlreadyExists)
	}
}

// The route and the body naming different objects is a bad request, not a silent
// preference for one of them.
func TestRouteAndBodyMustAgree(t *testing.T) {
	h := newTestHarness(t)
	h.createBucket(t, "analytics", s3tables.FormatLance)
	h.mustDo(t, http.MethodPost, "/v1/namespace/analytics$sales/create", `{}`, http.StatusOK)

	recorder := h.mustDo(t, http.MethodPost, "/v1/table/analytics$sales$orders/declare",
		`{"id":["analytics","sales","other"]}`, http.StatusBadRequest)
	if got := decode[errorResponse](t, recorder); got.Code != codeInvalidInput {
		t.Fatalf("error code = %d, want %d", got.Code, codeInvalidInput)
	}
}

// A client-supplied location must resolve inside its own bucket. Without this,
// a "../"-laden or cross-bucket location escapes when the marker path is joined
// under /buckets, letting the request reach another tenant's namespace, or
// outside /buckets entirely, and auto-create the parent directories on the way.
func TestLocationOutsideBucketIsRejected(t *testing.T) {
	cases := []struct {
		name     string
		location string
	}{
		{"another bucket", "s3://victim/secret"},
		{"traversal into another bucket", "s3://analytics/../victim/secret"},
		{"traversal above the buckets root", "s3://analytics/../../etc/cron.d"},
		{"not an s3 uri", "../../etc/cron.d"},
	}
	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			dir := s3tables.TableDataDirFromMetadataLocation(tc.location)
			body := `{"location":"` + tc.location + `"}`

			// declare writes .lance-reserved at the location: rejection must leave
			// nothing behind at the escaped directory.
			t.Run("declare", func(t *testing.T) {
				h := newTestHarness(t)
				h.createBucket(t, "analytics", s3tables.FormatLance)
				h.mustDo(t, http.MethodPost, "/v1/namespace/analytics$sales/create", `{}`, http.StatusOK)

				recorder := h.mustDo(t, http.MethodPost, "/v1/table/analytics$sales$orders/declare", body, http.StatusBadRequest)
				if got := decode[errorResponse](t, recorder); got.Code != codeInvalidInput {
					t.Fatalf("error code = %d, want %d", got.Code, codeInvalidInput)
				}
				if dir != "" && h.filer.Get(dir, reservedMarker) != nil {
					t.Fatalf("declare wrote %s into %s despite rejection", reservedMarker, dir)
				}
			})

			// register removes .lance-deregistered at the location: rejection must
			// leave a victim's marker in place rather than un-hiding their table.
			t.Run("register", func(t *testing.T) {
				h := newTestHarness(t)
				h.createBucket(t, "analytics", s3tables.FormatLance)
				h.mustDo(t, http.MethodPost, "/v1/namespace/analytics$sales/create", `{}`, http.StatusOK)
				if dir != "" {
					h.filer.PutFile(dir, deregisteredMarker, time.Unix(1, 0))
				}

				recorder := h.mustDo(t, http.MethodPost, "/v1/table/analytics$sales$orders/register", body, http.StatusBadRequest)
				if got := decode[errorResponse](t, recorder); got.Code != codeInvalidInput {
					t.Fatalf("error code = %d, want %d", got.Code, codeInvalidInput)
				}
				if dir != "" && h.filer.Get(dir, deregisteredMarker) == nil {
					t.Fatalf("register removed %s from %s despite rejection", deregisteredMarker, dir)
				}
			})
		})
	}
}

// The data plane needs Lance format support that does not exist in Go, so it
// answers with the spec's own Unsupported code rather than a bare 404.
func TestDataPlaneIsUnsupported(t *testing.T) {
	h := newTestHarness(t)
	recorder := h.mustDo(t, http.MethodPost, "/v1/table/analytics$sales$orders/query", `{}`, http.StatusNotImplemented)
	if got := decode[errorResponse](t, recorder); got.Code != codeUnsupported {
		t.Fatalf("error code = %d, want %d", got.Code, codeUnsupported)
	}
}

// createOnlyAuthenticator signs in an identity that may create and read tables
// but not update them.
type createOnlyAuthenticator struct{}

func (createOnlyAuthenticator) AuthenticateRequest(*http.Request) (string, interface{}, s3err.ErrorCode) {
	return "creator", &struct {
		Name    string
		Actions []string
	}{Name: "creator", Actions: []string{"s3tables:CreateTable", "s3tables:GetTable"}}, s3err.ErrNone
}

func (createOnlyAuthenticator) DefaultAllow() bool { return false }

func (createOnlyAuthenticator) AuthEnabled() bool { return true }

// The right to create a name is the write evidence for a table the declare
// creates, but not for one it only finds.
func TestDeclareVendsWriteOnlyForATableItCreates(t *testing.T) {
	h := newTestHarness(t)
	h.createBucket(t, "lnc", s3tables.FormatLance)
	h.mustDo(t, http.MethodPost, "/v1/namespace/lnc$ns/create", `{}`, http.StatusOK)
	h.mustDo(t, http.MethodPost, "/v1/table/lnc$ns$theirs/declare", `{}`, http.StatusOK)
	h.mustDo(t, http.MethodPost, "/v1/table/lnc$ns$theirs/deregister", `{}`, http.StatusOK)

	server := NewServer(s3tables.NewManagerClient(h.filer.Client), createOnlyAuthenticator{})
	server.SetCredentialVendor(recordingVendor{})
	router := mux.NewRouter().SkipClean(true)
	server.RegisterRoutes(router)
	creator := &testHarness{router: router, filer: h.filer, admin: h.admin, server: server}

	for _, tc := range []struct{ name, want string }{
		{"theirs", "read-only"},
		{"mine", "read-write"},
	} {
		t.Run(tc.name, func(t *testing.T) {
			if got := creator.vendedKey(t, "/v1/table/lnc$ns$"+tc.name+"/declare", ""); got != tc.want {
				t.Errorf("vended %q, want %q", got, tc.want)
			}
		})
	}
	if h.filer.Get("/buckets/lnc/ns/theirs", deregisteredMarker) == nil {
		t.Error("a declare that may not write the table brought it back")
	}
}