mirror of
https://github.com/jech/galene.git
synced 2024-11-22 16:45:58 +01:00
29e006037c
We now support or reasonable values for proxyURL, such as "http:" or "/galene".
2160 lines
45 KiB
Go
2160 lines
45 KiB
Go
package rtpconn
|
|
|
|
import (
|
|
crand "crypto/rand"
|
|
"encoding/base64"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"log"
|
|
"os"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/gorilla/websocket"
|
|
"github.com/pion/webrtc/v3"
|
|
|
|
"github.com/jech/galene/conn"
|
|
"github.com/jech/galene/diskwriter"
|
|
"github.com/jech/galene/estimator"
|
|
"github.com/jech/galene/group"
|
|
"github.com/jech/galene/ice"
|
|
"github.com/jech/galene/token"
|
|
"github.com/jech/galene/unbounded"
|
|
)
|
|
|
|
func errorToWSCloseMessage(id string, err error) (*clientMessage, []byte) {
|
|
var code int
|
|
var m *clientMessage
|
|
var text string
|
|
switch e := err.(type) {
|
|
case *websocket.CloseError:
|
|
code = websocket.CloseNormalClosure
|
|
case group.ProtocolError:
|
|
code = websocket.CloseProtocolError
|
|
m = &clientMessage{
|
|
Type: "usermessage",
|
|
Kind: "error",
|
|
Dest: id,
|
|
Privileged: true,
|
|
Value: e.Error(),
|
|
}
|
|
text = e.Error()
|
|
case group.UserError, group.KickError:
|
|
code = websocket.CloseNormalClosure
|
|
m = errorMessage(id, err)
|
|
text = e.Error()
|
|
default:
|
|
code = websocket.CloseInternalServerErr
|
|
}
|
|
return m, websocket.FormatCloseMessage(code, text)
|
|
}
|
|
|
|
func isWSNormalError(err error) bool {
|
|
return websocket.IsCloseError(err,
|
|
websocket.CloseNormalClosure,
|
|
websocket.CloseGoingAway)
|
|
}
|
|
|
|
type webClient struct {
|
|
group *group.Group
|
|
id string
|
|
username string
|
|
permissions []string
|
|
data map[string]interface{}
|
|
requested map[string][]string
|
|
done chan struct{}
|
|
writeCh chan interface{}
|
|
writerDone chan struct{}
|
|
actions *unbounded.Channel[any]
|
|
|
|
mu sync.Mutex
|
|
down map[string]*rtpDownConnection
|
|
up map[string]*rtpUpConnection
|
|
}
|
|
|
|
func (c *webClient) Group() *group.Group {
|
|
return c.group
|
|
}
|
|
|
|
func (c *webClient) Id() string {
|
|
return c.id
|
|
}
|
|
|
|
func (c *webClient) Username() string {
|
|
return c.username
|
|
}
|
|
|
|
func (c *webClient) SetUsername(username string) {
|
|
c.username = username
|
|
}
|
|
|
|
func (c *webClient) Permissions() []string {
|
|
return c.permissions
|
|
}
|
|
|
|
func (c *webClient) Data() map[string]interface{} {
|
|
return c.data
|
|
}
|
|
|
|
func (c *webClient) SetPermissions(perms []string) {
|
|
c.permissions = perms
|
|
}
|
|
|
|
func (c *webClient) PushClient(group, kind, id string, username string, perms []string, data map[string]interface{}) error {
|
|
c.action(pushClientAction{
|
|
group, kind, id, username, perms, data,
|
|
})
|
|
return nil
|
|
}
|
|
|
|
type clientMessage struct {
|
|
Type string `json:"type"`
|
|
Version []string `json:"version,omitempty"`
|
|
Kind string `json:"kind,omitempty"`
|
|
Error string `json:"error,omitempty"`
|
|
Id string `json:"id,omitempty"`
|
|
Replace string `json:"replace,omitempty"`
|
|
Source string `json:"source,omitempty"`
|
|
Dest string `json:"dest,omitempty"`
|
|
Username *string `json:"username,omitempty"`
|
|
Password string `json:"password,omitempty"`
|
|
Token string `json:"token,omitempty"`
|
|
Privileged bool `json:"privileged,omitempty"`
|
|
Permissions []string `json:"permissions,omitempty"`
|
|
Status *group.Status `json:"status,omitempty"`
|
|
Data map[string]interface{} `json:"data,omitempty"`
|
|
Group string `json:"group,omitempty"`
|
|
Value interface{} `json:"value,omitempty"`
|
|
NoEcho bool `json:"noecho,omitempty"`
|
|
Time string `json:"time,omitempty"`
|
|
SDP string `json:"sdp,omitempty"`
|
|
Candidate *webrtc.ICECandidateInit `json:"candidate,omitempty"`
|
|
Label string `json:"label,omitempty"`
|
|
Request interface{} `json:"request,omitempty"`
|
|
RTCConfiguration *webrtc.Configuration `json:"rtcConfiguration,omitempty"`
|
|
}
|
|
|
|
type closeMessage struct {
|
|
data []byte
|
|
}
|
|
|
|
func getUpConn(c *webClient, id string) *rtpUpConnection {
|
|
c.mu.Lock()
|
|
defer c.mu.Unlock()
|
|
|
|
if c.up == nil {
|
|
return nil
|
|
}
|
|
return c.up[id]
|
|
}
|
|
|
|
func getUpConns(c *webClient) []*rtpUpConnection {
|
|
c.mu.Lock()
|
|
defer c.mu.Unlock()
|
|
up := make([]*rtpUpConnection, 0, len(c.up))
|
|
for _, u := range c.up {
|
|
up = append(up, u)
|
|
}
|
|
return up
|
|
}
|
|
|
|
func addUpConn(c *webClient, id, label string, offer string) (*rtpUpConnection, bool, error) {
|
|
c.mu.Lock()
|
|
defer c.mu.Unlock()
|
|
|
|
if c.up == nil {
|
|
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, label, offer)
|
|
if err != nil {
|
|
return nil, false, err
|
|
}
|
|
|
|
c.up[id] = conn
|
|
|
|
conn.pc.OnICECandidate(func(candidate *webrtc.ICECandidate) {
|
|
sendICE(c, id, candidate)
|
|
})
|
|
|
|
conn.pc.OnICEConnectionStateChange(func(state webrtc.ICEConnectionState) {
|
|
if state == webrtc.ICEConnectionStateFailed {
|
|
c.action(connectionFailedAction{id: id})
|
|
}
|
|
})
|
|
|
|
return conn, true, nil
|
|
}
|
|
|
|
var ErrUserMismatch = errors.New("user id mismatch")
|
|
|
|
// delUpConn deletes an up connection. If push is closed, the close is
|
|
// pushed to all corresponding down connections.
|
|
func delUpConn(c *webClient, id string, userId string, push bool) error {
|
|
c.mu.Lock()
|
|
if c.up == nil {
|
|
c.mu.Unlock()
|
|
return os.ErrNotExist
|
|
}
|
|
conn := c.up[id]
|
|
if conn == nil {
|
|
c.mu.Unlock()
|
|
return os.ErrNotExist
|
|
}
|
|
if userId != "" {
|
|
id, _ := conn.User()
|
|
if id != userId {
|
|
c.mu.Unlock()
|
|
return ErrUserMismatch
|
|
}
|
|
}
|
|
|
|
replace := conn.getReplace(false)
|
|
|
|
delete(c.up, id)
|
|
g := c.group
|
|
c.mu.Unlock()
|
|
|
|
conn.mu.Lock()
|
|
conn.closed = true
|
|
conn.mu.Unlock()
|
|
|
|
conn.pc.Close()
|
|
|
|
if push && g != nil {
|
|
for _, c := range g.GetClients(c) {
|
|
err := c.PushConn(g, id, nil, nil, replace)
|
|
if err != nil {
|
|
log.Printf("PushConn: %v", err)
|
|
}
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func getDownConn(c *webClient, id string) *rtpDownConnection {
|
|
if c.down == nil {
|
|
return nil
|
|
}
|
|
|
|
c.mu.Lock()
|
|
defer c.mu.Unlock()
|
|
conn := c.down[id]
|
|
if conn == nil {
|
|
return nil
|
|
}
|
|
return conn
|
|
}
|
|
|
|
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
|
|
}
|
|
|
|
func addDownConn(c *webClient, remote conn.Up) (*rtpDownConnection, bool, error) {
|
|
id := remote.Id()
|
|
|
|
c.mu.Lock()
|
|
defer c.mu.Unlock()
|
|
|
|
if c.up != nil && c.up[id] != nil {
|
|
return nil, false, errors.New("adding duplicate connection")
|
|
}
|
|
|
|
if c.down == nil {
|
|
c.down = make(map[string]*rtpDownConnection)
|
|
}
|
|
|
|
if down := c.down[id]; down != nil {
|
|
return down, false, nil
|
|
}
|
|
|
|
down, err := newDownConn(c, id, remote)
|
|
if err != nil {
|
|
return nil, false, err
|
|
}
|
|
|
|
down.pc.OnICECandidate(func(candidate *webrtc.ICECandidate) {
|
|
sendICE(c, down.id, candidate)
|
|
})
|
|
|
|
down.pc.OnICEConnectionStateChange(func(state webrtc.ICEConnectionState) {
|
|
if state == webrtc.ICEConnectionStateFailed {
|
|
c.action(connectionFailedAction{id: down.id})
|
|
}
|
|
})
|
|
|
|
err = remote.AddLocal(down)
|
|
if err != nil {
|
|
down.pc.Close()
|
|
return nil, false, err
|
|
}
|
|
|
|
c.down[down.id] = down
|
|
|
|
go rtcpDownSender(down)
|
|
|
|
return down, true, nil
|
|
}
|
|
|
|
func delDownConn(c *webClient, id string) error {
|
|
conn := delDownConnHelper(c, id)
|
|
if conn != nil {
|
|
conn.pc.Close()
|
|
return nil
|
|
}
|
|
return os.ErrNotExist
|
|
}
|
|
|
|
func delDownConnHelper(c *webClient, id string) *rtpDownConnection {
|
|
c.mu.Lock()
|
|
defer c.mu.Unlock()
|
|
|
|
if c.down == nil {
|
|
return nil
|
|
}
|
|
conn := c.down[id]
|
|
if conn == nil {
|
|
return nil
|
|
}
|
|
|
|
conn.remote.DelLocal(conn)
|
|
for _, track := range conn.tracks {
|
|
// we only insert the track after we get an answer, so
|
|
// ignore errors here.
|
|
track.remote.DelLocal(track)
|
|
}
|
|
delete(c.down, id)
|
|
return conn
|
|
}
|
|
|
|
var errUnexpectedTrackType = errors.New("unexpected track type, this shouldn't happen")
|
|
|
|
func addDownTrackUnlocked(conn *rtpDownConnection, remoteTrack *rtpUpTrack) error {
|
|
for _, t := range conn.tracks {
|
|
tt, ok := t.remote.(*rtpUpTrack)
|
|
if !ok {
|
|
return errUnexpectedTrackType
|
|
}
|
|
if tt == remoteTrack {
|
|
return os.ErrExist
|
|
}
|
|
}
|
|
|
|
id := remoteTrack.track.ID()
|
|
if id == "" {
|
|
log.Println("Got track with empty id")
|
|
id = remoteTrack.track.RID()
|
|
}
|
|
if id == "" {
|
|
id = remoteTrack.track.Kind().String()
|
|
}
|
|
msid := remoteTrack.track.StreamID()
|
|
if msid == "" || msid == "-" {
|
|
log.Println("Got track with empty msid")
|
|
msid = remoteTrack.conn.Label()
|
|
}
|
|
if msid == "" {
|
|
msid = "dummy"
|
|
}
|
|
|
|
local, err := webrtc.NewTrackLocalStaticRTP(
|
|
remoteTrack.Codec(), id, msid,
|
|
)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
transceiver, err := conn.pc.AddTransceiverFromTrack(local,
|
|
webrtc.RTPTransceiverInit{
|
|
Direction: webrtc.RTPTransceiverDirectionSendonly,
|
|
},
|
|
)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
codec := local.Codec()
|
|
ptype, err := group.CodecPayloadType(local.Codec())
|
|
if err != nil {
|
|
log.Printf("Couldn't determine ptype for codec %v: %v",
|
|
codec.MimeType, err)
|
|
} else {
|
|
err := transceiver.SetCodecPreferences(
|
|
[]webrtc.RTPCodecParameters{
|
|
{
|
|
RTPCodecCapability: codec,
|
|
PayloadType: ptype,
|
|
},
|
|
},
|
|
)
|
|
if err != nil {
|
|
log.Printf("Couldn't set ptype for codec %v: %v",
|
|
codec.MimeType, err)
|
|
}
|
|
}
|
|
|
|
parms := transceiver.Sender().GetParameters()
|
|
if len(parms.Encodings) != 1 {
|
|
return errors.New("got multiple encodings")
|
|
}
|
|
|
|
track := &rtpDownTrack{
|
|
track: local,
|
|
sender: transceiver.Sender(),
|
|
ssrc: parms.Encodings[0].SSRC,
|
|
conn: conn,
|
|
remote: remoteTrack,
|
|
maxBitrate: new(bitrate),
|
|
maxREMBBitrate: new(bitrate),
|
|
stats: new(receiverStats),
|
|
rate: estimator.New(time.Second),
|
|
atomics: &downTrackAtomics{},
|
|
}
|
|
|
|
conn.tracks = append(conn.tracks, track)
|
|
|
|
go rtcpDownListener(track)
|
|
|
|
return nil
|
|
}
|
|
|
|
func delDownTrackUnlocked(conn *rtpDownConnection, track *rtpDownTrack) error {
|
|
for i := range conn.tracks {
|
|
if conn.tracks[i] == track {
|
|
track.remote.DelLocal(track)
|
|
conn.tracks =
|
|
append(conn.tracks[:i], conn.tracks[i+1:]...)
|
|
return conn.pc.RemoveTrack(track.sender)
|
|
}
|
|
}
|
|
return os.ErrNotExist
|
|
}
|
|
|
|
func replaceTracks(conn *rtpDownConnection, remote []conn.UpTrack, limitSid bool) (bool, error) {
|
|
conn.mu.Lock()
|
|
defer conn.mu.Unlock()
|
|
|
|
var add []*rtpUpTrack
|
|
var del []*rtpDownTrack
|
|
|
|
outer:
|
|
for _, rtrack := range remote {
|
|
rt, ok := rtrack.(*rtpUpTrack)
|
|
if !ok {
|
|
return false, errUnexpectedTrackType
|
|
}
|
|
for _, track := range conn.tracks {
|
|
rt2, ok := track.remote.(*rtpUpTrack)
|
|
if !ok {
|
|
return false, errUnexpectedTrackType
|
|
}
|
|
if rt == rt2 {
|
|
continue outer
|
|
}
|
|
}
|
|
add = append(add, rt)
|
|
}
|
|
|
|
outer2:
|
|
for _, track := range conn.tracks {
|
|
rt, ok := track.remote.(*rtpUpTrack)
|
|
if !ok {
|
|
return false, errUnexpectedTrackType
|
|
}
|
|
for _, rtrack := range remote {
|
|
rt2, ok := rtrack.(*rtpUpTrack)
|
|
if !ok {
|
|
return false, errUnexpectedTrackType
|
|
}
|
|
if rt == rt2 {
|
|
continue outer2
|
|
}
|
|
}
|
|
del = append(del, track)
|
|
}
|
|
|
|
defer func() {
|
|
for _, t := range conn.tracks {
|
|
layer := t.getLayerInfo()
|
|
layer.limitSid = limitSid
|
|
if limitSid {
|
|
layer.wantedSid = 0
|
|
}
|
|
t.setLayerInfo(layer)
|
|
}
|
|
}()
|
|
|
|
if len(del) == 0 && len(add) == 0 {
|
|
return false, nil
|
|
}
|
|
|
|
for _, t := range del {
|
|
err := delDownTrackUnlocked(conn, t)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
}
|
|
|
|
for _, rt := range add {
|
|
err := addDownTrackUnlocked(conn, rt)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
}
|
|
|
|
return true, nil
|
|
}
|
|
|
|
func negotiate(c *webClient, down *rtpDownConnection, restartIce bool, replace string) error {
|
|
if down.pc.SignalingState() == webrtc.SignalingStateHaveLocalOffer {
|
|
// avoid sending multiple offers back-to-back
|
|
if restartIce {
|
|
down.negotiationNeeded = negotiationRestartIce
|
|
} else if down.negotiationNeeded == negotiationUnneeded {
|
|
down.negotiationNeeded = negotiationNeeded
|
|
}
|
|
return nil
|
|
}
|
|
|
|
down.negotiationNeeded = negotiationUnneeded
|
|
|
|
options := webrtc.OfferOptions{ICERestart: restartIce}
|
|
offer, err := down.pc.CreateOffer(&options)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
err = down.pc.SetLocalDescription(offer)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
source, username := down.remote.User()
|
|
|
|
return c.write(clientMessage{
|
|
Type: "offer",
|
|
Id: down.id,
|
|
Label: down.remote.Label(),
|
|
Replace: replace,
|
|
Source: source,
|
|
Username: &username,
|
|
SDP: down.pc.LocalDescription().SDP,
|
|
})
|
|
}
|
|
|
|
func sendICE(c *webClient, id string, candidate *webrtc.ICECandidate) error {
|
|
if candidate == nil {
|
|
return nil
|
|
}
|
|
cand := candidate.ToJSON()
|
|
return c.write(clientMessage{
|
|
Type: "ice",
|
|
Id: id,
|
|
Candidate: &cand,
|
|
})
|
|
}
|
|
|
|
func gotOffer(c *webClient, id, label string, sdp string, replace string) error {
|
|
up, _, err := addUpConn(c, id, label, sdp)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
if replace != "" {
|
|
up.replace = replace
|
|
delUpConn(c, replace, c.Id(), false)
|
|
}
|
|
|
|
err = up.pc.SetRemoteDescription(webrtc.SessionDescription{
|
|
Type: webrtc.SDPTypeOffer,
|
|
SDP: sdp,
|
|
})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
answer, err := up.pc.CreateAnswer(nil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
err = up.pc.SetLocalDescription(answer)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
err = up.flushICECandidates()
|
|
if err != nil {
|
|
log.Printf("ICE: %v", err)
|
|
}
|
|
|
|
return c.write(clientMessage{
|
|
Type: "answer",
|
|
Id: id,
|
|
SDP: up.pc.LocalDescription().SDP,
|
|
})
|
|
}
|
|
|
|
var ErrUnknownId = errors.New("unknown id")
|
|
|
|
func gotAnswer(c *webClient, id string, sdp string) error {
|
|
down := getDownConn(c, id)
|
|
if down == nil {
|
|
return ErrUnknownId
|
|
}
|
|
|
|
err := down.pc.SetRemoteDescription(webrtc.SessionDescription{
|
|
Type: webrtc.SDPTypeAnswer,
|
|
SDP: sdp,
|
|
})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
err = down.flushICECandidates()
|
|
if err != nil {
|
|
log.Printf("ICE: %v", err)
|
|
}
|
|
|
|
add := func() {
|
|
down.pc.OnConnectionStateChange(nil)
|
|
for _, t := range down.tracks {
|
|
err := t.remote.AddLocal(t)
|
|
if err != nil && err != os.ErrClosed {
|
|
log.Printf("Add track: %v", err)
|
|
}
|
|
}
|
|
}
|
|
down.pc.OnConnectionStateChange(func(state webrtc.PeerConnectionState) {
|
|
if state == webrtc.PeerConnectionStateConnected {
|
|
add()
|
|
}
|
|
})
|
|
if down.pc.ConnectionState() == webrtc.PeerConnectionStateConnected {
|
|
add()
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func gotICE(c *webClient, candidate *webrtc.ICECandidateInit, id string) error {
|
|
conn := getConn(c, id)
|
|
if conn == nil {
|
|
return errors.New("unknown id in ICE")
|
|
}
|
|
return conn.addICECandidate(candidate)
|
|
}
|
|
|
|
var errBadType = errors.New("bad type")
|
|
|
|
func toStringArray(r interface{}) ([]string, error) {
|
|
if r == nil {
|
|
return nil, nil
|
|
}
|
|
rr, ok := r.([]interface{})
|
|
if !ok {
|
|
return nil, errBadType
|
|
}
|
|
if rr == nil {
|
|
return nil, nil
|
|
}
|
|
|
|
rrr := make([]string, len(rr))
|
|
for i, s := range rr {
|
|
rrr[i], ok = s.(string)
|
|
if !ok {
|
|
return nil, errBadType
|
|
}
|
|
}
|
|
return rrr, nil
|
|
}
|
|
|
|
func parseRequested(r interface{}) (map[string][]string, error) {
|
|
if r == nil {
|
|
return nil, nil
|
|
}
|
|
rr, ok := r.(map[string]interface{})
|
|
if !ok {
|
|
return nil, errBadType
|
|
}
|
|
if rr == nil {
|
|
return nil, nil
|
|
}
|
|
rrr := make(map[string][]string)
|
|
for k, v := range rr {
|
|
vv, err := toStringArray(v)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
rrr[k] = vv
|
|
}
|
|
return rrr, nil
|
|
}
|
|
|
|
func (c *webClient) setRequested(requested map[string][]string) error {
|
|
if c.group == nil {
|
|
return errors.New("attempted to request with no group joined")
|
|
}
|
|
c.requested = requested
|
|
|
|
requestConns(c, c.group, "")
|
|
return nil
|
|
}
|
|
|
|
func (c *webClient) setRequestedStream(down *rtpDownConnection, requested []string) error {
|
|
var remoteClient group.Client
|
|
remote, ok := down.remote.(*rtpUpConnection)
|
|
if ok {
|
|
remoteClient = remote.client
|
|
}
|
|
down.requested = requested
|
|
return remoteClient.RequestConns(c, c.group, remote.id)
|
|
}
|
|
|
|
func (c *webClient) RequestConns(target group.Client, g *group.Group, id string) error {
|
|
c.action(requestConnsAction{g, target, id})
|
|
return nil
|
|
}
|
|
|
|
func requestConns(target group.Client, g *group.Group, id string) {
|
|
clients := g.GetClients(target)
|
|
for _, c := range clients {
|
|
c.RequestConns(target, g, id)
|
|
}
|
|
}
|
|
|
|
func requestedTracks(c *webClient, requested []string, tracks []conn.UpTrack) ([]conn.UpTrack, bool) {
|
|
if len(requested) == 0 {
|
|
return nil, false
|
|
}
|
|
var audio, video, videoLow bool
|
|
for _, s := range requested {
|
|
switch s {
|
|
case "audio":
|
|
audio = true
|
|
case "video":
|
|
video = true
|
|
case "video-low":
|
|
videoLow = true
|
|
default:
|
|
log.Printf("client requested unknown value %v", s)
|
|
}
|
|
}
|
|
|
|
find := func(kind webrtc.RTPCodecType, last bool) (conn.UpTrack, int) {
|
|
var track conn.UpTrack
|
|
count := 0
|
|
for _, t := range tracks {
|
|
if t.Kind() != kind {
|
|
continue
|
|
}
|
|
track = t
|
|
count++
|
|
if !last {
|
|
break
|
|
}
|
|
}
|
|
return track, count
|
|
}
|
|
|
|
var ts []conn.UpTrack
|
|
limitSid := false
|
|
if audio {
|
|
t, _ := find(webrtc.RTPCodecTypeAudio, false)
|
|
if t != nil {
|
|
ts = append(ts, t)
|
|
}
|
|
}
|
|
if video {
|
|
t, _ := find(webrtc.RTPCodecTypeVideo, false)
|
|
if t != nil {
|
|
ts = append(ts, t)
|
|
}
|
|
} else if videoLow {
|
|
t, count := find(webrtc.RTPCodecTypeVideo, true)
|
|
if t != nil {
|
|
ts = append(ts, t)
|
|
}
|
|
if count < 2 {
|
|
limitSid = true
|
|
}
|
|
}
|
|
|
|
return ts, limitSid
|
|
}
|
|
|
|
func (c *webClient) PushConn(g *group.Group, id string, up conn.Up, tracks []conn.UpTrack, replace string) error {
|
|
c.action(pushConnAction{g, id, up, tracks, replace})
|
|
return nil
|
|
}
|
|
|
|
func readMessage(conn *websocket.Conn, m *clientMessage) error {
|
|
err := conn.SetReadDeadline(time.Now().Add(15 * time.Second))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer conn.SetReadDeadline(time.Time{})
|
|
|
|
return conn.ReadJSON(&m)
|
|
}
|
|
|
|
const protocolVersion = "2"
|
|
|
|
func StartClient(conn *websocket.Conn) (err error) {
|
|
var m clientMessage
|
|
|
|
err = readMessage(conn, &m)
|
|
if err != nil {
|
|
conn.Close()
|
|
return
|
|
}
|
|
|
|
if m.Type != "handshake" {
|
|
conn.WriteMessage(websocket.CloseMessage,
|
|
websocket.FormatCloseMessage(
|
|
websocket.CloseProtocolError,
|
|
"you must handshake first",
|
|
),
|
|
)
|
|
conn.Close()
|
|
err = group.ProtocolError("client didn't handshake")
|
|
return
|
|
}
|
|
|
|
versionError := true
|
|
if m.Version != nil {
|
|
for _, v := range m.Version {
|
|
if v == protocolVersion {
|
|
versionError = false
|
|
}
|
|
}
|
|
}
|
|
|
|
c := &webClient{
|
|
id: m.Id,
|
|
actions: unbounded.New[any](),
|
|
done: make(chan struct{}),
|
|
}
|
|
|
|
defer close(c.done)
|
|
|
|
c.writeCh = make(chan interface{}, 100)
|
|
c.writerDone = make(chan struct{})
|
|
go clientWriter(conn, c.writeCh, c.writerDone)
|
|
defer func() {
|
|
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)
|
|
}
|
|
c.close(e)
|
|
}()
|
|
|
|
return clientLoop(c, conn, versionError)
|
|
}
|
|
|
|
type pushConnAction struct {
|
|
group *group.Group
|
|
id string
|
|
conn conn.Up
|
|
tracks []conn.UpTrack
|
|
replace string
|
|
}
|
|
|
|
type requestConnsAction struct {
|
|
group *group.Group
|
|
target group.Client
|
|
id string
|
|
}
|
|
|
|
type connectionFailedAction struct {
|
|
id string
|
|
}
|
|
|
|
type pushClientAction struct {
|
|
group string
|
|
kind string
|
|
id string
|
|
username string
|
|
permissions []string
|
|
data map[string]interface{}
|
|
}
|
|
|
|
type permissionsChangedAction struct{}
|
|
|
|
type joinedAction struct {
|
|
group string
|
|
kind string
|
|
}
|
|
|
|
type kickAction struct {
|
|
id string
|
|
username *string
|
|
message string
|
|
}
|
|
|
|
var errEmptyId = group.ProtocolError("empty id")
|
|
|
|
func member(v string, l []string) bool {
|
|
for _, w := range l {
|
|
if v == w {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
func remove(v string, l []string) []string {
|
|
for i, w := range l {
|
|
if v == w {
|
|
l = append(l[:i], l[i+1:]...)
|
|
return l
|
|
}
|
|
}
|
|
return l
|
|
}
|
|
|
|
func addnew(v string, l []string) []string {
|
|
if member(v, l) {
|
|
return l
|
|
}
|
|
l = append(l, v)
|
|
return l
|
|
}
|
|
|
|
func clientLoop(c *webClient, ws *websocket.Conn, versionError bool) error {
|
|
read := make(chan interface{}, 1)
|
|
go clientReader(ws, read, c.done)
|
|
|
|
defer leaveGroup(c)
|
|
|
|
readTime := time.Now()
|
|
|
|
ticker := time.NewTicker(10 * time.Second)
|
|
defer ticker.Stop()
|
|
|
|
err := c.write(clientMessage{
|
|
Type: "handshake",
|
|
Version: []string{protocolVersion},
|
|
})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
if versionError {
|
|
c.write(clientMessage{
|
|
Type: "usermessage",
|
|
Kind: "warning",
|
|
Dest: c.id,
|
|
Privileged: true,
|
|
Value: "This client is using an unknown protocol version.\n" +
|
|
"Perhaps it needs upgrading?\n" +
|
|
"Trying to continue, things may break.",
|
|
})
|
|
}
|
|
|
|
for {
|
|
select {
|
|
case m, ok := <-read:
|
|
if !ok {
|
|
return errors.New("reader died")
|
|
}
|
|
switch m := m.(type) {
|
|
case clientMessage:
|
|
readTime = time.Now()
|
|
err := handleClientMessage(c, m)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
case error:
|
|
return m
|
|
}
|
|
case <-c.actions.Ch:
|
|
actions := c.actions.Get()
|
|
for _, a := range actions {
|
|
err := handleAction(c, a)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
case <-ticker.C:
|
|
if time.Since(readTime) > 75*time.Second {
|
|
return errors.New("client is dead")
|
|
}
|
|
// Some reverse proxies timeout connexions at 60
|
|
// seconds, make sure we generate some activity
|
|
// after 55s at most.
|
|
if time.Since(readTime) > 45*time.Second {
|
|
err := c.write(clientMessage{
|
|
Type: "ping",
|
|
})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func pushDownConn(c *webClient, id string, up conn.Up, tracks []conn.UpTrack, replace string) error {
|
|
var requested []conn.UpTrack
|
|
limitSid := false
|
|
if up != nil {
|
|
var old *rtpDownConnection
|
|
if replace != "" {
|
|
old = getDownConn(c, replace)
|
|
} else {
|
|
old = getDownConn(c, up.Id())
|
|
}
|
|
var req []string
|
|
if old != nil {
|
|
req = old.requested
|
|
}
|
|
if req == nil {
|
|
var ok bool
|
|
req, ok = c.requested[up.Label()]
|
|
if !ok {
|
|
req = c.requested[""]
|
|
}
|
|
}
|
|
requested, limitSid = requestedTracks(c, req, tracks)
|
|
}
|
|
|
|
if replace != "" {
|
|
err := delDownConn(c, replace)
|
|
if err != nil {
|
|
log.Printf("Replace: %v", err)
|
|
}
|
|
}
|
|
|
|
// closes over replace, which will be modified below
|
|
defer func() {
|
|
if replace != "" {
|
|
closeDownConn(c, replace, "")
|
|
}
|
|
}()
|
|
|
|
if len(requested) == 0 {
|
|
closeDownConn(c, id, "")
|
|
return nil
|
|
}
|
|
|
|
down, _, err := addDownConn(c, up)
|
|
if err != nil {
|
|
if err == os.ErrClosed {
|
|
return nil
|
|
}
|
|
return err
|
|
}
|
|
done, err := replaceTracks(down, requested, limitSid)
|
|
if err != nil || !done {
|
|
return err
|
|
}
|
|
err = negotiate(c, down, false, replace)
|
|
if err != nil {
|
|
log.Printf("Negotiation failed: %v", err)
|
|
closeDownConn(c, down.id, err.Error())
|
|
return err
|
|
}
|
|
replace = ""
|
|
return nil
|
|
}
|
|
|
|
func handleAction(c *webClient, a any) error {
|
|
switch a := a.(type) {
|
|
case pushConnAction:
|
|
if c.group == nil || c.group != a.group {
|
|
log.Printf("Got connectsions for wrong group")
|
|
return nil
|
|
}
|
|
return pushDownConn(c, a.id, a.conn, a.tracks, a.replace)
|
|
case requestConnsAction:
|
|
g := c.group
|
|
if g == nil || a.group != g {
|
|
log.Printf("Misdirected pushConns")
|
|
return nil
|
|
}
|
|
for _, u := range c.up {
|
|
if a.id != "" && a.id != u.id {
|
|
continue
|
|
}
|
|
tracks := u.getTracks()
|
|
replace := u.getReplace(false)
|
|
|
|
ts := make([]conn.UpTrack, len(tracks))
|
|
for i, t := range tracks {
|
|
ts[i] = t
|
|
}
|
|
err := a.target.PushConn(g, u.id, u, ts, replace)
|
|
if err != nil {
|
|
log.Printf("PushConn: %v", err)
|
|
}
|
|
}
|
|
case connectionFailedAction:
|
|
if down := getDownConn(c, a.id); down != nil {
|
|
err := negotiate(c, down, true, "")
|
|
if err != nil {
|
|
return err
|
|
}
|
|
tracks := make(
|
|
[]conn.UpTrack, len(down.tracks),
|
|
)
|
|
for i, t := range down.tracks {
|
|
tracks[i] = t.remote
|
|
}
|
|
c.PushConn(
|
|
c.group,
|
|
down.remote.Id(), down.remote,
|
|
tracks, "",
|
|
)
|
|
} 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 pushClientAction:
|
|
if a.group != c.group.Name() {
|
|
log.Printf("got client for wrong group")
|
|
return nil
|
|
}
|
|
perms := append([]string(nil), a.permissions...)
|
|
username := a.username
|
|
return c.write(clientMessage{
|
|
Type: "user",
|
|
Kind: a.kind,
|
|
Id: a.id,
|
|
Username: &username,
|
|
Permissions: perms,
|
|
Data: a.data,
|
|
})
|
|
case joinedAction:
|
|
var status *group.Status
|
|
var data map[string]interface{}
|
|
var g *group.Group
|
|
if a.group != "" {
|
|
g = group.Get(a.group)
|
|
if g != nil {
|
|
s := g.Status(true, nil)
|
|
status = &s
|
|
data = g.Data()
|
|
}
|
|
}
|
|
perms := append([]string(nil), c.permissions...)
|
|
username := c.username
|
|
err := c.write(clientMessage{
|
|
Type: "joined",
|
|
Kind: a.kind,
|
|
Group: a.group,
|
|
Username: &username,
|
|
Permissions: perms,
|
|
Status: status,
|
|
Data: data,
|
|
RTCConfiguration: ice.ICEConfiguration(),
|
|
})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if a.kind == "join" {
|
|
if g == nil {
|
|
log.Println("g is null when joining" +
|
|
"this shouldn't happen")
|
|
return nil
|
|
}
|
|
h := g.GetChatHistory()
|
|
for _, m := range h {
|
|
err := c.write(clientMessage{
|
|
Type: "chathistory",
|
|
Source: m.Id,
|
|
Username: m.User,
|
|
Time: m.Time.Format(time.RFC3339),
|
|
Value: m.Value,
|
|
Kind: m.Kind,
|
|
})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
}
|
|
case permissionsChangedAction:
|
|
g := c.Group()
|
|
if g == nil {
|
|
return errors.New("Permissions changed in no group")
|
|
}
|
|
perms := append([]string(nil), c.permissions...)
|
|
status := g.Status(true, nil)
|
|
username := c.username
|
|
c.write(clientMessage{
|
|
Type: "joined",
|
|
Kind: "change",
|
|
Group: g.Name(),
|
|
Username: &username,
|
|
Permissions: perms,
|
|
Status: &status,
|
|
RTCConfiguration: ice.ICEConfiguration(),
|
|
})
|
|
if !member("present", c.permissions) {
|
|
up := getUpConns(c)
|
|
for _, u := range up {
|
|
err := delUpConn(
|
|
c, u.id, c.id, true,
|
|
)
|
|
if err == nil {
|
|
failUpConnection(
|
|
c, u.id,
|
|
"permission denied",
|
|
)
|
|
}
|
|
}
|
|
}
|
|
id := c.Id()
|
|
user := c.Username()
|
|
d := c.Data()
|
|
clients := g.GetClients(nil)
|
|
go func(clients []group.Client) {
|
|
for _, cc := range clients {
|
|
cc.PushClient(
|
|
g.Name(), "change", id, user, perms, d,
|
|
)
|
|
}
|
|
}(clients)
|
|
case kickAction:
|
|
return group.KickError{
|
|
a.id, a.username, a.message,
|
|
}
|
|
default:
|
|
log.Printf("unexpected action %T", a)
|
|
return errors.New("unexpected action")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func failUpConnection(c *webClient, id string, message string) error {
|
|
if id != "" {
|
|
err := c.write(clientMessage{
|
|
Type: "abort",
|
|
Id: id,
|
|
})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
if message != "" {
|
|
err := c.error(group.UserError(message))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func leaveGroup(c *webClient) {
|
|
if c.group == nil {
|
|
return
|
|
}
|
|
|
|
if c.up != nil {
|
|
for id := range c.up {
|
|
delUpConn(c, id, c.id, true)
|
|
}
|
|
}
|
|
if c.down != nil {
|
|
for id := range c.down {
|
|
delDownConn(c, id)
|
|
}
|
|
}
|
|
|
|
group.DelClient(c)
|
|
c.permissions = nil
|
|
c.data = nil
|
|
c.requested = make(map[string][]string)
|
|
c.group = nil
|
|
}
|
|
|
|
func closeDownConn(c *webClient, id string, message string) error {
|
|
err := delDownConn(c, id)
|
|
if err != nil && !os.IsNotExist(err) {
|
|
log.Printf("Close down connection: %v", err)
|
|
}
|
|
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)
|
|
if client == nil {
|
|
return group.UserError("no such user")
|
|
}
|
|
|
|
c, ok := client.(*webClient)
|
|
if !ok {
|
|
return group.UserError("this is not a real user")
|
|
}
|
|
|
|
switch perm {
|
|
case "op":
|
|
c.permissions = addnew("op", c.permissions)
|
|
if g.Description().AllowRecording {
|
|
c.permissions = addnew("record", c.permissions)
|
|
}
|
|
case "unop":
|
|
c.permissions = remove("op", c.permissions)
|
|
c.permissions = remove("record", c.permissions)
|
|
case "present":
|
|
c.permissions = addnew("present", c.permissions)
|
|
case "unpresent":
|
|
c.permissions = remove("present", c.permissions)
|
|
default:
|
|
return group.UserError("unknown permission")
|
|
}
|
|
c.action(permissionsChangedAction{})
|
|
return nil
|
|
}
|
|
|
|
func (c *webClient) Kick(id string, user *string, message string) error {
|
|
c.action(kickAction{id, user, message})
|
|
return nil
|
|
}
|
|
|
|
func (c *webClient) Joined(group, kind string) error {
|
|
c.action(joinedAction{group, kind})
|
|
return nil
|
|
}
|
|
|
|
func kickClient(g *group.Group, id string, user *string, dest string, message string) error {
|
|
client := g.GetClient(dest)
|
|
if client == nil {
|
|
return group.UserError("no such user")
|
|
}
|
|
|
|
return client.Kick(id, user, message)
|
|
}
|
|
|
|
func handleClientMessage(c *webClient, m clientMessage) error {
|
|
if m.Source != "" {
|
|
if m.Source != c.Id() {
|
|
return group.ProtocolError("spoofed client id")
|
|
}
|
|
}
|
|
|
|
if m.Type != "join" {
|
|
if m.Username != nil {
|
|
if *m.Username != c.Username() {
|
|
return group.ProtocolError("spoofed username")
|
|
}
|
|
}
|
|
}
|
|
|
|
switch m.Type {
|
|
case "join":
|
|
if m.Kind == "leave" {
|
|
if c.group == nil || c.group.Name() != m.Group {
|
|
return group.UserError("you are not joined")
|
|
}
|
|
leaveGroup(c)
|
|
return nil
|
|
}
|
|
|
|
if m.Kind != "join" {
|
|
return group.ProtocolError("unknown kind")
|
|
}
|
|
|
|
if c.group != nil {
|
|
return group.ProtocolError(
|
|
"cannot join multiple groups",
|
|
)
|
|
}
|
|
c.data = m.Data
|
|
g, err := group.AddClient(m.Group, c,
|
|
group.ClientCredentials{
|
|
Username: m.Username,
|
|
Password: m.Password,
|
|
Token: m.Token,
|
|
},
|
|
)
|
|
if err != nil {
|
|
var e, 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 err == group.ErrAnonymousNotAuthorised {
|
|
s = "please choose a username"
|
|
} else if _, ok := err.(group.UserError); ok {
|
|
s = err.Error()
|
|
} else if err == token.ErrUsernameRequired {
|
|
s = err.Error()
|
|
e = "need-username"
|
|
} else if err == group.ErrDuplicateUsername {
|
|
s = err.Error()
|
|
e = "duplicate-username"
|
|
} else {
|
|
s = "internal server error"
|
|
log.Printf("Join group: %v", err)
|
|
}
|
|
username := c.username
|
|
return c.write(clientMessage{
|
|
Type: "joined",
|
|
Kind: "fail",
|
|
Error: e,
|
|
Group: m.Group,
|
|
Username: &username,
|
|
Value: s,
|
|
})
|
|
}
|
|
if redirect := g.Description().Redirect; redirect != "" {
|
|
// We normally redirect at the HTTP level, but the group
|
|
// description could have been edited in the meantime.
|
|
username := c.username
|
|
return c.write(clientMessage{
|
|
Type: "joined",
|
|
Kind: "redirect",
|
|
Group: m.Group,
|
|
Username: &username,
|
|
Value: redirect,
|
|
})
|
|
}
|
|
c.group = g
|
|
case "request":
|
|
requested, err := parseRequested(m.Request)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return c.setRequested(requested)
|
|
case "requestStream":
|
|
down := getDownConn(c, m.Id)
|
|
if down == nil {
|
|
return ErrUnknownId
|
|
}
|
|
requested, err := toStringArray(m.Request)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
c.setRequestedStream(down, requested)
|
|
case "offer":
|
|
if m.Id == "" {
|
|
return errEmptyId
|
|
}
|
|
if !member("present", c.permissions) {
|
|
if m.Replace != "" {
|
|
delUpConn(c, m.Replace, c.id, true)
|
|
}
|
|
c.write(clientMessage{
|
|
Type: "abort",
|
|
Id: m.Id,
|
|
})
|
|
return c.error(group.UserError("not authorised"))
|
|
}
|
|
err := gotOffer(c, m.Id, m.Label, m.SDP, m.Replace)
|
|
if err != nil {
|
|
log.Printf("gotOffer: %v", err)
|
|
return failUpConnection(c, m.Id, err.Error())
|
|
}
|
|
case "answer":
|
|
if m.Id == "" {
|
|
return errEmptyId
|
|
}
|
|
err := gotAnswer(c, m.Id, m.SDP)
|
|
if err != nil {
|
|
log.Printf("gotAnswer: %v", err)
|
|
message := ""
|
|
if err != ErrUnknownId {
|
|
message = err.Error()
|
|
}
|
|
return closeDownConn(c, m.Id, message)
|
|
}
|
|
down := getDownConn(c, m.Id)
|
|
if down == nil {
|
|
return ErrUnknownId
|
|
}
|
|
if down.negotiationNeeded > negotiationUnneeded {
|
|
err := negotiate(
|
|
c, down,
|
|
down.negotiationNeeded == negotiationRestartIce,
|
|
"",
|
|
)
|
|
if err != nil {
|
|
return closeDownConn(c, m.Id, err.Error())
|
|
}
|
|
}
|
|
case "renegotiate":
|
|
if m.Id == "" {
|
|
return errEmptyId
|
|
}
|
|
down := getDownConn(c, m.Id)
|
|
if down != nil {
|
|
err := negotiate(c, down, true, "")
|
|
if err != nil {
|
|
return closeDownConn(c, m.Id, err.Error())
|
|
}
|
|
} else {
|
|
log.Printf("Trying to renegotiate unknown connection")
|
|
}
|
|
case "close":
|
|
if m.Id == "" {
|
|
return errEmptyId
|
|
}
|
|
err := delUpConn(c, m.Id, c.id, true)
|
|
if err != nil {
|
|
log.Printf("Deleting up connection %v: %v",
|
|
m.Id, err)
|
|
return nil
|
|
}
|
|
case "abort":
|
|
if m.Id == "" {
|
|
return errEmptyId
|
|
}
|
|
return closeDownConn(c, m.Id, "")
|
|
case "ice":
|
|
if m.Id == "" {
|
|
return errEmptyId
|
|
}
|
|
if m.Candidate == nil {
|
|
return group.ProtocolError("null candidate")
|
|
}
|
|
err := gotICE(c, m.Candidate, m.Id)
|
|
if err != nil {
|
|
log.Printf("ICE: %v", err)
|
|
}
|
|
case "chat", "usermessage":
|
|
g := c.group
|
|
if g == nil {
|
|
return c.error(group.UserError("join a group first"))
|
|
}
|
|
|
|
now := time.Now()
|
|
|
|
if m.Type == "chat" {
|
|
if m.Dest == "" {
|
|
g.AddToChatHistory(
|
|
m.Source, m.Username, now, m.Kind, m.Value,
|
|
)
|
|
}
|
|
}
|
|
mm := clientMessage{
|
|
Type: m.Type,
|
|
Source: m.Source,
|
|
Dest: m.Dest,
|
|
Username: m.Username,
|
|
Privileged: member("op", c.permissions),
|
|
Time: now.Format(time.RFC3339),
|
|
Kind: m.Kind,
|
|
NoEcho: m.NoEcho,
|
|
Value: m.Value,
|
|
}
|
|
if m.Dest == "" {
|
|
var except group.Client
|
|
if m.NoEcho {
|
|
except = c
|
|
}
|
|
err := broadcast(g.GetClients(except), mm)
|
|
if err != nil {
|
|
log.Printf("broadcast(chat): %v", err)
|
|
}
|
|
} else {
|
|
cc := g.GetClient(m.Dest)
|
|
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",
|
|
))
|
|
}
|
|
ccc.write(mm)
|
|
}
|
|
case "groupaction":
|
|
g := c.group
|
|
if g == nil {
|
|
return c.error(group.UserError("join a group first"))
|
|
}
|
|
switch m.Kind {
|
|
case "clearchat":
|
|
g.ClearChatHistory()
|
|
m := clientMessage{
|
|
Type: "usermessage",
|
|
Kind: "clearchat",
|
|
Privileged: true,
|
|
}
|
|
err := broadcast(g.GetClients(nil), m)
|
|
if err != nil {
|
|
log.Printf("broadcast(clearchat): %v", err)
|
|
}
|
|
case "lock", "unlock":
|
|
if !member("op", c.permissions) {
|
|
return c.error(group.UserError("not authorised"))
|
|
}
|
|
message := ""
|
|
v, ok := m.Value.(string)
|
|
if ok {
|
|
message = v
|
|
}
|
|
g.SetLocked(m.Kind == "lock", message)
|
|
case "record":
|
|
if !member("record", c.permissions) {
|
|
return c.error(group.UserError("not authorised"))
|
|
}
|
|
for _, cc := range g.GetClients(c) {
|
|
_, ok := cc.(*diskwriter.Client)
|
|
if ok {
|
|
return c.error(group.UserError("already recording"))
|
|
}
|
|
}
|
|
disk := diskwriter.New(g)
|
|
_, err := group.AddClient(g.Name(), disk,
|
|
group.ClientCredentials{
|
|
System: true,
|
|
},
|
|
)
|
|
if err != nil {
|
|
disk.Close()
|
|
return c.error(err)
|
|
}
|
|
requestConns(disk, c.group, "")
|
|
case "unrecord":
|
|
if !member("record", c.permissions) {
|
|
return c.error(group.UserError("not authorised"))
|
|
}
|
|
for _, cc := range g.GetClients(c) {
|
|
disk, ok := cc.(*diskwriter.Client)
|
|
if ok {
|
|
disk.Close()
|
|
group.DelClient(disk)
|
|
}
|
|
}
|
|
case "subgroups":
|
|
if !member("op", c.permissions) {
|
|
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)\n",
|
|
sg.Name, sg.Clients, plural)
|
|
}
|
|
username := "Server"
|
|
c.write(clientMessage{
|
|
Type: "chat",
|
|
Dest: c.id,
|
|
Username: &username,
|
|
Time: time.Now().Format(time.RFC3339),
|
|
Value: s,
|
|
})
|
|
case "setdata":
|
|
if !member("op", c.permissions) {
|
|
return c.error(group.UserError("not authorised"))
|
|
}
|
|
data, ok := m.Value.(map[string]interface{})
|
|
if !ok {
|
|
return c.error(group.UserError(
|
|
"Bad value in setdata",
|
|
))
|
|
}
|
|
g.UpdateData(data)
|
|
case "maketoken":
|
|
terror := func(e, m string) error {
|
|
return c.write(clientMessage{
|
|
Type: "usermessage",
|
|
Kind: "token",
|
|
Privileged: true,
|
|
Error: e,
|
|
Value: m,
|
|
})
|
|
}
|
|
if !member("token", c.permissions) {
|
|
return terror("not-authorised", "not authorised")
|
|
}
|
|
tok, err := parseStatefulToken(m.Value)
|
|
if err != nil {
|
|
return terror("error", err.Error())
|
|
}
|
|
|
|
if tok.Token == "" {
|
|
buf := make([]byte, 8)
|
|
crand.Read(buf)
|
|
tok.Token =
|
|
base64.RawURLEncoding.EncodeToString(buf)
|
|
} else {
|
|
return terror("error", "client specified token")
|
|
}
|
|
|
|
if tok.Group != c.group.Name() {
|
|
return terror("error", "wrong group in token")
|
|
}
|
|
|
|
if tok.Expires == nil {
|
|
return terror("error", "token doesn't expire")
|
|
}
|
|
|
|
if tok.Username != nil &&
|
|
c.group.UserExists(*tok.Username) {
|
|
return terror("error", "that username is taken")
|
|
}
|
|
|
|
for _, p := range tok.Permissions {
|
|
if !member(p, c.permissions) {
|
|
return terror(
|
|
"not-authorised",
|
|
"not authorised",
|
|
)
|
|
}
|
|
}
|
|
|
|
user := c.username
|
|
if user != "" {
|
|
tok.IssuedBy = &user
|
|
}
|
|
|
|
now := time.Now()
|
|
tok.IssuedAt = &now
|
|
|
|
new, err := token.Add(tok)
|
|
if err != nil {
|
|
return terror("error", err.Error())
|
|
}
|
|
c.write(clientMessage{
|
|
Type: "usermessage",
|
|
Kind: "token",
|
|
Privileged: true,
|
|
Value: new,
|
|
})
|
|
case "edittoken":
|
|
terror := func(e, m string) error {
|
|
return c.write(clientMessage{
|
|
Type: "usermessage",
|
|
Kind: "token",
|
|
Privileged: true,
|
|
Error: e,
|
|
Value: m,
|
|
})
|
|
}
|
|
if !member("op", c.permissions) ||
|
|
!member("token", c.permissions) {
|
|
return terror("not-authorised", "not authorised")
|
|
}
|
|
tok, err := parseStatefulToken(m.Value)
|
|
if err != nil {
|
|
return terror("error", err.Error())
|
|
}
|
|
if tok.Group != "" || tok.Username != nil ||
|
|
tok.Permissions != nil ||
|
|
tok.NotBefore != nil ||
|
|
tok.IssuedBy != nil ||
|
|
tok.IssuedAt != nil {
|
|
return terror(
|
|
"error", "this field cannot be edited",
|
|
)
|
|
}
|
|
if tok.Expires == nil {
|
|
return terror("error", "trying to edit nothing")
|
|
}
|
|
new, err := token.Edit(
|
|
c.group.Name(), tok.Token, *tok.Expires,
|
|
)
|
|
if err != nil {
|
|
return terror("error", err.Error())
|
|
}
|
|
c.write(clientMessage{
|
|
Type: "usermessage",
|
|
Kind: "token",
|
|
Privileged: true,
|
|
Value: new,
|
|
})
|
|
case "listtokens":
|
|
terror := func(e, m string) error {
|
|
return c.write(clientMessage{
|
|
Type: "usermessage",
|
|
Kind: "tokenlist",
|
|
Privileged: true,
|
|
Error: e,
|
|
Value: m,
|
|
})
|
|
}
|
|
if !member("op", c.permissions) ||
|
|
!member("token", c.permissions) {
|
|
return terror("not-authorised", "not authorised")
|
|
}
|
|
tokens, err := token.List(c.group.Name())
|
|
if err != nil {
|
|
return terror("error", err.Error())
|
|
}
|
|
c.write(clientMessage{
|
|
Type: "usermessage",
|
|
Kind: "tokenlist",
|
|
Privileged: true,
|
|
Value: tokens,
|
|
})
|
|
default:
|
|
return group.UserError("unknown group action")
|
|
}
|
|
case "useraction":
|
|
g := c.group
|
|
if g == nil {
|
|
return c.error(group.UserError("join a group first"))
|
|
}
|
|
switch m.Kind {
|
|
case "op", "unop", "present", "unpresent":
|
|
if !member("op", c.permissions) {
|
|
return c.error(group.UserError("not authorised"))
|
|
}
|
|
err := setPermissions(g, m.Dest, m.Kind)
|
|
if err != nil {
|
|
return c.error(err)
|
|
}
|
|
case "kick":
|
|
if !member("op", c.permissions) {
|
|
return c.error(group.UserError("not authorised"))
|
|
}
|
|
message := ""
|
|
v, ok := m.Value.(string)
|
|
if ok {
|
|
message = v
|
|
}
|
|
err := kickClient(g, m.Source, m.Username, m.Dest, message)
|
|
if err != nil {
|
|
return c.error(err)
|
|
}
|
|
case "setdata":
|
|
if m.Dest != c.Id() {
|
|
return c.error(group.UserError("not authorised"))
|
|
}
|
|
data, ok := m.Value.(map[string]interface{})
|
|
if !ok {
|
|
return c.error(group.UserError(
|
|
"Bad value in setdata",
|
|
))
|
|
}
|
|
if c.data == nil {
|
|
c.data = make(map[string]interface{})
|
|
}
|
|
for k, v := range data {
|
|
if v == nil {
|
|
delete(c.data, k)
|
|
} else {
|
|
c.data[k] = v
|
|
}
|
|
}
|
|
id := c.Id()
|
|
user := c.Username()
|
|
perms := c.Permissions()
|
|
data = c.Data()
|
|
go func(clients []group.Client) {
|
|
for _, cc := range clients {
|
|
cc.PushClient(
|
|
g.Name(), "change",
|
|
id, user, perms, data,
|
|
)
|
|
}
|
|
}(g.GetClients(nil))
|
|
default:
|
|
return group.UserError("unknown user action")
|
|
}
|
|
case "pong":
|
|
// nothing
|
|
case "ping":
|
|
return c.write(clientMessage{
|
|
Type: "pong",
|
|
})
|
|
default:
|
|
log.Printf("unexpected message: %v", m.Type)
|
|
return group.ProtocolError("unexpected message")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func parseStatefulToken(value interface{}) (*token.Stateful, error) {
|
|
data, ok := value.(map[string]interface{})
|
|
if !ok || data == nil {
|
|
return nil, errors.New("bad token value")
|
|
}
|
|
parseString := func(key string) (*string, error) {
|
|
v := data[key]
|
|
if v == nil {
|
|
return nil, nil
|
|
}
|
|
vv, ok := v.(string)
|
|
if !ok {
|
|
return nil, errors.New("bad string value")
|
|
}
|
|
return &vv, nil
|
|
}
|
|
parseStringList := func(key string) ([]string, error) {
|
|
v := data[key]
|
|
if v == nil {
|
|
return nil, nil
|
|
}
|
|
vv, ok := v.([]interface{})
|
|
if !ok {
|
|
return nil, errors.New("bad string list")
|
|
}
|
|
vvv := make([]string, 0, len(vv))
|
|
for _, s := range vv {
|
|
ss, ok := s.(string)
|
|
if !ok {
|
|
return nil, errors.New("bad string list")
|
|
}
|
|
vvv = append(vvv, ss)
|
|
}
|
|
return vvv, nil
|
|
}
|
|
parseTime := func(key string) (*time.Time, error) {
|
|
v := data[key]
|
|
if v == nil {
|
|
return nil, nil
|
|
}
|
|
switch v := v.(type) {
|
|
case string:
|
|
vv, err := time.Parse(time.RFC3339, v)
|
|
if err != nil {
|
|
return nil, errors.New("bad time value")
|
|
}
|
|
return &vv, nil
|
|
case float64: // relative time
|
|
vv := time.Now().Add(time.Duration(v) * time.Millisecond)
|
|
return &vv, nil
|
|
default:
|
|
return nil, errors.New("bad time value")
|
|
}
|
|
}
|
|
|
|
t, err := parseString("token")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
tt := ""
|
|
if t != nil {
|
|
tt = *t
|
|
}
|
|
u, err := parseString("username")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
g, err := parseString("group")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
gg := ""
|
|
if g != nil {
|
|
gg = *g
|
|
}
|
|
p, err := parseStringList("permissions")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
e, err := parseTime("expires")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
n, err := parseTime("not-before")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return &token.Stateful{
|
|
Token: tt,
|
|
Group: gg,
|
|
Username: u,
|
|
Permissions: p,
|
|
Expires: e,
|
|
NotBefore: n,
|
|
}, 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(500 * time.Millisecond),
|
|
)
|
|
if err != nil {
|
|
return
|
|
}
|
|
switch m := m.(type) {
|
|
case clientMessage:
|
|
err := conn.WriteJSON(m)
|
|
if err != nil {
|
|
return
|
|
}
|
|
case []byte:
|
|
err := conn.WriteMessage(websocket.TextMessage, m)
|
|
if err != nil {
|
|
return
|
|
}
|
|
case closeMessage:
|
|
if m.data != nil {
|
|
conn.WriteMessage(
|
|
websocket.CloseMessage,
|
|
m.data,
|
|
)
|
|
}
|
|
return
|
|
default:
|
|
log.Printf("clientWriter: unexpected message %T", m)
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
func (c *webClient) Warn(oponly bool, message string) error {
|
|
if oponly && !member("op", c.permissions) {
|
|
return nil
|
|
}
|
|
|
|
return c.write(clientMessage{
|
|
Type: "usermessage",
|
|
Kind: "warning",
|
|
Dest: c.id,
|
|
Privileged: true,
|
|
Value: message,
|
|
})
|
|
}
|
|
|
|
var ErrClientDead = errors.New("client is dead")
|
|
|
|
func (c *webClient) action(a interface{}) {
|
|
c.actions.Put(a)
|
|
}
|
|
|
|
func (c *webClient) write(m clientMessage) error {
|
|
select {
|
|
case c.writeCh <- m:
|
|
return nil
|
|
case <-c.writerDone:
|
|
return ErrClientDead
|
|
}
|
|
}
|
|
|
|
func broadcast(cs []group.Client, m clientMessage) error {
|
|
b, err := json.Marshal(m)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
for _, c := range cs {
|
|
cc, ok := c.(*webClient)
|
|
if !ok {
|
|
continue
|
|
}
|
|
select {
|
|
case cc.writeCh <- b:
|
|
case <-cc.writerDone:
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (c *webClient) close(data []byte) error {
|
|
select {
|
|
case c.writeCh <- closeMessage{data}:
|
|
return nil
|
|
case <-c.writerDone:
|
|
return ErrClientDead
|
|
}
|
|
}
|
|
|
|
func errorMessage(id string, err error) *clientMessage {
|
|
switch e := err.(type) {
|
|
case group.UserError:
|
|
return &clientMessage{
|
|
Type: "usermessage",
|
|
Kind: "error",
|
|
Dest: id,
|
|
Privileged: true,
|
|
Value: e.Error(),
|
|
}
|
|
case group.KickError:
|
|
message := e.Message
|
|
if message == "" {
|
|
message = "you have been kicked out"
|
|
}
|
|
return &clientMessage{
|
|
Type: "usermessage",
|
|
Kind: "kicked",
|
|
Id: e.Id,
|
|
Username: e.Username,
|
|
Dest: id,
|
|
Privileged: true,
|
|
Value: message,
|
|
}
|
|
default:
|
|
return nil
|
|
}
|
|
}
|
|
|
|
func (c *webClient) error(err error) error {
|
|
m := errorMessage(c.id, err)
|
|
if m == nil {
|
|
return err
|
|
}
|
|
return c.write(*m)
|
|
}
|