Copilot commented on code in PR #14160: URL: https://github.com/apache/cloudstack/pull/14160#discussion_r4010843459
########## plugins/storage/object/seaweedfs/src/main/java/org/apache/cloudstack/storage/datastore/driver/SeaweedFSObjectStoreDriverImpl.java: ########## @@ -0,0 +1,554 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +// SPDX-License-Identifier: Apache-2.0 +package org.apache.cloudstack.storage.datastore.driver; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import javax.inject.Inject; + +import org.apache.cloudstack.engine.subsystem.api.storage.DataStore; +import org.apache.cloudstack.storage.datastore.db.ObjectStoreDao; +import org.apache.cloudstack.storage.datastore.db.ObjectStoreDetailsDao; +import org.apache.cloudstack.storage.datastore.db.ObjectStoreVO; +import org.apache.cloudstack.storage.datastore.util.SeaweedFSObjectStoreUtil; +import org.apache.cloudstack.storage.object.BaseObjectStoreDriverImpl; +import org.apache.cloudstack.storage.object.Bucket; +import org.apache.cloudstack.storage.object.BucketObject; + +import com.amazonaws.AmazonClientException; +import com.amazonaws.services.identitymanagement.AmazonIdentityManagement; +import com.amazonaws.services.identitymanagement.model.AccessKey; +import com.amazonaws.services.identitymanagement.model.AccessKeyMetadata; +import com.amazonaws.services.identitymanagement.model.CreateAccessKeyRequest; +import com.amazonaws.services.identitymanagement.model.CreateAccessKeyResult; +import com.amazonaws.services.identitymanagement.model.CreateUserRequest; +import com.amazonaws.services.identitymanagement.model.DeleteAccessKeyRequest; +import com.amazonaws.services.identitymanagement.model.EntityAlreadyExistsException; +import com.amazonaws.services.identitymanagement.model.ListAccessKeysRequest; +import com.amazonaws.services.identitymanagement.model.PutUserPolicyRequest; +import com.amazonaws.services.s3.AmazonS3; +import com.amazonaws.services.s3.model.AccessControlList; +import com.amazonaws.services.s3.model.BucketPolicy; +import com.amazonaws.services.s3.model.BucketVersioningConfiguration; +import com.amazonaws.services.s3.model.CreateBucketRequest; +import com.amazonaws.services.s3.model.DeleteBucketPolicyRequest; +import com.amazonaws.services.s3.model.GetBucketPolicyRequest; +import com.amazonaws.services.s3.model.SSEAlgorithm; +import com.amazonaws.services.s3.model.ServerSideEncryptionByDefault; +import com.amazonaws.services.s3.model.ServerSideEncryptionConfiguration; +import com.amazonaws.services.s3.model.ServerSideEncryptionRule; +import com.amazonaws.services.s3.model.SetBucketEncryptionRequest; +import com.amazonaws.services.s3.model.SetBucketVersioningConfigurationRequest; +import com.cloud.agent.api.to.BucketTO; +import com.cloud.agent.api.to.DataStoreTO; +import com.cloud.storage.BucketVO; +import com.cloud.storage.dao.BucketDao; +import com.cloud.user.Account; +import com.cloud.user.AccountDetailsDao; +import com.cloud.user.dao.AccountDao; +import com.cloud.utils.exception.CloudRuntimeException; + +/** + * SeaweedFS object store driver. + * + * Bucket operations use the AWS S3 SDK v1 (path-style access, endpoint-pinned). + * User/credential management uses the AWS IAM SDK v1, since SeaweedFS exposes a + * standard AWS IAM-compatible API. No proprietary admin client is needed. + * + * Modeled on CloudianHyperStoreObjectStoreDriverImpl, which uses the same + * S3 + IAM SDK pair. + */ +public class SeaweedFSObjectStoreDriverImpl extends BaseObjectStoreDriverImpl { + + @Inject + AccountDao _accountDao; + + @Inject + AccountDetailsDao _accountDetailsDao; + + @Inject + ObjectStoreDao _storeDao; + + @Inject + BucketDao _bucketDao; + + @Inject + ObjectStoreDetailsDao _storeDetailsDao; + + private static final String ACS_PREFIX = "acs"; + + @Override + public DataStoreTO getStoreTO(DataStore store) { + return null; + } + + /** + * Get the SeaweedFS IAM user name for the given CloudStack account. + * Uses the account UUID prefixed with "acs-" for namespacing. + */ + protected String getUserNameForAccount(Account account) { + return String.format("%s-%s", ACS_PREFIX, account.getUuid()); + } + + /** + * Create the IAM user for the CloudStack account if it doesn't exist, + * attach the restricted S3 policy, and ensure the account has a usable + * IAM access key persisted in its account details. + * + * <p>If a previously stored access key is still present in IAM, it is + * reused rather than rotated. A new key is only created when no stored + * key exists or the stored key is no longer found in IAM; in the latter + * case any unmanaged (leftover) keys for the user are deleted first to + * avoid hitting IAM access-key limits. This keeps bucket records that + * reference the stored credentials valid across repeated calls. + * + * @return true if the user exists or was created, false on failure. + */ + @Override + public boolean createUser(long accountId, long storeId) { + Account account = _accountDao.findById(accountId); + if (account == null) { + logger.error("Account {} not found", accountId); + return false; + } + String userName = getUserNameForAccount(account); + AmazonIdentityManagement iamClient = getIAMClient(storeId); + + // Create the IAM user if it doesn't already exist + try { + iamClient.createUser(new CreateUserRequest(userName)); + logger.info("Created IAM user {} for account {}", userName, account.getAccountName()); + } catch (EntityAlreadyExistsException e) { + logger.debug("IAM user {} already exists", userName); + } + + // Attach the restricted S3 policy (idempotent — overwrites if present) + iamClient.putUserPolicy(new PutUserPolicyRequest(userName, + "CloudStackPolicy", SeaweedFSObjectStoreUtil.IAM_USER_POLICY)); + + // Reuse the stored access key if it is still present in IAM; only + // create a new one when no usable key exists. + Map<String, String> details = _accountDetailsDao.findDetails(accountId); + String storedAccessKeyId = details.get(SeaweedFSObjectStoreUtil.KEY_ACCESS_KEY); + if (storedAccessKeyId != null && iamAccessKeyExists(iamClient, userName, storedAccessKeyId)) { + logger.debug("Reusing existing IAM access key {} for user {}", storedAccessKeyId, userName); + return true; + } + + // The stored key is missing or no longer in IAM. Clean up any + // unmanaged leftover keys before creating a replacement so we do not + // accumulate keys and hit IAM access-key limits. + deleteUnmanagedAccessKeys(iamClient, userName, storedAccessKeyId); + + CreateAccessKeyResult result = iamClient.createAccessKey( + new CreateAccessKeyRequest().withUserName(userName)); + AccessKey key = result.getAccessKey(); + + // Persist the credentials in the account details + details.put(SeaweedFSObjectStoreUtil.KEY_ACCESS_KEY, key.getAccessKeyId()); + details.put(SeaweedFSObjectStoreUtil.KEY_SECRET_KEY, key.getSecretAccessKey()); + _accountDetailsDao.persist(accountId, details); + + logger.info("Created IAM credentials {} for user {}", key.getAccessKeyId(), userName); + return true; + } + + /** + * Check whether the given access key id is still listed in IAM for the user. + */ + private boolean iamAccessKeyExists(AmazonIdentityManagement iamClient, String userName, String accessKeyId) { + try { + for (AccessKeyMetadata metadata : + iamClient.listAccessKeys(new ListAccessKeysRequest() + .withUserName(userName)).getAccessKeyMetadata()) { + if (accessKeyId.equals(metadata.getAccessKeyId())) { + return true; + } + } + } catch (AmazonClientException e) { + logger.warn("Failed to list IAM access keys for user {}: {}", userName, e.getMessage()); + } + return false; + } + + /** + * Delete access keys for the user other than the (optionally) preserved + * key id. Used to clean up unmanaged leftover keys before creating a + * replacement so repeated calls do not hit IAM access-key limits. + */ + private void deleteUnmanagedAccessKeys(AmazonIdentityManagement iamClient, String userName, String preserveAccessKeyId) { + try { + for (AccessKeyMetadata metadata : + iamClient.listAccessKeys(new ListAccessKeysRequest() + .withUserName(userName)).getAccessKeyMetadata()) { + String keyId = metadata.getAccessKeyId(); + if (preserveAccessKeyId != null && preserveAccessKeyId.equals(keyId)) { + continue; + } + DeleteAccessKeyRequest deleteReq = + new DeleteAccessKeyRequest() + .withUserName(userName) + .withAccessKeyId(keyId); + logger.info("Deleting un-managed IAM access key {} for user {}", keyId, userName); + iamClient.deleteAccessKey(deleteReq); + } + } catch (AmazonClientException e) { + logger.warn("Failed to clean up IAM access keys for user {}: {}", userName, e.getMessage()); + } + } + + @Override + public Bucket createBucket(Bucket bucket, boolean objectLock) { + String bucketName = bucket.getName(); + long storeId = bucket.getObjectStoreId(); + long accountId = bucket.getAccountId(); + + // Use the store's admin credentials to create the bucket + AmazonS3 s3client = getS3ClientByStoreId(storeId); + + // Check if the bucket already exists + try { + if (s3client.doesBucketExistV2(bucketName)) { + throw new CloudRuntimeException("Bucket already exists with name " + bucketName); + } + } catch (AmazonClientException e) { + throw new CloudRuntimeException(e); + } + + // Create the bucket + try { + CreateBucketRequest request = new CreateBucketRequest(bucketName); + if (objectLock) { + request.setObjectLockEnabledForBucket(true); + } + s3client.createBucket(request); + } catch (AmazonClientException e) { + logger.error("Create bucket failed", e); + throw new CloudRuntimeException(e); + } + + // Update the bucket record with the account's IAM credentials + Map<String, String> accountDetails = _accountDetailsDao.findDetails(accountId); + String accessKey = accountDetails.get(SeaweedFSObjectStoreUtil.KEY_ACCESS_KEY); + String secretKey = accountDetails.get(SeaweedFSObjectStoreUtil.KEY_SECRET_KEY); + if (accessKey == null || secretKey == null) { + logger.warn("No IAM credentials found for account {}. Bucket will be created without per-account credentials.", accountId); + } + + ObjectStoreVO store = _storeDao.findById(storeId); + String s3Url = getS3Url(storeId); + BucketVO bucketVO = _bucketDao.findById(bucket.getId()); + bucketVO.setAccessKey(accessKey); + bucketVO.setSecretKey(secretKey); + bucketVO.setBucketURL(s3Url + "/" + bucketName); + _bucketDao.update(bucket.getId(), bucketVO); + return bucket; + } + + @Override + public List<Bucket> listBuckets(long storeId) { + AmazonS3 s3client = getS3ClientByStoreId(storeId); + List<Bucket> bucketsList = new ArrayList<>(); + try { + List<com.amazonaws.services.s3.model.Bucket> s3Buckets = s3client.listBuckets(); + for (com.amazonaws.services.s3.model.Bucket s3Bucket : s3Buckets) { + Bucket bucket = new BucketObject(); + bucket.setName(s3Bucket.getName()); + bucketsList.add(bucket); + } + } catch (AmazonClientException e) { + throw new CloudRuntimeException(e); + } + return bucketsList; + } + + @Override + public boolean deleteBucket(BucketTO bucket, long storeId) { + AmazonS3 s3client = getS3ClientByStoreId(storeId); + try { + if (! s3client.doesBucketExistV2(bucket.getName())) { + throw new CloudRuntimeException("Bucket doesn't exist: " + bucket.getName()); + } + } catch (AmazonClientException e) { + throw new CloudRuntimeException(e); + } + try { + s3client.deleteBucket(bucket.getName()); + } catch (AmazonClientException e) { + throw new CloudRuntimeException(e); + } + return true; + } + + @Override + public AccessControlList getBucketAcl(BucketTO bucket, long storeId) { + AmazonS3 s3client = getS3ClientByStoreId(storeId); + try { + return s3client.getBucketAcl(bucket.getName()); + } catch (AmazonClientException e) { + throw new CloudRuntimeException(e); + } + } + + @Override + public void setBucketAcl(BucketTO bucket, AccessControlList acl, long storeId) { + AmazonS3 s3client = getS3ClientByStoreId(storeId); + try { + s3client.setBucketAcl(bucket.getName(), acl); + } catch (AmazonClientException e) { + throw new CloudRuntimeException(e); + } + } + + @Override + public void setBucketPolicy(BucketTO bucket, String policy, long storeId) { + if ("private".equalsIgnoreCase(policy)) { + deleteBucketPolicy(bucket, storeId); + return; + } + + StringBuilder sb = new StringBuilder(); + sb.append("{\n"); + sb.append(" \"Version\": \"2012-10-17\",\n"); + sb.append(" \"Statement\": [\n"); + sb.append(" {\n"); + sb.append(" \"Sid\": \"PublicReadForObjects\",\n"); + sb.append(" \"Effect\": \"Allow\",\n"); + sb.append(" \"Principal\": \"*\",\n"); + sb.append(" \"Action\": \"s3:GetObject\",\n"); + sb.append(" \"Resource\": \"arn:aws:s3:::%s/*\"\n"); + sb.append(" }\n"); + sb.append(" ]\n"); + sb.append("}\n"); + + String jsonPolicy = String.format(sb.toString(), bucket.getName()); + AmazonS3 s3client = getS3ClientByStoreId(storeId); + try { + s3client.setBucketPolicy(bucket.getName(), jsonPolicy); + } catch (AmazonClientException e) { + throw new CloudRuntimeException(e); + } + } + + @Override + public BucketPolicy getBucketPolicy(BucketTO bucket, long storeId) { + AmazonS3 s3client = getS3ClientByStoreId(storeId); + try { + return s3client.getBucketPolicy(new GetBucketPolicyRequest(bucket.getName())); + } catch (AmazonClientException e) { + throw new CloudRuntimeException(e); + } + } + + @Override + public void deleteBucketPolicy(BucketTO bucket, long storeId) { + AmazonS3 s3client = getS3ClientByStoreId(storeId); + try { + s3client.deleteBucketPolicy(new DeleteBucketPolicyRequest(bucket.getName())); + } catch (AmazonClientException e) { + throw new CloudRuntimeException(e); + } + } + + @Override + public boolean setBucketEncryption(BucketTO bucket, long storeId) { + AmazonS3 s3client = getS3ClientByStoreId(storeId); + try { + SetBucketEncryptionRequest eRequest = new SetBucketEncryptionRequest(); + eRequest.setBucketName(bucket.getName()); + + ServerSideEncryptionByDefault sseByDefault = new ServerSideEncryptionByDefault(); + sseByDefault.setSSEAlgorithm(SSEAlgorithm.AES256.toString()); + + ServerSideEncryptionRule sseRule = new ServerSideEncryptionRule(); + sseRule.setApplyServerSideEncryptionByDefault(sseByDefault); + + List<ServerSideEncryptionRule> sseRules = new ArrayList<>(); + sseRules.add(sseRule); + + ServerSideEncryptionConfiguration sseConf = new ServerSideEncryptionConfiguration(); + sseConf.setRules(sseRules); + + eRequest.setServerSideEncryptionConfiguration(sseConf); + s3client.setBucketEncryption(eRequest); + return true; + } catch (AmazonClientException e) { + throw new CloudRuntimeException(e); + } + } + + @Override + public boolean deleteBucketEncryption(BucketTO bucket, long storeId) { + AmazonS3 s3client = getS3ClientByStoreId(storeId); + try { + s3client.deleteBucketEncryption(bucket.getName()); + return true; + } catch (AmazonClientException e) { + throw new CloudRuntimeException(e); + } + } + + @Override + public boolean setBucketVersioning(BucketTO bucket, long storeId) { + AmazonS3 s3client = getS3ClientByStoreId(storeId); + try { + BucketVersioningConfiguration vConf = new BucketVersioningConfiguration(BucketVersioningConfiguration.ENABLED); + s3client.setBucketVersioningConfiguration( + new SetBucketVersioningConfigurationRequest(bucket.getName(), vConf)); + return true; + } catch (AmazonClientException e) { + throw new CloudRuntimeException(e); + } + } + + @Override + public boolean deleteBucketVersioning(BucketTO bucket, long storeId) { + AmazonS3 s3client = getS3ClientByStoreId(storeId); + try { + BucketVersioningConfiguration vConf = new BucketVersioningConfiguration(BucketVersioningConfiguration.SUSPENDED); + s3client.setBucketVersioningConfiguration( + new SetBucketVersioningConfigurationRequest(bucket.getName(), vConf)); + return true; + } catch (AmazonClientException e) { + throw new CloudRuntimeException(e); + } + } + + /** + * Set the bucket quota via the SeaweedFS admin REST API. + * + * SeaweedFS enforces bucket quota server-side by setting a read-only flag + * when usage exceeds the configured limit. The quota is configured via the + * SeaweedFS S3 extension endpoint PUT /{bucket}?seaweedfs-quota, + * authenticated via standard S3 SigV4 and authorized via the + * s3:PutBucketQuota IAM permission. + * + * @param size the GiB size to set the quota to. 0 disables quota. + * @throws CloudRuntimeException if the S3 endpoint or credentials are missing or the request fails. + */ + @Override + public void setBucketQuota(BucketTO bucket, long storeId, long size) { + String s3Url = getS3Url(storeId); + String accessKey = getAccessKey(storeId); + String secretKey = getSecretKey(storeId); + if (s3Url == null || s3Url.isEmpty() || accessKey == null || accessKey.isEmpty() || secretKey == null || secretKey.isEmpty()) { + throw new CloudRuntimeException("SeaweedFS S3 URL and credentials are required to set bucket quota. " + + "Configure 's3Url', 'accesskey', and 'secretkey' in the object store details."); + } + SeaweedFSObjectStoreUtil.setBucketQuotaViaS3Extension(s3Url, accessKey, secretKey, bucket.getName(), size, getS3ExtensionHttpClient()); + } + + /** + * Returns the HTTP client used to send SeaweedFS S3 extension requests + * (e.g. PUT /{bucket}?seaweedfs-quota). Exposed as a protected seam so + * tests can inject a mock client and assert the signed request without + * touching the network. + */ + protected java.net.http.HttpClient getS3ExtensionHttpClient() { + return java.net.http.HttpClient.newHttpClient(); Review Comment: `HttpClient.newHttpClient()` has no connect timeout, and `executeSignedS3Request` builds the request without a per-request timeout. A stalled or unreachable SeaweedFS endpoint can therefore block the synchronous bucket create/update API indefinitely. Configure bounded timeouts (preferably through the provider's configured client) before sending the quota request. ########## plugins/storage/object/seaweedfs/src/main/java/org/apache/cloudstack/storage/datastore/util/SeaweedFSObjectStoreUtil.java: ########## @@ -0,0 +1,362 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. +// SPDX-License-Identifier: Apache-2.0 +package org.apache.cloudstack.storage.datastore.util; + +import org.apache.commons.lang3.StringUtils; + +import com.amazonaws.AmazonServiceException; +import com.amazonaws.auth.AWSStaticCredentialsProvider; +import com.amazonaws.auth.BasicAWSCredentials; +import com.amazonaws.client.builder.AwsClientBuilder; +import com.amazonaws.services.identitymanagement.AmazonIdentityManagement; +import com.amazonaws.services.identitymanagement.AmazonIdentityManagementClientBuilder; +import com.amazonaws.services.s3.AmazonS3; +import com.amazonaws.services.s3.AmazonS3ClientBuilder; +import com.cloud.utils.exception.CloudRuntimeException; + +/** + * Utility class for the SeaweedFS object storage provider. + * + * SeaweedFS exposes both an S3-compatible API and an AWS IAM-compatible API, + * so this provider needs no proprietary admin client — only the AWS S3 and IAM + * SDKs, the same pair Cloudian HyperStore already uses in this tree. + */ +public class SeaweedFSObjectStoreUtil { + + /** The name of our Object Store Provider */ + public static final String OBJECT_STORE_PROVIDER_NAME = "SeaweedFS"; + + public static final String STORE_KEY_PROVIDER_NAME = "providerName"; + public static final String STORE_KEY_URL = "url"; + public static final String STORE_KEY_NAME = "name"; + public static final String STORE_KEY_SIZE = "size"; + public static final String STORE_KEY_DETAILS = "details"; + + // Store Details Map key names - managed outside of plugin + public static final String STORE_DETAILS_KEY_ACCESS_KEY = "accesskey"; // admin/root access key + public static final String STORE_DETAILS_KEY_SECRET_KEY = "secretkey"; // admin/root secret key + public static final String STORE_DETAILS_KEY_S3_URL = "s3Url"; // S3 endpoint URL + public static final String STORE_DETAILS_KEY_IAM_URL = "iamUrl"; // IAM endpoint URL + + // Account Detail Map key names - credentials created per CloudStack account + public static final String KEY_ACCESS_KEY = "swfs_AccessKey"; + public static final String KEY_SECRET_KEY = "swfs_SecretKey"; + + /** + * IAM user policy applied to each per-account IAM user. Grants full S3 + * access except bucket creation/deletion, so CloudStack retains control of + * bucket lifecycle while the account's IAM credentials can manage objects. + */ + public static final String IAM_USER_POLICY = "{\n" + + " \"Version\": \"2012-10-17\",\n" + + " \"Statement\": [\n" + + " {\n" + + " \"Sid\": \"AllowFullS3Access\",\n" + + " \"Effect\": \"Allow\",\n" + + " \"Action\": [\n" + + " \"s3:*\"\n" + + " ],\n" + + " \"Resource\": \"*\"\n" + + " },\n" + + " {\n" + + " \"Sid\": \"ExceptBucketCreationOrDeletion\",\n" + + " \"Effect\": \"Deny\",\n" + + " \"Action\": [\n" + + " \"s3:CreateBucket\",\n" + + " \"s3:DeleteBucket\"\n" + + " ],\n" + + " \"Resource\": \"*\"\n" + + " }\n" + + " ]\n" + + "}\n"; + + // The CloudStack service credential (the accesskey/secretkey configured on + // the object store) is the admin credential used for ALL driver operations: + // - AmazonS3 client: bucket CRUD, policy, versioning, encryption, listing + // - AmazonIdentityManagement client: per-account IAM user provisioning + // - setBucketQuotaViaS3Extension: PUT /{bucket}?seaweedfs-quota + // It must therefore have broad S3 and IAM permissions. It is NOT scoped + // down to only s3:PutBucketQuota/s3:GetBucketQuota — that was an earlier + // design idea that does not match the implementation. The per-account IAM + // users (created by createUser) are the ones with restricted permissions + // (see IAM_USER_POLICY above). + + /** + * Returns an S3 connection for the given endpoint and credentials. + * Uses path-style access, which SeaweedFS requires. + * + * @param url the url of the S3 service + * @param accessKey the credentials to use for the S3 connection. + * @param secretKey the matching secret key. + * @return an S3 connection (never null) + * @throws CloudRuntimeException on failure. + */ + public static AmazonS3 getS3Client(String url, String accessKey, String secretKey) { + AmazonS3 client = AmazonS3ClientBuilder.standard() + .enablePathStyleAccess() + .withCredentials(new AWSStaticCredentialsProvider(new BasicAWSCredentials(accessKey, secretKey))) + .withEndpointConfiguration(new AwsClientBuilder.EndpointConfiguration(url, "us-east-1")) + .build(); + if (client == null) { + throw new CloudRuntimeException("Error while creating SeaweedFS S3 client"); + } + return client; + } + + /** + * Returns an IAM connection for the given endpoint and credentials. + * + * @param url the url of the IAM service + * @param accessKey the credentials to use for the iam connection. + * @param secretKey the matching secret key. + * @return an IAM connection (never null) + * @throws CloudRuntimeException on failure. + */ + public static AmazonIdentityManagement getIAMClient(String url, String accessKey, String secretKey) { + AmazonIdentityManagement iamClient = AmazonIdentityManagementClientBuilder.standard() + .withCredentials(new AWSStaticCredentialsProvider(new BasicAWSCredentials(accessKey, secretKey))) + .withEndpointConfiguration(new AwsClientBuilder.EndpointConfiguration(url, "us-east-1")) + .build(); + if (iamClient == null) { + throw new CloudRuntimeException("Error while creating SeaweedFS IAM client"); + } + return iamClient; + } + + /** + * Test the S3Url to confirm it behaves like an S3 Service. + * + * Uses bad credentials and looks for the particular error from S3 that says + * InvalidAccessKeyId was used. Quietly returns if we connect and get the + * expected error back. + * + * @param s3Url the url to check + * @throws CloudRuntimeException if there is any unexpected issue. + */ + public static void validateS3Url(String s3Url) { + try { + AmazonS3 s3Client = SeaweedFSObjectStoreUtil.getS3Client(s3Url, "unknown", "unknown"); + s3Client.listBuckets(); + } catch (AmazonServiceException e) { + if (StringUtils.compareIgnoreCase(e.getErrorCode(), "InvalidAccessKeyId") != 0 + && StringUtils.compareIgnoreCase(e.getErrorCode(), "SignatureDoesNotMatch") != 0) { + throw new CloudRuntimeException("Unexpected response from S3 Endpoint.", e); + } + } + } + + /** + * Test the IAMUrl to confirm it behaves like an IAM Service. + * + * Uses bad credentials and looks for the particular error from IAM that says + * InvalidAccessKeyId or InvalidClientTokenId was used. Quietly returns if we + * connect and get the expected error back. + * + * @param iamUrl the url to check + * @throws CloudRuntimeException if there is any unexpected issue. + */ + public static void validateIAMUrl(String iamUrl) { + try { + AmazonIdentityManagement iamClient = SeaweedFSObjectStoreUtil.getIAMClient(iamUrl, "unknown", "unknown"); + iamClient.listAccessKeys(); + } catch (AmazonServiceException e) { + if (! StringUtils.equalsAnyIgnoreCase(e.getErrorCode(), "InvalidAccessKeyId", "InvalidClientTokenId", "SignatureDoesNotMatch")) { + throw new CloudRuntimeException("Unexpected response from IAM Endpoint.", e); + } + } + } + + /** + * Set bucket quota via the SeaweedFS S3 extension endpoint. + * + * SeaweedFS exposes a custom S3 subresource at + * PUT /{bucket}?seaweedfs-quota + * authenticated via standard S3 SigV4 and authorized via the + * s3:PutBucketQuota IAM permission. This avoids the need for a + * separate admin API credential. + * + * The request body is JSON: + * {"quota_size": <n>, "quota_unit": "GB", "quota_enabled": true} + * + * @param s3Url the S3 endpoint URL (e.g. http://host:8333) + * @param accessKey the S3 access key (must have s3:PutBucketQuota permission) + * @param secretKey the S3 secret key + * @param bucketName the bucket name + * @param sizeGiB the quota size in GiB (0 to disable quota) + * @throws CloudRuntimeException on any failure + */ + public static void setBucketQuotaViaS3Extension(String s3Url, String accessKey, String secretKey, String bucketName, long sizeGiB) { + setBucketQuotaViaS3Extension(s3Url, accessKey, secretKey, bucketName, sizeGiB, java.net.http.HttpClient.newHttpClient()); Review Comment: This quota request is executed synchronously from bucket create/update, but the default `HttpClient` and `HttpRequest` have neither a connect timeout nor a request timeout. If the SeaweedFS endpoint becomes unreachable, a management-server API thread can block indefinitely instead of failing promptly. Configure bounded connect and per-request timeouts (preferably using CloudStack's existing timeout configuration). ########## plugins/storage/object/seaweedfs/src/main/java/org/apache/cloudstack/storage/datastore/driver/SeaweedFSObjectStoreDriverImpl.java: ########## @@ -0,0 +1,554 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +// SPDX-License-Identifier: Apache-2.0 +package org.apache.cloudstack.storage.datastore.driver; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import javax.inject.Inject; + +import org.apache.cloudstack.engine.subsystem.api.storage.DataStore; +import org.apache.cloudstack.storage.datastore.db.ObjectStoreDao; +import org.apache.cloudstack.storage.datastore.db.ObjectStoreDetailsDao; +import org.apache.cloudstack.storage.datastore.db.ObjectStoreVO; +import org.apache.cloudstack.storage.datastore.util.SeaweedFSObjectStoreUtil; +import org.apache.cloudstack.storage.object.BaseObjectStoreDriverImpl; +import org.apache.cloudstack.storage.object.Bucket; +import org.apache.cloudstack.storage.object.BucketObject; + +import com.amazonaws.AmazonClientException; +import com.amazonaws.services.identitymanagement.AmazonIdentityManagement; +import com.amazonaws.services.identitymanagement.model.AccessKey; +import com.amazonaws.services.identitymanagement.model.AccessKeyMetadata; +import com.amazonaws.services.identitymanagement.model.CreateAccessKeyRequest; +import com.amazonaws.services.identitymanagement.model.CreateAccessKeyResult; +import com.amazonaws.services.identitymanagement.model.CreateUserRequest; +import com.amazonaws.services.identitymanagement.model.DeleteAccessKeyRequest; +import com.amazonaws.services.identitymanagement.model.EntityAlreadyExistsException; +import com.amazonaws.services.identitymanagement.model.ListAccessKeysRequest; +import com.amazonaws.services.identitymanagement.model.PutUserPolicyRequest; +import com.amazonaws.services.s3.AmazonS3; +import com.amazonaws.services.s3.model.AccessControlList; +import com.amazonaws.services.s3.model.BucketPolicy; +import com.amazonaws.services.s3.model.BucketVersioningConfiguration; +import com.amazonaws.services.s3.model.CreateBucketRequest; +import com.amazonaws.services.s3.model.DeleteBucketPolicyRequest; +import com.amazonaws.services.s3.model.GetBucketPolicyRequest; +import com.amazonaws.services.s3.model.SSEAlgorithm; +import com.amazonaws.services.s3.model.ServerSideEncryptionByDefault; +import com.amazonaws.services.s3.model.ServerSideEncryptionConfiguration; +import com.amazonaws.services.s3.model.ServerSideEncryptionRule; +import com.amazonaws.services.s3.model.SetBucketEncryptionRequest; +import com.amazonaws.services.s3.model.SetBucketVersioningConfigurationRequest; +import com.cloud.agent.api.to.BucketTO; +import com.cloud.agent.api.to.DataStoreTO; +import com.cloud.storage.BucketVO; +import com.cloud.storage.dao.BucketDao; +import com.cloud.user.Account; +import com.cloud.user.AccountDetailsDao; +import com.cloud.user.dao.AccountDao; +import com.cloud.utils.exception.CloudRuntimeException; + +/** + * SeaweedFS object store driver. + * + * Bucket operations use the AWS S3 SDK v1 (path-style access, endpoint-pinned). + * User/credential management uses the AWS IAM SDK v1, since SeaweedFS exposes a + * standard AWS IAM-compatible API. No proprietary admin client is needed. + * + * Modeled on CloudianHyperStoreObjectStoreDriverImpl, which uses the same + * S3 + IAM SDK pair. + */ +public class SeaweedFSObjectStoreDriverImpl extends BaseObjectStoreDriverImpl { + + @Inject + AccountDao _accountDao; + + @Inject + AccountDetailsDao _accountDetailsDao; + + @Inject + ObjectStoreDao _storeDao; + + @Inject + BucketDao _bucketDao; + + @Inject + ObjectStoreDetailsDao _storeDetailsDao; + + private static final String ACS_PREFIX = "acs"; + + @Override + public DataStoreTO getStoreTO(DataStore store) { + return null; + } + + /** + * Get the SeaweedFS IAM user name for the given CloudStack account. + * Uses the account UUID prefixed with "acs-" for namespacing. + */ + protected String getUserNameForAccount(Account account) { + return String.format("%s-%s", ACS_PREFIX, account.getUuid()); + } + + /** + * Create the IAM user for the CloudStack account if it doesn't exist, + * attach the restricted S3 policy, and ensure the account has a usable + * IAM access key persisted in its account details. + * + * <p>If a previously stored access key is still present in IAM, it is + * reused rather than rotated. A new key is only created when no stored + * key exists or the stored key is no longer found in IAM; in the latter + * case any unmanaged (leftover) keys for the user are deleted first to + * avoid hitting IAM access-key limits. This keeps bucket records that + * reference the stored credentials valid across repeated calls. + * + * @return true if the user exists or was created, false on failure. + */ + @Override + public boolean createUser(long accountId, long storeId) { + Account account = _accountDao.findById(accountId); + if (account == null) { + logger.error("Account {} not found", accountId); + return false; + } + String userName = getUserNameForAccount(account); + AmazonIdentityManagement iamClient = getIAMClient(storeId); + + // Create the IAM user if it doesn't already exist + try { + iamClient.createUser(new CreateUserRequest(userName)); + logger.info("Created IAM user {} for account {}", userName, account.getAccountName()); + } catch (EntityAlreadyExistsException e) { + logger.debug("IAM user {} already exists", userName); + } + + // Attach the restricted S3 policy (idempotent — overwrites if present) + iamClient.putUserPolicy(new PutUserPolicyRequest(userName, + "CloudStackPolicy", SeaweedFSObjectStoreUtil.IAM_USER_POLICY)); + + // Reuse the stored access key if it is still present in IAM; only + // create a new one when no usable key exists. + Map<String, String> details = _accountDetailsDao.findDetails(accountId); + String storedAccessKeyId = details.get(SeaweedFSObjectStoreUtil.KEY_ACCESS_KEY); + if (storedAccessKeyId != null && iamAccessKeyExists(iamClient, userName, storedAccessKeyId)) { + logger.debug("Reusing existing IAM access key {} for user {}", storedAccessKeyId, userName); + return true; + } + + // The stored key is missing or no longer in IAM. Clean up any + // unmanaged leftover keys before creating a replacement so we do not + // accumulate keys and hit IAM access-key limits. + deleteUnmanagedAccessKeys(iamClient, userName, storedAccessKeyId); + + CreateAccessKeyResult result = iamClient.createAccessKey( + new CreateAccessKeyRequest().withUserName(userName)); + AccessKey key = result.getAccessKey(); + + // Persist the credentials in the account details + details.put(SeaweedFSObjectStoreUtil.KEY_ACCESS_KEY, key.getAccessKeyId()); + details.put(SeaweedFSObjectStoreUtil.KEY_SECRET_KEY, key.getSecretAccessKey()); + _accountDetailsDao.persist(accountId, details); + + logger.info("Created IAM credentials {} for user {}", key.getAccessKeyId(), userName); + return true; + } + + /** + * Check whether the given access key id is still listed in IAM for the user. + */ + private boolean iamAccessKeyExists(AmazonIdentityManagement iamClient, String userName, String accessKeyId) { + try { + for (AccessKeyMetadata metadata : + iamClient.listAccessKeys(new ListAccessKeysRequest() + .withUserName(userName)).getAccessKeyMetadata()) { + if (accessKeyId.equals(metadata.getAccessKeyId())) { + return true; + } + } + } catch (AmazonClientException e) { + logger.warn("Failed to list IAM access keys for user {}: {}", userName, e.getMessage()); + } + return false; + } + + /** + * Delete access keys for the user other than the (optionally) preserved + * key id. Used to clean up unmanaged leftover keys before creating a + * replacement so repeated calls do not hit IAM access-key limits. + */ + private void deleteUnmanagedAccessKeys(AmazonIdentityManagement iamClient, String userName, String preserveAccessKeyId) { + try { + for (AccessKeyMetadata metadata : + iamClient.listAccessKeys(new ListAccessKeysRequest() + .withUserName(userName)).getAccessKeyMetadata()) { + String keyId = metadata.getAccessKeyId(); + if (preserveAccessKeyId != null && preserveAccessKeyId.equals(keyId)) { + continue; + } + DeleteAccessKeyRequest deleteReq = + new DeleteAccessKeyRequest() + .withUserName(userName) + .withAccessKeyId(keyId); + logger.info("Deleting un-managed IAM access key {} for user {}", keyId, userName); + iamClient.deleteAccessKey(deleteReq); + } + } catch (AmazonClientException e) { + logger.warn("Failed to clean up IAM access keys for user {}: {}", userName, e.getMessage()); + } + } + + @Override + public Bucket createBucket(Bucket bucket, boolean objectLock) { + String bucketName = bucket.getName(); + long storeId = bucket.getObjectStoreId(); + long accountId = bucket.getAccountId(); + + // Use the store's admin credentials to create the bucket + AmazonS3 s3client = getS3ClientByStoreId(storeId); + + // Check if the bucket already exists + try { + if (s3client.doesBucketExistV2(bucketName)) { + throw new CloudRuntimeException("Bucket already exists with name " + bucketName); + } + } catch (AmazonClientException e) { + throw new CloudRuntimeException(e); + } + + // Create the bucket + try { + CreateBucketRequest request = new CreateBucketRequest(bucketName); + if (objectLock) { + request.setObjectLockEnabledForBucket(true); + } + s3client.createBucket(request); + } catch (AmazonClientException e) { + logger.error("Create bucket failed", e); + throw new CloudRuntimeException(e); + } + + // Update the bucket record with the account's IAM credentials + Map<String, String> accountDetails = _accountDetailsDao.findDetails(accountId); + String accessKey = accountDetails.get(SeaweedFSObjectStoreUtil.KEY_ACCESS_KEY); + String secretKey = accountDetails.get(SeaweedFSObjectStoreUtil.KEY_SECRET_KEY); + if (accessKey == null || secretKey == null) { + logger.warn("No IAM credentials found for account {}. Bucket will be created without per-account credentials.", accountId); + } + + ObjectStoreVO store = _storeDao.findById(storeId); + String s3Url = getS3Url(storeId); + BucketVO bucketVO = _bucketDao.findById(bucket.getId()); + bucketVO.setAccessKey(accessKey); + bucketVO.setSecretKey(secretKey); + bucketVO.setBucketURL(s3Url + "/" + bucketName); + _bucketDao.update(bucket.getId(), bucketVO); + return bucket; + } + + @Override + public List<Bucket> listBuckets(long storeId) { + AmazonS3 s3client = getS3ClientByStoreId(storeId); + List<Bucket> bucketsList = new ArrayList<>(); + try { + List<com.amazonaws.services.s3.model.Bucket> s3Buckets = s3client.listBuckets(); + for (com.amazonaws.services.s3.model.Bucket s3Bucket : s3Buckets) { + Bucket bucket = new BucketObject(); + bucket.setName(s3Bucket.getName()); + bucketsList.add(bucket); + } + } catch (AmazonClientException e) { + throw new CloudRuntimeException(e); + } + return bucketsList; + } + + @Override + public boolean deleteBucket(BucketTO bucket, long storeId) { + AmazonS3 s3client = getS3ClientByStoreId(storeId); + try { + if (! s3client.doesBucketExistV2(bucket.getName())) { + throw new CloudRuntimeException("Bucket doesn't exist: " + bucket.getName()); + } + } catch (AmazonClientException e) { + throw new CloudRuntimeException(e); + } + try { + s3client.deleteBucket(bucket.getName()); + } catch (AmazonClientException e) { + throw new CloudRuntimeException(e); + } + return true; + } + + @Override + public AccessControlList getBucketAcl(BucketTO bucket, long storeId) { + AmazonS3 s3client = getS3ClientByStoreId(storeId); + try { + return s3client.getBucketAcl(bucket.getName()); + } catch (AmazonClientException e) { + throw new CloudRuntimeException(e); + } + } + + @Override + public void setBucketAcl(BucketTO bucket, AccessControlList acl, long storeId) { + AmazonS3 s3client = getS3ClientByStoreId(storeId); + try { + s3client.setBucketAcl(bucket.getName(), acl); + } catch (AmazonClientException e) { + throw new CloudRuntimeException(e); + } + } + + @Override + public void setBucketPolicy(BucketTO bucket, String policy, long storeId) { + if ("private".equalsIgnoreCase(policy)) { + deleteBucketPolicy(bucket, storeId); + return; + } + + StringBuilder sb = new StringBuilder(); + sb.append("{\n"); + sb.append(" \"Version\": \"2012-10-17\",\n"); + sb.append(" \"Statement\": [\n"); + sb.append(" {\n"); + sb.append(" \"Sid\": \"PublicReadForObjects\",\n"); + sb.append(" \"Effect\": \"Allow\",\n"); + sb.append(" \"Principal\": \"*\",\n"); + sb.append(" \"Action\": \"s3:GetObject\",\n"); + sb.append(" \"Resource\": \"arn:aws:s3:::%s/*\"\n"); + sb.append(" }\n"); + sb.append(" ]\n"); + sb.append("}\n"); + + String jsonPolicy = String.format(sb.toString(), bucket.getName()); + AmazonS3 s3client = getS3ClientByStoreId(storeId); + try { + s3client.setBucketPolicy(bucket.getName(), jsonPolicy); + } catch (AmazonClientException e) { + throw new CloudRuntimeException(e); + } + } + + @Override + public BucketPolicy getBucketPolicy(BucketTO bucket, long storeId) { + AmazonS3 s3client = getS3ClientByStoreId(storeId); + try { + return s3client.getBucketPolicy(new GetBucketPolicyRequest(bucket.getName())); + } catch (AmazonClientException e) { + throw new CloudRuntimeException(e); + } + } + + @Override + public void deleteBucketPolicy(BucketTO bucket, long storeId) { + AmazonS3 s3client = getS3ClientByStoreId(storeId); + try { + s3client.deleteBucketPolicy(new DeleteBucketPolicyRequest(bucket.getName())); + } catch (AmazonClientException e) { + throw new CloudRuntimeException(e); + } + } + + @Override + public boolean setBucketEncryption(BucketTO bucket, long storeId) { + AmazonS3 s3client = getS3ClientByStoreId(storeId); + try { + SetBucketEncryptionRequest eRequest = new SetBucketEncryptionRequest(); + eRequest.setBucketName(bucket.getName()); + + ServerSideEncryptionByDefault sseByDefault = new ServerSideEncryptionByDefault(); + sseByDefault.setSSEAlgorithm(SSEAlgorithm.AES256.toString()); + + ServerSideEncryptionRule sseRule = new ServerSideEncryptionRule(); + sseRule.setApplyServerSideEncryptionByDefault(sseByDefault); + + List<ServerSideEncryptionRule> sseRules = new ArrayList<>(); + sseRules.add(sseRule); + + ServerSideEncryptionConfiguration sseConf = new ServerSideEncryptionConfiguration(); + sseConf.setRules(sseRules); + + eRequest.setServerSideEncryptionConfiguration(sseConf); + s3client.setBucketEncryption(eRequest); + return true; + } catch (AmazonClientException e) { + throw new CloudRuntimeException(e); + } + } + + @Override + public boolean deleteBucketEncryption(BucketTO bucket, long storeId) { + AmazonS3 s3client = getS3ClientByStoreId(storeId); + try { + s3client.deleteBucketEncryption(bucket.getName()); + return true; + } catch (AmazonClientException e) { + throw new CloudRuntimeException(e); + } + } + + @Override + public boolean setBucketVersioning(BucketTO bucket, long storeId) { + AmazonS3 s3client = getS3ClientByStoreId(storeId); + try { + BucketVersioningConfiguration vConf = new BucketVersioningConfiguration(BucketVersioningConfiguration.ENABLED); + s3client.setBucketVersioningConfiguration( + new SetBucketVersioningConfigurationRequest(bucket.getName(), vConf)); + return true; + } catch (AmazonClientException e) { + throw new CloudRuntimeException(e); + } + } + + @Override + public boolean deleteBucketVersioning(BucketTO bucket, long storeId) { + AmazonS3 s3client = getS3ClientByStoreId(storeId); + try { + BucketVersioningConfiguration vConf = new BucketVersioningConfiguration(BucketVersioningConfiguration.SUSPENDED); + s3client.setBucketVersioningConfiguration( + new SetBucketVersioningConfigurationRequest(bucket.getName(), vConf)); + return true; + } catch (AmazonClientException e) { + throw new CloudRuntimeException(e); + } + } + + /** + * Set the bucket quota via the SeaweedFS admin REST API. + * + * SeaweedFS enforces bucket quota server-side by setting a read-only flag + * when usage exceeds the configured limit. The quota is configured via the + * SeaweedFS S3 extension endpoint PUT /{bucket}?seaweedfs-quota, + * authenticated via standard S3 SigV4 and authorized via the + * s3:PutBucketQuota IAM permission. + * + * @param size the GiB size to set the quota to. 0 disables quota. + * @throws CloudRuntimeException if the S3 endpoint or credentials are missing or the request fails. + */ + @Override + public void setBucketQuota(BucketTO bucket, long storeId, long size) { + String s3Url = getS3Url(storeId); + String accessKey = getAccessKey(storeId); + String secretKey = getSecretKey(storeId); + if (s3Url == null || s3Url.isEmpty() || accessKey == null || accessKey.isEmpty() || secretKey == null || secretKey.isEmpty()) { + throw new CloudRuntimeException("SeaweedFS S3 URL and credentials are required to set bucket quota. " + + "Configure 's3Url', 'accesskey', and 'secretkey' in the object store details."); + } + SeaweedFSObjectStoreUtil.setBucketQuotaViaS3Extension(s3Url, accessKey, secretKey, bucket.getName(), size, getS3ExtensionHttpClient()); + } + + /** + * Returns the HTTP client used to send SeaweedFS S3 extension requests + * (e.g. PUT /{bucket}?seaweedfs-quota). Exposed as a protected seam so + * tests can inject a mock client and assert the signed request without + * touching the network. + */ + protected java.net.http.HttpClient getS3ExtensionHttpClient() { + return java.net.http.HttpClient.newHttpClient(); + } + + @Override + public Map<String, Long> getAllBucketsUsage(long storeId) { + Map<String, Long> bucketUsage = new HashMap<>(); + List<BucketVO> bucketList = _bucketDao.listByObjectStoreId(storeId); + if (bucketList.isEmpty()) { + return bucketUsage; + } + + // List objects per bucket via S3 (no admin API needed). + // SeaweedFS also publishes per-bucket Prometheus metrics and an SOSAPI + // capacity.xml response; operators who need scalable usage reporting + // should consume those instead of S3 list-based aggregation. + AmazonS3 s3client = getS3ClientByStoreId(storeId); + for (BucketVO bucket : bucketList) { + try { + long size = 0L; + com.amazonaws.services.s3.model.ListObjectsV2Result result; + String continuationToken = null; + do { + com.amazonaws.services.s3.model.ListObjectsV2Request req = + new com.amazonaws.services.s3.model.ListObjectsV2Request() + .withBucketName(bucket.getName()) + .withMaxKeys(1000); + if (continuationToken != null) { + req.setContinuationToken(continuationToken); + } + result = s3client.listObjectsV2(req); + for (com.amazonaws.services.s3.model.S3ObjectSummary summary : result.getObjectSummaries()) { + size += summary.getSize(); + } + continuationToken = result.getNextContinuationToken(); + } while (result.isTruncated()); + bucketUsage.put(bucket.getName(), size); + } catch (AmazonClientException e) { + logger.warn("Failed to get usage for bucket {}: {}", bucket.getName(), e.getMessage()); + bucketUsage.put(bucket.getName(), 0L); + } Review Comment: Returning 0 for an S3 listing failure converts "unknown" into a valid usage value. `BucketApiServiceImpl` writes every returned size into `BucketVO.size`, so a transient endpoint or permission failure will erase the stored usage and under-report capacity. Omit the bucket or propagate the failure so the caller retains or skips the previous value. ########## plugins/storage/object/seaweedfs/src/test/java/org/apache/cloudstack/storage/datastore/driver/SeaweedFSObjectStoreDriverImplTest.java: ########## @@ -0,0 +1,549 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. +// SPDX-License-Identifier: Apache-2.0 +package org.apache.cloudstack.storage.datastore.driver; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.lenient; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.Flow; + +import java.io.ByteArrayOutputStream; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; + +import org.apache.cloudstack.storage.datastore.db.ObjectStoreDao; +import org.apache.cloudstack.storage.datastore.db.ObjectStoreDetailsDao; +import org.apache.cloudstack.storage.datastore.db.ObjectStoreVO; +import org.apache.cloudstack.storage.datastore.util.SeaweedFSObjectStoreUtil; +import org.apache.cloudstack.storage.object.Bucket; + +import com.amazonaws.services.identitymanagement.AmazonIdentityManagement; +import com.amazonaws.services.identitymanagement.model.AccessKey; +import com.amazonaws.services.identitymanagement.model.AccessKeyMetadata; +import com.amazonaws.services.identitymanagement.model.CreateAccessKeyRequest; +import com.amazonaws.services.identitymanagement.model.CreateAccessKeyResult; +import com.amazonaws.services.identitymanagement.model.CreateUserRequest; +import com.amazonaws.services.identitymanagement.model.DeleteAccessKeyRequest; +import com.amazonaws.services.identitymanagement.model.EntityAlreadyExistsException; +import com.amazonaws.services.identitymanagement.model.ListAccessKeysRequest; +import com.amazonaws.services.identitymanagement.model.ListAccessKeysResult; +import com.amazonaws.services.identitymanagement.model.PutUserPolicyRequest; +import com.amazonaws.services.s3.AmazonS3; +import com.amazonaws.services.s3.model.BucketVersioningConfiguration; +import com.amazonaws.services.s3.model.CreateBucketRequest; +import com.amazonaws.services.s3.model.ListObjectsV2Request; +import com.amazonaws.services.s3.model.ListObjectsV2Result; +import com.amazonaws.services.s3.model.S3ObjectSummary; +import com.amazonaws.services.s3.model.SetBucketVersioningConfigurationRequest; +import com.cloud.agent.api.to.BucketTO; +import com.cloud.storage.BucketVO; +import com.cloud.storage.dao.BucketDao; +import com.cloud.user.AccountDetailsDao; +import com.cloud.user.AccountVO; +import com.cloud.user.dao.AccountDao; +import com.cloud.utils.exception.CloudRuntimeException; + +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.ArgumentCaptor; +import org.mockito.ArgumentMatchers; +import org.mockito.Mock; +import org.mockito.MockitoAnnotations; +import org.mockito.Spy; +import org.mockito.junit.MockitoJUnitRunner; + +@RunWith(MockitoJUnitRunner.Silent.class) +public class SeaweedFSObjectStoreDriverImplTest { + + @Spy + SeaweedFSObjectStoreDriverImpl driver = new SeaweedFSObjectStoreDriverImpl(); + + @Mock + AmazonS3 s3Client; + @Mock + AmazonIdentityManagement iamClient; + @Mock + ObjectStoreDao objectStoreDao; + @Mock + ObjectStoreVO objectStoreVO; + @Mock + ObjectStoreDetailsDao objectStoreDetailsDao; + @Mock + AccountDao accountDao; + @Mock + BucketDao bucketDao; + @Mock + AccountDetailsDao accountDetailsDao; + @Mock + AccountVO account; + + BucketVO bucketVo; + Map<String, String> storeDetailsMap; + Map<String, String> accountDetailsMap; + + static long TEST_STORE_ID = 1010L; + static long TEST_ACCOUNT_ID = 2010L; + static long TEST_DOMAIN_ID = 3010L; + static String TEST_ACCESS_KEY = "test_access_key"; + static String TEST_SECRET_KEY = "test_secret_key"; + static String TEST_BUCKET_NAME = "testbucketname"; + static String TEST_S3_URL = "http://s3-endpoint";; + static String TEST_IAM_URL = "http://iam-endpoint";; + static String TEST_AK = "user_access_key"; + static String TEST_SK = "user_secret_key"; + static String TEST_BUCKET_URL = TEST_S3_URL + "/" + TEST_BUCKET_NAME; + static String TEST_ACCOUNT_UUID = "account-uuid-1234"; + + private AutoCloseable closeable; + + @Before + public void setUp() { + closeable = MockitoAnnotations.openMocks(this); + driver._storeDao = objectStoreDao; + driver._storeDetailsDao = objectStoreDetailsDao; + driver._accountDao = accountDao; + driver._bucketDao = bucketDao; + driver._accountDetailsDao = accountDetailsDao; + + lenient().when(objectStoreDao.findById(TEST_STORE_ID)).thenReturn(objectStoreVO); + lenient().when(objectStoreVO.getUrl()).thenReturn(TEST_S3_URL); + + storeDetailsMap = new HashMap<>(); + storeDetailsMap.put(SeaweedFSObjectStoreUtil.STORE_DETAILS_KEY_ACCESS_KEY, TEST_ACCESS_KEY); + storeDetailsMap.put(SeaweedFSObjectStoreUtil.STORE_DETAILS_KEY_SECRET_KEY, TEST_SECRET_KEY); + storeDetailsMap.put(SeaweedFSObjectStoreUtil.STORE_DETAILS_KEY_S3_URL, TEST_S3_URL); + storeDetailsMap.put(SeaweedFSObjectStoreUtil.STORE_DETAILS_KEY_IAM_URL, TEST_IAM_URL); + lenient().when(objectStoreDetailsDao.getDetails(TEST_STORE_ID)).thenReturn(storeDetailsMap); + + accountDetailsMap = new HashMap<>(); + accountDetailsMap.put(SeaweedFSObjectStoreUtil.KEY_ACCESS_KEY, TEST_AK); + accountDetailsMap.put(SeaweedFSObjectStoreUtil.KEY_SECRET_KEY, TEST_SK); + lenient().when(accountDetailsDao.findDetails(TEST_ACCOUNT_ID)).thenReturn(accountDetailsMap); + + bucketVo = new BucketVO(TEST_ACCOUNT_ID, TEST_DOMAIN_ID, TEST_STORE_ID, TEST_BUCKET_NAME, null, false, false, false, null); + } + + @After + public void tearDown() throws Exception { + closeable.close(); + } + + @Test + public void testGetStoreTO() { + assertNull(driver.getStoreTO(null)); + } + + @Test + public void testCreateBucket() throws Exception { + doReturn(s3Client).when(driver).getS3ClientByStoreId(TEST_STORE_ID); + when(s3Client.doesBucketExistV2(TEST_BUCKET_NAME)).thenReturn(false); + when(bucketDao.findById(anyLong())).thenReturn(bucketVo); + + Bucket result = driver.createBucket(bucketVo, false); + + assertEquals(TEST_BUCKET_NAME, result.getName()); + + ArgumentCaptor<BucketVO> captor = ArgumentCaptor.forClass(BucketVO.class); + verify(bucketDao, times(1)).update(any(), captor.capture()); + BucketVO updated = captor.getValue(); + assertEquals(TEST_AK, updated.getAccessKey()); + assertEquals(TEST_SK, updated.getSecretKey()); + assertEquals(TEST_BUCKET_URL, updated.getBucketURL()); + + verify(s3Client, times(1)).createBucket(any(CreateBucketRequest.class)); + } + + @Test + public void testCreateBucketAlreadyExists() throws Exception { + doReturn(s3Client).when(driver).getS3ClientByStoreId(TEST_STORE_ID); + when(s3Client.doesBucketExistV2(TEST_BUCKET_NAME)).thenReturn(true); + + assertThrows(CloudRuntimeException.class, () -> driver.createBucket(bucketVo, false)); + verify(s3Client, never()).createBucket(any(CreateBucketRequest.class)); + } + + @Test + public void testListBuckets() throws Exception { + doReturn(s3Client).when(driver).getS3ClientByStoreId(TEST_STORE_ID); + List<com.amazonaws.services.s3.model.Bucket> s3Buckets = new ArrayList<>(); + s3Buckets.add(new com.amazonaws.services.s3.model.Bucket("bucket1")); + s3Buckets.add(new com.amazonaws.services.s3.model.Bucket("bucket2")); + when(s3Client.listBuckets()).thenReturn(s3Buckets); + + List<Bucket> result = driver.listBuckets(TEST_STORE_ID); + + assertEquals(2, result.size()); + assertEquals("bucket1", result.get(0).getName()); + assertEquals("bucket2", result.get(1).getName()); + } + + @Test + public void testDeleteBucket() throws Exception { + doReturn(s3Client).when(driver).getS3ClientByStoreId(TEST_STORE_ID); + BucketTO bucketTO = mock(BucketTO.class); + when(bucketTO.getName()).thenReturn(TEST_BUCKET_NAME); + when(s3Client.doesBucketExistV2(TEST_BUCKET_NAME)).thenReturn(true); + + assertTrue(driver.deleteBucket(bucketTO, TEST_STORE_ID)); + verify(s3Client, times(1)).deleteBucket(TEST_BUCKET_NAME); + } + + @Test + public void testDeleteBucketNotFound() throws Exception { + doReturn(s3Client).when(driver).getS3ClientByStoreId(TEST_STORE_ID); + BucketTO bucketTO = mock(BucketTO.class); + when(bucketTO.getName()).thenReturn(TEST_BUCKET_NAME); + when(s3Client.doesBucketExistV2(TEST_BUCKET_NAME)).thenReturn(false); + + assertThrows(CloudRuntimeException.class, () -> driver.deleteBucket(bucketTO, TEST_STORE_ID)); + } + + @Test + public void testSetBucketVersioning() throws Exception { + doReturn(s3Client).when(driver).getS3ClientByStoreId(TEST_STORE_ID); + BucketTO bucketTO = mock(BucketTO.class); + when(bucketTO.getName()).thenReturn(TEST_BUCKET_NAME); + + assertTrue(driver.setBucketVersioning(bucketTO, TEST_STORE_ID)); + verify(s3Client, times(1)).setBucketVersioningConfiguration(any(SetBucketVersioningConfigurationRequest.class)); + } + + @Test + public void testDeleteBucketVersioning() throws Exception { + doReturn(s3Client).when(driver).getS3ClientByStoreId(TEST_STORE_ID); + BucketTO bucketTO = mock(BucketTO.class); + when(bucketTO.getName()).thenReturn(TEST_BUCKET_NAME); + + assertTrue(driver.deleteBucketVersioning(bucketTO, TEST_STORE_ID)); + ArgumentCaptor<SetBucketVersioningConfigurationRequest> captor = + ArgumentCaptor.forClass(SetBucketVersioningConfigurationRequest.class); + verify(s3Client, times(1)).setBucketVersioningConfiguration(captor.capture()); + assertEquals(BucketVersioningConfiguration.SUSPENDED, captor.getValue().getVersioningConfiguration().getStatus()); + } + + @Test + public void testSetBucketQuotaZero() throws Exception { + BucketTO bucketTO = mock(BucketTO.class); + when(bucketTO.getName()).thenReturn(TEST_BUCKET_NAME); + doReturn(TEST_S3_URL).when(driver).getS3Url(TEST_STORE_ID); + doReturn("access-key").when(driver).getAccessKey(TEST_STORE_ID); + doReturn("secret-key").when(driver).getSecretKey(TEST_STORE_ID); + + HttpClient mockHttpClient = mock(HttpClient.class); + HttpResponse<String> mockResponse = mock(HttpResponse.class); + when(mockResponse.statusCode()).thenReturn(200); + when(mockResponse.body()).thenReturn(""); + when(mockHttpClient.send(ArgumentMatchers.<HttpRequest>any(), + ArgumentMatchers.<HttpResponse.BodyHandler<String>>any())).thenReturn(mockResponse); + doReturn(mockHttpClient).when(driver).getS3ExtensionHttpClient(); + + driver.setBucketQuota(bucketTO, TEST_STORE_ID, 0); + + ArgumentCaptor<HttpRequest> reqCaptor = ArgumentCaptor.forClass(HttpRequest.class); + verify(mockHttpClient, times(1)).send(reqCaptor.capture(), + ArgumentMatchers.<HttpResponse.BodyHandler<String>>any()); + HttpRequest sent = reqCaptor.getValue(); + assertEquals("PUT", sent.method()); + assertEquals("/" + TEST_BUCKET_NAME, sent.uri().getPath()); + assertTrue("query must carry the seaweedfs-quota subresource", + sent.uri().getQuery().contains("seaweedfs-quota")); + assertNotNull("request must be SigV4-signed", sent.headers().firstValue("Authorization")); + assertEquals("{\"quota_size\":0,\"quota_unit\":\"B\",\"quota_enabled\":false}", extractBody(sent)); Review Comment: These tests only check that an `Authorization` header exists; they do not verify the SigV4 canonical query, payload hash, or signed headers against a known signature or S3-compatible verifier. A signing mismatch would therefore pass the suite and make every quota operation fail against SeaweedFS. Add a deterministic signature-verification test for the actual request. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
