1
Fork 0
mirror of https://github.com/jech/galene.git synced 2024-11-22 16:45:58 +01:00
galene/rtpconn/webclient.go

1485 lines
30 KiB
Go
Raw Normal View History

package rtpconn
2020-04-24 19:38:21 +02:00
import (
"encoding/json"
"errors"
"fmt"
2020-04-24 19:38:21 +02:00
"log"
"os"
"sync"
"time"
"github.com/gorilla/websocket"
2020-07-16 20:17:32 +02:00
"github.com/pion/webrtc/v3"
2020-04-24 19:38:21 +02:00
2020-12-19 17:37:48 +01:00
"github.com/jech/galene/conn"
"github.com/jech/galene/diskwriter"
"github.com/jech/galene/estimator"
"github.com/jech/galene/group"
2020-04-24 19:38:21 +02:00
)
2020-11-30 16:26:11 +01:00
func errorToWSCloseMessage(id string, err error) (*clientMessage, []byte) {
2020-04-24 19:38:21 +02:00
var code int
2020-11-30 16:26:11 +01:00
var m *clientMessage
2020-04-24 19:38:21 +02:00
var text string
switch e := err.(type) {
case *websocket.CloseError:
code = websocket.CloseNormalClosure
case group.ProtocolError:
2020-04-24 19:38:21 +02:00
code = websocket.CloseProtocolError
2020-12-02 19:33:53 +01:00
s := e.Error()
2020-11-30 16:26:11 +01:00
m = &clientMessage{
2020-12-19 17:38:47 +01:00
Type: "usermessage",
Kind: "error",
Dest: id,
Privileged: true,
Value: &s,
2020-11-30 16:26:11 +01:00
}
text = e.Error()
case group.UserError, group.KickError:
code = websocket.CloseNormalClosure
2020-11-30 16:26:11 +01:00
m = errorMessage(id, err)
text = e.Error()
2020-04-24 19:38:21 +02:00
default:
code = websocket.CloseInternalServerErr
}
2020-11-30 16:26:11 +01:00
return m, websocket.FormatCloseMessage(code, text)
2020-04-24 19:38:21 +02:00
}
func isWSNormalError(err error) bool {
return websocket.IsCloseError(err,
websocket.CloseNormalClosure,
websocket.CloseGoingAway)
}
2020-05-27 11:44:49 +02:00
type webClient struct {
group *group.Group
2020-05-27 11:44:49 +02:00
id string
2020-11-29 14:26:42 +01:00
username string
password string
permissions group.ClientPermissions
2020-05-27 11:44:49 +02:00
requested map[string]uint32
done chan struct{}
writeCh chan interface{}
writerDone chan struct{}
actionCh chan interface{}
mu sync.Mutex
down map[string]*rtpDownConnection
2020-06-08 22:14:28 +02:00
up map[string]*rtpUpConnection
2020-05-27 11:44:49 +02:00
}
func (c *webClient) Group() *group.Group {
2020-05-28 02:35:09 +02:00
return c.group
}
2020-06-08 22:14:28 +02:00
func (c *webClient) Id() string {
2020-05-28 02:35:09 +02:00
return c.id
}
2020-11-29 14:26:42 +01:00
func (c *webClient) Username() string {
return c.username
}
func (c *webClient) Challenge(group string, creds group.ClientCredentials) bool {
if creds.Password == nil {
return true
}
m, err := creds.Password.Match(c.password)
if err != nil {
log.Printf("Password match: %v", err)
return false
}
return m
2020-05-28 02:35:09 +02:00
}
func (c *webClient) SetPermissions(perms group.ClientPermissions) {
c.permissions = perms
}
func (c *webClient) OverridePermissions(g *group.Group) bool {
return false
}
func (c *webClient) PushClient(id, username string, add bool) error {
2020-08-12 13:51:31 +02:00
kind := "add"
if !add {
kind = "delete"
}
return c.write(clientMessage{
Type: "user",
2020-08-12 13:51:31 +02:00
Kind: kind,
Id: id,
Username: username,
})
}
type rateMap map[string]uint32
func (v *rateMap) UnmarshalJSON(b []byte) error {
var m map[string]interface{}
err := json.Unmarshal(b, &m)
if err != nil {
return err
}
n := make(map[string]uint32, len(m))
for k, w := range m {
switch w := w.(type) {
case bool:
if w {
n[k] = ^uint32(0)
} else {
n[k] = 0
}
case float64:
if w < 0 || w >= float64(^uint32(0)) {
return errors.New("overflow")
}
n[k] = uint32(w)
default:
return errors.New("unexpected type in JSON map")
}
}
*v = n
return nil
}
func (v rateMap) MarshalJSON() ([]byte, error) {
m := make(map[string]interface{}, len(v))
for k, w := range v {
switch w {
case 0:
m[k] = false
case ^uint32(0):
m[k] = true
default:
m[k] = w
}
}
return json.Marshal(m)
}
2020-04-24 19:38:21 +02:00
type clientMessage struct {
Type string `json:"type"`
2020-08-12 13:56:35 +02:00
Kind string `json:"kind,omitempty"`
Id string `json:"id,omitempty"`
2020-10-01 16:52:01 +02:00
Dest string `json:"dest,omitempty"`
Username string `json:"username,omitempty"`
Password string `json:"password,omitempty"`
2020-12-14 10:35:12 +01:00
Privileged bool `json:"privileged,omitempty"`
Permissions *group.ClientPermissions `json:"permissions,omitempty"`
Group string `json:"group,omitempty"`
2020-12-02 19:33:53 +01:00
Value *string `json:"value,omitempty"`
2020-10-08 14:38:33 +02:00
Time int64 `json:"time,omitempty"`
Offer *webrtc.SessionDescription `json:"offer,omitempty"`
Answer *webrtc.SessionDescription `json:"answer,omitempty"`
Candidate *webrtc.ICECandidateInit `json:"candidate,omitempty"`
Labels map[string]string `json:"labels,omitempty"`
Request rateMap `json:"request,omitempty"`
2020-04-24 19:38:21 +02:00
}
type closeMessage struct {
data []byte
}
2020-06-08 22:14:28 +02:00
func getUpConn(c *webClient, id string) *rtpUpConnection {
c.mu.Lock()
defer c.mu.Unlock()
2020-04-24 19:38:21 +02:00
if c.up == nil {
return nil
}
2020-06-08 22:14:28 +02:00
return c.up[id]
2020-04-24 19:38:21 +02:00
}
2020-06-08 22:14:28 +02:00
func getUpConns(c *webClient) []*rtpUpConnection {
c.mu.Lock()
defer c.mu.Unlock()
2020-06-08 22:14:28 +02:00
up := make([]*rtpUpConnection, 0, len(c.up))
for _, u := range c.up {
up = append(up, u)
}
return up
}
func addUpConn(c *webClient, id string) (*rtpUpConnection, bool, error) {
c.mu.Lock()
defer c.mu.Unlock()
if c.up == nil {
2020-06-08 22:14:28 +02:00
c.up = make(map[string]*rtpUpConnection)
}
if c.down != nil && c.down[id] != nil {
return nil, false, errors.New("Adding duplicate connection")
}
old := c.up[id]
if old != nil {
return old, false, nil
}
conn, err := newUpConn(c, id)
if err != nil {
return nil, false, err
}
c.up[id] = conn
2020-06-08 22:14:28 +02:00
conn.pc.OnICECandidate(func(candidate *webrtc.ICECandidate) {
2020-04-24 19:38:21 +02:00
sendICE(c, id, candidate)
})
conn.pc.OnICEConnectionStateChange(func(state webrtc.ICEConnectionState) {
if state == webrtc.ICEConnectionStateFailed {
c.action(connectionFailedAction{id: id})
}
})
return conn, true, nil
}
2020-04-24 19:38:21 +02:00
2020-05-27 11:44:49 +02:00
func delUpConn(c *webClient, id string) bool {
c.mu.Lock()
2020-04-24 19:38:21 +02:00
if c.up == nil {
c.mu.Unlock()
return false
2020-04-24 19:38:21 +02:00
}
conn := c.up[id]
if conn == nil {
c.mu.Unlock()
return false
2020-04-24 19:38:21 +02:00
}
delete(c.up, id)
c.mu.Unlock()
2020-04-24 19:38:21 +02:00
g := c.group
if g != nil {
go func(clients []group.Client) {
for _, c := range clients {
err := c.PushConn(g, conn.id, nil, nil, "")
2020-12-04 12:15:37 +01:00
if err != nil {
log.Printf("PushConn: %v", err)
}
}
}(g.GetClients(c))
} else {
log.Printf("Deleting connection for client with no group")
}
conn.pc.Close()
return true
}
2020-05-27 11:44:49 +02:00
func getDownConn(c *webClient, id string) *rtpDownConnection {
2020-04-24 19:38:21 +02:00
if c.down == nil {
return nil
}
c.mu.Lock()
defer c.mu.Unlock()
2020-04-24 19:38:21 +02:00
conn := c.down[id]
if conn == nil {
return nil
}
return conn
}
2020-05-27 11:44:49 +02:00
func getConn(c *webClient, id string) iceConnection {
up := getUpConn(c, id)
if up != nil {
return up
}
down := getDownConn(c, id)
if down != nil {
return down
}
return nil
}
2020-09-13 11:04:16 +02:00
func addDownConn(c *webClient, id string, remote conn.Up) (*rtpDownConnection, error) {
2020-09-13 13:01:06 +02:00
conn, err := newDownConn(c, id, remote)
2020-04-24 19:38:21 +02:00
if err != nil {
return nil, err
}
err = addDownConnHelper(c, conn, remote)
if err != nil {
conn.pc.Close()
return nil, err
}
return conn, err
}
2020-10-01 16:52:01 +02:00
func addDownConnHelper(c *webClient, conn *rtpDownConnection, remote conn.Up) error {
2020-06-08 22:14:28 +02:00
c.mu.Lock()
defer c.mu.Unlock()
2020-04-24 19:38:21 +02:00
if c.up != nil && c.up[conn.id] != nil {
return errors.New("Adding duplicate connection")
}
2020-04-24 19:38:21 +02:00
if c.down == nil {
c.down = make(map[string]*rtpDownConnection)
2020-04-24 19:38:21 +02:00
}
2020-05-23 02:22:43 +02:00
old := c.down[conn.id]
if old != nil {
// Avoid calling Close under a lock
go old.pc.Close()
2020-04-24 19:38:21 +02:00
}
2020-06-08 22:14:28 +02:00
conn.pc.OnICECandidate(func(candidate *webrtc.ICECandidate) {
sendICE(c, conn.id, candidate)
2020-06-08 22:14:28 +02:00
})
conn.pc.OnICEConnectionStateChange(func(state webrtc.ICEConnectionState) {
if state == webrtc.ICEConnectionStateFailed {
c.action(connectionFailedAction{id: conn.id})
}
})
err := remote.AddLocal(conn)
2020-05-30 03:36:15 +02:00
if err != nil {
return err
2020-05-30 03:36:15 +02:00
}
2020-05-23 02:22:43 +02:00
c.down[conn.id] = conn
return nil
2020-04-24 19:38:21 +02:00
}
2020-05-27 11:44:49 +02:00
func delDownConn(c *webClient, id string) bool {
conn := delDownConnHelper(c, id)
if conn != nil {
conn.pc.Close()
return true
}
return false
}
func delDownConnHelper(c *webClient, id string) *rtpDownConnection {
c.mu.Lock()
defer c.mu.Unlock()
2020-04-24 19:38:21 +02:00
if c.down == nil {
return nil
2020-04-24 19:38:21 +02:00
}
2020-04-25 02:25:51 +02:00
conn := c.down[id]
2020-04-24 19:38:21 +02:00
if conn == nil {
return nil
2020-04-24 19:38:21 +02:00
}
2020-09-13 11:04:16 +02:00
conn.remote.DelLocal(conn)
for _, track := range conn.tracks {
// we only insert the track after we get an answer, so
// ignore errors here.
2020-09-13 11:04:16 +02:00
track.remote.DelLocal(track)
}
2020-04-24 19:38:21 +02:00
delete(c.down, id)
return conn
2020-04-24 19:38:21 +02:00
}
2020-09-13 11:04:16 +02:00
func addDownTrack(c *webClient, conn *rtpDownConnection, remoteTrack conn.UpTrack, remoteConn conn.Up) (*webrtc.RTPSender, error) {
2020-12-04 01:15:52 +01:00
rt, ok := remoteTrack.(*rtpUpTrack)
if !ok {
return nil, errors.New("unexpected up track type")
2020-06-08 22:14:28 +02:00
}
2020-12-04 01:15:52 +01:00
local, err := webrtc.NewTrackLocalStaticRTP(
remoteTrack.Codec(),
rt.track.ID(), rt.track.StreamID(),
)
2020-04-24 19:38:21 +02:00
if err != nil {
return nil, err
2020-04-24 19:38:21 +02:00
}
2020-12-04 01:15:52 +01:00
sender, err := conn.pc.AddTrack(local)
if err != nil {
return nil, err
}
2020-04-26 01:33:18 +02:00
2020-12-04 01:15:52 +01:00
parms := sender.GetParameters()
if len(parms.Encodings) != 1 {
return nil, errors.New("got multiple encodings")
}
track := &rtpDownTrack{
track: local,
2020-12-04 01:15:52 +01:00
ssrc: parms.Encodings[0].SSRC,
remote: remoteTrack,
maxBitrate: new(bitrate),
stats: new(receiverStats),
rate: estimator.New(time.Second),
2020-04-28 15:26:50 +02:00
}
conn.tracks = append(conn.tracks, track)
2020-12-04 01:15:52 +01:00
go rtcpDownListener(conn, track, sender)
2020-04-24 19:38:21 +02:00
2020-12-04 01:15:52 +01:00
return sender, nil
2020-04-24 19:38:21 +02:00
}
func negotiate(c *webClient, down *rtpDownConnection, renegotiate, restartIce bool) error {
options := webrtc.OfferOptions{ICERestart: restartIce}
offer, err := down.pc.CreateOffer(&options)
2020-04-24 19:38:21 +02:00
if err != nil {
return err
2020-04-24 19:38:21 +02:00
}
err = down.pc.SetLocalDescription(offer)
2020-04-24 19:38:21 +02:00
if err != nil {
return err
}
labels := make(map[string]string)
for _, t := range down.pc.GetTransceivers() {
2020-12-04 01:15:52 +01:00
var track webrtc.TrackLocal
if t.Sender() != nil {
track = t.Sender().Track()
}
if track == nil {
continue
}
for _, tr := range down.tracks {
if tr.track == track {
2020-06-08 22:14:28 +02:00
labels[t.Mid()] = tr.remote.Label()
}
}
2020-04-24 19:38:21 +02:00
}
2020-08-12 13:56:35 +02:00
kind := ""
if renegotiate {
kind = "renegotiate"
}
2020-04-24 19:38:21 +02:00
return c.write(clientMessage{
2020-08-12 13:56:35 +02:00
Type: "offer",
Kind: kind,
Id: down.id,
Offer: &offer,
Labels: labels,
2020-04-24 19:38:21 +02:00
})
}
2020-05-27 11:44:49 +02:00
func sendICE(c *webClient, id string, candidate *webrtc.ICECandidate) error {
2020-04-24 19:38:21 +02:00
if candidate == nil {
return nil
}
cand := candidate.ToJSON()
return c.write(clientMessage{
Type: "ice",
Id: id,
Candidate: &cand,
})
}
func gotOffer(c *webClient, id string, offer webrtc.SessionDescription, renegotiate bool, labels map[string]string) error {
if !renegotiate {
// unless the client indicates that this is a compatible
// renegotiation, tear down the existing connection.
delUpConn(c, id)
}
up, isnew, err := addUpConn(c, id)
if err != nil {
return err
2020-04-24 19:38:21 +02:00
}
2020-11-29 14:26:42 +01:00
if u := c.Username(); u != "" {
up.label = u
2020-04-24 19:38:21 +02:00
}
err = up.pc.SetRemoteDescription(offer)
if err != nil {
if renegotiate && !isnew {
// create a new PC from scratch
log.Printf("SetRemoteDescription(offer): %v", err)
return gotOffer(c, id, offer, false, labels)
}
2020-04-24 19:38:21 +02:00
return err
}
answer, err := up.pc.CreateAnswer(nil)
if err != nil {
return err
}
err = up.pc.SetLocalDescription(answer)
if err != nil {
return err
}
up.labels = labels
2020-05-22 23:07:38 +02:00
err = up.flushICECandidates()
if err != nil {
log.Printf("ICE: %v", err)
}
2020-04-24 19:38:21 +02:00
return c.write(clientMessage{
Type: "answer",
Id: id,
Answer: &answer,
})
}
var ErrUnknownId = errors.New("unknown id")
2020-05-27 11:44:49 +02:00
func gotAnswer(c *webClient, id string, answer webrtc.SessionDescription) error {
2020-05-22 23:07:38 +02:00
down := getDownConn(c, id)
if down == nil {
return ErrUnknownId
2020-04-24 19:38:21 +02:00
}
2020-05-22 23:07:38 +02:00
err := down.pc.SetRemoteDescription(answer)
2020-04-24 19:38:21 +02:00
if err != nil {
return err
}
2020-05-21 00:55:00 +02:00
2020-05-22 23:07:38 +02:00
err = down.flushICECandidates()
if err != nil {
log.Printf("ICE: %v", err)
}
add := func() {
down.pc.OnConnectionStateChange(nil)
for _, t := range down.tracks {
t.remote.AddLocal(t)
}
2020-05-21 00:55:00 +02:00
}
down.pc.OnConnectionStateChange(func(state webrtc.PeerConnectionState) {
if state == webrtc.PeerConnectionStateConnected {
add()
}
})
if down.pc.ConnectionState() == webrtc.PeerConnectionStateConnected {
add()
}
2020-04-24 19:38:21 +02:00
return nil
}
2020-05-27 11:44:49 +02:00
func gotICE(c *webClient, candidate *webrtc.ICECandidateInit, id string) error {
conn := getConn(c, id)
if conn == nil {
return errors.New("unknown id in ICE")
2020-04-24 19:38:21 +02:00
}
2020-05-22 23:07:38 +02:00
return conn.addICECandidate(candidate)
2020-04-24 19:38:21 +02:00
}
2020-05-27 11:44:49 +02:00
func (c *webClient) setRequested(requested map[string]uint32) error {
2020-05-09 19:39:34 +02:00
if c.down != nil {
down := make([]string, 0, len(c.down))
2020-05-09 19:39:34 +02:00
for id := range c.down {
down = append(down, id)
}
for _, id := range down {
2020-05-09 19:39:34 +02:00
c.write(clientMessage{
Type: "close",
Id: id,
})
delDownConn(c, id)
}
}
c.requested = requested
2020-05-09 19:39:34 +02:00
go pushConns(c, c.group)
2020-05-09 19:39:34 +02:00
return nil
}
func pushConns(c group.Client, g *group.Group) {
clients := g.GetClients(c)
2020-05-30 00:23:54 +02:00
for _, cc := range clients {
ccc, ok := cc.(*webClient)
if ok {
ccc.action(pushConnsAction{g, c})
2020-05-30 00:23:54 +02:00
}
}
}
2020-05-27 11:44:49 +02:00
func (c *webClient) isRequested(label string) bool {
return c.requested[label] != 0
2020-05-09 19:39:34 +02:00
}
2020-09-13 11:04:16 +02:00
func addDownConnTracks(c *webClient, remote conn.Up, tracks []conn.UpTrack) (*rtpDownConnection, error) {
requested := false
for _, t := range tracks {
2020-06-08 22:14:28 +02:00
if c.isRequested(t.Label()) {
requested = true
break
}
}
if !requested {
delDownConn(c, remote.Id())
return nil, nil
}
2020-06-08 22:14:28 +02:00
down, err := addDownConn(c, remote.Id(), remote)
if err != nil {
return nil, err
}
for _, t := range tracks {
2020-06-08 22:14:28 +02:00
if !c.isRequested(t.Label()) {
continue
}
_, err = addDownTrack(c, down, t, remote)
if err != nil {
delDownConn(c, down.id)
return nil, err
}
}
2020-05-31 23:35:56 +02:00
go rtcpDownSender(down)
return down, nil
}
func (c *webClient) PushConn(g *group.Group, id string, up conn.Up, tracks []conn.UpTrack, label string) error {
err := c.action(pushConnAction{g, id, up, tracks})
2020-05-28 02:35:09 +02:00
if err != nil {
return err
}
2020-09-13 11:04:16 +02:00
if up != nil && label != "" {
err := c.action(addLabelAction{up.Id(), up.Label()})
2020-05-28 02:35:09 +02:00
if err != nil {
return err
}
}
2020-05-28 02:35:09 +02:00
return nil
}
func readMessage(conn *websocket.Conn, m *clientMessage) error {
err := conn.SetReadDeadline(time.Now().Add(15 * time.Second))
2020-06-08 22:14:28 +02:00
if err != nil {
return err
2020-06-08 22:14:28 +02:00
}
defer conn.SetReadDeadline(time.Time{})
return conn.ReadJSON(&m)
}
2020-12-05 01:18:27 +01:00
func StartClient(conn *websocket.Conn) (err error) {
var m clientMessage
2020-12-05 01:18:27 +01:00
err = readMessage(conn, &m)
2020-06-08 22:14:28 +02:00
if err != nil {
conn.Close()
2020-12-05 01:18:27 +01:00
return
2020-06-08 22:14:28 +02:00
}
if m.Type != "handshake" {
conn.WriteMessage(websocket.CloseMessage,
websocket.FormatCloseMessage(
websocket.CloseProtocolError,
"you must handshake first",
),
)
2020-06-08 22:14:28 +02:00
conn.Close()
2020-12-05 01:18:27 +01:00
err = group.ProtocolError("client didn't handshake")
return
2020-06-08 22:14:28 +02:00
}
c := &webClient{
2020-11-29 14:26:42 +01:00
id: m.Id,
2020-06-08 22:14:28 +02:00
actionCh: make(chan interface{}, 10),
done: make(chan struct{}),
}
defer close(c.done)
c.writeCh = make(chan interface{}, 25)
c.writerDone = make(chan struct{})
go clientWriter(conn, c.writeCh, c.writerDone)
2020-06-08 22:14:28 +02:00
defer func() {
2020-12-05 01:18:27 +01:00
m, e := errorToWSCloseMessage(c.id, err)
if isWSNormalError(err) {
err = nil
} else if _, ok := err.(group.KickError); ok {
err = nil
}
if m != nil {
c.write(*m)
2020-06-08 22:14:28 +02:00
}
c.close(e)
2020-06-08 22:14:28 +02:00
}()
return clientLoop(c, conn)
}
type pushConnAction struct {
2020-12-19 17:38:47 +01:00
group *group.Group
id string
2020-09-13 11:04:16 +02:00
conn conn.Up
tracks []conn.UpTrack
}
type addLabelAction struct {
id string
label string
}
type pushConnsAction struct {
2020-12-19 17:38:47 +01:00
group *group.Group
client group.Client
}
type connectionFailedAction struct {
id string
}
type permissionsChangedAction struct{}
type kickAction struct {
2020-11-30 16:26:11 +01:00
id string
username string
message string
}
2020-09-13 11:04:16 +02:00
func clientLoop(c *webClient, ws *websocket.Conn) error {
2020-04-24 19:38:21 +02:00
read := make(chan interface{}, 1)
2020-09-13 11:04:16 +02:00
go clientReader(ws, read, c.done)
2020-04-24 19:38:21 +02:00
defer leaveGroup(c)
2020-04-24 19:38:21 +02:00
readTime := time.Now()
ticker := time.NewTicker(10 * time.Second)
2020-04-24 19:38:21 +02:00
defer ticker.Stop()
for {
select {
case m, ok := <-read:
if !ok {
return errors.New("reader died")
}
switch m := m.(type) {
case clientMessage:
readTime = time.Now()
2020-04-24 19:38:21 +02:00
err := handleClientMessage(c, m)
if err != nil {
return err
}
case error:
return m
}
case a := <-c.actionCh:
switch a := a.(type) {
case pushConnAction:
g := c.group
if g == nil || a.group != g {
return nil
}
if a.conn == nil {
found := delDownConn(c, a.id)
if found {
c.write(clientMessage{
Type: "close",
Id: a.id,
})
} else {
log.Printf("Deleting unknown " +
"down connection")
}
continue
}
down, err := addDownConnTracks(
c, a.conn, a.tracks,
)
if err != nil {
return err
2020-04-24 19:38:21 +02:00
}
if down != nil {
err = negotiate(c, down, false, false)
2020-04-24 19:38:21 +02:00
if err != nil {
2020-05-24 13:36:42 +02:00
delDownConn(c, down.id)
c.error(group.UserError(
"Negotiation failed",
))
2020-05-24 13:36:42 +02:00
continue
2020-04-24 19:38:21 +02:00
}
}
case addLabelAction:
2020-12-02 19:33:53 +01:00
label := a.label
2020-04-24 19:38:21 +02:00
c.write(clientMessage{
Type: "label",
Id: a.id,
2020-12-02 19:33:53 +01:00
Value: &label,
2020-04-24 19:38:21 +02:00
})
case pushConnsAction:
g := c.group
if g == nil || a.group != g {
return nil
}
2020-04-24 19:38:21 +02:00
for _, u := range c.up {
if !u.complete() {
continue
}
tracks := u.getTracks()
2020-09-13 11:04:16 +02:00
ts := make([]conn.UpTrack, len(tracks))
2020-06-08 22:14:28 +02:00
for i, t := range tracks {
ts[i] = t
}
go func(u *rtpUpConnection, ts []conn.UpTrack) {
err := a.client.PushConn(
g, u.id, u, ts, u.label,
2020-12-04 12:15:37 +01:00
)
if err != nil {
log.Printf(
"PushConn: %v",
err,
)
}
}(u, ts)
2020-04-24 19:38:21 +02:00
}
case connectionFailedAction:
if down := getDownConn(c, a.id); down != nil {
err := negotiate(c, down, true, true)
2020-08-13 13:48:17 +02:00
if err != nil {
return err
}
tracks := make(
2020-09-13 11:04:16 +02:00
[]conn.UpTrack, len(down.tracks),
)
for i, t := range down.tracks {
tracks[i] = t.remote
}
go c.PushConn(
c.group,
down.remote.Id(), down.remote,
tracks, down.remote.Label(),
)
} else if up := getUpConn(c, a.id); up != nil {
c.write(clientMessage{
Type: "renegotiate",
Id: a.id,
})
} else {
log.Printf("Attempting to renegotiate " +
"unknown connection")
}
case permissionsChangedAction:
group := c.Group()
if group == nil {
return errors.New("Permissions changed in no group")
}
perms := c.permissions
2020-04-25 17:36:35 +02:00
c.write(clientMessage{
Type: "joined",
Kind: "change",
Group: group.Name(),
Permissions: &perms,
2020-04-25 17:36:35 +02:00
})
if !c.permissions.Present {
up := getUpConns(c)
for _, u := range up {
found := delUpConn(c, u.id)
2020-05-24 13:36:42 +02:00
if found {
failUpConnection(
c, u.id,
2020-05-24 13:36:42 +02:00
"permission denied",
)
}
}
}
2020-04-25 17:36:35 +02:00
case kickAction:
2020-11-30 16:26:11 +01:00
return group.KickError{
a.id, a.username, a.message,
}
2020-04-24 19:38:21 +02:00
default:
log.Printf("unexpected action %T", a)
return errors.New("unexpected action")
}
case <-ticker.C:
if time.Since(readTime) > 90*time.Second {
return errors.New("client is dead")
}
if time.Since(readTime) > 60*time.Second {
err := c.write(clientMessage{
Type: "ping",
})
if err != nil {
return err
}
}
2020-04-24 19:38:21 +02:00
}
}
}
func failUpConnection(c *webClient, id string, message string) error {
2020-05-24 13:36:42 +02:00
if id != "" {
err := c.write(clientMessage{
Type: "abort",
Id: id,
})
if err != nil {
return err
}
}
if message != "" {
err := c.error(group.UserError(message))
2020-05-24 13:36:42 +02:00
if err != nil {
return err
}
}
return nil
}
func leaveGroup(c *webClient) {
if c.group == nil {
return
}
c.setRequested(map[string]uint32{})
if c.up != nil {
for id := range c.up {
delUpConn(c, id)
}
}
group.DelClient(c)
c.permissions = group.ClientPermissions{}
c.group = nil
}
func failDownConnection(c *webClient, id string, message string) error {
if id != "" {
err := c.write(clientMessage{
Type: "close",
Id: id,
})
if err != nil {
return err
}
}
if message != "" {
err := c.error(group.UserError(message))
if err != nil {
return err
}
}
return nil
}
func setPermissions(g *group.Group, id string, perm string) error {
client := g.GetClient(id)
2020-05-28 02:35:09 +02:00
if client == nil {
return group.UserError("no such user")
2020-05-28 02:35:09 +02:00
}
c, ok := client.(*webClient)
if !ok {
return group.UserError("this is not a real user")
2020-05-28 02:35:09 +02:00
}
switch perm {
case "op":
c.permissions.Op = true
if g.AllowRecording() {
2020-05-30 00:23:54 +02:00
c.permissions.Record = true
}
2020-05-28 02:35:09 +02:00
case "unop":
c.permissions.Op = false
2020-05-30 00:23:54 +02:00
c.permissions.Record = false
2020-05-28 02:35:09 +02:00
case "present":
c.permissions.Present = true
case "unpresent":
c.permissions.Present = false
default:
return group.UserError("unknown permission")
2020-05-28 02:35:09 +02:00
}
return c.action(permissionsChangedAction{})
}
2020-11-30 16:26:11 +01:00
func (c *webClient) Kick(id, user, message string) error {
return c.action(kickAction{id, user, message})
}
2020-11-30 16:26:11 +01:00
func kickClient(g *group.Group, id, user, dest string, message string) error {
client := g.GetClient(dest)
2020-05-28 02:35:09 +02:00
if client == nil {
return group.UserError("no such user")
2020-05-28 02:35:09 +02:00
}
c, ok := client.(group.Kickable)
2020-05-28 02:35:09 +02:00
if !ok {
return group.UserError("this client is not kickable")
2020-05-28 02:35:09 +02:00
}
2020-11-30 16:26:11 +01:00
return c.Kick(id, user, message)
2020-05-28 02:35:09 +02:00
}
2020-05-27 11:44:49 +02:00
func handleClientMessage(c *webClient, m clientMessage) error {
2020-04-24 19:38:21 +02:00
switch m.Type {
case "join":
if m.Kind == "leave" {
if c.group == nil || c.group.Name() != m.Group {
return group.ProtocolError("you are not joined")
}
leaveGroup(c)
perms := c.permissions
return c.write(clientMessage{
Type: "joined",
Kind: "leave",
Group: m.Group,
Permissions: &perms,
})
}
if m.Kind != "join" {
return group.ProtocolError("unknown kind")
}
if c.group != nil {
return group.ProtocolError("cannot join multiple groups")
}
c.username = m.Username
c.password = m.Password
g, err := group.AddClient(m.Group, c)
2020-05-09 19:39:34 +02:00
if err != nil {
var s string
if os.IsNotExist(err) {
s = "group does not exist"
} else if err == group.ErrNotAuthorised {
s = "not authorised"
time.Sleep(200 * time.Millisecond)
} else if e, ok := err.(group.UserError); ok {
s = string(e)
} else {
s = "internal server error"
log.Printf("Join group: %v", err)
}
return c.write(clientMessage{
Type: "joined",
Kind: "fail",
Group: m.Group,
Permissions: &group.ClientPermissions{},
Value: &s,
})
2020-05-09 19:39:34 +02:00
}
if redirect := g.Redirect(); redirect != "" {
// We normally redirect at the HTTP level, but the group
// description could have been edited in the meantime.
return c.write(clientMessage{
Type: "joined",
Kind: "redirect",
Group: m.Group,
Permissions: &group.ClientPermissions{},
Value: &redirect,
})
}
c.group = g
perms := c.permissions
err = c.write(clientMessage{
Type: "joined",
Kind: "join",
Group: m.Group,
Permissions: &perms,
})
if err != nil {
return err
}
h := c.group.GetChatHistory()
for _, m := range h {
message := m.Value
err := c.write(clientMessage{
Type: "chat",
Id: m.Id,
Username: m.User,
Time: m.Time,
Value: &message,
Kind: m.Kind,
})
if err != nil {
return err
}
}
case "request":
return c.setRequested(m.Request)
2020-04-24 19:38:21 +02:00
case "offer":
2020-04-25 02:25:51 +02:00
if !c.permissions.Present {
c.write(clientMessage{
Type: "abort",
Id: m.Id,
})
return c.error(group.UserError("not authorised"))
2020-04-25 02:25:51 +02:00
}
2020-04-24 19:38:21 +02:00
if m.Offer == nil {
return group.ProtocolError("null offer")
2020-04-24 19:38:21 +02:00
}
2020-08-12 13:56:35 +02:00
err := gotOffer(
c, m.Id, *m.Offer, m.Kind == "renegotiate", m.Labels,
)
2020-04-24 19:38:21 +02:00
if err != nil {
2020-05-24 13:36:42 +02:00
log.Printf("gotOffer: %v", err)
return failUpConnection(c, m.Id, "negotiation failed")
2020-04-24 19:38:21 +02:00
}
case "answer":
if m.Answer == nil {
return group.ProtocolError("null answer")
2020-04-24 19:38:21 +02:00
}
err := gotAnswer(c, m.Id, *m.Answer)
2020-04-24 19:38:21 +02:00
if err != nil {
log.Printf("gotAnswer: %v", err)
message := ""
if err != ErrUnknownId {
message = "negotiation failed"
}
return failDownConnection(c, m.Id, message)
2020-04-24 19:38:21 +02:00
}
case "renegotiate":
down := getDownConn(c, m.Id)
if down != nil {
err := negotiate(c, down, true, true)
if err != nil {
return failDownConnection(c, m.Id,
"renegotiation failed")
}
} else {
log.Printf("Trying to renegotiate unknown connection")
}
2020-04-24 19:38:21 +02:00
case "close":
found := delUpConn(c, m.Id)
if !found {
log.Printf("Deleting unknown up connection %v", m.Id)
}
case "abort":
found := delDownConn(c, m.Id)
if !found {
log.Printf("Attempted to abort unknown connection")
}
c.write(clientMessage{
Type: "close",
Id: m.Id,
});
2020-04-24 19:38:21 +02:00
case "ice":
if m.Candidate == nil {
return group.ProtocolError("null candidate")
2020-04-24 19:38:21 +02:00
}
err := gotICE(c, m.Candidate, m.Id)
if err != nil {
log.Printf("ICE: %v", err)
}
case "chat", "usermessage":
2020-11-30 15:22:00 +01:00
if m.Id != c.id {
return group.ProtocolError("wrong sender id")
2020-11-30 15:22:00 +01:00
}
if m.Username != "" && m.Username != c.username {
return group.ProtocolError("wrong sender username")
}
g := c.group
if g == nil {
return c.error(group.UserError("join a group first"))
2020-11-30 15:22:00 +01:00
}
2020-10-08 14:38:33 +02:00
tm := group.ToJSTime(time.Now())
2020-12-02 19:33:53 +01:00
if m.Type == "chat" {
2020-12-02 19:33:53 +01:00
if m.Value == nil {
return group.ProtocolError("missing value")
}
if m.Dest == "" {
g.AddToChatHistory(
2020-12-02 19:33:53 +01:00
m.Id, m.Username, tm, m.Kind, *m.Value,
)
}
2020-10-01 16:52:01 +02:00
}
2020-09-30 00:33:23 +02:00
mm := clientMessage{
2020-12-19 17:38:47 +01:00
Type: m.Type,
Id: m.Id,
Dest: m.Dest,
Username: m.Username,
Privileged: c.permissions.Op,
Time: tm,
Kind: m.Kind,
Value: m.Value,
2020-09-30 00:33:23 +02:00
}
2020-10-01 16:52:01 +02:00
if m.Dest == "" {
clients := g.GetClients(nil)
2020-10-01 16:52:01 +02:00
for _, cc := range clients {
ccc, ok := cc.(*webClient)
if ok {
ccc.write(mm)
}
}
} else {
cc := g.GetClient(m.Dest)
2020-10-01 16:52:01 +02:00
if cc == nil {
return c.error(group.UserError("user unknown"))
}
ccc, ok := cc.(*webClient)
if !ok {
return c.error(group.UserError("this user doesn't chat"))
2020-05-28 02:35:09 +02:00
}
2020-10-01 16:52:01 +02:00
ccc.write(mm)
2020-04-24 19:38:21 +02:00
}
case "groupaction":
2020-11-30 15:39:44 +01:00
if m.Id != c.id {
return group.ProtocolError("wrong sender id")
2020-11-30 15:39:44 +01:00
}
if m.Username != "" && m.Username != c.username {
return group.ProtocolError("wrong sender username")
}
g := c.group
if g == nil {
return c.error(group.UserError("join a group first"))
2020-11-30 15:39:44 +01:00
}
switch m.Kind {
case "clearchat":
g.ClearChatHistory()
m := clientMessage{Type: "clearchat"}
clients := g.GetClients(nil)
for _, cc := range clients {
cc, ok := cc.(*webClient)
if ok {
cc.write(m)
}
2020-05-28 02:35:09 +02:00
}
case "lock", "unlock":
if !c.permissions.Op {
return c.error(group.UserError("not authorised"))
2020-05-30 00:23:54 +02:00
}
2020-12-02 19:33:53 +01:00
message := ""
if m.Value != nil {
message = *m.Value
}
g.SetLocked(m.Kind == "lock", message)
case "record":
if !c.permissions.Record {
return c.error(group.UserError("not authorised"))
}
for _, cc := range g.GetClients(c) {
2020-10-04 19:01:06 +02:00
_, ok := cc.(*diskwriter.Client)
if ok {
return c.error(group.UserError("already recording"))
}
}
disk := diskwriter.New(g)
_, err := group.AddClient(g.Name(), disk)
if err != nil {
2020-05-30 00:23:54 +02:00
disk.Close()
return c.error(err)
2020-05-30 00:23:54 +02:00
}
go pushConns(disk, c.group)
case "unrecord":
if !c.permissions.Record {
return c.error(group.UserError("not authorised"))
}
for _, cc := range g.GetClients(c) {
2020-10-04 19:01:06 +02:00
disk, ok := cc.(*diskwriter.Client)
if ok {
disk.Close()
group.DelClient(disk)
}
}
2020-12-02 19:47:32 +01:00
case "subgroups":
if !c.permissions.Op {
return c.error(group.UserError("not authorised"))
}
s := ""
for _, sg := range group.GetSubGroups(g.Name()) {
plural := ""
if sg.Clients > 1 {
plural = "s"
}
s = s + fmt.Sprintf("%v (%v client%v)",
sg.Name, sg.Clients, plural)
}
c.write(clientMessage{
Type: "chat",
Dest: c.id,
2020-12-02 19:47:32 +01:00
Username: "Server",
Time: group.ToJSTime(time.Now()),
Value: &s,
2020-12-02 19:47:32 +01:00
})
default:
return group.ProtocolError("unknown group action")
2020-05-30 00:23:54 +02:00
}
case "useraction":
2020-11-30 15:39:44 +01:00
if m.Id != c.id {
return group.ProtocolError("wrong sender id")
2020-11-30 15:39:44 +01:00
}
if m.Username != "" && m.Username != c.username {
return group.ProtocolError("wrong sender username")
}
g := c.group
if g == nil {
return c.error(group.UserError("join a group first"))
2020-11-30 15:39:44 +01:00
}
switch m.Kind {
case "op", "unop", "present", "unpresent":
if !c.permissions.Op {
return c.error(group.UserError("not authorised"))
}
err := setPermissions(g, m.Dest, m.Kind)
if err != nil {
return c.error(err)
}
case "kick":
if !c.permissions.Op {
return c.error(group.UserError("not authorised"))
}
2020-12-02 19:33:53 +01:00
message := ""
if m.Value != nil {
message = *m.Value
}
err := kickClient(g, m.Id, m.Username, m.Dest, message)
if err != nil {
return c.error(err)
}
default:
return group.ProtocolError("unknown user action")
2020-04-25 17:36:35 +02:00
}
case "pong":
// nothing
case "ping":
return c.write(clientMessage{
Type: "pong",
})
2020-04-24 19:38:21 +02:00
default:
log.Printf("unexpected message: %v", m.Type)
return group.ProtocolError("unexpected message")
2020-04-24 19:38:21 +02:00
}
return nil
}
func clientReader(conn *websocket.Conn, read chan<- interface{}, done <-chan struct{}) {
defer close(read)
for {
var m clientMessage
err := conn.ReadJSON(&m)
if err != nil {
select {
case read <- err:
return
case <-done:
return
}
}
select {
case read <- m:
case <-done:
return
}
}
}
func clientWriter(conn *websocket.Conn, ch <-chan interface{}, done chan<- struct{}) {
defer func() {
close(done)
conn.Close()
}()
for {
m, ok := <-ch
if !ok {
break
}
err := conn.SetWriteDeadline(
time.Now().Add(2 * time.Second))
if err != nil {
return
}
switch m := m.(type) {
case clientMessage:
err := conn.WriteJSON(m)
if err != nil {
return
}
case closeMessage:
if m.data != nil {
conn.WriteMessage(
websocket.CloseMessage,
m.data,
)
2020-04-24 19:38:21 +02:00
}
return
2020-04-24 19:38:21 +02:00
default:
log.Printf("clientWiter: unexpected message %T", m)
return
}
}
}
2020-05-28 02:35:09 +02:00
func (c *webClient) Warn(oponly bool, message string) error {
if oponly && !c.permissions.Op {
return nil
}
return c.write(clientMessage{
Type: "usermessage",
Kind: "warning",
Dest: c.id,
Privileged: true,
Value: &message,
})
}
var ErrClientDead = errors.New("client is dead")
2020-05-30 12:38:13 +02:00
2020-05-28 02:35:09 +02:00
func (c *webClient) action(m interface{}) error {
select {
case c.actionCh <- m:
return nil
case <-c.done:
2020-05-30 12:38:13 +02:00
return ErrClientDead
2020-05-28 02:35:09 +02:00
}
}
func (c *webClient) write(m clientMessage) error {
select {
case c.writeCh <- m:
return nil
case <-c.writerDone:
return ErrClientDead
2020-05-28 02:35:09 +02:00
}
}
func (c *webClient) close(data []byte) error {
select {
case c.writeCh <- closeMessage{data}:
return nil
case <-c.writerDone:
return ErrClientDead
}
}
2020-11-30 16:26:11 +01:00
func errorMessage(id string, err error) *clientMessage {
2020-05-28 02:35:09 +02:00
switch e := err.(type) {
case group.UserError:
2020-12-02 19:33:53 +01:00
message := e.Error()
2020-11-30 16:26:11 +01:00
return &clientMessage{
2020-12-19 17:38:47 +01:00
Type: "usermessage",
Kind: "error",
Dest: id,
Privileged: true,
Value: &message,
2020-11-30 16:26:11 +01:00
}
case group.KickError:
message := e.Message
if message == "" {
message = "you have been kicked out"
}
return &clientMessage{
2020-12-19 17:38:47 +01:00
Type: "usermessage",
Kind: "error",
Id: e.Id,
Username: e.Username,
Dest: id,
Privileged: true,
Value: &message,
2020-11-30 16:26:11 +01:00
}
2020-05-28 02:35:09 +02:00
default:
2020-11-30 16:26:11 +01:00
return nil
}
}
func (c *webClient) error(err error) error {
m := errorMessage(c.id, err)
if m == nil {
2020-05-28 02:35:09 +02:00
return err
}
2020-11-30 16:26:11 +01:00
return c.write(*m)
2020-05-28 02:35:09 +02:00
}