-
Notifications
You must be signed in to change notification settings - Fork 9
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Oplogtoredis: Add Configurable Denylist to HTTP Server (#64)
This adds a new feature- a configurable denylist. The denylist is a list of rules. Each rule is a list of keys and a regex. Every incoming oplog message will be checked against every rule. The keys will be used sequentially to index into the oplog message, and if the value found is a string, it will be checked against the regex. If the regex is found for _any_ rule in the denylist, the oplog message will be skipped and will not be reported to Redis. The denylist can be viewed by HTTP methods (GET /denylist and GET /denylist/:ruleID), and can be configured in the same manner (PUT /denylist and DELETE /denylist/:ruleID). Default behavior is unchanged, since the denylist is empty. This is useful when an external context knows that some oplog messages are not necessary to be transferred. The external context can inform Oplogtoredis via the denylist, and then monitor the internal state via the HTTP API.
- Loading branch information
1 parent
137014e
commit 38e75cd
Showing
9 changed files
with
323 additions
and
9 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,88 @@ | ||
| package main | ||
|
|
||
| import ( | ||
| "bytes" | ||
| "encoding/json" | ||
| "io" | ||
| "net/http" | ||
| "os" | ||
| "reflect" | ||
| "testing" | ||
| ) | ||
|
|
||
| func doRequest(method string, path string, t *testing.T, expectedCode int) interface{} { | ||
| req, err := http.NewRequest(method, os.Getenv("OTR_URL")+path, &bytes.Buffer{}) | ||
| if err != nil { | ||
| t.Fatalf("Error creating req: %s", err) | ||
| } | ||
| req.Header.Set("Content-Type", "application/json") | ||
| resp, err := (&http.Client{}).Do(req) | ||
| if err != nil { | ||
| t.Fatalf("Error sending request: %s", err) | ||
| } | ||
|
|
||
| defer resp.Body.Close() | ||
|
|
||
| respBody, err := io.ReadAll(resp.Body) | ||
| if err != nil { | ||
| t.Fatalf("Error eceiving response body: %s", err) | ||
| } | ||
|
|
||
| if resp.StatusCode != expectedCode { | ||
| t.Fatalf("Expected status code %d, but got %d.\nBody was: %s", expectedCode, resp.StatusCode, respBody) | ||
| } | ||
|
|
||
| if expectedCode == 200 { | ||
| var data interface{} | ||
| err = json.Unmarshal(respBody, &data) | ||
| if err != nil { | ||
| t.Fatalf("Error parsing JSON response: %s", err) | ||
| } | ||
|
|
||
| return data | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| // Test the /denylist HTTP operations | ||
| func TestDenyList(t *testing.T) { | ||
| // GET empty list of rules | ||
| data := doRequest("GET", "/denylist", t, 200) | ||
| if !reflect.DeepEqual(data, []interface{}{}) { | ||
| t.Fatalf("Expected empty list from blank GET, but got %#v", data) | ||
| } | ||
| // PUT new rule | ||
| doRequest("PUT", "/denylist/abc", t, 201) | ||
| // GET list with new rule in it | ||
| data = doRequest("GET", "/denylist", t, 200) | ||
| if !reflect.DeepEqual(data, []interface{}{"abc"}) { | ||
| t.Fatalf("Expected singleton from GET, but got %#v", data) | ||
| } | ||
| // GET existing rule | ||
| data = doRequest("GET", "/denylist/abc", t, 200) | ||
| if !reflect.DeepEqual(data, "abc") { | ||
| t.Fatalf("Expected matched body from GET, but got %#v", data) | ||
| } | ||
| // PUT second rule | ||
| doRequest("PUT", "/denylist/def", t, 201) | ||
| // GET second rule | ||
| data = doRequest("GET", "/denylist/def", t, 200) | ||
| if !reflect.DeepEqual(data, "def") { | ||
| t.Fatalf("Expected matched body from GET, but got %#v", data) | ||
| } | ||
| // GET list with both rules | ||
| data = doRequest("GET", "/denylist", t, 200) | ||
| // check both permutations, in case the server reordered them | ||
| if !reflect.DeepEqual(data, []interface{}{"abc", "def"}) && !reflect.DeepEqual(data, []interface{}{"def", "abc"}) { | ||
| t.Fatalf("Expected doubleton from GET, but got %#v", data) | ||
| } | ||
| // DELETE first rule | ||
| doRequest("DELETE", "/denylist/abc", t, 204) | ||
| // GET first rule | ||
| doRequest("GET", "/denylist/abc", t, 404) | ||
| // GET list with only second rule | ||
| data = doRequest("GET", "/denylist", t, 200) | ||
| if !reflect.DeepEqual(data, []interface{}{"def"}) { | ||
| t.Fatalf("Expected singleton from GET, but got %#V", data) | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,72 @@ | ||
| package main | ||
|
|
||
| import ( | ||
| "context" | ||
| "testing" | ||
|
|
||
| "github.com/tulip/oplogtoredis/integration-tests/helpers" | ||
| "go.mongodb.org/mongo-driver/bson" | ||
| ) | ||
|
|
||
| func TestDenyOplog(t *testing.T) { | ||
| harness := startHarness() | ||
| defer harness.stop() | ||
|
|
||
| _, err := harness.mongoClient.Collection("Foo").InsertOne(context.Background(), bson.M{ | ||
| "_id": "id1", | ||
| "f": "1", | ||
| }) | ||
| if err != nil { | ||
| panic(err) | ||
| } | ||
|
|
||
| expectedMessage1 := helpers.OTRMessage{ | ||
| Event: "i", | ||
| Document: map[string]interface{}{ | ||
| "_id": "id1", | ||
| }, | ||
| Fields: []string{"_id", "f"}, | ||
| } | ||
|
|
||
| harness.verify(t, map[string][]helpers.OTRMessage{ | ||
| "tests.Foo": {expectedMessage1}, | ||
| "tests.Foo::id1": {expectedMessage1}, | ||
| }) | ||
|
|
||
| doRequest("PUT", "/denylist/tests", t, 201) | ||
|
|
||
| _, err = harness.mongoClient.Collection("Foo").InsertOne(context.Background(), bson.M{ | ||
| "_id": "id2", | ||
| "g": "2", | ||
| }) | ||
| if err != nil { | ||
| panic(err) | ||
| } | ||
|
|
||
| // second message should not have been received, since it got denied | ||
| harness.verify(t, map[string][]helpers.OTRMessage{}) | ||
|
|
||
| doRequest("DELETE", "/denylist/tests", t, 204) | ||
|
|
||
| _, err = harness.mongoClient.Collection("Foo").InsertOne(context.Background(), bson.M{ | ||
| "_id": "id3", | ||
| "h": "3", | ||
| }) | ||
| if err != nil { | ||
| panic(err) | ||
| } | ||
|
|
||
| expectedMessage3 := helpers.OTRMessage{ | ||
| Event: "i", | ||
| Document: map[string]interface{}{ | ||
| "_id": "id3", | ||
| }, | ||
| Fields: []string{"_id", "h"}, | ||
| } | ||
|
|
||
| // back to normal now that the deny rule is gone | ||
| harness.verify(t, map[string][]helpers.OTRMessage{ | ||
| "tests.Foo": {expectedMessage3}, | ||
| "tests.Foo::id3": {expectedMessage3}, | ||
| }) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,127 @@ | ||
| package denylist | ||
|
|
||
| import ( | ||
| "encoding/json" | ||
| "net/http" | ||
| "strings" | ||
| "sync" | ||
|
|
||
| "github.com/prometheus/client_golang/prometheus" | ||
| "github.com/prometheus/client_golang/prometheus/promauto" | ||
| "github.com/tulip/oplogtoredis/lib/log" | ||
| ) | ||
|
|
||
| var metricFilterEnabled = promauto.NewGaugeVec(prometheus.GaugeOpts{ | ||
| Namespace: "otr", | ||
| Subsystem: "denylist", | ||
| Name: "filter_enabled", | ||
| Help: "Gauge indicating whether the denylist filter is enabled for a particular DB name", | ||
| }, []string{"db"}) | ||
|
|
||
| // CollectionEndpoint serves the endpoints for the whole Denylist at /denylist | ||
| func CollectionEndpoint(denylist *sync.Map) func(http.ResponseWriter, *http.Request) { | ||
| return func(response http.ResponseWriter, request *http.Request) { | ||
| switch request.Method { | ||
| case "GET": | ||
| listDenylistKeys(response, denylist) | ||
| default: | ||
| http.Error(response, http.StatusText(http.StatusNotFound), http.StatusNotFound) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // SingleEndpoint serves the endpoints for particular Denylist entries at /denylist/... | ||
| func SingleEndpoint(denylist *sync.Map) func(http.ResponseWriter, *http.Request) { | ||
| return func(response http.ResponseWriter, request *http.Request) { | ||
| switch request.Method { | ||
| case "GET": | ||
| getDenylistEntry(response, request, denylist) | ||
| case "PUT": | ||
| createDenylistEntry(response, request, denylist) | ||
| case "DELETE": | ||
| deleteDenylistEntry(response, request, denylist) | ||
| default: | ||
| http.Error(response, http.StatusText(http.StatusNotFound), http.StatusNotFound) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // GET /denylist | ||
| func listDenylistKeys(response http.ResponseWriter, denylist *sync.Map) { | ||
| keys := []interface{}{} | ||
|
|
||
| denylist.Range(func(key interface{}, value interface{}) bool { | ||
| keys = append(keys, key) | ||
| return true | ||
| }) | ||
|
|
||
| response.Header().Set("Content-Type", "application/json") | ||
| response.WriteHeader(http.StatusOK) | ||
| err := json.NewEncoder(response).Encode(keys) | ||
| if err != nil { | ||
| http.Error(response, "couldn't encode result", http.StatusInternalServerError) | ||
| return | ||
| } | ||
| } | ||
|
|
||
| // GET /denylist/... | ||
| func getDenylistEntry(response http.ResponseWriter, request *http.Request, denylist *sync.Map) { | ||
| id := request.URL.Path | ||
| if strings.Contains(id, "/") { | ||
| http.Error(response, http.StatusText(http.StatusNotFound), http.StatusNotFound) | ||
| return | ||
| } | ||
| _, exists := denylist.Load(id) | ||
| if !exists { | ||
| http.Error(response, "denylist entry not found with that id", http.StatusNotFound) | ||
| return | ||
| } | ||
|
|
||
| response.Header().Set("Content-Type", "application/json") | ||
| response.WriteHeader(http.StatusOK) | ||
| err := json.NewEncoder(response).Encode(id) | ||
| if err != nil { | ||
| http.Error(response, "couldn't encode result", http.StatusInternalServerError) | ||
| return | ||
| } | ||
| } | ||
|
|
||
| // PUT /denylist/... | ||
| func createDenylistEntry(response http.ResponseWriter, request *http.Request, denylist *sync.Map) { | ||
| id := request.URL.Path | ||
| if strings.Contains(id, "/") { | ||
| http.Error(response, http.StatusText(http.StatusNotFound), http.StatusNotFound) | ||
| return | ||
| } | ||
| _, exists := denylist.Load(id) | ||
| if exists { | ||
| response.WriteHeader(http.StatusNoContent) | ||
| return | ||
| } | ||
|
|
||
| denylist.Store(id, true) | ||
| log.Log.Infow("Created denylist entry", "id", id) | ||
| metricFilterEnabled.WithLabelValues(id).Set(1) | ||
|
|
||
| response.WriteHeader(http.StatusCreated) | ||
| } | ||
|
|
||
| // DELETE /denylist/... | ||
| func deleteDenylistEntry(response http.ResponseWriter, request *http.Request, denylist *sync.Map) { | ||
| id := request.URL.Path | ||
| if strings.Contains(id, "/") { | ||
| http.Error(response, http.StatusText(http.StatusNotFound), http.StatusNotFound) | ||
| return | ||
| } | ||
| _, exists := denylist.Load(id) | ||
| if !exists { | ||
| http.Error(response, "denylist entry not found with that id", http.StatusNotFound) | ||
| return | ||
| } | ||
|
|
||
| denylist.Delete(id) | ||
| log.Log.Infow("Deleted denylist entry", "id", id) | ||
| metricFilterEnabled.WithLabelValues(id).Set(0) | ||
|
|
||
| response.WriteHeader(http.StatusNoContent) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.