EQStream.cpp 37 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426
  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. uint16 EQStream::MaxWindowSize=2048;
  52. void EQStream::init(bool resetSession) {
  53. if (resetSession)
  54. {
  55. streamactive = false;
  56. sessionAttempts = 0;
  57. }
  58. timeout_delays = 0;
  59. MInUse.lock();
  60. active_users = 0;
  61. MInUse.unlock();
  62. Session=0;
  63. Key=0;
  64. MaxLen=0;
  65. NextInSeq=0;
  66. NextOutSeq=0;
  67. CombinedAppPacket=NULL;
  68. MAcks.lock();
  69. MaxAckReceived = -1;
  70. NextAckToSend = -1;
  71. LastAckSent = -1;
  72. MAcks.unlock();
  73. LastSeqSent=-1;
  74. MaxSends=5;
  75. LastPacket=Timer::GetCurrentTime2();
  76. oversize_buffer=NULL;
  77. oversize_length=0;
  78. oversize_offset=0;
  79. Factory = NULL;
  80. MRate.lock();
  81. RateThreshold=RATEBASE/250;
  82. DecayRate=DECAYBASE/250;
  83. MRate.unlock();
  84. BytesWritten=0;
  85. crypto->setRC4Key(0);
  86. }
  87. EQStream::EQStream(sockaddr_in addr){
  88. crypto = new Crypto();
  89. resend_que_timer = new Timer(1000);
  90. combine_timer = new Timer(250); //250 milliseconds
  91. combine_timer->Start();
  92. resend_que_timer->Start();
  93. init();
  94. remote_ip=addr.sin_addr.s_addr;
  95. remote_port=addr.sin_port;
  96. State=CLOSED;
  97. StreamType=UnknownStream;
  98. compressed=true;
  99. encoded=false;
  100. app_opcode_size=2;
  101. #ifdef WIN32
  102. ZeroMemory(&stream, sizeof(z_stream));
  103. #else
  104. bzero(&stream, sizeof(z_stream));
  105. #endif
  106. stream.zalloc = (alloc_func)0;
  107. stream.zfree = (free_func)0;
  108. stream.opaque = (voidpf)0;
  109. deflateInit2(&stream, 9, Z_DEFLATED, 13, 9, Z_DEFAULT_STRATEGY);
  110. //deflateInit(&stream, 5);
  111. compressed_offset = 0;
  112. client_version = 0;
  113. received_packets = 0;
  114. sent_packets = 0;
  115. }
  116. EQProtocolPacket* EQStream::ProcessEncryptedData(uchar* data, int32 size, int16 opcode){
  117. //cout << "B4:\n";
  118. //DumpPacket(data, size);
  119. /*if(size >= 2 && data[0] == 0 && data[1] == 0){
  120. cout << "Attempting to fix packet!\n";
  121. //Have to fix bad packet from client or it will screw up encryption :P
  122. size--;
  123. data++;
  124. }*/
  125. crypto->RC4Decrypt(data,size);
  126. int8 offset = 0;
  127. if(data[0] == 0xFF && size > 2){
  128. offset = 3;
  129. memcpy(&opcode, data+sizeof(int8), sizeof(int16));
  130. }
  131. else{
  132. offset = 1;
  133. memcpy(&opcode, data, sizeof(int8));
  134. }
  135. //cout << "After:\n";
  136. //DumpPacket(data, size);
  137. return new EQProtocolPacket(opcode, data+offset, size - offset);
  138. }
  139. EQProtocolPacket* EQStream::ProcessEncryptedPacket(EQProtocolPacket *p){
  140. EQProtocolPacket* ret = NULL;
  141. if(p->opcode == OP_Packet && p->size > 2)
  142. ret = ProcessEncryptedData(p->pBuffer+2, p->size-2, p->opcode);
  143. else
  144. ret = ProcessEncryptedData(p->pBuffer, p->size, p->opcode);
  145. return ret;
  146. }
  147. bool EQStream::HandleEmbeddedPacket(EQProtocolPacket *p, int16 offset, int16 length){
  148. if(p && p->size >= ((uint32)(offset+2))){
  149. if(p->pBuffer[offset] == 0 && p->pBuffer[offset+1] == 0x19){
  150. if(length == 0)
  151. length = p->size-2-offset;
  152. else
  153. length-=2;
  154. #ifdef LE_DEBUG
  155. LogWrite(PACKET__DEBUG, 0, "Packet", "Creating OP_AppCombined Packet!");
  156. #endif
  157. EQProtocolPacket *subp=new EQProtocolPacket(OP_AppCombined, p->pBuffer+2+offset, length);
  158. subp->copyInfo(p);
  159. ProcessPacket(subp);
  160. safe_delete(subp);
  161. return true;
  162. }
  163. else if(p->pBuffer[offset] == 0 && p->pBuffer[offset+1] == 0){
  164. if(length == 0)
  165. length = p->size-1-offset;
  166. else
  167. length--;
  168. #ifdef LE_DEBUG
  169. LogWrite(PACKET__DEBUG, 0, "Packet", "Creating Opcode 0 Packet!");
  170. DumpPacket(p->pBuffer+1+offset, length);
  171. #endif
  172. EQProtocolPacket* newpacket = ProcessEncryptedData(p->pBuffer+1+offset, length, OP_Packet);
  173. if(newpacket){
  174. #ifdef LE_DEBUG
  175. LogWrite(PACKET__DEBUG, 0, "Packet", "Result: ");
  176. DumpPacket(newpacket);
  177. #endif
  178. EQApplicationPacket *ap = newpacket->MakeApplicationPacket(2);
  179. InboundQueuePush(ap);
  180. safe_delete(newpacket);
  181. }
  182. else
  183. LogWrite(PACKET__ERROR, 0, "Packet", "No Packet!");
  184. return true;
  185. }
  186. }
  187. return false;
  188. }
  189. void EQStream::ProcessPacket(EQProtocolPacket *p)
  190. {
  191. uint32 processed=0,subpacket_length=0;
  192. if (p) {
  193. if (p->opcode!=OP_SessionRequest && p->opcode!=OP_SessionResponse && !Session) {
  194. #ifdef EQN_DEBUG
  195. LogWrite(PACKET__ERROR, 0, "Packet", "*** Session not initialized, packet ignored ");
  196. //p->DumpRaw();
  197. #endif
  198. return;
  199. }
  200. //cout << "Received " << (int)p->opcode << ":\n";
  201. //DumpPacket(p->pBuffer, p->size);
  202. switch (p->opcode) {
  203. case OP_Combined: {
  204. processed=0;
  205. int8 offset = 0;
  206. int count = 0;
  207. #ifdef LE_DEBUG
  208. LogWrite(PACKET__DEBUG, 0, "Packet", "OP_Combined: ");
  209. DumpPacket(p);
  210. #endif
  211. while(processed<p->size) {
  212. if ((subpacket_length=(unsigned char)*(p->pBuffer+processed))==0xff) {
  213. subpacket_length=ntohs(*(uint16 *)(p->pBuffer+processed+1));
  214. offset = 3;
  215. }
  216. else
  217. offset = 1;
  218. count++;
  219. #ifdef LE_DEBUG
  220. LogWrite(PACKET__DEBUG, 0, "Packet", "OP_Combined Packet %i (%u) (%u): ", count, subpacket_length, processed);
  221. #endif
  222. EQProtocolPacket *subp=new EQProtocolPacket(p->pBuffer+processed+offset,subpacket_length);
  223. subp->copyInfo(p);
  224. #ifdef LE_DEBUG
  225. LogWrite(PACKET__DEBUG, 0, "Packet", "Opcode %i:", subp->opcode);
  226. DumpPacket(subp);
  227. #endif
  228. ProcessPacket(subp);
  229. #ifdef LE_DEBUG
  230. DumpPacket(subp);
  231. #endif
  232. delete subp;
  233. processed+=subpacket_length+offset;
  234. }
  235. break;
  236. }
  237. case OP_AppCombined: {
  238. processed=0;
  239. EQProtocolPacket* newpacket = 0;
  240. int8 offset = 0;
  241. #ifdef LE_DEBUG
  242. LogWrite(PACKET__DEBUG, 0, "Packet", "OP_AppCombined: ");
  243. DumpPacket(p);
  244. #endif
  245. int count = 0;
  246. while(processed<p->size) {
  247. count++;
  248. if ((subpacket_length=(unsigned char)*(p->pBuffer+processed))==0xff) {
  249. subpacket_length=ntohs(*(uint16 *)(p->pBuffer+processed+1));
  250. offset = 3;
  251. } else
  252. offset = 1;
  253. if(crypto->getRC4Key()==0 && p->size >= 70){
  254. processRSAKey(p);
  255. }
  256. else if(crypto->isEncrypted()){
  257. #ifdef LE_DEBUG
  258. LogWrite(PACKET__DEBUG, 0, "Packet", "OP_AppCombined Packet %i (%u) (%u): ", count, subpacket_length, processed);
  259. DumpPacket(p->pBuffer+processed+offset, subpacket_length);
  260. #endif
  261. if(!HandleEmbeddedPacket(p, processed + offset, subpacket_length)){
  262. #ifdef LE_DEBUG
  263. LogWrite(PACKET__DEBUG, 0, "Packet", "OP_AppCombined Here:");
  264. #endif
  265. newpacket = ProcessEncryptedData(p->pBuffer+processed + offset, subpacket_length, OP_AppCombined);
  266. if(newpacket){
  267. #ifdef LE_DEBUG
  268. LogWrite(PACKET__DEBUG, 0, "Packet", "Opcode %i:", newpacket->opcode);
  269. DumpPacket(newpacket);
  270. #endif
  271. EQApplicationPacket* ap = newpacket->MakeApplicationPacket(2);
  272. #ifdef LE_DEBUG
  273. LogWrite(PACKET__DEBUG, 0, "Packet", "OP_AppCombined Here2:");
  274. DumpPacket(ap);
  275. #endif
  276. InboundQueuePush(ap);
  277. safe_delete(newpacket);
  278. }
  279. }
  280. }
  281. processed+=subpacket_length+offset;
  282. }
  283. }
  284. break;
  285. case OP_Packet: {
  286. if (!p->pBuffer || (p->Size() < 4))
  287. {
  288. break;
  289. }
  290. uint16 seq=ntohs(*(uint16 *)(p->pBuffer));
  291. sint8 check=CompareSequence(NextInSeq,seq);
  292. if (check>0) {
  293. #ifdef EQN_DEBUG
  294. LogWrite(PACKET__DEBUG, 1, "Packet", "*** Future packet: Expecting Seq=%i, but got Seq=%i", NextInSeq, seq);
  295. LogWrite(PACKET__DEBUG, 1, "Packet", "[Start]");
  296. p->DumpRawHeader(seq);
  297. LogWrite(PACKET__DEBUG, 1, "Packet", "[End]");
  298. #endif
  299. OutOfOrderpackets[seq] = p->Copy();
  300. // Image (2020): Removed as this is bad contributes to infinite loop
  301. //SendOutOfOrderAck(seq);
  302. } else if (check<0) {
  303. #ifdef EQN_DEBUG
  304. LogWrite(PACKET__DEBUG, 1, "Packet", "*** Duplicate packet: Expecting Seq=%i, but got Seq=%i", NextInSeq, seq);
  305. LogWrite(PACKET__DEBUG, 1, "Packet", "[Start]");
  306. p->DumpRawHeader(seq);
  307. LogWrite(PACKET__DEBUG, 1, "Packet", "[End]");
  308. #endif
  309. // Image (2020): Removed as this is bad contributes to infinite loop
  310. //OutOfOrderpackets[seq] = p->Copy();
  311. SendOutOfOrderAck(seq);
  312. } else {
  313. EQProtocolPacket* qp = RemoveQueue(seq);
  314. if (qp) {
  315. LogWrite(PACKET__DEBUG, 1, "Packet", "OP_Fragment: Removing older queued packet with sequence %i", seq);
  316. delete qp;
  317. }
  318. SetNextAckToSend(seq);
  319. NextInSeq++;
  320. if(HandleEmbeddedPacket(p))
  321. break;
  322. if(crypto->getRC4Key()==0 && p && p->size >= 70){
  323. processRSAKey(p);
  324. }
  325. else if(crypto->isEncrypted() && p){
  326. EQProtocolPacket* newpacket = ProcessEncryptedPacket(p);
  327. if(newpacket){
  328. EQApplicationPacket *ap = newpacket->MakeApplicationPacket(2);
  329. InboundQueuePush(ap);
  330. safe_delete(newpacket);
  331. }
  332. }
  333. }
  334. }
  335. break;
  336. case OP_Fragment: {
  337. if (!p->pBuffer || (p->Size() < 4))
  338. {
  339. break;
  340. }
  341. uint16 seq=ntohs(*(uint16 *)(p->pBuffer));
  342. sint8 check=CompareSequence(NextInSeq,seq);
  343. if (check>0) {
  344. #ifdef EQN_DEBUG
  345. LogWrite(PACKET__DEBUG, 1, "Packet", "*** Future packet2: Expecting Seq=%i, but got Seq=%i", NextInSeq, seq);
  346. LogWrite(PACKET__DEBUG, 1, "Packet", "[Start]");
  347. //p->DumpRawHeader(seq);
  348. LogWrite(PACKET__DEBUG, 1, "Packet", "[End]");
  349. #endif
  350. OutOfOrderpackets[seq] = p->Copy();
  351. //SendOutOfOrderAck(seq);
  352. } else if (check<0) {
  353. #ifdef EQN_DEBUG
  354. LogWrite(PACKET__DEBUG, 1, "Packet", "*** Duplicate packet2: Expecting Seq=%i, but got Seq=%i", NextInSeq, seq);
  355. LogWrite(PACKET__DEBUG, 1, "Packet", "[Start]");
  356. //p->DumpRawHeader(seq);
  357. LogWrite(PACKET__DEBUG, 1, "Packet", "[End]");
  358. #endif
  359. //OutOfOrderpackets[seq] = p->Copy();
  360. SendOutOfOrderAck(seq);
  361. } else {
  362. // In case we did queue one before as well.
  363. EQProtocolPacket* qp = RemoveQueue(seq);
  364. if (qp) {
  365. LogWrite(PACKET__DEBUG, 1, "Packet", "OP_Fragment: Removing older queued packet with sequence %i", seq);
  366. delete qp;
  367. }
  368. SetNextAckToSend(seq);
  369. NextInSeq++;
  370. if (oversize_buffer) {
  371. memcpy(oversize_buffer+oversize_offset,p->pBuffer+2,p->size-2);
  372. oversize_offset+=p->size-2;
  373. //cout << "Oversized is " << oversize_offset << "/" << oversize_length << " (" << (p->size-2) << ") Seq=" << seq << endl;
  374. if (oversize_offset==oversize_length) {
  375. if (*(p->pBuffer+2)==0x00 && *(p->pBuffer+3)==0x19) {
  376. EQProtocolPacket *subp=new EQProtocolPacket(oversize_buffer,oversize_offset);
  377. subp->copyInfo(p);
  378. ProcessPacket(subp);
  379. delete subp;
  380. } else {
  381. if(crypto->isEncrypted() && p && p->size > 2){
  382. EQProtocolPacket* p2 = ProcessEncryptedData(oversize_buffer, oversize_offset, p->opcode);
  383. EQApplicationPacket *ap = p2->MakeApplicationPacket(2);
  384. ap->copyInfo(p);
  385. InboundQueuePush(ap);
  386. safe_delete(p2);
  387. }
  388. }
  389. delete[] oversize_buffer;
  390. oversize_buffer=NULL;
  391. oversize_offset=0;
  392. }
  393. } else if (!oversize_buffer) {
  394. oversize_length=ntohl(*(uint32 *)(p->pBuffer+2));
  395. oversize_buffer=new unsigned char[oversize_length];
  396. memcpy(oversize_buffer,p->pBuffer+6,p->size-6);
  397. oversize_offset=p->size-6;
  398. //cout << "Oversized is " << oversize_offset << "/" << oversize_length << " (" << (p->size-6) << ") Seq=" << seq << endl;
  399. }
  400. }
  401. }
  402. break;
  403. case OP_KeepAlive: {
  404. #ifndef COLLECTOR
  405. NonSequencedPush(new EQProtocolPacket(p->opcode,p->pBuffer,p->size));
  406. #endif
  407. }
  408. break;
  409. case OP_Ack: {
  410. if (!p->pBuffer || (p->Size() < 4))
  411. {
  412. break;
  413. }
  414. uint16 seq=ntohs(*(uint16 *)(p->pBuffer));
  415. SetMaxAckReceived(seq);
  416. }
  417. break;
  418. case OP_SessionRequest: {
  419. if (p->Size() < sizeof(SessionRequest))
  420. {
  421. break;
  422. }
  423. if (GetState() == ESTABLISHED) {
  424. //_log(NET__ERROR, _L "Received OP_SessionRequest in ESTABLISHED state (%d) streamactive (%i) attempt (%i)" __L, GetState(), streamactive, sessionAttempts);
  425. // client seems to try a max of 4 times (initial +3 retries) then gives up, giving it a few more attempts just in case
  426. // streamactive means we identified the opcode, we cannot re-establish this connection
  427. if (streamactive || (sessionAttempts > 30))
  428. {
  429. SendDisconnect(false);
  430. SetState(CLOSED);
  431. break;
  432. }
  433. }
  434. sessionAttempts++;
  435. init(GetState() != ESTABLISHED);
  436. OutboundQueueClear();
  437. SessionRequest *Request=(SessionRequest *)p->pBuffer;
  438. Session=ntohl(Request->Session);
  439. SetMaxLen(ntohl(Request->MaxLength));
  440. #ifndef COLLECTOR
  441. NextInSeq=0;
  442. Key=0x33624702;
  443. SendSessionResponse();
  444. #endif
  445. SetState(ESTABLISHED);
  446. }
  447. break;
  448. case OP_SessionResponse: {
  449. if (p->Size() < sizeof(SessionResponse))
  450. {
  451. break;
  452. }
  453. init();
  454. OutboundQueueClear();
  455. SetActive(true);
  456. SessionResponse *Response=(SessionResponse *)p->pBuffer;
  457. SetMaxLen(ntohl(Response->MaxLength));
  458. Key=ntohl(Response->Key);
  459. NextInSeq=0;
  460. SetState(ESTABLISHED);
  461. if (!Session)
  462. Session=ntohl(Response->Session);
  463. compressed=(Response->Format&FLAG_COMPRESSED);
  464. encoded=(Response->Format&FLAG_ENCODED);
  465. // Kinda kludgy, but trie for now
  466. if (compressed) {
  467. if (remote_port==9000 || (remote_port==0 && p->src_port==9000))
  468. SetStreamType(WorldStream);
  469. else
  470. SetStreamType(ZoneStream);
  471. } else if (encoded)
  472. SetStreamType(ChatOrMailStream);
  473. else
  474. SetStreamType(LoginStream);
  475. }
  476. break;
  477. case OP_SessionDisconnect: {
  478. //NextInSeq=0;
  479. SendDisconnect();
  480. //SetState(CLOSED);
  481. }
  482. break;
  483. case OP_OutOfOrderAck: {
  484. if (!p->pBuffer || (p->Size() < 4))
  485. {
  486. break;
  487. }
  488. #ifndef COLLECTOR
  489. uint16 seq=ntohs(*(uint16 *)(p->pBuffer));
  490. if (CompareSequence(GetMaxAckReceived(),seq)>0 && CompareSequence(NextOutSeq,seq) < 0) {
  491. SetLastSeqSent(GetMaxAckReceived());
  492. }
  493. #endif
  494. }
  495. break;
  496. case OP_ServerKeyRequest:{
  497. if (p->Size() < sizeof(ClientSessionStats))
  498. {
  499. //_log(NET__ERROR, _L "Received OP_SessionStatRequest that was of malformed size" __L);
  500. break;
  501. }
  502. ClientSessionStats* Stats = (ClientSessionStats*)p->pBuffer;
  503. int16 request_id = Stats->RequestID;
  504. AdjustRates(ntohl(Stats->average_delta));
  505. ServerSessionStats* stats=(ServerSessionStats*)p->pBuffer;
  506. memset(stats, 0, sizeof(ServerSessionStats));
  507. stats->RequestID = request_id;
  508. stats->current_time = ntohl(Timer::GetCurrentTime2());
  509. stats->sent_packets = ntohl(sent_packets);
  510. stats->sent_packets2 = ntohl(sent_packets);
  511. stats->received_packets = ntohl(received_packets);
  512. stats->received_packets2 = ntohl(received_packets);
  513. NonSequencedPush(new EQProtocolPacket(OP_SessionStatResponse,p->pBuffer,p->size));
  514. if(!crypto->isEncrypted())
  515. SendKeyRequest();
  516. }
  517. break;
  518. case OP_SessionStatResponse: {
  519. LogWrite(PACKET__INFO, 0, "Packet", "OP_SessionStatResponse");
  520. }
  521. break;
  522. case OP_OutOfSession: {
  523. LogWrite(PACKET__INFO, 0, "Packet", "OP_OutOfSession");
  524. }
  525. break;
  526. default:
  527. EQApplicationPacket *ap = p->MakeApplicationPacket(app_opcode_size);
  528. InboundQueuePush(ap);
  529. LogWrite(PACKET__INFO, 0, "Packet", "Received unknown packet type, disconnecting client");
  530. //SendDisconnect();
  531. break;
  532. }
  533. }
  534. }
  535. int8 EQStream::EQ2_Compress(EQ2Packet* app, int8 offset){
  536. #ifdef LE_DEBUG
  537. LogWrite(PACKET__DEBUG, 0, "Packet", "Before Compress in %s, line %i:", __FUNCTION__, __LINE__);
  538. DumpPacket(app);
  539. #endif
  540. uchar* pDataPtr = app->pBuffer + offset;
  541. uchar* deflate_buff = new uchar[app->size];
  542. MCompressData.lock();
  543. stream.next_in = pDataPtr;
  544. stream.avail_in = app->size - offset;
  545. stream.next_out = deflate_buff;
  546. stream.avail_out = app->size;
  547. deflate(&stream, Z_SYNC_FLUSH);
  548. int32 newsize = app->size - stream.avail_out;
  549. safe_delete_array(app->pBuffer);
  550. app->size = newsize + offset;
  551. app->pBuffer = new uchar[app->size];
  552. app->pBuffer[(offset-1)] = 1;
  553. memcpy(app->pBuffer + offset, deflate_buff, newsize);
  554. MCompressData.unlock();
  555. safe_delete_array(deflate_buff);
  556. #ifdef LE_DEBUG
  557. LogWrite(PACKET__DEBUG, 0, "Packet", "After Compress in %s, line %i:", __FUNCTION__, __LINE__);
  558. DumpPacket(app);
  559. #endif
  560. return offset - 1;
  561. }
  562. int16 EQStream::processRSAKey(EQProtocolPacket *p){
  563. /*int16 limit = 0;
  564. int8 offset = 13;
  565. int8 offset2 = 0;
  566. if(p->pBuffer[2] == 0)
  567. limit = p->pBuffer[9];
  568. else{
  569. limit = p->pBuffer[5];
  570. offset2 = 5;
  571. offset-=1;
  572. }
  573. crypto->setRC4Key(Crypto::RSADecrypt(p->pBuffer + offset + (limit-8), 8));
  574. return (limit + offset +1) - offset2;*/
  575. if(p->pBuffer[0] == 0)
  576. crypto->setRC4Key(Crypto::RSADecrypt(p->pBuffer + 62, 8));
  577. else
  578. crypto->setRC4Key(Crypto::RSADecrypt(p->pBuffer + 61, 8));
  579. return 0;
  580. }
  581. void EQStream::SendKeyRequest(){
  582. int32 crypto_key_size = 60;
  583. int16 size = sizeof(KeyGen_Struct) + sizeof(KeyGen_End_Struct) + crypto_key_size;
  584. EQ2Packet *outapp=new EQ2Packet(OP_WSLoginRequestMsg,NULL,size);
  585. memcpy(&outapp->pBuffer[0], &crypto_key_size, sizeof(int32));
  586. memset(&outapp->pBuffer[4], 0xFF, crypto_key_size);
  587. memset(&outapp->pBuffer[size-5], 1, 1);
  588. memset(&outapp->pBuffer[size-1], 1, 1);
  589. EQ2QueuePacket(outapp);
  590. }
  591. void EQStream::EncryptPacket(EQ2Packet* app, int8 compress_offset, int8 offset){
  592. if(app->size>2 && crypto->isEncrypted()){
  593. app->packet_encrypted = true;
  594. uchar* crypt_buff = app->pBuffer;
  595. if(app->eq2_compressed)
  596. crypto->RC4Encrypt(crypt_buff + compress_offset, app->size - compress_offset);
  597. else
  598. crypto->RC4Encrypt(crypt_buff + 2 + offset, app->size - 2 - offset);
  599. }
  600. }
  601. void EQStream::EQ2QueuePacket(EQ2Packet* app, bool attempted_combine){
  602. if(CheckActive()){
  603. if(app->size < 600 && !attempted_combine){
  604. MCombineQueueLock.lock();
  605. combine_queue.push_back(app);
  606. MCombineQueueLock.unlock();
  607. }
  608. else{
  609. PreparePacket(app);
  610. #ifdef LE_DEBUG
  611. LogWrite(PACKET__DEBUG, 0, "Packet", "After B in %s, line %i:", __FUNCTION__, __LINE__);
  612. DumpPacket(app);
  613. #endif
  614. SendPacket(app);
  615. }
  616. }
  617. }
  618. void EQStream::UnPreparePacket(EQ2Packet* app){
  619. if(app->pBuffer[2] == 0 && app->pBuffer[3] == 19){
  620. uchar* new_buffer = new uchar[app->size-3];
  621. memcpy(new_buffer+2, app->pBuffer+5, app->size-3);
  622. delete[] app->pBuffer;
  623. app->size-=3;
  624. app->pBuffer = new_buffer;
  625. }
  626. }
  627. void EQStream::PreparePacket(EQ2Packet* app, int8 offset){
  628. app->setVersion(client_version);
  629. compressed_offset = 0;
  630. #ifdef LE_DEBUG
  631. LogWrite(PACKET__DEBUG, 0, "Packet", "Before A in %s, line %i:", __FUNCTION__, __LINE__);
  632. DumpPacket(app);
  633. #endif
  634. if(!app->packet_prepared){
  635. if(app->PreparePacket(MaxLen) == 255) //invalid version
  636. return;
  637. }
  638. #ifdef LE_DEBUG
  639. LogWrite(PACKET__DEBUG, 0, "Packet", "After Prepare in %s, line %i:", __FUNCTION__, __LINE__);
  640. DumpPacket(app);
  641. #endif
  642. if(!app->eq2_compressed && app->size>=0x80){
  643. compressed_offset = EQ2_Compress(app);
  644. app->eq2_compressed = true;
  645. }
  646. if(!app->packet_encrypted){
  647. EncryptPacket(app, compressed_offset, offset);
  648. if(app->size > 2 && app->pBuffer[2] == 0){
  649. uchar* new_buffer = new uchar[app->size+1];
  650. new_buffer[2] = 0;
  651. memcpy(new_buffer+3, app->pBuffer+2, app->size-2);
  652. delete[] app->pBuffer;
  653. app->pBuffer = new_buffer;
  654. app->size++;
  655. }
  656. }
  657. #ifdef LE_DEBUG
  658. LogWrite(PACKET__DEBUG, 0, "Packet", "After A in %s, line %i:", __FUNCTION__, __LINE__);
  659. DumpPacket(app);
  660. #endif
  661. }
  662. void EQStream::SendPacket(EQProtocolPacket *p)
  663. {
  664. uint32 chunksize,used;
  665. uint32 length;
  666. // Convert the EQApplicationPacket to 1 or more EQProtocolPackets
  667. if (p->size>( MaxLen-6)) { // proto-op(2), app-op(2) ... data ... crc(2)
  668. uchar* tmpbuff=p->pBuffer;
  669. length=p->size - 2;
  670. EQProtocolPacket *out=new EQProtocolPacket(OP_Fragment,NULL,MaxLen-4);
  671. *(uint32 *)(out->pBuffer+2)=htonl(length);
  672. memcpy(out->pBuffer+6,tmpbuff+2,MaxLen-10);
  673. used=MaxLen-10;
  674. #ifdef LE_DEBUG
  675. LogWrite(PACKET__DEBUG, 0, "Packet", "(%s, %i) New Fragment: ", __FUNCTION__, __LINE__);
  676. DumpPacket(out);
  677. #endif
  678. SequencedPush(out);
  679. while (used<length) {
  680. chunksize=min(length-used,MaxLen-6);
  681. out=new EQProtocolPacket(OP_Fragment,NULL,chunksize+2);
  682. //memcpy(out->pBuffer+2,tmpbuff,1);
  683. memcpy(out->pBuffer+2,tmpbuff+used+2,chunksize);
  684. #ifdef LE_DEBUG
  685. LogWrite(PACKET__DEBUG, 0, "Packet", "Chunk: ");
  686. DumpPacket(out);
  687. #endif
  688. SequencedPush(out);
  689. used+=chunksize;
  690. }
  691. #ifdef LE_DEBUG
  692. LogWrite(PACKET__DEBUG, 0, "Packet", "Chunk: ");
  693. DumpPacket(out);
  694. cerr << "1: Deleting 0x" << hex << (uint32)(p) << dec << endl;
  695. #endif
  696. delete p;
  697. } else {
  698. SequencedPush(p);
  699. }
  700. }
  701. void EQStream::SendPacket(EQApplicationPacket *p)
  702. {
  703. uint32 chunksize,used;
  704. uint32 length;
  705. // Convert the EQApplicationPacket to 1 or more EQProtocolPackets
  706. if (p->size>(MaxLen-8)) { // proto-op(2), seq(2), app-op(2) ... data ... crc(2)
  707. //cout << "Making oversized packet for: " << endl;
  708. //cout << p->size << endl;
  709. //p->DumpRawHeader();
  710. //dump_message(p->pBuffer,p->size,timestamp());
  711. //cout << p->size << endl;
  712. unsigned char *tmpbuff=new unsigned char[p->size+2];
  713. //cout << hex << (int)tmpbuff << dec << endl;
  714. length=p->serialize(tmpbuff);
  715. EQProtocolPacket *out=new EQProtocolPacket(OP_Fragment,NULL,MaxLen-4);
  716. *(uint32 *)(out->pBuffer+2)=htonl(p->Size());
  717. memcpy(out->pBuffer+6,tmpbuff,MaxLen-10);
  718. used=MaxLen-10;
  719. SequencedPush(out);
  720. //cout << "Chunk #" << ++i << " size=" << used << ", length-used=" << (length-used) << endl;
  721. while (used<length) {
  722. out=new EQProtocolPacket(OP_Fragment,NULL,MaxLen-4);
  723. chunksize=min(length-used,MaxLen-6);
  724. memcpy(out->pBuffer+2,tmpbuff+used,chunksize);
  725. out->size=chunksize+2;
  726. SequencedPush(out);
  727. used+=chunksize;
  728. //cout << "Chunk #"<< ++i << " size=" << chunksize << ", length-used=" << (length-used) << endl;
  729. }
  730. //cerr << "1: Deleting 0x" << hex << (uint32)(p) << dec << endl;
  731. delete p;
  732. delete[] tmpbuff;
  733. } else {
  734. EQProtocolPacket *out=new EQProtocolPacket(OP_Packet,NULL,p->Size()+2);
  735. p->serialize(out->pBuffer+2);
  736. SequencedPush(out);
  737. //cerr << "2: Deleting 0x" << hex << (uint32)(p) << dec << endl;
  738. delete p;
  739. }
  740. }
  741. void EQStream::SequencedPush(EQProtocolPacket *p)
  742. {
  743. p->setVersion(client_version);
  744. MOutboundQueue.lock();
  745. *(uint16 *)(p->pBuffer)=htons(NextOutSeq);
  746. SequencedQueue.push_back(p);
  747. p->sequence = NextOutSeq;
  748. NextOutSeq++;
  749. MOutboundQueue.unlock();
  750. }
  751. void EQStream::NonSequencedPush(EQProtocolPacket *p)
  752. {
  753. p->setVersion(client_version);
  754. MOutboundQueue.lock();
  755. NonSequencedQueue.push_back(p);
  756. MOutboundQueue.unlock();
  757. }
  758. void EQStream::SendAck(uint16 seq)
  759. {
  760. uint16 Seq=htons(seq);
  761. SetLastAckSent(seq);
  762. NonSequencedPush(new EQProtocolPacket(OP_Ack,(unsigned char *)&Seq,sizeof(uint16)));
  763. }
  764. void EQStream::SendOutOfOrderAck(uint16 seq)
  765. {
  766. uint16 Seq=htons(seq);
  767. NonSequencedPush(new EQProtocolPacket(OP_OutOfOrderAck,(unsigned char *)&Seq,sizeof(uint16)));
  768. }
  769. bool EQStream::CheckCombineQueue(){
  770. bool ret = true; //processed all packets
  771. MCombineQueueLock.lock();
  772. if(combine_queue.size() > 0){
  773. EQ2Packet* first = combine_queue.front();
  774. combine_queue.pop_front();
  775. if(combine_queue.size() == 0){ //nothing to combine this with
  776. EQ2QueuePacket(first, true);
  777. }
  778. else{
  779. PreparePacket(first);
  780. EQ2Packet* second = 0;
  781. bool combine_worked = false;
  782. int16 count = 0;
  783. while(combine_queue.size()){
  784. count++;
  785. second = combine_queue.front();
  786. combine_queue.pop_front();
  787. MCombineQueueLock.unlock();
  788. PreparePacket(second);
  789. /*if(first->GetRawOpcode() != OP_AppCombined && first->pBuffer[2] == 0){
  790. EQ2Packet* tmp = second;
  791. second = first;
  792. first = tmp;
  793. }*/
  794. if(!first->AppCombine(second)){
  795. first->SetProtocolOpcode(OP_Packet);
  796. if(combine_worked){
  797. SequencedPush(first);
  798. }
  799. else{
  800. EQ2QueuePacket(first, true);
  801. }
  802. first = second;
  803. combine_worked = false;
  804. }
  805. else{
  806. combine_worked = true;
  807. //DumpPacket(first);
  808. }
  809. MCombineQueueLock.lock();
  810. if(count >= 20){ //other clients need packets too
  811. ret = false;
  812. break;
  813. }
  814. }
  815. if(first){
  816. first->SetProtocolOpcode(OP_Packet);
  817. if(combine_worked){
  818. SequencedPush(first);
  819. }
  820. else{
  821. EQ2QueuePacket(first, true);
  822. }
  823. }
  824. }
  825. }
  826. MCombineQueueLock.unlock();
  827. return ret;
  828. }
  829. void EQStream::CheckResend(int eq_fd){
  830. int32 curr = Timer::GetCurrentTime2();
  831. EQProtocolPacket* packet = 0;
  832. deque<EQProtocolPacket*>::iterator itr;
  833. MResendQue.lock();
  834. for(itr=resend_que.begin();itr!=resend_que.end();itr++){
  835. packet = *itr;
  836. if(packet->attempt_count >= 5){//tried to resend this packet 5 times, client must already have it but didnt ack it
  837. safe_delete(packet);
  838. itr = resend_que.erase(itr);
  839. if(itr == resend_que.end())
  840. break;
  841. }
  842. else{
  843. if((curr - packet->sent_time) < 1000)
  844. continue;
  845. packet->sent_time -=1000;
  846. packet->attempt_count++;
  847. WritePacket(eq_fd, packet);
  848. }
  849. }
  850. MResendQue.unlock();
  851. }
  852. void EQStream::Write(int eq_fd)
  853. {
  854. deque<EQProtocolPacket *> ReadyToSend;
  855. deque<EQProtocolPacket *> SeqReadyToSend;
  856. long maxack;
  857. map<uint16, EQProtocolPacket *>::iterator sitr;
  858. // Check our rate to make sure we can send more
  859. MRate.lock();
  860. sint32 threshold=RateThreshold;
  861. MRate.unlock();
  862. if (BytesWritten > threshold) {
  863. //cout << "Over threshold: " << BytesWritten << " > " << threshold << endl;
  864. return;
  865. }
  866. MCombinedAppPacket.lock();
  867. EQApplicationPacket *CombPack=CombinedAppPacket;
  868. CombinedAppPacket=NULL;
  869. MCombinedAppPacket.unlock();
  870. if (CombPack) {
  871. SendPacket(CombPack);
  872. }
  873. // If we got more packets to we need to ack, send an ack on the highest one
  874. MAcks.lock();
  875. maxack=MaxAckReceived;
  876. // Added from peaks findings
  877. if (NextAckToSend>LastAckSent || LastAckSent == 0x0000ffff)
  878. SendAck(NextAckToSend);
  879. MAcks.unlock();
  880. // Lock the outbound queues while we process
  881. MOutboundQueue.lock();
  882. // Adjust where we start sending in case we get a late ack
  883. if (maxack>LastSeqSent)
  884. LastSeqSent=maxack;
  885. // Place to hold the base packet t combine into
  886. EQProtocolPacket *p=NULL;
  887. // Loop until both are empty or MaxSends is reached
  888. // See if there are more non-sequenced packets left
  889. while (NonSequencedQueue.size()) {
  890. if (!p) {
  891. // If we don't have a packet to try to combine into, use this one as the base
  892. // And remove it form the queue
  893. p = NonSequencedQueue.front();
  894. NonSequencedQueue.pop_front();
  895. }
  896. //Check if we have a packet to combine p with...
  897. if (NonSequencedQueue.size()){
  898. if (!p->combine(NonSequencedQueue.front())) {
  899. // Tryint to combine this packet with the base didn't work (too big maybe)
  900. // So just send the base packet (we'll try this packet again later)
  901. ReadyToSend.push_back(p);
  902. BytesWritten += p->size;
  903. p = NULL;
  904. }
  905. else {
  906. // Combine worked, so just remove this packet and it's spot in the queue
  907. delete NonSequencedQueue.front();
  908. NonSequencedQueue.pop_front();
  909. }
  910. }
  911. else {
  912. //We have no packets to combine p with so just send it...
  913. ReadyToSend.push_back(p);
  914. BytesWritten += p->size;
  915. p = NULL;
  916. }
  917. if (BytesWritten > threshold) {
  918. // Sent enough this round, lets stop to be fair
  919. break;
  920. }
  921. }
  922. //The non-seq loop must have broke before we sent this packet, send it now
  923. if (p){
  924. ReadyToSend.push_back(p);
  925. BytesWritten += p->size;
  926. }
  927. if (SequencedQueue.size() && BytesWritten < threshold) {
  928. while(SequencedQueue.size()) {
  929. p = SequencedQueue.front();
  930. BytesWritten+=p->size;
  931. SeqReadyToSend.push_back(p);
  932. p->sent_time = Timer::GetCurrentTime2();
  933. resend_que.push_back(p);
  934. SequencedQueue.pop_front();
  935. LastSeqSent=p->sequence;
  936. if (BytesWritten > threshold) {
  937. break;
  938. }
  939. }
  940. }
  941. // Unlock the queue
  942. MOutboundQueue.unlock();
  943. // Send all the packets we "made"
  944. while(ReadyToSend.size()) {
  945. WritePacket(eq_fd,ReadyToSend.front());
  946. delete ReadyToSend.front();
  947. ReadyToSend.pop_front();
  948. }
  949. while(SeqReadyToSend.size()){
  950. WritePacket(eq_fd,SeqReadyToSend.front());
  951. SeqReadyToSend.pop_front();
  952. }
  953. }
  954. void EQStream::WritePacket(int eq_fd, EQProtocolPacket *p)
  955. {
  956. uint32 length = 0;
  957. sockaddr_in address;
  958. unsigned char tmpbuffer[1024];
  959. address.sin_family = AF_INET;
  960. address.sin_addr.s_addr=remote_ip;
  961. address.sin_port=remote_port;
  962. #ifdef NOWAY
  963. uint32 ip=address.sin_addr.s_addr;
  964. cout << "Sending to: "
  965. << (int)*(unsigned char *)&ip
  966. << "." << (int)*((unsigned char *)&ip+1)
  967. << "." << (int)*((unsigned char *)&ip+2)
  968. << "." << (int)*((unsigned char *)&ip+3)
  969. << "," << (int)ntohs(address.sin_port) << "(" << p->size << ")" << endl;
  970. p->DumpRaw();
  971. cout << "-------------" << endl;
  972. #endif
  973. length=p->serialize(buffer);
  974. if (p->opcode!=OP_SessionRequest && p->opcode!=OP_SessionResponse) {
  975. if (compressed) {
  976. uint32 newlen=EQProtocolPacket::Compress(buffer,length,tmpbuffer,1024);
  977. memcpy(buffer,tmpbuffer,newlen);
  978. length=newlen;
  979. }
  980. if (encoded) {
  981. EQProtocolPacket::ChatEncode(buffer,length,Key);
  982. }
  983. *(uint16 *)(buffer+length)=htons(CRC16(buffer,length,Key));
  984. length+=2;
  985. }
  986. sent_packets++;
  987. //dump_message_column(buffer,length,"Writer: ");
  988. //cout << "Raw Data:\n";
  989. //DumpPacket(buffer, length);
  990. sendto(eq_fd,(char *)buffer,length,0,(sockaddr *)&address,sizeof(address));
  991. }
  992. EQProtocolPacket *EQStream::Read(int eq_fd, sockaddr_in *from)
  993. {
  994. int socklen;
  995. int length=0;
  996. unsigned char buffer[2048];
  997. EQProtocolPacket *p=NULL;
  998. char temp[15];
  999. socklen=sizeof(sockaddr);
  1000. #ifdef WIN32
  1001. length=recvfrom(eq_fd, (char *)buffer, 2048, 0, (struct sockaddr*)from, (int *)&socklen);
  1002. #else
  1003. length=recvfrom(eq_fd, buffer, 2048, 0, (struct sockaddr*)from, (socklen_t *)&socklen);
  1004. #endif
  1005. if (length>=2) {
  1006. DumpPacket(buffer, length);
  1007. p=new EQProtocolPacket(buffer[1],&buffer[2],length-2);
  1008. uint32 ip=from->sin_addr.s_addr;
  1009. sprintf(temp,"%d.%d.%d.%d:%d",
  1010. *(unsigned char *)&ip,
  1011. *((unsigned char *)&ip+1),
  1012. *((unsigned char *)&ip+2),
  1013. *((unsigned char *)&ip+3),
  1014. ntohs(from->sin_port));
  1015. //cout << timestamp() << "Data from: " << temp << " OpCode 0x" << hex << setw(2) << setfill('0') << (int)p->opcode << dec << endl;
  1016. //dump_message(p->pBuffer,p->size,timestamp());
  1017. }
  1018. return p;
  1019. }
  1020. void EQStream::SendSessionResponse()
  1021. {
  1022. EQProtocolPacket *out=new EQProtocolPacket(OP_SessionResponse,NULL,sizeof(SessionResponse));
  1023. SessionResponse *Response=(SessionResponse *)out->pBuffer;
  1024. Response->Session=htonl(Session);
  1025. Response->MaxLength=htonl(MaxLen);
  1026. Response->UnknownA=2;
  1027. Response->Format=0;
  1028. if (compressed)
  1029. Response->Format|=FLAG_COMPRESSED;
  1030. if (encoded)
  1031. Response->Format|=FLAG_ENCODED;
  1032. Response->Key=htonl(Key);
  1033. out->size=sizeof(SessionResponse);
  1034. NonSequencedPush(out);
  1035. }
  1036. void EQStream::SendSessionRequest()
  1037. {
  1038. EQProtocolPacket *out=new EQProtocolPacket(OP_SessionRequest,NULL,sizeof(SessionRequest));
  1039. SessionRequest *Request=(SessionRequest *)out->pBuffer;
  1040. memset(Request,0,sizeof(SessionRequest));
  1041. Request->Session=htonl(time(NULL));
  1042. Request->MaxLength=htonl(512);
  1043. NonSequencedPush(out);
  1044. }
  1045. void EQStream::SendDisconnect(bool setstate)
  1046. {
  1047. try{
  1048. if(GetState() != ESTABLISHED)
  1049. return;
  1050. EQProtocolPacket *out=new EQProtocolPacket(OP_SessionDisconnect,NULL,sizeof(uint32)+sizeof(int16));
  1051. *(uint32 *)out->pBuffer=htonl(Session);
  1052. out->pBuffer[4] = 0;
  1053. out->pBuffer[5] = 6;
  1054. NonSequencedPush(out);
  1055. if(setstate)
  1056. SetState(CLOSING);
  1057. }
  1058. catch(...){}
  1059. }
  1060. void EQStream::InboundQueuePush(EQApplicationPacket *p)
  1061. {
  1062. MInboundQueue.lock();
  1063. InboundQueue.push_back(p);
  1064. MInboundQueue.unlock();
  1065. }
  1066. EQApplicationPacket *EQStream::PopPacket()
  1067. {
  1068. EQApplicationPacket *p=NULL;
  1069. MInboundQueue.lock();
  1070. if (InboundQueue.size()) {
  1071. p=InboundQueue.front();
  1072. InboundQueue.pop_front();
  1073. }
  1074. MInboundQueue.unlock();
  1075. if(p)
  1076. p->setVersion(client_version);
  1077. return p;
  1078. }
  1079. void EQStream::InboundQueueClear()
  1080. {
  1081. MInboundQueue.lock();
  1082. while(InboundQueue.size()){
  1083. delete InboundQueue.front();
  1084. InboundQueue.pop_front();
  1085. }
  1086. MInboundQueue.unlock();
  1087. }
  1088. void EQStream::EncryptPacket(uchar* data, int16 size){
  1089. if(size>6){
  1090. }
  1091. }
  1092. bool EQStream::HasOutgoingData()
  1093. {
  1094. bool flag;
  1095. //once closed, we have nothing more to say
  1096. if(CheckClosed())
  1097. return(false);
  1098. MOutboundQueue.lock();
  1099. flag=(!NonSequencedQueue.empty());
  1100. if (!flag) {
  1101. flag = (!SequencedQueue.empty());
  1102. }
  1103. MOutboundQueue.unlock();
  1104. if (!flag) {
  1105. MAcks.lock();
  1106. flag= (NextAckToSend>LastAckSent);
  1107. MAcks.unlock();
  1108. }
  1109. if (!flag) {
  1110. MCombinedAppPacket.lock();
  1111. flag=(CombinedAppPacket!=NULL);
  1112. MCombinedAppPacket.unlock();
  1113. }
  1114. return flag;
  1115. }
  1116. void EQStream::OutboundQueueClear()
  1117. {
  1118. MOutboundQueue.lock();
  1119. while(NonSequencedQueue.size()) {
  1120. delete NonSequencedQueue.front();
  1121. NonSequencedQueue.pop_front();
  1122. }
  1123. while(SequencedQueue.size()) {
  1124. delete SequencedQueue.front();
  1125. SequencedQueue.pop_front();
  1126. }
  1127. MOutboundQueue.unlock();
  1128. }
  1129. void EQStream::Process(const unsigned char *buffer, const uint32 length)
  1130. {
  1131. received_packets++;
  1132. static unsigned char newbuffer[2048];
  1133. uint32 newlength=0;
  1134. if (EQProtocolPacket::ValidateCRC(buffer,length,Key)) {
  1135. if (compressed) {
  1136. newlength=EQProtocolPacket::Decompress(buffer,length,newbuffer,2048);
  1137. } else {
  1138. memcpy(newbuffer,buffer,length);
  1139. newlength=length;
  1140. if (encoded)
  1141. EQProtocolPacket::ChatDecode(newbuffer,newlength-2,Key);
  1142. }
  1143. if (buffer[1]!=0x01 && buffer[1]!=0x02 && buffer[1]!=0x1d)
  1144. newlength-=2;
  1145. EQProtocolPacket p(newbuffer,newlength);
  1146. ProcessPacket(&p);
  1147. ProcessQueue();
  1148. } else {
  1149. #ifdef EQN_DEBUG
  1150. cout << "Incoming packet failed checksum:" <<endl;
  1151. dump_message_column(const_cast<unsigned char *>(buffer),length,"CRC failed: ");
  1152. #endif
  1153. }
  1154. }
  1155. long EQStream::GetMaxAckReceived()
  1156. {
  1157. MAcks.lock();
  1158. long l=MaxAckReceived;
  1159. MAcks.unlock();
  1160. return l;
  1161. }
  1162. long EQStream::GetNextAckToSend()
  1163. {
  1164. MAcks.lock();
  1165. long l=NextAckToSend;
  1166. MAcks.unlock();
  1167. return l;
  1168. }
  1169. long EQStream::GetLastAckSent()
  1170. {
  1171. MAcks.lock();
  1172. long l=LastAckSent;
  1173. MAcks.unlock();
  1174. return l;
  1175. }
  1176. void EQStream::SetMaxAckReceived(uint32 seq)
  1177. {
  1178. deque<EQProtocolPacket *>::iterator itr;
  1179. MAcks.lock();
  1180. MaxAckReceived=seq;
  1181. MAcks.unlock();
  1182. MOutboundQueue.lock();
  1183. if (long(seq) > LastSeqSent)
  1184. LastSeqSent=seq;
  1185. MResendQue.lock();
  1186. EQProtocolPacket* packet = 0;
  1187. for(itr=resend_que.begin();itr!=resend_que.end();itr++){
  1188. packet = *itr;
  1189. if(packet && packet->sequence <= seq){
  1190. safe_delete(packet);
  1191. itr = resend_que.erase(itr);
  1192. if(itr == resend_que.end())
  1193. break;
  1194. }
  1195. }
  1196. MResendQue.unlock();
  1197. MOutboundQueue.unlock();
  1198. }
  1199. void EQStream::SetNextAckToSend(uint32 seq)
  1200. {
  1201. MAcks.lock();
  1202. NextAckToSend=seq;
  1203. MAcks.unlock();
  1204. }
  1205. void EQStream::SetLastAckSent(uint32 seq)
  1206. {
  1207. MAcks.lock();
  1208. LastAckSent=seq;
  1209. MAcks.unlock();
  1210. }
  1211. void EQStream::SetLastSeqSent(uint32 seq)
  1212. {
  1213. MOutboundQueue.lock();
  1214. LastSeqSent=seq;
  1215. MOutboundQueue.unlock();
  1216. }
  1217. void EQStream::SetStreamType(EQStreamType type)
  1218. {
  1219. StreamType=type;
  1220. switch (StreamType) {
  1221. case LoginStream:
  1222. app_opcode_size=1;
  1223. compressed=false;
  1224. encoded=false;
  1225. break;
  1226. case EQ2Stream:
  1227. app_opcode_size=2;
  1228. compressed=false;
  1229. encoded=false;
  1230. break;
  1231. case ChatOrMailStream:
  1232. case ChatStream:
  1233. case MailStream:
  1234. app_opcode_size=1;
  1235. compressed=false;
  1236. encoded=true;
  1237. break;
  1238. case ZoneStream:
  1239. case WorldStream:
  1240. default:
  1241. app_opcode_size=2;
  1242. compressed=true;
  1243. encoded=false;
  1244. break;
  1245. }
  1246. }
  1247. void EQStream::ProcessQueue()
  1248. {
  1249. if (OutOfOrderpackets.empty()) {
  1250. return;
  1251. }
  1252. EQProtocolPacket* qp = NULL;
  1253. while ((qp = RemoveQueue(NextInSeq)) != NULL) {
  1254. //_log(NET__DEBUG, _L "Processing Queued Packet: Seq=%d" __L, NextInSeq);
  1255. ProcessPacket(qp);
  1256. delete qp;
  1257. //_log(NET__APP_TRACE, _L "OP_Packet Queue size=%d" __L, PacketQueue.size());
  1258. }
  1259. }
  1260. EQProtocolPacket* EQStream::RemoveQueue(uint16 seq)
  1261. {
  1262. map<unsigned short, EQProtocolPacket*>::iterator itr;
  1263. EQProtocolPacket* qp = NULL;
  1264. if ((itr = OutOfOrderpackets.find(seq)) != OutOfOrderpackets.end()) {
  1265. qp = itr->second;
  1266. OutOfOrderpackets.erase(itr);
  1267. //_log(NET__APP_TRACE, _L "OP_Packet Queue size=%d" __L, PacketQueue.size());
  1268. }
  1269. return qp;
  1270. }
  1271. sint8 EQStream::CompareSequence(uint16 expected_seq , uint16 seq)
  1272. {
  1273. if (expected_seq==seq) {
  1274. // Curent
  1275. return 0;
  1276. } else if ((seq > expected_seq && (uint32)seq < ((uint32)expected_seq + EQStream::MaxWindowSize)) || seq < (expected_seq - EQStream::MaxWindowSize)) {
  1277. // Future
  1278. return 1;
  1279. } else {
  1280. // Past
  1281. return -1;
  1282. }
  1283. }
  1284. void EQStream::Decay()
  1285. {
  1286. MRate.lock();
  1287. uint32 rate=DecayRate;
  1288. MRate.unlock();
  1289. if (BytesWritten>0) {
  1290. BytesWritten-=rate;
  1291. if (BytesWritten<0)
  1292. BytesWritten=0;
  1293. }
  1294. }
  1295. void EQStream::AdjustRates(uint32 average_delta)
  1296. {
  1297. if (average_delta) {
  1298. MRate.lock();
  1299. RateThreshold=RATEBASE/average_delta;
  1300. DecayRate=DECAYBASE/average_delta;
  1301. MRate.unlock();
  1302. }
  1303. }