amadan

carlos / platform Public

clamd: a hand-rolled INSTREAM client with explicit deadlines

Stdlib only. The protocol is four commands wide and we use two, so a
dependency would cost more than it saved — and this way every read and
write carries a deadline we chose. A scan pass that blocks forever on a
wedged clamd is a stuck pass, not a slow one.

Two things the mapping has to get right. A size refusal arrives as
"INSTREAM size limit exceeded. ERROR" and is NOT a failure: the object
was never judged, and evidence that cannot tell those apart is not worth
keeping. And clamd refuses an over-large stream mid-upload by replying
and hanging up, so the next write fails — reporting that broken pipe as
a transport error would fail the whole pass over an object the engine
has already answered about.

The fakes carry a deadline. Without one a framing bug does not fail the
tests, it hangs them: verified by flipping the header to little-endian,
which wedged until the package timeout. Now it fails in five seconds
saying so.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01YJkJkxZZLgW8X8A5tk69Sq
Paul Campbell pushed by paul@keymail.dev 45619080cb05cca94d5fd1e2c1e73de33ff7e645 parent 3f0d155
2 files changed, +514 −0
  • internal/clamd/clamd.go +277 −0
  • internal/clamd/clamd_test.go +237 −0
diff --git a/internal/clamd/clamd.go b/internal/clamd/clamd.go
new file mode 100644
index 000000000..8b57fe288
--- /dev/null
+++ b/internal/clamd/clamd.go
@@ -0,0 +1,277 @@
+// Package clamd speaks just enough of the clamd protocol to stream one object
+// at a time past a scanning engine and read back a verdict.
+//
+// It is hand-rolled and stdlib-only on purpose. The protocol is four commands
+// wide and we use two of them; a dependency would cost more than it saved, and
+// this way every read and write on the socket carries a deadline we chose
+// rather than one a library chose for us. A scan pass that blocks forever on a
+// wedged clamd is a stuck pass, not a slow one.
+//
+// The wire format, for the record:
+//
+// zINSTREAM\0 the command, NUL-terminated
+// <uint32 be><payload> one chunk, repeated
+// <uint32 be 0> the terminator
+// stream: OK\0 the reply, NUL-terminated
+package clamd
+
+import (
+ "bufio"
+ "context"
+ "encoding/binary"
+ "errors"
+ "fmt"
+ "io"
+ "net"
+ "strings"
+ "time"
+)
+
+// Status is what the engine said about one stream.
+type Status string
+
+const (
+ // StatusClean is a scan that ran and found nothing.
+ StatusClean Status = "clean"
+ // StatusInfected is a scan that ran and matched a signature.
+ StatusInfected Status = "infected"
+ // StatusSizeExceeded is the engine refusing to judge an object at all
+ // because it is larger than StreamMaxLength. This is deliberately NOT
+ // StatusError: nothing failed, the object simply was not judged, and an
+ // audit that cannot tell those apart is not worth keeping.
+ StatusSizeExceeded Status = "size-exceeded"
+ // StatusError is the engine reading the stream and failing on it.
+ StatusError Status = "error"
+)
+
+// Result is one engine verdict.
+type Result struct {
+ Status Status
+ // Signature is the malware name on an infected result, empty otherwise.
+ Signature string
+ // Raw is the engine's reply with its NUL and whitespace trimmed, kept so
+ // a failure can be logged as the engine actually phrased it.
+ Raw string
+}
+
+// Timeouts bounds every blocking operation. There is no infinite option: a
+// zero field takes the default below, it does not mean "wait forever".
+type Timeouts struct {
+ // Dial bounds connecting to the socket.
+ Dial time.Duration
+ // Write bounds a single write — the command, one chunk, the terminator —
+ // not the whole stream. A large object legitimately takes many writes.
+ Write time.Duration
+ // Read bounds waiting for the reply once the stream is fully sent. clamd
+ // does its scanning work before answering, so this is the one that has to
+ // tolerate a real workload.
+ Read time.Duration
+}
+
+const (
+ defaultDial = 5 * time.Second
+ defaultWrite = 30 * time.Second
+ defaultRead = 5 * time.Minute
+
+ // chunkSize is what we hand the kernel per frame. clamd's own client uses
+ // the same order of magnitude and the frame header is 4 bytes, so this
+ // keeps the framing overhead under a thousandth.
+ chunkSize = 64 * 1024
+)
+
+func (t Timeouts) withDefaults() Timeouts {
+ if t.Dial <= 0 {
+ t.Dial = defaultDial
+ }
+ if t.Write <= 0 {
+ t.Write = defaultWrite
+ }
+ if t.Read <= 0 {
+ t.Read = defaultRead
+ }
+ return t
+}
+
+// Client talks to one clamd over a unix socket. It holds no connection between
+// calls: clamd's INSTREAM is one command per connection, so each call dials,
+// scans and hangs up. That also means a Client is safe to share.
+type Client struct {
+ socket string
+ timeouts Timeouts
+
+ // conn, when set, is used for exactly one call and then dropped. Tests set
+ // it through newWithConn to run the protocol over a net.Pipe; production
+ // never does.
+ conn net.Conn
+}
+
+// New returns a Client dialling socket, which is a unix socket path such as
+// /run/clamav/clamd.ctl.
+func New(socket string, timeouts Timeouts) *Client {
+ return &Client{socket: socket, timeouts: timeouts.withDefaults()}
+}
+
+// newWithConn returns a Client that will run its next — and only its next —
+// command over conn. Test seam.
+func newWithConn(conn net.Conn) *Client {
+ return &Client{conn: conn, timeouts: Timeouts{}.withDefaults()}
+}
+
+func (c *Client) dial(ctx context.Context) (net.Conn, error) {
+ if c.conn != nil {
+ conn := c.conn
+ c.conn = nil
+ return conn, nil
+ }
+ if c.socket == "" {
+ return nil, errors.New("clamd: no socket configured")
+ }
+ d := net.Dialer{Timeout: c.timeouts.Dial}
+ conn, err := d.DialContext(ctx, "unix", c.socket)
+ if err != nil {
+ return nil, fmt.Errorf("clamd: dial %s: %w", c.socket, err)
+ }
+ return conn, nil
+}
+
+// deadline is now+d, pulled in to the context's own deadline when that is
+// sooner, so a cancelled pass does not sit out a full timeout.
+func deadline(ctx context.Context, d time.Duration) time.Time {
+ t := time.Now().Add(d)
+ if ctxDeadline, ok := ctx.Deadline(); ok && ctxDeadline.Before(t) {
+ return ctxDeadline
+ }
+ return t
+}
+
+func (c *Client) write(ctx context.Context, conn net.Conn, b []byte) error {
+ if err := ctx.Err(); err != nil {
+ return err
+ }
+ if err := conn.SetWriteDeadline(deadline(ctx, c.timeouts.Write)); err != nil {
+ return err
+ }
+ _, err := conn.Write(b)
+ return err
+}
+
+// readReply reads up to the NUL that ends every clamd reply. A closed
+// connection with bytes already in hand is a complete reply, not a failure:
+// clamd hangs up promptly after answering.
+func (c *Client) readReply(ctx context.Context, conn net.Conn) (string, error) {
+ if err := conn.SetReadDeadline(deadline(ctx, c.timeouts.Read)); err != nil {
+ return "", err
+ }
+ r := bufio.NewReader(conn)
+ s, err := r.ReadString(0)
+ if err != nil && !(errors.Is(err, io.EOF) && s != "") {
+ return "", fmt.Errorf("clamd: read reply: %w", err)
+ }
+ return strings.TrimSpace(strings.TrimSuffix(s, "\x00")), nil
+}
+
+// ScanStream sends r past the engine and returns its verdict. It never buffers
+// the object: r is read a chunk at a time straight onto the socket, so a
+// gigabyte object costs a chunk of memory, not a gigabyte.
+func (c *Client) ScanStream(ctx context.Context, r io.Reader) (Result, error) {
+ conn, err := c.dial(ctx)
+ if err != nil {
+ return Result{}, err
+ }
+ defer conn.Close()
+
+ if err := c.write(ctx, conn, []byte("zINSTREAM\x00")); err != nil {
+ return Result{}, fmt.Errorf("clamd: send INSTREAM: %w", err)
+ }
+
+ buf := make([]byte, chunkSize)
+ var hdr [4]byte
+ for {
+ n, rerr := r.Read(buf)
+ if n > 0 {
+ binary.BigEndian.PutUint32(hdr[:], uint32(n))
+ if err := c.write(ctx, conn, hdr[:]); err != nil {
+ return c.replyAfterWriteFailure(ctx, conn, err)
+ }
+ if err := c.write(ctx, conn, buf[:n]); err != nil {
+ return c.replyAfterWriteFailure(ctx, conn, err)
+ }
+ }
+ if errors.Is(rerr, io.EOF) {
+ break
+ }
+ if rerr != nil {
+ return Result{}, fmt.Errorf("clamd: read object: %w", rerr)
+ }
+ }
+
+ binary.BigEndian.PutUint32(hdr[:], 0)
+ if err := c.write(ctx, conn, hdr[:]); err != nil {
+ return c.replyAfterWriteFailure(ctx, conn, err)
+ }
+
+ reply, err := c.readReply(ctx, conn)
+ if err != nil {
+ return Result{}, err
+ }
+ return parseReply(reply), nil
+}
+
+// replyAfterWriteFailure covers the one case where a broken pipe is an answer
+// rather than an accident: clamd refuses an over-large stream mid-upload by
+// replying and hanging up, so our next write fails. Reporting that as a
+// transport error would fail the whole pass over an object the engine has
+// already told us it will not judge. If nothing readable is waiting, the
+// original write error stands.
+func (c *Client) replyAfterWriteFailure(ctx context.Context, conn net.Conn, werr error) (Result, error) {
+ reply, rerr := c.readReply(ctx, conn)
+ if rerr != nil || reply == "" {
+ return Result{}, fmt.Errorf("clamd: send stream: %w", werr)
+ }
+ return parseReply(reply), nil
+}
+
+// Version returns the engine and signature-database line, which is what the
+// evidence records as the provenance of a sweep.
+func (c *Client) Version(ctx context.Context) (string, error) {
+ conn, err := c.dial(ctx)
+ if err != nil {
+ return "", err
+ }
+ defer conn.Close()
+
+ if err := c.write(ctx, conn, []byte("zVERSION\x00")); err != nil {
+ return "", fmt.Errorf("clamd: send VERSION: %w", err)
+ }
+ return c.readReply(ctx, conn)
+}
+
+// parseReply maps clamd's one-line answers onto a Status.
+//
+// The size check comes first and deliberately: the refusal arrives as
+// "INSTREAM size limit exceeded. ERROR", so an ERROR-suffix test that ran
+// first would swallow it and record a failure where nothing failed.
+func parseReply(reply string) Result {
+ res := Result{Raw: reply}
+ switch {
+ case strings.Contains(reply, "size limit exceeded"):
+ res.Status = StatusSizeExceeded
+ case strings.HasSuffix(reply, " OK"):
+ res.Status = StatusClean
+ case strings.HasSuffix(reply, " FOUND"):
+ res.Status = StatusInfected
+ res.Signature = signatureOf(reply)
+ default:
+ res.Status = StatusError
+ }
+ return res
+}
+
+// signatureOf pulls the malware name out of "stream: Eicar-Signature FOUND".
+func signatureOf(reply string) string {
+ body := strings.TrimSuffix(reply, " FOUND")
+ if i := strings.LastIndex(body, ": "); i >= 0 {
+ body = body[i+2:]
+ }
+ return strings.TrimSpace(body)
+}
diff --git a/internal/clamd/clamd_test.go b/internal/clamd/clamd_test.go
new file mode 100644
index 000000000..55fa03fc9
--- /dev/null
+++ b/internal/clamd/clamd_test.go
@@ -0,0 +1,237 @@
+package clamd
+
+import (
+ "bufio"
+ "bytes"
+ "context"
+ "encoding/binary"
+ "io"
+ "net"
+ "strings"
+ "testing"
+ "time"
+)
+
+// fakeDeadline bounds the fake clamd's side of every pipe, so a protocol bug
+// surfaces as a failing test rather than a hung one.
+const fakeDeadline = 5 * time.Second
+
+// frameReader decodes the INSTREAM framing off the wire: an optional
+// NUL-terminated command, then 4-byte big-endian length + payload chunks,
+// ending at the zero-length chunk. It exists so the tests assert the bytes we
+// actually put on the socket rather than trusting the writer to agree with
+// itself.
+type frameReader struct {
+ r *bufio.Reader
+ cmdRead bool
+ remain uint32
+ done bool
+}
+
+func newFrameReader(r io.Reader) *frameReader {
+ return &frameReader{r: bufio.NewReader(r)}
+}
+
+func (f *frameReader) Read(p []byte) (int, error) {
+ if !f.cmdRead {
+ if _, err := f.r.ReadString(0); err != nil {
+ return 0, err
+ }
+ f.cmdRead = true
+ }
+ for f.remain == 0 {
+ if f.done {
+ return 0, io.EOF
+ }
+ var hdr [4]byte
+ if _, err := io.ReadFull(f.r, hdr[:]); err != nil {
+ return 0, err
+ }
+ f.remain = binary.BigEndian.Uint32(hdr[:])
+ if f.remain == 0 {
+ f.done = true
+ return 0, io.EOF
+ }
+ }
+ if uint32(len(p)) > f.remain {
+ p = p[:f.remain]
+ }
+ n, err := f.r.Read(p)
+ f.remain -= uint32(n)
+ return n, err
+}
+
+// fakeClamd stands a clamd on one end of a net.Pipe. It reads the command,
+// drains any INSTREAM frames, hands back the payload, then writes reply.
+//
+// NOTE: over net.Pipe (unbuffered, synchronous) the server MUST finish reading
+// before it writes, or both ends deadlock — the client is still blocked in a
+// write nobody is reading while the server blocks in a write nobody is
+// reading. Every helper here keeps that order.
+func fakeClamd(t *testing.T, reply string) (*Client, <-chan []byte) {
+ t.Helper()
+ client, server := net.Pipe()
+ // A framing bug does not fail these tests, it HANGS them: a wrong length
+ // header has the fake waiting on megabytes the client will never send
+ // while the client waits on a reply. Verified by flipping the header to
+ // little-endian, which without this deadline wedges until the package
+ // test timeout ten minutes later. The deadline turns that into a clean
+ // failure in seconds.
+ _ = server.SetDeadline(time.Now().Add(fakeDeadline))
+ got := make(chan []byte, 1)
+ go func() {
+ defer server.Close()
+ br := bufio.NewReader(server)
+ cmd, err := br.ReadString(0)
+ if err != nil {
+ got <- nil
+ return
+ }
+ var payload []byte
+ if strings.Contains(cmd, "INSTREAM") {
+ payload, _ = io.ReadAll(&frameReader{r: br, cmdRead: true})
+ }
+ got <- payload
+ _, _ = server.Write([]byte(reply))
+ }()
+ t.Cleanup(func() { client.Close() })
+ return newWithConn(client), got
+}
+
+func replyFixture(t *testing.T, reply string) Result {
+ t.Helper()
+ c, _ := fakeClamd(t, reply)
+ res, err := c.ScanStream(context.Background(), bytes.NewReader([]byte("payload")))
+ if err != nil {
+ t.Fatalf("ScanStream: %v", err)
+ }
+ return res
+}
+
+func versionFixture(t *testing.T, reply string) string {
+ t.Helper()
+ c, _ := fakeClamd(t, reply)
+ v, err := c.Version(context.Background())
+ if err != nil {
+ t.Fatalf("Version: %v", err)
+ }
+ return v
+}
+
+// INSTREAM framing, asserted on the wire: zINSTREAM\0, then 4-byte
+// big-endian length + payload chunks, then a zero-length chunk.
+func TestScanStreamWireFormat(t *testing.T) {
+ client, server := net.Pipe()
+ _ = server.SetDeadline(time.Now().Add(fakeDeadline))
+ got := make(chan []byte, 1)
+ go func() {
+ defer server.Close()
+ buf, _ := io.ReadAll(newFrameReader(server)) // reads until the zero chunk
+ got <- buf
+ server.Write([]byte("stream: OK\x00"))
+ }()
+
+ c := newWithConn(client)
+ res, err := c.ScanStream(context.Background(), bytes.NewReader([]byte("hello")))
+ if err != nil {
+ t.Fatal(err)
+ }
+ if res.Status != StatusClean {
+ t.Fatalf("status = %v, want clean", res.Status)
+ }
+ if b := <-got; !bytes.Equal(b, []byte("hello")) {
+ t.Fatalf("payload on the wire = %q, want %q", b, "hello")
+ }
+}
+
+func TestScanStreamInfected(t *testing.T) {
+ res := replyFixture(t, "stream: Eicar-Signature FOUND\x00")
+ if res.Status != StatusInfected {
+ t.Fatalf("status = %v, want infected", res.Status)
+ }
+ if res.Signature != "Eicar-Signature" {
+ t.Fatalf("signature = %q", res.Signature)
+ }
+}
+
+// A size refusal is SKIPPED, not FAILED. An over-large object was not judged;
+// calling that a failure would make the evidence lie about what happened.
+func TestScanStreamSizeExceededIsItsOwnStatus(t *testing.T) {
+ res := replyFixture(t, "INSTREAM size limit exceeded. ERROR\x00")
+ if res.Status != StatusSizeExceeded {
+ t.Fatalf("status = %v, want size-exceeded", res.Status)
+ }
+}
+
+// Anything else ending ERROR is the engine reading the object and failing on
+// it — a per-object fact, distinct from the engine being unreachable.
+func TestScanStreamErrorReply(t *testing.T) {
+ res := replyFixture(t, "stream: Can't allocate memory ERROR\x00")
+ if res.Status != StatusError {
+ t.Fatalf("status = %v, want error", res.Status)
+ }
+ if !strings.Contains(res.Raw, "allocate memory") {
+ t.Fatalf("the engine's own words did not survive: %q", res.Raw)
+ }
+}
+
+func TestVersionParse(t *testing.T) {
+ // zVERSION answers engine/sig-db-version/sig-db-date.
+ v := versionFixture(t, "ClamAV 1.4.1/27500/Mon Sep 1 08:00:00 2026\x00")
+ if v != "ClamAV 1.4.1/27500/Mon Sep 1 08:00:00 2026" {
+ t.Fatalf("version = %q", v)
+ }
+}
+
+// A chunked object arrives whole and in order. One Read per chunk means an
+// off-by-one in the framing shows up as a corrupted stream, not as a slightly
+// wrong length — this asserts we never send a frame header we then fail to
+// fill.
+func TestScanStreamChunksLargeObject(t *testing.T) {
+ want := bytes.Repeat([]byte("carlos"), chunkSize) // several chunks' worth
+ c, got := fakeClamd(t, "stream: OK\x00")
+ if _, err := c.ScanStream(context.Background(), bytes.NewReader(want)); err != nil {
+ t.Fatal(err)
+ }
+ if b := <-got; !bytes.Equal(b, want) {
+ t.Fatalf("streamed %d bytes, want %d", len(b), len(want))
+ }
+}
+
+// Every blocking operation carries a deadline. A clamd that accepts the stream
+// and never answers must fail the object, not wedge the pass forever.
+func TestReadReplyHasADeadline(t *testing.T) {
+ client, server := net.Pipe()
+ defer server.Close()
+ go func() {
+ br := bufio.NewReader(server)
+ br.ReadString(0)
+ io.ReadAll(&frameReader{r: br, cmdRead: true})
+ // ... and then never reply.
+ select {}
+ }()
+
+ c := newWithConn(client)
+ c.timeouts.Read = 50 * time.Millisecond
+ done := make(chan error, 1)
+ go func() {
+ _, err := c.ScanStream(context.Background(), bytes.NewReader([]byte("x")))
+ done <- err
+ }()
+ select {
+ case err := <-done:
+ if err == nil {
+ t.Fatal("a silent clamd returned success")
+ }
+ case <-time.After(5 * time.Second):
+ t.Fatal("ScanStream blocked past its read deadline")
+ }
+}
+
+// A zero Timeouts means the defaults, never "wait forever".
+func TestZeroTimeoutsTakeDefaults(t *testing.T) {
+ c := New("/run/clamav/clamd.ctl", Timeouts{})
+ if c.timeouts.Dial <= 0 || c.timeouts.Write <= 0 || c.timeouts.Read <= 0 {
+ t.Fatalf("zero timeouts left an infinite wait: %+v", c.timeouts)
+ }
+}