Compare commits

...

40 Commits

Author SHA1 Message Date
Alexey Khit
4b27d119f0 Update version to 0.1-rc.3 2022-11-14 09:50:47 +03:00
Alexey Khit
dd55c03dc2 Add support multiple transcoding for ffmpeg 2022-11-14 02:26:14 +03:00
Alexey Khit
a4eab1944a Add ffmpeg async option for HomeKit audio 2022-11-14 01:29:45 +03:00
Alexey Khit
eea413a36c Support stream name as ffmpeg input 2022-11-14 01:22:07 +03:00
Alexey Khit
cdd42a8ed2 Change HomeKit codec from AAC to ELD 2022-11-14 00:58:34 +03:00
Alexey Khit
4815ce1baf Fix stop ffmpeg without matching tracks 2022-11-14 00:58:34 +03:00
Alexey Khit
e6d3939c78 Fix external producers 2022-11-13 21:33:09 +03:00
Alexey Khit
220b9ca318 Remove old example file 2022-11-13 20:54:19 +03:00
Alexey Khit
d625620dfd Fix ffmpeg profile warning 2022-11-13 20:09:18 +03:00
Alexey Khit
dd503f3410 Adds rotate template for ffmpeg 2022-11-13 20:05:54 +03:00
Alexey Khit
3e8e87bfcc Fix RTSP unknown response handler 2022-11-13 19:26:22 +03:00
Alexey Khit
64d218886e Add exec early exit handler 2022-11-13 19:24:26 +03:00
Alexey Khit
e91ccc211e Change ffmpeg verbose level to error 2022-11-13 19:18:53 +03:00
Alexey Khit
9f8a219483 Exec stderr will show with debug log 2022-11-13 19:15:12 +03:00
Alexey Khit
b617796941 Improve RTCP for HomeKit 2022-11-13 14:35:53 +03:00
Alexey Khit
77888fe086 Refactoring for HomeKit client 2022-11-11 22:47:14 +03:00
Alexey Khit
7bc3534bcb Add useful links to readmes 2022-11-11 22:44:54 +03:00
Alexey Khit
77bc0630d6 Add teaps to main readme 2022-11-11 22:44:49 +03:00
Alexey Khit
2f68711405 Fix MP4f test consumer 2022-11-11 22:44:34 +03:00
Alexey Khit
b8cab5db60 Remove aacparser from MP4 muxer 2022-11-11 22:44:05 +03:00
Alexey Khit
eae01be71f Add User-Agent to FFmpeg input and output 2022-11-11 18:04:31 +03:00
Alexey Khit
0127115180 Add User-Agent to go2rtc RTSP requests 2022-11-11 18:02:08 +03:00
Alexey Khit
aef84cef6b Add go2rtc version info 2022-11-11 18:01:38 +03:00
Alexey Khit
d478436758 Set video track for WebRTC always first 2022-11-11 16:33:08 +03:00
Alexey Khit
f77db44529 Refactoring for HomeKit client 2022-11-11 13:24:09 +03:00
Alex X
149d1bf235 Merge pull request #101 from felipecrs/patch-2
Mention alternative method to import hass cameras
2022-11-09 20:34:31 +03:00
Felipe Santos
b650475b10 Mention alternative method to import hass cameras 2022-11-09 13:00:30 -03:00
Alexey Khit
16e5406156 Update readme 2022-11-08 09:57:17 +03:00
Alexey Khit
49f6233bde Update RTSP server filters 2022-11-08 01:50:28 +03:00
Alexey Khit
78c5c70c73 Add duration API for MP4 file 2022-11-08 01:29:58 +03:00
Alexey Khit
32651c74ab Fix frame.mp4 support 2022-11-08 01:13:38 +03:00
Alexey Khit
5c64d1f847 Update MSE procession on JS side 2022-11-08 00:37:32 +03:00
Alexey Khit
717af29630 Refactoring 2022-11-08 00:37:13 +03:00
Alexey Khit
ea18475d31 Split MSE data on packets 2022-11-07 23:35:36 +03:00
Alexey Khit
701a9c69ec Update write websocket func 2022-11-07 23:35:08 +03:00
Alexey Khit
c06253c8b2 Fix producer request new track after start 2022-11-07 17:48:45 +03:00
Alexey Khit
3a07e9fa03 Fix lock on mp4 restarts 2022-11-07 13:32:27 +03:00
Alexey Khit
e1bc30fab3 Add support AAC for RTSP 2022-11-07 11:02:26 +03:00
Alexey Khit
d16ae0972f Code refactoring 2022-11-07 11:01:19 +03:00
Alexey Khit
8b93c97e69 Add support AAC for RTMP to MP4 2022-11-06 22:44:48 +03:00
57 changed files with 2085 additions and 1536 deletions

View File

@@ -140,8 +140,8 @@ Available modules:
Available source types:
- [rtsp](#source-rtsp) - `RTSP` and `RTSPS` cameras
- [rtmp](#source-rtmp) - `RTMP` streams
- [ffmpeg](#source-ffmpeg) - FFmpeg integration (`MJPEG`, `HLS`, `files` and source types)
- [rtmp](#source-rtmp) - `RTMP` and `HTTP-FLV` streams
- [ffmpeg](#source-ffmpeg) - FFmpeg integration (`MJPEG`, `HLS`, `files` and others)
- [ffmpeg:device](#source-ffmpeg-device) - local USB Camera or Webcam
- [exec](#source-exec) - advanced FFmpeg and GStreamer integration
- [echo](#source-echo) - get stream link from bash or python
@@ -306,6 +306,8 @@ streams:
aqara_g3: hass:Camera-Hub-G3-AB12
```
More cameras, like [Tuya](https://www.home-assistant.io/integrations/tuya/), [ONVIF](https://www.home-assistant.io/integrations/onvif/), and possibly others can also be imported by using [this method](https://github.com/felipecrs/hass-expose-camera-stream-source#importing-home-assistant-cameras-to-go2rtc-andor-frigate).
### Module: API
The HTTP API is the main part for interacting with the application. Default address: `http://127.0.0.1:1984/`.
@@ -323,9 +325,9 @@ api:
static_dir: "" # folder for static files (custom web interface)
```
**PS. go2rtc** don't provide HTTPS or password protection. Use [Nginx](https://nginx.org/) or [Ngrok](#module-ngrok) or [Home Assistant Add-on](#go2rtc-home-assistant-add-on) for this tasks.
**PS. go2rtc** doesn't provide HTTPS or password protection. Use [Nginx](https://nginx.org/) or [Ngrok](#module-ngrok) or [Home Assistant Add-on](#go2rtc-home-assistant-add-on) for this tasks.
**PS2.** You can access microphone (for 2-way audio) only with HTTPS
**PS2.** You can access microphone (for 2-way audio) only with HTTPS ([read more](https://stackoverflow.com/questions/52759992/how-to-access-camera-and-microphone-in-chrome-without-https)).
### Module: RTSP
@@ -566,13 +568,23 @@ iPad Safari | H264, H265* | H264 | no
Android Chrome | H264 | H264 | H264
masOS Hass App | no | no | no
- WebRTC audio codecs: `PCMU/8000`, `PCMA/8000`, `OPUS/48000/2`
- MSE/MP4 audio codecs: not supported yet (should be: `AAC`)
- Chrome H265: [read this](https://github.com/StaZhu/enable-chromium-hevc-hardware-decoding)
- Edge H265: [read this](https://www.reddit.com/r/MicrosoftEdge/comments/v9iw8k/enable_hevc_support_in_edge/)
- Desktop Safari H265: Menu > Develop > Experimental > WebRTC H265
- iOS Safari H265: Settings > Safari > Advanced > Experimental > WebRTC H265
**Audio**
- WebRTC audio codecs: `PCMU/8000`, `PCMA/8000`, `OPUS/48000/2`
- MSE/MP4 audio codecs: `AAC`
## TIPS
**Using apps for low RTSP delay**
- `ffplay -fflags nobuffer -flags low_delay "rtsp://192.168.1.123:8554/camera1"`
- VLC > Preferences > Input / Codecs > Default Caching Level: Lowest Latency
## FAQ
**Q. What's the difference between go2rtc, WebRTC Camera and RTSPtoWebRTC?**

View File

@@ -64,22 +64,14 @@ func (ctx *Context) Close() {
func (ctx *Context) Write(msg interface{}) {
ctx.mu.Lock()
defer ctx.mu.Unlock()
var err error
switch msg := msg.(type) {
case *streamer.Message:
err = ctx.Conn.WriteJSON(msg)
case []byte:
err = ctx.Conn.WriteMessage(websocket.BinaryMessage, msg)
default:
return
if data, ok := msg.([]byte); ok {
_ = ctx.Conn.WriteMessage(websocket.BinaryMessage, data)
} else {
_ = ctx.Conn.WriteJSON(msg)
}
if err != nil {
//panic(err) // TODO: fix panic
}
ctx.mu.Unlock()
}
func (ctx *Context) Error(err error) {

View File

@@ -10,6 +10,9 @@ import (
"runtime"
)
var Version = "0.1-rc.3"
var UserAgent = "go2rtc/" + Version
func Init() {
config := flag.String(
"config",
@@ -35,9 +38,10 @@ func Init() {
modules = cfg.Mod
log.Info().Msgf("go2rtc version %s %s/%s", Version, runtime.GOOS, runtime.GOARCH)
path, _ := os.Getwd()
log.Debug().Str("os", runtime.GOOS).Str("arch", runtime.GOARCH).
Str("cwd", path).Int("conf_size", len(data)).Msgf("[app]")
log.Debug().Str("cwd", path).Send()
}
func NewLogger(format string, level string) zerolog.Logger {

View File

@@ -4,6 +4,7 @@ import (
"crypto/md5"
"encoding/hex"
"errors"
"fmt"
"github.com/AlexxIT/go2rtc/cmd/app"
"github.com/AlexxIT/go2rtc/cmd/rtsp"
"github.com/AlexxIT/go2rtc/cmd/streams"
@@ -56,6 +57,8 @@ func Handle(url string) (streamer.Producer, error) {
if log.Trace().Enabled() {
cmd.Stdout = os.Stdout
}
if log.Debug().Enabled() {
cmd.Stderr = os.Stderr
}
@@ -80,11 +83,19 @@ func Handle(url string) (streamer.Producer, error) {
return nil, err
}
chErr := make(chan error)
go func() {
chErr <- cmd.Wait()
}()
select {
case <-time.After(time.Second * 15):
case <-time.After(time.Second * 60):
_ = cmd.Process.Kill()
log.Error().Str("url", url).Msg("[exec] timeout")
return nil, errors.New("timeout")
case err := <-chErr:
return nil, fmt.Errorf("exec: %s", err)
case prod := <-ch:
log.Debug().Stringer("launch", time.Since(ts)).Msg("[exec] run")
return prod, nil

View File

@@ -1,3 +1,17 @@
## FFplay output
[FFplay](https://stackoverflow.com/questions/27778678/what-are-mv-fd-aq-vq-sq-and-f-in-a-video-stream) `7.11 A-V: 0.003 fd= 1 aq= 21KB vq= 321KB sq= 0B f=0/0`:
- `7.11` - master clock, is the time from start of the stream/video
- `A-V` - av_diff, difference between audio and video timestamps
- `fd` - frames dropped
- `aq` - audio queue (0 - no delay)
- `vq` - video queue (0 - no delay)
- `sq` - subtitle queue
- `f` - timestamp error correction rate (Not 100% sure)
`M-V`, `M-A` means video stream only, audio stream only respectively.
## Devices Windows
```
@@ -41,3 +55,7 @@
- https://trac.ffmpeg.org/wiki/DirectShow
- https://stackoverflow.com/questions/53207692/libav-mjpeg-encoding-and-huffman-table
- https://github.com/tuupola/esp_video/blob/master/README.md
- https://github.com/leandromoreira/ffmpeg-libav-tutorial
- https://www.reddit.com/user/VeritablePornocopium/comments/okw130/ffmpeg_with_libfdk_aac_for_windows_x64/
- https://slhck.info/video/2017/02/24/vbr-settings.html
- [HomeKit audio samples problem](https://superuser.com/questions/1290996/non-monotonous-dts-with-igndts-flag)

View File

@@ -4,9 +4,11 @@ import (
"github.com/AlexxIT/go2rtc/cmd/app"
"github.com/AlexxIT/go2rtc/cmd/exec"
"github.com/AlexxIT/go2rtc/cmd/ffmpeg/device"
"github.com/AlexxIT/go2rtc/cmd/rtsp"
"github.com/AlexxIT/go2rtc/cmd/streams"
"github.com/AlexxIT/go2rtc/pkg/streamer"
"net/url"
"strconv"
"strings"
)
@@ -23,40 +25,48 @@ func Init() {
// inputs
"file": "-re -stream_loop -1 -i {input}",
"http": "-fflags nobuffer -flags low_delay -i {input}",
"rtsp": "-fflags nobuffer -flags low_delay -rtsp_transport tcp -timeout 5000000 -i {input}",
"rtsp": "-fflags nobuffer -flags low_delay -timeout 5000000 -user_agent go2rtc/ffmpeg -rtsp_transport tcp -i {input}",
// output
"output": "-rtsp_transport tcp -f rtsp {output}",
"output": "-user_agent ffmpeg/go2rtc -rtsp_transport tcp -f rtsp {output}",
// `-g 30` - group of picture, GOP, keyframe interval
// `-preset superfast` - we can't use ultrafast because it doesn't support `-profile main -level 4.1`
// `-tune zerolatency` - for minimal latency
// `-profile main -level 4.1` - most used streaming profile
// `-pix_fmt yuv420p` - if input pix format 4:2:2
"h264": "-codec:v libx264 -g 30 -preset superfast -tune zerolatency -profile main -level 4.1 -pix_fmt yuv420p",
"h264/ultra": "-codec:v libx264 -g 30 -preset ultrafast -tune zerolatency",
"h264/high": "-codec:v libx264 -g 30 -preset superfast -tune zerolatency",
"h265": "-codec:v libx265 -g 30 -preset ultrafast -tune zerolatency",
"mjpeg": "-codec:v mjpeg -force_duplicated_matrix 1 -huffman 0 -pix_fmt yuvj420p",
"opus": "-codec:a libopus -ar 48000 -ac 2",
"pcmu": "-codec:a pcm_mulaw -ar 8000 -ac 1",
"pcmu/16000": "-codec:a pcm_mulaw -ar 16000 -ac 1",
"pcmu/48000": "-codec:a pcm_mulaw -ar 48000 -ac 1",
"pcma": "-codec:a pcm_alaw -ar 8000 -ac 1",
"pcma/16000": "-codec:a pcm_alaw -ar 16000 -ac 1",
"pcma/48000": "-codec:a pcm_alaw -ar 48000 -ac 1",
"aac/16000": "-codec:a aac -ar 16000 -ac 1",
"h264": "-c:v libx264 -g:v 30 -preset:v superfast -tune:v zerolatency -profile:v main -level:v 4.1 -pix_fmt:v yuv420p",
"h264/ultra": "-c:v libx264 -g:v 30 -preset:v ultrafast -tune:v zerolatency",
"h264/high": "-c:v libx264 -g:v 30 -preset:v superfast -tune:v zerolatency",
"h265": "-c:v libx265 -g:v 30 -preset:v ultrafast -tune:v zerolatency",
"mjpeg": "-c:v mjpeg -force_duplicated_matrix:v 1 -huffman:v 0 -pix_fmt:v yuvj420p",
"opus": "-c:a libopus -ar:a 48000 -ac:a 2",
"pcmu": "-c:a pcm_mulaw -ar:a 8000 -ac:a 1",
"pcmu/16000": "-c:a pcm_mulaw -ar:a 16000 -ac:a 1",
"pcmu/48000": "-c:a pcm_mulaw -ar:a 48000 -ac:a 1",
"pcma": "-c:a pcm_alaw -ar:a 8000 -ac:a 1",
"pcma/16000": "-c:a pcm_alaw -ar:a 16000 -ac:a 1",
"pcma/48000": "-c:a pcm_alaw -ar:a 48000 -ac:a 1",
"aac": "-c:a aac", // keep sample rate and channels
"aac/16000": "-c:a aac -ar:a 16000 -ac:a 1",
}
app.LoadConfig(&cfg)
tpl := cfg.Mod
cmd := "exec:" + tpl["bin"] + " -hide_banner "
if app.GetLogger("exec").GetLevel() >= 0 {
cmd += "-v error "
}
streams.HandleFunc("ffmpeg", func(s string) (streamer.Producer, error) {
s = s[7:] // remove `ffmpeg:`
var query url.Values
var queryVideo, queryAudio bool
if i := strings.IndexByte(s, '#'); i > 0 {
query = parseQuery(s[i+1:])
queryVideo = query["video"] != nil
@@ -69,7 +79,7 @@ func Init() {
}
var input string
if i := strings.IndexByte(s, ':'); i > 0 {
if i := strings.Index(s, "://"); i > 0 {
switch s[:i] {
case "http", "https", "rtmp":
input = strings.Replace(tpl["http"], "{input}", s, 1)
@@ -86,52 +96,97 @@ func Init() {
}
input += strings.Replace(tpl["rtsp"], "{input}", s, 1)
default:
input = "-i " + s
}
} else if streams.Get(s) != nil {
s = "rtsp://localhost:" + rtsp.Port + "/" + s
switch {
case queryVideo && !queryAudio:
s += "?video"
case queryAudio && !queryVideo:
s += "?audio"
}
input = strings.Replace(tpl["rtsp"], "{input}", s, 1)
} else if strings.HasPrefix(s, "device?") {
var err error
input, err = device.GetInput(s)
if err != nil {
return nil, err
}
} else {
input = strings.Replace(tpl["file"], "{input}", s, 1)
}
if input == "" {
if strings.HasPrefix(s, "device?") {
var err error
input, err = device.GetInput(s)
if err != nil {
return nil, err
}
} else {
input = strings.Replace(tpl["file"], "{input}", s, 1)
}
if _, ok := query["async"]; ok {
input = "-use_wallclock_as_timestamps 1 -async 1 " + input
}
s = "exec:" + tpl["bin"] + " -hide_banner " + input
s = cmd + input
if query != nil {
for _, raw := range query["raw"] {
s += " " + raw
}
// TODO: multiple codecs via -map
// s += fmt.Sprintf(" -map 0:v:0 -c:v:%d copy", i)
for _, video := range query["video"] {
if video == "copy" {
s += " -codec:v copy"
} else {
s += " " + tpl[video]
for _, rotate := range query["rotate"] {
switch rotate {
case "90":
s += " -vf transpose=1" // 90 degrees clockwise
case "180":
s += " -vf transpose=1,transpose=1"
case "-90", "270":
s += " -vf transpose=2" // 90 degrees counterclockwise
}
break
}
for _, audio := range query["audio"] {
if audio == "copy" {
s += " -codec:a copy"
} else {
s += " " + tpl[audio]
}
}
switch {
case queryVideo && !queryAudio:
s += " -an"
case queryAudio && !queryVideo:
switch len(query["video"]) {
case 0:
s += " -vn"
case 1:
if len(query["audio"]) > 1 {
s += " -map 0:v:0"
}
for _, video := range query["video"] {
if video == "copy" {
s += " -c:v copy"
} else {
s += " " + tpl[video]
}
}
default:
for i, video := range query["video"] {
if video == "copy" {
s += " -map 0:v:0 -c:v:" + strconv.Itoa(i) + " copy"
} else {
s += " -map 0:v:0 " + strings.ReplaceAll(tpl[video], ":v ", ":v:"+strconv.Itoa(i)+" ")
}
}
}
switch len(query["audio"]) {
case 0:
s += " -an"
case 1:
if len(query["video"]) > 1 {
s += " -map 0:a:0"
}
for _, audio := range query["audio"] {
if audio == "copy" {
s += " -c:a copy"
} else {
s += " " + tpl[audio]
}
}
default:
for i, audio := range query["audio"] {
if audio == "copy" {
s += " -map 0:a:0 -c:a:" + strconv.Itoa(i) + " copy"
} else {
s += " -map 0:a:0 " + strings.ReplaceAll(tpl[audio], ":a ", ":a:"+strconv.Itoa(i)+" ")
}
}
}
} else {
s += " -c copy"

View File

@@ -5,8 +5,8 @@ import (
"fmt"
"github.com/AlexxIT/go2rtc/cmd/app/store"
"github.com/AlexxIT/go2rtc/cmd/streams"
"github.com/AlexxIT/go2rtc/pkg/homekit"
"github.com/AlexxIT/go2rtc/pkg/homekit/mdns"
"github.com/AlexxIT/go2rtc/pkg/hap"
"github.com/AlexxIT/go2rtc/pkg/hap/mdns"
"net/http"
"net/url"
"strings"
@@ -54,91 +54,82 @@ func apiHandler(w http.ResponseWriter, r *http.Request) {
items = append(items, device)
}
_= json.NewEncoder(w).Encode(items)
_ = json.NewEncoder(w).Encode(items)
case "POST":
// TODO: post params...
id := r.URL.Query().Get("id")
pin := r.URL.Query().Get("pin")
client, err := homekit.Pair(id, pin)
if err != nil {
// log error
log.Error().Err(err).Msg("[api.homekit] pair")
// response error
_, err = w.Write([]byte(err.Error()))
return
}
name := r.URL.Query().Get("name")
dict := store.GetDict("streams")
dict[name] = client.URL()
if err = store.Set("streams", dict); err != nil {
// log error
log.Error().Err(err).Msg("[api.homekit] save to store")
// response error
if err := hkPair(id, pin, name); err != nil {
log.Error().Err(err).Caller().Send()
_, err = w.Write([]byte(err.Error()))
}
streams.New(name, client.URL())
case "DELETE":
src := r.URL.Query().Get("src")
dict := store.GetDict("streams")
for name, rawURL := range dict {
if name != src {
continue
}
client, err := homekit.NewClient(rawURL.(string))
if err != nil {
// log error
log.Error().Err(err).Msg("[api.homekit] new client")
// response error
_, err = w.Write([]byte(err.Error()))
return
}
if err = client.Dial(); err != nil {
// log error
log.Error().Err(err).Msg("[api.homekit] client dial")
// response error
_, err = w.Write([]byte(err.Error()))
return
}
go client.Handle()
if err = client.ListPairings(); err != nil {
// log error
log.Error().Err(err).Msg("[api.homekit] unpair")
// response error
_, err = w.Write([]byte(err.Error()))
return
}
if err = client.DeletePairing(client.ClientID); err != nil {
// log error
log.Error().Err(err).Msg("[api.homekit] unpair")
// response error
_, err = w.Write([]byte(err.Error()))
}
delete(dict, name)
if err = store.Set("streams", dict); err != nil {
// log error
log.Error().Err(err).Msg("[api.homekit] store set")
// response error
_, err = w.Write([]byte(err.Error()))
}
return
if err := hkDelete(src); err != nil {
log.Error().Err(err).Caller().Send()
_, err = w.Write([]byte(err.Error()))
}
}
}
func hkPair(deviceID, pin, name string) (err error) {
var conn *hap.Conn
if conn, err = hap.Pair(deviceID, pin); err != nil {
return
}
streams.New(name, conn.URL())
dict := store.GetDict("streams")
dict[name] = conn.URL()
return store.Set("streams", dict)
}
func hkDelete(name string) (err error) {
dict := store.GetDict("streams")
for key, rawURL := range dict {
if key != name {
continue
}
var conn *hap.Conn
if conn, err = hap.NewConn(rawURL.(string)); err != nil {
return
}
if err = conn.Dial(); err != nil {
return
}
go func() {
if err = conn.Handle(); err != nil {
log.Warn().Err(err).Caller().Send()
}
}()
if err = conn.ListPairings(); err != nil {
return
}
if err = conn.DeletePairing(conn.ClientID); err != nil {
log.Error().Err(err).Caller().Send()
}
delete(dict, name)
return store.Set("streams", dict)
}
return nil
}
type Device struct {
ID string `json:"id"`
Name string `json:"name"`

View File

@@ -3,6 +3,7 @@ package homekit
import (
"github.com/AlexxIT/go2rtc/cmd/api"
"github.com/AlexxIT/go2rtc/cmd/app"
"github.com/AlexxIT/go2rtc/cmd/srtp"
"github.com/AlexxIT/go2rtc/cmd/streams"
"github.com/AlexxIT/go2rtc/pkg/homekit"
"github.com/AlexxIT/go2rtc/pkg/streamer"
@@ -20,20 +21,12 @@ func Init() {
var log zerolog.Logger
func streamHandler(url string) (streamer.Producer, error) {
client, err := homekit.NewClient(url)
conn, err := homekit.NewClient(url, srtp.Server)
if err != nil {
return nil, err
}
if err = client.Dial(); err != nil {
if err = conn.Dial();err!=nil{
return nil, err
}
// start gorutine for reading responses from camera
go func() {
if err = client.Handle(); err != nil {
log.Warn().Err(err).Msg("[homekit] client")
}
}()
return &Producer{client: client}, nil
return conn, nil
}

View File

@@ -1,189 +0,0 @@
package homekit
import (
"errors"
"fmt"
"github.com/AlexxIT/go2rtc/cmd/srtp"
"github.com/AlexxIT/go2rtc/pkg/homekit"
"github.com/AlexxIT/go2rtc/pkg/homekit/camera"
pkg "github.com/AlexxIT/go2rtc/pkg/srtp"
"github.com/AlexxIT/go2rtc/pkg/streamer"
"github.com/brutella/hap/characteristic"
"github.com/brutella/hap/rtp"
"net"
"strconv"
)
type Producer struct {
streamer.Element
client *homekit.Client
medias []*streamer.Media
tracks []*streamer.Track
sessions []*pkg.Session
}
func (c *Producer) GetMedias() []*streamer.Media {
if c.medias == nil {
c.medias = c.getMedias()
}
return c.medias
}
func (c *Producer) GetTrack(media *streamer.Media, codec *streamer.Codec) *streamer.Track {
for _, track := range c.tracks {
if track.Codec == codec {
return track
}
}
track := &streamer.Track{Codec: codec, Direction: media.Direction}
c.tracks = append(c.tracks, track)
return track
}
func (c *Producer) Start() error {
if c.tracks == nil {
return errors.New("producer without tracks")
}
// get our server local IP-address
host, _, err := net.SplitHostPort(c.client.LocalAddr())
if err != nil {
return err
}
// get our server SRTP port
port, err := strconv.Atoi(srtp.Port)
if err != nil {
return err
}
// setup HomeKit stream session
hkSession := camera.NewSession()
hkSession.SetLocalEndpoint(host, uint16(port))
// create client for processing camera accessory
cam := camera.NewClient(c.client)
// try to start HomeKit stream
if err = cam.StartStream2(hkSession); err != nil {
panic(err) // TODO: fixme
}
// SRTP Video Session
vs := &pkg.Session{
LocalSSRC: hkSession.Config.Video.RTP.Ssrc,
RemoteSSRC: hkSession.Answer.SsrcVideo,
Track: c.tracks[0],
}
if err = vs.SetKeys(
hkSession.Offer.Video.MasterKey, hkSession.Offer.Video.MasterSalt,
hkSession.Answer.Video.MasterKey, hkSession.Answer.Video.MasterSalt,
); err != nil {
return err
}
// SRTP Audio Session
as := &pkg.Session{
LocalSSRC: hkSession.Config.Audio.RTP.Ssrc,
RemoteSSRC: hkSession.Answer.SsrcAudio,
Track: &streamer.Track{},
}
if err = as.SetKeys(
hkSession.Offer.Audio.MasterKey, hkSession.Offer.Audio.MasterSalt,
hkSession.Answer.Audio.MasterKey, hkSession.Answer.Audio.MasterSalt,
); err != nil {
return err
}
srtp.AddSession(vs)
srtp.AddSession(as)
c.sessions = []*pkg.Session{vs, as}
return nil
}
func (c *Producer) Stop() error {
err := c.client.Close()
for _, session := range c.sessions {
srtp.RemoveSession(session)
}
return err
}
func (c *Producer) getMedias() []*streamer.Media {
var medias []*streamer.Media
accs, err := c.client.GetAccessories()
acc := accs[0]
if err != nil {
panic(err)
}
// get supported video config (not really necessary)
char := acc.GetCharacter(characteristic.TypeSupportedVideoStreamConfiguration)
v1 := &rtp.VideoStreamConfiguration{}
if err = char.ReadTLV8(v1); err != nil {
panic(err)
}
for _, hkCodec := range v1.Codecs {
codec := &streamer.Codec{ClockRate: 90000}
switch hkCodec.Type {
case rtp.VideoCodecType_H264:
codec.Name = streamer.CodecH264
default:
panic(fmt.Sprintf("unknown codec: %d", hkCodec.Type))
}
media := &streamer.Media{
Kind: streamer.KindVideo, Direction: streamer.DirectionSendonly,
Codecs: []*streamer.Codec{codec},
}
medias = append(medias, media)
}
char = acc.GetCharacter(characteristic.TypeSupportedAudioStreamConfiguration)
v2 := &rtp.AudioStreamConfiguration{}
if err = char.ReadTLV8(v2); err != nil {
panic(err)
}
for _, hkCodec := range v2.Codecs {
codec := &streamer.Codec{
Channels: uint16(hkCodec.Parameters.Channels),
}
switch hkCodec.Type {
case rtp.AudioCodecType_AAC_ELD:
codec.Name = streamer.CodecAAC
default:
panic(fmt.Sprintf("unknown codec: %d", hkCodec.Type))
}
switch hkCodec.Parameters.Samplerate {
case rtp.AudioCodecSampleRate8Khz:
codec.ClockRate = 8000
case rtp.AudioCodecSampleRate16Khz:
codec.ClockRate = 16000
case rtp.AudioCodecSampleRate24Khz:
codec.ClockRate = 24000
default:
panic(fmt.Sprintf("unknown clockrate: %d", hkCodec.Parameters.Samplerate))
}
media := &streamer.Media{
Kind: streamer.KindAudio, Direction: streamer.DirectionSendonly,
Codecs: []*streamer.Codec{codec},
}
medias = append(medias, media)
}
return medias
}

View File

@@ -9,6 +9,7 @@ import (
"net/http"
"strconv"
"strings"
"time"
)
func Init() {
@@ -35,35 +36,29 @@ func handlerKeyframe(w http.ResponseWriter, r *http.Request) {
exit := make(chan []byte)
cons := &mp4.Consumer{}
cons := &mp4.Keyframe{}
cons.Listen(func(msg interface{}) {
switch msg := msg.(type) {
case []byte:
exit <- msg
if data, ok := msg.([]byte); ok && exit != nil {
exit <- data
exit = nil
}
})
if err := stream.AddConsumer(cons); err != nil {
log.Error().Err(err).Msg("[api.keyframe] add consumer")
log.Error().Err(err).Caller().Send()
return
}
defer stream.RemoveConsumer(cons)
data := <-exit
w.Header().Set("Content-Type", cons.MimeType())
data, err := cons.Init()
if err != nil {
log.Error().Err(err).Msg("[api.keyframe] init")
return
}
data = append(data, <-exit...)
stream.RemoveConsumer(cons)
// Apple Safari won't show frame without length
w.Header().Set("Content-Length", strconv.Itoa(len(data)))
w.Header().Set("Content-Type", cons.MimeType)
if _, err := w.Write(data); err != nil {
log.Error().Err(err).Msg("[api.keyframe] add consumer")
log.Error().Err(err).Caller().Send()
}
}
@@ -80,20 +75,20 @@ func handlerMP4(w http.ResponseWriter, r *http.Request) {
return
}
exit := make(chan struct{})
exit := make(chan error)
cons := &mp4.Consumer{}
cons.Listen(func(msg interface{}) {
switch msg := msg.(type) {
case []byte:
if _, err := w.Write(msg); err != nil {
exit <- struct{}{}
if data, ok := msg.([]byte); ok {
if _, err := w.Write(data); err != nil && exit != nil {
exit <- err
exit = nil
}
}
})
if err := stream.AddConsumer(cons); err != nil {
log.Error().Err(err).Msg("[api.mp4] add consumer")
log.Error().Err(err).Caller().Send()
return
}
@@ -103,18 +98,36 @@ func handlerMP4(w http.ResponseWriter, r *http.Request) {
data, err := cons.Init()
if err != nil {
log.Error().Err(err).Msg("[api.mp4] init")
log.Error().Err(err).Caller().Send()
return
}
if _, err = w.Write(data); err != nil {
log.Error().Err(err).Msg("[api.mp4] write")
log.Error().Err(err).Caller().Send()
return
}
<-exit
cons.Start()
log.Trace().Msg("[api.mp4] close")
var duration *time.Timer
if s := r.URL.Query().Get("duration"); s != "" {
if i, _ := strconv.Atoi(s); i > 0 {
duration = time.AfterFunc(time.Second*time.Duration(i), func() {
if exit != nil {
exit <- nil
exit = nil
}
})
}
}
err = <-exit
log.Trace().Err(err).Caller().Send()
if duration != nil {
duration.Stop()
}
}
func isChromeFirst(w http.ResponseWriter, r *http.Request) bool {

View File

@@ -9,6 +9,8 @@ import (
const MsgTypeMSE = "mse" // fMP4
const packetSize = 8192
func handlerWS(ctx *api.Context, msg *streamer.Message) {
src := ctx.Request.URL.Query().Get("src")
stream := streams.GetOrNew(src)
@@ -21,14 +23,17 @@ func handlerWS(ctx *api.Context, msg *streamer.Message) {
cons.RemoteAddr = ctx.Request.RemoteAddr
cons.Listen(func(msg interface{}) {
switch msg.(type) {
case *streamer.Message, []byte:
ctx.Write(msg)
if data, ok := msg.([]byte); ok {
for len(data) > packetSize {
ctx.Write(data[:packetSize])
data = data[packetSize:]
}
ctx.Write(data)
}
})
if err := stream.AddConsumer(cons); err != nil {
log.Warn().Err(err).Msg("[api.mse] add consumer")
log.Warn().Err(err).Caller().Send()
ctx.Error(err)
return
}
@@ -37,16 +42,16 @@ func handlerWS(ctx *api.Context, msg *streamer.Message) {
stream.RemoveConsumer(cons)
})
ctx.Write(&streamer.Message{
Type: MsgTypeMSE, Value: cons.MimeType(),
})
ctx.Write(&streamer.Message{Type: MsgTypeMSE, Value: cons.MimeType()})
data, err := cons.Init()
if err != nil {
log.Warn().Err(err).Msg("[api.mse] init")
log.Warn().Err(err).Caller().Send()
ctx.Error(err)
return
}
ctx.Write(data)
cons.Start()
}

View File

@@ -85,6 +85,8 @@ func rtspHandler(url string) (streamer.Producer, error) {
return nil, err
}
conn.UserAgent = app.UserAgent
if log.Trace().Enabled() {
conn.Listen(func(msg interface{}) {
switch msg := msg.(type) {
@@ -209,16 +211,27 @@ func initMedias(conn *rtsp.Conn) {
for key, value := range conn.URL.Query() {
switch key {
case streamer.KindVideo, streamer.KindAudio:
for _, value := range value {
for _, name := range value {
name = strings.ToUpper(name)
// check aliases
switch name {
case "COPY":
name = "" // pass empty codecs list
case "MJPEG":
name = streamer.CodecJPEG
case "AAC":
name = streamer.CodecAAC
}
media := &streamer.Media{
Kind: key, Direction: streamer.DirectionRecvonly,
}
switch value {
case "", "copy": // pass empty codecs list
default:
codec := streamer.NewCodec(value)
media.Codecs = append(media.Codecs, codec)
// empty codecs match all codecs
if name != "" {
// empty clock rate and channels match any values
media.Codecs = []*streamer.Codec{{Name: name}}
}
conn.Medias = append(conn.Medias, media)

View File

@@ -3,7 +3,6 @@ package srtp
import (
"github.com/AlexxIT/go2rtc/cmd/app"
"github.com/AlexxIT/go2rtc/pkg/srtp"
"github.com/rs/zerolog"
"net"
)
@@ -24,36 +23,23 @@ func Init() {
return
}
log = app.GetLogger("srtp")
log := app.GetLogger("srtp")
// create SRTP server (endpoint) for receiving video from HomeKit camera
conn, err := net.ListenPacket("udp", cfg.Mod.Listen)
if err != nil {
log.Warn().Err(err).Msg("[srtp] listen")
log.Warn().Err(err).Caller().Send()
}
log.Info().Str("addr", cfg.Mod.Listen).Msg("[srtp] listen")
_, Port, _ = net.SplitHostPort(cfg.Mod.Listen)
// run server
go func() {
server = &srtp.Server{}
if err = server.Serve(conn); err != nil {
log.Warn().Err(err).Msg("[srtp] serve")
Server = &srtp.Server{}
if err = Server.Serve(conn); err != nil {
log.Warn().Err(err).Caller().Send()
}
}()
}
var log zerolog.Logger
var server *srtp.Server
var Port string
func AddSession(session *srtp.Session) {
server.AddSession(session)
}
func RemoveSession(session *srtp.Session) {
server.RemoveSession(session)
}
var Server *srtp.Server

View File

@@ -14,6 +14,7 @@ const (
stateMedias
stateTracks
stateStart
stateExternal
)
type Producer struct {
@@ -65,23 +66,23 @@ func (p *Producer) GetTrack(media *streamer.Media, codec *streamer.Codec) *strea
return nil
}
for _, track := range p.tracks {
if track.Codec == codec {
return track
}
}
track := p.element.GetTrack(media, codec)
if track == nil {
return nil
}
for _, t := range p.tracks {
if track == t {
return track
}
}
p.tracks = append(p.tracks, track)
if p.state == stateMedias {
p.state = stateTracks
}
p.tracks = append(p.tracks, track)
return track
}
@@ -157,6 +158,16 @@ func (p *Producer) reconnect() {
func (p *Producer) stop() {
p.mu.Lock()
defer p.mu.Unlock()
switch p.state {
case stateExternal:
log.Debug().Msgf("[streams] can't stop external producer")
return
case stateNone:
log.Debug().Msgf("[streams] can't stop none producer")
return
}
log.Debug().Msgf("[streams] stop producer url=%s", p.url)
@@ -171,6 +182,4 @@ func (p *Producer) stop() {
p.state = stateNone
p.tracks = nil
p.mu.Unlock()
}

View File

@@ -92,6 +92,7 @@ func (s *Stream) AddConsumer(cons streamer.Consumer) (err error) {
}
if len(producers) == 0 {
s.stopProducers()
return errors.New("couldn't find the matching tracks")
}
@@ -110,11 +111,6 @@ func (s *Stream) AddConsumer(cons streamer.Consumer) (err error) {
func (s *Stream) RemoveConsumer(cons streamer.Consumer) {
s.mu.Lock()
for i, consumer := range s.consumers {
if consumer == nil {
log.Warn().Msgf("empty consumer: %+v\n", s)
continue
}
if consumer.element == cons {
// remove consumer pads from all producers
for _, track := range consumer.tracks {
@@ -125,28 +121,13 @@ func (s *Stream) RemoveConsumer(cons streamer.Consumer) {
break
}
}
for _, producer := range s.producers {
if producer == nil {
log.Warn().Msgf("empty producer: %+v\n", s)
continue
}
var sink bool
for _, track := range producer.tracks {
if track.HasSink() {
sink = true
}
}
if !sink {
producer.stop()
}
}
s.mu.Unlock()
s.stopProducers()
}
func (s *Stream) AddProducer(prod streamer.Producer) {
producer := &Producer{element: prod, state: stateTracks}
producer := &Producer{element: prod, state: stateExternal}
s.mu.Lock()
s.producers = append(s.producers, producer)
s.mu.Unlock()
@@ -163,6 +144,20 @@ func (s *Stream) RemoveProducer(prod streamer.Producer) {
s.mu.Unlock()
}
func (s *Stream) stopProducers() {
s.mu.Lock()
producers:
for _, producer := range s.producers {
for _, track := range producer.tracks {
if track.HasSink() {
continue producers
}
}
producer.stop()
}
s.mu.Unlock()
}
//func (s *Stream) Active() bool {
// if len(s.consumers) > 0 {
// return true

View File

@@ -1,58 +0,0 @@
package main
import (
"fmt"
"github.com/AlexxIT/go2rtc/pkg/rtsp"
"github.com/pion/rtp"
"os"
"time"
)
func main() {
client, err := rtsp.NewClient(os.Args[1])
if err != nil {
panic(err)
}
if err = client.Dial(); err != nil {
panic(err)
}
if err = client.Describe(); err != nil {
panic(err)
}
for _, media := range client.GetMedias() {
fmt.Printf("Media: %v\n", media)
if media.AV() {
track := client.GetTrack(media, media.Codecs[0])
fmt.Printf("Track: %v, %v\n", track, track.Codec)
track.Bind(func(packet *rtp.Packet) error {
nalUnitType := packet.Payload[0] & 0x1F
fmt.Printf(
"[RTP] codec: %s, nalu: %2d, size: %6d, ts: %10d, pt: %2d, ssrc: %d\n",
track.Codec.Name, nalUnitType, len(packet.Payload), packet.Timestamp,
packet.PayloadType, packet.SSRC,
)
return nil
})
}
}
if err = client.Play(); err != nil {
panic(err)
}
time.AfterFunc(time.Second*5, func() {
if err = client.Close(); err != nil {
panic(err)
}
})
if err = client.Handle(); err != nil {
panic(err)
}
fmt.Println("The End")
}

4
go.mod
View File

@@ -53,5 +53,7 @@ replace (
// windows support: https://github.com/brutella/dnssd/pull/35
github.com/brutella/dnssd v1.2.2 => github.com/rblenkinsopp/dnssd v1.2.3-0.20220516082132-0923f3c787a1
// RTP tlv8 fix
github.com/brutella/hap v0.0.17 => github.com/AlexxIT/hap v0.0.15-0.20220823033740-ce7d1564e657
github.com/brutella/hap v0.0.17 => github.com/AlexxIT/hap v0.0.15-0.20221108133010-d8a45b7a7045
// fix reading AAC config bytes
github.com/deepch/vdk v0.0.19 => github.com/AlexxIT/vdk v0.0.18-0.20221108193131-6168555b4f92
)

9
go.sum
View File

@@ -1,5 +1,7 @@
github.com/AlexxIT/hap v0.0.15-0.20220823033740-ce7d1564e657 h1:FUzXAJfm6sRLJ8T6vfzvy/Hm3aioX8+fbxgx2VZoI78=
github.com/AlexxIT/hap v0.0.15-0.20220823033740-ce7d1564e657/go.mod h1:c2vEL5pzjRWEx07sa32kTVjzI9bBVlstrwBwKe3DlJ0=
github.com/AlexxIT/hap v0.0.15-0.20221108133010-d8a45b7a7045 h1:xJf3FxQJReJSDyYXJfI1NUWv8tUEAGNV9xigLqNtmrI=
github.com/AlexxIT/hap v0.0.15-0.20221108133010-d8a45b7a7045/go.mod h1:QNA3sm16zE5uUyC8+E/gNkMvQWjqQLuxQKkU5PMi8N4=
github.com/AlexxIT/vdk v0.0.18-0.20221108193131-6168555b4f92 h1:cIeYMGaAirSZnrKRDTb5VgZDDYqPLhYiczElMg4sQW0=
github.com/AlexxIT/vdk v0.0.18-0.20221108193131-6168555b4f92/go.mod h1:7ydHfSkflMZxBXfWR79dMjrT54xzvLxnPaByOa9Jpzg=
github.com/brutella/dnssd v1.2.3 h1:4fBLjZjPH7SbcHhEcIJhZcC9nOhIDZ0m3rn9bjl1/i0=
github.com/brutella/dnssd v1.2.3/go.mod h1:JoW2sJUrmVIef25G6lrLj7HS6Xdwh6q8WUIvMkkBYXs=
github.com/cheekybits/genny v1.0.0/go.mod h1:+tQajlRqAUrPI7DOSpB0XAqZYtQakVtB7wXkRAgjxjQ=
@@ -7,8 +9,6 @@ github.com/coreos/go-systemd/v22 v22.3.3-0.20220203105225-a9a7ef127534/go.mod h1
github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
github.com/deepch/vdk v0.0.19 h1:r6xYyBTtXEIEh+csO0XHT00sI7xLF+hQFkJE9/go5II=
github.com/deepch/vdk v0.0.19/go.mod h1:7ydHfSkflMZxBXfWR79dMjrT54xzvLxnPaByOa9Jpzg=
github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
github.com/fsnotify/fsnotify v1.4.9/go.mod h1:znqG4EE+3YCdAaPaxE2ZRY/06pZUdp0tY4IgpuI1SZQ=
github.com/go-chi/chi v1.5.4 h1:QHdzF2szwjqVV4wmByUnTcsbIg7UGaQ0tPF2t5GcAIs=
@@ -245,6 +245,7 @@ gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7/go.mod h1:dt/ZhP58zS4L8KSrWD
gopkg.in/yaml.v2 v2.2.1/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
gopkg.in/yaml.v2 v2.2.4/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
gopkg.in/yaml.v2 v2.3.0/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ=
gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

View File

@@ -2,4 +2,7 @@
- https://www.wowza.com/blog/streaming-protocols
- https://vimeo.com/blog/post/rtmp-stream/
- https://sanjeev-pandey.medium.com/understanding-the-mpeg-4-moov-atom-pseudo-streaming-in-mp4-93935e1b9e9a
- https://sanjeev-pandey.medium.com/understanding-the-mpeg-4-moov-atom-pseudo-streaming-in-mp4-93935e1b9e9a
- [Android Supported media formats](https://developer.android.com/guide/topics/media/media-formats)
- [THEOplayer](https://www.theoplayer.com/test-your-stream-hls-dash-hesp)
- [How Generate DTS/PTS](https://www.ramugedia.com/how-generate-dts-pts-from-elementary-stream)

20
pkg/aac/README.md Normal file
View File

@@ -0,0 +1,20 @@
## AAC-LD and AAC-ELD
Codec | Rate | QuickTime | ffmpeg | VLC
------|------|-----------|--------|----
AAC-LD | 8000 | yes | no | no
AAC-LD | 16000 | yes | no | no
AAC-LD | 22050 | yes | yes | no
AAC-LD | 24000 | yes | yes | no
AAC-LD | 32000 | yes | yes | no
AAC-ELD | 8000 | yes | no | no
AAC-ELD | 16000 | yes | no | no
AAC-ELD | 22050 | yes | yes | yes
AAC-ELD | 24000 | yes | yes | yes
AAC-ELD | 32000 | yes | yes | yes
## Useful links
- [4.6.20 Enhanced Low Delay Codec](https://csclub.uwaterloo.ca/~ehashman/ISO14496-3-2009.pdf)
- https://stackoverflow.com/questions/40014508/aac-adts-for-aacobject-eld-packets
- https://code.videolan.org/videolan/vlc/-/blob/master/modules/packetizer/mpeg4audio.c

57
pkg/aac/rtp.go Normal file
View File

@@ -0,0 +1,57 @@
package aac
import (
"encoding/binary"
"github.com/AlexxIT/go2rtc/pkg/streamer"
"github.com/pion/rtp"
)
const RTPPacketVersionAAC = 0
func RTPDepay(track *streamer.Track) streamer.WrapperFunc {
return func(push streamer.WriterFunc) streamer.WriterFunc {
return func(packet *rtp.Packet) error {
// support ONLY 2 bytes header size!
// streamtype=5;profile-level-id=1;mode=AAC-hbr;sizelength=13;indexlength=3;indexdeltalength=3;config=1408
headersSize := binary.BigEndian.Uint16(packet.Payload) >> 3
//log.Printf("[RTP/AAC] units: %d, size: %4d, ts: %10d, %t", headersSize/2, len(packet.Payload), packet.Timestamp, packet.Marker)
clone := *packet
clone.Version = RTPPacketVersionAAC
clone.Payload = packet.Payload[2+headersSize:]
return push(&clone)
}
}
}
func RTPPay(mtu uint16) streamer.WrapperFunc {
sequencer := rtp.NewRandomSequencer()
return func(push streamer.WriterFunc) streamer.WriterFunc {
return func(packet *rtp.Packet) error {
if packet.Version != RTPPacketVersionAAC {
return push(packet)
}
// support ONLY one unit in payload
size := uint16(len(packet.Payload))
// 2 bytes header size + 2 bytes first payload size
payload := make([]byte, 2+2+size)
payload[1] = 16 // header size in bits
binary.BigEndian.PutUint16(payload[2:], size<<3)
copy(payload[4:], packet.Payload)
clone := rtp.Packet{
Header: rtp.Header{
Version: 2,
Marker: true,
SequenceNumber: sequencer.NextSequenceNumber(),
Timestamp: packet.Timestamp,
},
Payload: payload,
}
return push(&clone)
}
}
}

View File

@@ -6,12 +6,6 @@ import (
"github.com/pion/rtp"
)
const PayloadTypeAVC = 255
func IsAVC(codec *streamer.Codec) bool {
return codec.PayloadType == PayloadTypeAVC
}
func EncodeAVC(nals ...[]byte) (avc []byte) {
var i, n int

View File

@@ -87,29 +87,28 @@ func RTPPay(mtu uint16) streamer.WrapperFunc {
return func(push streamer.WriterFunc) streamer.WriterFunc {
return func(packet *rtp.Packet) error {
if packet.Version == RTPPacketVersionAVC {
payloads := payloader.Payload(mtu, packet.Payload)
last := len(payloads) - 1
for i, payload := range payloads {
clone := rtp.Packet{
Header: rtp.Header{
Version: 2,
Marker: i == last,
//PayloadType: packet.PayloadType,
SequenceNumber: sequencer.NextSequenceNumber(),
Timestamp: packet.Timestamp,
//SSRC: packet.SSRC,
},
Payload: payload,
}
if err := push(&clone); err != nil {
return err
}
}
return nil
if packet.Version != RTPPacketVersionAVC {
return push(packet)
}
return push(packet)
payloads := payloader.Payload(mtu, packet.Payload)
last := len(payloads) - 1
for i, payload := range payloads {
clone := rtp.Packet{
Header: rtp.Header{
Version: 2,
Marker: i == last,
SequenceNumber: sequencer.NextSequenceNumber(),
Timestamp: packet.Timestamp,
},
Payload: payload,
}
if err := push(&clone); err != nil {
return err
}
}
return nil
}
}
}

View File

@@ -1,4 +1,4 @@
# Homekit
# Home Accessory Protocol
> PS. Character = Characteristic

View File

@@ -1,4 +1,4 @@
package homekit
package hap
type Accessory struct {
AID int `json:"aid"`

View File

@@ -2,22 +2,22 @@ package camera
import (
"errors"
"github.com/AlexxIT/go2rtc/pkg/homekit"
"github.com/AlexxIT/go2rtc/pkg/hap"
"github.com/brutella/hap/characteristic"
"github.com/brutella/hap/rtp"
)
type Client struct {
client *homekit.Client
client *hap.Conn
}
func NewClient(client *homekit.Client) *Client {
func NewClient(client *hap.Conn) *Client {
return &Client{client: client}
}
func (c *Client) StartStream2(ses *Session) (err error) {
func (c *Client) StartStream(ses *Session) (err error) {
// Step 1. Check if camera ready (free) to stream
var srv *homekit.Service
var srv *hap.Service
if srv, err = c.GetFreeStream(); err != nil {
return err
}
@@ -35,8 +35,8 @@ func (c *Client) StartStream2(ses *Session) (err error) {
// GetFreeStream search free streaming service.
// Usual every HomeKit camera can stream only to two clients simultaniosly.
// So it has two similar services for streaming.
func (c *Client) GetFreeStream() (srv *homekit.Service, err error) {
var accs []*homekit.Accessory
func (c *Client) GetFreeStream() (srv *hap.Service, err error) {
var accs []*hap.Accessory
if accs, err = c.client.GetAccessories(); err != nil {
return
}
@@ -60,7 +60,7 @@ func (c *Client) GetFreeStream() (srv *homekit.Service, err error) {
}
func (c *Client) SetupEndpoins(
srv *homekit.Service, req *rtp.SetupEndpoints,
srv *hap.Service, req *rtp.SetupEndpoints,
) (res *rtp.SetupEndpointsResponse, err error) {
// get setup endpoint character ID
char := srv.GetCharacter(characteristic.TypeSetupEndpoints)
@@ -87,7 +87,7 @@ func (c *Client) SetupEndpoins(
return
}
func (c *Client) SetConfig(srv *homekit.Service, config *rtp.StreamConfiguration) (err error) {
func (c *Client) SetConfig(srv *hap.Service, config *rtp.StreamConfiguration) (err error) {
// get setup endpoint character ID
char := srv.GetCharacter(characteristic.TypeSelectedStreamConfiguration)
char.Event = nil

75
pkg/hap/camera/session.go Normal file
View File

@@ -0,0 +1,75 @@
package camera
import (
cryptorand "crypto/rand"
"encoding/binary"
"github.com/brutella/hap/rtp"
)
type Session struct {
Offer *rtp.SetupEndpoints
Answer *rtp.SetupEndpointsResponse
Config *rtp.StreamConfiguration
}
func NewSession(vp *rtp.VideoParameters, ap *rtp.AudioParameters) *Session {
vp.RTP = rtp.RTPParams{
PayloadType: 99,
Ssrc: RandomUint32(),
Bitrate: 2048,
Interval: 10,
MTU: 1200, // like WebRTC
}
ap.RTP = rtp.RTPParams{
PayloadType: 110,
Ssrc: RandomUint32(),
Bitrate: 32,
Interval: 10,
ComfortNoisePayloadType: 98,
MTU: 0,
}
sessionID := RandomBytes(16)
s := &Session{
Offer: &rtp.SetupEndpoints{
SessionId: sessionID,
Video: rtp.CryptoSuite{
MasterKey: RandomBytes(16),
MasterSalt: RandomBytes(14),
},
Audio: rtp.CryptoSuite{
MasterKey: RandomBytes(16),
MasterSalt: RandomBytes(14),
},
},
Config: &rtp.StreamConfiguration{
Command: rtp.SessionControlCommand{
Identifier: sessionID,
Type: rtp.SessionControlCommandTypeStart,
},
Video: *vp,
Audio: *ap,
},
}
return s
}
func (s *Session) SetLocalEndpoint(host string, port uint16) {
s.Offer.ControllerAddr = rtp.Addr{
IPAddr: host,
VideoRtpPort: port,
AudioRtpPort: port,
}
}
func RandomBytes(size int) []byte {
data := make([]byte, size)
_, _ = cryptorand.Read(data)
return data
}
func RandomUint32() uint32 {
data := make([]byte, 4)
_, _ = cryptorand.Read(data)
return binary.BigEndian.Uint32(data)
}

View File

@@ -1,4 +1,4 @@
package homekit
package hap
import (
"bytes"

733
pkg/hap/conn.go Normal file
View File

@@ -0,0 +1,733 @@
package hap
import (
"bufio"
"crypto/sha512"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"github.com/AlexxIT/go2rtc/pkg/hap/mdns"
"github.com/AlexxIT/go2rtc/pkg/streamer"
"github.com/brutella/hap"
"github.com/brutella/hap/chacha20poly1305"
"github.com/brutella/hap/curve25519"
"github.com/brutella/hap/ed25519"
"github.com/brutella/hap/hkdf"
"github.com/brutella/hap/tlv8"
"github.com/tadglines/go-pkgs/crypto/srp"
"io"
"net"
"net/http"
"net/url"
"strings"
"time"
)
// Conn for HomeKit. DevicePublic can be null.
type Conn struct {
streamer.Element
DeviceAddress string // including port
DeviceID string
DevicePublic []byte
ClientID string
ClientPrivate []byte
OnEvent func(res *http.Response)
Output func(msg interface{})
conn net.Conn
secure *Secure
httpResponse chan *bufio.Reader
}
func NewConn(rawURL string) (*Conn, error) {
u, err := url.Parse(rawURL)
if err != nil {
return nil, err
}
query := u.Query()
c := &Conn{
DeviceAddress: u.Host,
DeviceID: query.Get("device_id"),
DevicePublic: DecodeKey(query.Get("device_public")),
ClientID: query.Get("client_id"),
ClientPrivate: DecodeKey(query.Get("client_private")),
}
return c, nil
}
func Pair(deviceID, pin string) (*Conn, error) {
entry := mdns.GetEntry(deviceID)
if entry == nil {
return nil, errors.New("can't find device via mDNS")
}
c := &Conn{
DeviceAddress: fmt.Sprintf("%s:%d", entry.AddrV4.String(), entry.Port),
DeviceID: deviceID,
ClientID: GenerateUUID(),
ClientPrivate: GenerateKey(),
}
var mfi bool
for _, field := range entry.InfoFields {
if field[:2] == "ff" {
if field[3] == '1' {
mfi = true
}
break
}
}
return c, c.Pair(mfi, pin)
}
func (c *Conn) ClientPublic() []byte {
return c.ClientPrivate[32:]
}
func (c *Conn) URL() string {
return fmt.Sprintf(
"homekit://%s?device_id=%s&device_public=%16x&client_id=%s&client_private=%32x",
c.DeviceAddress, c.DeviceID, c.DevicePublic, c.ClientID, c.ClientPrivate,
)
}
func (c *Conn) DialAndServe() error {
if err := c.Dial(); err != nil {
return err
}
return c.Handle()
}
func (c *Conn) Dial() error {
// update device host before dial
if host := mdns.GetAddress(c.DeviceID); host != "" {
c.DeviceAddress = host
}
var err error
c.conn, err = net.DialTimeout("tcp", c.DeviceAddress, time.Second*5)
if err != nil {
return err
}
// STEP M1: send our session public to device
sessionPublic, sessionPrivate := curve25519.GenerateKeyPair()
// 1. generate payload
// important not include other fields
requestM1 := struct {
State byte `tlv8:"6"`
PublicKey []byte `tlv8:"3"`
}{
State: hap.M1,
PublicKey: sessionPublic[:],
}
// 2. pack payload to TLV8
buf, err := tlv8.Marshal(requestM1)
if err != nil {
return err
}
// 3. send request
resp, err := c.Post(UriPairVerify, buf)
if err != nil {
return err
}
// STEP M2: unpack deviceID from response
responseM2 := PairVerifyPayload{}
if err = tlv8.UnmarshalReader(resp.Body, &responseM2); err != nil {
return err
}
// 1. generate session shared key
var deviceSessionPublic [32]byte
copy(deviceSessionPublic[:], responseM2.PublicKey)
sessionShared := curve25519.SharedSecret(sessionPrivate, deviceSessionPublic)
sessionKey, err := hkdf.Sha512(
sessionShared[:], []byte("Pair-Verify-Encrypt-Salt"),
[]byte("Pair-Verify-Encrypt-Info"),
)
// 2. decrypt M2 response with session key
msg := responseM2.EncryptedData[:len(responseM2.EncryptedData)-16]
var mac [16]byte
copy(mac[:], responseM2.EncryptedData[len(msg):]) // 16 byte (MAC)
buf, err = chacha20poly1305.DecryptAndVerify(
sessionKey[:], []byte("PV-Msg02"), msg, mac, nil,
)
// 3. unpack payload from TLV8
payloadM2 := PairVerifyPayload{}
if err = tlv8.Unmarshal(buf, &payloadM2); err != nil {
return err
}
// 4. verify signature for M2 response with device public
// device session + device id + our session
if c.DevicePublic != nil {
buf = nil
buf = append(buf, responseM2.PublicKey[:]...)
buf = append(buf, []byte(payloadM2.Identifier)...)
buf = append(buf, sessionPublic[:]...)
if !ed25519.ValidateSignature(
c.DevicePublic[:], buf, payloadM2.Signature,
) {
return errors.New("device public signature invalid")
}
}
// STEP M3: send our clientID to device
// 1. generate signature with our private key
// (our session + our ID + device session)
buf = nil
buf = append(buf, sessionPublic[:]...)
buf = append(buf, []byte(c.ClientID)...)
buf = append(buf, responseM2.PublicKey[:]...)
signature, err := ed25519.Signature(c.ClientPrivate[:], buf)
if err != nil {
return err
}
// 2. generate payload
payloadM3 := struct {
Identifier string `tlv8:"1"`
Signature []byte `tlv8:"10"`
}{
Identifier: c.ClientID,
Signature: signature,
}
// 3. pack payload to TLV8
buf, err = tlv8.Marshal(payloadM3)
if err != nil {
return err
}
// 4. encrypt payload with session key
msg, mac, _ = chacha20poly1305.EncryptAndSeal(
sessionKey[:], []byte("PV-Msg03"), buf, nil,
)
// 4. generate request
requestM3 := struct {
EncryptedData []byte `tlv8:"5"`
State byte `tlv8:"6"`
}{
State: hap.M3,
EncryptedData: append(msg, mac[:]...),
}
// 5. pack payload to TLV8
buf, err = tlv8.Marshal(requestM3)
if err != nil {
return err
}
resp, err = c.Post(UriPairVerify, buf)
if err != nil {
return err
}
// STEP M4. Read response
responseM4 := PairVerifyPayload{}
if err = tlv8.UnmarshalReader(resp.Body, &responseM4); err != nil {
return err
}
// 1. check response state
if responseM4.State != 4 || responseM4.Status != 0 {
return fmt.Errorf("wrong M4 response: %+v", responseM4)
}
c.secure, err = NewSecure(sessionShared, false)
//c.secure.Buffer = bytes.NewBuffer(nil)
c.secure.Conn = c.conn
c.httpResponse = make(chan *bufio.Reader, 10)
return err
}
// https://github.com/apple/HomeKitADK/blob/master/HAP/HAPPairingPairSetup.c
func (c *Conn) Pair(mfi bool, pin string) (err error) {
pin = strings.ReplaceAll(pin, "-", "")
if len(pin) != 8 {
return fmt.Errorf("wrong PIN format: %s", pin)
}
pin = pin[:3] + "-" + pin[3:5] + "-" + pin[5:]
c.conn, err = net.Dial("tcp", c.DeviceAddress)
if err != nil {
return
}
// STEP M1. Generate request
reqM1 := struct {
Method byte `tlv8:"0"`
State byte `tlv8:"6"`
}{
State: hap.M1,
}
if mfi {
reqM1.Method = 1 // ff=1 => method=1, ff=2 => method=0
}
buf, err := tlv8.Marshal(reqM1)
if err != nil {
return
}
// STEP M1. Send request
res, err := c.Post(UriPairSetup, buf)
if err != nil {
return
}
// STEP M2. Read response
resM2 := struct {
Salt []byte `tlv8:"2"`
PublicKey []byte `tlv8:"3"` // server public key, aka session.B
State byte `tlv8:"6"`
Error byte `tlv8:"7"`
}{}
if err = tlv8.UnmarshalReader(res.Body, &resM2); err != nil {
return
}
if resM2.State != 2 || resM2.Error > 0 {
return fmt.Errorf("wrong M2: %+v", resM2)
}
// STEP M3. Generate session using pin
username := []byte("Pair-Setup")
SRP, err := srp.NewSRP(
"rfc5054.3072", sha512.New, keyDerivativeFuncRFC2945(username),
)
if err != nil {
return
}
SRP.SaltLength = 16
// username: "Pair-Setup"
// password: PIN (with dashes)
session := SRP.NewClientSession(username, []byte(pin))
sessionShared, err := session.ComputeKey(resM2.Salt, resM2.PublicKey)
if err != nil {
return
}
// STEP M3. Generate request
reqM3 := struct {
PublicKey []byte `tlv8:"3"`
Proof []byte `tlv8:"4"`
State byte `tlv8:"6"`
}{
PublicKey: session.GetA(), // client public key, aka session.A
Proof: session.ComputeAuthenticator(),
State: hap.M3,
}
buf, err = tlv8.Marshal(reqM3)
if err != nil {
return err
}
// STEP M3. Send request
res, err = c.Post(UriPairSetup, buf)
if err != nil {
return
}
// STEP M4. Read response
resM4 := struct {
Proof []byte `tlv8:"4"` // server proof
State byte `tlv8:"6"`
Error byte `tlv8:"7"`
}{}
if err = tlv8.UnmarshalReader(res.Body, &resM4); err != nil {
return
}
if resM4.Error == 2 {
return fmt.Errorf("wrong PIN: %s", pin)
}
if resM4.State != 4 || resM4.Error > 0 {
return fmt.Errorf("wrong M4: %+v", resM4)
}
// STEP M4. Verify response
if !session.VerifyServerAuthenticator(resM4.Proof) {
return errors.New("verify server auth fail")
}
// STEP M5. Generate signature
saltKey, err := hkdf.Sha512(
sessionShared, []byte("Pair-Setup-Controller-Sign-Salt"),
[]byte("Pair-Setup-Controller-Sign-Info"),
)
if err != nil {
return
}
buf = nil
buf = append(buf, saltKey[:]...)
buf = append(buf, []byte(c.ClientID)...)
buf = append(buf, c.ClientPublic()...)
signature, err := ed25519.Signature(c.ClientPrivate, buf)
if err != nil {
return
}
// STEP M5. Generate payload
msgM5 := struct {
Identifier string `tlv8:"1"`
PublicKey []byte `tlv8:"3"`
Signature []byte `tlv8:"10"`
}{
Identifier: c.ClientID,
PublicKey: c.ClientPublic(),
Signature: signature,
}
buf, err = tlv8.Marshal(msgM5)
if err != nil {
return
}
// STEP M5. Encrypt payload
sessionKey, err := hkdf.Sha512(
sessionShared, []byte("Pair-Setup-Encrypt-Salt"),
[]byte("Pair-Setup-Encrypt-Info"),
)
if err != nil {
return
}
buf, mac, _ := chacha20poly1305.EncryptAndSeal(
sessionKey[:], []byte("PS-Msg05"), buf, nil,
)
// STEP M5. Generate request
reqM5 := struct {
EncryptedData []byte `tlv8:"5"`
State byte `tlv8:"6"`
}{
EncryptedData: append(buf, mac[:]...),
State: hap.M5,
}
buf, err = tlv8.Marshal(reqM5)
if err != nil {
return err
}
// STEP M5. Send request
res, err = c.Post(UriPairSetup, buf)
if err != nil {
return
}
// STEP M6. Read response
resM6 := struct {
EncryptedData []byte `tlv8:"5"`
State byte `tlv8:"6"`
Error byte `tlv8:"7"`
}{}
if err = tlv8.UnmarshalReader(res.Body, &resM6); err != nil {
return
}
if resM6.State != 6 || resM6.Error > 0 {
return fmt.Errorf("wrong M6: %+v", resM2)
}
// STEP M6. Decrypt payload
msg := resM6.EncryptedData[:len(resM6.EncryptedData)-16]
copy(mac[:], resM6.EncryptedData[len(msg):]) // 16 byte (MAC)
buf, err = chacha20poly1305.DecryptAndVerify(
sessionKey[:], []byte("PS-Msg06"), msg, mac, nil,
)
if err != nil {
return
}
msgM6 := struct {
Identifier []byte `tlv8:"1"`
PublicKey []byte `tlv8:"3"`
Signature []byte `tlv8:"10"`
}{}
if err = tlv8.Unmarshal(buf, &msgM6); err != nil {
return
}
// STEP M6. Verify payload
if saltKey, err = hkdf.Sha512(
sessionShared, []byte("Pair-Setup-Accessory-Sign-Salt"),
[]byte("Pair-Setup-Accessory-Sign-Info"),
); err != nil {
return
}
buf = nil
buf = append(buf, saltKey[:]...)
buf = append(buf, msgM6.Identifier...)
buf = append(buf, msgM6.PublicKey...)
if !ed25519.ValidateSignature(
msgM6.PublicKey[:], buf, msgM6.Signature,
) {
return errors.New("wrong server signature")
}
if c.DeviceID != string(msgM6.Identifier) {
return fmt.Errorf("wrong Device ID: %s", msgM6.Identifier)
}
c.DevicePublic = msgM6.PublicKey
return nil
}
func (c *Conn) Close() error {
if c.conn == nil {
return nil
}
conn := c.conn
c.conn = nil
return conn.Close()
}
func (c *Conn) GetAccessories() ([]*Accessory, error) {
res, err := c.Get("/accessories")
if err != nil {
return nil, err
}
data, err := io.ReadAll(res.Body)
if err != nil {
return nil, err
}
p := Accessories{}
if err = json.Unmarshal(data, &p); err != nil {
return nil, err
}
for _, accs := range p.Accessories {
for _, serv := range accs.Services {
for _, char := range serv.Characters {
char.AID = accs.AID
}
}
}
return p.Accessories, nil
}
func (c *Conn) GetCharacters(query string) ([]*Character, error) {
res, err := c.Get("/characteristics?id=" + query)
if err != nil {
return nil, err
}
data, err := io.ReadAll(res.Body)
if err != nil {
return nil, err
}
ch := Characters{}
if err = json.Unmarshal(data, &ch); err != nil {
return nil, err
}
return ch.Characters, nil
}
func (c *Conn) GetCharacter(char *Character) error {
query := fmt.Sprintf("%d.%d", char.AID, char.IID)
chars, err := c.GetCharacters(query)
if err != nil {
return err
}
char.Value = chars[0].Value
return nil
}
func (c *Conn) PutCharacters(characters ...*Character) (err error) {
for i, char := range characters {
if char.Event != nil {
char = &Character{AID: char.AID, IID: char.IID, Event: char.Event}
} else {
char = &Character{AID: char.AID, IID: char.IID, Value: char.Value}
}
characters[i] = char
}
var data []byte
if data, err = json.Marshal(Characters{characters}); err != nil {
return
}
var res *http.Response
if res, err = c.Put("/characteristics", data); err != nil {
return
}
if res.StatusCode >= 400 {
return errors.New("wrong response status")
}
return
}
func (c *Conn) GetImage(width, height int) ([]byte, error) {
res, err := c.Post(
"/resource", []byte(fmt.Sprintf(
`{"image-width":%d,"image-height":%d,"resource-type":"image","reason":0}`,
width, height,
)),
)
if err != nil {
return nil, err
}
return io.ReadAll(res.Body)
}
//func (c *Client) onEventData(r io.Reader) error {
// if c.OnEvent == nil {
// return nil
// }
//
// data, err := io.ReadAll(r)
//
// ch := Characters{}
// if err = json.Unmarshal(data, &ch); err != nil {
// return err
// }
//
// c.OnEvent(ch.Characters)
//
// return nil
//}
func (c *Conn) ListPairings() error {
pReq := struct {
Method byte `tlv8:"0"`
State byte `tlv8:"6"`
}{
Method: hap.MethodListPairings,
State: hap.M1,
}
data, err := tlv8.Marshal(pReq)
if err != nil {
return err
}
res, err := c.Post("/pairings", data)
if err != nil {
return err
}
data, err = io.ReadAll(res.Body)
// TODO: don't know how to fix array of items
var pRes struct {
State byte `tlv8:"6"`
Identifier string `tlv8:"1"`
PublicKey []byte `tlv8:"3"`
Permission byte `tlv8:"11"`
}
if err = tlv8.Unmarshal(data, &pRes); err != nil {
return err
}
return nil
}
func (c *Conn) PairingsAdd(clientID string, clientPublic []byte, admin bool) error {
pReq := struct {
Method byte `tlv8:"0"`
Identifier string `tlv8:"1"`
PublicKey []byte `tlv8:"3"`
State byte `tlv8:"6"`
Permission byte `tlv8:"11"`
}{
Method: hap.MethodAddPairing,
Identifier: clientID,
PublicKey: clientPublic,
State: hap.M1,
Permission: hap.PermissionUser,
}
if admin {
pReq.Permission = hap.PermissionAdmin
}
data, err := tlv8.Marshal(pReq)
if err != nil {
return err
}
res, err := c.Post("/pairings", data)
if err != nil {
return err
}
data, err = io.ReadAll(res.Body)
var pRes struct {
State byte `tlv8:"6"`
Unknown byte `tlv8:"7"`
}
if err = tlv8.Unmarshal(data, &pRes); err != nil {
return err
}
return nil
}
func (c *Conn) DeletePairing(id string) error {
reqM1 := struct {
State byte `tlv8:"6"`
Method byte `tlv8:"0"`
Identifier string `tlv8:"1"`
}{
State: hap.M1,
Method: hap.MethodDeletePairing,
Identifier: id,
}
data, err := tlv8.Marshal(reqM1)
if err != nil {
return err
}
res, err := c.Post("/pairings", data)
if err != nil {
return err
}
data, err = io.ReadAll(res.Body)
var resM2 struct {
State byte `tlv8:"6"`
}
if err = tlv8.Unmarshal(data, &resM2); err != nil {
return err
}
if resM2.State != hap.M2 {
return errors.New("wrong state")
}
return nil
}
func (c *Conn) LocalAddr() string {
return c.conn.LocalAddr().String()
}
func DecodeKey(s string) []byte {
if s == "" {
return nil
}
data, err := hex.DecodeString(s)
if err != nil {
return nil
}
return data
}

View File

@@ -1,4 +1,4 @@
package homekit
package hap
import (
"crypto/rand"
@@ -29,13 +29,13 @@ func GenerateUUID() string {
}
type PairVerifyPayload struct {
Method byte `tlv8:"0"`
Identifier string `tlv8:"1"`
PublicKey []byte `tlv8:"3"`
EncryptedData []byte `tlv8:"5"`
State byte `tlv8:"6"`
Status byte `tlv8:"7"`
Signature []byte `tlv8:"10"`
Method byte `tlv8:"0,optional"`
Identifier string `tlv8:"1,optional"`
PublicKey []byte `tlv8:"3,optional"`
EncryptedData []byte `tlv8:"5,optional"`
State byte `tlv8:"6,optional"`
Status byte `tlv8:"7,optional"`
Signature []byte `tlv8:"10,optional"`
}
//func (c *Character) Unmarshal(value interface{}) error {

View File

@@ -1,4 +1,4 @@
package homekit
package hap
import (
"bufio"
@@ -23,7 +23,7 @@ const (
UriResource = "/resource"
)
func (c *Client) Write(p []byte) (r io.Reader, err error) {
func (c *Conn) Write(p []byte) (r io.Reader, err error) {
if c.secure == nil {
if _, err = c.conn.Write(p); err == nil {
r = bufio.NewReader(c.conn)
@@ -36,7 +36,7 @@ func (c *Client) Write(p []byte) (r io.Reader, err error) {
return
}
func (c *Client) Do(req *http.Request) (*http.Response, error) {
func (c *Conn) Do(req *http.Request) (*http.Response, error) {
if c.secure == nil {
// insecure requests
if err := req.Write(c.conn); err != nil {
@@ -56,7 +56,7 @@ func (c *Client) Do(req *http.Request) (*http.Response, error) {
return http.ReadResponse(buf, req)
}
func (c *Client) Get(uri string) (*http.Response, error) {
func (c *Conn) Get(uri string) (*http.Response, error) {
req, err := http.NewRequest(
"GET", "http://"+c.DeviceAddress+uri, nil,
)
@@ -66,7 +66,7 @@ func (c *Client) Get(uri string) (*http.Response, error) {
return c.Do(req)
}
func (c *Client) Post(uri string, data []byte) (*http.Response, error) {
func (c *Conn) Post(uri string, data []byte) (*http.Response, error) {
req, err := http.NewRequest(
"POST", "http://"+c.DeviceAddress+uri,
bytes.NewReader(data),
@@ -85,7 +85,7 @@ func (c *Client) Post(uri string, data []byte) (*http.Response, error) {
return c.Do(req)
}
func (c *Client) Put(uri string, data []byte) (*http.Response, error) {
func (c *Conn) Put(uri string, data []byte) (*http.Response, error) {
req, err := http.NewRequest(
"PUT", "http://"+c.DeviceAddress+uri,
bytes.NewReader(data),
@@ -102,7 +102,7 @@ func (c *Client) Put(uri string, data []byte) (*http.Response, error) {
return c.Do(req)
}
func (c *Client) Handle() (err error) {
func (c *Conn) Handle() (err error) {
defer func() {
if c.conn == nil {
err = nil

View File

@@ -1,4 +1,4 @@
package homekit
package hap
import (
"bufio"

View File

@@ -1,4 +1,4 @@
package homekit
package hap
import (
"encoding/binary"

View File

@@ -1,4 +1,4 @@
package homekit
package hap
import (
"bufio"

View File

@@ -1,103 +0,0 @@
package camera
import (
cryptorand "crypto/rand"
"encoding/binary"
"github.com/brutella/hap/rtp"
)
type Session struct {
Offer *rtp.SetupEndpoints
Answer *rtp.SetupEndpointsResponse
Config *rtp.StreamConfiguration
}
func NewSession() *Session {
sessionID := RandomBytes(16)
s := &Session{
Offer: &rtp.SetupEndpoints{
SessionId: sessionID,
Video: rtp.CryptoSuite{
MasterKey: RandomBytes(16),
MasterSalt: RandomBytes(14),
},
Audio: rtp.CryptoSuite{
MasterKey: RandomBytes(16),
MasterSalt: RandomBytes(14),
},
},
Config: &rtp.StreamConfiguration{
Command: rtp.SessionControlCommand{
Identifier: sessionID,
Type: rtp.SessionControlCommandTypeStart,
},
Video: rtp.VideoParameters{
CodecType: rtp.VideoCodecType_H264,
CodecParams: rtp.VideoCodecParameters{
Profiles: []rtp.VideoCodecProfile{
{Id: rtp.VideoCodecProfileMain},
},
Levels: []rtp.VideoCodecLevel{
{Level: rtp.VideoCodecLevel4},
},
Packetizations: []rtp.VideoCodecPacketization{
{Mode: rtp.VideoCodecPacketizationModeNonInterleaved},
},
},
Attributes: rtp.VideoCodecAttributes{
Width: 1920, Height: 1080, Framerate: 30,
},
RTP: rtp.RTPParams{
PayloadType: 99,
Ssrc: RandomUint32(),
Bitrate: 299,
Interval: 0.5,
ComfortNoisePayloadType: 98,
MTU: 0,
},
},
Audio: rtp.AudioParameters{
CodecType: rtp.AudioCodecType_AAC_ELD,
CodecParams: rtp.AudioCodecParameters{
Channels: 1,
Bitrate: rtp.AudioCodecBitrateVariable,
Samplerate: rtp.AudioCodecSampleRate16Khz,
PacketTime: 30,
},
RTP: rtp.RTPParams{
PayloadType: 110,
Ssrc: RandomUint32(),
Bitrate: 24,
Interval: 5,
MTU: 13,
},
ComfortNoise: false,
},
},
}
return s
}
func (s *Session) SetLocalEndpoint(host string, port uint16) {
s.Offer.ControllerAddr = rtp.Addr{
IPAddr: host,
VideoRtpPort: port,
AudioRtpPort: port,
}
}
func (s *Session) SetVideo() {
}
func RandomBytes(size int) []byte {
data := make([]byte, size)
_, _ = cryptorand.Read(data)
return data
}
func RandomUint32() uint32 {
data := make([]byte, 4)
_, _ = cryptorand.Read(data)
return binary.BigEndian.Uint32(data)
}

View File

@@ -1,732 +1,265 @@
package homekit
import (
"bufio"
"crypto/sha512"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"github.com/AlexxIT/go2rtc/pkg/homekit/mdns"
"github.com/AlexxIT/go2rtc/pkg/hap"
"github.com/AlexxIT/go2rtc/pkg/hap/camera"
"github.com/AlexxIT/go2rtc/pkg/srtp"
"github.com/AlexxIT/go2rtc/pkg/streamer"
"github.com/brutella/hap"
"github.com/brutella/hap/chacha20poly1305"
"github.com/brutella/hap/curve25519"
"github.com/brutella/hap/ed25519"
"github.com/brutella/hap/hkdf"
"github.com/brutella/hap/tlv8"
"github.com/tadglines/go-pkgs/crypto/srp"
"io"
"github.com/brutella/hap/characteristic"
"github.com/brutella/hap/rtp"
"net"
"net/http"
"net/url"
"strings"
)
// Client for HomeKit. DevicePublic can be null.
type Client struct {
streamer.Element
DeviceAddress string // including port
DeviceID string
DevicePublic []byte
ClientID string
ClientPrivate []byte
conn *hap.Conn
exit chan error
server *srtp.Server
url string
OnEvent func(res *http.Response)
Output func(msg interface{})
medias []*streamer.Media
tracks []*streamer.Track
conn net.Conn
secure *Secure
httpResponse chan *bufio.Reader
sessions []*srtp.Session
}
func NewClient(rawURL string) (*Client, error) {
func NewClient(rawURL string, server *srtp.Server) (*Client, error) {
u, err := url.Parse(rawURL)
if err != nil {
return nil, err
}
query := u.Query()
c := &Client{
c := &hap.Conn{
DeviceAddress: u.Host,
DeviceID: query.Get("device_id"),
DevicePublic: DecodeKey(query.Get("device_public")),
DevicePublic: hap.DecodeKey(query.Get("device_public")),
ClientID: query.Get("client_id"),
ClientPrivate: DecodeKey(query.Get("client_private")),
ClientPrivate: hap.DecodeKey(query.Get("client_private")),
}
return c, nil
}
func Pair(deviceID, pin string) (*Client, error) {
entry := mdns.GetEntry(deviceID)
if entry == nil {
return nil, errors.New("can't find device via mDNS")
}
c := &Client{
DeviceAddress: fmt.Sprintf("%s:%d", entry.AddrV4.String(), entry.Port),
DeviceID: deviceID,
ClientID: GenerateUUID(),
ClientPrivate: GenerateKey(),
}
var mfi bool
for _, field := range entry.InfoFields {
if field[:2] == "ff" {
if field[3] == '1' {
mfi = true
}
break
}
}
return c, c.Pair(mfi, pin)
}
func (c *Client) ClientPublic() []byte {
return c.ClientPrivate[32:]
}
func (c *Client) URL() string {
return fmt.Sprintf(
"homekit://%s?device_id=%s&device_public=%16x&client_id=%s&client_private=%32x",
c.DeviceAddress, c.DeviceID, c.DevicePublic, c.ClientID, c.ClientPrivate,
)
}
func (c *Client) DialAndServe() error {
if err := c.Dial(); err != nil {
return err
}
return c.Handle()
return &Client{conn: c, server: server}, nil
}
func (c *Client) Dial() error {
// update device host before dial
if host := mdns.GetAddress(c.DeviceID); host != "" {
c.DeviceAddress = host
}
var err error
c.conn, err = net.Dial("tcp", c.DeviceAddress)
if err != nil {
if err := c.conn.Dial(); err != nil {
return err
}
// STEP M1: send our session public to device
sessionPublic, sessionPrivate := curve25519.GenerateKeyPair()
c.exit = make(chan error)
// 1. generate payload
// important not include other fields
requestM1 := struct {
State byte `tlv8:"6"`
PublicKey []byte `tlv8:"3"`
}{
State: hap.M1,
PublicKey: sessionPublic[:],
}
// 2. pack payload to TLV8
buf, err := tlv8.Marshal(requestM1)
if err != nil {
return err
go func() {
//start goroutine for reading responses from camera
c.exit <- c.conn.Handle()
}()
return nil
}
func (c *Client) GetMedias() []*streamer.Media {
if c.medias == nil {
c.medias = c.getMedias()
}
// 3. send request
resp, err := c.Post(UriPairVerify, buf)
if err != nil {
return err
}
return c.medias
}
// STEP M2: unpack deviceID from response
responseM2 := PairVerifyPayload{}
if err = tlv8.UnmarshalReader(resp.Body, &responseM2); err != nil {
return err
}
// 1. generate session shared key
var deviceSessionPublic [32]byte
copy(deviceSessionPublic[:], responseM2.PublicKey)
sessionShared := curve25519.SharedSecret(sessionPrivate, deviceSessionPublic)
sessionKey, err := hkdf.Sha512(
sessionShared[:], []byte("Pair-Verify-Encrypt-Salt"),
[]byte("Pair-Verify-Encrypt-Info"),
)
// 2. decrypt M2 response with session key
msg := responseM2.EncryptedData[:len(responseM2.EncryptedData)-16]
var mac [16]byte
copy(mac[:], responseM2.EncryptedData[len(msg):]) // 16 byte (MAC)
buf, err = chacha20poly1305.DecryptAndVerify(
sessionKey[:], []byte("PV-Msg02"), msg, mac, nil,
)
// 3. unpack payload from TLV8
payloadM2 := PairVerifyPayload{}
if err = tlv8.Unmarshal(buf, &payloadM2); err != nil {
return err
}
// 4. verify signature for M2 response with device public
// device session + device id + our session
if c.DevicePublic != nil {
buf = nil
buf = append(buf, responseM2.PublicKey[:]...)
buf = append(buf, []byte(payloadM2.Identifier)...)
buf = append(buf, sessionPublic[:]...)
if !ed25519.ValidateSignature(
c.DevicePublic[:], buf, payloadM2.Signature,
) {
return errors.New("device public signature invalid")
func (c *Client) GetTrack(media *streamer.Media, codec *streamer.Codec) *streamer.Track {
for _, track := range c.tracks {
if track.Codec == codec {
return track
}
}
// STEP M3: send our clientID to device
// 1. generate signature with our private key
// (our session + our ID + device session)
buf = nil
buf = append(buf, sessionPublic[:]...)
buf = append(buf, []byte(c.ClientID)...)
buf = append(buf, responseM2.PublicKey[:]...)
signature, err := ed25519.Signature(c.ClientPrivate[:], buf)
track := &streamer.Track{Codec: codec, Direction: media.Direction}
c.tracks = append(c.tracks, track)
return track
}
func (c *Client) Start() error {
if c.tracks == nil {
return errors.New("producer without tracks")
}
// get our server local IP-address
host, _, err := net.SplitHostPort(c.conn.LocalAddr())
if err != nil {
return err
}
// 2. generate payload
payloadM3 := struct {
Identifier string `tlv8:"1"`
Signature []byte `tlv8:"10"`
}{
Identifier: c.ClientID,
Signature: signature,
// TODO: set right config
vp := &rtp.VideoParameters{
CodecType: rtp.VideoCodecType_H264,
CodecParams: rtp.VideoCodecParameters{
Profiles: []rtp.VideoCodecProfile{
{Id: rtp.VideoCodecProfileMain},
},
Levels: []rtp.VideoCodecLevel{
{Level: rtp.VideoCodecLevel4},
},
Packetizations: []rtp.VideoCodecPacketization{
{Mode: rtp.VideoCodecPacketizationModeNonInterleaved},
},
},
Attributes: rtp.VideoCodecAttributes{
Width: 1920, Height: 1080, Framerate: 30,
},
}
// 3. pack payload to TLV8
buf, err = tlv8.Marshal(payloadM3)
if err != nil {
ap := &rtp.AudioParameters{
CodecType: rtp.AudioCodecType_AAC_ELD,
CodecParams: rtp.AudioCodecParameters{
Channels: 1,
Bitrate: rtp.AudioCodecBitrateVariable,
Samplerate: rtp.AudioCodecSampleRate16Khz,
// packet time=20 => AAC-ELD packet size=480
// packet time=30 => AAC-ELD packet size=480
// packet time=40 => AAC-ELD packet size=480
// packet time=60 => AAC-LD packet size=960
PacketTime: 40,
},
}
// setup HomeKit stream session
hkSession := camera.NewSession(vp, ap)
hkSession.SetLocalEndpoint(host, c.server.Port())
// create client for processing camera accessory
cam := camera.NewClient(c.conn)
// try to start HomeKit stream
if err = cam.StartStream(hkSession); err != nil {
return err
}
// 4. encrypt payload with session key
msg, mac, _ = chacha20poly1305.EncryptAndSeal(
sessionKey[:], []byte("PV-Msg03"), buf, nil,
)
// 4. generate request
requestM3 := struct {
EncryptedData []byte `tlv8:"5"`
State byte `tlv8:"6"`
}{
State: hap.M3,
EncryptedData: append(msg, mac[:]...),
// SRTP Video Session
vs := &srtp.Session{
LocalSSRC: hkSession.Config.Video.RTP.Ssrc,
RemoteSSRC: hkSession.Answer.SsrcVideo,
}
// 5. pack payload to TLV8
buf, err = tlv8.Marshal(requestM3)
if err != nil {
if err = vs.SetKeys(
hkSession.Offer.Video.MasterKey, hkSession.Offer.Video.MasterSalt,
hkSession.Answer.Video.MasterKey, hkSession.Answer.Video.MasterSalt,
); err != nil {
return err
}
resp, err = c.Post(UriPairVerify, buf)
if err != nil {
// SRTP Audio Session
as := &srtp.Session{
LocalSSRC: hkSession.Config.Audio.RTP.Ssrc,
RemoteSSRC: hkSession.Answer.SsrcAudio,
}
if err = as.SetKeys(
hkSession.Offer.Audio.MasterKey, hkSession.Offer.Audio.MasterSalt,
hkSession.Answer.Audio.MasterKey, hkSession.Answer.Audio.MasterSalt,
); err != nil {
return err
}
// STEP M4. Read response
responseM4 := PairVerifyPayload{}
if err = tlv8.UnmarshalReader(resp.Body, &responseM4); err != nil {
return err
for _, track := range c.tracks {
switch track.Codec.Name {
case streamer.CodecH264:
vs.Track = track
case streamer.CodecELD:
as.Track = track
}
}
// 1. check response state
if responseM4.State != 4 || responseM4.Status != 0 {
return fmt.Errorf("wrong M4 response: %+v", responseM4)
c.server.AddSession(vs)
c.server.AddSession(as)
c.sessions = []*srtp.Session{vs, as}
return <-c.exit
}
func (c *Client) Stop() error {
err := c.conn.Close()
for _, session := range c.sessions {
c.server.RemoveSession(session)
}
c.secure, err = NewSecure(sessionShared, false)
//c.secure.Buffer = bytes.NewBuffer(nil)
c.secure.Conn = c.conn
c.httpResponse = make(chan *bufio.Reader, 10)
return err
}
// https://github.com/apple/HomeKitADK/blob/master/HAP/HAPPairingPairSetup.c
func (c *Client) Pair(mfi bool, pin string) (err error) {
pin = strings.ReplaceAll(pin, "-", "")
if len(pin) != 8 {
return fmt.Errorf("wrong PIN format: %s", pin)
}
pin = pin[:3] + "-" + pin[3:5] + "-" + pin[5:]
func (c *Client) getMedias() []*streamer.Media {
var medias []*streamer.Media
c.conn, err = net.Dial("tcp", c.DeviceAddress)
accs, err := c.conn.GetAccessories()
if err != nil {
return
}
// STEP M1. Generate request
reqM1 := struct {
Method byte `tlv8:"0"`
State byte `tlv8:"6"`
}{
State: hap.M1,
}
if mfi {
reqM1.Method = 1 // ff=1 => method=1, ff=2 => method=0
}
buf, err := tlv8.Marshal(reqM1)
if err != nil {
return
}
// STEP M1. Send request
res, err := c.Post(UriPairSetup, buf)
if err != nil {
return
}
// STEP M2. Read response
resM2 := struct {
Salt []byte `tlv8:"2"`
PublicKey []byte `tlv8:"3"` // server public key, aka session.B
State byte `tlv8:"6"`
Error byte `tlv8:"7"`
}{}
if err = tlv8.UnmarshalReader(res.Body, &resM2); err != nil {
return
}
if resM2.State != 2 || resM2.Error > 0 {
return fmt.Errorf("wrong M2: %+v", resM2)
}
// STEP M3. Generate session using pin
username := []byte("Pair-Setup")
SRP, err := srp.NewSRP(
"rfc5054.3072", sha512.New, keyDerivativeFuncRFC2945(username),
)
if err != nil {
return
}
SRP.SaltLength = 16
// username: "Pair-Setup"
// password: PIN (with dashes)
session := SRP.NewClientSession(username, []byte(pin))
sessionShared, err := session.ComputeKey(resM2.Salt, resM2.PublicKey)
if err != nil {
return
}
// STEP M3. Generate request
reqM3 := struct {
PublicKey []byte `tlv8:"3"`
Proof []byte `tlv8:"4"`
State byte `tlv8:"6"`
}{
PublicKey: session.GetA(), // client public key, aka session.A
Proof: session.ComputeAuthenticator(),
State: hap.M3,
}
buf, err = tlv8.Marshal(reqM3)
if err != nil {
return err
}
// STEP M3. Send request
res, err = c.Post(UriPairSetup, buf)
if err != nil {
return
}
// STEP M4. Read response
resM4 := struct {
Proof []byte `tlv8:"4"` // server proof
State byte `tlv8:"6"`
Error byte `tlv8:"7"`
}{}
if err = tlv8.UnmarshalReader(res.Body, &resM4); err != nil {
return
}
if resM4.Error == 2 {
return fmt.Errorf("wrong PIN: %s", pin)
}
if resM4.State != 4 || resM4.Error > 0 {
return fmt.Errorf("wrong M4: %+v", resM4)
}
// STEP M4. Verify response
if !session.VerifyServerAuthenticator(resM4.Proof) {
return errors.New("verify server auth fail")
}
// STEP M5. Generate signature
saltKey, err := hkdf.Sha512(
sessionShared, []byte("Pair-Setup-Controller-Sign-Salt"),
[]byte("Pair-Setup-Controller-Sign-Info"),
)
if err != nil {
return
}
buf = nil
buf = append(buf, saltKey[:]...)
buf = append(buf, []byte(c.ClientID)...)
buf = append(buf, c.ClientPublic()...)
signature, err := ed25519.Signature(c.ClientPrivate, buf)
if err != nil {
return
}
// STEP M5. Generate payload
msgM5 := struct {
Identifier string `tlv8:"1"`
PublicKey []byte `tlv8:"3"`
Signature []byte `tlv8:"10"`
}{
Identifier: c.ClientID,
PublicKey: c.ClientPublic(),
Signature: signature,
}
buf, err = tlv8.Marshal(msgM5)
if err != nil {
return
}
// STEP M5. Encrypt payload
sessionKey, err := hkdf.Sha512(
sessionShared, []byte("Pair-Setup-Encrypt-Salt"),
[]byte("Pair-Setup-Encrypt-Info"),
)
if err != nil {
return
}
buf, mac, _ := chacha20poly1305.EncryptAndSeal(
sessionKey[:], []byte("PS-Msg05"), buf, nil,
)
// STEP M5. Generate request
reqM5 := struct {
EncryptedData []byte `tlv8:"5"`
State byte `tlv8:"6"`
}{
EncryptedData: append(buf, mac[:]...),
State: hap.M5,
}
buf, err = tlv8.Marshal(reqM5)
if err != nil {
return err
}
// STEP M5. Send request
res, err = c.Post(UriPairSetup, buf)
if err != nil {
return
}
// STEP M6. Read response
resM6 := struct {
EncryptedData []byte `tlv8:"5"`
State byte `tlv8:"6"`
Error byte `tlv8:"7"`
}{}
if err = tlv8.UnmarshalReader(res.Body, &resM6); err != nil {
return
}
if resM6.State != 6 || resM6.Error > 0 {
return fmt.Errorf("wrong M6: %+v", resM2)
}
// STEP M6. Decrypt payload
msg := resM6.EncryptedData[:len(resM6.EncryptedData)-16]
copy(mac[:], resM6.EncryptedData[len(msg):]) // 16 byte (MAC)
buf, err = chacha20poly1305.DecryptAndVerify(
sessionKey[:], []byte("PS-Msg06"), msg, mac, nil,
)
if err != nil {
return
}
msgM6 := struct {
Identifier []byte `tlv8:"1"`
PublicKey []byte `tlv8:"3"`
Signature []byte `tlv8:"10"`
}{}
if err = tlv8.Unmarshal(buf, &msgM6); err != nil {
return
}
// STEP M6. Verify payload
if saltKey, err = hkdf.Sha512(
sessionShared, []byte("Pair-Setup-Accessory-Sign-Salt"),
[]byte("Pair-Setup-Accessory-Sign-Info"),
); err != nil {
return
}
buf = nil
buf = append(buf, saltKey[:]...)
buf = append(buf, msgM6.Identifier...)
buf = append(buf, msgM6.PublicKey...)
if !ed25519.ValidateSignature(
msgM6.PublicKey[:], buf, msgM6.Signature,
) {
return errors.New("wrong server signature")
}
if c.DeviceID != string(msgM6.Identifier) {
return fmt.Errorf("wrong Device ID: %s", msgM6.Identifier)
}
c.DevicePublic = msgM6.PublicKey
return nil
}
func (c *Client) Close() error {
if c.conn == nil {
return nil
}
conn := c.conn
c.conn = nil
return conn.Close()
}
func (c *Client) GetAccessories() ([]*Accessory, error) {
res, err := c.Get("/accessories")
if err != nil {
return nil, err
acc := accs[0]
// get supported video config (not really necessary)
char := acc.GetCharacter(characteristic.TypeSupportedVideoStreamConfiguration)
v1 := &rtp.VideoStreamConfiguration{}
if err = char.ReadTLV8(v1); err != nil {
return nil
}
data, err := io.ReadAll(res.Body)
if err != nil {
return nil, err
}
for _, hkCodec := range v1.Codecs {
codec := &streamer.Codec{ClockRate: 90000}
p := Accessories{}
if err = json.Unmarshal(data, &p); err != nil {
return nil, err
}
for _, accs := range p.Accessories {
for _, serv := range accs.Services {
for _, char := range serv.Characters {
char.AID = accs.AID
}
switch hkCodec.Type {
case rtp.VideoCodecType_H264:
codec.Name = streamer.CodecH264
codec.FmtpLine = "profile-level-id=420029"
default:
fmt.Printf("unknown codec: %d", hkCodec.Type)
continue
}
}
return p.Accessories, nil
}
func (c *Client) GetCharacters(query string) ([]*Character, error) {
res, err := c.Get("/characteristics?id=" + query)
if err != nil {
return nil, err
}
data, err := io.ReadAll(res.Body)
if err != nil {
return nil, err
}
ch := Characters{}
if err = json.Unmarshal(data, &ch); err != nil {
return nil, err
}
return ch.Characters, nil
}
func (c *Client) GetCharacter(char *Character) error {
query := fmt.Sprintf("%d.%d", char.AID, char.IID)
chars, err := c.GetCharacters(query)
if err != nil {
return err
}
char.Value = chars[0].Value
return nil
}
func (c *Client) PutCharacters(characters ...*Character) (err error) {
for i, char := range characters {
if char.Event != nil {
char = &Character{AID: char.AID, IID: char.IID, Event: char.Event}
} else {
char = &Character{AID: char.AID, IID: char.IID, Value: char.Value}
media := &streamer.Media{
Kind: streamer.KindVideo, Direction: streamer.DirectionSendonly,
Codecs: []*streamer.Codec{codec},
}
characters[i] = char
}
var data []byte
if data, err = json.Marshal(Characters{characters}); err != nil {
return
medias = append(medias, media)
}
var res *http.Response
if res, err = c.Put("/characteristics", data); err != nil {
return
}
if res.StatusCode >= 400 {
return errors.New("wrong response status")
}
return
}
func (c *Client) GetImage(width, height int) ([]byte, error) {
res, err := c.Post(
"/resource", []byte(fmt.Sprintf(
`{"image-width":%d,"image-height":%d,"resource-type":"image","reason":0}`,
width, height,
)),
)
if err != nil {
return nil, err
}
return io.ReadAll(res.Body)
}
//func (c *Client) onEventData(r io.Reader) error {
// if c.OnEvent == nil {
// return nil
// }
//
// data, err := io.ReadAll(r)
//
// ch := Characters{}
// if err = json.Unmarshal(data, &ch); err != nil {
// return err
// }
//
// c.OnEvent(ch.Characters)
//
// return nil
//}
func (c *Client) ListPairings() error {
pReq := struct {
Method byte `tlv8:"0"`
State byte `tlv8:"6"`
}{
Method: hap.MethodListPairings,
State: hap.M1,
}
data, err := tlv8.Marshal(pReq)
if err != nil {
return err
}
res, err := c.Post("/pairings", data)
if err != nil {
return err
}
data, err = io.ReadAll(res.Body)
// TODO: don't know how to fix array of items
var pRes struct {
State byte `tlv8:"6"`
Identifier string `tlv8:"1"`
PublicKey []byte `tlv8:"3"`
Permission byte `tlv8:"11"`
}
if err = tlv8.Unmarshal(data, &pRes); err != nil {
return err
}
return nil
}
func (c *Client) PairingsAdd(clientID string, clientPublic []byte, admin bool) error {
pReq := struct {
Method byte `tlv8:"0"`
Identifier string `tlv8:"1"`
PublicKey []byte `tlv8:"3"`
State byte `tlv8:"6"`
Permission byte `tlv8:"11"`
}{
Method: hap.MethodAddPairing,
Identifier: clientID,
PublicKey: clientPublic,
State: hap.M1,
Permission: hap.PermissionUser,
}
if admin {
pReq.Permission = hap.PermissionAdmin
}
data, err := tlv8.Marshal(pReq)
if err != nil {
return err
}
res, err := c.Post("/pairings", data)
if err != nil {
return err
}
data, err = io.ReadAll(res.Body)
var pRes struct {
State byte `tlv8:"6"`
Unknown byte `tlv8:"7"`
}
if err = tlv8.Unmarshal(data, &pRes); err != nil {
return err
}
return nil
}
func (c *Client) DeletePairing(id string) error {
reqM1 := struct {
State byte `tlv8:"6"`
Method byte `tlv8:"0"`
Identifier string `tlv8:"1"`
}{
State: hap.M1,
Method: hap.MethodDeletePairing,
Identifier: id,
}
data, err := tlv8.Marshal(reqM1)
if err != nil {
return err
}
res, err := c.Post("/pairings", data)
if err != nil {
return err
}
data, err = io.ReadAll(res.Body)
var resM2 struct {
State byte `tlv8:"6"`
}
if err = tlv8.Unmarshal(data, &resM2); err != nil {
return err
}
if resM2.State != hap.M2 {
return errors.New("wrong state")
}
return nil
}
func (c *Client) LocalAddr() string {
return c.conn.LocalAddr().String()
}
func DecodeKey(s string) []byte {
if s == "" {
char = acc.GetCharacter(characteristic.TypeSupportedAudioStreamConfiguration)
v2 := &rtp.AudioStreamConfiguration{}
if err = char.ReadTLV8(v2); err != nil {
return nil
}
data, err := hex.DecodeString(s)
if err != nil {
return nil
for _, hkCodec := range v2.Codecs {
codec := &streamer.Codec{
Channels: uint16(hkCodec.Parameters.Channels),
}
switch hkCodec.Parameters.Samplerate {
case rtp.AudioCodecSampleRate8Khz:
codec.ClockRate = 8000
case rtp.AudioCodecSampleRate16Khz:
codec.ClockRate = 16000
case rtp.AudioCodecSampleRate24Khz:
codec.ClockRate = 24000
default:
panic(fmt.Sprintf("unknown clockrate: %d", hkCodec.Parameters.Samplerate))
}
switch hkCodec.Type {
case rtp.AudioCodecType_AAC_ELD:
codec.Name = streamer.CodecELD
// only this value supported by FFmpeg
codec.FmtpLine = "profile-level-id=1;mode=AAC-hbr;sizelength=13;indexlength=3;indexdeltalength=3;config=F8EC3000"
default:
fmt.Printf("unknown codec: %d", hkCodec.Type)
continue
}
media := &streamer.Media{
Kind: streamer.KindAudio, Direction: streamer.DirectionSendonly,
Codecs: []*streamer.Codec{codec},
}
medias = append(medias, media)
}
return data
return medias
}

View File

@@ -6,7 +6,6 @@ import (
"encoding/binary"
"encoding/json"
"fmt"
"github.com/AlexxIT/go2rtc/pkg/h264"
"github.com/AlexxIT/go2rtc/pkg/streamer"
"github.com/deepch/vdk/codec/h264parser"
"github.com/deepch/vdk/format/fmp4/fmp4io"
@@ -162,9 +161,12 @@ func (c *Client) getTracks() error {
continue
}
codec := streamer.NewCodec(streamer.CodecH264)
codec.FmtpLine = "profile-level-id=" + msg.CodecString[i+1:]
codec.PayloadType = h264.PayloadTypeAVC
codec := &streamer.Codec{
Name: streamer.CodecH264,
ClockRate: 90000,
FmtpLine: "profile-level-id=" + msg.CodecString[i+1:],
PayloadType: streamer.PayloadTypeMP4,
}
i = bytes.Index(msg.Data, []byte("avcC")) - 4
if i < 0 {

View File

@@ -21,3 +21,4 @@ Stream #0:0(eng): Video: hevc (Main) (hvc1 / 0x31637668), yuv420p(tv, progressiv
- https://stackoverflow.com/questions/32152090/encode-h265-to-hvc1-codec
- https://jellyfin.org/docs/general/clients/codec-support.html
- https://github.com/StaZhu/enable-chromium-hevc-hardware-decoding
- https://developer.mozilla.org/ru/docs/Web/Media/Formats/codecs_parameter

View File

@@ -3,6 +3,8 @@ package mp4
import (
"encoding/binary"
"github.com/deepch/vdk/format/mp4/mp4io"
"github.com/deepch/vdk/format/mp4f"
"github.com/deepch/vdk/format/mp4f/mp4fio"
"time"
)
@@ -37,25 +39,17 @@ func MOOV() *mp4io.Movie {
SelectionDuration: time0,
CurrentTime: time0,
},
MovieExtend: &mp4io.MovieExtend{
Tracks: []*mp4io.TrackExtend{
{
TrackId: 1,
DefaultSampleDescIdx: 1,
DefaultSampleDuration: 40,
},
},
},
MovieExtend: &mp4io.MovieExtend{},
}
}
func TRAK() *mp4io.Track {
func TRAK(id int) *mp4io.Track {
return &mp4io.Track{
// trak > tkhd
Header: &mp4io.TrackHeader{
TrackId: int32(1), // change me
Flags: 0x0007, // 7 ENABLED IN-MOVIE IN-PREVIEW
Duration: 0, // OK
TrackId: int32(id),
Flags: 0x0007, // 7 ENABLED IN-MOVIE IN-PREVIEW
Duration: 0, // OK
Matrix: matrix,
CreateTime: time0,
ModifyTime: time0,
@@ -92,3 +86,15 @@ func TRAK() *mp4io.Track {
},
}
}
func ESDS(conf []byte) *mp4f.FDummy {
esds := &mp4fio.ElemStreamDesc{DecConfig: conf}
b := make([]byte, esds.Len())
esds.Marshal(b)
return &mp4f.FDummy{
Data: b,
Tag_: mp4io.Tag(uint32(mp4io.ESDS)),
}
}

View File

@@ -2,7 +2,7 @@ package mp4
import (
"encoding/json"
"fmt"
"github.com/AlexxIT/go2rtc/pkg/aac"
"github.com/AlexxIT/go2rtc/pkg/h264"
"github.com/AlexxIT/go2rtc/pkg/h265"
"github.com/AlexxIT/go2rtc/pkg/streamer"
@@ -28,44 +28,37 @@ func (c *Consumer) GetMedias() []*streamer.Media {
Kind: streamer.KindVideo,
Direction: streamer.DirectionRecvonly,
Codecs: []*streamer.Codec{
{Name: streamer.CodecH264, ClockRate: 90000},
{Name: streamer.CodecH265, ClockRate: 90000},
{Name: streamer.CodecH264},
{Name: streamer.CodecH265},
},
},
{
Kind: streamer.KindAudio,
Direction: streamer.DirectionRecvonly,
Codecs: []*streamer.Codec{
{Name: streamer.CodecAAC},
},
},
//{
// Kind: streamer.KindAudio,
// Direction: streamer.DirectionRecvonly,
// Codecs: []*streamer.Codec{
// {Name: streamer.CodecAAC, ClockRate: 16000},
// },
//},
}
}
func (c *Consumer) AddTrack(media *streamer.Media, track *streamer.Track) *streamer.Track {
trackID := byte(len(c.codecs))
c.codecs = append(c.codecs, track.Codec)
codec := track.Codec
switch codec.Name {
case streamer.CodecH264:
c.codecs = append(c.codecs, track.Codec)
push := func(packet *rtp.Packet) error {
if packet.Version != h264.RTPPacketVersionAVC {
return nil
}
if c.muxer == nil {
if !c.start {
return nil
}
if !c.start {
if h264.IsKeyframe(packet.Payload) {
c.start = true
} else {
return nil
}
}
buf := c.muxer.Marshal(packet)
buf := c.muxer.Marshal(trackID, packet)
c.send += len(buf)
c.Fire(buf)
@@ -73,7 +66,7 @@ func (c *Consumer) AddTrack(media *streamer.Media, track *streamer.Track) *strea
}
var wrapper streamer.WrapperFunc
if h264.IsAVC(codec) {
if codec.IsMP4() {
wrapper = h264.RepairAVC(track)
} else {
wrapper = h264.RTPDepay(track)
@@ -83,39 +76,51 @@ func (c *Consumer) AddTrack(media *streamer.Media, track *streamer.Track) *strea
return track.Bind(push)
case streamer.CodecH265:
c.codecs = append(c.codecs, track.Codec)
push := func(packet *rtp.Packet) error {
if packet.Version != h264.RTPPacketVersionAVC {
return nil
}
if !c.start {
if h265.IsKeyframe(packet.Payload) {
c.start = true
} else {
return nil
}
return nil
}
buf := c.muxer.Marshal(packet)
buf := c.muxer.Marshal(trackID, packet)
c.send += len(buf)
c.Fire(buf)
return nil
}
if !h264.IsAVC(codec) {
if !codec.IsMP4() {
wrapper := h265.RTPDepay(track)
push = wrapper(push)
}
return track.Bind(push)
case streamer.CodecAAC:
push := func(packet *rtp.Packet) error {
if !c.start {
return nil
}
buf := c.muxer.Marshal(trackID, packet)
c.send += len(buf)
c.Fire(buf)
return nil
}
if !codec.IsMP4() {
wrapper := aac.RTPDepay(track)
push = wrapper(push)
}
return track.Bind(push)
}
fmt.Printf("[rtmp] unsupported codec: %+v\n", track.Codec)
return nil
panic("unsupported codec")
}
func (c *Consumer) MimeType() string {
@@ -123,12 +128,14 @@ func (c *Consumer) MimeType() string {
}
func (c *Consumer) Init() ([]byte, error) {
if c.muxer == nil {
c.muxer = &Muxer{}
}
c.muxer = &Muxer{}
return c.muxer.GetInit(c.codecs)
}
func (c *Consumer) Start() {
c.start = true
}
//
func (c *Consumer) MarshalJSON() ([]byte, error) {

85
pkg/mp4/keyframe.go Normal file
View File

@@ -0,0 +1,85 @@
package mp4
import (
"github.com/AlexxIT/go2rtc/pkg/h264"
"github.com/AlexxIT/go2rtc/pkg/h265"
"github.com/AlexxIT/go2rtc/pkg/streamer"
"github.com/pion/rtp"
)
type Keyframe struct {
streamer.Element
MimeType string
}
func (c *Keyframe) GetMedias() []*streamer.Media {
return []*streamer.Media{
{
Kind: streamer.KindVideo,
Direction: streamer.DirectionRecvonly,
Codecs: []*streamer.Codec{
{Name: streamer.CodecH264},
{Name: streamer.CodecH265},
},
},
}
}
func (c *Keyframe) AddTrack(media *streamer.Media, track *streamer.Track) *streamer.Track {
muxer := &Muxer{}
codecs := []*streamer.Codec{track.Codec}
init, err := muxer.GetInit(codecs)
if err != nil {
return nil
}
c.MimeType = muxer.MimeType(codecs)
switch track.Codec.Name {
case streamer.CodecH264:
push := func(packet *rtp.Packet) error {
if !h264.IsKeyframe(packet.Payload) {
return nil
}
buf := muxer.Marshal(0, packet)
c.Fire(append(init, buf...))
return nil
}
var wrapper streamer.WrapperFunc
if track.Codec.IsMP4() {
wrapper = h264.RepairAVC(track)
} else {
wrapper = h264.RTPDepay(track)
}
push = wrapper(push)
return track.Bind(push)
case streamer.CodecH265:
push := func(packet *rtp.Packet) error {
if !h265.IsKeyframe(packet.Payload) {
return nil
}
buf := muxer.Marshal(0, packet)
c.Fire(append(init, buf...))
return nil
}
if !track.Codec.IsMP4() {
wrapper := h265.RTPDepay(track)
push = wrapper(push)
}
return track.Bind(push)
}
panic("unsupported codec")
}

View File

@@ -2,10 +2,12 @@ package mp4
import (
"encoding/binary"
"encoding/hex"
"fmt"
"github.com/AlexxIT/go2rtc/pkg/h264"
"github.com/AlexxIT/go2rtc/pkg/h265"
"github.com/AlexxIT/go2rtc/pkg/streamer"
"github.com/deepch/vdk/av"
"github.com/deepch/vdk/codec/h264parser"
"github.com/deepch/vdk/codec/h265parser"
"github.com/deepch/vdk/format/fmp4/fmp4io"
@@ -16,22 +18,28 @@ import (
type Muxer struct {
fragIndex uint32
dts uint64
pts uint32
data []byte
total int
dts []uint64
pts []uint32
//data []byte
//total int
}
func (m *Muxer) MimeType(codecs []*streamer.Codec) string {
s := `video/mp4; codecs="`
for _, codec := range codecs {
for i, codec := range codecs {
if i > 0 {
s += ","
}
switch codec.Name {
case streamer.CodecH264:
s += "avc1." + h264.GetProfileLevelID(codec.FmtpLine)
case streamer.CodecH265:
// +Safari +Chrome +Edge -iOS15 -Android13
s += "hvc1.1.6.L93.B0" // hev1.1.6.L93.B0
case streamer.CodecAAC:
s += "mp4a.40.2"
}
}
@@ -41,7 +49,7 @@ func (m *Muxer) MimeType(codecs []*streamer.Codec) string {
func (m *Muxer) GetInit(codecs []*streamer.Codec) ([]byte, error) {
moov := MOOV()
for _, codec := range codecs {
for i, codec := range codecs {
switch codec.Name {
case streamer.CodecH264:
sps, pps := h264.GetParameterSet(codec.FmtpLine)
@@ -59,11 +67,14 @@ func (m *Muxer) GetInit(codecs []*streamer.Codec) ([]byte, error) {
width := codecData.Width()
height := codecData.Height()
trak := TRAK()
trak.Media.Header.TimeScale = int32(codec.ClockRate)
trak := TRAK(i + 1)
trak.Header.TrackWidth = float64(width)
trak.Header.TrackHeight = float64(height)
trak.Media.Header.TimeScale = int32(codec.ClockRate)
trak.Media.Handler = &mp4io.HandlerRefer{
SubType: [4]byte{'v', 'i', 'd', 'e'},
Name: []byte{0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 'm', 'a', 'i', 'n', 0},
}
trak.Media.Info.Video = &mp4io.VideoMediaInfo{
Flags: 0x000001,
}
@@ -81,11 +92,6 @@ func (m *Muxer) GetInit(codecs []*streamer.Codec) ([]byte, error) {
},
}
trak.Media.Handler = &mp4io.HandlerRefer{
SubType: [4]byte{'v', 'i', 'd', 'e'},
Name: []byte{0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 'm', 'a', 'i', 'n', 0},
}
moov.Tracks = append(moov.Tracks, trak)
case streamer.CodecH265:
@@ -102,11 +108,14 @@ func (m *Muxer) GetInit(codecs []*streamer.Codec) ([]byte, error) {
width := codecData.Width()
height := codecData.Height()
trak := TRAK()
trak.Media.Header.TimeScale = int32(codec.ClockRate)
trak := TRAK(i + 1)
trak.Header.TrackWidth = float64(width)
trak.Header.TrackHeight = float64(height)
trak.Media.Header.TimeScale = int32(codec.ClockRate)
trak.Media.Handler = &mp4io.HandlerRefer{
SubType: [4]byte{'v', 'i', 'd', 'e'},
Name: []byte{0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 'm', 'a', 'i', 'n', 0},
}
trak.Media.Info.Video = &mp4io.VideoMediaInfo{
Flags: 0x000001,
}
@@ -124,13 +133,47 @@ func (m *Muxer) GetInit(codecs []*streamer.Codec) ([]byte, error) {
},
}
moov.Tracks = append(moov.Tracks, trak)
case streamer.CodecAAC:
s := streamer.Between(codec.FmtpLine, "config=", ";")
b, err := hex.DecodeString(s)
if err != nil {
return nil, err
}
trak := TRAK(i + 1)
trak.Header.AlternateGroup = 1
trak.Header.Duration = 0
trak.Header.Volume = 1
trak.Media.Header.TimeScale = int32(codec.ClockRate)
trak.Media.Handler = &mp4io.HandlerRefer{
SubType: [4]byte{'v', 'i', 'd', 'e'},
SubType: [4]byte{'s', 'o', 'u', 'n'},
Name: []byte{0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 'm', 'a', 'i', 'n', 0},
}
trak.Media.Info.Sound = &mp4io.SoundMediaInfo{}
trak.Media.Info.Sample.SampleDesc.MP4ADesc = &mp4io.MP4ADesc{
DataRefIdx: 1,
NumberOfChannels: int16(codec.Channels),
SampleSize: int16(av.FLTP.BytesPerSample() * 4),
SampleRate: float64(codec.ClockRate),
Unknowns: []mp4io.Atom{ESDS(b)},
}
moov.Tracks = append(moov.Tracks, trak)
}
trex := &mp4io.TrackExtend{
TrackId: uint32(i + 1),
DefaultSampleDescIdx: 1,
DefaultSampleDuration: 0,
}
moov.MovieExtend.Tracks = append(moov.MovieExtend.Tracks, trex)
m.pts = append(m.pts, 0)
m.dts = append(m.dts, 0)
}
data := make([]byte, moov.Len())
@@ -139,14 +182,12 @@ func (m *Muxer) GetInit(codecs []*streamer.Codec) ([]byte, error) {
return append(FTYP(), data...), nil
}
func (m *Muxer) Rewind() {
m.dts = 0
m.pts = 0
}
func (m *Muxer) Marshal(packet *rtp.Packet) []byte {
trackID := uint8(1)
//func (m *Muxer) Rewind() {
// m.dts = 0
// m.pts = 0
//}
func (m *Muxer) Marshal(trackID byte, packet *rtp.Packet) []byte {
run := &mp4fio.TrackFragRun{
Flags: 0x000b05,
FirstSampleFlags: uint32(fmp4io.SampleNoDependencies),
@@ -161,12 +202,12 @@ func (m *Muxer) Marshal(packet *rtp.Packet) []byte {
Tracks: []*mp4fio.TrackFrag{
{
Header: &mp4fio.TrackFragHeader{
Data: []byte{0x00, 0x02, 0x00, 0x20, 0x00, 0x00, 0x00, trackID, 0x01, 0x01, 0x00, 0x00},
Data: []byte{0x00, 0x02, 0x00, 0x20, 0x00, 0x00, 0x00, trackID + 1, 0x01, 0x01, 0x00, 0x00},
},
DecodeTime: &mp4fio.TrackFragDecodeTime{
Version: 1,
Flags: 0,
Time: m.dts,
Time: m.dts[trackID],
},
Run: run,
},
@@ -179,12 +220,12 @@ func (m *Muxer) Marshal(packet *rtp.Packet) []byte {
}
newTime := packet.Timestamp
if m.pts > 0 {
if m.pts[trackID] > 0 {
//m.dts += uint64(newTime - m.pts)
entry.Duration = newTime - m.pts
m.dts += uint64(entry.Duration)
entry.Duration = newTime - m.pts[trackID]
m.dts[trackID] += uint64(entry.Duration)
}
m.pts = newTime
m.pts[trackID] = newTime
// important before moof.Len()
run.Entries = append(run.Entries, entry)
@@ -204,7 +245,7 @@ func (m *Muxer) Marshal(packet *rtp.Packet) []byte {
m.fragIndex++
m.total += moofLen + mdatLen
//m.total += moofLen + mdatLen
return buf
}

164
pkg/mp4f/consumer.go Normal file
View File

@@ -0,0 +1,164 @@
package mp4f
import (
"encoding/json"
"github.com/AlexxIT/go2rtc/pkg/h264"
"github.com/AlexxIT/go2rtc/pkg/streamer"
"github.com/deepch/vdk/av"
"github.com/deepch/vdk/codec/aacparser"
"github.com/deepch/vdk/codec/h264parser"
"github.com/deepch/vdk/format/mp4f"
"github.com/pion/rtp"
"time"
)
type Consumer struct {
streamer.Element
UserAgent string
RemoteAddr string
muxer *mp4f.Muxer
streams []av.CodecData
mimeType string
start bool
send int
}
func (c *Consumer) GetMedias() []*streamer.Media {
return []*streamer.Media{
{
Kind: streamer.KindVideo,
Direction: streamer.DirectionRecvonly,
Codecs: []*streamer.Codec{
{Name: streamer.CodecH264, ClockRate: 90000},
},
},
{
Kind: streamer.KindAudio,
Direction: streamer.DirectionRecvonly,
Codecs: []*streamer.Codec{
{Name: streamer.CodecAAC, ClockRate: 16000},
},
},
}
}
func (c *Consumer) AddTrack(media *streamer.Media, track *streamer.Track) *streamer.Track {
codec := track.Codec
trackID := int8(len(c.streams))
switch codec.Name {
case streamer.CodecH264:
sps, pps := h264.GetParameterSet(codec.FmtpLine)
stream, err := h264parser.NewCodecDataFromSPSAndPPS(sps, pps)
if err != nil {
return nil
}
c.mimeType += "avc1." + h264.GetProfileLevelID(codec.FmtpLine)
c.streams = append(c.streams, stream)
pkt := av.Packet{Idx: trackID, CompositionTime: time.Millisecond}
ts2time := time.Second / time.Duration(codec.ClockRate)
push := func(packet *rtp.Packet) error {
if packet.Version != h264.RTPPacketVersionAVC {
return nil
}
if !c.start {
return nil
}
pkt.Data = packet.Payload
newTime := time.Duration(packet.Timestamp) * ts2time
if pkt.Time > 0 {
pkt.Duration = newTime - pkt.Time
}
pkt.Time = newTime
ready, buf, _ := c.muxer.WritePacket(pkt, false)
if ready {
c.send += len(buf)
c.Fire(buf)
}
return nil
}
if !codec.IsMP4() {
wrapper := h264.RTPDepay(track)
push = wrapper(push)
}
return track.Bind(push)
case streamer.CodecAAC:
stream, _ := aacparser.NewCodecDataFromMPEG4AudioConfigBytes([]byte{20, 8})
c.mimeType += ",mp4a.40.2"
c.streams = append(c.streams, stream)
pkt := av.Packet{Idx: trackID, CompositionTime: time.Millisecond}
ts2time := time.Second / time.Duration(codec.ClockRate)
push := func(packet *rtp.Packet) error {
if !c.start {
return nil
}
pkt.Data = packet.Payload
newTime := time.Duration(packet.Timestamp) * ts2time
if pkt.Time > 0 {
pkt.Duration = newTime - pkt.Time
}
pkt.Time = newTime
ready, buf, _ := c.muxer.WritePacket(pkt, false)
if ready {
c.send += len(buf)
c.Fire(buf)
}
return nil
}
return track.Bind(push)
}
panic("unsupported codec")
}
func (c *Consumer) MimeType() string {
return `video/mp4; codecs="` + c.mimeType + `"`
}
func (c *Consumer) Init() ([]byte, error) {
c.muxer = mp4f.NewMuxer(nil)
if err := c.muxer.WriteHeader(c.streams); err != nil {
return nil, err
}
_, data := c.muxer.GetInit(c.streams)
return data, nil
}
func (c *Consumer) Start() {
c.start = true
}
//
func (c *Consumer) MarshalJSON() ([]byte, error) {
v := map[string]interface{}{
"type": "MSE server consumer",
"send": c.send,
"remote_addr": c.RemoteAddr,
"user_agent": c.UserAgent,
}
return json.Marshal(v)
}

View File

@@ -4,7 +4,6 @@ import (
"encoding/base64"
"encoding/hex"
"fmt"
"github.com/AlexxIT/go2rtc/pkg/h264"
"github.com/AlexxIT/go2rtc/pkg/httpflv"
"github.com/AlexxIT/go2rtc/pkg/streamer"
"github.com/deepch/vdk/av"
@@ -74,7 +73,7 @@ func (c *Client) Dial() (err error) {
Name: streamer.CodecH264,
ClockRate: 90000,
FmtpLine: fmtp,
PayloadType: h264.PayloadTypeAVC,
PayloadType: streamer.PayloadTypeMP4,
}
media := &streamer.Media{
@@ -93,17 +92,13 @@ func (c *Client) Dial() (err error) {
// TODO: fix support
cd := stream.(aacparser.CodecData)
// a=fmtp:97 streamtype=5;profile-level-id=1;mode=AAC-hbr;sizelength=13;indexlength=3;indexdeltalength=3;config=1588
fmtp := fmt.Sprintf(
"config=%s",
hex.EncodeToString(cd.ConfigBytes),
)
codec := &streamer.Codec{
Name: streamer.CodecAAC,
ClockRate: uint32(cd.Config.SampleRate),
Channels: uint16(cd.Config.ChannelConfig),
FmtpLine: fmtp,
// a=fmtp:97 streamtype=5;profile-level-id=1;mode=AAC-hbr;sizelength=13;indexlength=3;indexdeltalength=3;config=1588
FmtpLine: "streamtype=5;profile-level-id=1;mode=AAC-hbr;sizelength=13;indexlength=3;indexdeltalength=3;config=" + hex.EncodeToString(cd.ConfigBytes),
PayloadType: streamer.PayloadTypeMP4,
}
media := &streamer.Media{

View File

@@ -7,6 +7,7 @@ import (
"encoding/binary"
"errors"
"fmt"
"github.com/AlexxIT/go2rtc/pkg/aac"
"github.com/AlexxIT/go2rtc/pkg/h264"
"github.com/AlexxIT/go2rtc/pkg/streamer"
"github.com/AlexxIT/go2rtc/pkg/tcp"
@@ -43,6 +44,15 @@ const (
ModeServerConsumer
)
type State byte
const (
StateNone State = iota
StateConn
StateSetup
StatePlay
)
type Conn struct {
streamer.Element
@@ -58,9 +68,9 @@ type Conn struct {
// internal
auth *tcp.Auth
closed bool
conn net.Conn
mode Mode
state State
reader *bufio.Reader
sequence int
uri string
@@ -107,9 +117,6 @@ func (c *Conn) parseURI() (err error) {
}
func (c *Conn) Dial() (err error) {
//if c.state != StateClientInit {
// panic("wrong state")
//}
if c.conn != nil {
_ = c.parseURI()
}
@@ -136,6 +143,7 @@ func (c *Conn) Dial() (err error) {
}
c.reader = bufio.NewReader(c.conn)
c.state = StateConn
return nil
}
@@ -436,33 +444,35 @@ func (c *Conn) SetupMedia(
track = c.bindTrack(track, byte(ch), codec.PayloadType)
}
c.state = StateSetup
c.tracks = append(c.tracks, track)
return track, nil
}
func (c *Conn) Play() (err error) {
if c.state != StateSetup {
return fmt.Errorf("RTSP PLAY from wrong state: %s", c.state)
}
req := &tcp.Request{Method: MethodPlay, URL: c.URL}
return c.Request(req)
}
func (c *Conn) Teardown() (err error) {
//if c.state != StateClientPlay {
// panic("wrong state")
//}
// allow TEARDOWN from any state (ex. ANNOUNCE > SETUP)
req := &tcp.Request{Method: MethodTeardown, URL: c.URL}
return c.Request(req)
}
func (c *Conn) Close() error {
if c.closed {
if c.state == StateNone {
return nil
}
if err := c.Teardown(); err != nil {
return err
}
c.closed = true
c.state = StateNone
return c.conn.Close()
}
@@ -573,6 +583,7 @@ func (c *Conn) Accept() error {
if strings.HasPrefix(tr, transport) {
c.Session = "1" // TODO: fixme
c.state = StateSetup
res.Header.Set("Transport", tr[:len(transport)+3])
} else {
res.Status = "461 Unsupported transport"
@@ -593,14 +604,22 @@ func (c *Conn) Accept() error {
}
func (c *Conn) Handle() (err error) {
if c.state != StateSetup {
return fmt.Errorf("RTSP Handle from wrong state: %d", c.state)
}
c.state = StatePlay
defer func() {
if c.closed {
if c.state == StateNone {
err = nil
} else {
// may have gotten here because of the deadline
// so close the connection to stop keepalive
_ = c.conn.Close()
return
}
// may have gotten here because of the deadline
// so close the connection to stop keepalive
c.state = StateNone
_ = c.conn.Close()
}()
var timeout time.Duration
@@ -624,7 +643,7 @@ func (c *Conn) Handle() (err error) {
}
for {
if c.closed {
if c.state == StateNone {
return
}
@@ -643,22 +662,21 @@ func (c *Conn) Handle() (err error) {
}
if buf4[0] != '$' {
if string(buf4) == "RTSP" {
switch string(buf4) {
case "RTSP":
var res *tcp.Response
res, err = tcp.ReadResponse(c.reader)
if err != nil {
if res, err = tcp.ReadResponse(c.reader); err != nil {
return
}
c.Fire(res)
} else {
case "OPTI", "TEAR", "DESC", "SETU", "PLAY", "PAUS", "RECO", "ANNO", "GET_", "SET_":
var req *tcp.Request
req, err = tcp.ReadRequest(c.reader)
if err != nil {
if req, err = tcp.ReadRequest(c.reader); err != nil {
return
}
c.Fire(req)
default:
return fmt.Errorf("RTSP wrong input")
}
continue
}
@@ -717,7 +735,7 @@ func (c *Conn) keepalive() {
req := &tcp.Request{Method: MethodOptions, URL: c.URL}
for {
time.Sleep(time.Second * 25)
if c.closed {
if c.state == StateNone {
return
}
if err := c.Request(req); err != nil {
@@ -739,7 +757,7 @@ func (c *Conn) bindTrack(
track *streamer.Track, channel uint8, payloadType uint8,
) *streamer.Track {
push := func(packet *rtp.Packet) error {
if c.closed {
if c.state == StateNone {
return nil
}
packet.Header.PayloadType = payloadType
@@ -764,9 +782,15 @@ func (c *Conn) bindTrack(
return nil
}
if h264.IsAVC(track.Codec) {
wrapper := h264.RTPPay(1500)
push = wrapper(push)
if track.Codec.IsMP4() {
switch track.Codec.Name {
case streamer.CodecH264:
wrapper := h264.RTPPay(1500)
push = wrapper(push)
case streamer.CodecAAC:
wrapper := aac.RTPPay(1500)
push = wrapper(push)
}
}
return track.Bind(push)

View File

@@ -20,6 +20,11 @@ func (c *Conn) GetTrack(media *streamer.Media, codec *streamer.Codec) *streamer.
}
}
// can't setup new tracks from play state
if c.state == StatePlay {
return nil
}
track, err := c.SetupMedia(media, codec)
if err != nil {
return nil

View File

@@ -8,9 +8,19 @@ import (
// Server using same UDP port for SRTP and for SRTCP as the iPhone does
// this is not really necessary but anyway
type Server struct {
conn net.PacketConn
sessions map[uint32]*Session
}
func (s *Server) Port() uint16 {
addr := s.conn.LocalAddr().(*net.UDPAddr)
return uint16(addr.Port)
}
func (s *Server) Close() error {
return s.conn.Close()
}
func (s *Server) AddSession(session *Session) {
if s.sessions == nil {
s.sessions = map[uint32]*Session{}
@@ -23,6 +33,8 @@ func (s *Server) RemoveSession(session *Session) {
}
func (s *Server) Serve(conn net.PacketConn) error {
s.conn = conn
buf := make([]byte, 2048)
for {
n, addr, err := conn.ReadFrom(buf)

View File

@@ -5,6 +5,7 @@ import (
"github.com/pion/rtcp"
"github.com/pion/rtp"
"github.com/pion/srtp/v2"
"time"
)
type Session struct {
@@ -16,6 +17,14 @@ type Session struct {
Write func(b []byte) (int, error)
Track *streamer.Track
lastSequence uint32
lastTimestamp uint32
//lastPacket *rtp.Packet
lastTime time.Time
jitter float64
//sequenceCycle uint16
totalLost uint32
}
func (s *Session) SetKeys(
@@ -37,13 +46,42 @@ func (s *Session) HandleRTP(data []byte) (err error) {
return
}
if s.Track == nil {
return
}
packet := &rtp.Packet{}
if err = packet.Unmarshal(data); err != nil {
return
}
now := time.Now()
// https://www.ietf.org/rfc/rfc3550.txt
if s.lastTimestamp != 0 {
delta := packet.SequenceNumber - uint16(s.lastSequence)
// lost packet
if delta > 1 {
s.totalLost += uint32(delta - 1)
}
// D(i,j) = (Rj - Ri) - (Sj - Si) = (Rj - Sj) - (Ri - Si)
dTime := now.Sub(s.lastTime).Seconds()*float64(s.Track.Codec.ClockRate) -
float64(packet.Timestamp-s.lastTimestamp)
if dTime < 0 {
dTime = -dTime
}
// J(i) = J(i-1) + (|D(i-1,i)| - J(i-1))/16
s.jitter += (dTime - s.jitter) / 16
}
// keeping cycles (overflow)
s.lastSequence = s.lastSequence&0xFFFF0000 | uint32(packet.SequenceNumber)
s.lastTimestamp = packet.Timestamp
s.lastTime = now
_ = s.Track.WriteRTP(packet)
//s.Output(core.RTP{Channel: s.Channel, Packet: packet})
return
}
@@ -60,7 +98,6 @@ func (s *Session) HandleRTCP(data []byte) (err error) {
}
_ = packets
//s.Output(core.RTCP{Channel: s.Channel + 1, Header: header, Packets: packets})
if header.Type == rtcp.TypeSenderReport {
err = s.KeepAlive()
@@ -70,9 +107,25 @@ func (s *Session) HandleRTCP(data []byte) (err error) {
}
func (s *Session) KeepAlive() (err error) {
var data []byte
// we can send empty receiver response, but should send it to hold the connection
rep := rtcp.ReceiverReport{SSRC: s.LocalSSRC}
if s.lastTimestamp > 0 {
//log.Printf("[RTCP] ssrc=%d seq=%d lost=%d jit=%.2f", s.RemoteSSRC, s.lastSequence, s.totalLost, s.jitter)
rep.Reports = []rtcp.ReceptionReport{{
SSRC: s.RemoteSSRC,
LastSequenceNumber: s.lastSequence,
LastSenderReport: s.lastTimestamp,
FractionLost: 0, // TODO
TotalLost: s.totalLost,
Delay: 0, // send just after receive
Jitter: uint32(s.jitter),
}}
}
// we can send empty receiver response, but should send it to hold the connection
var data []byte
if data, err = rep.Marshal(); err != nil {
return
}
@@ -90,8 +143,8 @@ func GuessProfile(masterKey []byte) srtp.ProtectionProfile {
switch len(masterKey) {
case 16:
return srtp.ProtectionProfileAes128CmHmacSha1_80
//case 32:
// return srtp.ProtectionProfileAes256CmHmacSha1_80
//case 32:
// return srtp.ProtectionProfileAes256CmHmacSha1_80
}
return 0
}

View File

@@ -33,13 +33,17 @@ const (
CodecOpus = "OPUS" // payloadType: 111
CodecG722 = "G722"
CodecMPA = "MPA" // payload: 14
CodecELD = "ELD" // AAC-ELD
)
const PayloadTypeMP4 byte = 255
func GetKind(name string) string {
switch name {
case CodecH264, CodecH265, CodecVP8, CodecVP9, CodecAV1, CodecJPEG:
return KindVideo
case CodecPCMU, CodecPCMA, CodecAAC, CodecOpus, CodecG722, CodecMPA:
case CodecPCMU, CodecPCMA, CodecAAC, CodecOpus, CodecG722, CodecMPA, CodecELD:
return KindAudio
}
return ""
@@ -127,22 +131,6 @@ type Codec struct {
PayloadType uint8
}
func NewCodec(name string) *Codec {
name = strings.ToUpper(name)
switch name {
case CodecH264, CodecH265, CodecVP8, CodecVP9, CodecAV1, CodecJPEG:
return &Codec{Name: name, ClockRate: 90000}
case CodecPCMU, CodecPCMA:
return &Codec{Name: name, ClockRate: 8000}
case CodecOpus:
return &Codec{Name: name, ClockRate: 48000, Channels: 2}
case "MJPEG":
return &Codec{Name: CodecJPEG, ClockRate: 90000}
}
panic(fmt.Sprintf("unsupported codec: %s", name))
}
func (c *Codec) String() string {
s := fmt.Sprintf("%d %s/%d", c.PayloadType, c.Name, c.ClockRate)
if c.Channels > 0 {
@@ -151,6 +139,10 @@ func (c *Codec) String() string {
return s
}
func (c *Codec) IsMP4() bool {
return c.PayloadType == PayloadTypeMP4
}
func (c *Codec) Clone() *Codec {
clone := *c
return &clone
@@ -197,13 +189,19 @@ func MarshalSDP(medias []*Media) ([]byte, error) {
}
codec := media.Codecs[0]
name := codec.Name
if name == CodecELD {
name = CodecAAC
}
md := &sdp.MediaDescription{
MediaName: sdp.MediaName{
Media: media.Kind,
Protos: []string{"RTP", "AVP"},
},
}
md.WithCodec(payloadType, codec.Name, codec.ClockRate, codec.Channels, codec.FmtpLine)
md.WithCodec(payloadType, name, codec.ClockRate, codec.Channels, codec.FmtpLine)
sd.MediaDescriptions = append(sd.MediaDescriptions, md)

View File

@@ -13,7 +13,7 @@ type Track struct {
Codec *Codec
Direction string
sink map[*Track]WriterFunc
sinkMu sync.Mutex
sinkMu sync.RWMutex
}
func (t *Track) String() string {
@@ -23,11 +23,11 @@ func (t *Track) String() string {
}
func (t *Track) WriteRTP(p *rtp.Packet) error {
t.sinkMu.Lock()
t.sinkMu.RLock()
for _, f := range t.sink {
_ = f(p)
}
t.sinkMu.Unlock()
t.sinkMu.RUnlock()
return nil
}
@@ -59,7 +59,7 @@ func (t *Track) GetSink(from *Track) {
}
func (t *Track) HasSink() bool {
t.sinkMu.Lock()
defer t.sinkMu.Unlock()
t.sinkMu.RLock()
defer t.sinkMu.RUnlock()
return len(t.sink) > 0
}

View File

@@ -4,6 +4,7 @@ import (
"fmt"
"github.com/AlexxIT/go2rtc/pkg/streamer"
"github.com/pion/webrtc/v3"
"sort"
)
const (
@@ -100,6 +101,12 @@ func (c *Conn) SetOffer(offer string) (err error) {
}
rawSDP := []byte(c.Conn.RemoteDescription().SDP)
c.medias, err = streamer.UnmarshalSDP(rawSDP)
// sort medias, so video will always be before audio
sort.Slice(c.medias, func(i, j int) bool {
return c.medias[i].Kind == streamer.KindVideo
})
return
}

View File

@@ -57,7 +57,7 @@ func (c *Conn) AddTrack(media *streamer.Media, track *streamer.Track) *streamer.
wrapper := h264.RTPPay(1200)
push = wrapper(push)
if h264.IsAVC(codec) {
if codec.IsMP4() {
wrapper = h264.RepairAVC(track)
} else {
wrapper = h264.RTPDepay(track)

View File

@@ -53,3 +53,6 @@ pc.ontrack = ev => {
- https://www.webrtc-experiment.com/DetectRTC/
- https://divtable.com/table-styler/
- https://www.chromium.org/audio-video/
- https://web.dev/i18n/en/fast-playback-with-preload/#manual_buffering
- https://developer.mozilla.org/en-US/docs/Web/API/Media_Source_Extensions_API
- https://chromium.googlesource.com/external/w3c/web-platform-tests/+/refs/heads/master/media-source/mediasource-is-type-supported.html

View File

@@ -25,93 +25,75 @@
<!-- muted is important for autoplay -->
<video id="video" autoplay controls playsinline muted></video>
<script>
const video = document.querySelector('#video');
// support api_path
const baseUrl = location.origin + location.pathname.substr(
0, location.pathname.lastIndexOf("/")
);
const ws = new WebSocket(`ws${baseUrl.substr(4)}/api/ws${location.search}`);
ws.binaryType = "arraybuffer";
const video = document.querySelector('#video');
let mediaSource;
function init() {
let mediaSource, sourceBuffer, queueBuffer = [];
ws.onopen = () => {
console.log("Start WS");
const ws = new WebSocket(`ws${baseUrl.substr(4)}/api/ws${location.search}`);
ws.binaryType = "arraybuffer";
// https://web.dev/i18n/en/fast-playback-with-preload/#manual_buffering
// https://developer.mozilla.org/en-US/docs/Web/API/Media_Source_Extensions_API
mediaSource = new MediaSource();
video.src = URL.createObjectURL(mediaSource);
mediaSource.onsourceopen = () => {
console.debug("mediaSource.onsourceopen");
mediaSource.onsourceopen = null;
URL.revokeObjectURL(video.src);
ws.send(JSON.stringify({"type": "mse"}));
ws.onopen = () => {
mediaSource = new MediaSource();
video.src = URL.createObjectURL(mediaSource);
mediaSource.onsourceopen = () => {
mediaSource.onsourceopen = null;
URL.revokeObjectURL(video.src);
ws.send(JSON.stringify({"type": "mse"}));
};
};
};
let sourceBuffer, queueBuffer = [];
ws.onmessage = ev => {
if (typeof ev.data === 'string') {
const data = JSON.parse(ev.data);
console.debug("ws.onmessage", data);
ws.onmessage = ev => {
if (typeof ev.data === 'string') {
const data = JSON.parse(ev.data);
console.debug("ws.onmessage", data);
if (data.type === "mse") {
sourceBuffer = mediaSource.addSourceBuffer(data.value);
// important: segments supports TrackFragDecodeTime
// sequence supports only TrackFragRunEntry Duration
sourceBuffer.mode = "segments";
sourceBuffer.onupdateend = () => {
if (!sourceBuffer.updating && queueBuffer.length > 0) {
sourceBuffer.appendBuffer(queueBuffer.shift());
if (data.type === "mse") {
sourceBuffer = mediaSource.addSourceBuffer(data.value);
sourceBuffer.mode = "segments"; // segments or sequence
sourceBuffer.onupdateend = () => {
if (!sourceBuffer.updating && queueBuffer.length > 0) {
try {
sourceBuffer.appendBuffer(queueBuffer.shift());
} catch (e) {
// console.warn(e);
}
}
}
}
}
} else {
if (sourceBuffer.updating) {
queueBuffer.push(ev.data)
} else if (sourceBuffer.updating || queueBuffer.length > 0) {
queueBuffer.push(ev.data);
} else {
sourceBuffer.appendBuffer(ev.data);
try {
sourceBuffer.appendBuffer(ev.data);
} catch (e) {
// console.warn(e);
}
}
if (video.seekable.length > 0) {
const delay = video.seekable.end(video.seekable.length - 1) - video.currentTime;
if (delay < 1) {
video.playbackRate = 1;
} else if (delay > 10) {
video.playbackRate = 10;
} else if (delay > 2) {
video.playbackRate = Math.floor(delay);
}
}
}
video.onpause = () => {
ws.close();
setTimeout(init, 0);
}
}
let offsetTime = 1, noWaiting = 0;
setInterval(() => {
if (video.paused || video.seekable.length === 0) return;
if (noWaiting < 0) {
offsetTime = Math.min(offsetTime * 1.1, 5);
console.debug("offset time up:", offsetTime);
} else if (noWaiting >= 30) {
noWaiting = 0;
offsetTime = Math.max(offsetTime * 0.9, 0.5);
console.debug("offset time down:", offsetTime);
}
noWaiting += 1;
const endTime = video.seekable.end(video.seekable.length - 1);
let playbackRate = (endTime - video.currentTime) / offsetTime;
if (playbackRate < 0.1) {
// video.currentTime = endTime - offsetTime;
playbackRate = 0.1;
} else if (playbackRate > 10) {
// video.currentTime = endTime - offsetTime;
playbackRate = 10;
}
// https://github.com/GoogleChrome/developer.chrome.com/issues/135
video.playbackRate = playbackRate;
}, 1000);
video.onwaiting = () => {
const endTime = video.seekable.end(video.seekable.length - 1);
video.currentTime = endTime - offsetTime;
noWaiting = -1;
}
init();
</script>
</body>
</html>

View File

@@ -25,12 +25,12 @@
<body>
<video id="video" autoplay controls playsinline muted></video>
<script>
const baseUrl = location.origin + location.pathname.substr(
0, location.pathname.lastIndexOf("/")
);
function init(stream) {
// support api_path
const baseUrl = location.origin + location.pathname.substr(
0, location.pathname.lastIndexOf("/")
);
const ws = new WebSocket(`ws${baseUrl.substr(4)}/api/ws${location.search}`);
ws.onopen = () => {
console.debug('ws.onopen');