Merge pull request #12 from kerberos-io/feature/enhance-ptz-failure-logging

feature/enhance-ptz-failure-logging

.github/workflows/pr-build.yml

@@ -0,0 +1,51 @@
+name: Pull Request Build
+
+# Builds, vets and tests the Go library on every pull request.
+# This repository is a Go library (no Dockerfile), so unlike the Docker-oriented
+# uug-ai/workflows pr-build.yml it validates the module directly instead of
+# building and pushing a container image.
+
+on:
+  pull_request:
+    types: [opened, synchronize, reopened]
+
+permissions:
+  contents: read
+
+jobs:
+  build:
+    name: Build & Test (Go)
+    runs-on: ubuntu-22.04
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+
+      - name: Set up Go
+        uses: actions/setup-go@v5
+        with:
+          go-version: '1.24'
+          check-latest: true
+          cache-dependency-path: go.sum
+
+      - name: Build
+        run: go build -v ./...
+
+      - name: Vet (non-blocking)
+        continue-on-error: true
+        run: go vet ./...
+
+      - name: Gofmt check (non-blocking)
+        continue-on-error: true
+        run: |
+          unformatted="$(gofmt -l .)"
+          if [ -n "$unformatted" ]; then
+            echo "The following files are not gofmt-clean:"
+            echo "$unformatted"
+            exit 1
+          fi
+
+      - name: Test
+        # ws-discovery/TestDevicesFromProbeResponses is a network-dependent
+        # integration test that probes real ONVIF cameras on the LAN, so it
+        # cannot pass on hosted runners. Run it locally against real devices.
+        run: go test $(go list ./... | grep -v '/ws-discovery')

.github/workflows/release-bump.yml

@@ -0,0 +1,28 @@
+name: Bump release
+
+# Manually bump the semantic version: determines the next tag from the latest
+# v* tag, creates a GitHub release with generated notes and exposes the new tag.
+# Thin wrapper around the reusable uug-ai/workflows release-bump workflow.
+
+on:
+  workflow_dispatch:
+    inputs:
+      bump:
+        description: "Which part of the version to bump"
+        required: true
+        default: patch
+        type: choice
+        options:
+          - major
+          - minor
+          - patch
+
+permissions:
+  contents: write
+
+jobs:
+  bump-release:
+    uses: uug-ai/workflows/.github/workflows/release-bump.yml@main
+    with:
+      bump: ${{ github.event.inputs.bump }}
+    secrets: inherit

.github/workflows/release.yml

@@ -0,0 +1,36 @@
+name: Release
+
+# Validates the tagged code when a release is published (build + test).
+# For a Go library a "release" is simply a git tag consumers depend on, so this
+# workflow guards that a released tag builds and passes its tests rather than
+# publishing a container image.
+
+on:
+  release:
+    types: [published]
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+jobs:
+  validate:
+    name: Validate release (Go)
+    runs-on: ubuntu-22.04
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+
+      - name: Set up Go
+        uses: actions/setup-go@v5
+        with:
+          go-version: '1.24'
+          check-latest: true
+          cache-dependency-path: go.sum
+
+      - name: Build
+        run: go build -v ./...
+
+      - name: Test
+        # See pr-build.yml: the ws-discovery probe test needs real cameras.
+        run: go test $(go list ./... | grep -v '/ws-discovery')

.gitignore

@@ -1,3 +1,4 @@
+.idea
 *.iml
 .idea/
 .vscode/

Device.go

@@ -280,21 +280,45 @@ func (dev Device) callMethodDo(endpoint string, method interface{}) (*http.Respo
 	soap.AddRootNamespaces(Xlmns)
 	soap.AddAction()
 
-	//Auth Handling
-	if dev.params.Username != "" && dev.params.Password != "" {
-		soap.AddWSSecurity(dev.params.Username, dev.params.Password)
-	}
+	return dev.sendSOAP(endpoint, soap)
+}
 
-	servResp, err := networking.SendSoap(dev.params.HttpClient, endpoint, soap.String())
-	if err != nil {
-		// Close server response body to reuse the connection
-		if servResp != nil {
-			servResp.Body.Close()
+// sendSOAP dispatches an assembled SOAP message to the endpoint using the
+// authentication mechanism selected through DeviceParams.AuthMode.
+//
+// Behaviour per AuthMode:
+//   - NoAuth ("none"):            no authentication is added.
+//   - UsernameTokenAuth:          WS-Security UsernameToken only; never falls
+//     back to HTTP digest, which keeps cameras that authenticate exclusively
+//     through WS-Security working.
+//   - DigestAuth ("digest"):      HTTP digest only; no WS-Security header.
+//   - Both ("both") / unset (""): WS-Security credentials are added (when
+//     available) and HTTP digest is attempted only if the device answers with
+//     an authentication challenge (HTTP 401 Unauthorized). On that digest
+//     retry the WS-Security header is dropped so the credentials are not sent
+//     twice.
+func (dev Device) sendSOAP(endpoint string, soap gosoap.SoapMessage) (*http.Response, error) {
+	hasCredentials := dev.params.Username != "" || dev.params.Password != ""
+
+	switch dev.params.AuthMode {
+	case NoAuth:
+		return networking.SendSoap(dev.params.HttpClient, endpoint, soap.String())
+
+	case UsernameTokenAuth:
+		if hasCredentials {
+			soap.AddWSSecurity(dev.params.Username, dev.params.Password)
 		}
-		servResp, err = networking.SendSoapWithDigest(dev.params.HttpClient, endpoint, soap.String(), dev.params.Username, dev.params.Password)
-	}
+		return networking.SendSoap(dev.params.HttpClient, endpoint, soap.String())
 
-	return servResp, err
+	case DigestAuth:
+		return networking.SendSoapWithDigest(dev.params.HttpClient, endpoint, soap.String(), dev.params.Username, dev.params.Password)
+
+	default: // Both and the empty/unset default.
+		if hasCredentials {
+			soap.AddWSSecurity(dev.params.Username, dev.params.Password)
+		}
+		return networking.SendSoapWithDigest(dev.params.HttpClient, endpoint, soap.String(), dev.params.Username, dev.params.Password)
+	}
 }
 
 func (dev *Device) GetDeviceParams() DeviceParams {
@@ -405,7 +429,6 @@ func (dev Device) SendSoapWithOptions(endpoint, xmlRequestBody string, opts ...S
 	return servResp, err
 }
 
-
 func createHttpRequest(httpMethod string, endpoint string, soap string) (req *http.Request, err error) {
 	req, err = http.NewRequest(httpMethod, endpoint, bytes.NewBufferString(soap))
 	if err != nil {

event/stream/doc.go

@@ -18,7 +18,8 @@
 //	}
 //
 // NewStream performs network I/O so auth and reachability failures
-// surface synchronously. Events and Errors close when the Stream stops;
+// surface synchronously, and rejects a client timeout that cannot
+// outlast PullTimeout with ErrInvalidOptions. Events and Errors close when the Stream stops;
 // Errors sends are non-blocking so a stalled consumer drops older
 // errors rather than blocking the pull loop. After a silent reconnect,
 // the next batch's events carry Event.AfterReconnect=true.

event/stream/pulltimeout.go

@@ -0,0 +1,51 @@
+package stream
+
+import (
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/kerberos-io/onvif"
+)
+
+// ErrInvalidOptions marks a configuration that cannot succeed. Callers
+// retry the pull/renew/recreate errors; retrying this one never helps,
+// so it is a distinct sentinel they can short-circuit on.
+var ErrInvalidOptions = errors.New("stream: invalid options")
+
+// minClientHeadroom is how far http.Client.Timeout must exceed
+// PullTimeout. The client ceiling covers dial, TLS and the response
+// transfer on top of the poll it has to outlast, and starts before the
+// camera has parsed the request; on a cellular bearer that overhead
+// runs to hundreds of milliseconds.
+const minClientHeadroom = 5 * time.Second
+
+// validateClientTimeout rejects a client ceiling that cannot outlast
+// the PullMessages long-poll plus minClientHeadroom.
+//
+// Zero means unbounded and is accepted: it is the SDK's default when a
+// caller passes no client, so rejecting it would break every default
+// consumer. Note it is not risk-free — the caller interface documents
+// that ctx cannot interrupt an in-flight SOAP call, so only the client
+// timeout can unwedge a stalled camera.
+func validateClientTimeout(clientTimeout, pullTimeout time.Duration) error {
+	if clientTimeout == 0 || clientTimeout >= pullTimeout+minClientHeadroom {
+		return nil
+	}
+	return fmt.Errorf(
+		"%w: http.Client.Timeout (%s) must exceed PullTimeout (%s) by at least %s; PullMessages is a long-poll and the client would abort every quiet pull",
+		ErrInvalidOptions, clientTimeout, pullTimeout, minClientHeadroom)
+}
+
+// clientTimeoutOf reports the device's HTTP client ceiling, or 0 when
+// the SDK is using its own default (unbounded) client.
+func clientTimeoutOf(dev *onvif.Device) time.Duration {
+	if dev == nil {
+		return 0
+	}
+	c := dev.GetDeviceParams().HttpClient
+	if c == nil {
+		return 0
+	}
+	return c.Timeout
+}

event/stream/pulltimeout_test.go

@@ -0,0 +1,68 @@
+package stream
+
+import (
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// TestValidateClientTimeout — PullMessages is a long-poll: the camera
+// holds the connection open for PullTimeout waiting for an event. An
+// http.Client.Timeout covers the whole exchange and starts before the
+// camera has parsed the request, so a client ceiling at or below
+// PullTimeout loses the race on every quiet interval and the pull can
+// only ever fail. This shipped once (both were 5s) and presented as a
+// slow camera rather than a misconfiguration.
+//
+// Strict inequality is not enough: the client also has to cover dial,
+// TLS and the response transfer, which on a cellular bearer runs to
+// hundreds of milliseconds. Hence a real headroom floor.
+func TestValidateClientTimeout(t *testing.T) {
+	tests := []struct {
+		name    string
+		client  time.Duration
+		pull    time.Duration
+		wantErr bool
+	}{
+		{"unbounded client is the caller's risk, not an error", 0, 30 * time.Second, false},
+		{"comfortable headroom", 40 * time.Second, 30 * time.Second, false},
+		{"exactly the minimum headroom", 30*time.Second + minClientHeadroom, 30 * time.Second, false},
+		{"a hair under the minimum headroom", 30*time.Second + minClientHeadroom - time.Millisecond, 30 * time.Second, true},
+		{"strictly greater but no headroom", 30*time.Second + time.Millisecond, 30 * time.Second, true},
+		{"equal timeouts always lose", 5 * time.Second, 5 * time.Second, true},
+		{"client below pull", 4 * time.Second, 30 * time.Second, true},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			err := validateClientTimeout(tt.client, tt.pull)
+			if tt.wantErr {
+				require.Error(t, err, "client=%s pull=%s must be rejected", tt.client, tt.pull)
+				assert.ErrorIs(t, err, ErrInvalidOptions,
+					"callers need a sentinel to tell a permanent misconfiguration from a transient failure")
+				assert.Contains(t, err.Error(), "PullTimeout",
+					"the error must name the option the caller has to change")
+				return
+			}
+			assert.NoError(t, err, "client=%s pull=%s must be accepted", tt.client, tt.pull)
+		})
+	}
+}
+
+// TestErrInvalidOptions_IsDistinctFromStreamErrors — the pull/renew/
+// recreate errors are transient and callers retry them. A bad Options
+// never becomes valid by retrying, so it must not be mistaken for one.
+func TestErrInvalidOptions_IsDistinctFromStreamErrors(t *testing.T) {
+	err := validateClientTimeout(5*time.Second, 5*time.Second)
+	require.Error(t, err)
+
+	var pull ErrPullFailed
+	var renew ErrRenewFailed
+	var recreate ErrRecreateFailed
+	assert.False(t, errors.As(err, &pull))
+	assert.False(t, errors.As(err, &renew))
+	assert.False(t, errors.As(err, &recreate))
+}

event/stream/stream.go

@@ -39,7 +39,9 @@ type Options struct {
 	// server-side filtering is fragile across vendors and empty is
 	// required for AXIS.
 	RawTopicFilter string
-	// PullTimeout — zero means default (5s).
+	// PullTimeout — zero means default (5s). The device's
+	// http.Client.Timeout must exceed this by minClientHeadroom or
+	// NewStream returns ErrInvalidOptions.
 	PullTimeout time.Duration
 	// MessageLimit — zero means default (32). Busy AXIS cameras with
 	// many configured rules can burst beyond 10 per pull.
@@ -234,6 +236,12 @@ func (s *Stream) updateGrantedTerminationIfGen(gen uint64, t time.Time) {
 //
 // The returned Stream stops when ctx is cancelled or Close is called.
 func NewStream(ctx context.Context, dev *onvif.Device, opts Options) (*Stream, error) {
+	// Checked before the subscription call: this config can only fail,
+	// so surfacing it here beats a stream that appears to work and
+	// silently survives on reconnects alone.
+	if err := validateClientTimeout(clientTimeoutOf(dev), opts.withDefaults().PullTimeout); err != nil {
+		return nil, err
+	}
 	return newStream(ctx, deviceCaller{dev: dev}, opts)
 }
 

event/stream/topics.go

@@ -79,10 +79,19 @@ var topicRules = []struct {
 	// https://developer.axis.com/vapix/network-video/event-and-action-services/
 	{"MotionRegionDetector/Motion", KindMotion},
 
-	// AXIS Guard suite — vendor analytics apps with Camera<N>Profile<ID>
-	// suffixes. Treated as motion so they can drive motion-triggered
-	// recording on cameras using these apps instead of basic VMD.
+	// tns1:RuleEngine/tnsaxis:VMD3/vmd3_video_<N> — AXIS VMD 3, the
+	// firmware-builtin predecessor of the VMD 4 ACAP. It lives under
+	// RuleEngine, not CameraApplicationPlatform, so the VMD rule below
+	// does not cover it. Still shipping on deployed cameras.
+	// https://developer.axis.com/vapix/network-video/event-and-action-services/
+	{"RuleEngine/VMD3/", KindMotion},
+
+	// AXIS ACAP motion apps with Camera<N>Profile<ID> suffixes. VMD 4 is
+	// the stock app shipped on the camera; the Guard suite are the
+	// paid analytics products. Prefix-match because of the suffix.
+	// https://developer.axis.com/vapix/applications/vmd4
 	// https://developer.axis.com/vapix/applications/motion-guard
+	{"CameraApplicationPlatform/VMD/", KindMotion},
 	{"CameraApplicationPlatform/MotionGuard/", KindMotion},
 	{"CameraApplicationPlatform/FenceGuard/", KindMotion},
 	{"CameraApplicationPlatform/LoiteringGuard/", KindMotion},

event/stream/topics_test.go

@@ -25,6 +25,18 @@ func TestClassifyTopic(t *testing.T) {
 		{"axis_fence_guard", "tnsaxis:CameraApplicationPlatform/FenceGuard/Camera1ProfileANY", KindMotion},
 		{"axis_loitering_guard", "tnsaxis:CameraApplicationPlatform/LoiteringGuard/Camera1ProfileANY", KindMotion},
 
+		// AXIS VMD 4 — the stock motion app, and the one an installer
+		// reaches for before any Guard product. The profile suffix
+		// varies with the configured VMD profile.
+		{"axis_vmd4_profile_any", "tnsaxis:CameraApplicationPlatform/VMD/Camera1ProfileANY", KindMotion},
+		{"axis_vmd4_profile_numbered", "tnsaxis:CameraApplicationPlatform/VMD/Camera1Profile1", KindMotion},
+
+		// AXIS VMD 3 — the firmware-builtin predecessor, published under
+		// RuleEngine rather than CameraApplicationPlatform. Still
+		// shipping on deployed cameras.
+		{"axis_vmd3_video_1", "tns1:RuleEngine/tnsaxis:VMD3/vmd3_video_1", KindMotion},
+		{"axis_vmd3_video_2", "tns1:RuleEngine/tnsaxis:VMD3/vmd3_video_2", KindMotion},
+
 		// --- Tampering --------------------------------------------------
 
 		{"tamper_detector", "tns1:RuleEngine/TamperDetector/Tamper", KindTampering},
@@ -88,6 +100,16 @@ func TestClassifyTopic(t *testing.T) {
 		{"relay_failure_not_digital_output", "tns1:Device/HardwareFailure/RelayFailure", KindUnknown},
 		{"digital_input_config_not_digital_input", "tns1:Device/IO/DigitalInputConfiguration", KindUnknown},
 		{"tamper_detector_log_not_tampering", "tns1:Device/Diagnostics/TamperDetectorLog", KindUnknown},
+
+		// The two AXIS VMD needles carry a trailing slash so they match a
+		// whole path segment. Without it, any sibling app or rule whose
+		// name merely starts with VMD / VMD3 would classify as motion and
+		// drive recording.
+		{"vmd_statistics_app_not_motion", "tnsaxis:CameraApplicationPlatform/VMDStatistics/Camera1", KindUnknown},
+		{"vmd3_config_rule_not_motion", "tns1:RuleEngine/tnsaxis:VMD3Config/Changed", KindUnknown},
+		// VMD3 is scoped to RuleEngine; the same name under another
+		// container is a different thing.
+		{"vmd3_outside_rule_engine_not_motion", "tnsaxis:Storage/VMD3/Status", KindUnknown},
 	}
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {

networking/digest_test.go

@@ -0,0 +1,147 @@
+package networking
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+)
+
+func TestSendSoapWithDigestReturnsServerError(t *testing.T) {
+	server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, _ *http.Request) {
+		writer.WriteHeader(http.StatusInternalServerError)
+	}))
+	defer server.Close()
+
+	response, err := SendSoapWithDigest(server.Client(), server.URL, "<Envelope/>", "user", "password")
+	if response == nil {
+		t.Fatal("SendSoapWithDigest returned a nil response")
+	}
+	defer response.Body.Close()
+	if err == nil {
+		t.Fatal("SendSoapWithDigest returned nil error for HTTP 500")
+	}
+}
+
+func TestSendSoapWithDigestReturnsServerErrorAfterAuthentication(t *testing.T) {
+	requestCount := 0
+	server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+		requestCount++
+		if requestCount == 1 {
+			writer.Header().Set("WWW-Authenticate", `Digest realm="AXIS", nonce="nonce", qop="auth", algorithm=MD5`)
+			writer.WriteHeader(http.StatusUnauthorized)
+			return
+		}
+		if !strings.HasPrefix(request.Header.Get("Authorization"), "Digest ") {
+			t.Error("authenticated retry is missing Digest Authorization header")
+		}
+		writer.WriteHeader(http.StatusInternalServerError)
+	}))
+	defer server.Close()
+
+	response, err := SendSoapWithDigest(server.Client(), server.URL, "<Envelope/>", "user", "password")
+	if response == nil {
+		t.Fatal("SendSoapWithDigest returned a nil response")
+	}
+	defer response.Body.Close()
+	if err == nil {
+		t.Fatal("SendSoapWithDigest returned nil error for authenticated HTTP 500")
+	}
+	if requestCount != 2 {
+		t.Fatalf("request count = %d, want 2", requestCount)
+	}
+}
+
+func TestParseDigestChallenge(t *testing.T) {
+	challenge := `Digest realm="testrealm@host.com", qop="auth,auth-int", nonce="dcd98b7102dd2f0e8b11d0f600bfb0c093", opaque="5ccc069c403ebaf9f0171e9517f40e41", algorithm=MD5`
+	parts := parseDigestChallenge(challenge)
+
+	cases := map[string]string{
+		"realm":     "testrealm@host.com",
+		"qop":       "auth,auth-int",
+		"nonce":     "dcd98b7102dd2f0e8b11d0f600bfb0c093",
+		"opaque":    "5ccc069c403ebaf9f0171e9517f40e41",
+		"algorithm": "MD5",
+	}
+	for key, want := range cases {
+		if got := parts[key]; got != want {
+			t.Errorf("parseDigestChallenge()[%q] = %q, want %q", key, got, want)
+		}
+	}
+}
+
+// TestMD5DigestVectors validates md5Hex and the response assembly order against
+// the canonical RFC 2617 section 3.5 worked example.
+func TestMD5DigestVectors(t *testing.T) {
+	ha1 := md5Hex("Mufasa:testrealm@host.com:Circle Of Life")
+	if want := "939e7578ed9e3c518a452acee763bce9"; ha1 != want {
+		t.Fatalf("HA1 = %q, want %q", ha1, want)
+	}
+
+	ha2 := md5Hex("GET:/dir/index.html")
+	if want := "39aff3a2bab6126f332b942af96d3366"; ha2 != want {
+		t.Fatalf("HA2 = %q, want %q", ha2, want)
+	}
+
+	response := md5Hex(strings.Join([]string{
+		ha1,
+		"dcd98b7102dd2f0e8b11d0f600bfb0c093", // nonce
+		"00000001",                           // nc
+		"0a4f113b",                           // cnonce
+		"auth",                               // qop
+		ha2,
+	}, ":"))
+	if want := "6629fae49393a05397450978507c4ef1"; response != want {
+		t.Fatalf("response = %q, want %q", response, want)
+	}
+}
+
+// TestNewDigestAuthorizationConsistency builds an Authorization header and then
+// recomputes the response from the emitted cnonce/nc to confirm the header is
+// internally consistent (correct field wiring and formula).
+func TestNewDigestAuthorizationConsistency(t *testing.T) {
+	const (
+		username = "admin"
+		password = "s3cret"
+		realm    = "IP Camera"
+		nonce    = "0cc175b9c0f1b6a831c399e269772661"
+	)
+	challenge := `Digest realm="` + realm + `", nonce="` + nonce + `", qop="auth", algorithm=MD5`
+
+	header := newDigestAuthorization(challenge, "POST", "http://192.168.1.10/onvif/ptz_service", username, password)
+	if header == "" {
+		t.Fatal("newDigestAuthorization returned empty header")
+	}
+	if !strings.HasPrefix(header, "Digest ") {
+		t.Fatalf("header does not start with Digest scheme: %q", header)
+	}
+
+	fields := parseDigestChallenge(header)
+	if fields["username"] != username {
+		t.Errorf("username = %q, want %q", fields["username"], username)
+	}
+	if fields["uri"] != "/onvif/ptz_service" {
+		t.Errorf("uri = %q, want %q", fields["uri"], "/onvif/ptz_service")
+	}
+	if fields["qop"] != "auth" {
+		t.Errorf("qop = %q, want %q", fields["qop"], "auth")
+	}
+
+	ha1 := md5Hex(username + ":" + realm + ":" + password)
+	ha2 := md5Hex("POST:/onvif/ptz_service")
+	want := md5Hex(strings.Join([]string{ha1, nonce, fields["nc"], fields["cnonce"], "auth", ha2}, ":"))
+	if fields["response"] != want {
+		t.Errorf("response = %q, want %q", fields["response"], want)
+	}
+}
+
+// TestNewDigestAuthorizationRejectsUnsupported ensures an empty header is
+// returned when required parameters are missing or the qop is unsupported.
+func TestNewDigestAuthorizationRejectsUnsupported(t *testing.T) {
+	if got := newDigestAuthorization(`Digest realm="r"`, "POST", "http://host/x", "u", "p"); got != "" {
+		t.Errorf("expected empty header when nonce missing, got %q", got)
+	}
+	if got := newDigestAuthorization(`Digest realm="r", nonce="n", qop="auth-int"`, "POST", "http://host/x", "u", "p"); got != "" {
+		t.Errorf("expected empty header for unsupported qop, got %q", got)
+	}
+}

networking/networking.go

@@ -2,90 +2,212 @@ package networking
 
 import (
 	"bytes"
+	"crypto/md5"
+	"crypto/rand"
+	"encoding/hex"
 	"fmt"
-	"io"
 	"net/http"
+	"net/url"
+	"regexp"
+	"strings"
 
 	"github.com/beevik/etree"
-	"github.com/icholy/digest"
 	"github.com/juju/errors"
 )
 
+const soapContentType = "application/soap+xml; charset=utf-8"
+
 // SendSoap send soap message
 func SendSoap(httpClient *http.Client, endpoint, message string) (*http.Response, error) {
-	resp, err := httpClient.Post(endpoint, "application/soap+xml; charset=utf-8", bytes.NewBufferString(message))
+	resp, err := httpClient.Post(endpoint, soapContentType, bytes.NewBufferString(message))
 	if err != nil {
 		return resp, errors.Annotate(err, "Post")
 	}
 
-	// if resp.StatusCode is 4xx,5xx, return error
-	if resp.StatusCode >= 400 && resp.StatusCode < 600 {
-		return resp, errors.Errorf("Server error: %d: %s", resp.StatusCode, resp.Status)
-	}
-
-	return resp, nil
+	return resp, responseError(resp)
 }
 
+func responseError(resp *http.Response) error {
+	if resp.StatusCode >= 400 && resp.StatusCode < 600 {
+		return errors.Errorf("Server error: %d: %s", resp.StatusCode, resp.Status)
+	}
+	return nil
+}
+
+// SendSoapWithDigest sends a soap message and, when the device answers with an
+// HTTP 401 digest challenge, transparently retries the request with the
+// computed HTTP digest Authorization header.
+//
+// Any wsse:Security header present in the message is stripped before sending:
+// when a device requires HTTP digest the credentials travel in the
+// Authorization header, so keeping the WS-Security UsernameToken in the body
+// would put the credentials on the wire twice. Other SOAP header blocks (for
+// example WS-Addressing reference parameters) are preserved so vendor-specific
+// routing keeps working across the retry.
 func SendSoapWithDigest(httpClient *http.Client, endpoint, message, username, password string) (*http.Response, error) {
-	doc := etree.NewDocument()
-	if err := doc.ReadFromString(message); err != nil {
-		return nil, err
+	if httpClient == nil {
+		httpClient = new(http.Client)
 	}
 
-	e := doc.FindElement("./Envelope/Header/Security")
-	if e != nil {
-		bodyTag := doc.Root().SelectElement("Header")
-		bodyTag.RemoveChild(e)
-		data, err := doc.WriteToString()
-		if err != nil {
-			return nil, err
-		}
-		message = data
-	}
+	// Avoid sending the credentials twice (WS-Security + digest) on the retry.
+	message = stripWSSecurityHeader(message)
 
-	req, err := http.NewRequest("POST", endpoint, bytes.NewBufferString(message))
+	resp, err := httpClient.Post(endpoint, soapContentType, bytes.NewBufferString(message))
 	if err != nil {
-		return nil, err
+		return resp, errors.Annotate(err, "Post")
 	}
-	req.Header.Set("Content-Type", "application/soap+xml; charset=utf-8")
-	resp, err := httpClient.Do(req)
+
+	// Only escalate to HTTP digest when the device explicitly asks for it.
+	if resp.StatusCode != http.StatusUnauthorized {
+		return resp, responseError(resp)
+	}
+
+	challenge := resp.Header.Get("WWW-Authenticate")
+	if !strings.HasPrefix(strings.ToLower(strings.TrimSpace(challenge)), "digest") {
+		// Not a digest challenge (e.g. Basic) - nothing more we can do here.
+		return resp, responseError(resp)
+	}
+
+	authorization := newDigestAuthorization(challenge, http.MethodPost, endpoint, username, password)
+	if authorization == "" {
+		return resp, errors.New("unsupported digest challenge")
+	}
+
+	// Release the challenge response before issuing the authenticated retry.
+	resp.Body.Close()
+
+	req, err := http.NewRequest(http.MethodPost, endpoint, bytes.NewBufferString(message))
+	if err != nil {
+		return nil, errors.Annotate(err, "new digest request")
+	}
+	req.Header.Set("Content-Type", soapContentType)
+	req.Header.Set("Authorization", authorization)
+
+	resp, err = httpClient.Do(req)
 	if err != nil {
 		return resp, errors.Annotate(err, "Post with digest")
 	}
 
-	if resp.StatusCode != http.StatusUnauthorized {
-		return resp, err
-	}
-
-	wwwAuth := resp.Header.Get("WWW-Authenticate")
-	chal, err := digest.ParseChallenge(wwwAuth)
-	if err != nil {
-		return resp, fmt.Errorf("fail to parse challenge: %w", err)
-	}
-
-	cred, err := digest.Digest(chal, digest.Options{
-		Method:   "POST",
-		URI:      req.URL.RequestURI(),
-		Username: username,
-		Password: password,
-	})
-
-	if err != nil {
-		return resp, fmt.Errorf("fail to build digest: %w", err)
-	}
-
-	// Readout body to close the connection
-	resp.Body.Close()
-
-	req.Header.Add("Authorization", cred.String())
-	req.Body = io.NopCloser((bytes.NewBufferString(message)))
-	resp, err = httpClient.Do(req)
-	if err != nil {
-		return nil, errors.Annotate(err, "Post with digest")
-	}
-	if resp.StatusCode >= 400 && resp.StatusCode < 600 {
-		return resp, errors.Errorf("Post with digest error: %d: %s", resp.StatusCode, resp.Status)
-	}
-
-	return resp, nil
+	return resp, responseError(resp)
+}
+
+// stripWSSecurityHeader removes the wsse:Security header block from a SOAP
+// envelope, leaving all other header blocks intact. The message is returned
+// unchanged if it cannot be parsed as XML or has no such header.
+func stripWSSecurityHeader(message string) string {
+	doc := etree.NewDocument()
+	if err := doc.ReadFromString(message); err != nil {
+		return message
+	}
+	security := doc.FindElement("./Envelope/Header/Security")
+	if security == nil {
+		return message
+	}
+	header := doc.Root().SelectElement("Header")
+	if header == nil {
+		return message
+	}
+	header.RemoveChild(security)
+	data, err := doc.WriteToString()
+	if err != nil {
+		return message
+	}
+	return data
+}
+
+var digestParamRe = regexp.MustCompile(`(\w+)=(?:"([^"]*)"|([^,]+))`)
+
+// parseDigestChallenge parses the parameters of a WWW-Authenticate: Digest header.
+func parseDigestChallenge(challenge string) map[string]string {
+	challenge = strings.TrimSpace(challenge)
+	if i := strings.IndexAny(challenge, " \t"); i >= 0 && strings.EqualFold(challenge[:i], "Digest") {
+		challenge = challenge[i+1:]
+	}
+
+	result := make(map[string]string)
+	for _, m := range digestParamRe.FindAllStringSubmatch(challenge, -1) {
+		value := m[2]
+		if value == "" {
+			value = m[3]
+		}
+		result[strings.ToLower(m[1])] = strings.TrimSpace(value)
+	}
+	return result
+}
+
+// newDigestAuthorization builds an RFC 2617 HTTP digest Authorization header
+// value. It supports the MD5 and MD5-sess algorithms and the "auth" qop, which
+// covers the vast majority of ONVIF devices. It returns an empty string when the
+// challenge is missing required parameters or requires an unsupported qop.
+func newDigestAuthorization(challenge, method, uri, username, password string) string {
+	parts := parseDigestChallenge(challenge)
+	realm := parts["realm"]
+	nonce := parts["nonce"]
+	if realm == "" || nonce == "" {
+		return ""
+	}
+	opaque := parts["opaque"]
+	algorithm := parts["algorithm"]
+
+	qop := ""
+	if rawQop, ok := parts["qop"]; ok {
+		for _, candidate := range strings.Split(rawQop, ",") {
+			if strings.TrimSpace(candidate) == "auth" {
+				qop = "auth"
+				break
+			}
+		}
+		// The server offered qop but none we support (e.g. auth-int only).
+		if qop == "" {
+			return ""
+		}
+	}
+
+	digestURI := uri
+	if u, err := url.Parse(uri); err == nil {
+		digestURI = u.RequestURI()
+	}
+
+	cnonce := randomCnonce()
+	const nc = "00000001"
+
+	ha1 := md5Hex(username + ":" + realm + ":" + password)
+	if strings.EqualFold(algorithm, "MD5-sess") {
+		ha1 = md5Hex(ha1 + ":" + nonce + ":" + cnonce)
+	}
+	ha2 := md5Hex(method + ":" + digestURI)
+
+	var response string
+	if qop == "auth" {
+		response = md5Hex(strings.Join([]string{ha1, nonce, nc, cnonce, qop, ha2}, ":"))
+	} else {
+		response = md5Hex(ha1 + ":" + nonce + ":" + ha2)
+	}
+
+	var b strings.Builder
+	fmt.Fprintf(&b, `Digest username="%s", realm="%s", nonce="%s", uri="%s", response="%s"`,
+		username, realm, nonce, digestURI, response)
+	if qop == "auth" {
+		fmt.Fprintf(&b, `, qop=auth, nc=%s, cnonce="%s"`, nc, cnonce)
+	}
+	if algorithm != "" {
+		fmt.Fprintf(&b, `, algorithm=%s`, algorithm)
+	}
+	if opaque != "" {
+		fmt.Fprintf(&b, `, opaque="%s"`, opaque)
+	}
+	return b.String()
+}
+
+func md5Hex(s string) string {
+	sum := md5.Sum([]byte(s))
+	return hex.EncodeToString(sum[:])
+}
+
+func randomCnonce() string {
+	b := make([]byte, 8)
+	if _, err := rand.Read(b); err != nil {
+		return "00000000"
+	}
+	return hex.EncodeToString(b)
 }

ptz/types_test.go

@@ -0,0 +1,25 @@
+package ptz
+
+import (
+	"encoding/xml"
+	"strings"
+	"testing"
+
+	"github.com/kerberos-io/onvif/xsd/onvif"
+)
+
+func TestContinuousMoveIncludesZeroPanTiltCoordinates(t *testing.T) {
+	request := ContinuousMove{
+		Velocity: onvif.PTZSpeedPanTilt{
+			PanTilt: onvif.Vector2D{X: 0.5, Y: 0},
+		},
+	}
+
+	encoded, err := xml.Marshal(request)
+	if err != nil {
+		t.Fatalf("xml.Marshal() error = %v", err)
+	}
+	if !strings.Contains(string(encoded), `x="0.5" y="0"`) {
+		t.Fatalf("ContinuousMove PanTilt = %s, want explicit x and y attributes", encoded)
+	}
+}

sdk/event/PullMessages_auto.go

@@ -0,0 +1,31 @@
+// Code generated : DO NOT EDIT.
+// Copyright (c) 2022 Jean-Francois SMIGIELSKI
+// Distributed under the MIT License
+
+package event
+
+import (
+	"context"
+
+	"github.com/juju/errors"
+	"github.com/kerberos-io/onvif"
+	"github.com/kerberos-io/onvif/event"
+	"github.com/kerberos-io/onvif/sdk"
+)
+
+// Call_PullMessages forwards the call to dev.CallMethod() then parses the payload of the reply as a PullMessagesResponse.
+func Call_PullMessages(ctx context.Context, dev *onvif.Device, request event.PullMessages) (event.PullMessagesResponse, error) {
+	type Envelope struct {
+		Header struct{}
+		Body   struct {
+			PullMessagesResponse event.PullMessagesResponse
+		}
+	}
+	var reply Envelope
+	if httpReply, err := dev.CallMethod(request); err != nil {
+		return reply.Body.PullMessagesResponse, errors.Annotate(err, "call")
+	} else {
+		err = sdk.ReadAndParse(ctx, httpReply, &reply, "PullMessages")
+		return reply.Body.PullMessagesResponse, errors.Annotate(err, "reply")
+	}
+}

sdk/event/event.go

@@ -0,0 +1,8 @@
+package event
+
+//go:generate go run github.com/kerberos-io/onvif/sdk/codegen event event CreatePullPointSubscription
+//go:generate go run github.com/kerberos-io/onvif/sdk/codegen event event GetEventProperties
+//go:generate go run github.com/kerberos-io/onvif/sdk/codegen event event GetServiceCapabilities
+//go:generate go run github.com/kerberos-io/onvif/sdk/codegen event event Subscribe
+//go:generate go run github.com/kerberos-io/onvif/sdk/codegen event event Unsubscribe
+//go:generate go run github.com/kerberos-io/onvif/sdk/codegen event event PullMessages

sdk/sdk.go

@@ -0,0 +1,25 @@
+package sdk
+
+import (
+	"context"
+	"encoding/xml"
+	"io"
+	"net/http"
+
+	"github.com/juju/errors"
+)
+
+// ReadAndParse reads the body of the given HTTP reply and unmarshals it into
+// reply. The tag identifies the ONVIF action for diagnostic purposes.
+func ReadAndParse(ctx context.Context, httpReply *http.Response, reply interface{}, tag string) error {
+	// TODO(jfsmig): extract the deadline from ctx.Deadline() and apply it on the reply reading
+	b, err := io.ReadAll(httpReply.Body)
+	if err != nil {
+		return errors.Annotate(err, "read")
+	}
+
+	httpReply.Body.Close()
+
+	err = xml.Unmarshal(b, reply)
+	return errors.Annotate(err, "decode")
+}

xsd/onvif/onvif.go

@@ -666,13 +666,13 @@ type PTZSpeedPanTilt struct {
 }
 
 type Vector2D struct {
-	X     float64     `xml:"x,attr,omitempty"`
-	Y     float64     `xml:"y,attr,omitempty"`
+	X     float64     `xml:"x,attr"`
+	Y     float64     `xml:"y,attr"`
 	Space *xsd.AnyURI `xml:"space,attr,omitempty"`
 }
 
 type Vector1D struct {
-	X     float64     `xml:"x,attr,omitempty"`
+	X     float64     `xml:"x,attr"`
 	Space *xsd.AnyURI `xml:"space,attr,omitempty"`
 }
 
@@ -1176,7 +1176,7 @@ type PresetTour struct {
 	Status            PTZPresetTourStatus            `xml:"Status"`
 	AutoStart         xsd.Boolean                    `xml:"AutoStart"`
 	StartingCondition PTZPresetTourStartingCondition `xml:"StartingCondition"`
-	TourSpot          []PTZPresetTourSpot           	 `xml:"TourSpot"`
+	TourSpot          []PTZPresetTourSpot                    `xml:"TourSpot"`
 	Extension         PTZPresetTourExtension         `xml:"Extension"`
 }