Merge pull request #1037 from ninoseki/renew-parallel

Renew parallel

docs/analyzers/index.md

@@ -95,7 +95,7 @@ queries:
 
 ### Parallel
 
-`parallel` (`bool`) controls whether to do the parallel query execution or not. Optional. Defaults to `false`. Configurable via `PARALLEL` environment variable.
+`parallel` (`bool`) controls whether to allow parallel execution or not. Optional. Defaults to `false`. Configurable via `PARALLEL` environment variable.
 
 ### Pagination Interval
 

docs/emitters/index.md

@@ -17,6 +17,7 @@ options:
   retry_times: ...
   retry_interval: ...
   retry_exponential_backoff: ...
+  parallel: ...
 ```
 
 ### Timeout
@@ -34,3 +35,7 @@ options:
 ### Retry Exponential Backoff
 
 `retry_exponential_backoff` (`bool`) controls whether to do exponential backoff. Optional. Defaults to `true`. Configurable via `RETRY_EXPONENTIAL_BACKOFF` environment variable.
+
+### Parallel
+
+`parallel` (`bool`) controls whether to allow parallel execution or not. Optional. Defaults to `false`. Configurable via `PARALLEL` environment variable.

docs/enrichers/index.md

@@ -16,6 +16,7 @@ options:
   retry_times: ...
   retry_interval: ...
   retry_exponential_backoff: ...
+  parallel: ...
 ```
 
 ### Timeout
@@ -33,3 +34,7 @@ options:
 ### Retry Exponential Backoff
 
 `retry_exponential_backoff` (`bool`) controls whether to do exponential backoff. Optional. Defaults to `true`. Configurable via `RETRY_EXPONENTIAL_BACKOFF` environment variable.
+
+### Parallel
+
+`parallel` (`bool`) controls whether to allow parallel execution or not. Optional. Defaults to `false`. Configurable via `PARALLEL` environment variable.

lib/mihari/actor.rb

@@ -50,6 +50,13 @@ def timeout
       options[:timeout]
     end
 
+    #
+    # @return [Boolean]
+    #
+    def parallel?
+      options[:parallel] || Mihari.config.parallel
+    end
+
     def validate_configuration!
       return if configured?
 

lib/mihari/analyzers/base.rb

@@ -40,13 +40,6 @@ def ignore_error?
         options[:ignore_error] || Mihari.config.ignore_error
       end
 
-      #
-      # @return [Boolean]
-      #
-      def parallel?
-        options[:parallel] || Mihari.config.parallel
-      end
-
       # @return [Array<String>, Array<Mihari::Models::Artifact>]
       def artifacts
         raise NotImplementedError, "You must implement #{self.class}##{__method__}"

lib/mihari/enrichers/google_public_dns.rb

@@ -12,6 +12,8 @@ class GooglePublicDNS < Base
       # @return [Mihari::Models::Artifact]
       #
       def call(artifact)
+        return if artifact.domain.nil?
+
         res = client.query_all(artifact.domain)
 
         artifact.tap do |tapped|

lib/mihari/enrichers/whois.rb

@@ -15,10 +15,21 @@ class Whois < Base
       #
       # @param [Mihari::Models::Artifact] artifact
       #
-      # @return [Mihari::Models::WhoisRecord, nil]
-      #
       def call(artifact)
-        artifact.whois_record ||= memoized_call(PublicSuffix.domain(artifact.domain))
+        return if artifact.domain.nil?
+
+        domain = PublicSuffix.domain(artifact.domain)
+        record = memoized_lookup(domain)
+        return if record.parser.available?
+
+        artifact.whois_record ||= Models::WhoisRecord.new(
+          domain: domain,
+          created_on: get_created_on(record.parser),
+          updated_on: get_updated_on(record.parser),
+          expires_on: get_expires_on(record.parser),
+          registrar: get_registrar(record.parser),
+          contacts: get_contacts(record.parser)
+        )
       end
 
       private
@@ -41,21 +52,10 @@ def supported_data_types
       #
       # @return [Mihari::Models::WhoisRecord, nil]
       #
-      def memoized_call(domain)
-        record = whois.lookup(domain)
-        parser = record.parser
-        return nil if parser.available?
-
-        Models::WhoisRecord.new(
-          domain: domain,
-          created_on: get_created_on(parser),
-          updated_on: get_updated_on(parser),
-          expires_on: get_expires_on(parser),
-          registrar: get_registrar(parser),
-          contacts: get_contacts(parser)
-        )
+      def memoized_lookup(domain)
+        whois.lookup domain
       end
-      memo_wise :memoized_call
+      memo_wise :memoized_lookup
 
       #
       # @return [::Whois::Client]

lib/mihari/models/artifact.rb

@@ -190,7 +190,8 @@ def domain
         when "domain"
           data
         when "url"
-          Addressable::URI.parse(data).host
+          host = Addressable::URI.parse(data).host
+          (DataType.type(host) == "ip") ? nil : host
         end
       end
 

lib/mihari/rule.rb

@@ -174,11 +174,10 @@ def unique_artifacts
     # @return [Array<Mihari::Models::Artifact>]
     #
     def enriched_artifacts
-      # TODO: same whois query can be issued multiple times
-      @enriched_artifacts ||= Parallel.map(unique_artifacts) do |artifact|
-        enrichers.each do |enricher|
-          enricher.result(artifact) if enricher.callable?(artifact)
-        end
+      @enriched_artifacts ||= unique_artifacts.map do |artifact|
+        serial_enrichers.each { |enricher| enricher.result(artifact) }
+        Parallel.each(parallel_enrichers) { |enricher| enricher.result(artifact) }
+
         artifact
       end
     end
@@ -191,7 +190,10 @@ def enriched_artifacts
     def bulk_emit
       return [] if enriched_artifacts.empty?
 
-      Parallel.map(emitters) { |emitter| emitter.result(enriched_artifacts).value_or nil }.compact
+      [].tap do |out|
+        out << serial_emitters.map { |emitter| emitter.result(enriched_artifacts).value_or(nil) }
+        out << Parallel.map(parallel_emitters) { |emitter| emitter.result(enriched_artifacts).value_or(nil) }
+      end.flatten.compact
     end
 
     #
@@ -368,6 +370,14 @@ def emitters
       end
     end
 
+    def parallel_emitters
+      emitters.select(&:parallel?)
+    end
+
+    def serial_emitters
+      emitters.reject(&:parallel?)
+    end
+
     #
     # Get enricher class
     #
@@ -394,6 +404,14 @@ def enrichers
       end
     end
 
+    def parallel_enrichers
+      enrichers.select(&:parallel?)
+    end
+
+    def serial_enrichers
+      enrichers.reject(&:parallel?)
+    end
+
     #
     # Validate the data format
     #

lib/mihari/schemas/options.rb

@@ -7,17 +7,14 @@ module Schemas
       optional(:retry_interval).value(:integer).default(Mihari.config.retry_interval)
       optional(:retry_exponential_backoff).value(:bool).default(Mihari.config.retry_exponential_backoff)
       optional(:timeout).value(:integer)
+      optional(:parallel).value(:bool).default(Mihari.config.parallel)
     end
 
     IgnoreErrorOptions = Dry::Schema.Params do
       optional(:ignore_error).value(:bool).default(Mihari.config.ignore_error)
     end
 
-    ParallelOptions = Dry::Schema.Params do
-      optional(:parallel).value(:bool).default(Mihari.config.parallel)
-    end
-
-    AnalyzerOptions = Options | IgnoreErrorOptions | ParallelOptions
+    AnalyzerOptions = Options | IgnoreErrorOptions
 
     PaginationOptions = Dry::Schema.Params do
       optional(:pagination_interval).value(:integer).default(Mihari.config.pagination_interval)

spec/commands/search_spec.rb

@@ -13,7 +13,6 @@ class SearchCLI < Mihari::CLI::Base
 
   before do
     allow(rule).to receive(:enrichers).and_return([])
-    allow(Parallel).to receive(:processor_count).and_return(0)
   end
 
   describe "#search" do

spec/rule_spec.rb

@@ -39,10 +39,6 @@
     [artifact, artifact]
   end
 
-  before do
-    allow(Parallel).to receive(:processor_count).and_return(0)
-  end
-
   describe "#model" do
     it "returns a model" do
       expect(rule.model).to be_a Mihari::Models::Rule