cephrgw-operator/internal/ceph/client.go

// Package ceph is a small client for the Ceph RGW (radosgw) admin and S3 APIs,
// scoped to the user, bucket and policy operations the operator needs.
//
// It talks directly to radosgw (e.g. https://radosgw.service.consul:443) rather
// than the manager dashboard, via two native Go libraries:
//
//   - github.com/ceph/go-ceph/rgw/admin drives the RGW Admin Ops API
//     (/admin/...), signed with the operator's admin access/secret key, to
//     manage users, keys, quotas and bucket info/removal.
//   - github.com/aws/aws-sdk-go-v2/service/s3 drives the S3 API (/), signed as
//     the bucket's owner, to create buckets and set versioning, tagging, policy
//     and object lock — operations the Admin Ops API does not expose.
package ceph

import (
	"context"
	"crypto/tls"
	"crypto/x509"
	"errors"
	"fmt"
	"net/http"
	"strings"
	"sync"
	"time"

	"github.com/aws/aws-sdk-go-v2/aws"
	"github.com/aws/aws-sdk-go-v2/credentials"
	"github.com/aws/aws-sdk-go-v2/service/s3"
	smithy "github.com/aws/smithy-go"
	awshttp "github.com/aws/smithy-go/transport/http"
	"github.com/ceph/go-ceph/rgw/admin"
)

// Config configures a Client.
type Config struct {
	// Endpoint is the radosgw root, e.g. https://radosgw.service.consul:443.
	Endpoint string
	// AccessKey / SecretKey are the S3 credentials of an RGW user holding the
	// admin caps the operator needs (users=*, buckets=*).
	AccessKey string
	SecretKey string
	// Region is the SigV4 credential-scope region used for S3 requests. radosgw
	// verifies the signature against whatever region the client used, so any
	// consistent value works; defaults to "default". (The go-ceph admin client
	// always signs its own requests with region "default".)
	Region string
	// CACert is an optional PEM bundle used to verify the radosgw TLS cert.
	CACert []byte
	// Insecure disables TLS verification (not recommended).
	Insecure bool
	// Timeout bounds each HTTP request. Defaults to 30s.
	Timeout time.Duration
}

// Client talks to radosgw. It is safe for concurrent use.
type Client struct {
	admin  *admin.API
	s3     *s3.Client
	region string

	// keyCache memoises owner uid -> S3 credentials (via the Admin Ops API) so
	// per-owner S3 calls do not re-fetch keys on every reconcile.
	mu       sync.Mutex
	keyCache map[string]aws.CredentialsProvider
}

// notFoundCodes and conflictCodes classify RGW/S3 error codes that surface only
// as a generic smithy.APIError (i.e. not a modeled S3 error type).
var notFoundCodes = map[string]bool{
	"NoSuchUser": true, "NoSuchBucket": true, "NoSuchKey": true,
	"NoSuchBucketPolicy": true, "NoSuchTagSet": true,
	"NoSuchTagSetError": true, "NotFound": true,
}

var conflictCodes = map[string]bool{
	"BucketAlreadyExists": true, "BucketAlreadyOwnedByYou": true, "UserAlreadyExists": true,
}

// IsNotFound reports whether err represents a missing user, bucket, key, policy
// or tag set, on either the admin or the S3 path.
func IsNotFound(err error) bool {
	if err == nil {
		return false
	}
	if errors.Is(err, admin.ErrNoSuchUser) || errors.Is(err, admin.ErrNoSuchBucket) ||
		errors.Is(err, admin.ErrNoSuchKey) || errors.Is(err, admin.ErrNoSuchObject) {
		return true
	}
	var apiErr smithy.APIError
	if errors.As(err, &apiErr) && notFoundCodes[apiErr.ErrorCode()] {
		return true
	}
	var respErr *awshttp.ResponseError
	if errors.As(err, &respErr) && respErr.HTTPStatusCode() == http.StatusNotFound {
		return true
	}
	return false
}

// IsConflict reports whether err represents an already-exists conflict on either
// the admin or the S3 path.
func IsConflict(err error) bool {
	if err == nil {
		return false
	}
	if errors.Is(err, admin.ErrUserExists) || errors.Is(err, admin.ErrEmailExists) ||
		errors.Is(err, admin.ErrKeyExists) || errors.Is(err, admin.ErrBucketNotEmpty) {
		return true
	}
	var apiErr smithy.APIError
	if errors.As(err, &apiErr) && conflictCodes[apiErr.ErrorCode()] {
		return true
	}
	var respErr *awshttp.ResponseError
	if errors.As(err, &respErr) && respErr.HTTPStatusCode() == http.StatusConflict {
		return true
	}
	return false
}

// NewClient validates cfg and builds a Client.
func NewClient(cfg Config) (*Client, error) {
	endpoint := strings.TrimRight(cfg.Endpoint, "/")
	if endpoint == "" {
		return nil, fmt.Errorf("ceph: radosgw endpoint is required")
	}
	if cfg.AccessKey == "" || cfg.SecretKey == "" {
		return nil, fmt.Errorf("ceph: radosgw admin access and secret key are required")
	}

	tlsCfg := &tls.Config{InsecureSkipVerify: cfg.Insecure} //nolint:gosec // opt-in via config
	if len(cfg.CACert) > 0 {
		pool := x509.NewCertPool()
		if !pool.AppendCertsFromPEM(cfg.CACert) {
			return nil, fmt.Errorf("ceph: failed to parse CA certificate PEM")
		}
		tlsCfg.RootCAs = pool
	}

	timeout := cfg.Timeout
	if timeout == 0 {
		timeout = 30 * time.Second
	}
	region := cfg.Region
	if region == "" {
		region = "default"
	}

	httpClient := &http.Client{
		Timeout:   timeout,
		Transport: &http.Transport{TLSClientConfig: tlsCfg},
	}

	adminAPI, err := admin.New(endpoint, cfg.AccessKey, cfg.SecretKey, httpClient)
	if err != nil {
		return nil, fmt.Errorf("ceph: build admin client: %w", err)
	}

	s3Client := s3.New(s3.Options{
		Region:       region,
		Credentials:  credentials.NewStaticCredentialsProvider(cfg.AccessKey, cfg.SecretKey, ""),
		HTTPClient:   httpClient,
		BaseEndpoint: aws.String(endpoint),
		// radosgw serves buckets path-style, not virtual-host style.
		UsePathStyle: true,
		// radosgw (pre-Reef backports) rejects the SDK's default CRC32 /
		// aws-chunked integrity protections; only send checksums when the API
		// requires them.
		RequestChecksumCalculation: aws.RequestChecksumCalculationWhenRequired,
		ResponseChecksumValidation: aws.ResponseChecksumValidationWhenRequired,
	})

	return &Client{
		admin:    adminAPI,
		s3:       s3Client,
		region:   region,
		keyCache: map[string]aws.CredentialsProvider{},
	}, nil
}

// asOwner returns a per-call S3 option that signs the request as the RGW user
// uid, looking up (and caching) the user's first key pair via the Admin Ops API.
// Signing S3 sub-resource operations as the bucket owner (rather than the admin
// user) makes the owner the bucket owner directly and keeps RGW's per-user S3
// authorization intact.
func (c *Client) asOwner(ctx context.Context, uid string) (func(*s3.Options), error) {
	c.mu.Lock()
	provider, ok := c.keyCache[uid]
	c.mu.Unlock()
	if !ok {
		user, err := c.GetUser(ctx, uid)
		if err != nil {
			return nil, err
		}
		key, has := user.S3Key()
		if !has {
			return nil, fmt.Errorf("ceph: user %s has no S3 keys to sign bucket operations", uid)
		}
		provider = credentials.NewStaticCredentialsProvider(key.AccessKey, key.SecretKey, "")
		c.mu.Lock()
		c.keyCache[uid] = provider
		c.mu.Unlock()
	}
	return func(o *s3.Options) { o.Credentials = provider }, nil
}

// forgetIdentity drops any cached S3 credentials for uid, e.g. after its keys
// may have changed.
func (c *Client) forgetIdentity(uid string) {
	c.mu.Lock()
	delete(c.keyCache, uid)
	c.mu.Unlock()
}

// Ping verifies connectivity and that the admin credentials sign correctly. It
// asks the Admin Ops API for a sentinel user: a NoSuchUser answer still proves
// the request authenticated, so only transport/auth errors fail the check.
func (c *Client) Ping(ctx context.Context) error {
	_, err := c.GetUser(ctx, "cephrgw-operator-ping-nonexistent")
	if err == nil || IsNotFound(err) {
		return nil
	}
	return err
}