~psiphon-inc/psiphon/trunk

/*
 * Copyright (c) 2010, Psiphon Inc.
 * All rights reserved.
 *
 * 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/>.
 *
 */

#include "stdafx.h"
#include "psiclient.h"
#include "sshtunnel.h"
#include "locallistener.h"
#include "libssh2.h"
#include "sshsession.h"
#include "config.h"
#include "embeddedconfigvalues.h"

#include <ws2tcpip.h>
#include <winsock2.h>

struct Connection;
DWORD WINAPI listen(LPVOID pData);
void process_connection(Connection connection, bool readFromClient);

SSHTunnel::SSHTunnel(void)
{
    m_listenThread = 0;
    m_shuttingDown = false;
}

SSHTunnel::~SSHTunnel(void)
{
    TearDown();
}

void SSHTunnel::StartUp(void)
{
    TearDown();

    m_shuttingDown = false;

    int rc = libssh2_init(0);
    if (rc)
    {
        my_print(false, _T("libssh2 initialization failed: %d"), rc);
        return;
    }

    // Start the listener
    m_listenThread = CreateThread(NULL, 0, listen, (LPVOID)this, 0, NULL);

    // TODO: should wait for listener to start before changing proxy settings and launching IE?

    // Configure the proxy settings
    if (!m_proxySettings.Configure())
    {
        my_print(false, _T("error configuring proxy settings"));
    }
    else
    {
        // Start IE
        m_webBrowser.Open();

        my_print(false, _T("PsiphonX started"));
    }
}

void SSHTunnel::ShutDown(void)
{
    TearDown();

    my_print(false, _T("PsiphonX stopped"));
}

bool SSHTunnel::ShuttingDown(void)
{
    return m_shuttingDown;
}

void SSHTunnel::TearDown(void)
{
    // Stop IE
    m_webBrowser.Close();

    // Reset the proxy settings
    m_proxySettings.Revert();

    // Stop the listener by setting close flag
    // TODO: signal event?

    m_shuttingDown = true;

    if (m_listenThread)
    {
        WaitForSingleObject(m_listenThread, INFINITE);

        m_listenThread = 0;
    }

    m_shuttingDown = false;

    libssh2_exit();
}

struct Connection
{
    int socket;
    LIBSSH2_CHANNEL *channel;
};
typedef vector<Connection>::const_iterator connection_iter;

DWORD WINAPI listen(LPVOID pData)
{
    SSHTunnel* pTunnel = (SSHTunnel*)pData;
    LocalListener listener(LOCAL_LISTEN_IP, LOCAL_LISTEN_PORT);
    ssh_credentials credentials = {
        UnpadValue(SSH_SERVER_HOST),
        atoi(UnpadValue(SSH_SERVER_PORT_NUMBER).c_str()),
        UnpadValue(SSH_USER_NAME),
        UnpadValue(SSH_USER_PASSWORD),
        UnpadValue(SSH_SERVER_FINGERPRINT)};
    SSHSession ssh_session(credentials);
    vector<Connection> connections;
    bool bTearDownSession = false;

    WSADATA wsadata;
    int rc = WSAStartup(MAKEWORD(2,0), &wsadata);

    if (rc)
    {
        my_print(false, _T("WSAStartup failed: %d"), rc);
        return 0;
    }
 
    try
    {
        // initialize local listener

        listener.Initialize();

        while (true)
        {
            if (pTunnel->ShuttingDown())
            {
                // The main thread is signalling us to shut down
                throw 0;
            }

            // to recover from major session errors, reset everything

            if (bTearDownSession)
            {
                my_print(false, _T("reset session"));

                // destroy an existing session (and connections)
                for (connection_iter conn = connections.begin(); conn != connections.end(); ++conn)
                {
                    libssh2_channel_free(conn->channel); 
                    closesocket(conn->socket);
                }
                connections.clear();

                ssh_session.TearDown();

                bTearDownSession = false;
            }

            fd_set fds;
            FD_ZERO(&fds);

            // - select on the listen socket
            FD_SET(listener.ListenSocket(), &fds);

            // - select on all connection sockets to see if there's anything to read from the clients
            for (connection_iter conn = connections.begin(); conn != connections.end(); ++conn)
            {
                FD_SET(conn->socket, &fds);
            }

            struct timeval tv;
            tv.tv_sec = 0;
            tv.tv_usec = 1;
            int rc = select(0, &fds, NULL, NULL, &tv);
            if (SOCKET_ERROR == rc)
            {
                my_print(false, _T("select error: %d"), WSAGetLastError());
                throw 0;
            }

            // accept connections and add to connection list

            if (FD_ISSET(listener.ListenSocket(), &fds))
            {
                int forwardSocket = -1;
                listener.Accept(forwardSocket);

                // initialize ssh connection/session on demand

                if (!ssh_session.IsConnected() && !ssh_session.Connect())
                {
                    my_print(false, _T("ssh connection failed"));

                    // wait for more work (or shutdown signal) before retrying to connect
                    continue;
                }

                // create channel for this connection

                // in libssh2, session needs to be in blocking mode for this operation
                libssh2_session_set_blocking(ssh_session.GetSession(), 1);

                // TODO: do we need a new channel each time?
                LIBSSH2_CHANNEL *channel = libssh2_channel_direct_tcpip_ex(
                                    ssh_session.GetSession(), 
                                    HTTP_PROXY_HOST,
                                    atoi(UnpadValue(HTTP_PROXY_PORT_NUMBER).c_str()),
                                    LOCAL_LISTEN_IP,
                                    LOCAL_LISTEN_PORT);

                if (!channel)
                {
                    // after "Could not open the direct-tcpip channel" the session seems broken,
                    // so force a complete reset
                    bTearDownSession = true;

                    my_print(false, _T("Could not open the direct-tcpip channel"));
                            //"(Note that this can be a problem at the server!"
                            //" Please review the server logs.)");
                    
                    // go to the top of the loop where the session will be restarted
                    continue;
                }
 
                // Must use non-blocking IO hereafter due to the current libssh2 API
                libssh2_session_set_blocking(ssh_session.GetSession(), 0);

                // add this connection to list for processing

                Connection connection = {forwardSocket, channel};
                connections.push_back(connection);
            }

            // process each connection:
            // - the check for browser-side reads has already been done by select above
            // - also, check for ssh-side reads

            connection_iter conn = connections.begin();
            while (conn != connections.end())
            {
                // try block for errors related to this connection: will reset connection "conn" only

                try
                {
                    process_connection(*conn, 0 != FD_ISSET(conn->socket, &fds));
                    ++conn;
                }
                catch (...)
                {
                    // connection was closed or error; shutdown and remove from list

                    my_print(true, _T("closing connection"));

                    libssh2_channel_free(conn->channel);
 
                    closesocket(conn->socket);

                    conn = connections.erase(conn);
                }
            }
        }
    }
    catch (...)
    {
    }
    // TODO: Don't throw 0.  Throw and catch a std::exception.
    // Maybe put the error message in the exception.

    // close all connections
    
    for (connection_iter conn = connections.begin(); conn != connections.end(); ++conn)
    {
        libssh2_channel_free(conn->channel); 
        closesocket(conn->socket);
    }
    connections.clear();

    return 0;
}

void process_connection(Connection connection, bool readFromClient)
{
    ssize_t len, wr;
    char buf[16384];

    if (readFromClient)
    {
        len = recv(connection.socket, buf, sizeof(buf), 0);
        if (len < 0)
        {
            // errno=11 corresponds to "resource is not available" which apparently means
            // the peer closed the the socket seems to be reported a lot; perhaps IE is abandoning
            // the request due to delays.  for now, since IE seems to retry, only showing this
            // error in debug
            my_print((errno == 11), _T("read error: %s: %d"), _tcserror(errno), WSAGetLastError());
            throw 0;
        }
        else if (0 == len)
        {
            // TODO: correctly detect client shutdown?
            //my_print(_T("the client at %hs:%d disconnected"), shost, sport);
            my_print(true, _T("the client disconnected"));
            throw 0;
        }
        else
        {
            wr = 0;
            while (wr < len)
            {
                int i = libssh2_channel_write(connection.channel, buf, len);
                if (i == LIBSSH2_ERROR_EAGAIN)
                {
                    // TODO: wait?
                    continue;
                }
                else if (i < 0)
                {
                    my_print(false, _T("libssh2_channel_write error: %d"), i);
                    throw 0;
                }
                else if (i == 0)
                {
                    // TODO: need this condition?
                    break;
                }
                wr += i;
            }
        }
    }
    while (true)
    {
        len = libssh2_channel_read(connection.channel, buf, sizeof(buf));

        if (LIBSSH2_ERROR_EAGAIN == len)
        {
            break;
        }
        else if (len < 0)
        {
            my_print(false, _T("libssh2_channel_read error: %d"), len);
            throw 0;
        }
        wr = 0;
        while (wr < len)
        {
            int i = send(connection.socket, buf + wr, len - wr, 0);
            if (i <= 0)
            {
                // see comment above about errno=11
                my_print((errno == 11), _T("write error: %s: %d"), _tcserror(errno), WSAGetLastError());
                throw 0;
            }
            wr += i;
        }
        if (libssh2_channel_eof(connection.channel))
        {
            my_print(true,
                     _T("the server at %hs:%hs disconnected"),
                     HTTP_PROXY_HOST,
                     UnpadValue(HTTP_PROXY_PORT_NUMBER).c_str());
            throw 0;
        }
    }
}