diff --git a/.github/workflows/pull-request.yaml b/.github/workflows/pull-request.yaml index 20cb91eb..e276517a 100644 --- a/.github/workflows/pull-request.yaml +++ b/.github/workflows/pull-request.yaml @@ -78,6 +78,7 @@ jobs: retry() { for i in $(seq 1 30); do if "$@"; then return 0; fi; echo "attempt $i failed, retrying in 5s..."; sleep 5; done; return 1; } retry sh -c 'curl --fail -s http://127.0.0.1:8282/v1/bundles -H "Authorization: bearer sesame" | jq -e ".result | length > 0"' retry sh -c 'curl --fail -s http://127.0.0.1:8181/v1/data/k8s/authz/decision/status/allowed -d "{\"input\":{\"apiVersion\":\"v1\"}}" | jq -e ".result == true"' + retry sh -c 'curl --fail -s http://127.0.0.1:8181/v1/data/example/allow -d "{\"input\":{\"method\":\"POST\",\"path\":[\"posts\"],\"subject\":{\"user\":\"alice\"}}}" | jq -e ".result == true"' - name: dump logs run: docker compose logs working-directory: examples/docker diff --git a/cmd/backtest/backtest.go b/cmd/backtest/backtest.go index d265cc9d..5d020b01 100644 --- a/cmd/backtest/backtest.go +++ b/cmd/backtest/backtest.go @@ -28,6 +28,7 @@ import ( "github.com/open-policy-agent/opa-control-plane/internal/logging" "github.com/open-policy-agent/opa-control-plane/internal/progress" "github.com/open-policy-agent/opa-control-plane/internal/s3" + objectstore "github.com/open-policy-agent/opa-control-plane/pkg/objectstorage" "github.com/open-policy-agent/opa/ast" // nolint:staticcheck "github.com/open-policy-agent/opa/bundle" // nolint:staticcheck "github.com/open-policy-agent/opa/sdk" // nolint:staticcheck @@ -311,7 +312,12 @@ func backtestBundle(ctx context.Context, opts Options, styra *das.Client, b *con return err } - r, err := s.Download(ctx) + d, ok := s.(objectstore.Downloader) + if !ok { + return fmt.Errorf("object storage for %q does not support download", b.Name) + } + + r, err := d.Download(ctx) if err != nil { return err } diff --git a/cmd/compare/compare.go b/cmd/compare/compare.go index 20e76e5e..a4002e44 100644 --- a/cmd/compare/compare.go +++ b/cmd/compare/compare.go @@ -25,6 +25,7 @@ import ( "github.com/open-policy-agent/opa-control-plane/internal/config" "github.com/open-policy-agent/opa-control-plane/internal/logging" "github.com/open-policy-agent/opa-control-plane/internal/s3" + objectstore "github.com/open-policy-agent/opa-control-plane/pkg/objectstorage" ) var log *logging.Logger @@ -214,7 +215,11 @@ func compareSystem(ctx context.Context, client *das.Client, v1 *das.V1System, sy if err != nil { return nil, err } - r, err := s.Download(ctx) + d, ok := s.(objectstore.Downloader) + if !ok { + return nil, fmt.Errorf("object storage for %q does not support download", system.Name) + } + r, err := d.Download(ctx) if err != nil { return nil, err } diff --git a/cmd/run/run.go b/cmd/run/run.go index 26a878c1..45d0559e 100644 --- a/cmd/run/run.go +++ b/cmd/run/run.go @@ -71,7 +71,13 @@ func init() { } go func() { - if err := server.New().WithDatabase(svc.Database()).WithReadiness(svc.Ready).WithConfig(config.Service).Init().ListenAndServe(params.addr); err != nil { + srv := server.New(). + WithDatabase(svc.Database()). + WithReadiness(svc.Ready). + WithConfig(config.Service). + WithBundleStorages(svc.BundleStorages()). + Init() + if err := srv.ListenAndServe(params.addr); err != nil { log.Fatalf("failed to start server: %v", err) } }() diff --git a/config/schema.json b/config/schema.json index a0b07117..577a9979 100644 --- a/config/schema.json +++ b/config/schema.json @@ -220,6 +220,14 @@ }, "type": "object" }, + "ConfigHTTPServer": { + "properties": { + "path": { + "type": "string" + } + }, + "type": "object" + }, "ConfigLabels": { "additionalProperties": { "type": "string" @@ -251,6 +259,9 @@ }, "gcp": { "$ref": "#/definitions/ConfigGCPCloudStorage" + }, + "http_server": { + "$ref": "#/definitions/ConfigHTTPServer" } }, "type": "object" @@ -345,7 +356,8 @@ "administrator", "viewer", "owner", - "stack_owner" + "stack_owner", + "downloader" ], "type": "string" } diff --git a/e2e/cli/run_http_server.txtar b/e2e/cli/run_http_server.txtar new file mode 100644 index 00000000..32335ad5 --- /dev/null +++ b/e2e/cli/run_http_server.txtar @@ -0,0 +1,63 @@ +! exec $OPACTL run --addr ./ocp.sock --config config.d/bundle.yml --data-dir tmp &opactl& + +exec curl --retry 5 --retry-all-errors --unix-socket ocp.sock http://localhost/health + +# Wait for bundle to be built and download it +retry curl -f --unix-socket ocp.sock -H 'Authorization: Bearer test-token' http://localhost/v1/external/bundles/hello-world/bundle.tar.gz -o bundle.tar.gz + +# Verify the bundle is a valid tarball with expected contents +exec tar tf bundle.tar.gz +cmp stdout exp/tarball + +# Verify ETag header is returned +exec curl -f -D headers.txt --unix-socket ocp.sock -H 'Authorization: Bearer test-token' http://localhost/v1/external/bundles/hello-world/bundle.tar.gz -o /dev/null +exec grep Etag headers.txt + +# Verify unauthenticated request is rejected +! exec curl -f --unix-socket ocp.sock http://localhost/v1/external/bundles/hello-world/bundle.tar.gz + +# Verify bad token is rejected +! exec curl -f --unix-socket ocp.sock -H 'Authorization: Bearer wrong-token' http://localhost/v1/external/bundles/hello-world/bundle.tar.gz + +# Verify downloader-only token can download bundles +exec curl -f --unix-socket ocp.sock -H 'Authorization: Bearer dl-token' http://localhost/v1/external/bundles/hello-world/bundle.tar.gz -o /dev/null + +# Verify downloader-only token cannot access the API +! exec curl -f --unix-socket ocp.sock -H 'Authorization: Bearer dl-token' http://localhost/v1/bundles/hello-world + +# Verify unknown bundle path returns 404 +! exec curl -f --unix-socket ocp.sock -H 'Authorization: Bearer test-token' http://localhost/v1/external/bundles/nonexistent/bundle.tar.gz + +kill opactl +wait opactl + +-- files/sources/hello-world/rules/rules.rego -- +package rules +import rego.v1 +result if input.yay +-- config.d/bundle.yml -- +tokens: + test: + api_key: test-token + scopes: + - role: viewer + downloader: + api_key: dl-token + scopes: + - role: downloader +bundles: + hello-world: + object_storage: + http_server: + path: bundles/hello-world/bundle.tar.gz + requirements: + - source: hello-world +sources: + hello-world: + directory: files/sources/hello-world + paths: + - rules/rules.rego +-- exp/tarball -- +/data.json +/hello-world/rules/rules.rego +/.manifest diff --git a/e2e/migrate_e2e_test.go b/e2e/migrate_e2e_test.go index 23a00161..665af89c 100644 --- a/e2e/migrate_e2e_test.go +++ b/e2e/migrate_e2e_test.go @@ -28,6 +28,7 @@ import ( ocp_fs "github.com/open-policy-agent/opa-control-plane/internal/fs" "github.com/open-policy-agent/opa-control-plane/internal/logging" "github.com/open-policy-agent/opa-control-plane/internal/s3" + objectstore "github.com/open-policy-agent/opa-control-plane/pkg/objectstorage" "github.com/open-policy-agent/opa-control-plane/pkg/service" "github.com/open-policy-agent/opa-control-plane/internal/test/tempfs" "github.com/open-policy-agent/opa-control-plane/libraries" @@ -653,7 +654,12 @@ func TestMigration(t *testing.T) { t.Fatal(err) } - r, err := s.Download(ctx) + d, ok := s.(objectstore.Downloader) + if !ok { + t.Fatal("object storage does not support download") + } + + r, err := d.Download(ctx) if err != nil { t.Fatal(err) } diff --git a/examples/docker/README.md b/examples/docker/README.md index cb047eeb..648be006 100644 --- a/examples/docker/README.md +++ b/examples/docker/README.md @@ -27,7 +27,12 @@ This will generate the certs needed for the examples (via `tls/gencerts.sh`), an When it's running, you can go to http://127.0.0.1:9090 to examine the published Prometheus metrics. Enter `ocp_` in the expression field to see completion options for the various metrics in the expression field to see completion options for the various metrics in the expression field to see completion options for the various metrics. -The OCP configuration already contains a bundle, pulling some rego from https://github.com/open-policy-agent/contrib, so that there are some metrics to explore. +The OCP configuration contains two bundles: + +1. `hello-world` — pushed to S3, OPA pulls from s3proxy +2. `hello-http` — served directly by OCP via in-memory `http_server` storage, OPA pulls from OCP using a `downloader` token + +Both pull rego from https://github.com/open-policy-agent/contrib, so that there are some metrics to explore. > [!WARNING] > Note that on startup, it will take a while until the system settles: diff --git a/examples/docker/ocp.yml b/examples/docker/ocp.yml index 59399ded..dcc8e3be 100644 --- a/examples/docker/ocp.yml +++ b/examples/docker/ocp.yml @@ -9,6 +9,12 @@ bundles: credentials: s3-creds requirements: - source: git-policies + hello-http: + object_storage: + http_server: + path: bundles/hello-http/bundle.tar.gz + requirements: + - source: http-policies sources: git-policies: git: @@ -16,11 +22,21 @@ sources: commit: 0f81d9a0018451d98dcd3f4bb885ee676f49f6fe included_files: - k8s_authorization/policy/policy.rego + http-policies: + git: + repo: https://github.com/open-policy-agent/contrib + commit: 0f81d9a0018451d98dcd3f4bb885ee676f49f6fe + included_files: + - data_filter_example/example.rego tokens: admin: api_key: sesame scopes: - role: administrator + opa: + api_key: opa-token + scopes: + - role: downloader database: sql: driver: postgres diff --git a/examples/docker/opa.yml b/examples/docker/opa.yml index 843e3dcd..fcf19ebc 100644 --- a/examples/docker/opa.yml +++ b/examples/docker/opa.yml @@ -1,7 +1,14 @@ services: s3: url: http://s3proxy:80 + ocp: + url: http://ocp:8282 + headers: + Authorization: Bearer opa-token bundles: hello-world: service: s3 resource: bundles/hello-world + hello-http: + service: ocp + resource: v1/external/bundles/hello-http/bundle.tar.gz diff --git a/internal/authz/authz.rego b/internal/authz/authz.rego index 6bdd61a1..898c2374 100644 --- a/internal/authz/authz.rego +++ b/internal/authz/authz.rego @@ -14,6 +14,7 @@ allow if { in_tenant(data.principals.tenant_id) input.permission in [ "bundles.view", + "bundles.download", "sources.view", "secrets.view", "stacks.view", @@ -40,6 +41,13 @@ allow if { input.permission == "stacks.create" } +allow if { + data.principals.id == input.principal + data.principals.role == "downloader" + in_tenant(data.principals.tenant_id) + input.permission == "bundles.download" +} + allow if { data.resource_permissions.name == input.name data.resource_permissions.resource == input.resource diff --git a/internal/authz/authz_test.rego b/internal/authz/authz_test.rego index 83188b61..a25c007a 100644 --- a/internal/authz/authz_test.rego +++ b/internal/authz/authz_test.rego @@ -14,6 +14,7 @@ test_admin_can_do_anything if { read_permissions := { "bundles.view", + "bundles.download", "sources.view", "stacks.view", "secrets.view", @@ -129,3 +130,36 @@ test_explicit_permission_grant if { with input.resource as "sources" with input.tenant as "ten10" } + +test_downloader_can_download_bundles if { + data.authz.allow with input.principal as "testuser" + with data.principals.id as "testuser" + with data.principals.role as "downloader" + with data.principals.tenant_id as 10 + with data.tenants.id as 10 + with data.tenants.name as "ten10" + with input.tenant as "ten10" + with input.permission as "bundles.download" +} + +test_downloader_cannot_view_bundles if { + not data.authz.allow with input.principal as "testuser" + with data.principals.id as "testuser" + with data.principals.role as "downloader" + with data.principals.tenant_id as 10 + with data.tenants.id as 10 + with data.tenants.name as "ten10" + with input.tenant as "ten10" + with input.permission as "bundles.view" +} + +test_downloader_cannot_view_sources if { + not data.authz.allow with input.principal as "testuser" + with data.principals.id as "testuser" + with data.principals.role as "downloader" + with data.principals.tenant_id as 10 + with data.tenants.id as 10 + with data.tenants.name as "ten10" + with input.tenant as "ten10" + with input.permission as "sources.view" +} diff --git a/internal/config/config.go b/internal/config/config.go index 6e8401a0..85115316 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -35,6 +35,7 @@ type ( GCPCloudStorage = extconfig.GCPCloudStorage AzureBlobStorage = extconfig.AzureBlobStorage FileSystemStorage = extconfig.FileSystemStorage + HTTPServer = extconfig.HTTPServer StringSet = extconfig.StringSet Requirements = extconfig.Requirements Files = extconfig.Files @@ -449,7 +450,7 @@ func (t *Token) Equal(other *Token) bool { } type Scope struct { - Role string `json:"role" enum:"administrator,viewer,owner,stack_owner"` + Role string `json:"role" enum:"administrator,viewer,owner,stack_owner,downloader"` } func scopesEqual(a, b []Scope) bool { diff --git a/internal/database/database.go b/internal/database/database.go index 0f051d2d..e85bc158 100644 --- a/internal/database/database.go +++ b/internal/database/database.go @@ -646,6 +646,20 @@ func (d *Database) GetBundle(ctx context.Context, principal, tenant, name string return bundles[0], nil } +func (d *Database) CheckBundleDownload(ctx context.Context, principal, tenant, name string) error { + return tx1(ctx, d, func(tx *sql.Tx) error { + if err := d.resourceExists(ctx, tx, tenant, "bundles", name); err != nil { + return err + } + + ad := d.accessFactory().WithPrincipal(principal).WithTenant(tenant).WithResource("bundles").WithPermission("bundles.download").WithName(name) + if !d.authorizer.Check(ctx, tx, d.arg, ad) { + return ErrNotAuthorized + } + return nil + }) +} + func (d *Database) DeleteBundle(ctx context.Context, principal, tenant, name string) error { return tx1(ctx, d, func(tx *sql.Tx) error { if err := d.prepareDelete(ctx, tx, principal, tenant, "bundles", name, "bundles.manage"); err != nil { @@ -698,6 +712,7 @@ func (d *Database) ListBundles(ctx context.Context, principal, tenant string, op bundles.azure_container, bundles.azure_path, bundles.filepath, + bundles.http_server_path, bundles.excluded, bundles.rebuild_interval, bundles.options @@ -756,6 +771,7 @@ LEFT JOIN gcpProject, gcpObject *string // GCP object storage azureAccountURL, azureContainer, azurePath *string // Azure object storage filepath *string // File system storage + httpServerPath *string // HTTP server storage excluded *string interval *string options *string @@ -775,6 +791,7 @@ LEFT JOIN &row.gcpProject, &row.gcpObject, // GCP &row.azureAccountURL, &row.azureContainer, &row.azurePath, // Azure &row.filepath, + &row.httpServerPath, &row.excluded, &row.interval, &row.options, @@ -857,6 +874,10 @@ LEFT JOIN bundle.ObjectStorage.FileSystemStorage = &config.FileSystemStorage{ Path: *row.filepath, } + } else if row.httpServerPath != nil { + bundle.ObjectStorage.HTTPServer = &config.HTTPServer{ + Path: *row.httpServerPath, + } } if row.excluded != nil { @@ -1549,7 +1570,7 @@ func (d *Database) UpsertBundle(ctx context.Context, principal, tenant string, b return err } - var s3url, s3region, s3bucket, s3key, gcpProject, gcpObject, azureAccountURL, azureContainer, azurePath, filepath *string + var s3url, s3region, s3bucket, s3key, gcpProject, gcpObject, azureAccountURL, azureContainer, azurePath, filepath, httpServerPath *string if bundle.ObjectStorage.AmazonS3 != nil { s3url = &bundle.ObjectStorage.AmazonS3.URL s3region = &bundle.ObjectStorage.AmazonS3.Region @@ -1569,6 +1590,9 @@ func (d *Database) UpsertBundle(ctx context.Context, principal, tenant string, b if bundle.ObjectStorage.FileSystemStorage != nil { filepath = &bundle.ObjectStorage.FileSystemStorage.Path } + if bundle.ObjectStorage.HTTPServer != nil { + httpServerPath = &bundle.ObjectStorage.HTTPServer.Path + } labels, err := json.Marshal(bundle.Labels) if err != nil { @@ -1591,13 +1615,13 @@ func (d *Database) UpsertBundle(ctx context.Context, principal, tenant string, b "s3url", "s3region", "s3bucket", "s3key", "gcp_project", "gcp_object", "azure_account_url", "azure_container", "azure_path", - "filepath", "excluded", + "filepath", "http_server_path", "excluded", "rebuild_interval", "options"}, []string{"name"}, bundle.Name, string(labels), bundle.Revision, s3url, s3region, s3bucket, s3key, gcpProject, gcpObject, azureAccountURL, azureContainer, azurePath, - filepath, string(excluded), bundle.Interval.String(), + filepath, httpServerPath, string(excluded), bundle.Interval.String(), options) if err != nil { return err diff --git a/internal/migrations/migrations.go b/internal/migrations/migrations.go index f83c4655..0d99bf83 100644 --- a/internal/migrations/migrations.go +++ b/internal/migrations/migrations.go @@ -42,6 +42,7 @@ func Migrations(dialect string) (fs.FS, error) { addBundlesRevision(23, dialect), addDatasourcesCredentialsName(24, dialect), addSourcesGitCredentialsName(25, dialect), + addBundlesHTTPServerPath(26, dialect), ), nil } diff --git a/internal/migrations/migrations_next.go b/internal/migrations/migrations_next.go index 262d6018..514eb8e5 100644 --- a/internal/migrations/migrations_next.go +++ b/internal/migrations/migrations_next.go @@ -23,6 +23,20 @@ func addBundlesRevision(offset int, dialect string) fs.FS { }) } +func addBundlesHTTPServerPath(offset int, dialect string) fs.FS { + var stmt string + switch dialect { + case "sqlite", "postgresql", "cockroachdb": + stmt = `ALTER TABLE bundles ADD http_server_path TEXT` + case "mysql": + stmt = `ALTER TABLE bundles ADD http_server_path VARCHAR(255)` + } + + return ocp_fs.MapFS(map[string]string{ + fmt.Sprintf("%03d_add_bundles_http_server_path.up.sql", offset): stmt, + }) +} + func addSourcesGitCredentialsName(offset int, dialect string) fs.FS { var stmt string switch dialect { diff --git a/internal/s3/s3.go b/internal/s3/s3.go index 3c46c4bb..d96ca473 100644 --- a/internal/s3/s3.go +++ b/internal/s3/s3.go @@ -9,8 +9,10 @@ import ( "errors" "fmt" "io" + "net/http" "os" "path/filepath" + "sync" "cloud.google.com/go/storage" "github.com/Azure/azure-sdk-for-go/sdk/azidentity" @@ -29,8 +31,67 @@ var ( _ ext_os.ObjectStorage = (*GCPCloudStorage)(nil) _ ext_os.ObjectStorage = (*AzureBlobStorage)(nil) _ ext_os.ObjectStorage = (*FileSystemStorage)(nil) + _ ext_os.ObjectStorage = (*InMemoryStorage)(nil) ) +// InMemoryStorage is an ObjectStorage implementation that holds the bundle +// bytes in memory. It also implements http.Handler so it can serve bundles +// directly via the OPA bundle protocol. +type InMemoryStorage struct { + mu sync.RWMutex + data []byte + etag string + revision string +} + +// NewInMemoryStorage creates a new InMemoryStorage instance. +func NewInMemoryStorage() *InMemoryStorage { + return &InMemoryStorage{} +} + +func (m *InMemoryStorage) Upload(_ context.Context, body io.ReadSeeker, _ string, revision string, _ int64) error { + data, err := io.ReadAll(body) + if err != nil { + return err + } + + hash := sha256.Sum256(data) + etag := hex.EncodeToString(hash[:]) + + m.mu.Lock() + defer m.mu.Unlock() + + m.data = data + m.etag = etag + m.revision = revision + + return nil +} + +func (m *InMemoryStorage) ServeHTTP(w http.ResponseWriter, r *http.Request) { + m.mu.RLock() + defer m.mu.RUnlock() + + if m.data == nil { + http.Error(w, "no bundle available", http.StatusNotFound) + return + } + + if match := r.Header.Get("If-None-Match"); match == m.etag { + w.WriteHeader(http.StatusNotModified) + return + } + + w.Header().Set("Content-Type", "application/gzip") + w.Header().Set("ETag", m.etag) + if m.revision != "" { + w.Header().Set("X-OPA-Revision", m.revision) + } + + w.WriteHeader(http.StatusOK) + _, _ = w.Write(m.data) +} + type ( AmazonS3 struct { bucket string @@ -185,6 +246,8 @@ func New(ctx context.Context, c config.ObjectStorage) (ext_os.ObjectStorage, err return &AzureBlobStorage{container: c.AzureBlobStorage.Container, path: c.AzureBlobStorage.Path, client: client}, nil case c.FileSystemStorage != nil: return &FileSystemStorage{path: c.FileSystemStorage.Path}, nil + case c.HTTPServer != nil: + return NewInMemoryStorage(), nil default: return nil, ErrUnsupportedProvider } @@ -290,10 +353,6 @@ func (s *GCPCloudStorage) Upload(ctx context.Context, body io.ReadSeeker, _ stri return w.Close() } -func (*GCPCloudStorage) Download(context.Context) (io.Reader, error) { - return nil, errors.New("not implemented") -} - func (s *AzureBlobStorage) Upload(ctx context.Context, body io.ReadSeeker, _ string, revision string, _ int64) error { opts := &azblob.UploadStreamOptions{} if revision != "" { @@ -305,10 +364,6 @@ func (s *AzureBlobStorage) Upload(ctx context.Context, body io.ReadSeeker, _ str return err } -func (*AzureBlobStorage) Download(context.Context) (io.Reader, error) { - return nil, errors.New("not implemented") -} - func (s *FileSystemStorage) Upload(ctx context.Context, body io.ReadSeeker, _ string, _ string, _ int64) error { digest, equal, err := s.check(ctx, body) if equal || err != nil { diff --git a/internal/s3/s3_test.go b/internal/s3/s3_test.go index 7c43fdd8..bb5a9426 100644 --- a/internal/s3/s3_test.go +++ b/internal/s3/s3_test.go @@ -6,6 +6,7 @@ import ( "crypto/sha256" "encoding/hex" "io" + "net/http" "net/http/httptest" "testing" @@ -69,7 +70,7 @@ func TestS3(t *testing.T) { t.Fatalf("expected object contents to be 'bundle content', got '%s'", contents) } - reader, err := storage.Download(ctx) + reader, err := storage.(*AmazonS3).Download(ctx) if err != nil { t.Fatal(err) } @@ -212,3 +213,98 @@ func TestS3WithoutRevision(t *testing.T) { t.Errorf("expected revision metadata to not be present, but got %q", output.Metadata["revision"]) } } + +func TestInMemoryStorage(t *testing.T) { + ms := NewInMemoryStorage() + + // Upload content. + content := []byte("test bundle data") + err := ms.Upload(t.Context(), bytes.NewReader(content), "mybundle", "rev1", int64(len(content))) + if err != nil { + t.Fatalf("unexpected upload error: %v", err) + } + + // Verify via HTTP handler. + rec := httptest.NewRecorder() + req := httptest.NewRequest(http.MethodGet, "/bundles/mybundle", nil) + ms.ServeHTTP(rec, req) + if !bytes.Equal(rec.Body.Bytes(), content) { + t.Fatalf("expected %q, got %q", content, rec.Body.Bytes()) + } + + // Upload new content replaces old. + content2 := []byte("updated bundle") + if err := ms.Upload(t.Context(), bytes.NewReader(content2), "mybundle", "rev2", int64(len(content2))); err != nil { + t.Fatalf("unexpected upload error: %v", err) + } + rec = httptest.NewRecorder() + req = httptest.NewRequest(http.MethodGet, "/bundles/mybundle", nil) + ms.ServeHTTP(rec, req) + if !bytes.Equal(rec.Body.Bytes(), content2) { + t.Fatalf("expected %q, got %q", content2, rec.Body.Bytes()) + } +} + +func TestInMemoryStorageServeHTTP(t *testing.T) { + ms := NewInMemoryStorage() + + // Serve before upload should return 404. + req := httptest.NewRequest(http.MethodGet, "/bundles/test", nil) + rec := httptest.NewRecorder() + ms.ServeHTTP(rec, req) + if rec.Code != http.StatusNotFound { + t.Fatalf("expected 404, got %d", rec.Code) + } + + // Upload content. + content := []byte("gzipped bundle content") + if err := ms.Upload(t.Context(), bytes.NewReader(content), "test", "rev1", int64(len(content))); err != nil { + t.Fatal(err) + } + + // Serve should return content with correct headers. + rec = httptest.NewRecorder() + req = httptest.NewRequest(http.MethodGet, "/bundles/test", nil) + ms.ServeHTTP(rec, req) + + if rec.Code != http.StatusOK { + t.Fatalf("expected 200, got %d", rec.Code) + } + if ct := rec.Header().Get("Content-Type"); ct != "application/gzip" { + t.Fatalf("expected Content-Type application/gzip, got %q", ct) + } + etag := rec.Header().Get("ETag") + if etag == "" { + t.Fatal("expected ETag header") + } + + hash := sha256.Sum256(content) + expectedETag := hex.EncodeToString(hash[:]) + if etag != expectedETag { + t.Fatalf("expected ETag %q, got %q", expectedETag, etag) + } + if rev := rec.Header().Get("X-OPA-Revision"); rev != "rev1" { + t.Fatalf("expected X-OPA-Revision rev1, got %q", rev) + } + if !bytes.Equal(rec.Body.Bytes(), content) { + t.Fatal("body mismatch") + } + + // If-None-Match with matching ETag should return 304. + rec = httptest.NewRecorder() + req = httptest.NewRequest(http.MethodGet, "/bundles/test", nil) + req.Header.Set("If-None-Match", etag) + ms.ServeHTTP(rec, req) + if rec.Code != http.StatusNotModified { + t.Fatalf("expected 304, got %d", rec.Code) + } + + // If-None-Match with non-matching ETag should return 200. + rec = httptest.NewRecorder() + req = httptest.NewRequest(http.MethodGet, "/bundles/test", nil) + req.Header.Set("If-None-Match", "stale-etag") + ms.ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("expected 200, got %d", rec.Code) + } +} diff --git a/internal/server/server.go b/internal/server/server.go index 41b1f0a7..145604ad 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -21,15 +21,17 @@ import ( "github.com/open-policy-agent/opa-control-plane/internal/metrics" "github.com/open-policy-agent/opa-control-plane/internal/server/chain" "github.com/open-policy-agent/opa-control-plane/internal/server/types" + "github.com/open-policy-agent/opa-control-plane/pkg/service" ) const defaultTenant = "default" type Server struct { - router *http.ServeMux - db *database.Database - readyFn func(context.Context) error - apiPrefix string + router *http.ServeMux + db *database.Database + readyFn func(context.Context) error + apiPrefix string + bundleStorages []service.BundleStorage } func New() *Server { @@ -77,6 +79,22 @@ func (s *Server) Init() *Server { setup("PUT", "/v1/secrets/{secret}", s.v1SecretsPut) setup("DELETE", "/v1/secrets/{secret}", s.v1SecretsDelete) + if len(s.bundleStorages) > 0 { + for _, bs := range s.bundleStorages { + urlPath := "/v1/external/" + path.Clean(bs.Path) + h := bs.Handler // capture for closure + name := bs.BundleName // capture for closure + s.router.Handle("GET "+apiPrefix+urlPath, append(base, metrics.InstrumentHandler(apiPrefix+urlPath)).ThenFunc(func(w http.ResponseWriter, r *http.Request) { + principal, tenant := s.auth(r) + if err := s.db.CheckBundleDownload(r.Context(), principal, tenant, name); err != nil { + errorAuto(w, err) + return + } + h.ServeHTTP(w, r) + })) + } + } + return s } @@ -96,8 +114,8 @@ func (s *Server) WithReadiness(fn func(context.Context) error) *Server { } func (s *Server) ListenAndServe(addr string) error { - if strings.HasPrefix(addr, "unix://") { - socketPath := strings.TrimPrefix(addr, "unix://") + if after, ok := strings.CutPrefix(addr, "unix://"); ok { + socketPath := after return s.listenAndServeUnix(socketPath) } if strings.HasPrefix(addr, "/") || strings.HasPrefix(addr, "./") { @@ -131,6 +149,11 @@ func (s *Server) WithConfig(config *config.Service) *Server { return s } +func (s *Server) WithBundleStorages(storages []service.BundleStorage) *Server { + s.bundleStorages = storages + return s +} + func (s *Server) health(w http.ResponseWriter, r *http.Request) { err := s.readyFn(r.Context()) diff --git a/pkg/config/config.go b/pkg/config/config.go index 45ecfba8..e04028d4 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -335,6 +335,7 @@ type ObjectStorage struct { GCPCloudStorage *GCPCloudStorage `json:"gcp,omitempty"` AzureBlobStorage *AzureBlobStorage `json:"azure,omitempty"` FileSystemStorage *FileSystemStorage `json:"filesystem,omitempty"` + HTTPServer *HTTPServer `json:"http_server,omitempty"` } func (o *ObjectStorage) validate() error { @@ -347,7 +348,10 @@ func (o *ObjectStorage) validate() error { if err := o.AzureBlobStorage.validate(); err != nil { return err } - return o.FileSystemStorage.validate() + if err := o.FileSystemStorage.validate(); err != nil { + return err + } + return o.HTTPServer.validate() } // AmazonS3 defines the configuration for an Amazon S3-compatible object storage. @@ -452,6 +456,24 @@ func (f *FileSystemStorage) validate() error { return nil } +// HTTPServer defines the configuration for serving bundles directly from OCP's HTTP server. +// When configured, the bundle is held in memory and served at the specified path. +type HTTPServer struct { + Path string `json:"path"` +} + +func (h *HTTPServer) validate() error { + if h == nil { + return nil + } + + if h.Path == "" { + return errors.New("http server path is required") + } + + return nil +} + // Git defines the Git synchronization configuration used by OPA Control Plane Sources. type Git struct { Repo string `json:"repo"` diff --git a/pkg/objectstorage/objectstorage.go b/pkg/objectstorage/objectstorage.go index 3f9ba36c..5aca0c09 100644 --- a/pkg/objectstorage/objectstorage.go +++ b/pkg/objectstorage/objectstorage.go @@ -5,12 +5,14 @@ import ( "io" ) -// ObjectStorage defines the interface for uploading and downloading bundle artifacts -// to/from object storage systems (e.g., S3, GCS, Azure Blob Storage). +// ObjectStorage defines the interface for uploading bundle artifacts +// to object storage systems (e.g., S3, GCS, Azure Blob Storage). type ObjectStorage interface { // Upload stores a bundle artifact in object storage. Upload(ctx context.Context, body io.ReadSeeker, name string, revision string, totalSize int64) error +} - // Download retrieves a bundle artifact from object storage. +// Downloader is an optional interface for retrieving bundle artifacts from object storage. +type Downloader interface { Download(ctx context.Context) (io.Reader, error) } diff --git a/pkg/service/service.go b/pkg/service/service.go index 1288bf18..307fe409 100644 --- a/pkg/service/service.go +++ b/pkg/service/service.go @@ -9,6 +9,7 @@ import ( "fmt" "io/fs" "maps" + "net/http" "path" "path/filepath" "slices" @@ -65,6 +66,12 @@ type Service struct { initialized bool storage ext_os.ObjectStorage secretProvider pkgsync.SecretProvider + bundleStorages map[bundleStorageKey]*s3.InMemoryStorage +} + +type bundleStorageKey struct { + path string + bundleName string } type Report struct { @@ -176,13 +183,57 @@ func (s *Service) WithSecretProvider(provider pkgsync.SecretProvider) *Service { return s } +// BundleStorages returns the list of bundle storages served via in-memory storage. +// Returns nil if no http_server storages are configured. +func (s *Service) BundleStorages() []BundleStorage { + if len(s.bundleStorages) == 0 { + return nil + } + result := make([]BundleStorage, 0, len(s.bundleStorages)) + for key, storage := range s.bundleStorages { + result = append(result, BundleStorage{ + Path: key.path, + BundleName: key.bundleName, + Handler: storage, + }) + } + return result +} + +// BundleStorage associates a URL path with a bundle name and its HTTP handler. +type BundleStorage struct { + Path string + BundleName string + Handler http.Handler +} + func (s *Service) Init(ctx context.Context) error { if s.initialized { return nil } err := s.initDB(ctx) - s.initialized = err == nil - return err + if err != nil { + return err + } + s.initialized = true + + // Pre-create InMemoryStorage instances for bundles configured with + // http_server object storage, so they are available before Run() is called. + if s.config != nil { + for _, b := range s.config.Bundles { + if b.ObjectStorage.HTTPServer != nil { + if s.bundleStorages == nil { + s.bundleStorages = make(map[bundleStorageKey]*s3.InMemoryStorage) + } + s.bundleStorages[bundleStorageKey{ + path: b.ObjectStorage.HTTPServer.Path, + bundleName: b.Name, + }] = s3.NewInMemoryStorage() + } + } + } + + return nil } func (s *Service) Run(ctx context.Context) error { @@ -424,6 +475,14 @@ func (s *Service) launchWorkers(ctx context.Context) { if s.storage != nil { w.WithStorage(s.storage) + } else if b.ObjectStorage.HTTPServer != nil { + ms := s.bundleStorages[bundleStorageKey{path: b.ObjectStorage.HTTPServer.Path, bundleName: b.Name}] + if ms == nil { + s.log.Errorf("no in-memory storage found for bundle %q", b.Name) + failures[b.Name] = Status{State: BuildStateConfigError, Message: "http_server storage not initialized"} + continue + } + w.WithStorage(ms) } else { storage, err := s3.New(ctx, b.ObjectStorage) if err != nil {