wvdbusconn.cc

/* -*- Mode: C++ -*-
 * Worldvisions Weaver Software:
 *   Copyright (C) 2004-2006 Net Integration Technologies, Inc.
 *
 * Pathfinder Software:
 *   Copyright (C) 2007, Carillon Information Security Inc.
 *
 * This library is licensed under the LGPL, please read LICENSE for details.
 *
 */
#include "wvdbusconn.h"
#include "wvmoniker.h"
#include "wvstrutils.h"
#undef interface // windows
#include <dbus/dbus.h>
 
 
static WvString translate(WvStringParm dbus_moniker)
{
    WvStringList l;
    WvStringList::Iter i(l);
 
    if (!strncasecmp(dbus_moniker, "unix:", 5))
    {
        WvString path, tmpdir;
        l.split(dbus_moniker+5, ",");
        for (i.rewind(); i.next(); )
        {
            if (!strncasecmp(*i, "path=", 5))
                path = *i + 5;
            else if (!strncasecmp(*i, "abstract=", 9))
                path = WvString("@%s", *i + 9);
            else if (!strncasecmp(*i, "tmpdir=", 7))
                tmpdir = *i + 7;
        }
        if (!!path)
            return WvString("unix:%s", path);
        else if (!!tmpdir)
            return WvString("unix:%s/dbus.sock", tmpdir);
    }
    else if (!strncasecmp(dbus_moniker, "tcp:", 4))
    {
        WvString host, port, family;
        l.split(dbus_moniker+4, ",");
        for (i.rewind(); i.next(); )
        {
            if (!strncasecmp(*i, "family=", 7))
                family = *i + 7;
            else if (!strncasecmp(*i, "host=", 5))
                host = *i + 5;
            else if (!strncasecmp(*i, "port=", 5))
                port = *i + 5;
        }
        if (!!host && !!port)
            return WvString("tcp:%s:%s", host, port);
        else if (!!host)
            return WvString("tcp:%s", host);
        else if (!!port)
            return WvString("tcp:0.0.0.0:%s", port); // localhost
    }
 
    return dbus_moniker; // unrecognized
}
 
 
static IWvStream *stream_creator(WvStringParm _s, IObject *)
{
    WvString s(_s);
 
    if (!strcasecmp(s, "starter"))
    {
        WvString startbus(getenv("DBUS_STARTER_ADDRESS"));
        if (!!startbus)
            return IWvStream::create(translate(startbus));
        else
        {
            WvString starttype(getenv("DBUS_STARTER_BUS_TYPE"));
            if (!!starttype && !strcasecmp(starttype, "system"))
                s = "system";
            else if (!!starttype && !strcasecmp(starttype, "session"))
                s = "session";
        }
    }
 
    if (!strcasecmp(s, "system"))
    {
        // NOTE: the environment variable for the address of the system
        // bus is very often not set-- in that case, look in your dbus 
        // system bus config file (e.g. /etc/dbus-1/system.conf) for the 
        // raw address and either set this environment variable to that, or 
        // pass in the address directly
        WvString bus(getenv("DBUS_SYSTEM_BUS_ADDRESS"));
        if (!!bus)
            return IWvStream::create(translate(bus));
    }
 
    if (!strcasecmp(s, "session"))
    {
        WvString bus(getenv("DBUS_SESSION_BUS_ADDRESS"));
        if (!!bus)
            return IWvStream::create(translate(bus));
    }
 
    return IWvStream::create(translate(s));
}
 
static WvMoniker<IWvStream> reg("dbus", stream_creator);
 
 
static int conncount;
 
WvDBusConn::WvDBusConn(IWvStream *_cloned, IWvDBusAuth *_auth, bool _client)
    : WvStreamClone(_cloned),
        log(WvString("DBus %s%s",
                     _client ? "" : "s",
                     ++conncount), WvLog::Debug5),
        pending(10)
{
    init(_auth, _client);
}
 
 
WvDBusConn::WvDBusConn(WvStringParm moniker, IWvDBusAuth *_auth, bool _client)
    : WvStreamClone(IWvStream::create(moniker)),
        log(WvString("DBus %s%s",
                     _client ? "" : "s",
                     ++conncount), WvLog::Debug5),
        pending(10)
{
    log("Connecting to '%s'\n", moniker);
    init(_auth, _client);
}
 
 
void WvDBusConn::init(IWvDBusAuth *_auth, bool _client)
{
    log("Initializing.\n");
    client = _client;
    auth = _auth ? _auth : new WvDBusClientAuth;
    authorized = in_post_select = false;
    if (!client) set_uniquename(WvString(":%s.0", conncount));
 
    if (!isok()) return;
    
    delay_output(true);
 
    // this will get enqueued until later, but we want to make sure it
    // comes before anything the user tries to send - including anything
    // goofy they enqueue in the authorization part.
    if (client)
        send_hello();
 
    try_auth();
}
 
WvDBusConn::~WvDBusConn()
{
    log("Shutting down.\n");
    if (geterr())
        log("Error was: %s\n", errstr());
 
    close();
 
    delete auth;
}
 
 
void WvDBusConn::close()
{
    if (!closed)
        log("Closing.\n");
    WvStreamClone::close();
}
 
 
WvString WvDBusConn::uniquename() const
{
    return _uniquename;
}
 
 
void WvDBusConn::request_name(WvStringParm name, const WvDBusCallback &onreply,
                              time_t msec_timeout)
{
    uint32_t flags = (DBUS_NAME_FLAG_ALLOW_REPLACEMENT |
                      DBUS_NAME_FLAG_REPLACE_EXISTING);
    WvDBusMsg msg("org.freedesktop.DBus", "/org/freedesktop/DBus",
                  "org.freedesktop.DBus", "RequestName");
    msg.append(name).append(flags);
    send(msg, onreply, msec_timeout);
}
 
 
uint32_t WvDBusConn::send(WvDBusMsg msg)
{
    msg.marshal(out_queue);
    if (authorized)
    {
        log(" >> %s\n", msg);
        write(out_queue);
    }
    else
        log(" .> %s\n", msg);
    return msg.get_serial();
}
 
 
void WvDBusConn::send(WvDBusMsg msg, const WvDBusCallback &onreply,
                      time_t msec_timeout)
{
    send(msg);
    if (onreply)
        add_pending(msg, onreply, msec_timeout);
}
 
 
class xxReplyWaiter
{
public:
    WvDBusMsg *reply;
    
    xxReplyWaiter()
        { reply = NULL; }
    ~xxReplyWaiter()
        { delete reply; }
    bool reply_wait(WvDBusMsg &msg)
        { reply = new WvDBusMsg(msg); return true; }
};
 
 
WvDBusMsg WvDBusConn::send_and_wait(WvDBusMsg msg, time_t msec_timeout,
                                wv::function<void(uint32_t)> serial_cb)
{
    xxReplyWaiter rw;
    
    send(msg, wv::bind(&xxReplyWaiter::reply_wait, &rw, _1),
         msec_timeout);
    if (serial_cb)
        serial_cb(msg.get_serial());
    while (!rw.reply && isok())
        runonce();
    if (!rw.reply)
        return WvDBusError(msg, DBUS_ERROR_FAILED,
                           WvString("Connection closed (%s) "
                                    "while waiting for reply.",
                                    errstr()));
    else
        return *rw.reply;
}
 
 
void WvDBusConn::out(WvStringParm s)
{
    log(" >> %s", s);
    print(s);
}
 
 
const char *WvDBusConn::in()
{
    const char *s = trim_string(getline(0));
    if (s)
        log("<<  %s\n", s);
    return s;
}
 
 
void WvDBusConn::send_hello()
{
    WvDBusMsg msg("org.freedesktop.DBus", "/org/freedesktop/DBus",
                  "org.freedesktop.DBus", "Hello");
    send(msg, wv::bind(&WvDBusConn::_registered, this, _1));
    WvDBusMsg msg2("org.freedesktop.DBus", "/org/freedesktop/DBus",
                   "org.freedesktop.DBus", "AddMatch");
    msg2.append("type='signal'");
    send(msg2); // don't need to monitor this for completion
}
 
 
void WvDBusConn::set_uniquename(WvStringParm s)
{
    // we want to print the message before switching log.app, so that we
    // can trace which log.app turned into which
    log("Assigned name '%s'\n", s);
    _uniquename = s;
    log.app = WvString("DBus %s%s", client ? "" : "s", uniquename());
}
 
 
void WvDBusConn::try_auth()
{
    bool done = auth->authorize(*this);
    if (done)
    {
        // ready to send messages!
        if (out_queue.used())
        {
            log(" >> (sending enqueued messages)\n");
            write(out_queue);
        }
 
        authorized = true;
    }
}
 
 
void WvDBusConn::add_callback(CallbackPri pri, WvDBusCallback cb, void *cookie)
{
    callbacks.append(new CallbackInfo(pri, cb, cookie), true);
}
 
 
void WvDBusConn::del_callback(void *cookie)
{
    // remember, there might be more than one callback with the same cookie.
    CallbackInfoList::Iter i(callbacks);
    for (i.rewind(); i.next(); )
        if (i->cookie == cookie)
            i.xunlink();
}
 
 
int WvDBusConn::priority_order(const CallbackInfo *a, const CallbackInfo *b)
{
    return a->pri - b->pri;
}
 
bool WvDBusConn::filter_func(WvDBusMsg &msg)
{
    log("<<  %s\n", msg);
 
    // handle replies
    uint32_t rserial = msg.get_replyserial();
    if (rserial)
    {
        Pending *p = pending[rserial];
        if (p)
        {
            p->cb(msg);
            pending.remove(p);
            return true; // handled it
        }
    }
 
    // handle all the generic filters
    CallbackInfoList::Sorter i(callbacks, priority_order);
    for (i.rewind(); i.next(); )
    {
        bool handled = i->cb(msg);
        if (handled) return true;
    }
 
    return false; // couldn't handle the message, sorry
}
 
 
WvDBusClientAuth::WvDBusClientAuth()
{
    sent_request = false;
}
 
 
wvuid_t WvDBusClientAuth::get_uid()
{
    return wvgetuid();
}
 
 
bool WvDBusClientAuth::authorize(WvDBusConn &c)
{
    if (!sent_request)
    {
        c.write("\0", 1);
        WvString uid = get_uid();
        c.out("AUTH EXTERNAL %s\r\n\0", WvHexEncoder().strflushstr(uid));
        sent_request = true;
    }
    else
    {
        const char *line = c.in();
        if (line)
        {
            if (!strncasecmp(line, "OK ", 3))
            {
                c.out("BEGIN\r\n");
                return true;
            }
            else if (!strncasecmp(line, "ERROR ", 6))
                c.seterr("Auth failed: %s", line);
            else
                c.seterr("Unknown AUTH response: '%s'", line);
        }
    }
 
    return false;
}
 
 
time_t WvDBusConn::mintimeout_msec()
{
    WvTime when = 0;
    PendingDict::Iter i(pending);
    for (i.rewind(); i.next(); )
    {
        if (!when || when > i->valid_until)
            when = i->valid_until;
    }
    if (!when)
        return -1;
    else if (when <= wvstime())
        return 0;
    else
        return msecdiff(when, wvstime());
}
 
 
bool WvDBusConn::post_select(SelectInfo &si)
{
    bool ready = WvStreamClone::post_select(si);
    if (si.inherit_request) return ready;
    
    if (in_post_select) return false;
    in_post_select = true;
 
    if (!authorized && ready)
        try_auth();
 
    if (!alarm_remaining())
    {
        WvTime now = wvstime();
        PendingDict::Iter i(pending);
        for (i.rewind(); i.next(); )
        {
            if (now > i->valid_until)
            {
                log("Expiring %s\n", i->msg);
                expire_pending(i.ptr());
                i.rewind();
            }
        }
    }
 
    if (authorized && ready)
    {
        // put this in a loop so that wvdbusd can forward packets rapidly.
        // Otherwise TCP_NODELAY kicks in, because we do a select() loop
        // between packets, which causes delay_output() to flush.
        bool ran;
        do
        {
            ran = false;
            size_t needed = WvDBusMsg::demarshal_bytes_needed(in_queue);
            size_t amt = needed - in_queue.used();
            if (amt < 4096)
                amt = 4096;
            read(in_queue, amt);
            WvDBusMsg *m;
            while ((m = WvDBusMsg::demarshal(in_queue)) != NULL)
            {
                ran = true;
                filter_func(*m);
                delete m;
            }
        } while (ran);
    }
 
    alarm(mintimeout_msec());
    in_post_select = false;
    return false;
}
 
 
bool WvDBusConn::isidle()
{
    return !out_queue.used() && pending.isempty();
}
 
 
void WvDBusConn::expire_pending(Pending *p)
{
    if (p)
    {
        WvDBusCallback xcb(p->cb);
        pending.remove(p); // prevent accidental recursion
        WvDBusError e(p->msg, DBUS_ERROR_FAILED,
                      "Timed out while waiting for reply");
        xcb(e);
    }
}
 
 
void WvDBusConn::cancel_pending(uint32_t serial)
{
    Pending *p = pending[serial];
    if (p)
    {
        WvDBusCallback xcb(p->cb);
        WvDBusMsg msg(p->msg);
        pending.remove(p); // prevent accidental recursion
        WvDBusError e(msg, DBUS_ERROR_FAILED,
                      "Canceled while waiting for reply");
        xcb(e);
    }
}
 
 
void WvDBusConn::add_pending(WvDBusMsg &msg, WvDBusCallback cb,
                 time_t msec_timeout)
{
    uint32_t serial = msg.get_serial();
    assert(serial);
    if (pending[serial])
        cancel_pending(serial);
    pending.add(new Pending(msg, cb, msec_timeout), true);
    alarm(mintimeout_msec());
}
 
 
bool WvDBusConn::_registered(WvDBusMsg &msg)
{
    WvDBusMsg::Iter i(msg);
    _uniquename = i.getnext().get_str();
    set_uniquename(_uniquename);
    return true;
}