go-sip/internal/agent/upload.go

package agent

import (
	"bytes"
	"context"
	"crypto/sha256"
	"encoding/hex"
	"errors"
	"fmt"
	"hash"
	"io"
	"net/http"
	"net/url"
	"os"
	"strings"
	"sync"
	"time"

	agentpb "git.ipao.vip/rogee/go-sip/gen/agent"
)

// UploadClient performs the Agent-direct data-plane upload using a restricted
// Dispatcher grant. It never sends file bytes through Dispatcher or writes an
// OSS credential to logs. It never deletes or moves the source asset; the
// lifecycle owner retains it until upload-fact delivery to MQ is durably recorded.
type UploadClient struct {
	HTTPClient          *http.Client
	AllowedHosts        map[string]struct{}
	AllowInsecureHTTP   bool
	MaxResponseBodySize int64
	Now                 func() time.Time
}

var ErrUploadGrantExpired = errors.New("upload grant is expired")
var ErrUploadGrantInvalid = errors.New("upload grant is invalid")
var ErrUploadChecksumMismatch = errors.New("upload checksum mismatch")
var ErrUploadOutcomeUnknown = errors.New("upload outcome unknown")

// UploadHTTPError records only the status, never the signed URL or OSS body.
type UploadHTTPError struct{ StatusCode int }

func (e *UploadHTTPError) Error() string { return fmt.Sprintf("upload returned HTTP %d", e.StatusCode) }

type UploadResult struct {
	StatusCode int
	SizeBytes  int64
	SHA256     string
	ETag       string
}

// UploadBytes sends an in-memory recording directly; the success path never
// creates a recording file or a local call-result journal.
func (c UploadClient) UploadBytes(ctx context.Context, grant *agentpb.UploadGrant, recording []byte) (UploadResult, error) {
	parsed, err := c.validateGrant(ctx, grant)
	if err != nil {
		return UploadResult{}, err
	}
	if len(recording) == 0 {
		return UploadResult{}, errors.New("recording audio is empty")
	}
	sum := sha256.Sum256(recording)
	return c.putValidated(ctx, grant, parsed, bytes.NewReader(recording), int64(len(recording)), hex.EncodeToString(sum[:]))
}

func (c UploadClient) validateGrant(ctx context.Context, grant *agentpb.UploadGrant) (*url.URL, error) {
	if grant == nil {
		return nil, fmt.Errorf("%w: grant is required", ErrUploadGrantInvalid)
	}
	if grant.TargetUrl == "" || grant.UploadId == "" || grant.ObjectKey == "" {
		return nil, fmt.Errorf("%w: URL, ID and object key are required", ErrUploadGrantInvalid)
	}
	if grant.ExpiresAtUnixMs <= 0 {
		return nil, fmt.Errorf("%w: expiry is required", ErrUploadGrantInvalid)
	}
	now := time.Now
	if c.Now != nil {
		now = c.Now
	}
	if !now().Before(time.UnixMilli(grant.ExpiresAtUnixMs)) {
		return nil, ErrUploadGrantExpired
	}
	parsed, err := url.Parse(grant.TargetUrl)
	if err != nil || parsed.Host == "" {
		return nil, fmt.Errorf("%w: URL is invalid", ErrUploadGrantInvalid)
	}
	if parsed.Scheme != "https" && !(c.AllowInsecureHTTP && parsed.Scheme == "http") {
		return nil, fmt.Errorf("%w: URL must use HTTPS", ErrUploadGrantInvalid)
	}
	if len(c.AllowedHosts) > 0 {
		if _, ok := c.AllowedHosts[strings.ToLower(parsed.Host)]; !ok {
			return nil, fmt.Errorf("%w: host %q is not allowed", ErrUploadGrantInvalid, parsed.Host)
		}
	}
	if err := ctx.Err(); err != nil {
		return nil, err
	}
	return parsed, nil
}

func (c UploadClient) UploadFile(ctx context.Context, grant *agentpb.UploadGrant, path string) (result UploadResult, err error) {
	parsed, err := c.validateGrant(ctx, grant)
	if err != nil {
		return UploadResult{}, err
	}
	file, err := os.Open(path)
	if err != nil {
		return UploadResult{}, err
	}
	stat, err := file.Stat()
	if err != nil {
		_ = file.Close()
		return UploadResult{}, err
	}
	if stat.IsDir() {
		_ = file.Close()
		return UploadResult{}, errors.New("upload path is a directory")
	}
	if grant.MaxBytes > 0 && stat.Size() > grant.MaxBytes {
		_ = file.Close()
		return UploadResult{}, fmt.Errorf("asset exceeds grant limit: %d > %d", stat.Size(), grant.MaxBytes)
	}
	defer func() {
		if closeErr := file.Close(); closeErr != nil && !errors.Is(closeErr, os.ErrClosed) {
			result = UploadResult{}
			err = errors.Join(err, closeErr)
		}
	}()
	digest, err := digestFile(file)
	if err != nil {
		return UploadResult{}, err
	}
	if _, err := file.Seek(0, io.SeekStart); err != nil {
		return UploadResult{}, err
	}
	return c.putValidated(ctx, grant, parsed, file, stat.Size(), digest)
}

func (c UploadClient) putValidated(ctx context.Context, grant *agentpb.UploadGrant, parsed *url.URL, source io.Reader, size int64, digest string) (UploadResult, error) {
	if grant.MaxBytes > 0 && size > grant.MaxBytes {
		return UploadResult{}, fmt.Errorf("asset exceeds grant limit: %d > %d", size, grant.MaxBytes)
	}
	if grant.RequiredChecksumSha256 != "" && !strings.EqualFold(grant.RequiredChecksumSha256, digest) {
		return UploadResult{}, fmt.Errorf("%w: asset does not match grant", ErrUploadChecksumMismatch)
	}
	transmitted := &uploadChecksum{hash: sha256.New()}
	body := io.TeeReader(io.LimitReader(source, size), transmitted)
	req, err := http.NewRequestWithContext(ctx, http.MethodPut, parsed.String(), body)
	if err != nil {
		return UploadResult{}, fmt.Errorf("%w: cannot construct PUT request", ErrUploadGrantInvalid)
	}
	req.ContentLength = size
	for _, header := range grant.Headers {
		if header == nil {
			return UploadResult{}, fmt.Errorf("%w: header is missing", ErrUploadGrantInvalid)
		}
		if strings.EqualFold(header.Name, "host") || strings.EqualFold(header.Name, "content-length") {
			return UploadResult{}, fmt.Errorf("%w: forbidden header", ErrUploadGrantInvalid)
		}
		req.Header.Set(header.Name, header.Value)
	}
	client := c.HTTPClient
	if client == nil {
		client = &http.Client{}
	}
	copyClient := *client
	copyClient.CheckRedirect = func(_ *http.Request, _ []*http.Request) error { return http.ErrUseLastResponse }
	resp, err := copyClient.Do(req)
	if err != nil {
		// A transport failure after sending bytes has an unknown OSS outcome.
		// Never expose the signed URL, even if a custom transport includes it.
		var requestError *url.Error
		if errors.As(err, &requestError) {
			err = requestError.Err
		}
		kind := fmt.Sprintf("%T", err)
		switch {
		case errors.Is(err, context.Canceled):
			kind = "canceled"
		case errors.Is(err, context.DeadlineExceeded):
			kind = "deadline"
		case errors.Is(err, io.ErrUnexpectedEOF):
			kind = "unexpected_eof"
		}
		return UploadResult{}, fmt.Errorf("%w: PUT transport failure (%s)", ErrUploadOutcomeUnknown, kind)
	}
	defer resp.Body.Close()
	maxResponse := c.MaxResponseBodySize
	if maxResponse <= 0 {
		maxResponse = 64 << 10
	}
	if resp.StatusCode < http.StatusOK || resp.StatusCode >= http.StatusMultipleChoices {
		_, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, maxResponse))
		return UploadResult{}, &UploadHTTPError{StatusCode: resp.StatusCode}
	}
	if _, err := io.Copy(io.Discard, io.LimitReader(resp.Body, maxResponse)); err != nil {
		return UploadResult{}, fmt.Errorf("%w: read PUT response (%T)", ErrUploadOutcomeUnknown, err)
	}
	sentDigest, sentBytes := transmitted.result()
	if sentBytes != size || sentDigest != digest {
		return UploadResult{}, errors.Join(ErrUploadOutcomeUnknown, fmt.Errorf("%w: transmitted bytes differ from validated asset", ErrUploadChecksumMismatch))
	}
	return UploadResult{StatusCode: resp.StatusCode, SizeBytes: sentBytes, SHA256: sentDigest, ETag: resp.Header.Get("ETag")}, nil
}

// HTTP transports may still be writing the request when response headers arrive.
// Synchronize observation so early responses cannot race checksum calculation.
type uploadChecksum struct {
	mu    sync.Mutex
	hash  hash.Hash
	bytes int64
}

func (c *uploadChecksum) Write(p []byte) (int, error) {
	c.mu.Lock()
	defer c.mu.Unlock()
	n, err := c.hash.Write(p)
	c.bytes += int64(n)
	return n, err
}
func (c *uploadChecksum) result() (string, int64) {
	c.mu.Lock()
	defer c.mu.Unlock()
	return hex.EncodeToString(c.hash.Sum(nil)), c.bytes
}

func digestFile(file *os.File) (string, error) {
	if _, err := file.Seek(0, io.SeekStart); err != nil {
		return "", err
	}
	hash := sha256.New()
	if _, err := io.Copy(hash, file); err != nil {
		return "", err
	}
	return hex.EncodeToString(hash.Sum(nil)), nil
}