Fixed blocking issues with security scanning

Unless you provided enough Puma workers, there was a problem with HTTP
requests getting blocked. This is a similar issue as pointed out in SUSE#1353
with registry events.

The solution has been to move security scanning in the background, in a
different process. This is similar to SUSE#1353, but more polished (and if
this proves to really work we might move SUSE#1353 to this background runner
as well). Now the Tag model has two more columns: `vulnerabilities` and
`scanned`. The former contains the cached result of the latest security
check. The latter is an integer that can have three different values: 0
(scanning not performed), 1 (working on it) and 2 (scanning done). This
has also been exposed on the API, so the client side can be updated.

Signed-off-by: Miquel Sabaté Solà <[email protected]>

app/controllers/tags_controller.rb

@@ -6,9 +6,7 @@ def show
     authorize @tag
 
     @names = Tag.where(digest: @tag.digest).sort.map(&:name)
-
-    sec = ::Portus::Security.new(@tag.repository.full_name, @tag.name)
-    @vulnerabilities = sec.vulnerabilities
+    @vulnerabilities = @tag.fetch_vulnerabilities
   end
 
   # Removes all tags that match the digest of the tag with the given ID.

app/helpers/repositories_helper.rb

@@ -131,7 +131,7 @@ def name_and_link(tr, activity)
 
   # Returns if any security module is enabled
   def security_vulns_enabled?
-    ::Portus::Security.new(nil, nil).enabled?
+    ::Portus::Security.enabled?
   end
 
   # Returns true if any vulnerability is found

app/models/repository.rb

@@ -136,10 +136,13 @@ def self.add_repo(event, namespace, repo, tag)
     if repository.nil?
       repository = Repository.create(namespace: namespace, name: repo)
     elsif repository.tags.exists?(name: tag)
-      # Update digest if the given tag already exists.
+      # Update digest and status if the given tag already exists.
       id, digest = Repository.id_and_digest_from_event(event, repository.full_name)
       tag = repository.tags.find_by(name: tag)
-      tag.update_columns(image_id: id, digest: digest, updated_at: Time.current)
+      tag.update_columns(image_id:   id,
+                         digest:     digest,
+                         scanned:    Tag.statuses[:scan_none],
+                         updated_at: Time.current)
       repository.create_activity(:push, owner: actor, recipient: tag)
       return
     end

app/models/tag.rb

@@ -2,16 +2,18 @@
 #
 # Table name: tags
 #
-#  id            :integer          not null, primary key
-#  name          :string(255)      default("latest"), not null
-#  repository_id :integer          not null
-#  created_at    :datetime         not null
-#  updated_at    :datetime         not null
-#  user_id       :integer
-#  digest        :string(255)
-#  image_id      :string(255)      default("")
-#  marked        :boolean          default(FALSE)
-#  username      :string(255)
+#  id              :integer          not null, primary key
+#  name            :string(255)      default("latest"), not null
+#  repository_id   :integer          not null
+#  created_at      :datetime         not null
+#  updated_at      :datetime         not null
+#  user_id         :integer
+#  digest          :string(255)
+#  image_id        :string(255)      default("")
+#  marked          :boolean          default(FALSE)
+#  username        :string(255)
+#  scanned         :integer          default(0)
+#  vulnerabilities :text(65535)
 #
 # Indexes
 #
@@ -23,6 +25,11 @@
 # name follows the format as defined in registry/api/v2/names.go from Docker's
 # Distribution project. The default name for a tag is "latest".
 class Tag < ActiveRecord::Base
+  serialize :vulnerabilities
+
+  # NOTE: as a Hash because in Rails 5 we'll be able to pass a proper prefix.
+  enum status: { scan_none: 0, scan_working: 1, scan_done: 2 }
+
   belongs_to :repository
   belongs_to :author, class_name: "User", foreign_key: "user_id"
 
@@ -86,9 +93,11 @@ def delete_and_update!(actor)
     create_delete_activities!(actor)
   end
 
-  def vulnerabilities
-    sec = ::Portus::Security.new(repository.full_name, name)
-    sec.vulnerabilities
+  # Returns vulnerabilities if there are any available and security scanning is
+  # enabled.
+  def fetch_vulnerabilities
+    return unless ::Portus::Security.enabled?
+    vulnerabilities if scanned == Tag.statuses[:scan_done]
   end
 
   protected

bin/background.rb

@@ -0,0 +1,47 @@
+require "portus/db"
+
+#
+# First of all, wait until the database is up and running. This is useful in
+# containerized scenarios.
+#
+
+count = 0
+TIMEOUT = 90
+
+while ::Portus::DB.ping != :ready
+  if count >= TIMEOUT
+    puts "Timeout reached, exiting with error. Check the logs..."
+    exit 1
+  end
+
+  puts "Waiting for DB to be ready"
+  sleep 5
+  count += 5
+end
+
+#
+# The DB is up, now let's define the different background jobs as classes.
+#
+
+require "portus/background/security_scanning"
+
+they = [::Portus::Background::SecurityScanning.new]
+values = they.map { |v| "'#{v}'" }.join(", ")
+Rails.logger.info "Running: #{values}"
+
+#
+# Finally, we will loop infinitely like this:
+#   1. Each background job will execute its task.
+#   2. Then we will sleep until there's more work to be done.
+#
+
+SLEEP_VALUE = 10
+
+# Loop forever executing the given tasks. It will go to sleep for spans of
+# `SLEEP_VALUE` seconds, if there's nothing else to be done.
+loop do
+  they.each { |t| t.execute! if t.work? }
+  sleep SLEEP_VALUE until they.any?(&:work?)
+end
+
+exit 0

db/migrate/20171130095821_add_scanned_and_vulns_to_tag.rb

@@ -0,0 +1,6 @@
+class AddScannedAndVulnsToTag < ActiveRecord::Migration
+  def change
+    add_column :tags, :scanned, :integer, default: 0
+    add_column :tags, :vulnerabilities, :text
+  end
+end

db/schema.rb

@@ -11,7 +11,7 @@
 #
 # It's strongly recommended that you check this file into your version control system.
 
-ActiveRecord::Schema.define(version: 20171031112721) do
+ActiveRecord::Schema.define(version: 20171130095821) do
 
   create_table "activities", force: :cascade do |t|
     t.integer  "trackable_id",   limit: 4
@@ -119,15 +119,17 @@
   add_index "stars", ["user_id"], name: "index_stars_on_user_id", using: :btree
 
   create_table "tags", force: :cascade do |t|
-    t.string   "name",          limit: 255, default: "latest", null: false
-    t.integer  "repository_id", limit: 4,                      null: false
-    t.datetime "created_at",                                   null: false
-    t.datetime "updated_at",                                   null: false
-    t.integer  "user_id",       limit: 4
-    t.string   "digest",        limit: 255
-    t.string   "image_id",      limit: 255, default: ""
-    t.boolean  "marked",                    default: false
-    t.string   "username",      limit: 255
+    t.string   "name",            limit: 255,   default: "latest", null: false
+    t.integer  "repository_id",   limit: 4,                        null: false
+    t.datetime "created_at",                                       null: false
+    t.datetime "updated_at",                                       null: false
+    t.integer  "user_id",         limit: 4
+    t.string   "digest",          limit: 255
+    t.string   "image_id",        limit: 255,   default: ""
+    t.boolean  "marked",                        default: false
+    t.string   "username",        limit: 255
+    t.integer  "scanned",         limit: 4,     default: 0
+    t.text     "vulnerabilities", limit: 65535
   end
 
   add_index "tags", ["repository_id"], name: "index_tags_on_repository_id", using: :btree

lib/api/entities.rb

@@ -81,6 +81,11 @@ class Tags < Grape::Entity
       expose :digest, documentation: { type: String, desc: "The digest of the tag" }
       expose :image_id, documentation: { type: String, desc: "The internal image ID" }
       expose :created_at, :updated_at, documentation: { type: DateTime }
+      expose :scanned, documentation: {
+        type: Integer,
+        desc: "Whether vulnerabilities have been scanned or not. The values available are: " \
+              "0 (not scanned), 1 (work in progress) and 2 (scanning done)."
+      }
       expose :vulnerabilities, documentation: {
         is_array: true,
         desc:     "An array of vulnerabilities for this tag, or null if the feature is not enabled"

lib/portus/background/security_scanning.rb

@@ -0,0 +1,48 @@
+require "portus/security"
+
+module Portus
+  module Background
+    # SecurityScanning represents a task for checking vulnerabilities of tags that
+    # have not been scanned yet.
+    class SecurityScanning
+      def work?
+        ::Portus::Security.enabled? && Tag.exists?(scanned: Tag.statuses[:scan_none])
+      end
+
+      def execute!
+        Tag.where(scanned: Tag.statuses[:scan_none]).find_each do |tag|
+          # Mark as work in progress. This is important in case there is a push in
+          # progress.
+          tag.update_columns(scanned: Tag.statuses[:scan_working])
+
+          # Fetch vulnerabilities.
+          sec = ::Portus::Security.new(tag.repository.full_name, tag.name)
+          vulns = sec.vulnerabilities
+
+          # Now it's time to update the columns and mark the scanning as done. That
+          # being said, it may have happened that during the scanning process a push
+          # or a delete action has been performed against this tag. For this reason,
+          # in a transaction we will reload it and check if any of these conditions
+          # changed. If not, then we will proceed with the change.
+          ActiveRecord::Base.transaction do
+            # If the tag no longer exists, then we need to raise a Rollback
+            # exception to leave early and cleanly from the transaction.
+            begin
+              tag = tag.reload
+            rescue ActiveRecord::RecordNotFound
+              raise ActiveRecord::Rollback
+            end
+
+            if tag&.scanned != Tag.statuses[:scan_none]
+              tag.update_columns(vulnerabilities: vulns, scanned: Tag.statuses[:scan_done])
+            end
+          end
+        end
+      end
+
+      def to_s
+        "Security scanning"
+      end
+    end
+  end
+end

lib/portus/db.rb

@@ -21,7 +21,11 @@ def self.ping
     # trivial, but this gives us a nice way to test this module.
     def self.migrations?
       ActiveRecord::Base.connection
-      ActiveRecord::Base.connection.table_exists? "schema_migrations"
+      return unless ActiveRecord::Base.connection.table_exists? "schema_migrations"
+
+      # If db:migrate:status does not return a migration as "down", then all
+      # migrations are up and ready.
+      !`bundle exec rake db:migrate:status`.include?("down")
     end
 
     # Returns true if the given configured adapter is MariaDB.

lib/portus/security.rb

@@ -21,16 +21,23 @@ def initialize(repo, tag)
       BACKENDS.each { |b| @backends << b.new(repo, tag) if b.enabled? }
     end
 
-    # Returns true if security scanning is enabled, false otherwise.
-    def enabled?
+    # Returns true if there is backends that work available, false otherwise.
+    def available?
       @backends.present?
     end
 
+    # Returns true if security scanning is enabled, false otherwise.
+    def self.enabled?
+      ::Portus::Security::BACKENDS.any? do |b|
+        APP_CONFIG["security"][b.config_key]["server"].present?
+      end
+    end
+
     # Returns a hash with the results from all the backends. The results are a
     # list of hashes. If security vulnerability checking is not enabled, then
     # nil is returned.
     def vulnerabilities
-      return unless enabled?
+      return unless available?
 
       # First get all the layers composing the given image.
       client = Registry.get.client

lib/portus/security_backends/clair.rb

@@ -30,6 +30,10 @@ def vulnerabilities(params)
         layer_vulnerabilities(@layers.last)
       end
 
+      def self.config_key
+        "clair"
+      end
+
       protected
 
       # Returns an array with all the vulnerabilities found by Clair for the
@@ -45,20 +49,23 @@ def layer_vulnerabilities(digest)
           next if vulns.nil?
 
           vulns.each do |v|
-            if v && v["Name"] && !known.include?(v["Name"])
-              known << v["Name"]
-              res << v
-            end
+            name = Hash(v)["Name"]
+            next if name.blank? || known.include?(name)
+
+            known << v["Name"]
+            # Skipping some fields that we don't use and can be very long...
+            res << v.except("Metadata", "Description")
           end
         end
-        res
+        res.sort_by { |el| el["Name"] }
       end
 
       # Fetches the layer information from Clair for the given digest as a Hash.
       # If nothing could be extracted, then nil is returned.
       def fetch_layer(digest)
         # Now we fetch the vulnerabilities discovered by clair on that layer.
         uri, req = get_request("/v1/layers/#{digest}?features=false&vulnerabilities=true", "get")
+        req["Accept"] = "application/json"
         begin
           res = get_response_token(uri, req)
         rescue SocketError, Errno::ECONNREFUSED => e
@@ -67,13 +74,11 @@ def fetch_layer(digest)
         end
 
         # Parse the given response and return the result.
-        msg = JSON.parse(res.body)
         if res.code.to_i == 200
+          msg = JSON.parse(res.body)
           msg["Layer"]
         else
-          msg = error_message(msg)
-          Rails.logger.tagged("clair.get") { Rails.logger.debug "Error for '#{digest}': #{msg}" }
-          nil
+          handle_response(res, digest, "clair.get")
         end
       end
 
@@ -92,13 +97,7 @@ def post_layer(index)
           return
         end
 
-        code = res.code.to_i
-        return if code == 200 || code == 201
-
-        msg = error_message(JSON.parse(res.body))
-        Rails.logger.tagged("clair.post") do
-          Rails.logger.debug "Could not post '#{digest}': #{msg}"
-        end
+        handle_response(res, digest, "clair.post")
       end
 
       # Returns a hash that has to be used as the body of a POST request. This
@@ -122,6 +121,21 @@ def layer_body(digest, parent)
         }
       end
 
+      # Handle a response from a Clair request. The first parameter is the HTTP
+      # response itself, the `digest` holds a string with the digest ID, and
+      # finally `kind` is a string that identifies the kind of request.
+      def handle_response(response, digest, kind)
+        code = response.code.to_i
+        return if code == 200 || code == 201
+
+        msg = code == 404 ? response.body : error_message(JSON.parse(response.body))
+        Rails.logger.tagged(kind) do
+          Rails.logger.debug "Could not post '#{digest}': #{msg}"
+        end
+
+        nil
+      end
+
       # Returns a proper error message for the given JSON response.
       def error_message(msg)
         msg["Error"] && msg["Error"]["Message"] ? msg["Error"]["Message"] : msg

lib/portus/security_backends/dummy.rb

@@ -21,6 +21,10 @@ def vulnerabilities(_params)
         path = Rails.root.join("lib", "portus", "security_backends", "fixtures", DUMMY_FIXTURE)
         JSON.parse(File.read(path))
       end
+
+      def self.config_key
+        "dummy"
+      end
     end
   end
 end

lib/portus/security_backends/zypper.rb

@@ -27,6 +27,10 @@ def vulnerabilities(_params)
         end
       end
 
+      def self.config_key
+        "zypper"
+      end
+
       protected
 
       def consume_response(response)