root/nntp/connection.cc

/*  This file is part of Alue, the multiprotocol Internet discussion daemon

    Copyright © 2009, 2010 Antti-Juhani Kaijanaho

    Alue 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.

    Alue 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 Alue.  If not, see <http://www.gnu.org/licenses/>.

 */

#include "connection.hh"
#include "lexutils.hh"

#include "../db/user.hh"
#include "../logger/logline.hh"
#include "../tls/session_impl.hh"

#include <boost/version.hpp>
#include <boost/date_time.hpp>
#include <map>
#include <sstream>

#if BOOST_VERSION < 103600
#  include "../buffers_iterator.hpp"
#endif

#include "../assert.hh"

namespace nntp
{
        connection::connection(std::string port_name,
                               bool initial_ssl,
                               boost::asio::ip::tcp::acceptor &acc,
                               tls::init(tls_init),
                               server::conn_cb srv_cb)
                : port_name(port_name)
                , initial_ssl(initial_ssl)
                , current_article(0)
                , identified(false)
                , authenticated(false)
                , startup_timestamp
                  (boost::posix_time::second_clock::universal_time())
                , client_activity_timestamp
                  (boost::posix_time::second_clock::universal_time())
                , num_commands_served(0)
                , num_500_responses(0)
                , capabilities_used(false)
                , srv_cb(srv_cb)
                , ios(acc.get_io_service())
                , peer(ios)
                , speer(peer, tls_init, tls::server)
                , tls_active(false)
        {
                acc.async_accept(peer, ep,
                                 boost::bind(&connection::accept, this, _1));
        }

        connection::cmds_type *connection::cmds;

        const connection::overview_fmt_type connection::overview_fmt[] = {
                { "Subject", false, },
                { "From", false, },
                { "Date", false, },
                { "Message-ID", false, },
                { "References", false, },
                { "Bytes", false, },
                { "Lines", false, },
                { "Xref", true }
        };
        const size_t connection::overview_fmt_count =
                sizeof overview_fmt / sizeof *overview_fmt;

        void connection::abort(boost::system::error_code ec)
        {
                logger::logline ll;
                ll << loghead
                   << "failure (" 
                   << ec.message()
                   << "), aborting connection";
                ll.close();
                kill();
        }

        void connection::kill()
        {
                stats();
                speer.kill();
                peer.cancel();
                srv_cb.terminal(shared_from_this());
                // reset these pointers to break pointer loops
                terminate.reset();
                dispatch.reset();
                starttls.reset();
        }

        void connection::terminator::done(ptr, boost::system::error_code)
        {
                cp->kill();
        }

        void connection::terminator::sent(ptr lifeline,
                                          boost::system::error_code ec)
        {
                if (!cp->tls_active) 
                {
                        done(lifeline, ec); 
                        return; 
                }
                
                logger::logline ll;
                ll << cp->loghead << "shutting down TLS";
                ll.close();

                cp->speer.async_shutdown
                        (boost::bind(&connection::terminator::done,
                                     this, lifeline, _1)); 
        }

        void connection::accept(boost::system::error_code ec)
        {
                if (ec) { abort(ec); return; }

                terminate.reset(new terminator(shared_from_this()));
                dispatch.reset(new dispatcher(shared_from_this()));
                starttls.reset(new tls_starter(shared_from_this()));

                startup_timestamp =
                        boost::posix_time::second_clock::universal_time();

                std::ostringstream ss;
                ss << "nntpd for " << ep << " ";
                loghead = ss.str();
                
                logger::logline ll;
                ll << loghead << "starts";
                ll.close();
                logger::logger.flush();

                srv_cb.active(shared_from_this());

                if (initial_ssl)
                {
                        ll.open();
                        ll << loghead << "starting TLS negotiation";
                        ll.close();

                        tls_active = true;
                        speer.async_handshake(bind(&connection::greet,
                                                   this, _1));
                }
                else
                {
                        greet(ec);
                }
        }

        void connection::greet(boost::system::error_code ec)
        {
                using boost::asio::async_write;

                if (ec) { abort(ec); return; }

                std::string msg;

                send_line("200 posting requires authentication");
                flush_writebuf(dispatch);
        }

        void connection::flush_writebuf(continuation::ptr cont)
        {
                using boost::asio::async_write;
                using boost::bind;
                using boost::asio::buffer;

                if (!cont)
                {
                        logger::logline ll;
                        ll << loghead << "has a null continuation";
                        return;
                }

                // to work around a weird bug in the TLS wrapper...
                if (writebuf.size() == 0) 
                {
                        ios.post(bind(&continuation::sent,
                                      cont.get(), cont,
                                      boost::system::error_code()));
                        return;
                }

                if (srv_cb.do_log_protocol_details())
                {
                        logger::logline ll;
                        ll << loghead << "==> ";
                        for (size_t i = 0; i < writebuf.size(); i++)
                        {
                                if (writebuf[i] == '\r') continue;
                                if (writebuf[i] == '\n')
                                {
                                        if (i + 1 < writebuf.size())
                                        {
                                                ll.nl();
                                                ll << loghead << "==> ";
                                        }
                                }
                                else
                                        ll << writebuf[i];
                        }
                        ll.close();
                }

                async_write(speer, buffer(writebuf.data(), writebuf.size()),
                            bind(&continuation::sent,
                                 cont.get(), cont, _1));
        }

        void connection::dispatcher::received(ptr, boost::system::error_code ec,
                                              std::size_t received_bytes)
        {
                typedef boost::asio::buffers_iterator
                        <boost::asio::streambuf::const_buffers_type> sbit;
                using boost::bind;

                if (ec) { cp->abort(ec); return; }

                cp->client_activity_timestamp =
                        boost::posix_time::second_clock::universal_time();

                cp->readbuf.commit(received_bytes);

                // if no CRLF yet, read more
                boost::asio::streambuf::const_buffers_type sbuf = 
                        cp->readbuf.data();
                sbit beginit = sbit::begin(sbuf);
                sbit endit = sbit::end(sbuf);
                sbit crlf = sbit::begin(sbuf);
                for (/**/; crlf != endit; crlf++) 
                {
                        if (crlf == beginit) continue;
                        if (crlf[-1] != '\r' || crlf[0] != '\n') continue;
                        break;
                }
                if (crlf == endit)
                {
                        std::string line;
                        for (sbit it = beginit; it != crlf; it++)
                        {
                                line += *it;
                        }
                        cp->speer.async_read_some
                                (cp->readbuf.prepare(1024),
                                 bind(&connection::dispatcher::received,
                                      this, shared_from_this(), 
                                      _1, _2));

                        return;
                }
                
                assert(crlf != beginit);
                assert(crlf != endit);
                assert(crlf[-1] == '\r');
                assert(crlf[ 0] == '\n');
                crlf--;

                std::string line;
                for (sbit it = beginit; it != crlf; it++)
                {
                        assert(it != endit);
                        line += *it;
                }
                cp->readbuf.consume(line.length() + 2); // 2 is for the CRLF

                if (cp->srv_cb.do_log_protocol_details())
                {
                        std::string li = line;
                        util::strip_crlf(li);
                        logger::logline ll;
                        ll << cp->loghead << "<== " << li;
                        ll.close();
                }


                if (line.length() != 0 && line[line.length()-1] == '\r')
                        line.erase(line.length()-1);

                std::string cmd = split(line);
                boost::to_lower(cmd);

                std::map<std::string, command *>::const_iterator it =
                        cp->cmds->find(cmd);

                cp->num_commands_served += 1;

                if (it == cp->cmds->end()) {
                        logger::logline ll;
                        ll << cp->loghead << "received unknown command " << cmd;
                        ll.close();
                        cp->num_500_responses += 1;
                        cp->send_line("500 unrecognized command"); 
                        cp->flush_writebuf(shared_from_this());
                        return;
                }

                std::vector<std::string> args;
                while (!line.empty()) {
                        std::string arg = split(line);
                        if (!arg.empty()) args.push_back(arg);
                }
                continuation::ptr cont;
                try
                {
                        cont = it->second->perform(cp->shared_from_this(),
                                                   args.data(), args.size());
                }
                catch (std::exception &e)
                {
                        logger::logline ll;
                        ll << cp->loghead
                           << "exception caught: "
                           << e.what()
                           << " (" << typeid(e).name() << ")";
                        ll.close();
                        cp->writebuf.clear();
                        cp->send_line("403 server error, please try again");
                        cont = shared_from_this();
                }
                cp->flush_writebuf(cont);
        }

        void connection::dispatcher::sent(ptr, boost::system::error_code ec)
        {
                using boost::asio::async_read_until;
                using boost::bind;

                if (ec) { cp->abort(ec); return; }

                cp->writebuf.clear();  /* we are only called after a
                                        * complete writebuf flush */


                if (cp->readbuf.size() == 0)
                {
                        cp->speer.async_read_some
                                (cp->readbuf.prepare(8),
                                 bind(&connection::dispatcher::received,
                                      this, shared_from_this(), 
                                      _1, _2));
                }
                else
                        received(shared_from_this(), ec, 0);
        }

        void connection::reader::received(ptr lifeline,
                                          size_t scanned,
                                          boost::system::error_code ec,
                                          std::size_t received_bytes)
        {
                using boost::bind;
                typedef boost::asio::buffers_iterator
                        <boost::asio::streambuf::const_buffers_type> sbit;

                if (ec) { cp->abort(ec); return; }

                cp->readbuf.commit(received_bytes);

                cp->client_activity_timestamp =
                        boost::posix_time::second_clock::universal_time();

                bool done = false;
                boost::asio::streambuf::const_buffers_type sbuf = 
                        cp->readbuf.data();
                sbit it = sbit::begin(sbuf) + scanned;
                sbit endit = sbit::end(sbuf);
                for (/**/; it != endit; it++, scanned++)
                {
                        if (*it != '.') continue;
                        if (0 < scanned && scanned < 2) continue;
                        if (it[-2] != '\r' || it[-1] != '\n') continue;
                        sbit it1 = it+1;
                        if (it1 == endit) continue;
                        sbit it2 = it1+1;
                        if (it2 == endit) continue;
                        if (*it1 != '\r' || *it2 != '\n') continue;
                        // we now have "." CRLF at the beginning of line
                        done = true;
                        break;
                }
                if (!done)
                {
                        /* The above loop may end in a state where
                         * it's past the '.' but hasn't seen enough to
                         * decide that it's starting the
                         * end-of-message mark.  */
                        /**/ if (scanned >= 2 && it[-2] == '.') scanned -= 2;
                        else if (scanned >= 1 && it[-1] == '.') scanned -= 1;
                        
                        cp->speer.async_read_some
                                (cp->readbuf.prepare(64),
                                 bind(&connection::reader::received,
                                      this, lifeline, scanned, _1, _2));
                        return;
                }

                continuation::ptr cont;

                std::string data;
                bool cr = false;
                it = sbit::begin(sbuf);
                for (size_t i = 0; i < scanned; i++, it++)
                {
                        bool oldcr = cr;
                        cr = false;
                        char c = *it;
                        switch (c) 
                        {
                        default:
                                break;
                        case '\r':
                                if (oldcr) goto fail;
                                cr = true;
                                break;
                        case '\n':
                                if (!oldcr) goto fail;
                                break;
                        case '\0':
                                goto fail;
                        }
                        data.push_back(c);
                }
                cp->readbuf.consume(scanned+3); // the 3 is the final "." CRLF

                if (cp->srv_cb.do_log_protocol_details())
                {
                        logger::logline ll;
                        ll << cp->loghead << "<== ";
                        for (size_t i = 0; i < data.size(); i++)
                        {
                                if (data[i] == '\r') continue;
                                if (data[i] == '\n')
                                {
                                        if (i + 1 < data.size())
                                        {
                                                ll.nl();
                                                ll << cp->loghead << "<== ";
                                        }
                                }
                                else
                                        ll << data[i];
                        }
                        ll.close();
                }


                cont = (*h)(data);
                cp->flush_writebuf(cont);
                return;
        fail:
                logger::logline ll;
                ll << cp->loghead << "received invalid multi-line data";
                ll.close();
                cp->srv_cb.terminal(cp->shared_from_this());
        }

        void connection::reader::sent(ptr lifeline,
                                      boost::system::error_code ec)
        {
                using boost::asio::async_read_until;
                using boost::bind;

                if (ec) { cp->abort(ec); return; }

                cp->writebuf.clear();  /* we are only called after a
                                        * complete writebuf flush */


                cp->speer.async_read_some
                        (cp->readbuf.prepare(64),
                         bind(&connection::reader::received,
                              this, lifeline, 0, _1, _2));
        }

        void connection::tls_starter::sent(ptr, boost::system::error_code ec)
        {
                using boost::asio::async_read_until;
                using boost::bind;

                if (ec) { cp->abort(ec); return; }

                cp->writebuf.clear();  /* we are only called after a
                                        * complete writebuf flush */

                logger::logline ll;
                ll << cp->loghead << "starting TLS negotiation";
                ll.close();

                cp->current_group.reset();
                cp->current_article = 0;
                cp->tls_active = true;
                cp->speer.async_handshake(bind(&connection::dispatcher::sent,
                                               (dispatcher*)cp->dispatch.get(),
                                               cp->dispatch,
                                               _1));
        }

        void connection::failure_delayer::waited(ptr, boost::system::error_code)
        {
                cp->send_line(line);
                cp->flush_writebuf(cp->dispatch);
        }

        void connection::failure_delayer::sent(ptr lifeline, 
                                               boost::system::error_code ec)
        {
                if (ec) { cp->abort(ec); return; }

                cp->writebuf.clear();  /* we are only called after a
                                        * complete writebuf flush */

                timer.expires_from_now(boost::posix_time::seconds(5));
                timer.async_wait(boost::bind
                                 (&connection::failure_delayer::waited,
                                  this, lifeline, _1));
        }

        connection::command::~command() {}

        void connection::stats()
        {
                using namespace boost::posix_time;
                ptime now = second_clock::universal_time();
                time_duration idle = now - client_activity_timestamp;

                logger::logline ll;
                ll << loghead 
                   << (tls_active ? "TLS; " : "")
                   << (capabilities_used ? "CAPABILITIES; " : "")
                   << "age " 
                   << now - startup_timestamp
                   << ", no client activity in "
                   << idle
                   << "; "
                   << num_commands_served 
                   << " commands served; "
                   << num_500_responses
                   << " unknown commands; "
                   << (identified ? "identified; " : "not identified; ")
                   << (authenticated ? "authenticated" : "not authenticated");
        }

        void connection::tick(bool log)
        {
                if (log) stats();
                using namespace boost::posix_time;
                ptime now = second_clock::universal_time();
                time_duration idle = now - client_activity_timestamp;
                if (idle > hours(2) && dispatch)
                {
                        logger::logline ll;
                        ll << loghead
                           << "dropping idle connection";
                        ll.close();
                        terminate.reset();
                        dispatch.reset();
                        starttls.reset();
                        kill();
                }
        }

        connection::~connection()
        {
                try
                {
                        logger::logline ll;
                        ll << loghead << "dying";
                        ll.close();
                } catch (std::exception)
                {}
        }

};