1/* 2 * RTMP network protocol 3 * Copyright (c) 2009 Kostya Shishkov 4 * 5 * This file is part of Libav. 6 * 7 * Libav is free software; you can redistribute it and/or 8 * modify it under the terms of the GNU Lesser General Public 9 * License as published by the Free Software Foundation; either 10 * version 2.1 of the License, or (at your option) any later version. 11 * 12 * Libav is distributed in the hope that it will be useful, 13 * but WITHOUT ANY WARRANTY; without even the implied warranty of 14 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU 15 * Lesser General Public License for more details. 16 * 17 * You should have received a copy of the GNU Lesser General Public 18 * License along with Libav; if not, write to the Free Software 19 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA 20 */ 21 22/** 23 * @file 24 * RTMP protocol 25 */ 26 27#include "libavcodec/bytestream.h" 28#include "libavutil/avstring.h" 29#include "libavutil/intfloat.h" 30#include "libavutil/lfg.h" 31#include "libavutil/sha.h" 32#include "avformat.h" 33#include "internal.h" 34 35#include "network.h" 36 37#include "flv.h" 38#include "rtmp.h" 39#include "rtmppkt.h" 40#include "url.h" 41 42//#define DEBUG 43 44/** RTMP protocol handler state */ 45typedef enum { 46 STATE_START, ///< client has not done anything yet 47 STATE_HANDSHAKED, ///< client has performed handshake 48 STATE_RELEASING, ///< client releasing stream before publish it (for output) 49 STATE_FCPUBLISH, ///< client FCPublishing stream (for output) 50 STATE_CONNECTING, ///< client connected to server successfully 51 STATE_READY, ///< client has sent all needed commands and waits for server reply 52 STATE_PLAYING, ///< client has started receiving multimedia data from server 53 STATE_PUBLISHING, ///< client has started sending multimedia data to server (for output) 54 STATE_STOPPED, ///< the broadcast has been stopped 55} ClientState; 56 57/** protocol handler context */ 58typedef struct RTMPContext { 59 URLContext* stream; ///< TCP stream used in interactions with RTMP server 60 RTMPPacket prev_pkt[2][RTMP_CHANNELS]; ///< packet history used when reading and sending packets 61 int chunk_size; ///< size of the chunks RTMP packets are divided into 62 int is_input; ///< input/output flag 63 char playpath[256]; ///< path to filename to play (with possible "mp4:" prefix) 64 char app[128]; ///< application 65 ClientState state; ///< current state 66 int main_channel_id; ///< an additional channel ID which is used for some invocations 67 uint8_t* flv_data; ///< buffer with data for demuxer 68 int flv_size; ///< current buffer size 69 int flv_off; ///< number of bytes read from current buffer 70 RTMPPacket out_pkt; ///< rtmp packet, created from flv a/v or metadata (for output) 71 uint32_t client_report_size; ///< number of bytes after which client should report to server 72 uint32_t bytes_read; ///< number of bytes read from server 73 uint32_t last_bytes_read; ///< number of bytes read last reported to server 74 int skip_bytes; ///< number of bytes to skip from the input FLV stream in the next write call 75 uint8_t flv_header[11]; ///< partial incoming flv packet header 76 int flv_header_bytes; ///< number of initialized bytes in flv_header 77 int nb_invokes; ///< keeps track of invoke messages 78 int create_stream_invoke; ///< invoke id for the create stream command 79} RTMPContext; 80 81#define PLAYER_KEY_OPEN_PART_LEN 30 ///< length of partial key used for first client digest signing 82/** Client key used for digest signing */ 83static const uint8_t rtmp_player_key[] = { 84 'G', 'e', 'n', 'u', 'i', 'n', 'e', ' ', 'A', 'd', 'o', 'b', 'e', ' ', 85 'F', 'l', 'a', 's', 'h', ' ', 'P', 'l', 'a', 'y', 'e', 'r', ' ', '0', '0', '1', 86 87 0xF0, 0xEE, 0xC2, 0x4A, 0x80, 0x68, 0xBE, 0xE8, 0x2E, 0x00, 0xD0, 0xD1, 0x02, 88 0x9E, 0x7E, 0x57, 0x6E, 0xEC, 0x5D, 0x2D, 0x29, 0x80, 0x6F, 0xAB, 0x93, 0xB8, 89 0xE6, 0x36, 0xCF, 0xEB, 0x31, 0xAE 90}; 91 92#define SERVER_KEY_OPEN_PART_LEN 36 ///< length of partial key used for first server digest signing 93/** Key used for RTMP server digest signing */ 94static const uint8_t rtmp_server_key[] = { 95 'G', 'e', 'n', 'u', 'i', 'n', 'e', ' ', 'A', 'd', 'o', 'b', 'e', ' ', 96 'F', 'l', 'a', 's', 'h', ' ', 'M', 'e', 'd', 'i', 'a', ' ', 97 'S', 'e', 'r', 'v', 'e', 'r', ' ', '0', '0', '1', 98 99 0xF0, 0xEE, 0xC2, 0x4A, 0x80, 0x68, 0xBE, 0xE8, 0x2E, 0x00, 0xD0, 0xD1, 0x02, 100 0x9E, 0x7E, 0x57, 0x6E, 0xEC, 0x5D, 0x2D, 0x29, 0x80, 0x6F, 0xAB, 0x93, 0xB8, 101 0xE6, 0x36, 0xCF, 0xEB, 0x31, 0xAE 102}; 103 104/** 105 * Generate 'connect' call and send it to the server. 106 */ 107static void gen_connect(URLContext *s, RTMPContext *rt, const char *proto, 108 const char *host, int port) 109{ 110 RTMPPacket pkt; 111 uint8_t ver[64], *p; 112 char tcurl[512]; 113 114 ff_rtmp_packet_create(&pkt, RTMP_SYSTEM_CHANNEL, RTMP_PT_INVOKE, 0, 4096); 115 p = pkt.data; 116 117 ff_url_join(tcurl, sizeof(tcurl), proto, NULL, host, port, "/%s", rt->app); 118 ff_amf_write_string(&p, "connect"); 119 ff_amf_write_number(&p, ++rt->nb_invokes); 120 ff_amf_write_object_start(&p); 121 ff_amf_write_field_name(&p, "app"); 122 ff_amf_write_string(&p, rt->app); 123 124 if (rt->is_input) { 125 snprintf(ver, sizeof(ver), "%s %d,%d,%d,%d", RTMP_CLIENT_PLATFORM, RTMP_CLIENT_VER1, 126 RTMP_CLIENT_VER2, RTMP_CLIENT_VER3, RTMP_CLIENT_VER4); 127 } else { 128 snprintf(ver, sizeof(ver), "FMLE/3.0 (compatible; %s)", LIBAVFORMAT_IDENT); 129 ff_amf_write_field_name(&p, "type"); 130 ff_amf_write_string(&p, "nonprivate"); 131 } 132 ff_amf_write_field_name(&p, "flashVer"); 133 ff_amf_write_string(&p, ver); 134 ff_amf_write_field_name(&p, "tcUrl"); 135 ff_amf_write_string(&p, tcurl); 136 if (rt->is_input) { 137 ff_amf_write_field_name(&p, "fpad"); 138 ff_amf_write_bool(&p, 0); 139 ff_amf_write_field_name(&p, "capabilities"); 140 ff_amf_write_number(&p, 15.0); 141 ff_amf_write_field_name(&p, "audioCodecs"); 142 ff_amf_write_number(&p, 1639.0); 143 ff_amf_write_field_name(&p, "videoCodecs"); 144 ff_amf_write_number(&p, 252.0); 145 ff_amf_write_field_name(&p, "videoFunction"); 146 ff_amf_write_number(&p, 1.0); 147 } 148 ff_amf_write_object_end(&p); 149 150 pkt.data_size = p - pkt.data; 151 152 ff_rtmp_packet_write(rt->stream, &pkt, rt->chunk_size, rt->prev_pkt[1]); 153 ff_rtmp_packet_destroy(&pkt); 154} 155 156/** 157 * Generate 'releaseStream' call and send it to the server. It should make 158 * the server release some channel for media streams. 159 */ 160static void gen_release_stream(URLContext *s, RTMPContext *rt) 161{ 162 RTMPPacket pkt; 163 uint8_t *p; 164 165 ff_rtmp_packet_create(&pkt, RTMP_SYSTEM_CHANNEL, RTMP_PT_INVOKE, 0, 166 29 + strlen(rt->playpath)); 167 168 av_log(s, AV_LOG_DEBUG, "Releasing stream...\n"); 169 p = pkt.data; 170 ff_amf_write_string(&p, "releaseStream"); 171 ff_amf_write_number(&p, ++rt->nb_invokes); 172 ff_amf_write_null(&p); 173 ff_amf_write_string(&p, rt->playpath); 174 175 ff_rtmp_packet_write(rt->stream, &pkt, rt->chunk_size, rt->prev_pkt[1]); 176 ff_rtmp_packet_destroy(&pkt); 177} 178 179/** 180 * Generate 'FCPublish' call and send it to the server. It should make 181 * the server preapare for receiving media streams. 182 */ 183static void gen_fcpublish_stream(URLContext *s, RTMPContext *rt) 184{ 185 RTMPPacket pkt; 186 uint8_t *p; 187 188 ff_rtmp_packet_create(&pkt, RTMP_SYSTEM_CHANNEL, RTMP_PT_INVOKE, 0, 189 25 + strlen(rt->playpath)); 190 191 av_log(s, AV_LOG_DEBUG, "FCPublish stream...\n"); 192 p = pkt.data; 193 ff_amf_write_string(&p, "FCPublish"); 194 ff_amf_write_number(&p, ++rt->nb_invokes); 195 ff_amf_write_null(&p); 196 ff_amf_write_string(&p, rt->playpath); 197 198 ff_rtmp_packet_write(rt->stream, &pkt, rt->chunk_size, rt->prev_pkt[1]); 199 ff_rtmp_packet_destroy(&pkt); 200} 201 202/** 203 * Generate 'FCUnpublish' call and send it to the server. It should make 204 * the server destroy stream. 205 */ 206static void gen_fcunpublish_stream(URLContext *s, RTMPContext *rt) 207{ 208 RTMPPacket pkt; 209 uint8_t *p; 210 211 ff_rtmp_packet_create(&pkt, RTMP_SYSTEM_CHANNEL, RTMP_PT_INVOKE, 0, 212 27 + strlen(rt->playpath)); 213 214 av_log(s, AV_LOG_DEBUG, "UnPublishing stream...\n"); 215 p = pkt.data; 216 ff_amf_write_string(&p, "FCUnpublish"); 217 ff_amf_write_number(&p, ++rt->nb_invokes); 218 ff_amf_write_null(&p); 219 ff_amf_write_string(&p, rt->playpath); 220 221 ff_rtmp_packet_write(rt->stream, &pkt, rt->chunk_size, rt->prev_pkt[1]); 222 ff_rtmp_packet_destroy(&pkt); 223} 224 225/** 226 * Generate 'createStream' call and send it to the server. It should make 227 * the server allocate some channel for media streams. 228 */ 229static void gen_create_stream(URLContext *s, RTMPContext *rt) 230{ 231 RTMPPacket pkt; 232 uint8_t *p; 233 234 av_log(s, AV_LOG_DEBUG, "Creating stream...\n"); 235 ff_rtmp_packet_create(&pkt, RTMP_SYSTEM_CHANNEL, RTMP_PT_INVOKE, 0, 25); 236 237 p = pkt.data; 238 ff_amf_write_string(&p, "createStream"); 239 ff_amf_write_number(&p, ++rt->nb_invokes); 240 ff_amf_write_null(&p); 241 rt->create_stream_invoke = rt->nb_invokes; 242 243 ff_rtmp_packet_write(rt->stream, &pkt, rt->chunk_size, rt->prev_pkt[1]); 244 ff_rtmp_packet_destroy(&pkt); 245} 246 247 248/** 249 * Generate 'deleteStream' call and send it to the server. It should make 250 * the server remove some channel for media streams. 251 */ 252static void gen_delete_stream(URLContext *s, RTMPContext *rt) 253{ 254 RTMPPacket pkt; 255 uint8_t *p; 256 257 av_log(s, AV_LOG_DEBUG, "Deleting stream...\n"); 258 ff_rtmp_packet_create(&pkt, RTMP_SYSTEM_CHANNEL, RTMP_PT_INVOKE, 0, 34); 259 260 p = pkt.data; 261 ff_amf_write_string(&p, "deleteStream"); 262 ff_amf_write_number(&p, ++rt->nb_invokes); 263 ff_amf_write_null(&p); 264 ff_amf_write_number(&p, rt->main_channel_id); 265 266 ff_rtmp_packet_write(rt->stream, &pkt, rt->chunk_size, rt->prev_pkt[1]); 267 ff_rtmp_packet_destroy(&pkt); 268} 269 270/** 271 * Generate 'play' call and send it to the server, then ping the server 272 * to start actual playing. 273 */ 274static void gen_play(URLContext *s, RTMPContext *rt) 275{ 276 RTMPPacket pkt; 277 uint8_t *p; 278 279 av_log(s, AV_LOG_DEBUG, "Sending play command for '%s'\n", rt->playpath); 280 ff_rtmp_packet_create(&pkt, RTMP_VIDEO_CHANNEL, RTMP_PT_INVOKE, 0, 281 20 + strlen(rt->playpath)); 282 pkt.extra = rt->main_channel_id; 283 284 p = pkt.data; 285 ff_amf_write_string(&p, "play"); 286 ff_amf_write_number(&p, ++rt->nb_invokes); 287 ff_amf_write_null(&p); 288 ff_amf_write_string(&p, rt->playpath); 289 290 ff_rtmp_packet_write(rt->stream, &pkt, rt->chunk_size, rt->prev_pkt[1]); 291 ff_rtmp_packet_destroy(&pkt); 292 293 // set client buffer time disguised in ping packet 294 ff_rtmp_packet_create(&pkt, RTMP_NETWORK_CHANNEL, RTMP_PT_PING, 1, 10); 295 296 p = pkt.data; 297 bytestream_put_be16(&p, 3); 298 bytestream_put_be32(&p, 1); 299 bytestream_put_be32(&p, 256); //TODO: what is a good value here? 300 301 ff_rtmp_packet_write(rt->stream, &pkt, rt->chunk_size, rt->prev_pkt[1]); 302 ff_rtmp_packet_destroy(&pkt); 303} 304 305/** 306 * Generate 'publish' call and send it to the server. 307 */ 308static void gen_publish(URLContext *s, RTMPContext *rt) 309{ 310 RTMPPacket pkt; 311 uint8_t *p; 312 313 av_log(s, AV_LOG_DEBUG, "Sending publish command for '%s'\n", rt->playpath); 314 ff_rtmp_packet_create(&pkt, RTMP_SOURCE_CHANNEL, RTMP_PT_INVOKE, 0, 315 30 + strlen(rt->playpath)); 316 pkt.extra = rt->main_channel_id; 317 318 p = pkt.data; 319 ff_amf_write_string(&p, "publish"); 320 ff_amf_write_number(&p, ++rt->nb_invokes); 321 ff_amf_write_null(&p); 322 ff_amf_write_string(&p, rt->playpath); 323 ff_amf_write_string(&p, "live"); 324 325 ff_rtmp_packet_write(rt->stream, &pkt, rt->chunk_size, rt->prev_pkt[1]); 326 ff_rtmp_packet_destroy(&pkt); 327} 328 329/** 330 * Generate ping reply and send it to the server. 331 */ 332static void gen_pong(URLContext *s, RTMPContext *rt, RTMPPacket *ppkt) 333{ 334 RTMPPacket pkt; 335 uint8_t *p; 336 337 ff_rtmp_packet_create(&pkt, RTMP_NETWORK_CHANNEL, RTMP_PT_PING, ppkt->timestamp + 1, 6); 338 p = pkt.data; 339 bytestream_put_be16(&p, 7); 340 bytestream_put_be32(&p, AV_RB32(ppkt->data+2)); 341 ff_rtmp_packet_write(rt->stream, &pkt, rt->chunk_size, rt->prev_pkt[1]); 342 ff_rtmp_packet_destroy(&pkt); 343} 344 345/** 346 * Generate report on bytes read so far and send it to the server. 347 */ 348static void gen_bytes_read(URLContext *s, RTMPContext *rt, uint32_t ts) 349{ 350 RTMPPacket pkt; 351 uint8_t *p; 352 353 ff_rtmp_packet_create(&pkt, RTMP_NETWORK_CHANNEL, RTMP_PT_BYTES_READ, ts, 4); 354 p = pkt.data; 355 bytestream_put_be32(&p, rt->bytes_read); 356 ff_rtmp_packet_write(rt->stream, &pkt, rt->chunk_size, rt->prev_pkt[1]); 357 ff_rtmp_packet_destroy(&pkt); 358} 359 360//TODO: Move HMAC code somewhere. Eventually. 361#define HMAC_IPAD_VAL 0x36 362#define HMAC_OPAD_VAL 0x5C 363 364/** 365 * Calculate HMAC-SHA2 digest for RTMP handshake packets. 366 * 367 * @param src input buffer 368 * @param len input buffer length (should be 1536) 369 * @param gap offset in buffer where 32 bytes should not be taken into account 370 * when calculating digest (since it will be used to store that digest) 371 * @param key digest key 372 * @param keylen digest key length 373 * @param dst buffer where calculated digest will be stored (32 bytes) 374 */ 375static void rtmp_calc_digest(const uint8_t *src, int len, int gap, 376 const uint8_t *key, int keylen, uint8_t *dst) 377{ 378 struct AVSHA *sha; 379 uint8_t hmac_buf[64+32] = {0}; 380 int i; 381 382 sha = av_mallocz(av_sha_size); 383 384 if (keylen < 64) { 385 memcpy(hmac_buf, key, keylen); 386 } else { 387 av_sha_init(sha, 256); 388 av_sha_update(sha,key, keylen); 389 av_sha_final(sha, hmac_buf); 390 } 391 for (i = 0; i < 64; i++) 392 hmac_buf[i] ^= HMAC_IPAD_VAL; 393 394 av_sha_init(sha, 256); 395 av_sha_update(sha, hmac_buf, 64); 396 if (gap <= 0) { 397 av_sha_update(sha, src, len); 398 } else { //skip 32 bytes used for storing digest 399 av_sha_update(sha, src, gap); 400 av_sha_update(sha, src + gap + 32, len - gap - 32); 401 } 402 av_sha_final(sha, hmac_buf + 64); 403 404 for (i = 0; i < 64; i++) 405 hmac_buf[i] ^= HMAC_IPAD_VAL ^ HMAC_OPAD_VAL; //reuse XORed key for opad 406 av_sha_init(sha, 256); 407 av_sha_update(sha, hmac_buf, 64+32); 408 av_sha_final(sha, dst); 409 410 av_free(sha); 411} 412 413/** 414 * Put HMAC-SHA2 digest of packet data (except for the bytes where this digest 415 * will be stored) into that packet. 416 * 417 * @param buf handshake data (1536 bytes) 418 * @return offset to the digest inside input data 419 */ 420static int rtmp_handshake_imprint_with_digest(uint8_t *buf) 421{ 422 int i, digest_pos = 0; 423 424 for (i = 8; i < 12; i++) 425 digest_pos += buf[i]; 426 digest_pos = (digest_pos % 728) + 12; 427 428 rtmp_calc_digest(buf, RTMP_HANDSHAKE_PACKET_SIZE, digest_pos, 429 rtmp_player_key, PLAYER_KEY_OPEN_PART_LEN, 430 buf + digest_pos); 431 return digest_pos; 432} 433 434/** 435 * Verify that the received server response has the expected digest value. 436 * 437 * @param buf handshake data received from the server (1536 bytes) 438 * @param off position to search digest offset from 439 * @return 0 if digest is valid, digest position otherwise 440 */ 441static int rtmp_validate_digest(uint8_t *buf, int off) 442{ 443 int i, digest_pos = 0; 444 uint8_t digest[32]; 445 446 for (i = 0; i < 4; i++) 447 digest_pos += buf[i + off]; 448 digest_pos = (digest_pos % 728) + off + 4; 449 450 rtmp_calc_digest(buf, RTMP_HANDSHAKE_PACKET_SIZE, digest_pos, 451 rtmp_server_key, SERVER_KEY_OPEN_PART_LEN, 452 digest); 453 if (!memcmp(digest, buf + digest_pos, 32)) 454 return digest_pos; 455 return 0; 456} 457 458/** 459 * Perform handshake with the server by means of exchanging pseudorandom data 460 * signed with HMAC-SHA2 digest. 461 * 462 * @return 0 if handshake succeeds, negative value otherwise 463 */ 464static int rtmp_handshake(URLContext *s, RTMPContext *rt) 465{ 466 AVLFG rnd; 467 uint8_t tosend [RTMP_HANDSHAKE_PACKET_SIZE+1] = { 468 3, // unencrypted data 469 0, 0, 0, 0, // client uptime 470 RTMP_CLIENT_VER1, 471 RTMP_CLIENT_VER2, 472 RTMP_CLIENT_VER3, 473 RTMP_CLIENT_VER4, 474 }; 475 uint8_t clientdata[RTMP_HANDSHAKE_PACKET_SIZE]; 476 uint8_t serverdata[RTMP_HANDSHAKE_PACKET_SIZE+1]; 477 int i; 478 int server_pos, client_pos; 479 uint8_t digest[32]; 480 481 av_log(s, AV_LOG_DEBUG, "Handshaking...\n"); 482 483 av_lfg_init(&rnd, 0xDEADC0DE); 484 // generate handshake packet - 1536 bytes of pseudorandom data 485 for (i = 9; i <= RTMP_HANDSHAKE_PACKET_SIZE; i++) 486 tosend[i] = av_lfg_get(&rnd) >> 24; 487 client_pos = rtmp_handshake_imprint_with_digest(tosend + 1); 488 489 ffurl_write(rt->stream, tosend, RTMP_HANDSHAKE_PACKET_SIZE + 1); 490 i = ffurl_read_complete(rt->stream, serverdata, RTMP_HANDSHAKE_PACKET_SIZE + 1); 491 if (i != RTMP_HANDSHAKE_PACKET_SIZE + 1) { 492 av_log(s, AV_LOG_ERROR, "Cannot read RTMP handshake response\n"); 493 return -1; 494 } 495 i = ffurl_read_complete(rt->stream, clientdata, RTMP_HANDSHAKE_PACKET_SIZE); 496 if (i != RTMP_HANDSHAKE_PACKET_SIZE) { 497 av_log(s, AV_LOG_ERROR, "Cannot read RTMP handshake response\n"); 498 return -1; 499 } 500 501 av_log(s, AV_LOG_DEBUG, "Server version %d.%d.%d.%d\n", 502 serverdata[5], serverdata[6], serverdata[7], serverdata[8]); 503 504 if (rt->is_input && serverdata[5] >= 3) { 505 server_pos = rtmp_validate_digest(serverdata + 1, 772); 506 if (!server_pos) { 507 server_pos = rtmp_validate_digest(serverdata + 1, 8); 508 if (!server_pos) { 509 av_log(s, AV_LOG_ERROR, "Server response validating failed\n"); 510 return -1; 511 } 512 } 513 514 rtmp_calc_digest(tosend + 1 + client_pos, 32, 0, 515 rtmp_server_key, sizeof(rtmp_server_key), 516 digest); 517 rtmp_calc_digest(clientdata, RTMP_HANDSHAKE_PACKET_SIZE-32, 0, 518 digest, 32, 519 digest); 520 if (memcmp(digest, clientdata + RTMP_HANDSHAKE_PACKET_SIZE - 32, 32)) { 521 av_log(s, AV_LOG_ERROR, "Signature mismatch\n"); 522 return -1; 523 } 524 525 for (i = 0; i < RTMP_HANDSHAKE_PACKET_SIZE; i++) 526 tosend[i] = av_lfg_get(&rnd) >> 24; 527 rtmp_calc_digest(serverdata + 1 + server_pos, 32, 0, 528 rtmp_player_key, sizeof(rtmp_player_key), 529 digest); 530 rtmp_calc_digest(tosend, RTMP_HANDSHAKE_PACKET_SIZE - 32, 0, 531 digest, 32, 532 tosend + RTMP_HANDSHAKE_PACKET_SIZE - 32); 533 534 // write reply back to the server 535 ffurl_write(rt->stream, tosend, RTMP_HANDSHAKE_PACKET_SIZE); 536 } else { 537 ffurl_write(rt->stream, serverdata+1, RTMP_HANDSHAKE_PACKET_SIZE); 538 } 539 540 return 0; 541} 542 543/** 544 * Parse received packet and possibly perform some action depending on 545 * the packet contents. 546 * @return 0 for no errors, negative values for serious errors which prevent 547 * further communications, positive values for uncritical errors 548 */ 549static int rtmp_parse_result(URLContext *s, RTMPContext *rt, RTMPPacket *pkt) 550{ 551 int i, t; 552 const uint8_t *data_end = pkt->data + pkt->data_size; 553 554#ifdef DEBUG 555 ff_rtmp_packet_dump(s, pkt); 556#endif 557 558 switch (pkt->type) { 559 case RTMP_PT_CHUNK_SIZE: 560 if (pkt->data_size != 4) { 561 av_log(s, AV_LOG_ERROR, 562 "Chunk size change packet is not 4 bytes long (%d)\n", pkt->data_size); 563 return -1; 564 } 565 if (!rt->is_input) 566 ff_rtmp_packet_write(rt->stream, pkt, rt->chunk_size, rt->prev_pkt[1]); 567 rt->chunk_size = AV_RB32(pkt->data); 568 if (rt->chunk_size <= 0) { 569 av_log(s, AV_LOG_ERROR, "Incorrect chunk size %d\n", rt->chunk_size); 570 return -1; 571 } 572 av_log(s, AV_LOG_DEBUG, "New chunk size = %d\n", rt->chunk_size); 573 break; 574 case RTMP_PT_PING: 575 t = AV_RB16(pkt->data); 576 if (t == 6) 577 gen_pong(s, rt, pkt); 578 break; 579 case RTMP_PT_CLIENT_BW: 580 if (pkt->data_size < 4) { 581 av_log(s, AV_LOG_ERROR, 582 "Client bandwidth report packet is less than 4 bytes long (%d)\n", 583 pkt->data_size); 584 return -1; 585 } 586 av_log(s, AV_LOG_DEBUG, "Client bandwidth = %d\n", AV_RB32(pkt->data)); 587 rt->client_report_size = AV_RB32(pkt->data) >> 1; 588 break; 589 case RTMP_PT_INVOKE: 590 //TODO: check for the messages sent for wrong state? 591 if (!memcmp(pkt->data, "\002\000\006_error", 9)) { 592 uint8_t tmpstr[256]; 593 594 if (!ff_amf_get_field_value(pkt->data + 9, data_end, 595 "description", tmpstr, sizeof(tmpstr))) 596 av_log(s, AV_LOG_ERROR, "Server error: %s\n",tmpstr); 597 return -1; 598 } else if (!memcmp(pkt->data, "\002\000\007_result", 10)) { 599 switch (rt->state) { 600 case STATE_HANDSHAKED: 601 if (!rt->is_input) { 602 gen_release_stream(s, rt); 603 gen_fcpublish_stream(s, rt); 604 rt->state = STATE_RELEASING; 605 } else { 606 rt->state = STATE_CONNECTING; 607 } 608 gen_create_stream(s, rt); 609 break; 610 case STATE_FCPUBLISH: 611 rt->state = STATE_CONNECTING; 612 break; 613 case STATE_RELEASING: 614 rt->state = STATE_FCPUBLISH; 615 /* hack for Wowza Media Server, it does not send result for 616 * releaseStream and FCPublish calls */ 617 if (!pkt->data[10]) { 618 int pkt_id = av_int2double(AV_RB64(pkt->data + 11)); 619 if (pkt_id == rt->create_stream_invoke) 620 rt->state = STATE_CONNECTING; 621 } 622 if (rt->state != STATE_CONNECTING) 623 break; 624 case STATE_CONNECTING: 625 //extract a number from the result 626 if (pkt->data[10] || pkt->data[19] != 5 || pkt->data[20]) { 627 av_log(s, AV_LOG_WARNING, "Unexpected reply on connect()\n"); 628 } else { 629 rt->main_channel_id = av_int2double(AV_RB64(pkt->data + 21)); 630 } 631 if (rt->is_input) { 632 gen_play(s, rt); 633 } else { 634 gen_publish(s, rt); 635 } 636 rt->state = STATE_READY; 637 break; 638 } 639 } else if (!memcmp(pkt->data, "\002\000\010onStatus", 11)) { 640 const uint8_t* ptr = pkt->data + 11; 641 uint8_t tmpstr[256]; 642 643 for (i = 0; i < 2; i++) { 644 t = ff_amf_tag_size(ptr, data_end); 645 if (t < 0) 646 return 1; 647 ptr += t; 648 } 649 t = ff_amf_get_field_value(ptr, data_end, 650 "level", tmpstr, sizeof(tmpstr)); 651 if (!t && !strcmp(tmpstr, "error")) { 652 if (!ff_amf_get_field_value(ptr, data_end, 653 "description", tmpstr, sizeof(tmpstr))) 654 av_log(s, AV_LOG_ERROR, "Server error: %s\n",tmpstr); 655 return -1; 656 } 657 t = ff_amf_get_field_value(ptr, data_end, 658 "code", tmpstr, sizeof(tmpstr)); 659 if (!t && !strcmp(tmpstr, "NetStream.Play.Start")) rt->state = STATE_PLAYING; 660 if (!t && !strcmp(tmpstr, "NetStream.Play.Stop")) rt->state = STATE_STOPPED; 661 if (!t && !strcmp(tmpstr, "NetStream.Play.UnpublishNotify")) rt->state = STATE_STOPPED; 662 if (!t && !strcmp(tmpstr, "NetStream.Publish.Start")) rt->state = STATE_PUBLISHING; 663 } 664 break; 665 } 666 return 0; 667} 668 669/** 670 * Interact with the server by receiving and sending RTMP packets until 671 * there is some significant data (media data or expected status notification). 672 * 673 * @param s reading context 674 * @param for_header non-zero value tells function to work until it 675 * gets notification from the server that playing has been started, 676 * otherwise function will work until some media data is received (or 677 * an error happens) 678 * @return 0 for successful operation, negative value in case of error 679 */ 680static int get_packet(URLContext *s, int for_header) 681{ 682 RTMPContext *rt = s->priv_data; 683 int ret; 684 uint8_t *p; 685 const uint8_t *next; 686 uint32_t data_size; 687 uint32_t ts, cts, pts=0; 688 689 if (rt->state == STATE_STOPPED) 690 return AVERROR_EOF; 691 692 for (;;) { 693 RTMPPacket rpkt = { 0 }; 694 if ((ret = ff_rtmp_packet_read(rt->stream, &rpkt, 695 rt->chunk_size, rt->prev_pkt[0])) <= 0) { 696 if (ret == 0) { 697 return AVERROR(EAGAIN); 698 } else { 699 return AVERROR(EIO); 700 } 701 } 702 rt->bytes_read += ret; 703 if (rt->bytes_read > rt->last_bytes_read + rt->client_report_size) { 704 av_log(s, AV_LOG_DEBUG, "Sending bytes read report\n"); 705 gen_bytes_read(s, rt, rpkt.timestamp + 1); 706 rt->last_bytes_read = rt->bytes_read; 707 } 708 709 ret = rtmp_parse_result(s, rt, &rpkt); 710 if (ret < 0) {//serious error in current packet 711 ff_rtmp_packet_destroy(&rpkt); 712 return -1; 713 } 714 if (rt->state == STATE_STOPPED) { 715 ff_rtmp_packet_destroy(&rpkt); 716 return AVERROR_EOF; 717 } 718 if (for_header && (rt->state == STATE_PLAYING || rt->state == STATE_PUBLISHING)) { 719 ff_rtmp_packet_destroy(&rpkt); 720 return 0; 721 } 722 if (!rpkt.data_size || !rt->is_input) { 723 ff_rtmp_packet_destroy(&rpkt); 724 continue; 725 } 726 if (rpkt.type == RTMP_PT_VIDEO || rpkt.type == RTMP_PT_AUDIO || 727 (rpkt.type == RTMP_PT_NOTIFY && !memcmp("\002\000\012onMetaData", rpkt.data, 13))) { 728 ts = rpkt.timestamp; 729 730 // generate packet header and put data into buffer for FLV demuxer 731 rt->flv_off = 0; 732 rt->flv_size = rpkt.data_size + 15; 733 rt->flv_data = p = av_realloc(rt->flv_data, rt->flv_size); 734 bytestream_put_byte(&p, rpkt.type); 735 bytestream_put_be24(&p, rpkt.data_size); 736 bytestream_put_be24(&p, ts); 737 bytestream_put_byte(&p, ts >> 24); 738 bytestream_put_be24(&p, 0); 739 bytestream_put_buffer(&p, rpkt.data, rpkt.data_size); 740 bytestream_put_be32(&p, 0); 741 ff_rtmp_packet_destroy(&rpkt); 742 return 0; 743 } else if (rpkt.type == RTMP_PT_METADATA) { 744 // we got raw FLV data, make it available for FLV demuxer 745 rt->flv_off = 0; 746 rt->flv_size = rpkt.data_size; 747 rt->flv_data = av_realloc(rt->flv_data, rt->flv_size); 748 /* rewrite timestamps */ 749 next = rpkt.data; 750 ts = rpkt.timestamp; 751 while (next - rpkt.data < rpkt.data_size - 11) { 752 next++; 753 data_size = bytestream_get_be24(&next); 754 p=next; 755 cts = bytestream_get_be24(&next); 756 cts |= bytestream_get_byte(&next) << 24; 757 if (pts==0) 758 pts=cts; 759 ts += cts - pts; 760 pts = cts; 761 bytestream_put_be24(&p, ts); 762 bytestream_put_byte(&p, ts >> 24); 763 next += data_size + 3 + 4; 764 } 765 memcpy(rt->flv_data, rpkt.data, rpkt.data_size); 766 ff_rtmp_packet_destroy(&rpkt); 767 return 0; 768 } 769 ff_rtmp_packet_destroy(&rpkt); 770 } 771} 772 773static int rtmp_close(URLContext *h) 774{ 775 RTMPContext *rt = h->priv_data; 776 777 if (!rt->is_input) { 778 rt->flv_data = NULL; 779 if (rt->out_pkt.data_size) 780 ff_rtmp_packet_destroy(&rt->out_pkt); 781 if (rt->state > STATE_FCPUBLISH) 782 gen_fcunpublish_stream(h, rt); 783 } 784 if (rt->state > STATE_HANDSHAKED) 785 gen_delete_stream(h, rt); 786 787 av_freep(&rt->flv_data); 788 ffurl_close(rt->stream); 789 return 0; 790} 791 792/** 793 * Open RTMP connection and verify that the stream can be played. 794 * 795 * URL syntax: rtmp://server[:port][/app][/playpath] 796 * where 'app' is first one or two directories in the path 797 * (e.g. /ondemand/, /flash/live/, etc.) 798 * and 'playpath' is a file name (the rest of the path, 799 * may be prefixed with "mp4:") 800 */ 801static int rtmp_open(URLContext *s, const char *uri, int flags) 802{ 803 RTMPContext *rt = s->priv_data; 804 char proto[8], hostname[256], path[1024], *fname; 805 uint8_t buf[2048]; 806 int port; 807 int ret; 808 809 rt->is_input = !(flags & AVIO_FLAG_WRITE); 810 811 av_url_split(proto, sizeof(proto), NULL, 0, hostname, sizeof(hostname), &port, 812 path, sizeof(path), s->filename); 813 814 if (port < 0) 815 port = RTMP_DEFAULT_PORT; 816 ff_url_join(buf, sizeof(buf), "tcp", NULL, hostname, port, NULL); 817 818 if (ffurl_open(&rt->stream, buf, AVIO_FLAG_READ_WRITE, 819 &s->interrupt_callback, NULL) < 0) { 820 av_log(s , AV_LOG_ERROR, "Cannot open connection %s\n", buf); 821 goto fail; 822 } 823 824 rt->state = STATE_START; 825 if (rtmp_handshake(s, rt)) 826 goto fail; 827 828 rt->chunk_size = 128; 829 rt->state = STATE_HANDSHAKED; 830 //extract "app" part from path 831 if (!strncmp(path, "/ondemand/", 10)) { 832 fname = path + 10; 833 memcpy(rt->app, "ondemand", 9); 834 } else { 835 char *p = strchr(path + 1, '/'); 836 if (!p) { 837 fname = path + 1; 838 rt->app[0] = '\0'; 839 } else { 840 char *c = strchr(p + 1, ':'); 841 fname = strchr(p + 1, '/'); 842 if (!fname || c < fname) { 843 fname = p + 1; 844 av_strlcpy(rt->app, path + 1, p - path); 845 } else { 846 fname++; 847 av_strlcpy(rt->app, path + 1, fname - path - 1); 848 } 849 } 850 } 851 if (!strchr(fname, ':') && 852 (!strcmp(fname + strlen(fname) - 4, ".f4v") || 853 !strcmp(fname + strlen(fname) - 4, ".mp4"))) { 854 memcpy(rt->playpath, "mp4:", 5); 855 } else { 856 rt->playpath[0] = 0; 857 } 858 strncat(rt->playpath, fname, sizeof(rt->playpath) - 5); 859 860 rt->client_report_size = 1048576; 861 rt->bytes_read = 0; 862 rt->last_bytes_read = 0; 863 864 av_log(s, AV_LOG_DEBUG, "Proto = %s, path = %s, app = %s, fname = %s\n", 865 proto, path, rt->app, rt->playpath); 866 gen_connect(s, rt, proto, hostname, port); 867 868 do { 869 ret = get_packet(s, 1); 870 } while (ret == EAGAIN); 871 if (ret < 0) 872 goto fail; 873 874 if (rt->is_input) { 875 // generate FLV header for demuxer 876 rt->flv_size = 13; 877 rt->flv_data = av_realloc(rt->flv_data, rt->flv_size); 878 rt->flv_off = 0; 879 memcpy(rt->flv_data, "FLV\1\5\0\0\0\011\0\0\0\0", rt->flv_size); 880 } else { 881 rt->flv_size = 0; 882 rt->flv_data = NULL; 883 rt->flv_off = 0; 884 rt->skip_bytes = 13; 885 } 886 887 s->max_packet_size = rt->stream->max_packet_size; 888 s->is_streamed = 1; 889 return 0; 890 891fail: 892 rtmp_close(s); 893 return AVERROR(EIO); 894} 895 896static int rtmp_read(URLContext *s, uint8_t *buf, int size) 897{ 898 RTMPContext *rt = s->priv_data; 899 int orig_size = size; 900 int ret; 901 902 while (size > 0) { 903 int data_left = rt->flv_size - rt->flv_off; 904 905 if (data_left >= size) { 906 memcpy(buf, rt->flv_data + rt->flv_off, size); 907 rt->flv_off += size; 908 return orig_size; 909 } 910 if (data_left > 0) { 911 memcpy(buf, rt->flv_data + rt->flv_off, data_left); 912 buf += data_left; 913 size -= data_left; 914 rt->flv_off = rt->flv_size; 915 return data_left; 916 } 917 if ((ret = get_packet(s, 0)) < 0) 918 return ret; 919 } 920 return orig_size; 921} 922 923static int rtmp_write(URLContext *s, const uint8_t *buf, int size) 924{ 925 RTMPContext *rt = s->priv_data; 926 int size_temp = size; 927 int pktsize, pkttype; 928 uint32_t ts; 929 const uint8_t *buf_temp = buf; 930 931 do { 932 if (rt->skip_bytes) { 933 int skip = FFMIN(rt->skip_bytes, size_temp); 934 buf_temp += skip; 935 size_temp -= skip; 936 rt->skip_bytes -= skip; 937 continue; 938 } 939 940 if (rt->flv_header_bytes < 11) { 941 const uint8_t *header = rt->flv_header; 942 int copy = FFMIN(11 - rt->flv_header_bytes, size_temp); 943 bytestream_get_buffer(&buf_temp, rt->flv_header + rt->flv_header_bytes, copy); 944 rt->flv_header_bytes += copy; 945 size_temp -= copy; 946 if (rt->flv_header_bytes < 11) 947 break; 948 949 pkttype = bytestream_get_byte(&header); 950 pktsize = bytestream_get_be24(&header); 951 ts = bytestream_get_be24(&header); 952 ts |= bytestream_get_byte(&header) << 24; 953 bytestream_get_be24(&header); 954 rt->flv_size = pktsize; 955 956 //force 12bytes header 957 if (((pkttype == RTMP_PT_VIDEO || pkttype == RTMP_PT_AUDIO) && ts == 0) || 958 pkttype == RTMP_PT_NOTIFY) { 959 if (pkttype == RTMP_PT_NOTIFY) 960 pktsize += 16; 961 rt->prev_pkt[1][RTMP_SOURCE_CHANNEL].channel_id = 0; 962 } 963 964 //this can be a big packet, it's better to send it right here 965 ff_rtmp_packet_create(&rt->out_pkt, RTMP_SOURCE_CHANNEL, pkttype, ts, pktsize); 966 rt->out_pkt.extra = rt->main_channel_id; 967 rt->flv_data = rt->out_pkt.data; 968 969 if (pkttype == RTMP_PT_NOTIFY) 970 ff_amf_write_string(&rt->flv_data, "@setDataFrame"); 971 } 972 973 if (rt->flv_size - rt->flv_off > size_temp) { 974 bytestream_get_buffer(&buf_temp, rt->flv_data + rt->flv_off, size_temp); 975 rt->flv_off += size_temp; 976 size_temp = 0; 977 } else { 978 bytestream_get_buffer(&buf_temp, rt->flv_data + rt->flv_off, rt->flv_size - rt->flv_off); 979 size_temp -= rt->flv_size - rt->flv_off; 980 rt->flv_off += rt->flv_size - rt->flv_off; 981 } 982 983 if (rt->flv_off == rt->flv_size) { 984 rt->skip_bytes = 4; 985 986 ff_rtmp_packet_write(rt->stream, &rt->out_pkt, rt->chunk_size, rt->prev_pkt[1]); 987 ff_rtmp_packet_destroy(&rt->out_pkt); 988 rt->flv_size = 0; 989 rt->flv_off = 0; 990 rt->flv_header_bytes = 0; 991 } 992 } while (buf_temp - buf < size); 993 return size; 994} 995 996URLProtocol ff_rtmp_protocol = { 997 .name = "rtmp", 998 .url_open = rtmp_open, 999 .url_read = rtmp_read, 1000 .url_write = rtmp_write, 1001 .url_close = rtmp_close, 1002 .priv_data_size = sizeof(RTMPContext), 1003 .flags = URL_PROTOCOL_FLAG_NETWORK, 1004}; 1005