EQStream.cpp 54 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818181918201821182218231824182518261827182818291830183118321833183418351836183718381839184018411842184318441845184618471848184918501851185218531854185518561857185818591860186118621863186418651866186718681869187018711872187318741875187618771878187918801881188218831884188518861887188818891890189118921893189418951896189718981899190019011902190319041905
  1. /*
  2. EQ2Emulator: Everquest II Server Emulator
  3. Copyright (C) 2007 EQ2EMulator Development Team (http://www.eq2emulator.net)
  4. This file is part of EQ2Emulator.
  5. EQ2Emulator is free software: you can redistribute it and/or modify
  6. it under the terms of the GNU General Public License as published by
  7. the Free Software Foundation, either version 3 of the License, or
  8. (at your option) any later version.
  9. EQ2Emulator is distributed in the hope that it will be useful,
  10. but WITHOUT ANY WARRANTY; without even the implied warranty of
  11. MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  12. GNU General Public License for more details.
  13. You should have received a copy of the GNU General Public License
  14. along with EQ2Emulator. If not, see <http://www.gnu.org/licenses/>.
  15. */
  16. #ifdef WIN32
  17. #include <WinSock2.h>
  18. #include <windows.h>
  19. #endif
  20. #include "debug.h"
  21. #include <string>
  22. #include <iomanip>
  23. #include <iostream>
  24. #include <vector>
  25. #include <time.h>
  26. #include <sys/types.h>
  27. #ifdef WIN32
  28. #include <time.h>
  29. #else
  30. #include <sys/socket.h>
  31. #include <netinet/in.h>
  32. #include <sys/time.h>
  33. #include <sys/socket.h>
  34. #include <netdb.h>
  35. #include <fcntl.h>
  36. #include <arpa/inet.h>
  37. #endif
  38. #include "EQPacket.h"
  39. #include "EQStream.h"
  40. #include "EQStreamFactory.h"
  41. #include "misc.h"
  42. #include "Mutex.h"
  43. #include "op_codes.h"
  44. #include "CRC16.h"
  45. #include "packet_dump.h"
  46. #ifdef LOGIN
  47. #include "../LoginServer/login_structs.h"
  48. #endif
  49. #include "EQ2_Common_Structs.h"
  50. #include "Log.h"
  51. //#define DEBUG_EMBEDDED_PACKETS 1
  52. uint16 EQStream::MaxWindowSize=2048;
  53. void EQStream::init(bool resetSession) {
  54. if (resetSession)
  55. {
  56. streamactive = false;
  57. sessionAttempts = 0;
  58. }
  59. timeout_delays = 0;
  60. MInUse.lock();
  61. active_users = 0;
  62. MInUse.unlock();
  63. Session=0;
  64. Key=0;
  65. MaxLen=0;
  66. NextInSeq=0;
  67. NextOutSeq=0;
  68. CombinedAppPacket=NULL;
  69. MAcks.lock();
  70. MaxAckReceived = -1;
  71. NextAckToSend = -1;
  72. LastAckSent = -1;
  73. MAcks.unlock();
  74. LastSeqSent=-1;
  75. MaxSends=5;
  76. LastPacket=Timer::GetCurrentTime2();
  77. oversize_buffer=NULL;
  78. oversize_length=0;
  79. oversize_offset=0;
  80. Factory = NULL;
  81. rogue_buffer=NULL;
  82. roguebuf_offset=0;
  83. roguebuf_size=0;
  84. MRate.lock();
  85. RateThreshold=RATEBASE/250;
  86. DecayRate=DECAYBASE/250;
  87. MRate.unlock();
  88. BytesWritten=0;
  89. SequencedBase = 0;
  90. AverageDelta = 500;
  91. crypto->setRC4Key(0);
  92. retransmittimer = Timer::GetCurrentTime2();
  93. retransmittimeout = 500 * RETRANSMIT_TIMEOUT_MULT;
  94. reconnectAttempt = 0;
  95. if (uint16(SequencedBase + SequencedQueue.size()) != NextOutSeq) {
  96. LogWrite(PACKET__DEBUG, 9, "Packet", "init Invalid Sequenced queue: BS %u + SQ %u != NOS %u", SequencedBase, SequencedQueue.size(), NextOutSeq);
  97. }
  98. }
  99. EQStream::EQStream(sockaddr_in addr){
  100. crypto = new Crypto();
  101. resend_que_timer = new Timer(1000);
  102. combine_timer = new Timer(250); //250 milliseconds
  103. combine_timer->Start();
  104. resend_que_timer->Start();
  105. init();
  106. remote_ip=addr.sin_addr.s_addr;
  107. remote_port=addr.sin_port;
  108. State=CLOSED;
  109. StreamType=UnknownStream;
  110. compressed=true;
  111. encoded=false;
  112. app_opcode_size=2;
  113. #ifdef WIN32
  114. ZeroMemory(&stream, sizeof(z_stream));
  115. #else
  116. bzero(&stream, sizeof(z_stream));
  117. #endif
  118. stream.zalloc = (alloc_func)0;
  119. stream.zfree = (free_func)0;
  120. stream.opaque = (voidpf)0;
  121. deflateInit2(&stream, 9, Z_DEFLATED, 13, 9, Z_DEFAULT_STRATEGY);
  122. //deflateInit(&stream, 5);
  123. compressed_offset = 0;
  124. client_version = 0;
  125. received_packets = 0;
  126. sent_packets = 0;
  127. #ifdef WRITE_PACKETS
  128. write_packets = 0;
  129. char write_packets_filename[64];
  130. snprintf(write_packets_filename, sizeof(write_packets_filename), "PacketLog%i.log", Timer::GetCurrentTime2());
  131. write_packets = fopen(write_packets_filename, "w+");
  132. #endif
  133. }
  134. EQProtocolPacket* EQStream::ProcessEncryptedData(uchar* data, int32 size, int16 opcode){
  135. //cout << "B4:\n";
  136. //DumpPacket(data, size);
  137. /*if(size >= 2 && data[0] == 0 && data[1] == 0){
  138. cout << "Attempting to fix packet!\n";
  139. //Have to fix bad packet from client or it will screw up encryption :P
  140. size--;
  141. data++;
  142. }*/
  143. crypto->RC4Decrypt(data,size);
  144. int8 offset = 0;
  145. if(data[0] == 0xFF && size > 2){
  146. offset = 3;
  147. memcpy(&opcode, data+sizeof(int8), sizeof(int16));
  148. }
  149. else{
  150. offset = 1;
  151. memcpy(&opcode, data, sizeof(int8));
  152. }
  153. //cout << "After:\n";
  154. //DumpPacket(data, size);
  155. return new EQProtocolPacket(opcode, data+offset, size - offset);
  156. }
  157. EQProtocolPacket* EQStream::ProcessEncryptedPacket(EQProtocolPacket *p){
  158. EQProtocolPacket* ret = NULL;
  159. if(p->opcode == OP_Packet && p->size > 2)
  160. ret = ProcessEncryptedData(p->pBuffer+2, p->size-2, p->opcode);
  161. else
  162. ret = ProcessEncryptedData(p->pBuffer, p->size, p->opcode);
  163. return ret;
  164. }
  165. bool EQStream::ProcessEmbeddedPacket(uchar* pBuffer, int16 length,int8 opcode) {
  166. if(!pBuffer || !crypto->isEncrypted())
  167. return false;
  168. MCombineQueueLock.lock();
  169. EQProtocolPacket* newpacket = ProcessEncryptedData(pBuffer, length, opcode);
  170. MCombineQueueLock.unlock();
  171. if (newpacket) {
  172. #ifdef DEBUG_EMBEDDED_PACKETS
  173. printf("Opcode: %u\n", newpacket->opcode);
  174. DumpPacket(newpacket->pBuffer, newpacket->size);
  175. #endif
  176. EQApplicationPacket* ap = newpacket->MakeApplicationPacket(2);
  177. if (ap->version == 0)
  178. ap->version = client_version;
  179. InboundQueuePush(ap);
  180. #ifdef WRITE_PACKETS
  181. WritePackets(ap->GetOpcodeName(), pBuffer, length, false);
  182. #endif
  183. safe_delete(newpacket);
  184. return true;
  185. }
  186. return false;
  187. }
  188. bool EQStream::HandleEmbeddedPacket(EQProtocolPacket *p, int16 offset, int16 length){
  189. if(!p)
  190. return false;
  191. #ifdef DEBUG_EMBEDDED_PACKETS
  192. // printf works better with DumpPacket
  193. printf( "Start Packet with offset %u, length %u, p->size %u\n", offset, length, p->size);
  194. #endif
  195. if(p->size >= ((uint32)(offset+2))){
  196. if(p->pBuffer[offset] == 0 && p->pBuffer[offset+1] == 0x19){
  197. if(length == 0)
  198. length = p->size-2-offset;
  199. else
  200. length-=2;
  201. #ifdef DEBUG_EMBEDDED_PACKETS
  202. printf( "Creating OP_AppCombined Packet with offset %u, length %u, p->size %u\n", offset, length, p->size);
  203. DumpPacket(p->pBuffer, p->size);
  204. #endif
  205. EQProtocolPacket *subp=new EQProtocolPacket(OP_AppCombined, p->pBuffer+2+offset, length);
  206. subp->copyInfo(p);
  207. ProcessPacket(subp, p);
  208. safe_delete(subp);
  209. return true;
  210. }
  211. else if (p->pBuffer[offset] == 0 && p->pBuffer[offset + 1] == 0) {
  212. if (length == 0)
  213. length = p->size - 1 - offset;
  214. else
  215. length--;
  216. #ifdef DEBUG_EMBEDDED_PACKETS
  217. printf( "Creating Opcode 0 Packet!");
  218. DumpPacket(p->pBuffer + 1 + offset, length);
  219. #endif
  220. uchar* buffer = (p->pBuffer + 1 + offset);
  221. bool valid = ProcessEmbeddedPacket(buffer, length);
  222. if(valid)
  223. return true;
  224. }
  225. else if(offset+4 < p->size && ntohl(*(uint32 *)(p->pBuffer+offset)) != 0xffffffff) {
  226. #ifdef DEBUG_EMBEDDED_PACKETS
  227. uint16 seq = NextInSeq-1;
  228. sint8 check = 0;
  229. if(offset == 2) {
  230. seq=ntohs(*(uint16 *)(p->pBuffer));
  231. check=CompareSequence(NextInSeq,seq);
  232. }
  233. printf( "Unhandled Packet with offset %u, length %u, p->size %u, check: %i, nextinseq: %u, seq: %u\n", offset, length, p->size, check, NextInSeq, seq);
  234. DumpPacket(p->pBuffer, p->size);
  235. #endif
  236. if(length == 0)
  237. length = p->size - offset;
  238. uchar* buffer = (p->pBuffer + offset);
  239. bool valid = ProcessEmbeddedPacket(buffer, length);
  240. if(valid)
  241. return true;
  242. }
  243. else if(p->pBuffer[offset] != 0xff && p->pBuffer[offset+1] == 0xff) {
  244. uint8 new_length = 0;
  245. memcpy(&new_length, p->pBuffer+offset, sizeof(int8));
  246. if((new_length+offset+2) == p->size) {
  247. new_length -= 2;
  248. EQProtocolPacket *subp=new EQProtocolPacket(p->pBuffer+offset+2, new_length, OP_Packet);
  249. subp->copyInfo(p);
  250. ProcessPacket(subp, p);
  251. delete subp;
  252. return true;
  253. }
  254. }
  255. }
  256. return false;
  257. }
  258. void EQStream::ProcessPacket(EQProtocolPacket *p, EQProtocolPacket* lastp)
  259. {
  260. uint32 processed=0,subpacket_length=0;
  261. if (p) {
  262. if (p->opcode!=OP_SessionRequest && p->opcode!=OP_SessionResponse && !Session) {
  263. #ifdef EQN_DEBUG
  264. LogWrite(PACKET__ERROR, 0, "Packet", "*** Session not initialized, packet ignored ");
  265. //p->DumpRaw();
  266. #endif
  267. return;
  268. }
  269. //cout << "Received " << (int)p->opcode << ":\n";
  270. //DumpPacket(p->pBuffer, p->size);
  271. switch (p->opcode) {
  272. case OP_Combined: {
  273. processed=0;
  274. int8 offset = 0;
  275. int count = 0;
  276. #ifdef LE_DEBUG
  277. printf( "OP_Combined:\n");
  278. DumpPacket(p);
  279. #endif
  280. while(processed<p->size) {
  281. if ((subpacket_length=(unsigned char)*(p->pBuffer+processed))==0xff) {
  282. subpacket_length = ntohs(*(uint16*)(p->pBuffer + processed + 1));
  283. //printf("OP_Combined subpacket_length %u\n",subpacket_length);
  284. offset = 3;
  285. }
  286. else {
  287. offset = 1;
  288. }
  289. //printf("OP_Combined processed %u p->size %u subpacket length %u count %i\n",processed, p->size, subpacket_length, count);
  290. count++;
  291. #ifdef LE_DEBUG
  292. printf( "OP_Combined Packet %i (%u) (%u):\n", count, subpacket_length, processed);
  293. #endif
  294. bool isSubPacket = EQProtocolPacket::IsProtocolPacket(p->pBuffer + processed + offset, subpacket_length, false);
  295. if (isSubPacket) {
  296. EQProtocolPacket* subp = new EQProtocolPacket(p->pBuffer + processed + offset, subpacket_length);
  297. subp->copyInfo(p);
  298. #ifdef LE_DEBUG
  299. printf( "Opcode %i:\n", subp->opcode);
  300. DumpPacket(subp);
  301. #endif
  302. ProcessPacket(subp, p);
  303. #ifdef LE_DEBUG
  304. DumpPacket(subp);
  305. #endif
  306. delete subp;
  307. }
  308. else {
  309. offset = 1; // 0xFF in this case means it is actually 255 bytes of encrypted data after a 00 09 packet
  310. //Garbage packet?
  311. if(ntohs(*reinterpret_cast<uint16_t*>(p->pBuffer + processed + offset)) <= 0x1e) {
  312. subpacket_length=(unsigned char)*(p->pBuffer+processed);
  313. LogWrite(PACKET__ERROR, 0, "Packet", "!!!!!!!!!Garbage Packet Unknown Process as OP_Packet!!!!!!!!!!!!!\n");
  314. DumpPacket(p->pBuffer + processed + offset, subpacket_length);
  315. uchar* newbuf = p->pBuffer;
  316. newbuf += processed + offset;
  317. EQProtocolPacket *subp=new EQProtocolPacket(newbuf,subpacket_length);
  318. subp->copyInfo(p);
  319. ProcessPacket(subp, p);
  320. delete subp;
  321. }
  322. else {
  323. crypto->RC4Decrypt(p->pBuffer + processed + offset, subpacket_length);
  324. LogWrite(PACKET__ERROR, 0, "Packet", "!!!!!!!!!Garbage Packet!!!!!!!!!!!!! processed: %u, offset: %u, count: %i, subpacket_length: %u, offset_pos_1: %u, oversized_buffer_present: %u, offset size: %u, offset length: %u\n",
  325. processed, offset, count, subpacket_length, p->pBuffer[processed + offset], oversize_buffer ? 1 : 0, oversize_offset, oversize_length);
  326. if(p->pBuffer[processed + offset] == 0xff)
  327. {
  328. uchar* newbuf = p->pBuffer;
  329. newbuf += processed + offset + 1;
  330. DumpPacket(p->pBuffer + processed + offset, subpacket_length);
  331. EQProtocolPacket *subp=new EQProtocolPacket(newbuf, subpacket_length, OP_Packet);
  332. subp->copyInfo(p);
  333. ProcessPacket(subp, p);
  334. delete subp;
  335. }
  336. }
  337. }
  338. processed+=subpacket_length+offset;
  339. }
  340. break;
  341. }
  342. case OP_AppCombined: {
  343. processed=0;
  344. EQProtocolPacket* newpacket = 0;
  345. int8 offset = 0;
  346. #ifdef DEBUG_EMBEDDED_PACKETS
  347. printf( "OP_AppCombined: \n");
  348. DumpPacket(p);
  349. #endif
  350. int count = 0;
  351. while(processed<p->size) {
  352. count++;
  353. if ((subpacket_length=(unsigned char)*(p->pBuffer+processed))==0xff) {
  354. subpacket_length=ntohs(*(uint16 *)(p->pBuffer+processed+1));
  355. offset = 3;
  356. } else
  357. offset = 1;
  358. if(crypto->getRC4Key()==0 && p && subpacket_length > 8+offset){
  359. #ifdef DEBUG_EMBEDDED_PACKETS
  360. DumpPacket(p->pBuffer, p->size);
  361. #endif
  362. p->pBuffer += offset;
  363. processRSAKey(p, subpacket_length);
  364. p->pBuffer -= offset;
  365. }
  366. else if(crypto->isEncrypted()){
  367. #ifdef DEBUG_EMBEDDED_PACKETS
  368. printf( "OP_AppCombined Packet %i (%u) (%u): \n", count, subpacket_length, processed);
  369. DumpPacket(p->pBuffer+processed+offset, subpacket_length);
  370. #endif
  371. if(!HandleEmbeddedPacket(p, processed + offset, subpacket_length)){
  372. uchar* buffer = (p->pBuffer + processed + offset);
  373. if(!ProcessEmbeddedPacket(buffer, subpacket_length, OP_AppCombined)) {
  374. LogWrite(PACKET__ERROR, 0, "Packet", "*** This is bad, ProcessEmbeddedPacket failed, report to Image!");
  375. }
  376. }
  377. }
  378. processed+=subpacket_length+offset;
  379. }
  380. }
  381. break;
  382. case OP_Packet: {
  383. if (!p->pBuffer || (p->Size() < 4))
  384. {
  385. break;
  386. }
  387. uint16 seq=ntohs(*(uint16 *)(p->pBuffer));
  388. sint8 check=CompareSequence(NextInSeq,seq);
  389. if (check == SeqFuture) {
  390. #ifdef EQN_DEBUG
  391. LogWrite(PACKET__DEBUG, 1, "Packet", "*** Future packet: Expecting Seq=%i, but got Seq=%i", NextInSeq, seq);
  392. LogWrite(PACKET__DEBUG, 1, "Packet", "[Start]");
  393. p->DumpRawHeader(seq);
  394. LogWrite(PACKET__DEBUG, 1, "Packet", "[End]");
  395. #endif
  396. OutOfOrderpackets[seq] = p->Copy();
  397. // Image (2020): Removed as this is bad contributes to infinite loop
  398. //SendOutOfOrderAck(seq);
  399. } else if (check == SeqPast) {
  400. #ifdef EQN_DEBUG
  401. LogWrite(PACKET__DEBUG, 1, "Packet", "*** Duplicate packet: Expecting Seq=%i, but got Seq=%i", NextInSeq, seq);
  402. LogWrite(PACKET__DEBUG, 1, "Packet", "[Start]");
  403. p->DumpRawHeader(seq);
  404. LogWrite(PACKET__DEBUG, 1, "Packet", "[End]");
  405. #endif
  406. // Image (2020): Removed as this is bad contributes to infinite loop
  407. //OutOfOrderpackets[seq] = p->Copy();
  408. SendOutOfOrderAck(seq);
  409. } else {
  410. EQProtocolPacket* qp = RemoveQueue(seq);
  411. if (qp) {
  412. LogWrite(PACKET__DEBUG, 1, "Packet", "OP_Fragment: Removing older queued packet with sequence %i", seq);
  413. delete qp;
  414. }
  415. SetNextAckToSend(seq);
  416. NextInSeq++;
  417. if(HandleEmbeddedPacket(p))
  418. break;
  419. if(crypto->getRC4Key()==0 && p && p->size >= 69){
  420. #ifdef DEBUG_EMBEDDED_PACKETS
  421. DumpPacket(p->pBuffer, p->size);
  422. #endif
  423. processRSAKey(p);
  424. }
  425. else if(crypto->isEncrypted() && p){
  426. MCombineQueueLock.lock();
  427. EQProtocolPacket* newpacket = ProcessEncryptedPacket(p);
  428. MCombineQueueLock.unlock();
  429. if(newpacket){
  430. EQApplicationPacket* ap = newpacket->MakeApplicationPacket(2);
  431. if (ap->version == 0)
  432. ap->version = client_version;
  433. #ifdef WRITE_PACKETS
  434. WritePackets(ap->GetOpcodeName(), p->pBuffer, p->size, false);
  435. #endif
  436. InboundQueuePush(ap);
  437. safe_delete(newpacket);
  438. }
  439. }
  440. }
  441. }
  442. break;
  443. case OP_Fragment: {
  444. if (!p->pBuffer || (p->Size() < 4))
  445. {
  446. break;
  447. }
  448. uint16 seq=ntohs(*(uint16 *)(p->pBuffer));
  449. sint8 check=CompareSequence(NextInSeq,seq);
  450. if (check == SeqFuture) {
  451. #ifdef EQN_DEBUG
  452. LogWrite(PACKET__DEBUG, 1, "Packet", "*** Future packet2: Expecting Seq=%i, but got Seq=%i", NextInSeq, seq);
  453. LogWrite(PACKET__DEBUG, 1, "Packet", "[Start]");
  454. //p->DumpRawHeader(seq);
  455. LogWrite(PACKET__DEBUG, 1, "Packet", "[End]");
  456. #endif
  457. OutOfOrderpackets[seq] = p->Copy();
  458. //SendOutOfOrderAck(seq);
  459. } else if (check == SeqPast) {
  460. #ifdef EQN_DEBUG
  461. LogWrite(PACKET__DEBUG, 1, "Packet", "*** Duplicate packet2: Expecting Seq=%i, but got Seq=%i", NextInSeq, seq);
  462. LogWrite(PACKET__DEBUG, 1, "Packet", "[Start]");
  463. //p->DumpRawHeader(seq);
  464. LogWrite(PACKET__DEBUG, 1, "Packet", "[End]");
  465. #endif
  466. //OutOfOrderpackets[seq] = p->Copy();
  467. SendOutOfOrderAck(seq);
  468. } else {
  469. // In case we did queue one before as well.
  470. EQProtocolPacket* qp = RemoveQueue(seq);
  471. if (qp) {
  472. LogWrite(PACKET__DEBUG, 1, "Packet", "OP_Fragment: Removing older queued packet with sequence %i", seq);
  473. delete qp;
  474. }
  475. SetNextAckToSend(seq);
  476. NextInSeq++;
  477. if (oversize_buffer) {
  478. memcpy(oversize_buffer+oversize_offset,p->pBuffer+2,p->size-2);
  479. oversize_offset+=p->size-2;
  480. //cout << "Oversized is " << oversize_offset << "/" << oversize_length << " (" << (p->size-2) << ") Seq=" << seq << endl;
  481. if (oversize_offset==oversize_length) {
  482. if (*(p->pBuffer+2)==0x00 && *(p->pBuffer+3)==0x19) {
  483. EQProtocolPacket *subp=new EQProtocolPacket(oversize_buffer,oversize_offset);
  484. subp->copyInfo(p);
  485. ProcessPacket(subp, p);
  486. delete subp;
  487. } else {
  488. if(crypto->isEncrypted() && p && p->size > 2){
  489. MCombineQueueLock.lock();
  490. EQProtocolPacket* p2 = ProcessEncryptedData(oversize_buffer, oversize_offset, p->opcode);
  491. MCombineQueueLock.unlock();
  492. EQApplicationPacket* ap = p2->MakeApplicationPacket(2);
  493. ap->copyInfo(p);
  494. if (ap->version == 0)
  495. ap->version = client_version;
  496. #ifdef WRITE_PACKETS
  497. WritePackets(ap->GetOpcodeName(), oversize_buffer, oversize_offset, false);
  498. #endif
  499. ap->copyInfo(p);
  500. InboundQueuePush(ap);
  501. safe_delete(p2);
  502. }
  503. }
  504. delete[] oversize_buffer;
  505. oversize_buffer=NULL;
  506. oversize_offset=0;
  507. }
  508. } else if (!oversize_buffer) {
  509. oversize_length=ntohl(*(uint32 *)(p->pBuffer+2));
  510. oversize_buffer=new unsigned char[oversize_length];
  511. memcpy(oversize_buffer,p->pBuffer+6,p->size-6);
  512. oversize_offset=p->size-6;
  513. //cout << "Oversized is " << oversize_offset << "/" << oversize_length << " (" << (p->size-6) << ") Seq=" << seq << endl;
  514. }
  515. }
  516. }
  517. break;
  518. case OP_KeepAlive: {
  519. #ifndef COLLECTOR
  520. NonSequencedPush(new EQProtocolPacket(p->opcode,p->pBuffer,p->size));
  521. #endif
  522. }
  523. break;
  524. case OP_Ack: {
  525. if (!p->pBuffer || (p->Size() < 4))
  526. {
  527. LogWrite(PACKET__DEBUG, 9, "Packet", "Received OP_Ack that was of malformed size");
  528. break;
  529. }
  530. uint16 seq = ntohs(*(uint16*)(p->pBuffer));
  531. AckPackets(seq);
  532. retransmittimer = Timer::GetCurrentTime2();
  533. }
  534. break;
  535. case OP_SessionRequest: {
  536. if (p->Size() < sizeof(SessionRequest))
  537. {
  538. break;
  539. }
  540. if (GetState() == ESTABLISHED) {
  541. //_log(NET__ERROR, _L "Received OP_SessionRequest in ESTABLISHED state (%d) streamactive (%i) attempt (%i)" __L, GetState(), streamactive, sessionAttempts);
  542. // client seems to try a max of 4 times (initial +3 retries) then gives up, giving it a few more attempts just in case
  543. // streamactive means we identified the opcode, we cannot re-establish this connection
  544. if (streamactive || (sessionAttempts > 30))
  545. {
  546. SendDisconnect(false);
  547. SetState(CLOSED);
  548. break;
  549. }
  550. }
  551. sessionAttempts++;
  552. if(GetState() == WAIT_CLOSE) {
  553. printf("WAIT_CLOSE Reconnect with streamactive %u, sessionAttempts %u\n", streamactive, sessionAttempts);
  554. reconnectAttempt++;
  555. }
  556. init(GetState() != ESTABLISHED);
  557. OutboundQueueClear();
  558. SessionRequest *Request=(SessionRequest *)p->pBuffer;
  559. Session=ntohl(Request->Session);
  560. SetMaxLen(ntohl(Request->MaxLength));
  561. #ifndef COLLECTOR
  562. NextInSeq=0;
  563. Key=0x33624702;
  564. SendSessionResponse();
  565. #endif
  566. SetState(ESTABLISHED);
  567. }
  568. break;
  569. case OP_SessionResponse: {
  570. if (p->Size() < sizeof(SessionResponse))
  571. {
  572. break;
  573. }
  574. init();
  575. OutboundQueueClear();
  576. SetActive(true);
  577. SessionResponse *Response=(SessionResponse *)p->pBuffer;
  578. SetMaxLen(ntohl(Response->MaxLength));
  579. Key=ntohl(Response->Key);
  580. NextInSeq=0;
  581. SetState(ESTABLISHED);
  582. if (!Session)
  583. Session=ntohl(Response->Session);
  584. compressed=(Response->Format&FLAG_COMPRESSED);
  585. encoded=(Response->Format&FLAG_ENCODED);
  586. // Kinda kludgy, but trie for now
  587. if (compressed) {
  588. if (remote_port==9000 || (remote_port==0 && p->src_port==9000))
  589. SetStreamType(WorldStream);
  590. else
  591. SetStreamType(ZoneStream);
  592. } else if (encoded)
  593. SetStreamType(ChatOrMailStream);
  594. else
  595. SetStreamType(LoginStream);
  596. }
  597. break;
  598. case OP_SessionDisconnect: {
  599. //NextInSeq=0;
  600. SendDisconnect();
  601. //SetState(CLOSED);
  602. }
  603. break;
  604. case OP_OutOfOrderAck: {
  605. if (!p->pBuffer || (p->Size() < 4))
  606. {
  607. LogWrite(PACKET__DEBUG, 9, "Packet", "Received OP_OutOfOrderAck that was of malformed size");
  608. break;
  609. }
  610. uint16 seq = ntohs(*(uint16*)(p->pBuffer));
  611. MOutboundQueue.lock();
  612. if (uint16(SequencedBase + SequencedQueue.size()) != NextOutSeq) {
  613. LogWrite(PACKET__DEBUG, 9, "Packet", "Pre-OOA Invalid Sequenced queue: BS %u + SQ %u != NOS %u", SequencedBase, SequencedQueue.size(), NextOutSeq);
  614. }
  615. //if the packet they got out of order is between our last acked packet and the last sent packet, then its valid.
  616. if (CompareSequence(SequencedBase, seq) != SeqPast && CompareSequence(NextOutSeq, seq) == SeqPast) {
  617. uint16 sqsize = SequencedQueue.size();
  618. uint16 index = seq - SequencedBase;
  619. LogWrite(PACKET__DEBUG, 9, "Packet", "OP_OutOfOrderAck marking packet acked in queue (queue index = %u, queue size = %u)", index, sqsize);
  620. if (index < sqsize) {
  621. SequencedQueue[index]->acked = true;
  622. // flag packets for a resend
  623. uint16 count = 0;
  624. uint32 timeout = AverageDelta * 2 + 100;
  625. for (auto sitr = SequencedQueue.begin(); sitr != SequencedQueue.end() && count < index; ++sitr, ++count) {
  626. if (!(*sitr)->acked && (*sitr)->sent_time > 0 && (((*sitr)->sent_time + timeout) < Timer::GetCurrentTime2())) {
  627. (*sitr)->sent_time = 0;
  628. LogWrite(PACKET__DEBUG, 9, "Packet", "OP_OutOfOrderAck Flagging packet %u for retransmission", SequencedBase + count);
  629. }
  630. }
  631. }
  632. if (RETRANSMIT_TIMEOUT_MULT) {
  633. retransmittimer = Timer::GetCurrentTime2();
  634. }
  635. }
  636. else {
  637. LogWrite(PACKET__DEBUG, 9, "Packet", "Received OP_OutOfOrderAck for out-of-window %u. Window (%u->%u)", seq, SequencedBase, NextOutSeq);
  638. }
  639. if (uint16(SequencedBase + SequencedQueue.size()) != NextOutSeq) {
  640. LogWrite(PACKET__DEBUG, 9, "Packet", "Post-OOA Invalid Sequenced queue: BS %u + SQ %u != NOS %u", SequencedBase, SequencedQueue.size(), NextOutSeq);
  641. }
  642. MOutboundQueue.unlock();
  643. }
  644. break;
  645. case OP_ServerKeyRequest:{
  646. if (p->Size() < sizeof(ClientSessionStats))
  647. {
  648. //_log(NET__ERROR, _L "Received OP_SessionStatRequest that was of malformed size" __L);
  649. break;
  650. }
  651. ClientSessionStats* Stats = (ClientSessionStats*)p->pBuffer;
  652. int16 request_id = Stats->RequestID;
  653. AdjustRates(ntohl(Stats->average_delta));
  654. ServerSessionStats* stats=(ServerSessionStats*)p->pBuffer;
  655. memset(stats, 0, sizeof(ServerSessionStats));
  656. stats->RequestID = request_id;
  657. stats->current_time = ntohl(Timer::GetCurrentTime2());
  658. stats->sent_packets = ntohl(sent_packets);
  659. stats->sent_packets2 = ntohl(sent_packets);
  660. stats->received_packets = ntohl(received_packets);
  661. stats->received_packets2 = ntohl(received_packets);
  662. NonSequencedPush(new EQProtocolPacket(OP_SessionStatResponse,p->pBuffer,p->size));
  663. if(!crypto->isEncrypted())
  664. SendKeyRequest();
  665. else
  666. SendSessionResponse();
  667. }
  668. break;
  669. case OP_SessionStatResponse: {
  670. LogWrite(PACKET__INFO, 0, "Packet", "OP_SessionStatResponse");
  671. }
  672. break;
  673. case OP_OutOfSession: {
  674. LogWrite(PACKET__INFO, 0, "Packet", "OP_OutOfSession");
  675. SendDisconnect();
  676. SetState(CLOSED);
  677. }
  678. break;
  679. default:
  680. //EQApplicationPacket *ap = p->MakeApplicationPacket(app_opcode_size);
  681. //InboundQueuePush(ap);
  682. cout << "Orig Packet: " << p->opcode << endl;
  683. DumpPacket(p->pBuffer, p->size);
  684. if(p && p->size >= 69){
  685. processRSAKey(p);
  686. }
  687. MCombineQueueLock.lock();
  688. EQProtocolPacket* p2 = ProcessEncryptedData(p->pBuffer, p->size, OP_Packet);
  689. MCombineQueueLock.unlock();
  690. cout << "Decrypted Packet: " << p2->opcode << endl;
  691. DumpPacket(p2->pBuffer, p2->size);
  692. safe_delete(p2);
  693. /* if(p2)
  694. {
  695. EQApplicationPacket* ap = p2->MakeApplicationPacket(2);
  696. if (ap->version == 0)
  697. ap->version = client_version;
  698. InboundQueuePush(ap);
  699. safe_delete(p2);
  700. }*/
  701. //EQProtocolPacket* puse = p2;
  702. /* if (!rogue_buffer) {
  703. roguebuf_size=puse->size;
  704. rogue_buffer=new unsigned char[roguebuf_size];
  705. memcpy(rogue_buffer,puse->pBuffer,puse->size);
  706. roguebuf_offset=puse->size;
  707. cout << "RogueBuf is " << roguebuf_offset << "/" << roguebuf_size << " (" << (p->size-6) << ") NextInSeq=" << NextInSeq << endl;
  708. }
  709. else {
  710. int32 new_size = roguebuf_size + puse->size;
  711. uchar* tmp_buffer = new unsigned char[new_size];
  712. uchar* ptr = tmp_buffer;
  713. memcpy(ptr,rogue_buffer,roguebuf_size);
  714. ptr += roguebuf_size;
  715. memcpy(ptr,puse->pBuffer,puse->size);
  716. roguebuf_offset=puse->size;
  717. safe_delete_array(rogue_buffer);
  718. rogue_buffer = tmp_buffer;
  719. roguebuf_size = new_size;
  720. roguebuf_offset = new_size;
  721. cout << "RogueBuf is " << roguebuf_offset << "/" << roguebuf_size << " (" << (p->size-6) << ") NextInSeq=" << NextInSeq << endl;
  722. }*/
  723. #ifdef WRITE_PACKETS
  724. WritePackets(ap->GetOpcodeName(), p->pBuffer, p->size, false);
  725. #endif
  726. //InboundQueuePush(ap);
  727. LogWrite(PACKET__INFO, 0, "Packet", "Received unknown packet type, not adding to inbound queue");
  728. //safe_delete(p2);
  729. //SendDisconnect();
  730. break;
  731. }
  732. }
  733. }
  734. int8 EQStream::EQ2_Compress(EQ2Packet* app, int8 offset){
  735. #ifdef LE_DEBUG
  736. printf( "Before Compress in %s, line %i:\n", __FUNCTION__, __LINE__);
  737. DumpPacket(app);
  738. #endif
  739. uchar* pDataPtr = app->pBuffer + offset;
  740. int xpandSize = app->size * 2;
  741. uchar* deflate_buff = new uchar[xpandSize];
  742. MCompressData.lock();
  743. stream.next_in = pDataPtr;
  744. stream.avail_in = app->size - offset;
  745. stream.next_out = deflate_buff;
  746. stream.avail_out = xpandSize;
  747. int ret = deflate(&stream, Z_SYNC_FLUSH);
  748. if (ret != Z_OK)
  749. {
  750. printf("ZLIB COMPRESSION RETFAIL: %i, %i (Ret: %i)\n", app->size, stream.avail_out, ret);
  751. MCompressData.unlock();
  752. safe_delete_array(deflate_buff);
  753. return 0;
  754. }
  755. int32 newsize = xpandSize - stream.avail_out;
  756. safe_delete_array(app->pBuffer);
  757. app->size = newsize + offset;
  758. app->pBuffer = new uchar[app->size];
  759. app->pBuffer[(offset - 1)] = 1;
  760. memcpy(app->pBuffer + offset, deflate_buff, newsize);
  761. MCompressData.unlock();
  762. safe_delete_array(deflate_buff);
  763. #ifdef LE_DEBUG
  764. printf( "After Compress in %s, line %i:\n", __FUNCTION__, __LINE__);
  765. DumpPacket(app);
  766. #endif
  767. return offset - 1;
  768. }
  769. int16 EQStream::processRSAKey(EQProtocolPacket *p, uint16 subpacket_length){
  770. /*int16 limit = 0;
  771. int8 offset = 13;
  772. int8 offset2 = 0;
  773. if(p->pBuffer[2] == 0)
  774. limit = p->pBuffer[9];
  775. else{
  776. limit = p->pBuffer[5];
  777. offset2 = 5;
  778. offset-=1;
  779. }
  780. crypto->setRC4Key(Crypto::RSADecrypt(p->pBuffer + offset + (limit-8), 8));
  781. return (limit + offset +1) - offset2;*/
  782. if(subpacket_length)
  783. crypto->setRC4Key(Crypto::RSADecrypt(p->pBuffer + subpacket_length - 8, 8));
  784. else
  785. crypto->setRC4Key(Crypto::RSADecrypt(p->pBuffer + p->size - 8, 8));
  786. return 0;
  787. }
  788. void EQStream::SendKeyRequest(){
  789. int32 crypto_key_size = 60;
  790. int16 size = sizeof(KeyGen_Struct) + sizeof(KeyGen_End_Struct) + crypto_key_size;
  791. EQ2Packet *outapp=new EQ2Packet(OP_WSLoginRequestMsg,NULL,size);
  792. memcpy(&outapp->pBuffer[0], &crypto_key_size, sizeof(int32));
  793. memset(&outapp->pBuffer[4], 0xFF, crypto_key_size);
  794. memset(&outapp->pBuffer[size-5], 1, 1);
  795. memset(&outapp->pBuffer[size-1], 1, 1);
  796. EQ2QueuePacket(outapp, true);
  797. }
  798. void EQStream::EncryptPacket(EQ2Packet* app, int8 compress_offset, int8 offset){
  799. if(app->size>2 && crypto->isEncrypted()){
  800. app->packet_encrypted = true;
  801. uchar* crypt_buff = app->pBuffer;
  802. if(app->eq2_compressed)
  803. crypto->RC4Encrypt(crypt_buff + compress_offset, app->size - compress_offset);
  804. else
  805. crypto->RC4Encrypt(crypt_buff + 2 + offset, app->size - 2 - offset);
  806. }
  807. }
  808. void EQStream::EQ2QueuePacket(EQ2Packet* app, bool attempted_combine){
  809. if(CheckActive()){
  810. if(!attempted_combine){
  811. MCombineQueueLock.lock();
  812. combine_queue.push_back(app);
  813. MCombineQueueLock.unlock();
  814. }
  815. else{
  816. MCombineQueueLock.lock();
  817. PreparePacket(app);
  818. MCombineQueueLock.unlock();
  819. #ifdef LE_DEBUG
  820. printf( "After B in %s, line %i:\n", __FUNCTION__, __LINE__);
  821. DumpPacket(app);
  822. #endif
  823. SendPacket(app);
  824. }
  825. }
  826. }
  827. void EQStream::UnPreparePacket(EQ2Packet* app){
  828. if(app->pBuffer[2] == 0 && app->pBuffer[3] == 19){
  829. uchar* new_buffer = new uchar[app->size-3];
  830. memcpy(new_buffer+2, app->pBuffer+5, app->size-3);
  831. delete[] app->pBuffer;
  832. app->size-=3;
  833. app->pBuffer = new_buffer;
  834. }
  835. }
  836. #ifdef WRITE_PACKETS
  837. char EQStream::GetChar(uchar in)
  838. {
  839. if (in < ' ' || in > '~')
  840. return '.';
  841. return (char)in;
  842. }
  843. void EQStream::WriteToFile(char* pFormat, ...) {
  844. va_list args;
  845. va_start(args, pFormat);
  846. vfprintf(write_packets, pFormat, args);
  847. va_end(args);
  848. }
  849. void EQStream::WritePackets(const char* opcodeName, uchar* data, int32 size, bool outgoing) {
  850. MWritePackets.lock();
  851. struct in_addr ip_addr;
  852. ip_addr.s_addr = remote_ip;
  853. char timebuffer[80];
  854. time_t rawtime;
  855. struct tm* timeinfo;
  856. time(&rawtime);
  857. timeinfo = localtime(&rawtime);
  858. strftime(timebuffer, 80, "%m/%d/%Y %H:%M:%S", timeinfo);
  859. if (outgoing)
  860. WriteToFile("-- %s --\n%s\nSERVER -> %s\n", opcodeName, timebuffer, inet_ntoa(ip_addr));
  861. else
  862. WriteToFile("-- %s --\n%s\n%s -> SERVER\n", opcodeName, timebuffer, inet_ntoa(ip_addr));
  863. int i;
  864. int nLines = size / 16;
  865. int nExtra = size % 16;
  866. uchar* pPtr = data;
  867. for (i = 0; i < nLines; i++)
  868. {
  869. WriteToFile("%4.4X:\t%2.2X %2.2X %2.2X %2.2X %2.2X %2.2X %2.2X %2.2X %2.2X %2.2X %2.2X %2.2X %2.2X %2.2X %2.2X %2.2X %c%c%c%c%c%c%c%c%c%c%c%c%c%c%c%c\n", i * 16, pPtr[0], pPtr[1], pPtr[2], pPtr[3], pPtr[4], pPtr[5], pPtr[6], pPtr[7], pPtr[8], pPtr[9], pPtr[10], pPtr[11], pPtr[12], pPtr[13], pPtr[14], pPtr[15], GetChar(pPtr[0]), GetChar(pPtr[1]), GetChar(pPtr[2]), GetChar(pPtr[3]), GetChar(pPtr[4]), GetChar(pPtr[5]), GetChar(pPtr[6]), GetChar(pPtr[7]), GetChar(pPtr[8]), GetChar(pPtr[9]), GetChar(pPtr[10]), GetChar(pPtr[11]), GetChar(pPtr[12]), GetChar(pPtr[13]), GetChar(pPtr[14]), GetChar(pPtr[15]));
  870. pPtr += 16;
  871. }
  872. if (nExtra)
  873. {
  874. WriteToFile("%4.4X\t", nLines * 16);
  875. for (i = 0; i < nExtra; i++)
  876. {
  877. WriteToFile("%2.2X ", pPtr[i]);
  878. }
  879. for (i; i < 16; i++)
  880. WriteToFile(" ");
  881. for (i = 0; i < nExtra; i++)
  882. {
  883. WriteToFile("%c", GetChar(pPtr[i]));
  884. }
  885. WriteToFile("\n");
  886. }
  887. WriteToFile("\n\n");
  888. fflush(write_packets);
  889. MWritePackets.unlock();
  890. }
  891. void EQStream::WritePackets(EQ2Packet* app, bool outgoing) {
  892. if (app->version == 0)
  893. app->version = client_version;
  894. WritePackets(app->GetOpcodeName(), app->pBuffer, app->size, outgoing);
  895. }
  896. #endif
  897. void EQStream::PreparePacket(EQ2Packet* app, int8 offset){
  898. app->setVersion(client_version);
  899. compressed_offset = 0;
  900. #ifdef LE_DEBUG
  901. printf( "Before A in %s, line %i:\n", __FUNCTION__, __LINE__);
  902. DumpPacket(app);
  903. #endif
  904. if(!app->packet_prepared){
  905. if(app->PreparePacket(MaxLen) == 255) //invalid version
  906. return;
  907. }
  908. #ifdef LE_DEBUG
  909. printf( "After Prepare in %s, line %i:\n", __FUNCTION__, __LINE__);
  910. DumpPacket(app);
  911. #endif
  912. #ifdef WRITE_PACKETS
  913. if (!app->eq2_compressed && !app->packet_encrypted)
  914. WritePackets(app, true);
  915. #endif
  916. if(!app->eq2_compressed && app->size>128){
  917. compressed_offset = EQ2_Compress(app);
  918. if (compressed_offset)
  919. app->eq2_compressed = true;
  920. }
  921. if(!app->packet_encrypted){
  922. EncryptPacket(app, compressed_offset, offset);
  923. if(app->size > 2 && app->pBuffer[2] == 0){
  924. uchar* new_buffer = new uchar[app->size+1];
  925. new_buffer[2] = 0;
  926. memcpy(new_buffer+3, app->pBuffer+2, app->size-2);
  927. delete[] app->pBuffer;
  928. app->pBuffer = new_buffer;
  929. app->size++;
  930. }
  931. }
  932. #ifdef LE_DEBUG
  933. printf( "After A in %s, line %i:\n", __FUNCTION__, __LINE__);
  934. DumpPacket(app);
  935. #endif
  936. }
  937. void EQStream::SendPacket(EQProtocolPacket *p)
  938. {
  939. uint32 chunksize,used;
  940. uint32 length;
  941. // Convert the EQApplicationPacket to 1 or more EQProtocolPackets
  942. if (p->size>( MaxLen-8)) { // proto-op(2), seq(2), app-op(2) ... data ... crc(2)
  943. uchar* tmpbuff=p->pBuffer;
  944. length=p->size - 2;
  945. EQProtocolPacket *out=new EQProtocolPacket(OP_Fragment,NULL,MaxLen-4);
  946. *(uint32 *)(out->pBuffer+2)=htonl(length);
  947. used=MaxLen-10;
  948. memcpy(out->pBuffer+6,tmpbuff+2,used);
  949. #ifdef LE_DEBUG
  950. printf("(%s, %i) New Fragment:\n ", __FUNCTION__, __LINE__);
  951. DumpPacket(out);
  952. #endif
  953. SequencedPush(out);
  954. while (used<length) {
  955. chunksize=min(length-used,MaxLen-6);
  956. out=new EQProtocolPacket(OP_Fragment,NULL,chunksize+2);
  957. //memcpy(out->pBuffer+2,tmpbuff,1);
  958. memcpy(out->pBuffer+2,tmpbuff+used+2,chunksize);
  959. #ifdef LE_DEBUG
  960. printf("Chunk: \n");
  961. DumpPacket(out);
  962. #endif
  963. SequencedPush(out);
  964. used+=chunksize;
  965. }
  966. #ifdef LE_DEBUG
  967. printf( "ChunkDelete: \n");
  968. DumpPacket(out);
  969. //cerr << "1: Deleting 0x" << hex << (uint32)(p) << dec << endl;
  970. #endif
  971. delete p;
  972. } else {
  973. SequencedPush(p);
  974. }
  975. }
  976. void EQStream::SendPacket(EQApplicationPacket *p)
  977. {
  978. uint32 chunksize,used;
  979. uint32 length;
  980. // Convert the EQApplicationPacket to 1 or more EQProtocolPackets
  981. if (p->size>(MaxLen-8)) { // proto-op(2), seq(2), app-op(2) ... data ... crc(2)
  982. //cout << "Making oversized packet for: " << endl;
  983. //cout << p->size << endl;
  984. //p->DumpRawHeader();
  985. //dump_message(p->pBuffer,p->size,timestamp());
  986. //cout << p->size << endl;
  987. unsigned char *tmpbuff=new unsigned char[p->size+2];
  988. //cout << hex << (int)tmpbuff << dec << endl;
  989. length=p->serialize(tmpbuff);
  990. EQProtocolPacket *out=new EQProtocolPacket(OP_Fragment,NULL,MaxLen-4);
  991. *(uint32 *)(out->pBuffer+2)=htonl(p->Size());
  992. memcpy(out->pBuffer+6,tmpbuff,MaxLen-10);
  993. used=MaxLen-10;
  994. SequencedPush(out);
  995. //cout << "Chunk #" << ++i << " size=" << used << ", length-used=" << (length-used) << endl;
  996. while (used<length) {
  997. out=new EQProtocolPacket(OP_Fragment,NULL,MaxLen-4);
  998. chunksize=min(length-used,MaxLen-6);
  999. memcpy(out->pBuffer+2,tmpbuff+used,chunksize);
  1000. out->size=chunksize+2;
  1001. SequencedPush(out);
  1002. used+=chunksize;
  1003. //cout << "Chunk #"<< ++i << " size=" << chunksize << ", length-used=" << (length-used) << endl;
  1004. }
  1005. //cerr << "1: Deleting 0x" << hex << (uint32)(p) << dec << endl;
  1006. delete p;
  1007. delete[] tmpbuff;
  1008. } else {
  1009. EQProtocolPacket *out=new EQProtocolPacket(OP_Packet,NULL,p->Size()+2);
  1010. p->serialize(out->pBuffer+2);
  1011. SequencedPush(out);
  1012. //cerr << "2: Deleting 0x" << hex << (uint32)(p) << dec << endl;
  1013. delete p;
  1014. }
  1015. }
  1016. void EQStream::SequencedPush(EQProtocolPacket *p)
  1017. {
  1018. p->setVersion(client_version);
  1019. MOutboundQueue.lock();
  1020. *(uint16 *)(p->pBuffer)=htons(NextOutSeq);
  1021. SequencedQueue.push_back(p);
  1022. p->sequence = NextOutSeq;
  1023. NextOutSeq++;
  1024. MOutboundQueue.unlock();
  1025. }
  1026. void EQStream::NonSequencedPush(EQProtocolPacket *p)
  1027. {
  1028. p->setVersion(client_version);
  1029. MOutboundQueue.lock();
  1030. NonSequencedQueue.push(p);
  1031. MOutboundQueue.unlock();
  1032. }
  1033. void EQStream::SendAck(uint16 seq)
  1034. {
  1035. uint16 Seq=htons(seq);
  1036. SetLastAckSent(seq);
  1037. NonSequencedPush(new EQProtocolPacket(OP_Ack,(unsigned char *)&Seq,sizeof(uint16)));
  1038. }
  1039. void EQStream::SendOutOfOrderAck(uint16 seq)
  1040. {
  1041. uint16 Seq=htons(seq);
  1042. NonSequencedPush(new EQProtocolPacket(OP_OutOfOrderAck,(unsigned char *)&Seq,sizeof(uint16)));
  1043. }
  1044. bool EQStream::CheckCombineQueue(){
  1045. bool ret = true; //processed all packets
  1046. MCombineQueueLock.lock();
  1047. if(combine_queue.size() > 0){
  1048. EQ2Packet* first = combine_queue.front();
  1049. combine_queue.pop_front();
  1050. if(combine_queue.size() == 0){ //nothing to combine this with
  1051. EQ2QueuePacket(first, true);
  1052. }
  1053. else{
  1054. PreparePacket(first);
  1055. EQ2Packet* second = 0;
  1056. bool combine_worked = false;
  1057. int16 count = 0;
  1058. while(combine_queue.size()){
  1059. count++;
  1060. second = combine_queue.front();
  1061. combine_queue.pop_front();
  1062. PreparePacket(second);
  1063. /*if(first->GetRawOpcode() != OP_AppCombined && first->pBuffer[2] == 0){
  1064. EQ2Packet* tmp = second;
  1065. second = first;
  1066. first = tmp;
  1067. }*/
  1068. if(!first->AppCombine(second)){
  1069. first->SetProtocolOpcode(OP_Packet);
  1070. if(combine_worked){
  1071. SequencedPush(first);
  1072. }
  1073. else{
  1074. EQ2QueuePacket(first, true);
  1075. }
  1076. first = second;
  1077. combine_worked = false;
  1078. }
  1079. else{
  1080. combine_worked = true;
  1081. //DumpPacket(first);
  1082. }
  1083. if(count >= 60 || first->size > 4000){ //other clients need packets too
  1084. ret = false;
  1085. break;
  1086. }
  1087. }
  1088. if(first){
  1089. first->SetProtocolOpcode(OP_Packet);
  1090. if(combine_worked){
  1091. SequencedPush(first);
  1092. }
  1093. else{
  1094. EQ2QueuePacket(first, true);
  1095. }
  1096. }
  1097. }
  1098. }
  1099. MCombineQueueLock.unlock();
  1100. return ret;
  1101. }
  1102. void EQStream::CheckResend(int eq_fd){
  1103. int32 curr = Timer::GetCurrentTime2();
  1104. EQProtocolPacket* packet = 0;
  1105. deque<EQProtocolPacket*>::iterator itr;
  1106. MResendQue.lock();
  1107. for(itr=resend_que.begin();itr!=resend_que.end();itr++){
  1108. packet = *itr;
  1109. if(packet->attempt_count >= 5){//tried to resend this packet 5 times, client must already have it but didnt ack it
  1110. safe_delete(packet);
  1111. itr = resend_que.erase(itr);
  1112. if(itr == resend_que.end())
  1113. break;
  1114. }
  1115. else{
  1116. if((curr - packet->sent_time) < 1000)
  1117. continue;
  1118. packet->sent_time -=1000;
  1119. packet->attempt_count++;
  1120. WritePacket(eq_fd, packet);
  1121. }
  1122. }
  1123. MResendQue.unlock();
  1124. }
  1125. //returns SeqFuture if `seq` is later than `expected_seq`
  1126. EQStream::SeqOrder EQStream::CompareSequence(uint16 expected_seq, uint16 seq)
  1127. {
  1128. if (expected_seq == seq) {
  1129. // Curent
  1130. return SeqInOrder;
  1131. }
  1132. else if ((seq > expected_seq && (uint32)seq < ((uint32)expected_seq + EQStream::MaxWindowSize)) || seq < (expected_seq - EQStream::MaxWindowSize)) {
  1133. // Future
  1134. return SeqFuture;
  1135. }
  1136. else {
  1137. // Past
  1138. return SeqPast;
  1139. }
  1140. }
  1141. void EQStream::AckPackets(uint16 seq)
  1142. {
  1143. std::deque<EQProtocolPacket*>::iterator itr, tmp;
  1144. MOutboundQueue.lock();
  1145. SeqOrder ord = CompareSequence(SequencedBase, seq);
  1146. if (ord == SeqInOrder) {
  1147. //they are not acking anything new...
  1148. LogWrite(PACKET__DEBUG, 9, "Packet", "Received an ack with no window advancement (seq %u)", seq);
  1149. }
  1150. else if (ord == SeqPast) {
  1151. //they are nacking blocks going back before our buffer, wtf?
  1152. LogWrite(PACKET__DEBUG, 9, "Packet", "Received an ack with backward window advancement (they gave %u, our window starts at %u). This is bad" , seq, SequencedBase);
  1153. }
  1154. else {
  1155. LogWrite(PACKET__DEBUG, 9, "Packet", "Received an ack up through sequence %u. Our base is %u", seq, SequencedBase);
  1156. //this is a good ack, we get to ack some blocks.
  1157. seq++; //we stop at the block right after their ack, counting on the wrap of both numbers.
  1158. while (SequencedBase != seq) {
  1159. if (SequencedQueue.empty()) {
  1160. LogWrite(PACKET__DEBUG, 9, "Packet", "OUT OF PACKETS acked packet with sequence %u. Next send is %u before this", (unsigned long)SequencedBase, SequencedQueue.size());
  1161. SequencedBase = NextOutSeq;
  1162. break;
  1163. }
  1164. LogWrite(PACKET__DEBUG, 9, "Packet", "Removing acked packet with sequence %u", (unsigned long)SequencedBase);
  1165. //clean out the acked packet
  1166. delete SequencedQueue.front();
  1167. SequencedQueue.pop_front();
  1168. //advance the base sequence number to the seq of the block after the one we just got rid of.
  1169. SequencedBase++;
  1170. }
  1171. if (uint16(SequencedBase + SequencedQueue.size()) != NextOutSeq) {
  1172. LogWrite(PACKET__DEBUG, 9, "Packet", "Post-Ack on %u Invalid Sequenced queue: BS %u + SQ %u != NOS %u", seq, SequencedBase, SequencedQueue.size(), NextOutSeq);
  1173. }
  1174. }
  1175. MOutboundQueue.unlock();
  1176. }
  1177. void EQStream::Write(int eq_fd)
  1178. {
  1179. queue<EQProtocolPacket *> ReadyToSend;
  1180. long maxack;
  1181. // Check our rate to make sure we can send more
  1182. MRate.lock();
  1183. sint32 threshold=RateThreshold;
  1184. MRate.unlock();
  1185. if (BytesWritten > threshold) {
  1186. //cout << "Over threshold: " << BytesWritten << " > " << threshold << endl;
  1187. return;
  1188. }
  1189. MCombinedAppPacket.lock();
  1190. EQApplicationPacket *CombPack=CombinedAppPacket;
  1191. CombinedAppPacket=NULL;
  1192. MCombinedAppPacket.unlock();
  1193. if (CombPack) {
  1194. SendPacket(CombPack);
  1195. }
  1196. // If we got more packets to we need to ack, send an ack on the highest one
  1197. MAcks.lock();
  1198. maxack=MaxAckReceived;
  1199. // Added from peaks findings
  1200. if (NextAckToSend>LastAckSent || LastAckSent == 0x0000ffff)
  1201. SendAck(NextAckToSend);
  1202. MAcks.unlock();
  1203. // Lock the outbound queues while we process
  1204. MOutboundQueue.lock();
  1205. // Adjust where we start sending in case we get a late ack
  1206. //if (maxack>LastSeqSent)
  1207. // LastSeqSent=maxack;
  1208. // Place to hold the base packet t combine into
  1209. EQProtocolPacket *p=NULL;
  1210. std::deque<EQProtocolPacket*>::iterator sitr;
  1211. // Find the next sequenced packet to send from the "queue"
  1212. sitr = SequencedQueue.begin();
  1213. uint16 count = 0;
  1214. // get to start of packets
  1215. while (sitr != SequencedQueue.end() && (*sitr)->sent_time > 0) {
  1216. ++sitr;
  1217. ++count;
  1218. }
  1219. bool SeqEmpty = false, NonSeqEmpty = false;
  1220. // Loop until both are empty or MaxSends is reached
  1221. while (!SeqEmpty || !NonSeqEmpty) {
  1222. // See if there are more non-sequenced packets left
  1223. if (!NonSequencedQueue.empty()) {
  1224. if (!p) {
  1225. // If we don't have a packet to try to combine into, use this one as the base
  1226. // And remove it form the queue
  1227. p = NonSequencedQueue.front();
  1228. LogWrite(PACKET__DEBUG, 9, "Packet", "Starting combined packet with non-seq packet of len %u",p->size);
  1229. NonSequencedQueue.pop();
  1230. }
  1231. else if (!p->combine(NonSequencedQueue.front())) {
  1232. // Trying to combine this packet with the base didn't work (too big maybe)
  1233. // So just send the base packet (we'll try this packet again later)
  1234. LogWrite(PACKET__DEBUG, 9, "Packet", "Combined packet full at len %u, next non-seq packet is len %u", p->size, (NonSequencedQueue.front())->size);
  1235. ReadyToSend.push(p);
  1236. BytesWritten += p->size;
  1237. p = nullptr;
  1238. if (BytesWritten > threshold) {
  1239. // Sent enough this round, lets stop to be fair
  1240. LogWrite(PACKET__DEBUG, 9, "Packet", "Exceeded write threshold in nonseq (%u > %u)", BytesWritten, threshold);
  1241. break;
  1242. }
  1243. }
  1244. else {
  1245. // Combine worked, so just remove this packet and it's spot in the queue
  1246. LogWrite(PACKET__DEBUG, 9, "Packet", "Combined non-seq packet of len %u, yeilding %u combined", (NonSequencedQueue.front())->size, p->size);
  1247. delete NonSequencedQueue.front();
  1248. NonSequencedQueue.pop();
  1249. }
  1250. }
  1251. else {
  1252. // No more non-sequenced packets
  1253. NonSeqEmpty = true;
  1254. }
  1255. if (sitr != SequencedQueue.end()) {
  1256. uint16 seq_send = SequencedBase + count; //just for logging...
  1257. if (SequencedQueue.empty()) {
  1258. LogWrite(PACKET__DEBUG, 9, "Packet", "Tried to write a packet with an empty queue (%u is past next out %u)", seq_send, NextOutSeq);
  1259. SeqEmpty = true;
  1260. continue;
  1261. }
  1262. if ((*sitr)->acked || (*sitr)->sent_time != 0) {
  1263. ++sitr;
  1264. ++count;
  1265. if (p) {
  1266. LogWrite(PACKET__DEBUG, 9, "Packet", "Final combined packet not full, len %u", p->size);
  1267. ReadyToSend.push(p);
  1268. BytesWritten += p->size;
  1269. p = nullptr;
  1270. }
  1271. LogWrite(PACKET__DEBUG, 9, "Packet", "Not retransmitting seq packet %u because already marked as acked", seq_send);
  1272. }
  1273. else if (!p) {
  1274. // If we don't have a packet to try to combine into, use this one as the base
  1275. // Copy it first as it will still live until it is acked
  1276. p = (*sitr)->Copy();
  1277. LogWrite(PACKET__DEBUG, 9, "Packet", "Starting combined packet with seq packet %u of len %u", seq_send, p->size);
  1278. (*sitr)->sent_time = Timer::GetCurrentTime2();
  1279. ++sitr;
  1280. ++count;
  1281. }
  1282. else if (!p->combine(*sitr)) {
  1283. // Trying to combine this packet with the base didn't work (too big maybe)
  1284. // So just send the base packet (we'll try this packet again later)
  1285. LogWrite(PACKET__DEBUG, 9, "Packet", "Combined packet full at len %u, next seq packet %u is len %u", p->size, seq_send + 1, (*sitr)->size);
  1286. ReadyToSend.push(p);
  1287. BytesWritten += p->size;
  1288. p = nullptr;
  1289. if ((*sitr)->opcode != OP_Fragment && BytesWritten > threshold) {
  1290. // Sent enough this round, lets stop to be fair
  1291. LogWrite(PACKET__DEBUG, 9, "Packet", "Exceeded write threshold in seq (%u > %u)", BytesWritten, threshold);
  1292. break;
  1293. }
  1294. }
  1295. else {
  1296. // Combine worked
  1297. LogWrite(PACKET__DEBUG, 9, "Packet", "Combined seq packet %u of len %u, yeilding %u combined", seq_send, (*sitr)->size, p->size);
  1298. (*sitr)->sent_time = Timer::GetCurrentTime2();
  1299. ++sitr;
  1300. ++count;
  1301. }
  1302. if (uint16(SequencedBase + SequencedQueue.size()) != NextOutSeq) {
  1303. LogWrite(PACKET__DEBUG, 9, "Packet", "Post send Invalid Sequenced queue: BS %u + SQ %u != NOS %u", SequencedBase, SequencedQueue.size(), NextOutSeq);
  1304. }
  1305. }
  1306. else {
  1307. // No more sequenced packets
  1308. SeqEmpty = true;
  1309. }
  1310. }
  1311. MOutboundQueue.unlock(); // Unlock the queue
  1312. // We have a packet still, must have run out of both seq and non-seq, so send it
  1313. if (p) {
  1314. LogWrite(PACKET__DEBUG, 9, "Packet", "Final combined packet not full, len %u", p->size);
  1315. ReadyToSend.push(p);
  1316. BytesWritten += p->size;
  1317. }
  1318. // Send all the packets we "made"
  1319. while (!ReadyToSend.empty()) {
  1320. p = ReadyToSend.front();
  1321. WritePacket(eq_fd, p);
  1322. delete p;
  1323. ReadyToSend.pop();
  1324. }
  1325. //see if we need to send our disconnect and finish our close
  1326. if (SeqEmpty && NonSeqEmpty) {
  1327. //no more data to send
  1328. if (GetState() == CLOSING) {
  1329. MOutboundQueue.lock();
  1330. if (SequencedQueue.size() > 0 )
  1331. LogWrite(PACKET__DEBUG, 9, "Packet", "All outgoing data flushed, client should be disconnecting, awaiting acknowledgement of SequencedQueue.");
  1332. else
  1333. {
  1334. LogWrite(PACKET__DEBUG, 9, "Packet", "All outgoing data flushed, disconnecting client.");
  1335. //we are waiting for the queues to empty, now we can do our disconnect.
  1336. //this packet will not actually go out until the next call to Write().
  1337. SendDisconnect();
  1338. //SetState(CLOSED);
  1339. }
  1340. MOutboundQueue.unlock();
  1341. }
  1342. }
  1343. }
  1344. void EQStream::WritePacket(int eq_fd, EQProtocolPacket *p)
  1345. {
  1346. uint32 length = 0;
  1347. sockaddr_in address;
  1348. unsigned char tmpbuffer[2048];
  1349. address.sin_family = AF_INET;
  1350. address.sin_addr.s_addr=remote_ip;
  1351. address.sin_port=remote_port;
  1352. #ifdef NOWAY
  1353. uint32 ip=address.sin_addr.s_addr;
  1354. cout << "Sending to: "
  1355. << (int)*(unsigned char *)&ip
  1356. << "." << (int)*((unsigned char *)&ip+1)
  1357. << "." << (int)*((unsigned char *)&ip+2)
  1358. << "." << (int)*((unsigned char *)&ip+3)
  1359. << "," << (int)ntohs(address.sin_port) << "(" << p->size << ")" << endl;
  1360. p->DumpRaw();
  1361. cout << "-------------" << endl;
  1362. #endif
  1363. length=p->serialize(buffer);
  1364. if (p->opcode!=OP_SessionRequest && p->opcode!=OP_SessionResponse) {
  1365. if (compressed) {
  1366. BytesWritten -= p->size;
  1367. uint32 newlen=EQProtocolPacket::Compress(buffer,length,tmpbuffer,2048);
  1368. memcpy(buffer,tmpbuffer,newlen);
  1369. length=newlen;
  1370. BytesWritten += newlen;
  1371. }
  1372. if (encoded) {
  1373. EQProtocolPacket::ChatEncode(buffer,length,Key);
  1374. }
  1375. *(uint16 *)(buffer+length)=htons(CRC16(buffer,length,Key));
  1376. length+=2;
  1377. }
  1378. sent_packets++;
  1379. //dump_message_column(buffer,length,"Writer: ");
  1380. //cout << "Raw Data:\n";
  1381. //DumpPacket(buffer, length);
  1382. sendto(eq_fd,(char *)buffer,length,0,(sockaddr *)&address,sizeof(address));
  1383. }
  1384. EQProtocolPacket *EQStream::Read(int eq_fd, sockaddr_in *from)
  1385. {
  1386. int socklen;
  1387. int length=0;
  1388. unsigned char buffer[2048];
  1389. EQProtocolPacket *p=NULL;
  1390. char temp[15];
  1391. socklen=sizeof(sockaddr);
  1392. #ifdef WIN32
  1393. length=recvfrom(eq_fd, (char *)buffer, 2048, 0, (struct sockaddr*)from, (int *)&socklen);
  1394. #else
  1395. length=recvfrom(eq_fd, buffer, 2048, 0, (struct sockaddr*)from, (socklen_t *)&socklen);
  1396. #endif
  1397. if (length>=2) {
  1398. DumpPacket(buffer, length);
  1399. p=new EQProtocolPacket(buffer[1],&buffer[2],length-2);
  1400. //printf("Read packet: opcode %i length %u, expected-length: %u\n",buffer[1], length, p->size);
  1401. uint32 ip=from->sin_addr.s_addr;
  1402. sprintf(temp,"%d.%d.%d.%d:%d",
  1403. *(unsigned char *)&ip,
  1404. *((unsigned char *)&ip+1),
  1405. *((unsigned char *)&ip+2),
  1406. *((unsigned char *)&ip+3),
  1407. ntohs(from->sin_port));
  1408. //cout << timestamp() << "Data from: " << temp << " OpCode 0x" << hex << setw(2) << setfill('0') << (int)p->opcode << dec << endl;
  1409. //dump_message(p->pBuffer,p->size,timestamp());
  1410. }
  1411. return p;
  1412. }
  1413. void EQStream::SendSessionResponse()
  1414. {
  1415. EQProtocolPacket *out=new EQProtocolPacket(OP_SessionResponse,NULL,sizeof(SessionResponse));
  1416. SessionResponse *Response=(SessionResponse *)out->pBuffer;
  1417. Response->Session=htonl(Session);
  1418. Response->MaxLength=htonl(MaxLen);
  1419. Response->UnknownA=2;
  1420. Response->Format=0;
  1421. if (compressed)
  1422. Response->Format|=FLAG_COMPRESSED;
  1423. if (encoded)
  1424. Response->Format|=FLAG_ENCODED;
  1425. Response->Key=htonl(Key);
  1426. out->size=sizeof(SessionResponse);
  1427. NonSequencedPush(out);
  1428. }
  1429. void EQStream::SendSessionRequest()
  1430. {
  1431. EQProtocolPacket *out=new EQProtocolPacket(OP_SessionRequest,NULL,sizeof(SessionRequest));
  1432. SessionRequest *Request=(SessionRequest *)out->pBuffer;
  1433. memset(Request,0,sizeof(SessionRequest));
  1434. Request->Session=htonl(time(NULL));
  1435. Request->MaxLength=htonl(512);
  1436. NonSequencedPush(out);
  1437. }
  1438. void EQStream::SendDisconnect(bool setstate)
  1439. {
  1440. try{
  1441. if(GetState() != ESTABLISHED && GetState() != WAIT_CLOSE)
  1442. return;
  1443. EQProtocolPacket *out=new EQProtocolPacket(OP_SessionDisconnect,NULL,sizeof(uint32)+sizeof(int16));
  1444. *(uint32 *)out->pBuffer=htonl(Session);
  1445. out->pBuffer[4] = 0;
  1446. out->pBuffer[5] = 6;
  1447. NonSequencedPush(out);
  1448. if(setstate)
  1449. SetState(CLOSING);
  1450. }
  1451. catch(...){}
  1452. }
  1453. void EQStream::InboundQueuePush(EQApplicationPacket *p)
  1454. {
  1455. MInboundQueue.lock();
  1456. InboundQueue.push_back(p);
  1457. MInboundQueue.unlock();
  1458. }
  1459. EQApplicationPacket *EQStream::PopPacket()
  1460. {
  1461. EQApplicationPacket *p=NULL;
  1462. MInboundQueue.lock();
  1463. if (InboundQueue.size()) {
  1464. p=InboundQueue.front();
  1465. InboundQueue.pop_front();
  1466. }
  1467. MInboundQueue.unlock();
  1468. if(p)
  1469. p->setVersion(client_version);
  1470. return p;
  1471. }
  1472. void EQStream::InboundQueueClear()
  1473. {
  1474. MInboundQueue.lock();
  1475. while(InboundQueue.size()){
  1476. delete InboundQueue.front();
  1477. InboundQueue.pop_front();
  1478. }
  1479. MInboundQueue.unlock();
  1480. }
  1481. void EQStream::EncryptPacket(uchar* data, int16 size){
  1482. if(size>6){
  1483. }
  1484. }
  1485. bool EQStream::HasOutgoingData()
  1486. {
  1487. bool flag;
  1488. //once closed, we have nothing more to say
  1489. if(CheckClosed())
  1490. return(false);
  1491. MOutboundQueue.lock();
  1492. flag=(!NonSequencedQueue.empty());
  1493. if (!flag) {
  1494. flag = (!SequencedQueue.empty());
  1495. }
  1496. MOutboundQueue.unlock();
  1497. if (!flag) {
  1498. MAcks.lock();
  1499. flag= (NextAckToSend>LastAckSent);
  1500. MAcks.unlock();
  1501. }
  1502. if (!flag) {
  1503. MCombinedAppPacket.lock();
  1504. flag=(CombinedAppPacket!=NULL);
  1505. MCombinedAppPacket.unlock();
  1506. }
  1507. return flag;
  1508. }
  1509. void EQStream::OutboundQueueClear()
  1510. {
  1511. MOutboundQueue.lock();
  1512. while(NonSequencedQueue.size()) {
  1513. delete NonSequencedQueue.front();
  1514. NonSequencedQueue.pop();
  1515. }
  1516. while(SequencedQueue.size()) {
  1517. delete SequencedQueue.front();
  1518. SequencedQueue.pop_front();
  1519. }
  1520. MOutboundQueue.unlock();
  1521. }
  1522. void EQStream::Process(const unsigned char *buffer, const uint32 length)
  1523. {
  1524. received_packets++;
  1525. static unsigned char newbuffer[2048];
  1526. uint32 newlength=0;
  1527. #ifdef LE_DEBUG
  1528. printf("ProcessBuffer:\n");
  1529. DumpPacket(buffer, length);
  1530. #endif
  1531. if (EQProtocolPacket::ValidateCRC(buffer,length,Key)) {
  1532. if (compressed) {
  1533. newlength=EQProtocolPacket::Decompress(buffer,length,newbuffer,2048);
  1534. #ifdef LE_DEBUG
  1535. printf("ProcessBufferDecompress:\n");
  1536. DumpPacket(buffer, newlength);
  1537. #endif
  1538. } else {
  1539. memcpy(newbuffer,buffer,length);
  1540. newlength=length;
  1541. if (encoded)
  1542. EQProtocolPacket::ChatDecode(newbuffer,newlength-2,Key);
  1543. }
  1544. #ifdef LE_DEBUG
  1545. printf("ResultProcessBuffer:\n");
  1546. DumpPacket(buffer, newlength);
  1547. #endif
  1548. uint16 opcode=ntohs(*(const uint16 *)newbuffer);
  1549. //printf("Read packet: opcode %i newlength %u, newbuffer2len: %u, newbuffer3len: %u\n",opcode, newlength, newbuffer[2], newbuffer[3]);
  1550. if(opcode > 0 && opcode <= OP_OutOfSession)
  1551. {
  1552. if (buffer[1]!=0x01 && buffer[1]!=0x02 && buffer[1]!=0x1d)
  1553. newlength-=2;
  1554. EQProtocolPacket p(newbuffer,newlength);
  1555. ProcessPacket(&p);
  1556. }
  1557. else
  1558. {
  1559. cout << "2Orig Packet: " << opcode << endl;
  1560. DumpPacket(newbuffer, newlength);
  1561. ProcessEmbeddedPacket(newbuffer, newlength, OP_Fragment);
  1562. }
  1563. ProcessQueue();
  1564. } else {
  1565. cout << "Incoming packet failed checksum:" <<endl;
  1566. dump_message_column(const_cast<unsigned char *>(buffer),length,"CRC failed: ");
  1567. }
  1568. }
  1569. long EQStream::GetMaxAckReceived()
  1570. {
  1571. MAcks.lock();
  1572. long l=MaxAckReceived;
  1573. MAcks.unlock();
  1574. return l;
  1575. }
  1576. long EQStream::GetNextAckToSend()
  1577. {
  1578. MAcks.lock();
  1579. long l=NextAckToSend;
  1580. MAcks.unlock();
  1581. return l;
  1582. }
  1583. long EQStream::GetLastAckSent()
  1584. {
  1585. MAcks.lock();
  1586. long l=LastAckSent;
  1587. MAcks.unlock();
  1588. return l;
  1589. }
  1590. void EQStream::SetMaxAckReceived(uint32 seq)
  1591. {
  1592. deque<EQProtocolPacket *>::iterator itr;
  1593. MAcks.lock();
  1594. MaxAckReceived=seq;
  1595. MAcks.unlock();
  1596. MOutboundQueue.lock();
  1597. if (long(seq) > LastSeqSent)
  1598. LastSeqSent=seq;
  1599. MResendQue.lock();
  1600. EQProtocolPacket* packet = 0;
  1601. for(itr=resend_que.begin();itr!=resend_que.end();itr++){
  1602. packet = *itr;
  1603. if(packet && packet->sequence <= seq){
  1604. safe_delete(packet);
  1605. itr = resend_que.erase(itr);
  1606. if(itr == resend_que.end())
  1607. break;
  1608. }
  1609. }
  1610. MResendQue.unlock();
  1611. MOutboundQueue.unlock();
  1612. }
  1613. void EQStream::SetNextAckToSend(uint32 seq)
  1614. {
  1615. MAcks.lock();
  1616. NextAckToSend=seq;
  1617. MAcks.unlock();
  1618. }
  1619. void EQStream::SetLastAckSent(uint32 seq)
  1620. {
  1621. MAcks.lock();
  1622. LastAckSent=seq;
  1623. MAcks.unlock();
  1624. }
  1625. void EQStream::SetLastSeqSent(uint32 seq)
  1626. {
  1627. MOutboundQueue.lock();
  1628. LastSeqSent=seq;
  1629. MOutboundQueue.unlock();
  1630. }
  1631. void EQStream::SetStreamType(EQStreamType type)
  1632. {
  1633. StreamType=type;
  1634. switch (StreamType) {
  1635. case LoginStream:
  1636. app_opcode_size=1;
  1637. compressed=false;
  1638. encoded=false;
  1639. break;
  1640. case EQ2Stream:
  1641. app_opcode_size=2;
  1642. compressed=false;
  1643. encoded=false;
  1644. break;
  1645. case ChatOrMailStream:
  1646. case ChatStream:
  1647. case MailStream:
  1648. app_opcode_size=1;
  1649. compressed=false;
  1650. encoded=true;
  1651. break;
  1652. case ZoneStream:
  1653. case WorldStream:
  1654. default:
  1655. app_opcode_size=2;
  1656. compressed=true;
  1657. encoded=false;
  1658. break;
  1659. }
  1660. }
  1661. void EQStream::ProcessQueue()
  1662. {
  1663. if (OutOfOrderpackets.empty()) {
  1664. return;
  1665. }
  1666. EQProtocolPacket* qp = NULL;
  1667. while ((qp = RemoveQueue(NextInSeq)) != NULL) {
  1668. //_log(NET__DEBUG, _L "Processing Queued Packet: Seq=%d" __L, NextInSeq);
  1669. ProcessPacket(qp);
  1670. delete qp;
  1671. //_log(NET__APP_TRACE, _L "OP_Packet Queue size=%d" __L, PacketQueue.size());
  1672. }
  1673. }
  1674. EQProtocolPacket* EQStream::RemoveQueue(uint16 seq)
  1675. {
  1676. map<unsigned short, EQProtocolPacket*>::iterator itr;
  1677. EQProtocolPacket* qp = NULL;
  1678. if ((itr = OutOfOrderpackets.find(seq)) != OutOfOrderpackets.end()) {
  1679. qp = itr->second;
  1680. OutOfOrderpackets.erase(itr);
  1681. //_log(NET__APP_TRACE, _L "OP_Packet Queue size=%d" __L, PacketQueue.size());
  1682. }
  1683. return qp;
  1684. }
  1685. void EQStream::Decay()
  1686. {
  1687. MRate.lock();
  1688. uint32 rate=DecayRate;
  1689. MRate.unlock();
  1690. if (BytesWritten>0) {
  1691. BytesWritten-=rate;
  1692. if (BytesWritten<0)
  1693. BytesWritten=0;
  1694. }
  1695. int count = 0;
  1696. MOutboundQueue.lock();
  1697. for (auto sitr = SequencedQueue.begin(); sitr != SequencedQueue.end(); ++sitr, count++) {
  1698. if (!(*sitr)->acked && (*sitr)->sent_time > 0 && ((*sitr)->sent_time + retransmittimeout) < Timer::GetCurrentTime2()) {
  1699. (*sitr)->sent_time = 0;
  1700. LogWrite(PACKET__DEBUG, 9, "Packet", "Timeout exceeded for seq %u. Flagging packet for retransmission", SequencedBase + count);
  1701. }
  1702. }
  1703. MOutboundQueue.unlock();
  1704. }
  1705. void EQStream::AdjustRates(uint32 average_delta)
  1706. {
  1707. if (average_delta && (average_delta <= AVERAGE_DELTA_MAX)) {
  1708. MRate.lock();
  1709. AverageDelta = average_delta;
  1710. RateThreshold = RATEBASE / average_delta;
  1711. DecayRate = DECAYBASE / average_delta;
  1712. if (BytesWritten > RateThreshold)
  1713. BytesWritten = RateThreshold + DecayRate;
  1714. MRate.unlock();
  1715. }
  1716. else {
  1717. AverageDelta = AVERAGE_DELTA_MAX;
  1718. }
  1719. }