Skip to content

Storage SDK: streaming, security and fixes #69

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
maheshkumargangula wants to merge 1 commit into
base: v1.4.10
Choose a base branch
from
Open

Storage SDK: streaming, security and fixes #69

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
48 changes: 32 additions & 16 deletions cloud-storage-sdk-api/src/main/java/org/sunbird/cloud/storage/AbstractStorageService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -229,23 +229,25 @@ public void download(String container, String objectKey, String localPath, boole
if (isDirectory) {
List<String> objects = listObjectKeys(container, objectKey);
for (String obj : objects) {
InputStream stream = getObjectStream(container, obj);
String relativePath = obj.startsWith(objectKey) ? obj.substring(objectKey.length()) : obj;
File targetFile = new File(localPath, relativePath);
File parentDir = targetFile.getParentFile();
if (parentDir != null && !parentDir.exists()) {
parentDir.mkdirs();
try (InputStream stream = getObjectStream(container, obj)) {
String relativePath = obj.startsWith(objectKey) ? obj.substring(objectKey.length()) : obj;
File targetFile = new File(localPath, relativePath);
File parentDir = targetFile.getParentFile();
if (parentDir != null && !parentDir.exists()) {
parentDir.mkdirs();
}
String fileName = targetFile.getName();
String dirPath = parentDir != null ? parentDir.getAbsolutePath() + "/" : localPath;
FileUtil.copyStream(stream, dirPath, fileName);
}
String fileName = targetFile.getName();
String dirPath = parentDir != null ? parentDir.getAbsolutePath() + "/" : localPath;
FileUtil.copyStream(stream, dirPath, fileName);
}
} else {
InputStream stream = getObjectStream(container, objectKey);
String fileName = objectKey.contains("/")
? objectKey.substring(objectKey.lastIndexOf('/') + 1)
: objectKey;
FileUtil.copyStream(stream, localPath, fileName);
try (InputStream stream = getObjectStream(container, objectKey)) {
String fileName = objectKey.contains("/")
? objectKey.substring(objectKey.lastIndexOf('/') + 1)
: objectKey;
FileUtil.copyStream(stream, localPath, fileName);
}
}
} catch (StorageServiceException e) {
throw e;
Expand Down Expand Up @@ -285,6 +287,11 @@ public Blob getObject(String container, String objectKey, boolean withPayload) {
BlobDetail detail = getObjectDetail(container, objectKey);
byte[] payload = null;
if (withPayload) {
if (detail.contentLength > 100 * 1024 * 1024) { // 100MB limit
throw new StorageServiceException(
"Object too large for in-memory payload (" + detail.contentLength + " bytes). " +
"Use getObjectStream() for large objects.");
}
try (InputStream is = getObjectStream(container, objectKey)) {
payload = is.readAllBytes();
}
Expand Down Expand Up @@ -379,9 +386,9 @@ public void copyObjects(String fromContainer, String fromKey,

@Override
public void extractArchive(String container, String objectKey, String toKey) {
String localExtractPath = System.getProperty("local_extract_path",
System.getenv().getOrDefault("local_extract_path", "/tmp/extract"));
try {
String localExtractPath = System.getProperty("local_extract_path",
System.getenv().getOrDefault("local_extract_path", "/tmp/extract"));
download(container, objectKey, localExtractPath, false);
String archiveName = objectKey.contains("/")
? objectKey.substring(objectKey.lastIndexOf('/') + 1)
Expand All @@ -396,6 +403,15 @@ public void extractArchive(String container, String objectKey, String toKey) {
throw e;
} catch (Exception e) {
throw new StorageServiceException("Extract archive failed: " + e.getMessage(), e);
} finally {
try {
java.nio.file.Files.walk(java.nio.file.Paths.get(localExtractPath))
.sorted(java.util.Comparator.reverseOrder())
.map(java.nio.file.Path::toFile)
.forEach(File::delete);
} catch (IOException cleanupEx) {
logger.warn("Failed to clean up extraction directory: {}", localExtractPath, cleanupEx);
}
}
}

Expand Down
21 changes: 17 additions & 4 deletions cloud-storage-sdk-api/src/main/java/org/sunbird/cloud/storage/model/Blob.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,9 +19,9 @@ public Blob(String key, long contentLength, Date lastModified,
Map<String, Object> metadata, byte[] payload) {
this.key = key;
this.contentLength = contentLength;
this.lastModified = lastModified;
this.lastModified = lastModified != null ? new Date(lastModified.getTime()) : null;
this.metadata = metadata != null ? Collections.unmodifiableMap(metadata) : Collections.emptyMap();
this.payload = payload;
this.payload = payload != null ? payload.clone() : null;
}

public Blob(String key, long contentLength, Date lastModified, Map<String, Object> metadata) {
Expand All @@ -37,15 +37,28 @@ public long getContentLength() {
}

public Date getLastModified() {
return lastModified;
return lastModified != null ? new Date(lastModified.getTime()) : null;
}

public Map<String, Object> getMetadata() {
return metadata;
}

public byte[] getPayload() {
return payload;
return payload != null ? payload.clone() : null;
}

@Override
public boolean equals(Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
Blob blob = (Blob) o;
return java.util.Objects.equals(key, blob.key);
}

@Override
public int hashCode() {
return java.util.Objects.hash(key);
}

@Override
Expand Down
13 changes: 13 additions & 0 deletions cloud-storage-sdk-api/src/main/java/org/sunbird/cloud/storage/model/DeleteTarget.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,19 @@ public boolean isDirectory() {
return directory;
}

@Override
public boolean equals(Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
DeleteTarget that = (DeleteTarget) o;
return directory == that.directory && java.util.Objects.equals(objectKey, that.objectKey);
}

@Override
public int hashCode() {
return java.util.Objects.hash(objectKey, directory);
}

@Override
public String toString() {
return "DeleteTarget{key='" + objectKey + "', directory=" + directory + "}";
Expand Down
5 changes: 5 additions & 0 deletions cloud-storage-sdk-api/src/main/java/org/sunbird/cloud/storage/util/FileUtil.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -85,6 +85,11 @@ public static void unZip(String zipFile, String outputFolder) throws IOException
}

File newFile = new File(outputFolder + File.separator + entryName);
String canonicalDestDir = folder.getCanonicalPath();
String canonicalNewFile = newFile.getCanonicalPath();
if (!canonicalNewFile.startsWith(canonicalDestDir + File.separator)) {
throw new IOException("Zip entry is outside of the target dir: " + entryName);
}
File parent = newFile.getParentFile();
if (parent != null && !parent.exists()) {
parent.mkdirs();
Expand Down
3 changes: 2 additions & 1 deletion ...torage-sdk-aws/src/main/java/org/sunbird/cloud/storage/service/aws/AwsStorageService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -102,6 +102,7 @@ public AwsStorageService(StorageConfig config) {
.pathStyleAccessEnabled(true)
.build();
clientBuilder.serviceConfiguration(s3Config);
presignerBuilder.serviceConfiguration(s3Config);
}

this.s3Client = clientBuilder.build();
Expand Down Expand Up @@ -167,7 +168,7 @@ protected String putObject(String container, String objectKey, byte[] content) {
PutObjectRequest request = PutObjectRequest.builder()
.bucket(container)
.key(objectKey)
.contentType("application/octet-stream")
.contentType(tika.detect(new java.io.ByteArrayInputStream(content), objectKey))
.build();
s3Client.putObject(request, RequestBody.fromBytes(content));
return getObjectUri(container, objectKey);
Expand Down
5 changes: 3 additions & 2 deletions ...-sdk-azure/src/main/java/org/sunbird/cloud/storage/service/azure/AzureStorageService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -239,7 +239,7 @@ protected void copyObject(String fromContainer, String fromKey,
try {
BlobClient sourceBlobClient = getBlobClient(fromContainer, fromKey);
BlobClient destBlobClient = getBlobClient(toContainer, toKey);
destBlobClient.copyFromUrl(sourceBlobClient.getBlobUrl());
destBlobClient.copyFromUrl(decodeBlobUrl(sourceBlobClient.getBlobUrl()));
} catch (Exception e) {
throw new StorageServiceException(
"Failed to copy object from " + fromContainer + "/" + fromKey
Expand Down Expand Up @@ -307,7 +307,8 @@ protected String getHdfsPrefix(String container) {

@Override
public void close() {
// BlobServiceClient does not implement Closeable; no resource cleanup needed
// BlobServiceClient manages its own HTTP connection pool internally
// and does not implement Closeable. No explicit cleanup is needed.
logger.info("AzureStorageService closed");
}
}
29 changes: 22 additions & 7 deletions ...torage-sdk-gcp/src/main/java/org/sunbird/cloud/storage/service/gcp/GcpStorageService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@ public class GcpStorageService extends AbstractStorageService {

private final StorageConfig config;
private final Storage storage;
private Storage signingStorage;

public GcpStorageService(StorageConfig config) {
this.config = config;
Expand All @@ -52,6 +53,7 @@ private Storage buildStorage(StorageConfig config) {
// For access key auth, storageKey is the project ID and storageSecret
// is a service account JSON key. Parse as JSON credentials.
if (config.getStorageSecret() != null && !config.getStorageSecret().isEmpty()) {
logger.warn("Using service account JSON from config. For production, prefer GOOGLE_APPLICATION_CREDENTIALS env var.");
GoogleCredentials credentials = GoogleCredentials.fromStream(
new ByteArrayInputStream(config.getStorageSecret().getBytes()));
builder.setCredentials(credentials);
Expand Down Expand Up @@ -99,7 +101,16 @@ protected String putObject(String container, String objectKey, File file) {
BlobInfo blobInfo = BlobInfo.newBuilder(blobId)
.setContentType(contentType)
.build();
storage.create(blobInfo, Files.readAllBytes(file.toPath()));
try (java.nio.channels.WritableByteChannel writer = storage.writer(blobInfo);
java.io.FileInputStream fis = new java.io.FileInputStream(file);
java.nio.channels.ReadableByteChannel reader = java.nio.channels.Channels.newChannel(fis)) {
java.nio.ByteBuffer buffer = java.nio.ByteBuffer.allocate(64 * 1024);
while (reader.read(buffer) > 0) {
buffer.flip();
writer.write(buffer);
buffer.clear();
}
}
return GCS_HOST + container + "/" + objectKey;
} catch (IOException e) {
throw new StorageServiceException(
Expand All @@ -118,8 +129,8 @@ protected String putObject(String container, String objectKey, byte[] content) {
@Override
protected InputStream getObjectStream(String container, String objectKey) {
try {
byte[] content = storage.readAllBytes(BlobId.of(container, objectKey));
return new ByteArrayInputStream(content);
com.google.cloud.ReadChannel reader = storage.reader(BlobId.of(container, objectKey));
return java.nio.channels.Channels.newInputStream(reader);
} catch (Exception e) {
throw new StorageServiceException(
"Failed to get object stream: " + objectKey + " - " + e.getMessage(), e);
Expand Down Expand Up @@ -270,30 +281,34 @@ protected String generateSignedPutUrl(String container, String objectKey,
*/
private Storage resolveSigningStorage(Map<String, String> additionalParams) {
if (additionalParams == null || additionalParams.isEmpty()) {
return storage;
return this.storage;
}
String clientId = additionalParams.get("clientId");
String clientEmail = additionalParams.get("clientEmail");
String privateKeyPkcs8 = additionalParams.get("privateKeyPkcs8");
String privateKeyId = additionalParams.get("privateKeyIds");
String privateKeyId = additionalParams.get("privateKeyId");
String projectId = additionalParams.get("projectId");

if (clientEmail != null && privateKeyPkcs8 != null && privateKeyId != null) {
if (this.signingStorage != null) {
return this.signingStorage;
}
try {
ServiceAccountCredentials credentials = ServiceAccountCredentials.fromPkcs8(
clientId, clientEmail, privateKeyPkcs8, privateKeyId,
new ArrayList<>());
return StorageOptions.newBuilder()
this.signingStorage = StorageOptions.newBuilder()
.setProjectId(projectId)
.setCredentials(credentials)
.build()
.getService();
return this.signingStorage;
} catch (IOException e) {
throw new StorageServiceException(
"Failed to create signing credentials from additionalParams: " + e.getMessage(), e);
}
}
return storage;
return this.storage;
}

@Override
Expand Down
46 changes: 18 additions & 28 deletions ...torage-sdk-oci/src/main/java/org/sunbird/cloud/storage/service/oci/OciStorageService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,8 +20,8 @@

import java.io.ByteArrayInputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.InputStream;
import java.nio.file.Files;
import java.security.MessageDigest;
import java.util.ArrayList;
import java.util.Base64;
Expand Down Expand Up @@ -54,12 +54,17 @@ public OciStorageService(StorageConfig config) {
clientBuilder.region(Region.fromRegionId(config.getRegion()));
}

this.objectStorageClient = clientBuilder.build(authProvider);

// Retrieve namespace
GetNamespaceResponse namespaceResponse = objectStorageClient.getNamespace(
GetNamespaceRequest.builder().build());
this.namespace = namespaceResponse.getValue();
ObjectStorageClient tempClient = clientBuilder.build(authProvider);
try {
GetNamespaceResponse namespaceResponse = tempClient.getNamespace(
GetNamespaceRequest.builder().build());
this.namespace = namespaceResponse.getValue();
this.objectStorageClient = tempClient;
} catch (Exception e) {
tempClient.close();
throw new StorageServiceException(
"Failed to initialize OCI storage - could not get namespace: " + e.getMessage(), e);
}

logger.info("Initialized OciStorageService, namespace={}, authType={}",
namespace, config.getAuthType());
Expand Down Expand Up @@ -126,18 +131,19 @@ protected void ensureContainerExists(String container) {

@Override
protected String putObject(String container, String objectKey, File file) {
try (FileInputStream fis = new FileInputStream(file)) {
try {
String contentType = tika.detect(file);
String md5 = computeMd5Base64(file);
byte[] data = Files.readAllBytes(file.toPath());
String md5 = computeMd5Base64(data);

PutObjectRequest request = PutObjectRequest.builder()
.namespaceName(namespace)
.bucketName(container)
.objectName(objectKey)
.contentType(contentType)
.contentLength(file.length())
.contentLength((long) data.length)
.contentMD5(md5)
.putObjectBody(fis)
.putObjectBody(new ByteArrayInputStream(data))
.build();
objectStorageClient.putObject(request);
return buildObjectUri(container, objectKey);
Expand Down Expand Up @@ -305,7 +311,7 @@ private String generatePreauthenticatedRequest(String container, String objectKe
int ttlSeconds,
CreatePreauthenticatedRequestDetails.AccessType accessType) {
try {
Date expiry = new Date(System.currentTimeMillis() + (long) ttlSeconds * 1000);
Date expiry = Date.from(java.time.Instant.now().plusSeconds(ttlSeconds));

CreatePreauthenticatedRequestDetails details = CreatePreauthenticatedRequestDetails.builder()
.name("par-" + objectKey + "-" + System.currentTimeMillis())
Expand Down Expand Up @@ -358,22 +364,6 @@ private String buildObjectUri(String container, String objectKey) {
namespace + "/b/" + container + "/o/" + objectKey;
}

private String computeMd5Base64(File file) {
try {
MessageDigest md = MessageDigest.getInstance("MD5");
try (FileInputStream fis = new FileInputStream(file)) {
byte[] buffer = new byte[8192];
int read;
while ((read = fis.read(buffer)) != -1) {
md.update(buffer, 0, read);
}
}
return Base64.getEncoder().encodeToString(md.digest());
} catch (Exception e) {
throw new StorageServiceException("Failed to compute MD5: " + e.getMessage(), e);
}
}

private String computeMd5Base64(byte[] content) {
try {
MessageDigest md = MessageDigest.getInstance("MD5");
Expand Down
Loading