feat(sense): complete T-006 five-stream integration

Sense/internal/config/config.go

@@ -7,6 +7,7 @@ import (
 	"net/url"
 	"os"
 	"strconv"
+	"strings"
 	"time"
 )
 
@@ -16,6 +17,7 @@ const (
 	defaultMediaMTXURL     = "http://127.0.0.1:9997"
 	defaultReconcilePeriod = 5 * time.Second
 	defaultProbePeriod     = 10 * time.Second
+	defaultONVIFMode       = "disabled"
 )
 
 type Config struct {
@@ -25,6 +27,10 @@ type Config struct {
 	MediaMTXURL       string
 	ReconcileInterval time.Duration
 	ProbeInterval     time.Duration
+	ONVIFMode         string
+	RTSPRewriteHost   string
+	RTSPRewritePort   int
+	RTSPStripQuery    bool
 }
 
 func Load() (Config, error) {
@@ -40,6 +46,14 @@ func Load() (Config, error) {
 	if err != nil {
 		return Config{}, err
 	}
+	rewritePort, err := intEnv("SENSE_ONVIF_RTSP_REWRITE_PORT", 0)
+	if err != nil {
+		return Config{}, err
+	}
+	stripQuery, err := boolEnv("SENSE_ONVIF_RTSP_STRIP_QUERY", false)
+	if err != nil {
+		return Config{}, err
+	}
 
 	cfg := Config{
 		HTTPAddress:       stringEnv("SENSE_HTTP_ADDR", defaultHTTPAddress),
@@ -48,6 +62,10 @@ func Load() (Config, error) {
 		MediaMTXURL:       stringEnv("SENSE_MEDIAMTX_URL", defaultMediaMTXURL),
 		ReconcileInterval: reconcilePeriod,
 		ProbeInterval:     probePeriod,
+		ONVIFMode:         stringEnv("SENSE_ONVIF_MODE", defaultONVIFMode),
+		RTSPRewriteHost:   stringEnv("SENSE_ONVIF_RTSP_REWRITE_HOST", ""),
+		RTSPRewritePort:   rewritePort,
+		RTSPStripQuery:    stripQuery,
 	}
 	if err := cfg.Validate(); err != nil {
 		return Config{}, err
@@ -78,6 +96,18 @@ func (c Config) Validate() error {
 	if c.ReconcileInterval <= 0 || c.ProbeInterval <= 0 {
 		return fmt.Errorf("loop intervals must be positive")
 	}
+	if c.ONVIFMode != "" && c.ONVIFMode != "disabled" && c.ONVIFMode != "standard" {
+		return fmt.Errorf("SENSE_ONVIF_MODE must be disabled or standard")
+	}
+	if c.RTSPRewritePort < 0 || c.RTSPRewritePort > 65535 {
+		return fmt.Errorf("SENSE_ONVIF_RTSP_REWRITE_PORT must be between 0 and 65535")
+	}
+	if c.RTSPRewriteHost != "" {
+		if strings.TrimSpace(c.RTSPRewriteHost) != c.RTSPRewriteHost ||
+			strings.ContainsAny(c.RTSPRewriteHost, "/@") {
+			return fmt.Errorf("invalid SENSE_ONVIF_RTSP_REWRITE_HOST")
+		}
+	}
 	return nil
 }
 
@@ -111,3 +141,15 @@ func durationEnv(name string, fallback time.Duration) (time.Duration, error) {
 	}
 	return parsed, nil
 }
+
+func intEnv(name string, fallback int) (int, error) {
+	value, ok := os.LookupEnv(name)
+	if !ok {
+		return fallback, nil
+	}
+	parsed, err := strconv.Atoi(value)
+	if err != nil {
+		return 0, fmt.Errorf("invalid %s: %w", name, err)
+	}
+	return parsed, nil
+}

Sense/internal/config/config_test.go

@@ -33,3 +33,40 @@ func TestValidateRejectsCredentialsInMediaMTXURL(t *testing.T) {
 		t.Fatal("expected credentials in MediaMTX URL to be rejected")
 	}
 }
+
+func TestValidateONVIFModeAndRewritePort(t *testing.T) {
+	t.Parallel()
+	cfg := Config{
+		HTTPAddress:       "127.0.0.1:8080",
+		DatabaseDSN:       "file:test.db",
+		MediaMTXURL:       "http://127.0.0.1:9997",
+		ReconcileInterval: 1,
+		ProbeInterval:     1,
+		ONVIFMode:         "standard",
+		RTSPRewriteHost:   "127.0.0.1",
+		RTSPRewritePort:   10554,
+	}
+	if err := cfg.Validate(); err != nil {
+		t.Fatalf("valid ONVIF configuration failed: %v", err)
+	}
+	cfg.ONVIFMode = "vendor"
+	if err := cfg.Validate(); err == nil {
+		t.Fatal("unknown ONVIF mode must be rejected")
+	}
+	cfg.ONVIFMode = "standard"
+	cfg.RTSPRewritePort = 65536
+	if err := cfg.Validate(); err == nil {
+		t.Fatal("invalid RTSP rewrite port must be rejected")
+	}
+}
+
+func TestLoadRTSPStripQueryOptIn(t *testing.T) {
+	t.Setenv("SENSE_ONVIF_RTSP_STRIP_QUERY", "true")
+	cfg, err := Load()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !cfg.RTSPStripQuery {
+		t.Fatal("explicit RTSP query stripping was not loaded")
+	}
+}

Sense/internal/onvif/credentials.go

@@ -0,0 +1,62 @@
+package onvif
+
+import (
+	"fmt"
+	"net/url"
+	"os"
+	"regexp"
+	"strings"
+)
+
+type Credentials struct {
+	ONVIFUsername string
+	ONVIFPassword string
+	RTSPUsername  string
+	RTSPPassword  string
+}
+
+type CredentialProvider interface {
+	Resolve(reference string) (Credentials, error)
+}
+
+type EnvCredentials struct {
+	LookupEnv func(string) (string, bool)
+}
+
+var credentialKey = regexp.MustCompile(`^[A-Za-z0-9_-]+$`)
+
+func (p EnvCredentials) Resolve(reference string) (Credentials, error) {
+	parsed, err := url.Parse(reference)
+	if err != nil || parsed.Scheme != "env" || parsed.User != nil || parsed.RawQuery != "" || parsed.Fragment != "" {
+		return Credentials{}, fmt.Errorf("credential reference must use env://<key>")
+	}
+	key := strings.Trim(strings.TrimSpace(parsed.Host+parsed.Path), "/")
+	if !credentialKey.MatchString(key) {
+		return Credentials{}, fmt.Errorf("credential reference contains an invalid key")
+	}
+	lookup := p.LookupEnv
+	if lookup == nil {
+		lookup = os.LookupEnv
+	}
+	prefix := "SENSE_CREDENTIAL_" + strings.ToUpper(strings.ReplaceAll(key, "-", "_"))
+	read := func(suffix string) string {
+		value, _ := lookup(prefix + suffix)
+		return value
+	}
+	result := Credentials{
+		ONVIFUsername: read("_ONVIF_USERNAME"),
+		ONVIFPassword: read("_ONVIF_PASSWORD"),
+		RTSPUsername:  read("_RTSP_USERNAME"),
+		RTSPPassword:  read("_RTSP_PASSWORD"),
+	}
+	if result.ONVIFUsername == "" || result.ONVIFPassword == "" {
+		return Credentials{}, fmt.Errorf("ONVIF credentials are not configured for reference")
+	}
+	if result.RTSPUsername == "" {
+		result.RTSPUsername = result.ONVIFUsername
+	}
+	if result.RTSPPassword == "" {
+		result.RTSPPassword = result.ONVIFPassword
+	}
+	return result, nil
+}

Sense/internal/onvif/http.go

@@ -0,0 +1,342 @@
+package onvif
+
+import (
+	"bytes"
+	"context"
+	"crypto/rand"
+	"crypto/sha1"
+	"encoding/base64"
+	"encoding/xml"
+	"errors"
+	"fmt"
+	"io"
+	"net"
+	"net/http"
+	"net/url"
+	"strconv"
+	"strings"
+	"time"
+)
+
+const (
+	deviceNamespace = "http://www.onvif.org/ver10/device/wsdl"
+	mediaNamespace  = "http://www.onvif.org/ver10/media/wsdl"
+)
+
+type HTTPOptions struct {
+	RTSPRewriteHost string
+	RTSPRewritePort int
+	StripRTSPQuery  bool
+}
+
+type HTTPAdapter struct {
+	credentials CredentialProvider
+	client      *http.Client
+	options     HTTPOptions
+	now         func() time.Time
+	random      io.Reader
+}
+
+func NewHTTPAdapter(credentials CredentialProvider, client *http.Client, options HTTPOptions) *HTTPAdapter {
+	if client == nil {
+		client = &http.Client{Timeout: 10 * time.Second}
+	}
+	return &HTTPAdapter{
+		credentials: credentials,
+		client:      client,
+		options:     options,
+		now:         time.Now,
+		random:      rand.Reader,
+	}
+}
+
+func (a *HTTPAdapter) Probe(ctx context.Context, target Target) (ProbeResult, error) {
+	endpoint, credentials, err := a.target(target)
+	if err != nil {
+		return ProbeResult{}, err
+	}
+	infoBody, err := a.call(ctx, endpoint, deviceNamespace+"/GetDeviceInformation",
+		`<tds:GetDeviceInformation xmlns:tds="`+deviceNamespace+`"/>`, credentials)
+	if err != nil {
+		return ProbeResult{}, err
+	}
+	var info deviceInformationEnvelope
+	if err := xml.Unmarshal(infoBody, &info); err != nil {
+		return ProbeResult{}, invalidResponse("decode device information")
+	}
+
+	servicesBody, err := a.call(ctx, endpoint, deviceNamespace+"/GetServices",
+		`<tds:GetServices xmlns:tds="`+deviceNamespace+`"><tds:IncludeCapability>false</tds:IncludeCapability></tds:GetServices>`, credentials)
+	if err != nil {
+		return ProbeResult{}, err
+	}
+	var services servicesEnvelope
+	if err := xml.Unmarshal(servicesBody, &services); err != nil {
+		return ProbeResult{}, invalidResponse("decode services")
+	}
+	mediaEndpoint, err := externalMediaEndpoint(endpoint, services.Body.Response.Services)
+	if err != nil {
+		return ProbeResult{}, err
+	}
+
+	profilesBody, err := a.call(ctx, mediaEndpoint, mediaNamespace+"/GetProfiles",
+		`<trt:GetProfiles xmlns:trt="`+mediaNamespace+`"/>`, credentials)
+	if err != nil {
+		return ProbeResult{}, err
+	}
+	var profilesResponse profilesEnvelope
+	if err := xml.Unmarshal(profilesBody, &profilesResponse); err != nil {
+		return ProbeResult{}, invalidResponse("decode profiles")
+	}
+	profiles := make([]Profile, 0, len(profilesResponse.Body.Response.Profiles))
+	selectedToken := ""
+	for _, value := range profilesResponse.Body.Response.Profiles {
+		video := value.VideoEncoder != nil
+		profiles = append(profiles, Profile{Token: value.Token, Name: value.Name, VideoEncoder: video})
+		if selectedToken == "" && video && value.Token != "" {
+			selectedToken = value.Token
+		}
+	}
+	if selectedToken == "" {
+		return ProbeResult{}, invalidResponse("no video profile")
+	}
+
+	streamRequest := `<trt:GetStreamUri xmlns:trt="` + mediaNamespace + `" xmlns:tt="http://www.onvif.org/ver10/schema">` +
+		`<trt:StreamSetup><tt:Stream>RTP-Unicast</tt:Stream><tt:Transport><tt:Protocol>RTSP</tt:Protocol></tt:Transport></trt:StreamSetup>` +
+		`<trt:ProfileToken>` + escapeXML(selectedToken) + `</trt:ProfileToken></trt:GetStreamUri>`
+	streamBody, err := a.call(ctx, mediaEndpoint, mediaNamespace+"/GetStreamUri", streamRequest, credentials)
+	if err != nil {
+		return ProbeResult{}, err
+	}
+	var streamResponse streamURIEnvelope
+	if err := xml.Unmarshal(streamBody, &streamResponse); err != nil {
+		return ProbeResult{}, invalidResponse("decode stream URI")
+	}
+	streamURI, err := a.rewriteStreamURI(endpoint, streamResponse.Body.Response.MediaURI.URI, credentials)
+	if err != nil {
+		return ProbeResult{}, err
+	}
+	return ProbeResult{
+		Manufacturer:    info.Body.Response.Manufacturer,
+		Model:           info.Body.Response.Model,
+		FirmwareVersion: info.Body.Response.FirmwareVersion,
+		SerialNumber:    info.Body.Response.SerialNumber,
+		Profiles:        profiles,
+		StreamURI:       streamURI,
+	}, nil
+}
+
+func (a *HTTPAdapter) SetSystemDateAndTime(ctx context.Context, target Target, value time.Time) error {
+	endpoint, credentials, err := a.target(target)
+	if err != nil {
+		return err
+	}
+	utc := value.UTC()
+	body := `<tds:SetSystemDateAndTime xmlns:tds="` + deviceNamespace + `" xmlns:tt="http://www.onvif.org/ver10/schema">` +
+		`<tds:DateTimeType>Manual</tds:DateTimeType><tds:DaylightSavings>false</tds:DaylightSavings>` +
+		`<tds:UTCDateTime><tt:Time><tt:Hour>` + strconv.Itoa(utc.Hour()) + `</tt:Hour><tt:Minute>` + strconv.Itoa(utc.Minute()) +
+		`</tt:Minute><tt:Second>` + strconv.Itoa(utc.Second()) + `</tt:Second></tt:Time><tt:Date><tt:Year>` + strconv.Itoa(utc.Year()) +
+		`</tt:Year><tt:Month>` + strconv.Itoa(int(utc.Month())) + `</tt:Month><tt:Day>` + strconv.Itoa(utc.Day()) +
+		`</tt:Day></tt:Date></tds:UTCDateTime></tds:SetSystemDateAndTime>`
+	_, err = a.call(ctx, endpoint, deviceNamespace+"/SetSystemDateAndTime", body, credentials)
+	return err
+}
+
+func (a *HTTPAdapter) target(target Target) (*url.URL, Credentials, error) {
+	endpoint, err := url.Parse(target.EndpointRef)
+	if err != nil || endpoint.Host == "" || endpoint.User != nil || (endpoint.Scheme != "http" && endpoint.Scheme != "https") {
+		return nil, Credentials{}, invalidResponse("invalid ONVIF endpoint")
+	}
+	credentials, err := a.credentials.Resolve(target.CredentialRef)
+	if err != nil {
+		return nil, Credentials{}, &Error{Code: ErrorAuthentication, Err: err}
+	}
+	return endpoint, credentials, nil
+}
+
+func (a *HTTPAdapter) call(ctx context.Context, endpoint *url.URL, action, body string, credentials Credentials) ([]byte, error) {
+	nonce := make([]byte, 20)
+	if _, err := io.ReadFull(a.random, nonce); err != nil {
+		return nil, &Error{Code: ErrorUnavailable, Err: fmt.Errorf("create authentication nonce")}
+	}
+	created := a.now().UTC().Format("2006-01-02T15:04:05Z")
+	digestInput := append(append(append([]byte{}, nonce...), []byte(created)...), []byte(credentials.ONVIFPassword)...)
+	digest := sha1.Sum(digestInput)
+	envelope := `<?xml version="1.0" encoding="UTF-8"?>` +
+		`<s:Envelope xmlns:s="http://www.w3.org/2003/05/soap-envelope" xmlns:wsse="http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-wssecurity-secext-1.0.xsd" xmlns:wsu="http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-wssecurity-utility-1.0.xsd">` +
+		`<s:Header><wsse:Security s:mustUnderstand="1"><wsse:UsernameToken><wsse:Username>` + escapeXML(credentials.ONVIFUsername) +
+		`</wsse:Username><wsse:Password Type="http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-username-token-profile-1.0#PasswordDigest">` +
+		base64.StdEncoding.EncodeToString(digest[:]) + `</wsse:Password><wsse:Nonce EncodingType="http://docs.oasis-open.org/wss/2004/01/oasis-200401-wss-soap-message-security-1.0#Base64Binary">` +
+		base64.StdEncoding.EncodeToString(nonce) + `</wsse:Nonce><wsu:Created>` + created +
+		`</wsu:Created></wsse:UsernameToken></wsse:Security></s:Header><s:Body>` + body + `</s:Body></s:Envelope>`
+	request, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint.String(), strings.NewReader(envelope))
+	if err != nil {
+		return nil, &Error{Code: ErrorUnavailable, Err: fmt.Errorf("create ONVIF request")}
+	}
+	request.Header.Set("Content-Type", `application/soap+xml; charset=utf-8; action="`+action+`"`)
+	response, err := a.client.Do(request)
+	if err != nil {
+		code := ErrorUnavailable
+		if errors.Is(err, context.DeadlineExceeded) || errors.Is(ctx.Err(), context.DeadlineExceeded) {
+			code = ErrorTimeout
+		}
+		return nil, &Error{Code: code, Err: fmt.Errorf("ONVIF transport failed")}
+	}
+	defer response.Body.Close()
+	responseBody, err := io.ReadAll(io.LimitReader(response.Body, 2<<20))
+	if err != nil {
+		return nil, &Error{Code: ErrorUnavailable, Err: fmt.Errorf("read ONVIF response")}
+	}
+	if response.StatusCode == http.StatusUnauthorized || response.StatusCode == http.StatusForbidden {
+		return nil, &Error{Code: ErrorAuthentication, Err: fmt.Errorf("ONVIF authorization failed")}
+	}
+	if fault := soapFault(responseBody); fault != "" {
+		code := ErrorInvalidReply
+		lower := strings.ToLower(fault)
+		if strings.Contains(lower, "authoriz") || strings.Contains(lower, "notauthorized") {
+			code = ErrorAuthentication
+		}
+		return nil, &Error{Code: code, Err: fmt.Errorf("ONVIF SOAP fault")}
+	}
+	if response.StatusCode != http.StatusOK {
+		return nil, &Error{Code: ErrorUnavailable, Err: fmt.Errorf("ONVIF returned HTTP status %d", response.StatusCode)}
+	}
+	return responseBody, nil
+}
+
+func externalMediaEndpoint(deviceEndpoint *url.URL, services []service) (*url.URL, error) {
+	for _, value := range services {
+		if value.Namespace != mediaNamespace || value.XAddr == "" {
+			continue
+		}
+		mediaEndpoint, err := url.Parse(value.XAddr)
+		if err != nil || mediaEndpoint.Host == "" {
+			return nil, invalidResponse("invalid media service address")
+		}
+		mediaEndpoint.Scheme = deviceEndpoint.Scheme
+		mediaEndpoint.Host = deviceEndpoint.Host
+		mediaEndpoint.User = nil
+		return mediaEndpoint, nil
+	}
+	return nil, invalidResponse("media service is unavailable")
+}
+
+func (a *HTTPAdapter) rewriteStreamURI(deviceEndpoint *url.URL, raw string, credentials Credentials) (string, error) {
+	stream, err := url.Parse(raw)
+	if err != nil || stream.Host == "" || (stream.Scheme != "rtsp" && stream.Scheme != "rtsps") {
+		return "", invalidResponse("invalid stream URI")
+	}
+	host := a.options.RTSPRewriteHost
+	if host == "" {
+		host = deviceEndpoint.Hostname()
+	}
+	port := a.options.RTSPRewritePort
+	if port == 0 {
+		if parsedPort := stream.Port(); parsedPort != "" {
+			value, parseErr := strconv.Atoi(parsedPort)
+			if parseErr != nil {
+				return "", invalidResponse("invalid stream port")
+			}
+			port = value
+		}
+	}
+	if port > 0 {
+		stream.Host = net.JoinHostPort(host, strconv.Itoa(port))
+	} else {
+		stream.Host = host
+	}
+	stream.User = url.UserPassword(credentials.RTSPUsername, credentials.RTSPPassword)
+	if a.options.StripRTSPQuery {
+		stream.RawQuery = ""
+		stream.ForceQuery = false
+	}
+	return stream.String(), nil
+}
+
+func invalidResponse(message string) error {
+	return &Error{Code: ErrorInvalidReply, Err: fmt.Errorf("%s", message)}
+}
+
+func escapeXML(value string) string {
+	var buffer bytes.Buffer
+	_ = xml.EscapeText(&buffer, []byte(value))
+	return buffer.String()
+}
+
+func soapFault(body []byte) string {
+	decoder := xml.NewDecoder(bytes.NewReader(body))
+	inFault := false
+	for {
+		token, err := decoder.Token()
+		if errors.Is(err, io.EOF) {
+			return ""
+		}
+		if err != nil {
+			return ""
+		}
+		switch value := token.(type) {
+		case xml.StartElement:
+			if value.Name.Local == "Fault" {
+				inFault = true
+			}
+			if inFault && (value.Name.Local == "Text" || value.Name.Local == "faultstring") {
+				var message string
+				if decoder.DecodeElement(&message, &value) == nil {
+					return message
+				}
+			}
+		case xml.EndElement:
+			if value.Name.Local == "Fault" {
+				return "SOAP fault"
+			}
+		}
+	}
+}
+
+type deviceInformationEnvelope struct {
+	Body struct {
+		Response struct {
+			Manufacturer    string `xml:"Manufacturer"`
+			Model           string `xml:"Model"`
+			FirmwareVersion string `xml:"FirmwareVersion"`
+			SerialNumber    string `xml:"SerialNumber"`
+		} `xml:"GetDeviceInformationResponse"`
+	} `xml:"Body"`
+}
+
+type service struct {
+	Namespace string `xml:"Namespace"`
+	XAddr     string `xml:"XAddr"`
+}
+
+type servicesEnvelope struct {
+	Body struct {
+		Response struct {
+			Services []service `xml:"Service"`
+		} `xml:"GetServicesResponse"`
+	} `xml:"Body"`
+}
+
+type profileResponse struct {
+	Token        string    `xml:"token,attr"`
+	Name         string    `xml:"Name"`
+	VideoEncoder *struct{} `xml:"VideoEncoderConfiguration"`
+}
+
+type profilesEnvelope struct {
+	Body struct {
+		Response struct {
+			Profiles []profileResponse `xml:"Profiles"`
+		} `xml:"GetProfilesResponse"`
+	} `xml:"Body"`
+}
+
+type streamURIEnvelope struct {
+	Body struct {
+		Response struct {
+			MediaURI struct {
+				URI string `xml:"Uri"`
+			} `xml:"MediaUri"`
+		} `xml:"GetStreamUriResponse"`
+	} `xml:"Body"`
+}

Sense/internal/onvif/http_test.go

@@ -0,0 +1,136 @@
+package onvif
+
+import (
+	"context"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"net/url"
+	"strings"
+	"testing"
+)
+
+type staticCredentials struct {
+	value Credentials
+	err   error
+}
+
+func (s staticCredentials) Resolve(string) (Credentials, error) {
+	return s.value, s.err
+}
+
+func TestHTTPAdapterDiscoversMediaAndRewritesNATStream(t *testing.T) {
+	t.Parallel()
+	server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+		if !strings.Contains(request.Header.Get("Content-Type"), "action=") {
+			t.Fatal("SOAP action is required")
+		}
+		body := ""
+		switch {
+		case strings.Contains(request.Header.Get("Content-Type"), "GetDeviceInformation"):
+			body = `<tds:GetDeviceInformationResponse xmlns:tds="http://www.onvif.org/ver10/device/wsdl"><tds:Manufacturer>HIKVISION</tds:Manufacturer><tds:Model>camera</tds:Model><tds:FirmwareVersion>v1</tds:FirmwareVersion><tds:SerialNumber>serial</tds:SerialNumber></tds:GetDeviceInformationResponse>`
+		case strings.Contains(request.Header.Get("Content-Type"), "GetServices"):
+			body = `<tds:GetServicesResponse xmlns:tds="http://www.onvif.org/ver10/device/wsdl"><tds:Service><tds:Namespace>http://www.onvif.org/ver10/media/wsdl</tds:Namespace><tds:XAddr>http://192.0.2.10/onvif/Media</tds:XAddr></tds:Service></tds:GetServicesResponse>`
+		case strings.Contains(request.Header.Get("Content-Type"), "GetProfiles"):
+			body = `<trt:GetProfilesResponse xmlns:trt="http://www.onvif.org/ver10/media/wsdl"><trt:Profiles token="main"><trt:Name>Main</trt:Name><trt:VideoEncoderConfiguration/></trt:Profiles></trt:GetProfilesResponse>`
+		case strings.Contains(request.Header.Get("Content-Type"), "GetStreamUri"):
+			body = `<trt:GetStreamUriResponse xmlns:trt="http://www.onvif.org/ver10/media/wsdl"><trt:MediaUri><trt:Uri>rtsp://192.0.2.10:554/Streaming/Channels/101?transportmode=unicast&amp;profile=Profile_1</trt:Uri></trt:MediaUri></trt:GetStreamUriResponse>`
+		default:
+			http.Error(writer, "unexpected action", http.StatusBadRequest)
+			return
+		}
+		writer.Header().Set("Content-Type", "application/soap+xml")
+		_, _ = fmt.Fprintf(writer, `<s:Envelope xmlns:s="http://www.w3.org/2003/05/soap-envelope"><s:Body>%s</s:Body></s:Envelope>`, body)
+	}))
+	defer server.Close()
+
+	credentials := Credentials{
+		ONVIFUsername: "onvif-user", ONVIFPassword: "onvif-password",
+		RTSPUsername: "rtsp-user", RTSPPassword: "rtsp-password",
+	}
+	adapter := NewHTTPAdapter(staticCredentials{value: credentials}, server.Client(), HTTPOptions{
+		RTSPRewriteHost: "127.0.0.1", RTSPRewritePort: 10554, StripRTSPQuery: true,
+	})
+	result, err := adapter.Probe(context.Background(), Target{
+		EndpointRef: server.URL + "/onvif/device_service", CredentialRef: "env://camera",
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if result.Manufacturer != "HIKVISION" || result.Model != "camera" || len(result.Profiles) != 1 {
+		t.Fatalf("unexpected probe result: %+v", result)
+	}
+	stream, err := url.Parse(result.StreamURI)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if stream.Host != "127.0.0.1:10554" || stream.Path != "/Streaming/Channels/101" || stream.RawQuery != "" {
+		t.Fatalf("unexpected rewritten stream address: host=%s path=%s", stream.Host, stream.Path)
+	}
+	password, _ := stream.User.Password()
+	if stream.User.Username() != credentials.RTSPUsername || password != credentials.RTSPPassword {
+		t.Fatal("RTSP credentials were not injected")
+	}
+}
+
+func TestHTTPAdapterPreservesRTSPQueryByDefault(t *testing.T) {
+	t.Parallel()
+	adapter := NewHTTPAdapter(staticCredentials{}, nil, HTTPOptions{})
+	streamURI, err := adapter.rewriteStreamURI(
+		&url.URL{Scheme: "http", Host: "camera.example:8008"},
+		"rtsp://192.0.2.10:554/live?profile=main",
+		Credentials{RTSPUsername: "user", RTSPPassword: "secret"},
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+	stream, err := url.Parse(streamURI)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if stream.RawQuery != "profile=main" {
+		t.Fatalf("RTSP query was unexpectedly changed: %q", stream.RawQuery)
+	}
+}
+
+func TestHTTPAdapterMapsAuthorizationFaultWithoutLeakingSecret(t *testing.T) {
+	t.Parallel()
+	server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, _ *http.Request) {
+		writer.WriteHeader(http.StatusBadRequest)
+		_, _ = writer.Write([]byte(`<s:Envelope xmlns:s="http://www.w3.org/2003/05/soap-envelope"><s:Body><s:Fault><s:Reason><s:Text>The action requires authorization</s:Text></s:Reason></s:Fault></s:Body></s:Envelope>`))
+	}))
+	defer server.Close()
+	secret := "not-for-errors"
+	adapter := NewHTTPAdapter(staticCredentials{value: Credentials{
+		ONVIFUsername: "user", ONVIFPassword: secret, RTSPUsername: "user", RTSPPassword: secret,
+	}}, server.Client(), HTTPOptions{})
+	_, err := adapter.Probe(context.Background(), Target{EndpointRef: server.URL, CredentialRef: "env://camera"})
+	if CodeOf(err) != ErrorAuthentication || strings.Contains(err.Error(), secret) {
+		t.Fatalf("expected redacted authentication error, got %v", err)
+	}
+}
+
+func TestEnvCredentialsAndDirectRTSPRouting(t *testing.T) {
+	t.Parallel()
+	values := map[string]string{
+		"SENSE_CREDENTIAL_CAMERA_ONVIF_USERNAME": "onvif",
+		"SENSE_CREDENTIAL_CAMERA_ONVIF_PASSWORD": "onvif-secret",
+		"SENSE_CREDENTIAL_CAMERA_RTSP_USERNAME":  "rtsp",
+		"SENSE_CREDENTIAL_CAMERA_RTSP_PASSWORD":  "rtsp-secret",
+	}
+	provider := EnvCredentials{LookupEnv: func(name string) (string, bool) {
+		value, ok := values[name]
+		return value, ok
+	}}
+	resolved, err := provider.Resolve("env://camera")
+	if err != nil || resolved.RTSPUsername != "rtsp" {
+		t.Fatalf("resolve credentials: %+v err=%v", resolved, err)
+	}
+	router := NewRouter(UnavailableAdapter{}, provider)
+	result, err := router.Probe(context.Background(), Target{
+		EndpointRef: "rtsp://127.0.0.1:8555/synthetic-1",
+	})
+	if err != nil || result.StreamURI != "rtsp://127.0.0.1:8555/synthetic-1" {
+		t.Fatalf("route direct RTSP: %+v err=%v", result, err)
+	}
+}

Sense/internal/onvif/router.go

@@ -0,0 +1,50 @@
+package onvif
+
+import (
+	"context"
+	"fmt"
+	"net/url"
+	"time"
+)
+
+type Router struct {
+	camera      Adapter
+	credentials CredentialProvider
+}
+
+func NewRouter(camera Adapter, credentials CredentialProvider) *Router {
+	return &Router{camera: camera, credentials: credentials}
+}
+
+func (r *Router) Probe(ctx context.Context, target Target) (ProbeResult, error) {
+	endpoint, err := url.Parse(target.EndpointRef)
+	if err != nil || endpoint.Host == "" || endpoint.User != nil {
+		return ProbeResult{}, &Error{Code: ErrorInvalidReply, Err: fmt.Errorf("endpoint reference is invalid")}
+	}
+	switch endpoint.Scheme {
+	case "http", "https":
+		return r.camera.Probe(ctx, target)
+	case "rtsp", "rtsps":
+		if target.CredentialRef != "" {
+			credentials, resolveErr := r.credentials.Resolve(target.CredentialRef)
+			if resolveErr != nil {
+				return ProbeResult{}, &Error{Code: ErrorAuthentication, Err: resolveErr}
+			}
+			endpoint.User = url.UserPassword(credentials.RTSPUsername, credentials.RTSPPassword)
+		}
+		return ProbeResult{
+			Profiles:  []Profile{{Token: "direct", Name: "direct", VideoEncoder: true}},
+			StreamURI: endpoint.String(),
+		}, nil
+	default:
+		return ProbeResult{}, &Error{Code: ErrorInvalidReply, Err: fmt.Errorf("unsupported endpoint scheme")}
+	}
+}
+
+func (r *Router) SetSystemDateAndTime(ctx context.Context, target Target, value time.Time) error {
+	endpoint, err := url.Parse(target.EndpointRef)
+	if err != nil || (endpoint.Scheme != "http" && endpoint.Scheme != "https") {
+		return &Error{Code: ErrorInvalidReply, Err: fmt.Errorf("clock sync requires an ONVIF endpoint")}
+	}
+	return r.camera.SetSystemDateAndTime(ctx, target, value)
+}

Sense/internal/probe/probe.go

@@ -15,6 +15,7 @@ const defaultBatchSize = 128
 type Repository interface {
 	ListEnabledVideoDevices(ctx context.Context, limit int) ([]device.Device, error)
 	UpdateActualState(ctx context.Context, id string, state device.ActualState, now time.Time) error
+	RequestReconcile(ctx context.Context, id string, now time.Time) error
 }
 
 type RuntimePaths interface {
@@ -50,7 +51,13 @@ func (c *Checker) RunOnce(ctx context.Context) error {
 		if probeErr == nil && ready {
 			state = device.ActualOnline
 		}
-		if updateErr := c.repository.UpdateActualState(ctx, value.ID, state, c.now().UTC()); updateErr != nil {
+		now := c.now().UTC()
+		if probeErr != nil {
+			if requestErr := c.repository.RequestReconcile(ctx, value.ID, now); requestErr != nil {
+				runErrors = append(runErrors, fmt.Errorf("request device %s reconciliation: %w", value.ID, requestErr))
+			}
+		}
+		if updateErr := c.repository.UpdateActualState(ctx, value.ID, state, now); updateErr != nil {
 			runErrors = append(runErrors, fmt.Errorf("update device %s health: %w", value.ID, updateErr))
 		}
 		if probeErr != nil {

Sense/internal/probe/probe_test.go

@@ -10,8 +10,17 @@ import (
 )
 
 type fakeRepository struct {
-	devices []device.Device
-	states  map[string]device.ActualState
+	devices   []device.Device
+	states    map[string]device.ActualState
+	requested map[string]int
+}
+
+func (f *fakeRepository) RequestReconcile(_ context.Context, id string, _ time.Time) error {
+	if f.requested == nil {
+		f.requested = make(map[string]int)
+	}
+	f.requested[id]++
+	return nil
 }
 
 func (f *fakeRepository) ListEnabledVideoDevices(context.Context, int) ([]device.Device, error) {
@@ -50,4 +59,7 @@ func TestCheckerMapsReadyAndUnavailablePaths(t *testing.T) {
 	if repository.states["online"] != device.ActualOnline || repository.states["offline"] != device.ActualOffline {
 		t.Fatalf("unexpected actual states: %+v", repository.states)
 	}
+	if repository.requested["offline"] != 1 || repository.requested["online"] != 0 {
+		t.Fatalf("unexpected reconcile requests: %+v", repository.requested)
+	}
 }

Sense/internal/store/sqlite.go

@@ -25,6 +25,25 @@ type ReconcileCandidate struct {
 	NextAttempt  *time.Time
 }
 
+type DeviceConvergence struct {
+	ID                 string              `json:"id"`
+	PathName           string              `json:"path_name"`
+	DesiredState       device.DesiredState `json:"desired_state"`
+	ActualState        device.ActualState  `json:"actual_state"`
+	Generation         int64               `json:"generation"`
+	ObservedGeneration int64               `json:"observed_generation"`
+	FailureCount       int                 `json:"failure_count"`
+	NextAttemptAt      *time.Time          `json:"next_attempt_at,omitempty"`
+	LastErrorCode      string              `json:"last_error_code,omitempty"`
+	Converged          bool                `json:"converged"`
+}
+
+type ConvergenceSnapshot struct {
+	Total       int                 `json:"total"`
+	Unconverged int                 `json:"unconverged"`
+	Devices     []DeviceConvergence `json:"devices"`
+}
+
 type SQLite struct {
 	db *sql.DB
 }
@@ -521,6 +540,76 @@ func (s *SQLite) UpdateActualState(ctx context.Context, id string, state device.
 	return nil
 }
 
+// RequestReconcile invalidates the observed generation without changing the
+// desired state or retry backoff. Runtime probes use it when MediaMTX loses a
+// configured path, including after a MediaMTX process restart.
+func (s *SQLite) RequestReconcile(ctx context.Context, id string, now time.Time) error {
+	result, err := s.db.ExecContext(ctx, `
+		UPDATE sense_reconcile_state
+		SET observed_generation = 0, updated_at = ?
+		WHERE device_id = ?`, formatTime(now), id)
+	if err != nil {
+		return fmt.Errorf("request device reconciliation: %w", err)
+	}
+	if affected, _ := result.RowsAffected(); affected != 1 {
+		return ErrNotFound
+	}
+	return nil
+}
+
+// ConvergenceSnapshot returns only identifiers, state and counters. Endpoint
+// and credential references are deliberately excluded from diagnostics.
+func (s *SQLite) ConvergenceSnapshot(ctx context.Context) (ConvergenceSnapshot, error) {
+	rows, err := s.db.QueryContext(ctx, `
+		SELECT d.id, d.path_name, d.desired_state, d.actual_state, d.generation,
+			r.observed_generation, r.failure_count, r.next_attempt_at, r.last_error_code
+		FROM sense_devices d
+		JOIN sense_reconcile_state r ON r.device_id = d.id
+		WHERE d.desired_state = 'enabled'
+			AND EXISTS (
+				SELECT 1 FROM sense_device_capabilities c
+				WHERE c.device_id = d.id AND c.capability = 'video_capture'
+			)
+		ORDER BY d.id`)
+	if err != nil {
+		return ConvergenceSnapshot{}, fmt.Errorf("query convergence snapshot: %w", err)
+	}
+	defer rows.Close()
+	snapshot := ConvergenceSnapshot{Devices: make([]DeviceConvergence, 0)}
+	for rows.Next() {
+		var value DeviceConvergence
+		var nextAttempt, lastError sql.NullString
+		if err := rows.Scan(
+			&value.ID, &value.PathName, &value.DesiredState, &value.ActualState,
+			&value.Generation, &value.ObservedGeneration, &value.FailureCount,
+			&nextAttempt, &lastError,
+		); err != nil {
+			return ConvergenceSnapshot{}, fmt.Errorf("scan convergence snapshot: %w", err)
+		}
+		if nextAttempt.Valid {
+			parsed, parseErr := parseTime(nextAttempt.String)
+			if parseErr != nil {
+				return ConvergenceSnapshot{}, parseErr
+			}
+			value.NextAttemptAt = &parsed
+		}
+		if lastError.Valid {
+			value.LastErrorCode = lastError.String
+		}
+		value.Converged = value.ObservedGeneration == value.Generation &&
+			value.FailureCount == 0 && value.ActualState == device.ActualOnline
+		if !value.Converged {
+			snapshot.Unconverged++
+		}
+		snapshot.Devices = append(snapshot.Devices, value)
+	}
+	if err := rows.Err(); err != nil {
+		return ConvergenceSnapshot{}, fmt.Errorf("iterate convergence snapshot: %w", err)
+	}
+	snapshot.Total = len(snapshot.Devices)
+	return snapshot, nil
+}
+
 const deviceColumns = `d.id, d.tenant_id, d.site_id, d.serial_number, d.name, d.modality,
 		d.desired_state, d.actual_state, d.endpoint_ref, d.credential_ref,
 		d.path_name, d.generation, d.created_at, d.updated_at`

Sense/internal/store/sqlite_test.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"path/filepath"
 	"testing"
+	"time"
 
 	"yovision/sense/internal/device"
 )
@@ -144,6 +145,42 @@ func TestLowerQuotaDoesNotDisableExistingStreams(t *testing.T) {
 	}
 }
 
+func TestConvergenceSnapshotAndRuntimeReconcileRequest(t *testing.T) {
+	t.Parallel()
+	store := openTestStore(t)
+	ctx := context.Background()
+	if err := store.EnsureSite(ctx, device.Site{TenantID: "tenant", ID: "site", Name: "Site"}); err != nil {
+		t.Fatal(err)
+	}
+	if err := store.CreateDevice(ctx, videoDevice(1, "tenant", "site")); err != nil {
+		t.Fatal(err)
+	}
+	now := time.Date(2026, 8, 7, 0, 0, 0, 0, time.UTC)
+	if err := store.MarkReconciled(ctx, "camera-001", 1, now); err != nil {
+		t.Fatal(err)
+	}
+	if err := store.UpdateActualState(ctx, "camera-001", device.ActualOnline, now); err != nil {
+		t.Fatal(err)
+	}
+	snapshot, err := store.ConvergenceSnapshot(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if snapshot.Total != 1 || snapshot.Unconverged != 0 {
+		t.Fatalf("expected converged snapshot, got %+v", snapshot)
+	}
+	if err := store.RequestReconcile(ctx, "camera-001", now.Add(time.Second)); err != nil {
+		t.Fatal(err)
+	}
+	snapshot, err = store.ConvergenceSnapshot(ctx)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if snapshot.Unconverged != 1 || snapshot.Devices[0].ObservedGeneration != 0 {
+		t.Fatalf("runtime loss must invalidate convergence: %+v", snapshot)
+	}
+}
+
 func openTestStore(t *testing.T) *SQLite {
 	t.Helper()
 	dsn := "file:" + filepath.ToSlash(filepath.Join(t.TempDir(), "sense.db"))