Project

General

Profile

« Previous | Next » 

Revision d7ae0ab2

Added by Marc Dequènes almost 14 years ago

  • ID d7ae0ab2845854a57e2d73e943d88eff5a7051bb

[layout] reorganization

View differences:

bin/librarian
# load config before modules are included
Config.load(self.human_name)
include BotNetServer
include BotNet
def interface
LibrarianInterface.instance
bin/mapmaker
# load config before modules are included
Config.load(self.human_name)
include BotNetServer
include BotNet
def interface
MapMakerInterface.instance
bin/test_client
$: << File.join(File.dirname(__FILE__), "..", "lib")
require 'cyborghood/cyborg'
require 'cyborghood/cyborg/dsl'
module CyborgHood
lib/cyborghood/cyborg.rb
require 'cyborghood'
require 'eventmachine'
require 'cyborghood/cyborg/dsl'
module CyborgHood
......
end
autoload :BotNet, 'cyborghood/cyborg/botnet'
autoload :BotNetServer, 'cyborghood/cyborg/botnet_server'
end
lib/cyborghood/cyborg/botnet.rb
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#++
require 'cyborghood/cyborg/interface'
require 'cyborghood/cyborg/conversation'
require 'cyborghood/cyborg/botnet_dsl'
require 'cyborghood/cyborg/botnet/interface'
require 'cyborghood/cyborg/botnet/conversation'
require 'cyborghood/cyborg/botnet/dsl'
require 'set'
module CyborgHood
......
include CyborgServerRootInterfaceAddon
end
module BotNetClientUNIXSocket
def identifier_prefix
"unix_socket"
end
def setup
super
@pending_conversation_close = Set.new
end
def peer_socket(peer)
File.join(Config::RUN_DIR, peer.downcase + ".sock")
end
def contact_peer(peer, &block)
super(peer, block) do |callback|
EventMachine.connect_unix_domain(peer_socket(peer), Conversation, self, callback)
end
end
end
module BotNet
attr_reader :interface
def self.included(base)
case Config.instance.botnet.connection_type
when 'unix_socket'
return base.class_eval("include BotNetClientUNIXSocket")
return base.class_eval("include BotNetUNIXSocket")
else
raise CyberError.new(:unrecoverable, "config", "Unknown botnet connection type")
end
......
def setup
super
@comm_list = {}
@comm_list_attempt = {}
self.interface.bot = self
end
def contact_peer(peer, block)
......
@comm_list.empty? and super
end
end
autoload :BotNetUNIXSocket, 'cyborghood/cyborg/botnet/backend/unix'
end
lib/cyborghood/cyborg/botnet/backend/unix.rb
#--
# CyborgHood, a distributed system management software.
# Copyright (c) 2009-2010 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/>.
#++
module CyborgHood
module BotNetUNIXSocket
def identifier_prefix
"unix_socket"
end
def setup
super
@pending_conversation_close = Set.new
@socket = peer_socket(@config.bot_name)
at_exit { remove_socket_file }
end
def start_work
super
EventMachine.start_unix_domain_server(@socket, Conversation, self)
end
def peer_socket(peer)
File.join(Config::RUN_DIR, peer.downcase + ".sock")
end
def contact_peer(peer, &block)
super(peer, block) do |callback|
EventMachine.connect_unix_domain(peer_socket(peer), Conversation, self, callback)
end
end
# backend-specific capabilities
def capabilities
super + []
end
def remove_socket_file
File.delete(@socket) if @socket && File.exist?(@socket)
end
end
end
lib/cyborghood/cyborg/botnet/conversation.rb
#--
# CyborgHood, a distributed system management software.
# Copyright (c) 2009-2010 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'
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 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
@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
@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?
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
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 = YAML.load(@split_data.join("\n"))
reply_syntax_error("bad parameters format") if parameters.nil?
message = @split_data_message.conv_thread.new_message(@split_data_message.action_code, parameters, @split_data_message.action_id)
receive_message(message)
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
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 @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
lib/cyborghood/cyborg/botnet/dsl.rb
#--
# CyborgHood, a distributed system management software.
# Copyright (c) 2009-2010 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/>.
#++
module CyborgHood
module DSL
module BotnetTask
def ask(peer, key, cmd, *args)
_add_subtask("botnet/peer/#{peer}/out") do |subtask|
logger.debug "Task '#{@name}': Trying to contact peer '#{peer}'"
@bot.contact_peer(peer) do |conv|
if conv
logger.debug "Task '#{@name}': Peer '#{peer}' contacted, starting conversation"
# don't use the block call to leave the conversation thread open
conv_thread = conv.thread(@notification_name)
conv_thread.call(cmd, *args) do |reply|
case reply[:status]
when :ok
subtask.results = {key => reply[:result]}
when :decline
# TODO: remove this case ???
when :error
subtask.errors << reply[:exception]
end
subtask.finish
end
else
logger.debug "Task '#{@name}': Could not contact peer '#{peer}'"
subtask.errors << CyberError.new(:unrecoverable, "botnet/client/dsl", "Task '#{@name}': could not contact peer '#{peer}'")
subtask.finish
end
end
end
end
end
Task.class_eval do
include BotnetTask
end
end
end
lib/cyborghood/cyborg/botnet/interface.rb
#--
# CyborgHood, a distributed system management software.
# Copyright (c) 2009-2010 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 'singleton'
module CyborgHood
# the base mixin (not intended to be used directly, but...)
module CyborgServerInterfaceBase
NODE_PATTERN = "((?:\/|(?:\/[a-zA-Z0-9._]+)+[?=]?))"
attr_accessor :bot
def self.included(base)
base.extend(ClassMethods)
end
module ClassMethods
attr_accessor :exported_methods
attr_accessor :auto_export_public_instance_methods
def export_method(*syms)
syms = [syms] unless syms.is_a? Array
self.exported_methods ||= []
self.exported_methods += syms.collect{|m| m.to_s }
end
def unexport_method(*syms)
syms = [syms] unless syms.is_a? Array
self.exported_methods ||= []
self.exported_methods -= syms.collect{|m| m.to_s }
end
def export_parent_methods
self.export_method *self.superclass.public_instance_methods(false)
end
def is_node?(node)
(node =~ Regexp.new(NODE_PATTERN)) ? true : false
end
end
def initialize(*args)
super(*args)
@config = Config.instance
self.class.exported_methods ||= []
self.class.auto_export_public_instance_methods = true
end
# convenience method
def is_node?(node)
self.class.is_node?(node)
end
def api_klasses
list = self.class.constants.collect do |c|
cc = self.class.const_get(c)
(cc.class == Class and cc.ancestors.include? CyborgHood::CyborgServerInterfaceBase) ? [c, cc] : nil
end.compact
Hash[list]
end
def api_methods
methods = []
methods += self.class.public_instance_methods(false) if self.class.auto_export_public_instance_methods
methods -= ["initialize", "__destroy", "method_missing"]
methods &= self.methods
methods += self.class.exported_methods
end
def api_container_methods
[]
end
def api_containers
(api_klasses.keys + api_container_methods).sort
end
def api_leafs
(api_methods - api_container_methods).sort
end
def api_nodes
(api_klasses.keys + api_methods).sort
end
def find_node_action(session, node_name)
node_name.gsub!(/^\//, "")
next_node_name, other_nodes_names = node_name.split('/', 2)
next_node_klass = next_node_name.nil? ? self.class : api_klasses[next_node_name]
# inner class or method ?
if next_node_klass.nil?
# method is declared ?
if api_methods.include? next_node_name
# final node ?
if other_nodes_names.blank?
# cannot use method(), as this method may not exist at all (but still
# be usuable through metaprogramming
return lambda do |*args|
r = child_node(next_node_name, session, *args)
# dynamic tree construction: method may return a node
if r.is_a? CyborgHood::CyborgServerInterfaceBase
r.api_nodes
else
r
end
end
end
# not a container, leaving
return unless self.api_container_methods.include? next_node_name
next_node = child_node(next_node_name, session)
else
# unknown method
return
end
else
next_node = next_node_klass.instance
# final node ?
return next_node.method('api_nodes') if other_nodes_names.blank?
end
# search deeper
if next_node.is_a? CyborgHood::CyborgServerInterfaceBase
next_node.find_node_action(session, other_nodes_names)
else
# it is not a node, so there are no children
return
end
end
def child_node(next_node_name, session, *args)
args.unshift session if self.is_a? CyborgHood::CyborgServerStatefulInterface
self.send(next_node_name, *args)
end
def has_node?(cmd)
not find_node_action(nil, cmd).nil?
end
# preliminary incoming message handling
def call(session, cmd, data = nil)
action = find_node_action(session, cmd)
raise CyberError.new(:unrecoverable, 'api/cyborghood', "unknown node") if action.nil?
data ||= []
raise CyberError.new(:unrecoverable, 'api/cyborghood', "wrong format for arguments") unless data.is_a? Array
begin
action.call(*data)
rescue
logger.debug "node action error message: " + $!
logger.debug "node action error backtrace: " + $!.backtrace.join("\n")
raise CyberError.new(:unrecoverable, 'api/cyborghood', "method call failed: " + $!)
end
end
end
# structural mixins
module CyborgServerInterface
def self.included(base)
base.class_eval("include CyborgServerInterfaceBase")
base.class_eval("include Singleton")
base.extend(ClassMethods)
end
module ClassMethods
def dynamic_interface(&resource_generator)
class_eval do
class_inheritable_reader :resource_generator
def method_missing(method_name, *args)
node_name = method_name.to_s
if api_methods.include?(node_name)
self.resource_generator.call(node_name)
else
super
end
end
end
write_inheritable_attribute(:resource_generator, resource_generator)
end
end
end
module CyborgServerEmbeddedInterface
def self.included(base)
base.class_eval("include CyborgServerInterfaceBase")
base.export_parent_methods
end
end
module CyborgServerStatefulInterface
def self.included(base)
base.class_eval("include CyborgServerInterfaceBase")
base.class_eval("include Singleton")
base.extend(ClassMethods)
end
module ClassMethods
def stateful_dynamic_interface(resource_key_pattern, &resource_generator)
class_eval do
class_inheritable_reader :resource_key_pattern, :resource_generator
def method_missing(method_name, *args)
session = args.shift
node_name = method_name.to_s
if api_methods.include?(node_name)
resource_key = self.resource_key_pattern.gsub("#NODE#", node_name)
session.store.get(resource_key) { self.resource_generator.call(node_name) }
else
super
end
end
end
write_inheritable_attribute(:resource_key_pattern, resource_key_pattern)
write_inheritable_attribute(:resource_generator, resource_generator)
end
end
end
# additional mixin
module CyborgServerRootInterfaceAddon
API_VERSION = "0.1~"
def self.included(base)
list = self.public_instance_methods(false)
base.class_eval do
export_method *list
end
end
def product_name
PRODUCT
end
def product_version
VERSION
end
def api_version
API_VERSION
end
def bot_name
@bot.name
end
end
end
lib/cyborghood/cyborg/botnet/protocol.rb
#--
# CyborgHood, a distributed system management software.
# Copyright (c) 2009-2010 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/>.
#++
# Note: Event machine core is mono-thread, so this code is not threadsafe.
# Only long tasks will be run in threads, which do not care about all this.
module CyborgHood
class BotProtocol
VERSION = "0.1"
CAPABILITIES = []
@@request_callback = proc do |result|
protocol = result[:reply_message].conv_thread.conversation.protocol
protocol.process_request_result(result)
end
def initialize(conversation)
@conversation = conversation
@negociation_received = false
@negociation_sent = false
@negociation_ok = false
end
def negociation_ok?
@negociation_ok
end
def process_received_message(message)
method = "receive_" + message.action_code.downcase.tr(" ", "_")
if respond_to? method
send(method, message)
else
send_error_protocol "unknown action"
end
end
def process_request_result(result)
if result[:error]
send_error_action(result[:reply_message], result[:error])
else
send_reply_result(result[:reply_message], result[:action_result])
end
end
def receive_announce_helo(message)
unless message.conv_thread.id == 0
return send_quit_decline "bad negociation"
end
if message.action_parameters.nil?
return send_quit_decline "missing parameters"
end
unless message.action_parameters[:bot_name] =~ Regexp.new(Conversation::BOT_NAME_PATTERN)
return send_quit_decline "bad bot name"
end
unless message.action_parameters[:protocol_version] == VERSION
return send_quit_decline "protocol version does not match"
end
@negociation_received = true
@conversation.set_peer_info(message.action_parameters[:bot_name], message.action_parameters[:capabilities])
if @negociation_sent
send_announce_ok(message)
@negociation_ok = true
@conversation.set_comm_ready
else
send_announce_helo(message)
end
end
def receive_announce_ok(message)
unless @negociation_sent and @negociation_received
send_quit_decline "bad negociation"
end
@negociation_ok = true
end
def receive_request_capabilities(message)
send_reply_result(message, @conversation.bot.capabilities + CAPABILITIES)
end
def receive_request_call(message)
if message.action_parameters.nil?
return send_error_action(message, "missing parameters")
end
unless @conversation.bot.interface.is_node? message.action_parameters[:node]
return send_error_action(message, "bad node")
end
send_reply_ack(message)
@conversation.bot.schedule_task(@@request_callback) do
result = {
:reply_message => message
}
begin
result[:action_result] = @conversation.bot.interface.call(message.conv_thread.session,
message.action_parameters[:node],
message.action_parameters[:parameters])
rescue CyberError => e
result[:error] = {
:category => e.category,
:severity => e.severity,
:message => e.message
}
rescue
result[:error] = {
:category => 'unknown',
:severity => :unrecoverable,
:message => $!.to_s
}
end
result
end
end
def receive_request_exists(message)
if message.action_parameters.nil?
return send_error_action(message, "missing parameters")
end
unless @conversation.bot.interface.is_node? message.action_parameters[:node]
return send_error_action(message, "bad node")
end
send_reply_ack(message)
@conversation.bot.schedule_task(@@request_callback) do
{
:reply_message => message,
:action_result => @conversation.bot.interface.has_node?(message.action_parameters[:node])
}
end
end
def receive_request_describe(message)
# TODO: implement when ready in the interface
send_reply_decline(message, "not implemented")
end
def receive_error_protocol(message)
logger.error "received protocol error notification from '#{@conversation.peer_name}': #{message.action_parameters[:error]}"
end
def receive_error_action(message)
message.pop_callback do |cb|
if cb
error = message.action_parameters[:error]
exception = CyberError.new(error[:severity], error[:category], error[:message])
cb.call({:status => :error, :exception => exception})
else
send_error_protocol("received reply for unknown action")
end
end
end
# TODO: what if the peer close a thread i have opened ?
# send error to all actions ?
def receive_notify_thread_closed(message)
message.conv_thread.close(false)
end
def receive_notify_event(message)
@conversation.bot.get_channel("peer/#{@conversation.peer_name}/incoming") << {
:from => message.conversation.peer_id,
:topic => message.action_parameters[:name],
:info => message.action_parameters[:info]
}
end
def receive_reply_ack(message)
# TODO: cancel timeout (which does not exist yet)
end
def receive_reply_decline(message)
message.pop_callback do |cb|
if cb
cb.call({:status => :decline, :reason => message.action_parameters[:reason]})
else
send_error_protocol("received reply for unknown action")
end
end
end
def receive_reply_result(message)
message.pop_callback do |cb|
if cb
cb.call({:status => :ok, :result => message.action_parameters[:result]})
else
send_error_protocol("received reply for unknown action")
end
end
end
def receive_quit_decline(message)
logger.warning "peer '#{@conversation.peer_name}' refused more conversation: #{message.action_parameters[:reason]}"
@conversation.set_comm_stop(true)
# TODO: notify client
end
def receive_quit_leaving(message)
logger.info "peer '#{@conversation.peer_name}' is leaving"
@conversation.set_comm_stop(true)
# TODO: notify client
end
def send_announce_helo(recv_message = nil)
action_code = "ANNOUNCE HELO"
action_parameters = {
:bot_name => @conversation.bot.name,
:protocol_version => VERSION
}
message = (recv_message.nil? ? @conversation.thread('system').new_message(action_code, action_parameters) :
recv_message.create_reply(action_code, action_parameters))
message.send
@negociation_sent = true
end
def send_announce_ok(recv_message)
recv_message.create_reply("ANNOUNCE OK").send
end
def send_request_capabilities
@conversation.thread('system').new_message("REQUEST EXISTS", { :node => node }).send
end
def send_request_call(conv_thread, node, *parameters, &callback)
message = conv_thread.new_message("REQUEST CALL", { :node => node, :parameters => parameters }).send
message.register_callback(callback)
end
def send_request_exists(conv_thread, node)
message = conv_thread.new_message("REQUEST EXISTS", { :node => node }).send
message.register_callback(callback)
end
def send_request_describe(conv_thread, node)
message = conv_thread.new_message("REQUEST DESCRIBE", { :node => node }).send
message.register_callback(callback)
end
def send_error_protocol(error, fatal = false)
@conversation.thread('system').new_message("ERROR PROTOCOL", { :error => error }).send
@conversation.set_error_status(fatal)
end
def send_error_action(recv_message, error)
recv_message.create_reply("ERROR ACTION", { :error => error }).send
end
def send_notify_thread_closed(conv_thread)
conv_thread.new_message("NOTIFY THREAD CLOSED").send
end
def send_notify_event(conv_thread, event_name, event_info)
conv_thread.new_message.create_reply("NOTIFY EVENT", { :name => event_name, :info => event_info }).send
end
def send_reply_ack(recv_message)
recv_message.create_reply("REPLY ACK").send
end
def send_reply_decline(recv_message, reason)
recv_message.create_reply("REPLY DECLINE", { :reason => reason }).send
end
def send_reply_result(recv_message, result)
recv_message.create_reply("REPLY RESULT", { :result => result }).send
end
def send_quit_decline(reason)
@conversation.set_comm_stop do
@conversation.thread('system').new_message("QUIT DECLINE", { :reason => reason }).send
end
end
def send_quit_leaving
@conversation.set_comm_stop do
@conversation.thread('system').new_message("QUIT LEAVING").send
end
end
end
end
lib/cyborghood/cyborg/botnet/session.rb
#--
# CyborgHood, a distributed system management software.
# Copyright (c) 2009-2010 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/>.
#++
module CyborgHood
class Session
attr_reader :store
def initialize
@store = Store.new
end
def clear
@store.values.each do |obj|
obj.__destroy if obj.respond_to? :__destroy
end
@store.clear
end
class Store < Hash
def get(key)
obj = self[key]
if obj.nil? and block_given?
obj = yield
self[key] = obj
end
obj
end
end
end
end
lib/cyborghood/cyborg/botnet_dsl.rb
#--
# CyborgHood, a distributed system management software.
# Copyright (c) 2009-2010 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/>.
#++
module CyborgHood
module BotnetDSL
module BotnetTask
def ask(peer, key, cmd, *args)
_add_subtask("botnet/peer/#{peer}/out") do |subtask|
logger.debug "Task '#{@name}': Trying to contact peer '#{peer}'"
@bot.contact_peer(peer) do |conv|
if conv
logger.debug "Task '#{@name}': Peer '#{peer}' contacted, starting conversation"
# don't use the block call to leave the conversation thread open
conv_thread = conv.thread(@notification_name)
conv_thread.call(cmd, *args) do |reply|
case reply[:status]
when :ok
subtask.results = {key => reply[:result]}
when :decline
# TODO: remove this case ???
when :error
subtask.errors << reply[:exception]
end
subtask.finish
end
else
logger.debug "Task '#{@name}': Could not contact peer '#{peer}'"
subtask.errors << CyberError.new(:unrecoverable, "botnet/client/dsl", "Task '#{@name}': could not contact peer '#{peer}'")
subtask.finish
end
end
end
end
end
::CyborgHood::DSL::Task.class_eval do
include BotnetTask
end
end
end
lib/cyborghood/cyborg/botnet_server.rb
#--
# CyborgHood, a distributed system management software.
# Copyright (c) 2009-2010 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 'cyborghood/cyborg/botnet'
module CyborgHood
module BotNetServerUNIXSocket
def setup
... This diff was truncated because it exceeds the maximum size that can be displayed.

Also available in: Unified diff