libbase/RTMP.cpp

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