tend/internal/transport/protocol.go

// SPDX-License-Identifier: AGPL-3.0-only

// Package transport implements Tend's bounded, versioned deployment stream.
package transport

import (
	"bufio"
	"bytes"
	"crypto/sha256"
	"encoding/binary"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"regexp"
	"strings"
)

const (
	Protocol         = "tend-receive-v1"
	MaxArtifactBytes = 512 << 20
	maxHeaderBytes   = 64 << 10
	maxArtifactName  = 128
)

var (
	servicePattern  = regexp.MustCompile(`^[a-z][a-z0-9-]{1,62}$`)
	artifactPattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$`)
)

type Header struct {
	Protocol       string `json:"protocol"`
	Service        string `json:"service"`
	ArtifactName   string `json:"artifact_name"`
	Size           int64  `json:"size"`
	SHA256         string `json:"sha256"`
	ApprovedSHA256 string `json:"approved_sha256"`
	Activate       bool   `json:"activate"`
}

func (h Header) Validate(maxBytes int64) error {
	if h.Protocol != Protocol {
		return errors.New("unsupported receive protocol")
	}
	if !servicePattern.MatchString(h.Service) {
		return errors.New("invalid service name")
	}
	if len(h.ArtifactName) > maxArtifactName || !artifactPattern.MatchString(h.ArtifactName) {
		return errors.New("invalid artifact name")
	}
	if h.Size <= 0 || h.Size > maxBytes {
		return errors.New("artifact size exceeds policy")
	}
	if h.SHA256 != h.ApprovedSHA256 || len(h.SHA256) != 64 || strings.ToLower(h.SHA256) != h.SHA256 {
		return errors.New("artifact digest was not explicitly approved")
	}
	if _, err := hex.DecodeString(h.SHA256); err != nil {
		return errors.New("artifact digest is not hexadecimal")
	}
	return nil
}

func Prefix(header Header) ([]byte, error) {
	if err := header.Validate(MaxArtifactBytes); err != nil {
		return nil, err
	}
	body, err := json.Marshal(header)
	if err != nil {
		return nil, err
	}
	if len(body) > maxHeaderBytes {
		return nil, errors.New("receive header exceeds limit")
	}
	prefix := make([]byte, 4+len(body))
	binary.BigEndian.PutUint32(prefix[:4], uint32(len(body)))
	copy(prefix[4:], body)
	return prefix, nil
}

func ReadHeader(reader *bufio.Reader) (Header, error) {
	var size [4]byte
	if _, err := io.ReadFull(reader, size[:]); err != nil {
		return Header{}, fmt.Errorf("read receive header length: %w", err)
	}
	length := binary.BigEndian.Uint32(size[:])
	if length == 0 || length > maxHeaderBytes {
		return Header{}, errors.New("receive header length is invalid")
	}
	body := make([]byte, length)
	if _, err := io.ReadFull(reader, body); err != nil {
		return Header{}, fmt.Errorf("read receive header: %w", err)
	}
	decoder := json.NewDecoder(bytes.NewReader(body))
	decoder.DisallowUnknownFields()
	var header Header
	if err := decoder.Decode(&header); err != nil {
		return Header{}, fmt.Errorf("decode receive header: %w", err)
	}
	var trailing any
	if err := decoder.Decode(&trailing); !errors.Is(err, io.EOF) {
		return Header{}, errors.New("receive header contains trailing data")
	}
	return header, nil
}

func CopyArtifact(destination io.Writer, reader *bufio.Reader, header Header, maxBytes int64) error {
	if err := header.Validate(maxBytes); err != nil {
		return err
	}
	hash := sha256.New()
	written, err := io.CopyN(io.MultiWriter(destination, hash), reader, header.Size)
	if err != nil || written != header.Size {
		return errors.New("artifact stream ended before declared size")
	}
	if _, err = reader.ReadByte(); !errors.Is(err, io.EOF) {
		return errors.New("artifact stream contains trailing bytes")
	}
	if hex.EncodeToString(hash.Sum(nil)) != header.SHA256 {
		return errors.New("artifact stream digest does not match")
	}
	return nil
}