Merge pull request #239 from LucDeCaf/feat/validate-crud-row-fields

Validate input for CrudEntry.fromRow explicitly

Author: simolus3

packages/powersync_core/analysis_options.yaml

@@ -13,6 +13,12 @@
 
 include: package:lints/recommended.yaml
 
+analyzer:
+  language:
+    strict-casts: true
+    strict-inference: true
+    strict-raw-types: true
+
 # Uncomment the following section to specify additional rules.
 
 # linter:

packages/powersync_core/example/getting_started.dart

@@ -91,7 +91,7 @@ Future<String> getDatabasePath() async {
   return join(dir, dbFilename);
 }
 
-openDatabase() async {
+Future<void> openDatabase() async {
   // Setup the database.
   final psFactory = PowerSyncDartOpenFactory(path: await getDatabasePath());
   db = PowerSyncDatabase.withFactory(psFactory, schema: schema);

packages/powersync_core/lib/src/bucket_storage.dart

@@ -21,7 +21,7 @@ class BucketStorage {
     _init();
   }
 
-  _init() {}
+  void _init() {}
 
   // Use only for read statements
   Future<ResultSet> select(String query,
@@ -36,7 +36,8 @@ class BucketStorage {
         'SELECT name as bucket, cast(last_op as TEXT) as op_id FROM ps_buckets WHERE pending_delete = 0 AND name != \'\$local\'');
     return [
       for (var row in rows)
-        BucketState(bucket: row['bucket'], opId: row['op_id'])
+        BucketState(
+            bucket: row['bucket'] as String, opId: row['op_id'] as String)
     ];
   }
 
@@ -157,14 +158,16 @@ class BucketStorage {
       Checkpoint checkpoint) async {
     final rs = await select("SELECT powersync_validate_checkpoint(?) as result",
         [jsonEncode(checkpoint)]);
-    final result = jsonDecode(rs[0]['result']);
-    if (result['valid']) {
+    final result =
+        jsonDecode(rs[0]['result'] as String) as Map<String, dynamic>;
+    if (result['valid'] as bool) {
       return SyncLocalDatabaseResult(ready: true);
     } else {
       return SyncLocalDatabaseResult(
           checkpointValid: false,
           ready: false,
-          checkpointFailures: result['failed_buckets'].cast<String>());
+          checkpointFailures:
+              (result['failed_buckets'] as List).cast<String>());
     }
   }
 
@@ -232,7 +235,7 @@ class BucketStorage {
       // Nothing to update
       return false;
     }
-    int seqBefore = rs.first['seq'];
+    int seqBefore = rs.first['seq'] as int;
     var opId = await checkpointCallback();
 
     return await writeTransaction((tx) async {
@@ -244,7 +247,7 @@ class BucketStorage {
           .execute('SELECT seq FROM sqlite_sequence WHERE name = \'ps_crud\'');
       assert(rs.isNotEmpty);
 
-      int seqAfter = rs.first['seq'];
+      int seqAfter = rs.first['seq'] as int;
       if (seqAfter != seqBefore) {
         // New crud data may have been uploaded since we got the checkpoint. Abort.
         return false;
@@ -362,12 +365,13 @@ class SyncBucketData {
       this.nextAfter});
 
   SyncBucketData.fromJson(Map<String, dynamic> json)
-      : bucket = json['bucket'],
-        hasMore = json['has_more'] ?? false,
-        after = json['after'],
-        nextAfter = json['next_after'],
-        data =
-            (json['data'] as List).map((e) => OplogEntry.fromJson(e)).toList();
+      : bucket = json['bucket'] as String,
+        hasMore = json['has_more'] as bool? ?? false,
+        after = json['after'] as String?,
+        nextAfter = json['next_after'] as String?,
+        data = (json['data'] as List)
+            .map((e) => OplogEntry.fromJson(e as Map<String, dynamic>))
+            .toList();
 
   Map<String, dynamic> toJson() {
     return {
@@ -407,16 +411,25 @@ class OplogEntry {
       required this.checksum});
 
   OplogEntry.fromJson(Map<String, dynamic> json)
-      : opId = json['op_id'],
-        op = OpType.fromJson(json['op']),
-        rowType = json['object_type'],
-        rowId = json['object_id'],
-        checksum = json['checksum'],
-        data = json['data'] is String ? json['data'] : jsonEncode(json['data']),
-        subkey = json['subkey'] is String ? json['subkey'] : null;
+      : opId = json['op_id'] as String,
+        op = OpType.fromJson(json['op'] as String),
+        rowType = json['object_type'] as String?,
+        rowId = json['object_id'] as String?,
+        checksum = json['checksum'] as int,
+        data = switch (json['data']) {
+          String data => data,
+          var other => jsonEncode(other),
+        },
+        subkey = switch (json['subkey']) {
+          String subkey => subkey,
+          _ => null,
+        };
 
   Map<String, dynamic>? get parsedData {
-    return data == null ? null : jsonDecode(data!);
+    return switch (data) {
+      final data? => jsonDecode(data) as Map<String, dynamic>,
+      null => null,
+    };
   }
 
   /// Key to uniquely represent a source entry in a bucket.
@@ -463,16 +476,16 @@ class SyncLocalDatabaseResult {
 
   @override
   int get hashCode {
-    return Object.hash(
-        ready, checkpointValid, const ListEquality().hash(checkpointFailures));
+    return Object.hash(ready, checkpointValid,
+        const ListEquality<String?>().hash(checkpointFailures));
   }
 
   @override
   bool operator ==(Object other) {
     return other is SyncLocalDatabaseResult &&
         other.ready == ready &&
         other.checkpointValid == checkpointValid &&
-        const ListEquality()
+        const ListEquality<String?>()
             .equals(other.checkpointFailures, checkpointFailures);
   }
 }

packages/powersync_core/lib/src/connector.dart

@@ -90,13 +90,13 @@ class PowerSyncCredentials {
   }
 
   factory PowerSyncCredentials.fromJson(Map<String, dynamic> parsed) {
-    String token = parsed['token'];
+    String token = parsed['token'] as String;
     DateTime? expiresAt = getExpiryDate(token);
 
     return PowerSyncCredentials(
-        endpoint: parsed['endpoint'],
-        token: parsed['token'],
-        userId: parsed['user_id'],
+        endpoint: parsed['endpoint'] as String,
+        token: token,
+        userId: parsed['user_id'] as String?,
         expiresAt: expiresAt);
   }
 
@@ -110,9 +110,9 @@ class PowerSyncCredentials {
         // dart:convert doesn't like missing padding
         final rawData = base64Url.decode(base64.normalize(parts[1]));
         final text = Utf8Decoder().convert(rawData);
-        Map<String, dynamic> payload = jsonDecode(text);
-        if (payload.containsKey('exp') && payload['exp'] is int) {
-          return DateTime.fromMillisecondsSinceEpoch(payload['exp'] * 1000);
+        final payload = jsonDecode(text) as Map<String, dynamic>;
+        if (payload['exp'] case int exp) {
+          return DateTime.fromMillisecondsSinceEpoch(exp * 1000);
         }
       }
       return null;
@@ -131,7 +131,7 @@ class PowerSyncCredentials {
     return Uri.parse(endpoint).resolve(path);
   }
 
-  _validateEndpoint() {
+  void _validateEndpoint() {
     final parsed = Uri.parse(endpoint);
     if ((!parsed.isScheme('http') && !parsed.isScheme('https')) ||
         parsed.host.isEmpty) {
@@ -162,14 +162,14 @@ class DevCredentials {
 
   factory DevCredentials.fromJson(Map<String, dynamic> parsed) {
     return DevCredentials(
-        endpoint: parsed['endpoint'],
-        token: parsed['token'],
-        userId: parsed['user_id']);
+        endpoint: parsed['endpoint'] as String,
+        token: parsed['token'] as String?,
+        userId: parsed['user_id'] as String?);
   }
 
   factory DevCredentials.fromString(String credentials) {
     var parsed = jsonDecode(credentials);
-    return DevCredentials.fromJson(parsed);
+    return DevCredentials.fromJson(parsed as Map<String, dynamic>);
   }
 
   static DevCredentials? fromOptionalString(String? credentials) {
@@ -255,10 +255,12 @@ class DevConnector extends PowerSyncBackendConnector {
 
     if (res.statusCode == 200) {
       var parsed = jsonDecode(res.body);
+      var data = parsed['data'] as Map<String, dynamic>;
+
       storeDevCredentials(DevCredentials(
           endpoint: endpoint,
-          token: parsed['data']['token'],
-          userId: parsed['data']['user_id']));
+          token: data['token'] as String?,
+          userId: data['user_id'] as String?));
     } else {
       throw http.ClientException(res.reasonPhrase ?? 'Request failed', uri);
     }
@@ -281,7 +283,8 @@ class DevConnector extends PowerSyncBackendConnector {
       throw http.ClientException(res.reasonPhrase ?? 'Request failed', uri);
     }
 
-    return PowerSyncCredentials.fromJson(jsonDecode(res.body)['data']);
+    return PowerSyncCredentials.fromJson(
+        jsonDecode(res.body)['data'] as Map<String, dynamic>);
   }
 
   /// Upload changes using the PowerSync dev API.
@@ -319,7 +322,7 @@ class DevConnector extends PowerSyncBackendConnector {
     final body = jsonDecode(response.body);
     // writeCheckpoint is optional, but reduces latency between writing,
     // and reading back the same change.
-    final String? writeCheckpoint = body['data']['write_checkpoint'];
+    final writeCheckpoint = body['data']['write_checkpoint'] as String?;
     await batch.complete(writeCheckpoint: writeCheckpoint);
   }
 }

packages/powersync_core/lib/src/crud.dart

@@ -81,9 +81,15 @@ class CrudEntry {
       this.opData);
 
   factory CrudEntry.fromRow(sqlite.Row row) {
-    final data = jsonDecode(row['data']);
-    return CrudEntry(row['id'], UpdateType.fromJsonChecked(data['op'])!,
-        data['type'], data['id'], row['tx_id'], data['data']);
+    final data = jsonDecode(row['data'] as String);
+    return CrudEntry(
+      row['id'] as int,
+      UpdateType.fromJsonChecked(data['op'] as String)!,
+      data['type'] as String,
+      data['id'] as String,
+      row['tx_id'] as int,
+      data['data'] as Map<String, Object?>?,
+    );
   }
 
   /// Converts the change to JSON format, as required by the dev crud API.
@@ -111,13 +117,13 @@ class CrudEntry {
         other.op == op &&
         other.table == table &&
         other.id == id &&
-        const MapEquality().equals(other.opData, opData));
+        const MapEquality<String, dynamic>().equals(other.opData, opData));
   }
 
   @override
   int get hashCode {
     return Object.hash(transactionId, clientId, op.toJson(), table, id,
-        const MapEquality().hash(opData));
+        const MapEquality<String, dynamic>().hash(opData));
   }
 }
 

packages/powersync_core/lib/src/database/native/native_powersync_database.dart

@@ -143,10 +143,10 @@ class PowerSyncDatabaseImpl
     await isInitialized;
     final dbRef = database.isolateConnectionFactory();
     ReceivePort rPort = ReceivePort();
-    StreamSubscription? crudUpdateSubscription;
+    StreamSubscription<UpdateNotification>? crudUpdateSubscription;
     rPort.listen((data) async {
       if (data is List) {
-        String action = data[0];
+        String action = data[0] as String;
         if (action == "getCredentials") {
           await (data[1] as PortCompleter).handle(() async {
             final token = await connector.getCredentialsCached();
@@ -159,7 +159,7 @@ class PowerSyncDatabaseImpl
             await connector.prefetchCredentials();
           });
         } else if (action == 'init') {
-          SendPort port = data[1];
+          SendPort port = data[1] as SendPort;
           var crudStream =
               database.onChange(['ps_crud'], throttle: crudThrottleTime);
           crudUpdateSubscription = crudStream.listen((event) {
@@ -173,15 +173,15 @@ class PowerSyncDatabaseImpl
             await connector.uploadData(this);
           });
         } else if (action == 'status') {
-          final SyncStatus status = data[1];
+          final SyncStatus status = data[1] as SyncStatus;
           setStatus(status);
         } else if (action == 'close') {
           // Clear status apart from lastSyncedAt
           setStatus(SyncStatus(lastSyncedAt: currentStatus.lastSyncedAt));
           rPort.close();
           crudUpdateSubscription?.cancel();
         } else if (action == 'log') {
-          LogRecord record = data[1];
+          LogRecord record = data[1] as LogRecord;
           logger.log(
               record.level, record.message, record.error, record.stackTrace);
         }
@@ -290,7 +290,7 @@ Future<void> _powerSyncDatabaseIsolate(
 
   rPort.listen((message) async {
     if (message is List) {
-      String action = message[0];
+      String action = message[0] as String;
       if (action == 'update') {
         crudUpdateController.add('update');
       } else if (action == 'close') {

packages/powersync_core/lib/src/database/powersync_db_mixin.dart

@@ -87,7 +87,7 @@ mixin PowerSyncDatabaseMixin implements SqliteConnection {
     try {
       final row =
           await database.get('SELECT powersync_rs_version() as version');
-      version = row['version'];
+      version = row['version'] as String;
     } catch (e) {
       throw SqliteException(
           1, 'The powersync extension is not loaded correctly. Details: $e');
@@ -291,7 +291,7 @@ mixin PowerSyncDatabaseMixin implements SqliteConnection {
   /// A connection factory that can be passed to different isolates.
   ///
   /// Use this to access the database in background isolates.
-  isolateConnectionFactory() {
+  IsolateConnectionFactory<CommonDatabase> isolateConnectionFactory() {
     return database.isolateConnectionFactory();
   }
 
@@ -309,10 +309,10 @@ mixin PowerSyncDatabaseMixin implements SqliteConnection {
       final row = await getOptional(
           'SELECT SUM(cast(data as blob) + 20) as size, count(*) as count FROM ps_crud');
       return UploadQueueStats(
-          count: row?['count'] ?? 0, size: row?['size'] ?? 0);
+          count: row?['count'] as int? ?? 0, size: row?['size'] as int? ?? 0);
     } else {
       final row = await getOptional('SELECT count(*) as count FROM ps_crud');
-      return UploadQueueStats(count: row?['count'] ?? 0);
+      return UploadQueueStats(count: row?['count'] as int? ?? 0);
     }
   }
 
@@ -331,7 +331,7 @@ mixin PowerSyncDatabaseMixin implements SqliteConnection {
   /// This method does include transaction ids in the result, but does not group
   /// data by transaction. One batch may contain data from multiple transactions,
   /// and a single transaction may be split over multiple batches.
-  Future<CrudBatch?> getCrudBatch({limit = 100}) async {
+  Future<CrudBatch?> getCrudBatch({int limit = 100}) async {
     final rows = await getAll(
         'SELECT id, tx_id, data FROM ps_crud ORDER BY id ASC LIMIT ?',
         [limit + 1]);
@@ -384,7 +384,7 @@ mixin PowerSyncDatabaseMixin implements SqliteConnection {
       if (first == null) {
         return null;
       }
-      final int? txId = first['tx_id'];
+      final txId = first['tx_id'] as int?;
       List<CrudEntry> all;
       if (txId == null) {
         all = [CrudEntry.fromRow(first)];

packages/powersync_core/lib/src/exceptions.dart

@@ -87,8 +87,8 @@ String? _stringOrFirst(Object? details) {
     return null;
   } else if (details is String) {
     return details;
-  } else if (details is List && details[0] is String) {
-    return details[0];
+  } else if (details case [final String first, ...]) {
+    return first;
   } else {
     return null;
   }