Talk to radosgw directly via go-ceph + aws-sdk-go-v2

The operator drove the Ceph manager dashboard REST API to manage RGW users,
buckets and policies. That coupled it to a dashboard login, the dashboard's RGW
wiring, and the dashboard's bucket API surface. Rebuild the Ceph integration to
talk directly to radosgw the way the CLI does, using native Go libraries, while
keeping every operator capability identical.

The exported surface of internal/ceph is unchanged, so the three controllers
and cmd/operator's structure are untouched (bar the CEPH_RGW_* config plumbing).

- replace the internal/ceph client internals with github.com/ceph/go-ceph
  rgw/admin (Admin Ops API) for users, keys, quotas and bucket info/removal
- add github.com/aws/aws-sdk-go-v2 S3 client for bucket create, versioning,
  policy, tagging and object lock, signed as the bucket owner
- map go-ceph admin.ErrNoSuch*/ErrUserExists and smithy APIError codes into
  IsNotFound/IsConflict so controller create-vs-update branching is preserved
- set S3 path-style addressing and WhenRequired checksum modes for RGW
- delete the hand-rolled dashboard client, token auth and JSON plumbing
- keep policy.go/BuildBucketPolicy/BuildTagJSON as pure builders
- replace the client tests with NewClient validation and error-classifier tests
- keep CGO_ENABLED=0 distroless: only go-ceph's pure-Go rgw/admin is imported
- switch env/config to CEPH_RGW_* (endpoint, admin endpoint, access/secret key,
  region, CA, insecure) and update the deployment manifest
- rewrite README and docs/ceph-setup.md for the single RGW admin user
  (caps users=*;buckets=*), keeping Vault/VSO as the primary credential source

Claude-Session: https://claude.ai/code/session_016CEncETbf8cvy1PhsHfFHM

internal/ceph/buckets.go

@@ -2,22 +2,25 @@ package ceph
 
 import (
 	"context"
-	"net/http"
-	"net/url"
-	"strconv"
+	"encoding/json"
+	"fmt"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/service/s3"
+	s3types "github.com/aws/aws-sdk-go-v2/service/s3/types"
+	"github.com/ceph/go-ceph/rgw/admin"
 )
 
-// BucketInfo is the subset of an RGW bucket record the operator consumes.
-// Different Ceph releases name the id/name fields slightly differently, so the
-// struct captures the known variants and Name/ID normalise them.
+// BucketInfo is the subset of an RGW bucket record the operator consumes, as
+// returned by the Admin Ops API (GET /admin/bucket).
 type BucketInfo struct {
-	Bucket string `json:"bucket"`
-	Bid    string `json:"bid"`
-	ID     string `json:"id"`
-	Owner  string `json:"owner"`
+	Bucket string
+	Bid    string
+	ID     string
+	Owner  string
 }
 
-// Name returns the bucket name regardless of the field the dashboard used.
+// Name returns the bucket name regardless of the field radosgw used.
 func (b *BucketInfo) Name() string {
 	if b.Bucket != "" {
 		return b.Bucket
@@ -40,90 +43,160 @@ type CreateBucketSpec struct {
 	LockYears       *int32
 }
 
-type createBucketRequest struct {
-	Bucket          string `json:"bucket"`
-	UID             string `json:"uid"`
-	Zonegroup       string `json:"zonegroup,omitempty"`
-	PlacementTarget string `json:"placement_target,omitempty"`
-	LockEnabled     string `json:"lock_enabled"`
-	LockMode        string `json:"lock_mode,omitempty"`
-	LockDays        string `json:"lock_retention_period_days,omitempty"`
-	LockYears       string `json:"lock_retention_period_years,omitempty"`
-}
-
-// GetBucket fetches a bucket by name, returning an *APIError with status 404
-// (see IsNotFound) when it does not exist.
+// GetBucket fetches a bucket by name via the Admin Ops API, returning an error
+// classified by IsNotFound (admin.ErrNoSuchBucket) when it does not exist.
 func (c *Client) GetBucket(ctx context.Context, name string) (*BucketInfo, error) {
-	var b BucketInfo
-	if err := c.do(ctx, http.MethodGet, "/api/rgw/bucket/"+url.PathEscape(name), nil, &b, ""); err != nil {
+	b, err := c.admin.GetBucketInfo(ctx, admin.Bucket{Bucket: name})
+	if err != nil {
 		return nil, err
 	}
-	return &b, nil
+	return &BucketInfo{Bucket: b.Bucket, ID: b.ID, Owner: b.Owner}, nil
 }
 
-// CreateBucket provisions a bucket owned by spec.OwnerUID.
+// CreateBucket provisions a bucket owned by spec.OwnerUID. The Admin Ops API
+// cannot create buckets, so the operator issues an S3 CreateBucket signed as the
+// owner (which makes the owner the bucket owner directly).
 func (c *Client) CreateBucket(ctx context.Context, spec CreateBucketSpec) (*BucketInfo, error) {
-	req := createBucketRequest{
-		Bucket:          spec.Bucket,
-		UID:             spec.OwnerUID,
-		Zonegroup:       spec.Zonegroup,
-		PlacementTarget: spec.PlacementTarget,
-		LockEnabled:     strconv.FormatBool(spec.LockEnabled),
-		LockMode:        spec.LockMode,
-	}
-	if spec.LockDays != nil {
-		req.LockDays = strconv.Itoa(int(*spec.LockDays))
-	}
-	if spec.LockYears != nil {
-		req.LockYears = strconv.Itoa(int(*spec.LockYears))
-	}
-	var b BucketInfo
-	if err := c.do(ctx, http.MethodPost, "/api/rgw/bucket", req, &b, ""); err != nil {
+	owner, err := c.asOwner(ctx, spec.OwnerUID)
+	if err != nil {
 		return nil, err
 	}
-	return &b, nil
-}
 
-type setBucketRequest struct {
-	BucketID        string  `json:"bucket_id"`
-	UID             string  `json:"uid"`
-	VersioningState *string `json:"versioning_state,omitempty"`
-	BucketPolicy    *string `json:"bucket_policy,omitempty"`
-	Tags            *string `json:"tags,omitempty"`
-}
-
-// SetBucketVersioning enables or suspends S3 versioning on a bucket.
-func (c *Client) SetBucketVersioning(ctx context.Context, name, bucketID, ownerUID string, enabled bool) error {
-	state := "Suspended"
-	if enabled {
-		state = "Enabled"
+	input := &s3.CreateBucketInput{Bucket: aws.String(spec.Bucket)}
+	if loc := locationConstraint(spec.Zonegroup, spec.PlacementTarget); loc != "" {
+		input.CreateBucketConfiguration = &s3types.CreateBucketConfiguration{
+			LocationConstraint: s3types.BucketLocationConstraint(loc),
+		}
 	}
-	req := setBucketRequest{BucketID: bucketID, UID: ownerUID, VersioningState: &state}
-	return c.do(ctx, http.MethodPut, "/api/rgw/bucket/"+url.PathEscape(name), req, nil, "")
+	if spec.LockEnabled {
+		input.ObjectLockEnabledForBucket = aws.Bool(true)
+	}
+
+	if _, err := c.s3.CreateBucket(ctx, input, owner); err != nil && !IsConflict(err) {
+		return nil, err
+	}
+
+	// Apply a default object-lock retention when requested.
+	if spec.LockEnabled && spec.LockMode != "" && (spec.LockDays != nil || spec.LockYears != nil) {
+		if err := c.setObjectLockDefault(ctx, owner, spec); err != nil {
+			return nil, err
+		}
+	}
+
+	return c.GetBucket(ctx, spec.Bucket)
 }
 
-// SetBucketPolicy replaces the S3 bucket policy. An empty policy string asks the
-// dashboard to clear it; not every release honours clearing, so callers should
-// treat a clear as best-effort.
+// SetBucketVersioning enables or suspends S3 versioning on a bucket. bucketID is
+// unused (kept for call-site stability).
+func (c *Client) SetBucketVersioning(ctx context.Context, name, bucketID, ownerUID string, enabled bool) error {
+	owner, err := c.asOwner(ctx, ownerUID)
+	if err != nil {
+		return err
+	}
+	status := s3types.BucketVersioningStatusSuspended
+	if enabled {
+		status = s3types.BucketVersioningStatusEnabled
+	}
+	_, err = c.s3.PutBucketVersioning(ctx, &s3.PutBucketVersioningInput{
+		Bucket:                  aws.String(name),
+		VersioningConfiguration: &s3types.VersioningConfiguration{Status: status},
+	}, owner)
+	return err
+}
+
+// SetBucketPolicy replaces the S3 bucket policy. An empty policy clears it.
+// bucketID is unused (kept for call-site stability).
 func (c *Client) SetBucketPolicy(ctx context.Context, name, bucketID, ownerUID, policy string) error {
-	req := setBucketRequest{BucketID: bucketID, UID: ownerUID, BucketPolicy: &policy}
-	return c.do(ctx, http.MethodPut, "/api/rgw/bucket/"+url.PathEscape(name), req, nil, "")
+	owner, err := c.asOwner(ctx, ownerUID)
+	if err != nil {
+		return err
+	}
+	if policy == "" {
+		_, err := c.s3.DeleteBucketPolicy(ctx, &s3.DeleteBucketPolicyInput{Bucket: aws.String(name)}, owner)
+		if IsNotFound(err) {
+			return nil
+		}
+		return err
+	}
+	_, err = c.s3.PutBucketPolicy(ctx, &s3.PutBucketPolicyInput{
+		Bucket: aws.String(name),
+		Policy: aws.String(policy),
+	}, owner)
+	return err
 }
 
-// SetBucketTags replaces the bucket tag set. tagsJSON is the RGW/S3 tag JSON
-// (a list of {"Key","Value"} objects).
+// SetBucketTags replaces the bucket tag set. tagsJSON is the JSON produced by
+// BuildTagJSON (a list of {"Key","Value"} objects). bucketID is unused (kept for
+// call-site stability).
 func (c *Client) SetBucketTags(ctx context.Context, name, bucketID, ownerUID, tagsJSON string) error {
-	req := setBucketRequest{BucketID: bucketID, UID: ownerUID, Tags: &tagsJSON}
-	return c.do(ctx, http.MethodPut, "/api/rgw/bucket/"+url.PathEscape(name), req, nil, "")
+	owner, err := c.asOwner(ctx, ownerUID)
+	if err != nil {
+		return err
+	}
+
+	type tag struct {
+		Key   string `json:"Key"`
+		Value string `json:"Value"`
+	}
+	var tags []tag
+	if tagsJSON != "" {
+		if err := json.Unmarshal([]byte(tagsJSON), &tags); err != nil {
+			return fmt.Errorf("ceph: parse bucket tags: %w", err)
+		}
+	}
+	if len(tags) == 0 {
+		_, err := c.s3.DeleteBucketTagging(ctx, &s3.DeleteBucketTaggingInput{Bucket: aws.String(name)}, owner)
+		if IsNotFound(err) {
+			return nil
+		}
+		return err
+	}
+
+	tagSet := make([]s3types.Tag, 0, len(tags))
+	for _, t := range tags {
+		tagSet = append(tagSet, s3types.Tag{Key: aws.String(t.Key), Value: aws.String(t.Value)})
+	}
+	_, err = c.s3.PutBucketTagging(ctx, &s3.PutBucketTaggingInput{
+		Bucket:  aws.String(name),
+		Tagging: &s3types.Tagging{TagSet: tagSet},
+	}, owner)
+	return err
 }
 
-// DeleteBucket removes a bucket. When purge is true its objects are deleted too;
-// otherwise deletion of a non-empty bucket fails. A 404 is treated as success.
+// DeleteBucket removes a bucket via the Admin Ops API. When purge is true its
+// objects are deleted too. A NoSuchBucket response is treated as success.
 func (c *Client) DeleteBucket(ctx context.Context, name string, purge bool) error {
-	path := "/api/rgw/bucket/" + url.PathEscape(name) + "?purge_objects=" + strconv.FormatBool(purge)
-	err := c.do(ctx, http.MethodDelete, path, nil, nil, "")
+	err := c.admin.RemoveBucket(ctx, admin.Bucket{Bucket: name, PurgeObject: &purge})
 	if IsNotFound(err) {
 		return nil
 	}
 	return err
 }
+
+// setObjectLockDefault sets the bucket's default object-lock retention.
+func (c *Client) setObjectLockDefault(ctx context.Context, owner func(*s3.Options), spec CreateBucketSpec) error {
+	_, err := c.s3.PutObjectLockConfiguration(ctx, &s3.PutObjectLockConfigurationInput{
+		Bucket: aws.String(spec.Bucket),
+		ObjectLockConfiguration: &s3types.ObjectLockConfiguration{
+			ObjectLockEnabled: s3types.ObjectLockEnabledEnabled,
+			Rule: &s3types.ObjectLockRule{
+				DefaultRetention: &s3types.DefaultRetention{
+					Mode:  s3types.ObjectLockRetentionMode(spec.LockMode),
+					Days:  spec.LockDays,
+					Years: spec.LockYears,
+				},
+			},
+		},
+	}, owner)
+	return err
+}
+
+// locationConstraint renders the RGW LocationConstraint from a zonegroup and
+// placement target ("<zonegroup>:<placement>"), or "" for default placement.
+func locationConstraint(zonegroup, placement string) string {
+	loc := zonegroup
+	if placement != "" {
+		loc = zonegroup + ":" + placement
+	}
+	return loc
+}

internal/ceph/client.go

@@ -1,39 +1,50 @@
-// Package ceph is a small client for the Ceph manager dashboard REST API,
-// scoped to the RGW (S3) user and bucket endpoints the operator needs.
+// 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.
 //
-// The dashboard authenticates with a username/password to POST /api/auth, which
-// returns a bearer (JWT) token. The client caches that token and transparently
-// re-authenticates when the server returns 401 (expired/invalid token).
+// 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 (
-	"bytes"
 	"context"
 	"crypto/tls"
 	"crypto/x509"
-	"encoding/json"
 	"errors"
 	"fmt"
-	"io"
 	"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"
 )
 
-// defaultAccept is the versioned media type the Ceph dashboard requires on its
-// RGW endpoints. The dashboard rejects requests without a matching version.
-const defaultAccept = "application/vnd.ceph.api.v1.0+json"
-
-// Config configures a dashboard Client.
+// Config configures a Client.
 type Config struct {
-	// BaseURL is the dashboard root, e.g. https://dashboard.ceph.unkin.net.
-	BaseURL string
-	// Username / Password authenticate to POST /api/auth. The account needs the
-	// rgw-manager role (or admin) on the dashboard.
-	Username string
-	Password string
-	// CACert is an optional PEM bundle used to verify the dashboard TLS cert.
+	// 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
@@ -41,49 +52,80 @@ type Config struct {
 	Timeout time.Duration
 }
 
-// Client talks to the Ceph dashboard API. It is safe for concurrent use.
+// Client talks to radosgw. It is safe for concurrent use.
 type Client struct {
-	base string
-	user string
-	pass string
-	http *http.Client
+	admin  *admin.API
+	s3     *s3.Client
+	region string
 
-	mu    sync.Mutex
-	token 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
 }
 
-// APIError is returned for any non-2xx dashboard response.
-type APIError struct {
-	Status int
-	Method string
-	Path   string
-	Body   string
+// 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,
 }
 
-func (e *APIError) Error() string {
-	return fmt.Sprintf("ceph dashboard %s %s: status %d: %s", e.Method, e.Path, e.Status, e.Body)
+var conflictCodes = map[string]bool{
+	"BucketAlreadyExists": true, "BucketAlreadyOwnedByYou": true, "UserAlreadyExists": true,
 }
 
-// IsNotFound reports whether err is a 404 from the dashboard.
+// 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 {
-	var a *APIError
-	return errors.As(err, &a) && a.Status == http.StatusNotFound
+	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 is a 409 from the dashboard.
+// IsConflict reports whether err represents an already-exists conflict on either
+// the admin or the S3 path.
 func IsConflict(err error) bool {
-	var a *APIError
-	return errors.As(err, &a) && a.Status == http.StatusConflict
+	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) {
-	base := strings.TrimRight(cfg.BaseURL, "/")
-	if base == "" {
-		return nil, fmt.Errorf("ceph: dashboard base URL is required")
+	endpoint := strings.TrimRight(cfg.Endpoint, "/")
+	if endpoint == "" {
+		return nil, fmt.Errorf("ceph: radosgw endpoint is required")
 	}
-	if cfg.Username == "" || cfg.Password == "" {
-		return nil, fmt.Errorf("ceph: dashboard username and password are 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
@@ -99,124 +141,84 @@ func NewClient(cfg Config) (*Client, error) {
 	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{
-		base: base,
-		user: cfg.Username,
-		pass: cfg.Password,
-		http: &http.Client{
-			Timeout:   timeout,
-			Transport: &http.Transport{TLSClientConfig: tlsCfg},
-		},
+		admin:    adminAPI,
+		s3:       s3Client,
+		region:   region,
+		keyCache: map[string]aws.CredentialsProvider{},
 	}, nil
 }
 
-// do performs an authenticated request, decoding a 2xx JSON body into out (when
-// non-nil). On a 401 it drops the cached token, re-authenticates, and retries
-// once. accept overrides the Accept header version when non-empty.
-func (c *Client) do(ctx context.Context, method, path string, body, out any, accept string) error {
-	if accept == "" {
-		accept = defaultAccept
-	}
-	tok, err := c.ensureToken(ctx)
-	if err != nil {
-		return err
-	}
-	status, err := c.execute(ctx, method, path, body, accept, tok, out)
-	if status == http.StatusUnauthorized {
-		c.clearToken()
-		tok, err = c.ensureToken(ctx)
+// 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 err
+			return nil, err
 		}
-		_, err = c.execute(ctx, method, path, body, accept, tok, out)
+		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 err
+	return func(o *s3.Options) { o.Credentials = provider }, nil
 }
 
-func (c *Client) ensureToken(ctx context.Context) (string, error) {
+// 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()
-	defer c.mu.Unlock()
-	if c.token != "" {
-		return c.token, nil
-	}
-	tok, err := c.login(ctx)
-	if err != nil {
-		return "", err
-	}
-	c.token = tok
-	return tok, nil
-}
-
-func (c *Client) clearToken() {
-	c.mu.Lock()
-	c.token = ""
+	delete(c.keyCache, uid)
 	c.mu.Unlock()
 }
 
-func (c *Client) login(ctx context.Context) (string, error) {
-	var out struct {
-		Token string `json:"token"`
-	}
-	payload := map[string]string{"username": c.user, "password": c.pass}
-	if _, err := c.execute(ctx, http.MethodPost, "/api/auth", payload, defaultAccept, "", &out); err != nil {
-		return "", fmt.Errorf("dashboard login failed: %w", err)
-	}
-	if out.Token == "" {
-		return "", fmt.Errorf("dashboard login returned no token")
-	}
-	return out.Token, nil
-}
-
-// execute runs a single request and returns the HTTP status. A non-2xx status
-// yields an *APIError. token is sent as a bearer when non-empty.
-func (c *Client) execute(ctx context.Context, method, path string, body any, accept, token string, out any) (int, error) {
-	var reader io.Reader
-	if body != nil {
-		b, err := json.Marshal(body)
-		if err != nil {
-			return 0, fmt.Errorf("marshal request body: %w", err)
-		}
-		reader = bytes.NewReader(b)
-	}
-
-	req, err := http.NewRequestWithContext(ctx, method, c.base+path, reader)
-	if err != nil {
-		return 0, err
-	}
-	req.Header.Set("Accept", accept)
-	if body != nil {
-		req.Header.Set("Content-Type", "application/json")
-	}
-	if token != "" {
-		req.Header.Set("Authorization", "Bearer "+token)
-	}
-
-	resp, err := c.http.Do(req)
-	if err != nil {
-		return 0, err
-	}
-	defer resp.Body.Close()
-
-	data, _ := io.ReadAll(resp.Body)
-	if resp.StatusCode < 200 || resp.StatusCode >= 300 {
-		return resp.StatusCode, &APIError{
-			Status: resp.StatusCode,
-			Method: method,
-			Path:   path,
-			Body:   strings.TrimSpace(string(data)),
-		}
-	}
-	if out != nil && len(data) > 0 {
-		if err := json.Unmarshal(data, out); err != nil {
-			return resp.StatusCode, fmt.Errorf("decode %s %s response: %w", method, path, err)
-		}
-	}
-	return resp.StatusCode, nil
-}
-
-// Ping verifies connectivity and credentials by authenticating.
+// 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.ensureToken(ctx)
+	_, err := c.GetUser(ctx, "cephrgw-operator-ping-nonexistent")
+	if err == nil || IsNotFound(err) {
+		return nil
+	}
 	return err
 }

internal/ceph/client_test.go

@@ -0,0 +1,81 @@
+package ceph
+
+import (
+	"testing"
+
+	s3types "github.com/aws/aws-sdk-go-v2/service/s3/types"
+	smithy "github.com/aws/smithy-go"
+	"github.com/ceph/go-ceph/rgw/admin"
+)
+
+func TestNewClientValidation(t *testing.T) {
+	cases := []struct {
+		name    string
+		cfg     Config
+		wantErr bool
+	}{
+		{"ok", Config{Endpoint: "https://rgw:443", AccessKey: "a", SecretKey: "s"}, false},
+		{"no endpoint", Config{AccessKey: "a", SecretKey: "s"}, true},
+		{"no access key", Config{Endpoint: "https://rgw:443", SecretKey: "s"}, true},
+		{"no secret key", Config{Endpoint: "https://rgw:443", AccessKey: "a"}, true},
+		{"bad ca", Config{Endpoint: "https://rgw:443", AccessKey: "a", SecretKey: "s", CACert: []byte("not pem")}, true},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			_, err := NewClient(tc.cfg)
+			if (err != nil) != tc.wantErr {
+				t.Fatalf("NewClient err=%v wantErr=%v", err, tc.wantErr)
+			}
+		})
+	}
+}
+
+func TestIsNotFound(t *testing.T) {
+	cases := []struct {
+		name string
+		err  error
+		want bool
+	}{
+		{"nil", nil, false},
+		{"admin no such user", admin.ErrNoSuchUser, true},
+		{"admin no such bucket", admin.ErrNoSuchBucket, true},
+		{"admin no such key", admin.ErrNoSuchKey, true},
+		{"s3 no such bucket", &s3types.NoSuchBucket{}, true},
+		{"s3 no such key", &s3types.NoSuchKey{}, true},
+		{"generic no such bucket policy", &smithy.GenericAPIError{Code: "NoSuchBucketPolicy"}, true},
+		{"generic no such tag set", &smithy.GenericAPIError{Code: "NoSuchTagSet"}, true},
+		{"admin user exists is not notfound", admin.ErrUserExists, false},
+		{"unrelated", &smithy.GenericAPIError{Code: "AccessDenied"}, false},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			if got := IsNotFound(tc.err); got != tc.want {
+				t.Errorf("IsNotFound(%v)=%v want %v", tc.err, got, tc.want)
+			}
+		})
+	}
+}
+
+func TestIsConflict(t *testing.T) {
+	cases := []struct {
+		name string
+		err  error
+		want bool
+	}{
+		{"nil", nil, false},
+		{"admin user exists", admin.ErrUserExists, true},
+		{"admin bucket not empty", admin.ErrBucketNotEmpty, true},
+		{"s3 bucket already owned by you", &s3types.BucketAlreadyOwnedByYou{}, true},
+		{"s3 bucket already exists", &s3types.BucketAlreadyExists{}, true},
+		{"generic bucket already exists", &smithy.GenericAPIError{Code: "BucketAlreadyExists"}, true},
+		{"admin no such user is not conflict", admin.ErrNoSuchUser, false},
+		{"unrelated", &smithy.GenericAPIError{Code: "AccessDenied"}, false},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			if got := IsConflict(tc.err); got != tc.want {
+				t.Errorf("IsConflict(%v)=%v want %v", tc.err, got, tc.want)
+			}
+		})
+	}
+}

internal/ceph/policy.go

@@ -139,7 +139,8 @@ func sid(prefix, uid string) string {
 	return b.String()
 }
 
-// BuildTagJSON renders bucket tags in the JSON form the dashboard expects.
+// BuildTagJSON renders bucket tags as a {Key,Value} JSON list, the intermediate
+// form SetBucketTags parses and re-encodes into the S3 Tagging XML document.
 func BuildTagJSON(tags map[string]string) (string, error) {
 	if len(tags) == 0 {
 		return "", nil

internal/ceph/users.go

@@ -2,25 +2,25 @@ package ceph
 
 import (
 	"context"
-	"net/http"
-	"net/url"
+
+	"github.com/ceph/go-ceph/rgw/admin"
 )
 
 // UserKey is an S3 access/secret key pair belonging to an RGW user.
 type UserKey struct {
-	User      string `json:"user"`
-	AccessKey string `json:"access_key"`
-	SecretKey string `json:"secret_key"`
+	User      string
+	AccessKey string
+	SecretKey string
 }
 
 // User is the subset of an RGW user record the operator consumes.
 type User struct {
-	UID         string    `json:"user_id"`
-	DisplayName string    `json:"display_name"`
-	Email       string    `json:"email"`
-	MaxBuckets  int       `json:"max_buckets"`
-	Suspended   int       `json:"suspended"`
-	Keys        []UserKey `json:"keys"`
+	UID         string
+	DisplayName string
+	Email       string
+	MaxBuckets  int
+	Suspended   int
+	Keys        []UserKey
 }
 
 // S3Key returns the first access/secret key pair, if any.
@@ -40,92 +40,87 @@ type UserSpec struct {
 	Suspended   bool
 }
 
-type createUserRequest struct {
-	UID         string `json:"uid"`
-	DisplayName string `json:"display_name"`
-	Email       string `json:"email,omitempty"`
-	MaxBuckets  *int32 `json:"max_buckets,omitempty"`
-	Suspended   bool   `json:"suspended"`
-	GenerateKey bool   `json:"generate_key"`
+// fromAdminUser converts a go-ceph admin.User into the subset the operator uses.
+func fromAdminUser(u admin.User) *User {
+	out := &User{
+		UID:         u.ID,
+		DisplayName: u.DisplayName,
+		Email:       u.Email,
+		MaxBuckets:  derefInt(u.MaxBuckets),
+		Suspended:   derefInt(u.Suspended),
+	}
+	for _, k := range u.Keys {
+		out.Keys = append(out.Keys, UserKey{User: k.User, AccessKey: k.AccessKey, SecretKey: k.SecretKey})
+	}
+	return out
 }
 
-type updateUserRequest struct {
-	DisplayName string `json:"display_name"`
-	Email       string `json:"email,omitempty"`
-	MaxBuckets  *int32 `json:"max_buckets,omitempty"`
-	Suspended   bool   `json:"suspended"`
-}
-
-// GetUser fetches an RGW user by uid, returning an *APIError with status 404
-// (see IsNotFound) when it does not exist.
+// GetUser fetches an RGW user by uid, returning an error classified by
+// IsNotFound (admin.ErrNoSuchUser) when it does not exist.
 func (c *Client) GetUser(ctx context.Context, uid string) (*User, error) {
-	var u User
-	if err := c.do(ctx, http.MethodGet, "/api/rgw/user/"+url.PathEscape(uid), nil, &u, ""); err != nil {
+	u, err := c.admin.GetUser(ctx, admin.User{ID: uid})
+	if err != nil {
 		return nil, err
 	}
-	return &u, nil
+	return fromAdminUser(u), nil
 }
 
-// CreateUser creates an RGW user, asking the dashboard to generate an S3 key
-// pair. The returned User carries the generated keys.
+// CreateUser creates an RGW user, asking radosgw to generate an S3 key pair. The
+// returned User carries the generated keys.
 func (c *Client) CreateUser(ctx context.Context, spec UserSpec) (*User, error) {
-	req := createUserRequest{
-		UID:         spec.UID,
+	u, err := c.admin.CreateUser(ctx, admin.User{
+		ID:          spec.UID,
 		DisplayName: firstNonEmpty(spec.DisplayName, spec.UID),
 		Email:       spec.Email,
-		MaxBuckets:  spec.MaxBuckets,
-		Suspended:   spec.Suspended,
-		GenerateKey: true,
-	}
-	var u User
-	if err := c.do(ctx, http.MethodPost, "/api/rgw/user", req, &u, ""); err != nil {
+		MaxBuckets:  int32PtrToIntPtr(spec.MaxBuckets),
+		Suspended:   boolToIntPtr(spec.Suspended),
+		GenerateKey: boolPtr(true),
+	})
+	if err != nil {
 		return nil, err
 	}
-	return &u, nil
+	c.forgetIdentity(spec.UID)
+	return fromAdminUser(u), nil
 }
 
 // UpdateUser reconciles the mutable attributes of an existing RGW user.
 func (c *Client) UpdateUser(ctx context.Context, spec UserSpec) (*User, error) {
-	req := updateUserRequest{
+	u, err := c.admin.ModifyUser(ctx, admin.User{
+		ID:          spec.UID,
 		DisplayName: firstNonEmpty(spec.DisplayName, spec.UID),
 		Email:       spec.Email,
-		MaxBuckets:  spec.MaxBuckets,
-		Suspended:   spec.Suspended,
-	}
-	var u User
-	if err := c.do(ctx, http.MethodPut, "/api/rgw/user/"+url.PathEscape(spec.UID), req, &u, ""); err != nil {
+		MaxBuckets:  int32PtrToIntPtr(spec.MaxBuckets),
+		Suspended:   boolToIntPtr(spec.Suspended),
+	})
+	if err != nil {
 		return nil, err
 	}
-	return &u, nil
+	return fromAdminUser(u), nil
 }
 
-// DeleteUser removes an RGW user. A 404 is treated as success.
+// DeleteUser removes an RGW user. A NoSuchUser response is treated as success.
 func (c *Client) DeleteUser(ctx context.Context, uid string) error {
-	err := c.do(ctx, http.MethodDelete, "/api/rgw/user/"+url.PathEscape(uid), nil, nil, "")
+	err := c.admin.RemoveUser(ctx, admin.User{ID: uid})
+	c.forgetIdentity(uid)
 	if IsNotFound(err) {
 		return nil
 	}
 	return err
 }
 
-type quotaRequest struct {
-	QuotaType  string `json:"quota_type"`
-	Enabled    bool   `json:"enabled"`
-	MaxSizeKb  int64  `json:"max_size_kb"`
-	MaxObjects int64  `json:"max_objects"`
-}
-
 // SetUserQuota applies a quota to a user. quotaType is "user" or "bucket" (the
 // latter sets the per-bucket default for buckets the user owns). A nil or
 // negative limit means unlimited for that dimension.
 func (c *Client) SetUserQuota(ctx context.Context, uid, quotaType string, enabled bool, maxSizeBytes, maxObjects *int64) error {
-	req := quotaRequest{
+	maxSize := valueOr(maxSizeBytes, -1)
+	maxObj := valueOr(maxObjects, -1)
+	return c.admin.SetUserQuota(ctx, admin.QuotaSpec{
+		UID:        uid,
 		QuotaType:  quotaType,
-		Enabled:    enabled,
-		MaxSizeKb:  bytesToKb(maxSizeBytes),
-		MaxObjects: valueOr(maxObjects, -1),
-	}
-	return c.do(ctx, http.MethodPut, "/api/rgw/user/"+url.PathEscape(uid)+"/quota", req, nil, "")
+		Enabled:    &enabled,
+		MaxSize:    &maxSize,
+		MaxObjects: &maxObj,
+	})
 }
 
 func firstNonEmpty(vals ...string) string {
@@ -137,16 +132,34 @@ func firstNonEmpty(vals ...string) string {
 	return ""
 }
 
-func bytesToKb(b *int64) int64 {
-	if b == nil || *b < 0 {
-		return -1
-	}
-	return *b / 1024
-}
-
 func valueOr(v *int64, fallback int64) int64 {
 	if v == nil || *v < 0 {
 		return fallback
 	}
 	return *v
 }
+
+func derefInt(p *int) int {
+	if p == nil {
+		return 0
+	}
+	return *p
+}
+
+func boolPtr(b bool) *bool { return &b }
+
+func int32PtrToIntPtr(p *int32) *int {
+	if p == nil {
+		return nil
+	}
+	v := int(*p)
+	return &v
+}
+
+func boolToIntPtr(b bool) *int {
+	v := 0
+	if b {
+		v = 1
+	}
+	return &v
+}