mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-11-10 05:48:19 +01:00
iceberg: withhold the S3 endpoint from credential-vending clients (#10570)
* iceberg: withhold the S3 endpoint from credential-vending clients A client that sends X-Iceberg-Access-Delegation: vended-credentials builds its storage credential out of the LoadTable config and drops the one it was configured with. We vend no credentials, so the endpoint we advertised left DuckDB signing nothing: every metadata and data file came back 403, and its attempt to refresh the empty credential 404ed on stage-created tables. Answer those clients with no config at all so they keep their own credentials. Clients that do not ask for delegation still get the endpoint. * iceberg: mark load responses as varying on the delegation header The FileIO config in a table or view load response now depends on whether the client asked for vended credentials, so a cache between us and the client must key on that header rather than on the URL alone. * test: cover the DuckDB vended-credentials access pattern Runs weed mini with -s3.externalUrl, which is what makes the catalog advertise an endpoint at all, and checks both halves: a plain LoadTable still gets the endpoint, while one asking for vended credentials never gets an endpoint without the credentials to sign with. The DuckDB round trip creates a table from a query and reads it back, which is the flow that failed with 403 on every data file.
This commit is contained in:
1 parent
474a0713b0
commit
d448e9db7b
9 files changed
+382
-36
No files matched your search
@@ -246,13 +246,13 @@ func (s *Server) handleCreateTable(w http.ResponseWriter, r *http.Request) {
|
||||
if existsErr == nil {
|
||||
// Table already registered. Return the existing definition so CTAS/IF NOT
|
||||
// EXISTS flows see a stable response instead of a 409.
|
||||
result, buildErr := s.buildLoadTableResult(existsResp, bucketName, namespace, tableName)
|
||||
result, buildErr := s.buildLoadTableResult(r, existsResp, bucketName, namespace, tableName)
|
||||
if buildErr != nil {
|
||||
glog.Errorf("Iceberg: CreateTable load existing %s: %v", tableName, buildErr)
|
||||
writeError(w, http.StatusInternalServerError, "InternalServerError", "Failed to build table metadata")
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusOK, result)
|
||||
writeLoadResult(w, http.StatusOK, result)
|
||||
return
|
||||
}
|
||||
if !isNoSuchTableError(existsErr) {
|
||||
@@ -275,9 +275,9 @@ func (s *Server) handleCreateTable(w http.ResponseWriter, r *http.Request) {
|
||||
result := LoadTableResult{
|
||||
MetadataLocation: metadataLocation,
|
||||
Metadata: metadata,
|
||||
Config: s.buildFileIOConfig(),
|
||||
Config: s.buildFileIOConfig(r),
|
||||
}
|
||||
writeJSON(w, http.StatusOK, result)
|
||||
writeLoadResult(w, http.StatusOK, result)
|
||||
return
|
||||
}
|
||||
if err := s.saveMetadataFile(r.Context(), metadataBucket, metadataPath, metadataFileName, metadataBytes); err != nil {
|
||||
@@ -324,13 +324,13 @@ func (s *Server) handleCreateTable(w http.ResponseWriter, r *http.Request) {
|
||||
writeError(w, http.StatusConflict, "AlreadyExistsException", err.Error())
|
||||
return
|
||||
}
|
||||
result, buildErr := s.buildLoadTableResult(getResp, bucketName, namespace, tableName)
|
||||
result, buildErr := s.buildLoadTableResult(r, getResp, bucketName, namespace, tableName)
|
||||
if buildErr != nil {
|
||||
glog.Errorf("Iceberg: CreateTable load existing %s: %v", tableName, buildErr)
|
||||
writeError(w, http.StatusInternalServerError, "InternalServerError", "Failed to build table metadata")
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusOK, result)
|
||||
writeLoadResult(w, http.StatusOK, result)
|
||||
return
|
||||
}
|
||||
if strings.Contains(err.Error(), "already exists") {
|
||||
@@ -348,13 +348,13 @@ func (s *Server) handleCreateTable(w http.ResponseWriter, r *http.Request) {
|
||||
writeError(w, http.StatusConflict, "AlreadyExistsException", err.Error())
|
||||
return
|
||||
}
|
||||
result, buildErr := s.buildLoadTableResult(getResp, bucketName, namespace, tableName)
|
||||
result, buildErr := s.buildLoadTableResult(r, getResp, bucketName, namespace, tableName)
|
||||
if buildErr != nil {
|
||||
glog.Errorf("Iceberg: CreateTable load existing %s: %v", tableName, buildErr)
|
||||
writeError(w, http.StatusInternalServerError, "InternalServerError", "Failed to build table metadata")
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusOK, result)
|
||||
writeLoadResult(w, http.StatusOK, result)
|
||||
return
|
||||
}
|
||||
glog.V(1).Infof("Iceberg: CreateTable error: %v", err)
|
||||
@@ -374,9 +374,9 @@ func (s *Server) handleCreateTable(w http.ResponseWriter, r *http.Request) {
|
||||
result := LoadTableResult{
|
||||
MetadataLocation: finalLocation,
|
||||
Metadata: metadata,
|
||||
Config: s.buildFileIOConfig(),
|
||||
Config: s.buildFileIOConfig(r),
|
||||
}
|
||||
writeJSON(w, http.StatusOK, result)
|
||||
writeLoadResult(w, http.StatusOK, result)
|
||||
}
|
||||
|
||||
// handleRegisterTable registers an existing metadata.json under a new catalog
|
||||
@@ -465,13 +465,13 @@ func (s *Server) handleRegisterTable(w http.ResponseWriter, r *http.Request) {
|
||||
MetadataLocation: req.MetadataLocation,
|
||||
Metadata: &s3tables.TableMetadata{FullMetadata: json.RawMessage(metadataBytes)},
|
||||
}
|
||||
result, buildErr := s.buildLoadTableResult(getResp, bucketName, namespace, req.Name)
|
||||
result, buildErr := s.buildLoadTableResult(r, getResp, bucketName, namespace, req.Name)
|
||||
if buildErr != nil {
|
||||
glog.Errorf("Iceberg: RegisterTable %s: %v", req.Name, buildErr)
|
||||
writeError(w, http.StatusInternalServerError, "InternalServerError", "Failed to build table metadata")
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusOK, result)
|
||||
writeLoadResult(w, http.StatusOK, result)
|
||||
}
|
||||
|
||||
// handleLoadTable loads table metadata.
|
||||
@@ -514,16 +514,16 @@ func (s *Server) handleLoadTable(w http.ResponseWriter, r *http.Request) {
|
||||
return
|
||||
}
|
||||
|
||||
result, buildErr := s.buildLoadTableResult(getResp, bucketName, namespace, tableName)
|
||||
result, buildErr := s.buildLoadTableResult(r, getResp, bucketName, namespace, tableName)
|
||||
if buildErr != nil {
|
||||
glog.Errorf("Iceberg: LoadTable %s: %v", tableName, buildErr)
|
||||
writeError(w, http.StatusInternalServerError, "InternalServerError", "Failed to build table metadata")
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusOK, result)
|
||||
writeLoadResult(w, http.StatusOK, result)
|
||||
}
|
||||
|
||||
func (s *Server) buildLoadTableResult(getResp s3tables.GetTableResponse, bucketName string, namespace []string, tableName string) (LoadTableResult, error) {
|
||||
func (s *Server) buildLoadTableResult(r *http.Request, getResp s3tables.GetTableResponse, bucketName string, namespace []string, tableName string) (LoadTableResult, error) {
|
||||
location := tableLocationFromMetadataLocation(getResp.MetadataLocation)
|
||||
if location == "" {
|
||||
location = fmt.Sprintf("s3://%s/%s", bucketName, path.Join(flattenNamespacePath(namespace), tableName))
|
||||
@@ -562,7 +562,7 @@ func (s *Server) buildLoadTableResult(getResp s3tables.GetTableResponse, bucketN
|
||||
return LoadTableResult{
|
||||
MetadataLocation: getResp.MetadataLocation,
|
||||
Metadata: metadata,
|
||||
Config: s.buildFileIOConfig(),
|
||||
Config: s.buildFileIOConfig(r),
|
||||
}, nil
|
||||
}
|
||||
|
||||
@@ -571,13 +571,21 @@ func (s *Server) buildLoadTableResult(getResp s3tables.GetTableResponse, bucketN
|
||||
// separately discovering the endpoint. The region defaults to the same
|
||||
// value baked into table bucket ARNs so clients like DuckDB that require
|
||||
// a region on attach don't need to be told it out-of-band. See issue #9103.
|
||||
func (s *Server) buildFileIOConfig() iceberg.Properties {
|
||||
func (s *Server) buildFileIOConfig(r *http.Request) iceberg.Properties {
|
||||
config := make(iceberg.Properties)
|
||||
if s.s3Endpoint != "" {
|
||||
config["s3.endpoint"] = s.s3Endpoint
|
||||
config["s3.path-style-access"] = "true"
|
||||
config["s3.region"] = s3tables.DefaultRegion
|
||||
if s.s3Endpoint == "" {
|
||||
return config
|
||||
}
|
||||
// A client asking for vended credentials builds its storage credential out
|
||||
// of whatever comes back here and stops using the one it was configured
|
||||
// with. We vend none, so an endpoint on its own leaves it sending unsigned
|
||||
// requests; say nothing instead and let it keep its own credentials.
|
||||
if wantsVendedCredentials(r) {
|
||||
return config
|
||||
}
|
||||
config["s3.endpoint"] = s.s3Endpoint
|
||||
config["s3.path-style-access"] = "true"
|
||||
config["s3.region"] = s3tables.DefaultRegion
|
||||
return config
|
||||
}
|
||||
|
||||
|
||||
@@ -158,12 +158,12 @@ func (s *Server) handleCreateView(w http.ResponseWriter, r *http.Request) {
|
||||
// short-circuits with its stored definition (idempotent CreateView) so we
|
||||
// never overwrite the persisted metadata of an existing view.
|
||||
if existsResp, existsErr := s.getView(r, namespace, req.Name); existsErr == nil {
|
||||
result, buildErr := s.buildViewResponse(existsResp, bucketName, namespace, req.Name)
|
||||
result, buildErr := s.buildViewResponse(r, existsResp, bucketName, namespace, req.Name)
|
||||
if buildErr != nil {
|
||||
writeError(w, http.StatusInternalServerError, "InternalServerError", buildErr.Error())
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusOK, result)
|
||||
writeLoadResult(w, http.StatusOK, result)
|
||||
return
|
||||
} else if !isViewNotFound(existsErr) {
|
||||
glog.V(1).Infof("Iceberg: CreateView existence check failed for %s.%s: %v", flattenNamespacePath(namespace), req.Name, existsErr)
|
||||
@@ -218,10 +218,10 @@ func (s *Server) handleCreateView(w http.ResponseWriter, r *http.Request) {
|
||||
if finalLocation == "" {
|
||||
finalLocation = metadataLocation
|
||||
}
|
||||
writeJSON(w, http.StatusOK, ViewResponse{
|
||||
writeLoadResult(w, http.StatusOK, ViewResponse{
|
||||
MetadataLocation: finalLocation,
|
||||
Metadata: metadata,
|
||||
Config: s.buildFileIOConfig(),
|
||||
Config: s.buildFileIOConfig(r),
|
||||
})
|
||||
}
|
||||
|
||||
@@ -246,12 +246,12 @@ func (s *Server) handleLoadView(w http.ResponseWriter, r *http.Request) {
|
||||
return
|
||||
}
|
||||
|
||||
result, err := s.buildViewResponse(getResp, getBucketFromPrefix(r), namespace, viewName)
|
||||
result, err := s.buildViewResponse(r, getResp, getBucketFromPrefix(r), namespace, viewName)
|
||||
if err != nil {
|
||||
writeError(w, http.StatusInternalServerError, "InternalServerError", err.Error())
|
||||
return
|
||||
}
|
||||
writeJSON(w, http.StatusOK, result)
|
||||
writeLoadResult(w, http.StatusOK, result)
|
||||
}
|
||||
|
||||
// handleViewExists checks if a view exists.
|
||||
@@ -340,7 +340,7 @@ func (s *Server) getView(r *http.Request, namespace []string, viewName string) (
|
||||
}
|
||||
|
||||
// buildViewResponse parses the stored view metadata into a ViewResponse.
|
||||
func (s *Server) buildViewResponse(getResp s3tables.GetViewResponse, bucketName string, namespace []string, viewName string) (ViewResponse, error) {
|
||||
func (s *Server) buildViewResponse(r *http.Request, getResp s3tables.GetViewResponse, bucketName string, namespace []string, viewName string) (ViewResponse, error) {
|
||||
if getResp.Metadata == nil || len(getResp.Metadata.FullMetadata) == 0 {
|
||||
return ViewResponse{}, fmt.Errorf("view %s has no metadata", viewName)
|
||||
}
|
||||
@@ -351,7 +351,7 @@ func (s *Server) buildViewResponse(getResp s3tables.GetViewResponse, bucketName
|
||||
return ViewResponse{
|
||||
MetadataLocation: getResp.MetadataLocation,
|
||||
Metadata: metadata,
|
||||
Config: s.buildFileIOConfig(),
|
||||
Config: s.buildFileIOConfig(r),
|
||||
}, nil
|
||||
}
|
||||
|
||||
|
||||
@@ -130,10 +130,10 @@ func (s *Server) handleUpdateView(w http.ResponseWriter, r *http.Request) {
|
||||
return s.tablesManager.Execute(r.Context(), mgrClient, "UpdateView", updateReq, nil, identityName)
|
||||
})
|
||||
if err == nil {
|
||||
writeJSON(w, http.StatusOK, ViewResponse{
|
||||
writeLoadResult(w, http.StatusOK, ViewResponse{
|
||||
MetadataLocation: newMetadataLocation,
|
||||
Metadata: newMetadata,
|
||||
Config: s.buildFileIOConfig(),
|
||||
Config: s.buildFileIOConfig(r),
|
||||
})
|
||||
return
|
||||
}
|
||||
|
||||
@@ -0,0 +1,79 @@
|
||||
package iceberg
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/s3api/s3tables"
|
||||
)
|
||||
|
||||
func TestWantsVendedCredentials(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
values []string
|
||||
want bool
|
||||
}{
|
||||
{name: "no header", want: false},
|
||||
{name: "vended credentials", values: []string{"vended-credentials"}, want: true},
|
||||
{name: "mechanism list", values: []string{"remote-signing,vended-credentials"}, want: true},
|
||||
{name: "spaced and mixed case", values: []string{"Remote-Signing, Vended-Credentials"}, want: true},
|
||||
{name: "repeated header", values: []string{"remote-signing", "vended-credentials"}, want: true},
|
||||
{name: "remote signing only", values: []string{"remote-signing"}, want: false},
|
||||
}
|
||||
for _, tc := range tests {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
r := httptest.NewRequest(http.MethodGet, "/v1/namespaces/ns/tables/t", nil)
|
||||
for _, v := range tc.values {
|
||||
r.Header.Add(accessDelegationHeader, v)
|
||||
}
|
||||
if got := wantsVendedCredentials(r); got != tc.want {
|
||||
t.Fatalf("wantsVendedCredentials(%v) = %v, want %v", tc.values, got, tc.want)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// A client that asked for vended credentials replaces its own storage
|
||||
// credentials with whatever the catalog returns. Since the catalog vends none,
|
||||
// an endpoint by itself would leave the client signing nothing and every data
|
||||
// file read and write would come back 403.
|
||||
func TestBuildFileIOConfigWithholdsEndpointFromCredentialVendingClients(t *testing.T) {
|
||||
s := &Server{s3Endpoint: "http://seaweed.example:8333"}
|
||||
|
||||
r := httptest.NewRequest(http.MethodGet, "/v1/namespaces/ns/tables/t", nil)
|
||||
r.Header.Set(accessDelegationHeader, "vended-credentials")
|
||||
if got := s.buildFileIOConfig(r); len(got) != 0 {
|
||||
t.Fatalf("buildFileIOConfig() = %v, want empty for a vended-credentials request", got)
|
||||
}
|
||||
|
||||
plain := httptest.NewRequest(http.MethodGet, "/v1/namespaces/ns/tables/t", nil)
|
||||
if got := s.buildFileIOConfig(plain); got["s3.endpoint"] != s.s3Endpoint {
|
||||
t.Fatalf("s3.endpoint = %q, want %q", got["s3.endpoint"], s.s3Endpoint)
|
||||
}
|
||||
}
|
||||
|
||||
// The body of a load response depends on the delegation header, so a cache in
|
||||
// front of the catalog must not key on the URL alone.
|
||||
func TestWriteLoadResultVariesOnTheDelegationHeader(t *testing.T) {
|
||||
rec := httptest.NewRecorder()
|
||||
writeLoadResult(rec, http.StatusOK, LoadTableResult{})
|
||||
if got := rec.Header().Get("Vary"); got != accessDelegationHeader {
|
||||
t.Fatalf("Vary = %q, want %q", got, accessDelegationHeader)
|
||||
}
|
||||
}
|
||||
|
||||
func TestLoadTableResultOmitsConfigForCredentialVendingClients(t *testing.T) {
|
||||
s := &Server{s3Endpoint: "http://seaweed.example:8333"}
|
||||
r := httptest.NewRequest(http.MethodGet, "/v1/namespaces/ns/tables/t", nil)
|
||||
r.Header.Set(accessDelegationHeader, "vended-credentials")
|
||||
|
||||
getResp := s3tables.GetTableResponse{MetadataLocation: "s3://bkt/ns/t/metadata/v1.metadata.json"}
|
||||
result, err := s.buildLoadTableResult(r, getResp, "bkt", []string{"ns"}, "t")
|
||||
if err != nil {
|
||||
t.Fatalf("buildLoadTableResult() error = %v", err)
|
||||
}
|
||||
if len(result.Config) != 0 {
|
||||
t.Fatalf("LoadTableResult.Config = %v, want empty", result.Config)
|
||||
}
|
||||
}
|
||||
@@ -3,6 +3,8 @@ package iceberg
|
||||
import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
@@ -96,7 +98,8 @@ func TestBuildLoadTableResultNeverReturnsNilMetadata(t *testing.T) {
|
||||
}
|
||||
for name, getResp := range cases {
|
||||
t.Run(name, func(t *testing.T) {
|
||||
result, err := (&Server{}).buildLoadTableResult(getResp, "bkt", []string{"ns"}, "t")
|
||||
r := httptest.NewRequest(http.MethodGet, "/v1/namespaces/ns/tables/t", nil)
|
||||
result, err := (&Server{}).buildLoadTableResult(r, getResp, "bkt", []string{"ns"}, "t")
|
||||
if err != nil {
|
||||
t.Fatalf("buildLoadTableResult() error = %v, want nil", err)
|
||||
}
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
package iceberg
|
||||
|
||||
import (
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"testing"
|
||||
)
|
||||
@@ -54,9 +55,13 @@ func TestGetBucketFromPrefix_WarehouseQueryFallback(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestBuildFileIOConfig(t *testing.T) {
|
||||
loadTable := func() *http.Request {
|
||||
return httptest.NewRequest(http.MethodGet, "/v1/namespaces/ns/tables/t", nil)
|
||||
}
|
||||
|
||||
t.Run("no endpoint configured yields empty config", func(t *testing.T) {
|
||||
s := &Server{}
|
||||
got := s.buildFileIOConfig()
|
||||
got := s.buildFileIOConfig(loadTable())
|
||||
if len(got) != 0 {
|
||||
t.Fatalf("buildFileIOConfig() = %v, want empty", got)
|
||||
}
|
||||
@@ -64,7 +69,7 @@ func TestBuildFileIOConfig(t *testing.T) {
|
||||
|
||||
t.Run("endpoint is advertised with path-style-access and region", func(t *testing.T) {
|
||||
s := &Server{s3Endpoint: "http://seaweed.example:8333"}
|
||||
got := s.buildFileIOConfig()
|
||||
got := s.buildFileIOConfig(loadTable())
|
||||
if got["s3.endpoint"] != "http://seaweed.example:8333" {
|
||||
t.Fatalf("s3.endpoint = %q, want %q", got["s3.endpoint"], "http://seaweed.example:8333")
|
||||
}
|
||||
|
||||
@@ -75,6 +75,34 @@ func tableLocationFromMetadataLocation(metadataLocation string) string {
|
||||
return trimmed
|
||||
}
|
||||
|
||||
// accessDelegationHeader is how a client asks the catalog to hand back storage
|
||||
// credentials along with the table metadata.
|
||||
const accessDelegationHeader = "X-Iceberg-Access-Delegation"
|
||||
|
||||
// wantsVendedCredentials reports whether the client asked for vended
|
||||
// credentials. The header carries a comma-separated list of mechanisms.
|
||||
func wantsVendedCredentials(r *http.Request) bool {
|
||||
if r == nil {
|
||||
return false
|
||||
}
|
||||
for _, value := range r.Header.Values(accessDelegationHeader) {
|
||||
for _, mechanism := range strings.Split(value, ",") {
|
||||
if strings.EqualFold(strings.TrimSpace(mechanism), "vended-credentials") {
|
||||
return true
|
||||
}
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// writeLoadResult writes a table or view load response. The FileIO config it
|
||||
// carries depends on the delegation the client asked for, so a cache between
|
||||
// us and the client must key on that header and not on the URL alone.
|
||||
func writeLoadResult(w http.ResponseWriter, status int, v interface{}) {
|
||||
w.Header().Set("Vary", accessDelegationHeader)
|
||||
writeJSON(w, status, v)
|
||||
}
|
||||
|
||||
// writeJSON writes a JSON response.
|
||||
func writeJSON(w http.ResponseWriter, status int, v interface{}) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
|
||||
Reference in new issue
Block a user