Skip to content

Commit 7088f7b

Browse files
crozzycrozzy
committed
vex: fetcher add changes before archived data
But switching how the VEX data is appended to the output spool we allow the fetcher to ignore the files in the archive that have been changed. Signed-off-by: crozzy <joseph.crosland@gmail.com>
1 parent 1602d4b commit 7088f7b

2 files changed

Lines changed: 123 additions & 58 deletions

File tree

‎rhel/vex/fetcher.go‎

Lines changed: 102 additions & 44 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,7 @@ import (
1010
"fmt"
1111
"io"
1212
"net/http"
13+
"net/url"
1314
"path"
1415
"regexp"
1516
"strconv"
@@ -31,6 +32,17 @@ var (
3132
cvePathRegex = regexp.MustCompile(`^\d{4}/(cve-\d{4}-\d{4,}).json$`)
3233
)
3334

35+
// Fetch pulls data down from the Red Hat VEX endpoints. The order of operations is:
36+
// 1. Check if we need to process the entire archive of data. If yes:
37+
// - Make a request to discover the latest archive endpoint.
38+
// - Make a HEAD request to archive endpoint to get the last-modified header.
39+
// - Save the last-modified time in the fingerprint's requestTime.
40+
// 2. Process the changes.csv file, requesting and appending the entries that changed since the finderprint's requestTime.
41+
// 3. Process the deletions.csv file, processing the entries that changed since the finderprint's requestTime.
42+
// 4. If we need to process entire archive, request the archive data and append the entries that have not been changed or deleted.
43+
//
44+
// This helps to ensure that we only persist one copy of an advisory in the worst possible case. In most cases,
45+
// after the initial load, the number of processed files should be very small.
3446
func (u *Updater) Fetch(ctx context.Context, hint driver.Fingerprint) (io.ReadCloser, driver.Fingerprint, error) {
3547
ctx = zlog.ContextWithValues(ctx, "component", "rhel/vex/Updater.Fetch")
3648
fp, err := parseFingerprint(hint)
@@ -62,48 +74,46 @@ func (u *Updater) Fetch(ctx context.Context, hint driver.Fingerprint) (io.ReadCl
6274
}
6375
}()
6476

77+
var compressedURL *url.URL
6578
// Is this the first run or has the updater changed since the last run?
66-
if fp.changesEtag == "" || fp.version != updaterVersion {
79+
processArchive := fp.changesEtag == "" || fp.version != updaterVersion
80+
if processArchive {
6781
// We need to go after the full corpus of vulnerabilities
68-
// First we target the archive_latest.txt file
69-
latestURI, err := u.url.Parse(latestFile)
82+
// First we target the archive_latest.txt file.
83+
var err error
84+
compressedURL, err = u.getCompressedFileURL(ctx)
7085
if err != nil {
71-
return nil, hint, err
72-
}
73-
latestReq, err := http.NewRequestWithContext(ctx, http.MethodGet, latestURI.String(), nil)
74-
if err != nil {
75-
return nil, hint, err
76-
}
77-
latestRes, err := u.client.Do(latestReq)
78-
if err != nil {
79-
return nil, hint, err
80-
}
81-
defer latestRes.Body.Close()
82-
83-
err = httputil.CheckResponse(latestRes, http.StatusOK)
84-
if err != nil {
85-
return nil, hint, fmt.Errorf("unexpected response from archive_latest.txt: %w", err)
86+
return nil, hint, fmt.Errorf("could not get compressed file URL: %w", err)
8687
}
88+
zlog.Debug(ctx).
89+
Str("url", compressedURL.String()).
90+
Msg("got compressed URL")
8791

88-
body, err := io.ReadAll(latestRes.Body) // Fine to use as expecting small number of bytes.
92+
fp.requestTime, err = u.getLastModified(ctx, compressedURL)
8993
if err != nil {
90-
return nil, hint, err
94+
return nil, hint, fmt.Errorf("could not get last-modified header: %w", err)
9195
}
96+
}
9297

93-
compressedFilename := string(body)
94-
zlog.Debug(ctx).
95-
Str("filename", compressedFilename).
96-
Msg("requesting latest compressed file")
98+
changed := map[string]bool{}
99+
err = u.processChanges(ctx, cw, fp, changed)
100+
if err != nil {
101+
return nil, hint, err
102+
}
97103

98-
uri, err := u.url.Parse(compressedFilename)
99-
if err != nil {
100-
return nil, hint, err
101-
}
104+
err = u.processDeletions(ctx, cw, fp, changed)
105+
if err != nil {
106+
return nil, hint, err
107+
}
102108

109+
if processArchive {
103110
rctx, cancel := context.WithTimeout(ctx, compressedFileTimeout)
104111
defer cancel()
105112

106-
req, err := http.NewRequestWithContext(rctx, http.MethodGet, uri.String(), nil)
113+
if compressedURL == nil {
114+
return nil, hint, fmt.Errorf("compressed file URL needs to be populated")
115+
}
116+
req, err := http.NewRequestWithContext(rctx, http.MethodGet, compressedURL.String(), nil)
107117
if err != nil {
108118
return nil, hint, err
109119
}
@@ -119,11 +129,6 @@ func (u *Updater) Fetch(ctx context.Context, hint driver.Fingerprint) (io.ReadCl
119129
return nil, hint, fmt.Errorf("unexpected response from latest compressed file: %w", err)
120130
}
121131

122-
lm := res.Header.Get("last-modified")
123-
fp.requestTime, err = time.Parse(http.TimeFormat, lm)
124-
if err != nil {
125-
return nil, hint, fmt.Errorf("could not parse last-modified header %s: %w", lm, err)
126-
}
127132
z, err := zreader.Reader(res.Body)
128133
if err != nil {
129134
return nil, hint, err
@@ -149,6 +154,10 @@ func (u *Updater) Fetch(ctx context.Context, hint driver.Fingerprint) (io.ReadCl
149154
if year < lookBackToYear {
150155
continue
151156
}
157+
if changed[path.Base(h.Name)] {
158+
// We've already processed this file don't bother appending it to the output
159+
continue
160+
}
152161
buf.Grow(int(h.Size))
153162
if _, err := buf.ReadFrom(r); err != nil {
154163
return nil, hint, err
@@ -176,26 +185,71 @@ func (u *Updater) Fetch(ctx context.Context, hint driver.Fingerprint) (io.ReadCl
176185
Msg("finished writing compressed data to spool")
177186
}
178187

179-
err = u.processChanges(ctx, cw, fp)
188+
fp.version = updaterVersion
189+
fp.requestTime = time.Now()
190+
success = true
191+
return f, driver.Fingerprint(fp.String()), nil
192+
}
193+
194+
func (u *Updater) getCompressedFileURL(ctx context.Context) (*url.URL, error) {
195+
latestURI, err := u.url.Parse(latestFile)
196+
if err != nil {
197+
return nil, err
198+
}
199+
latestReq, err := http.NewRequestWithContext(ctx, http.MethodGet, latestURI.String(), nil)
200+
if err != nil {
201+
return nil, err
202+
}
203+
latestRes, err := u.client.Do(latestReq)
204+
if err != nil {
205+
return nil, err
206+
}
207+
defer latestRes.Body.Close()
208+
209+
err = httputil.CheckResponse(latestRes, http.StatusOK)
180210
if err != nil {
181-
return nil, hint, err
211+
return nil, fmt.Errorf("unexpected response from archive_latest.txt: %w", err)
182212
}
183213

184-
err = u.processDeletions(ctx, cw, fp)
214+
body, err := io.ReadAll(latestRes.Body) // Fine to use as expecting small number of bytes.
185215
if err != nil {
186-
return nil, hint, err
216+
return nil, err
187217
}
188218

189-
fp.version = updaterVersion
190-
fp.requestTime = time.Now()
191-
success = true
192-
return f, driver.Fingerprint(fp.String()), nil
219+
compressedFilename := string(body)
220+
compressedURL, err := u.url.Parse(compressedFilename)
221+
if err != nil {
222+
return nil, err
223+
}
224+
return compressedURL, nil
225+
}
226+
227+
func (u *Updater) getLastModified(ctx context.Context, cu *url.URL) (time.Time, error) {
228+
var empty time.Time
229+
req, err := http.NewRequestWithContext(ctx, http.MethodHead, cu.String(), nil)
230+
if err != nil {
231+
return empty, err
232+
}
233+
234+
res, err := u.client.Do(req)
235+
if err != nil {
236+
return empty, err
237+
}
238+
defer res.Body.Close()
239+
240+
err = httputil.CheckResponse(res, http.StatusOK)
241+
if err != nil {
242+
return empty, fmt.Errorf("unexpected HEAD response from latest compressed file: %w", err)
243+
}
244+
245+
lm := res.Header.Get("last-modified")
246+
return time.Parse(http.TimeFormat, lm)
193247
}
194248

195249
// ProcessChanges deals with the published changes.csv, adding records
196250
// to w means they are deemed to have changed since the compressed
197251
// file was last processed. w and fp can be modified.
198-
func (u *Updater) processChanges(ctx context.Context, w io.Writer, fp *fingerprint) error {
252+
func (u *Updater) processChanges(ctx context.Context, w io.Writer, fp *fingerprint, changed map[string]bool) error {
199253
tf, err := tmp.NewFile("", "rhel-vex-changes.")
200254
if err != nil {
201255
return err
@@ -271,6 +325,8 @@ func (u *Updater) processChanges(ctx context.Context, w io.Writer, fp *fingerpri
271325
continue
272326
}
273327

328+
changed[path.Base(cvePath)] = true
329+
274330
advisoryURI, err := u.url.Parse(cvePath)
275331
if err != nil {
276332
return err
@@ -323,7 +379,7 @@ func (u *Updater) processChanges(ctx context.Context, w io.Writer, fp *fingerpri
323379
// ProcessDeletions deals with the published deletions.csv, adding records
324380
// to w mean they are deemed to have been deleted since the last compressed
325381
// file was last processed. w and fp can be modified.
326-
func (u *Updater) processDeletions(ctx context.Context, w io.Writer, fp *fingerprint) error {
382+
func (u *Updater) processDeletions(ctx context.Context, w io.Writer, fp *fingerprint, changed map[string]bool) error {
327383
deletionURI, err := u.url.Parse(deletionsFile)
328384
if err != nil {
329385
return err
@@ -375,6 +431,8 @@ func (u *Updater) processDeletions(ctx context.Context, w io.Writer, fp *fingerp
375431
if updatedTime.Before(fp.requestTime) {
376432
continue
377433
}
434+
changed[path.Base(cvePath)] = true
435+
378436
deletedJSON, err := createDeletedJSON(cvePath)
379437
if err != nil {
380438
zlog.Warn(ctx).Err(err).Msg("error creating JSON object denoting deletion")

‎rhel/vex/fetcher_test.go‎

Lines changed: 21 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@ import (
1616
"github.com/quay/zlog"
1717
"golang.org/x/tools/txtar"
1818

19+
"github.com/quay/claircore/libvuln/driver"
1920
"github.com/quay/claircore/toolkit/types/csaf"
2021
)
2122

@@ -43,17 +44,20 @@ func serveSecDB(t *testing.T, txtarFile string) (string, *http.Client) {
4344
t.Fatal(err)
4445
}
4546
filename := filepath.Base(relFilepath)
46-
mux.HandleFunc("/"+filename, func(w http.ResponseWriter, _ *http.Request) {
47+
mux.HandleFunc("/"+filename, func(w http.ResponseWriter, r *http.Request) {
4748
for k, v := range headers {
4849
w.Header().Set(k, v[0])
4950
}
50-
51-
f, err := os.Open("testdata/" + relFilepath)
52-
if err != nil {
53-
t.Fatal(err)
54-
}
55-
if _, err := io.Copy(w, f); err != nil {
56-
t.Fatal(err)
51+
switch r.Method {
52+
case http.MethodHead:
53+
case http.MethodGet:
54+
f, err := os.Open("testdata/" + relFilepath)
55+
if err != nil {
56+
t.Fatal(err)
57+
}
58+
if _, err := io.Copy(w, f); err != nil {
59+
t.Fatal(err)
60+
}
5761
}
5862
})
5963
for _, f := range archive.Files {
@@ -111,9 +115,12 @@ func TestFactory(t *testing.T) {
111115
if f.changesEtag != "something" {
112116
t.Errorf("bad etag for the changes.csv endpoint: %s", f.changesEtag)
113117
}
118+
if f.deletionsEtag != "somethingelse" {
119+
t.Errorf("bad etag for the deletions.csv endpoint: %s", f.deletionsEtag)
120+
}
114121

115122
// Check saved vulns
116-
expectedLnCt := 8
123+
expectedLnCt := 7
117124
lnCt := 0
118125
r := bufio.NewReader(snappy.NewReader(data))
119126
for b, err := r.ReadBytes('\n'); err == nil; b, err = r.ReadBytes('\n') {
@@ -127,7 +134,7 @@ func TestFactory(t *testing.T) {
127134
t.Errorf("got %d entries but expected %d", lnCt, expectedLnCt)
128135
}
129136

130-
newData, newFP, err := s.Updaters()[0].Fetch(ctx, "")
137+
newData, newFP, err := s.Updaters()[0].Fetch(ctx, driver.Fingerprint(f.String()))
131138
if err != nil {
132139
t.Fatalf("error re-Fetching, cannot continue: %v", err)
133140
}
@@ -143,9 +150,9 @@ func TestFactory(t *testing.T) {
143150
if f.deletionsEtag != "somethingelse" {
144151
t.Errorf("bad etag for the deletions.csv endpoint: %s", f.deletionsEtag)
145152
}
146-
buf := &bytes.Buffer{}
147-
sz, _ := newData.Read(buf.Bytes())
148-
if sz != 0 {
149-
t.Errorf("got too much data: %s", buf.String())
153+
154+
r = bufio.NewReader(snappy.NewReader(newData))
155+
for _, err := r.ReadBytes('\n'); err == nil; _, err = r.ReadBytes('\n') {
156+
t.Fatal("should not have anymore data")
150157
}
151158
}

0 commit comments

Comments
 (0)