commit bf19fbe42f8f291e04a1ded34c3261a5c18007c3
parent ce90339d90026caf72cdf791283ad97687f574e7
Author: Jack Mordaunt <jackmordaunt.dev@gmail.com>
Date: Fri, 18 Sep 2026 13:16:17 -0400
mf: make a wedged decode abandonable
Fuzzing found a half-megabyte file that Media Foundation never finishes
decoding. It opens cleanly, reports a plausible format, then grinds out
a few hundred kilobytes of PCM and stops answering. The decode could not
be recovered: ReadSample was synchronous, so the blocked call owned the
calling thread and nothing could interrupt it.
The source reader now runs asynchronously, which needs a COM callback,
which means implementing IMFSourceReaderCallback in Go. A COM interface
pointer is a struct whose first word points to a vtable, so a Go struct
with a vtable of syscall.NewCallback trampolines as its first field is a
usable COM object. The vtable is a package singleton because those
trampolines are never freed, the object is refcounted and pinned because
Media Foundation holds its address, and every method runs under a
recover: a panic crossing those frames would take the process down. The
64-bit timestamp argument is the only thing that differs by word size,
so it lives in two small build-tagged files and is ignored in favour of
reading the time back off the sample. The retry loop is bounded too,
since asking for a sample and being given nothing is otherwise unlimited.
Waiting is then ours to bound. Limits caps how long a decode may run and
how much it may produce, with a finite default, and the deadline reaches
the backend so a read already parked on the decoder gives up with
everything else. The buffered API is routed through the streaming one so
both are covered.
That leaves teardown. Releasing a source reader waits for Media
Foundation to finish, which after a clean end of stream is immediate and
after a wedged decode is never: flushing the reader and closing the byte
stream underneath it were both measured to make no difference, and one
blocked release was watched for ten minutes without returning.
Such a reader is leaked rather than released on a goroutine. A release
that never returns never frees anything either, so waiting on it leaks
the same objects and adds a thread: waiting is a superset of not
waiting. The thread costs no CPU, since one parked in a blocking system
call is never scheduled, but the runtime caps a process at 10000 threads
and crossing that is fatal. At roughly one thread per bad file that
ceiling is reachable by a program decoding untrusted input.
Diffstat:
10 files changed, 597 insertions(+), 90 deletions(-)
diff --git a/README.md b/README.md
@@ -69,6 +69,10 @@ play.File("audio.m4a")
always 2.
- A `Decoder` is safe for concurrent use, and `Close` waits for decodes
already in flight.
+- `New(WithLimits(...))` bounds how long a decode may run and how much PCM
+ it may produce. Malformed audio can otherwise make a native decoder
+ grind indefinitely, so the default budget is finite. Untrusted input
+ should set its own.
- `FFmpegLoad`, `FFmpegDecode` and `FFmpegStream` shell out to ffmpeg
regardless of platform, and need no `Decoder`.
- The `play` subpackage plays PCM through oto. The core package builds
diff --git a/audio.go b/audio.go
@@ -15,6 +15,7 @@ package nativeaudio
import (
"errors"
"fmt"
+ "io"
"sync"
)
@@ -35,15 +36,28 @@ type Decoder struct {
mu sync.RWMutex
closed bool
streams sync.WaitGroup
+ limits Limits
+}
+
+// Option configures a Decoder at construction.
+type Option func(*Decoder)
+
+// WithLimits bounds what a single decode may consume. See [Limits].
+func WithLimits(l Limits) Option {
+ return func(d *Decoder) { d.limits = l }
}
// New creates a Decoder, initialising any platform state the backend
// needs. Call Close when you are finished with it.
-func New() (*Decoder, error) {
+func New(opts ...Option) (*Decoder, error) {
if err := start(); err != nil {
return nil, fmt.Errorf("initialising platform decoder: %w", err)
}
- return &Decoder{}, nil
+ d := &Decoder{limits: DefaultLimits()}
+ for _, opt := range opts {
+ opt(d)
+ }
+ return d, nil
}
// Close releases the platform state held by the Decoder. It is
@@ -72,7 +86,7 @@ func (d *Decoder) DecodeFile(path string) (pcm []byte, format Format, err error)
if d.closed {
return nil, format, ErrClosed
}
- return load(path)
+ return d.drain(openStreamFile(path))
}
// Decode decodes compressed audio held in memory, returning s16le PCM
@@ -83,7 +97,22 @@ func (d *Decoder) Decode(compressed []byte) (pcm []byte, format Format, err erro
if d.closed {
return nil, format, ErrClosed
}
- return decode(compressed)
+ return d.drain(openStream(compressed))
+}
+
+// drain reads a stream to completion under the decoder's limits, always
+// closing it. Routing the buffered API through the streaming one is what
+// lets limits apply to both.
+func (d *Decoder) drain(s *Stream, err error) ([]byte, Format, error) {
+ if err != nil {
+ return nil, Format{}, err
+ }
+ defer s.Close()
+ pcm, err := io.ReadAll(newLimited(s.r, d.limits))
+ if err != nil {
+ return nil, s.format, err
+ }
+ return pcm, s.format, nil
}
// Format describes the PCM a decode produced, and is everything needed
@@ -110,6 +139,7 @@ func (d *Decoder) Stream(compressed []byte) (*Stream, error) {
if err != nil {
return nil, err
}
+ s.r = newLimited(s.r, d.limits)
d.track(s)
return s, nil
}
@@ -131,6 +161,7 @@ func (d *Decoder) StreamFile(path string) (*Stream, error) {
if err != nil {
return nil, err
}
+ s.r = newLimited(s.r, d.limits)
d.track(s)
return s, nil
}
diff --git a/audio_linux.go b/audio_linux.go
@@ -2,16 +2,6 @@
package nativeaudio
-// load stub for Linux.
-func load(path string) ([]byte, Format, error) {
- return FFmpegLoad(path)
-}
-
-// decode stub for Linux.
-func decode(by []byte) ([]byte, Format, error) {
- return FFmpegDecode(by)
-}
-
func start() error {
return nil
}
diff --git a/audio_unknown.go b/audio_unknown.go
@@ -5,16 +5,6 @@
package nativeaudio
-// load and decode an audio file with ffmpeg.
-func load(path string) ([]byte, Format, error) {
- return FFmpegLoad(path)
-}
-
-// decode a bytes with ffmpeg.
-func decode(by []byte) ([]byte, Format, error) {
- return FFmpegDecode(by)
-}
-
// start stub.
func start() error {
return nil
diff --git a/audio_windows.go b/audio_windows.go
@@ -2,7 +2,6 @@ package nativeaudio
import (
"fmt"
- "io"
"os"
"git.sr.ht/~jackmordaunt/nativeaudio/internal/mf"
@@ -16,34 +15,6 @@ func end() error {
return mf.Shutdown()
}
-// load raw pcm data from the Windows Media Foundation.
-func load(path string) (uncompressed []byte, format Format, err error) {
- f, err := os.Open(path)
- if err != nil {
- return nil, format, fmt.Errorf("opening input file: %w", err)
- }
- defer f.Close()
- by, err := io.ReadAll(f)
- if err != nil {
- return nil, format, fmt.Errorf("buffering input file: %w", err)
- }
- return decode(by)
-}
-
-// decode compressed data, returning the uncompressed data as PCM data
-// (s16le) and details about the PCM required to playback correctly.
-func decode(compressed []byte) (uncompressed []byte, format Format, err error) {
- pcm, f, err := mf.Decode(compressed)
- if err != nil {
- return nil, format, err
- }
- return pcm, Format{
- SampleRate: f.SampleRate,
- Channels: f.Channels,
- BytesPerSample: f.BytesPerSample,
- }, nil
-}
-
// openStream decodes incrementally through Media Foundation.
func openStream(compressed []byte) (*Stream, error) {
ms, err := mf.Open(compressed)
diff --git a/internal/mf/callback.go b/internal/mf/callback.go
@@ -0,0 +1,215 @@
+//go:build windows
+
+package mf
+
+import (
+ "errors"
+ "fmt"
+ "os"
+ "runtime"
+ "runtime/debug"
+ "sync/atomic"
+ "syscall"
+ "time"
+ "unsafe"
+)
+
+// readTimeout bounds the wait for one asynchronous read.
+//
+// A healthy decode delivers samples in milliseconds. Media Foundation can
+// block forever on a malformed stream, and in synchronous mode that hang
+// is unrecoverable: the blocked call owns the calling thread and there is
+// nothing to interrupt it. Driving the reader asynchronously makes the
+// wait ours to abandon.
+const readTimeout = 10 * time.Second
+
+// errReadTimeout reports that Media Foundation never answered a read.
+var errReadTimeout = errors.New("timed out waiting for the decoder")
+
+// HRESULTs the callback needs beyond S_OK.
+const (
+ E_NOINTERFACE HRESULT = 0x80004002
+ E_POINTER HRESULT = 0x80004003
+ E_FAIL HRESULT = 0x80004005
+)
+
+var (
+ IID_IUnknown = GUID{0x00000000, 0x0000, 0x0000, [8]uint8{0xc0, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x46}}
+ IID_IMFSourceReaderCallback = GUID{0xdeec8d99, 0xfa1d, 0x4d82, [8]uint8{0x84, 0xc2, 0x2c, 0x89, 0x69, 0x94, 0x48, 0x67}}
+
+ MF_SOURCE_READER_ASYNC_CALLBACK = GUID{0x1e3dbeac, 0xbb43, 0x4c35, [8]uint8{0xb5, 0x07, 0xcd, 0x64, 0x44, 0x64, 0xc9, 0x65}}
+)
+
+// failed reports whether an HRESULT denotes failure, which is the sign bit
+// of the 32-bit code.
+func failed(hr HRESULT) bool {
+ return int32(uint32(hr)) < 0
+}
+
+// IUnknownVtbl is the head of every COM vtable.
+type IUnknownVtbl struct {
+ QueryInterface uintptr
+ AddRef uintptr
+ Release uintptr
+}
+
+// sourceReaderCallbackVtbl is the vtable of IMFSourceReaderCallback.
+type sourceReaderCallbackVtbl struct {
+ IUnknownVtbl
+ OnReadSample uintptr
+ OnFlush uintptr
+ OnEvent uintptr
+}
+
+// readResult is one delivery from OnReadSample. Its sample carries a
+// reference taken by the callback, which the receiver must release.
+type readResult struct {
+ status HRESULT
+ flags uint32
+ sample *IMFSample
+}
+
+// callback implements IMFSourceReaderCallback so the source reader can be
+// driven asynchronously.
+//
+// A COM interface pointer points at a struct whose first word points to a
+// vtable of function pointers. Handing Media Foundation the address of a
+// Go struct whose first field is that vtable pointer therefore makes it a
+// usable COM object, and each trampoline receives the struct back as its
+// first argument. The vtable is a package-level singleton because
+// [syscall.NewCallback] never frees what it creates.
+//
+// Media Foundation invokes the callback on its own worker thread, so
+// deliveries arrive over a channel rather than being handled in place.
+type callback struct {
+ vtbl *sourceReaderCallbackVtbl // must remain the first field.
+
+ refs atomic.Int32
+ pin runtime.Pinner
+
+ // ch holds at most one undelivered read. Reads are issued one at a
+ // time, so a deeper buffer would only hide a protocol mistake.
+ ch chan readResult
+}
+
+// newCallback returns a callback with one reference, pinned so Media
+// Foundation may hold its address.
+func newCallback() *callback {
+ c := &callback{vtbl: sourceReaderCallbackVtable, ch: make(chan readResult, 1)}
+ c.refs.Store(1)
+ c.pin.Pin(c)
+ return c
+}
+
+// AddRef implements IUnknown::AddRef.
+func (c *callback) AddRef() uint32 {
+ return uint32(c.refs.Add(1))
+}
+
+// Release implements IUnknown::Release, unpinning on the last reference.
+//
+// Media Foundation holds its own reference for as long as it might still
+// call us, so the object outlives the stream that created it whenever a
+// read is still outstanding.
+func (c *callback) Release() uint32 {
+ n := c.refs.Add(-1)
+ if n == 0 {
+ c.drain()
+ c.pin.Unpin()
+ }
+ return uint32(n)
+}
+
+// QueryInterface implements IUnknown::QueryInterface.
+func (c *callback) QueryInterface(riid *GUID, ppv *unsafe.Pointer) HRESULT {
+ if ppv == nil {
+ return E_POINTER
+ }
+ if riid == nil || (*riid != IID_IUnknown && *riid != IID_IMFSourceReaderCallback) {
+ *ppv = nil
+ return E_NOINTERFACE
+ }
+ *ppv = unsafe.Pointer(c)
+ c.AddRef()
+ return S_OK
+}
+
+// onReadSample takes delivery of one read.
+//
+// The sample is only guaranteed for the duration of this call, so it is
+// retained before being handed over. When nothing is waiting, which is
+// what happens once the reader has given up on a read, the sample is
+// released here rather than leaked.
+func (c *callback) onReadSample(status HRESULT, flags uint32, sample *IMFSample) HRESULT {
+ sample.AddRef()
+ select {
+ case c.ch <- readResult{status: status, flags: flags, sample: sample}:
+ default:
+ sample.Release()
+ }
+ return S_OK
+}
+
+// wait blocks for the next delivery, giving up after d.
+func (c *callback) wait(d time.Duration) (readResult, error) {
+ timer := time.NewTimer(d)
+ defer timer.Stop()
+ select {
+ case r := <-c.ch:
+ return r, nil
+ case <-timer.C:
+ return readResult{}, errReadTimeout
+ }
+}
+
+// drain releases any sample left undelivered.
+func (c *callback) drain() {
+ for {
+ select {
+ case r := <-c.ch:
+ r.sample.Release()
+ default:
+ return
+ }
+ }
+}
+
+// guard runs a COM method body and turns a panic into a failed HRESULT.
+//
+// A panic unwinding out of a [syscall.NewCallback] trampoline crosses C
+// frames and takes the process down with it. COM callers expect failure
+// as a status code, so report it as one.
+func guard(method string, body func() HRESULT) (hr HRESULT) {
+ defer func() {
+ if p := recover(); p != nil {
+ fmt.Fprintf(os.Stderr, "nativeaudio: panic in %s: %v\n%s\n", method, p, debug.Stack())
+ hr = E_FAIL
+ }
+ }()
+ return body()
+}
+
+// sourceReaderCallbackVtable is shared by every callback instance. The
+// trampolines recover the instance from the pointer COM passes back.
+var sourceReaderCallbackVtable = &sourceReaderCallbackVtbl{
+ IUnknownVtbl: IUnknownVtbl{
+ QueryInterface: syscall.NewCallback(func(this *callback, riid *GUID, ppv *unsafe.Pointer) uintptr {
+ return guard("IMFSourceReaderCallback::QueryInterface", func() HRESULT { return this.QueryInterface(riid, ppv) })
+ }),
+ AddRef: syscall.NewCallback(func(this *callback) uintptr {
+ return guard("IMFSourceReaderCallback::AddRef", func() HRESULT { return HRESULT(this.AddRef()) })
+ }),
+ Release: syscall.NewCallback(func(this *callback) uintptr {
+ return guard("IMFSourceReaderCallback::Release", func() HRESULT { return HRESULT(this.Release()) })
+ }),
+ },
+ OnReadSample: onReadSampleTrampoline,
+ // Nothing is flushed and no events are acted on, but both slots must
+ // be populated and must succeed.
+ OnFlush: syscall.NewCallback(func(this *callback, streamIndex uint32) uintptr {
+ return S_OK
+ }),
+ OnEvent: syscall.NewCallback(func(this *callback, streamIndex uint32, event uintptr) uintptr {
+ return S_OK
+ }),
+}
diff --git a/internal/mf/callback_32.go b/internal/mf/callback_32.go
@@ -0,0 +1,18 @@
+//go:build windows && !amd64 && !arm64
+
+package mf
+
+import "syscall"
+
+// onReadSampleTrampoline implements IMFSourceReaderCallback::OnReadSample.
+//
+// See the 64-bit variant for the interface signature. Where a word is four
+// bytes wide the LONGLONG timestamp is passed as two slots, so it arrives
+// as two halves and is ignored, exactly as it is there.
+var onReadSampleTrampoline = syscall.NewCallback(
+ func(this *callback, status uintptr, streamIndex, flags, timestampLow, timestampHigh uint32, sample *IMFSample) uintptr {
+ return guard("IMFSourceReaderCallback::OnReadSample", func() HRESULT {
+ return this.onReadSample(HRESULT(status), flags, sample)
+ })
+ },
+)
diff --git a/internal/mf/callback_64.go b/internal/mf/callback_64.go
@@ -0,0 +1,23 @@
+//go:build windows && (amd64 || arm64)
+
+package mf
+
+import "syscall"
+
+// onReadSampleTrampoline implements IMFSourceReaderCallback::OnReadSample:
+//
+// HRESULT OnReadSample(HRESULT hrStatus, DWORD dwStreamIndex,
+// DWORD dwStreamFlags, LONGLONG llTimestamp,
+// IMFSample *pSample)
+//
+// Every argument occupies one pointer-sized slot here, so the 64-bit
+// timestamp needs no special handling. It is ignored in any case, because
+// the timestamp is read back from the sample itself. That keeps this
+// signature the only thing that differs between word sizes.
+var onReadSampleTrampoline = syscall.NewCallback(
+ func(this *callback, status uintptr, streamIndex, flags uint32, timestamp int64, sample *IMFSample) uintptr {
+ return guard("IMFSourceReaderCallback::OnReadSample", func() HRESULT {
+ return this.onReadSample(HRESULT(status), flags, sample)
+ })
+ },
+)
diff --git a/internal/mf/mf.go b/internal/mf/mf.go
@@ -15,6 +15,7 @@ import (
"runtime"
"sync"
"syscall"
+ "time"
"unicode/utf16"
"unsafe"
@@ -52,6 +53,7 @@ type Stream struct {
attributes *IMFAttributes
reader *IMFSourceReader
samples *SourceReader
+ cb *callback
format Format
closed bool
@@ -79,14 +81,23 @@ func Open(compressed []byte) (_ *Stream, err error) {
return nil, fmt.Errorf("creating MFByteStream from IStream: %w", err)
}
- // Attributes to configure the source reader with; specifically, enable hardware codecs.
- if err := MFCreateAttributes(&s.attributes, 1); err != nil {
+ // Attributes to configure the source reader with: hardware codecs,
+ // and the callback that puts the reader in asynchronous mode.
+ if err := MFCreateAttributes(&s.attributes, 2); err != nil {
return nil, fmt.Errorf("creating attributes to apply to source reader: %w", err)
}
if err := s.attributes.SetUINT32(&MF_READWRITE_ENABLE_HARDWARE_TRANSFORMS, 1); err != nil {
return nil, fmt.Errorf("enabling hardware transforms: %w", err)
}
+ // Without a callback the reader is synchronous, and a synchronous
+ // ReadSample on a malformed stream can block forever with no way to
+ // interrupt it.
+ s.cb = newCallback()
+ if err := s.attributes.SetUnknown(&MF_SOURCE_READER_ASYNC_CALLBACK, unsafe.Pointer(s.cb)); err != nil {
+ return nil, fmt.Errorf("setting async callback: %w", err)
+ }
+
// Create the source reader using the byte stream.
if err := MFCreateSourceReaderFromByteStream(s.byteStream, s.attributes, &s.reader); err != nil {
return nil, fmt.Errorf("creating IMFSourceReader from IMFByteStream: %w", err)
@@ -101,7 +112,7 @@ func Open(compressed []byte) (_ *Stream, err error) {
return nil, fmt.Errorf("getting format: %w", err)
}
- s.samples = NewSourceReader(s.reader)
+ s.samples = NewSourceReader(s.reader, s.cb)
return s, nil
}
@@ -112,6 +123,18 @@ func (s *Stream) Format() Format {
return s.format
}
+// SetDeadline bounds how long the stream may go on decoding. Reads fail
+// once it passes, including a read already waiting on the decoder. A
+// zero time clears it.
+//
+// Without this the only bound is the per-read backstop, which a file
+// that decodes slowly but steadily never trips.
+func (s *Stream) SetDeadline(t time.Time) {
+ if s.samples != nil {
+ s.samples.deadline = t
+ }
+}
+
// Read fills p with decoded PCM, returning [io.EOF] once the source is
// exhausted.
func (s *Stream) Read(p []byte) (int, error) {
@@ -125,11 +148,38 @@ func (s *Stream) Read(p []byte) (int, error) {
// Close releases the Media Foundation objects backing the stream. It is
// idempotent.
+//
+// Releasing a source reader waits for Media Foundation to finish what it
+// is doing. After a clean end of stream that is immediate, and after the
+// caller simply stops reading early it is still prompt.
+//
+// It is not prompt when the decoder stopped answering, which malformed
+// audio can cause: the release blocks with no way to cancel it. Flushing
+// the reader and closing the byte stream underneath it were both measured
+// to make no difference, and one blocked release was watched for ten
+// minutes without returning, so treat it as permanent.
+//
+// Such a reader is leaked rather than released on a goroutine. A release
+// that never returns never frees anything either, so waiting on it leaks
+// the same objects and adds a thread: waiting is a superset of not
+// waiting, which is the whole argument. The thread itself costs no CPU,
+// since one parked in a blocking system call is never scheduled, and it
+// does not slow the scheduler, which tracks processors rather than
+// threads. What it does is count against the runtime limit of 10000
+// operating system threads, and crossing that is a fatal thread
+// exhaustion rather than a slowdown. At roughly one thread per bad file
+// that ceiling is reachable by a program decoding untrusted input.
+//
+// Everything else is released either way, and the reader keeps its own
+// references to what it still needs.
func (s *Stream) Close() error {
if s.closed {
return nil
}
s.closed = true
+ if s.samples != nil && s.samples.gaveUp && s.reader != nil {
+ s.reader = nil
+ }
s.release()
return nil
}
@@ -157,27 +207,16 @@ func (s *Stream) release() {
s.istream.Release()
s.istream = nil
}
+ // Last: the reader is gone by now, so Media Foundation is finished
+ // calling us. Any reference it still holds keeps the object alive.
+ if s.cb != nil {
+ s.cb.Release()
+ s.cb = nil
+ }
runtime.KeepAlive(s.compressed)
s.compressed = nil
}
-// Decode compressed audio, returning the uncompressed data as PCM data
-// (s16le) and details about the PCM required to playback correctly.
-func Decode(compressed []byte) (uncompressed []byte, format Format, err error) {
- s, err := Open(compressed)
- if err != nil {
- return nil, format, err
- }
- defer s.Close()
-
- buf, err := io.ReadAll(s)
- if err != nil {
- return nil, s.format, err
- }
-
- return buf, s.format, nil
-}
-
// configureAudioStream selects the first audio stream and configures it output PCM.
func configureAudioStream(pReader *IMFSourceReader) error {
var pPartialType *IMFMediaType
@@ -262,13 +301,24 @@ type SourceReader struct {
prevTimestamp int64 // timestamp of the last emitted sample.
hasPrev bool // whether any sample has been emitted yet.
+ cb *callback // receives asynchronous reads.
+
+ // gaveUp records that a read was abandoned because the decoder
+ // stopped answering, as opposed to the caller simply stopping early.
+ gaveUp bool
+
+ // deadline bounds the whole decode, not just one read. Zero means
+ // only the per-read backstop applies.
+ deadline time.Time
+
data []byte // Go view of the audio data, backed by native memory.
}
-// NewSourceReader allocates a [SourceReader].
-// The caller is responsible for releasing the underlying [IMFSourceReader].
-func NewSourceReader(r *IMFSourceReader) *SourceReader {
- return &SourceReader{source: r}
+// NewSourceReader allocates a [SourceReader] that pulls samples from r,
+// which must have been created with cb as its asynchronous callback.
+// The caller is responsible for releasing both.
+func NewSourceReader(r *IMFSourceReader, cb *callback) *SourceReader {
+ return &SourceReader{source: r, cb: cb}
}
func (s *SourceReader) Read(p []byte) (int, error) {
@@ -294,24 +344,56 @@ func (s *SourceReader) Read(p []byte) (int, error) {
// next reads the next audio sample into s, returning false at end of
// stream. Any error from the source reader ends the stream.
+// maxEmptyReads bounds how many times next will ask for a sample and
+// be given nothing usable.
+//
+// ReadSample can legitimately return success with no sample and no
+// terminal flag while a decoder primes, and it can repeat a timestamp,
+// so a few unproductive reads are normal. An unbounded number is a
+// hang: fuzzing found malformed streams that spin here forever,
+// burning a core and never returning. The limit sits far above what any
+// healthy stream needs.
+const maxEmptyReads = 1024
+
func (s *SourceReader) next() (bool, error) {
+ empty := 0
for {
- var (
- flags uint32
- timestamp int64
- sample *IMFSample
- )
-
- err := s.source.ReadSample(MF_SOURCE_READER_FIRST_AUDIO_STREAM, 0, nil, &flags, ×tamp, &sample)
+ if empty > maxEmptyReads {
+ return false, fmt.Errorf("reading sample: gave up after %d reads without a usable sample", empty)
+ }
+ if err := s.source.ReadSampleAsync(MF_SOURCE_READER_FIRST_AUDIO_STREAM); err != nil {
+ return false, fmt.Errorf("requesting sample: %w", err)
+ }
- // A sample can accompany a flag we treat as terminal, so release
- // it before deciding whether to stop.
- if err != nil || flags&(MF_SOURCE_READERF_ERROR|MF_SOURCE_READERF_ENDOFSTREAM|MF_SOURCE_READERF_CURRENTMEDIATYPECHANGED) != 0 {
- sample.Release()
+ // Wait no longer than whichever of the caller's deadline and the
+ // backstop comes first.
+ wait := readTimeout
+ if !s.deadline.IsZero() {
+ remaining := time.Until(s.deadline)
+ if remaining <= 0 {
+ s.gaveUp = true
+ return false, fmt.Errorf("reading sample: %w", errReadTimeout)
+ }
+ if remaining < wait {
+ wait = remaining
+ }
}
+
+ res, err := s.cb.wait(wait)
if err != nil {
+ s.gaveUp = true
return false, fmt.Errorf("reading sample: %w", err)
}
+ flags, sample := res.flags, res.sample
+
+ // A sample can accompany a status or flag we treat as terminal,
+ // so release it before deciding whether to stop.
+ if failed(res.status) || flags&(MF_SOURCE_READERF_ERROR|MF_SOURCE_READERF_ENDOFSTREAM|MF_SOURCE_READERF_CURRENTMEDIATYPECHANGED) != 0 {
+ sample.Release()
+ }
+ if failed(res.status) {
+ return false, fmt.Errorf("reading sample: %w", MFErr{Code: res.status})
+ }
if flags&MF_SOURCE_READERF_ERROR != 0 {
return false, fmt.Errorf("reading sample: source reader reported an error")
}
@@ -323,16 +405,25 @@ func (s *SourceReader) next() (bool, error) {
}
// The reader can legitimately return no sample and no terminal
- // flag (for example, while a decoder is buffering). Ask again.
+ // flag (for example, while a decoder is buffering). Ask again,
+ // up to the bound above.
if sample == nil {
+ empty++
continue
}
- // Skip samples that repeat the previous timestamp. ReadSample can
+ var timestamp int64
+ if err := sample.GetSampleTime(×tamp); err != nil {
+ sample.Release()
+ return false, fmt.Errorf("reading sample timestamp: %w", err)
+ }
+
+ // Skip samples that repeat the previous timestamp. The reader can
// produce more than one sample at time 0; emitting all of them
// produces larger output and audible artifacts.
if s.hasPrev && timestamp == s.prevTimestamp {
sample.Release()
+ empty++
continue
}
@@ -481,6 +572,22 @@ func (v *IMFByteStream) Release() error {
return nil
}
+// Close closes the byte stream, failing any I/O the media source has in
+// flight against it.
+func (v *IMFByteStream) Close() error {
+ if v == nil {
+ return nil
+ }
+ r, _, _ := syscall.SyscallN(
+ v.VTable.Close,
+ uintptr(unsafe.Pointer(v)),
+ )
+ if r != S_OK {
+ return MFErr{Code: r}
+ }
+ return nil
+}
+
type IStream struct {
VTable *IStreamVTable
}
@@ -583,6 +690,20 @@ func (v *IMFAttributes) SetUINT32(guid *GUID, unValue uint32) error {
return nil
}
+// SetUnknown stores a COM interface pointer under guid, retaining it.
+func (v *IMFAttributes) SetUnknown(guid *GUID, unknown unsafe.Pointer) error {
+ r, _, _ := syscall.SyscallN(
+ v.VTable.SetUnknown,
+ uintptr(unsafe.Pointer(v)),
+ uintptr(unsafe.Pointer(guid)),
+ uintptr(unknown),
+ )
+ if r != S_OK {
+ return MFErr{Code: r}
+ }
+ return nil
+}
+
type IMFMediaType struct {
VTable *IMFMediaTypeVTable
}
@@ -759,6 +880,43 @@ func (v *IMFSourceReader) ReadSample(index, controlFlags uint32, actualIndex *ui
return nil
}
+// ReadSampleAsync requests the next sample without waiting for it.
+//
+// A reader configured with a callback rejects the synchronous form, and
+// every out-parameter must be nil: the result is delivered to
+// IMFSourceReaderCallback::OnReadSample instead. Only one read may be
+// outstanding at a time.
+func (v *IMFSourceReader) ReadSampleAsync(index uint32) error {
+ r, _, _ := syscall.SyscallN(
+ v.VTable.ReadSample,
+ uintptr(unsafe.Pointer(v)),
+ uintptr(index),
+ 0,
+ 0,
+ 0,
+ 0,
+ 0,
+ )
+ if r != S_OK {
+ return MFErr{Code: r}
+ }
+ return nil
+}
+
+// Flush discards queued samples and cancels pending reads on a stream.
+// With a callback attached it completes asynchronously, through OnFlush.
+func (v *IMFSourceReader) Flush(index uint32) error {
+ r, _, _ := syscall.SyscallN(
+ v.VTable.Flush,
+ uintptr(unsafe.Pointer(v)),
+ uintptr(index),
+ )
+ if r != S_OK {
+ return MFErr{Code: r}
+ }
+ return nil
+}
+
type IMFSample struct {
VTable *IMFSampleVTable
}
@@ -827,6 +985,36 @@ func (v *IMFSample) Release() error {
return nil
}
+// AddRef implements IUnknown::AddRef, retaining a sample past the
+// callback that delivered it.
+func (v *IMFSample) AddRef() uint32 {
+ if v == nil {
+ return 0
+ }
+ r, _, _ := syscall.SyscallN(
+ v.VTable.AddRef,
+ uintptr(unsafe.Pointer(v)),
+ )
+ return uint32(r)
+}
+
+// GetSampleTime returns the presentation time of the sample.
+//
+// Asynchronous delivery also carries a timestamp, but reading it back
+// from the sample keeps the callback signature free of a 64-bit argument
+// whose slot count varies by word size.
+func (v *IMFSample) GetSampleTime(t *int64) error {
+ r, _, _ := syscall.SyscallN(
+ v.VTable.GetSampleTime,
+ uintptr(unsafe.Pointer(v)),
+ uintptr(unsafe.Pointer(t)),
+ )
+ if r != S_OK {
+ return MFErr{Code: r}
+ }
+ return nil
+}
+
func (v *IMFSample) ConvertToContiguousBuffer(b **IMFMediaBuffer) error {
r, _, _ := syscall.SyscallN(
v.VTable.ConvertToContiguousBuffer,
diff --git a/stream.go b/stream.go
@@ -2,7 +2,10 @@ package nativeaudio
import (
"bytes"
+ "errors"
+ "fmt"
"io"
+ "time"
)
// Stream decodes audio incrementally, yielding signed 16-bit
@@ -56,3 +59,77 @@ func newBufferedStream(pcm []byte, format Format) *Stream {
format: format,
}
}
+
+// Limits bound a single decode. A zero field means no limit.
+//
+// Malformed audio can make a decoder grind: fuzzing found a half-megabyte
+// file that Media Foundation decodes indefinitely, a few hundred
+// kilobytes of PCM at a time, never finishing and never failing. No
+// per-read timeout catches that, because every individual read completes.
+// Limits are the backstop, and matter whenever the input is untrusted.
+type Limits struct {
+ // MaxBytes caps the PCM a decode may produce.
+ MaxBytes int64
+
+ // MaxDuration caps how long a decode may run.
+ MaxDuration time.Duration
+}
+
+// DefaultLimits is applied to a Decoder created without WithLimits.
+//
+// The duration is generous by orders of magnitude: a healthy backend
+// decodes half an hour of audio in well under a second, so anything
+// still running after this is not making meaningful progress. No byte
+// limit is imposed by default, since a legitimately long recording is
+// legitimately large.
+func DefaultLimits() Limits {
+ return Limits{MaxDuration: 2 * time.Minute}
+}
+
+// ErrLimitExceeded reports that a decode hit its [Limits].
+var ErrLimitExceeded = errors.New("nativeaudio: decode exceeded its limits")
+
+// limited enforces Limits over a stream.
+type limited struct {
+ r io.ReadCloser
+ limits Limits
+ read int64
+ deadline time.Time
+}
+
+func newLimited(r io.ReadCloser, l Limits) io.ReadCloser {
+ if l.MaxBytes <= 0 && l.MaxDuration <= 0 {
+ return r
+ }
+ lr := &limited{r: r, limits: l}
+ if l.MaxDuration > 0 {
+ lr.deadline = time.Now().Add(l.MaxDuration)
+ // Hand the deadline to the backend too, so a read already
+ // waiting on the decoder gives up with everything else rather
+ // than running on to its own backstop.
+ if d, ok := r.(interface{ SetDeadline(time.Time) }); ok {
+ d.SetDeadline(lr.deadline)
+ }
+ }
+ return lr
+}
+
+func (l *limited) Read(p []byte) (int, error) {
+ if !l.deadline.IsZero() && time.Now().After(l.deadline) {
+ return 0, fmt.Errorf("%w: still decoding after %s", ErrLimitExceeded, l.limits.MaxDuration)
+ }
+ if l.limits.MaxBytes > 0 {
+ if remaining := l.limits.MaxBytes - l.read; remaining <= 0 {
+ return 0, fmt.Errorf("%w: produced more than %d bytes", ErrLimitExceeded, l.limits.MaxBytes)
+ } else if int64(len(p)) > remaining {
+ p = p[:remaining]
+ }
+ }
+ n, err := l.r.Read(p)
+ l.read += int64(n)
+ return n, err
+}
+
+func (l *limited) Close() error {
+ return l.r.Close()
+}
