9
3

EQStream.cpp 54 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858185918601861186218631864186518661867186818691870187118721873187418751876187718781879188018811882188318841885188618871888188918901891189218931894189518961897189818991900190119021903190419051906
  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 && p->size > (1+offset)) {
  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. // retransmission attempts
  1332. }
  1333. else
  1334. {
  1335. LogWrite(PACKET__DEBUG, 9, "Packet", "All outgoing data flushed, disconnecting client.");
  1336. //we are waiting for the queues to empty, now we can do our disconnect.
  1337. //this packet will not actually go out until the next call to Write().
  1338. SendDisconnect();
  1339. //SetState(CLOSED);
  1340. }
  1341. MOutboundQueue.unlock();
  1342. }
  1343. }
  1344. }
  1345. void EQStream::WritePacket(int eq_fd, EQProtocolPacket *p)
  1346. {
  1347. uint32 length = 0;
  1348. sockaddr_in address;
  1349. unsigned char tmpbuffer[2048];
  1350. address.sin_family = AF_INET;
  1351. address.sin_addr.s_addr=remote_ip;
  1352. address.sin_port=remote_port;
  1353. #ifdef NOWAY
  1354. uint32 ip=address.sin_addr.s_addr;
  1355. cout << "Sending to: "
  1356. << (int)*(unsigned char *)&ip
  1357. << "." << (int)*((unsigned char *)&ip+1)
  1358. << "." << (int)*((unsigned char *)&ip+2)
  1359. << "." << (int)*((unsigned char *)&ip+3)
  1360. << "," << (int)ntohs(address.sin_port) << "(" << p->size << ")" << endl;
  1361. p->DumpRaw();
  1362. cout << "-------------" << endl;
  1363. #endif
  1364. length=p->serialize(buffer);
  1365. if (p->opcode!=OP_SessionRequest && p->opcode!=OP_SessionResponse) {
  1366. if (compressed) {
  1367. BytesWritten -= p->size;
  1368. uint32 newlen=EQProtocolPacket::Compress(buffer,length,tmpbuffer,2048);
  1369. memcpy(buffer,tmpbuffer,newlen);
  1370. length=newlen;
  1371. BytesWritten += newlen;
  1372. }
  1373. if (encoded) {
  1374. EQProtocolPacket::ChatEncode(buffer,length,Key);
  1375. }
  1376. *(uint16 *)(buffer+length)=htons(CRC16(buffer,length,Key));
  1377. length+=2;
  1378. }
  1379. sent_packets++;
  1380. //dump_message_column(buffer,length,"Writer: ");
  1381. //cout << "Raw Data:\n";
  1382. //DumpPacket(buffer, length);
  1383. sendto(eq_fd,(char *)buffer,length,0,(sockaddr *)&address,sizeof(address));
  1384. }
  1385. EQProtocolPacket *EQStream::Read(int eq_fd, sockaddr_in *from)
  1386. {
  1387. int socklen;
  1388. int length=0;
  1389. unsigned char buffer[2048];
  1390. EQProtocolPacket *p=NULL;
  1391. char temp[15];
  1392. socklen=sizeof(sockaddr);
  1393. #ifdef WIN32
  1394. length=recvfrom(eq_fd, (char *)buffer, 2048, 0, (struct sockaddr*)from, (int *)&socklen);
  1395. #else
  1396. length=recvfrom(eq_fd, buffer, 2048, 0, (struct sockaddr*)from, (socklen_t *)&socklen);
  1397. #endif
  1398. if (length>=2) {
  1399. DumpPacket(buffer, length);
  1400. p=new EQProtocolPacket(buffer[1],&buffer[2],length-2);
  1401. //printf("Read packet: opcode %i length %u, expected-length: %u\n",buffer[1], length, p->size);
  1402. uint32 ip=from->sin_addr.s_addr;
  1403. sprintf(temp,"%d.%d.%d.%d:%d",
  1404. *(unsigned char *)&ip,
  1405. *((unsigned char *)&ip+1),
  1406. *((unsigned char *)&ip+2),
  1407. *((unsigned char *)&ip+3),
  1408. ntohs(from->sin_port));
  1409. //cout << timestamp() << "Data from: " << temp << " OpCode 0x" << hex << setw(2) << setfill('0') << (int)p->opcode << dec << endl;
  1410. //dump_message(p->pBuffer,p->size,timestamp());
  1411. }
  1412. return p;
  1413. }
  1414. void EQStream::SendSessionResponse()
  1415. {
  1416. EQProtocolPacket *out=new EQProtocolPacket(OP_SessionResponse,NULL,sizeof(SessionResponse));
  1417. SessionResponse *Response=(SessionResponse *)out->pBuffer;
  1418. Response->Session=htonl(Session);
  1419. Response->MaxLength=htonl(MaxLen);
  1420. Response->UnknownA=2;
  1421. Response->Format=0;
  1422. if (compressed)
  1423. Response->Format|=FLAG_COMPRESSED;
  1424. if (encoded)
  1425. Response->Format|=FLAG_ENCODED;
  1426. Response->Key=htonl(Key);
  1427. out->size=sizeof(SessionResponse);
  1428. NonSequencedPush(out);
  1429. }
  1430. void EQStream::SendSessionRequest()
  1431. {
  1432. EQProtocolPacket *out=new EQProtocolPacket(OP_SessionRequest,NULL,sizeof(SessionRequest));
  1433. SessionRequest *Request=(SessionRequest *)out->pBuffer;
  1434. memset(Request,0,sizeof(SessionRequest));
  1435. Request->Session=htonl(time(NULL));
  1436. Request->MaxLength=htonl(512);
  1437. NonSequencedPush(out);
  1438. }
  1439. void EQStream::SendDisconnect(bool setstate)
  1440. {
  1441. try{
  1442. if(GetState() != ESTABLISHED && GetState() != WAIT_CLOSE)
  1443. return;
  1444. EQProtocolPacket *out=new EQProtocolPacket(OP_SessionDisconnect,NULL,sizeof(uint32)+sizeof(int16));
  1445. *(uint32 *)out->pBuffer=htonl(Session);
  1446. out->pBuffer[4] = 0;
  1447. out->pBuffer[5] = 6;
  1448. NonSequencedPush(out);
  1449. if(setstate)
  1450. SetState(CLOSING);
  1451. }
  1452. catch(...){}
  1453. }
  1454. void EQStream::InboundQueuePush(EQApplicationPacket *p)
  1455. {
  1456. MInboundQueue.lock();
  1457. InboundQueue.push_back(p);
  1458. MInboundQueue.unlock();
  1459. }
  1460. EQApplicationPacket *EQStream::PopPacket()
  1461. {
  1462. EQApplicationPacket *p=NULL;
  1463. MInboundQueue.lock();
  1464. if (InboundQueue.size()) {
  1465. p=InboundQueue.front();
  1466. InboundQueue.pop_front();
  1467. }
  1468. MInboundQueue.unlock();
  1469. if(p)
  1470. p->setVersion(client_version);
  1471. return p;
  1472. }
  1473. void EQStream::InboundQueueClear()
  1474. {
  1475. MInboundQueue.lock();
  1476. while(InboundQueue.size()){
  1477. delete InboundQueue.front();
  1478. InboundQueue.pop_front();
  1479. }
  1480. MInboundQueue.unlock();
  1481. }
  1482. void EQStream::EncryptPacket(uchar* data, int16 size){
  1483. if(size>6){
  1484. }
  1485. }
  1486. bool EQStream::HasOutgoingData()
  1487. {
  1488. bool flag;
  1489. //once closed, we have nothing more to say
  1490. if(CheckClosed())
  1491. return(false);
  1492. MOutboundQueue.lock();
  1493. flag=(!NonSequencedQueue.empty());
  1494. if (!flag) {
  1495. flag = (!SequencedQueue.empty());
  1496. }
  1497. MOutboundQueue.unlock();
  1498. if (!flag) {
  1499. MAcks.lock();
  1500. flag= (NextAckToSend>LastAckSent);
  1501. MAcks.unlock();
  1502. }
  1503. if (!flag) {
  1504. MCombinedAppPacket.lock();
  1505. flag=(CombinedAppPacket!=NULL);
  1506. MCombinedAppPacket.unlock();
  1507. }
  1508. return flag;
  1509. }
  1510. void EQStream::OutboundQueueClear()
  1511. {
  1512. MOutboundQueue.lock();
  1513. while(NonSequencedQueue.size()) {
  1514. delete NonSequencedQueue.front();
  1515. NonSequencedQueue.pop();
  1516. }
  1517. while(SequencedQueue.size()) {
  1518. delete SequencedQueue.front();
  1519. SequencedQueue.pop_front();
  1520. }
  1521. MOutboundQueue.unlock();
  1522. }
  1523. void EQStream::Process(const unsigned char *buffer, const uint32 length)
  1524. {
  1525. received_packets++;
  1526. static unsigned char newbuffer[2048];
  1527. uint32 newlength=0;
  1528. #ifdef LE_DEBUG
  1529. printf("ProcessBuffer:\n");
  1530. DumpPacket(buffer, length);
  1531. #endif
  1532. if (EQProtocolPacket::ValidateCRC(buffer,length,Key)) {
  1533. if (compressed) {
  1534. newlength=EQProtocolPacket::Decompress(buffer,length,newbuffer,2048);
  1535. #ifdef LE_DEBUG
  1536. printf("ProcessBufferDecompress:\n");
  1537. DumpPacket(buffer, newlength);
  1538. #endif
  1539. } else {
  1540. memcpy(newbuffer,buffer,length);
  1541. newlength=length;
  1542. if (encoded)
  1543. EQProtocolPacket::ChatDecode(newbuffer,newlength-2,Key);
  1544. }
  1545. #ifdef LE_DEBUG
  1546. printf("ResultProcessBuffer:\n");
  1547. DumpPacket(buffer, newlength);
  1548. #endif
  1549. uint16 opcode=ntohs(*(const uint16 *)newbuffer);
  1550. //printf("Read packet: opcode %i newlength %u, newbuffer2len: %u, newbuffer3len: %u\n",opcode, newlength, newbuffer[2], newbuffer[3]);
  1551. if(opcode > 0 && opcode <= OP_OutOfSession)
  1552. {
  1553. if (buffer[1]!=0x01 && buffer[1]!=0x02 && buffer[1]!=0x1d)
  1554. newlength-=2;
  1555. EQProtocolPacket p(newbuffer,newlength);
  1556. ProcessPacket(&p);
  1557. }
  1558. else
  1559. {
  1560. cout << "2Orig Packet: " << opcode << endl;
  1561. DumpPacket(newbuffer, newlength);
  1562. ProcessEmbeddedPacket(newbuffer, newlength, OP_Fragment);
  1563. }
  1564. ProcessQueue();
  1565. } else {
  1566. cout << "Incoming packet failed checksum:" <<endl;
  1567. dump_message_column(const_cast<unsigned char *>(buffer),length,"CRC failed: ");
  1568. }
  1569. }
  1570. long EQStream::GetMaxAckReceived()
  1571. {
  1572. MAcks.lock();
  1573. long l=MaxAckReceived;
  1574. MAcks.unlock();
  1575. return l;
  1576. }
  1577. long EQStream::GetNextAckToSend()
  1578. {
  1579. MAcks.lock();
  1580. long l=NextAckToSend;
  1581. MAcks.unlock();
  1582. return l;
  1583. }
  1584. long EQStream::GetLastAckSent()
  1585. {
  1586. MAcks.lock();
  1587. long l=LastAckSent;
  1588. MAcks.unlock();
  1589. return l;
  1590. }
  1591. void EQStream::SetMaxAckReceived(uint32 seq)
  1592. {
  1593. deque<EQProtocolPacket *>::iterator itr;
  1594. MAcks.lock();
  1595. MaxAckReceived=seq;
  1596. MAcks.unlock();
  1597. MOutboundQueue.lock();
  1598. if (long(seq) > LastSeqSent)
  1599. LastSeqSent=seq;
  1600. MResendQue.lock();
  1601. EQProtocolPacket* packet = 0;
  1602. for(itr=resend_que.begin();itr!=resend_que.end();itr++){
  1603. packet = *itr;
  1604. if(packet && packet->sequence <= seq){
  1605. safe_delete(packet);
  1606. itr = resend_que.erase(itr);
  1607. if(itr == resend_que.end())
  1608. break;
  1609. }
  1610. }
  1611. MResendQue.unlock();
  1612. MOutboundQueue.unlock();
  1613. }
  1614. void EQStream::SetNextAckToSend(uint32 seq)
  1615. {
  1616. MAcks.lock();
  1617. NextAckToSend=seq;
  1618. MAcks.unlock();
  1619. }
  1620. void EQStream::SetLastAckSent(uint32 seq)
  1621. {
  1622. MAcks.lock();
  1623. LastAckSent=seq;
  1624. MAcks.unlock();
  1625. }
  1626. void EQStream::SetLastSeqSent(uint32 seq)
  1627. {
  1628. MOutboundQueue.lock();
  1629. LastSeqSent=seq;
  1630. MOutboundQueue.unlock();
  1631. }
  1632. void EQStream::SetStreamType(EQStreamType type)
  1633. {
  1634. StreamType=type;
  1635. switch (StreamType) {
  1636. case LoginStream:
  1637. app_opcode_size=1;
  1638. compressed=false;
  1639. encoded=false;
  1640. break;
  1641. case EQ2Stream:
  1642. app_opcode_size=2;
  1643. compressed=false;
  1644. encoded=false;
  1645. break;
  1646. case ChatOrMailStream:
  1647. case ChatStream:
  1648. case MailStream:
  1649. app_opcode_size=1;
  1650. compressed=false;
  1651. encoded=true;
  1652. break;
  1653. case ZoneStream:
  1654. case WorldStream:
  1655. default:
  1656. app_opcode_size=2;
  1657. compressed=true;
  1658. encoded=false;
  1659. break;
  1660. }
  1661. }
  1662. void EQStream::ProcessQueue()
  1663. {
  1664. if (OutOfOrderpackets.empty()) {
  1665. return;
  1666. }
  1667. EQProtocolPacket* qp = NULL;
  1668. while ((qp = RemoveQueue(NextInSeq)) != NULL) {
  1669. //_log(NET__DEBUG, _L "Processing Queued Packet: Seq=%d" __L, NextInSeq);
  1670. ProcessPacket(qp);
  1671. delete qp;
  1672. //_log(NET__APP_TRACE, _L "OP_Packet Queue size=%d" __L, PacketQueue.size());
  1673. }
  1674. }
  1675. EQProtocolPacket* EQStream::RemoveQueue(uint16 seq)
  1676. {
  1677. map<unsigned short, EQProtocolPacket*>::iterator itr;
  1678. EQProtocolPacket* qp = NULL;
  1679. if ((itr = OutOfOrderpackets.find(seq)) != OutOfOrderpackets.end()) {
  1680. qp = itr->second;
  1681. OutOfOrderpackets.erase(itr);
  1682. //_log(NET__APP_TRACE, _L "OP_Packet Queue size=%d" __L, PacketQueue.size());
  1683. }
  1684. return qp;
  1685. }
  1686. void EQStream::Decay()
  1687. {
  1688. MRate.lock();
  1689. uint32 rate=DecayRate;
  1690. MRate.unlock();
  1691. if (BytesWritten>0) {
  1692. BytesWritten-=rate;
  1693. if (BytesWritten<0)
  1694. BytesWritten=0;
  1695. }
  1696. int count = 0;
  1697. MOutboundQueue.lock();
  1698. for (auto sitr = SequencedQueue.begin(); sitr != SequencedQueue.end(); ++sitr, count++) {
  1699. if (!(*sitr)->acked && (*sitr)->sent_time > 0 && ((*sitr)->sent_time + retransmittimeout) < Timer::GetCurrentTime2()) {
  1700. (*sitr)->sent_time = 0;
  1701. LogWrite(PACKET__DEBUG, 9, "Packet", "Timeout exceeded for seq %u. Flagging packet for retransmission", SequencedBase + count);
  1702. }
  1703. }
  1704. MOutboundQueue.unlock();
  1705. }
  1706. void EQStream::AdjustRates(uint32 average_delta)
  1707. {
  1708. if (average_delta && (average_delta <= AVERAGE_DELTA_MAX)) {
  1709. MRate.lock();
  1710. AverageDelta = average_delta;
  1711. RateThreshold = RATEBASE / average_delta;
  1712. DecayRate = DECAYBASE / average_delta;
  1713. if (BytesWritten > RateThreshold)
  1714. BytesWritten = RateThreshold + DecayRate;
  1715. MRate.unlock();
  1716. }
  1717. else {
  1718. AverageDelta = AVERAGE_DELTA_MAX;
  1719. }
  1720. }