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_compound_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_compound_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"]
159 ddl = DDL.new(pluginname, :data)
161 # parses numbers and booleans entered as strings into proper
162 # types of data so that DDL validation will pass
163 statement["fstatement"]["params"] = Data.ddl_transform_input(ddl, statement["fstatement"]["params"])
165 Data.ddl_validate(ddl, statement["fstatement"]["params"])
167 unless value && Data.ddl_has_output?(ddl, value)
168 DDL.validation_fail!(:PLMC41, "Data plugin '%{functionname}()' does not return a '%{value}' value", :error, {:functionname => functionname, :value => value})
176 raise "Cannot decode message type #{type}" unless [:request, :reply].include?(type)
178 @payload = PluginManager["security_plugin"].decodemsg(self)
181 raise 'callerid in request is not valid, surpressing reply to potentially forged request' unless PluginManager["security_plugin"].valid_callerid?(payload[:callerid])
184 [:collective, :agent, :filter, :requestid, :ttl, :msgtime].each do |prop|
185 instance_variable_set("@#{prop}", payload[prop]) if payload.include?(prop)
189 # Perform validation against the message by checking filters and ttl
191 raise "Can only validate request messages" unless type == :request
193 msg_age = Time.now.utc.to_i - msgtime
197 cid += payload[:callerid] + "@" if payload.include?(:callerid)
198 cid += payload[:senderid]
201 PluginManager["global_stats"].ttlexpired
203 raise(MsgTTLExpired, "message #{requestid} from #{cid} created at #{msgtime} is #{msg_age} seconds old, TTL is #{ttl}")
207 raise(NotTargettedAtUs, "Received message is not targetted to us") unless PluginManager["security_plugin"].validate_filter?(payload[:filter])
212 # publish a reply message by creating a target name and sending it
214 # If we've been specificaly told about hosts that were discovered
215 # use that information to do P2P calls if appropriate else just
217 config = Config.instance
218 if @discovered_hosts && config.direct_addressing && (@discovered_hosts.size <= config.direct_addressing_threshold)
219 self.type = :direct_request
220 Log.debug("Handling #{requestid} as a direct request")
223 PluginManager['connector_plugin'].publish(self)
227 # we gsub out the -s so that the format of the id does not
228 # change from previous versions, these should just be more
229 # unique than previous ones
230 SSL.uuid.gsub("-", "")