Add support for pluggable Authorizer, default to NoopAuthorizer.

G8XSU · G8XSU · commit 424a3859795b · 2024-07-30T15:04:01.000-07:00
Different authorizers can act upon headers to return userToken
or throw AuthException if auth fails. If hosting providers,
want to support alternative auths or add additional checks in auth,
they can just implement this interface.
diff --git a/app/src/main/java/org/vss/KVStore.java b/app/src/main/java/org/vss/KVStore.java
@@ -4,11 +4,11 @@ public interface KVStore {
 
   String GLOBAL_VERSION_KEY = "vss_global_version";
 
-  GetObjectResponse get(GetObjectRequest request);
+  GetObjectResponse get(String userToken, GetObjectRequest request);
 
-  PutObjectResponse put(PutObjectRequest request);
+  PutObjectResponse put(String userToken, PutObjectRequest request);
 
-  DeleteObjectResponse delete(DeleteObjectRequest request);
+  DeleteObjectResponse delete(String userToken, DeleteObjectRequest request);
 
-  ListKeyVersionsResponse listKeyVersions(ListKeyVersionsRequest request);
+  ListKeyVersionsResponse listKeyVersions(String userToken, ListKeyVersionsRequest request);
 }
diff --git a/app/src/main/java/org/vss/api/AbstractVssApi.java b/app/src/main/java/org/vss/api/AbstractVssApi.java
@@ -7,16 +7,19 @@
 import org.vss.ErrorCode;
 import org.vss.ErrorResponse;
 import org.vss.KVStore;
+import org.vss.auth.Authorizer;
 import org.vss.exception.AuthException;
 import org.vss.exception.ConflictException;
 import org.vss.exception.NoSuchKeyException;
 
 public abstract class AbstractVssApi {
   final KVStore kvStore;
+  final Authorizer authorizer;
 
   @Inject
-  public AbstractVssApi(KVStore kvStore) {
+  public AbstractVssApi(KVStore kvStore, Authorizer authorizer) {
     this.kvStore = kvStore;
+    this.authorizer = authorizer;
   }
 
   Response toResponse(GeneratedMessageV3 protoResponse) {
diff --git a/app/src/main/java/org/vss/api/DeleteObjectApi.java b/app/src/main/java/org/vss/api/DeleteObjectApi.java
@@ -4,27 +4,32 @@
 import jakarta.ws.rs.POST;
 import jakarta.ws.rs.Path;
 import jakarta.ws.rs.Produces;
+import jakarta.ws.rs.core.Context;
+import jakarta.ws.rs.core.HttpHeaders;
 import jakarta.ws.rs.core.MediaType;
 import jakarta.ws.rs.core.Response;
 import lombok.extern.slf4j.Slf4j;
 import org.vss.DeleteObjectRequest;
 import org.vss.DeleteObjectResponse;
 import org.vss.KVStore;
+import org.vss.auth.AuthResponse;
+import org.vss.auth.Authorizer;
 
 @Path(VssApiEndpoint.DELETE_OBJECT)
 @Slf4j
 public class DeleteObjectApi extends AbstractVssApi {
   @Inject
-  public DeleteObjectApi(KVStore kvstore) {
-    super(kvstore);
+  public DeleteObjectApi(KVStore kvstore, Authorizer authorizer) {
+    super(kvstore, authorizer);
   }
 
   @POST
   @Produces(MediaType.APPLICATION_OCTET_STREAM)
-  public Response execute(byte[] payload) {
+  public Response execute(byte[] payload, @Context HttpHeaders headers) {
     try {
+      AuthResponse authResponse = authorizer.verify(headers);
       DeleteObjectRequest request = DeleteObjectRequest.parseFrom(payload);
-      DeleteObjectResponse response = kvStore.delete(request);
+      DeleteObjectResponse response = kvStore.delete(authResponse.getUserToken(), request);
       return toResponse(response);
     } catch (Exception e) {
       log.error("Exception in DeleteObjectApi: ", e);
diff --git a/app/src/main/java/org/vss/api/GetObjectApi.java b/app/src/main/java/org/vss/api/GetObjectApi.java
@@ -4,28 +4,33 @@
 import jakarta.ws.rs.POST;
 import jakarta.ws.rs.Path;
 import jakarta.ws.rs.Produces;
+import jakarta.ws.rs.core.Context;
+import jakarta.ws.rs.core.HttpHeaders;
 import jakarta.ws.rs.core.MediaType;
 import jakarta.ws.rs.core.Response;
 import lombok.extern.slf4j.Slf4j;
 import org.vss.GetObjectRequest;
 import org.vss.GetObjectResponse;
 import org.vss.KVStore;
+import org.vss.auth.AuthResponse;
+import org.vss.auth.Authorizer;
 
 @Path(VssApiEndpoint.GET_OBJECT)
 @Slf4j
 public class GetObjectApi extends AbstractVssApi {
 
   @Inject
-  public GetObjectApi(KVStore kvstore) {
-    super(kvstore);
+  public GetObjectApi(KVStore kvstore, Authorizer authorizer) {
+    super(kvstore, authorizer);
   }
 
   @POST
   @Produces(MediaType.APPLICATION_OCTET_STREAM)
-  public Response execute(byte[] payload) {
+  public Response execute(byte[] payload, @Context HttpHeaders headers) {
     try {
+      AuthResponse authResponse = authorizer.verify(headers);
       GetObjectRequest request = GetObjectRequest.parseFrom(payload);
-      GetObjectResponse response = kvStore.get(request);
+      GetObjectResponse response = kvStore.get(authResponse.getUserToken(), request);
       return toResponse(response);
     } catch (Exception e) {
       log.error("Exception in GetObjectApi: ", e);
diff --git a/app/src/main/java/org/vss/api/ListKeyVersionsApi.java b/app/src/main/java/org/vss/api/ListKeyVersionsApi.java
@@ -4,28 +4,33 @@
 import jakarta.ws.rs.POST;
 import jakarta.ws.rs.Path;
 import jakarta.ws.rs.Produces;
+import jakarta.ws.rs.core.Context;
+import jakarta.ws.rs.core.HttpHeaders;
 import jakarta.ws.rs.core.MediaType;
 import jakarta.ws.rs.core.Response;
 import lombok.extern.slf4j.Slf4j;
 import org.vss.KVStore;
 import org.vss.ListKeyVersionsRequest;
 import org.vss.ListKeyVersionsResponse;
+import org.vss.auth.AuthResponse;
+import org.vss.auth.Authorizer;
 
 @Path(VssApiEndpoint.LIST_KEY_VERSIONS)
 @Slf4j
 public class ListKeyVersionsApi extends AbstractVssApi {
 
   @Inject
-  public ListKeyVersionsApi(KVStore kvStore) {
-    super(kvStore);
+  public ListKeyVersionsApi(KVStore kvStore, Authorizer authorizer) {
+    super(kvStore, authorizer);
   }
 
   @POST
   @Produces(MediaType.APPLICATION_OCTET_STREAM)
-  public Response execute(byte[] payload) {
+  public Response execute(byte[] payload, @Context HttpHeaders headers) {
     try {
+      AuthResponse authResponse = authorizer.verify(headers);
       ListKeyVersionsRequest request = ListKeyVersionsRequest.parseFrom(payload);
-      ListKeyVersionsResponse response = kvStore.listKeyVersions(request);
+      ListKeyVersionsResponse response = kvStore.listKeyVersions(authResponse.getUserToken(), request);
       return toResponse(response);
     } catch (Exception e) {
       log.error("Exception in ListKeyVersionsApi: ", e);
diff --git a/app/src/main/java/org/vss/api/PutObjectsApi.java b/app/src/main/java/org/vss/api/PutObjectsApi.java
@@ -4,28 +4,33 @@
 import jakarta.ws.rs.POST;
 import jakarta.ws.rs.Path;
 import jakarta.ws.rs.Produces;
+import jakarta.ws.rs.core.Context;
+import jakarta.ws.rs.core.HttpHeaders;
 import jakarta.ws.rs.core.MediaType;
 import jakarta.ws.rs.core.Response;
 import lombok.extern.slf4j.Slf4j;
 import org.vss.KVStore;
 import org.vss.PutObjectRequest;
 import org.vss.PutObjectResponse;
+import org.vss.auth.AuthResponse;
+import org.vss.auth.Authorizer;
 
 @Path(VssApiEndpoint.PUT_OBJECTS)
 @Slf4j
 public class PutObjectsApi extends AbstractVssApi {
 
   @Inject
-  public PutObjectsApi(KVStore kvStore) {
-    super(kvStore);
+  public PutObjectsApi(KVStore kvStore, Authorizer authorizer) {
+    super(kvStore, authorizer);
   }
 
   @POST
   @Produces(MediaType.APPLICATION_OCTET_STREAM)
-  public Response execute(byte[] payload) {
+  public Response execute(byte[] payload, @Context HttpHeaders headers) {
     try {
+      AuthResponse authResponse = authorizer.verify(headers);
       PutObjectRequest putObjectRequest = PutObjectRequest.parseFrom(payload);
-      PutObjectResponse response = kvStore.put(putObjectRequest);
+      PutObjectResponse response = kvStore.put(authResponse.getUserToken(), putObjectRequest);
       return toResponse(response);
     } catch (Exception e) {
       log.error("Exception in PutObjectsApi: ", e);
diff --git a/app/src/main/java/org/vss/auth/AuthResponse.java b/app/src/main/java/org/vss/auth/AuthResponse.java
@@ -0,0 +1,10 @@
+package org.vss.auth;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+
+@Data
+@AllArgsConstructor
+public class AuthResponse {
+    private String userToken;
+}
diff --git a/app/src/main/java/org/vss/auth/Authorizer.java b/app/src/main/java/org/vss/auth/Authorizer.java
@@ -0,0 +1,9 @@
+package org.vss.auth;
+
+import jakarta.ws.rs.core.HttpHeaders;
+import org.vss.exception.AuthException;
+
+// Interface for authorizer that is run before every request.
+public interface Authorizer {
+    AuthResponse verify(HttpHeaders headers) throws AuthException;
+}
diff --git a/app/src/main/java/org/vss/auth/NoopAuthorizer.java b/app/src/main/java/org/vss/auth/NoopAuthorizer.java
@@ -0,0 +1,14 @@
+package org.vss.auth;
+
+import jakarta.ws.rs.core.HttpHeaders;
+import org.vss.exception.AuthException;
+
+// A no-operation authorizer, that lets any user-request go through.
+public class NoopAuthorizer implements Authorizer {
+    private static String UNAUTHENTICATED_USER = "unauth-user";
+
+    @Override
+    public AuthResponse verify(HttpHeaders headers) throws AuthException {
+        return new AuthResponse(UNAUTHENTICATED_USER);
+    }
+}
diff --git a/app/src/main/java/org/vss/guice/BaseModule.java b/app/src/main/java/org/vss/guice/BaseModule.java
@@ -11,7 +11,10 @@
 import org.jooq.DSLContext;
 import org.jooq.SQLDialect;
 import org.jooq.impl.DSL;
+import org.jooq.tools.StringUtils;
 import org.vss.KVStore;
+import org.vss.auth.Authorizer;
+import org.vss.auth.NoopAuthorizer;
 import org.vss.impl.postgres.PostgresBackendImpl;
 
 public class BaseModule extends AbstractModule {
@@ -20,6 +23,9 @@ public class BaseModule extends AbstractModule {
   protected void configure() {
     // Provide PostgresBackend as default implementation for KVStore.
     bind(KVStore.class).to(PostgresBackendImpl.class).in(Singleton.class);
+
+    // Default to Noop Authorizer.
+    bind(Authorizer.class).to(NoopAuthorizer.class).in(Singleton.class);
   }
 
   @Provides
diff --git a/app/src/main/java/org/vss/impl/postgres/PostgresBackendImpl.java b/app/src/main/java/org/vss/impl/postgres/PostgresBackendImpl.java
@@ -48,11 +48,12 @@ public PostgresBackendImpl(DSLContext context) {
   }
 
   @Override
-  public GetObjectResponse get(GetObjectRequest request) {
+  public GetObjectResponse get(String userToken, GetObjectRequest request) {
 
     VssDbRecord vssDbRecord = context.selectFrom(VSS_DB)
-        .where(VSS_DB.STORE_ID.eq(request.getStoreId())
-            .and(VSS_DB.KEY.eq(request.getKey())))
+        .where(VSS_DB.USER_TOKEN.eq(userToken)
+            .and(VSS_DB.STORE_ID.eq(request.getStoreId())
+                .and(VSS_DB.KEY.eq(request.getKey()))))
         .fetchOne();
 
     final KeyValue keyValue;
@@ -77,18 +78,18 @@ public GetObjectResponse get(GetObjectRequest request) {
   }
 
   @Override
-  public PutObjectResponse put(PutObjectRequest request) {
+  public PutObjectResponse put(String userToken, PutObjectRequest request) {
 
     String storeId = request.getStoreId();
 
     List<VssDbRecord> vssPutRecords = new ArrayList<>(request.getTransactionItemsList().stream()
-        .map(kv -> buildVssRecord(storeId, kv)).toList());
+        .map(kv -> buildVssRecord(userToken, storeId, kv)).toList());
 
     List<VssDbRecord> vssDeleteRecords = new ArrayList<>(request.getDeleteItemsList().stream()
-        .map(kv -> buildVssRecord(storeId, kv)).toList());
+        .map(kv -> buildVssRecord(userToken, storeId, kv)).toList());
 
     if (request.hasGlobalVersion()) {
-      VssDbRecord globalVersionRecord = buildVssRecord(storeId,
+      VssDbRecord globalVersionRecord = buildVssRecord(userToken, storeId,
           KeyValue.newBuilder()
               .setKey(GLOBAL_VERSION_KEY)
               .setVersion(request.getGlobalVersion())
@@ -130,15 +131,17 @@ private Query buildDeleteObjectQuery(DSLContext dsl, VssDbRecord vssRecord) {
 
   private static DeleteConditionStep<VssDbRecord> buildNonConditionalDeleteQuery(DSLContext dsl,
       VssDbRecord vssRecord) {
-    return dsl.deleteFrom(VSS_DB).where(VSS_DB.STORE_ID.eq(vssRecord.getStoreId())
-        .and(VSS_DB.KEY.eq(vssRecord.getKey())));
+    return dsl.deleteFrom(VSS_DB).where(VSS_DB.USER_TOKEN.eq(vssRecord.getUserToken())
+        .and(VSS_DB.STORE_ID.eq(vssRecord.getStoreId())
+        .and(VSS_DB.KEY.eq(vssRecord.getKey()))));
   }
 
   private static DeleteConditionStep<VssDbRecord> buildConditionalDeleteQuery(DSLContext dsl,
       VssDbRecord vssRecord) {
-    return dsl.deleteFrom(VSS_DB).where(VSS_DB.STORE_ID.eq(vssRecord.getStoreId())
+    return dsl.deleteFrom(VSS_DB).where(VSS_DB.USER_TOKEN.eq(vssRecord.getUserToken())
+        .and(VSS_DB.STORE_ID.eq(vssRecord.getStoreId())
         .and(VSS_DB.KEY.eq(vssRecord.getKey()))
-        .and(VSS_DB.VERSION.eq(vssRecord.getVersion())));
+        .and(VSS_DB.VERSION.eq(vssRecord.getVersion()))));
   }
 
   private Query buildPutObjectQuery(DSLContext dsl, VssDbRecord vssRecord) {
@@ -153,9 +156,9 @@ private Query buildPutObjectQuery(DSLContext dsl, VssDbRecord vssRecord) {
 
   private Query buildNonConditionalUpsertRecordQuery(DSLContext dsl, VssDbRecord vssRecord) {
     return dsl.insertInto(VSS_DB)
-        .values(vssRecord.getStoreId(), vssRecord.getKey(),
+        .values(vssRecord.getUserToken(), vssRecord.getStoreId(), vssRecord.getKey(),
             vssRecord.getValue(), 1, vssRecord.getCreatedAt(), vssRecord.getLastUpdatedAt())
-        .onConflict(VSS_DB.STORE_ID, VSS_DB.KEY)
+        .onConflict(VSS_DB.USER_TOKEN, VSS_DB.STORE_ID, VSS_DB.KEY)
         .doUpdate()
         .set(VSS_DB.VALUE, vssRecord.getValue())
         .set(VSS_DB.VERSION, 1L)
@@ -165,7 +168,7 @@ private Query buildNonConditionalUpsertRecordQuery(DSLContext dsl, VssDbRecord v
   private Insert<VssDbRecord> buildConditionalInsertRecordQuery(DSLContext dsl,
       VssDbRecord vssRecord) {
     return dsl.insertInto(VSS_DB)
-        .values(vssRecord.getStoreId(), vssRecord.getKey(),
+        .values(vssRecord.getUserToken(), vssRecord.getStoreId(), vssRecord.getKey(),
             vssRecord.getValue(), 1, vssRecord.getCreatedAt(), vssRecord.getLastUpdatedAt())
         .onDuplicateKeyIgnore();
   }
@@ -175,14 +178,16 @@ private Update<VssDbRecord> buildConditionalUpdateRecordQuery(DSLContext dsl, Vs
         .set(Map.of(VSS_DB.VALUE, vssRecord.getValue(),
             VSS_DB.VERSION, vssRecord.getVersion() + 1,
             VSS_DB.LAST_UPDATED_AT, vssRecord.getLastUpdatedAt()))
-        .where(VSS_DB.STORE_ID.eq(vssRecord.getStoreId())
+        .where(VSS_DB.USER_TOKEN.eq(vssRecord.getUserToken())
+            .and(VSS_DB.STORE_ID.eq(vssRecord.getStoreId())
             .and(VSS_DB.KEY.eq(vssRecord.getKey()))
-            .and(VSS_DB.VERSION.eq(vssRecord.getVersion())));
+            .and(VSS_DB.VERSION.eq(vssRecord.getVersion()))));
   }
 
-  private VssDbRecord buildVssRecord(String storeId, KeyValue kv) {
+  private VssDbRecord buildVssRecord(String userToken, String storeId, KeyValue kv) {
     OffsetDateTime today = OffsetDateTime.now(ZoneOffset.UTC).truncatedTo(ChronoUnit.DAYS);
     return new VssDbRecord()
+        .setUserToken(userToken)
         .setStoreId(storeId)
         .setKey(kv.getKey())
         .setValue(kv.getValue().toByteArray())
@@ -192,9 +197,9 @@ private VssDbRecord buildVssRecord(String storeId, KeyValue kv) {
   }
 
   @Override
-  public DeleteObjectResponse delete(DeleteObjectRequest request) {
+  public DeleteObjectResponse delete(String userToken, DeleteObjectRequest request) {
     String storeId = request.getStoreId();
-    VssDbRecord vssDbRecord = buildVssRecord(storeId, request.getKeyValue());
+    VssDbRecord vssDbRecord = buildVssRecord(userToken, storeId, request.getKeyValue());
 
     context.transaction((ctx) -> {
       DSLContext dsl = ctx.dsl();
@@ -206,7 +211,7 @@ public DeleteObjectResponse delete(DeleteObjectRequest request) {
   }
 
   @Override
-  public ListKeyVersionsResponse listKeyVersions(ListKeyVersionsRequest request) {
+  public ListKeyVersionsResponse listKeyVersions(String userToken, ListKeyVersionsRequest request) {
     String storeId = request.getStoreId();
     String keyPrefix = request.getKeyPrefix();
     String pageToken = request.getPageToken();
@@ -221,12 +226,13 @@ public ListKeyVersionsResponse listKeyVersions(ListKeyVersionsRequest request) {
           .setStoreId(storeId)
           .setKey(GLOBAL_VERSION_KEY)
           .build();
-      globalVersion = get(getGlobalVersionRequest).getValue().getVersion();
+      globalVersion = get(userToken, getGlobalVersionRequest).getValue().getVersion();
     }
 
     List<VssDbRecord> vssDbRecords = context.select(VSS_DB.KEY, VSS_DB.VERSION).from(VSS_DB)
-        .where(VSS_DB.STORE_ID.eq(storeId)
-            .and(VSS_DB.KEY.startsWith(keyPrefix)))
+        .where(VSS_DB.USER_TOKEN.eq(userToken)
+            .and(VSS_DB.STORE_ID.eq(storeId)
+            .and(VSS_DB.KEY.startsWith(keyPrefix))))
         .orderBy(VSS_DB.KEY)
         .seek(pageToken)
         .limit(Math.min(pageSize, LIST_KEY_VERSIONS_MAX_PAGE_SIZE))
diff --git a/app/src/main/java/org/vss/impl/postgres/sql/v0_create_vss_db.sql b/app/src/main/java/org/vss/impl/postgres/sql/v0_create_vss_db.sql
@@ -1,9 +1,10 @@
 CREATE TABLE vss_db (
+    user_token character varying(120) NOT NULL CHECK (user_token <> ''),
     store_id character varying(120) NOT NULL CHECK (store_id <> ''),
     key character varying(600) NOT NULL,
     value bytea NULL,
     version bigint NOT NULL,
     created_at TIMESTAMP WITH TIME ZONE,
     last_updated_at TIMESTAMP WITH TIME ZONE,
-    PRIMARY KEY (store_id, key)
+    PRIMARY KEY (user_token, store_id, key)
 );
diff --git a/app/src/test/java/org/vss/AbstractKVStoreIntegrationTest.java b/app/src/test/java/org/vss/AbstractKVStoreIntegrationTest.java
diff --git a/app/src/test/java/org/vss/api/DeleteObjectApiTest.java b/app/src/test/java/org/vss/api/DeleteObjectApiTest.java
diff --git a/app/src/test/java/org/vss/api/GetObjectApiTest.java b/app/src/test/java/org/vss/api/GetObjectApiTest.java
diff --git a/app/src/test/java/org/vss/api/ListKeyVersionsApiTest.java b/app/src/test/java/org/vss/api/ListKeyVersionsApiTest.java
diff --git a/app/src/test/java/org/vss/api/PutObjectsApiTest.java b/app/src/test/java/org/vss/api/PutObjectsApiTest.java
diff --git a/app/src/test/java/org/vss/impl/postgres/PostgresBackendImplIntegrationTest.java b/app/src/test/java/org/vss/impl/postgres/PostgresBackendImplIntegrationTest.java
