mangos-mirror/server
0
0
Fork 0
mirror of https://github.com/mangosfour/server.git synced 2025-12-17 16:37:00 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

[9517] rewrote RA console using ACE

Browse source 
- moved SQL pinging to SqlDelayThread
- use sAccountMgr instead of explict sql queries
This commit is contained in:
arrai 2010-03-01 23:27:05 +01:00
parent bc5092686e
commit 39559fc73a
8 changed files with 307 additions and 179 deletions
Download patch file Download diff file Expand all files Collapse all files

119
src/mangosd/Master.cpp
View file

@ -20,8 +20,6 @@
\ingroup mangosd
*/
#include <ace/OS_NS_signal.h>
#include "WorldSocketMgr.h"
#include "Common.h"
#include "Master.h"
@ -41,12 +39,9 @@
#include "revision_sql.h"
#include "MaNGOSsoap.h"
#include "sockets/TcpSocket.h"
#include "sockets/Utility.h"
#include "sockets/Parse.h"
#include "sockets/Socket.h"
#include "sockets/SocketHandler.h"
#include "sockets/ListenSocket.h"
#include <ace/OS_NS_signal.h>
#include <ace/TP_Reactor.h>
#include <ace/Dev_Poll_Reactor.h>
#ifdef WIN32
#include "ServiceWin32.h"
@ -115,74 +110,68 @@ public:
class RARunnable : public ACE_Based::Runnable
{
private:
ACE_Reactor *m_Reactor;
RASocket::Acceptor *m_Acceptor;
public:
uint32 numLoops, loopCounter;
RARunnable()
{
uint32 socketSelecttime = sWorld.getConfig (CONFIG_UINT32_SOCKET_SELECTTIME);
numLoops = (sConfig.GetIntDefault ("MaxPingTime", 30) * (MINUTE * 1000000 / socketSelecttime));
loopCounter = 0;
ACE_Reactor_Impl* imp = 0;
#if defined (ACE_HAS_EVENT_POLL) || defined (ACE_HAS_DEV_POLL)
imp = new ACE_Dev_Poll_Reactor ();
imp->max_notify_iterations (128);
imp->restart (1);
#else
imp = new ACE_TP_Reactor ();
imp->max_notify_iterations (128);
#endif
m_Reactor = new ACE_Reactor (imp, 1 /* 1= delete implementation so we don't have to care */);
m_Acceptor = new RASocket::Acceptor;
}
void checkping ()
~RARunnable()
{
// ping if need
if ((++loopCounter) == numLoops)
{
loopCounter = 0;
sLog.outDetail ("Ping MySQL to keep connection alive");
delete WorldDatabase.Query ("SELECT 1 FROM command LIMIT 1");
delete loginDatabase.Query ("SELECT 1 FROM realmlist LIMIT 1");
delete CharacterDatabase.Query ("SELECT 1 FROM bugreport LIMIT 1");
}
delete m_Reactor;
delete m_Acceptor;
}
void run ()
{
SocketHandler h;
// Launch the RA listener socket
ListenSocket<RASocket> RAListenSocket (h);
bool usera = sConfig.GetBoolDefault ("Ra.Enable", false);
if (usera)
{
port_t raport = sConfig.GetIntDefault ("Ra.Port", 3443);
uint16 raport = sConfig.GetIntDefault ("Ra.Port", 3443);
std::string stringip = sConfig.GetStringDefault ("Ra.IP", "0.0.0.0");
ipaddr_t raip;
if (!Utility::u2ip (stringip, raip))
sLog.outError ("MaNGOS RA can not bind to ip %s", stringip.c_str ());
else if (RAListenSocket.Bind (raip, raport))
sLog.outError ("MaNGOS RA can not bind to port %d on %s", raport, stringip.c_str ());
else
ACE_INET_Addr listen_addr(raport, stringip.c_str());
if (m_Acceptor->open (listen_addr, m_Reactor, ACE_NONBLOCK) == -1)
{
h.Add (&RAListenSocket);
sLog.outError ("MaNGOS RA can not bind to port %d on %s", raport, stringip.c_str ());
}
sLog.outString ("Starting Remote access listner on port %d on %s", raport, stringip.c_str ());
}
}
// Socket Selet time is in microseconds , not miliseconds!!
uint32 socketSelecttime = sWorld.getConfig (CONFIG_UINT32_SOCKET_SELECTTIME);
while (!m_Reactor->reactor_event_loop_done())
{
ACE_Time_Value interval (0, 10000);
// if use ra spend time waiting for io, if not use ra ,just sleep
if (usera)
if (m_Reactor->run_reactor_event_loop (interval) == -1)
break;
if(World::IsStopped())
{
while (!World::IsStopped())
{
h.Select (0, socketSelecttime);
checkping ();
}
}
else
{
while (!World::IsStopped())
{
ACE_Based::Thread::Sleep(static_cast<unsigned long> (socketSelecttime / 1000));
checkping ();
m_Acceptor->close();
break;
}
}
sLog.outString("RARunnable thread ended");
}
};
@ -244,7 +233,11 @@ int Master::Run()
cliThread = new ACE_Based::Thread(new CliRunnable);
}
ACE_Based::Thread rar_thread(new RARunnable);
ACE_Based::Thread* rar_thread = NULL;
if(sConfig.GetBoolDefault ("Ra.Enable", false))
{
rar_thread = new ACE_Based::Thread(new RARunnable);
}
///- Handle affinity for multiple processors and process priority on Windows
#ifdef WIN32
@ -316,7 +309,7 @@ int Master::Run()
}
///- Launch the world listener socket
port_t wsport = sWorld.getConfig (CONFIG_UINT32_PORT_WORLD);
uint16 wsport = sWorld.getConfig (CONFIG_UINT32_PORT_WORLD);
std::string bind_ip = sConfig.GetStringDefault ("BindIP", "0.0.0.0");
if (sWorldSocketMgr->StartNetwork (wsport, bind_ip.c_str ()) == -1)
@ -352,7 +345,13 @@ int Master::Run()
// when the main thread closes the singletons get unloaded
// since worldrunnable uses them, it will crash if unloaded after master
world_thread.wait();
rar_thread.wait ();
if(rar_thread)
{
rar_thread->wait();
rar_thread->destroy();
delete rar_thread;
}
///- Clean account database before leaving
clearOnlineAccounts();

290
src/mangosd/RASocket.cpp
View file

@ -31,186 +31,250 @@
#include "Language.h"
#include "ObjectMgr.h"
// TODO: drop old socket library and implement RASocket using ACE
/// RASocket constructor
RASocket::RASocket(ISocketHandler &h): TcpSocket(h)
RASocket::RASocket()
:RAHandler(),
pendingCommands(0, USYNC_THREAD, "pendingCommands"),
outActive(false),
inputBufferLen(0),
outputBufferLen(0),
stage(NONE)
{
///- Get the config parameters
bSecure = sConfig.GetBoolDefault( "RA.Secure", true );
iMinLevel = sConfig.GetIntDefault( "RA.MinLevel", 3 );
///- Initialize buffer and data
iInputLength=0;
stage=NONE;
iMinLevel = sConfig.GetIntDefault( "RA.MinLevel", SEC_ADMINISTRATOR );
reference_counting_policy ().value (ACE_Event_Handler::Reference_Counting_Policy::ENABLED);
}
/// RASocket destructor
RASocket::~RASocket()
{
sLog.outRALog("Connection was closed.\n");
peer().close();
sLog.outRALog("Connection was closed.");
}
/// Accept an incoming connection
void RASocket::OnAccept()
int RASocket::open(void* )
{
std::string ss=GetRemoteAddress();
sLog.outRALog("Incoming connection from %s.\n",ss.c_str());
if (reactor ()->register_handler(this, ACE_Event_Handler::READ_MASK | ACE_Event_Handler::WRITE_MASK) == -1)
{
sLog.outError ("RASocket::open: unable to register client handler errno = %s", ACE_OS::strerror (errno));
return -1;
}
ACE_INET_Addr remote_addr;
if (peer ().get_remote_addr (remote_addr) == -1)
{
sLog.outError ("RASocket::open: peer ().get_remote_addr errno = %s", ACE_OS::strerror (errno));
return -1;
}
sLog.outRALog("Incoming connection from %s.",remote_addr.get_host_addr());
///- print Motd
Sendf("%s\r\n",sWorld.GetMotd());
Sendf("\r\n%s",sObjectMgr.GetMangosStringForDBCLocale(LANG_RA_USER));
sendf(sWorld.GetMotd());
sendf("\r\n");
sendf(sObjectMgr.GetMangosStringForDBCLocale(LANG_RA_USER));
return 0;
}
int RASocket::close(int)
{
if(closing_)
return -1;
sLog.outDebug("RASocket::close");
shutdown();
closing_ = true;
remove_reference();
return 0;
}
int RASocket::handle_close (ACE_HANDLE h, ACE_Reactor_Mask)
{
if(closing_)
return -1;
sLog.outDebug("RASocket::handle_close");
ACE_GUARD_RETURN (ACE_Thread_Mutex, Guard, outBufferLock, -1);
closing_ = true;
if (h == ACE_INVALID_HANDLE)
peer ().close_writer ();
remove_reference();
return 0;
}
int RASocket::handle_output (ACE_HANDLE)
{
ACE_GUARD_RETURN (ACE_Thread_Mutex, Guard, outBufferLock, -1);
if(closing_)
return -1;
if (!outputBufferLen)
{
if(reactor()->cancel_wakeup(this, ACE_Event_Handler::WRITE_MASK) == -1)
{
sLog.outError ("RASocket::handle_output: error while cancel_wakeup");
return -1;
}
outActive = false;
return 0;
}
#ifdef MSG_NOSIGNAL
ssize_t n = peer ().send (outputBuffer, outputBufferLen, MSG_NOSIGNAL);
#else
ssize_t n = peer ().send (outputBuffer, outputBufferLen);
#endif // MSG_NOSIGNAL
if(n<=0)
return -1;
ACE_OS::memmove(outputBuffer, outputBuffer+n, outputBufferLen-n);
outputBufferLen -= n;
return 0;
}
/// Read data from the network
void RASocket::OnRead()
int RASocket::handle_input(ACE_HANDLE)
{
///- Read data and check input length
TcpSocket::OnRead();
unsigned int sz=ibuf.GetLength();
if (iInputLength+sz>=RA_BUFF_SIZE)
sLog.outDebug("RASocket::handle_input");
if(closing_)
{
sLog.outRALog("Input buffer overflow, possible DOS attack.\n");
SetCloseAndDelete();
return;
sLog.outError("Called RASocket::handle_input with closing_ = true");
return -1;
}
char *inp = new char [sz+1];
ibuf.Read(inp,sz);
size_t readBytes = peer().recv(inputBuffer+inputBufferLen, RA_BUFF_SIZE-inputBufferLen-1);
if(readBytes <= 0)
{
sLog.outDebug("read %u bytes in RASocket::handle_input", readBytes);
return -1;
}
///- Discard data after line break or line feed
bool gotenter=false;
unsigned int y=0;
for(;y<sz;y++)
for(; readBytes > 0 ; --readBytes)
{
if (inp[y]=='\r'||inp[y]=='\n')
char c = inputBuffer[inputBufferLen];
if (c=='\r'|| c=='\n')
{
gotenter=true;
break;
}
++inputBufferLen;
}
//No buffer overflow (checked above)
memcpy(&buff[iInputLength],inp,y);
iInputLength+=y;
delete [] inp;
if (gotenter)
{
buff[iInputLength]=0;
iInputLength=0;
inputBuffer[inputBufferLen]=0;
inputBufferLen=0;
switch(stage)
{
/// <ul> <li> If the input is 'USER <username>'
/// <ul> <li> If the input is '<username>'
case NONE:
{
///- If we're interactive we don't expect "USER " to be there
szLogin=&buff[0];
std::string szLogin=inputBuffer;
///- Get the gmlevel from the account table
std::string login = szLogin;
///- Convert Account name to Upper Format
AccountMgr::normalizeString(login);
///- Escape the Login to allow quotes in names
loginDatabase.escape_string(login);
QueryResult* result = loginDatabase.PQuery("SELECT gmlevel FROM account WHERE username = '%s'",login.c_str());
accId = sAccountMgr.GetId(szLogin);
///- If the user is not found, deny access
if(!result)
if(!accId)
{
Sendf("-No such user.\r\n");
sLog.outRALog("User %s does not exist.\n",szLogin.c_str());
sendf("-No such user.\r\n");
sLog.outRALog("User %s does not exist.",szLogin.c_str());
if(bSecure)
SetCloseAndDelete();
Sendf("\r\n%s",sObjectMgr.GetMangosStringForDBCLocale(LANG_RA_USER));
{
handle_output();
return -1;
}
sendf("\r\n");
sendf(sObjectMgr.GetMangosStringForDBCLocale(LANG_RA_USER));
}
else
{
Field *fields = result->Fetch();
AccountTypes sec = sAccountMgr.GetSecurity(accId);
///- if gmlevel is too low, deny access
if (fields[0].GetUInt32()<iMinLevel)
if (sec < iMinLevel)
{
Sendf("-Not enough privileges.\r\n");
sLog.outRALog("User %s has no privilege.\n",szLogin.c_str());
sendf("-Not enough privileges.\r\n");
sLog.outRALog("User %s has no privilege.",szLogin.c_str());
if(bSecure)
SetCloseAndDelete();
Sendf("\r\n%s",sObjectMgr.GetMangosStringForDBCLocale(LANG_RA_USER));
{
handle_output();
return -1;
}
sendf("\r\n");
sendf(sObjectMgr.GetMangosStringForDBCLocale(LANG_RA_USER));
}
else
{
stage=LG;
Sendf(sObjectMgr.GetMangosStringForDBCLocale(LANG_RA_PASS));
sendf(sObjectMgr.GetMangosStringForDBCLocale(LANG_RA_PASS));
}
delete result;
}
break;
}
///<li> If the input is 'PASS <password>' (and the user already gave his username)
///<li> If the input is '<password>' (and the user already gave his username)
case LG:
{ //login+pass ok
///- If password is correct, increment the number of active administrators
std::string login = szLogin;
std::string pw = inputBuffer;
///- If we're interactive we don't expect "PASS " to be there
std::string pw = &buff[0];
AccountMgr::normalizeString(login);
AccountMgr::normalizeString(pw);
loginDatabase.escape_string(login);
loginDatabase.escape_string(pw);
QueryResult *check = loginDatabase.PQuery(
"SELECT 1 FROM account WHERE username = '%s' AND sha_pass_hash=SHA1(CONCAT(username,':','%s'))",
login.c_str(), pw.c_str());
if (check)
if (sAccountMgr.CheckPassword(accId, pw))
{
delete check;
GetSocket();
stage=OK;
Sendf("+Logged in.\r\n");
sLog.outRALog("User %s has logged in.\n",szLogin.c_str());
Sendf("mangos>");
sendf("+Logged in.\r\n");
sLog.outRALog("User account %u has logged in.", accId);
sendf("mangos>");
}
else
{
///- Else deny access
Sendf("-Wrong pass.\r\n");
sLog.outRALog("User %s has failed to log in.\n",szLogin.c_str());
sendf("-Wrong pass.\r\n");
sLog.outRALog("User account %u has failed to log in.", accId);
if(bSecure)
SetCloseAndDelete();
Sendf("\r\n%s",sObjectMgr.GetMangosStringForDBCLocale(LANG_RA_PASS));
{
handle_output();
return -1;
}
sendf("\r\n");
sendf(sObjectMgr.GetMangosStringForDBCLocale(LANG_RA_PASS));
}
break;
}
///<li> If user is logged, parse and execute the command
case OK:
if (strlen(buff))
if (strlen(inputBuffer))
{
sLog.outRALog("Got '%s' cmd.\n",buff);
if (strncmp(buff,"quit",4)==0)
SetCloseAndDelete();
sLog.outRALog("Got '%s' cmd.",inputBuffer);
if (strncmp(inputBuffer,"quit",4)==0)
return -1;
else
{
SetDeleteByHandler(false);
CliCommandHolder* cmd = new CliCommandHolder(this, buff, &RASocket::zprint, &RASocket::commandFinished);
CliCommandHolder* cmd = new CliCommandHolder(this, inputBuffer, &RASocket::zprint, &RASocket::commandFinished);
sWorld.QueueCliCommand(cmd);
++pendingCommands;
pendingCommands.acquire();
}
}
else
Sendf("mangos>");
sendf("mangos>");
break;
///</ul>
};
}
// no enter yet? wait for next input...
return 0;
}
/// Output function
@ -219,17 +283,41 @@ void RASocket::zprint(void* callbackArg, const char * szText )
if( !szText )
return;
unsigned int sz=strlen(szText);
send(((RASocket*)callbackArg)->GetSocket(), szText, sz, 0);
((RASocket*)callbackArg)->sendf(szText);
}
void RASocket::commandFinished(void* callbackArg, bool success)
{
RASocket* raSocket = (RASocket*)callbackArg;
raSocket->Sendf("mangos>");
uint64 remainingCommands = --raSocket->pendingCommands;
if(remainingCommands == 0)
raSocket->SetDeleteByHandler(true);
raSocket->sendf("mangos>");
raSocket->pendingCommands.release();
}
int RASocket::sendf(const char* msg)
{
ACE_GUARD_RETURN (ACE_Thread_Mutex, Guard, outBufferLock, -1);
if(closing_)
return -1;
int msgLen = strlen(msg);
if(msgLen+outputBufferLen > RA_BUFF_SIZE)
return -1;
ACE_OS::memcpy(outputBuffer+outputBufferLen, msg, msgLen);
outputBufferLen += msgLen;
if(!outActive)
{
if (reactor ()->schedule_wakeup
(this, ACE_Event_Handler::WRITE_MASK) == -1)
{
sLog.outError ("RASocket::sendf error while schedule_wakeup");
return -1;
}
outActive = true;
}
return 0;
}

54
src/mangosd/RASocket.h
View file

@ -24,33 +24,59 @@
#define _RASOCKET_H
#include "Common.h"
#include "sockets/TcpSocket.h"
#include <ace/Synch_Traits.h>
#include <ace/Svc_Handler.h>
#include <ace/SOCK_Acceptor.h>
#include <ace/Acceptor.h>
#include <ace/Thread_Mutex.h>
#include <ace/Semaphore.h>
#define RA_BUFF_SIZE 1024
#define RA_BUFF_SIZE 8192
class ISocketHandler;
typedef ACE_Atomic_Op<ACE_SYNCH_MUTEX, uint64> AtomicInt;
/// Remote Administration socket
class RASocket: public TcpSocket
typedef ACE_Svc_Handler < ACE_SOCK_STREAM, ACE_NULL_SYNCH> RAHandler;
class RASocket: protected RAHandler
{
public:
ACE_Semaphore pendingCommands;
typedef ACE_Acceptor<RASocket, ACE_SOCK_ACCEPTOR > Acceptor;
friend class ACE_Acceptor<RASocket, ACE_SOCK_ACCEPTOR >;
RASocket(ISocketHandler& h);
~RASocket();
int sendf(const char*);
void OnAccept();
void OnRead();
protected:
/// things called by ACE framework.
RASocket(void);
virtual ~RASocket(void);
/// Called on open ,the void* is the acceptor.
virtual int open (void *);
/// Called on failures inside of the acceptor, don't call from your code.
virtual int close (int);
/// Called when we can read from the socket.
virtual int handle_input (ACE_HANDLE = ACE_INVALID_HANDLE);
/// Called when the socket can write.
virtual int handle_output (ACE_HANDLE = ACE_INVALID_HANDLE);
/// Called when connection is closed or error happens.
virtual int handle_close (ACE_HANDLE = ACE_INVALID_HANDLE,
ACE_Reactor_Mask = ACE_Event_Handler::ALL_EVENTS_MASK);
AtomicInt pendingCommands;
private:
bool outActive;
char buff[RA_BUFF_SIZE];
char inputBuffer[RA_BUFF_SIZE];
uint32 inputBufferLen;
std::string szLogin;
unsigned int iInputLength;
ACE_Thread_Mutex outBufferLock;
char outputBuffer[RA_BUFF_SIZE];
uint32 outputBufferLen;
uint32 accId;
bool bSecure; //kick on wrong pass, non exist. user OR user with no priv
//will protect from DOS, bruteforce attacks
//some 'smart' protection must be added for more security

1
src/shared/Database/Database.cpp
View file

@ -40,6 +40,7 @@ bool Database::Initialize(const char *)
m_logsDir.append("/");
}
m_pingIntervallms = sConfig.GetIntDefault ("MaxPingTime", 30) * (MINUTE * 1000);
return true;
}

3
src/shared/Database/Database.h
View file

@ -130,8 +130,11 @@ class MANGOS_DLL_SPEC Database
void SetResultQueue(SqlResultQueue * queue);
bool CheckRequiredField(char const* table_name, char const* required_name);
uint32 GetPingIntervall() { return m_pingIntervallms;}
private:
bool m_logSQL;
std::string m_logsDir;
uint32 m_pingIntervallms;
};
#endif

13
src/shared/Database/SqlDelayThread.cpp
View file

@ -30,17 +30,28 @@ void SqlDelayThread::run()
mysql_thread_init();
#endif
const uint32 loopSleepms = 10;
const uint32 pingEveryLoop = m_dbEngine->GetPingIntervall()/loopSleepms;
uint32 loopCounter = 0;
while (m_running)
{
// if the running state gets turned off while sleeping
// empty the queue before exiting
ACE_Based::Thread::Sleep(10);
ACE_Based::Thread::Sleep(loopSleepms);
SqlOperation* s;
while (m_sqlQueue.next(s))
{
s->Execute(m_dbEngine);
delete s;
}
if((loopCounter++) >= pingEveryLoop)
{
loopCounter = 0;
delete m_dbEngine->Query("SELECT 1");
}
}
#ifndef DO_POSTGRESQL

2
src/shared/Util.cpp
View file

@ -18,10 +18,10 @@
#include "Util.h"
#include "sockets/socket_include.h"
#include "utf8cpp/utf8.h"
#include "mersennetwister/MersenneTwister.h"
#include <ace/TSS_T.h>
#include <ace/INET_Addr.h>
typedef ACE_TSS<MTRand> MTRandTSS;
static MTRandTSS mtRand;

2
src/shared/revision_nr.h
View file

@ -1,4 +1,4 @@
#ifndef __REVISION_NR_H__
#define __REVISION_NR_H__
#define REVISION_NR "9516"
#define REVISION_NR "9517"
#endif // __REVISION_NR_H__