ÿØÿà JFIF    ÿÛ „  ( %"1!%)+...383,7(-.+  -+++--++++---+-+-----+---------------+---+-++7-----ÿÀ  ß â" ÿÄ     ÿÄ H    !1AQaq"‘¡2B±ÁÑð#R“Ò Tbr‚²á3csƒ’ÂñDS¢³$CÿÄ   ÿÄ %  !1AQa"23‘ÿÚ   ? ôÿ ¨pŸªáÿ —åYõõ\?àÒü©ŠÄï¨pŸªáÿ —åYõõ\?àÓü©ŠÄá 0Ÿªáÿ Ÿå[úƒ ú®ði~TÁbqÐ8OÕpÿ ƒOò¤Oè`–RÂáœá™êi€ßÉ< FtŸI“öÌ8úDf´°å}“¾œ6  öFá°y¥jñÇh†ˆ¢ã/ÃÐ:ªcÈ "Y¡ðÑl>ÿ ”ÏËte:qž\oäŠe÷󲍷˜HT4&ÿ ÓÐü6ö®¿øþßèô Ÿ•7Ñi’•j|“ñì>b…þS?*Óôÿ ÓÐü*h¥£ír¶ü UãSç‚Ÿ[AÐaè[ûª•õ&õj?†Éö+EzP—WeÒírJFt ‘BŒ†Ï‡%#tE Øz ¥OÛ«!1›üä±Í™%ºÍãö]°î(–:@<‹ŒÊö×òÆt¦ãº+‡¦%ÌÁ²h´OƒJŒtMÜ>ÀÜÊw3Y´•牋4ǍýʏTì>œú=Íwhyë,¾Ôò×õ¿ßÊa»«þˆѪQ|%6ž™A õ%:øj<>É—ÿ Å_ˆCbõ¥š±ý¯Ýƒï…¶|RëócÍf溪“t.СøTÿ *Ä¿-{†çàczůŽ_–^XþŒ±miB[X±d 1,é”zEù»& î9gœf™9Ð'.;—™i}!ôšåîqêÛ٤ёý£½ÆA–àôe"A$˝Úsäÿ ÷Û #°xŸëí(l »ý3—¥5m! rt`†0~'j2(]S¦¦kv,ÚÇ l¦øJA£Šƒ J3E8ÙiŽ:cÉžúeZ°€¯\®kÖ(79«Ž:¯X”¾³Š&¡* ….‰Ž(ÜíŸ2¥ª‡×Hi²TF¤ò[¨íÈRëÉ䢍mgÑ.Ÿ<öäS0í„ǹÁU´f#Vß;Õ–…P@3ío<ä-±»Ž.L|kªÀê›fÂ6@»eu‚|ÓaÞÆŸ…¨ááå>åŠ?cKü6ùTÍÆ”†sĤÚ;H2RÚ†õ\Ö·Ÿn'¾ ñ#ºI¤Å´%çÁ­‚â7›‹qT3Iï¨ÖÚ5I7Ë!ÅOóŸ¶øÝñØôת¦$Tcö‘[«Ö³šÒ';Aþ ¸èíg A2Z"i¸vdÄ÷.iõ®§)¿]¤À†–‡É&ä{V¶iŽ”.Ó×Õÿ û?h¬Mt–íª[ÿ Ñÿ ÌV(í}=ibÔ¡›¥¢±b Lô¥‡piη_Z<‡z§èŒ)iÖwiÇ 2hÙ3·=’d÷8éŽ1¦¸c¤µ€7›7Ø ð\á)} ¹fËí›pAÃL%âc2 í§æQz¿;T8sæ°qø)QFMð‰XŒÂ±N¢aF¨…8¯!U  Z©RÊ ÖPVÄÀÍin™Ì-GˆªÅËŠ›•zË}º±ŽÍFò¹}Uw×#ä5B¤{î}Ð<ÙD é©¤&‡ïDbàÁôMÁ." ¤‡ú*õ'VŽ|¼´Úgllº¼klz[Æüï÷Aób‡Eÿ dÑ»Xx9ÃÜ£ÁT/`¼¸vI±Ýµ·Ë‚“G³þ*Ÿû´r|*}<¨îºœ @¦mÄ’M¹”.œ«Y–|6ÏU¤jç¥ÕÞqO ˜kDÆÁ¨5ÿ š;Њ¦¦€GÙk \ –Þ=â¼=SͧµªS°ÚÍpÜãQűÀõ¬?ÃÁ1Ñ•õZà?hóœ€ L¦l{Y*K˜Ù›zc˜–ˆâ ø+¾ ­-Ök¥%ùEÜA'}ˆ><ÊIè“bpÍ/qÞâvoX€w,\úªò6Z[XdÒæ­@Ö—€$òJí#é>'°Ú ôª˜<)4ryÙ£|óAÅn5žêŸyÒäMÝ2{"}‰–¤l÷ûWX\l¾Á¸góÉOÔ /óñB¤f¸çñ[.P˜ZsÊË*ßT܈§QN¢’¡¨§V¼(Üù*eÕ“”5T¨‹Âê¥FŒã½Dü[8'Ò¥a…Ú¶k7a *•›¼'Ò·\8¨ª\@\õ¢¦íq+DÙrmÎ…_ªæ»ŠÓœ¡¯’Ré9MÅ×D™lælffc+ŒÑ,ý™ÿ ¯þǤ=Å’Á7µ÷ÚÛ/“Ü€ñýã¼àí¾ÕÑ+ƒ,uµMâÀÄbm:ÒÎPæ{˜Gz[ƒ¯«® KHà`ßØŠéí¯P8Aq.C‰ à€kòpj´kN¶qô€…Õ,ÜNŠª-­{Zö’æû44‰sŽè‰îVíRœÕm" 6?³D9¡ÇTíÅꋇ`4«¸ÝÁô ï’ýorqКÇZ«x4Žâéþuïf¹µö[P ,Q£éaX±`PÉÍZ ¸äYúg üAx ’6Lê‚xÝÓ*äQ  Ï’¨hÍ =²,6ï#rÃ<¯–£»ƒ‹,–ê•€ aÛsñ'%Æ"®ÛüìBᝠHÚ3ß°©$“XnœÖ’î2ËTeûìxîß ¦å¿çÉ ðK§þ{‘t‚Ϋ¬jéîZ[ ”š7L¥4VÚCE×]m¤Øy”ä4-dz£œ§¸x.*ãÊÊ b÷•h:©‡¦s`BTÁRû¾g⻩‹jø sF¢àJøFl‘È•Xᓁà~*j¯ +(ÚÕ6-£¯÷GŠØy‚<Ç’.F‹Hœw(+)ÜÜâÈzÄäT§FߘãÏ;DmVœ3Àu@mÚüXÝü•3B¨òÌÁÛ<·ÃÜ z,Ì@õÅ·d2]ü8s÷IôÞ¯^Ç9¢u„~ëAŸï4«M? K]­ÅàPl@s_ p:°¬ZR”´›JC[CS.h‹ƒïËœ«Æ]–÷ó‚wR×k7X‰k›‘´ù¦=¡«‰¨¨Â')—71ó’c‡Ðúµ `é.{§p¹ój\Ž{1h{o±Ý=áUÊïGÖŒõ–-BÄm+AZX¶¡ ïHðæ¥JmÙ;…ä¡Ÿˆ¦ ° äšiÉg«$üMk5¤L“’çÊvïâï ,=f“"íἊ5ô¬x6{ɏžID0e¸vçmi'︧ºð9$ò¹÷*£’9ÿ ²TÔ…×>JV¥}Œ}$p[bÔ®*[jzS*8 ”·T›Í–ñUîƒwo$áè=LT™ç—~ô·¤ÈÚ$榍q‰„+´kFm)ž‹©i–ËqÞŠ‰à¶ü( ‚•§ •°ò·‡#5ª•µÊ﯅¡X¨šÁ*F#TXJÊ ušJVÍ&=iÄs1‚3•'fý§5Ñ<=[íÞ­ PÚ;ѱÌ_~Ä££8rÞ ²w;’hDT°>ÈG¬8Á²ÚzŽ®ò®qZcqJêäÞ-ö[ܘbň±çb“Š¶31²n×iƒðÕ;1¶þÉ ªX‰,ßqÏ$>•î íZ¥Z 1{ç൵+ƒÕµ¥°T$§K]á»Ûï*·¤tMI’ÂZbŽÕiÒ˜}bÓ0£ª5›¨ [5Ž^ÝœWøÂÝh° ¢OWun£¤5 a2Z.G2³YL]jåtì”ä ÁÓ‘%"©<ÔúÊ°sº UZvä‡ÄiÆÒM .÷V·™ø#kèýiíÌ–ª)µT[)BˆõÑ xB¾B€ÖT¨.¥~ð@VĶr#¸ü*åZNDŽH;âi ],©£öØpù(šºãö¼T.uCê•4@ÿ GÕÛ)Cx›®0ø#:ÏðFÒbR\(€€Ä®fã4Þ‰Fä¯HXƒÅ,†öEÑÔÜ]Öv²?tLÃvBY£ú6Êu5ÅAQ³1‘’¬x–HŒÐ‡ ^ ¸KwJôÖŽ5×CÚ¨vÜ«/B0$×k°=ðbÇ(Ï)w±A†Á† 11Í=èQšµ626ŒÜ/`G«µ<}—-Ö7KEHÈÉðóȤmݱû±·ø«Snmá=“ä«šmݱŸ¡¶~ó·“äUóJæúòB|E LêŽy´jDÔ$G¢þÐñ7óR8ýÒ…Ç› WVe#·Ÿ p·Fx~•ݤF÷0Èÿ K¯æS<6’¡WШ; ´ÿ ¥Êø\Òuî†åÝ–VNœkÒ7oòX¨Á­Ø÷FÎÑä±g÷ÿ M~Çî=p,X´ ÝÌÚÅ‹’ÃjÖ.ØöÏñ qïQ¤ÓZE†° =6·]܈ s¸>v•Ž^Ý\wq9r‰Î\¸¡kURÒ$­*‹Nq?Þª*!sŠÆ:TU_u±T+øX¡ ®¹¡,ÄâÃBTsÜ$Ø›4m椴zÜK]’’›Pƒ @€#â˜`é¹=I‡fiV•Ôî“nRm+µFPOhÍ0B£ €+¬5c v•:P'ÒyÎ ‰V~‚Ó†ÖuókDoh$å\*ö%Ю=£«…aȼ½÷Û.-½VŒŠ¼'lyî±1¬3ó#ÞE¿ÔS¤gV£m›=§\û"—WU¤ÚǼÿ ÂnÁGŒÃ ‚õN D³õNÚíŒÕ;HôyÄÈ©P¹Ä{:?R‘Ô¨âF÷ø£bÅó® JS|‚R÷ivýáâ€Æé¡è³´IئÑT!§˜•Øª‚¬â@q€wnïCWÄ@JU€ê¯m6]Ï:£âx'+ÒðXvÓ¦Úm=–´7œ $ì“B£~p%ÕŸUþ« N@¼üï~w˜ñø5®—'Ôe»¤5ã//€ž~‰Tþ›Å7•#¤× Íö pÄ$ùeåì*«ÓŠEØWEÈsßg ¦ûvžSsLpºÊW–âµEWöˬH; ™!CYõZ ÃÄf æ#1W. \uWâ\,\Çf j’<qTbên›Î[vxx£ë 'ö¨1›˜ÀM¼Pÿ H)ƒêêŒA7s,|F“ 꺸k³9Ìö*ç®;Ö!Ö$Eiž•¹ÒÚ†ýóéÝû¾ÕS®ó$’NÝäŸz¤5r¦ãÄÃD÷Üø!°ø‡Ô&@m™Ì^Ãä­d q5Lnÿ N;.6½·N|#ä"1Nƒx“ã<3('&ñßt  ~ªu”1Tb㫨9ê–›–bìd$ߣ=#ÕãÒmU¯eí$EFù5ýYô櫨æ왍Ǘ±ssM]·á¿0ÕåJRÓªîiƒ+O58ÖñªŠÒx" \µâá¨i’¤i —Ö ” M+M¤ë9‚‰A¦°Qõ¾ßøK~¼Ã‘g…Ö´~÷Ï[3GUœÒ½#…kàÔ®Ò”‰³·dWV‰IP‰Ú8u¹”E ÖqLj¾êÕCBš{A^Âß;–¨`¯¬ìö ˼ ×tìø.tƐm*n¨y4o&Àx¥n¦×î‡aupáÛj8¿m›è¶ã!o½;ß0y^ý×^EÑ¿ÒjzŒ­)vÚÑnÄL …^ªô× ‡—‚3k Îý­hï]içå–îÏ*÷ñþ»Ô CÒjøjÍznˆ´ ¹#b'Fô‹ ‰v¥'’à'T´ƒHýÍ%M‰ ƒ&ÆÇŒï1 ‘ –Þ ‰i¬s žR-Ÿ kŠ¬á¬7:þ 0ŒÅÒÕ/aÙ¬ÃÝ#Úøœ ©aiVc‰. ¹¦ãµ” ›Yg¦›ÆÎýº°f³7ƒhá·¸­}&D9¡ÂsÉÙÞèŠõØàC™¨ñbFC|´Ü(ŸƒÚÒ-%»'a Ì¿)ËÇn¿úÿ ÞŽX…4ÊÅH^ôΑí@ù¹Eh¶“L8Çjù ¼ÎåVªóR©Ï5uà V4lZß®=€xÖŸ–ÑÈ ÷”¨°¾__yM1tÉ?uÆþIkÄgæ@þ[¢†°XÃJ£j·:nkÅ¢u ‘}âGzö­/IµèŠ¬¼48q¦F°ŽR¼=ûì{´¯RýicS ÕÛ íNtÍÙï£,w4rêì®»~x(©Uñ§#Ñ&œÕ¤>ÎåÍÓ9’Ö{9eV­[Öjâ²ãu]˜å2›qÑšÕJç0€sÄ|Êëè0튔bÁ>“{×_F`Ø©ºê:µä,v¤ðfc1±"«ÔÍän1#=· Âøv~H½ÐßA¾¿Ü€Óš]Õ; I¾÷ç‚Qi†î¹9ywÔKG˜áñ zQY—§ÃÕZ07§X‚ Áh;ÁM)iÌCH-¯T‘ë|A0{Ò½LÚ–TâÖkÜ’dÀ“rmm»”جPF³ÖcbE§T€ÒxKºû’Ó®7±²(\4ŽÃ¸Uu@j™yĵ;³µ!Á¢b.W¤=mõ´êµK k ¸K^ÜÛ#p*Ü14qkZç5ïë †°5Ï%ÍÛ<Õ¤×Ô¥ê†C Õ´¼ú$ƒÖ“”]Ù¬qÞÚ[4©ý!ûÏ—Áb쳐XµA¬â~`›Çr¸8ìùÝ䫦<>ä÷«?xs´ÇÑ /á;¹øüÊÈÙà{"@Žïzâ¬[âß‚ U_<ÇŸ½4èN˜ú61®qŠu ¦þF£»äJ_ˆÙÎ~ ÞAã–Ý„Ï—rŠD;xTž‘ô`É«…suãO`?³à™ô Lý#Íc5öoæØ‚y´´÷«ZR§<&JÇ+éâô´€i!Àˆ0æAoàðLèÖ-2ŸõW.’t^–(KÁmHµV@xÜÇy®Ñø­â^:Ú3w· 7½¹°ñ¸â¹®:',«Mœ—n­Á+Ãbš LÈ‘ÄnRÓÅœ%¦²‰¨ùQ:¤f‚ "PÕtô¸…cæl…&˜Ú˜Ôkv‹ž+vŠ,=¢v­6—Xy*¥t£«<™:“aîϲ=¦6rO]XI¿Œ÷¤zÚ­›¶ 6÷”w\d ü~v®ˆÌk«^m<ÿ ¢‰Õ\)ùºŽ;… lîÙÅEŠ®cѾ@vnMÏ,¼“ñ•ŽBxðÃzãÇç%3ˆ"}Ù•Åî> BÉú;Ò]V+P˜F_´ßé> Øše|ï‡ÄOmFæÇ ãqÞ$/xÐx­z`ï9"œÜij‚!7.\Td…9M‡•iŽ‹¾‘50ÞŽn¥ß4ÉôO ¹*í^QêËÜÇÌ8=Ž§s‰'ÂëÙ«á%Pú[O †ÅP¯VsŽ°.‰,kc¶ ¬A9n˜XÎ-ÞšN["¹QÕ‰ƒMýÁߺXJæÍaLj¾×Ãmã¾ãÚ uñÒþåQô¦¥ /ÄUx:‚ÍÜ’ Đ©ØÝ3V¨‰ÕnÐ6ó*óúK­«…c ¯U òhsý­jóÔj#,ímŒRµ«lbïUTŒÑ8†Ä0œÏr`ð¡¬É Ї ë"À² ™ 6¥ f¶ ¢ÚoܱԷ-<Àî)†a¶ž'Ú»¨TXqØæ¶÷YÄHy˜9ÈIW­YÀuMFë ºÏ’AqÌ4·/Ú †ô'i$øä­=Ä Ý|öK×40è|È6p‘0§)o¥ctî§H+CA-“ xØ|ÐXŠç l8íºð3Ø:³¤¬KX¯UÿÙ # -*- test-case-name: twisted.conch.test.test_telnet -*- # Copyright (c) Twisted Matrix Laboratories. # See LICENSE for details. """ Tests for L{twisted.conch.telnet}. """ from __future__ import absolute_import, division from zope.interface import implementer from zope.interface.verify import verifyObject from twisted.internet import defer from twisted.conch import telnet from twisted.trial import unittest from twisted.test import proto_helpers from twisted.python.compat import iterbytes @implementer(telnet.ITelnetProtocol) class TestProtocol: localEnableable = () remoteEnableable = () def __init__(self): self.data = b'' self.subcmd = [] self.calls = [] self.enabledLocal = [] self.enabledRemote = [] self.disabledLocal = [] self.disabledRemote = [] def makeConnection(self, transport): d = transport.negotiationMap = {} d[b'\x12'] = self.neg_TEST_COMMAND d = transport.commandMap = transport.commandMap.copy() for cmd in ('NOP', 'DM', 'BRK', 'IP', 'AO', 'AYT', 'EC', 'EL', 'GA'): d[getattr(telnet, cmd)] = lambda arg, cmd=cmd: self.calls.append(cmd) def dataReceived(self, data): self.data += data def connectionLost(self, reason): pass def neg_TEST_COMMAND(self, payload): self.subcmd = payload def enableLocal(self, option): if option in self.localEnableable: self.enabledLocal.append(option) return True return False def disableLocal(self, option): self.disabledLocal.append(option) def enableRemote(self, option): if option in self.remoteEnableable: self.enabledRemote.append(option) return True return False def disableRemote(self, option): self.disabledRemote.append(option) class InterfacesTests(unittest.TestCase): def test_interface(self): """ L{telnet.TelnetProtocol} implements L{telnet.ITelnetProtocol} """ p = telnet.TelnetProtocol() verifyObject(telnet.ITelnetProtocol, p) class TelnetTransportTests(unittest.TestCase): """ Tests for L{telnet.TelnetTransport}. """ def setUp(self): self.p = telnet.TelnetTransport(TestProtocol) self.t = proto_helpers.StringTransport() self.p.makeConnection(self.t) def testRegularBytes(self): # Just send a bunch of bytes. None of these do anything # with telnet. They should pass right through to the # application layer. h = self.p.protocol L = [b"here are some bytes la la la", b"some more arrive here", b"lots of bytes to play with", b"la la la", b"ta de da", b"dum"] for b in L: self.p.dataReceived(b) self.assertEqual(h.data, b''.join(L)) def testNewlineHandling(self): # Send various kinds of newlines and make sure they get translated # into \n. h = self.p.protocol L = [b"here is the first line\r\n", b"here is the second line\r\0", b"here is the third line\r\n", b"here is the last line\r\0"] for b in L: self.p.dataReceived(b) self.assertEqual(h.data, L[0][:-2] + b'\n' + L[1][:-2] + b'\r' + L[2][:-2] + b'\n' + L[3][:-2] + b'\r') def testIACEscape(self): # Send a bunch of bytes and a couple quoted \xFFs. Unquoted, # \xFF is a telnet command. Quoted, one of them from each pair # should be passed through to the application layer. h = self.p.protocol L = [b"here are some bytes\xff\xff with an embedded IAC", b"and here is a test of a border escape\xff", b"\xff did you get that IAC?"] for b in L: self.p.dataReceived(b) self.assertEqual(h.data, b''.join(L).replace(b'\xff\xff', b'\xff')) def _simpleCommandTest(self, cmdName): # Send a single simple telnet command and make sure # it gets noticed and the appropriate method gets # called. h = self.p.protocol cmd = telnet.IAC + getattr(telnet, cmdName) L = [b"Here's some bytes, tra la la", b"But ono!" + cmd + b" an interrupt"] for b in L: self.p.dataReceived(b) self.assertEqual(h.calls, [cmdName]) self.assertEqual(h.data, b''.join(L).replace(cmd, b'')) def testInterrupt(self): self._simpleCommandTest("IP") def testNoOperation(self): self._simpleCommandTest("NOP") def testDataMark(self): self._simpleCommandTest("DM") def testBreak(self): self._simpleCommandTest("BRK") def testAbortOutput(self): self._simpleCommandTest("AO") def testAreYouThere(self): self._simpleCommandTest("AYT") def testEraseCharacter(self): self._simpleCommandTest("EC") def testEraseLine(self): self._simpleCommandTest("EL") def testGoAhead(self): self._simpleCommandTest("GA") def testSubnegotiation(self): # Send a subnegotiation command and make sure it gets # parsed and that the correct method is called. h = self.p.protocol cmd = telnet.IAC + telnet.SB + b'\x12hello world' + telnet.IAC + telnet.SE L = [b"These are some bytes but soon" + cmd, b"there will be some more"] for b in L: self.p.dataReceived(b) self.assertEqual(h.data, b''.join(L).replace(cmd, b'')) self.assertEqual(h.subcmd, list(iterbytes(b"hello world"))) def testSubnegotiationWithEmbeddedSE(self): # Send a subnegotiation command with an embedded SE. Make sure # that SE gets passed to the correct method. h = self.p.protocol cmd = (telnet.IAC + telnet.SB + b'\x12' + telnet.SE + telnet.IAC + telnet.SE) L = [b"Some bytes are here" + cmd + b"and here", b"and here"] for b in L: self.p.dataReceived(b) self.assertEqual(h.data, b''.join(L).replace(cmd, b'')) self.assertEqual(h.subcmd, [telnet.SE]) def testBoundarySubnegotiation(self): # Send a subnegotiation command. Split it at every possible byte boundary # and make sure it always gets parsed and that it is passed to the correct # method. cmd = (telnet.IAC + telnet.SB + b'\x12' + telnet.SE + b'hello' + telnet.IAC + telnet.SE) for i in range(len(cmd)): h = self.p.protocol = TestProtocol() h.makeConnection(self.p) a, b = cmd[:i], cmd[i:] L = [b"first part" + a, b + b"last part"] for data in L: self.p.dataReceived(data) self.assertEqual(h.data, b''.join(L).replace(cmd, b'')) self.assertEqual(h.subcmd, [telnet.SE] + list(iterbytes(b'hello'))) def _enabledHelper(self, o, eL=[], eR=[], dL=[], dR=[]): self.assertEqual(o.enabledLocal, eL) self.assertEqual(o.enabledRemote, eR) self.assertEqual(o.disabledLocal, dL) self.assertEqual(o.disabledRemote, dR) def testRefuseWill(self): # Try to enable an option. The server should refuse to enable it. cmd = telnet.IAC + telnet.WILL + b'\x12' data = b"surrounding bytes" + cmd + b"to spice things up" self.p.dataReceived(data) self.assertEqual(self.p.protocol.data, data.replace(cmd, b'')) self.assertEqual(self.t.value(), telnet.IAC + telnet.DONT + b'\x12') self._enabledHelper(self.p.protocol) def testRefuseDo(self): # Try to enable an option. The server should refuse to enable it. cmd = telnet.IAC + telnet.DO + b'\x12' data = b"surrounding bytes" + cmd + b"to spice things up" self.p.dataReceived(data) self.assertEqual(self.p.protocol.data, data.replace(cmd, b'')) self.assertEqual(self.t.value(), telnet.IAC + telnet.WONT + b'\x12') self._enabledHelper(self.p.protocol) def testAcceptDo(self): # Try to enable an option. The option is in our allowEnable # list, so we will allow it to be enabled. cmd = telnet.IAC + telnet.DO + b'\x19' data = b'padding' + cmd + b'trailer' h = self.p.protocol h.localEnableable = (b'\x19',) self.p.dataReceived(data) self.assertEqual(self.t.value(), telnet.IAC + telnet.WILL + b'\x19') self._enabledHelper(h, eL=[b'\x19']) def testAcceptWill(self): # Same as testAcceptDo, but reversed. cmd = telnet.IAC + telnet.WILL + b'\x91' data = b'header' + cmd + b'padding' h = self.p.protocol h.remoteEnableable = (b'\x91',) self.p.dataReceived(data) self.assertEqual(self.t.value(), telnet.IAC + telnet.DO + b'\x91') self._enabledHelper(h, eR=[b'\x91']) def testAcceptWont(self): # Try to disable an option. The server must allow any option to # be disabled at any time. Make sure it disables it and sends # back an acknowledgement of this. cmd = telnet.IAC + telnet.WONT + b'\x29' # Jimmy it - after these two lines, the server will be in a state # such that it believes the option to have been previously enabled # via normal negotiation. s = self.p.getOptionState(b'\x29') s.him.state = 'yes' data = b"fiddle dee" + cmd self.p.dataReceived(data) self.assertEqual(self.p.protocol.data, data.replace(cmd, b'')) self.assertEqual(self.t.value(), telnet.IAC + telnet.DONT + b'\x29') self.assertEqual(s.him.state, 'no') self._enabledHelper(self.p.protocol, dR=[b'\x29']) def testAcceptDont(self): # Try to disable an option. The server must allow any option to # be disabled at any time. Make sure it disables it and sends # back an acknowledgement of this. cmd = telnet.IAC + telnet.DONT + b'\x29' # Jimmy it - after these two lines, the server will be in a state # such that it believes the option to have beenp previously enabled # via normal negotiation. s = self.p.getOptionState(b'\x29') s.us.state = 'yes' data = b"fiddle dum " + cmd self.p.dataReceived(data) self.assertEqual(self.p.protocol.data, data.replace(cmd, b'')) self.assertEqual(self.t.value(), telnet.IAC + telnet.WONT + b'\x29') self.assertEqual(s.us.state, 'no') self._enabledHelper(self.p.protocol, dL=[b'\x29']) def testIgnoreWont(self): # Try to disable an option. The option is already disabled. The # server should send nothing in response to this. cmd = telnet.IAC + telnet.WONT + b'\x47' data = b"dum de dum" + cmd + b"tra la la" self.p.dataReceived(data) self.assertEqual(self.p.protocol.data, data.replace(cmd, b'')) self.assertEqual(self.t.value(), b'') self._enabledHelper(self.p.protocol) def testIgnoreDont(self): # Try to disable an option. The option is already disabled. The # server should send nothing in response to this. Doing so could # lead to a negotiation loop. cmd = telnet.IAC + telnet.DONT + b'\x47' data = b"dum de dum" + cmd + b"tra la la" self.p.dataReceived(data) self.assertEqual(self.p.protocol.data, data.replace(cmd, b'')) self.assertEqual(self.t.value(), b'') self._enabledHelper(self.p.protocol) def testIgnoreWill(self): # Try to enable an option. The option is already enabled. The # server should send nothing in response to this. Doing so could # lead to a negotiation loop. cmd = telnet.IAC + telnet.WILL + b'\x56' # Jimmy it - after these two lines, the server will be in a state # such that it believes the option to have been previously enabled # via normal negotiation. s = self.p.getOptionState(b'\x56') s.him.state = 'yes' data = b"tra la la" + cmd + b"dum de dum" self.p.dataReceived(data) self.assertEqual(self.p.protocol.data, data.replace(cmd, b'')) self.assertEqual(self.t.value(), b'') self._enabledHelper(self.p.protocol) def testIgnoreDo(self): # Try to enable an option. The option is already enabled. The # server should send nothing in response to this. Doing so could # lead to a negotiation loop. cmd = telnet.IAC + telnet.DO + b'\x56' # Jimmy it - after these two lines, the server will be in a state # such that it believes the option to have been previously enabled # via normal negotiation. s = self.p.getOptionState(b'\x56') s.us.state = 'yes' data = b"tra la la" + cmd + b"dum de dum" self.p.dataReceived(data) self.assertEqual(self.p.protocol.data, data.replace(cmd, b'')) self.assertEqual(self.t.value(), b'') self._enabledHelper(self.p.protocol) def testAcceptedEnableRequest(self): # Try to enable an option through the user-level API. This # returns a Deferred that fires when negotiation about the option # finishes. Make sure it fires, make sure state gets updated # properly, make sure the result indicates the option was enabled. d = self.p.do(b'\x42') h = self.p.protocol h.remoteEnableable = (b'\x42',) self.assertEqual(self.t.value(), telnet.IAC + telnet.DO + b'\x42') self.p.dataReceived(telnet.IAC + telnet.WILL + b'\x42') d.addCallback(self.assertEqual, True) d.addCallback(lambda _: self._enabledHelper(h, eR=[b'\x42'])) return d def test_refusedEnableRequest(self): """ If the peer refuses to enable an option we request it to enable, the L{Deferred} returned by L{TelnetProtocol.do} fires with an L{OptionRefused} L{Failure}. """ # Try to enable an option through the user-level API. This returns a # Deferred that fires when negotiation about the option finishes. Make # sure it fires, make sure state gets updated properly, make sure the # result indicates the option was enabled. self.p.protocol.remoteEnableable = (b'\x42',) d = self.p.do(b'\x42') self.assertEqual(self.t.value(), telnet.IAC + telnet.DO + b'\x42') s = self.p.getOptionState(b'\x42') self.assertEqual(s.him.state, 'no') self.assertEqual(s.us.state, 'no') self.assertTrue(s.him.negotiating) self.assertFalse(s.us.negotiating) self.p.dataReceived(telnet.IAC + telnet.WONT + b'\x42') d = self.assertFailure(d, telnet.OptionRefused) d.addCallback(lambda ignored: self._enabledHelper(self.p.protocol)) d.addCallback( lambda ignored: self.assertFalse(s.him.negotiating)) return d def test_refusedEnableOffer(self): """ If the peer refuses to allow us to enable an option, the L{Deferred} returned by L{TelnetProtocol.will} fires with an L{OptionRefused} L{Failure}. """ # Try to offer an option through the user-level API. This returns a # Deferred that fires when negotiation about the option finishes. Make # sure it fires, make sure state gets updated properly, make sure the # result indicates the option was enabled. self.p.protocol.localEnableable = (b'\x42',) d = self.p.will(b'\x42') self.assertEqual(self.t.value(), telnet.IAC + telnet.WILL + b'\x42') s = self.p.getOptionState(b'\x42') self.assertEqual(s.him.state, 'no') self.assertEqual(s.us.state, 'no') self.assertFalse(s.him.negotiating) self.assertTrue(s.us.negotiating) self.p.dataReceived(telnet.IAC + telnet.DONT + b'\x42') d = self.assertFailure(d, telnet.OptionRefused) d.addCallback(lambda ignored: self._enabledHelper(self.p.protocol)) d.addCallback( lambda ignored: self.assertFalse(s.us.negotiating)) return d def testAcceptedDisableRequest(self): # Try to disable an option through the user-level API. This # returns a Deferred that fires when negotiation about the option # finishes. Make sure it fires, make sure state gets updated # properly, make sure the result indicates the option was enabled. s = self.p.getOptionState(b'\x42') s.him.state = 'yes' d = self.p.dont(b'\x42') self.assertEqual(self.t.value(), telnet.IAC + telnet.DONT + b'\x42') self.p.dataReceived(telnet.IAC + telnet.WONT + b'\x42') d.addCallback(self.assertEqual, True) d.addCallback(lambda _: self._enabledHelper(self.p.protocol, dR=[b'\x42'])) return d def testNegotiationBlocksFurtherNegotiation(self): # Try to disable an option, then immediately try to enable it, then # immediately try to disable it. Ensure that the 2nd and 3rd calls # fail quickly with the right exception. s = self.p.getOptionState(b'\x24') s.him.state = 'yes' self.p.dont(b'\x24') # fires after the first line of _final def _do(x): d = self.p.do(b'\x24') return self.assertFailure(d, telnet.AlreadyNegotiating) def _dont(x): d = self.p.dont(b'\x24') return self.assertFailure(d, telnet.AlreadyNegotiating) def _final(x): self.p.dataReceived(telnet.IAC + telnet.WONT + b'\x24') # an assertion that only passes if d2 has fired self._enabledHelper(self.p.protocol, dR=[b'\x24']) # Make sure we allow this self.p.protocol.remoteEnableable = (b'\x24',) d = self.p.do(b'\x24') self.p.dataReceived(telnet.IAC + telnet.WILL + b'\x24') d.addCallback(self.assertEqual, True) d.addCallback(lambda _: self._enabledHelper(self.p.protocol, eR=[b'\x24'], dR=[b'\x24'])) return d d = _do(None) d.addCallback(_dont) d.addCallback(_final) return d def testSuperfluousDisableRequestRaises(self): # Try to disable a disabled option. Make sure it fails properly. d = self.p.dont(b'\xab') return self.assertFailure(d, telnet.AlreadyDisabled) def testSuperfluousEnableRequestRaises(self): # Try to disable a disabled option. Make sure it fails properly. s = self.p.getOptionState(b'\xab') s.him.state = 'yes' d = self.p.do(b'\xab') return self.assertFailure(d, telnet.AlreadyEnabled) def testLostConnectionFailsDeferreds(self): d1 = self.p.do(b'\x12') d2 = self.p.do(b'\x23') d3 = self.p.do(b'\x34') class TestException(Exception): pass self.p.connectionLost(TestException("Total failure!")) d1 = self.assertFailure(d1, TestException) d2 = self.assertFailure(d2, TestException) d3 = self.assertFailure(d3, TestException) return defer.gatherResults([d1, d2, d3]) class TestTelnet(telnet.Telnet): """ A trivial extension of the telnet protocol class useful to unit tests. """ def __init__(self): telnet.Telnet.__init__(self) self.events = [] def applicationDataReceived(self, data): """ Record the given data in C{self.events}. """ self.events.append(('bytes', data)) def unhandledCommand(self, command, data): """ Record the given command in C{self.events}. """ self.events.append(('command', command, data)) def unhandledSubnegotiation(self, command, data): """ Record the given subnegotiation command in C{self.events}. """ self.events.append(('negotiate', command, data)) class TelnetTests(unittest.TestCase): """ Tests for L{telnet.Telnet}. L{telnet.Telnet} implements the TELNET protocol (RFC 854), including option and suboption negotiation, and option state tracking. """ def setUp(self): """ Create an unconnected L{telnet.Telnet} to be used by tests. """ self.protocol = TestTelnet() def test_enableLocal(self): """ L{telnet.Telnet.enableLocal} should reject all options, since L{telnet.Telnet} does not know how to implement any options. """ self.assertFalse(self.protocol.enableLocal(b'\0')) def test_enableRemote(self): """ L{telnet.Telnet.enableRemote} should reject all options, since L{telnet.Telnet} does not know how to implement any options. """ self.assertFalse(self.protocol.enableRemote(b'\0')) def test_disableLocal(self): """ It is an error for L{telnet.Telnet.disableLocal} to be called, since L{telnet.Telnet.enableLocal} will never allow any options to be enabled locally. If a subclass overrides enableLocal, it must also override disableLocal. """ self.assertRaises(NotImplementedError, self.protocol.disableLocal, b'\0') def test_disableRemote(self): """ It is an error for L{telnet.Telnet.disableRemote} to be called, since L{telnet.Telnet.enableRemote} will never allow any options to be enabled remotely. If a subclass overrides enableRemote, it must also override disableRemote. """ self.assertRaises(NotImplementedError, self.protocol.disableRemote, b'\0') def test_requestNegotiation(self): """ L{telnet.Telnet.requestNegotiation} formats the feature byte and the payload bytes into the subnegotiation format and sends them. See RFC 855. """ transport = proto_helpers.StringTransport() self.protocol.makeConnection(transport) self.protocol.requestNegotiation(b'\x01', b'\x02\x03') self.assertEqual( transport.value(), # IAC SB feature bytes IAC SE b'\xff\xfa\x01\x02\x03\xff\xf0') def test_requestNegotiationEscapesIAC(self): """ If the payload for a subnegotiation includes I{IAC}, it is escaped by L{telnet.Telnet.requestNegotiation} with another I{IAC}. See RFC 855. """ transport = proto_helpers.StringTransport() self.protocol.makeConnection(transport) self.protocol.requestNegotiation(b'\x01', b'\xff') self.assertEqual( transport.value(), b'\xff\xfa\x01\xff\xff\xff\xf0') def _deliver(self, data, *expected): """ Pass the given bytes to the protocol's C{dataReceived} method and assert that the given events occur. """ received = self.protocol.events = [] self.protocol.dataReceived(data) self.assertEqual(received, list(expected)) def test_oneApplicationDataByte(self): """ One application-data byte in the default state gets delivered right away. """ self._deliver(b'a', ('bytes', b'a')) def test_twoApplicationDataBytes(self): """ Two application-data bytes in the default state get delivered together. """ self._deliver(b'bc', ('bytes', b'bc')) def test_threeApplicationDataBytes(self): """ Three application-data bytes followed by a control byte get delivered, but the control byte doesn't. """ self._deliver(b'def' + telnet.IAC, ('bytes', b'def')) def test_escapedControl(self): """ IAC in the escaped state gets delivered and so does another application-data byte following it. """ self._deliver(telnet.IAC) self._deliver(telnet.IAC + b'g', ('bytes', telnet.IAC + b'g')) def test_carriageReturn(self): """ A carriage return only puts the protocol into the newline state. A linefeed in the newline state causes just the newline to be delivered. A nul in the newline state causes a carriage return to be delivered. An IAC in the newline state causes a carriage return to be delivered and puts the protocol into the escaped state. Anything else causes a carriage return and that thing to be delivered. """ self._deliver(b'\r') self._deliver(b'\n', ('bytes', b'\n')) self._deliver(b'\r\n', ('bytes', b'\n')) self._deliver(b'\r') self._deliver(b'\0', ('bytes', b'\r')) self._deliver(b'\r\0', ('bytes', b'\r')) self._deliver(b'\r') self._deliver(b'a', ('bytes', b'\ra')) self._deliver(b'\ra', ('bytes', b'\ra')) self._deliver(b'\r') self._deliver( telnet.IAC + telnet.IAC + b'x', ('bytes', b'\r' + telnet.IAC + b'x')) def test_applicationDataBeforeSimpleCommand(self): """ Application bytes received before a command are delivered before the command is processed. """ self._deliver( b'x' + telnet.IAC + telnet.NOP, ('bytes', b'x'), ('command', telnet.NOP, None)) def test_applicationDataBeforeCommand(self): """ Application bytes received before a WILL/WONT/DO/DONT are delivered before the command is processed. """ self.protocol.commandMap = {} self._deliver( b'y' + telnet.IAC + telnet.WILL + b'\x00', ('bytes', b'y'), ('command', telnet.WILL, b'\x00')) def test_applicationDataBeforeSubnegotiation(self): """ Application bytes received before a subnegotiation command are delivered before the negotiation is processed. """ self._deliver( b'z' + telnet.IAC + telnet.SB + b'Qx' + telnet.IAC + telnet.SE, ('bytes', b'z'), ('negotiate', b'Q', [b'x']))