print when emulator is running

Author: michaelkedar

affectedcoarse.py

@@ -0,0 +1,47 @@
+from google.cloud import ndb
+from osv import AffectedVersions, ecosystems
+from osv.models import MIN_COARSE_VERSION, MAX_COARSE_VERSION, _get_coarse_min_max
+import logging
+
+logging.getLogger().setLevel(logging.INFO)
+
+@ndb.tasklet
+def main():
+  q = AffectedVersions.query()
+  it: ndb.QueryIterator = q.iter()
+  fs = []
+  while (yield it.has_next_async()):
+    av: AffectedVersions = it.next()
+    if av.coarse_min:
+      continue
+    if av.ecosystem == 'GIT':
+      continue
+    coarse_min = MIN_COARSE_VERSION
+    coarse_max = MAX_COARSE_VERSION
+    e_helper = ecosystems.get(av.ecosystem)
+    if e_helper is not None:
+      if len(av.versions) > 0:
+        try:
+          all_coarse = [e_helper.coarse_version(v) for v in av.versions]
+          coarse_min = min(all_coarse)
+          coarse_max = max(all_coarse)
+        except NotImplementedError:
+          # Coarse versioning not yet implemented for this ecosystem.
+          pass
+        except ValueError:
+          logging.warning('Invalid version in %s', av.vuln_id)
+      elif len(av.events) > 0:
+        coarse_min, coarse_max = _get_coarse_min_max(av.events, e_helper, av.vuln_id)
+      else:
+        logging.warning('AffectedVersions with no versions: %s', av.vuln_id)
+
+    print(coarse_min, coarse_max, av.vuln_id)
+    av.coarse_min = coarse_min
+    av.coarse_max = coarse_max
+    fs.append(av.put_async())
+  yield fs
+
+
+with ndb.Client(project='oss-vdb').context(cache_policy=False):
+  logging.info('starting')
+  main().result()

go/cmd/importer/bucket.go

@@ -0,0 +1,37 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"log/slog"
+
+	"cloud.google.com/go/storage"
+	"github.com/google/osv.dev/go/internal/models"
+	"github.com/google/osv.dev/go/logger"
+	"google.golang.org/api/iterator"
+)
+
+func handleImportBucket(ctx context.Context, config ImporterConfig, sourceRepo *models.SourceRepository) error {
+	if sourceRepo.Type != models.SourceRepositoryTypeBucket || sourceRepo.Bucket == nil {
+		return errors.New("invalid SourceRepository for bucket import")
+	}
+	logger.Info("Importing bucket source repository",
+		slog.String("source_repository", sourceRepo.Name), slog.String("bucket", sourceRepo.Bucket.Name))
+
+	bucket := config.GCSClient.Bucket(sourceRepo.Bucket.Name)
+	query := &storage.Query{Prefix: sourceRepo.Bucket.Path}
+	it := bucket.Objects(ctx, query)
+	for {
+		object, err := it.Next()
+		if errors.Is(err, iterator.Done) {
+			break
+		}
+		if err != nil {
+			return err
+		}
+		if sourceRepo.Bucket.IgnoreLastImportTime || sourceRepo.Bucket.LastUpdated == nil || object.Created.After(*sourceRepo.Bucket.LastUpdated) {
+			logger.Info("Found object", slog.String("object", object.Name))
+		}
+	}
+	return nil
+}

go/cmd/importer/git.go

@@ -0,0 +1,15 @@
+package main
+
+import (
+	"context"
+	"log/slog"
+
+	"github.com/google/osv.dev/go/internal/models"
+	"github.com/google/osv.dev/go/logger"
+)
+
+func handleImportGit(ctx context.Context, config ImporterConfig, sourceRepo *models.SourceRepository) error {
+	logger.Info("Importing git source repository",
+		slog.String("source_repository", sourceRepo.Name), slog.String("repo", sourceRepo.Git.URL))
+	return nil
+}

go/cmd/importer/importer.go

@@ -0,0 +1,98 @@
+package main
+
+import (
+	"context"
+	"flag"
+	"log/slog"
+	"os"
+	"sync"
+
+	"cloud.google.com/go/datastore"
+	"cloud.google.com/go/pubsub/v2"
+	"cloud.google.com/go/storage"
+	db "github.com/google/osv.dev/go/internal/database/datastore"
+	"github.com/google/osv.dev/go/internal/models"
+	"github.com/google/osv.dev/go/logger"
+)
+
+type ImporterConfig struct {
+	SourceRepoStore models.SourceRepositoryStore
+	PubSubClient    *pubsub.Client
+	GCSClient       *storage.Client
+
+	StrictValidation bool
+}
+
+func main() {
+	logger.InitGlobalLogger()
+
+	strictValidation := flag.Bool("strict-validation", false, "Fail to import entries that do not pass validation.")
+	delete := flag.Bool("delete", false, "Bypass importing and propagate record deletions from source to Datastore")
+	deleteThresholdPct := flag.Float64("delete-threshold-pct", 10.0, "More than this percent of records for a given source being deleted triggers an error")
+
+	flag.Parse()
+
+	project := os.Getenv("GOOGLE_CLOUD_PROJECT")
+	if project == "" {
+		logger.Fatal("GOOGLE_CLOUD_PROJECT environment variable is not set")
+	}
+
+	config := ImporterConfig{
+		StrictValidation: *strictValidation,
+	}
+
+	datastoreClient, err := datastore.NewClient(context.Background(), project)
+	if err != nil {
+		logger.Fatal("Failed to create datastore client", slog.Any("error", err))
+	}
+	config.SourceRepoStore = db.NewSourceRepositoryStore(datastoreClient)
+
+	config.PubSubClient, err = pubsub.NewClient(context.Background(), project)
+	if err != nil {
+		logger.Fatal("Failed to create pubsub client", slog.Any("error", err))
+	}
+
+	config.GCSClient, err = storage.NewClient(context.Background())
+	if err != nil {
+		logger.Fatal("Failed to create GCS client", slog.Any("error", err))
+	}
+
+	if *delete {
+		_ = deleteThresholdPct
+		logger.Fatal("delete not implemented yet")
+	}
+
+	if err := RunImporter(context.Background(), config); err != nil {
+		logger.Fatal("Importer failed", slog.Any("error", err))
+	}
+}
+
+func RunImporter(ctx context.Context, config ImporterConfig) error {
+	logger.Info("Importer started")
+	var wg sync.WaitGroup
+	for sourceRepo, err := range config.SourceRepoStore.All(ctx) {
+		if err != nil {
+			return err
+		}
+		wg.Go(func() {
+			switch sourceRepo.Type {
+			case models.SourceRepositoryTypeGit:
+				if err := handleImportGit(ctx, config, sourceRepo); err != nil {
+					logger.Error("Failed to import git source repository", slog.Any("error", err), slog.String("source_repository", sourceRepo.Name))
+				}
+			case models.SourceRepositoryTypeBucket:
+				if err := handleImportBucket(ctx, config, sourceRepo); err != nil {
+					logger.Error("Failed to import bucket source repository", slog.Any("error", err), slog.String("source_repository", sourceRepo.Name))
+				}
+			case models.SourceRepositoryTypeREST:
+				if err := handleImportREST(ctx, config, sourceRepo); err != nil {
+					logger.Error("Failed to import REST source repository", slog.Any("error", err), slog.String("source_repository", sourceRepo.Name))
+				}
+			default:
+				logger.Error("Unsupported source repository type", slog.String("source_repository", sourceRepo.Name), slog.Any("type", sourceRepo.Type))
+			}
+		})
+	}
+	wg.Wait()
+	return nil
+}

go/cmd/importer/rest.go

@@ -0,0 +1,15 @@
+package main
+
+import (
+	"context"
+	"log/slog"
+
+	"github.com/google/osv.dev/go/internal/models"
+	"github.com/google/osv.dev/go/logger"
+)
+
+func handleImportREST(ctx context.Context, config ImporterConfig, sourceRepo *models.SourceRepository) error {
+	logger.Info("Importing REST source repository",
+		slog.String("source_repository", sourceRepo.Name), slog.String("url", sourceRepo.REST.URL))
+	return nil
+}

go/internal/database/datastore/internal/validate/validate.py

@@ -158,5 +158,6 @@ def main() -> int:
 
 if __name__ == '__main__':
   with osv.tests.datastore_emulator(), ndb.Client().context():
+    print('Datastore emulator running')
     ret = main()
   sys.exit(ret)