Files
seaweedFS/weed/s3api/s3tables/utils.go
Chris Lu 995dfc4d5d chore: remove ~50k lines of unreachable dead code (#8913)
* chore: remove unreachable dead code across the codebase

Remove ~50,000 lines of unreachable code identified by static analysis.

Major removals:
- weed/filer/redis_lua: entire unused Redis Lua filer store implementation
- weed/wdclient/net2, resource_pool: unused connection/resource pool packages
- weed/plugin/worker/lifecycle: unused lifecycle plugin worker
- weed/s3api: unused S3 policy templates, presigned URL IAM, streaming copy,
  multipart IAM, key rotation, and various SSE helper functions
- weed/mq/kafka: unused partition mapping, compression, schema, and protocol functions
- weed/mq/offset: unused SQL storage and migration code
- weed/worker: unused registry, task, and monitoring functions
- weed/query: unused SQL engine, parquet scanner, and type functions
- weed/shell: unused EC proportional rebalance functions
- weed/storage/erasure_coding/distribution: unused distribution analysis functions
- Individual unreachable functions removed from 150+ files across admin,
  credential, filer, iam, kms, mount, mq, operation, pb, s3api, server,
  shell, storage, topology, and util packages

* fix(s3): reset shared memory store in IAM test to prevent flaky failure

TestLoadIAMManagerFromConfig_EmptyConfigWithFallbackKey was flaky because
the MemoryStore credential backend is a singleton registered via init().
Earlier tests that create anonymous identities pollute the shared store,
causing LookupAnonymous() to unexpectedly return true.

Fix by calling Reset() on the memory store before the test runs.

* style: run gofmt on changed files

* fix: restore KMS functions used by integration tests

* fix(plugin): prevent panic on send to closed worker session channel

The Plugin.sendToWorker method could panic with "send on closed channel"
when a worker disconnected while a message was being sent. The race was
between streamSession.close() closing the outgoing channel and sendToWorker
writing to it concurrently.

Add a done channel to streamSession that is closed before the outgoing
channel, and check it in sendToWorker's select to safely detect closed
sessions without panicking.
2026-04-03 16:04:27 -07:00

418 lines
13 KiB
Go

package s3tables
import (
"crypto/rand"
"encoding/hex"
"fmt"
"net/url"
"path"
"regexp"
"strings"
"time"
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
)
const (
bucketNamePatternStr = `[a-z0-9-]+`
tableNamespacePatternStr = `[a-z0-9_.]+`
tableNamePatternStr = `[a-z0-9_]+`
)
const (
tableObjectRootDirName = ".objects"
)
var (
bucketARNPattern = regexp.MustCompile(`^arn:aws:s3tables:[^:]*:[^:]*:bucket/(` + bucketNamePatternStr + `)$`)
tableARNPattern = regexp.MustCompile(`^arn:aws:s3tables:[^:]*:[^:]*:bucket/(` + bucketNamePatternStr + `)/table/(` + tableNamespacePatternStr + `)/(` + tableNamePatternStr + `)$`)
tagPattern = regexp.MustCompile(`^([\p{L}\p{Z}\p{N}_.:/=+\-@]*)$`)
)
// ARN parsing functions
// parseBucketNameFromARN extracts bucket name from table bucket ARN
// ARN format: arn:aws:s3tables:{region}:{account}:bucket/{bucket-name}
func parseBucketNameFromARN(arn string) (string, error) {
matches := bucketARNPattern.FindStringSubmatch(arn)
if len(matches) != 2 {
return "", fmt.Errorf("invalid bucket ARN: %s", arn)
}
bucketName := matches[1]
if !isValidBucketName(bucketName) {
return "", fmt.Errorf("invalid bucket name in ARN: %s", bucketName)
}
return bucketName, nil
}
// ParseBucketNameFromARN is a wrapper to validate bucket ARN for other packages.
func ParseBucketNameFromARN(arn string) (string, error) {
return parseBucketNameFromARN(arn)
}
// parseTableFromARN extracts bucket name, namespace, and table name from ARN
// ARN format: arn:aws:s3tables:{region}:{account}:bucket/{bucket-name}/table/{namespace}/{table-name}
func parseTableFromARN(arn string) (bucketName, namespace, tableName string, err error) {
matches := tableARNPattern.FindStringSubmatch(arn)
if len(matches) != 4 {
return "", "", "", fmt.Errorf("invalid table ARN: %s", arn)
}
// Validate bucket name
bucketName = matches[1]
if err := validateBucketName(bucketName); err != nil {
return "", "", "", fmt.Errorf("invalid bucket name in ARN: %v", err)
}
namespace, err = validateNamespace([]string{matches[2]})
if err != nil {
return "", "", "", fmt.Errorf("invalid namespace in ARN: %v", err)
}
// URL decode and validate the table name from the ARN path component
tableNameUnescaped, err := url.PathUnescape(matches[3])
if err != nil {
return "", "", "", fmt.Errorf("invalid table name encoding in ARN: %v", err)
}
if _, err := validateTableName(tableNameUnescaped); err != nil {
return "", "", "", fmt.Errorf("invalid table name in ARN: %v", err)
}
return bucketName, namespace, tableNameUnescaped, nil
}
// Path helpers
// GetTableBucketPath returns the filer path for a table bucket
func GetTableBucketPath(bucketName string) string {
return path.Join(TablesPath, bucketName)
}
// GetNamespacePath returns the filer path for a namespace
func GetNamespacePath(bucketName, namespace string) string {
return path.Join(TablesPath, bucketName, namespace)
}
// GetTablePath returns the filer path for a table
func GetTablePath(bucketName, namespace, tableName string) string {
return path.Join(TablesPath, bucketName, namespace, tableName)
}
// GetTableObjectRootDir returns the root path for table bucket object storage
func GetTableObjectRootDir() string {
return path.Join(TablesPath, tableObjectRootDirName)
}
// GetTableObjectBucketPath returns the filer path for table bucket object storage
func GetTableObjectBucketPath(bucketName string) string {
return path.Join(GetTableObjectRootDir(), bucketName)
}
// Metadata structures
type tableBucketMetadata struct {
Name string `json:"name"`
CreatedAt time.Time `json:"createdAt"`
OwnerAccountID string `json:"ownerAccountId"`
}
// namespaceMetadata stores metadata for a namespace
type namespaceMetadata struct {
Namespace []string `json:"namespace"`
CreatedAt time.Time `json:"createdAt"`
OwnerAccountID string `json:"ownerAccountId"`
Properties map[string]string `json:"properties,omitempty"`
}
// tableMetadataInternal stores metadata for a table
type tableMetadataInternal struct {
Name string `json:"name"`
Namespace string `json:"namespace"`
Format string `json:"format"`
CreatedAt time.Time `json:"createdAt"`
ModifiedAt time.Time `json:"modifiedAt"`
OwnerAccountID string `json:"ownerAccountId"`
VersionToken string `json:"versionToken"`
MetadataVersion int `json:"metadataVersion"`
MetadataLocation string `json:"metadataLocation,omitempty"`
Metadata *TableMetadata `json:"metadata,omitempty"`
}
// IsTableBucketEntry returns true when the entry is marked as a table bucket.
func IsTableBucketEntry(entry *filer_pb.Entry) bool {
if entry == nil || entry.Extended == nil {
return false
}
_, ok := entry.Extended[ExtendedKeyTableBucket]
return ok
}
// Utility functions
// validateBucketName validates bucket name and returns an error if invalid.
// Bucket names must contain only lowercase letters, numbers, and hyphens.
// Length must be between 3 and 63 characters.
// Must start and end with a letter or digit.
// Reserved prefixes/suffixes are rejected.
func validateBucketName(name string) error {
if name == "" {
return fmt.Errorf("bucket name is required")
}
if len(name) < 3 || len(name) > 63 {
return fmt.Errorf("bucket name must be between 3 and 63 characters")
}
// Must start and end with a letter or digit
start := name[0]
end := name[len(name)-1]
if !((start >= 'a' && start <= 'z') || (start >= '0' && start <= '9')) {
return fmt.Errorf("bucket name must start with a letter or digit")
}
if !((end >= 'a' && end <= 'z') || (end >= '0' && end <= '9')) {
return fmt.Errorf("bucket name must end with a letter or digit")
}
// Allowed characters: a-z, 0-9, -
for i := 0; i < len(name); i++ {
ch := name[i]
if (ch >= 'a' && ch <= 'z') || (ch >= '0' && ch <= '9') || ch == '-' {
continue
}
return fmt.Errorf("bucket name can only contain lowercase letters, numbers, and hyphens")
}
// Reserved prefixes
reservedPrefixes := []string{"xn--", "sthree-", "amzn-s3-demo-", "aws"}
for _, p := range reservedPrefixes {
if strings.HasPrefix(name, p) {
return fmt.Errorf("bucket name cannot start with reserved prefix: %s", p)
}
}
// Reserved suffixes
reservedSuffixes := []string{"-s3alias", "--ol-s3", "--x-s3", "--table-s3"}
for _, s := range reservedSuffixes {
if strings.HasSuffix(name, s) {
return fmt.Errorf("bucket name cannot end with reserved suffix: %s", s)
}
}
return nil
}
// BuildBucketARN builds a bucket ARN with the provided region and account ID.
// If region is empty, the ARN will omit the region field.
func BuildBucketARN(region, accountID, bucketName string) (string, error) {
if bucketName == "" {
return "", fmt.Errorf("bucket name is required")
}
if err := validateBucketName(bucketName); err != nil {
return "", err
}
if accountID == "" {
accountID = DefaultAccountID
}
return buildARN(region, accountID, fmt.Sprintf("bucket/%s", bucketName)), nil
}
// BuildTableARN builds a table ARN with the provided region and account ID.
func BuildTableARN(region, accountID, bucketName, namespace, tableName string) (string, error) {
if bucketName == "" {
return "", fmt.Errorf("bucket name is required")
}
if err := validateBucketName(bucketName); err != nil {
return "", err
}
if namespace == "" {
return "", fmt.Errorf("namespace is required")
}
normalizedNamespace, err := validateNamespace([]string{namespace})
if err != nil {
return "", err
}
if tableName == "" {
return "", fmt.Errorf("table name is required")
}
normalizedTable, err := validateTableName(tableName)
if err != nil {
return "", err
}
if accountID == "" {
accountID = DefaultAccountID
}
return buildARN(region, accountID, fmt.Sprintf("bucket/%s/table/%s/%s", bucketName, normalizedNamespace, normalizedTable)), nil
}
func buildARN(region, accountID, resourcePath string) string {
return fmt.Sprintf("arn:aws:s3tables:%s:%s:%s", region, accountID, resourcePath)
}
// ValidateTags validates tags for S3 Tables.
func ValidateTags(tags map[string]string) error {
if len(tags) > 10 {
return fmt.Errorf("validate tags: %d tags more than 10", len(tags))
}
for k, v := range tags {
if len(k) > 128 {
return fmt.Errorf("validate tags: tag key longer than 128")
}
if !tagPattern.MatchString(k) {
return fmt.Errorf("validate tags key %s error, incorrect key", k)
}
if len(v) > 256 {
return fmt.Errorf("validate tags: tag value longer than 256")
}
if !tagPattern.MatchString(v) {
return fmt.Errorf("validate tags value %s error, incorrect value", v)
}
}
return nil
}
// isValidBucketName validates bucket name characters (kept for compatibility)
// Deprecated: use validateBucketName instead
func isValidBucketName(name string) bool {
return validateBucketName(name) == nil
}
// generateVersionToken generates a unique, unpredictable version token
func generateVersionToken() string {
b := make([]byte, 16)
if _, err := rand.Read(b); err != nil {
// Fallback to timestamp if crypto/rand fails
return fmt.Sprintf("%x", time.Now().UnixNano())
}
return hex.EncodeToString(b)
}
// splitPath splits a path into directory and name components using stdlib
func splitPath(p string) (dir, name string) {
dir = path.Dir(p)
name = path.Base(p)
return
}
func validateNamespacePart(name string) error {
if len(name) < 1 || len(name) > 255 {
return fmt.Errorf("namespace name must be between 1 and 255 characters")
}
// Prevent path traversal and multi-segment paths
if name == "." || name == ".." {
return fmt.Errorf("namespace name cannot be '.' or '..'")
}
if strings.Contains(name, "/") {
return fmt.Errorf("namespace name cannot contain '/'")
}
// Must start and end with a letter or digit
start := name[0]
end := name[len(name)-1]
if !((start >= 'a' && start <= 'z') || (start >= '0' && start <= '9')) {
return fmt.Errorf("namespace name must start with a letter or digit")
}
if !((end >= 'a' && end <= 'z') || (end >= '0' && end <= '9')) {
return fmt.Errorf("namespace name must end with a letter or digit")
}
// Allowed characters: a-z, 0-9, _
for _, ch := range name {
if (ch >= 'a' && ch <= 'z') || (ch >= '0' && ch <= '9') || ch == '_' {
continue
}
return fmt.Errorf("invalid namespace name: only 'a-z', '0-9', and '_' are allowed")
}
// Reserved prefix
if strings.HasPrefix(name, "aws") {
return fmt.Errorf("namespace name cannot start with reserved prefix 'aws'")
}
return nil
}
func normalizeNamespace(namespace []string) ([]string, error) {
if len(namespace) == 0 {
return nil, fmt.Errorf("namespace is required")
}
parts := namespace
if len(namespace) == 1 {
parts = strings.Split(namespace[0], ".")
}
normalized := make([]string, 0, len(parts))
for _, part := range parts {
if err := validateNamespacePart(part); err != nil {
return nil, err
}
normalized = append(normalized, part)
}
return normalized, nil
}
// validateNamespace validates namespace identifiers and returns an internal namespace key.
// A single dotted namespace value is interpreted as multi-level namespace for compatibility
// with path-style APIs, for example "analytics.daily" => ["analytics", "daily"].
func validateNamespace(namespace []string) (string, error) {
parts, err := normalizeNamespace(namespace)
if err != nil {
return "", err
}
return flattenNamespace(parts), nil
}
// ParseNamespace parses a namespace string into namespace parts.
func ParseNamespace(namespace string) ([]string, error) {
return normalizeNamespace([]string{namespace})
}
// validateTableName validates a table name
func validateTableName(name string) (string, error) {
if len(name) < 1 || len(name) > 255 {
return "", fmt.Errorf("table name must be between 1 and 255 characters")
}
if name == "." || name == ".." || strings.Contains(name, "/") {
return "", fmt.Errorf("invalid table name: cannot be '.', '..' or contain '/'")
}
// First character must be a letter or digit
start := name[0]
if !((start >= 'a' && start <= 'z') || (start >= '0' && start <= '9')) {
return "", fmt.Errorf("table name must start with a letter or digit")
}
// Allowed characters: a-z, 0-9, _
for _, ch := range name {
if (ch >= 'a' && ch <= 'z') || (ch >= '0' && ch <= '9') || ch == '_' {
continue
}
return "", fmt.Errorf("invalid table name: only 'a-z', '0-9', and '_' are allowed")
}
return name, nil
}
// ValidateTableName is a wrapper to validate table name for other packages.
func ValidateTableName(name string) (string, error) {
return validateTableName(name)
}
// flattenNamespace joins namespace elements into a single string (using dots as per AWS S3 Tables)
func flattenNamespace(namespace []string) string {
if len(namespace) == 0 {
return ""
}
return strings.Join(namespace, ".")
}
func expandNamespace(namespace string) []string {
if namespace == "" {
return nil
}
parts, err := ParseNamespace(namespace)
if err != nil {
return []string{namespace}
}
return parts
}