libbase/RTMP.cpp

   1 //
   2 //   Copyright (C) 2007, 2008, 2009, 2010 Free Software Foundation, Inc.
   3 //
   4 // This program is free software; you can redistribute it and/or modify
   5 // it under the terms of the GNU General Public License as published by
   6 // the Free Software Foundation; either version 3 of the License, or
   7 // (at your option) any later version.
   8 //
   9 // This program 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 //
  14 // You should have received a copy of the GNU General Public License
  15 // along with this program; if not, write to the Free Software
  16 // Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301  USA
  17
  18 #include <cstdlib>
  19 #include <cstring>
  20 #include <cassert>
  21 #include <cstdio>
  22 #include <boost/lexical_cast.hpp>
  23
  24 #include "GnashSystemNetHeaders.h"
  25
  26 // Replace!!
  27 #ifndef _WIN32
  28 # include <sys/times.h>
  29 #else
  30 // TODO: use uptime properly on win32.
  31 # include <ctime>
  32 #endif
  33
  34 #include "RTMP.h"
  35 #include "log.h"
  36 #include "AMF.h"
  37 #include "GnashAlgorithm.h"
  38 #include "URL.h"
  39 #include "ClockTime.h"
  40
  41 namespace gnash {
  42 namespace rtmp {
  43
  44 namespace {
  45
  46     bool sendBytesReceived(RTMP* r);
  47
  48     void handleMetadata(RTMP& r, const boost::uint8_t *payload,
  49             unsigned int len);
  50     void handleChangeChunkSize(RTMP& r, const RTMPPacket& packet);
  51     void handleControl(RTMP& r, const RTMPPacket& packet);
  52     void handleServerBW(RTMP& r, const RTMPPacket& packet);
  53     void handleClientBW(RTMP& r, const RTMPPacket& packet);
  54
  55     void setupInvokePacket(RTMPPacket& packet);
  56     boost::uint32_t getUptime();
  57
  58     boost::int32_t decodeInt32LE(const boost::uint8_t* c);
  59     int encodeInt32LE(boost::uint8_t *output, int nVal);
  60     unsigned int decodeInt24(const boost::uint8_t* c);
  61     boost::uint8_t* encodeInt16(boost::uint8_t *output, boost::uint8_t *outend,
  62             short nVal);
  63     boost::uint8_t* encodeInt24(boost::uint8_t *output, boost::uint8_t *outend,
  64             int nVal);
  65     boost::uint8_t* encodeInt32(boost::uint8_t *output, boost::uint8_t *outend,
  66             int nVal);
  67
  68     static const int packetSize[] = { 12, 8, 4, 1 };
  69
  70 }
  71
  72 namespace {
  73
  74 /// A random generator for generating the signature.
  75 //
  76 /// TODO: do this properly (it's currently not very random).
  77 struct RandomByte
  78 {
  79     bool operator()() const {
  80         return std::rand() % 256;
  81     }
  82 };
  83
  84 }
  85
  86 /// A utility functor for carrying out the handshake.
  87 class HandShaker
  88 {
  89 public:
  90
  91     static const int sigSize = 1536;
  92
  93     HandShaker(Socket& s);
  94
  95     /// Calls the next stage in the handshake process.
  96     void call();
  97
  98     bool success() const {
  99         return _complete;
 100     }
 101
 102     bool error() const {
 103         return _error || _socket.bad();
 104     }
 105
 106 private:
 107
 108     /// These are the stages of the handshake.
 109     //
 110     /// If the socket is not ready, they will return false. If the socket
 111     /// is in error, they will set _error.
 112     bool stage0();
 113     bool stage1();
 114     bool stage2();
 115     bool stage3();
 116
 117     Socket _socket;
 118     std::vector<boost::uint8_t> _sendBuf;
 119     std::vector<boost::uint8_t> _recvBuf;
 120     bool _error;
 121     bool _complete;
 122     size_t _stage;
 123 };
 124
 125 RTMPPacket::RTMPPacket(size_t reserve)
 126     :
 127     header(),
 128     buffer(new SimpleBuffer(reserve + RTMPHeader::headerSize)),
 129     bytesRead(0)
 130 {
 131     // This is space for the header be filled in later.
 132     buffer->resize(RTMPHeader::headerSize);
 133 }
 134
 135 RTMPPacket::RTMPPacket(const RTMPPacket& other)
 136     :
 137     header(other.header),
 138     buffer(other.buffer)
 139 {}
 140
 141 const size_t RTMPHeader::headerSize;
 142
 143 RTMP::RTMP()
 144     :
 145     _inChunkSize(RTMP_DEFAULT_CHUNKSIZE),
 146     m_mediaChannel(0),
 147     m_nClientBW2(2),
 148     _bytesIn(0),
 149     _bytesInSent(0),
 150     _serverBandwidth(2500000),
 151     _bandwidth(2500000),
 152     _outChunkSize(RTMP_DEFAULT_CHUNKSIZE),
 153     _connected(false),
 154     _error(false)
 155 {
 156 }
 157
 158 RTMP::~RTMP()
 159 {
 160 }
 161
 162 bool
 163 RTMP::hasPacket(ChannelType t, size_t channel) const
 164 {
 165     const ChannelSet& set = (t == CHANNELS_OUT) ? _outChannels : _inChannels;
 166     return set.find(channel) != set.end();
 167 }
 168
 169 RTMPPacket&
 170 RTMP::getPacket(ChannelType t, size_t channel)
 171 {
 172     ChannelSet& set = (t == CHANNELS_OUT) ? _outChannels : _inChannels;
 173     return set[channel];
 174 }
 175
 176 RTMPPacket&
 177 RTMP::storePacket(ChannelType t, size_t channel, const RTMPPacket& p)
 178 {
 179     ChannelSet& set = (t == CHANNELS_OUT) ? _outChannels : _inChannels;
 180     RTMPPacket& stored = set[channel];
 181     stored = p;
 182     return stored;
 183 }
 184
 185 void
 186 RTMP::setBufferTime(size_t size, int streamID)
 187 {
 188     sendCtrl(*this, CONTROL_BUFFER_TIME, streamID, size);
 189 }
 190
 191 void
 192 RTMP::call(const SimpleBuffer& amf)
 193 {
 194     RTMPPacket p(amf.size());
 195     setupInvokePacket(p);
 196
 197     // Copy the data.
 198     p.buffer->append(amf.data(), amf.size());
 199     sendPacket(p);
 200 }
 201
 202 bool
 203 RTMP::connect(const URL& url)
 204 {
 205     log_debug("Connecting to %s", url.str());
 206
 207     const std::string& hostname = url.hostname();
 208     const std::string& p = url.port();
 209
 210     // Default port.
 211     boost::uint16_t port = 1935;
 212     if (!p.empty()) {
 213         try {
 214             port = boost::lexical_cast<boost::uint16_t>(p);
 215         }
 216         catch (const boost::bad_lexical_cast&) {}
 217     }
 218
 219     // Basic connection attempt.
 220     if (!_socket.connect(hostname, port)) {
 221         log_error("Initial connection failed");
 222         return false;
 223     }
 224
 225     _handShaker.reset(new HandShaker(_socket));
 226
 227     // Start handshake attempt immediately.
 228     _handShaker->call();
 229
 230     return true;
 231 }
 232
 233 void
 234 RTMP::update()
 235 {
 236     if (!connected()) {
 237         _handShaker->call();
 238         if (_handShaker->error()) {
 239             _error = true;
 240         }
 241         if (!_handShaker->success()) return;
 242         _connected = true;
 243     }
 244
 245     const size_t reads = 10;
 246
 247     for (size_t i = 0; i < reads; ++i) {
 248
 249         /// No need to continue reading (though it should do no harm).
 250         if (error()) return;
 251
 252         RTMPPacket p;
 253
 254         // If we haven't finished reading a packet, retrieve it; otherwise
 255         // use an empty one.
 256         if (_incompletePacket.get()) {
 257             log_debug("Doing incomplete packet");
 258             p = *_incompletePacket;
 259             _incompletePacket.reset();
 260         }
 261         else {
 262             if (!readPacketHeader(p)) continue;
 263         }
 264
 265         // Get the payload if possible.
 266         if (hasPayload(p) && !readPacketPayload(p)) {
 267             // If the payload is not completely readable, store it and
 268             // continue.
 269             _incompletePacket.reset(new RTMPPacket(p));
 270             continue;
 271         }
 272
 273         // Store a copy of the packet for later additions and as a reference for
 274         // future sends.
 275         RTMPPacket& stored = storePacket(CHANNELS_IN, p.header.channel, p);
 276
 277         // If the packet is complete, the stored packet no longer needs to
 278         // keep the data alive.
 279         if (isReady(p)) {
 280             clearPayload(stored);
 281             handlePacket(p);
 282             return;
 283         }
 284     }
 285 }
 286
 287 void
 288 RTMP::handlePacket(const RTMPPacket& packet)
 289 {
 290     const PacketType t = packet.header.packetType;
 291
 292     log_debug("Received %s", t);
 293
 294     switch (t) {
 295
 296         case PACKET_TYPE_CHUNK_SIZE:
 297             handleChangeChunkSize(*this, packet);
 298             break;
 299
 300         case PACKET_TYPE_BYTES_READ:
 301             break;
 302
 303         case PACKET_TYPE_CONTROL:
 304             handleControl(*this, packet);
 305             break;
 306
 307         case PACKET_TYPE_SERVERBW:
 308             handleServerBW(*this, packet);
 309             break;
 310
 311         case PACKET_TYPE_CLIENTBW:
 312             handleClientBW(*this, packet);
 313             break;
 314
 315         case PACKET_TYPE_AUDIO:
 316             if (!m_mediaChannel) m_mediaChannel = packet.header.channel;
 317             break;
 318
 319         case PACKET_TYPE_VIDEO:
 320             if (!m_mediaChannel) m_mediaChannel = packet.header.channel;
 321             break;
 322
 323         case PACKET_TYPE_FLEX_STREAM_SEND:
 324             LOG_ONCE(log_unimpl("unsupported packet %s received"));
 325             break;
 326
 327         case PACKET_TYPE_FLEX_SHARED_OBJECT:
 328             LOG_ONCE(log_unimpl("unsupported packet %s received"));
 329             break;
 330
 331         case PACKET_TYPE_FLEX_MESSAGE:
 332         {
 333             LOG_ONCE(log_unimpl("partially supported packet %s received"));
 334             _messageQueue.push_back(packet.buffer);
 335             break;
 336         }
 337
 338         case PACKET_TYPE_METADATA:
 339             handleMetadata(*this, payloadData(packet), payloadSize(packet));
 340             break;
 341
 342         case PACKET_TYPE_SHARED_OBJECT:
 343             LOG_ONCE(log_unimpl("packet %s received"));
 344             break;
 345
 346         case PACKET_TYPE_INVOKE:
 347             _messageQueue.push_back(packet.buffer);
 348             break;
 349
 350         case PACKET_TYPE_FLV:
 351             _flvQueue.push_back(packet.buffer);
 352             break;
 353
 354         default:
 355             log_error("Unknown packet %s received", t);
 356
 357     }
 358
 359 }
 360
 361 int
 362 RTMP::readSocket(boost::uint8_t* buffer, int n)
 363 {
 364
 365     assert(n >= 0);
 366
 367     const std::streamsize bytesRead = _socket.read(buffer, n);
 368
 369     if (_socket.bad() || _socket.eof() || !_socket.connected()) {
 370         _error = true;
 371         return 0;
 372     }
 373
 374     if (!bytesRead) return 0;
 375
 376     _bytesIn += bytesRead;
 377
 378     // Report bytes recieved every time we reach half the bandwidth.
 379     // Doesn't seem very likely to be the way the pp does it.
 380     if (_bytesIn > _bytesInSent + _bandwidth / 2) {
 381         sendBytesReceived(this);
 382         log_debug("Sent bytes received");
 383     }
 384
 385     buffer += bytesRead;
 386     return bytesRead;
 387 }
 388
 389 void
 390 RTMP::play(const SimpleBuffer& buf, int streamID)
 391 {
 392     RTMPPacket packet(buf.size());
 393
 394     packet.header.channel = CHANNEL_VIDEO;
 395     packet.header.packetType = PACKET_TYPE_INVOKE;
 396
 397     packet.header._streamID = streamID;
 398
 399     packet.buffer->append(buf.data(), buf.size());
 400     sendPacket(packet);
 401 }
 402
 403 /// Send the server bandwidth.
 404 //
 405 /// Why would we want to send this?
 406 bool
 407 sendServerBW(RTMP& r)
 408 {
 409     RTMPPacket packet(4);
 410
 411     packet.header.channel = CHANNEL_CONTROL1;
 412     packet.header.packetType = PACKET_TYPE_SERVERBW;
 413
 414     SimpleBuffer& buf = *packet.buffer;
 415
 416     buf.appendNetworkLong(r.serverBandwidth());
 417     return r.sendPacket(packet);
 418 }
 419
 420
 421 /// Fills a pre-existent RTMPPacket with information.
 422 //
 423 /// This is either read entirely from incoming data, or copied from a
 424 /// previous packet in the same channel. This happens when the header type
 425 /// is less than RTMP_PACKET_SIZE_LARGE.
 426 //
 427 /// It seems as if new packets can add to the data of old ones if they have
 428 /// a minimal, small header.
 429 bool
 430 RTMP::readPacketHeader(RTMPPacket& packet)
 431 {
 432
 433     RTMPHeader& hr = packet.header;
 434
 435     boost::uint8_t hbuf[RTMPHeader::headerSize] = { 0 };
 436     boost::uint8_t* header = hbuf;
 437
 438     // The first read may fail, but otherwise we expect a complete header.
 439     if (readSocket(hbuf, 1) == 0) {
 440         return false;
 441     }
 442
 443     //log_debug("Packet is %s", boost::io::group(std::hex, (unsigned)hbuf[0]));
 444
 445     const int htype = ((hbuf[0] & 0xc0) >> 6);
 446     //log_debug("Thingy whatsit (packet size type): %s", htype);
 447
 448     const int channel = (hbuf[0] & 0x3f);
 449     //log_debug("Channel: %s", channel);
 450
 451     hr.headerType = static_cast<PacketSize>(htype);
 452     hr.channel = channel;
 453     ++header;
 454
 455     if (hr.channel == 0) {
 456         if (readSocket(&hbuf[1], 1) != 1) {
 457           log_error("failed to read RTMP packet header 2nd byte");
 458           return false;
 459         }
 460         hr.channel = hbuf[1] + 64;
 461         ++header;
 462     }
 463     else if (hr.channel == 1) {
 464         if (readSocket(&hbuf[1], 2) != 2) {
 465             log_error("Failed to read RTMP packet header 3nd byte");
 466              return false;
 467         }
 468
 469         const boost::uint32_t tmp = (hbuf[2] << 8) + hbuf[1];
 470         hr.channel = tmp + 64;
 471         log_debug( "%s, channel: %0x", __FUNCTION__, hr.channel);
 472         header += 2;
 473     }
 474
 475     // This is the size in bytes of the packet header according to the
 476     // type.
 477     int nSize = packetSize[htype];
 478
 479     /// If we didn't receive a large header, the timestamp is relative
 480     if (htype != RTMP_PACKET_SIZE_LARGE) {
 481
 482         if (!hasPacket(CHANNELS_IN, hr.channel)) {
 483             log_error("Incomplete packet received on channel %s", channel);
 484             return false;
 485         }
 486
 487         // For all other header types, copy values from the last message of
 488         // this channel. This includes any payload data from incomplete
 489         // messages.
 490         packet = getPacket(CHANNELS_IN, hr.channel);
 491     }
 492
 493     --nSize;
 494
 495     if (nSize > 0 && readSocket(header, nSize) != nSize) {
 496         log_error( "Failed to read RTMP packet header. type: %s",
 497                 static_cast<unsigned>(hbuf[0]));
 498         return false;
 499     }
 500
 501     // nSize is predicted size - 1. Add what we've read already.
 502     int hSize = nSize + (header - hbuf);
 503
 504     if (nSize >= 3) {
 505
 506         const boost::uint32_t timestamp = decodeInt24(header);
 507
 508         // Make our packet timestamp absolute. If the value is 0xffffff,
 509         // the absolute value comes later.
 510         if (timestamp != 0xffffff) {
 511             if (htype != RTMP_PACKET_SIZE_LARGE) {
 512                 packet.header._timestamp += timestamp;
 513             }
 514             else {
 515                 packet.header._timestamp = timestamp;
 516             }
 517         }
 518
 519         // Have at least a different size payload from the last packet.
 520         if (nSize >= 6) {
 521
 522             // We do this in case there was an incomplete packet in the
 523             // channel already.
 524             clearPayload(packet);
 525             hr.dataSize = decodeInt24(header + 3);
 526
 527             // More than six: read packet type
 528             if (nSize > 6) {
 529                 hr.packetType = static_cast<PacketType>(header[6]);
 530
 531                 // Large packets have a streamID.
 532                 if (nSize == 11) {
 533                     hr._streamID = decodeInt32LE(header + 7);
 534                 }
 535             }
 536         }
 537     }
 538
 539     if (hr._timestamp == 0xffffff) {
 540       if (readSocket(header+nSize, 4) != 4) {
 541               log_error( "%s, failed to read extended timestamp",
 542               __FUNCTION__);
 543               return false;
 544             }
 545           hr._timestamp = amf::readNetworkLong(header+nSize);
 546           hSize += 4;
 547
 548     }
 549
 550     const size_t bufSize = hr.dataSize + RTMPHeader::headerSize;
 551
 552     // If the packet does not have a payload, it was a complete packet stored in
 553     // the channel for reference. This is the only case when a packet should
 554     // exist but have no payload. We re-allocate in this case.
 555     if (!hasPayload(packet)) {
 556         packet.buffer.reset(new SimpleBuffer(bufSize));
 557
 558         // Why do this again? In case it was copied from the old packet?
 559         hr.headerType = static_cast<PacketSize>(htype);
 560     }
 561
 562     // Resize anyway. If it's different from what it was before, we should
 563     // already have cleared it.
 564     packet.buffer->resize(bufSize);
 565     return true;
 566 }
 567
 568 bool
 569 RTMP::readPacketPayload(RTMPPacket& packet)
 570 {
 571     RTMPHeader& hr = packet.header;
 572
 573     const size_t bytesRead = packet.bytesRead;
 574
 575     const int nToRead = hr.dataSize - bytesRead;
 576
 577     const int nChunk = std::min<int>(nToRead, _inChunkSize);
 578     assert(nChunk >= 0);
 579
 580     // This is fine. We'll keep trying to read this payload until there
 581     // is enough data.
 582     if (readSocket(payloadData(packet) + bytesRead, nChunk) != nChunk) {
 583         return false;
 584     }
 585
 586     packet.bytesRead += nChunk;
 587
 588     return true;
 589 }
 590
 591 bool
 592 RTMP::handShake()
 593 {
 594
 595     /// It is a size type, but our socket functions return int.
 596     const int sigSize = 1536;
 597
 598     boost::uint8_t clientbuf[sigSize + 1];
 599     boost::uint8_t* ourSig = clientbuf + 1;
 600
 601     // Not encrypted
 602     clientbuf[0] = 0x03;
 603
 604     // TODO: do this properly.
 605     boost::uint32_t uptime = htonl(getUptime());
 606     std::memcpy(ourSig, &uptime, 4);
 607
 608     std::fill_n(ourSig + 4, 4, 0);
 609
 610     // Generate 1536 random bytes.
 611     std::generate(ourSig + 8, ourSig + sigSize, RandomByte());
 612
 613     // Send it to server.
 614     if (_socket.write(clientbuf, sigSize + 1) != sigSize + 1) {
 615         return false;
 616     }
 617
 618     // Expect the same byte as we sent.
 619     boost::uint8_t type;
 620     if (readSocket(&type, 1) != 1) {
 621         return false;
 622     }
 623
 624     log_debug( "%s: Type Answer   : %02X", __FUNCTION__, (int)type);
 625
 626     if (type != clientbuf[0]) {
 627         log_error( "%s: Type mismatch: client sent %d, server answered %d",
 628             __FUNCTION__, clientbuf[0], type);
 629     }
 630
 631     boost::uint8_t serverSig[sigSize];
 632
 633     // Read from server.
 634     if (readSocket(serverSig, sigSize) != sigSize) {
 635         return false;
 636     }
 637
 638     // decode server response
 639     boost::uint32_t suptime;
 640
 641     memcpy(&suptime, serverSig, 4);
 642     suptime = ntohl(suptime);
 643
 644     log_debug("Server Uptime : %d", suptime);
 645     log_debug("FMS Version   : %d.%d.%d.%d",
 646             +serverSig[4], +serverSig[5], +serverSig[6], +serverSig[7]);
 647
 648     // Send what we received from server.
 649     if (_socket.write(serverSig, sigSize) != sigSize) {
 650         return false;
 651     }
 652
 653     // Expect it back again.
 654     if (readSocket(serverSig, sigSize) != sigSize) {
 655         return false;
 656     }
 657
 658     const bool match = std::equal(serverSig, serverSig + arraySize(serverSig),
 659                                   ourSig);
 660
 661     if (!match) {
 662         log_error( "Signatures do not match during handshake!");
 663     }
 664     return true;
 665 }
 666
 667
 668 bool
 669 RTMP::sendPacket(RTMPPacket& packet)
 670 {
 671     // Set the data size of the packet to send.
 672     RTMPHeader& hr = packet.header;
 673
 674     hr.dataSize = payloadSize(packet);
 675
 676     // This is the timestamp for our message.
 677     const boost::uint32_t uptime = getUptime();
 678
 679     // Look at the previous packet on the channel.
 680     bool prev = hasPacket(CHANNELS_OUT, hr.channel);
 681
 682     // The packet shall be large if it contains an absolute timestamp.
 683     //      * This is necessary if there is no previous packet, or if the
 684     //        timestamp is smaller than the last packet.
 685     // Else it shall be medium if data size and packet type are the same
 686     // It shall be small if ...
 687     // It shall be minimal if it is exactly the same as its predecessor.
 688
 689     // All packets should start off as large. They will stay large if there
 690     // is no previous packet.
 691     assert(hr.headerType == RTMP_PACKET_SIZE_LARGE);
 692
 693     if (!prev) {
 694         hr._timestamp = uptime;
 695     }
 696     else {
 697
 698         const RTMPPacket& prevPacket = getPacket(CHANNELS_OUT, hr.channel);
 699         const RTMPHeader& oldh = prevPacket.header;
 700         const boost::uint32_t prevTimestamp = oldh._timestamp;
 701
 702         // If this timestamp is later than the other and the difference fits
 703         // in 3 bytes, encode a relative one.
 704         if (uptime >= oldh._timestamp && uptime - prevTimestamp < 0xffffff) {
 705             //log_debug("Shrinking to medium");
 706             hr.headerType = RTMP_PACKET_SIZE_MEDIUM;
 707             hr._timestamp = uptime - prevTimestamp;
 708
 709             // It can be still smaller if the data size is the same.
 710             if (oldh.dataSize == hr.dataSize &&
 711                     oldh.packetType == hr.packetType) {
 712                 //log_debug("Shrinking to small");
 713                 hr.headerType = RTMP_PACKET_SIZE_SMALL;
 714                 // If there is no timestamp difference, the minimum size
 715                 // is possible.
 716                 if (hr._timestamp == 0) {
 717                     //log_debug("Shrinking to minimum");
 718                     hr.headerType = RTMP_PACKET_SIZE_MINIMUM;
 719                 }
 720             }
 721         }
 722         else {
 723             // Otherwise we need an absolute one, so a large header.
 724             hr.headerType = RTMP_PACKET_SIZE_LARGE;
 725             hr._timestamp = uptime;
 726         }
 727     }
 728
 729     assert (hr.headerType < 4);
 730
 731     int nSize = packetSize[hr.headerType];
 732
 733     int hSize = nSize;
 734     boost::uint8_t* header;
 735     boost::uint8_t* hptr;
 736     boost::uint8_t* hend;
 737     boost::uint8_t c;
 738
 739     // If there is a payload, the same buffer is used to write the header.
 740     // Otherwise a separate buffer is used. But as we write them separately
 741     // anyway, why do we do that?
 742
 743     // Work out where the beginning of the header is.
 744     header = payloadData(packet) - nSize;
 745     hend = payloadData(packet);
 746
 747     // The header size includes only a single channel/type. If we need more,
 748     // they have to be added on.
 749     const int channelSize = hr.channel > 319 ? 3 : hr.channel > 63 ? 1 : 0;
 750     header -= channelSize;
 751     hSize += channelSize;
 752
 753     /// Add space for absolute timestamp if necessary.
 754     if (hr.headerType == RTMP_PACKET_SIZE_LARGE && hr._timestamp >= 0xffffff) {
 755         header -= 4;
 756         hSize += 4;
 757     }
 758
 759     hptr = header;
 760     c = hr.headerType << 6;
 761     switch (channelSize) {
 762         case 0:
 763             c |= hr.channel;
 764             break;
 765         case 1:
 766             break;
 767         case 2:
 768             c |= 1;
 769             break;
 770     }
 771     *hptr++ = c;
 772
 773     if (channelSize) {
 774         const int tmp = hr.channel - 64;
 775         *hptr++ = tmp & 0xff;
 776         if (channelSize == 2) *hptr++ = tmp >> 8;
 777     }
 778
 779     if (hr.headerType == RTMP_PACKET_SIZE_LARGE && hr._timestamp >= 0xffffff) {
 780         // Signify that the extended timestamp field is present.
 781         const boost::uint32_t t = 0xffffff;
 782         hptr = encodeInt24(hptr, hend, t);
 783     }
 784     else if (hr.headerType != RTMP_PACKET_SIZE_MINIMUM) {
 785         // Write absolute or relative timestamp. Only minimal packets have
 786         // no timestamp.
 787         hptr = encodeInt24(hptr, hend, hr._timestamp);
 788     }
 789
 790     /// Encode dataSize and packet type for medium packets.
 791     if (nSize > 4) {
 792         hptr = encodeInt24(hptr, hend, hr.dataSize);
 793         *hptr++ = hr.packetType;
 794     }
 795
 796     /// Encode streamID for large packets.
 797     if (hr.headerType == RTMP_PACKET_SIZE_LARGE) {
 798         hptr += encodeInt32LE(hptr, hr._streamID);
 799     }
 800
 801     // Encode extended absolute timestamp if needed.
 802     if (hr.headerType == RTMP_PACKET_SIZE_LARGE && hr._timestamp >= 0xffffff) {
 803         hptr += encodeInt32LE(hptr, hr._timestamp);
 804     }
 805
 806     nSize = hr.dataSize;
 807     boost::uint8_t *buffer = payloadData(packet);
 808     int nChunkSize = _outChunkSize;
 809
 810     std::string hx = hexify(header, payloadEnd(packet) - header, false);
 811
 812     while (nSize + hSize) {
 813
 814         if (nSize < nChunkSize) nChunkSize = nSize;
 815
 816         // First write header.
 817         if (header) {
 818             const int chunk = nChunkSize + hSize;
 819             if (_socket.write(header, chunk) != chunk) {
 820                 return false;
 821             }
 822             header = NULL;
 823             hSize = 0;
 824         }
 825
 826         else {
 827             // Then write data.
 828             if (_socket.write(buffer, nChunkSize) != nChunkSize) {
 829                 return false;
 830           }
 831
 832         }
 833
 834         nSize -= nChunkSize;
 835         buffer += nChunkSize;
 836
 837         if (nSize > 0) {
 838             header = buffer - 1;
 839             hSize = 1;
 840             if (channelSize) {
 841                 header -= channelSize;
 842                 hSize += channelSize;
 843             }
 844
 845             *header = (0xc0 | c);
 846             if (channelSize) {
 847                 int tmp = hr.channel - 64;
 848                 header[1] = tmp & 0xff;
 849                 if (channelSize == 2) header[2] = tmp >> 8;
 850             }
 851         }
 852     }
 853
 854     /* we invoked a remote method */
 855     if (hr.packetType == PACKET_TYPE_INVOKE) {
 856         assert(payloadData(packet)[0] == amf::STRING_AMF0);
 857         const boost::uint8_t* pos = payloadData(packet) + 1;
 858         const boost::uint8_t* end = payloadEnd(packet);
 859         const std::string& s = amf::readString(pos, end);
 860         log_debug( "Calling remote method %s", s);
 861     }
 862
 863     RTMPPacket& storedpacket = storePacket(CHANNELS_OUT, hr.channel, packet);
 864
 865     // Make it absolute for the next delta.
 866     storedpacket.header._timestamp = uptime;
 867
 868     return true;
 869 }
 870
 871 void
 872 RTMP::close()
 873 {
 874     _socket.close();
 875     _inChannels.clear();
 876     _outChannels.clear();
 877     _inChunkSize = RTMP_DEFAULT_CHUNKSIZE;
 878     _outChunkSize = RTMP_DEFAULT_CHUNKSIZE;
 879     _bytesIn = 0;
 880     _bytesInSent = 0;
 881     _bandwidth = 2500000;
 882     m_nClientBW2 = 2;
 883     _serverBandwidth = 2500000;
 884 }
 885
 886
 887 /////////////////////////////////////
 888 /// HandShaker implementation
 889 /////////////////////////////////////
 890
 891 HandShaker::HandShaker(Socket& s)
 892     :
 893     _socket(s),
 894     _sendBuf(sigSize + 1),
 895     _recvBuf(sigSize + 1),
 896     _error(false),
 897     _complete(false),
 898     _stage(0)
 899 {
 900     // Not encrypted
 901     _sendBuf[0] = 0x03;
 902
 903     // TODO: do this properly.
 904     boost::uint32_t uptime = htonl(getUptime());
 905
 906     boost::uint8_t* ourSig = &_sendBuf.front() + 1;
 907     std::memcpy(ourSig, &uptime, 4);
 908     std::fill_n(ourSig + 4, 4, 0);
 909
 910     // Generate 1536 random bytes.
 911     std::generate(ourSig + 8, ourSig + sigSize, RandomByte());
 912
 913 }
 914
 915
 916 /// Calls the next stage in the handshake process.
 917 void
 918 HandShaker::call()
 919 {
 920     if (error() || !_socket.connected()) return;
 921
 922     switch (_stage) {
 923         case 0:
 924             if (!stage0()) return;
 925             _stage = 1;
 926         case 1:
 927             if (!stage1()) return;
 928             _stage = 2;
 929         case 2:
 930             if (!stage2()) return;
 931             _stage = 3;
 932         case 3:
 933             if (!stage3()) return;
 934             log_debug("Handshake completed");
 935             _complete = true;
 936     }
 937 }
 938
 939 bool
 940 HandShaker::stage0()
 941 {
 942     std::streamsize sent = _socket.write(&_sendBuf.front(), sigSize + 1);
 943
 944     // This should probably not happen, but we can try again. An error will
 945     // be signalled later if the socket is no longer usable.
 946     if (!sent) {
 947         log_error("Stage 1 socket not ready. This should not happen.");
 948         return false;
 949     }
 950
 951     /// If we sent the wrong amount of data, we can't recover.
 952     if (sent != sigSize + 1) {
 953         log_error("Could not send stage 1 data");
 954         _error = true;
 955         return false;
 956     }
 957     return true;
 958 }
 959
 960 bool
 961 HandShaker::stage1()
 962 {
 963
 964     std::streamsize read = _socket.read(&_recvBuf.front(), sigSize + 1);
 965
 966     if (!read) {
 967         // If we receive nothing, wait until the next try.
 968         return false;
 969     }
 970
 971     // The read should never return anything but 0 or what we asked for.
 972     assert (read == sigSize + 1);
 973
 974     if (_recvBuf[0] != _sendBuf[0]) {
 975         log_error( "Type mismatch: client sent %d, server answered %d",
 976                 _recvBuf[0], _sendBuf[0]);
 977     }
 978
 979     const boost::uint8_t* serverSig = &_recvBuf.front() + 1;
 980
 981     // decode server response
 982     boost::uint32_t suptime;
 983     std::memcpy(&suptime, serverSig, 4);
 984     suptime = ntohl(suptime);
 985
 986     log_debug("Server Uptime : %d", suptime);
 987     log_debug("FMS Version   : %d.%d.%d.%d",
 988             +serverSig[4], +serverSig[5], +serverSig[6], +serverSig[7]);
 989
 990     return true;
 991 }
 992
 993 bool
 994 HandShaker::stage2()
 995 {
 996
 997     std::streamsize sent = _socket.write(&_recvBuf.front() + 1, sigSize);
 998
 999     // This should probably not happen.
1000     if (!sent) return false;
1001
1002     if (sent != sigSize) {
1003         log_error("Could not send complete signature.");
1004         _error = true;
1005         return false;
1006     }
1007
1008     return true;
1009 }
1010
1011 bool
1012 HandShaker::stage3()
1013 {
1014
1015     // Expect it back again.
1016     std::streamsize got = _socket.read(&_recvBuf.front(), sigSize);
1017
1018     if (!got) return false;
1019
1020     assert (got == sigSize);
1021
1022     const boost::uint8_t* serverSig = &_recvBuf.front();
1023     const boost::uint8_t* ourSig = &_sendBuf.front() + 1;
1024
1025     const bool match = std::equal(serverSig, serverSig + sigSize, ourSig);
1026
1027     // Should we set an error here?
1028     if (!match) {
1029         log_error( "Signatures do not match during handshake!");
1030     }
1031     return true;
1032 }
1033
1034 /// The type of Ping packet is 0x4 and contains two mandatory parameters
1035 /// and two optional parameters. The first parameter is
1036 /// the type of Ping and in short integer. The second parameter is the
1037 /// target of the ping. As Ping is always sent in Channel 2
1038 /// (control channel) and the target object in RTMP header is always 0 whicj
1039 /// means the Connection object, it's necessary to put an extra parameter
1040 /// to indicate the exact target object the Ping is sent to. The second
1041 /// parameter takes this responsibility. The value has the same meaning
1042 /// as the target object field in RTMP header. (The second value could also
1043 /// be used as other purposes, like RTT Ping/Pong. It is used as the
1044 /// timestamp.) The third and fourth parameters are optional and could be
1045 /// looked upon as the parameter of the Ping packet.
1046 bool
1047 sendCtrl(RTMP& r, ControlType t, unsigned int nObject, unsigned int nTime)
1048 {
1049     log_debug( "Sending control type %s %s", +t, t);
1050
1051     RTMPPacket packet(256);
1052
1053     packet.header.channel = CHANNEL_CONTROL1;
1054     packet.header.headerType = RTMP_PACKET_SIZE_LARGE;
1055     packet.header.packetType = PACKET_TYPE_CONTROL;
1056
1057     // type 3 is the buffer time and requires all 3 parameters.
1058     // all in all 10 bytes.
1059     int nSize = (t == CONTROL_BUFFER_TIME ? 10 : 6);
1060     if (t == CONTROL_RESPOND_VERIFY) nSize = 44;
1061
1062     SimpleBuffer& buf = *packet.buffer;
1063
1064     buf.appendNetworkShort(t);
1065
1066     if (t == CONTROL_RESPOND_VERIFY) { }
1067     else {
1068         if (nSize > 2) buf.appendNetworkLong(nObject);
1069         if (nSize > 6) buf.appendNetworkLong(nTime);
1070     }
1071     return r.sendPacket(packet);
1072 }
1073
1074 namespace {
1075
1076
1077 bool
1078 sendBytesReceived(RTMP* r)
1079 {
1080     RTMPPacket packet(4);
1081
1082     packet.header.channel = CHANNEL_CONTROL1;
1083     packet.header.packetType = PACKET_TYPE_BYTES_READ;
1084
1085     SimpleBuffer& buf = *packet.buffer;
1086
1087     buf.appendNetworkLong(r->_bytesIn);
1088     r->_bytesInSent = r->_bytesIn;
1089
1090     return r->sendPacket(packet);
1091 }
1092
1093
1094 void
1095 handleMetadata(RTMP& /*r*/, const boost::uint8_t* /* payload*/,
1096         unsigned int /*len*/)
1097 {
1098     return;
1099 }
1100
1101 void
1102 handleChangeChunkSize(RTMP& r, const RTMPPacket& packet)
1103 {
1104     if (payloadSize(packet) >= 4) {
1105         r._inChunkSize = amf::readNetworkLong(payloadData(packet));
1106         log_debug( "Changed chunk size to %d", r._inChunkSize);
1107     }
1108 }
1109
1110 void
1111 handleControl(RTMP& r, const RTMPPacket& packet)
1112 {
1113
1114     const size_t size = payloadSize(packet);
1115
1116     if (size < 2) {
1117         log_error("Control packet too short");
1118         return;
1119     }
1120
1121     const ControlType t =
1122         static_cast<ControlType>(amf::readNetworkShort(payloadData(packet)));
1123
1124     if (size < 6) {
1125         log_error("Control packet (%s) data too short", t);
1126         return;
1127     }
1128
1129     const int arg = amf::readNetworkLong(payloadData(packet) + 2);
1130     log_debug( "Received control packet %s with argument %s", t, arg);
1131
1132     switch (t)
1133     {
1134
1135         case CONTROL_CLEAR_STREAM:
1136             // TODO: handle this.
1137             break;
1138
1139         case CONTROL_CLEAR_BUFFER:
1140             // TODO: handle this.
1141             break;
1142
1143         case CONTROL_STREAM_DRY:
1144             break;
1145
1146         case CONTROL_RESET_STREAM:
1147             log_debug("Stream is recorded: %s", arg);
1148             break;
1149
1150         case CONTROL_PING:
1151             sendCtrl(r, CONTROL_PONG, arg, 0);
1152             break;
1153
1154         case CONTROL_BUFFER_EMPTY:
1155             // TODO: handle.
1156             break;
1157
1158         case CONTROL_BUFFER_READY:
1159             // TODO: handle
1160             break;
1161
1162         default:
1163             log_error("Received unknown or unhandled control %s", t);
1164             break;
1165     }
1166
1167 }
1168
1169 void
1170 handleServerBW(RTMP& r, const RTMPPacket& packet)
1171 {
1172     const boost::uint32_t bw = amf::readNetworkLong(payloadData(packet));
1173     log_debug( "Server bandwidth is %s", bw);
1174     r.setServerBandwidth(bw);
1175 }
1176
1177 void
1178 handleClientBW(RTMP& r, const RTMPPacket& packet)
1179 {
1180     const boost::uint32_t bw = amf::readNetworkLong(payloadData(packet));
1181
1182     r.setBandwidth(bw);
1183
1184     if (payloadSize(packet) > 4) r.m_nClientBW2 = payloadData(packet)[4];
1185     else r.m_nClientBW2 = -1;
1186
1187     log_debug( "Client bandwidth is %d %d", r.bandwidth(), +r.m_nClientBW2);
1188 }
1189
1190
1191
1192 boost::int32_t
1193 decodeInt32LE(const boost::uint8_t* c)
1194 {
1195     return (c[3] << 24) | (c[2] << 16) | (c[1] << 8) | c[0];
1196 }
1197
1198 int
1199 encodeInt32LE(boost::uint8_t *output, int nVal)
1200 {
1201     output[0] = nVal;
1202     nVal >>= 8;
1203     output[1] = nVal;
1204     nVal >>= 8;
1205     output[2] = nVal;
1206     nVal >>= 8;
1207     output[3] = nVal;
1208     return 4;
1209 }
1210
1211 void
1212 setupInvokePacket(RTMPPacket& packet)
1213 {
1214     RTMPHeader& hr = packet.header;
1215     // Control channel
1216     hr.channel = CHANNEL_CONTROL2;
1217     // Invoke
1218     hr.packetType = PACKET_TYPE_INVOKE;
1219 }
1220
1221 unsigned int
1222 decodeInt24(const boost::uint8_t *c)
1223 {
1224     unsigned int val;
1225     val = (c[0] << 16) | (c[1] << 8) | c[2];
1226     return val;
1227 }
1228
1229 boost::uint8_t*
1230 encodeInt16(boost::uint8_t *output, boost::uint8_t *outend, short nVal)
1231 {
1232     if (output+2 > outend) return NULL;
1233
1234     output[1] = nVal & 0xff;
1235     output[0] = nVal >> 8;
1236     return output + 2;
1237 }
1238
1239 boost::uint8_t*
1240 encodeInt24(boost::uint8_t *output, boost::uint8_t *outend, int nVal)
1241 {
1242     if (output + 3 > outend) return NULL;
1243
1244     output[2] = nVal & 0xff;
1245     output[1] = nVal >> 8;
1246     output[0] = nVal >> 16;
1247     return output+3;
1248 }
1249
1250 boost::uint8_t*
1251 encodeInt32(boost::uint8_t *output, boost::uint8_t *outend, int nVal)
1252 {
1253     if (output+4 > outend) return NULL;
1254
1255     output[3] = nVal & 0xff;
1256     output[2] = nVal >> 8;
1257     output[1] = nVal >> 16;
1258     output[0] = nVal >> 24;
1259     return output + 4;
1260 }
1261
1262 boost::uint32_t
1263 getUptime()
1264 {
1265 #if !defined(_WIN32) && !defined(__amigaos4__)
1266     struct tms t;
1267     return times(&t) * 1000 / sysconf(_SC_CLK_TCK);
1268 #elif defined(__amigaos4__)
1269     struct tms t;
1270     return times(&t) * 1000 / 50;
1271 #else
1272     return std::clock() * 100 / CLOCKS_PER_SEC;
1273 #endif
1274 }
1275
1276 } // anonymous namespace
1277
1278 std::ostream&
1279 operator<<(std::ostream& o, PacketType p)
1280 {
1281     switch(p) {
1282         case PACKET_TYPE_CHUNK_SIZE:
1283             return o << "<chunk size packet>";
1284         case PACKET_TYPE_BYTES_READ:
1285             return o << "<bytes read packet>";
1286         case PACKET_TYPE_CONTROL:
1287             return o << "<control packet>";
1288         case PACKET_TYPE_SERVERBW:
1289             return o << "<server bw packet>";
1290         case PACKET_TYPE_CLIENTBW:
1291             return o << "<client bw packet>";
1292         case PACKET_TYPE_AUDIO:
1293             return o << "<audio packet>";
1294         case PACKET_TYPE_VIDEO:
1295             return o << "<video packet>";
1296         case PACKET_TYPE_FLEX_STREAM_SEND:
1297             return o << "<flex stream send packet>";
1298         case PACKET_TYPE_FLEX_SHARED_OBJECT:
1299             return o << "<flex sharedobject packet>";
1300         case PACKET_TYPE_FLEX_MESSAGE:
1301             return o << "<flex message packet>";
1302         case PACKET_TYPE_METADATA:
1303             return o << "<metadata packet>";
1304         case PACKET_TYPE_SHARED_OBJECT:
1305             return o << "<sharedobject packet>";
1306         case PACKET_TYPE_INVOKE:
1307             return o << "<invoke packet>";
1308         case PACKET_TYPE_FLV:
1309             return o << "<flv packet>";
1310         default:
1311             return o << "<unknown packet type " << +p << ">";
1312     }
1313 }
1314
1315 std::ostream&
1316 operator<<(std::ostream& o, ControlType t)
1317 {
1318     switch (t) {
1319
1320         case CONTROL_CLEAR_STREAM:
1321             return o << "<clear stream>";
1322         case CONTROL_CLEAR_BUFFER:
1323             return o << "<clear buffer>";
1324         case CONTROL_STREAM_DRY:
1325             return o << "<stream dry>";
1326         case CONTROL_BUFFER_TIME:
1327             return o << "<buffer time>";
1328         case CONTROL_RESET_STREAM:
1329             return o << "<reset stream>";
1330         case CONTROL_PING:
1331             return o << "<ping>";
1332         case CONTROL_PONG:
1333             return o << "<pong>";
1334         case CONTROL_REQUEST_VERIFY:
1335             return o << "<verify request>";
1336         case CONTROL_RESPOND_VERIFY:
1337             return o << "<verify response>";
1338         case CONTROL_BUFFER_EMPTY:
1339             return o << "<buffer empty>";
1340         case CONTROL_BUFFER_READY:
1341             return o << "<buffer ready>";
1342         default:
1343             return o << "<unknown control " << +t << ">";
1344     }
1345 }
1346
1347 } // namespace rtmp
1348 } // namespace gnash