socket_pair.cpp 15.9 KB
Newer Older
1
/*
2
 *  Copyright (C) 2004-2019 Savoir-faire Linux Inc.
3 4 5
 *  Copyright (c) 2002 Fabrice Bellard
 *
 *  Author: Tristan Matthews <tristan.matthews@savoirfairelinux.com>
6
 *  Author: Guillaume Roguez <guillaume.roguez@savoirfairelinux.com>
7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22
 *
 *  This program is free software; you can redistribute it and/or modify
 *  it under the terms of the GNU General Public License as published by
 *  the Free Software Foundation; either version 3 of the License, or
 *  (at your option) any later version.
 *
 *  This program is distributed in the hope that it will be useful,
 *  but WITHOUT ANY WARRANTY; without even the implied warranty of
 *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *  GNU General Public License for more details.
 *
 *  You should have received a copy of the GNU General Public License
 *  along with this program; if not, write to the Free Software
 *  Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301 USA.
 */

23 24 25
#include "ip_utils.h" // MUST BE INCLUDED FIRST
#include "libav_deps.h" // THEN THIS ONE AFTER

26
#include "socket_pair.h"
27
#include "ice_socket.h"
28
#include "libav_utils.h"
29
#include "logger.h"
30
#include "security/memory.h"
31

32 33 34 35 36
#include <iostream>
#include <string>
#include <algorithm>
#include <iterator>

37 38 39 40
extern "C" {
#include "srtp.h"
}

41 42 43 44
#include <cstring>
#include <stdexcept>
#include <unistd.h>
#include <sys/types.h>
45
#include <ciso646> // fix windows compiler bug
Edric Milaret's avatar
Edric Milaret committed
46 47 48 49

#ifdef _WIN32
#define SOCK_NONBLOCK FIONBIO
#define poll WSAPoll
50
#define close(x) closesocket(x)
Edric Milaret's avatar
Edric Milaret committed
51
#endif
52

53
#ifdef __ANDROID__
54
#include <asm-generic/fcntl.h>
55 56 57 58 59
#define SOCK_NONBLOCK O_NONBLOCK
#endif

#ifdef __APPLE__
#include <fcntl.h>
60
#endif
61

Pierre Lespagnol's avatar
Pierre Lespagnol committed
62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79

// Swap 2 byte, 16 bit values:
#define Swap2Bytes(val) \
	 ( (((val) >> 8) & 0x00FF) | (((val) << 8) & 0xFF00) )

// Swap 4 byte, 32 bit values:
#define Swap4Bytes(val) \
    ( (((val) >> 24) & 0x000000FF) | (((val) >>  8) & 0x0000FF00) | \
    (((val) <<  8) & 0x00FF0000) | (((val) << 24) & 0xFF000000) )

// Swap 8 byte, 64 bit values:
#define Swap8Bytes(val) \
    ( (((val) >> 56) & 0x00000000000000FF) | (((val) >> 40) & 0x000000000000FF00) | \
    (((val) >> 24) & 0x0000000000FF0000) | (((val) >>  8) & 0x00000000FF000000) | \
    (((val) <<  8) & 0x000000FF00000000) | (((val) << 24) & 0x0000FF0000000000) | \
    (((val) << 40) & 0x00FF000000000000) | (((val) << 56) & 0xFF00000000000000) )


Adrien Béraud's avatar
Adrien Béraud committed
80
namespace jami {
81

82
static constexpr int NET_POLL_TIMEOUT = 100; /* poll() timeout in ms */
83
static constexpr int RTP_MAX_PACKET_LENGTH = 2048;
84 85
static constexpr auto UDP_HEADER_SIZE = 8;
static constexpr auto SRTP_OVERHEAD = 10;
86
static constexpr uint32_t RTCP_RR_FRACTION_MASK = 0xFF000000;
87

88 89
enum class DataType : unsigned { RTP=1<<0, RTCP=1<<1 };

90
class SRTPProtoContext {
91 92 93
public:
    SRTPProtoContext(const char* out_suite, const char* out_key,
                     const char* in_suite, const char* in_key) {
94 95
        ring_secure_memzero(&srtp_out, sizeof(srtp_out));
        ring_secure_memzero(&srtp_in, sizeof(srtp_in));
96 97 98 99 100
        if (out_suite && out_key) {
            // XXX: see srtp_open from libavformat/srtpproto.c
            if (ff_srtp_set_crypto(&srtp_out, out_suite, out_key) < 0) {
                srtp_close();
                throw std::runtime_error("Could not set crypto on output");
101
            }
102
        }
103

104 105 106 107
        if (in_suite && in_key) {
            if (ff_srtp_set_crypto(&srtp_in, in_suite, in_key) < 0) {
                srtp_close();
                throw std::runtime_error("Could not set crypto on input");
108 109
            }
        }
110
    }
111

112 113 114
    ~SRTPProtoContext() {
        srtp_close();
    }
115

116 117 118
    SRTPContext srtp_out {};
    SRTPContext srtp_in {};
    uint8_t encryptbuf[RTP_MAX_PACKET_LENGTH];
119

120 121 122 123 124
private:
    void srtp_close() noexcept {
        ff_srtp_free(&srtp_out);
        ff_srtp_free(&srtp_in);
    }
125
};
126

127 128
static int
ff_network_wait_fd(int fd)
129
{
130
    struct pollfd p = { fd, POLLOUT, 0 };
131
    auto ret = poll(&p, 1, NET_POLL_TIMEOUT);
132
    return ret < 0 ? errno : p.revents & (POLLOUT | POLLERR | POLLHUP) ? 0 : -EAGAIN;
133 134
}

135
static int
136
udp_socket_create(int family, int port)
137 138 139
{
    int udp_fd = -1;

140
#ifdef __APPLE__
141 142 143 144 145 146 147 148 149 150 151 152
    udp_fd = socket(family, SOCK_DGRAM, 0);
    if (udp_fd >= 0 && fcntl(udp_fd, F_SETFL, O_NONBLOCK) < 0) {
        close(udp_fd);
        udp_fd = -1;
    }
#elif defined _WIN32
    udp_fd = socket(family, SOCK_DGRAM, 0);
    u_long block = 1;
    if (udp_fd >= 0 && ioctlsocket(udp_fd, FIONBIO, &block) < 0) {
        close(udp_fd);
        udp_fd = -1;
    }
153
#else
154
    udp_fd = socket(family, SOCK_DGRAM | SOCK_NONBLOCK, 0);
155
#endif
156

157
    if (udp_fd < 0) {
Adrien Béraud's avatar
Adrien Béraud committed
158
        JAMI_ERR("socket() failed");
159
        strErr();
160 161
        return -1;
    }
162

163 164
    auto bind_addr = ip_utils::getAnyHostAddr(family);
    if (not bind_addr.isIpv4() and not bind_addr.isIpv6()) {
Adrien Béraud's avatar
Adrien Béraud committed
165
        JAMI_ERR("No IPv4/IPv6 host found for family %u", family);
166 167 168
        close(udp_fd);
        return -1;
    }
169

170
    bind_addr.setPort(port);
Adrien Béraud's avatar
Adrien Béraud committed
171
    JAMI_DBG("use local address: %s", bind_addr.toString(true, true).c_str());
172
    if (::bind(udp_fd, bind_addr, bind_addr.getLength()) < 0) {
Adrien Béraud's avatar
Adrien Béraud committed
173
        JAMI_ERR("bind() failed");
Vittorio Giovara's avatar
Vittorio Giovara committed
174
        strErr();
175 176 177
        close(udp_fd);
        udp_fd = -1;
    }
178 179 180 181

    return udp_fd;
}

182
SocketPair::SocketPair(const char *uri, int localPort)
183 184 185 186
{
    openSockets(uri, localPort);
}

187 188
SocketPair::SocketPair(std::unique_ptr<IceSocket> rtp_sock,
                       std::unique_ptr<IceSocket> rtcp_sock)
189 190
    : rtp_sock_(std::move(rtp_sock))
    , rtcp_sock_(std::move(rtcp_sock))
191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208
{
    auto queueRtpPacket = [this](uint8_t* buf, size_t len) {
        std::lock_guard<std::mutex> l(dataBuffMutex_);
        rtpDataBuff_.emplace_back(buf, buf+len);
        cv_.notify_one();
        return len;
    };

    auto queueRtcpPacket = [this](uint8_t* buf, size_t len) {
        std::lock_guard<std::mutex> l(dataBuffMutex_);
        rtcpDataBuff_.emplace_back(buf, buf+len);
        cv_.notify_one();
        return len;
    };

    rtp_sock_->setOnRecv(queueRtpPacket);
    rtcp_sock_->setOnRecv(queueRtcpPacket);
}
209

210 211
SocketPair::~SocketPair()
{
212 213
    interrupt();
    closeSockets();
214 215
}

Pierre Lespagnol's avatar
Pierre Lespagnol committed
216 217 218 219 220 221 222 223 224
bool
SocketPair::waitForRTCP(std::chrono::seconds interval)
{
    std::unique_lock<std::mutex> lock(rtcpInfo_mutex_);
    return cvRtcpPacketReadyToRead_.wait_for(lock, interval, [this]{
        return interrupted_ or not listRtcpHeader_.empty();
    });
}

225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241
void
SocketPair::saveRtcpPacket(uint8_t* buf, size_t len)
{
    if (len < sizeof(rtcpRRHeader))
        return;

    auto header = reinterpret_cast<rtcpRRHeader*>(buf);
    if(header->pt != 201) //201 = RR PT
        return;

    std::lock_guard<std::mutex> lock(rtcpInfo_mutex_);

    if (listRtcpHeader_.size() >= MAX_LIST_SIZE) {
        listRtcpHeader_.pop_front();
    }

    listRtcpHeader_.push_back(*header);
Pierre Lespagnol's avatar
Pierre Lespagnol committed
242 243

    cvRtcpPacketReadyToRead_.notify_one();
244 245
}

Pierre Lespagnol's avatar
Pierre Lespagnol committed
246
std::list<rtcpRRHeader>
247 248
SocketPair::getRtcpInfo()
{
Pierre Lespagnol's avatar
Pierre Lespagnol committed
249 250
    std::lock_guard<std::mutex> lock(rtcpInfo_mutex_);
    return std::move(listRtcpHeader_);
251 252
}

253 254 255
void
SocketPair::createSRTP(const char* out_suite, const char* out_key,
                       const char* in_suite, const char* in_key)
256
{
257
    srtpContext_.reset(new SRTPProtoContext(out_suite, out_key, in_suite, in_key));
258 259
}

260 261
void
SocketPair::interrupt()
262 263
{
    interrupted_ = true;
264 265 266
    if (rtp_sock_) rtp_sock_->setOnRecv(nullptr);
    if (rtcp_sock_) rtcp_sock_->setOnRecv(nullptr);
    cv_.notify_all();
267 268
}

269 270 271 272 273 274
void
SocketPair::stopSendOp(bool state)
{
    noWrite_ = state;
}

275 276
void
SocketPair::closeSockets()
277 278
{
    if (rtcpHandle_ > 0 and close(rtcpHandle_))
Vittorio Giovara's avatar
Vittorio Giovara committed
279
        strErr();
280
    if (rtpHandle_ > 0 and close(rtpHandle_))
Vittorio Giovara's avatar
Vittorio Giovara committed
281
        strErr();
282 283
}

284 285
void
SocketPair::openSockets(const char* uri, int local_rtp_port)
286 287 288
{
    char hostname[256];
    char path[1024];
289
    int dst_rtp_port;
290

291
    libav_utils::ring_url_split(uri, hostname, sizeof(hostname), &dst_rtp_port, path, sizeof(path));
292 293

    const int local_rtcp_port = local_rtp_port + 1;
294
    const int dst_rtcp_port = dst_rtp_port + 1;
295

296 297
    rtpDestAddr_ = IpAddr {hostname}; rtpDestAddr_.setPort(dst_rtp_port);
    rtcpDestAddr_ = IpAddr {hostname}; rtcpDestAddr_.setPort(dst_rtcp_port);
298

299 300 301
    // Open local sockets (RTP/RTCP)
    if ((rtpHandle_ = udp_socket_create(rtpDestAddr_.getFamily(), local_rtp_port)) == -1 or
        (rtcpHandle_ = udp_socket_create(rtcpDestAddr_.getFamily(), local_rtcp_port)) == -1) {
302
        closeSockets();
303
        throw std::runtime_error("Sockets creation failed");
304
    }
305

Adrien Béraud's avatar
Adrien Béraud committed
306
    JAMI_WARN("SocketPair: local{%d,%d} / %s{%d,%d}",
307
              local_rtp_port, local_rtcp_port, hostname, dst_rtp_port, dst_rtcp_port);
308 309
}

310
MediaIOHandle*
311
SocketPair::createIOContext(const uint16_t mtu)
312
{
313 314 315
    unsigned ip_header_size;
    if (rtp_sock_)
        ip_header_size = rtp_sock_->getTransportOverhead();
316
    else if (rtpDestAddr_.getFamily() == AF_INET6)
317 318 319
        ip_header_size = 40;
    else
        ip_header_size = 20;
320 321
    return new MediaIOHandle( mtu - (srtpContext_ ? SRTP_OVERHEAD : 0) - UDP_HEADER_SIZE - ip_header_size,
                              true,
322 323 324
                             [](void* sp, uint8_t* buf, int len){ return static_cast<SocketPair*>(sp)->readCallback(buf, len); },
                             [](void* sp, uint8_t* buf, int len){ return static_cast<SocketPair*>(sp)->writeCallback(buf, len); },
                             0, reinterpret_cast<void*>(this));
325 326
}

327 328 329
int
SocketPair::waitForData()
{
330
    // System sockets
331
    if (rtpHandle_ >= 0) {
332 333 334 335 336 337 338 339 340 341 342
        int ret;
        do {
            if (interrupted_) {
                errno = EINTR;
                return -1;
            }

            // work with system socket
            struct pollfd p[2] = { {rtpHandle_, POLLIN, 0},
                                   {rtcpHandle_, POLLIN, 0} };
            ret = poll(p, 2, NET_POLL_TIMEOUT);
343 344 345 346 347 348 349 350
            if (ret > 0) {
                ret = 0;
                if (p[0].revents & POLLIN)
                    ret |= static_cast<int>(DataType::RTP);
                if (p[1].revents & POLLIN)
                    ret |= static_cast<int>(DataType::RTCP);
            }
        } while (!ret or (ret < 0 and errno == EAGAIN));
351 352

        return ret;
353

354 355 356
    }

    // work with IceSocket
357 358 359 360 361 362 363
    {
        std::unique_lock<std::mutex> lk(dataBuffMutex_);
        cv_.wait(lk, [this]{ return interrupted_ or not rtpDataBuff_.empty() or not rtcpDataBuff_.empty(); });
    }

    if (interrupted_) {
        errno = EINTR;
364 365 366
        return -1;
    }

367
    return static_cast<int>(DataType::RTP) | static_cast<int>(DataType::RTCP);
368 369 370
}

int
371
SocketPair::readRtpData(void* buf, int buf_size)
372
{
373
    // handle system socket
374 375 376
    if (rtpHandle_ >= 0) {
        struct sockaddr_storage from;
        socklen_t from_len = sizeof(from);
377 378
        return recvfrom(rtpHandle_, static_cast<char*>(buf), buf_size, 0,
                        reinterpret_cast<struct sockaddr*>(&from), &from_len);
379 380
    }

381 382 383 384 385 386 387 388 389 390
    // handle ICE
    std::unique_lock<std::mutex> lk(dataBuffMutex_);
    if (not rtpDataBuff_.empty()) {
        auto pkt = rtpDataBuff_.front();
        rtpDataBuff_.pop_front();
        lk.unlock(); // to not block our ICE callbacks
        int pkt_size = pkt.size();
        int len = std::min(pkt_size, buf_size);
        std::copy_n(pkt.begin(), len, static_cast<char*>(buf));
        return len;
391
    }
392

393
    return 0;
394 395 396
}

int
397
SocketPair::readRtcpData(void* buf, int buf_size)
398
{
399
    // handle system socket
400 401 402
    if (rtcpHandle_ >= 0) {
        struct sockaddr_storage from;
        socklen_t from_len = sizeof(from);
403 404
        return recvfrom(rtcpHandle_, static_cast<char*>(buf), buf_size, 0,
                        reinterpret_cast<struct sockaddr*>(&from), &from_len);
405 406
    }

407 408 409 410 411 412 413 414 415 416
    // handle ICE
    std::unique_lock<std::mutex> lk(dataBuffMutex_);
    if (not rtcpDataBuff_.empty()) {
        auto pkt = rtcpDataBuff_.front();
        rtcpDataBuff_.pop_front();
        lk.unlock();
        int pkt_size = pkt.size();
        int len = std::min(pkt_size, buf_size);
        std::copy_n(pkt.begin(), len, static_cast<char*>(buf));
        return len;
417
    }
418 419

    return 0;
420 421 422
}

int
423
SocketPair::readCallback(uint8_t* buf, int buf_size)
424
{
425 426 427
    auto datatype = waitForData();
    if (datatype < 0)
        return datatype;
428

429 430
    int len = 0;
    bool fromRTCP = false;
431

Pierre Lespagnol's avatar
Pierre Lespagnol committed
432 433 434 435 436 437 438 439
    auto header = reinterpret_cast<rtcpRRHeader*>(buf);
    if(header->pt == 201) //201 = RR PT
    {
        lastDLSR_ = Swap4Bytes(header->dlsr);
        //JAMI_WARN("Read RR, lastDLSR : %d", lastDLSR_);
        lastRR_time = std::chrono::steady_clock::now();
    }

440 441 442
    // Priority to RTCP as its less invasive in bandwidth
    if (datatype & static_cast<int>(DataType::RTCP)) {
        len = readRtcpData(buf, buf_size);
443
        saveRtcpPacket(buf, len);
444 445
        fromRTCP = true;
    }
446

447 448 449 450 451
    // No RTCP... try RTP
    if (!len and (datatype & static_cast<int>(DataType::RTP))) {
        len = readRtpData(buf, buf_size);
        fromRTCP = false;
    }
452

453
    if (len <= 0)
454
        return len;
455 456 457 458 459

    // SRTP decrypt
    if (not fromRTCP and srtpContext_ and srtpContext_->srtp_in.aes) {
        auto err = ff_srtp_decrypt(&srtpContext_->srtp_in, buf, &len);
        if (err < 0)
Adrien Béraud's avatar
Adrien Béraud committed
460
            JAMI_WARN("decrypt error %d", err);
461
    }
462

463 464 465 466
    if (len != 0)
        return len;
    else
        return AVERROR_EOF;
467
}
468

469 470 471 472 473 474 475 476
int
SocketPair::writeData(uint8_t* buf, int buf_size)
{
    bool isRTCP = RTP_PT_IS_RTCP(buf[1]);

    // System sockets?
    if (rtpHandle_ >= 0) {
        int fd;
477
        IpAddr* dest_addr;
478 479 480

        if (isRTCP) {
            fd = rtcpHandle_;
481
            dest_addr = &rtcpDestAddr_;
482 483
        } else {
            fd = rtpHandle_;
484
            dest_addr = &rtpDestAddr_;
485
        }
486

487
        auto ret = ff_network_wait_fd(fd);
488 489 490
        if (ret < 0)
            return ret;

491 492
        if (noWrite_)
            return buf_size;
493 494
        return ::sendto(fd, reinterpret_cast<const char*>(buf), buf_size, 0,
                        *dest_addr, dest_addr->getLength());
495
    }
496

497 498 499
    if (noWrite_)
        return buf_size;

500 501 502 503 504
    // IceSocket
    if (isRTCP)
        return rtcp_sock_->send(buf, buf_size);
    else
        return rtp_sock_->send(buf, buf_size);
505 506
}

507 508
int
SocketPair::writeCallback(uint8_t* buf, int buf_size)
509 510
{
    int ret;
511
    bool isRTCP = RTP_PT_IS_RTCP(buf[1]);
Pierre Lespagnol's avatar
Pierre Lespagnol committed
512 513
    unsigned int ts_LSB, ts_MSB;
    double currentSRTS, currentLatency;
514 515 516 517 518 519 520

    // Encrypt?
    if (not isRTCP and srtpContext_ and srtpContext_->srtp_out.aes) {
        buf_size = ff_srtp_encrypt(&srtpContext_->srtp_out, buf,
                                   buf_size, srtpContext_->encryptbuf,
                                   sizeof(srtpContext_->encryptbuf));
        if (buf_size < 0) {
Adrien Béraud's avatar
Adrien Béraud committed
521
            JAMI_WARN("encrypt error %d", buf_size);
522 523 524 525 526 527
            return buf_size;
        }

        buf = srtpContext_->encryptbuf;
    }

528 529 530 531 532 533 534
    // check if we're sending an RR, if so, detect packet loss
    // buf_size gives length of buffer, not just header
    if (isRTCP && static_cast<unsigned>(buf_size) >= sizeof(rtcpRRHeader)) {
        auto header = reinterpret_cast<rtcpRRHeader*>(buf);
        rtcpPacketLoss_ = (header->pt == 201 && ntohl(header->fraction_lost) & RTCP_RR_FRACTION_MASK);
    }

535 536 537 538 539
    do {
        if (interrupted_)
            return -EINTR;
        ret = writeData(buf, buf_size);
    } while (ret < 0 and errno == EAGAIN);
540

Pierre Lespagnol's avatar
Pierre Lespagnol committed
541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570
    if(buf[1] == 200) //Sender Report
    {
        auto header = reinterpret_cast<rtcpSRHeader*>(buf);
        ts_LSB = Swap4Bytes(header->timestampLSB);
        ts_MSB = Swap4Bytes(header->timestampMSB);

        currentSRTS = ts_MSB + (ts_LSB / pow(2,32));

        if(lastSRTS_ != 0 && lastDLSR_ != 0)
        {
            if (histoLatency_.size() >= MAX_LIST_SIZE)
                histoLatency_.pop_front();

            currentLatency = (currentSRTS - lastSRTS_) / 2;
            //JAMI_WARN("Current Latency : %f from sender %X", currentLatency, header->ssrc);
            histoLatency_.push_back(currentLatency);
        }

        lastSRTS_ = currentSRTS;

        // JAMI_WARN("SENDING NEW RTCP SR !! ");

    }
    else if(buf[1] == 201) //Receiver Report
    {
        //auto header = reinterpret_cast<rtcpRRHeader*>(buf);
        //JAMI_WARN("SENDING NEW RTCP RR !! ");

    }

571
    return ret < 0 ? -errno : ret;
572 573
}

Pierre Lespagnol's avatar
Pierre Lespagnol committed
574 575
double
SocketPair::getLastLatency()
576
{
Pierre Lespagnol's avatar
Pierre Lespagnol committed
577 578 579 580
    if(not histoLatency_.empty())
        return histoLatency_.back();
    else
        return -1;
581 582
}

Adrien Béraud's avatar
Adrien Béraud committed
583
} // namespace jami