src/modules/extra/m_pgsql.cpp

   1 /*
   2  * InspIRCd -- Internet Relay Chat Daemon
   3  *
   4  *   Copyright (C) 2016 Adam <Adam@anope.org>
   5  *   Copyright (C) 2015 Daniel Vassdal <shutter@canternet.org>
   6  *   Copyright (C) 2013, 2016-2019 Sadie Powell <sadie@witchery.services>
   7  *   Copyright (C) 2012-2015 Attila Molnar <attilamolnar@hush.com>
   8  *   Copyright (C) 2012 Robby <robby@chatbelgie.be>
   9  *   Copyright (C) 2009-2010 Daniel De Graaf <danieldg@inspircd.org>
  10  *   Copyright (C) 2009 Uli Schlachter <psychon@inspircd.org>
  11  *   Copyright (C) 2008 Thomas Stagner <aquanight@inspircd.org>
  12  *   Copyright (C) 2007, 2009-2010 Craig Edwards <brain@inspircd.org>
  13  *   Copyright (C) 2007 Robin Burchell <robin+git@viroteck.net>
  14  *   Copyright (C) 2007 Dennis Friis <peavey@inspircd.org>
  15  *   Copyright (C) 2006 Oliver Lupton <om@inspircd.org>
  16  *
  17  * This file is part of InspIRCd.  InspIRCd is free software: you can
  18  * redistribute it and/or modify it under the terms of the GNU General Public
  19  * License as published by the Free Software Foundation, version 2.
  20  *
  21  * This program is distributed in the hope that it will be useful, but WITHOUT
  22  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
  23  * FOR A PARTICULAR PURPOSE.  See the GNU General Public License for more
  24  * details.
  25  *
  26  * You should have received a copy of the GNU General Public License
  27  * along with this program.  If not, see <http://www.gnu.org/licenses/>.
  28  */
  29
  30 /// $CompilerFlags: -Iexecute("pg_config --includedir" "POSTGRESQL_INCLUDE_DIR")
  31 /// $LinkerFlags: -Lexecute("pg_config --libdir" "POSTGRESQL_LIBRARY_DIR") -lpq
  32
  33 /// $PackageInfo: require_system("arch") postgresql-libs
  34 /// $PackageInfo: require_system("centos") postgresql-devel
  35 /// $PackageInfo: require_system("darwin") postgresql
  36 /// $PackageInfo: require_system("debian") libpq-dev
  37 /// $PackageInfo: require_system("ubuntu") libpq-dev
  38
  39
  40 #include "inspircd.h"
  41 #include <cstdlib>
  42 #include <libpq-fe.h>
  43 #include "modules/sql.h"
  44
  45 /* SQLConn rewritten by peavey to
  46  * use EventHandler instead of
  47  * BufferedSocket. This is much neater
  48  * and gives total control of destroy
  49  * and delete of resources.
  50  */
  51
  52 /* Forward declare, so we can have the typedef neatly at the top */
  53 class SQLConn;
  54 class ModulePgSQL;
  55
  56 typedef insp::flat_map<std::string, SQLConn*> ConnMap;
  57
  58 enum SQLstatus
  59 {
  60         // Connecting and wants read event.
  61         CREAD,
  62
  63         // Connecting and wants write event.
  64         CWRITE,
  65
  66         // Connected/working and wants read event.
  67         WREAD,
  68
  69         // Connected/working and wants write event.
  70         WWRITE
  71 };
  72
  73 class ReconnectTimer : public Timer
  74 {
  75  private:
  76         ModulePgSQL* mod;
  77  public:
  78         ReconnectTimer(ModulePgSQL* m) : Timer(5, false), mod(m)
  79         {
  80         }
  81         bool Tick(time_t TIME) CXX11_OVERRIDE;
  82 };
  83
  84 struct QueueItem
  85 {
  86         SQL::Query* c;
  87         std::string q;
  88         QueueItem(SQL::Query* C, const std::string& Q) : c(C), q(Q) {}
  89 };
  90
  91 /** PgSQLresult is a subclass of the mostly-pure-virtual class SQLresult.
  92  * All SQL providers must create their own subclass and define it's methods using that
  93  * database library's data retriveal functions. The aim is to avoid a slow and inefficient process
  94  * of converting all data to a common format before it reaches the result structure. This way
  95  * data is passes to the module nearly as directly as if it was using the API directly itself.
  96  */
  97
  98 class PgSQLresult : public SQL::Result
  99 {
 100         PGresult* res;
 101         int currentrow;
 102         int rows;
 103         std::vector<std::string> colnames;
 104
 105         void getColNames()
 106         {
 107                 colnames.resize(PQnfields(res));
 108                 for(unsigned int i=0; i < colnames.size(); i++)
 109                 {
 110                         colnames[i] = PQfname(res, i);
 111                 }
 112         }
 113  public:
 114         PgSQLresult(PGresult* result) : res(result), currentrow(0)
 115         {
 116                 rows = PQntuples(res);
 117                 if (!rows)
 118                         rows = ConvToNum<int>(PQcmdTuples(res));
 119         }
 120
 121         ~PgSQLresult()
 122         {
 123                 PQclear(res);
 124         }
 125
 126         int Rows() CXX11_OVERRIDE
 127         {
 128                 return rows;
 129         }
 130
 131         void GetCols(std::vector<std::string>& result) CXX11_OVERRIDE
 132         {
 133                 if (colnames.empty())
 134                         getColNames();
 135                 result = colnames;
 136         }
 137
 138         bool HasColumn(const std::string& column, size_t& index) CXX11_OVERRIDE
 139         {
 140                 if (colnames.empty())
 141                         getColNames();
 142
 143                 for (size_t i = 0; i < colnames.size(); ++i)
 144                 {
 145                         if (colnames[i] == column)
 146                         {
 147                                 index = i;
 148                                 return true;
 149                         }
 150                 }
 151                 return false;
 152         }
 153
 154         SQL::Field GetValue(int row, int column)
 155         {
 156                 char* v = PQgetvalue(res, row, column);
 157                 if (!v || PQgetisnull(res, row, column))
 158                         return SQL::Field();
 159
 160                 return SQL::Field(std::string(v, PQgetlength(res, row, column)));
 161         }
 162
 163         bool GetRow(SQL::Row& result) CXX11_OVERRIDE
 164         {
 165                 if (currentrow >= PQntuples(res))
 166                         return false;
 167                 int ncols = PQnfields(res);
 168
 169                 for(int i = 0; i < ncols; i++)
 170                 {
 171                         result.push_back(GetValue(currentrow, i));
 172                 }
 173                 currentrow++;
 174
 175                 return true;
 176         }
 177 };
 178
 179 /** SQLConn represents one SQL session.
 180  */
 181 class SQLConn : public SQL::Provider, public EventHandler
 182 {
 183  public:
 184         reference<ConfigTag> conf;      /* The <database> entry */
 185         std::deque<QueueItem> queue;
 186         PGconn*                 sql;            /* PgSQL database connection handle */
 187         SQLstatus               status;         /* PgSQL database connection status */
 188         QueueItem               qinprog;        /* If there is currently a query in progress */
 189
 190         SQLConn(Module* Creator, ConfigTag* tag)
 191                 : SQL::Provider(Creator, tag->getString("id"))
 192                 , conf(tag)
 193                 , sql(NULL)
 194                 , status(CWRITE)
 195                 , qinprog(NULL, "")
 196         {
 197                 if (!DoConnect())
 198                 {
 199                         ServerInstance->Logs->Log(MODNAME, LOG_DEFAULT, "WARNING: Could not connect to database " + tag->getString("id"));
 200                         DelayReconnect();
 201                 }
 202         }
 203
 204         CullResult cull() CXX11_OVERRIDE
 205         {
 206                 this->SQL::Provider::cull();
 207                 ServerInstance->Modules->DelService(*this);
 208                 return this->EventHandler::cull();
 209         }
 210
 211         ~SQLConn()
 212         {
 213                 SQL::Error err(SQL::BAD_DBID);
 214                 if (qinprog.c)
 215                 {
 216                         qinprog.c->OnError(err);
 217                         delete qinprog.c;
 218                 }
 219                 for(std::deque<QueueItem>::iterator i = queue.begin(); i != queue.end(); i++)
 220                 {
 221                         SQL::Query* q = i->c;
 222                         q->OnError(err);
 223                         delete q;
 224                 }
 225         }
 226
 227         void OnEventHandlerRead() CXX11_OVERRIDE
 228         {
 229                 DoEvent();
 230         }
 231
 232         void OnEventHandlerWrite() CXX11_OVERRIDE
 233         {
 234                 DoEvent();
 235         }
 236
 237         void OnEventHandlerError(int errornum) CXX11_OVERRIDE
 238         {
 239                 DelayReconnect();
 240         }
 241
 242         std::string GetDSN()
 243         {
 244                 std::ostringstream conninfo("connect_timeout = '5'");
 245                 std::string item;
 246
 247                 if (conf->readString("host", item))
 248                         conninfo << " host = '" << item << "'";
 249
 250                 if (conf->readString("port", item))
 251                         conninfo << " port = '" << item << "'";
 252
 253                 if (conf->readString("name", item))
 254                         conninfo << " dbname = '" << item << "'";
 255
 256                 if (conf->readString("user", item))
 257                         conninfo << " user = '" << item << "'";
 258
 259                 if (conf->readString("pass", item))
 260                         conninfo << " password = '" << item << "'";
 261
 262                 if (conf->getBool("ssl"))
 263                         conninfo << " sslmode = 'require'";
 264                 else
 265                         conninfo << " sslmode = 'disable'";
 266
 267                 return conninfo.str();
 268         }
 269
 270         bool DoConnect()
 271         {
 272                 sql = PQconnectStart(GetDSN().c_str());
 273                 if (!sql)
 274                         return false;
 275
 276                 if(PQstatus(sql) == CONNECTION_BAD)
 277                         return false;
 278
 279                 if(PQsetnonblocking(sql, 1) == -1)
 280                         return false;
 281
 282                 /* OK, we've initialised the connection, now to get it hooked into the socket engine
 283                 * and then start polling it.
 284                 */
 285                 this->fd = PQsocket(sql);
 286
 287                 if(this->fd <= -1)
 288                         return false;
 289
 290                 if (!SocketEngine::AddFd(this, FD_WANT_NO_WRITE | FD_WANT_NO_READ))
 291                 {
 292                         ServerInstance->Logs->Log(MODNAME, LOG_DEBUG, "BUG: Couldn't add pgsql socket to socket engine");
 293                         return false;
 294                 }
 295
 296                 /* Socket all hooked into the engine, now to tell PgSQL to start connecting */
 297                 return DoPoll();
 298         }
 299
 300         bool DoPoll()
 301         {
 302                 switch(PQconnectPoll(sql))
 303                 {
 304                         case PGRES_POLLING_WRITING:
 305                                 SocketEngine::ChangeEventMask(this, FD_WANT_POLL_WRITE | FD_WANT_NO_READ);
 306                                 status = CWRITE;
 307                                 return true;
 308                         case PGRES_POLLING_READING:
 309                                 SocketEngine::ChangeEventMask(this, FD_WANT_POLL_READ | FD_WANT_NO_WRITE);
 310                                 status = CREAD;
 311                                 return true;
 312                         case PGRES_POLLING_FAILED:
 313                                 return false;
 314                         case PGRES_POLLING_OK:
 315                                 SocketEngine::ChangeEventMask(this, FD_WANT_POLL_READ | FD_WANT_NO_WRITE);
 316                                 status = WWRITE;
 317                                 DoConnectedPoll();
 318                                 return true;
 319                         default:
 320                                 return true;
 321                 }
 322         }
 323
 324         void DoConnectedPoll()
 325         {
 326 restart:
 327                 while (qinprog.q.empty() && !queue.empty())
 328                 {
 329                         /* There's no query currently in progress, and there's queries in the queue. */
 330                         DoQuery(queue.front());
 331                         queue.pop_front();
 332                 }
 333
 334                 if (PQconsumeInput(sql))
 335                 {
 336                         if (PQisBusy(sql))
 337                         {
 338                                 /* Nothing happens here */
 339                         }
 340                         else if (qinprog.c)
 341                         {
 342                                 /* Fetch the result.. */
 343                                 PGresult* result = PQgetResult(sql);
 344
 345                                 /* PgSQL would allow a query string to be sent which has multiple
 346                                  * queries in it, this isn't portable across database backends and
 347                                  * we don't want modules doing it. But just in case we make sure we
 348                                  * drain any results there are and just use the last one.
 349                                  * If the module devs are behaving there will only be one result.
 350                                  */
 351                                 while (PGresult* temp = PQgetResult(sql))
 352                                 {
 353                                         PQclear(result);
 354                                         result = temp;
 355                                 }
 356
 357                                 /* ..and the result */
 358                                 PgSQLresult reply(result);
 359                                 switch(PQresultStatus(result))
 360                                 {
 361                                         case PGRES_EMPTY_QUERY:
 362                                         case PGRES_BAD_RESPONSE:
 363                                         case PGRES_FATAL_ERROR:
 364                                         {
 365                                                 SQL::Error err(SQL::QREPLY_FAIL, PQresultErrorMessage(result));
 366                                                 qinprog.c->OnError(err);
 367                                                 break;
 368                                         }
 369                                         default:
 370                                                 /* Other values are not errors */
 371                                                 qinprog.c->OnResult(reply);
 372                                 }
 373
 374                                 delete qinprog.c;
 375                                 qinprog = QueueItem(NULL, "");
 376                                 goto restart;
 377                         }
 378                         else
 379                         {
 380                                 qinprog.q.clear();
 381                         }
 382                 }
 383                 else
 384                 {
 385                         /* I think we'll assume this means the server died...it might not,
 386                          * but I think that any error serious enough we actually get here
 387                          * deserves to reconnect [/excuse]
 388                          * Returning true so the core doesn't try and close the connection.
 389                          */
 390                         DelayReconnect();
 391                 }
 392         }
 393
 394         void DelayReconnect();
 395
 396         void DoEvent()
 397         {
 398                 if((status == CREAD) || (status == CWRITE))
 399                 {
 400                         DoPoll();
 401                 }
 402                 else
 403                 {
 404                         DoConnectedPoll();
 405                 }
 406         }
 407
 408         void Submit(SQL::Query *req, const std::string& q) CXX11_OVERRIDE
 409         {
 410                 ServerInstance->Logs->Log(MODNAME, LOG_DEBUG, "Executing PostgreSQL query: " + q);
 411                 if (qinprog.q.empty())
 412                 {
 413                         DoQuery(QueueItem(req,q));
 414                 }
 415                 else
 416                 {
 417                         // wait your turn.
 418                         queue.push_back(QueueItem(req,q));
 419                 }
 420         }
 421
 422         void Submit(SQL::Query *req, const std::string& q, const SQL::ParamList& p) CXX11_OVERRIDE
 423         {
 424                 std::string res;
 425                 unsigned int param = 0;
 426                 for(std::string::size_type i = 0; i < q.length(); i++)
 427                 {
 428                         if (q[i] != '?')
 429                                 res.push_back(q[i]);
 430                         else
 431                         {
 432                                 if (param < p.size())
 433                                 {
 434                                         std::string parm = p[param++];
 435                                         std::vector<char> buffer(parm.length() * 2 + 1);
 436                                         int error;
 437                                         size_t escapedsize = PQescapeStringConn(sql, &buffer[0], parm.data(), parm.length(), &error);
 438                                         if (error)
 439                                                 ServerInstance->Logs->Log(MODNAME, LOG_DEBUG, "BUG: Apparently PQescapeStringConn() failed");
 440                                         res.append(&buffer[0], escapedsize);
 441                                 }
 442                         }
 443                 }
 444                 Submit(req, res);
 445         }
 446
 447         void Submit(SQL::Query *req, const std::string& q, const SQL::ParamMap& p) CXX11_OVERRIDE
 448         {
 449                 std::string res;
 450                 for(std::string::size_type i = 0; i < q.length(); i++)
 451                 {
 452                         if (q[i] != '$')
 453                                 res.push_back(q[i]);
 454                         else
 455                         {
 456                                 std::string field;
 457                                 i++;
 458                                 while (i < q.length() && isalnum(q[i]))
 459                                         field.push_back(q[i++]);
 460                                 i--;
 461
 462                                 SQL::ParamMap::const_iterator it = p.find(field);
 463                                 if (it != p.end())
 464                                 {
 465                                         std::string parm = it->second;
 466                                         std::vector<char> buffer(parm.length() * 2 + 1);
 467                                         int error;
 468                                         size_t escapedsize = PQescapeStringConn(sql, &buffer[0], parm.data(), parm.length(), &error);
 469                                         if (error)
 470                                                 ServerInstance->Logs->Log(MODNAME, LOG_DEBUG, "BUG: Apparently PQescapeStringConn() failed");
 471                                         res.append(&buffer[0], escapedsize);
 472                                 }
 473                         }
 474                 }
 475                 Submit(req, res);
 476         }
 477
 478         void DoQuery(const QueueItem& req)
 479         {
 480                 if (status != WREAD && status != WWRITE)
 481                 {
 482                         // whoops, not connected...
 483                         SQL::Error err(SQL::BAD_CONN);
 484                         req.c->OnError(err);
 485                         delete req.c;
 486                         return;
 487                 }
 488
 489                 if(PQsendQuery(sql, req.q.c_str()))
 490                 {
 491                         qinprog = req;
 492                 }
 493                 else
 494                 {
 495                         SQL::Error err(SQL::QSEND_FAIL, PQerrorMessage(sql));
 496                         req.c->OnError(err);
 497                         delete req.c;
 498                 }
 499         }
 500
 501         void Close()
 502         {
 503                 SocketEngine::DelFd(this);
 504
 505                 if(sql)
 506                 {
 507                         PQfinish(sql);
 508                         sql = NULL;
 509                 }
 510         }
 511 };
 512
 513 class ModulePgSQL : public Module
 514 {
 515  public:
 516         ConnMap connections;
 517         ReconnectTimer* retimer;
 518
 519         ModulePgSQL()
 520                 : retimer(NULL)
 521         {
 522         }
 523
 524         ~ModulePgSQL()
 525         {
 526                 delete retimer;
 527                 ClearAllConnections();
 528         }
 529
 530         void ReadConfig(ConfigStatus& status) CXX11_OVERRIDE
 531         {
 532                 ReadConf();
 533         }
 534
 535         void ReadConf()
 536         {
 537                 ConnMap conns;
 538                 ConfigTagList tags = ServerInstance->Config->ConfTags("database");
 539                 for(ConfigIter i = tags.first; i != tags.second; i++)
 540                 {
 541                         if (!stdalgo::string::equalsci(i->second->getString("module"), "pgsql"))
 542                                 continue;
 543                         std::string id = i->second->getString("id");
 544                         ConnMap::iterator curr = connections.find(id);
 545                         if (curr == connections.end())
 546                         {
 547                                 SQLConn* conn = new SQLConn(this, i->second);
 548                                 conns.insert(std::make_pair(id, conn));
 549                                 ServerInstance->Modules->AddService(*conn);
 550                         }
 551                         else
 552                         {
 553                                 conns.insert(*curr);
 554                                 connections.erase(curr);
 555                         }
 556                 }
 557                 ClearAllConnections();
 558                 conns.swap(connections);
 559         }
 560
 561         void ClearAllConnections()
 562         {
 563                 for(ConnMap::iterator i = connections.begin(); i != connections.end(); i++)
 564                 {
 565                         i->second->cull();
 566                         delete i->second;
 567                 }
 568                 connections.clear();
 569         }
 570
 571         void OnUnloadModule(Module* mod) CXX11_OVERRIDE
 572         {
 573                 SQL::Error err(SQL::BAD_DBID);
 574                 for(ConnMap::iterator i = connections.begin(); i != connections.end(); i++)
 575                 {
 576                         SQLConn* conn = i->second;
 577                         if (conn->qinprog.c && conn->qinprog.c->creator == mod)
 578                         {
 579                                 conn->qinprog.c->OnError(err);
 580                                 delete conn->qinprog.c;
 581                                 conn->qinprog.c = NULL;
 582                         }
 583                         std::deque<QueueItem>::iterator j = conn->queue.begin();
 584                         while (j != conn->queue.end())
 585                         {
 586                                 SQL::Query* q = j->c;
 587                                 if (q->creator == mod)
 588                                 {
 589                                         q->OnError(err);
 590                                         delete q;
 591                                         j = conn->queue.erase(j);
 592                                 }
 593                                 else
 594                                         j++;
 595                         }
 596                 }
 597         }
 598
 599         Version GetVersion() CXX11_OVERRIDE
 600         {
 601                 return Version("Provides the ability for SQL modules to query a PostgreSQL database.", VF_VENDOR);
 602         }
 603 };
 604
 605 bool ReconnectTimer::Tick(time_t time)
 606 {
 607         mod->retimer = NULL;
 608         mod->ReadConf();
 609         delete this;
 610         return false;
 611 }
 612
 613 void SQLConn::DelayReconnect()
 614 {
 615         ModulePgSQL* mod = (ModulePgSQL*)(Module*)creator;
 616         ConnMap::iterator it = mod->connections.find(conf->getString("id"));
 617         if (it != mod->connections.end())
 618         {
 619                 mod->connections.erase(it);
 620                 ServerInstance->GlobalCulls.AddItem((EventHandler*)this);
 621                 if (!mod->retimer)
 622                 {
 623                         mod->retimer = new ReconnectTimer(mod);
 624                         ServerInstance->Timers.AddTimer(mod->retimer);
 625                 }
 626         }
 627 }
 628
 629 MODULE_INIT(ModulePgSQL)