TA-Deployment-Broker/internal/app/upload.go

package app

import (
	"crypto/sha256"
	"database/sql"
	"encoding/hex"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"strings"
)

const uploadFormOverhead = 1 << 20

type byteCounter int64

type packageUploadTarget struct {
	Slug            string
	DisplayName     string
	PackageName     string
	PackageVersion  string
	PreviousName    string
	PreviousVersion string
	Enabled         bool
	Replacing       bool
}

func (counter *byteCounter) Write(value []byte) (int, error) {
	*counter += byteCounter(len(value))
	return len(value), nil
}

func (s *Server) handleUploadPackage(w http.ResponseWriter, r *http.Request) {
	tech, _ := s.currentTechnician(r)
	if r.ContentLength > s.cfg.MaxUploadBytes+uploadFormOverhead {
		writeJSON(w, http.StatusRequestEntityTooLarge, map[string]string{"error": "package exceeds the upload limit"})
		return
	}
	r.Body = http.MaxBytesReader(w, r.Body, s.cfg.MaxUploadBytes+uploadFormOverhead)

	reader, err := r.MultipartReader()
	if err != nil {
		writeJSON(w, http.StatusBadRequest, map[string]string{"error": "multipart upload is required"})
		return
	}
	fields := make(map[string]string)
	for {
		part, err := reader.NextPart()
		if errors.Is(err, io.EOF) {
			writeJSON(w, http.StatusBadRequest, map[string]string{"error": "package file is required"})
			return
		}
		if err != nil {
			writeJSON(w, http.StatusBadRequest, map[string]string{"error": "unable to read upload"})
			return
		}

		if part.FileName() == "" {
			value, err := io.ReadAll(io.LimitReader(part, 4097))
			_ = part.Close()
			if err != nil || len(value) > 4096 {
				writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid upload field"})
				return
			}
			fields[part.FormName()] = strings.TrimSpace(string(value))
			continue
		}
		if part.FormName() != "package_file" {
			_ = part.Close()
			writeJSON(w, http.StatusBadRequest, map[string]string{"error": "unexpected uploaded file"})
			return
		}

		fileName := part.FileName()
		if !validRegistrySegment(fileName, 255) {
			_ = part.Close()
			writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid package filename"})
			return
		}
		target, status, err := s.preparePackageUpload(r, fields)
		if err != nil {
			_ = part.Close()
			writeJSON(w, status, map[string]string{"error": err.Error()})
			return
		}

		checksum, size, status, err := s.streamPackageToGitea(
			r,
			target.PackageName,
			target.PackageVersion,
			fileName,
			part,
		)
		_ = part.Close()
		if err != nil {
			var maxBytesError *http.MaxBytesError
			if errors.As(err, &maxBytesError) {
				status = http.StatusRequestEntityTooLarge
			}
			_ = s.audit(r.Context(), "package_upload_failed", tech.Login, nil, "", fields["slug"], s.clientIP(r), err.Error())
			writeJSON(w, status, map[string]string{"error": err.Error()})
			return
		}

		if err := s.savePackage(
			r,
			target.Slug,
			target.DisplayName,
			target.PackageName,
			target.PackageVersion,
			fileName,
			checksum,
			target.Enabled,
		); err != nil {
			_ = s.deletePackageVersion(r, target.PackageName, target.PackageVersion)
			_ = s.audit(r.Context(), "package_upload_failed", tech.Login, nil, "", fields["slug"], s.clientIP(r), err.Error())
			writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "package uploaded but catalog update failed"})
			return
		}

		warning := ""
		if target.Replacing {
			if err := s.deletePackageVersion(r, target.PreviousName, target.PreviousVersion); err != nil {
				warning = fmt.Sprintf(
					"Package updated, but Gitea could not remove superseded version %s: %s",
					target.PreviousVersion,
					err,
				)
				_ = s.audit(
					r.Context(),
					"package_cleanup_failed",
					tech.Login,
					nil,
					"",
					target.Slug,
					s.clientIP(r),
					warning,
				)
			}
		}
		_ = s.audit(
			r.Context(),
			map[bool]string{true: "package_updated", false: "package_uploaded"}[target.Replacing],
			tech.Login,
			nil,
			"",
			target.Slug,
			s.clientIP(r),
			fmt.Sprintf(
				"version=%s filename=%s bytes=%d sha256=%s enabled=%t",
				target.PackageVersion, fileName, size, checksum, target.Enabled,
			),
		)
		writeJSON(w, http.StatusCreated, map[string]any{
			"status":   "uploaded",
			"sha256":   checksum,
			"bytes":    size,
			"filename": fileName,
			"warning":  warning,
		})
		return
	}
}

func (s *Server) preparePackageUpload(
	r *http.Request,
	fields map[string]string,
) (packageUploadTarget, int, error) {
	var target packageUploadTarget
	if !validSlug(fields["slug"]) {
		return target, http.StatusBadRequest, errors.New("invalid package ID")
	}
	if !validRegistrySegment(fields["package_version"], 100) {
		return target, http.StatusBadRequest, errors.New("invalid package version")
	}

	target.Slug = fields["slug"]
	target.PackageVersion = fields["package_version"]
	target.Enabled = fields["enabled"] == "1"

	err := s.db.QueryRowContext(
		r.Context(),
		`SELECT display_name, package_name, package_version
		   FROM packages
		  WHERE slug = ?`,
		target.Slug,
	).Scan(&target.DisplayName, &target.PackageName, &target.PreviousVersion)
	switch {
	case err == nil:
		if fields["mode"] == "create" {
			return target, http.StatusConflict, errors.New("that package ID already exists; use Update package")
		}
		if target.PackageVersion == target.PreviousVersion {
			return target, http.StatusConflict, errors.New("new version must differ from the current version")
		}
		target.PreviousName = target.PackageName
		target.Replacing = true
	case errors.Is(err, sql.ErrNoRows):
		if fields["mode"] == "update" {
			return target, http.StatusNotFound, errors.New("package ID was not found")
		}
		target.DisplayName = fields["display_name"]
		if target.DisplayName == "" || len(target.DisplayName) > 255 {
			return target, http.StatusBadRequest, errors.New("invalid display name")
		}
		target.PackageName = target.Slug
	default:
		return target, http.StatusInternalServerError, errors.New("unable to read package catalog")
	}

	if !validRegistrySegment(target.PackageName, 255) {
		return target, http.StatusInternalServerError, errors.New("catalog contains an invalid registry package name")
	}
	return target, http.StatusOK, nil
}

func validRegistrySegment(value string, maxLength int) bool {
	if value == "" || len(value) > maxLength {
		return false
	}
	for _, character := range value {
		if (character < 'a' || character > 'z') &&
			(character < 'A' || character > 'Z') &&
			(character < '0' || character > '9') &&
			character != '.' && character != '-' &&
			character != '+' && character != '_' {
			return false
		}
	}
	return true
}

func (s *Server) streamPackageToGitea(
	r *http.Request,
	packageName string,
	packageVersion string,
	fileName string,
	source io.Reader,
) (string, int64, int, error) {
	registryURL := joinURL(
		s.cfg.GiteaURL,
		fmt.Sprintf(
			"/api/packages/%s/generic/%s/%s/%s",
			url.PathEscape(s.cfg.GiteaPackageOwner),
			url.PathEscape(packageName),
			url.PathEscape(packageVersion),
			url.PathEscape(fileName),
		),
	)

	hasher := sha256.New()
	var size byteCounter
	body := io.TeeReader(source, io.MultiWriter(hasher, &size))
	request, err := http.NewRequestWithContext(r.Context(), http.MethodPut, registryURL, body)
	if err != nil {
		return "", 0, http.StatusInternalServerError, errors.New("unable to prepare registry upload")
	}
	request.Header.Set("Content-Type", "application/octet-stream")
	request.SetBasicAuth(s.cfg.GiteaWriteUser, s.cfg.GiteaWriteToken)

	response, err := s.packageClient.Do(request)
	if err != nil {
		var maxBytesError *http.MaxBytesError
		if errors.As(err, &maxBytesError) {
			return "", int64(size), http.StatusRequestEntityTooLarge, maxBytesError
		}
		return "", int64(size), http.StatusBadGateway, errors.New("package registry upload failed")
	}
	defer response.Body.Close()
	_, _ = io.Copy(io.Discard, io.LimitReader(response.Body, 8192))

	switch response.StatusCode {
	case http.StatusCreated:
	case http.StatusConflict:
		return "", int64(size), http.StatusConflict, errors.New("that package version and filename already exist")
	case http.StatusUnauthorized, http.StatusForbidden:
		return "", int64(size), http.StatusBadGateway, errors.New("package registry rejected the publisher credentials")
	default:
		return "", int64(size), http.StatusBadGateway, fmt.Errorf("package registry returned %s", response.Status)
	}
	return hex.EncodeToString(hasher.Sum(nil)), int64(size), http.StatusCreated, nil
}

func (s *Server) deletePackageVersion(
	r *http.Request,
	packageName string,
	packageVersion string,
) error {
	registryURL := joinURL(
		s.cfg.GiteaURL,
		fmt.Sprintf(
			"/api/packages/%s/generic/%s/%s",
			url.PathEscape(s.cfg.GiteaPackageOwner),
			url.PathEscape(packageName),
			url.PathEscape(packageVersion),
		),
	)
	request, err := http.NewRequestWithContext(r.Context(), http.MethodDelete, registryURL, nil)
	if err != nil {
		return errors.New("unable to prepare registry cleanup")
	}
	request.SetBasicAuth(s.cfg.GiteaWriteUser, s.cfg.GiteaWriteToken)

	response, err := s.packageClient.Do(request)
	if err != nil {
		return errors.New("package registry cleanup failed")
	}
	defer response.Body.Close()
	_, _ = io.Copy(io.Discard, io.LimitReader(response.Body, 8192))

	switch response.StatusCode {
	case http.StatusNoContent, http.StatusNotFound:
		return nil
	case http.StatusUnauthorized, http.StatusForbidden:
		return errors.New("package registry rejected the publisher credentials")
	default:
		return fmt.Errorf("package registry returned %s", response.Status)
	}
}