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