hydrar/mistserver
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge branch 'master' of projectlivestream.com:pls

Browse source
This commit is contained in:
Erik Zandvliet 2010-12-14 10:41:36 +01:00
commit e6b0cbe634
28 changed files with 420 additions and 2194 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
Buffer/Makefile
View file

@ -1,4 +1,4 @@
SRC = main.cpp ../sockets/sw_base.cpp ../sockets/sw_inet.cpp ../sockets/sw_unix.cpp
SRC = main.cpp
OBJ = $(SRC:.cpp=.o)
OUT = Buffer
INCLUDES =

7
Buffer/buffer.h
View file

@ -1,7 +0,0 @@
#pragma once
struct buffer{
int number;
bool iskeyframe;
FLV_Pack * FLV;
};//buffer

196
Buffer/main.cpp
View file

@ -1,62 +1,146 @@
#include <unistd.h>
#include <fcntl.h>
#include <iostream>
#include "../sockets/SocketW.h"
#include <string>
#include <vector>
#include <cstdlib>
#include <cstdio>
#include <string.h>
#include <unistd.h>
#include <signal.h>
#include "../util/flv.cpp" //FLV format parser
#include "user.cpp"
#include "../util/ddv_socket.cpp" //DDV Socket lib
#include <sys/epoll.h>
int get_empty( user ** list, int amount ) {
for (int i = 0; i < amount; i++ ){
if (!list[i]->is_connected){return i;}
void termination_handler (int signum){
switch (signum){
case SIGPIPE: return; break;
default: return; break;
}
return -1;
}
struct buffer{
int number;
bool iskeyframe;
FLV_Pack * FLV;
buffer(){
number = -1;
iskeyframe = false;
FLV = 0;
}//constructor
};//buffer
class user{
public:
int MyBuffer;
int MyBuffer_num;
int MyBuffer_len;
int MyNum;
int currsend;
void * lastpointer;
static int UserCount;
int s;
user(int fd){
s = fd;
MyNum = UserCount++;
std::cout << "User " << MyNum << " connected" << std::endl;
}//constructor
void Disconnect(std::string reason) {
if (s != -1) {
close(s);
s = -1;
std::cout << "Disconnected user " << MyNum << ": " << reason << std::endl;
}
}//Disconnect
bool doSend(char * buffer, int todo){
int r = send(s, buffer+currsend, todo-currsend, 0);
if (r <= 0){
if ((r < 0) && (errno == EWOULDBLOCK)){return false;}
Disconnect("Connection closed");
return false;
}
currsend += r;
return (currsend == todo);
}
void Send(buffer ** ringbuf, int buffers){
//not connected? cancel
if (s < 0){return;}
//still waiting for next buffer? check it
if (MyBuffer_num < 0){
MyBuffer_num = ringbuf[MyBuffer]->number;
//still waiting? don't crash - wait longer.
if (MyBuffer_num < 0){
return;
}else{
MyBuffer_len = ringbuf[MyBuffer]->FLV->len;
lastpointer = ringbuf[MyBuffer]->FLV->data;
}
}
if (lastpointer != ringbuf[MyBuffer]->FLV->data){
Disconnect("Buffer resize at wrong time... had to disconnect");
return;
}
if (doSend(ringbuf[MyBuffer]->FLV->data, MyBuffer_len)){
//completed a send - switch to next buffer
if ((ringbuf[MyBuffer]->number != MyBuffer_num)){
std::cout << "Warning: User " << MyNum << " was send corrupt video data and send to the next keyframe!" << std::endl;
int nocrashcount = 0;
do{
MyBuffer++;
nocrashcount++;
MyBuffer %= buffers;
}while(!ringbuf[MyBuffer]->FLV->isKeyframe && (nocrashcount < buffers));
if (nocrashcount >= buffers){
std::cout << "Warning: No keyframe found in buffers! Skipping search for now..." << std::endl;
return;
}
}else{
MyBuffer++;
MyBuffer %= buffers;
}
MyBuffer_num = -1;
lastpointer = 0;
currsend = 0;
}//completed a send
}//send
};
int user::UserCount = 0;
int main( int argc, char * argv[] ) {
struct sigaction new_action;
new_action.sa_handler = termination_handler;
sigemptyset (&new_action.sa_mask);
new_action.sa_flags = 0;
sigaction (SIGPIPE, &new_action, NULL);
if (argc < 2) {
std::cout << "usage: " << argv[0] << " buffers_count [streamname]" << std::endl;
return 1;
}
int metabuflen = 0;
char * metabuffer = 0;
int buffers = atoi(argv[1]);
buffer ** ringbuf = (buffer**) calloc (buffers,sizeof(buffer*));
std::vector<user> connectionList;
std::vector<user>::iterator connIt;
for (int i = 0; i < buffers; ++i) ringbuf[i] = new buffer;
int current_buffer = 0;
int lastproper = 0;//last properly finished buffer number
unsigned int loopcount = 0;
SWUnixSocket listener(SWBaseSocket::nonblocking);
SWBaseSocket * incoming = 0;
SWBaseSocket::SWBaseError BError;
std::string shared_socket = "/tmp/shared_socket";
if (argc > 2){
shared_socket = argv[2];
shared_socket = "/tmp/shared_socket_" + shared_socket;
}
unlink(shared_socket.c_str());
listener.bind(shared_socket.c_str());
listener.listen(50);
listener.set_timeout(0,50000);
int metabuflen = 0;
char * metabuffer = 0;
int buffers = atoi(argv[1]);
buffer ** ringbuf = (buffer**) calloc (buffers,sizeof(buffer*));
std::vector<user> users;
std::vector<user>::iterator usersIt;
for (int i = 0; i < buffers; ++i) ringbuf[i] = new buffer;
int current_buffer = 0;
int lastproper = 0;//last properly finished buffer number
unsigned int loopcount = 0;
int listener = DDV_UnixListen(shared_socket, true);
int incoming = 0;
unsigned char packtype;
bool gotVideoInfo = false;
bool gotAudioInfo = false;
//set stdin to be nonblocking
//int flags = fcntl(0, F_GETFL, 0);
//flags |= O_NONBLOCK;
//fcntl(0, F_SETFL, flags);
int infile = fileno(stdin);
int poller = epoll_create(1);
struct epoll_event ev;
@ -69,8 +153,8 @@ int main( int argc, char * argv[] ) {
while(!feof(stdin) && !All_Hell_Broke_Loose){
//invalidate the current buffer
ringbuf[current_buffer]->number = -1;
if ((epoll_wait(poller, events, 1, 100) > 0) && FLV_GetPacket(ringbuf[current_buffer]->FLV)){
loopcount ++;
if ((epoll_wait(poller, events, 1, 10) > 0) && FLV_GetPacket(ringbuf[current_buffer]->FLV)){
loopcount++;
packtype = ringbuf[current_buffer]->FLV->data[0];
//store metadata, if available
if (packtype == 0x12){
@ -113,44 +197,54 @@ int main( int argc, char * argv[] ) {
}
//on keyframe set start point
if (packtype == 0x09){
if (((ringbuf[current_buffer]->FLV->data[11] & 0xf0) >> 4) == 1){lastproper = current_buffer;}
if (((ringbuf[current_buffer]->FLV->data[11] & 0xf0) >> 4) == 1){
lastproper = current_buffer;
}
}
//keep track of buffers
ringbuf[current_buffer]->number = loopcount;
current_buffer++;
current_buffer %= buffers;
ringbuf[current_buffer]->number = loopcount;
}
//check for new connections, accept them if there are any
incoming = listener.accept(&BError);
if (incoming){
connectionList.push_back(user(incoming));
incoming = DDV_Accept(listener, true);
if (incoming >= 0){
users.push_back(incoming);
//send the FLV header
connectionList.back().MyBuffer = lastproper;
connectionList.back().MyBuffer_num = -1;
users.back().currsend = 0;
users.back().MyBuffer = lastproper;
users.back().MyBuffer_num = -1;
//TODO: Do this more nicely?
if (connectionList.back().Conn->send(FLVHeader,13,&BError) != 13){
connectionList.back().disconnect("failed to receive the header!");
if (!DDV_write(FLVHeader, 13, incoming)){
users.back().Disconnect("failed to receive the header!");
}else{
if (connectionList.back().Conn->send(metabuffer,metabuflen,&BError) != metabuflen){
connectionList.back().disconnect("failed to receive metadata!");
if (!DDV_write(metabuffer, metabuflen, incoming)){
users.back().Disconnect("failed to receive metadata!");
}
}
if (BError != SWBaseSocket::ok){
connectionList.back().disconnect("Socket error: " + BError.get_error());
}
}
//send all connections what they need, if and when they need it
if (connectionList.size() > 0){
for (connIt = connectionList.begin(); connIt != connectionList.end(); connIt++){
if (!(*connIt).is_connected){connectionList.erase(connIt);break;}
(*connIt).Send(ringbuf, buffers);
if (users.size() > 0){
for (usersIt = users.begin(); usersIt != users.end(); usersIt++){
if ((*usersIt).s == -1){
users.erase(usersIt); break;
}else{
(*usersIt).Send(ringbuf, buffers);
}
}
}
}//main loop
// disconnect listener
std::cout << "Reached EOF of input" << std::endl;
listener.disconnect(&BError);
close(listener);
while (users.size() > 0){
for (usersIt = users.begin(); usersIt != users.end(); usersIt++){
(*usersIt).Disconnect("Shutting down...");
if ((*usersIt).s == -1){users.erase(usersIt);break;}
}
}
return 0;
}

2
Buffer/playh264.sh
View file

@ -5,5 +5,5 @@
#ffmpeg -y -i "$1" -ar 44100 -vcodec libx264 -b 1000k -g 150 -r 20 -f flv - | ./Buffer 500
ffmpeg -i "$1" -re -acodec aac -ar 11025 -vcodec libx264 -b 700k -vpre ultrafast -refs 1 -bf 0 -g 150 -f flv - 2> /dev/null | ./Buffer 500
ffmpeg -i "$1" -re -acodec aac -ar 11025 -vcodec libx264 -b 700k -vpre ultrafast -refs 1 -bf 0 -g 150 -f flv - 2> /dev/null | ./Buffer 500 $2

84
Buffer/user.cpp
View file

@ -1,84 +0,0 @@
#include "buffer.h"
#include "../sockets/SocketW.h"
#include <iostream>
class user{
public:
user(SWBaseSocket * newConn);
void disconnect(std::string reason);
void Send(buffer ** ringbuf, int buffers);
bool is_connected;
SWUnixSocket * Conn;
int MyBuffer;
int MyBuffer_num;
int MyBuffer_len;
int MyNum;
void * lastpointer;
static int UserCount;
static SWBaseSocket::SWBaseError err;
};//user
int user::UserCount = 0;
SWBaseSocket::SWBaseError user::err;
user::user(SWBaseSocket * newConn) {
Conn = (SWUnixSocket*)newConn;
is_connected = (Conn != 0);
MyNum = UserCount++;
std::cout << "User " << MyNum << " connected" << std::endl;
}
void user::disconnect(std::string reason) {
if (Conn) {
Conn->disconnect(&err);
Conn = NULL;
std::cout << "Disconnected user " << MyNum << ": " << reason << std::endl;
}
is_connected = false;
}
void user::Send(buffer ** ringbuf, int buffers){
//not connected? cancel
if (!is_connected){return;}
//still waiting for next buffer? check it
if (MyBuffer_num < 0){
MyBuffer_num = ringbuf[MyBuffer]->number;
//still waiting? don't crash - wait longer.
if (MyBuffer_num < 0){
return;
}else{
MyBuffer_len = ringbuf[MyBuffer]->FLV->len;
lastpointer = ringbuf[MyBuffer]->FLV->data;
}
}
if (lastpointer != ringbuf[MyBuffer]->FLV->data){
disconnect("Buffer resize at wrong time... had to disconnect");
return;
}
int ret = Conn->fsend(ringbuf[MyBuffer]->FLV->data, MyBuffer_len, &err);
if ((err != SWBaseSocket::ok) && (err != SWBaseSocket::notReady)){
disconnect("Socket error: " + err.get_error());
return;
}
if (ret == MyBuffer_len){
//completed a send - switch to next buffer
if ((ringbuf[MyBuffer]->number != MyBuffer_num)){
std::cout << "Warning: User " << MyNum << " was send corrupt video data and send to the next keyframe!" << std::endl;
int nocrashcount = 0;
do{
MyBuffer++;
nocrashcount++;
MyBuffer %= buffers;
}while(!ringbuf[MyBuffer]->FLV->isKeyframe && (nocrashcount < buffers));
if (nocrashcount >= buffers){
std::cout << "Warning: No keyframe found in buffers! Skipping search for now..." << std::endl;
return;
}
}else{
MyBuffer++;
MyBuffer %= buffers;
}
MyBuffer_num = -1;
lastpointer = 0;
}
}

34
Connector_RTMP/Conn_RTMP
View file

@ -1,32 +1,22 @@
#!/bin/sh
#
# chkconfig: 345 92 8
# description: DDVTech RTMP Connector
#
# processname: Connector_RTMP
. /etc/rc.d/init.d/functions
prog="Connector_RTMP"
fullprog="/usr/bin/Connector_RTMP"
RETVAL=0
start() {
gprintf "Starting %s: " $prog
daemon --user=root $fullprog
RETVAL=$?
echo
[ $RETVAL -eq 0 ] && touch /var/lock/subsys/$prog
return $RETVAL
echo "Starting $prog"
$fullprog
return $?
}
stop() {
gprintf "Stopping %s: " $prog
killproc $fullprog
RETVAL=$?
echo
[ $RETVAL -eq 0 ] && rm -f /var/lock/subsys/$prog
return $RETVAL
echo "Stopping $prog"
killall $prog
return $?
}
case "$1" in
@ -40,18 +30,8 @@ case "$1" in
stop
start
;;
condrestart)
if [ -f /var/lock/subsys/$prog ]; then
stop
start
fi
;;
status)
status $fullprog
RETVAL=$?
;;
*)
gprintf "Usage: %s {start|stop|restart|status}" $0
echo "Usage: $0 {start|stop|restart}"
RETVAL=1
esac

4
Connector_RTMP/Makefile
View file

@ -1,4 +1,4 @@
SRC = main.cpp ../sockets/sw_base.cpp ../sockets/sw_inet.cpp ../sockets/sw_unix.cpp
SRC = main.cpp
OBJ = $(SRC:.cpp=.o)
OUT = Connector_RTMP
INCLUDES =
@ -11,7 +11,7 @@ LIBS = -lssl -lcrypto
.PHONY: clean default
default: $(OUT)
.cpp.o:
$(CC) $(INCLUDES) $(CCFLAGS) $(LIBS) -c $< -o $@
$(CC) $(INCLUDES) $(CCFLAGS) -c $< -o $@
$(OUT): $(OBJ) chunkstream.cpp parsechunks.cpp handshake.cpp crypto.cpp amf.cpp
$(CC) $(LIBS) -o $(OUT) $(OBJ)
clean:

2
Connector_RTMP/amf.cpp
View file

@ -270,7 +270,7 @@ AMFType parseOneAMF(const unsigned char *& data, unsigned int &len, unsigned int
return ret;
} break;
}
#ifdef DEBUG
#if DEBUG >= 2
fprintf(stderr, "Error: Unimplemented AMF type %hhx - returning.\n", data[i]);
#endif
return AMFType("error", (unsigned char)0xFF);

2
Connector_RTMP/chunkstream.cpp
View file

@ -457,7 +457,7 @@ chunkpack * AddChunkPart(chunkpack newchunk){
p = it->second;
tmpdata = (unsigned char*)realloc(p->data, p->real_len + newchunk.real_len);
if (tmpdata == 0){
#ifdef DEBUG
#if DEBUG >= 1
fprintf(stderr, "Error allocating memory!\n");
#endif
return 0;

2
Connector_RTMP/crypto.cpp
View file

@ -497,7 +497,7 @@ bool ValidateClientScheme(uint8_t * pBuffer, uint8_t scheme) {
uint8_t *pTempHash = new uint8_t[512];
HMACsha256(pTempBuffer, 1536 - 32, genuineFPKey, 30, pTempHash);
bool result = (memcmp(pBuffer+clientDigestOffset, pTempHash, 32) == 0);
#ifdef DEBUG
#if DEBUG >= 4
fprintf(stderr, "Client scheme validation %hhi %s\n", scheme, result?"success":"failed");
#endif
delete[] pTempBuffer;

8
Connector_RTMP/handshake.cpp
View file

@ -1,6 +1,6 @@
#undef OLDHANDSHAKE //change to #define for old handshake method
char versionstring[] = "PLSRTMPServer";
char versionstring[] = "WWW.DDVTECH.COM ";
#ifdef OLDHANDSHAKE
struct Handshake {
@ -24,7 +24,7 @@ bool doHandshake(){
Server.Time[0] = 0; Server.Time[1] = 0; Server.Time[2] = 0; Server.Time[3] = 0;
Server.Zero[0] = 0; Server.Zero[1] = 0; Server.Zero[2] = 0; Server.Zero[3] = 0;
for (int i = 0; i < 1528; i++){
Server.Random[i] = versionstring[i%13];
Server.Random[i] = versionstring[i%sizeof(versionstring)];
}
/** Send S0 **/
DDV_write(&(Version), 1, 1, CONN_fd);
@ -69,14 +69,14 @@ bool doHandshake(){
for (int i = 8; i < 3072; ++i){Server[i] = versionstring[i%13];}//"random" data
bool encrypted = (Version == 6);
#ifdef DEBUG
#if DEBUG >= 4
fprintf(stderr, "Handshake version is %hhi\n", Version);
#endif
uint8_t _validationScheme = 5;
if (ValidateClientScheme(Client, 0)) _validationScheme = 0;
if (ValidateClientScheme(Client, 1)) _validationScheme = 1;
#ifdef DEBUG
#if DEBUG >= 4
fprintf(stderr, "Handshake type is %hhi, encryption is %s\n", _validationScheme, encrypted?"on":"off");
#endif

182
Connector_RTMP/main.cpp
View file

@ -1,4 +1,10 @@
#undef DEBUG
//debugging level 0 = nothing
//debugging level 1 = critical errors
//debugging level 2 = errors
//debugging level 3 = status information
//debugging level 4 = extremely verbose status information
#define DEBUG 3
#include <iostream>
#include <cstdlib>
#include <cstdio>
@ -8,6 +14,7 @@
#include <sys/types.h>
#include <sys/wait.h>
#include <sys/epoll.h>
#include <getopt.h>
//for connection to server
bool ready4data = false;//set to true when streaming starts
@ -27,6 +34,12 @@ int server_socket = 0;
void termination_handler (int signum){
if (server_socket == 0) return;
switch (signum){
case SIGINT: break;
case SIGHUP: break;
case SIGTERM: break;
default: return; break;
}
close(server_socket);
server_socket = 0;
}
@ -37,13 +50,48 @@ int main(int argc, char ** argv){
new_action.sa_handler = termination_handler;
sigemptyset (&new_action.sa_mask);
new_action.sa_flags = 0;
sigaction (SIGINT, &new_action, NULL);
sigaction (SIGHUP, &new_action, NULL);
sigaction (SIGTERM, &new_action, NULL);
sigaction(SIGINT, &new_action, NULL);
sigaction(SIGHUP, &new_action, NULL);
sigaction(SIGTERM, &new_action, NULL);
sigaction(SIGPIPE, &new_action, NULL);
server_socket = DDV_Listen(1936);
if ((argc < 2) || (argv[1] == "nd")){
if (server_socket > 0){daemon(1, 0);}else{return 1;}
int listen_port = 1935;
bool daemon_mode = true;
int opt = 0;
static const char *optString = "np:h?";
static const struct option longOpts[] = {
{"help",0,0,'h'},
{"port",1,0,'p'},
{"no-daemon",0,0,'n'}
};
while ((opt = getopt_long(argc, argv, optString, longOpts, 0)) != -1){
switch (opt){
case 'p':
listen_port = atoi(optarg);
break;
case 'n':
daemon_mode = false;
break;
case 'h':
case '?':
printf("Options: -h[elp], -?, -n[o-daemon], -p[ort] #\n");
return 1;
break;
}
}
server_socket = DDV_Listen(listen_port);
fprintf(stderr, "Made a listening socket on port %i...\n", listen_port);
if (server_socket > 0){
if (daemon_mode){
daemon(1, 0);
fprintf(stderr, "Going into background mode...");
}
}else{
fprintf(stderr, "Error: could not make listening socket");
return 1;
}
int status;
while (server_socket > 0){
@ -54,7 +102,7 @@ int main(int argc, char ** argv){
if (myid == 0){
break;
}else{
printf("Spawned new process %i for handling socket %i\n", (int)myid, CONN_fd);
fprintf(stderr, "Spawned new process %i for handling socket %i\n", (int)myid, CONN_fd);
}
}
}
@ -69,25 +117,19 @@ int main(int argc, char ** argv){
FLV_Pack * tag = 0;
//first timestamp set
int lastcheck = getNowMS();
firsttime = getNowMS();
#ifdef DEBUG
fprintf(stderr, "Doing handshake...\n");
#endif
if (doHandshake()){
#ifdef DEBUG
#if DEBUG >= 4
fprintf(stderr, "Handshake succcess!\n");
#endif
}else{
#ifdef DEBUG
#if DEBUG >= 1
fprintf(stderr, "Handshake fail!\n");
#endif
return 0;
}
#ifdef DEBUG
fprintf(stderr, "Starting processing...\n");
#endif
int retval;
int poller = epoll_create(1);
@ -104,10 +146,20 @@ int main(int argc, char ** argv){
while (!socketError && !All_Hell_Broke_Loose){
//only parse input if available or not yet init'ed
//rightnow = getNowMS();
retval = epoll_wait(poller, events, 1, 0);
if ((retval > 0) || !ready4data || (snd_cnt - snd_window_at >= snd_window_size)){
if (DDV_ready(CONN_fd)){
parseChunk();
retval = epoll_wait(poller, events, 1, 1);
if ((retval > 0) || !ready4data){// || (snd_cnt - snd_window_at >= snd_window_size)
switch (DDV_ready(CONN_fd)){
case 0:
socketError = true;
#if DEBUG >= 1
fprintf(stderr, "User socket is disconnected.\n");
#endif
break;
case -1: break;//not ready yet
default:
parseChunk();
lastcheck = getNowMS();
break;
}
}
if (ready4data){
@ -115,59 +167,81 @@ int main(int argc, char ** argv){
//we are ready, connect the socket!
ss = DDV_OpenUnix(streamname);
if (ss <= 0){
#ifdef DEBUG
#if DEBUG >= 1
fprintf(stderr, "Could not connect to server!\n");
#endif
return 0;
socketError = 1;
break;
}
ev.events = EPOLLIN;
ev.data.fd = ss;
epoll_ctl(sspoller, EPOLL_CTL_ADD, ss, &ev);
#ifdef DEBUG
#if DEBUG >= 3
fprintf(stderr, "Everything connected, starting to send video data...\n");
#endif
inited = true;
}
retval = epoll_wait(poller, events, 1, 50);
if (DDV_ready(ss)){
if (FLV_GetPacket(tag, ss)){//able to read a full packet?
ts = tag->data[7] * 256*256*256;
ts += tag->data[4] * 256*256;
ts += tag->data[5] * 256;
ts += tag->data[6];
if (ts != 0){
if (fts == 0){fts = ts;ftst = getNowMS();}
ts -= fts;
tag->data[7] = ts / (256*256*256);
tag->data[4] = ts / (256*256);
tag->data[5] = ts / 256;
tag->data[6] = ts % 256;
ts += ftst;
}else{
ftst = getNowMS();
tag->data[7] = ftst / (256*256*256);
tag->data[4] = ftst / (256*256);
tag->data[5] = ftst / 256;
tag->data[6] = ftst % 256;
}
SendMedia((unsigned char)tag->data[0], (unsigned char *)tag->data+11, tag->len-15, ts);
#ifdef DEBUG
fprintf(stderr, "Sent a tag to %i\n", CONN_fd);
retval = epoll_wait(poller, events, 1, 1);
switch (DDV_ready(ss)){
case 0:
socketError = true;
#if DEBUG >= 1
fprintf(stderr, "Source socket is disconnected.\n");
#endif
}
break;
case -1: break;//not ready yet
default:
if (FLV_GetPacket(tag, ss)){//able to read a full packet?
ts = tag->data[7] * 256*256*256;
ts += tag->data[4] * 256*256;
ts += tag->data[5] * 256;
ts += tag->data[6];
if (ts != 0){
if (fts == 0){fts = ts;ftst = getNowMS();}
ts -= fts;
tag->data[7] = ts / (256*256*256);
tag->data[4] = ts / (256*256);
tag->data[5] = ts / 256;
tag->data[6] = ts % 256;
ts += ftst;
}else{
ftst = getNowMS();
tag->data[7] = ftst / (256*256*256);
tag->data[4] = ftst / (256*256);
tag->data[5] = ftst / 256;
tag->data[6] = ftst % 256;
}
SendMedia((unsigned char)tag->data[0], (unsigned char *)tag->data+11, tag->len-15, ts);
lastcheck = getNowMS();
#if DEBUG >= 4
fprintf(stderr, "Sent a tag to %i\n", CONN_fd);
#endif
}
break;
}
}
//send ACK if we received a whole window
if (rec_cnt - rec_window_at > rec_window_size){
if ((rec_cnt - rec_window_at > rec_window_size) || (getNowMS() - lastcheck > 1)){
rec_window_at = rec_cnt;
SendCTL(3, rec_cnt);//send ack (msg 3)
lastcheck = getNowMS();
}
}
//#ifdef DEBUG
if (socketError){fprintf(stderr, "socketError\n");}
close(CONN_fd);
if (inited) close(ss);
#if DEBUG >= 1
if (All_Hell_Broke_Loose){fprintf(stderr, "All Hell Broke Loose\n");}
fprintf(stderr, "User %i disconnected.\n", CONN_fd);
//#endif
if (inited){
fprintf(stderr, "Status was: inited\n");
}else{
if (ready4data){
fprintf(stderr, "Status was: ready4data\n");
}else{
fprintf(stderr, "Status was: connected\n");
}
}
#endif
return 0;
}//main

53
Connector_RTMP/parsechunks.cpp
View file

@ -13,25 +13,25 @@ void parseChunk(){
break;//happens when connection breaks unexpectedly
case 1://set chunk size
chunk_rec_max = ntohl(*(int*)next.data);
#ifdef DEBUG
#if DEBUG >= 4
fprintf(stderr, "CTRL: Set chunk size: %i\n", chunk_rec_max);
#endif
break;
case 2://abort message - we ignore this one
#ifdef DEBUG
#if DEBUG >= 4
fprintf(stderr, "CTRL: Abort message\n");
#endif
//4 bytes of stream id to drop
break;
case 3://ack
#ifdef DEBUG
#if DEBUG >= 4
fprintf(stderr, "CTRL: Acknowledgement\n");
#endif
snd_window_at = ntohl(*(int*)next.data);
snd_window_at = snd_cnt;
break;
case 4:{
#ifdef DEBUG
#if DEBUG >= 4
short int ucmtype = ntohs(*(short int*)next.data);
fprintf(stderr, "CTRL: User control message %hi\n", ucmtype);
#endif
@ -47,7 +47,7 @@ void parseChunk(){
//we don't need to process this
} break;
case 5://window size of other end
#ifdef DEBUG
#if DEBUG >= 4
fprintf(stderr, "CTRL: Window size\n");
#endif
rec_window_size = ntohl(*(int*)next.data);
@ -55,7 +55,7 @@ void parseChunk(){
SendCTL(3, rec_cnt);//send ack (msg 3)
break;
case 6:
#ifdef DEBUG
#if DEBUG >= 4
fprintf(stderr, "CTRL: Set peer bandwidth\n");
#endif
//4 bytes window size, 1 byte limit type (ignored)
@ -63,49 +63,45 @@ void parseChunk(){
SendCTL(5, snd_window_size);//send window acknowledgement size (msg 5)
break;
case 8:
#ifdef DEBUG
#if DEBUG >= 4
fprintf(stderr, "Received audio data\n");
#endif
break;
case 9:
#ifdef DEBUG
#if DEBUG >= 4
fprintf(stderr, "Received video data\n");
#endif
break;
case 15:
#ifdef DEBUG
#if DEBUG >= 4
fprintf(stderr, "Received AFM3 data message\n");
#endif
break;
case 16:
#ifdef DEBUG
#if DEBUG >= 4
fprintf(stderr, "Received AFM3 shared object\n");
#endif
break;
case 17:
#ifdef DEBUG
#if DEBUG >= 4
fprintf(stderr, "Received AFM3 command message\n");
#endif
break;
case 18:
#ifdef DEBUG
#if DEBUG >= 4
fprintf(stderr, "Received AFM0 data message\n");
#endif
break;
case 19:
#ifdef DEBUG
#if DEBUG >= 4
fprintf(stderr, "Received AFM0 shared object\n");
#endif
break;
case 20:{//AMF0 command message
bool parsed = false;
amfdata = parseAMF(next.data, next.real_len);
#ifdef DEBUG
fprintf(stderr, "Received AFM0 command message:\n");
amfdata.Print();
#endif
if (amfdata.getContentP(0)->StrValue() == "connect"){
#ifdef DEBUG
#if DEBUG >= 4
int tmpint;
tmpint = amfdata.getContentP(2)->getContentP("videoCodecs")->NumValue();
if (tmpint & 0x04){fprintf(stderr, "Sorensen video support detected\n");}
@ -149,9 +145,6 @@ void parseChunk(){
amfreply.addContent(AMFType("", (double)1));//stream ID - we use 1
SendChunk(3, 20, next.msg_stream_id, amfreply.Pack());
SendUSR(0, 0);//send UCM StreamBegin (0), stream 0
#ifdef DEBUG
fprintf(stderr, "AMF0 command: createStream result\n");
#endif
parsed = true;
}//createStream
if ((amfdata.getContentP(0)->StrValue() == "getStreamLength") || (amfdata.getContentP(0)->StrValue() == "getMovLen")){
@ -162,9 +155,6 @@ void parseChunk(){
amfreply.addContent(AMFType("", (double)0, 0x05));//null - command info
amfreply.addContent(AMFType("", (double)0));//zero length
SendChunk(3, 20, next.msg_stream_id, amfreply.Pack());
#ifdef DEBUG
fprintf(stderr, "AMF0 command: getStreamLength result\n");
#endif
parsed = true;
}//getStreamLength
if (amfdata.getContentP(0)->StrValue() == "checkBandwidth"){
@ -175,9 +165,6 @@ void parseChunk(){
amfreply.addContent(AMFType("", (double)0, 0x05));//null - command info
amfreply.addContent(AMFType("", (double)0, 0x05));//null - command info
SendChunk(3, 20, 1, amfreply.Pack());
#ifdef DEBUG
fprintf(stderr, "AMF0 command: checkBandwidth result\n");
#endif
parsed = true;
}//checkBandwidth
if ((amfdata.getContentP(0)->StrValue() == "play") || (amfdata.getContentP(0)->StrValue() == "play2")){
@ -220,24 +207,24 @@ void parseChunk(){
chunk_snd_max = 1024*1024;
SendCTL(1, chunk_snd_max);//send chunk size max (msg 1)
ready4data = true;//start sending video data!
#ifdef DEBUG
fprintf(stderr, "AMF0 command: play result (%s)\n", streamname.c_str());
#endif
parsed = true;
}//createStream
#if DEBUG >= 3
fprintf(stderr, "AMF0 command: %s\n", amfdata.getContentP(0)->StrValue().c_str());
#endif
if (!parsed){
#ifdef DEBUG
#if DEBUG >= 2
fprintf(stderr, "AMF0 command not processed! :(\n");
#endif
}
} break;
case 22:
#ifdef DEBUG
#if DEBUG >= 4
fprintf(stderr, "Received aggregate message\n");
#endif
break;
default:
#ifdef DEBUG
#if DEBUG >= 1
fprintf(stderr, "Unknown chunk received! Probably protocol corruption, stopping parsing of incoming data.\n");
#endif
stopparsing = true;

23
Connector_RTMPf/Makefile
View file

@ -1,23 +0,0 @@
SRC = main.cpp ../sockets/sw_base.cpp ../sockets/sw_inet.cpp ../sockets/sw_unix.cpp
OBJ = $(SRC:.cpp=.o)
OUT = Connector_RTMPf
INCLUDES =
CCFLAGS = -Wall -Wextra -funsigned-char -g
CC = $(CROSS)g++
LD = $(CROSS)ld
AR = $(CROSS)ar
LIBS = -lssl -lcrypto
.SUFFIXES: .cpp
.PHONY: clean default
default: $(OUT)
.cpp.o:
$(CC) $(INCLUDES) $(CCFLAGS) $(LIBS) -c $< -o $@
$(OUT): $(OBJ) chunkstream.cpp parsechunks.cpp handshake.cpp crypto.cpp amf.cpp
$(CC) $(LIBS) -o $(OUT) $(OBJ)
clean:
rm -rf $(OBJ) $(OUT) Makefile.bak *~
run-test: $(OUT)
rm -rf ./meh
mkfifo ./meh
cat ./meh &
nc -l -p 1935 -e './Connector_RTMPf 2>./meh'

285
Connector_RTMPf/amf.cpp
View file

@ -1,285 +0,0 @@
#include <vector>
#include <string.h>
#include <string>
class AMFType {
public:
std::string Indice(){return myIndice;};
unsigned char GetType(){return myType;};
double NumValue(){return numval;};
std::string StrValue(){return strval;};
const char * Str(){return strval.c_str();};
int hasContent(){
if (!contents){return 0;}
return contents->size();
};
void addContent(AMFType c){if (contents != 0){contents->push_back(c);}};
AMFType* getContentP(int i){if (contents != 0){return &contents->at(i);}else{return 0;}};
AMFType getContent(int i){if (contents != 0){return contents->at(i);}else{return AMFType("error");}};
AMFType* getContentP(std::string s){
if (contents != 0){
for (std::vector<AMFType>::iterator it = contents->begin(); it != contents->end(); it++){
if (it->Indice() == s){
return &(*it);
}
}
}
return this;
};
AMFType getContent(std::string s){
if (contents != 0){
for (std::vector<AMFType>::iterator it = contents->begin(); it != contents->end(); it++){
if (it->Indice() == s){
return *it;
}
}
}
return AMFType("error");
};
AMFType(std::string indice, double val, unsigned char setType = 0x00){//num type initializer
myIndice = indice;
myType = setType;
strval = "";
numval = val;
contents = 0;
};
AMFType(std::string indice, std::string val, unsigned char setType = 0x02){//str type initializer
myIndice = indice;
myType = setType;
strval = val;
numval = 0;
contents = 0;
};
AMFType(std::string indice, unsigned char setType = 0x03){//object type initializer
myIndice = indice;
myType = setType;
strval = "";
numval = 0;
contents = new std::vector<AMFType>;
};
~AMFType(){if (contents != 0){delete contents;contents=0;}};
AMFType& operator=(const AMFType &a) {
myIndice = a.myIndice;
myType = a.myType;
strval = a.strval;
numval = a.numval;
if (contents){
if (a.contents != contents){
delete contents;
if (a.contents){
contents = new std::vector<AMFType>;
for (std::vector<AMFType>::iterator it = a.contents->begin(); it < a.contents->end(); it++){
contents->push_back(*it);
}
}else{
contents = 0;
}
}
}else{
if (a.contents){
contents = new std::vector<AMFType>;
for (std::vector<AMFType>::iterator it = a.contents->begin(); it < a.contents->end(); it++){
contents->push_back(*it);
}
}
}
return *this;
};//= operator
AMFType(const AMFType &a){
myIndice = a.myIndice;
myType = a.myType;
strval = a.strval;
numval = a.numval;
if (a.contents){
contents = new std::vector<AMFType>;
for (std::vector<AMFType>::iterator it = a.contents->begin(); it < a.contents->end(); it++){
contents->push_back(*it);
}
}else{contents = 0;}
};//copy constructor
void Print(std::string indent = ""){
std::cerr << indent;
switch (myType){
case 0x00: std::cerr << "Number"; break;
case 0x01: std::cerr << "Bool"; break;
case 0x02://short string
case 0x0C: std::cerr << "String"; break;
case 0x03: std::cerr << "Object"; break;
case 0x08: std::cerr << "ECMA Array"; break;
case 0x05: std::cerr << "Null"; break;
case 0x06: std::cerr << "Undefined"; break;
case 0x0D: std::cerr << "Unsupported"; break;
case 0xFF: std::cerr << "Container"; break;
}
std::cerr << " " << myIndice << " ";
switch (myType){
case 0x00: case 0x01: std::cerr << numval; break;
case 0x02: case 0x0C: std::cerr << strval; break;
}
std::cerr << std::endl;
if (contents){
for (std::vector<AMFType>::iterator it = contents->begin(); it != contents->end(); it++){it->Print(indent+" ");}
}
};//print
std::string Pack(){
std::string r = "";
if ((myType == 0x02) && (strval.size() > 0xFFFF)){myType = 0x0C;}
if (myType != 0xFF){r += myType;}
switch (myType){
case 0x00://number
r += *(((char*)&numval)+7); r += *(((char*)&numval)+6);
r += *(((char*)&numval)+5); r += *(((char*)&numval)+4);
r += *(((char*)&numval)+3); r += *(((char*)&numval)+2);
r += *(((char*)&numval)+1); r += *(((char*)&numval));
break;
case 0x01://bool
r += (char)numval;
break;
case 0x02://short string
r += strval.size() / 256;
r += strval.size() % 256;
r += strval;
break;
case 0x0C://long string
r += strval.size() / (256*256*256);
r += strval.size() / (256*256);
r += strval.size() / 256;
r += strval.size() % 256;
r += strval;
break;
case 0x03://object
if (contents){
for (std::vector<AMFType>::iterator it = contents->begin(); it != contents->end(); it++){
r += it->Indice().size() / 256;
r += it->Indice().size() % 256;
r += it->Indice();
r += it->Pack();
}
}
r += (char)0; r += (char)0; r += (char)9;
break;
case 0x08:{//array
int arrlen = 0;
if (contents){
arrlen = getContentP("length")->NumValue();
r += arrlen / (256*256*256); r += arrlen / (256*256); r += arrlen / 256; r += arrlen % 256;
for (std::vector<AMFType>::iterator it = contents->begin(); it != contents->end(); it++){
r += it->Indice().size() / 256;
r += it->Indice().size() % 256;
r += it->Indice();
r += it->Pack();
}
}else{
r += (char)0; r += (char)0; r += (char)0; r += (char)0;
}
r += (char)0; r += (char)0; r += (char)9;
} break;
case 0xFF://container - our own type - do not send, only send contents
if (contents){
for (std::vector<AMFType>::iterator it = contents->begin(); it != contents->end(); it++){
r += it->Pack();
}
}
break;
}
return r;
};//pack
protected:
std::string myIndice;
unsigned char myType;
std::string strval;
double numval;
std::vector<AMFType> * contents;
};//AMFType
AMFType parseOneAMF(const unsigned char *& data, unsigned int &len, unsigned int &i, std::string name){
char * helperchar = 0;
std::string tmpstr;
unsigned int tmpi = 0;
unsigned char tmpdbl[8];
switch (data[i]){
case 0x00://number
tmpdbl[7] = data[i+1];
tmpdbl[6] = data[i+2];
tmpdbl[5] = data[i+3];
tmpdbl[4] = data[i+4];
tmpdbl[3] = data[i+5];
tmpdbl[2] = data[i+6];
tmpdbl[1] = data[i+7];
tmpdbl[0] = data[i+8];
i+=9;
return AMFType(name, *(double*)tmpdbl, 0x00);
break;
case 0x01://bool
i+=2;
if (data[i-1] == 0){
return AMFType(name, (double)0, 0x01);
}else{
return AMFType(name, (double)1, 0x01);
}
break;
case 0x0C://long string
tmpi = data[i+1]*256*256*256+data[i+2]*256*256+data[i+3]*256+data[i+4];
helperchar = (char*)malloc(tmpi+1);
memcpy(helperchar, data+i+5, tmpi);
helperchar[tmpi] = 0;
tmpstr = helperchar;
free(helperchar);
i += tmpi + 5;
return AMFType(name, tmpstr, 0x0C);
break;
case 0x02://string
tmpi = data[i+1]*256+data[i+2];
helperchar = (char*)malloc(tmpi+1);
memcpy(helperchar, data+i+3, tmpi);
helperchar[tmpi] = 0;
tmpstr = helperchar;
free(helperchar);
i += tmpi + 3;
return AMFType(name, tmpstr, 0x02);
break;
case 0x05://null
case 0x06://undefined
case 0x0D://unsupported
++i;
return AMFType(name, (double)0, data[i-1]);
break;
case 0x03:{//object
++i;
AMFType ret = AMFType(name, data[i-1]);
while (data[i] + data[i+1] != 0){
tmpi = data[i]*256+data[i+1];
tmpstr = (char*)(data+i+2);
i += tmpi + 2;
ret.addContent(parseOneAMF(data, len, i, tmpstr));
}
i += 3;
return ret;
} break;
case 0x08:{//ECMA array
++i;
AMFType ret = AMFType(name, data[i-1]);
i += 4;
while (data[i] + data[i+1] != 0){
tmpi = data[i]*256+data[i+1];
tmpstr = (char*)(data+i+2);
i += tmpi + 2;
ret.addContent(parseOneAMF(data, len, i, tmpstr));
}
i += 3;
return ret;
} break;
}
#ifdef DEBUG
fprintf(stderr, "Error: Unimplemented AMF type %hhx - returning.\n", data[i]);
#endif
return AMFType("error", (unsigned char)0xFF);
}//parseOneAMF
AMFType parseAMF(const unsigned char * data, unsigned int len){
AMFType ret("returned", (unsigned char)0xFF);//container type
unsigned int i = 0;
while (i < len){ret.addContent(parseOneAMF(data, len, i, ""));}
return ret;
}//parseAMF
AMFType parseAMF(std::string data){return parseAMF((const unsigned char*)data.c_str(), data.size());}

501
Connector_RTMPf/chunkstream.cpp
View file

@ -1,501 +0,0 @@
#include <map>
#include <string.h>
#include <stdlib.h>
#include <sys/time.h>
#include <arpa/inet.h>
unsigned int getNowMS(){
timeval t;
gettimeofday(&t, 0);
return t.tv_sec + t.tv_usec/1000;
}
unsigned int chunk_rec_max = 128;
unsigned int chunk_snd_max = 128;
unsigned int rec_window_size = 0xFA00;
unsigned int snd_window_size = 1024*500;
unsigned int rec_window_at = 0;
unsigned int snd_window_at = 0;
unsigned int rec_cnt = 0;
unsigned int snd_cnt = 0;
unsigned int firsttime;
struct chunkinfo {
unsigned int cs_id;
unsigned int timestamp;
unsigned int len;
unsigned int real_len;
unsigned int len_left;
unsigned char msg_type_id;
unsigned int msg_stream_id;
};//chunkinfo
struct chunkpack {
unsigned char chunktype;
unsigned int cs_id;
unsigned int timestamp;
unsigned int len;
unsigned int real_len;
unsigned int len_left;
unsigned char msg_type_id;
unsigned int msg_stream_id;
unsigned char * data;
};//chunkpack
//clean a chunk so that it may be re-used without memory leaks
void scrubChunk(struct chunkpack c){
if (c.data){free(c.data);}
c.data = 0;
c.real_len = 0;
}//scrubChunk
//ugly global, but who cares...
std::map<unsigned int, chunkinfo> prevmap;
//return previous packet of this cs_id
chunkinfo GetPrev(unsigned int cs_id){
return prevmap[cs_id];
}//GetPrev
//store packet information of last packet of this cs_id
void PutPrev(chunkpack prev){
prevmap[prev.cs_id].timestamp = prev.timestamp;
prevmap[prev.cs_id].len = prev.len;
prevmap[prev.cs_id].real_len = prev.real_len;
prevmap[prev.cs_id].len_left = prev.len_left;
prevmap[prev.cs_id].msg_type_id = prev.msg_type_id;
prevmap[prev.cs_id].msg_stream_id = prev.msg_stream_id;
}//PutPrev
//ugly global, but who cares...
std::map<unsigned int, chunkinfo> sndprevmap;
//return previous packet of this cs_id
chunkinfo GetSndPrev(unsigned int cs_id){
return sndprevmap[cs_id];
}//GetPrev
//store packet information of last packet of this cs_id
void PutSndPrev(chunkpack prev){
sndprevmap[prev.cs_id].cs_id = prev.cs_id;
sndprevmap[prev.cs_id].timestamp = prev.timestamp;
sndprevmap[prev.cs_id].len = prev.len;
sndprevmap[prev.cs_id].real_len = prev.real_len;
sndprevmap[prev.cs_id].len_left = prev.len_left;
sndprevmap[prev.cs_id].msg_type_id = prev.msg_type_id;
sndprevmap[prev.cs_id].msg_stream_id = prev.msg_stream_id;
}//PutPrev
//sends the chunk over the network
void SendChunk(chunkpack ch){
unsigned char tmp;
unsigned int tmpi;
unsigned char chtype = 0x00;
chunkinfo prev = GetSndPrev(ch.cs_id);
ch.timestamp -= firsttime;
if (prev.cs_id == ch.cs_id){
if (ch.msg_stream_id == prev.msg_stream_id){
chtype = 0x40;//do not send msg_stream_id
if (ch.len == prev.len){
if (ch.msg_type_id == prev.msg_type_id){
chtype = 0x80;//do not send len and msg_type_id
if (ch.timestamp == prev.timestamp){
chtype = 0xC0;//do not send timestamp
}
}
}
}
}
if (ch.cs_id <= 63){
tmp = chtype | ch.cs_id; fwrite(&tmp, 1, 1, stdout);
snd_cnt+=1;
}else{
if (ch.cs_id <= 255+64){
tmp = chtype | 0; fwrite(&tmp, 1, 1, stdout);
tmp = ch.cs_id - 64; fwrite(&tmp, 1, 1, stdout);
snd_cnt+=2;
}else{
tmp = chtype | 1; fwrite(&tmp, 1, 1, stdout);
tmpi = ch.cs_id - 64;
tmp = tmpi % 256; fwrite(&tmp, 1, 1, stdout);
tmp = tmpi / 256; fwrite(&tmp, 1, 1, stdout);
snd_cnt+=3;
}
}
unsigned int ntime = 0;
if (chtype != 0xC0){
//timestamp or timestamp diff
if (chtype == 0x00){
tmpi = ch.timestamp;
if (tmpi >= 0x00ffffff){ntime = tmpi; tmpi = 0x00ffffff;}
tmp = tmpi / (256*256); fwrite(&tmp, 1, 1, stdout);
tmp = tmpi / 256; fwrite(&tmp, 1, 1, stdout);
tmp = tmpi % 256; fwrite(&tmp, 1, 1, stdout);
snd_cnt+=3;
}else{
tmpi = ch.timestamp - prev.timestamp;
if (tmpi >= 0x00ffffff){ntime = tmpi; tmpi = 0x00ffffff;}
tmp = tmpi / (256*256); fwrite(&tmp, 1, 1, stdout);
tmp = tmpi / 256; fwrite(&tmp, 1, 1, stdout);
tmp = tmpi % 256; fwrite(&tmp, 1, 1, stdout);
snd_cnt+=3;
}
if (chtype != 0x80){
//len
tmpi = ch.len;
tmp = tmpi / (256*256); fwrite(&tmp, 1, 1, stdout);
tmp = tmpi / 256; fwrite(&tmp, 1, 1, stdout);
tmp = tmpi % 256; fwrite(&tmp, 1, 1, stdout);
snd_cnt+=3;
//msg type id
tmp = ch.msg_type_id; fwrite(&tmp, 1, 1, stdout);
snd_cnt+=1;
if (chtype != 0x40){
//msg stream id
tmp = ch.msg_stream_id % 256; fwrite(&tmp, 1, 1, stdout);
tmp = ch.msg_stream_id / 256; fwrite(&tmp, 1, 1, stdout);
tmp = ch.msg_stream_id / (256*256); fwrite(&tmp, 1, 1, stdout);
tmp = ch.msg_stream_id / (256*256*256); fwrite(&tmp, 1, 1, stdout);
snd_cnt+=4;
}
}
}
//support for 0x00ffffff timestamps
if (ntime){
tmp = ntime / (256*256*256); fwrite(&tmp, 1, 1, stdout);
tmp = ntime / (256*256); fwrite(&tmp, 1, 1, stdout);
tmp = ntime / 256; fwrite(&tmp, 1, 1, stdout);
tmp = ntime % 256; fwrite(&tmp, 1, 1, stdout);
snd_cnt+=4;
}
ch.len_left = 0;
while (ch.len_left < ch.len){
tmpi = ch.len - ch.len_left;
if (tmpi > chunk_snd_max){tmpi = chunk_snd_max;}
fwrite((ch.data + ch.len_left), 1, tmpi, stdout);
snd_cnt+=tmpi;
ch.len_left += tmpi;
if (ch.len_left < ch.len){
if (ch.cs_id <= 63){
tmp = 0xC0 + ch.cs_id; fwrite(&tmp, 1, 1, stdout);
snd_cnt+=1;
}else{
if (ch.cs_id <= 255+64){
tmp = 0xC0; fwrite(&tmp, 1, 1, stdout);
tmp = ch.cs_id - 64; fwrite(&tmp, 1, 1, stdout);
snd_cnt+=2;
}else{
tmp = 0xC1; fwrite(&tmp, 1, 1, stdout);
tmpi = ch.cs_id - 64;
tmp = tmpi % 256; fwrite(&tmp, 1, 1, stdout);
tmp = tmpi / 256; fwrite(&tmp, 1, 1, stdout);
snd_cnt+=4;
}
}
}
}
PutSndPrev(ch);
}//SendChunk
//sends a chunk
void SendChunk(unsigned int cs_id, unsigned char msg_type_id, unsigned int msg_stream_id, std::string data){
chunkpack ch;
ch.cs_id = cs_id;
ch.timestamp = getNowMS();
ch.len = data.size();
ch.real_len = data.size();
ch.len_left = 0;
ch.msg_type_id = msg_type_id;
ch.msg_stream_id = msg_stream_id;
ch.data = (unsigned char*)malloc(data.size());
memcpy(ch.data, data.c_str(), data.size());
SendChunk(ch);
free(ch.data);
}//SendChunk
//sends a media chunk
void SendMedia(unsigned char msg_type_id, unsigned char * data, int len, unsigned int ts){
chunkpack ch;
ch.cs_id = msg_type_id;
ch.timestamp = ts;
ch.len = len;
ch.real_len = len;
ch.len_left = 0;
ch.msg_type_id = msg_type_id;
ch.msg_stream_id = 1;
ch.data = (unsigned char*)malloc(len);
memcpy(ch.data, data, len);
SendChunk(ch);
free(ch.data);
}//SendMedia
//sends a control message
void SendCTL(unsigned char type, unsigned int data){
chunkpack ch;
ch.cs_id = 2;
ch.timestamp = getNowMS();
ch.len = 4;
ch.real_len = 4;
ch.len_left = 0;
ch.msg_type_id = type;
ch.msg_stream_id = 0;
ch.data = (unsigned char*)malloc(4);
data = htonl(data);
memcpy(ch.data, &data, 4);
SendChunk(ch);
free(ch.data);
}//SendCTL
//sends a control message
void SendCTL(unsigned char type, unsigned int data, unsigned char data2){
chunkpack ch;
ch.cs_id = 2;
ch.timestamp = getNowMS();
ch.len = 5;
ch.real_len = 5;
ch.len_left = 0;
ch.msg_type_id = type;
ch.msg_stream_id = 0;
ch.data = (unsigned char*)malloc(5);
data = htonl(data);
memcpy(ch.data, &data, 4);
ch.data[4] = data2;
SendChunk(ch);
free(ch.data);
}//SendCTL
//sends a usr control message
void SendUSR(unsigned char type, unsigned int data){
chunkpack ch;
ch.cs_id = 2;
ch.timestamp = getNowMS();
ch.len = 6;
ch.real_len = 6;
ch.len_left = 0;
ch.msg_type_id = 4;
ch.msg_stream_id = 0;
ch.data = (unsigned char*)malloc(6);
data = htonl(data);
memcpy(ch.data+2, &data, 4);
ch.data[0] = 0;
ch.data[1] = type;
SendChunk(ch);
free(ch.data);
}//SendUSR
//sends a usr control message
void SendUSR(unsigned char type, unsigned int data, unsigned int data2){
chunkpack ch;
ch.cs_id = 2;
ch.timestamp = getNowMS();
ch.len = 10;
ch.real_len = 10;
ch.len_left = 0;
ch.msg_type_id = 4;
ch.msg_stream_id = 0;
ch.data = (unsigned char*)malloc(10);
data = htonl(data);
data2 = htonl(data2);
memcpy(ch.data+2, &data, 4);
memcpy(ch.data+6, &data2, 4);
ch.data[0] = 0;
ch.data[1] = type;
SendChunk(ch);
free(ch.data);
}//SendUSR
//get a chunk from standard input
struct chunkpack getChunk(){
gettimeofday(&lastrec, 0);
struct chunkpack ret;
unsigned char temp;
fread(&(ret.chunktype), 1, 1, stdin);
rec_cnt++;
//read the chunkstream ID properly
switch (ret.chunktype & 0x3F){
case 0:
fread(&temp, 1, 1, stdin);
rec_cnt++;
ret.cs_id = temp + 64;
break;
case 1:
fread(&temp, 1, 1, stdin);
ret.cs_id = temp + 64;
fread(&temp, 1, 1, stdin);
ret.cs_id += temp * 256;
rec_cnt+=2;
break;
default:
ret.cs_id = ret.chunktype & 0x3F;
break;
}
chunkinfo prev = GetPrev(ret.cs_id);
//process the rest of the header, for each chunk type
switch (ret.chunktype & 0xC0){
case 0x00:
fread(&temp, 1, 1, stdin);
ret.timestamp = temp*256*256;
fread(&temp, 1, 1, stdin);
ret.timestamp += temp*256;
fread(&temp, 1, 1, stdin);
ret.timestamp += temp;
fread(&temp, 1, 1, stdin);
ret.len = temp*256*256;
fread(&temp, 1, 1, stdin);
ret.len += temp*256;
fread(&temp, 1, 1, stdin);
ret.len += temp;
ret.len_left = 0;
fread(&temp, 1, 1, stdin);
ret.msg_type_id = temp;
fread(&temp, 1, 1, stdin);
ret.msg_stream_id = temp;
fread(&temp, 1, 1, stdin);
ret.msg_stream_id += temp*256;
fread(&temp, 1, 1, stdin);
ret.msg_stream_id += temp*256*256;
fread(&temp, 1, 1, stdin);
ret.msg_stream_id += temp*256*256*256;
rec_cnt+=11;
break;
case 0x40:
fread(&temp, 1, 1, stdin);
ret.timestamp = temp*256*256;
fread(&temp, 1, 1, stdin);
ret.timestamp += temp*256;
fread(&temp, 1, 1, stdin);
ret.timestamp += temp;
ret.timestamp += prev.timestamp;
fread(&temp, 1, 1, stdin);
ret.len = temp*256*256;
fread(&temp, 1, 1, stdin);
ret.len += temp*256;
fread(&temp, 1, 1, stdin);
ret.len += temp;
ret.len_left = 0;
fread(&temp, 1, 1, stdin);
ret.msg_type_id = temp;
ret.msg_stream_id = prev.msg_stream_id;
rec_cnt+=7;
break;
case 0x80:
fread(&temp, 1, 1, stdin);
ret.timestamp = temp*256*256;
fread(&temp, 1, 1, stdin);
ret.timestamp += temp*256;
fread(&temp, 1, 1, stdin);
ret.timestamp += temp;
ret.timestamp += prev.timestamp;
ret.len = prev.len;
ret.len_left = prev.len_left;
ret.msg_type_id = prev.msg_type_id;
ret.msg_stream_id = prev.msg_stream_id;
rec_cnt+=3;
break;
case 0xC0:
ret.timestamp = prev.timestamp;
ret.len = prev.len;
ret.len_left = prev.len_left;
ret.msg_type_id = prev.msg_type_id;
ret.msg_stream_id = prev.msg_stream_id;
break;
}
//calculate chunk length, real length, and length left till complete
if (ret.len_left > 0){
ret.real_len = ret.len_left;
ret.len_left -= ret.real_len;
}else{
ret.real_len = ret.len;
}
if (ret.real_len > chunk_rec_max){
ret.len_left += ret.real_len - chunk_rec_max;
ret.real_len = chunk_rec_max;
}
//read extended timestamp, if neccesary
if (ret.timestamp == 0x00ffffff){
fread(&temp, 1, 1, stdin);
ret.timestamp = temp*256*256*256;
fread(&temp, 1, 1, stdin);
ret.timestamp += temp*256*256;
fread(&temp, 1, 1, stdin);
ret.timestamp += temp*256;
fread(&temp, 1, 1, stdin);
ret.timestamp += temp;
rec_cnt+=4;
}
//read data if length > 0, and allocate it
if (ret.real_len > 0){
ret.data = (unsigned char*)malloc(ret.real_len);
fread(ret.data, 1, ret.real_len, stdin);
rec_cnt+=ret.real_len;
}else{
ret.data = 0;
}
PutPrev(ret);
return ret;
}//getChunk
//adds newchunk to global list of unfinished chunks, re-assembling them complete
//returns pointer to chunk when a chunk is finished, 0 otherwise
//removes pointed to chunk from internal list if returned, without cleanup
// (cleanup performed in getWholeChunk function)
chunkpack * AddChunkPart(chunkpack newchunk){
chunkpack * p;
unsigned char * tmpdata = 0;
static std::map<unsigned int, chunkpack *> ch_lst;
std::map<unsigned int, chunkpack *>::iterator it;
it = ch_lst.find(newchunk.cs_id);
if (it == ch_lst.end()){
p = (chunkpack*)malloc(sizeof(chunkpack));
*p = newchunk;
p->data = (unsigned char*)malloc(p->real_len);
memcpy(p->data, newchunk.data, p->real_len);
if (p->len_left == 0){return p;}
ch_lst[newchunk.cs_id] = p;
}else{
p = it->second;
tmpdata = (unsigned char*)realloc(p->data, p->real_len + newchunk.real_len);
if (tmpdata == 0){
#ifdef DEBUG
fprintf(stderr, "Error allocating memory!\n");
#endif
return 0;
}
p->data = tmpdata;
memcpy(p->data+p->real_len, newchunk.data, newchunk.real_len);
p->real_len += newchunk.real_len;
p->len_left -= newchunk.real_len;
if (p->len_left <= 0){
ch_lst.erase(it);
return p;
}else{
ch_lst[newchunk.cs_id] = p;//pointer may have changed
}
}
return 0;
}//AddChunkPart
//grabs chunks until a whole one comes in, then returns that
chunkpack getWholeChunk(){
static chunkpack gwc_next, gwc_complete;
static bool clean = false;
int counter = 0;
if (!clean){gwc_complete.data = 0; clean = true;}//prevent brain damage
chunkpack * ret = 0;
scrubChunk(gwc_complete);
while (counter < 10000){
gwc_next = getChunk();
ret = AddChunkPart(gwc_next);
scrubChunk(gwc_next);
if (ret){
gwc_complete = *ret;
free(ret);//cleanup returned chunk
return gwc_complete;
}
if (feof(stdin) != 0){break;}
counter++;
}
gwc_complete.msg_type_id = 0;
return gwc_complete;
}//getWholeChunk

506
Connector_RTMPf/crypto.cpp
View file

@ -1,506 +0,0 @@
#define STR(x) (((std::string)(x)).c_str())
#include "crypto.h"
#define P768 \
"FFFFFFFFFFFFFFFFC90FDAA22168C234C4C6628B80DC1CD1" \
"29024E088A67CC74020BBEA63B139B22514A08798E3404DD" \
"EF9519B3CD3A431B302B0A6DF25F14374FE1356D6D51C245" \
"E485B576625E7EC6F44C42E9A63A3620FFFFFFFFFFFFFFFF"
#define P1024 \
"FFFFFFFFFFFFFFFFC90FDAA22168C234C4C6628B80DC1CD1" \
"29024E088A67CC74020BBEA63B139B22514A08798E3404DD" \
"EF9519B3CD3A431B302B0A6DF25F14374FE1356D6D51C245" \
"E485B576625E7EC6F44C42E9A637ED6B0BFF5CB6F406B7ED" \
"EE386BFB5A899FA5AE9F24117C4B1FE649286651ECE65381" \
"FFFFFFFFFFFFFFFF"
#define Q1024 \
"7FFFFFFFFFFFFFFFE487ED5110B4611A62633145C06E0E68" \
"948127044533E63A0105DF531D89CD9128A5043CC71A026E" \
"F7CA8CD9E69D218D98158536F92F8A1BA7F09AB6B6A8E122" \
"F242DABB312F3F637A262174D31BF6B585FFAE5B7A035BF6" \
"F71C35FDAD44CFD2D74F9208BE258FF324943328F67329C0" \
"FFFFFFFFFFFFFFFF"
#define P1536 \
"FFFFFFFFFFFFFFFFC90FDAA22168C234C4C6628B80DC1CD1" \
"29024E088A67CC74020BBEA63B139B22514A08798E3404DD" \
"EF9519B3CD3A431B302B0A6DF25F14374FE1356D6D51C245" \
"E485B576625E7EC6F44C42E9A637ED6B0BFF5CB6F406B7ED" \
"EE386BFB5A899FA5AE9F24117C4B1FE649286651ECE45B3D" \
"C2007CB8A163BF0598DA48361C55D39A69163FA8FD24CF5F" \
"83655D23DCA3AD961C62F356208552BB9ED529077096966D" \
"670C354E4ABC9804F1746C08CA237327FFFFFFFFFFFFFFFF"
#define P2048 \
"FFFFFFFFFFFFFFFFC90FDAA22168C234C4C6628B80DC1CD1" \
"29024E088A67CC74020BBEA63B139B22514A08798E3404DD" \
"EF9519B3CD3A431B302B0A6DF25F14374FE1356D6D51C245" \
"E485B576625E7EC6F44C42E9A637ED6B0BFF5CB6F406B7ED" \
"EE386BFB5A899FA5AE9F24117C4B1FE649286651ECE45B3D" \
"C2007CB8A163BF0598DA48361C55D39A69163FA8FD24CF5F" \
"83655D23DCA3AD961C62F356208552BB9ED529077096966D" \
"670C354E4ABC9804F1746C08CA18217C32905E462E36CE3B" \
"E39E772C180E86039B2783A2EC07A28FB5C55DF06F4C52C9" \
"DE2BCBF6955817183995497CEA956AE515D2261898FA0510" \
"15728E5A8AACAA68FFFFFFFFFFFFFFFF"
#define P3072 \
"FFFFFFFFFFFFFFFFC90FDAA22168C234C4C6628B80DC1CD1" \
"29024E088A67CC74020BBEA63B139B22514A08798E3404DD" \
"EF9519B3CD3A431B302B0A6DF25F14374FE1356D6D51C245" \
"E485B576625E7EC6F44C42E9A637ED6B0BFF5CB6F406B7ED" \
"EE386BFB5A899FA5AE9F24117C4B1FE649286651ECE45B3D" \
"C2007CB8A163BF0598DA48361C55D39A69163FA8FD24CF5F" \
"83655D23DCA3AD961C62F356208552BB9ED529077096966D" \
"670C354E4ABC9804F1746C08CA18217C32905E462E36CE3B" \
"E39E772C180E86039B2783A2EC07A28FB5C55DF06F4C52C9" \
"DE2BCBF6955817183995497CEA956AE515D2261898FA0510" \
"15728E5A8AAAC42DAD33170D04507A33A85521ABDF1CBA64" \
"ECFB850458DBEF0A8AEA71575D060C7DB3970F85A6E1E4C7" \
"ABF5AE8CDB0933D71E8C94E04A25619DCEE3D2261AD2EE6B" \
"F12FFA06D98A0864D87602733EC86A64521F2B18177B200C" \
"BBE117577A615D6C770988C0BAD946E208E24FA074E5AB31" \
"43DB5BFCE0FD108E4B82D120A93AD2CAFFFFFFFFFFFFFFFF"
#define P4096 \
"FFFFFFFFFFFFFFFFC90FDAA22168C234C4C6628B80DC1CD1" \
"29024E088A67CC74020BBEA63B139B22514A08798E3404DD" \
"EF9519B3CD3A431B302B0A6DF25F14374FE1356D6D51C245" \
"E485B576625E7EC6F44C42E9A637ED6B0BFF5CB6F406B7ED" \
"EE386BFB5A899FA5AE9F24117C4B1FE649286651ECE45B3D" \
"C2007CB8A163BF0598DA48361C55D39A69163FA8FD24CF5F" \
"83655D23DCA3AD961C62F356208552BB9ED529077096966D" \
"670C354E4ABC9804F1746C08CA18217C32905E462E36CE3B" \
"E39E772C180E86039B2783A2EC07A28FB5C55DF06F4C52C9" \
"DE2BCBF6955817183995497CEA956AE515D2261898FA0510" \
"15728E5A8AAAC42DAD33170D04507A33A85521ABDF1CBA64" \
"ECFB850458DBEF0A8AEA71575D060C7DB3970F85A6E1E4C7" \
"ABF5AE8CDB0933D71E8C94E04A25619DCEE3D2261AD2EE6B" \
"F12FFA06D98A0864D87602733EC86A64521F2B18177B200C" \
"BBE117577A615D6C770988C0BAD946E208E24FA074E5AB31" \
"43DB5BFCE0FD108E4B82D120A92108011A723C12A787E6D7" \
"88719A10BDBA5B2699C327186AF4E23C1A946834B6150BDA" \
"2583E9CA2AD44CE8DBBBC2DB04DE8EF92E8EFC141FBECAA6" \
"287C59474E6BC05D99B2964FA090C3A2233BA186515BE7ED" \
"1F612970CEE2D7AFB81BDD762170481CD0069127D5B05AA9" \
"93B4EA988D8FDDC186FFB7DC90A6C08F4DF435C934063199" \
"FFFFFFFFFFFFFFFF"
#define P6144 \
"FFFFFFFFFFFFFFFFC90FDAA22168C234C4C6628B80DC1CD1" \
"29024E088A67CC74020BBEA63B139B22514A08798E3404DD" \
"EF9519B3CD3A431B302B0A6DF25F14374FE1356D6D51C245" \
"E485B576625E7EC6F44C42E9A637ED6B0BFF5CB6F406B7ED" \
"EE386BFB5A899FA5AE9F24117C4B1FE649286651ECE45B3D" \
"C2007CB8A163BF0598DA48361C55D39A69163FA8FD24CF5F" \
"83655D23DCA3AD961C62F356208552BB9ED529077096966D" \
"670C354E4ABC9804F1746C08CA18217C32905E462E36CE3B" \
"E39E772C180E86039B2783A2EC07A28FB5C55DF06F4C52C9" \
"DE2BCBF6955817183995497CEA956AE515D2261898FA0510" \
"15728E5A8AAAC42DAD33170D04507A33A85521ABDF1CBA64" \
"ECFB850458DBEF0A8AEA71575D060C7DB3970F85A6E1E4C7" \
"ABF5AE8CDB0933D71E8C94E04A25619DCEE3D2261AD2EE6B" \
"F12FFA06D98A0864D87602733EC86A64521F2B18177B200C" \
"BBE117577A615D6C770988C0BAD946E208E24FA074E5AB31" \
"43DB5BFCE0FD108E4B82D120A92108011A723C12A787E6D7" \
"88719A10BDBA5B2699C327186AF4E23C1A946834B6150BDA" \
"2583E9CA2AD44CE8DBBBC2DB04DE8EF92E8EFC141FBECAA6" \
"287C59474E6BC05D99B2964FA090C3A2233BA186515BE7ED" \
"1F612970CEE2D7AFB81BDD762170481CD0069127D5B05AA9" \
"93B4EA988D8FDDC186FFB7DC90A6C08F4DF435C934028492" \
"36C3FAB4D27C7026C1D4DCB2602646DEC9751E763DBA37BD" \
"F8FF9406AD9E530EE5DB382F413001AEB06A53ED9027D831" \
"179727B0865A8918DA3EDBEBCF9B14ED44CE6CBACED4BB1B" \
"DB7F1447E6CC254B332051512BD7AF426FB8F401378CD2BF" \
"5983CA01C64B92ECF032EA15D1721D03F482D7CE6E74FEF6" \
"D55E702F46980C82B5A84031900B1C9E59E7C97FBEC7E8F3" \
"23A97A7E36CC88BE0F1D45B7FF585AC54BD407B22B4154AA" \
"CC8F6D7EBF48E1D814CC5ED20F8037E0A79715EEF29BE328" \
"06A1D58BB7C5DA76F550AA3D8A1FBFF0EB19CCB1A313D55C" \
"DA56C9EC2EF29632387FE8D76E3C0468043E8F663F4860EE" \
"12BF2D5B0B7474D6E694F91E6DCC4024FFFFFFFFFFFFFFFF"
#define P8192 \
"FFFFFFFFFFFFFFFFC90FDAA22168C234C4C6628B80DC1CD1" \
"29024E088A67CC74020BBEA63B139B22514A08798E3404DD" \
"EF9519B3CD3A431B302B0A6DF25F14374FE1356D6D51C245" \
"E485B576625E7EC6F44C42E9A637ED6B0BFF5CB6F406B7ED" \
"EE386BFB5A899FA5AE9F24117C4B1FE649286651ECE45B3D" \
"C2007CB8A163BF0598DA48361C55D39A69163FA8FD24CF5F" \
"83655D23DCA3AD961C62F356208552BB9ED529077096966D" \
"670C354E4ABC9804F1746C08CA18217C32905E462E36CE3B" \
"E39E772C180E86039B2783A2EC07A28FB5C55DF06F4C52C9" \
"DE2BCBF6955817183995497CEA956AE515D2261898FA0510" \
"15728E5A8AAAC42DAD33170D04507A33A85521ABDF1CBA64" \
"ECFB850458DBEF0A8AEA71575D060C7DB3970F85A6E1E4C7" \
"ABF5AE8CDB0933D71E8C94E04A25619DCEE3D2261AD2EE6B" \
"F12FFA06D98A0864D87602733EC86A64521F2B18177B200C" \
"BBE117577A615D6C770988C0BAD946E208E24FA074E5AB31" \
"43DB5BFCE0FD108E4B82D120A92108011A723C12A787E6D7" \
"88719A10BDBA5B2699C327186AF4E23C1A946834B6150BDA" \
"2583E9CA2AD44CE8DBBBC2DB04DE8EF92E8EFC141FBECAA6" \
"287C59474E6BC05D99B2964FA090C3A2233BA186515BE7ED" \
"1F612970CEE2D7AFB81BDD762170481CD0069127D5B05AA9" \
"93B4EA988D8FDDC186FFB7DC90A6C08F4DF435C934028492" \
"36C3FAB4D27C7026C1D4DCB2602646DEC9751E763DBA37BD" \
"F8FF9406AD9E530EE5DB382F413001AEB06A53ED9027D831" \
"179727B0865A8918DA3EDBEBCF9B14ED44CE6CBACED4BB1B" \
"DB7F1447E6CC254B332051512BD7AF426FB8F401378CD2BF" \
"5983CA01C64B92ECF032EA15D1721D03F482D7CE6E74FEF6" \
"D55E702F46980C82B5A84031900B1C9E59E7C97FBEC7E8F3" \
"23A97A7E36CC88BE0F1D45B7FF585AC54BD407B22B4154AA" \
"CC8F6D7EBF48E1D814CC5ED20F8037E0A79715EEF29BE328" \
"06A1D58BB7C5DA76F550AA3D8A1FBFF0EB19CCB1A313D55C" \
"DA56C9EC2EF29632387FE8D76E3C0468043E8F663F4860EE" \
"12BF2D5B0B7474D6E694F91E6DBE115974A3926F12FEE5E4" \
"38777CB6A932DF8CD8BEC4D073B931BA3BC832B68D9DD300" \
"741FA7BF8AFC47ED2576F6936BA424663AAB639C5AE4F568" \
"3423B4742BF1C978238F16CBE39D652DE3FDB8BEFC848AD9" \
"22222E04A4037C0713EB57A81A23F0C73473FC646CEA306B" \
"4BCBC8862F8385DDFA9D4B7FA2C087E879683303ED5BDD3A" \
"062B3CF5B3A278A66D2A13F83F44F82DDF310EE074AB6A36" \
"4597E899A0255DC164F31CC50846851DF9AB48195DED7EA1" \
"B1D510BD7EE74D73FAF36BC31ECFA268359046F4EB879F92" \
"4009438B481C6CD7889A002ED5EE382BC9190DA6FC026E47" \
"9558E4475677E9AA9E3050E2765694DFC81F56E880B96E71" \
"60C980DD98EDD3DFFFFFFFFFFFFFFFFF"
uint8_t genuineFMSKey[] = {
0x47, 0x65, 0x6e, 0x75, 0x69, 0x6e, 0x65, 0x20,
0x41, 0x64, 0x6f, 0x62, 0x65, 0x20, 0x46, 0x6c,
0x61, 0x73, 0x68, 0x20, 0x4d, 0x65, 0x64, 0x69,
0x61, 0x20, 0x53, 0x65, 0x72, 0x76, 0x65, 0x72,
0x20, 0x30, 0x30, 0x31, // Genuine Adobe Flash Media Server 001
0xf0, 0xee, 0xc2, 0x4a, 0x80, 0x68, 0xbe, 0xe8,
0x2e, 0x00, 0xd0, 0xd1, 0x02, 0x9e, 0x7e, 0x57,
0x6e, 0xec, 0x5d, 0x2d, 0x29, 0x80, 0x6f, 0xab,
0x93, 0xb8, 0xe6, 0x36, 0xcf, 0xeb, 0x31, 0xae
}; // 68
uint8_t genuineFPKey[] = {
0x47, 0x65, 0x6E, 0x75, 0x69, 0x6E, 0x65, 0x20,
0x41, 0x64, 0x6F, 0x62, 0x65, 0x20, 0x46, 0x6C,
0x61, 0x73, 0x68, 0x20, 0x50, 0x6C, 0x61, 0x79,
0x65, 0x72, 0x20, 0x30, 0x30, 0x31, // Genuine Adobe Flash Player 001
0xF0, 0xEE, 0xC2, 0x4A, 0x80, 0x68, 0xBE, 0xE8,
0x2E, 0x00, 0xD0, 0xD1, 0x02, 0x9E, 0x7E, 0x57,
0x6E, 0xEC, 0x5D, 0x2D, 0x29, 0x80, 0x6F, 0xAB,
0x93, 0xB8, 0xE6, 0x36, 0xCF, 0xEB, 0x31, 0xAE
}; // 62
void replace(std::string &target, std::string search, std::string replacement) {
if (search == replacement)
return;
if (search == "")
return;
std::string::size_type i = std::string::npos;
while ((i = target.find(search)) != std::string::npos) {
target.replace(i, search.length(), replacement);
}
}
DHWrapper::DHWrapper(int32_t bitsCount) {
_bitsCount = bitsCount;
_pDH = NULL;
_pSharedKey = NULL;
_sharedKeyLength = 0;
_peerPublickey = NULL;
}
DHWrapper::~DHWrapper() {
Cleanup();
}
bool DHWrapper::Initialize() {
Cleanup();
//1. Create the DH
_pDH = DH_new();
if (_pDH == NULL) {
Cleanup();
return false;
}
//2. Create his internal p and g
_pDH->p = BN_new();
if (_pDH->p == NULL) {
Cleanup();
return false;
}
_pDH->g = BN_new();
if (_pDH->g == NULL) {
Cleanup();
return false;
}
//3. initialize p, g and key length
if (BN_hex2bn(&_pDH->p, P1024) == 0) {
Cleanup();
return false;
}
if (BN_set_word(_pDH->g, 2) != 1) {
Cleanup();
return false;
}
//4. Set the key length
_pDH->length = _bitsCount;
//5. Generate private and public key
if (DH_generate_key(_pDH) != 1) {
Cleanup();
return false;
}
return true;
}
bool DHWrapper::CopyPublicKey(uint8_t *pDst, int32_t dstLength) {
if (_pDH == NULL) {
return false;
}
return CopyKey(_pDH->pub_key, pDst, dstLength);
}
bool DHWrapper::CopyPrivateKey(uint8_t *pDst, int32_t dstLength) {
if (_pDH == NULL) {
return false;
}
return CopyKey(_pDH->priv_key, pDst, dstLength);
}
bool DHWrapper::CreateSharedKey(uint8_t *pPeerPublicKey, int32_t length) {
if (_pDH == NULL) {
return false;
}
if (_sharedKeyLength != 0 || _pSharedKey != NULL) {
return false;
}
_sharedKeyLength = DH_size(_pDH);
if (_sharedKeyLength <= 0 || _sharedKeyLength > 1024) {
return false;
}
_pSharedKey = new uint8_t[_sharedKeyLength];
_peerPublickey = BN_bin2bn(pPeerPublicKey, length, 0);
if (_peerPublickey == NULL) {
return false;
}
if (DH_compute_key(_pSharedKey, _peerPublickey, _pDH) != _sharedKeyLength) {
return false;
}
return true;
}
bool DHWrapper::CopySharedKey(uint8_t *pDst, int32_t dstLength) {
if (_pDH == NULL) {
return false;
}
if (dstLength != _sharedKeyLength) {
return false;
}
memcpy(pDst, _pSharedKey, _sharedKeyLength);
return true;
}
void DHWrapper::Cleanup() {
if (_pDH != NULL) {
if (_pDH->p != NULL) {
BN_free(_pDH->p);
_pDH->p = NULL;
}
if (_pDH->g != NULL) {
BN_free(_pDH->g);
_pDH->g = NULL;
}
DH_free(_pDH);
_pDH = NULL;
}
if (_pSharedKey != NULL) {
delete[] _pSharedKey;
_pSharedKey = NULL;
}
_sharedKeyLength = 0;
if (_peerPublickey != NULL) {
BN_free(_peerPublickey);
_peerPublickey = NULL;
}
}
bool DHWrapper::CopyKey(BIGNUM *pNum, uint8_t *pDst, int32_t dstLength) {
int32_t keySize = BN_num_bytes(pNum);
if ((keySize <= 0) || (dstLength <= 0) || (keySize > dstLength)) {
return false;
}
if (BN_bn2bin(pNum, pDst) != keySize) {
return false;
}
return true;
}
void InitRC4Encryption(uint8_t *secretKey, uint8_t *pubKeyIn, uint8_t *pubKeyOut, RC4_KEY *rc4keyIn, RC4_KEY *rc4keyOut) {
uint8_t digest[SHA256_DIGEST_LENGTH];
unsigned int digestLen = 0;
HMAC_CTX ctx;
HMAC_CTX_init(&ctx);
HMAC_Init_ex(&ctx, secretKey, 128, EVP_sha256(), 0);
HMAC_Update(&ctx, pubKeyIn, 128);
HMAC_Final(&ctx, digest, &digestLen);
HMAC_CTX_cleanup(&ctx);
RC4_set_key(rc4keyOut, 16, digest);
HMAC_CTX_init(&ctx);
HMAC_Init_ex(&ctx, secretKey, 128, EVP_sha256(), 0);
HMAC_Update(&ctx, pubKeyOut, 128);
HMAC_Final(&ctx, digest, &digestLen);
HMAC_CTX_cleanup(&ctx);
RC4_set_key(rc4keyIn, 16, digest);
}
std::string md5(std::string source, bool textResult) {
EVP_MD_CTX mdctx;
unsigned char md_value[EVP_MAX_MD_SIZE];
unsigned int md_len;
EVP_DigestInit(&mdctx, EVP_md5());
EVP_DigestUpdate(&mdctx, STR(source), source.length());
EVP_DigestFinal_ex(&mdctx, md_value, &md_len);
EVP_MD_CTX_cleanup(&mdctx);
if (textResult) {
std::string result = "";
char tmp[3];
for (uint32_t i = 0; i < md_len; i++) {
sprintf(tmp, "%02x", md_value[i]);
result += tmp;
}
return result;
} else {
return std::string((char *) md_value, md_len);
}
}
std::string b64(std::string source) {
return b64((uint8_t *) STR(source), source.size());
}
std::string b64(uint8_t *pBuffer, uint32_t length) {
BIO *bmem;
BIO *b64;
BUF_MEM *bptr;
b64 = BIO_new(BIO_f_base64());
bmem = BIO_new(BIO_s_mem());
b64 = BIO_push(b64, bmem);
BIO_write(b64, pBuffer, length);
std::string result = "";
if (BIO_flush(b64) == 1) {
BIO_get_mem_ptr(b64, &bptr);
result = std::string(bptr->data, bptr->length);
}
BIO_free_all(b64);
replace(result, "\n", "");
replace(result, "\r", "");
return result;
}
std::string unb64(std::string source) {
return unb64((uint8_t *)STR(source),source.length());
}
std::string unb64(uint8_t *pBuffer, uint32_t length){
// create a memory buffer containing base64 encoded data
//BIO* bmem = BIO_new_mem_buf((void*) STR(source), source.length());
BIO* bmem = BIO_new_mem_buf((void *)pBuffer, length);
// push a Base64 filter so that reading from buffer decodes it
BIO *bioCmd = BIO_new(BIO_f_base64());
// we don't want newlines
BIO_set_flags(bioCmd, BIO_FLAGS_BASE64_NO_NL);
bmem = BIO_push(bioCmd, bmem);
char *pOut = new char[length];
int finalLen = BIO_read(bmem, (void*) pOut, length);
BIO_free_all(bmem);
std::string result(pOut, finalLen);
delete[] pOut;
return result;
}
void HMACsha256(const void *pData, uint32_t dataLength, const void *pKey, uint32_t keyLength, void *pResult) {
unsigned int digestLen;
HMAC_CTX ctx;
HMAC_CTX_init(&ctx);
HMAC_Init_ex(&ctx, (unsigned char*) pKey, keyLength, EVP_sha256(), NULL);
HMAC_Update(&ctx, (unsigned char *) pData, dataLength);
HMAC_Final(&ctx, (unsigned char *) pResult, &digestLen);
HMAC_CTX_cleanup(&ctx);
}
uint32_t GetDigestOffset0(uint8_t *pBuffer) {
uint32_t offset = pBuffer[8] + pBuffer[9] + pBuffer[10] + pBuffer[11];
return (offset % 728) + 12;
}
uint32_t GetDigestOffset1(uint8_t *pBuffer) {
uint32_t offset = pBuffer[772] + pBuffer[773] + pBuffer[774] + pBuffer[775];
return (offset % 728) + 776;
}
uint32_t GetDigestOffset(uint8_t *pBuffer, uint8_t scheme){
if (scheme == 0){return GetDigestOffset0(pBuffer);}else{return GetDigestOffset1(pBuffer);}
}
uint32_t GetDHOffset0(uint8_t *pBuffer) {
uint32_t offset = pBuffer[1532] + pBuffer[1533] + pBuffer[1534] + pBuffer[1535];
return (offset % 632) + 772;
}
uint32_t GetDHOffset1(uint8_t *pBuffer) {
uint32_t offset = pBuffer[768] + pBuffer[769] + pBuffer[770] + pBuffer[771];
return (offset % 632) + 8;
}
uint32_t GetDHOffset(uint8_t *pBuffer, uint8_t scheme){
if (scheme == 0){return GetDHOffset0(pBuffer);}else{return GetDHOffset1(pBuffer);}
}
bool ValidateClientScheme(uint8_t * pBuffer, uint8_t scheme) {
uint32_t clientDigestOffset = GetDigestOffset(pBuffer, scheme);
uint8_t *pTempBuffer = new uint8_t[1536 - 32];
memcpy(pTempBuffer, pBuffer, clientDigestOffset);
memcpy(pTempBuffer + clientDigestOffset, pBuffer + clientDigestOffset + 32, 1536 - clientDigestOffset - 32);
uint8_t *pTempHash = new uint8_t[512];
HMACsha256(pTempBuffer, 1536 - 32, genuineFPKey, 30, pTempHash);
bool result = (memcmp(pBuffer+clientDigestOffset, pTempHash, 32) == 0);
#ifdef DEBUG
fprintf(stderr, "Client scheme validation %hhi %s\n", scheme, result?"success":"failed");
#endif
delete[] pTempBuffer;
delete[] pTempHash;
return result;
}

45
Connector_RTMPf/crypto.h
View file

@ -1,45 +0,0 @@
#ifndef _CRYPTO_H
#define _CRYPTO_H
#define DLLEXP
#include <openssl/bn.h>
#include <openssl/dh.h>
#include <openssl/rc4.h>
#include <openssl/ssl.h>
#include <openssl/rand.h>
#include <openssl/err.h>
#include <openssl/bio.h>
#include <openssl/hmac.h>
class DLLEXP DHWrapper {
private:
int32_t _bitsCount;
DH *_pDH;
uint8_t *_pSharedKey;
int32_t _sharedKeyLength;
BIGNUM *_peerPublickey;
public:
DHWrapper(int32_t bitsCount);
virtual ~DHWrapper();
bool Initialize();
bool CopyPublicKey(uint8_t *pDst, int32_t dstLength);
bool CopyPrivateKey(uint8_t *pDst, int32_t dstLength);
bool CreateSharedKey(uint8_t *pPeerPublicKey, int32_t length);
bool CopySharedKey(uint8_t *pDst, int32_t dstLength);
private:
void Cleanup();
bool CopyKey(BIGNUM *pNum, uint8_t *pDst, int32_t dstLength);
};
DLLEXP void InitRC4Encryption(uint8_t *secretKey, uint8_t *pubKeyIn, uint8_t *pubKeyOut,
RC4_KEY *rc4keyIn, RC4_KEY *rc4keyOut);
DLLEXP std::string md5(std::string source, bool textResult);
DLLEXP std::string b64(std::string source);
DLLEXP std::string b64(uint8_t *pBuffer, uint32_t length);
DLLEXP std::string unb64(std::string source);
DLLEXP std::string unb64(uint8_t *pBuffer, uint32_t length);
#endif /* _CRYPTO_H */

30
Connector_RTMPf/flv_sock.cpp
View file

@ -1,30 +0,0 @@
SWBaseSocket::SWBaseError SWBerr;
char * FLVbuffer;
int FLV_len;
int FLVbs = 0;
void FLV_Readheader(SWUnixSocket & ss){
static char header[13];
while (ss.frecv(header, 13, &SWBerr) != 13){
//wait
}
}//FLV_Readheader
void FLV_Dump(){FLV_len = 0;}
bool FLV_GetPacket(SWUnixSocket & ss){
if (FLVbs < 15){FLVbuffer = (char*)realloc(FLVbuffer, 15); FLVbs = 15;}
//if received a whole header, receive a whole packet
//if not, retry header next pass
if (FLV_len == 0){
if (ss.frecv(FLVbuffer, 11, &SWBerr) == 11){
FLV_len = FLVbuffer[3] + 15;
FLV_len += (FLVbuffer[2] << 8);
FLV_len += (FLVbuffer[1] << 16);
if (FLVbs < FLV_len){FLVbuffer = (char*)realloc(FLVbuffer, FLV_len);FLVbs = FLV_len;}
}
}else{
if (ss.frecv(FLVbuffer+11, FLV_len-11, &SWBerr) == FLV_len-11){return true;}
}
return false;
}//FLV_GetPacket

137
Connector_RTMPf/handshake.cpp
View file

@ -1,137 +0,0 @@
#undef OLDHANDSHAKE //change to #define for old handshake method
char versionstring[] = "PLSRTMPServer";
#ifdef OLDHANDSHAKE
struct Handshake {
char Time[4];
char Zero[4];
char Random[1528];
};//Handshake
bool doHandshake(){
char Version;
Handshake Client;
Handshake Server;
/** Read C0 **/
fread(&(Version), 1, 1, stdin);
/** Read C1 **/
fread(Client.Time, 1, 4, stdin);
fread(Client.Zero, 1, 4, stdin);
fread(Client.Random, 1, 1528, stdin);
rec_cnt+=1537;
/** Build S1 Packet **/
Server.Time[0] = 0; Server.Time[1] = 0; Server.Time[2] = 0; Server.Time[3] = 0;
Server.Zero[0] = 0; Server.Zero[1] = 0; Server.Zero[2] = 0; Server.Zero[3] = 0;
for (int i = 0; i < 1528; i++){
Server.Random[i] = versionstring[i%13];
}
/** Send S0 **/
fwrite(&(Version), 1, 1, stdout);
/** Send S1 **/
fwrite(Server.Time, 1, 4, stdout);
fwrite(Server.Zero, 1, 4, stdout);
fwrite(Server.Random, 1, 1528, stdout);
/** Flush output, just for certainty **/
fflush(stdout);
snd_cnt+=1537;
/** Send S2 **/
fwrite(Client.Time, 1, 4, stdout);
fwrite(Client.Time, 1, 4, stdout);
fwrite(Client.Random, 1, 1528, stdout);
snd_cnt+=1536;
/** Flush, necessary in order to work **/
fflush(stdout);
/** Read and discard C2 **/
fread(Client.Time, 1, 4, stdin);
fread(Client.Zero, 1, 4, stdin);
fread(Client.Random, 1, 1528, stdin);
rec_cnt+=1536;
return true;
}//doHandshake
#else
#include "crypto.cpp" //cryptography for handshaking
bool doHandshake(){
char Version;
/** Read C0 **/
fread(&Version, 1, 1, stdin);
uint8_t Client[1536];
uint8_t Server[3072];
fread(&Client, 1, 1536, stdin);
rec_cnt+=1537;
/** Build S1 Packet **/
*((uint32_t*)Server) = 0;//time zero
*(((uint32_t*)(Server+4))) = htonl(0x01020304);//version 1 2 3 4
for (int i = 8; i < 3072; ++i){Server[i] = versionstring[i%13];}//"random" data
bool encrypted = (Version == 6);
#ifdef DEBUG
fprintf(stderr, "Handshake version is %hhi\n", Version);
#endif
uint8_t _validationScheme = 5;
if (ValidateClientScheme(Client, 0)) _validationScheme = 0;
if (ValidateClientScheme(Client, 1)) _validationScheme = 1;
#ifdef DEBUG
fprintf(stderr, "Handshake type is %hhi, encryption is %s\n", _validationScheme, encrypted?"on":"off");
#endif
//**** FIRST 1536 bytes from server response ****//
//compute DH key position
uint32_t serverDHOffset = GetDHOffset(Server, _validationScheme);
uint32_t clientDHOffset = GetDHOffset(Client, _validationScheme);
//generate DH key
DHWrapper dhWrapper(1024);
if (!dhWrapper.Initialize()) return false;
if (!dhWrapper.CreateSharedKey(Client + clientDHOffset, 128)) return false;
if (!dhWrapper.CopyPublicKey(Server + serverDHOffset, 128)) return false;
if (encrypted) {
uint8_t secretKey[128];
if (!dhWrapper.CopySharedKey(secretKey, sizeof (secretKey))) return false;
RC4_KEY _pKeyIn;
RC4_KEY _pKeyOut;
InitRC4Encryption(secretKey, (uint8_t*) & Client[clientDHOffset], (uint8_t*) & Server[serverDHOffset], &_pKeyIn, &_pKeyOut);
uint8_t data[1536];
RC4(&_pKeyIn, 1536, data, data);
RC4(&_pKeyOut, 1536, data, data);
}
//generate the digest
uint32_t serverDigestOffset = GetDigestOffset(Server, _validationScheme);
uint8_t *pTempBuffer = new uint8_t[1536 - 32];
memcpy(pTempBuffer, Server, serverDigestOffset);
memcpy(pTempBuffer + serverDigestOffset, Server + serverDigestOffset + 32, 1536 - serverDigestOffset - 32);
uint8_t *pTempHash = new uint8_t[512];
HMACsha256(pTempBuffer, 1536 - 32, genuineFMSKey, 36, pTempHash);
memcpy(Server + serverDigestOffset, pTempHash, 32);
delete[] pTempBuffer;
delete[] pTempHash;
//**** SECOND 1536 bytes from server response ****//
uint32_t keyChallengeIndex = GetDigestOffset(Client, _validationScheme);
pTempHash = new uint8_t[512];
HMACsha256(Client + keyChallengeIndex, 32, genuineFMSKey, 68, pTempHash);
uint8_t *pLastHash = new uint8_t[512];
HMACsha256(Server + 1536, 1536 - 32, pTempHash, 32, pLastHash);
memcpy(Server + 1536 * 2 - 32, pLastHash, 32);
delete[] pTempHash;
delete[] pLastHash;
//***** DONE BUILDING THE RESPONSE ***//
/** Send response **/
fwrite(&Version, 1, 1, stdout);
fwrite(&Server, 1, 3072, stdout);
snd_cnt+=3073;
/** Flush, necessary in order to work **/
fflush(stdout);
/** Read and discard C2 **/
fread(Client, 1, 1536, stdin);
rec_cnt+=1536;
return true;
}
#endif

119
Connector_RTMPf/main.cpp
View file

@ -1,119 +0,0 @@
#undef DEBUG
#include <iostream>
#include <cstdlib>
#include <cstdio>
#include <cmath>
//needed for select
#include <stdio.h>
#include <stdlib.h>
#include <sys/time.h>
#include <sys/types.h>
#include <unistd.h>
//for connection to server
#include "../sockets/SocketW.h"
bool ready4data = false;//set to true when streaming starts
bool inited = false;
bool stopparsing = false;
timeval lastrec;
#include "parsechunks.cpp" //chunkstream parsing
#include "handshake.cpp" //handshaking
#include "flv_sock.cpp" //FLV parsing with SocketW
int main(){
unsigned int ts;
unsigned int fts = 0;
unsigned int ftst;
SWUnixSocket ss;
fd_set pollset;
struct timeval timeout;
//0 timeout - return immediately after select call
timeout.tv_sec = 1; timeout.tv_usec = 0;
FD_ZERO(&pollset);//clear the polling set
FD_SET(0, &pollset);//add stdin to polling set
//first timestamp set
firsttime = getNowMS();
#ifdef DEBUG
fprintf(stderr, "Doing handshake...\n");
#endif
if (doHandshake()){
#ifdef DEBUG
fprintf(stderr, "Handshake succcess!\n");
#endif
}else{
#ifdef DEBUG
fprintf(stderr, "Handshake fail!\n");
#endif
return 0;
}
#ifdef DEBUG
fprintf(stderr, "Starting processing...\n");
#endif
while (std::cin.good() && std::cout.good()){
//select(1, &pollset, 0, 0, &timeout);
//only parse input from stdin if available or not yet init'ed
//FD_ISSET(0, &pollset) || //NOTE: Polling does not work? WHY?!? WHY DAMN IT?!?
if ((!ready4data || (snd_cnt - snd_window_at >= snd_window_size)) && !stopparsing){parseChunk();fflush(stdout);}
if (ready4data){
if (!inited){
//we are ready, connect the socket!
if (!ss.connect(streamname.c_str())){
#ifdef DEBUG
fprintf(stderr, "Could not connect to server!\n");
#endif
return 0;
}
FLV_Readheader(ss);//read the header, we don't want it
#ifdef DEBUG
fprintf(stderr, "Header read, starting to send video data...\n");
#endif
inited = true;
}
//only send data if previous data has been ACK'ed...
if (snd_cnt - snd_window_at < snd_window_size){
if (FLV_GetPacket(ss)){//able to read a full packet?
ts = FLVbuffer[7] * 256*256*256;
ts += FLVbuffer[4] * 256*256;
ts += FLVbuffer[5] * 256;
ts += FLVbuffer[6];
if (ts != 0){
if (fts == 0){fts = ts;ftst = getNowMS();}
ts -= fts;
FLVbuffer[7] = ts / (256*256*256);
FLVbuffer[4] = ts / (256*256);
FLVbuffer[5] = ts / 256;
FLVbuffer[6] = ts % 256;
ts += ftst;
}else{
ftst = getNowMS();
FLVbuffer[7] = ftst / (256*256*256);
FLVbuffer[4] = ftst / (256*256);
FLVbuffer[5] = ftst / 256;
FLVbuffer[6] = ftst % 256;
}
SendMedia((unsigned char)FLVbuffer[0], (unsigned char *)FLVbuffer+11, FLV_len-15, ts);
FLV_Dump();//dump packet and get ready for next
}
if ((SWBerr != SWBaseSocket::ok) && (SWBerr != SWBaseSocket::notReady)){
#ifdef DEBUG
fprintf(stderr, "No more data! :-( (%s)\n", SWBerr.get_error().c_str());
#endif
return 0;//no more input possible! Fail immediately.
}
}
}
//send ACK if we received a whole window
if (rec_cnt - rec_window_at > rec_window_size){
rec_window_at = rec_cnt;
SendCTL(3, rec_cnt);//send ack (msg 3)
}
}
#ifdef DEBUG
fprintf(stderr, "User disconnected.\n");
#endif
return 0;
}//main

246
Connector_RTMPf/parsechunks.cpp
View file

@ -1,246 +0,0 @@
#include "chunkstream.cpp" //chunkstream decoding
#include "amf.cpp" //simple AMF0 parsing
std::string streamname = "/tmp/shared_socket";
//gets and parses one chunk
void parseChunk(){
static chunkpack next;
static AMFType amfdata("empty", (unsigned char)0xFF);
static AMFType amfelem("empty", (unsigned char)0xFF);
next = getWholeChunk();
switch (next.msg_type_id){
case 0://does not exist
break;//happens when connection breaks unexpectedly
case 1://set chunk size
chunk_rec_max = ntohl(*(int*)next.data);
#ifdef DEBUG
fprintf(stderr, "CTRL: Set chunk size: %i\n", chunk_rec_max);
#endif
break;
case 2://abort message - we ignore this one
#ifdef DEBUG
fprintf(stderr, "CTRL: Abort message\n");
#endif
//4 bytes of stream id to drop
break;
case 3://ack
#ifdef DEBUG
fprintf(stderr, "CTRL: Acknowledgement\n");
#endif
snd_window_at = ntohl(*(int*)next.data);
snd_window_at = snd_cnt;
break;
case 4:{
#ifdef DEBUG
short int ucmtype = ntohs(*(short int*)next.data);
fprintf(stderr, "CTRL: User control message %hi\n", ucmtype);
#endif
//2 bytes event type, rest = event data
//types:
//0 = stream begin, 4 bytes ID
//1 = stream EOF, 4 bytes ID
//2 = stream dry, 4 bytes ID
//3 = setbufferlen, 4 bytes ID, 4 bytes length
//4 = streamisrecorded, 4 bytes ID
//6 = pingrequest, 4 bytes data
//7 = pingresponse, 4 bytes data
//we don't need to process this
} break;
case 5://window size of other end
#ifdef DEBUG
fprintf(stderr, "CTRL: Window size\n");
#endif
rec_window_size = ntohl(*(int*)next.data);
rec_window_at = rec_cnt;
SendCTL(3, rec_cnt);//send ack (msg 3)
break;
case 6:
#ifdef DEBUG
fprintf(stderr, "CTRL: Set peer bandwidth\n");
#endif
//4 bytes window size, 1 byte limit type (ignored)
snd_window_size = ntohl(*(int*)next.data);
SendCTL(5, snd_window_size);//send window acknowledgement size (msg 5)
break;
case 8:
#ifdef DEBUG
fprintf(stderr, "Received audio data\n");
#endif
break;
case 9:
#ifdef DEBUG
fprintf(stderr, "Received video data\n");
#endif
break;
case 15:
#ifdef DEBUG
fprintf(stderr, "Received AFM3 data message\n");
#endif
break;
case 16:
#ifdef DEBUG
fprintf(stderr, "Received AFM3 shared object\n");
#endif
break;
case 17:
#ifdef DEBUG
fprintf(stderr, "Received AFM3 command message\n");
#endif
break;
case 18:
#ifdef DEBUG
fprintf(stderr, "Received AFM0 data message\n");
#endif
break;
case 19:
#ifdef DEBUG
fprintf(stderr, "Received AFM0 shared object\n");
#endif
break;
case 20:{//AMF0 command message
bool parsed = false;
amfdata = parseAMF(next.data, next.real_len);
#ifdef DEBUG
fprintf(stderr, "Received AFM0 command message:\n");
amfdata.Print();
#endif
if (amfdata.getContentP(0)->StrValue() == "connect"){
#ifdef DEBUG
int tmpint;
tmpint = amfdata.getContentP(2)->getContentP("videoCodecs")->NumValue();
if (tmpint & 0x04){fprintf(stderr, "Sorensen video support detected\n");}
if (tmpint & 0x80){fprintf(stderr, "H264 video support detected\n");}
tmpint = amfdata.getContentP(2)->getContentP("audioCodecs")->NumValue();
if (tmpint & 0x04){fprintf(stderr, "MP3 audio support detected\n");}
if (tmpint & 0x400){fprintf(stderr, "AAC video support detected\n");}
#endif
SendCTL(6, rec_window_size, 0);//send peer bandwidth (msg 6)
SendCTL(5, snd_window_size);//send window acknowledgement size (msg 5)
SendUSR(0, 0);//send UCM StreamBegin (0), stream 0
//send a _result reply
AMFType amfreply("container", (unsigned char)0xFF);
amfreply.addContent(AMFType("", "_result"));//result success
amfreply.addContent(amfdata.getContent(1));//same transaction ID
// amfreply.addContent(AMFType("", (double)0, 0x05));//null - command info
amfreply.addContent(AMFType(""));//server properties
amfreply.getContentP(2)->addContent(AMFType("fmsVer", "FMS/3,0,1,123"));//stolen from examples
amfreply.getContentP(2)->addContent(AMFType("capabilities", (double)31));//stolen from examples
amfreply.addContent(AMFType(""));//info
amfreply.getContentP(3)->addContent(AMFType("level", "status"));
amfreply.getContentP(3)->addContent(AMFType("code", "NetConnection.Connect.Success"));
amfreply.getContentP(3)->addContent(AMFType("description", "Connection succeeded."));
amfreply.getContentP(3)->addContent(AMFType("capabilities", (double)33));//from red5 server
amfreply.getContentP(3)->addContent(AMFType("fmsVer", "PLS/1,0,0,0"));//from red5 server
SendChunk(3, 20, next.msg_stream_id, amfreply.Pack());
//send onBWDone packet
//amfreply = AMFType("container", (unsigned char)0xFF);
//amfreply.addContent(AMFType("", "onBWDone"));//result success
//amfreply.addContent(AMFType("", (double)0));//zero
//amfreply.addContent(AMFType("", (double)0, 0x05));//null
//SendChunk(3, 20, next.msg_stream_id, amfreply.Pack());
parsed = true;
}//connect
if (amfdata.getContentP(0)->StrValue() == "createStream"){
//send a _result reply
AMFType amfreply("container", (unsigned char)0xFF);
amfreply.addContent(AMFType("", "_result"));//result success
amfreply.addContent(amfdata.getContent(1));//same transaction ID
amfreply.addContent(AMFType("", (double)0, 0x05));//null - command info
amfreply.addContent(AMFType("", (double)1));//stream ID - we use 1
SendChunk(3, 20, next.msg_stream_id, amfreply.Pack());
SendUSR(0, 0);//send UCM StreamBegin (0), stream 0
#ifdef DEBUG
fprintf(stderr, "AMF0 command: createStream result\n");
#endif
parsed = true;
}//createStream
if ((amfdata.getContentP(0)->StrValue() == "getStreamLength") || (amfdata.getContentP(0)->StrValue() == "getMovLen")){
//send a _result reply
AMFType amfreply("container", (unsigned char)0xFF);
amfreply.addContent(AMFType("", "_result"));//result success
amfreply.addContent(amfdata.getContent(1));//same transaction ID
amfreply.addContent(AMFType("", (double)0, 0x05));//null - command info
amfreply.addContent(AMFType("", (double)0));//zero length
SendChunk(3, 20, next.msg_stream_id, amfreply.Pack());
#ifdef DEBUG
fprintf(stderr, "AMF0 command: getStreamLength result\n");
#endif
parsed = true;
}//getStreamLength
if (amfdata.getContentP(0)->StrValue() == "checkBandwidth"){
//send a _result reply
AMFType amfreply("container", (unsigned char)0xFF);
amfreply.addContent(AMFType("", "_result"));//result success
amfreply.addContent(amfdata.getContent(1));//same transaction ID
amfreply.addContent(AMFType("", (double)0, 0x05));//null - command info
amfreply.addContent(AMFType("", (double)0, 0x05));//null - command info
SendChunk(3, 20, 1, amfreply.Pack());
#ifdef DEBUG
fprintf(stderr, "AMF0 command: checkBandwidth result\n");
#endif
parsed = true;
}//checkBandwidth
if ((amfdata.getContentP(0)->StrValue() == "play") || (amfdata.getContentP(0)->StrValue() == "play2")){
//send streambegin
streamname = amfdata.getContentP(3)->StrValue();
for (std::string::iterator i=streamname.end()-1; i>=streamname.begin(); --i){
if (!isalpha(*i) && !isdigit(*i)){streamname.erase(i);}else{*i=tolower(*i);}
}
streamname = "/tmp/shared_socket_" + streamname;
SendUSR(0, 1);//send UCM StreamBegin (0), stream 1
//send a status reply
AMFType amfreply("container", (unsigned char)0xFF);
amfreply.addContent(AMFType("", "onStatus"));//status reply
amfreply.addContent(amfdata.getContent(1));//same transaction ID
amfreply.addContent(AMFType("", (double)0, 0x05));//null - command info
amfreply.addContent(AMFType(""));//info
amfreply.getContentP(3)->addContent(AMFType("level", "status"));
amfreply.getContentP(3)->addContent(AMFType("code", "NetStream.Play.Reset"));
amfreply.getContentP(3)->addContent(AMFType("description", "Playing and resetting..."));
amfreply.getContentP(3)->addContent(AMFType("details", "PLS"));
amfreply.getContentP(3)->addContent(AMFType("clientid", (double)1));
SendChunk(4, 20, next.msg_stream_id, amfreply.Pack());
amfreply = AMFType("container", (unsigned char)0xFF);
amfreply.addContent(AMFType("", "onStatus"));//status reply
amfreply.addContent(amfdata.getContent(1));//same transaction ID
amfreply.addContent(AMFType("", (double)0, 0x05));//null - command info
amfreply.addContent(AMFType(""));//info
amfreply.getContentP(3)->addContent(AMFType("level", "status"));
amfreply.getContentP(3)->addContent(AMFType("code", "NetStream.Play.Start"));
amfreply.getContentP(3)->addContent(AMFType("description", "Playing!"));
amfreply.getContentP(3)->addContent(AMFType("details", "PLS"));
amfreply.getContentP(3)->addContent(AMFType("clientid", (double)1));
SendChunk(4, 20, 1, amfreply.Pack());
//No clue what this does. Most real servers send it, though...
// amfreply = AMFType("container", (unsigned char)0xFF);
// amfreply.addContent(AMFType("", "|RtmpSampleAccess"));//status reply
// amfreply.addContent(AMFType("", (double)1, 0x01));//bool true - audioaccess
// amfreply.addContent(AMFType("", (double)1, 0x01));//bool true - videoaccess
// SendChunk(4, 20, next.msg_stream_id, amfreply.Pack());
chunk_snd_max = 1024*1024;
SendCTL(1, chunk_snd_max);//send chunk size max (msg 1)
ready4data = true;//start sending video data!
#ifdef DEBUG
fprintf(stderr, "AMF0 command: play result (%s)\n", streamname.c_str());
#endif
parsed = true;
}//createStream
if (!parsed){
#ifdef DEBUG
fprintf(stderr, "AMF0 command not processed! :(\n");
#endif
}
} break;
case 22:
#ifdef DEBUG
fprintf(stderr, "Received aggregate message\n");
#endif
break;
default:
#ifdef DEBUG
fprintf(stderr, "Unknown chunk received! Probably protocol corruption, stopping parsing of incoming data.\n");
#endif
stopparsing = true;
break;
}
}//parseChunk

58
MP4/box.h Normal file
View file

@ -0,0 +1,58 @@
#include "boxheader.h"
class Box {
public:
Box();
Box(uint32_t BoxType);
~Box();
void SetBoxType(uint32_t BoxType);
uint32_t GetBoxType();
void SetPayload(uint32_t PayloadSize, uint8_t * Data);
uint8_t * GetPayload();
uint8_t * GetPayload(uint32_t Index, uint32_t Size);
private:
BoxHeader header;
uint8_t * Payload;
};//Box Class
Box::Box() {
Payload = NULL;
}
Box::Box(uint32_t BoxType) {
header.BoxType = BoxType;
Payload = NULL;
}
Box::~Box() {
}
void Box::SetBoxType(uint32_t BoxType) {
header.BoxType = BoxType;
}
uint32_t Box::GetBoxType() {
return header.BoxType;
}
void Box::SetPayload(uint32_t PayloadSize, uint8_t * Data ) {
if ( Payload != NULL ) { delete Payload; }
Payload = new uint8_t[PayloadSize];
memcpy( Payload, Data, PayloadSize );
header.TotalSize = PayloadSize + 8;
}
uint8_t * Box::GetPayload() {
uint8_t * temp = new uint8_t[header.TotalSize - 8];
memcpy( temp, Payload, header.TotalSize - 8 );
return temp;
}
uint8_t * Box::GetPayload(uint32_t Index, uint32_t Size) {
if(
uint8_t * temp = new uint8_t[header.TotalSize - 8];
memcpy( temp, Payload, header.TotalSize - 8 );
return temp;
}

4
MP4/boxheader.h Normal file
View file

@ -0,0 +1,4 @@
struct BoxHeader {
uint32_t TotalSize;
uint32_t BoxType;
};//BoxHeader struct

6
Makefile
View file

@ -3,24 +3,24 @@ default: client-install
client:
cd Connector_HTTP; $(MAKE)
cd Connector_RTMP; $(MAKE)
cd Connector_RTMPf; $(MAKE)
cd Connector_RAW; $(MAKE)
#cd Connector_RTSP; $(MAKE)
cd Buffer; $(MAKE)
client-clean:
cd Connector_HTTP; $(MAKE) clean
cd Connector_RTMP; $(MAKE) clean
cd Connector_RTMPf; $(MAKE) clean
cd Connector_RAW; $(MAKE) clean
#cd Connector_RTSP; $(MAKE) clean
cd Buffer; $(MAKE) clean
clean: client-clean
client-install: client-clean client
mkdir -p /tmp/cores
chmod 777 /tmp/cores
echo "/tmp/cores/%e.%s.%p" > /proc/sys/kernel/core_pattern
service xinetd stop
cp -f ./Connector_HTTP/Connector_HTTP /usr/bin/
cd Connector_RTMP; $(MAKE) install
cp -f ./Connector_RAW/Connector_RAW /usr/bin/
cp -f ./Connector_RTMPf/Connector_RTMPf /usr/bin/
#cp -f ./Connector_RTSP/Connector_RTSP /usr/bin/
cp -f ./Buffer/Buffer /usr/bin/
cp -f ./PLS /etc/xinetd.d/

15
PLS
View file

@ -25,18 +25,3 @@ service ddvtechraw
per_source = 10
cps = 100 5
}
service ddvtechrtmp
{
disable = no
type = UNLISTED
protocol = tcp
socket_type = stream
user = root
server = /usr/bin/Connector_RTMPf
port = 1935
wait = no
per_source = 10
cps = 100 5
}

53
util/ddv_socket.cpp
View file

@ -33,7 +33,6 @@ int DDV_OpenUnix(std::string adres, bool nonblock = false){
int DDV_Listen(int port){
int s = socket(AF_INET, SOCK_STREAM, 0);
int on = 1;
setsockopt(s, SOL_SOCKET, SO_REUSEADDR, &on, sizeof(on));
struct sockaddr_in addr;
@ -46,12 +45,40 @@ int DDV_Listen(int port){
if (ret == 0){
return s;
}else{
printf("Listen failed! Error: %s\n", strerror(errno));
fprintf(stderr, "Listen failed! Error: %s\n", strerror(errno));
close(s);
return 0;
}
}else{
printf("Binding failed! Error: %s\n", strerror(errno));
fprintf(stderr, "Binding failed! Error: %s\n", strerror(errno));
close(s);
return 0;
}
}
int DDV_UnixListen(std::string adres, bool nonblock = false){
unlink(adres.c_str());
int s = socket(AF_UNIX, SOCK_STREAM, 0);
if (nonblock){
int flags = fcntl(s, F_GETFL, 0);
flags |= O_NONBLOCK;
fcntl(s, F_SETFL, flags);
}
sockaddr_un addr;
addr.sun_family = AF_UNIX;
strncpy(addr.sun_path, adres.c_str(), adres.size()+1);
int ret = bind(s, (sockaddr*)&addr, sizeof(addr));
if (ret == 0){
ret = listen(s, 100);//start listening, backlog of 100 allowed
if (ret == 0){
return s;
}else{
fprintf(stderr, "Listen failed! Error: %s\n", strerror(errno));
close(s);
return 0;
}
}else{
fprintf(stderr, "Binding failed! Error: %s\n", strerror(errno));
close(s);
return 0;
}
@ -59,7 +86,7 @@ int DDV_Listen(int port){
int DDV_Accept(int sock, bool nonblock = false){
int r = accept(sock, 0, 0);
if ((r > 0) && nonblock){
if ((r >= 0) && nonblock){
int flags = fcntl(r, F_GETFL, 0);
flags |= O_NONBLOCK;
fcntl(r, F_SETFL, flags);
@ -77,7 +104,7 @@ bool DDV_write(void * buffer, int todo, int sock){
case EWOULDBLOCK: socketBlocking = true; break;
default:
socketError = true;
printf("Could not write! %s\n", strerror(errno));
fprintf(stderr, "Could not write! %s\n", strerror(errno));
return false;
break;
}
@ -87,14 +114,14 @@ bool DDV_write(void * buffer, int todo, int sock){
return true;
}
bool DDV_ready(int sock){
signed int DDV_ready(int sock){
char tmp;
int preflags = fcntl(sock, F_GETFL, 0);
int postflags = preflags | O_NONBLOCK;
fcntl(sock, F_SETFL, postflags);
int r = recv(sock, &tmp, 1, MSG_PEEK);
fcntl(sock, F_SETFL, preflags);
return (r == 1);
return r;
}
bool DDV_read(void * buffer, int todo, int sock){
@ -107,7 +134,7 @@ bool DDV_read(void * buffer, int todo, int sock){
case EWOULDBLOCK: socketBlocking = true; break;
default:
socketError = true;
printf("Could not read! %s\n", strerror(errno));
fprintf(stderr, "Could not read! %s\n", strerror(errno));
return false;
break;
}
@ -126,11 +153,11 @@ int DDV_iwrite(void * buffer, int todo, int sock){
int r = send(sock, buffer, todo, 0);
if (r < 0){
switch (errno){
case EWOULDBLOCK: break;
case EWOULDBLOCK: return 0; break;
default:
socketError = true;
printf("Could not write! %s\n", strerror(errno));
return false;
fprintf(stderr, "Could not write! %s\n", strerror(errno));
return 0;
break;
}
}
@ -144,8 +171,8 @@ int DDV_iread(void * buffer, int todo, int sock){
case EWOULDBLOCK: break;
default:
socketError = true;
printf("Could not read! %s\n", strerror(errno));
return false;
fprintf(stderr, "Could not read! %s\n", strerror(errno));
return 0;
break;
}
}

8
util/flv_sock.cpp
View file

@ -69,6 +69,7 @@ bool FLV_GetPacket(FLV_Pack *& p, int sock){
if (FLV_Checkheader(p->data)){
sofar = 0;
memcpy(FLVHeader, p->data, 13);
//fwrite(p->data, 13, 1, stdout);//output raw stream
}else{
All_Hell_Broke_Loose = true;
fprintf(stderr, "Invalid FLV header. All Hell Broke Loose!\n");
@ -94,12 +95,11 @@ bool FLV_GetPacket(FLV_Pack *& p, int sock){
testlen += (p->data[p->len-2] << 8);
testlen += (p->data[p->len-3] << 16);
testlen += (p->data[p->len-4] << 24);
if (p->len == testlen){
fprintf(stderr, "Correct length tag...\n");
}else{
//fwrite(p->data, p->len, 1, stdout);//output raw stream
if (p->len != testlen){
fprintf(stderr, "Len: %i, testlen: %i\n", p->len, testlen);
All_Hell_Broke_Loose = true;
fprintf(stderr, "ReadUntil fail: > 500kb tag? All Hell Broke Loose!\n", strerror(errno));
fprintf(stderr, "ReadUntil fail: Wrong size tag? All Hell Broke Loose!\n");
return false;
}
done = true;