refactor(driver_store): Add checks to ensure validity of existing + d…

docker-compose.yml

@@ -50,7 +50,13 @@ services:
       # Environment
       GITHUB_ACTION: ${GITHUB_ACTION:-}
       # Service Hosts
-      <<: [*redis-client-env, *postgresdb-client-env,*deployment-env, *build-api-env]
+      <<:
+        [
+          *redis-client-env,
+          *postgresdb-client-env,
+          *deployment-env,
+          *build-api-env,
+        ]
 
   redis:
     image: eqalpha/keydb
@@ -99,7 +105,7 @@ services:
       - 9000:9000
       - 9090:9090
     environment:
-      << : *s3-client-env
+      <<: *s3-client-env
       MINIO_ROOT_USER: $AWS_KEY
       MINIO_ROOT_PASSWORD: $AWS_SECRET
     command: server /data --console-address ":9090"
@@ -109,11 +115,11 @@ services:
     depends_on:
       - minio
     environment:
-      << : *s3-client-env
+      <<: *s3-client-env
     entrypoint: >
       sh -c '
       sleep 3 &&
-      mc config host add s3 $AWS_S3_ENDPOINT $AWS_KEY $AWS_SECRET &&
+      mc alias set s3 $AWS_S3_ENDPOINT $AWS_KEY $AWS_SECRET &&
       mc mb -p s3/$AWS_S3_BUCKET &&
       exit 0
       '

spec/helper.cr

@@ -55,7 +55,7 @@ macro around_suite(block)
   end
 end
 
-around_suite ->{
+around_suite -> {
   clear_tables
 }
 

src/api/chaos.cr

@@ -14,7 +14,7 @@ module PlaceOS::Core::Api
       @[AC::Param::Info(name: path, description: "the driver executable name", example: "drivers_place_meet_c54390a")]
       driver_key : String,
       @[AC::Param::Info(description: "optionally provide the edge id the driver is running on", example: "edge-12345")]
-      edge_id : String? = nil
+      edge_id : String? = nil,
     ) : Nil
       raise Error::NotFound.new("no process manager found for #{driver_key}") unless manager = module_manager.process_manager(driver_key, edge_id)
       manager.kill(driver_key)

src/api/command.cr

@@ -12,7 +12,7 @@ module PlaceOS::Core::Api
     @[AC::Route::POST("/:module_id/load")]
     def load(
       @[AC::Param::Info(description: "the module id we want to load", example: "mod-1234")]
-      module_id : String
+      module_id : String,
     ) : Nil
       mod = Model::Module.find(module_id)
       raise Error::NotFound.new("module #{module_id} not found in database") unless mod
@@ -25,7 +25,7 @@ module PlaceOS::Core::Api
       @[AC::Param::Info(description: "the module id we want to send an execute request to", example: "mod-1234")]
       module_id : String,
       @[AC::Param::Info(description: "the user context for the execution", example: "user-1234")]
-      user_id : String? = nil
+      user_id : String? = nil,
     ) : Nil
       unless module_manager.process_manager(module_id, &.module_loaded?(module_id))
         Log.info { {module_id: module_id, message: "module not loaded"} }
@@ -62,7 +62,7 @@ module PlaceOS::Core::Api
     def module_debugger(
       socket,
       @[AC::Param::Info(description: "the module we want to debug", example: "mod-1234")]
-      module_id : String
+      module_id : String,
     ) : Nil
       # Forward debug messages to the websocket
       module_manager.process_manager(module_id, &.attach_debugger(module_id, socket))

src/api/drivers.cr

@@ -16,7 +16,7 @@ module PlaceOS::Core::Api
       @[AC::Param::Info(description: "the commit hash of the driver to check is compiled", example: "e901494")]
       commit : String,
       @[AC::Param::Info(description: "the driver database id", example: "driver-GFEaAlJB5")]
-      tag : String
+      tag : String,
     ) : Bool
       driver = Model::Driver.find!(tag)
       repository = driver.repository!
@@ -31,7 +31,7 @@ module PlaceOS::Core::Api
       @[AC::Param::Info(description: "the commit hash of the driver to check is compiled", example: "e901494")]
       commit : String,
       @[AC::Param::Info(description: "the driver database id", example: "driver-GFEaAlJB5")]
-      tag : String
+      tag : String,
     ) : String
       driver = Model::Driver.find!(tag)
       repository = driver.repository!
@@ -47,7 +47,7 @@ module PlaceOS::Core::Api
     @[AC::Route::POST("/:driver_id/reload")]
     def reload(
       @[AC::Param::Info(description: "the driver database id", example: "driver-GFEaAlJB5")]
-      driver_id : String
+      driver_id : String,
     ) : String
       result = store.reload_driver(driver_id)
       render status: result[:status], text: result[:message]
@@ -63,7 +63,7 @@ module PlaceOS::Core::Api
       @[AC::Param::Info(description: "the commit hash of the driver to be built", example: "e901494")]
       commit : String,
       @[AC::Param::Info(description: "the branch of the repository", example: "main")]
-      branch : String = "master"
+      branch : String = "master",
     ) : Nil
       Log.context.set(driver: driver_file, repository: repository, commit: commit, branch: branch)
       repo = Model::Repository.find!(repository)

src/api/edge.cr

@@ -13,7 +13,7 @@ module PlaceOS::Core::Api
     def edge_control(
       socket,
       @[AC::Param::Info(description: "the edge this device is handling", example: "edge-1234")]
-      edge_id : String
+      edge_id : String,
     ) : Nil
       module_manager.manage_edge(edge_id, socket)
     end

src/api/status.cr

@@ -42,7 +42,7 @@ module PlaceOS::Core::Api
     @[AC::Route::GET("/driver")]
     def driver(
       @[AC::Param::Info(name: "path", description: "the path of the compiled driver", example: "/path/to/compiled_driver")]
-      driver_path : String
+      driver_path : String,
     ) : DriverStatus
       DriverStatus.new(
         local: module_manager.local_processes.driver_status(driver_path),

src/constants.cr

@@ -23,8 +23,8 @@ module PlaceOS::Core
   # In k8s we can grab the Pod information from the environment
   # https://kubernetes.io/docs/tasks/inject-data-application/environment-variable-expose-pod-information/#use-pod-fields-as-values-for-environment-variables
   CORE_HOST_RAW = ENV["CORE_HOST"]? || System.hostname
-  CORE_HOST = !CORE_HOST_RAW.starts_with?('[') && CORE_HOST_RAW.includes?(':') ? "[#{CORE_HOST_RAW}]" : CORE_HOST_RAW
-  CORE_PORT = (ENV["CORE_PORT"]? || "3000").to_i
+  CORE_HOST     = !CORE_HOST_RAW.starts_with?('[') && CORE_HOST_RAW.includes?(':') ? "[#{CORE_HOST_RAW}]" : CORE_HOST_RAW
+  CORE_PORT     = (ENV["CORE_PORT"]? || "3000").to_i
 
   PROD = ENV["SG_ENV"]?.try(&.downcase) == "production"
 

src/placeos-core/driver_manager/driver_store.cr

@@ -18,7 +18,17 @@ module PlaceOS::Core
     def compiled?(file_name : String, commit : String, branch : String, uri : String) : Bool
       Log.debug { {message: "Checking whether driver is compiled or not?", driver: file_name, commit: commit, branch: branch, repo: uri} }
       path = Path[binary_path, executable_name(file_name, commit)]
-      return true if File.exists?(path)
+
+      if File.exists?(path)
+        # Validate that the local file is a valid executable by running it with -h
+        if validate_binary(path)
+          return true
+        else
+          Log.warn { {message: "Local binary exists but is corrupted, removing and re-downloading", driver_file: file_name, path: path.to_s} }
+          File.delete(path) rescue nil
+        end
+      end
+
       resp = BuildApi.compiled?(file_name, commit, branch, uri)
       return false unless resp.success?
       ret = fetch_binary(LinkData.from_json(resp.body)) rescue nil
@@ -90,6 +100,16 @@ module PlaceOS::Core
       {driver_source, commit, Core::ARCH}.join("_").downcase
     end
 
+    private def validate_binary(path : Path) : Bool
+      # Try to execute the binary with -h flag to validate it's a working executable
+      result = Process.run(path.to_s, ["-h"], output: Process::Redirect::Close, error: Process::Redirect::Close)
+      # If the process runs without crashing, consider it valid
+      result.exit_code == 0
+    rescue ex : Exception
+      Log.error(exception: ex) { {message: "Driver binary validation failed", path: path.to_s} }
+      false
+    end
+
     def reload_driver(driver_id : String)
       if driver = Model::Driver.find?(driver_id)
         repo = driver.repository!
@@ -120,16 +140,27 @@ module PlaceOS::Core
                ConnectProxy::HTTPClient.new(url.host.not_nil!, 9000).get(uri.to_s)
              end
       if resp.success?
-        unless link.size == resp.headers.fetch("Content-Length", "0").to_i
-          Log.error { {message: "Expected content length #{link.size}, but received #{resp.headers.fetch("Content-Length", "0")}", driver_file: driver_file} }
+        # Check Content-Length header first if available
+        content_length = resp.headers.fetch("Content-Length", "0").to_i64
+        if content_length > 0 && link.size != content_length
+          Log.error { {message: "Expected content length #{link.size}, but received #{content_length}", driver_file: driver_file} }
           raise Error.new("Response size doesn't match with build service returned result")
         end
 
         body_io = IO::Digest.new(resp.body_io? || IO::Memory.new(resp.body), Digest::MD5.new)
+        bytes_written = 0_i64
         File.open(filename, "wb+") do |f|
-          IO.copy(body_io, f)
+          bytes_written = IO.copy(body_io, f)
           f.chmod(0o755)
         end
+
+        # Verify actual downloaded size matches expected size
+        unless link.size == bytes_written
+          Log.error { {message: "Expected download size #{link.size}, but actually downloaded #{bytes_written} bytes", driver_file: driver_file} }
+          File.delete(filename) if File.exists?(filename)
+          raise Error.new("Downloaded size doesn't match expected size from build service")
+        end
+
         filename.to_s
       else
         raise Error.new("Unable to fetch driver. Error : #{resp.body}")

src/placeos-core/mappings/control_system_modules.cr

@@ -12,7 +12,7 @@ module PlaceOS::Core
 
     def initialize(
       @startup : Bool = true,
-      @module_manager : ModuleManager = ModuleManager.instance
+      @module_manager : ModuleManager = ModuleManager.instance,
     )
       super()
     end
@@ -28,7 +28,7 @@ module PlaceOS::Core
     def self.update_mapping(
       system : Model::ControlSystem,
       startup : Bool = false,
-      module_manager : ModuleManager = ModuleManager.instance
+      module_manager : ModuleManager = ModuleManager.instance,
     ) : Resource::Result
       relevant_node = startup || module_manager.discovery.own_node?(system.id.as(String))
       unless relevant_node
@@ -57,7 +57,7 @@ module PlaceOS::Core
     #
     def self.update_logic_modules(
       system : Model::ControlSystem,
-      module_manager : ModuleManager = ModuleManager.instance
+      module_manager : ModuleManager = ModuleManager.instance,
     ) : Int32
       return 0 if system.destroyed?
 
@@ -90,7 +90,7 @@ module PlaceOS::Core
     # Pass module_id and updated_name to overrride a lookup
     def self.set_mappings(
       control_system : Model::ControlSystem,
-      mod : Model::Module?
+      mod : Model::Module?,
     ) : Hash(String, String)
       system_id = control_system.id.as(String)
       storage = Driver::RedisStorage.new(system_id, "system")

src/placeos-core/mappings/module_names.cr

@@ -25,7 +25,7 @@ module PlaceOS::Core
 
     def self.update_module_mapping(
       mod : Model::Module,
-      module_manager : ModuleManager = ModuleManager.instance
+      module_manager : ModuleManager = ModuleManager.instance,
     ) : Resource::Result
       module_id = mod.id.as(String)
       # Only consider name change events

src/placeos-core/module_manager.cr

@@ -60,7 +60,7 @@ module PlaceOS::Core
     # - once load complete, mark in etcd that load is complete
     def initialize(
       uri : String | URI,
-      clustering : Clustering? = nil
+      clustering : Clustering? = nil,
     )
       @uri = uri.is_a?(URI) ? uri : URI.parse(uri)
       ModuleManager.uri = @uri

src/placeos-core/resource_manager.cr

@@ -32,7 +32,7 @@ module PlaceOS::Core
       @module_names : Mappings::ModuleNames = Mappings::ModuleNames.new,
       @settings_updates : SettingsUpdate = SettingsUpdate.new,
       @driver_module_names : Mappings::DriverModuleNames = Mappings::DriverModuleNames.new,
-      testing : Bool = false
+      testing : Bool = false,
     )
     end
 

src/placeos-core/settings_update.cr

@@ -24,7 +24,7 @@ module PlaceOS
 
     def self.update_modules(
       settings : Model::Settings,
-      module_manager : ModuleManager
+      module_manager : ModuleManager,
     )
       Log.context.set(settings_id: settings.id)
       result = Result::Success

src/placeos-edge/client.cr

@@ -49,7 +49,7 @@ module PlaceOS::Edge
       @sequence_id : UInt64 = 0,
       @skip_handshake : Bool = false,
       @ping : Bool = true,
-      @store = Core::DriverStore.new
+      @store = Core::DriverStore.new,
     )
       @secret = if secret && secret.presence
                   secret
@@ -83,7 +83,7 @@ module PlaceOS::Edge
           end
           nil
         },
-        on_connect: ->{
+        on_connect: -> {
           handshake unless skip_handshake?
           nil
         }

src/placeos-edge/protocol.cr

@@ -21,8 +21,8 @@ module PlaceOS::Edge::Protocol
     field sequence_id : UInt64
 
     field status : Status = Status::Success
-    field length : Int32, value: ->{ key.bytesize }
-    field key : String, length: ->{ length }
+    field length : Int32, value: -> { key.bytesize }
+    field key : String, length: -> { length }
 
     # Keep a reference to the remainder of the message
     protected setter binary : IO
@@ -407,7 +407,7 @@ module PlaceOS::Edge::Protocol
         @remove_drivers = [] of String,
         @add_modules = [] of Module,
         @remove_modules = [] of String,
-        @running_modules = [] of Tuple(String, String)
+        @running_modules = [] of Tuple(String, String),
       )
       end
     end