func NewHttpFilterFactory()

in src/internal/filter/opa/opafilterfactory.go [57:241]
157 lines of code
28 McCabe index (conditional complexity)

func NewHttpFilterFactory(
	ctx context.Context,
	tracer trace.Tracer,
	config configuration.PolicyEngine) (filter.HttpFilterFactory, error) {
	policySet := make(map[string]string)
	var bundle *bundle.Bundle = nil
	var err error
	defer filter.RecordSpanError(ctx, &err)

	if config.PoliciesDirectory != "" {
		var files []fs.DirEntry
		files, err = os.ReadDir(config.PoliciesDirectory)
		if err != nil {
			log.Errorf("failed to ReadDir %s: %s", config.PoliciesDirectory, err)
			return nil, err
		}

		for _, file := range files {
			if !file.IsDir() && filepath.Ext(file.Name()) == ".rego" {
				filePath := filepath.Join(config.PoliciesDirectory, file.Name())
				var contents []byte
				contents, err = os.ReadFile(filePath)
				if err != nil {
					log.Errorf("failed to ReadFile %q: %s", filePath, err)
					return nil, err
				}

				policySet[file.Name()] = string(contents)
			}
		}
	} else if config.BundleResource != "" {
		ctx2, bundleDownloadSpan := tracer.Start(ctx, "downloadOpaPolicyBundle")
		defer bundleDownloadSpan.End()
		defer filter.RecordSpanError(ctx2, &err)

		t := plugins.TriggerManual
		dlConfig := download.Config{
			Trigger: &t,
		}

		err = dlConfig.ValidateAndInjectDefaults()
		if err != nil {
			log.Errorf("failed to create a valid download config: %s", err)
			return nil, err
		}

		bundleProtocol := "https"

		envUseHttp, present := os.LookupEnv("USE_HTTP")
		if present {
			var useHttp bool
			useHttp, err = strconv.ParseBool(envUseHttp)
			if err == nil && useHttp {
				bundleProtocol = "http"
			}
		}

		bundleServiceUrl := config.BundleServiceUrl
		if bundleServiceUrl == "" {
			bundleServiceUrl = bundleProtocol + "://" + strings.Split(config.BundleResource, "/")[0]
		}

		var restConfig []byte
		if config.BundleServiceCredentialsToken != "" {
			scheme := config.BundleServiceCredentialsScheme
			if scheme == "" {
				scheme = "Bearer"
			}

			restConfig = []byte(fmt.Sprintf(`{
				"url": %q,
				"credentials": {
					"bearer": {
						"scheme": %q,
						"token": %q
					}
				}
			}`,
				bundleServiceUrl,
				scheme,
				config.BundleServiceCredentialsToken))
		} else {
			restConfig = []byte(fmt.Sprintf(`{
				"url": %q
			}`, bundleServiceUrl))
		}

		var client rest.Client
		client, err = rest.New(restConfig, map[string]*keys.Config{})
		if err != nil {
			log.Errorf("failed to create rest client: %s", err)
			return nil, err
		}

		var update *download.Update
		d := download.NewOCI(dlConfig, client, config.BundleResource, "/tmp/opa/oci/").
			WithCallback(func(_ context.Context, u download.Update) {
				update = &u
			})

		log.Infof("Triggering policy bundle download from oci registry.")
		err = d.Trigger(ctx)
		if err != nil {
			log.Errorf("failed to trigger bundle download: %s", err)
			return nil, err
		}

		err = update.Error
		if err != nil {
			log.Errorf("failed to download bundle: %s", err)
			return nil, err
		}

		if update.Bundle == nil || len(update.Bundle.Modules) == 0 {
			err = fmt.Errorf("expected bundle with at least one module but got none")
			log.Errorf("expected bundle with at least one module but got none")
			return nil, err
		}

		bundle = update.Bundle
		log.Infof("Bundle downloaded successfully.")
	} else if config.Modules != nil {
		policySet = config.Modules
	} else if config.AllowAll == "true" {
		module := `
			package ccr.policy

			import future.keywords

			default on_request_headers = true
			default on_request_body = true
			default on_response_headers = true
			default on_response_body = true
		`
		policySet = map[string]string{
			"allow-all.rego": module,
		}
	} else {
		return nil, fmt.Errorf("Need to specify a bundle_resource.")
	}

	factory := &opaFilterFactory{
		config:        config,
		policyQueries: make(map[rule]rego.PreparedEvalQuery),
		tracer:        tracer,
	}

	// Check if sev device exists on the platform; if not then ccr is being hosted on
	// non-confidential compute.
	if isSevSnp() {
		factory.teeType = "sevsnpvm"
	} else {
		factory.teeType = "none"
	}

	// Load the data in the policy if it was specified.
	var jsonConfig map[string]interface{} = make(map[string]interface{})
	if config.Data != nil {
		jsonConfig = config.Data
	}

	store := inmem.NewFromObject(jsonConfig)
	txn, err := store.NewTransaction(context.Background(), storage.WriteParams)
	if err != nil {
		log.Errorf("failed to create new transaction: %s", err)
		return nil, err
	}

	for _, r := range rules {
		query := fmt.Sprintf("data.ccr.policy.%s", r)
		factory.policyQueries[r], err = preparePolicyEval(query, policySet, bundle, store, txn)
		if err != nil {
			log.Errorf("failed to prepare %s query: %s", r, err)
			return nil, err
		}
	}

	err = store.Commit(context.Background(), txn)
	if err != nil {
		log.Errorf("failed to commit store transaction: %s", err)
		return nil, err
	}

	return factory, nil
}