feat: add controlled ONVIF and RTSP admission (#24)

This commit is contained in:
QiuSW
2026-08-12 17:14:34 +08:00
parent 5b8f4a4c8c
commit 4f17cca91f
18 changed files with 713 additions and 0 deletions
@@ -0,0 +1,109 @@
package onvif
import (
"bytes"
"context"
"encoding/xml"
"fmt"
"io"
"net/http"
"net/url"
"strings"
"time"
)
type Credential struct {
Username string
Password string
}
type Profile struct {
Token string `json:"token"`
Name string `json:"name"`
Width int `json:"width"`
Height int `json:"height"`
Encoding string `json:"encoding"`
StreamURI string `json:"stream_uri"`
}
type Client interface {
Profiles(context.Context, string, Credential) ([]Profile, error)
}
type HTTPClient struct{ client *http.Client }
func NewHTTPClient(timeout time.Duration) *HTTPClient {
if timeout <= 0 {
timeout = 8 * time.Second
}
return &HTTPClient{client: &http.Client{Timeout: timeout}}
}
func (c *HTTPClient) Profiles(ctx context.Context, address string, credential Credential) ([]Profile, error) {
endpoint, err := validateEndpoint(address)
if err != nil {
return nil, err
}
body := `<?xml version="1.0"?><s:Envelope xmlns:s="http://www.w3.org/2003/05/soap-envelope"><s:Body><GetProfiles xmlns="http://www.onvif.org/ver10/media/wsdl"/></s:Body></s:Envelope>`
data, err := c.soap(ctx, endpoint, credential, body)
if err != nil {
return nil, err
}
profiles, err := ParseProfiles(data)
if err != nil {
return nil, err
}
for i := range profiles {
uriBody := fmt.Sprintf(`<?xml version="1.0"?><s:Envelope xmlns:s="http://www.w3.org/2003/05/soap-envelope"><s:Body><GetStreamUri xmlns="http://www.onvif.org/ver10/media/wsdl"><StreamSetup><Stream xmlns="http://www.onvif.org/ver10/schema">RTP-Unicast</Stream><Transport xmlns="http://www.onvif.org/ver10/schema"><Protocol>RTSP</Protocol></Transport></StreamSetup><ProfileToken>%s</ProfileToken></GetStreamUri></s:Body></s:Envelope>`, xmlEscape(profiles[i].Token))
streamData, err := c.soap(ctx, endpoint, credential, uriBody)
if err != nil {
return nil, err
}
profiles[i].StreamURI, err = ParseStreamURI(streamData)
if err != nil {
return nil, err
}
}
return profiles, nil
}
func (c *HTTPClient) soap(ctx context.Context, endpoint string, credential Credential, body string) ([]byte, error) {
req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, bytes.NewBufferString(body))
if err != nil {
return nil, err
}
req.Header.Set("Content-Type", "application/soap+xml; charset=utf-8")
if credential.Username != "" {
req.SetBasicAuth(credential.Username, credential.Password)
}
res, err := c.client.Do(req)
if err != nil {
return nil, fmt.Errorf("onvif request failed: %w", err)
}
defer res.Body.Close()
data, err := io.ReadAll(io.LimitReader(res.Body, 2<<20))
if err != nil {
return nil, err
}
if res.StatusCode == http.StatusUnauthorized {
return nil, fmt.Errorf("authentication_failed")
}
if res.StatusCode < 200 || res.StatusCode >= 300 {
return nil, fmt.Errorf("onvif_http_%d", res.StatusCode)
}
return data, nil
}
func validateEndpoint(value string) (string, error) {
parsed, err := url.Parse(value)
if err != nil || parsed.Scheme == "" || parsed.Host == "" {
return "", fmt.Errorf("invalid ONVIF address")
}
if parsed.User != nil {
return "", fmt.Errorf("credentials must not appear in ONVIF address")
}
if parsed.Scheme != "http" && parsed.Scheme != "https" {
return "", fmt.Errorf("unsupported ONVIF scheme")
}
return parsed.String(), nil
}
func xmlEscape(value string) string {
var b strings.Builder
_ = xml.EscapeText(&b, []byte(value))
return b.String()
}
@@ -0,0 +1,80 @@
package onvif
import (
"context"
"fmt"
"net"
"strings"
"time"
)
const discoveryProbe = `<?xml version="1.0"?><e:Envelope xmlns:e="http://www.w3.org/2003/05/soap-envelope" xmlns:w="http://schemas.xmlsoap.org/ws/2004/08/addressing" xmlns:d="http://schemas.xmlsoap.org/ws/2005/04/discovery" xmlns:dn="http://www.onvif.org/ver10/network/wsdl"><e:Header><w:MessageID>uuid:sense-controlled-discovery</w:MessageID><w:To e:mustUnderstand="true">urn:schemas-xmlsoap-org:ws:2005:04:discovery</w:To><w:Action e:mustUnderstand="true">http://schemas.xmlsoap.org/ws/2005/04/discovery/Probe</w:Action></e:Header><e:Body><d:Probe><d:Types>dn:NetworkVideoTransmitter</d:Types></d:Probe></e:Body></e:Envelope>`
func Discover(ctx context.Context, localIP string, timeout time.Duration) ([]string, error) {
ip := net.ParseIP(localIP)
if ip == nil {
return nil, fmt.Errorf("approved discovery interface is not configured")
}
approved := false
interfaces, _ := net.Interfaces()
for _, iface := range interfaces {
addresses, _ := iface.Addrs()
for _, address := range addresses {
if strings.HasPrefix(address.String(), ip.String()+"/") {
approved = true
}
}
}
if !approved {
return nil, fmt.Errorf("approved discovery IP is not a local interface")
}
connection, err := net.ListenUDP("udp4", &net.UDPAddr{IP: ip, Port: 0})
if err != nil {
return nil, err
}
defer connection.Close()
deadline := time.Now().Add(timeout)
_ = connection.SetDeadline(deadline)
target := &net.UDPAddr{IP: net.ParseIP("239.255.255.250"), Port: 3702}
if _, err := connection.WriteToUDP([]byte(discoveryProbe), target); err != nil {
return nil, err
}
found := map[string]bool{}
buffer := make([]byte, 65535)
for {
select {
case <-ctx.Done():
return keys(found), nil
default:
}
n, _, err := connection.ReadFromUDP(buffer)
if err != nil {
if networkErr, ok := err.(net.Error); ok && networkErr.Timeout() {
return keys(found), nil
}
return nil, err
}
for _, value := range extractXAddrs(string(buffer[:n])) {
found[value] = true
}
}
}
func extractXAddrs(value string) []string {
startTag, endTag := "<d:XAddrs>", "</d:XAddrs>"
start := strings.Index(value, startTag)
if start < 0 {
return nil
}
end := strings.Index(value[start:], endTag)
if end < 0 {
return nil
}
return strings.Fields(value[start+len(startTag) : start+end])
}
func keys(values map[string]bool) []string {
result := make([]string, 0, len(values))
for value := range values {
result = append(result, value)
}
return result
}
@@ -0,0 +1,56 @@
package onvif
import (
"encoding/xml"
"fmt"
"strings"
)
type profileEnvelope struct {
Profiles []struct {
Token string `xml:"token,attr"`
Name string `xml:"Name"`
Encoder struct {
Encoding string `xml:"Encoding"`
Resolution struct {
Width int `xml:"Width"`
Height int `xml:"Height"`
} `xml:"Resolution"`
} `xml:"VideoEncoderConfiguration"`
} `xml:"Body>GetProfilesResponse>Profiles"`
}
func ParseProfiles(data []byte) ([]Profile, error) {
var envelope profileEnvelope
if err := xml.Unmarshal(data, &envelope); err != nil {
return nil, fmt.Errorf("parse ONVIF profiles: %w", err)
}
if len(envelope.Profiles) == 0 {
return nil, fmt.Errorf("no_profiles")
}
result := make([]Profile, 0, len(envelope.Profiles))
for _, value := range envelope.Profiles {
result = append(result, Profile{Token: value.Token, Name: value.Name, Width: value.Encoder.Resolution.Width, Height: value.Encoder.Resolution.Height, Encoding: value.Encoder.Encoding})
}
return result, nil
}
func ParseStreamURI(data []byte) (string, error) {
decoder := xml.NewDecoder(strings.NewReader(string(data)))
for {
token, err := decoder.Token()
if err != nil {
return "", fmt.Errorf("stream_uri_not_found")
}
start, ok := token.(xml.StartElement)
if ok && start.Name.Local == "Uri" {
var value string
if err := decoder.DecodeElement(&value, &start); err != nil {
return "", err
}
if strings.Contains(value, "@") {
return "", fmt.Errorf("stream_uri_contains_credentials")
}
return value, nil
}
}
}
@@ -0,0 +1,31 @@
package onvif
import (
"os"
"path/filepath"
"testing"
)
func TestParseFixtureProfilesAndStreamURI(t *testing.T) {
profilesData, err := os.ReadFile(filepath.Join("..", "..", "..", "..", "..", "testdata", "onvif", "profiles.xml"))
if err != nil {
t.Fatal(err)
}
profiles, err := ParseProfiles(profilesData)
if err != nil || len(profiles) != 2 || profiles[0].Width != 1920 {
t.Fatalf("profiles=%#v err=%v", profiles, err)
}
uriData, err := os.ReadFile(filepath.Join("..", "..", "..", "..", "..", "testdata", "onvif", "stream-uri.xml"))
if err != nil {
t.Fatal(err)
}
uri, err := ParseStreamURI(uriData)
if err != nil || uri != "rtsp://camera.invalid/live/main" {
t.Fatalf("uri=%q err=%v", uri, err)
}
}
func TestRejectCredentialInStreamURI(t *testing.T) {
if _, err := ParseStreamURI([]byte(`<Uri>rtsp://user:pass@camera.invalid/live</Uri>`)); err == nil {
t.Fatal("credential URI accepted")
}
}
@@ -0,0 +1,74 @@
package rtsp
import (
"bufio"
"context"
"encoding/base64"
"fmt"
"net"
"net/url"
"strings"
"time"
)
type Credential struct {
Username string
Password string
}
type Result struct {
Status string `json:"status"`
LatencyMS int64 `json:"latency_ms"`
Detail string `json:"detail,omitempty"`
}
type Verifier interface {
Verify(context.Context, string, Credential) (Result, error)
}
type NetVerifier struct{ Timeout time.Duration }
func (v NetVerifier) Verify(ctx context.Context, raw string, credential Credential) (Result, error) {
parsed, err := url.Parse(raw)
if err != nil || parsed.Scheme != "rtsp" || parsed.Host == "" {
return Result{}, fmt.Errorf("invalid_rtsp_uri")
}
if parsed.User != nil {
return Result{}, fmt.Errorf("rtsp_uri_contains_credentials")
}
address := parsed.Host
if !strings.Contains(address, ":") {
address += ":554"
}
timeout := v.Timeout
if timeout <= 0 {
timeout = 5 * time.Second
}
dialer := net.Dialer{Timeout: timeout}
started := time.Now()
connection, err := dialer.DialContext(ctx, "tcp", address)
if err != nil {
return Result{Status: "unreachable", Detail: "无法连接视频端口"}, nil
}
defer connection.Close()
_ = connection.SetDeadline(time.Now().Add(timeout))
authorization := ""
if credential.Username != "" {
authorization = "Authorization: Basic " + base64.StdEncoding.EncodeToString([]byte(credential.Username+":"+credential.Password)) + "\r\n"
}
request := fmt.Sprintf("OPTIONS %s RTSP/1.0\r\nCSeq: 1\r\nUser-Agent: YoVision-Sense\r\n%s\r\n", parsed.String(), authorization)
if _, err := connection.Write([]byte(request)); err != nil {
return Result{}, err
}
line, err := bufio.NewReader(connection).ReadString('\n')
if err != nil {
return Result{Status: "timeout", Detail: "等待视频响应超时"}, nil
}
status := "ready"
detail := "码流可访问"
if strings.Contains(line, " 401 ") {
status = "authentication_failed"
detail = "设备拒绝了当前凭据"
} else if !strings.Contains(line, " 200 ") {
status = "failed"
detail = "设备返回非成功状态"
}
return Result{Status: status, LatencyMS: time.Since(started).Milliseconds(), Detail: detail}, nil
}
+49
View File
@@ -0,0 +1,49 @@
package admission
import (
"net/http"
"yovision.local/sense/app/sense/identity"
"yovision.local/sense/internal/platform"
)
type Module struct{ service *Service }
func NewModule(service *Service) *Module { return &Module{service: service} }
func (m *Module) Register(app *platform.App) {
app.Handle("GET /api/v1/admission/discover", identity.Require(identity.PermissionDeviceWrite, http.HandlerFunc(m.discover)))
app.Handle("POST /api/v1/admission/probe", identity.Require(identity.PermissionDeviceWrite, http.HandlerFunc(m.probe)))
app.Handle("GET /api/v1/admission/devices/{id}", identity.Require(identity.PermissionDeviceRead, http.HandlerFunc(m.get)))
}
func (m *Module) discover(w http.ResponseWriter, r *http.Request) {
items, err := m.service.Discover(r.Context())
if err != nil {
platform.WriteError(w, &platform.APIError{Status: http.StatusServiceUnavailable, Code: "discovery_unavailable", Message: "未配置获准的发现网卡"})
return
}
platform.WriteJSON(w, http.StatusOK, map[string]any{"items": items})
}
func (m *Module) probe(w http.ResponseWriter, r *http.Request) {
var request struct {
DeviceID string `json:"device_id"`
Address string `json:"address"`
}
if err := platform.DecodeJSON(r, &request); err != nil {
platform.WriteError(w, err)
return
}
actor, _ := identity.PrincipalFromContext(r.Context())
result, err := m.service.Probe(r.Context(), actor, request.DeviceID, request.Address)
if err != nil {
platform.WriteError(w, &platform.APIError{Status: http.StatusBadRequest, Code: "admission_failed", Message: err.Error()})
return
}
platform.WriteJSON(w, http.StatusOK, result)
}
func (m *Module) get(w http.ResponseWriter, r *http.Request) {
result, ok := m.service.Get(r.PathValue("id"))
if !ok {
platform.WriteError(w, &platform.APIError{Status: http.StatusNotFound, Code: "admission_not_found", Message: "尚未执行接入验证"})
return
}
platform.WriteJSON(w, http.StatusOK, result)
}
+119
View File
@@ -0,0 +1,119 @@
package admission
import (
"context"
"fmt"
"sort"
"strings"
"sync"
"time"
"yovision.local/sense/app/sense/adapters/onvif"
"yovision.local/sense/app/sense/adapters/rtsp"
"yovision.local/sense/app/sense/device"
"yovision.local/sense/app/sense/identity"
)
type Profile struct {
Token string `json:"token"`
Name string `json:"name"`
Width int `json:"width"`
Height int `json:"height"`
Encoding string `json:"encoding"`
StreamURI string `json:"stream_uri"`
Kind string `json:"kind"`
Verification rtsp.Result `json:"verification"`
}
type Result struct {
DeviceID string `json:"device_id"`
Address string `json:"address"`
Status string `json:"status"`
Detail string `json:"detail"`
Profiles []Profile `json:"profiles"`
CheckedAt time.Time `json:"checked_at"`
}
type Service struct {
onvif onvif.Client
rtsp rtsp.Verifier
discoveryIP string
discoveryTimeout time.Duration
mu sync.RWMutex
results map[string]Result
now func() time.Time
}
func NewService(client onvif.Client, verifier rtsp.Verifier, discoveryIP string) *Service {
return &Service{onvif: client, rtsp: verifier, discoveryIP: discoveryIP, discoveryTimeout: 3 * time.Second, results: map[string]Result{}, now: time.Now}
}
func (s *Service) Discover(ctx context.Context) ([]string, error) {
if strings.TrimSpace(s.discoveryIP) == "" {
return nil, fmt.Errorf("discovery_not_configured")
}
return onvif.Discover(ctx, s.discoveryIP, s.discoveryTimeout)
}
func (s *Service) Probe(ctx context.Context, actor identity.Principal, deviceID, address string) (Result, error) {
credential, err := device.ReadCredential(ctx, deviceID)
if err != nil {
return Result{}, fmt.Errorf("credential_required")
}
profiles, err := s.onvif.Profiles(ctx, address, onvif.Credential{Username: credential.Username, Password: credential.Password})
if err != nil {
status, detail := classify(err)
result := Result{DeviceID: deviceID, Address: address, Status: status, Detail: detail, CheckedAt: s.now().UTC()}
s.save(result)
identity.RecordAudit(ctx, actor.UserID, "admission.probe", deviceID, "failure", map[string]any{"status": status})
return result, nil
}
items := make([]Profile, 0, len(profiles))
for _, profile := range profiles {
verification, verifyErr := s.rtsp.Verify(ctx, profile.StreamURI, rtsp.Credential{Username: credential.Username, Password: credential.Password})
if verifyErr != nil {
verification = rtsp.Result{Status: "failed", Detail: "视频地址格式不正确"}
}
kind := "other"
items = append(items, Profile{Token: profile.Token, Name: profile.Name, Width: profile.Width, Height: profile.Height, Encoding: profile.Encoding, StreamURI: profile.StreamURI, Kind: kind, Verification: verification})
}
sort.Slice(items, func(i, j int) bool { return items[i].Width*items[i].Height > items[j].Width*items[j].Height })
if len(items) > 0 {
items[0].Kind = "main"
}
if len(items) > 1 {
items[len(items)-1].Kind = "sub"
}
status := "ready"
detail := "设备与视频 Profile 已验证"
for _, item := range items {
if item.Verification.Status != "ready" {
status = "profile_failed"
detail = "部分视频 Profile 验证失败"
}
}
result := Result{DeviceID: deviceID, Address: address, Status: status, Detail: detail, Profiles: items, CheckedAt: s.now().UTC()}
s.save(result)
identity.RecordAudit(ctx, actor.UserID, "admission.probe", deviceID, "success", map[string]any{"profile_count": len(items), "status": status})
return result, nil
}
func (s *Service) Get(deviceID string) (Result, bool) {
s.mu.RLock()
defer s.mu.RUnlock()
value, ok := s.results[deviceID]
return value, ok
}
func (s *Service) save(result Result) {
s.mu.Lock()
defer s.mu.Unlock()
s.results[result.DeviceID] = result
}
func classify(err error) (string, string) {
value := strings.ToLower(err.Error())
switch {
case strings.Contains(value, "authentication"):
return "authentication_failed", "设备拒绝了当前凭据,请更新后重试"
case strings.Contains(value, "deadline") || strings.Contains(value, "timeout"):
return "timeout", "设备响应超时"
case strings.Contains(value, "clock") || strings.Contains(value, "time"):
return "clock_skew", "设备时间可能不准确,请校时后重试"
default:
return "unreachable", "无法读取设备信息,请检查地址和网络"
}
}
@@ -0,0 +1,46 @@
package admission
import (
"context"
"encoding/base64"
"testing"
"yovision.local/sense/app/sense/adapters/onvif"
"yovision.local/sense/app/sense/adapters/rtsp"
"yovision.local/sense/app/sense/device"
"yovision.local/sense/app/sense/identity"
"yovision.local/sense/internal/platform"
)
type fakeONVIF struct{}
func (fakeONVIF) Profiles(context.Context, string, onvif.Credential) ([]onvif.Profile, error) {
return []onvif.Profile{{Token: "main", Name: "主码流", Width: 1920, Height: 1080, Encoding: "H264", StreamURI: "rtsp://camera.invalid/main"}, {Token: "sub", Name: "子码流", Width: 640, Height: 360, Encoding: "H264", StreamURI: "rtsp://camera.invalid/sub"}}, nil
}
type fakeRTSP struct{}
func (fakeRTSP) Verify(context.Context, string, rtsp.Credential) (rtsp.Result, error) {
return rtsp.Result{Status: "ready"}, nil
}
func TestProbeProfilesWithoutCredentialURI(t *testing.T) {
vault, err := device.NewCredentialVault(base64.StdEncoding.EncodeToString([]byte("0123456789abcdef0123456789abcdef")), false)
if err != nil {
t.Fatal(err)
}
store := device.NewMemoryStore()
deviceService := device.NewService(store, vault)
module := device.NewModule(deviceService)
module.Register(platform.NewApp(platform.Config{DatabaseMode: platform.DatabaseModeMemory}, nil, nil))
item, err := deviceService.Create(context.Background(), identity.Principal{}, "camera", "", device.ModalityVideo, nil)
if err != nil {
t.Fatal(err)
}
if _, err = deviceService.SetCredential(context.Background(), identity.Principal{}, item.ID, "fixture-user", "fixture-password"); err != nil {
t.Fatal(err)
}
service := NewService(fakeONVIF{}, fakeRTSP{}, "")
result, err := service.Probe(context.Background(), identity.Principal{}, item.ID, "http://camera.invalid/onvif/device_service")
if err != nil || result.Status != "ready" || result.Profiles[0].Kind != "main" || result.Profiles[1].Kind != "sub" {
t.Fatalf("result=%#v err=%v", result, err)
}
}
@@ -0,0 +1,32 @@
package device
import (
"context"
"fmt"
"sync/atomic"
)
type Credential struct {
Username string
Password string
}
var activeService atomic.Pointer[Service]
// ReadCredential is an internal adapter port. Credentials must never be
// returned from HTTP handlers, logged, or placed in a URL.
func ReadCredential(ctx context.Context, id string) (Credential, error) {
service := activeService.Load()
if service == nil {
return Credential{}, fmt.Errorf("device service is not ready")
}
item, err := service.store.Get(ctx, id)
if err != nil {
return Credential{}, err
}
if len(item.CredentialCiphertext) == 0 {
return Credential{}, fmt.Errorf("device credential is not configured")
}
username, password, err := service.vault.Decrypt(item.CredentialCiphertext)
return Credential{Username: username, Password: password}, err
}
+27
View File
@@ -51,3 +51,30 @@ func (v *CredentialVault) Encrypt(username, password string) ([]byte, error) {
}
return gcm.Seal(nonce, nonce, plaintext, []byte("sense-device-credential-v1")), nil
}
func (v *CredentialVault) Decrypt(ciphertext []byte) (string, string, error) {
block, err := aes.NewCipher(v.key)
if err != nil {
return "", "", err
}
gcm, err := cipher.NewGCM(block)
if err != nil {
return "", "", err
}
if len(ciphertext) < gcm.NonceSize() {
return "", "", fmt.Errorf("invalid credential ciphertext")
}
nonce, encrypted := ciphertext[:gcm.NonceSize()], ciphertext[gcm.NonceSize():]
plaintext, err := gcm.Open(nil, nonce, encrypted, []byte("sense-device-credential-v1"))
if err != nil {
return "", "", fmt.Errorf("decrypt device credential: %w", err)
}
var value struct {
Username string `json:"username"`
Password string `json:"password"`
}
if err := json.Unmarshal(plaintext, &value); err != nil {
return "", "", err
}
return value.Username, value.Password, nil
}
+1
View File
@@ -12,6 +12,7 @@ type Module struct{ service *Service }
func NewModule(service *Service) *Module { return &Module{service: service} }
func (m *Module) Register(app *platform.App) {
activeService.Store(m.service)
app.Handle("GET /api/v1/devices", identity.Require(identity.PermissionDeviceRead, http.HandlerFunc(m.list)))
app.Handle("POST /api/v1/devices", identity.Require(identity.PermissionDeviceWrite, http.HandlerFunc(m.create)))
app.Handle("GET /api/v1/devices/{id}", identity.Require(identity.PermissionDeviceRead, http.HandlerFunc(m.get)))
@@ -0,0 +1,18 @@
package sense
import (
"os"
"time"
"yovision.local/sense/app/sense/adapters/onvif"
"yovision.local/sense/app/sense/adapters/rtsp"
"yovision.local/sense/app/sense/admission"
"yovision.local/sense/internal/platform"
)
func init() {
registerModule(func(app *platform.App) error {
service := admission.NewService(onvif.NewHTTPClient(8*time.Second), rtsp.NetVerifier{Timeout: 5 * time.Second}, os.Getenv("SENSE_ONVIF_DISCOVERY_IP"))
admission.NewModule(service).Register(app)
return nil
})
}