diff options
| author | 2026-03-26 09:14:59 +0000 | |
|---|---|---|
| committer | 2026-03-26 09:14:59 +0000 | |
| commit | 3d25bda9d5da6814661828adabe8a09f9d01aefb (patch) | |
| tree | d034e28079333f85e5d7b96d921282eddd4798d6 /network/receivepack/service | |
| parent | object/id: Empty tree (diff) | |
| signature | No signature | |
network/receivepack: Rename from receivepack
Diffstat (limited to 'network/receivepack/service')
| -rw-r--r-- | network/receivepack/service/apply.go | 134 | ||||
| -rw-r--r-- | network/receivepack/service/command.go | 32 | ||||
| -rw-r--r-- | network/receivepack/service/command_result.go | 13 | ||||
| -rw-r--r-- | network/receivepack/service/doc.go | 6 | ||||
| -rw-r--r-- | network/receivepack/service/execute.go | 123 | ||||
| -rw-r--r-- | network/receivepack/service/hook.go | 45 | ||||
| -rw-r--r-- | network/receivepack/service/hook_apply.go | 44 | ||||
| -rw-r--r-- | network/receivepack/service/ingest_quarantine.go | 144 | ||||
| -rw-r--r-- | network/receivepack/service/options.go | 36 | ||||
| -rw-r--r-- | network/receivepack/service/quarantine.go | 274 | ||||
| -rw-r--r-- | network/receivepack/service/quarantine_test.go | 184 | ||||
| -rw-r--r-- | network/receivepack/service/request.go | 16 | ||||
| -rw-r--r-- | network/receivepack/service/result.go | 14 | ||||
| -rw-r--r-- | network/receivepack/service/run_hook.go | 168 | ||||
| -rw-r--r-- | network/receivepack/service/service.go | 16 | ||||
| -rw-r--r-- | network/receivepack/service/service_test.go | 99 | ||||
| -rw-r--r-- | network/receivepack/service/update.go | 12 |
17 files changed, 1360 insertions, 0 deletions
diff --git a/network/receivepack/service/apply.go b/network/receivepack/service/apply.go new file mode 100644 index 00000000..8fa500ca --- /dev/null +++ b/network/receivepack/service/apply.go @@ -0,0 +1,134 @@ +package service + +import ( + "codeberg.org/lindenii/furgit/internal/utils" + objectid "codeberg.org/lindenii/furgit/object/id" + refstore "codeberg.org/lindenii/furgit/ref/store" +) + +func (service *Service) applyAtomic(result *Result, commands []Command) error { + total := len(commands) + utils.BestEffortFprintf(service.opts.Progress, "updating refs: 0/%d\r", total) + + tx, err := service.opts.Refs.BeginTransaction() + if err != nil { + return err + } + + for i, command := range commands { + err = queueWriteTransaction(tx, command) + if err != nil { + _ = tx.Abort() + + fillCommandErrors(result, commands, err.Error()) + utils.BestEffortFprintf(service.opts.Progress, "updating refs: failed at %d/%d.\n", i+1, total) + + return nil + } + + utils.BestEffortFprintf(service.opts.Progress, "updating refs: %d/%d\r", i+1, total) + } + + err = tx.Commit() + if err != nil { + fillCommandErrors(result, commands, err.Error()) + utils.BestEffortFprintf(service.opts.Progress, "updating refs: failed at commit.\n") + + return nil + } + + result.Applied = true + for _, command := range commands { + result.Commands = append(result.Commands, successCommandResult(command)) + } + + utils.BestEffortFprintf(service.opts.Progress, "updating refs: done.\n") + + return nil +} + +func (service *Service) applyBatch(result *Result, commands []Command) error { + total := len(commands) + + utils.BestEffortFprintf(service.opts.Progress, "updating refs...\r") + + batch, err := service.opts.Refs.BeginBatch() + if err != nil { + return err + } + + for _, command := range commands { + queueWriteBatch(batch, command) + } + + batchResults, err := batch.Apply() + if err != nil && len(batchResults) == 0 { + utils.BestEffortFprintf(service.opts.Progress, "updating refs: failed at apply.\n") + + return err + } + + appliedAny := false + failedCount := 0 + + for i, command := range commands { + item := successCommandResult(command) + if i < len(batchResults) && batchResults[i].Error != nil { + item.Error = batchResults[i].Error.Error() + failedCount++ + } else { + appliedAny = true + } + + result.Commands = append(result.Commands, item) + + utils.BestEffortFprintf(service.opts.Progress, "updating refs: %d/%d\r", i+1, total) + } + + result.Applied = appliedAny + + if failedCount == 0 { + utils.BestEffortFprintf(service.opts.Progress, "updating refs: done.\n") + } else { + utils.BestEffortFprintf(service.opts.Progress, "updating refs: failed (%d/%d).\n", failedCount, total) + } + + return nil +} + +func queueWriteTransaction(tx refstore.Transaction, command Command) error { + if isDelete(command) { + return tx.Delete(command.Name, command.OldID) + } + + if command.OldID == objectid.Zero(command.OldID.Algorithm()) { + return tx.Create(command.Name, command.NewID) + } + + return tx.Update(command.Name, command.NewID, command.OldID) +} + +func queueWriteBatch(batch refstore.Batch, command Command) { + if isDelete(command) { + batch.Delete(command.Name, command.OldID) + + return + } + + if command.OldID == objectid.Zero(command.OldID.Algorithm()) { + batch.Create(command.Name, command.NewID) + + return + } + + batch.Update(command.Name, command.NewID, command.OldID) +} + +func successCommandResult(command Command) CommandResult { + return CommandResult{ + Name: command.Name, + RefName: command.Name, + OldID: objectIDPointer(command.OldID), + NewID: objectIDPointer(command.NewID), + } +} diff --git a/network/receivepack/service/command.go b/network/receivepack/service/command.go new file mode 100644 index 00000000..0fd8961e --- /dev/null +++ b/network/receivepack/service/command.go @@ -0,0 +1,32 @@ +package service + +import objectid "codeberg.org/lindenii/furgit/object/id" + +// Command is one protocol-independent requested ref update. +type Command struct { + OldID objectid.ObjectID + NewID objectid.ObjectID + Name string +} + +func fillCommandErrors(result *Result, commands []Command, errText string) { + for _, command := range commands { + result.Commands = append(result.Commands, CommandResult{ + Name: command.Name, + Error: errText, + RefName: command.Name, + OldID: objectIDPointer(command.OldID), + NewID: objectIDPointer(command.NewID), + }) + } +} + +func isDelete(command Command) bool { + return command.NewID == objectid.Zero(command.NewID.Algorithm()) +} + +func objectIDPointer(id objectid.ObjectID) *objectid.ObjectID { + out := id + + return &out +} diff --git a/network/receivepack/service/command_result.go b/network/receivepack/service/command_result.go new file mode 100644 index 00000000..37549f08 --- /dev/null +++ b/network/receivepack/service/command_result.go @@ -0,0 +1,13 @@ +package service + +import objectid "codeberg.org/lindenii/furgit/object/id" + +// CommandResult is one per-command execution result. +type CommandResult struct { + Name string + Error string + RefName string + OldID *objectid.ObjectID + NewID *objectid.ObjectID + ForcedUpdate bool +} diff --git a/network/receivepack/service/doc.go b/network/receivepack/service/doc.go new file mode 100644 index 00000000..37be23f4 --- /dev/null +++ b/network/receivepack/service/doc.go @@ -0,0 +1,6 @@ +// Package service implements the protocol-independent receive-pack service. +// +// A Service borrows the stores, roots, hooks, and I/O endpoints supplied in +// Options. Callers retain ownership of those dependencies and must keep them +// valid for each Execute call that uses them. +package service diff --git a/network/receivepack/service/execute.go b/network/receivepack/service/execute.go new file mode 100644 index 00000000..9f373e0d --- /dev/null +++ b/network/receivepack/service/execute.go @@ -0,0 +1,123 @@ +package service + +import ( + "context" + "os" + + "codeberg.org/lindenii/furgit/internal/utils" +) + +// Execute validates one receive-pack request, optionally ingests its pack into +// quarantine, runs the optional hook, and applies allowed ref updates. +func (service *Service) Execute(ctx context.Context, req *Request) (*Result, error) { + result := &Result{ + Commands: make([]CommandResult, 0, len(req.Commands)), + } + + var ( + quarantineName string + quarantineRoot *os.Root + err error + ) + + quarantineName, quarantineRoot, ok := service.ingestQuarantine(result, req.Commands, req) + if !ok { + return result, nil + } + + if quarantineRoot != nil { + defer func() { + _ = quarantineRoot.Close() + _ = service.opts.ObjectsRoot.RemoveAll(quarantineName) + }() + } + + for _, command := range req.Commands { + result.Planned = append(result.Planned, PlannedUpdate{ + Name: command.Name, + OldID: command.OldID, + NewID: command.NewID, + Delete: isDelete(command), + }) + } + + if len(req.Commands) == 0 { + return result, nil + } + + allowedCommands, allowedIndices, rejected, ok, errText := service.runHook( + ctx, + req, + req.Commands, + quarantineName, + ) + if !ok { + fillCommandErrors(result, req.Commands, errText) + + return result, nil + } + + if req.Atomic && len(rejected) != 0 { + result.Commands = make([]CommandResult, 0, len(req.Commands)) + for index, command := range req.Commands { + message := rejected[index] + if message == "" { + message = "atomic push rejected by hook" + } + + result.Commands = append(result.Commands, resultForHookRejection(command, message)) + } + + return result, nil + } + + if len(allowedCommands) == 0 { + result.Commands = mergeCommandResults(req.Commands, rejected, nil, nil) + + return result, nil + } + + if req.PackExpected && quarantineRoot != nil { + // Git migrates quarantined objects into permanent storage immediately + // before starting ref updates. + utils.BestEffortFprintf(service.opts.Progress, "promoting quarantine...\r") + + err = service.promoteQuarantine(quarantineName, quarantineRoot) + if err != nil { + utils.BestEffortFprintf(service.opts.Progress, "promoting quarantine: failed: %v.\n", err) + + result.UnpackError = err.Error() + fillCommandErrors(result, req.Commands, err.Error()) + + return result, nil + } + + utils.BestEffortFprintf(service.opts.Progress, "promoting quarantine: done.\n") + } + + if req.Atomic { + subresult := &Result{} + + err := service.applyAtomic(subresult, allowedCommands) + if err != nil { + return result, err + } + + result.Commands = mergeCommandResults(req.Commands, rejected, subresult.Commands, allowedIndices) + result.Applied = subresult.Applied + + return result, nil + } + + subresult := &Result{} + + err = service.applyBatch(subresult, allowedCommands) + if err != nil { + return result, err + } + + result.Commands = mergeCommandResults(req.Commands, rejected, subresult.Commands, allowedIndices) + result.Applied = subresult.Applied + + return result, nil +} diff --git a/network/receivepack/service/hook.go b/network/receivepack/service/hook.go new file mode 100644 index 00000000..750720dd --- /dev/null +++ b/network/receivepack/service/hook.go @@ -0,0 +1,45 @@ +package service + +import ( + "context" + "io" + + objectid "codeberg.org/lindenii/furgit/object/id" + objectstorer "codeberg.org/lindenii/furgit/object/storer" + refstore "codeberg.org/lindenii/furgit/ref/store" +) + +type HookIO struct { + Progress io.Writer + Error io.Writer +} + +type RefUpdate struct { + Name string + OldID objectid.ObjectID + NewID objectid.ObjectID +} + +type UpdateDecision struct { + Accept bool + Message string +} + +// HookRequest is the borrowed view passed to one Hook invocation. +// +// Refs, ExistingObjects, and QuarantinedObjects are borrowed and are only +// valid for the duration of the hook call. +type HookRequest struct { + Refs refstore.ReadingStore + ExistingObjects objectstorer.Store + QuarantinedObjects objectstorer.Store + Updates []RefUpdate + PushOptions []string + IO HookIO +} + +// Hook is an optional per-request validation hook. +// +// Hook borrows the data and stores in HookRequest only for the duration of the +// call. +type Hook func(context.Context, HookRequest) ([]UpdateDecision, error) diff --git a/network/receivepack/service/hook_apply.go b/network/receivepack/service/hook_apply.go new file mode 100644 index 00000000..5bd8f596 --- /dev/null +++ b/network/receivepack/service/hook_apply.go @@ -0,0 +1,44 @@ +package service + +func buildHookUpdates(commands []Command) []RefUpdate { + updates := make([]RefUpdate, 0, len(commands)) + for _, command := range commands { + updates = append(updates, RefUpdate{ + Name: command.Name, + OldID: command.OldID, + NewID: command.NewID, + }) + } + + return updates +} + +func resultForHookRejection(command Command, message string) CommandResult { + result := successCommandResult(command) + result.Error = message + + return result +} + +func mergeCommandResults( + commands []Command, + rejected map[int]string, + applied []CommandResult, + appliedIndices []int, +) []CommandResult { + out := make([]CommandResult, len(commands)) + + for index, message := range rejected { + out[index] = resultForHookRejection(commands[index], message) + } + + for i, appliedResult := range applied { + if i >= len(appliedIndices) { + break + } + + out[appliedIndices[i]] = appliedResult + } + + return out +} diff --git a/network/receivepack/service/ingest_quarantine.go b/network/receivepack/service/ingest_quarantine.go new file mode 100644 index 00000000..8e3e2455 --- /dev/null +++ b/network/receivepack/service/ingest_quarantine.go @@ -0,0 +1,144 @@ +package service + +import ( + "os" + + "codeberg.org/lindenii/furgit/internal/utils" + "codeberg.org/lindenii/furgit/packfile/ingest" +) + +func (service *Service) ingestQuarantine( + result *Result, + commands []Command, + req *Request, +) (string, *os.Root, bool) { + if !req.PackExpected { + return "", nil, true + } + + if req.Pack == nil { + utils.BestEffortFprintf(service.opts.Progress, "unpack failed: missing pack stream.\n") + + result.UnpackError = "missing pack stream" + fillCommandErrors(result, commands, "missing pack stream") + + return "", nil, false + } + + if service.opts.ObjectsRoot == nil { + utils.BestEffortFprintf(service.opts.Progress, "unpack failed: objects root not configured.\n") + + result.UnpackError = "objects root not configured" + fillCommandErrors(result, commands, "objects root not configured") + + return "", nil, false + } + + var err error + + err = service.opts.ExistingObjects.Refresh() + if err != nil { + utils.BestEffortFprintf(service.opts.Progress, "unpack failed: refresh existing objects: %v.\n", err) + + result.UnpackError = err.Error() + fillCommandErrors(result, commands, err.Error()) + + return "", nil, false + } + + pending, err := ingest.Ingest( + req.Pack, + service.opts.Algorithm, + ingest.Options{ + FixThin: true, + WriteRev: true, + Base: service.opts.ExistingObjects, + Progress: service.opts.Progress, + ProgressFlush: service.opts.ProgressFlush, + }, + ) + if err != nil { + utils.BestEffortFprintf(service.opts.Progress, "unpack failed: %v.\n", err) + + result.UnpackError = err.Error() + fillCommandErrors(result, commands, err.Error()) + + return "", nil, false + } + + if pending.Header().ObjectCount == 0 { + discarded, err := pending.Discard() + if err != nil { + utils.BestEffortFprintf(service.opts.Progress, "unpack failed: %v.\n", err) + + result.UnpackError = err.Error() + fillCommandErrors(result, commands, err.Error()) + + return "", nil, false + } + + result.Ingest = &ingest.Result{ + PackHash: discarded.PackHash, + ObjectCount: discarded.ObjectCount, + } + + utils.BestEffortFprintf( + service.opts.Progress, + "unpacking: done (%d objects, %s).\n", + discarded.ObjectCount, + discarded.PackHash, + ) + + return "", nil, true + } + + utils.BestEffortFprintf(service.opts.Progress, "creating quarantine...\r") + + quarantineName, quarantineRoot, err := service.createQuarantineRoot() + if err != nil { + utils.BestEffortFprintf(service.opts.Progress, "unpack failed: %v.\n", err) + + result.UnpackError = err.Error() + fillCommandErrors(result, commands, err.Error()) + + return "", nil, false + } + + quarantinePackRoot, err := service.openQuarantinePackRoot(quarantineRoot) + if err != nil { + utils.BestEffortFprintf(service.opts.Progress, "unpack failed: %v.\n", err) + + result.UnpackError = err.Error() + fillCommandErrors(result, commands, err.Error()) + + _ = quarantineRoot.Close() + _ = service.opts.ObjectsRoot.RemoveAll(quarantineName) + + return "", nil, false + } + + utils.BestEffortFprintf(service.opts.Progress, "creating quarantine: done.\n") + utils.BestEffortFprintf(service.opts.Progress, "unpacking...\r") + + ingested, err := pending.Continue(quarantinePackRoot) + + _ = quarantinePackRoot.Close() + + if err != nil { + utils.BestEffortFprintf(service.opts.Progress, "unpack failed: %v.\n", err) + + result.UnpackError = err.Error() + fillCommandErrors(result, commands, err.Error()) + + _ = quarantineRoot.Close() + _ = service.opts.ObjectsRoot.RemoveAll(quarantineName) + + return "", nil, false + } + + utils.BestEffortFprintf(service.opts.Progress, "unpacking: done (%d objects, %s).\n", ingested.ObjectCount, ingested.PackHash) + + result.Ingest = &ingested + + return quarantineName, quarantineRoot, true +} diff --git a/network/receivepack/service/options.go b/network/receivepack/service/options.go new file mode 100644 index 00000000..7edf4f06 --- /dev/null +++ b/network/receivepack/service/options.go @@ -0,0 +1,36 @@ +package service + +import ( + "io" + "io/fs" + "os" + + objectid "codeberg.org/lindenii/furgit/object/id" + objectstorer "codeberg.org/lindenii/furgit/object/storer" + refstore "codeberg.org/lindenii/furgit/ref/store" +) + +type PromotedObjectPermissions struct { + DirMode fs.FileMode + FileMode fs.FileMode +} + +// Options configures one protocol-independent receive-pack service. +// +// Service borrows all configured dependencies. +// +// Refs and ExistingObjects are required and must be non-nil. +// ObjectsRoot is required if Execute may need to ingest or promote a pack. +// Progress, ProgressFlush, Hook, and HookIO are optional; when provided they +// are also borrowed for the duration of Execute. +type Options struct { + Algorithm objectid.Algorithm + Refs refstore.ReadWriteStore + ExistingObjects objectstorer.Store + ObjectsRoot *os.Root + Progress io.Writer + ProgressFlush func() error + PromotedObjectPermissions *PromotedObjectPermissions + Hook Hook + HookIO HookIO +} diff --git a/network/receivepack/service/quarantine.go b/network/receivepack/service/quarantine.go new file mode 100644 index 00000000..0bd98aeb --- /dev/null +++ b/network/receivepack/service/quarantine.go @@ -0,0 +1,274 @@ +package service + +import ( + "bytes" + "crypto/rand" + "errors" + "fmt" + "io" + "io/fs" + "os" + "path" + "slices" +) + +// createQuarantineRoot creates one per-push quarantine directory beneath the +// permanent objects root. +// +// It returns both the quarantine directory name relative to ObjectsRoot and an +// opened root for that directory. Callers use the name for later promotion or +// removal relative to ObjectsRoot, and use the opened root for capability-based +// access within the quarantine itself. +func (service *Service) createQuarantineRoot() (string, *os.Root, error) { + name := "tmp_objdir-incoming-" + rand.Text() + + err := service.opts.ObjectsRoot.Mkdir(name, 0o700) + if err != nil { + return "", nil, err + } + + root, err := service.opts.ObjectsRoot.OpenRoot(name) + if err != nil { + _ = service.opts.ObjectsRoot.RemoveAll(name) + + return "", nil, err + } + + return name, root, nil +} + +func (service *Service) openQuarantinePackRoot(quarantineRoot *os.Root) (*os.Root, error) { + err := quarantineRoot.Mkdir("pack", 0o755) + if err != nil && !os.IsExist(err) { + return nil, err + } + + return quarantineRoot.OpenRoot("pack") +} + +func (service *Service) promoteQuarantine(quarantineName string, quarantineRoot *os.Root) error { + if quarantineName == "" || quarantineRoot == nil { + return nil + } + + return service.promoteQuarantineDir(quarantineName, quarantineRoot, ".") +} + +func (service *Service) promoteQuarantineDir(quarantineName string, quarantineRoot *os.Root, rel string) error { + entries, err := fs.ReadDir(quarantineRoot.FS(), rel) + if err != nil && !os.IsNotExist(err) { + return err + } + + slices.SortFunc(entries, func(left, right fs.DirEntry) int { + return packCopyPriority(left.Name()) - packCopyPriority(right.Name()) + }) + + for _, entry := range entries { + childRel := entry.Name() + if rel != "." { + childRel = path.Join(rel, entry.Name()) + } + + if entry.IsDir() { + err = service.opts.ObjectsRoot.Mkdir(childRel, 0o755) + if err != nil && !os.IsExist(err) { + return err + } + + err = service.applyPromotedDirectoryPermissions(childRel) + if err != nil { + return err + } + + err = service.promoteQuarantineDir(quarantineName, quarantineRoot, childRel) + if err != nil { + return err + } + + continue + } + + err = finalizeQuarantineFile( + service.opts.ObjectsRoot, + path.Join(quarantineName, childRel), + childRel, + isLooseObjectShardPath(rel), + service.opts.PromotedObjectPermissions, + ) + if err == nil { + continue + } + + return err + } + + return nil +} + +func packCopyPriority(name string) int { + if !pathHasPackPrefix(name) { + return 0 + } + + switch { + case path.Ext(name) == ".keep": + return 1 + case path.Ext(name) == ".pack": + return 2 + case path.Ext(name) == ".rev": + return 3 + case path.Ext(name) == ".idx": + return 4 + default: + return 5 + } +} + +func pathHasPackPrefix(name string) bool { + return len(name) >= 4 && name[:4] == "pack" +} + +func isLooseObjectShardPath(rel string) bool { + return len(rel) == 2 && isHex(rel[0]) && isHex(rel[1]) +} + +func isHex(ch byte) bool { + return ('0' <= ch && ch <= '9') || ('a' <= ch && ch <= 'f') || ('A' <= ch && ch <= 'F') +} + +func (service *Service) applyPromotedDirectoryPermissions(name string) error { + if service.opts.PromotedObjectPermissions == nil { + return nil + } + + return service.opts.ObjectsRoot.Chmod(name, service.opts.PromotedObjectPermissions.DirMode) +} + +func applyPromotedFilePermissions( + root *os.Root, + name string, + perms *PromotedObjectPermissions, +) error { + if perms == nil { + return nil + } + + return root.Chmod(name, perms.FileMode) +} + +func finalizeQuarantineFile( + root *os.Root, + src, dst string, + skipCollisionCheck bool, + perms *PromotedObjectPermissions, +) error { + const maxVanishedRetries = 5 + + for retries := 0; ; retries++ { + err := root.Link(src, dst) + switch { + case err == nil: + _ = root.Remove(src) + + return applyPromotedFilePermissions(root, dst, perms) + case !errors.Is(err, fs.ErrExist): + _, statErr := root.Stat(dst) + switch { + case statErr == nil: + err = fs.ErrExist + case errors.Is(statErr, fs.ErrNotExist): + renameErr := root.Rename(src, dst) + if renameErr == nil { + return applyPromotedFilePermissions(root, dst, perms) + } + + err = renameErr + default: + _ = root.Remove(src) + + return statErr + } + } + + if !errors.Is(err, fs.ErrExist) { + _ = root.Remove(src) + + return fmt.Errorf("promote quarantine %q -> %q: %w", src, dst, err) + } + + if skipCollisionCheck { + _ = root.Remove(src) + + return applyPromotedFilePermissions(root, dst, perms) + } + + equal, vanished, cmpErr := compareRootFiles(root, src, dst) + if vanished { + if retries >= maxVanishedRetries { + return fmt.Errorf("promote quarantine %q -> %q: destination repeatedly vanished", src, dst) + } + + continue + } + + if cmpErr != nil { + return cmpErr + } + + if !equal { + return fmt.Errorf("promote quarantine %q -> %q: files differ in contents", src, dst) + } + + _ = root.Remove(src) + + return applyPromotedFilePermissions(root, dst, perms) + } +} + +func compareRootFiles(root *os.Root, left, right string) (equal bool, vanished bool, err error) { + leftFile, err := root.Open(left) + if err != nil { + return false, false, err + } + + defer func() { + _ = leftFile.Close() + }() + + rightFile, err := root.Open(right) + if err != nil { + if errors.Is(err, fs.ErrNotExist) { + return false, true, nil + } + + return false, false, err + } + + defer func() { + _ = rightFile.Close() + }() + + var leftBuf, rightBuf [4096]byte + + for { + leftN, leftErr := leftFile.Read(leftBuf[:]) + rightN, rightErr := rightFile.Read(rightBuf[:]) + + if leftErr != nil && !errors.Is(leftErr, io.EOF) { + return false, false, leftErr + } + + if rightErr != nil && !errors.Is(rightErr, io.EOF) { + return false, false, rightErr + } + + if leftN != rightN || !bytes.Equal(leftBuf[:leftN], rightBuf[:rightN]) { + return false, false, nil + } + + if leftErr != nil || rightErr != nil { + return true, false, nil + } + } +} diff --git a/network/receivepack/service/quarantine_test.go b/network/receivepack/service/quarantine_test.go new file mode 100644 index 00000000..86299be2 --- /dev/null +++ b/network/receivepack/service/quarantine_test.go @@ -0,0 +1,184 @@ +package service //nolint:testpackage + +// because we need access to quarantine internals + +import ( + "os" + "path" + "testing" + + objectid "codeberg.org/lindenii/furgit/object/id" + "codeberg.org/lindenii/furgit/object/storer/memory" +) + +type quarantineFixture struct { + svc *Service + objectsRoot *os.Root + quarantineName string + quarantineRoot *os.Root +} + +func newQuarantineFixture(tb testing.TB, opts Options) *quarantineFixture { + tb.Helper() + + objectsRoot, err := os.OpenRoot(tb.TempDir()) + if err != nil { + tb.Fatalf("os.OpenRoot: %v", err) + } + + tb.Cleanup(func() { + _ = objectsRoot.Close() + }) + + opts.Algorithm = objectid.AlgorithmSHA1 + opts.ExistingObjects = memory.New(objectid.AlgorithmSHA1) + opts.ObjectsRoot = objectsRoot + + svc := New(opts) + + quarantineName, quarantineRoot, err := svc.createQuarantineRoot() + if err != nil { + tb.Fatalf("createQuarantineRoot: %v", err) + } + + tb.Cleanup(func() { + _ = quarantineRoot.Close() + _ = objectsRoot.RemoveAll(quarantineName) + }) + + return &quarantineFixture{ + svc: svc, + objectsRoot: objectsRoot, + quarantineName: quarantineName, + quarantineRoot: quarantineRoot, + } +} + +func writeMatchingPromotedFile( + tb testing.TB, + quarantineRoot, objectsRoot *os.Root, + dir, name, payload string, +) { + tb.Helper() + + err := quarantineRoot.Mkdir(dir, 0o755) + if err != nil { + tb.Fatalf("Mkdir(%s): %v", dir, err) + } + + err = objectsRoot.Mkdir(dir, 0o755) + if err != nil { + tb.Fatalf("Mkdir(dst %s): %v", dir, err) + } + + rel := path.Join(dir, name) + + err = quarantineRoot.WriteFile(rel, []byte(payload), 0o644) + if err != nil { + tb.Fatalf("WriteFile(quarantine %s): %v", rel, err) + } + + err = objectsRoot.WriteFile(rel, []byte(payload), 0o644) + if err != nil { + tb.Fatalf("WriteFile(permanent %s): %v", rel, err) + } +} + +func TestPromoteQuarantineAppliesConfiguredPermissions(t *testing.T) { + t.Parallel() + + fx := newQuarantineFixture(t, Options{ + PromotedObjectPermissions: &PromotedObjectPermissions{ + DirMode: 0o751, + FileMode: 0o640, + }, + }) + + err := fx.quarantineRoot.Mkdir("ab", 0o700) + if err != nil { + t.Fatalf("Mkdir(ab): %v", err) + } + + err = fx.quarantineRoot.WriteFile(path.Join("ab", "cdef"), []byte("payload"), 0o600) + if err != nil { + t.Fatalf("WriteFile(quarantine loose): %v", err) + } + + err = fx.svc.promoteQuarantine(fx.quarantineName, fx.quarantineRoot) + if err != nil { + t.Fatalf("promoteQuarantine: %v", err) + } + + dirInfo, err := fx.objectsRoot.Stat("ab") + if err != nil { + t.Fatalf("Stat(ab): %v", err) + } + + if got := dirInfo.Mode().Perm(); got != 0o751 { + t.Fatalf("dir mode = %o, want 751", got) + } + + fileInfo, err := fx.objectsRoot.Stat(path.Join("ab", "cdef")) + if err != nil { + t.Fatalf("Stat(ab/cdef): %v", err) + } + + if got := fileInfo.Mode().Perm(); got != 0o640 { + t.Fatalf("file mode = %o, want 640", got) + } +} + +func TestPromoteQuarantineTreatsExistingLooseObjectAsSuccess(t *testing.T) { + t.Parallel() + + fx := newQuarantineFixture(t, Options{}) + writeMatchingPromotedFile(t, fx.quarantineRoot, fx.objectsRoot, "ab", "cdef", "same object bytes") + + err := fx.svc.promoteQuarantine(fx.quarantineName, fx.quarantineRoot) + if err != nil { + t.Fatalf("promoteQuarantine: %v", err) + } +} + +func TestPromoteQuarantineRejectsDifferentExistingPackFile(t *testing.T) { + t.Parallel() + + fx := newQuarantineFixture(t, Options{}) + + err := fx.quarantineRoot.Mkdir("pack", 0o755) + if err != nil { + t.Fatalf("Mkdir(pack): %v", err) + } + + err = fx.objectsRoot.Mkdir("pack", 0o755) + if err != nil { + t.Fatalf("Mkdir(dst pack): %v", err) + } + + err = fx.quarantineRoot.WriteFile(path.Join("pack", "pack-a.pack"), []byte("new bytes"), 0o644) + if err != nil { + t.Fatalf("WriteFile(quarantine pack): %v", err) + } + + err = fx.objectsRoot.WriteFile(path.Join("pack", "pack-a.pack"), []byte("old bytes"), 0o644) + if err != nil { + t.Fatalf("WriteFile(permanent pack): %v", err) + } + + err = fx.svc.promoteQuarantine(fx.quarantineName, fx.quarantineRoot) + if err == nil { + t.Fatal("promoteQuarantine unexpectedly succeeded") + } +} + +func TestPromoteQuarantineAcceptsMatchingExistingPackFile(t *testing.T) { + t.Parallel() + + fx := newQuarantineFixture(t, Options{}) + writeMatchingPromotedFile(t, fx.quarantineRoot, fx.objectsRoot, "pack", "pack-a.pack", "identical pack bytes") + + err := fx.svc.promoteQuarantine(fx.quarantineName, fx.quarantineRoot) + if err != nil { + t.Fatalf("promoteQuarantine: %v", err) + } +} diff --git a/network/receivepack/service/request.go b/network/receivepack/service/request.go new file mode 100644 index 00000000..7f9bcc2f --- /dev/null +++ b/network/receivepack/service/request.go @@ -0,0 +1,16 @@ +package service + +import "io" + +// Request is one protocol-independent receive-pack execution request. +// +// If PackExpected is true, Pack must be non-nil and remain valid until +// Execute finishes consuming it. +type Request struct { + Commands []Command + PushOptions []string + Atomic bool + DeleteOnly bool + PackExpected bool + Pack io.Reader +} diff --git a/network/receivepack/service/result.go b/network/receivepack/service/result.go new file mode 100644 index 00000000..17fc0b6b --- /dev/null +++ b/network/receivepack/service/result.go @@ -0,0 +1,14 @@ +package service + +import ( + "codeberg.org/lindenii/furgit/packfile/ingest" +) + +// Result is one receive-pack execution result. +type Result struct { + UnpackError string + Commands []CommandResult + Ingest *ingest.Result + Planned []PlannedUpdate + Applied bool +} diff --git a/network/receivepack/service/run_hook.go b/network/receivepack/service/run_hook.go new file mode 100644 index 00000000..94467078 --- /dev/null +++ b/network/receivepack/service/run_hook.go @@ -0,0 +1,168 @@ +package service + +import ( + "context" + "os" + + "codeberg.org/lindenii/furgit/internal/utils" + objectstorer "codeberg.org/lindenii/furgit/object/storer" + "codeberg.org/lindenii/furgit/object/storer/loose" + objectmix "codeberg.org/lindenii/furgit/object/storer/mix" + "codeberg.org/lindenii/furgit/object/storer/packed" +) + +func (service *Service) runHook( + ctx context.Context, + req *Request, + commands []Command, + quarantineName string, +) ( + allowedCommands []Command, + allowedIndices []int, + rejected map[int]string, + ok bool, + errText string, +) { + allowedCommands = append([]Command(nil), commands...) + + allowedIndices = make([]int, 0, len(commands)) + for index := range commands { + allowedIndices = append(allowedIndices, index) + } + + rejected = make(map[int]string) + if service.opts.Hook == nil { + return allowedCommands, allowedIndices, rejected, true, "" + } + + utils.BestEffortFprintf(service.opts.Progress, "running hooks...\r") + + quarantinedObjects := service.opts.ExistingObjects + + var ( + quarantineObjectsStore objectstorer.Store + quarantineLooseStore *loose.Store + quarantinePackedStore *packed.Store + quarantineLooseRoot *os.Root + quarantinePackRoot *os.Root + err error + ) + + //nolint:nestif + if quarantineName != "" { + quarantineLooseRoot, err = service.opts.ObjectsRoot.OpenRoot(quarantineName) + if err != nil { + utils.BestEffortFprintf(service.opts.Progress, "running hooks: failed: %v.\n", err) + + return nil, nil, nil, false, err.Error() + } + + quarantineLooseStore, err = loose.New(quarantineLooseRoot, service.opts.Algorithm) + if err != nil { + _ = quarantineLooseRoot.Close() + + utils.BestEffortFprintf(service.opts.Progress, "running hooks: failed: %v.\n", err) + + return nil, nil, nil, false, err.Error() + } + + quarantineObjectsStore = quarantineLooseStore + quarantinedObjects = quarantineLooseStore + + quarantinePackRoot, err = quarantineLooseRoot.OpenRoot("pack") + if err == nil { + var packedErr error + + quarantinePackedStore, packedErr = packed.New(quarantinePackRoot, service.opts.Algorithm, packed.Options{}) + if packedErr != nil { + _ = quarantineLooseStore.Close() + _ = quarantinePackRoot.Close() + _ = quarantineLooseRoot.Close() + + utils.BestEffortFprintf(service.opts.Progress, "running hooks: failed: %v.\n", packedErr) + + return nil, nil, nil, false, packedErr.Error() + } + + quarantineObjectsStore = objectmix.New(quarantineLooseStore, quarantinePackedStore) + quarantinedObjects = quarantineObjectsStore + } else if !os.IsNotExist(err) { + _ = quarantineLooseStore.Close() + _ = quarantineLooseRoot.Close() + + utils.BestEffortFprintf(service.opts.Progress, "running hooks: failed: %v.\n", err) + + return nil, nil, nil, false, err.Error() + } + + defer func() { + if quarantineObjectsStore != nil { + _ = quarantineObjectsStore.Close() + } + + if quarantinePackedStore != nil { + _ = quarantinePackedStore.Close() + } + + if quarantineLooseStore != nil { + _ = quarantineLooseStore.Close() + } + + if quarantinePackRoot != nil { + _ = quarantinePackRoot.Close() + } + + if quarantineLooseRoot != nil { + _ = quarantineLooseRoot.Close() + } + }() + } + + decisions, err := service.opts.Hook(ctx, HookRequest{ + Refs: service.opts.Refs, + ExistingObjects: service.opts.ExistingObjects, + QuarantinedObjects: quarantinedObjects, + Updates: buildHookUpdates(commands), + PushOptions: append([]string(nil), req.PushOptions...), + IO: service.opts.HookIO, + }) + if err != nil { + utils.BestEffortFprintf(service.opts.Progress, "running hooks: failed: %v.\n", err) + + return nil, nil, nil, false, err.Error() + } + + if len(decisions) != len(commands) { + utils.BestEffortFprintf(service.opts.Progress, "running hooks: failed: wrong decision count.\n") + + return nil, nil, nil, false, "hook returned wrong number of update decisions" + } + + allowedCommands = allowedCommands[:0] + allowedIndices = allowedIndices[:0] + + for index, decision := range decisions { + if decision.Accept { + allowedCommands = append(allowedCommands, commands[index]) + allowedIndices = append(allowedIndices, index) + + continue + } + + message := decision.Message + if message == "" { + message = "rejected by hook" + } + + rejected[index] = message + } + + utils.BestEffortFprintf( + service.opts.Progress, + "running hooks: done (%d/%d accepted).\n", + len(allowedCommands), + len(commands), + ) + + return allowedCommands, allowedIndices, rejected, true, "" +} diff --git a/network/receivepack/service/service.go b/network/receivepack/service/service.go new file mode 100644 index 00000000..a57fd354 --- /dev/null +++ b/network/receivepack/service/service.go @@ -0,0 +1,16 @@ +package service + +// Service executes protocol-independent receive-pack requests. +// +// Service borrows all dependencies supplied in Options. +type Service struct { + opts Options +} + +// New creates one receive-pack service. +// +// The returned service borrows opts and does not take ownership of any stores, +// roots, hooks, or I/O endpoints reachable through it. +func New(opts Options) *Service { + return &Service{opts: opts} +} diff --git a/network/receivepack/service/service_test.go b/network/receivepack/service/service_test.go new file mode 100644 index 00000000..e59113be --- /dev/null +++ b/network/receivepack/service/service_test.go @@ -0,0 +1,99 @@ +package service_test + +import ( + "context" + "io/fs" + "os" + "strings" + "testing" + + "codeberg.org/lindenii/furgit/internal/testgit" + objectid "codeberg.org/lindenii/furgit/object/id" + "codeberg.org/lindenii/furgit/object/storer/memory" + "codeberg.org/lindenii/furgit/network/receivepack/service" +) + +func TestExecutePackExpectedWithoutObjectsRoot(t *testing.T) { + t.Parallel() + + //nolint:thelper + testgit.ForEachAlgorithm(t, func(t *testing.T, algo objectid.Algorithm) { + t.Parallel() + + store := memory.New(algo) + svc := service.New(service.Options{ + Algorithm: algo, + ExistingObjects: store, + }) + + result, err := svc.Execute(context.Background(), &service.Request{ + Commands: []service.Command{{ + Name: "refs/heads/main", + OldID: objectid.Zero(algo), + NewID: objectid.Zero(algo), + }}, + PackExpected: true, + Pack: strings.NewReader("not a pack"), + }) + if err != nil { + t.Fatalf("Execute: %v", err) + } + + if result.UnpackError != "objects root not configured" { + t.Fatalf("unexpected unpack error %q", result.UnpackError) + } + }) +} + +func TestExecuteRemovesDerivedQuarantineAfterIngestFailure(t *testing.T) { + t.Parallel() + + //nolint:thelper + testgit.ForEachAlgorithm(t, func(t *testing.T, algo objectid.Algorithm) { + t.Parallel() + + store := memory.New(algo) + objectsDir := t.TempDir() + + objectsRoot, err := os.OpenRoot(objectsDir) + if err != nil { + t.Fatalf("os.OpenRoot: %v", err) + } + + t.Cleanup(func() { + _ = objectsRoot.Close() + }) + + svc := service.New(service.Options{ + Algorithm: algo, + ExistingObjects: store, + ObjectsRoot: objectsRoot, + }) + + result, err := svc.Execute(context.Background(), &service.Request{ + Commands: []service.Command{{ + Name: "refs/heads/main", + OldID: objectid.Zero(algo), + NewID: objectid.Zero(algo), + }}, + PackExpected: true, + Pack: strings.NewReader("not a pack"), + }) + if err != nil { + t.Fatalf("Execute: %v", err) + } + + if result.UnpackError == "" { + t.Fatal("Execute returned empty unpack error for invalid pack") + } + + entries, err := fs.ReadDir(objectsRoot.FS(), ".") + if err != nil { + t.Fatalf("fs.ReadDir: %v", err) + } + + if len(entries) != 0 { + t.Fatalf("objects root still has entries after failed ingest: %d", len(entries)) + } + }) +} diff --git a/network/receivepack/service/update.go b/network/receivepack/service/update.go new file mode 100644 index 00000000..043f3d51 --- /dev/null +++ b/network/receivepack/service/update.go @@ -0,0 +1,12 @@ +package service + +import objectid "codeberg.org/lindenii/furgit/object/id" + +// PlannedUpdate is one ref update that would be applied once ref writing +// exists. +type PlannedUpdate struct { + Name string + OldID objectid.ObjectID + NewID objectid.ObjectID + Delete bool +} |
