An experimental feature to support different credentials for per bucket

pom.xml

@@ -452,5 +452,17 @@
       <artifactId>javax.annotation-api</artifactId>
       <version>1.3.2</version>
     </dependency>
+    <dependency>
+      <groupId>org.xerial</groupId>
+      <artifactId>sqlite-jdbc</artifactId>
+      <version>3.36.0.1</version>
+    </dependency>
+    <!-- https://mvnrepository.com/artifact/org.yaml/snakeyaml -->
+    <dependency>
+      <groupId>org.yaml</groupId>
+      <artifactId>snakeyaml</artifactId>
+      <version>1.29</version>
+    </dependency>
+
   </dependencies>
 </project>

src/main/java/org/gaul/s3proxy/Main.java

@@ -16,30 +16,17 @@
 
 package org.gaul.s3proxy;
 
-import java.io.Console;
-import java.io.File;
-import java.io.FileInputStream;
-import java.io.IOException;
-import java.io.InputStream;
-import java.io.PrintStream;
-import java.nio.charset.StandardCharsets;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-import java.util.Properties;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ThreadFactory;
-import java.util.concurrent.TimeUnit;
-
 import com.google.common.base.Strings;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Maps;
 import com.google.common.io.Files;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import com.google.inject.Module;
-
+import org.gaul.s3proxy.extend.AbStractAKSKManager;
+import org.gaul.s3proxy.extend.AkSkPair;
+import org.gaul.s3proxy.extend.SqliteAKSKManager;
+import org.gaul.s3proxy.extend.YamlAKSKManager;
 import org.jclouds.Constants;
 import org.jclouds.ContextBuilder;
 import org.jclouds.JcloudsVersion;
@@ -56,8 +43,17 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.*;
+import java.nio.charset.StandardCharsets;
+import java.util.*;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+
 public final class Main {
     private static final Logger logger = LoggerFactory.getLogger(Main.class);
+
     private Main() {
         throw new AssertionError("intentionally not implemented");
     }
@@ -69,8 +65,27 @@ private static final class Options {
 
         @Option(name = "--version", usage = "display version")
         private boolean version;
+
+        @Option(name = "--enable_bucket_ak_sk",
+                usage = "experimental feature to provide different ak/sk for per bucket"
+        )
+        private boolean enableBucketAkSk;
+
+        @Option(name = "--config_store",
+                usage = "yaml,sqlite.. etc."
+        )
+        private String configStoreType = "yaml";
+
+
+        @Option(name = "--config_path",
+                usage = "config file path. for sqlite is db file path, and for yaml is local yml file absolute path."
+        )
+        private String configPath = "";
+        ;
+
     }
 
+
     public static void main(String[] args) throws Exception {
         Console console = System.console();
         if (console == null) {
@@ -100,8 +115,26 @@ public static void main(String[] args) throws Exception {
                 .build();
         ExecutorService executorService = DynamicExecutors.newScalingThreadPool(
                 1, 20, 60 * 1000, factory);
-        ImmutableMap.Builder<String, Map.Entry<String, BlobStore>> locators =
-                ImmutableMap.builder();
+        Map<String, Map.Entry<String, BlobStore>> locators =
+                new LinkedHashMap<>();
+        // load bucket access db
+        AbStractAKSKManager accessSecretManager = null;
+        Map<String, AkSkPair> accessCache = null;
+        if (options.enableBucketAkSk) {
+            switch (options.configStoreType) {
+                case "sqlite":
+                    accessSecretManager = new SqliteAKSKManager(options.configPath);
+                    break;
+                case "yaml":
+                    accessSecretManager = new YamlAKSKManager(options.configPath);
+                    break;
+                default:
+                    accessSecretManager = new YamlAKSKManager(options.configPath);
+            }
+            accessCache = accessSecretManager.getBucketAkSkList();
+
+        }
+
         for (File propertiesFile : options.propertiesFiles) {
             Properties properties = new Properties();
             try (InputStream is = new FileInputStream(propertiesFile)) {
@@ -111,6 +144,7 @@ public static void main(String[] args) throws Exception {
 
             BlobStore blobStore = createBlobStore(properties, executorService);
 
+
             blobStore = parseMiddlewareProperties(blobStore, executorService,
                     properties);
 
@@ -122,8 +156,20 @@ public static void main(String[] args) throws Exception {
                         S3ProxyConstants.PROPERTY_IDENTITY);
                 String localCredential = properties.getProperty(
                         S3ProxyConstants.PROPERTY_CREDENTIAL);
+                Map<String, Map.Entry<String, BlobStore>> save = new HashMap<>();
                 locators.put(localIdentity, Maps.immutableEntry(
                         localCredential, blobStore));
+                // register blobStore to access manager
+                if (accessSecretManager != null) {
+                    accessSecretManager.registerBlobStore(blobStore);
+                }
+                if (accessCache != null) {
+                    for (Map.Entry<String, AkSkPair> entry : accessCache.entrySet()) {
+                        // add sqlite access key to locators
+                        locators.put(entry.getValue().getAccess_key(), Maps.immutableEntry(
+                                entry.getValue().getSecret_key(), blobStore));
+                    }
+                }
             }
 
             S3Proxy.Builder s3ProxyBuilder2 = S3Proxy.Builder
@@ -139,6 +185,7 @@ public static void main(String[] args) throws Exception {
             s3ProxyBuilder = s3ProxyBuilder2;
         }
 
+
         S3Proxy s3Proxy;
         try {
             s3Proxy = s3ProxyBuilder.build();
@@ -149,8 +196,12 @@ public static void main(String[] args) throws Exception {
         }
 
         final Map<String, Map.Entry<String, BlobStore>> locator =
-                locators.build();
+                locators;
+        if (accessSecretManager != null) {
+            accessSecretManager.setLocator(locator);
+        }
         if (!locator.isEmpty()) {
+            AbStractAKSKManager finalAbStractAKSKManager = accessSecretManager;
             s3Proxy.setBlobStoreLocator(new BlobStoreLocator() {
                 @Override
                 public Map.Entry<String, BlobStore> locateBlobStore(
@@ -161,10 +212,22 @@ public Map.Entry<String, BlobStore> locateBlobStore(
                                     .getValue();
                         }
                         throw new IllegalArgumentException(
-                            "cannot use anonymous access with multiple" +
-                            " backends");
+                                "cannot use anonymous access with multiple" +
+                                        " backends");
+                    }
+                    // verify key
+                    Map.Entry<String, BlobStore> provider = locator.get(identity);
+                    if (finalAbStractAKSKManager != null) {
+                        String bucket = finalAbStractAKSKManager.getBucketFromAccessKey(identity);
+                        if (!bucket.equalsIgnoreCase("")) {
+                            // if bucket and access_key not match. [1:1 in db] return null
+                            if (!container.equalsIgnoreCase(bucket)) {
+                                return null;
+                            }
+                        }
                     }
-                    return locator.get(identity);
+
+                    return provider;
                 }
             });
         }
@@ -178,7 +241,7 @@ public Map.Entry<String, BlobStore> locateBlobStore(
     }
 
     private static BlobStore parseMiddlewareProperties(BlobStore blobStore,
-            ExecutorService executorService, Properties properties)
+                                                       ExecutorService executorService, Properties properties)
             throws IOException {
         Properties altProperties = new Properties();
         for (Map.Entry<Object, Object> entry : properties.entrySet()) {
@@ -266,7 +329,7 @@ public void write(byte[] buf, int off, int len) {
     }
 
     private static BlobStore createBlobStore(Properties properties,
-            ExecutorService executorService) throws IOException {
+                                             ExecutorService executorService) throws IOException {
         String provider = properties.getProperty(Constants.PROPERTY_PROVIDER);
         String identity = properties.getProperty(Constants.PROPERTY_IDENTITY);
         String credential = properties.getProperty(
@@ -279,7 +342,7 @@ private static BlobStore createBlobStore(Properties properties,
         if (provider == null) {
             System.err.println(
                     "Properties file must contain: " +
-                    Constants.PROPERTY_PROVIDER);
+                            Constants.PROPERTY_PROVIDER);
             System.exit(1);
         }
 
@@ -299,8 +362,8 @@ private static BlobStore createBlobStore(Properties properties,
         if (identity == null || credential == null) {
             System.err.println(
                     "Properties file must contain: " +
-                    Constants.PROPERTY_IDENTITY + " and " +
-                    Constants.PROPERTY_CREDENTIAL);
+                            Constants.PROPERTY_IDENTITY + " and " +
+                            Constants.PROPERTY_CREDENTIAL);
             System.exit(1);
         }
 

src/main/java/org/gaul/s3proxy/S3ProxyHandler.java

@@ -608,6 +608,7 @@ public final void doHandle(HttpServletRequest baseRequest,
 
             if (!constantTimeEquals(expectedSignature,
                     authHeader.getSignature())) {
+                // verify access key check
                 throw new S3Exception(S3ErrorCode.SIGNATURE_DOES_NOT_MATCH);
             }
         }

src/main/java/org/gaul/s3proxy/extend/AbStractAKSKManager.java

@@ -0,0 +1,93 @@
+package org.gaul.s3proxy.extend;
+
+import org.jclouds.blobstore.BlobStore;
+import sun.reflect.generics.reflectiveObjects.NotImplementedException;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * @author yangyanbo
+ */
+public class AbStractAKSKManager implements AccessSecretManager {
+
+    public String cacheType = "memory";
+
+    static public List<BlobStore> blobStores = new ArrayList<>();
+
+    public CacheManager cacheManager;
+    Map<String, Map.Entry<String, BlobStore>> locator;
+
+    /**
+     * static variable save last config
+     */
+    public static Map<String, AkSkPair> save;
+
+    public AbStractAKSKManager() {
+        cacheManager = createCacheManager();
+    }
+
+    public Map<String, Map.Entry<String, BlobStore>> getLocator() {
+        return locator;
+    }
+
+    public void setLocator(Map<String, Map.Entry<String, BlobStore>> locator) {
+        this.locator = locator;
+    }
+
+    @Override
+    public CacheManager createCacheManager() {
+        switch (cacheType) {
+            case "memory":
+                cacheManager = new MemoryCache();
+                break;
+            default:
+                cacheManager = new MemoryCache();
+        }
+        return cacheManager;
+    }
+
+    @Override
+    public void createAKSKForBucket(String bucket, String access_key, String secret_key) {
+        throw new NotImplementedException();
+    }
+
+    @Override
+    public Map<String, AkSkPair> loads2Cache() {
+        throw new NotImplementedException();
+
+    }
+
+    @Override
+    public Map<String, AkSkPair> getBucketAkSkList() throws Exception {
+        throw new NotImplementedException();
+    }
+
+    @Override
+    public String getBucketFromAccessKey(String ak) {
+        throw new NotImplementedException();
+    }
+
+    @Override
+    public void registerBlobStore(BlobStore blobStore) {
+        blobStores.add(blobStore);
+    }
+
+    @Override
+    public List<BlobStore> listBlobStores() {
+        return blobStores;
+    }
+
+    public CacheManager getCacheManager() {
+        return cacheManager;
+    }
+
+    public static Map<String, AkSkPair> getSave() {
+        return save;
+    }
+
+    public static void setSave(Map<String, AkSkPair> save) {
+        AbStractAKSKManager.save = save;
+    }
+}

src/main/java/org/gaul/s3proxy/extend/AccessSecretManager.java

@@ -0,0 +1,29 @@
+package org.gaul.s3proxy.extend;
+
+import org.jclouds.blobstore.BlobStore;
+
+import java.util.List;
+import java.util.Map;
+
+/**
+ * @author yangyanbo
+ */
+public interface AccessSecretManager {
+
+
+    CacheManager createCacheManager() throws Exception;
+
+    void createAKSKForBucket(String bucket, String access_key, String secret_key) throws Exception;
+
+    Map<String, AkSkPair> loads2Cache() throws Exception;
+
+    Map<String, AkSkPair> getBucketAkSkList() throws Exception;
+
+    String getBucketFromAccessKey(String ak);
+
+    void registerBlobStore(BlobStore blobStore);
+
+    List<BlobStore> listBlobStores();
+
+
+}