Add JWT array payload support for authorization (#1434)

* Add JWT array payload support for authorization

Signed-off-by: Marvin Froeder <[email protected]>

* Add FIXME comment to ArrayContainsSqlTranslation for unwrapCast removal

Signed-off-by: Marvin Froeder <[email protected]>

* Update functionParameterTest.txt snapshot

Signed-off-by: Marvin Froeder <[email protected]>

---------

Signed-off-by: Marvin Froeder <[email protected]>

Author: velo

sqrl-planner/src/main/java/com/datasqrl/function/PgSpecificOperatorTable.java

@@ -18,8 +18,10 @@
 import static com.datasqrl.function.CalciteFunctionUtil.lightweightOp;
 
 import org.apache.calcite.sql.SqlBinaryOperator;
+import org.apache.calcite.sql.SqlCall;
 import org.apache.calcite.sql.SqlKind;
 import org.apache.calcite.sql.SqlUnresolvedFunction;
+import org.apache.calcite.sql.SqlWriter;
 import org.apache.calcite.sql.type.ReturnTypes;
 import org.apache.calcite.sql.type.SqlTypeName;
 
@@ -67,4 +69,27 @@ public class PgSpecificOperatorTable {
           ReturnTypes.explicit(SqlTypeName.ANY),
           null,
           null);
+
+  public static final SqlBinaryOperator EqualsAny =
+      new SqlBinaryOperator(
+          "= ANY",
+          SqlKind.OTHER_FUNCTION,
+          22,
+          true,
+          ReturnTypes.explicit(SqlTypeName.BOOLEAN),
+          null,
+          null) {
+
+        @Override
+        public void unparse(SqlWriter writer, SqlCall call, int leftPrec, int rightPrec) {
+          var left = call.operand(0);
+          var right = call.operand(1);
+
+          left.unparse(writer, leftPrec, getLeftPrec());
+          writer.keyword("= ANY");
+          writer.print("(");
+          right.unparse(writer, getRightPrec(), rightPrec);
+          writer.print(")");
+        }
+      };
 }

sqrl-planner/src/main/java/com/datasqrl/function/builtinflink/ArrayContainsSqlTranslation.java

@@ -0,0 +1,59 @@
+/*
+ * Copyright © 2021 DataSQRL ([email protected])
+ *
+ * Licensed 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.
+ */
+package com.datasqrl.function.builtinflink;
+
+import com.datasqrl.function.CalciteFunctionUtil;
+import com.datasqrl.function.PgSpecificOperatorTable;
+import com.datasqrl.function.translations.PostgresSqlTranslation;
+import com.datasqrl.function.translations.SqlTranslation;
+import com.google.auto.service.AutoService;
+import org.apache.calcite.sql.SqlBasicCall;
+import org.apache.calcite.sql.SqlCall;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlWriter;
+import org.apache.calcite.sql.parser.SqlParserPos;
+import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
+
+@AutoService(SqlTranslation.class)
+public class ArrayContainsSqlTranslation extends PostgresSqlTranslation {
+
+  public ArrayContainsSqlTranslation() {
+    super(CalciteFunctionUtil.lightweightOp(BuiltInFunctionDefinitions.ARRAY_CONTAINS));
+  }
+
+  @Override
+  public void unparse(SqlCall call, SqlWriter writer, int leftPrec, int rightPrec) {
+    var rawArray = call.getOperandList().get(0);
+    var value = call.getOperandList().get(1);
+
+    var array = unwrapCast(rawArray);
+
+    // Emit: value = ANY(array)
+    PgSpecificOperatorTable.EqualsAny.createCall(SqlParserPos.ZERO, value, array)
+        .unparse(writer, leftPrec, rightPrec);
+  }
+
+  // FIXME: Remove unwrapCast method
+  private SqlNode unwrapCast(SqlNode node) {
+    if (node instanceof SqlBasicCall) {
+      SqlBasicCall call = (SqlBasicCall) node;
+      if (call.getOperator().getName().equalsIgnoreCase("CAST")) {
+        return call.getOperandList().get(0);
+      }
+    }
+    return node;
+  }
+}

sqrl-server/sqrl-server-vertx-base/src/main/java/com/datasqrl/graphql/auth/AuthMetadataReader.java

@@ -19,6 +19,7 @@
 
 import com.datasqrl.graphql.server.MetadataReader;
 import graphql.schema.DataFetchingEnvironment;
+import io.vertx.core.json.JsonArray;
 import io.vertx.ext.web.RoutingContext;
 import lombok.extern.slf4j.Slf4j;
 
@@ -49,6 +50,11 @@ public Object read(DataFetchingEnvironment env, String name, boolean isRequired)
       checkNotNull(value, "Claim '%s' must not be null", name);
     }
 
+    if (value instanceof JsonArray array) {
+      // Unwrap JsonArray to plain Java array to avoid pgclient treating it as JSONB
+      return array.getList().toArray(new Object[0]);
+    }
+
     return value;
   }
 }

sqrl-testing/sqrl-integration-tests/src/test/java/com/datasqrl/FullUseCasesIT.java

@@ -339,7 +339,7 @@ void printUseCaseNumbers(UseCaseTestParameter param) {
   public void runTestCaseByName() {
     var param =
         getSpecificUseCase(
-            p -> p.sqrlFileName.startsWith("flink_kafka.sqrl") && p.goal.equals("test"));
+            p -> p.sqrlFileName.startsWith("jwt-authorized.sqrl") && p.goal.equals("test"));
 
     useCase(param);
   }

sqrl-testing/sqrl-integration-tests/src/test/resources/snapshots/com/datasqrl/DAGPlannerTest/functionParameterTest.txt

@@ -351,7 +351,7 @@ INSERT INTO `default_catalog`.`default_database`.`CustomerByNothing_2`
           ],
           "query" : {
             "type" : "SqlQuery",
-            "sql" : "SELECT *\nFROM \"Customer_1\"\nWHERE \"array_contains\"(CAST($1 AS JSONB), \"customerid\")",
+            "sql" : "SELECT *\nFROM \"Customer_1\"\nWHERE (\"customerid\" = ANY ($1))",
             "parameters" : [
               {
                 "type" : "arg",

sqrl-testing/sqrl-integration-tests/src/test/resources/snapshots/com/datasqrl/UseCaseCompileTest/jwt-authorized--package.txt

@@ -18,6 +18,25 @@ SQL: CREATE VIEW AuthMyTable AS
      WHERE t.val = ?   
      ORDER BY t.val ASC;
 
+=== AuthMyTableValues
+ID:     default_catalog.default_database.AuthMyTableValues
+Type:   query
+Stage:  postgres
+Inputs: default_catalog.default_database.MyTable
+Annotations:
+ - parameters: vals
+ - base-table: MyTable
+Plan:
+LogicalSort(sort0=[$0], dir0=[ASC-nulls-first])
+  LogicalProject(val=[$0])
+    LogicalFilter(condition=[array_contains(CAST(?0):BIGINT ARRAY, CAST($0):BIGINT)])
+      LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
+SQL: CREATE VIEW AuthMyTableValues AS 
+    SELECT t.* 
+      FROM MyTable t 
+     WHERE array_contains(cast(?     as ARRAY<BIGINT>), t.val)
+     ORDER BY t.val ASC;
+
 === MyTable
 ID:     default_catalog.default_database.MyTable
 Type:   state
@@ -139,6 +158,39 @@ CREATE TABLE IF NOT EXISTS "MyTable" ("val" INTEGER NOT NULL , PRIMARY KEY ("val
             "database" : "POSTGRES"
           }
         }
+      },
+      {
+        "type" : "args",
+        "parentType" : "Query",
+        "fieldName" : "AuthMyTableValues",
+        "exec" : {
+          "arguments" : [
+            {
+              "type" : "variable",
+              "path" : "offset"
+            },
+            {
+              "type" : "variable",
+              "path" : "limit"
+            }
+          ],
+          "query" : {
+            "type" : "SqlQuery",
+            "sql" : "SELECT *\nFROM (SELECT \"val\"\n  FROM \"MyTable\"\n  ORDER BY \"val\" NULLS FIRST) AS \"t\"\nWHERE (CAST(\"val\" AS BIGINT) = ANY ($1))\nORDER BY \"val\" NULLS FIRST",
+            "parameters" : [
+              {
+                "type" : "metadata",
+                "metadata" : {
+                  "metadataType" : "AUTH",
+                  "name" : "values",
+                  "isRequired" : true
+                }
+              }
+            ],
+            "pagination" : "LIMIT_AND_OFFSET",
+            "database" : "POSTGRES"
+          }
+        }
       }
     ],
     "mutations" : [ ],
@@ -195,11 +247,37 @@ CREATE TABLE IF NOT EXISTS "MyTable" ("val" INTEGER NOT NULL , PRIMARY KEY ("val
         "mcpMethod" : "TOOL",
         "restMethod" : "GET",
         "uriTemplate" : "queries/AuthMyTable{?offset,limit}"
+      },
+      {
+        "function" : {
+          "name" : "GetAuthMyTableValues",
+          "parameters" : {
+            "type" : "object",
+            "properties" : {
+              "offset" : {
+                "type" : "integer"
+              },
+              "limit" : {
+                "type" : "integer"
+              }
+            },
+            "required" : [ ]
+          }
+        },
+        "format" : "JSON",
+        "apiQuery" : {
+          "query" : "query AuthMyTableValues($limit: Int = 10, $offset: Int = 0) {\nAuthMyTableValues(limit: $limit, offset: $offset) {\nval\n}\n\n}",
+          "queryName" : "AuthMyTableValues",
+          "operationType" : "QUERY"
+        },
+        "mcpMethod" : "TOOL",
+        "restMethod" : "GET",
+        "uriTemplate" : "queries/AuthMyTableValues{?offset,limit}"
       }
     ],
     "schema" : {
       "type" : "string",
-      "schema" : "\"An RFC-3339 compliant Full Date Scalar\"\nscalar Date\n\n\"A slightly refined version of RFC-3339 compliant DateTime Scalar\"\nscalar DateTime\n\n\"A JSON scalar\"\nscalar JSON\n\n\"24-hour clock time value string in the format `hh:mm:ss` or `hh:mm:ss.sss`.\"\nscalar LocalTime\n\n\"A 64-bit signed integer\"\nscalar Long\n\ntype MyTable {\n  val: Int!\n}\n\ntype Query {\n  MyTable(limit: Int = 10, offset: Int = 0): [MyTable!]\n  AuthMyTable(limit: Int = 10, offset: Int = 0): [MyTable!]\n}\n\nenum _McpMethodType {\n  NONE\n  TOOL\n  RESOURCE\n}\n\nenum _RestMethodType {\n  NONE\n  GET\n  POST\n}\n\ndirective @api(mcp: _McpMethodType, rest: _RestMethodType, uri: String) on QUERY | MUTATION | FIELD_DEFINITION\n"
+      "schema" : "\"An RFC-3339 compliant Full Date Scalar\"\nscalar Date\n\n\"A slightly refined version of RFC-3339 compliant DateTime Scalar\"\nscalar DateTime\n\n\"A JSON scalar\"\nscalar JSON\n\n\"24-hour clock time value string in the format `hh:mm:ss` or `hh:mm:ss.sss`.\"\nscalar LocalTime\n\n\"A 64-bit signed integer\"\nscalar Long\n\ntype MyTable {\n  val: Int!\n}\n\ntype Query {\n  MyTable(limit: Int = 10, offset: Int = 0): [MyTable!]\n  AuthMyTable(limit: Int = 10, offset: Int = 0): [MyTable!]\n  AuthMyTableValues(limit: Int = 10, offset: Int = 0): [MyTable!]\n}\n\nenum _McpMethodType {\n  NONE\n  TOOL\n  RESOURCE\n}\n\nenum _RestMethodType {\n  NONE\n  GET\n  POST\n}\n\ndirective @api(mcp: _McpMethodType, rest: _RestMethodType, uri: String) on QUERY | MUTATION | FIELD_DEFINITION\n"
     }
   }
 }

sqrl-testing/sqrl-integration-tests/src/test/resources/usecases/jwt-authorized/generate_jwt_tokens.py

@@ -0,0 +1,118 @@
+#!/usr/bin/env python3
+"""
+JWT Token Generation Utility for DataSQRL JWT Authorization Tests
+
+This utility generates JWT tokens for testing JWT-based authorization in DataSQRL.
+The tokens are compatible with the jwt-authorized test suite configuration.
+
+REQUIREMENTS:
+    pip install PyJWT
+
+USAGE:
+    1. Edit the PAYLOAD variable below to set the desired JWT claims
+    2. Run: python3 generate_jwt_tokens.py
+    3. Copy the generated token to your test .properties file
+
+JWT CONFIGURATION:
+    The tokens are generated using the secret key and parameters from package.json:
+    - Secret: testSecretThatIsAtLeast256BitsLong32Chars (base64: dGVzdFNlY3JldFRoYXRJc0F0TGVhc3QyNTZCaXRzTG9uZzMyQ2hhcnM=)
+    - Algorithm: HS256
+    - Issuer: my-test-issuer  
+    - Audience: ["my-test-audience"]
+    - Default Expiration: 9999999999 (far future)
+
+SQRL FUNCTIONS:
+    - AuthMyTable(val BIGINT): Expects JWT claim "val" with single integer value
+    - AuthMyTableValues(val ARRAY<BIGINT>): Expects JWT claim "values" with array of integers
+
+EXAMPLES:
+    # For AuthMyTable with val=73, edit PAYLOAD to:
+    # PAYLOAD = {
+    #     "iss": "my-test-issuer",
+    #     "aud": ["my-test-audience"],
+    #     "exp": 9999999999,
+    #     "val": 73
+    # }
+    
+    # For AuthMyTableValues with values=[1,2,82], edit PAYLOAD to:
+    # PAYLOAD = {
+    #     "iss": "my-test-issuer", 
+    #     "aud": ["my-test-audience"],
+    #     "exp": 9999999999,
+    #     "values": [1, 2, 82]
+    # }
+"""
+
+import jwt
+import json
+
+# JWT configuration matching package.json
+SECRET = "testSecretThatIsAtLeast256BitsLong32Chars"
+ALGORITHM = "HS256"
+
+# ==============================================================================
+# EDIT THIS PAYLOAD TO GENERATE DIFFERENT TOKENS
+# ==============================================================================
+# Change the payload below and run this script to generate a new JWT token
+PAYLOAD = {
+    "iss": "my-test-issuer",
+    "aud": ["my-test-audience"],
+    "exp": 9999999999,
+    "values": [1, 2, 82]  # Example: for AuthMyTableValues with values 1, 2, 82
+    # "val": 73           # Example: for AuthMyTable with single value 73
+    # "val": None         # Example: for null value
+    # (omit val/values)   # Example: for no-val scenarios
+}
+# ==============================================================================
+
+
+def generate_token():
+    """Generate JWT token using the hardcoded PAYLOAD."""
+    token = jwt.encode(PAYLOAD, SECRET, algorithm=ALGORITHM)
+    return token
+
+
+def decode_token(token):
+    """Decode and verify JWT token."""
+    try:
+        decoded = jwt.decode(token, SECRET, algorithms=[ALGORITHM], 
+                           audience=PAYLOAD["aud"], issuer=PAYLOAD["iss"])
+        return decoded
+    except jwt.InvalidTokenError as e:
+        return f"Invalid token: {e}"
+
+
+def main():
+    """Generate token using the hardcoded PAYLOAD and display results."""
+    print("=" * 80)
+    print("JWT Token Generator for DataSQRL JWT Authorization Tests")
+    print("=" * 80)
+    
+    print("\nCurrent Payload:")
+    print(json.dumps(PAYLOAD, indent=2))
+    
+    token = generate_token()
+    
+    print(f"\nGenerated JWT Token:")
+    print(token)
+    
+    print(f"\nUse in .properties file:")
+    print(f"Authorization: Bearer {token}")
+    
+    # Verify by decoding
+    print(f"\nVerification (decoded payload):")
+    decoded = decode_token(token)
+    if isinstance(decoded, dict):
+        print(json.dumps(decoded, indent=2))
+    else:
+        print(f"Error: {decoded}")
+    
+    print("\n" + "=" * 80)
+    print("To generate different tokens:")
+    print("1. Edit the PAYLOAD variable in this script")
+    print("2. Run this script again")
+    print("=" * 80)
+
+
+if __name__ == "__main__":
+    main()

sqrl-testing/sqrl-integration-tests/src/test/resources/usecases/jwt-authorized/jwt-authorized.sqrl

@@ -10,3 +10,9 @@ AuthMyTable(val BIGINT NOT NULL METADATA FROM 'auth.val') :=
       FROM MyTable t 
      WHERE t.val = :val
      ORDER BY t.val ASC;
+
+AuthMyTableValues(vals ARRAY<BIGINT> NOT NULL METADATA FROM 'auth.values') :=
+    SELECT t.* 
+      FROM MyTable t 
+     WHERE array_contains(cast(:vals as ARRAY<BIGINT>), t.val)
+     ORDER BY t.val ASC;

sqrl-testing/sqrl-integration-tests/src/test/resources/usecases/jwt-authorized/snapshots-jwt-authorized/AuthMyTableValues-valid-values.snapshot

@@ -0,0 +1,9 @@
+{
+  "data" : {
+    "AuthMyTableValues" : [ {
+      "val" : 1
+    }, {
+      "val" : 2
+    } ]
+  }
+}

sqrl-testing/sqrl-integration-tests/src/test/resources/usecases/jwt-authorized/tests-jwt-authorized/AuthMyTableValues-valid-values.graphql

@@ -0,0 +1,5 @@
+query {
+  AuthMyTableValues {
+    val
+  }
+}