Project

General

Profile

Download (15.4 KB) Statistics
| Branch: | Tag: | Revision:
#--
# CyborgHood, a distributed system management software.
# Copyright (c) 2009-2011 Marc Dequènes (Duck) <Duck@DuckCorp.org>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#++

require 'yaml'
require 'cyborghood/cyborg/botnet/session'
require 'cyborghood/cyborg/botnet/protocol'
require 'set'
require 'thread'


module CyborgHood
class ConversationThread
attr_reader :conversation, :name, :id, :session

def initialize(conversation, id, name)
@conversation = conversation
@name = name
@id = id

# no need for session for system thread
@session = Session.new unless name == 'system'
@next_action_id = 0

@callbacks = {}
@locks = Set.new
end

def new_message(action_code, parameters = nil, action_id = nil)
Message.new(self, action_code, parameters, action_id)
end

def next_action_id
id = @next_action_id
@next_action_id +=1
id
end

def close(notify = true)
# the system thread cannot be closed
return if name == 'system'
@conversation.protocol.send_notify_thread_closed(self) if notify
@conversation.delete_thread(self)
@session.clear
end

# convenience method
def call(*args, &callback)
@conversation.protocol.send_request_call(self, *args, &callback)
end

# convenience method
def exists?(node, &callback)
@conversation.protocol.send_request_exists(self, node, &callback)
end

# convenience method
def notify(event_name, event_info)
@conversation.protocol.send_notify_event(self, event_name, event_info)
end

def register_callback(message, callback)
@callbacks[message.action_id] = callback
end

def pop_callback(message)
cb = @callbacks[message.action_id]
yield cb if block_given?
@callbacks.delete(message.action_id)
cb
end

def lock(name)
@locks << name
end

def unlock(name)
@locks.delete(name)
end

def locked?
not @locks.empty?
end

# check for known actions only, conversation API messages and
# attempts in error are not taken into account, that's why it is
# not sufficent to deduce idleness at connection level
def idle?
@callbacks.empty? and @locks.empty?
end
end

class Message
attr_reader :conv_thread, :action_code, :action_parameters, :action_id

def initialize(conv_thread, action_code, action_parameters = nil, action_id = nil)
@conv_thread = conv_thread
@action_code = action_code
@action_parameters = action_parameters
# reply with the matching action id
# (the namespace for requests is on our side, the namespace for replies is on peer side,
# and we may end up using the same action id is a server acts as client)
@action_id = action_id

@sent = false
end

def new?
@action_id.nil?
end

def sent?
@sent
end

def send
raise CyberError.new(:unrecoverable, "bot/conversation", "Not sending twice the same message") if self.sent?
@action_id = @conv_thread.next_action_id if @action_id.nil?
@conv_thread.conversation.send_message(self)
@sent = true

# return message (convenience)
self
end

def create_reply(action_code, parameters = nil)
raise CyberError.new(:unrecoverable, "bot/conversation", "Cannot reply to a newly created message") if self.new?
self.class.new(@conv_thread, action_code, parameters, @action_id)
end

# convenience method
def pop_callback(&block)
@conv_thread.pop_callback(self, &block)
end

# convenience method
def register_callback(callback)
@conv_thread.register_callback(self, callback)
end
end

class Conversation < EventMachine::Protocols::LineAndTextProtocol
# don't rely on EventMachine's default, it may change one day
MaxLineLength = 16*1024

EOD = "\033[0J"
BOT_NAME_PATTERN = "[A-Z][a-zA-Z0-9]+"
ACTION_WORD_PATTERN = "[a-zA-Z0-9]+"
ACTION_PATTERN = "^(#{BOT_NAME_PATTERN})-([0-9]{4})-([0-9]{4})([+]?) (#{ACTION_WORD_PATTERN}( #{ACTION_WORD_PATTERN})*)$"
MAXIMUM_ERROR_COUNT = 3
MAXIMUM_LINES = 1024

attr_reader :bot, :peer_name, :peer_capabilities, :protocol, :auto_close_threads

def initialize(bot, block = nil)
@bot = bot
@comm_logic_block = block

super

@config = Config.instance

@message_send = Mutex.new
@error_count = 0
@split_data_mode = false
@split_data_message = nil
@split_data = []
@comm_stop = false

# associated conversation threads
@auto_close_threads = 60 # max idle time before closing (nil => never close threads)
@conv_threads = {}
@conv_threads_index = {}
@conv_threads_timers = {}
@conv_threads_closing = []
# thread 0 is reserved
@next_thread_id = 0
@system_thread = self.thread('system')

# post-negociation peer info
@peer_name = nil
@peer_capabilities = []

@protocol = BotProtocol.new(self)

# we don't know the peer name yet
@system_notification_name = "peer/#{identifier}/system"
@system_notification = @bot.get_channel(@system_notification_name)
@system_notification_processing = @system_notification.subscribe do |msg|
process_system_notification(msg)
end
end

def post_init
logger.info "New conversation with #{identifier}"
@protocol.send_announce_helo unless @comm_logic_block.nil?
rescue
logger.error "Conversation post_init: " + $!
connection_close
end

def receive_line(data)
return if @comm_stop or data.empty?

clear_receive_info

if data == EOD
logger.debug "Received EOD [#{identifier}]"
exit_split_mode
else
if @split_data_mode
logger.debug "Received data (split mode) [#{identifier}]: #{data}"

if @split_data.size > MAXIMUM_LINES
reply_fatal_error "overflow"
else
@split_data << data
end
else
logger.debug "Received data [#{identifier}]: #{data}"

if data =~ Regexp.new(ACTION_PATTERN)
peer_name = $1
conv_thread_id = $2.to_i
action_id = $3.to_i
flags = $4 || ""
action_code = $5

conv_thread = self.thread_by_id(conv_thread_id)
message = conv_thread.new_message(action_code, nil, action_id)

if flags.index '+'
enter_split_mode(message)
else
receive_message(message)
end
else
reply_syntax_error "bad action format"
end
end
end

check_errors
end

def receive_message(message)
logger.debug "Received message '#{message.action_code}' [#{identifier}]"

reset_idle_thread_check(message.conv_thread)

@protocol.process_received_message(message)

check_idle_thread(message.conv_thread)
end

def receive_error(msg)
logger.error "Error [#{identifier}]: #{msg}"
@error_count += 1
end

def unbind
logger.info "Conversation finished with #{identifier} (#{@peer_name})"
@bot.unregister_communication @peer_name unless @peer_name.nil?
@bot.drop_channel(@system_notification_name)
@conv_threads.each_value {|s| s.close(false) }
@conv_threads = {}
@conv_threads_index = {}
@comm_logic_block.call false unless @comm_logic_block.nil? or @protocol.negociation_ok?
end

def bye
@protocol.send_quit_leaving
end

def set_peer_info(name, capabilities)
@peer_name = name
@peer_capabilities = capabilities || []
logger.info "Peer name for #{identifier}: #{@peer_name}"
logger.info "Peer capabilities for #{identifier}: #{@peer_capabilities.join(", ")}"
end

def set_comm_ready
logger.info "Protocol negociation with '#{@peer_name}' on #{identifier} succeeded"
@bot.register_communication @peer_name, self
@comm_logic_block.call self unless @comm_logic_block.nil?
end

def set_comm_stop(peer_left = false)
@comm_stop = true
@system_notification.unsubscribe(@system_notification_processing)
yield if block_given?
peer_left ? close_connection : close_connection_after_writing
end

def set_error_status(fatal = false)
# fatal status is conservative, it cannot be canceled
@reveive_fatal_error = @reveive_fatal_error || fatal
@receive_error = true
end

# threads opened using the block variant are locked
# don't forget to unlock it when it is not needed anymore
def thread(name = 'default')
th = @conv_threads[name] || new_thread(name)

if block_given?
th.lock

yield th
else
th
end
end

def thread_by_id(id)
name = @conv_threads_index[id]
name.nil? ? new_thread("noname/#{id}", id) : @conv_threads[name]
end

def close_thread(name)
return if name == 'system'
# ignore mistakes
return unless @conv_threads.has_key? name

@conv_threads_closing << name

stop_idle_thread_check(@conv_threads[name])
@conv_threads[name].close

# if only the system thread remains, notify idleness
if @conv_threads.size == 1
@bot.get_channel('global/system') << {
:topic => 'CONVERSATION IDLE',
:peer => peer_name
}
end
end

def delete_thread(th)
@conv_threads_index.delete(th.id)
@conv_threads.delete(th.name)
@conv_threads_closing.delete(th.name)
end

def send_message(message)
raise CyberError.new(:unrecoverable, "bot/conversation", "Cannot send message without action id") if message.action_id.nil?

reset_idle_thread_check(message.conv_thread)

flags = ""
flags += "+" unless message.action_parameters.nil?

@message_send.synchronize do
send_line sprintf("%s-%04d-%04d%s %s", @bot.name, message.conv_thread.id, message.action_id, flags, message.action_code)
unless message.action_parameters.nil?
message.action_parameters.to_yaml.each_line {|l| send_line l }
send_line EOD
end
end

check_idle_thread(message.conv_thread)
end

def identifier
"#{@bot.identifier_prefix}/#{@signature}"
end

protected

def clear_receive_info
@receive_error = false
@receive_fatal_error = false
end

def send_line(msg)
return if error?

logger.debug "Sending data [#{identifier}]: #{msg}"
send_data "#{msg}\n"
end

def check_errors
@error_count += 1 if @receive_error

msg_quit = nil
if @error_count >= MAXIMUM_ERROR_COUNT
msg_quit = "too much errors, terminating"
elsif @fatal_error
msg_quit = "previous fatal error"
end

@protocol.send_quit_decline msg_quit unless msg_quit.nil?
end

def new_thread(name, id = nil)
id ||= @next_thread_id
th = ConversationThread.new(self, id, name)
@next_thread_id = [@next_thread_id, id + 1].max

@conv_threads[th.name] = th
# allow searching by id too
@conv_threads_index[th.id] = name
th
end

def reply_syntax_error(msg = nil)
logger.error "Protocol error [#{identifier}]: syntax error (#{msg})"

msg = "syntax error" + (msg ? ": " + msg : "")
@protocol.send_error_protocol(msg)
end

def reply_fatal_error(msg = nil)
logger.error "Protocol error [#{identifier}]: fatal error (#{msg})"

msg = "fatal error" + (msg ? ": " + msg : "")
@protocol.send_error_protocol(msg, true)
end

def enter_split_mode(message)
if @split_data_mode
reply_fatal_error "already in split mode"
@split_data_mode = false
@split_data_message = nil
else
logger.debug "Protocol info [#{identifier}]: entered split mode for action '#{message.action_id}'"
@split_data_mode = true
@split_data_message = message
end
@split_data = []
end

def exit_split_mode
if @split_data_mode
logger.debug "Protocol info [#{identifier}]: quit split mode for action '#{@split_data_message.action_id}'"

parameters = nil
begin
parameters = YAML.load(@split_data.join("\n"))
rescue
logger.debug @split_data.join("\n").inspect
end

if parameters.nil?
reply_syntax_error("bad parameters format")
else
message = @split_data_message.conv_thread.new_message(@split_data_message.action_code, parameters, @split_data_message.action_id)
receive_message(message)
end
else
reply_fatal_error "not in split mode"
end
@split_data_mode = false
@split_data_message = nil
@split_data = []
end

def process_system_notification(msg)
# TODO: process other things, like remote notifications
end

def reset_idle_thread_check(conv_thread)
return if @auto_close_threads.nil?
return if @conv_threads_closing.include? conv_thread.name

# ignore system thread
return if conv_thread.name == "system"

logger.debug "Thread '#{conv_thread.name}@#{@peer_name}' is working, canceling idle check"

# cancel time if exists
stop_idle_thread_check(conv_thread)
end

def stop_idle_thread_check(conv_thread)
timer = @conv_threads_timers[conv_thread.id]
EventMachine.cancel_timer(timer) unless timer.nil?
@conv_threads_timers[conv_thread.id] = nil
end

# check for idleness at connection level: not only actions done in a thread,
# but any communication through the thread (errors, protocol stuff, ...)
# that's why it is checked here and not in the ConversationThread class
def check_idle_thread(conv_thread)
return if @comm_stop
return if @auto_close_threads.nil?
return if @conv_threads_closing.include? conv_thread.name

# ignore system thread
return if conv_thread.name == "system"

# if a timer is running, do nothing
return unless @conv_threads_timers[conv_thread.id].nil?

# test for idleness
return unless conv_thread.idle?
logger.debug "Thread '#{conv_thread.name}@#{@peer_name}' is currently idle, doing another check in #{@auto_close_threads}s"

# send notification
@bot.get_channel(@system_notification_name) << {
:topic => 'THREAD IDLE',
:thread => conv_thread.name
}

# set timer for closing
@conv_threads_timers[conv_thread.id] = EventMachine.add_timer(@auto_close_threads) do
@conv_threads_timers[conv_thread.id] = nil
# if it is not idle, then do nothing, wait for another call to check_idle_thread
if conv_thread.idle?
logger.debug "Thread '#{conv_thread.name}@#{@peer_name}' is still idle, closing"
close_thread(conv_thread.name)
else
logger.debug "Thread '#{conv_thread.name}@#{@peer_name}' is no more idle"
end
end
end
end
end
(1-1/5)