meow/test/telnet_test.rb

# encoding: UTF-8

require "zlib"

require_relative "test_helper"
require_relative "../lib/meow/telnet"

module MeowTelnetTest
  Telnet = MeowClient::Telnet

  def self.bytes(*values)
    values.pack("C*")
  end

  def self.collect(parser, chunks)
    chunks.each_with_object({:data => +"".b, :replies => [], :events => []}) do |chunk, result|
      parsed = parser.feed(chunk)
      result[:data] << parsed.data
      result[:replies].concat(parsed.replies)
      result[:events].concat(parsed.events)
    end
  end

  MeowTest.test("Telnet preserves data and unescapes IAC") do
    parser = Telnet::Parser.new
    result = parser.feed("look".b + bytes(Telnet::IAC, Telnet::IAC) + "!".b)
    MeowTest.assert_equal("look".b + bytes(Telnet::IAC) + "!".b, result.data)
    MeowTest.assert_empty(result.replies)
    MeowTest.assert_empty(result.events)
  end

  MeowTest.test("Telnet negotiation survives every input boundary") do
    packet = bytes(Telnet::IAC, Telnet::WILL, Telnet::ECHO)
    whole = collect(Telnet::Parser.new, [packet])
    split = collect(Telnet::Parser.new, packet.bytes.map { |byte| bytes(byte) })
    MeowTest.assert_equal(whole, split)
    MeowTest.assert_equal([bytes(Telnet::IAC, Telnet::DO, Telnet::ECHO)], split[:replies])
    MeowTest.assert_equal([[:remote_echo, true]], split[:events])
  end

  MeowTest.test("Telnet tracks echo transitions without duplicate replies") do
    parser = Telnet::Parser.new
    enabled = parser.feed(bytes(Telnet::IAC, Telnet::WILL, Telnet::ECHO))
    duplicate = parser.feed(bytes(Telnet::IAC, Telnet::WILL, Telnet::ECHO))
    disabled = parser.feed(bytes(Telnet::IAC, Telnet::WONT, Telnet::ECHO))
    disabled_duplicate = parser.feed(bytes(Telnet::IAC, Telnet::WONT, Telnet::ECHO))

    MeowTest.assert_equal([bytes(Telnet::IAC, Telnet::DO, Telnet::ECHO)], enabled.replies)
    MeowTest.assert_empty(duplicate.replies)
    MeowTest.assert_equal([bytes(Telnet::IAC, Telnet::DONT, Telnet::ECHO)], disabled.replies)
    MeowTest.assert_equal([[:remote_echo, false]], disabled.events)
    MeowTest.assert_empty(disabled_duplicate.replies)
    MeowTest.refute(parser.remote_echo)
  end

  MeowTest.test("Telnet accepts supported options and rejects unknown options") do
    parser = Telnet::Parser.new
    remote = parser.feed(bytes(Telnet::IAC, Telnet::WILL, Telnet::SUPPRESS_GO_AHEAD))
    remote_unknown = parser.feed(bytes(Telnet::IAC, Telnet::WILL, 42))
    local = parser.feed(bytes(Telnet::IAC, Telnet::DO, Telnet::TERMINAL_TYPE))
    local_unknown = parser.feed(bytes(Telnet::IAC, Telnet::DO, 43))

    MeowTest.assert_equal([bytes(Telnet::IAC, Telnet::DO, Telnet::SUPPRESS_GO_AHEAD)], remote.replies)
    MeowTest.assert_equal([bytes(Telnet::IAC, Telnet::DONT, 42)], remote_unknown.replies)
    MeowTest.assert_equal([bytes(Telnet::IAC, Telnet::WILL, Telnet::TERMINAL_TYPE)], local.replies)
    MeowTest.assert_equal([bytes(Telnet::IAC, Telnet::WONT, 43)], local_unknown.replies)
  end

  MeowTest.test("Telnet sends terminal type only after accepting the option") do
    request = bytes(Telnet::IAC, Telnet::SB, Telnet::TERMINAL_TYPE, Telnet::TTYPE_SEND, Telnet::IAC, Telnet::SE)
    parser = Telnet::Parser.new
    before = parser.feed(request)
    parser.feed(bytes(Telnet::IAC, Telnet::DO, Telnet::TERMINAL_TYPE))
    after = collect(parser, request.bytes.map { |byte| bytes(byte) })
    expected = bytes(Telnet::IAC, Telnet::SB, Telnet::TERMINAL_TYPE, Telnet::TTYPE_IS) + "MEOW".b + bytes(Telnet::IAC, Telnet::SE)

    MeowTest.assert_empty(before.replies)
    MeowTest.assert_equal([[:subnegotiation, Telnet::TERMINAL_TYPE]], before.events)
    MeowTest.assert_equal([expected], after[:replies])
    MeowTest.assert_equal([[:terminal_type, "MEOW"]], after[:events])
  end

  MeowTest.test("Telnet handles escaped IAC inside subnegotiation") do
    parser = Telnet::Parser.new
    packet = bytes(Telnet::IAC, Telnet::SB, 77, 1, Telnet::IAC, Telnet::IAC, 2, Telnet::IAC, Telnet::SE)
    result = collect(parser, packet.bytes.map { |byte| bytes(byte) })
    MeowTest.assert_empty(result[:data])
    MeowTest.assert_empty(result[:replies])
    MeowTest.assert_equal([[:subnegotiation, 77]], result[:events])
  end

  MeowTest.test("Telnet ignores unknown commands and resumes ordinary data") do
    parser = Telnet::Parser.new
    result = parser.feed("a".b + bytes(Telnet::IAC, 241) + "b".b)
    MeowTest.assert_equal("ab".b, result.data)
    MeowTest.assert_empty(result.replies)
    MeowTest.assert_empty(result.events)
  end

  MeowTest.test("Telnet negotiates MCCP2 once") do
    parser = Telnet::Parser.new
    packet = bytes(Telnet::IAC, Telnet::WILL, Telnet::COMPRESS2)
    first = parser.feed(packet)
    duplicate = parser.feed(packet)

    MeowTest.assert_equal([bytes(Telnet::IAC, Telnet::DO, Telnet::COMPRESS2)], first.replies)
    MeowTest.assert_empty(duplicate.replies)
  end

  MeowTest.test("Telnet starts MCCP2 only after negotiation") do
    marker = bytes(Telnet::IAC, Telnet::SB, Telnet::COMPRESS2, Telnet::IAC, Telnet::SE)
    parser = Telnet::Parser.new
    before = parser.feed(marker)
    parser.feed(bytes(Telnet::IAC, Telnet::WILL, Telnet::COMPRESS2))
    after = parser.feed(marker + Zlib::Deflate.deflate("compressed text".b))

    MeowTest.assert_equal([[:subnegotiation, Telnet::COMPRESS2]], before.events)
    MeowTest.assert_equal("compressed text".b, after.data)
    MeowTest.assert_equal([[:compression, true]], after.events)
  ensure
    parser.close if parser != nil
  end

  MeowTest.test("MCCP2 survives every activation and compressed input boundary") do
    negotiation = bytes(Telnet::IAC, Telnet::WILL, Telnet::COMPRESS2)
    marker = bytes(Telnet::IAC, Telnet::SB, Telnet::COMPRESS2, Telnet::IAC, Telnet::SE)
    payload = "room description\r\n".b
    packet = negotiation + marker + Zlib::Deflate.deflate(payload)
    whole_parser = Telnet::Parser.new
    split_parser = Telnet::Parser.new
    whole = collect(whole_parser, [packet])
    split = collect(split_parser, packet.bytes.map { |byte| bytes(byte) })

    MeowTest.assert_equal(whole, split)
    MeowTest.assert_equal(payload, split[:data])
    MeowTest.assert_equal([[:compression, true]], split[:events])
  ensure
    whole_parser.close if whole_parser != nil
    split_parser.close if split_parser != nil
  end

  MeowTest.test("MCCP2 inflated bytes continue through Telnet parsing") do
    parser = Telnet::Parser.new
    parser.feed(bytes(Telnet::IAC, Telnet::WILL, Telnet::COMPRESS2))
    marker = bytes(Telnet::IAC, Telnet::SB, Telnet::COMPRESS2, Telnet::IAC, Telnet::SE)
    payload = "hello".b + bytes(Telnet::IAC, Telnet::WILL, Telnet::ECHO) + " world".b
    result = parser.feed(marker + Zlib::Deflate.deflate(payload))

    MeowTest.assert_equal("hello world".b, result.data)
    MeowTest.assert_equal([bytes(Telnet::IAC, Telnet::DO, Telnet::ECHO)], result.replies)
    MeowTest.assert_equal([[:compression, true], [:remote_echo, true]], result.events)
  ensure
    parser.close if parser != nil
  end



  MeowTest.test("MCCP2 keeps one compressed stream across input chunks") do
    parser = Telnet::Parser.new
    deflater = Zlib::Deflate.new
    parser.feed(bytes(Telnet::IAC, Telnet::WILL, Telnet::COMPRESS2))
    marker = bytes(Telnet::IAC, Telnet::SB, Telnet::COMPRESS2, Telnet::IAC, Telnet::SE)
    first = deflater.deflate("first ".b, Zlib::SYNC_FLUSH)
    MeowTest.assert_equal("first ".b, parser.feed(marker + first).data)

    second = deflater.deflate("second".b, Zlib::SYNC_FLUSH)
    MeowTest.assert_equal("second".b, parser.feed(second).data)
  ensure
    parser.close if parser != nil
    deflater.close if deflater != nil && !deflater.closed?
  end

  MeowTest.test("MCCP2 rejects malformed streams and data after stream end") do
    marker = bytes(Telnet::IAC, Telnet::SB, Telnet::COMPRESS2, Telnet::IAC, Telnet::SE)
    malformed = Telnet::Parser.new
    malformed.feed(bytes(Telnet::IAC, Telnet::WILL, Telnet::COMPRESS2))
    MeowTest.assert_raises(Telnet::CompressionError) { malformed.feed(marker + "not zlib".b) }

    finished = Telnet::Parser.new
    finished.feed(bytes(Telnet::IAC, Telnet::WILL, Telnet::COMPRESS2))
    finished.feed(marker + Zlib::Deflate.deflate("done".b))
    MeowTest.assert_raises(Telnet::CompressionError) { finished.feed("extra".b) }
  ensure
    malformed.close if malformed != nil
    finished.close if finished != nil
  end

  MeowTest.test("MCCP2 limits inflated output per feed") do
    parser = Telnet::Parser.new
    parser.feed(bytes(Telnet::IAC, Telnet::WILL, Telnet::COMPRESS2))
    marker = bytes(Telnet::IAC, Telnet::SB, Telnet::COMPRESS2, Telnet::IAC, Telnet::SE)
    oversized = "a".b * (Telnet::MAX_INFLATED_BYTES_PER_FEED + 1)
    MeowTest.assert_raises(Telnet::CompressionError) do
      parser.feed(marker + Zlib::Deflate.deflate(oversized))
    end
  ensure
    parser.close if parser != nil
  end
end