2 # container for a message, its headers, agent, collective and other meta data
4 attr_reader :message, :request, :validated, :msgtime, :payload, :type, :expected_msgid, :reply_to
5 attr_accessor :headers, :agent, :collective, :filter
6 attr_accessor :requestid, :discovered_hosts, :options, :ttl
8 VALIDTYPES = [:message, :request, :direct_request, :reply]
10 # payload - the message body without headers etc, just the text
11 # message - the original message received from the middleware
12 # options[:base64] - if the body base64 encoded?
13 # options[:agent] - the agent the message is for/from
14 # options[:collective] - the collective its for/from
15 # options[:headers] - the message headers
16 # options[:type] - an indicator about the type of message, :message, :request, :direct_request or :reply
17 # options[:request] - if this is a reply this should old the message we are replying to
18 # options[:filter] - for requests, the filter to encode into the message
19 # options[:options] - the normal client options hash
20 # options[:ttl] - the maximum amount of seconds this message can be valid for
21 # options[:expected_msgid] - in the case of replies this is the msgid it is expecting in the replies
22 # options[:requestid] - specific request id to use else one will be generated
23 def initialize(payload, message, options = {})
24 options = {:base64 => false,
29 :filter => Util.empty_filter,
32 :expected_msgid => nil,
34 :collective => nil}.merge(options)
38 @requestid = options[:requestid]
39 @discovered_hosts = nil
42 @type = options[:type]
43 @headers = options[:headers]
44 @base64 = options[:base64]
45 @filter = options[:filter]
46 @expected_msgid = options[:expected_msgid]
47 @options = options[:options]
49 @ttl = @options[:ttl] || Config.instance.ttl
55 @request = options[:request]
56 @agent = request.agent
57 @collective = request.collective
60 @agent = options[:agent]
61 @collective = options[:collective]
67 # Sets the message type to one of the known types. In the case of :direct_request
68 # the list of hosts to communicate with should have been set with #discovered_hosts
69 # else an exception will be raised. This is for extra security, we never accidentally
70 # want to send a direct request without a list of hosts or something weird like that
71 # as it might result in a filterless broadcast being sent.
73 # Additionally you simply cannot set :direct_request if direct_addressing was not enabled
74 # this is to force a workflow that doesnt not yield in a mistake when someone might assume
75 # direct_addressing is enabled when its not.
77 raise "Unknown message type #{type}" unless VALIDTYPES.include?(type)
79 if type == :direct_request
80 raise "Direct requests is not enabled using the direct_addressing config option" unless Config.instance.direct_addressing
82 unless @discovered_hosts && !@discovered_hosts.empty?
83 raise "Can only set type to :direct_request if discovered_hosts have been set"
86 # clear out the filter, custom discovery sources might interpret the filters
87 # different than the remote mcollectived and in directed mode really the only
88 # filter that matters is the agent filter
89 @filter = Util.empty_filter
90 @filter["agent"] << @agent
96 # Sets a custom reply-to target for requests. The connector plugin should inspect this
97 # when constructing requests and set this header ensuring replies will go to the custom target
98 # otherwise the connector should just do what it usually does
100 raise "Custom reply targets can only be set on requests" unless [:request, :direct_request].include?(@type)
105 # in the case of reply messages we are expecting replies to a previously
106 # created message. This stores a hint to that previously sent message id
107 # and can be used by other classes like the security plugins as a means
108 # of optimizing their behavior like by ignoring messages not directed
110 def expected_msgid=(msgid)
111 raise "Can only store the expected msgid for reply messages" unless @type == :reply
112 @expected_msgid = msgid
116 return unless @base64
118 @payload = SSL.base64_decode(@payload)
125 @payload = SSL.base64_encode(@payload)
136 raise "Cannot encode a reply message if no request has been associated with it" unless request
137 raise 'callerid in original request is not valid, surpressing reply to potentially forged request' unless PluginManager["security_plugin"].valid_callerid?(request.payload[:callerid])
139 @requestid = request.payload[:requestid]
140 @payload = PluginManager["security_plugin"].encodereply(agent, payload, requestid, request.payload[:callerid])
141 when :request, :direct_request
142 validate_compount_filter(@filter["compound"]) unless @filter["compound"].empty?
144 @requestid = create_reqid unless @requestid
145 @payload = PluginManager["security_plugin"].encoderequest(Config.instance.identity, payload, requestid, filter, agent, collective, ttl)
147 raise "Cannot encode #{type} messages"
151 def validate_compount_filter(compound_filter)
152 compound_filter.each do |filter|
153 filter.each do |statement|
154 if statement["fstatement"]
155 functionname = statement["fstatement"]["name"]
156 pluginname = Data.pluginname(functionname)
157 value = statement["fstatement"]["value"]
160 ddl = DDL.new(pluginname, :data)
162 raise DDLValidationError, "Could not find DDL for data plugin #{pluginname}, cannot use #{functionname}() in discovery"
165 # parses numbers and booleans entered as strings into proper
166 # types of data so that DDL validation will pass
167 statement["fstatement"]["params"] = Data.ddl_transform_input(ddl, statement["fstatement"]["params"])
169 Data.ddl_validate(ddl, statement["fstatement"]["params"])
171 unless value && Data.ddl_has_output?(ddl, value)
172 raise DDLValidationError, "#{functionname}() does not return a #{value} value"
180 raise "Cannot decode message type #{type}" unless [:request, :reply].include?(type)
182 @payload = PluginManager["security_plugin"].decodemsg(self)
185 raise 'callerid in request is not valid, surpressing reply to potentially forged request' unless PluginManager["security_plugin"].valid_callerid?(payload[:callerid])
188 [:collective, :agent, :filter, :requestid, :ttl, :msgtime].each do |prop|
189 instance_variable_set("@#{prop}", payload[prop]) if payload.include?(prop)
193 # Perform validation against the message by checking filters and ttl
195 raise "Can only validate request messages" unless type == :request
197 msg_age = Time.now.utc.to_i - msgtime
201 cid += payload[:callerid] + "@" if payload.include?(:callerid)
202 cid += payload[:senderid]
205 PluginManager["global_stats"].ttlexpired
207 raise(MsgTTLExpired, "message #{requestid} from #{cid} created at #{msgtime} is #{msg_age} seconds old, TTL is #{ttl}")
211 raise(NotTargettedAtUs, "Received message is not targetted to us") unless PluginManager["security_plugin"].validate_filter?(payload[:filter])
216 # publish a reply message by creating a target name and sending it
218 Timeout.timeout(2) do
219 # If we've been specificaly told about hosts that were discovered
220 # use that information to do P2P calls if appropriate else just
222 if @discovered_hosts && Config.instance.direct_addressing
223 if @discovered_hosts.size <= Config.instance.direct_addressing_threshold
224 self.type = :direct_request
225 Log.debug("Handling #{requestid} as a direct request")
228 PluginManager["connector_plugin"].publish(self)
230 PluginManager["connector_plugin"].publish(self)
236 # we gsub out the -s so that the format of the id does not
237 # change from previous versions, these should just be more
238 # unique than previous ones
239 SSL.uuid.gsub("-", "")