Add upload archive staging
This commit is contained in:
346
internal/ingest/archive.go
Normal file
346
internal/ingest/archive.go
Normal file
@@ -0,0 +1,346 @@
|
||||
package ingest
|
||||
|
||||
import (
|
||||
"archive/tar"
|
||||
"compress/gzip"
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"mime"
|
||||
"os"
|
||||
"path"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
|
||||
sourcebundle "gitea.maximumdirect.net/eric/distributor/pkg/bundle"
|
||||
)
|
||||
|
||||
const (
|
||||
ContentTypeTar = "application/x-tar"
|
||||
ContentTypeGzip = "application/gzip"
|
||||
ContentTypeXGzip = "application/x-gzip"
|
||||
)
|
||||
|
||||
var (
|
||||
ErrUnsupportedContentType = errors.New("unsupported archive content type")
|
||||
ErrUploadTooLarge = errors.New("upload exceeds maximum size")
|
||||
ErrExtractedTooLarge = errors.New("extracted bundle exceeds maximum size")
|
||||
ErrTooManyFiles = errors.New("extracted bundle has too many files")
|
||||
ErrUnsafeArchivePath = errors.New("unsafe archive path")
|
||||
)
|
||||
|
||||
type StageOptions struct {
|
||||
Body io.Reader
|
||||
ContentType string
|
||||
PipelineStagingPath string
|
||||
RunID string
|
||||
MaxUploadSize int64
|
||||
MaxExtractedSize int64
|
||||
MaxFileCount int
|
||||
}
|
||||
|
||||
type StagedBundle struct {
|
||||
Root string
|
||||
Manifest sourcebundle.Manifest
|
||||
}
|
||||
|
||||
func StageArchive(ctx context.Context, opts StageOptions) (StagedBundle, error) {
|
||||
if ctx == nil {
|
||||
ctx = context.Background()
|
||||
}
|
||||
if err := validateStageOptions(opts); err != nil {
|
||||
return StagedBundle{}, err
|
||||
}
|
||||
format, err := archiveFormat(opts.ContentType)
|
||||
if err != nil {
|
||||
return StagedBundle{}, err
|
||||
}
|
||||
|
||||
if err := os.MkdirAll(opts.PipelineStagingPath, 0o755); err != nil {
|
||||
return StagedBundle{}, fmt.Errorf("create pipeline staging path: %w", err)
|
||||
}
|
||||
tempDir, err := os.MkdirTemp(opts.PipelineStagingPath, "."+opts.RunID+"-")
|
||||
if err != nil {
|
||||
return StagedBundle{}, fmt.Errorf("create staging temp dir: %w", err)
|
||||
}
|
||||
cleanupTemp := true
|
||||
defer func() {
|
||||
if cleanupTemp {
|
||||
_ = os.RemoveAll(tempDir)
|
||||
}
|
||||
}()
|
||||
|
||||
archivePath := filepath.Join(tempDir, "upload.archive")
|
||||
if err := writeLimited(ctx, archivePath, opts.Body, opts.MaxUploadSize); err != nil {
|
||||
return StagedBundle{}, err
|
||||
}
|
||||
|
||||
extractRoot := filepath.Join(tempDir, "bundle")
|
||||
if err := os.Mkdir(extractRoot, 0o755); err != nil {
|
||||
return StagedBundle{}, fmt.Errorf("create extraction root: %w", err)
|
||||
}
|
||||
if err := extractArchive(ctx, archivePath, extractRoot, format, opts.MaxExtractedSize, opts.MaxFileCount); err != nil {
|
||||
return StagedBundle{}, err
|
||||
}
|
||||
|
||||
manifest, err := sourcebundle.LoadManifest(extractRoot)
|
||||
if err != nil {
|
||||
return StagedBundle{}, err
|
||||
}
|
||||
if err := sourcebundle.ValidateBundle(extractRoot, manifest); err != nil {
|
||||
return StagedBundle{}, err
|
||||
}
|
||||
|
||||
finalRoot := filepath.Join(opts.PipelineStagingPath, opts.RunID)
|
||||
if err := os.Rename(extractRoot, finalRoot); err != nil {
|
||||
return StagedBundle{}, fmt.Errorf("commit staged bundle: %w", err)
|
||||
}
|
||||
cleanupTemp = false
|
||||
if err := os.RemoveAll(tempDir); err != nil {
|
||||
return StagedBundle{}, fmt.Errorf("remove staging temp dir: %w", err)
|
||||
}
|
||||
return StagedBundle{
|
||||
Root: finalRoot,
|
||||
Manifest: manifest,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func validateStageOptions(opts StageOptions) error {
|
||||
if opts.Body == nil {
|
||||
return fmt.Errorf("body is required")
|
||||
}
|
||||
if opts.PipelineStagingPath == "" {
|
||||
return fmt.Errorf("pipeline staging path is required")
|
||||
}
|
||||
if err := validateRunID(opts.RunID); err != nil {
|
||||
return err
|
||||
}
|
||||
if opts.MaxUploadSize <= 0 {
|
||||
return fmt.Errorf("max upload size must be greater than zero")
|
||||
}
|
||||
if opts.MaxExtractedSize <= 0 {
|
||||
return fmt.Errorf("max extracted size must be greater than zero")
|
||||
}
|
||||
if opts.MaxFileCount <= 0 {
|
||||
return fmt.Errorf("max file count must be greater than zero")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func validateRunID(value string) error {
|
||||
if value == "" {
|
||||
return fmt.Errorf("run id is required")
|
||||
}
|
||||
if value == "." || value == ".." || strings.ContainsAny(value, `/\`) {
|
||||
return fmt.Errorf("run id must be a single filesystem path segment")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
type archiveKind int
|
||||
|
||||
const (
|
||||
archiveKindTar archiveKind = iota + 1
|
||||
archiveKindGzip
|
||||
)
|
||||
|
||||
func archiveFormat(contentType string) (archiveKind, error) {
|
||||
mediaType, _, err := mime.ParseMediaType(contentType)
|
||||
if err != nil {
|
||||
mediaType = contentType
|
||||
}
|
||||
switch mediaType {
|
||||
case ContentTypeTar:
|
||||
return archiveKindTar, nil
|
||||
case ContentTypeGzip, ContentTypeXGzip:
|
||||
return archiveKindGzip, nil
|
||||
default:
|
||||
return 0, ErrUnsupportedContentType
|
||||
}
|
||||
}
|
||||
|
||||
func writeLimited(ctx context.Context, destination string, body io.Reader, maxSize int64) error {
|
||||
file, err := os.OpenFile(destination, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o600)
|
||||
if err != nil {
|
||||
return fmt.Errorf("create upload archive: %w", err)
|
||||
}
|
||||
defer file.Close()
|
||||
|
||||
limited := &limitedReader{ctx: ctx, reader: body, limit: maxSize}
|
||||
if _, err := io.Copy(file, limited); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := file.Close(); err != nil {
|
||||
return fmt.Errorf("write upload archive: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
type limitedReader struct {
|
||||
ctx context.Context
|
||||
reader io.Reader
|
||||
limit int64
|
||||
read int64
|
||||
}
|
||||
|
||||
func (r *limitedReader) Read(data []byte) (int, error) {
|
||||
if err := r.ctx.Err(); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
if r.read == r.limit {
|
||||
var probe [1]byte
|
||||
n, err := r.reader.Read(probe[:])
|
||||
if n > 0 {
|
||||
return 0, ErrUploadTooLarge
|
||||
}
|
||||
return 0, err
|
||||
}
|
||||
remaining := r.limit - r.read
|
||||
if int64(len(data)) > remaining+1 {
|
||||
data = data[:remaining+1]
|
||||
}
|
||||
n, err := r.reader.Read(data)
|
||||
if r.read+int64(n) > r.limit {
|
||||
allowed := int(r.limit - r.read)
|
||||
r.read = r.limit
|
||||
return allowed, ErrUploadTooLarge
|
||||
}
|
||||
r.read += int64(n)
|
||||
return n, err
|
||||
}
|
||||
|
||||
func extractArchive(ctx context.Context, archivePath, destination string, format archiveKind, maxExtractedSize int64, maxFileCount int) error {
|
||||
file, err := os.Open(archivePath)
|
||||
if err != nil {
|
||||
return fmt.Errorf("open upload archive: %w", err)
|
||||
}
|
||||
defer file.Close()
|
||||
|
||||
var reader io.Reader = file
|
||||
var gzipReader *gzip.Reader
|
||||
if format == archiveKindGzip {
|
||||
gzipReader, err = gzip.NewReader(file)
|
||||
if err != nil {
|
||||
return fmt.Errorf("open gzip archive: %w", err)
|
||||
}
|
||||
defer gzipReader.Close()
|
||||
reader = gzipReader
|
||||
}
|
||||
|
||||
extractor := archiveExtractor{
|
||||
ctx: ctx,
|
||||
destination: destination,
|
||||
maxExtractedSize: maxExtractedSize,
|
||||
maxFileCount: maxFileCount,
|
||||
}
|
||||
if err := extractor.extract(tar.NewReader(reader)); err != nil {
|
||||
return err
|
||||
}
|
||||
if extractor.rootManifestCount != 1 {
|
||||
return fmt.Errorf("archive must contain exactly one root-level manifest.json")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
type archiveExtractor struct {
|
||||
ctx context.Context
|
||||
destination string
|
||||
maxExtractedSize int64
|
||||
maxFileCount int
|
||||
extractedSize int64
|
||||
fileCount int
|
||||
rootManifestCount int
|
||||
seenFiles map[string]struct{}
|
||||
}
|
||||
|
||||
func (e *archiveExtractor) extract(reader *tar.Reader) error {
|
||||
e.seenFiles = make(map[string]struct{})
|
||||
for {
|
||||
if err := e.ctx.Err(); err != nil {
|
||||
return err
|
||||
}
|
||||
header, err := reader.Next()
|
||||
if errors.Is(err, io.EOF) {
|
||||
return nil
|
||||
}
|
||||
if err != nil {
|
||||
return fmt.Errorf("read tar archive: %w", err)
|
||||
}
|
||||
name, err := cleanArchivePath(header.Name)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if path.Base(name) == sourcebundle.ManifestName {
|
||||
if name != sourcebundle.ManifestName {
|
||||
return fmt.Errorf("nested manifest %q is not allowed", name)
|
||||
}
|
||||
e.rootManifestCount++
|
||||
}
|
||||
|
||||
switch header.Typeflag {
|
||||
case tar.TypeDir:
|
||||
if err := os.MkdirAll(filepath.Join(e.destination, filepath.FromSlash(name)), 0o755); err != nil {
|
||||
return fmt.Errorf("create archive directory %q: %w", name, err)
|
||||
}
|
||||
case tar.TypeReg, tar.TypeRegA:
|
||||
if err := e.extractFile(reader, name, header.Size); err != nil {
|
||||
return err
|
||||
}
|
||||
default:
|
||||
return fmt.Errorf("archive entry %q has unsupported type %c", name, header.Typeflag)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (e *archiveExtractor) extractFile(reader *tar.Reader, name string, size int64) error {
|
||||
if size < 0 {
|
||||
return fmt.Errorf("archive entry %q has invalid size", name)
|
||||
}
|
||||
e.fileCount++
|
||||
if e.fileCount > e.maxFileCount {
|
||||
return ErrTooManyFiles
|
||||
}
|
||||
e.extractedSize += size
|
||||
if e.extractedSize > e.maxExtractedSize {
|
||||
return ErrExtractedTooLarge
|
||||
}
|
||||
if _, exists := e.seenFiles[name]; exists {
|
||||
return fmt.Errorf("archive entry %q is duplicated", name)
|
||||
}
|
||||
e.seenFiles[name] = struct{}{}
|
||||
|
||||
fullPath := filepath.Join(e.destination, filepath.FromSlash(name))
|
||||
if err := os.MkdirAll(filepath.Dir(fullPath), 0o755); err != nil {
|
||||
return fmt.Errorf("create archive parent for %q: %w", name, err)
|
||||
}
|
||||
file, err := os.OpenFile(fullPath, os.O_WRONLY|os.O_CREATE|os.O_EXCL, 0o644)
|
||||
if err != nil {
|
||||
return fmt.Errorf("create archive file %q: %w", name, err)
|
||||
}
|
||||
defer file.Close()
|
||||
|
||||
if _, err := io.CopyN(file, reader, size); err != nil {
|
||||
return fmt.Errorf("extract archive file %q: %w", name, err)
|
||||
}
|
||||
if err := file.Close(); err != nil {
|
||||
return fmt.Errorf("extract archive file %q: %w", name, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func cleanArchivePath(value string) (string, error) {
|
||||
if value == "" || strings.Contains(value, `\`) || strings.HasPrefix(value, "/") {
|
||||
return "", ErrUnsafeArchivePath
|
||||
}
|
||||
cleaned := path.Clean(value)
|
||||
if cleaned != value {
|
||||
return "", ErrUnsafeArchivePath
|
||||
}
|
||||
for _, segment := range strings.Split(value, "/") {
|
||||
if segment == "" || segment == "." || segment == ".." {
|
||||
return "", ErrUnsafeArchivePath
|
||||
}
|
||||
}
|
||||
return value, nil
|
||||
}
|
||||
Reference in New Issue
Block a user