261 lines
7.3 KiB
Go
261 lines
7.3 KiB
Go
package http
|
|
|
|
import (
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"log/slog"
|
|
"net/http"
|
|
"time"
|
|
|
|
"git.clan.lol/clan/data-mesher/pkg/model"
|
|
"github.com/labstack/echo/v4"
|
|
)
|
|
|
|
func (s *Server) initFiles() {
|
|
s.echo.GET("/files", s.listFiles)
|
|
s.echo.PUT("/files/:name", s.putFile)
|
|
s.echo.DELETE("/files/:name", s.deleteFile)
|
|
}
|
|
|
|
func (s *Server) listFiles(c echo.Context) error {
|
|
// set content type for NDJSON
|
|
c.Response().Header().Set(echo.HeaderContentType, "application/x-ndjson")
|
|
c.Response().WriteHeader(http.StatusOK)
|
|
|
|
tx, err := s.files.BeginTx(false)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to begin read transaction: %w", err)
|
|
}
|
|
|
|
defer func() {
|
|
_ = tx.Rollback()
|
|
}()
|
|
|
|
r, err := s.files.ListSignatures(tx)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to list signatures: %w", err)
|
|
}
|
|
|
|
enc := json.NewEncoder(c.Response())
|
|
buf := make([]model.Signature, 16)
|
|
now := s.cfg.Now()
|
|
|
|
for {
|
|
n, readErr := r.Read(buf)
|
|
|
|
for _, sig := range buf[:n] {
|
|
// Hide expired files from listing — they'll be reaped by the
|
|
// sweeper but we don't want consumers to see them in the meantime.
|
|
if sig.Expired(now) {
|
|
continue
|
|
}
|
|
|
|
if err = enc.Encode(sig); err != nil {
|
|
return fmt.Errorf("failed to encode signature: %w", err)
|
|
}
|
|
|
|
// add a newline separator between each entry
|
|
if _, err = c.Response().Write([]byte("\n")); err != nil {
|
|
return fmt.Errorf("failed to write newline: %w", err)
|
|
}
|
|
}
|
|
|
|
if errors.Is(readErr, io.EOF) {
|
|
break
|
|
}
|
|
|
|
if readErr != nil {
|
|
return fmt.Errorf("failed to read signatures: %w", readErr)
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func (s *Server) putFile(ctx echo.Context) error {
|
|
reqCtx := ctx.Request().Context()
|
|
|
|
// create a container for the signature
|
|
sig := model.Signature{
|
|
Type: model.SignatureTypeFile,
|
|
}
|
|
|
|
// create a new binder for reading the signature fields from the http request
|
|
binder := echo.DefaultBinder{}
|
|
|
|
// bind path params (name)
|
|
if err := binder.BindPathParams(ctx, &sig); err != nil {
|
|
slog.ErrorContext(reqCtx, "failed to bind path params", "err", err)
|
|
|
|
return ctx.String(http.StatusBadRequest, "bad request") //nolint:wrapcheck
|
|
}
|
|
|
|
// bind headers (remaining signature fields)
|
|
if err := binder.BindHeaders(ctx, &sig); err != nil {
|
|
slog.ErrorContext(reqCtx, "failed to bind headers", "err", err)
|
|
|
|
return ctx.String(http.StatusBadRequest, "bad request") //nolint:wrapcheck
|
|
}
|
|
|
|
// Echo's default binder doesn't natively handle time.Duration, so we
|
|
// parse the X-Validfor header manually. Empty header means "no expiry".
|
|
if v := ctx.Request().Header.Get(HeaderValidFor); v != "" {
|
|
d, err := time.ParseDuration(v)
|
|
if err != nil {
|
|
slog.WarnContext(reqCtx, "invalid X-Validfor header", "value", v, "err", err)
|
|
|
|
return ctx.String(http.StatusBadRequest, "invalid X-Validfor header") //nolint:wrapcheck
|
|
}
|
|
|
|
if d < 0 {
|
|
return ctx.String(http.StatusBadRequest, "X-Validfor must be non-negative") //nolint:wrapcheck
|
|
}
|
|
|
|
if d > s.cfg.MaxValidFor {
|
|
return ctx.String(http.StatusBadRequest, //nolint:wrapcheck
|
|
fmt.Sprintf("X-Validfor exceeds maximum (%s)", s.cfg.MaxValidFor))
|
|
}
|
|
|
|
sig.ValidFor = d
|
|
}
|
|
|
|
// default to home network if no X-Network-Id header provided
|
|
if sig.NetworkID == nil {
|
|
sig.NetworkID = s.cfg.Network.ID
|
|
}
|
|
|
|
// get the size from the body's content length
|
|
sig.FileSize = ctx.Request().ContentLength
|
|
|
|
// check the file is not too large
|
|
if sig.FileSize > s.cfg.MaxFileSize {
|
|
slog.WarnContext(
|
|
reqCtx, "file too large",
|
|
slog.Int64("content_length", sig.FileSize),
|
|
slog.Int64("max_file_size", s.cfg.MaxFileSize),
|
|
)
|
|
|
|
return ctx.String(http.StatusRequestEntityTooLarge, //nolint:wrapcheck
|
|
fmt.Sprintf("file too large: max size is %d bytes", s.cfg.MaxFileSize))
|
|
}
|
|
|
|
// Validate required signature fields
|
|
// todo better validation
|
|
if sig.SignedBy == nil || len(sig.Signature) == 0 {
|
|
slog.ErrorContext(reqCtx, "missing required signature fields")
|
|
|
|
return ctx.String(http.StatusBadRequest, "bad request") //nolint:wrapcheck
|
|
}
|
|
|
|
// Check the signer is allowed for this file name
|
|
if err := s.cfg.SignatureAuth.Authorize(&sig); err != nil {
|
|
return ctx.String( //nolint:wrapcheck
|
|
http.StatusForbidden,
|
|
fmt.Sprintf("signer %s is not authorized for this file %s", sig.SignedBy, sig.Name),
|
|
)
|
|
}
|
|
|
|
// Reject files that are already expired on arrival.
|
|
// No clock-skew tolerance here: this is a local-client → local-daemon path, not gossip.
|
|
if sig.Expired(s.cfg.Now()) {
|
|
slog.WarnContext(
|
|
reqCtx, "rejecting already-expired file",
|
|
slog.String("name", sig.Name),
|
|
slog.Time("signed_at", sig.SignedAt),
|
|
slog.Duration("valid_for", sig.ValidFor),
|
|
)
|
|
|
|
return ctx.String(http.StatusBadRequest, "file is already expired") //nolint:wrapcheck
|
|
}
|
|
|
|
// Write the file
|
|
body := ctx.Request().Body
|
|
|
|
defer func() {
|
|
if err := body.Close(); err != nil {
|
|
slog.Error("failed to close body", slog.Any("err", err))
|
|
}
|
|
}()
|
|
|
|
// Wrap body with LimitReader to enforce Content-Length and detect overflow
|
|
limitedBody := io.LimitReader(body, sig.FileSize+1)
|
|
|
|
written, err := s.files.Put(&sig, limitedBody)
|
|
if err != nil {
|
|
slog.ErrorContext(reqCtx, "failed to write file", slog.Any("err", err))
|
|
|
|
return ctx.String(http.StatusInternalServerError, "internal server error") //nolint:wrapcheck
|
|
}
|
|
|
|
if !written {
|
|
//nolint:wrapcheck
|
|
return ctx.String(http.StatusConflict, fmt.Sprintf("newer version of file %s already exists", sig.Name))
|
|
}
|
|
|
|
return ctx.NoContent(http.StatusNoContent) //nolint:wrapcheck
|
|
}
|
|
|
|
// deleteFile creates a tombstone for the specified file.
|
|
// The tombstone will propagate across the cluster to delete the file everywhere.
|
|
// Request headers: X-Signedat, X-Signedby, X-Signature (signature over deletion).
|
|
func (s *Server) deleteFile(ctx echo.Context) error {
|
|
reqCtx := ctx.Request().Context()
|
|
|
|
sig := model.Signature{
|
|
Type: model.SignatureTypeTombstone,
|
|
}
|
|
|
|
binder := echo.DefaultBinder{}
|
|
|
|
// Bind path params (name)
|
|
if err := binder.BindPathParams(ctx, &sig); err != nil {
|
|
slog.ErrorContext(reqCtx, "failed to bind path params", "err", err)
|
|
|
|
return ctx.String(http.StatusBadRequest, "bad request") //nolint:wrapcheck
|
|
}
|
|
|
|
// Bind headers (signature)
|
|
if err := binder.BindHeaders(ctx, &sig); err != nil {
|
|
slog.ErrorContext(reqCtx, "failed to bind headers", "err", err)
|
|
|
|
return ctx.String(http.StatusBadRequest, "bad request") //nolint:wrapcheck
|
|
}
|
|
|
|
// default to home network if no X-Network-Id header provided
|
|
if sig.NetworkID == nil {
|
|
sig.NetworkID = s.cfg.Network.ID
|
|
}
|
|
|
|
// Validate required signature fields
|
|
if sig.SignedBy == nil || len(sig.Signature) == 0 {
|
|
slog.ErrorContext(reqCtx, "missing required signature fields")
|
|
|
|
return ctx.String(http.StatusBadRequest, "bad request") //nolint:wrapcheck
|
|
}
|
|
|
|
// Check the signer is allowed for this file name
|
|
if err := s.cfg.SignatureAuth.Authorize(&sig); err != nil {
|
|
return ctx.String( //nolint:wrapcheck
|
|
http.StatusForbidden,
|
|
fmt.Sprintf("signer %s is not authorized for this file %s", sig.SignedBy, sig.Name),
|
|
)
|
|
}
|
|
|
|
// Write the tombstone
|
|
written, err := s.files.PutTombstone(&sig)
|
|
if err != nil {
|
|
slog.ErrorContext(reqCtx, "failed to write tombstone", "err", err)
|
|
|
|
return ctx.String(http.StatusInternalServerError, "internal server error") //nolint:wrapcheck
|
|
}
|
|
|
|
if !written {
|
|
//nolint:wrapcheck
|
|
return ctx.String(http.StatusConflict, fmt.Sprintf("newer version of file %s already exists", sig.Name))
|
|
}
|
|
|
|
return ctx.NoContent(http.StatusNoContent) //nolint:wrapcheck
|
|
}
|