Merge branch 'AlexxIT:master' into onvif-client

internal/yandex/README.md

@@ -0,0 +1,22 @@
+# Yandex
+
+Source for receiving stream from new [Yandex IP camera](https://alice.yandex.ru/smart-home/security/ipcamera).
+
+## Get Yandex token
+
+1. Install HomeAssistant integration [YandexStation](https://github.com/AlexxIT/YandexStation).
+2. Copy token from HomeAssistant config folder: `/config/.storage/core.config_entries`, key: `"x_token"`.
+
+## Get device ID
+
+1. Open this link in any browser: https://iot.quasar.yandex.ru/m/v3/user/devices
+2. Copy ID of your camera, key: `"id"`.
+
+## Config examples
+
+```yaml
+streams:
+  yandex_stream: yandex:?x_token=XXXX&device_id=XXXX
+  yandex_snapshot: yandex:?x_token=XXXX&device_id=XXXX&snapshot
+  yandex_snapshot_custom_size: yandex:?x_token=XXXX&device_id=XXXX&snapshot=h=540
+```

internal/yandex/goloom.go

@@ -0,0 +1,152 @@
+package yandex
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/AlexxIT/go2rtc/internal/webrtc"
+	"github.com/AlexxIT/go2rtc/pkg/core"
+	xwebrtc "github.com/AlexxIT/go2rtc/pkg/webrtc"
+	"github.com/google/uuid"
+	"github.com/gorilla/websocket"
+	pion "github.com/pion/webrtc/v4"
+)
+
+func goloomClient(serviceURL, serviceName, roomId, participantId, credentials string) (core.Producer, error) {
+	conn, _, err := websocket.DefaultDialer.Dial(serviceURL, nil)
+	if err != nil {
+		return nil, err
+	}
+	defer func() {
+		time.Sleep(time.Second)
+		_ = conn.Close()
+	}()
+
+	s := fmt.Sprintf(`{"hello": {
+"credentials":"%s","participantId":"%s","roomId":"%s","serviceName":"%s","sdkInitializationId":"%s",
+"capabilitiesOffer":{},"sendAudio":false,"sendSharing":false,"sendVideo":false,
+"sdkInfo":{"hwConcurrency":4,"implementation":"browser","version":"5.4.0"},
+"participantAttributes":{"description":"","name":"mike","role":"SPEAKER"},
+"participantMeta":{"description":"","name":"mike","role":"SPEAKER","sendAudio":false,"sendVideo":false}
+},"uid":"%s"}`,
+		credentials, participantId, roomId, serviceName,
+		uuid.NewString(), uuid.NewString(),
+	)
+
+	err = conn.WriteMessage(websocket.TextMessage, []byte(s))
+	if err != nil {
+		return nil, err
+	}
+
+	if _, _, err = conn.ReadMessage(); err != nil {
+		return nil, err
+	}
+
+	pc, err := webrtc.PeerConnection(true)
+	if err != nil {
+		return nil, err
+	}
+
+	prod := xwebrtc.NewConn(pc)
+	prod.FormatName = "yandex"
+	prod.Mode = core.ModeActiveProducer
+	prod.Protocol = "wss"
+
+	var connState core.Waiter
+
+	prod.Listen(func(msg any) {
+		switch msg := msg.(type) {
+		case pion.PeerConnectionState:
+			switch msg {
+			case pion.PeerConnectionStateConnecting:
+			case pion.PeerConnectionStateConnected:
+				connState.Done(nil)
+			default:
+				connState.Done(errors.New("webrtc: " + msg.String()))
+			}
+		}
+	})
+
+	go func() {
+		for {
+			var msg map[string]json.RawMessage
+			if err = conn.ReadJSON(&msg); err != nil {
+				return
+			}
+
+			for k, v := range msg {
+				switch k {
+				case "uid":
+					continue
+				case "serverHello":
+				case "subscriberSdpOffer":
+					var sdp subscriberSdp
+					if err = json.Unmarshal(v, &sdp); err != nil {
+						return
+					}
+					//log.Trace().Msgf("offer:\n%s", sdp.Sdp)
+					if err = prod.SetOffer(sdp.Sdp); err != nil {
+						return
+					}
+					if sdp.Sdp, err = prod.GetAnswer(); err != nil {
+						return
+					}
+					//log.Trace().Msgf("answer:\n%s", sdp.Sdp)
+
+					var raw []byte
+					if raw, err = json.Marshal(sdp); err != nil {
+						return
+					}
+					s = fmt.Sprintf(`{"uid":"%s","subscriberSdpAnswer":%s}`, uuid.NewString(), raw)
+					if err = conn.WriteMessage(websocket.TextMessage, []byte(s)); err != nil {
+						return
+					}
+				case "webrtcIceCandidate":
+					var candidate webrtcIceCandidate
+					if err = json.Unmarshal(v, &candidate); err != nil {
+						return
+					}
+					if err = prod.AddCandidate(candidate.Candidate); err != nil {
+						return
+					}
+				}
+				//log.Trace().Msgf("%s : %s", k, v)
+			}
+
+			if msg["ack"] != nil {
+				continue
+			}
+
+			s = fmt.Sprintf(`{"uid":%s,"ack":{"status":{"code":"OK"}}}`, msg["uid"])
+			if err = conn.WriteMessage(websocket.TextMessage, []byte(s)); err != nil {
+				return
+			}
+		}
+	}()
+
+	if err = connState.Wait(); err != nil {
+		return nil, err
+	}
+
+	s = fmt.Sprintf(`{"uid":"%s","setSlots":{"slots":[{"width":0,"height":0}],"audioSlotsCount":0,"key":1,"shutdownAllVideo":false,"withSelfView":false,"selfViewVisibility":"ON_LOADING_THEN_HIDE","gridConfig":{}}}`, uuid.NewString())
+	if err = conn.WriteMessage(websocket.TextMessage, []byte(s)); err != nil {
+		return nil, err
+	}
+
+	return prod, nil
+}
+
+type subscriberSdp struct {
+	PcSeq int    `json:"pcSeq"`
+	Sdp   string `json:"sdp"`
+}
+
+type webrtcIceCandidate struct {
+	PcSeq         int    `json:"pcSeq"`
+	Target        string `json:"target"`
+	Candidate     string `json:"candidate"`
+	SdpMid        string `json:"sdpMid"`
+	SdpMlineIndex int    `json:"sdpMlineIndex"`
+}

internal/yandex/yandex.go

@@ -0,0 +1,44 @@
+package yandex
+
+import (
+	"net/url"
+
+	"github.com/AlexxIT/go2rtc/internal/streams"
+	"github.com/AlexxIT/go2rtc/pkg/core"
+	"github.com/AlexxIT/go2rtc/pkg/yandex"
+)
+
+func Init() {
+	streams.HandleFunc("yandex", func(source string) (core.Producer, error) {
+		u, err := url.Parse(source)
+		if err != nil {
+			return nil, err
+		}
+
+		query := u.Query()
+		token := query.Get("x_token")
+
+		session, err := yandex.GetSession(token)
+		if err != nil {
+			return nil, err
+		}
+
+		deviceID := query.Get("device_id")
+
+		if query.Has("snapshot") {
+			rawURL, err := session.GetSnapshotURL(deviceID)
+			if err != nil {
+				return nil, err
+			}
+			rawURL += "/current.jpg?" + query.Get("snapshot") + "#header=Cookie:" + session.GetCookieString(rawURL)
+			return streams.GetProducer(rawURL)
+		}
+
+		room, err := session.WebrtcCreateRoom(deviceID)
+		if err != nil {
+			return nil, err
+		}
+
+		return goloomClient(room.ServiceUrl, room.ServiceName, room.RoomId, room.ParticipantId, room.Credentials)
+	})
+}

main.go

@@ -39,6 +39,7 @@ import (
 	"github.com/AlexxIT/go2rtc/internal/webrtc"
 	"github.com/AlexxIT/go2rtc/internal/webtorrent"
 	"github.com/AlexxIT/go2rtc/internal/wyoming"
+	"github.com/AlexxIT/go2rtc/internal/yandex"
 	"github.com/AlexxIT/go2rtc/pkg/shell"
 )
 
@@ -96,6 +97,7 @@ func main() {
 	alsa.Init()     // alsa source
 	flussonic.Init()
 	eseecloud.Init()
+	yandex.Init()
 
 	// 6. Helper modules
 

pkg/webrtc/server.go

@@ -65,7 +65,8 @@ transeivers:
 
 		switch tr.Direction() {
 		case webrtc.RTPTransceiverDirectionSendrecv:
-			_ = tr.Sender().Stop()
+			_ = tr.Sender().Stop()             // don't know if necessary
+			_ = tr.SetSender(tr.Sender(), nil) // set direction to recvonly
 		case webrtc.RTPTransceiverDirectionSendonly:
 			_ = tr.Stop()
 		}

pkg/yandex/session.go

@@ -0,0 +1,203 @@
+package yandex
+
+import (
+	"encoding/json"
+	"errors"
+	"io"
+	"net/http"
+	"net/http/cookiejar"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/AlexxIT/go2rtc/pkg/core"
+)
+
+type Session struct {
+	token  string
+	client *http.Client
+}
+
+var sessions = map[string]*Session{}
+var sessionsMu sync.Mutex
+
+func GetSession(token string) (*Session, error) {
+	sessionsMu.Lock()
+	defer sessionsMu.Unlock()
+
+	if session, ok := sessions[token]; ok {
+		return session, nil
+	}
+
+	session := &Session{token: token}
+	if err := session.Login(); err != nil {
+		return nil, err
+	}
+
+	sessions[token] = session
+
+	return session, nil
+}
+
+func (s *Session) Login() error {
+	req, err := http.NewRequest(
+		"POST", "https://mobileproxy.passport.yandex.net/1/bundle/auth/x_token/",
+		strings.NewReader("type=x-token&retpath=https%3A%2F%2Fwww.yandex.ru"),
+	)
+	if err != nil {
+		return err
+	}
+
+	req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
+	req.Header.Set("Ya-Consumer-Authorization", "OAuth "+s.token)
+
+	res, err := http.DefaultClient.Do(req)
+	if err != nil {
+		return err
+	}
+
+	var auth struct {
+		PassportHost string `json:"passport_host"`
+		Status       string `json:"status"`
+		TrackId      string `json:"track_id"`
+	}
+	if err = json.NewDecoder(res.Body).Decode(&auth); err != nil {
+		return err
+	}
+
+	if auth.Status != "ok" {
+		return errors.New("yandex: login error: " + auth.Status)
+	}
+
+	s.client = &http.Client{Timeout: 15 * time.Second}
+	s.client.CheckRedirect = func(req *http.Request, via []*http.Request) error {
+		return http.ErrUseLastResponse
+	}
+	s.client.Jar, _ = cookiejar.New(nil)
+
+	res, err = s.client.Get(auth.PassportHost + "/auth/session/?track_id=" + auth.TrackId)
+	if err != nil {
+		return err
+	}
+
+	s.client.CheckRedirect = nil
+
+	return nil
+}
+
+func (s *Session) Get(url string) (*http.Response, error) {
+	return s.client.Get(url)
+}
+
+func (s *Session) GetCSRF() (string, error) {
+	res, err := s.Get("https://yandex.ru/quasar")
+	if err != nil {
+		return "", err
+	}
+
+	body, err := io.ReadAll(res.Body)
+	if err != nil {
+		return "", err
+	}
+
+	token := core.Between(string(body), `"csrfToken2":"`, `"`)
+	return token, nil
+}
+
+func (s *Session) GetCookieString(url string) string {
+	req, err := http.NewRequest("GET", url, nil)
+	if err != nil {
+		return ""
+	}
+	for _, cookie := range s.client.Jar.Cookies(req.URL) {
+		req.AddCookie(cookie)
+	}
+	return req.Header.Get("Cookie")
+}
+
+func (s *Session) GetDevices() ([]Device, error) {
+	res, err := s.Get("https://iot.quasar.yandex.ru/m/v3/user/devices")
+	if err != nil {
+		return nil, err
+	}
+
+	var data struct {
+		Households []struct {
+			All []Device `json:"all"`
+		} `json:"households"`
+	}
+
+	if err = json.NewDecoder(res.Body).Decode(&data); err != nil {
+		return nil, err
+	}
+
+	var devices []Device
+	for _, household := range data.Households {
+		devices = append(devices, household.All...)
+	}
+	return devices, nil
+}
+
+func (s *Session) GetSnapshotURL(deviceID string) (string, error) {
+	devices, err := s.GetDevices()
+	if err != nil {
+		return "", err
+	}
+
+	for _, device := range devices {
+		if device.Id == deviceID {
+			return device.Parameters.SnapshotUrl, nil
+		}
+	}
+
+	return "", errors.New("yandex: can't get snapshot url for device: " + deviceID)
+}
+
+func (s *Session) WebrtcCreateRoom(deviceID string) (*Room, error) {
+	csrf, err := s.GetCSRF()
+	if err != nil {
+		return nil, err
+	}
+
+	req, err := http.NewRequest(
+		"POST", "https://iot.quasar.yandex.ru/m/v3/user/devices/"+deviceID+"/webrtc/create-room",
+		strings.NewReader(`{"protocol":"whip"}`),
+	)
+	if err != nil {
+		return nil, err
+	}
+
+	req.Header.Add("Content-Type", "application/json")
+	req.Header.Add("X-CSRF-Token", csrf)
+
+	res, err := s.client.Do(req)
+	if err != nil {
+		return nil, err
+	}
+
+	var data struct {
+		Result Room `json:"result"`
+	}
+	if err = json.NewDecoder(res.Body).Decode(&data); err != nil {
+		return nil, err
+	}
+
+	return &data.Result, nil
+}
+
+type Device struct {
+	Id         string `json:"id"`
+	Name       string `json:"name"`
+	Type       string `json:"type"`
+	Parameters struct {
+		SnapshotUrl string `json:"snapshot_url,omitempty"`
+	} `json:"parameters"`
+}
+
+type Room struct {
+	ServiceUrl    string `json:"service_url"`
+	ServiceName   string `json:"service_name"`
+	RoomId        string `json:"room_id"`
+	ParticipantId string `json:"participant_id"`
+	Credentials   string `json:"jwt"`
+}