package mirror import ( "context" "encoding/xml" "errors" "fmt" "io" "os" "github.com/grmrgecko/repo-sync/fetch" log "github.com/sirupsen/logrus" ) const ( rpmRepomdPath = "repodata/repomd.xml" rpmSigPath = "repodata/repomd.xml.asc" rpmKeyPath = "repodata/repomd.xml.key" ) // rpmRoot is one staged repomd generation and its optional signature // material. type rpmRoot struct { repomdDst string sigDst string keyDst string sigMissing bool keyMissing bool authenticated bool } // rpmPublishedFile is a rollback copy of one live root metadata file. type rpmPublishedFile struct { path string data []byte mode os.FileMode exists bool } // rpmWithdrawnPackage is a stale package hidden until root publication // succeeds or restored if publication fails. type rpmWithdrawnPackage struct { path string backup string } // cleanup removes unpublished files left from staging a repomd generation. func (r *rpmRoot) cleanup() { for _, name := range []string{r.repomdDst, r.sigDst, r.keyDst} { _ = os.Remove(name + fetch.StagedSuffix) } } // stageRPMRoot stages and optionally verifies repomd.xml with its detached // signature. A failed check refetches the complete set once because an // upstream rotation can occur between the requests. func stageRPMRoot(ctx context.Context, src *fetch.Source, destDir string, opts *Options) (*rpmRoot, error) { r := &rpmRoot{} var err error if r.repomdDst, err = fetch.LocalJoin(destDir, rpmRepomdPath); err != nil { return nil, err } if r.sigDst, err = fetch.LocalJoin(destDir, rpmSigPath); err != nil { return nil, err } if r.keyDst, err = fetch.LocalJoin(destDir, rpmKeyPath); err != nil { return nil, err } mode := opts.SignatureMode if mode == "" { mode = SignatureOff } var verifier *signatureVerifier if mode != SignatureOff { verifier, err = newSignatureVerifier(opts) if err != nil { return nil, err } } for attempt := 0; attempt < 2; attempt++ { fresh := attempt > 0 file := fetch.File if fresh { file = func(ctx context.Context, src *fetch.Source, reqPath, dst string, want *fetch.Expect, stage, _ bool) (fetch.FileState, error) { return fetch.FileFresh(ctx, src, reqPath, dst, want, stage) } } if _, err := file(ctx, src, rpmRepomdPath, r.repomdDst, nil, true, false); err != nil { r.cleanup() return nil, fmt.Errorf("fetch repomd.xml: %w", err) } r.sigMissing = false r.keyMissing = false for _, extra := range []struct { req string dst string missing *bool }{ {rpmSigPath, r.sigDst, &r.sigMissing}, {rpmKeyPath, r.keyDst, &r.keyMissing}, } { _, err := file(ctx, src, extra.req, extra.dst, nil, true, false) switch { case err == nil: case errors.Is(err, fetch.ErrNotFound), errors.Is(err, fetch.ErrForbidden): *extra.missing = true case err != nil: r.cleanup() return nil, fmt.Errorf("fetch %s: %w", extra.req, err) } } if mode == SignatureOff { return r, nil } if r.sigMissing { if mode == SignatureRequired { r.cleanup() return nil, errors.New("repository signature is required but repomd.xml.asc is missing") } return r, nil } keyPath := "" if !r.keyMissing { keyPath = fetch.StagedOrFinal(r.keyDst) } fingerprint, verifyErr := verifier.verifyDetached( ctx, fetch.StagedOrFinal(r.repomdDst), fetch.StagedOrFinal(r.sigDst), keyPath, ) if verifyErr == nil { r.authenticated = true log.WithField("fingerprint", fingerprint).Debug("Verified repomd.xml signature.") return r, nil } if !fresh { log.WithError(verifyErr).Warn("Staged repomd.xml signature failed verification; refetching the pair.") continue } r.cleanup() return nil, fmt.Errorf("verify repomd.xml signature after refetch: %w", verifyErr) } return nil, errors.New("unable to stage repomd.xml") } // snapshotRPMRoot reads the current live root set before publication. func snapshotRPMRoot(paths ...string) ([]rpmPublishedFile, error) { files := make([]rpmPublishedFile, 0, len(paths)) for _, name := range paths { file := rpmPublishedFile{path: name} info, err := os.Stat(name) if os.IsNotExist(err) { files = append(files, file) continue } if err != nil { return nil, err } file.data, err = os.ReadFile(name) if err != nil { return nil, err } file.mode = info.Mode() file.exists = true files = append(files, file) } return files, nil } // restoreRPMRoot replaces a partially published root with its prior files. func restoreRPMRoot(files []rpmPublishedFile) error { var errs []error for _, file := range files { if !file.exists { if err := os.Remove(file.path); err != nil && !os.IsNotExist(err) { errs = append(errs, err) } continue } tmp := file.path + ".rollback" + fetch.StagedSuffix if err := os.WriteFile(tmp, file.data, file.mode); err != nil { errs = append(errs, err) continue } if err := os.Rename(tmp, file.path); err != nil { _ = os.Remove(tmp) errs = append(errs, err) } } return errors.Join(errs...) } // withdrawMissingPackages hides checksum-invalid packages the authenticated // root no longer provides upstream. func withdrawMissingPackages(jobs []fetch.Job, states []fetch.FileState) ([]rpmWithdrawnPackage, error) { var withdrawn []rpmWithdrawnPackage for i, state := range states { if state != fetch.FileMissing { continue } if _, err := os.Stat(jobs[i].Dst); os.IsNotExist(err) { continue } else if err != nil { return withdrawn, err } backup := jobs[i].Dst + ".withdrawn" + fetch.StagedSuffix if err := os.Remove(backup); err != nil && !os.IsNotExist(err) { return withdrawn, err } if err := os.Rename(jobs[i].Dst, backup); err != nil { return withdrawn, err } withdrawn = append(withdrawn, rpmWithdrawnPackage{path: jobs[i].Dst, backup: backup}) } return withdrawn, nil } // finishWithdrawnPackages removes hidden packages after publication or puts // them back when publication fails. func finishWithdrawnPackages(files []rpmWithdrawnPackage, published bool) error { var errs []error for _, file := range files { if published { if err := os.Remove(file.backup); err != nil && !os.IsNotExist(err) { errs = append(errs, err) } continue } if err := os.Rename(file.backup, file.path); err != nil && !os.IsNotExist(err) { errs = append(errs, err) } } return errors.Join(errs...) } // publishRPMRoot publishes signature material immediately before repomd.xml. // Missing optional files are removed only after the replacement is ready. func publishRPMRoot(r *rpmRoot, keep *fetch.KeepSet, dryRun bool) (retErr error) { var published []rpmPublishedFile if !dryRun { var err error published, err = snapshotRPMRoot(r.keyDst, r.sigDst, r.repomdDst) if err != nil { return err } defer func() { if retErr != nil { retErr = errors.Join(retErr, restoreRPMRoot(published)) } }() } for _, extra := range []struct { dst string missing bool }{ {r.keyDst, r.keyMissing}, {r.sigDst, r.sigMissing}, } { if extra.missing { if !dryRun { if err := os.Remove(extra.dst); err != nil && !os.IsNotExist(err) { return err } } continue } if err := fetch.PromoteOrDiscard(extra.dst, dryRun); err != nil { return err } if _, err := os.Stat(extra.dst); err == nil { keep.Add(extra.dst) } } if err := fetch.PromoteOrDiscard(r.repomdDst, dryRun); err != nil { return err } keep.Add(r.repomdDst) return nil } // repoMD is the partial XML schema for a yum repomd.xml file. Only the // fields the sync needs are decoded; unknown elements are ignored so // upstream additions do not break parsing. type repoMD struct { Data []repoData `xml:"data"` } // repoData is one metadata file referenced from repomd.xml. type repoData struct { Type string `xml:"type,attr"` Checksum repoChecksum `xml:"checksum"` Location repoLocation `xml:"location"` Size int64 `xml:"size"` } // repoChecksum is a checksum element with its algorithm type. type repoChecksum struct { Type string `xml:"type,attr"` Value string `xml:",chardata"` } // repoLocation captures the relative href of a metadata or package file. type repoLocation struct { Href string `xml:"href,attr"` } // rpmPackage is the minimal package element decoded from primary metadata, // carrying the location, checksum, and size needed to mirror the file. type rpmPackage struct { Checksum repoChecksum `xml:"checksum"` Size rpmSize `xml:"size"` Location repoLocation `xml:"location"` } // rpmSize carries the on-disk package size from primary metadata. type rpmSize struct { Package int64 `xml:"package,attr"` } // syncRPM synchronizes one RPM repository from src into destDir. func syncRPM(ctx context.Context, src *fetch.Source, destDir string, opts *Options) error { keep := fetch.NewKeepSet(opts.Prune) miss := opts.newMissing(destDir) tr := newTrace(opts) ctx = tr.track(ctx) // Authenticate the root metadata before using it to select any files. root, err := stageRPMRoot(ctx, src, destDir, opts) if err != nil { return err } defer root.cleanup() md, err := readRepomd(fetch.StagedOrFinal(root.repomdDst)) if err != nil { return err } // Download the metadata files referenced by repomd. Their names embed // content hashes, so they can be written directly into place. A dry run // only fetches the indexes that must be parsed, staged for discard. var jobs, planned []fetch.Job var primary *repoData var deltaMDs []*repoData var parseDsts []string for i, data := range md.Data { if data.Location.Href == "" { continue } dst, err := fetch.LocalJoin(destDir, data.Location.Href) if err != nil { return err } job := fetch.Job{ ReqPath: data.Location.Href, Dst: dst, Want: fetch.MakeExpect(data.Size, data.Checksum.Type, data.Checksum.Value), } parsed := false switch data.Type { case "primary": primary = &md.Data[i] parsed = true case "prestodelta", "deltainfo": deltaMDs = append(deltaMDs, &md.Data[i]) parsed = true } if opts.DryRun { if !parsed { planned = append(planned, job) continue } job.Stage = true parseDsts = append(parseDsts, dst) } jobs = append(jobs, job) } verifyMetadata := opts.Verify || root.authenticated fetch.PlanJobs(planned, verifyMetadata, keep) if _, err := fetch.Many(ctx, src, jobs, opts.Workers, verifyMetadata, keep, nil); err != nil { return fmt.Errorf("fetch repository metadata: %w", err) } if primary == nil { return errors.New("repository metadata lists no primary index") } // Collect package locations from the primary index, plus any delta // packages referenced by prestodelta metadata. primaryPath, err := fetch.LocalJoin(destDir, primary.Location.Href) if err != nil { return err } pkgs, err := readPrimary(fetch.StagedOrFinal(primaryPath)) if err != nil { return err } jobs = jobs[:0] for _, pkg := range pkgs { if pkg.Location.Href == "" { continue } dst, err := fetch.LocalJoin(destDir, pkg.Location.Href) if err != nil { return err } jobs = append(jobs, fetch.Job{ ReqPath: pkg.Location.Href, Dst: dst, Want: fetch.MakeExpect(pkg.Size.Package, pkg.Checksum.Type, pkg.Checksum.Value), }) } for _, deltaMD := range deltaMDs { deltaPath, err := fetch.LocalJoin(destDir, deltaMD.Location.Href) if err != nil { return err } deltas, err := readDeltas(fetch.StagedOrFinal(deltaPath)) if err != nil { return err } for _, delta := range deltas { if delta.Filename == "" { continue } dst, err := fetch.LocalJoin(destDir, delta.Filename) if err != nil { return err } jobs = append(jobs, fetch.Job{ ReqPath: delta.Filename, Dst: dst, Want: fetch.MakeExpect(delta.Size, delta.Checksum.Type, delta.Checksum.Value), }) } } log.WithField("packages", len(jobs)).Info("Synchronizing packages.") verifyPackages := opts.Verify || root.authenticated var packageStates []fetch.FileState if opts.DryRun { fetch.PlanJobs(jobs, verifyPackages, keep) } else { packageStates, err = fetch.Many(ctx, src, jobs, opts.Workers, verifyPackages, keep, miss) if err != nil { return fmt.Errorf("fetch packages: %w", err) } } // A dry run discards the indexes it staged for parsing. for _, dst := range parseDsts { _ = os.Remove(dst + fetch.StagedSuffix) } // Keep the previous root live until every referenced package is present // under the configured missing-file policy. if err := miss.Finish(); err != nil { return err } var withdrawn []rpmWithdrawnPackage if root.authenticated { withdrawn, err = withdrawMissingPackages(jobs, packageStates) if err != nil { _ = finishWithdrawnPackages(withdrawn, false) return fmt.Errorf("withdraw stale packages: %w", err) } } // Publish the verified root only after every referenced file is ready. if err := publishRPMRoot(root, keep, opts.DryRun); err != nil { return errors.Join(err, finishWithdrawnPackages(withdrawn, false)) } if err := finishWithdrawnPackages(withdrawn, true); err != nil { log.WithError(err).Warn("Unable to remove withdrawn package files.") } // Publish the traces, upstream's included, before pruning so the keep // set covers them. tr.publish(ctx, destDir, src, keep, opts) // Remove files that are no longer part of the repository. if opts.Prune { fetch.PruneTree(destDir, keep, opts.PruneGrace, opts.DryRun) } return nil } // rpmDelta is one delta package referenced from prestodelta metadata. type rpmDelta struct { Filename string `xml:"filename"` Size int64 `xml:"size"` Checksum repoChecksum `xml:"checksum"` } // readDeltas streams prestodelta metadata and collects each delta package // reference so drpm files are mirrored alongside full packages. func readDeltas(filename string) ([]rpmDelta, error) { f, err := os.Open(filename) if err != nil { return nil, err } defer f.Close() r, closeFn, err := fetch.Decompressor(f, filename) if err != nil { return nil, err } if closeFn != nil { defer closeFn() } dec := xml.NewDecoder(r) var deltas []rpmDelta for { tok, err := dec.Token() if errors.Is(err, io.EOF) { break } if err != nil { return nil, fmt.Errorf("parse delta metadata %s: %w", filename, err) } start, ok := tok.(xml.StartElement) if !ok || start.Name.Local != "delta" { continue } var delta rpmDelta if err := dec.DecodeElement(&delta, &start); err != nil { return nil, fmt.Errorf("parse delta entry %s: %w", filename, err) } if delta.Filename != "" { deltas = append(deltas, delta) } } return deltas, nil } // readRepomd decodes a repomd.xml file. func readRepomd(filename string) (*repoMD, error) { f, err := os.Open(filename) if err != nil { return nil, err } defer f.Close() var md repoMD if err := xml.NewDecoder(f).Decode(&md); err != nil { return nil, fmt.Errorf("parse repomd %s: %w", filename, err) } return &md, nil } // readPrimary streams a primary metadata index and collects each package // entry without loading the whole document into memory. func readPrimary(filename string) ([]rpmPackage, error) { f, err := os.Open(filename) if err != nil { return nil, err } defer f.Close() r, closeFn, err := fetch.Decompressor(f, filename) if err != nil { return nil, err } if closeFn != nil { defer closeFn() } dec := xml.NewDecoder(r) var pkgs []rpmPackage for { tok, err := dec.Token() if errors.Is(err, io.EOF) { break } if err != nil { return nil, fmt.Errorf("parse primary metadata %s: %w", filename, err) } start, ok := tok.(xml.StartElement) if !ok || start.Name.Local != "package" { continue } var pkg rpmPackage if err := dec.DecodeElement(&pkg, &start); err != nil { return nil, fmt.Errorf("parse package entry %s: %w", filename, err) } if pkg.Location.Href != "" { pkgs = append(pkgs, pkg) } } return pkgs, nil }