Compare commits

...

8 Commits

Author SHA1 Message Date
Jan Vidar Krey d98f013a51 Attempt to send messages via the Operations bot as private messages. 2013-02-14 02:36:37 +01:00
Jan Vidar Krey 2be48a13f7 Made sure we now have an 'Operations' user logged in.
This is still pretty much work in progress.
2013-02-14 00:16:18 +01:00
Jan Vidar Krey 70ba9d5831 Create a bot user system. 2013-02-13 21:10:13 +01:00
Jan Vidar Krey f91f3ea68c Added funcions for sending notifications to operators.
Notifications could be compared to some sort of a log message,
but since it is broadcasted to operators live in the hub
it draws attention to them.
2013-02-13 21:10:30 +01:00
Jan Vidar Krey 5b78c0826d Added functions to route messages to operators only.
Technially the messages are routed to users with the
flag_opnotify set, but this is intended to be set only for
operators (and above) - and it allows operators to opt out
of these messages if they wish to.
2013-02-13 21:10:48 +01:00
Jan Vidar Krey 5c31f47bea Added methods for adding a string to be escaped automatically.
Previously only existed for named parameters.
2013-02-13 21:11:05 +01:00
Jan Vidar Krey 51a8e785c0 Attempt to address issues with timers overflowing (bug #198) 2013-02-13 21:11:22 +01:00
Jan Vidar Krey 8899e49f73 Added some auto-tests for bulk timeouts. 2013-02-13 21:11:39 +01:00
15 changed files with 306 additions and 24 deletions

View File

@ -761,6 +761,9 @@ int main(int argc, char** argv)
exotic_add_test(&handle, &exotic_test_timer_add_5_events_1, "timer_add_5_events_1");
exotic_add_test(&handle, &exotic_test_timer_check_5_events_1, "timer_check_5_events_1");
exotic_add_test(&handle, &exotic_test_timer_process_5_events_1, "timer_process_5_events_1");
exotic_add_test(&handle, &exotic_test_timer_clear_1, "timer_clear_1");
exotic_add_test(&handle, &exotic_test_timer_bulk_1, "timer_bulk_1");
exotic_add_test(&handle, &exotic_test_timer_bulk_2, "timer_bulk_2");
exotic_add_test(&handle, &exotic_test_tokenizer_basic_0, "tokenizer_basic_0");
exotic_add_test(&handle, &exotic_test_tokenizer_basic_1, "tokenizer_basic_1");
exotic_add_test(&handle, &exotic_test_tokenizer_basic_1a, "tokenizer_basic_1a");
@ -951,7 +954,6 @@ int exotic_initialize(struct exotic_handle* handle, int argc, char** argv)
void exotic_add_test(struct exotic_handle* handle, exo_test_t func, const char* name)
{
struct exo_test_data* test;
if (!handle)
{
fprintf(stderr, "exotic_add_test: failed, no handle!\n");

View File

@ -1,6 +1,6 @@
#include <uhub.h>
#define MAX_EVENTS 15
#define MAX_EVENTS 100
static struct timeout_queue* g_queue;
static time_t g_now;
static size_t g_max;
@ -117,3 +117,28 @@ EXO_TEST(timer_process_5_events_1,{
g_now = 4;
return timeout_queue_process(g_queue, g_now) == g_triggered;
});
EXO_TEST(timer_clear_1,{
size_t n;
for (n = 0; n < MAX_EVENTS; n++)
timeout_queue_remove(g_queue, &g_events[n]);
return timeout_queue_get_next_timeout(g_queue, g_now) == g_max;
});
EXO_TEST(timer_bulk_1,{
size_t n;
g_now = 10;
for (n = 0; n < MAX_EVENTS; n++)
{
timeout_queue_insert(g_queue, &g_events[0], 0);
}
return timeout_queue_process(g_queue, g_now) == 10;
});
EXO_TEST(timer_bulk_2,{
g_now = 110;
return timeout_queue_process(g_queue, g_now) == 90;
});

View File

@ -787,6 +787,13 @@ int adc_msg_add_argument(struct adc_message* cmd, const char* string)
return 0;
}
int adc_msg_add_argument_string(struct adc_message* cmd, const char* string)
{
char* escaped = adc_msg_escape(string);
int ret = adc_msg_add_argument(cmd, escaped);
hub_free(escaped);
return ret;
}
char* adc_msg_get_argument(struct adc_message* cmd, int offset)
{
@ -866,8 +873,6 @@ int adc_msg_get_argument_index(struct adc_message* cmd, const char prefix[2])
return -1;
}
int adc_msg_escape_length(const char* str)
{
int add = 0;

View File

@ -171,6 +171,12 @@ extern int adc_msg_replace_named_argument(struct adc_message* cmd, const char pr
*/
extern int adc_msg_add_argument(struct adc_message* cmd, const char* string);
/**
* Add an argument string, the string will be automatcally escaped.
* @return 0 if successful, or -1 if an error occured (out of memory).
*/
extern int adc_msg_add_argument_string(struct adc_message* cmd, const char* string);
/**
* Append a named argument
*
@ -209,6 +215,13 @@ extern char* adc_msg_unescape(const char* string);
*/
extern int adc_msg_unescape_to_target(const char* string, char* target, size_t target_size);
/**
* Returns the length of the string once escaped with
* adc_msg_escape().
*
* The string must be NULL terminated.
*/
extern int adc_msg_escape_length(const char* str);
/**
* Convert a string to a ADC command escaped string.
@ -234,4 +247,5 @@ void adc_msg_unterminate(struct adc_message* cmd);
*/
int adc_msg_get_arg_offset(struct adc_message* msg);
#endif /* HAVE_UHUB_COMMAND_H */

View File

@ -353,6 +353,7 @@ static int command_kick(struct command_base* cbase, struct hub_user* user, struc
}
else
{
hub_notify(cbase->hub, notify_info, "Kicking user \"%s\" (%s)", target->id.nick, user->id.nick);
cbuf_append_format(buf, "Kicking user \"%s\".", target->id.nick);
hub_disconnect_user(cbase->hub, target, quit_kicked);
}
@ -362,6 +363,7 @@ static int command_kick(struct command_base* cbase, struct hub_user* user, struc
static int command_reload(struct command_base* cbase, struct hub_user* user, struct hub_command* cmd)
{
cbase->hub->status = hub_status_restart;
hub_notify(cbase->hub, notify_info, "Reloading configuration (%s)", user->id.nick);
return command_status(cbase, user, cmd, cbuf_create_const("Reloading configuration..."));
}
@ -384,6 +386,7 @@ static int command_unload(struct command_base* cbase, struct hub_user* user, str
static int command_shutdown_hub(struct command_base* cbase, struct hub_user* user, struct hub_command* cmd)
{
hub_notify(cbase->hub, notify_warn, "Shutting down hub (%s)", user->id.nick);
cbase->hub->status = hub_status_shutdown;
return command_status(cbase, user, cmd, cbuf_create_const("Hub shutting down..."));
}

View File

@ -653,6 +653,7 @@ static struct net_connection* start_listening_socket(const char* bind_addr, uint
if (ret == -1)
{
LOG_ERROR("hub_start_service(): Unable to bind to TCP local address. errno=%d, str=%s", net_error(), net_error_string(net_error()));
hub_notify(hub, notify_error, "Unable to bind to network address %s on port %d: %s (%d)", bind_addr, port, net_error_string(net_error()), net_error());
net_close(sd);
return 0;
}
@ -671,6 +672,60 @@ static struct net_connection* start_listening_socket(const char* bind_addr, uint
return server;
}
int hub_is_running(struct hub_info* hub)
{
return hub->status == hub_status_running || hub->status == hub_status_restart;
}
void hub_notify(struct hub_info* hub, enum notify_verbosity verbosity, const char* fmt, ...)
{
struct cbuffer* buf;
struct adc_message* msg;
va_list args;
char temp[1024];
va_start(args, fmt);
vsnprintf(temp, sizeof(temp), fmt, args);
va_end(args);
buf = cbuf_create(strlen(temp) + 8);
switch (verbosity)
{
case notify_error:
cbuf_append(buf, "ERROR: ");
LOG_ERROR(temp);
break;
case notify_warn:
cbuf_append(buf, "WARN: ");
LOG_WARN(temp);
break;
case notify_info:
cbuf_append(buf, "INFO: ");
LOG_INFO(temp);
break;
case notify_debug:
cbuf_append(buf, "DEBUG: ");
LOG_DEBUG(temp);
break;
}
cbuf_append(buf, temp);
if (hub_is_running(hub))
{
msg = adc_msg_construct(ADC_CMD_IMSG, 5 + adc_msg_escape_length(cbuf_get(buf)) + 2);
adc_msg_add_argument_string(msg, cbuf_get(buf));
route_to_operators(hub, msg);
adc_msg_free(msg);
}
cbuf_destroy(buf);
}
struct server_alt_port_data
{
struct hub_info* hub;
@ -751,6 +806,54 @@ static void unload_ssl_certificates(struct hub_info* hub)
}
#endif /* SSL_SUPPORT */
// #ifdef BOT_SUPPORT
static void route_privmsg_to_operators(struct hub_user* bot, sid_t from, const char* escaped_msg, int action)
{
struct hub_info* hub = bot->hub;
struct hub_user* user = (struct hub_user*) list_get_first(hub->users->list);
while (user)
{
if (from != user->id.sid && user_flag_get(user, flag_opnotify))
{
struct adc_message* msg = adc_msg_construct_source_dest(ADC_CMD_EMSG, from, user->id.sid, strlen(escaped_msg) + (action * 4) + 7);
adc_msg_add_argument(msg, escaped_msg);
adc_msg_add_named_argument(msg, ADC_MSG_FLAG_PRIVATE, sid_to_string(bot->id.sid));
if (action) adc_msg_add_named_argument(msg, ADC_MSG_FLAG_ACTION, "1");
route_to_user(hub, user, msg);
adc_msg_free(msg);
}
user = (struct hub_user*) list_get_next(hub->users->list);
}
}
/// This receives private messages and transmits them to the connected operators.
static void hub_bot_op_notify_handle(struct hub_user* bot, struct adc_message* msg)
{
char* chat;
LOG_TRACE("Invoked hub_bot_op_notify_handle()");
switch (msg->cmd)
{
case ADC_CMD_EMSG:
case ADC_CMD_DMSG:
chat = adc_msg_get_argument(msg, 0);
LOG_DEBUG("Hub chat: \"%s\"", chat);
route_privmsg_to_operators(bot, msg->source, chat, adc_msg_has_named_argument(msg, ADC_MSG_FLAG_ACTION) ? 1 : 0);
hub_free(chat);
break;
default:
/* ignore these messages! */
break;
}
}
static void hub_bot_op_notify_create(struct hub_info* hub)
{
struct hub_user* opcom = user_create_bot(hub, "Operations", "Hub operators", hub_bot_op_notify_handle);
uman_add(hub->users, opcom);
}
// #endif
struct hub_info* hub_start_service(struct hub_config* config)
{
struct hub_info* hub = 0;
@ -834,6 +937,8 @@ struct hub_info* hub_start_service(struct hub_config* config)
// Start the hub command sub-system
hub->commands = command_initialize(hub);
hub_bot_op_notify_create(hub);
return hub;
}

View File

@ -106,7 +106,7 @@ struct hub_info
struct adc_message* command_support; /* The hub's SUP command */
struct adc_message* command_banner; /* The default welcome message */
time_t tm_started;
int status;
enum hub_state status;
char* recvbuf; /* Global receive buffer */
char* sendbuf; /* Global send buffer */
@ -233,6 +233,20 @@ extern void hub_set_variables(struct hub_info* hub, struct acl_handle* acl);
*/
extern void hub_free_variables(struct hub_info* hub);
enum notify_verbosity
{
notify_error = 0,
notify_warn = 1,
notify_info = 2,
notify_debug = 3,
};
/**
* Send a notification message to operators.
*/
extern void hub_notify(struct hub_info* hub, enum notify_verbosity verbosity, const char* fmt, ...);
/**
* Returns a string for the given status_message (See enum status_message).
*/
@ -350,6 +364,10 @@ extern void hub_disconnect_user(struct hub_info* hub, struct hub_user* user, int
*/
extern void hub_logout_log(struct hub_info* hub, struct hub_user* user);
/**
* Returns 1 if the hub is running, and 0 otherwise.
*/
extern int hub_is_running(struct hub_info* hub);
#endif /* HAVE_UHUB_HUB_H */

View File

@ -31,6 +31,10 @@ void on_login_success(struct hub_info* hub, struct hub_user* u)
user_set_state(u, state_normal);
uman_add(hub->users, u);
// Make operators receive hub notifications by default.
if (user_is_protected(u))
user_flag_set(u, flag_opnotify);
/* Announce new user to all connected users */
if (user_is_logged_in(u))
route_info_message(hub, u);

View File

@ -104,7 +104,7 @@ static int cbfunc_command_add(struct plugin_handle* plugin, struct plugin_comman
cmdh->internal_handle = command;
list_append(data->commands, cmdh);
command_add(plugin_get_hub(plugin)->commands, command, (void*) plugin);
printf("*** Add plugin command: %s (%p, %p)\n", command->prefix, command, cmdh);
LOG_DEBUG("*** Add plugin command: %s (%p, %p)\n", command->prefix, command, cmdh);
return 0;
}
@ -113,7 +113,7 @@ static int cbfunc_command_del(struct plugin_handle* plugin, struct plugin_comman
struct plugin_callback_data* data = get_callback_data(plugin);
struct command_handle* command = (struct command_handle*) cmdh->internal_handle;
printf("*** Del plugin command: %s (%p, %p)\n", command->prefix, command, cmdh);
LOG_DEBUG("*** Del plugin command: %s (%p, %p)\n", command->prefix, command, cmdh);
list_remove(data->commands, cmdh);
command_del(plugin_get_hub(plugin)->commands, command);
hub_free(command);

View File

@ -104,7 +104,19 @@ int route_to_user(struct hub_info* hub, struct hub_user* user, struct adc_messag
#endif
if (!user->connection)
return 0;
{
switch (user->type)
{
case user_type_client:
return 0; // No connection - we're about to drop this user.
case user_type_bot:
{
bot_recv_msg handler = (bot_recv_msg) user->ptr;
handler(user, msg);
return 0;
}
}
}
uhub_assert(msg->cache && *msg->cache);
@ -149,6 +161,19 @@ int route_to_all(struct hub_info* hub, struct adc_message* command) /* iterate u
return 0;
}
int route_to_operators(struct hub_info* hub, struct adc_message* command) /* iterate users */
{
struct hub_user* user = (struct hub_user*) list_get_first(hub->users->list);
while (user)
{
if (user_flag_get(user, flag_opnotify))
route_to_user(hub, user, command);
user = (struct hub_user*) list_get_next(hub->users->list);
}
return 0;
}
int route_to_subscribers(struct hub_info* hub, struct adc_message* command) /* iterate users */
{
int do_send;

View File

@ -40,6 +40,14 @@ extern int route_to_user(struct hub_info* hub, struct hub_user*, struct adc_mess
*/
extern int route_to_all(struct hub_info* hub, struct adc_message* command);
/**
* Send a message to all operators.
* Technically it sends to all users with the flag_opnotify set
* which is intended to be set for only operators - allowing operators to opt-out
* of these messages.
*/
extern int route_to_operators(struct hub_info* hub, struct adc_message* command);
/**
* Broadcast message to all users subscribing to the type of message.
*/

View File

@ -42,7 +42,6 @@ struct hub_user* user_create(struct hub_info* hub, struct net_connection* con, s
LOG_TRACE("user_create(), hub=%p, con[sd=%d]", hub, net_con_get_sd(con));
user = (struct hub_user*) hub_malloc_zero(sizeof(struct hub_user));
if (user == NULL)
return NULL; /* OOM */
@ -54,6 +53,7 @@ struct hub_user* user_create(struct hub_info* hub, struct net_connection* con, s
memcpy(&user->id.addr, addr, sizeof(struct ip_addr_encap));
user_set_state(user, state_protocol);
user->type = user_type_client;
flood_control_reset(&user->flood_chat);
flood_control_reset(&user->flood_connect);
@ -65,6 +65,54 @@ struct hub_user* user_create(struct hub_info* hub, struct net_connection* con, s
return user;
}
/// generate a semi-stable CID: base32_encode(tiger({hub name} + {nick}))
static const char* generate_bot_cid(struct hub_info* hub, const char* nick)
{
static char result[MAX_CID_LEN+1];
char buf[(MAX_NICK_LEN*2)+1];
uint64_t tiger_res[3];
memset(buf, 0, sizeof(buf));
snprintf(buf, sizeof(buf), "%s%s", hub->config->hub_name, nick);
tiger((uint64_t*) buf, sizeof(buf), (uint64_t*) tiger_res);
base32_encode((unsigned char*) tiger_res, TIGERSIZE, result);
result[MAX_CID_LEN] = 0;
return result;
}
struct hub_user* user_create_bot(struct hub_info* hub, const char* nick, const char* description, bot_recv_msg msg_handler)
{
struct hub_user* user = NULL;
LOG_TRACE("user_create_bot(), hub=%p, nick=\"%s\"", hub, nick);
user = (struct hub_user*) hub_malloc_zero(sizeof(struct hub_user));
if (user == NULL)
return NULL; /* OOM */
strcpy(user->id.nick, nick);
uman_get_free_sid(hub->users, user);
user_set_state(user, state_normal);
user->type = user_type_bot;
user->credentials = auth_cred_bot;
// The message handler
user->ptr = (void*) msg_handler;
user->info = adc_msg_construct(ADC_CMD_BINF, 15);
if (user->info)
{
adc_msg_add_argument(user->info, sid_to_string(user->id.sid));
adc_msg_add_named_argument(user->info, ADC_INF_FLAG_CLIENT_TYPE, ADC_CLIENT_TYPE_BOT);
adc_msg_add_named_argument_string(user->info, ADC_INF_FLAG_USER_AGENT, PRODUCT_STRING);
adc_msg_add_named_argument_string(user->info, ADC_INF_FLAG_NICK, nick);
adc_msg_add_named_argument_string(user->info, ADC_INF_FLAG_DESCRIPTION, description);
adc_msg_add_named_argument(user->info, ADC_INF_FLAG_CLIENT_ID, generate_bot_cid(hub, nick));
}
user->hub = hub;
return user;
}
void user_destroy(struct hub_user* user)
{

View File

@ -47,6 +47,7 @@ enum user_flags
feature_link = 0x00000100, /** LINK: Hub link (not supported) */
feature_adcs = 0x00000200, /** ADCS: ADC over TLS/SSL */
feature_bas0 = 0x00000400, /** BAS0: Obsolete pre-ADC/1.0 protocol version */
flag_opnotify = 0x00200000, /** User should receive operation notifications. NOTE: Only operators should have this flag! */
flag_flood = 0x00400000, /** User has been notified about flooding. */
flag_muted = 0x00800000, /** User is muted (cannot chat) */
flag_ignore = 0x01000000, /** Ignore further reads */
@ -104,21 +105,30 @@ struct hub_user_limits
size_t hub_count_total; /** The number of hubs connected to in total */
};
enum user_type
{
user_type_client, /** A user connected normally as an ADC client */
user_type_bot, /** Not really a user, but a bot inside the hub */
};
typedef void (*bot_recv_msg)(struct hub_user*, struct adc_message* msg);
struct hub_user
{
struct hub_user_info id; /** Contains nick name and CID */
enum auth_credentials credentials; /** see enum user_credentials */
enum user_state state; /** see enum user_state */
enum user_type type;
uint32_t flags; /** see enum user_flags */
struct linked_list* feature_cast; /** Features supported by feature cast */
struct adc_message* info; /** ADC 'INF' message (broadcasted to everyone joining the hub) */
struct hub_info* hub; /** The hub instance this user belong to */
void* ptr;
struct ioq_recv* recv_queue;
struct ioq_send* send_queue;
struct net_connection* connection; /** Connection data */
struct hub_user_limits limits; /** Data used for limitation */
enum user_quit_reason quit_reason; /** Quit reason (see user_quit_reason) */
struct flood_control flood_chat;
struct flood_control flood_connect;
struct flood_control flood_search;
@ -138,6 +148,7 @@ struct hub_user
* @return User object or NULL if not enough memory is available.
*/
extern struct hub_user* user_create(struct hub_info* hub, struct net_connection* con, struct ip_addr_encap* addr);
extern struct hub_user* user_create_bot(struct hub_info* hub, const char* nick, const char* description, bot_recv_msg msg_handler);
/**
* Delete a user.

View File

@ -1,6 +1,6 @@
/*
* uhub - A tiny ADC p2p connection hub
* Copyright (C) 2007-2010, Jan Vidar Krey
* Copyright (C) 2007-2013, Jan Vidar Krey
*
* 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
@ -57,14 +57,30 @@ size_t timeout_queue_process(struct timeout_queue* t, time_t now)
size_t pos = (size_t) t->last;
size_t events = 0;
struct timeout_evt* evt = 0;
uhub_assert(t->last <= now);
t->last = now;
// We can optimize in case we need to wrap around
// the buffer, so we only do it once
if (MAX(pos, now) - MIN(pos, now) > t->max)
{
// FIXME: Double check this calculation
pos = (now - t->max);
}
for (; pos <= now; pos++)
{
while ((evt = t->events[pos % t->max]))
evt = t->events[pos % t->max];
while (evt)
{
timeout_queue_remove(t, evt);
evt->callback(evt);
events++;
if (evt->timestamp < pos)
{
timeout_queue_remove(t, evt);
evt->callback(evt);
events++;
}
evt = evt->next;
}
}
return events;
@ -85,24 +101,22 @@ size_t timeout_queue_get_next_timeout(struct timeout_queue* t, time_t now)
void timeout_queue_insert(struct timeout_queue* t, struct timeout_evt* evt, size_t seconds)
{
struct timeout_evt* first;
struct timeout_evt* it, *first;
size_t pos = ((t->last + seconds) % t->max);
evt->timestamp = t->last + seconds;
evt->next = 0;
first = t->events[pos];
if (first)
{
uhub_assert(first->timestamp == evt->timestamp);
first->prev->next = evt;
evt->prev = first->prev;
first->prev = evt;
for (it = first; it->next; it = it->next) { }
it->next = evt;
evt->prev = it;
}
else
{
t->events[pos] = evt;
evt->prev = evt;
evt->prev = evt; // point to self.
}
evt->next = 0;
}

View File

@ -1,6 +1,6 @@
/*
* uhub - A tiny ADC p2p connection hub
* Copyright (C) 2007-2010, Jan Vidar Krey
* Copyright (C) 2007-2013, Jan Vidar Krey
*
* 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