av/rtmp/rtmp.go

/*
NAME
  rtmp.go

DESCRIPTION
  RTMP command functionality.

AUTHORS
  Saxon Nelson-Milton <saxon@ausocean.org>
  Dan Kortschak <dan@ausocean.org>!
  Alan Noble <alan@ausocean.org>

LICENSE
  rtmp.go is Copyright (C) 2017-2019 the Australian Ocean Lab (AusOcean)

  It is free software: you can redistribute it and/or modify them
  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.

  It 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 revid in gpl.txt. If not, see http://www.gnu.org/licenses.

  Derived from librtmp under the GNU Lesser General Public License 2.1
  Copyright (C) 2005-2008 Team XBMC http://www.xbmc.org
  Copyright (C) 2008-2009 Andrej Stepanchuk
  Copyright (C) 2009-2010 Howard Chu
*/

package rtmp

import (
	"bytes"
	"encoding/binary"
	"errors"
	"math/rand"
	"net"
	"strconv"
	"time"

	"bitbucket.org/ausocean/av/rtmp/amf"
)

const (
	pkg            = "rtmp:"
	signatureSize  = 1536
	fullHeaderSize = 12
)

// Link flags.
const (
	linkAuth     = 0x0001 // using auth param
	linkLive     = 0x0002 // stream is live
	linkSWF      = 0x0004 // do SWF verification - not implemented
	linkPlaylist = 0x0008 // send playlist before play - not implemented
	linkBufx     = 0x0010 // toggle stream on BufferEmpty msg - not implemented
)

// Protocol features.
const (
	featureHTTP   = 0x01 // not implemented
	featureEncode = 0x02 // not implemented
	featureSSL    = 0x04 // not implemented
	featureMFP    = 0x08 // not implemented
	featureWrite  = 0x10 // publish, not play
	featureHTTP2  = 0x20 // server-side RTMPT - not implemented
)

// RTMP protocols.
const (
	protoRTMP   = 0
	protoRTMPE  = featureEncode
	protoRTMPT  = featureHTTP
	protoRTMPS  = featureSSL
	protoRTMPTE = (featureHTTP | featureEncode)
	protoRTMPTS = (featureHTTP | featureSSL)
	protoRTMFP  = featureMFP
)

// RTMP tokens (lexemes).
// NB: Underscores are deliberately preserved in const names where they exist in the corresponding tokens.
const (
	av_checkbw                       = "_checkbw"
	av_onbwcheck                     = "_onbwcheck"
	av_onbwdone                      = "_onbwdone"
	av_result                        = "_result"
	avApp                            = "app"
	avAudioCodecs                    = "audioCodecs"
	avCapabilities                   = "capabilities"
	avClose                          = "close"
	avCode                           = "code"
	avConnect                        = "connect"
	avCreatestream                   = "createStream"
	avDeletestream                   = "deleteStream"
	avFCPublish                      = "FCPublish"
	avFCUnpublish                    = "FCUnpublish"
	avFlashver                       = "flashVer"
	avFpad                           = "fpad"
	avLevel                          = "level"
	avLive                           = "live"
	avNetConnectionConnectInvalidApp = "NetConnection.Connect.InvalidApp"
	avNetStreamFailed                = "NetStream.Failed"
	avNetStreamPauseNotify           = "NetStream.Pause.Notify"
	avNetStreamPlayComplete          = "NetStream.Play.Complete"
	avNetStreamPlayFailed            = "NetStream.Play.Failed"
	avNetStreamPlayPublishNotify     = "NetStream.Play.PublishNotify"
	avNetStreamPlayStart             = "NetStream.Play.Start"
	avNetStreamPlayStop              = "NetStream.Play.Stop"
	avNetStreamPlayStreamNotFound    = "NetStream.Play.StreamNotFound"
	avNetStreamPlayUnpublishNotify   = "NetStream.Play.UnpublishNotify"
	avNetStreamPublish_Start         = "NetStream.Publish.Start"
	avNetStreamSeekNotify            = "NetStream.Seek.Notify"
	avNonprivate                     = "nonprivate"
	avObjectEncoding                 = "objectEncoding"
	avOnBWDone                       = "onBWDone"
	avOnFCSubscribe                  = "onFCSubscribe"
	avOnFCUnsubscribe                = "onFCUnsubscribe"
	avOnStatus                       = "onStatus"
	avPageUrl                        = "pageUrl"
	avPing                           = "ping"
	avPlay                           = "play"
	avPlaylist_ready                 = "playlist_ready"
	avPublish                        = "publish"
	avReleasestream                  = "releaseStream"
	avSecureToken                    = "secureToken"
	avSet_playlist                   = "set_playlist"
	avSwfUrl                         = "swfUrl"
	avTcUrl                          = "tcUrl"
	avType                           = "type"
	avVideoCodecs                    = "videoCodecs"
	avVideoFunction                  = "videoFunction"
)

// RTMP protocol strings.
var rtmpProtocolStrings = [...]string{
	"rtmp",
	"rtmpt",
	"rtmpe",
	"rtmpte",
	"rtmps",
	"rtmpts",
	"",
	"",
	"rtmfp",
}

// RTMP errors.
var (
	errUnknownScheme = errors.New("rtmp: unknown scheme")
	errInvalidURL    = errors.New("rtmp: invalid URL")
	errConnected     = errors.New("rtmp: already connected")
	errNotConnected  = errors.New("rtmp: not connected")
	errNotWritable   = errors.New("rtmp: connection not writable")
	errInvalidHeader = errors.New("rtmp: invalid header")
	errInvalidBody   = errors.New("rtmp: invalid body")
	errInvalidFlvTag = errors.New("rtmp: invalid FLV tag")
	errUnimplemented = errors.New("rtmp: unimplemented feature")
)

// init initialises the Session link
func (s *Session) init() (err error) {
	s.link.protocol, s.link.host, s.link.port, s.link.app, s.link.playpath, err = parseURL(s.url)
	if err != nil {
		return err
	}
	if s.link.app == "" {
		return errInvalidURL
	}
	if s.link.port == 0 {
		switch {
		case (s.link.protocol & featureSSL) != 0:
			s.link.port = 433
			s.log(FatalLevel, pkg+"SSL not supported")
		case (s.link.protocol & featureHTTP) != 0:
			s.link.port = 80
		default:
			s.link.port = 1935
		}
	}
	s.link.url = rtmpProtocolStrings[s.link.protocol] + "://" + s.link.host + ":" + strconv.Itoa(int(s.link.port)) + "/" + s.link.app
	s.link.protocol |= featureWrite
	return nil
}

// connect establishes an RTMP connection.
func connect(s *Session) error {
	addr, err := net.ResolveTCPAddr("tcp4", s.link.host+":"+strconv.Itoa(int(s.link.port)))
	if err != nil {
		return err
	}
	s.link.conn, err = net.DialTCP("tcp4", nil, addr)
	if err != nil {
		s.log(WarnLevel, pkg+"dial failed", "error", err.Error())
		return err
	}
	s.log(DebugLevel, pkg+"connected")
	err = handshake(s)
	if err != nil {
		s.log(WarnLevel, pkg+"handshake failed", "error", err.Error())
		return err
	}
	s.log(DebugLevel, pkg+"handshaked")
	err = sendConnectPacket(s)
	if err != nil {
		s.log(WarnLevel, pkg+"sendConnect failed", "error", err.Error())
		return err
	}
	return nil
}

// connectStream reads a packet and handles it
func connectStream(s *Session) error {
	var err error
	for !s.isPlaying {
		pkt := packet{}
		err = pkt.readFrom(s)
		if err != nil {
			break
		}

		switch pkt.packetType {
		case packetTypeAudio, packetTypeVideo, packetTypeInfo:
			s.log(WarnLevel, pkg+"got packet before play; ignoring", "type", pkt.packetType)
		default:
			err = handlePacket(s, &pkt)
			if err != nil {
				break
			}
		}
	}

	if !s.isPlaying {
		return err
	}
	return nil
}

// handlePacket handles a packet that the client has received.
// NB: Unsupported packet types are logged fatally.
func handlePacket(s *Session, pkt *packet) error {
	if pkt.bodySize < 4 {
		return errInvalidBody
	}

	switch pkt.packetType {
	case packetTypeChunkSize:
		s.inChunkSize = int32(amf.DecodeInt32(pkt.body[:4]))

	case packetTypeBytesReadReport:
		s.serverBW = int32(amf.DecodeInt32(pkt.body[:4]))

	case packetTypeServerBW:
		s.serverBW = int32(amf.DecodeInt32(pkt.body[:4]))

	case packetTypeClientBW:
		s.clientBW = int32(amf.DecodeInt32(pkt.body[:4]))
		if pkt.bodySize > 4 {
			s.clientBW2 = pkt.body[4]
		} else {
			s.clientBW2 = 0xff
		}

	case packetTypeInvoke:
		err := handleInvoke(s, pkt.body[:pkt.bodySize])
		if err != nil {
			s.log(WarnLevel, pkg+"unexpected error from handleInvoke", "error", err.Error())
			return err
		}

	case packetTypeControl, packetTypeAudio, packetTypeVideo, packetTypeFlashVideo, packetTypeFlexMessage, packetTypeInfo:
		s.log(FatalLevel, pkg+"unsupported packet type "+strconv.Itoa(int(pkt.packetType)))

	default:
		s.log(WarnLevel, pkg+"unknown packet type", "type", pkt.packetType)
	}
	return nil
}

func sendConnectPacket(s *Session) error {
	var pbuf [4096]byte
	pkt := packet{
		channel:    chanControl,
		headerType: headerSizeLarge,
		packetType: packetTypeInvoke,
		header:     pbuf[:],
		body:       pbuf[fullHeaderSize:],
	}
	enc := pkt.body

	enc, err := amf.EncodeString(enc, avConnect)
	if err != nil {
		return err
	}
	s.numInvokes += 1
	enc, err = amf.EncodeNumber(enc, float64(s.numInvokes))
	if err != nil {
		return err
	}

	enc[0] = amf.TypeObject
	enc = enc[1:]
	enc, err = amf.EncodeNamedString(enc, avApp, s.link.app)
	if err != nil {
		return err
	}
	enc, err = amf.EncodeNamedString(enc, avType, avNonprivate)
	if err != nil {
		return err
	}
	enc, err = amf.EncodeNamedString(enc, avTcUrl, s.link.url)
	if err != nil {
		return err
	}
	enc, err = amf.EncodeInt24(enc, amf.TypeObjectEnd)
	if err != nil {
		return err
	}

	// add auth string, if any
	if s.link.auth != "" {
		enc, err = amf.EncodeBoolean(enc, s.link.flags&linkAuth != 0)
		if err != nil {
			return err
		}
		enc, err = amf.EncodeString(enc, s.link.auth)
		if err != nil {
			return err
		}
	}

	pkt.bodySize = uint32((len(pbuf) - fullHeaderSize) - len(enc))

	return pkt.writeTo(s, true) // response expected
}

func sendCreateStream(s *Session) error {
	var pbuf [256]byte
	pkt := packet{
		channel:    chanControl,
		headerType: headerSizeMedium,
		packetType: packetTypeInvoke,
		header:     pbuf[:],
		body:       pbuf[fullHeaderSize:],
	}
	enc := pkt.body

	enc, err := amf.EncodeString(enc, avCreatestream)
	if err != nil {
		return err
	}
	s.numInvokes++
	enc, err = amf.EncodeNumber(enc, float64(s.numInvokes))
	if err != nil {
		return err
	}
	enc[0] = amf.TypeNull
	enc = enc[1:]

	pkt.bodySize = uint32((len(pbuf) - fullHeaderSize) - len(enc))

	return pkt.writeTo(s, true) // response expected
}

func sendReleaseStream(s *Session) error {
	var pbuf [1024]byte
	pkt := packet{
		channel:    chanControl,
		headerType: headerSizeMedium,
		packetType: packetTypeInvoke,
		header:     pbuf[:],
		body:       pbuf[fullHeaderSize:],
	}
	enc := pkt.body

	enc, err := amf.EncodeString(enc, avReleasestream)
	if err != nil {
		return err
	}
	s.numInvokes++
	enc, err = amf.EncodeNumber(enc, float64(s.numInvokes))
	if err != nil {
		return err
	}
	enc[0] = amf.TypeNull
	enc = enc[1:]
	enc, err = amf.EncodeString(enc, s.link.playpath)
	if err != nil {
		return err
	}
	pkt.bodySize = uint32((len(pbuf) - fullHeaderSize) - len(enc))

	return pkt.writeTo(s, false)
}

func sendFCPublish(s *Session) error {
	var pbuf [1024]byte
	pkt := packet{
		channel:    chanControl,
		headerType: headerSizeMedium,
		packetType: packetTypeInvoke,
		header:     pbuf[:],
		body:       pbuf[fullHeaderSize:],
	}
	enc := pkt.body

	enc, err := amf.EncodeString(enc, avFCPublish)
	if err != nil {
		return err
	}
	s.numInvokes++
	enc, err = amf.EncodeNumber(enc, float64(s.numInvokes))
	if err != nil {
		return err
	}
	enc[0] = amf.TypeNull
	enc = enc[1:]
	enc, err = amf.EncodeString(enc, s.link.playpath)
	if err != nil {
		return err
	}

	pkt.bodySize = uint32((len(pbuf) - fullHeaderSize) - len(enc))

	return pkt.writeTo(s, false)
}

func sendFCUnpublish(s *Session) error {
	var pbuf [1024]byte
	pkt := packet{
		channel:    chanControl,
		headerType: headerSizeMedium,
		packetType: packetTypeInvoke,
		header:     pbuf[:],
		body:       pbuf[fullHeaderSize:],
	}
	enc := pkt.body

	enc, err := amf.EncodeString(enc, avFCUnpublish)
	if err != nil {
		return err
	}
	s.numInvokes++
	enc, err = amf.EncodeNumber(enc, float64(s.numInvokes))
	if err != nil {
		return err
	}
	enc[0] = amf.TypeNull
	enc = enc[1:]
	enc, err = amf.EncodeString(enc, s.link.playpath)
	if err != nil {
		return err
	}

	pkt.bodySize = uint32((len(pbuf) - fullHeaderSize) - len(enc))

	return pkt.writeTo(s, false)
}

func sendPublish(s *Session) error {
	var pbuf [1024]byte
	pkt := packet{
		channel:    chanSource,
		headerType: headerSizeLarge,
		packetType: packetTypeInvoke,
		header:     pbuf[:],
		body:       pbuf[fullHeaderSize:],
	}
	enc := pkt.body

	enc, err := amf.EncodeString(enc, avPublish)
	if err != nil {
		return err
	}
	s.numInvokes++
	enc, err = amf.EncodeNumber(enc, float64(s.numInvokes))
	if err != nil {
		return err
	}
	enc[0] = amf.TypeNull
	enc = enc[1:]
	enc, err = amf.EncodeString(enc, s.link.playpath)
	if err != nil {
		return err
	}
	enc, err = amf.EncodeString(enc, avLive)
	if err != nil {
		return err
	}

	pkt.bodySize = uint32((len(pbuf) - fullHeaderSize) - len(enc))

	return pkt.writeTo(s, true) // response expected
}

func sendDeleteStream(s *Session, dStreamId float64) error {
	var pbuf [256]byte
	pkt := packet{
		channel:    chanControl,
		headerType: headerSizeMedium,
		packetType: packetTypeInvoke,
		header:     pbuf[:],
		body:       pbuf[fullHeaderSize:],
	}
	enc := pkt.body

	enc, err := amf.EncodeString(enc, avDeletestream)
	if err != nil {
		return err
	}
	s.numInvokes++
	enc, err = amf.EncodeNumber(enc, float64(s.numInvokes))
	if err != nil {
		return err
	}
	enc[0] = amf.TypeNull
	enc = enc[1:]
	enc, err = amf.EncodeNumber(enc, dStreamId)
	if err != nil {
		return err
	}
	pkt.bodySize = uint32((len(pbuf) - fullHeaderSize) - len(enc))

	return pkt.writeTo(s, false)
}

// sendBytesReceived tells the server how many bytes the client has received.
func sendBytesReceived(s *Session) error {
	var pbuf [256]byte
	pkt := packet{
		channel:    chanBytesRead,
		headerType: headerSizeMedium,
		packetType: packetTypeBytesReadReport,
		header:     pbuf[:],
		body:       pbuf[fullHeaderSize:],
	}
	enc := pkt.body

	s.nBytesInSent = s.nBytesIn

	enc, err := amf.EncodeInt32(enc, s.nBytesIn)
	if err != nil {
		return err
	}
	pkt.bodySize = 4

	return pkt.writeTo(s, false)
}

func sendCheckBW(s *Session) error {
	var pbuf [256]byte
	pkt := packet{
		channel:    chanControl,
		headerType: headerSizeLarge,
		packetType: packetTypeInvoke,
		header:     pbuf[:],
		body:       pbuf[fullHeaderSize:],
	}
	enc := pkt.body

	enc, err := amf.EncodeString(enc, av_checkbw)
	if err != nil {
		return err
	}
	s.numInvokes++
	enc, err = amf.EncodeNumber(enc, float64(s.numInvokes))
	if err != nil {
		return err
	}
	enc[0] = amf.TypeNull
	enc = enc[1:]

	pkt.bodySize = uint32((len(pbuf) - fullHeaderSize) - len(enc))

	return pkt.writeTo(s, false)
}

func eraseMethod(m []method, i int) []method {
	copy(m[i:], m[i+1:])
	m[len(m)-1] = method{}
	return m[:len(m)-1]
}

// int handleInvoke handles a packet invoke request
// Side effects: s.isPlaying set to true upon avNetStreamPublish_Start
func handleInvoke(s *Session, body []byte) error {
	if body[0] != 0x02 {
		return errInvalidBody
	}
	var obj amf.Object
	_, err := amf.Decode(&obj, body, false)
	if err != nil {
		return err
	}

	meth, err := obj.StringProperty("", 0)
	if err != nil {
		return err
	}
	txn, err := obj.NumberProperty("", 1)
	if err != nil {
		return err
	}

	s.log(DebugLevel, pkg+"invoking method "+meth)
	switch meth {
	case av_result:
		if (s.link.protocol & featureWrite) == 0 {
			return errNotWritable
		}
		var methodInvoked string
		for i, m := range s.methodCalls {
			if float64(m.num) == txn {
				methodInvoked = m.name
				s.methodCalls = eraseMethod(s.methodCalls, i)
				break
			}
		}
		if methodInvoked == "" {
			s.log(WarnLevel, pkg+"received result without matching request", "id", txn)
			return nil
		}
		s.log(DebugLevel, pkg+"received result for "+methodInvoked)

		switch methodInvoked {
		case avConnect:
			err := sendReleaseStream(s)
			if err != nil {
				return err
			}
			err = sendFCPublish(s)
			if err != nil {
				return err
			}
			err = sendCreateStream(s)
			if err != nil {
				return err
			}

		case avCreatestream:
			n, err := obj.NumberProperty("", 3)
			if err != nil {
				return err
			}
			s.streamID = int32(n)
			err = sendPublish(s)
			if err != nil {
				return err
			}

		default:
			s.log(FatalLevel, pkg+"unexpected method invoked"+methodInvoked)
		}

	case avOnBWDone:
		err := sendCheckBW(s)
		if err != nil {
			return err
		}

	case avOnStatus:
		obj2, err := obj.ObjectProperty("", 3)
		if err != nil {
			return err
		}
		code, err := obj2.StringProperty(avCode, -1)
		if err != nil {
			return err
		}
		level, err := obj2.StringProperty(avLevel, -1)
		if err != nil {
			return err
		}
		s.log(DebugLevel, pkg+"onStatus", "code", code, "level", level)

		if code != avNetStreamPublish_Start {
			s.log(ErrorLevel, pkg+"unexpected response "+code)
			return errUnimplemented
		}
		s.log(DebugLevel, pkg+"playing")
		s.isPlaying = true
		for i, m := range s.methodCalls {
			if m.name == avPublish {
				s.methodCalls = eraseMethod(s.methodCalls, i)
			}
		}

	default:
		s.log(FatalLevel, pkg+"unsuppoted method "+meth)
	}
	return nil
}

func handshake(s *Session) error {
	var clientbuf [signatureSize + 1]byte
	clientsig := clientbuf[1:]

	var serversig [signatureSize]byte
	clientbuf[0] = chanControl
	binary.BigEndian.PutUint32(clientsig, uint32(time.Now().UnixNano()/1000000))
	copy(clientsig[4:8], []byte{0, 0, 0, 0})

	for i := 8; i < signatureSize; i++ {
		clientsig[i] = byte(rand.Intn(256))
	}

	_, err := s.write(clientbuf[:])
	if err != nil {
		return err
	}
	s.log(DebugLevel, pkg+"handshake sent")

	var typ [1]byte
	_, err = s.read(typ[:])
	if err != nil {
		return err
	}
	s.log(DebugLevel, pkg+"handshake received")

	if typ[0] != clientbuf[0] {
		s.log(WarnLevel, pkg+"handshake type mismatch", "sent", clientbuf[0], "received", typ)
	}
	_, err = s.read(serversig[:])
	if err != nil {
		return err
	}

	// decode server response
	suptime := binary.BigEndian.Uint32(serversig[:4])
	s.log(DebugLevel, pkg+"server uptime", "uptime", suptime)

	// 2nd part of handshake
	_, err = s.write(serversig[:])
	if err != nil {
		return err
	}

	_, err = s.read(serversig[:])
	if err != nil {
		return err
	}

	if !bytes.Equal(serversig[:signatureSize], clientbuf[1:signatureSize+1]) {
		s.log(WarnLevel, pkg+"signature mismatch", "serversig", serversig[:signatureSize], "clientsig", clientbuf[1:signatureSize+1])
	}
	return nil
}