vos/corelibs/vos/localsocketsite.cc

Go to the documentation of this file.

/*
    This file is part of the Virtual Object System of
    the Interreality project (http://interreality.org).

    Copyright (C) 2001-2003 Peter Amstutz

    This library is free software; you can redistribute it and/or
    modify it under the terms of the GNU Lesser General Public
    License as published by the Free Software Foundation; either
    version 2 of the License, or (at your option) any later version.

    This library 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
    Lesser General Public License for more details.

    You should have received a copy of the GNU Lesser General Public
    License along with this library; if not, write to the Free Software
    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307  USA

    Peter Amstutz <tetron@interreality.org>
*/

#include "localsocketsite.hh"
#include "accesscontrol.hh"
#include "remotesocketsite.hh"
#include "localmetaobject.hh"
#include "message.hh"
#include "timer.hh"

using namespace VOS;

#include <stdio.h>
#include <fcntl.h>
#include <errno.h>
#include <math.h>

#include <deque>
#include <string>
#include <map>

#ifdef SOCKS_SUPPORT
#undef __P
#if defined (__STDC__) || defined (_AIX) \
        || (defined (__mips) && defined (_SYSTYPE_SVR4)) \
        || defined(WIN32) || defined(__cplusplus)
# define __P(protos) protos
#else
# define __P(protos) ()
#endif
extern "C" {
    int Rbind __P((int, const struct sockaddr *, socklen_t));
    int Raccept __P((int, struct sockaddr *, socklen_t *));
    void checkmodule(const char*) {} // silly undefined symbol is libsocks...
}
#endif

#ifdef WIN32
#include <windows.h>
#endif


#ifndef SOL_TCP
//#define SOL_TCP (getprotobyname("tcp")->p_proto)
#define SOL_TCP 6
#endif

// work around an annoying difference between winsock2.h and sys/socket.h
#ifdef HAVE_LIBWS2_32
# define SETSOCKOPT_PARAM4_CAST char*
#else
# define SETSOCKOPT_PARAM4_CAST int*
#endif


#if !defined(HAVE_SOCKLEN_T) && !defined(__socklen_t_defined)
//#warning socklen_t not defined. Defining it as int.
typedef int socklen_t;
#define __socklen_t_defined 1
#endif

/** @file
    Implements LocalSocketSite.
*/

void LocalSocketSite::init(const string& defaultHostname, unsigned short int port) throw (PortBindingError)
{
    srand((unsigned int)time(0));

#ifdef WIN32 // @@@ should go somewhere else
    WSADATA WSAData;

    WORD wVersionRequested = MAKEWORD(2, 0);
    WSAStartup(wVersionRequested, &WSAData);
#endif

    struct sockaddr_in sa;
    int set=1;

    listenport=port;
    memset(&sa, 0, sizeof(struct sockaddr_in));
    sa.sin_port=htons(listenport);
    sa.sin_family=AF_INET;
    sa.sin_addr.s_addr = INADDR_ANY;
    if((listensocket = (int)socket(AF_INET, SOCK_STREAM, 0)) < 0)
        throw PortBindingError(strerror(errno));
#ifndef WIN32
    setsockopt(listensocket, SOL_SOCKET, SO_REUSEADDR, (SETSOCKOPT_PARAM4_CAST)&set, sizeof(int));
#endif

#ifdef SOCKS_SUPPORT
    if(getenv("VOS_USE_SOCKS")) {
        if(Rbind(listensocket, (struct sockaddr *)&sa, sizeof(struct sockaddr_in)) < 0)
            throw PortBindingError(strerror(errno));
        listen(listensocket, 5);
        usingSOCKS = true;
    } else {
#endif
        if(bind(listensocket, (struct sockaddr *)&sa, sizeof(struct sockaddr_in)) < 0)
            throw PortBindingError(strerror(errno));
        listen(listensocket, 5);
#ifdef SOCKS_SUPPORT
        usingSOCKS = false;
    }
#endif

    string host;
    char c[256];
    if(defaultHostname == "") {
        gethostname(c, sizeof(c));
        host=c;
        if(host.find('.') == string::npos) {
            LOG("LocalSocketSite", 1, "WARNING: Your host name '" << host << "' does not appear to include a domain name. You will probably not be able to use this application on the Internet! Use the VOS_HOSTNAME environment variable to specify a fully qualified domain name or IP address.");
        }
    } else host=defaultHostname;

    snprintf(c, sizeof(c), "%s:%i", host.c_str(), port);
    addHostAlias(c);
    if(host != "localhost") {
        snprintf(c, sizeof(c), "localhost:%i", port);
        addHostAlias(c);
    }
    snprintf(c, sizeof(c), "%i", port);
    url.setHost(host);
    url.setPort(c);
    addSite(this);
    snprintf(c, sizeof(c), "%s:%i", host.c_str(), port);
    name = c;

    iteratorsUsingPeerSites=0;

    selectwait->tv_sec=0;
    selectwait->tv_usec=0;

    if(LocalSite* s = Site::getDefaultPeer()) {
        s->release(); // acquired by: getDefaultPeer()
    } else {
        Site::setDefaultPeer(this);
    }
}

LocalSocketSite::LocalSocketSite(SiteAccessControl* ac) throw (PortBindingError)
    : VobjectImplementation("", 0, true), LocalMetaObject("", 0, ac),
      LocalVobject("", 0, ac), MetaObject(0), alreadyDoingFlushIncoming(false)
{
    bool foundPort = false;
    int c=VOS_DEFAULT_PORT;
    char hostname[256] = "";
    if(getenv("VOS_HOSTNAME")) strncpy(hostname, getenv("VOS_HOSTNAME"), sizeof(hostname));
    if (*hostname) {
        int i;
        for(i = 0; hostname[i]; i++) {
            if(hostname[i] == ':') {
                c = atoi(hostname+i+1);
                break;
            }
        }
        hostname[i] = 0;
    }
    for(; !foundPort && c < 0xFFFF; c++) {
        try {
            init(hostname, c);
            foundPort=true;
        } catch(PortBindingError) {
            foundPort=false;
        }
    }
    // something is obviously fucked up big time
    if(c == 0xFFFF) throw PortBindingError("Cannot allocate port");
}

LocalSocketSite::LocalSocketSite(const string& defaultHostname, unsigned short int port, SiteAccessControl* ac)
    throw (PortBindingError)
    : VobjectImplementation(defaultHostname, 0, true), LocalMetaObject(defaultHostname, 0, ac),
      LocalVobject(defaultHostname, 0, ac), MetaObject(0), alreadyDoingFlushIncoming(false)
{
    init(defaultHostname, port);
}

LocalSocketSite::LocalSocketSite(const string& defaultHostname, SiteAccessControl* ac) throw (PortBindingError)
    : VobjectImplementation(defaultHostname, 0, true), LocalMetaObject(defaultHostname, 0, ac),
      LocalVobject(defaultHostname, 0, ac), MetaObject(0), alreadyDoingFlushIncoming(false)
{
    bool foundPort = false;
    int c;
    for(c=VOS_DEFAULT_PORT; !foundPort && c < 0xFFFF; c++) {
        try {
            init(defaultHostname, c);
            foundPort=true;
        } catch(PortBindingError) {
            foundPort=false;
        }
    }
    // something is obviously fucked up big time
    if(c == 0xFFFF) throw PortBindingError("Cannot allocate port");
}

LocalSocketSite::LocalSocketSite(unsigned short int port, SiteAccessControl* ac) throw (PortBindingError)
    :     VobjectImplementation("", 0, true), LocalVobject("", 0, ac),
          LocalMetaObject("", 0, ac), MetaObject(0), alreadyDoingFlushIncoming(false)
{
    char *hostname = getenv("VOS_HOSTNAME");
    if(hostname == NULL)
        hostname = "";  // init will call gethostname().
    init(hostname, port);
}

LocalSocketSite::~LocalSocketSite()
{
#ifdef WIN32
    WSACleanup();
#endif
}

void LocalSocketSite::acceptConnectionRequests()
{
    fd_set f;
    struct timeval tv = {0, 0};
    struct sockaddr_in sa;

    socklen_t len=sizeof(sa);

    FD_ZERO(&f);
    FD_SET(listensocket, &f);
    while(select(listensocket+1, &f, 0, 0, &tv) > 0
          && FD_ISSET(listensocket, &f)) {
        int newsock;
#ifdef SOCKS_SUPPORT
        if(usingSOCKS) newsock=Raccept(listensocket, (sockaddr*)&sa, &len);
        else
#endif
            newsock = (int)accept(listensocket, (sockaddr*)&sa, &len);
        int x = 1;
        setsockopt(newsock, SOL_TCP, TCP_NODELAY, (char*)&x, sizeof(int));
#ifdef WIN32
    unsigned long one = 1;
    ioctlsocket(newsock, FIONBIO, &one);
#else
    unsigned int one = fcntl(newsock, F_GETFL);
    one = one | O_NONBLOCK;
    fcntl(newsock, F_SETFL, one);
#endif
        LOG("localsite", 3, "got new connection, fd " << newsock);
        if(newsock > -1) {
            allOpenSockets.insert(newsock);
            vRef<RemoteSocketSite> rs = new RemoteSocketSite(newsock, &sa);

            if(getenv("VOS_HOSTNAME") == 0) {
                unsigned char ipaddr[4];
                string host = detectHostname(rs->getReadingFD(), ipaddr);
                URL u(getURL());
                u.setHost(host);
                addHostAlias(u.getHostAndPort());
                if(!(ipaddr[0] == 10 || ipaddr[0] == 127 || (ipaddr[0] == 192 && ipaddr[1] == 168))) {
                    setPrimaryHostname(host);
                }
            }
            try {
                LOG("refcount", 5, "count on new rs is " << rs->getCount());
                // at this point we have 2 references:
                // 1 in the local scope
                // 1 in the site table
                LOG("localsocketsite", 4, "Beginning site peering of inbound connection");
                doSitePeering(this, &rs, false, true);
                LOG("localsocketsite", 4, "Done site peering of inbound connection");
                LOG("refcount", 5, "count on rs post-peering is " << rs->getCount());
                // rs should be at least ?:
                // 1 reference in this scope
                // 1 in the site table
                // 1 reference in the peer sites

                //rs->getTypes();

                vRef<Message> m = new Message();
                m->setType("message");
                m->setMethod("core:get-types");
                m->setTo(rs->getURL().getString());
                m->setFrom(getURL().getString());
                rs->sendMessage(&m);
            } catch(RemoteError x) {
                LOG("localsite", 3, "Error peering with remote site: " << x.what());
            }
            LOG("refcount", 5, "count on rs getTypes n stuff is " << rs->getCount());
        } else {
            LOG("localsite", 3, "could not accept connection request: " << strerror(errno));
            break;
        }
        FD_ZERO(&f);
        FD_SET(listensocket, &f);
    }
}

int LocalSocketSite::getFDset(fd_set* readset, fd_set* writeset, fd_set* errorset)
{
    FD_SET(listensocket, readset);
    int max=listensocket;

    for(set<int>::iterator i = allOpenSockets.begin(); i != allOpenSockets.end(); i++) {
        if(*i > -1) {
            FD_SET(*i, readset);
            if(*i > max) max=*i;
        }
    }
    /* the following is for user-added file descriptors (such as  waiting
       on input events like a keypress) */
    for(map<int, pair<int, int> >::iterator i = extra_fds.begin(); i != extra_fds.end(); i++) {
        if((*i).first > max) max = (*i).first;
        if((*i).second.first & SELECTREAD) FD_SET((*i).first, readset);
        if((*i).second.first & SELECTWRITE) FD_SET((*i).first, writeset);
        if((*i).second.first & SELECTERROR) FD_SET((*i).first, errorset);
    }
    return max;
}

struct timeval* LocalSocketSite::calculateTimeout()
{
    struct timeval *tv;

    if(selectwait) {
        tv = new struct timeval();
        *tv = *selectwait;
    } else tv=0;

    if(!siteMessageQueue.empty() || !callbacks.empty()) {
        LOG("localsocketsite", 5, "smq " << siteMessageQueue.empty() << " cb " << callbacks.empty());

        double now = getTimer();

        double nextEvent;
        if(siteMessageQueue.empty()) {
            nextEvent = (*callbacks.begin()).first;
        } else if(callbacks.empty()) {
            nextEvent = (*siteMessageQueue.begin()).first;
        } else {
            if((*siteMessageQueue.begin()).first < (*callbacks.begin()).first)
                nextEvent = (*siteMessageQueue.begin()).first;
            else
                nextEvent = (*callbacks.begin()).first;
        }

        //LOG("main", 4, "first waiting message is "
        //     << (*siteMessageQueue.begin()).second->getFormattedString());

        nextEvent -= now;

        if(nextEvent <= 0) {
            // event happens right now,
            // or should have already happened
            if(! tv) tv = new struct timeval();
            tv->tv_sec = 0;
            tv->tv_usec = 0;
        } else {
            if(tv) {
                // event happens in the future.
                // wait for the max wait time
                // or the time until the next event,
                // whichever is smaller
                double maxWait = tv->tv_sec + (tv->tv_usec / 1000000.0);
                if(nextEvent < maxWait) {
                    tv->tv_sec = (long)floor(nextEvent);
                    tv->tv_usec = (long)((nextEvent - floor(nextEvent)) * 1000000.0);
                }
            } else {
                // wait just long enough for the next event
                tv = new struct timeval();
                tv->tv_sec = (long)floor(nextEvent);
                tv->tv_usec = (long)((nextEvent - floor(nextEvent)) * 1000000.0);
            }
        }
    }

    return tv;
}

void LocalSocketSite::flushIncomingBuffers()
{
    bool more;

    // flush this before we do anything else
    for(list< pair<string, RemoteSite*> >::iterator i = needSpoofIDreply.begin();
        i != needSpoofIDreply.end(); i++)
    {
        string spid = getAntiSpoofIDMapping((*i).first);
        if(spid != "") {
            vRef<Message> reply = new Message();
            reply->setType("message");
            reply->setFrom("");
            reply->setTo("");
            reply->setMethod("core:anti-spoof-reply");
            reply->insertField(-1, "poke", spid);
            (*i).second->sendMessage(&reply);

            (*i).second->release(); // acquired by: anti-spoof-check handler is LocalSite::sendMessage

            list< pair<string, RemoteSite*> >::iterator n = i;
            i++;
            needSpoofIDreply.erase(n);
        }
    }

//  if(!alreadyDoingFlushIncoming) {
//      alreadyDoingFlushIncoming = true;
        do {
            more=false;

            struct timeval *tv = 0;
            fd_set readset, writeset, errorset;

            do {
                FD_ZERO(&readset);
                FD_ZERO(&writeset);
                FD_ZERO(&errorset);

                int max = getFDset(&readset, &writeset, &errorset);

                tv = calculateTimeout();

                errno = 0;
                select(max+1, &readset, &writeset, &errorset, tv);

                if(tv) delete tv;
            } while(errno == EINTR);

            for(map<int, pair<int, int> >::iterator i = extra_fds.begin(); i != extra_fds.end(); i++) {
                (*i).second.second = 0;
                if(FD_ISSET((*i).first, &readset)) (*i).second.second |= SELECTREAD;
                if(FD_ISSET((*i).first, &writeset)) (*i).second.second |= SELECTWRITE;
                if(FD_ISSET((*i).first, &errorset)) (*i).second.second |= SELECTERROR;
            }

            if(FD_ISSET(listensocket, &readset)) acceptConnectionRequests();

            iteratorsUsingPeerSites++;
            for(map<RemoteSite*, SiteTableEntry*>::iterator i = peerSites.begin(); i != peerSites.end(); i++) {
                if(!((*i).first && (*i).second)) continue;
                if(RemoteSocketSite* rss = dynamic_cast<RemoteSocketSite*>((*i).first)) {
                    LOG("localsocketsite", 5, "waaagh, rss->getReadingFD is " << rss->getReadingFD() << " and "
                        << (FD_ISSET(rss->getReadingFD(), &readset) != 0));

                    if(rss->getReadingFD() > -1 && FD_ISSET(rss->getReadingFD(), &readset))
                        rss->flushIncomingBuffers();

                    if(rss->getWritingFD() > -1 && FD_ISSET(rss->getWritingFD(), &writeset))
                        rss->flushOutgoingBuffers();
                } else (*i).first->flushIncomingBuffers();
            }
            if(--iteratorsUsingPeerSites == 0) {
                boost::mutex::scoped_lock lock(peerSitesBuffer_mutex);

                while(peerSitesBuffer_remove.size() > 0) {
                    removeRemotePeer(peerSitesBuffer_remove.front());
                    peerSitesBuffer_remove.pop_front();
                }
                deque<RemoteSite*>::iterator i = peerSitesBuffer_rs.begin();
                deque<struct SiteTableEntry*>::iterator n = peerSitesBuffer_st.begin();
                while(i != peerSitesBuffer_rs.end()) {
                    peerSites[*i] = *n;
                    i++;
                    n++;
                }
                peerSitesBuffer_rs.clear();
                peerSitesBuffer_st.clear();
            }
        } while(more);

//      alreadyDoingFlushIncoming = false;
//}
    runSchedule();
}

const int LocalSocketSite::SELECTREAD = 0x1;
const int LocalSocketSite::SELECTWRITE = 0x2;
const int LocalSocketSite::SELECTERROR = 0x4;

void LocalSocketSite::addFDtoSelect(int fd, int which)
{
    extra_fds[fd] = pair<int, int>(which, 0);
}

void LocalSocketSite::removeFDfromSelect(int fd, int which)
{
    extra_fds[fd].first &= ~which;
    if(extra_fds[fd].first == 0) extra_fds.erase(fd);
}

int LocalSocketSite::lastSelectResultForFD(int fd)
{
    return extra_fds[fd].second;
}

SSL_CTX* LocalSocketSite::getSSLContext(const string& protocol)
{
    if(protocol == "TLSv1") {
        return TLSv1context;
    } else if(protocol == "SSLv3") {
        return SSLv3context;
    } else return 0;
}

void LocalSocketSite::sendMessage(Message* m)
{
    LocalSite::sendMessage(m);

    if(m->getMethod() == "core:protocol-switch") {
#ifdef SSL_SUPPORT
        try {
            pREF(RemoteSocketSite*, rs, meta_cast<RemoteSocketSite*>(m->getSourceSite()),
                 if(rs->getLocalPeer() == this) {
                     SSL_CTX* ctx = getSSLContext(m->getField("protocol").value);
                     pREF(Message*, reply, new Message(),
                          rREF(Vobject&, from, findObject(m->getFrom()),
                               initReply(this, reply, m, "core:protocol-switch-reply");
                               if(ctx) {
                                   reply->insertField(-1, "result", "ready");
                                   from.sendMessage(reply);
                                   rs->switchProtocol(ctx);
                               } else {
                                   reply->insertField(-1, "result", "unsupported protocol "
                                                  + m->getField("protocol").value + string("; supports TLSv1, SSLv3"));
                                   from.sendMessage(reply);
                               }
                              );
                         );
                 }
                );
        } catch(Message::NoSuchFieldError) {
        } catch(bad_cast) {
        } catch(NoSuchObjectError) {
        }
#else
        pREF(Message*, reply, new Message(),
             rREF(Vobject&, from, findObject(m->getFrom()),
                  initReply(this, reply, m, "core:protocol-switch-reply");
                  reply->insertField(-1, "result", "this site does not support secure protocols");
                  from.sendMessage(reply);
                 );
            );
#endif
    }
}

int LocalSocketSite::pemPasswordCallback(char *buf, int size, int rwflag, void *password)
{
    string* pw = (string*)password;
    strncpy(buf, pw->c_str(), (pw->size() < size ? pw->size() : size));
    buf[size - 1] = '\0';
    return (int)strlen(buf);
}

int LocalSocketSite::sslVerifyCallback(int preverify_ok, X509_STORE_CTX *ctx)
{
    return 1;
}

void LocalSocketSite::setupSSL(const string& certificateFile, const string& privateKeyFile, const string& password)
{
#ifdef SSL_SUPPORT
    static bool needInitSSL = true;

    if(needInitSSL) {
        SSL_load_error_strings();
        SSL_library_init();
        srand(time(NULL));
        int rnd = rand();
        RAND_add(&rnd, 4, 0);

        needInitSSL = false;
    }

    static string pw = password;

    TLSv1context = SSL_CTX_new(TLSv1_method());
    if(! TLSv1context) {
        throw SSLError("Could not set up a TLSv1 context");
    }

    SSLv3context = SSL_CTX_new(SSLv3_method());
    if(! SSLv3context) {
        throw SSLError("Could not set up a SSLv3 context");
    }

    SSL_CTX_set_default_passwd_cb(TLSv1context, pemPasswordCallback);
    SSL_CTX_set_default_passwd_cb(SSLv3context, pemPasswordCallback);

    SSL_CTX_set_default_passwd_cb_userdata(TLSv1context, &pw);
    SSL_CTX_set_default_passwd_cb_userdata(SSLv3context, &pw);

    int ret = SSL_CTX_use_RSAPrivateKey_file (TLSv1context, privateKeyFile.c_str(),  SSL_FILETYPE_PEM);
    if(!ret) {
        throw SSLError("Bad private key (cannot read or decrypt file)");
    }
    SSL_CTX_use_RSAPrivateKey_file (SSLv3context, privateKeyFile.c_str(),  SSL_FILETYPE_PEM);

    ret = SSL_CTX_use_certificate_file   (TLSv1context, certificateFile.c_str(), SSL_FILETYPE_PEM);
    if(!ret) {
        throw SSLError("Bad certificate key (cannot read file)");
    }
    SSL_CTX_use_certificate_file   (SSLv3context, certificateFile.c_str(), SSL_FILETYPE_PEM);

    SSL_CTX_set_verify(TLSv1context, SSL_VERIFY_PEER | SSL_VERIFY_FAIL_IF_NO_PEER_CERT, sslVerifyCallback);
    SSL_CTX_set_verify(SSLv3context, SSL_VERIFY_PEER | SSL_VERIFY_FAIL_IF_NO_PEER_CERT, sslVerifyCallback);
#else
    throw RemoteError("libvos was not compiled with SSL/TLS support");
#endif
}

---