222twotwotwo commented on code in PR #947: URL: https://github.com/apache/dubbo-go-pixiu/pull/947#discussion_r3510501285
########## admin/initialize/e2e_opa_test.go: ########## @@ -0,0 +1,626 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package initialize + +import ( + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" +) + +import ( + "github.com/gin-gonic/gin" + + "github.com/open-policy-agent/opa/rego" +) + +import ( + adminconfig "github.com/apache/dubbo-go-pixiu/admin/config" + "github.com/apache/dubbo-go-pixiu/pkg/common/extension/filter" + contextHttp "github.com/apache/dubbo-go-pixiu/pkg/context/http" + opaFilter "github.com/apache/dubbo-go-pixiu/pkg/filter/opa" +) + +// This file is the PR2 deliverable: a CI-runnable end-to-end test that exercises +// the full admin → OPA Server → gateway filter chain in a single process. +// +// admin REST PUT /config/api/opa/policy +// → JWT auth → controller → logic → HTTP PUT /v1/policies/<id> +// ↓ +// regoMockOPA (httptest) — stores rego module text AND compiles it +// ↑ +// gateway OPA filter POST {server_url}/v1/data/<path> +// ← input(method, path, headers, ...) → rego evaluation +// → filter.Continue (allow) | filter.Stop+403 (deny) +// +// The mock OPA is "smart" — it uses the real github.com/open-policy-agent/opa +// rego library that pkg/filter/opa already depends on, so the policy +// evaluation in the test is identical to what a real OPA server would do. +// No docker, no etcd, no external process needed. +// --------------------------------------------------------------------------- +// regoMockOPA: an in-process OPA server that speaks the subset of the OPA REST +// API exercised by the admin + gateway: PUT/GET/DELETE /v1/policies/{id} and +// POST /v1/data/<any/path>. +// --------------------------------------------------------------------------- +type regoMockOPA struct { + srv *httptest.Server + + mu sync.Mutex + policies map[string]string // policy_id -> rego module text + + // putRequests records every PUT for assertions (auth header, body, etc.). + putRequests []recordedRequest + + // decisionDelay, if set, makes the POST /v1/data/... handler sleep before + // evaluating. Lets tests exercise gateway-side timeouts. + decisionDelay time.Duration +} + +func newRegoMockOPA(t *testing.T) *regoMockOPA { + t.Helper() + m := ®oMockOPA{policies: map[string]string{}} + m.srv = httptest.NewServer(http.HandlerFunc(m.handle)) + t.Cleanup(m.srv.Close) + return m +} + +func (m *regoMockOPA) URL() string { return m.srv.URL } + +func (m *regoMockOPA) handle(w http.ResponseWriter, r *http.Request) { + switch { + case strings.HasPrefix(r.URL.Path, "/v1/policies/"): + m.handlePolicy(w, r) + case strings.HasPrefix(r.URL.Path, "/v1/data/"): + m.handleDecision(w, r) + default: + w.WriteHeader(http.StatusNotFound) + } +} + +func (m *regoMockOPA) handlePolicy(w http.ResponseWriter, r *http.Request) { + id := strings.TrimPrefix(r.URL.Path, "/v1/policies/") + body, _ := io.ReadAll(r.Body) + + switch r.Method { + case http.MethodPut: + // Sanity-check the rego compiles before accepting; mirrors real OPA's + // behavior of returning 400 with a compile error transcript. + if _, err := rego.New( + rego.Query("data"), + rego.Module(id, string(body)), + ).PrepareForEval(context.Background()); err != nil { + w.WriteHeader(http.StatusBadRequest) + _, _ = w.Write([]byte(err.Error())) + return + } + m.mu.Lock() + m.policies[id] = string(body) + m.putRequests = append(m.putRequests, recordedRequest{ + method: r.Method, + path: r.URL.Path, + ctype: r.Header.Get("Content-Type"), + auth: r.Header.Get("Authorization"), + body: string(body), + }) + m.mu.Unlock() + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte("{}")) + + case http.MethodGet: + m.mu.Lock() + raw, ok := m.policies[id] + m.mu.Unlock() + if !ok { + w.WriteHeader(http.StatusNotFound) + return + } + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]any{ + "result": map[string]any{"id": id, "raw": raw}, + }) + + case http.MethodDelete: + m.mu.Lock() + delete(m.policies, id) + m.mu.Unlock() + w.WriteHeader(http.StatusOK) + + default: + w.WriteHeader(http.StatusMethodNotAllowed) + } +} + +// handleDecision implements POST /v1/data/<rule path>. It bundles every stored +// policy into a single rego module set, runs the query, and replies with +// {"result": <value>} — matching the OPA REST contract that +// pkg/filter/opa/opa.go expects in evaluateServer(). +func (m *regoMockOPA) handleDecision(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodPost { + w.WriteHeader(http.StatusMethodNotAllowed) + return + } + + if m.decisionDelay > 0 { + time.Sleep(m.decisionDelay) + } + + var reqBody map[string]any + if err := json.NewDecoder(r.Body).Decode(&reqBody); err != nil { + w.WriteHeader(http.StatusBadRequest) + return + } + input := reqBody["input"] + + // /v1/data/http/authz/allow → data.http.authz.allow + rulePath := strings.TrimPrefix(r.URL.Path, "/v1/data/") + query := "data." + strings.ReplaceAll(rulePath, "/", ".") + + m.mu.Lock() + modules := make(map[string]string, len(m.policies)) + for id, raw := range m.policies { + modules[id] = raw + } + m.mu.Unlock() + + opts := []func(r *rego.Rego){rego.Query(query)} + for id, raw := range modules { + opts = append(opts, rego.Module(id, raw)) + } + pq, err := rego.New(opts...).PrepareForEval(context.Background()) + if err != nil { + // Real OPA returns 500 on bundle compile errors at decision time; the + // filter treats non-200 as BadGateway, which is what we want. + w.WriteHeader(http.StatusInternalServerError) + _, _ = w.Write([]byte(err.Error())) + return + } + + results, err := pq.Eval(r.Context(), rego.EvalInput(input)) + if err != nil { + w.WriteHeader(http.StatusInternalServerError) + _, _ = w.Write([]byte(err.Error())) + return + } + + resp := map[string]any{} + // No rules matched → omit "result" entirely. This is exactly what real + // OPA does, and it triggers the gateway's "missing 'result' field" branch + // — the same fail-closed behavior documented in test_opa.md §6.6. + if len(results) > 0 && len(results[0].Expressions) > 0 { + resp["result"] = results[0].Expressions[0].Value + } + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(resp) +} + +func (m *regoMockOPA) recordedPUTs() []recordedRequest { + m.mu.Lock() + defer m.mu.Unlock() + out := make([]recordedRequest, len(m.putRequests)) + copy(out, m.putRequests) + return out +} + +// --------------------------------------------------------------------------- +// Helpers for wiring admin router and gateway OPA filter against the mock. +// --------------------------------------------------------------------------- + +// installAdminRouterWithRegoMock mounts the real admin router with +// adminconfig.Bootstrap pointed at the given mock, mirroring installRouter() +// in router_opa_test.go but accepting a regoMockOPA instead. +func installAdminRouterWithRegoMock(t *testing.T, m *regoMockOPA, opaCfg adminconfig.OPAConfig) *gin.Engine { + t.Helper() + gin.SetMode(gin.TestMode) + t.Cleanup(func() { gin.SetMode(gin.DebugMode) }) + + prev := adminconfig.Bootstrap + cfg := opaCfg + if cfg.ServerURL == "" { + cfg.ServerURL = m.URL() + } + adminconfig.Bootstrap = &adminconfig.AdminBootstrap{OPA: cfg} + t.Cleanup(func() { adminconfig.Bootstrap = prev }) + + return Routers() +} + +// adminPutPolicy uses the real /config/api/opa/policy PUT route, signed with +// the same JWT key the middleware reads. The full request travels through +// gin → JWT middleware → controller → logic → mock OPA, just like in prod. +func adminPutPolicy(t *testing.T, r *gin.Engine, policyID, content string) { + t.Helper() + fields := map[string]string{"content": content} + if policyID != "" { + fields["policy_id"] = policyID + } + ctype, body := putMultipart(t, fields) + req := httptest.NewRequest(http.MethodPut, "/config/api/opa/policy", body) + req.Header.Set("Content-Type", ctype) + req.Header.Set("token", signToken(t)) + + w := doReq(t, r, req) + if w.Code != http.StatusOK { + t.Fatalf("admin PUT failed: status=%d body=%s", w.Code, w.Body.String()) + } + if !strings.Contains(w.Body.String(), "Update Success") { + t.Fatalf("admin PUT expected Update Success, got %s", w.Body.String()) + } +} + +// adminDeletePolicy hits DELETE /config/api/opa/policy. +func adminDeletePolicy(t *testing.T, r *gin.Engine, policyID string) { + t.Helper() + url := "/config/api/opa/policy" + if policyID != "" { + url = url + "?policy_id=" + policyID + } + req := httptest.NewRequest(http.MethodDelete, url, nil) + req.Header.Set("token", signToken(t)) + w := doReq(t, r, req) + if w.Code != http.StatusOK { + t.Fatalf("admin DELETE failed: status=%d body=%s", w.Code, w.Body.String()) + } +} + +// buildGatewayFilter constructs and applies the real gateway OPA filter +// (pkg/filter/opa) pointed at the same mock OPA the admin writes to. +func buildGatewayFilter(t *testing.T, mockURL, decisionPath string, timeoutMs int) filter.HttpDecodeFilter { + t.Helper() + plugin := &opaFilter.Plugin{} + factory, err := plugin.CreateFilterFactory() + if err != nil { + t.Fatalf("create filter factory: %v", err) + } + cfg := factory.Config().(*opaFilter.Config) + cfg.ServerURL = mockURL + cfg.DecisionPath = decisionPath + cfg.TimeoutMs = timeoutMs + if err := factory.Apply(); err != nil { + t.Fatalf("apply gateway filter: %v", err) + } + + chain := &e2eFilterChain{} + ctxStub := &contextHttp.HttpContext{ + Request: httptest.NewRequest(http.MethodGet, "/", nil), + Writer: httptest.NewRecorder(), + Ctx: context.Background(), + } + if err := factory.PrepareFilterChain(ctxStub, chain); err != nil { + t.Fatalf("prepare gateway filter chain: %v", err) + } + if len(chain.filters) != 1 { + t.Fatalf("expected 1 decode filter, got %d", len(chain.filters)) + } + return chain.filters[0] +} + +// driveGatewayRequest runs one HTTP request through the gateway OPA filter +// and returns the FilterStatus plus the captured HttpContext for status code +// and response body inspection. +func driveGatewayRequest(t *testing.T, f filter.HttpDecodeFilter, method, path string, headers map[string]string) (filter.FilterStatus, *contextHttp.HttpContext) { + t.Helper() + req := httptest.NewRequest(method, path, nil) + for k, v := range headers { + req.Header.Set(k, v) + } + ctx := &contextHttp.HttpContext{ + Writer: httptest.NewRecorder(), + Request: req, + Ctx: context.Background(), + } + return f.Decode(ctx), ctx +} + +type e2eFilterChain struct { + filters []filter.HttpDecodeFilter +} + +func (c *e2eFilterChain) AppendDecodeFilters(f ...filter.HttpDecodeFilter) { + c.filters = append(c.filters, f...) +} +func (c *e2eFilterChain) AppendEncodeFilters(f ...filter.HttpEncodeFilter) {} +func (c *e2eFilterChain) OnDecode(ctx *contextHttp.HttpContext) {} +func (c *e2eFilterChain) OnEncode(ctx *contextHttp.HttpContext) {} + +// --------------------------------------------------------------------------- +// Scenarios +// --------------------------------------------------------------------------- + +const ( + e2ePolicyID = "pixiu-authz" + e2eDecisionPath = "/v1/data/pixiu/authz/allow" + + // "allow GET only" — covers the headline allow case. + allowGETPolicy = `package pixiu.authz +import future.keywords.if +default allow := false +allow if input.method == "GET" +` + + // "default allow := false" only — every request denied. + denyAllPolicy = `package pixiu.authz +import future.keywords.if +default allow := false +` + + // "allow if header X-Role == admin" — exercises header propagation. + headerRolePolicy = `package pixiu.authz +import future.keywords.if +default allow := false +allow if input.headers["X-Role"][0] == "admin" +` +) + +// 1. Admin PUTs an "allow GET" policy. Gateway GET request is allowed +// end-to-end: filter returns Continue, no local reply written. +func TestE2E_AllowedThroughFullChain(t *testing.T) { + mock := newRegoMockOPA(t) + r := installAdminRouterWithRegoMock(t, mock, adminconfig.OPAConfig{ + PolicyID: e2ePolicyID, + RequestTimeout: 2 * time.Second, + }) + + adminPutPolicy(t, r, e2ePolicyID, allowGETPolicy) Review Comment: 收到 ########## admin/logic/opa_test.go: ########## @@ -0,0 +1,257 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package logic + +import ( + "io" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" +) + +import ( + adminconfig "github.com/apache/dubbo-go-pixiu/admin/config" +) + +// setBootstrap installs a temporary Bootstrap for the test and restores it after. +func setBootstrap(t *testing.T, b *adminconfig.AdminBootstrap) { + t.Helper() + prev := adminconfig.Bootstrap + adminconfig.Bootstrap = b + t.Cleanup(func() { adminconfig.Bootstrap = prev }) +} + +func TestGetOPATimeout_FallsBackToDefaultWhenUnset(t *testing.T) { + setBootstrap(t, nil) + if got := getOPATimeout(); got != adminconfig.DefaultOPAPolicyTimeout { + t.Fatalf("nil Bootstrap: want default %v, got %v", adminconfig.DefaultOPAPolicyTimeout, got) + } + + setBootstrap(t, &adminconfig.AdminBootstrap{}) // RequestTimeout == 0 + if got := getOPATimeout(); got != adminconfig.DefaultOPAPolicyTimeout { + t.Fatalf("zero RequestTimeout: want default %v, got %v", adminconfig.DefaultOPAPolicyTimeout, got) + } +} + +func TestGetOPATimeout_UsesConfigValue(t *testing.T) { + setBootstrap(t, &adminconfig.AdminBootstrap{ + OPA: adminconfig.OPAConfig{RequestTimeout: 3 * time.Second}, + }) + if got := getOPATimeout(); got != 3*time.Second { + t.Fatalf("want 3s from config, got %v", got) + } +} + +func TestBuildOPAPolicyURL(t *testing.T) { + cases := []struct { + name, server, policy, want string + wantErr bool + }{ + {"basic", "http://opa:8181";, "pid", "http://opa:8181/v1/policies/pid";, false}, + {"trims trailing slash", "http://opa:8181/";, "pid", "http://opa:8181/v1/policies/pid";, false}, + {"trims whitespace", " http://opa:8181 ", " pid ", "http://opa:8181/v1/policies/pid";, false}, + {"empty server", "", "pid", "", true}, + {"empty policy", "http://opa:8181";, "", "", true}, + } + for _, c := range cases { + t.Run(c.name, func(t *testing.T) { + got, err := buildOPAPolicyURL(c.server, c.policy) + if c.wantErr { + if err == nil { + t.Fatalf("want error, got url=%q", got) + } + return + } + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if got != c.want { + t.Fatalf("want %q, got %q", c.want, got) + } + }) + } +} + +type recordedRequest struct { + method string + path string + contentType string + authorization string + body string +} + +// startMockOPA spins up an httptest server that records each request and +// responds with `status` and `body` (body may be empty). +func startMockOPA(t *testing.T, status int, body string) (*httptest.Server, *[]recordedRequest, *sync.Mutex) { + t.Helper() + var ( + mu sync.Mutex + recs []recordedRequest + ) + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + b, _ := io.ReadAll(r.Body) + mu.Lock() + recs = append(recs, recordedRequest{ + method: r.Method, + path: r.URL.Path, + contentType: r.Header.Get("Content-Type"), + authorization: r.Header.Get("Authorization"), + body: string(b), + }) + mu.Unlock() + w.WriteHeader(status) + if body != "" { + _, _ = w.Write([]byte(body)) + } + })) + t.Cleanup(srv.Close) + return srv, &recs, &mu +} + +func TestBizPutOPAPolicy_SendsCorrectRequest(t *testing.T) { Review Comment: 收到 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
