commit 78a0c832d7613975b4d5f443765e4178f2a3abdf Author: Yaser Date: Thu Aug 13 20:20:47 2026 +0330 Initial commit diff --git a/.drone.yml b/.drone.yml new file mode 100644 index 0000000..a075c92 --- /dev/null +++ b/.drone.yml @@ -0,0 +1,87 @@ +--- +kind: pipeline +name: default + +steps: + - name: Go test + image: golang:1.18 + environment: + GO111MODULE: on + commands: + - go mod download + - go test ./... --cover --count=100 + + - name: "Bump & Tag" + image: quay.io/openware/sdk-citools:2.6.7 + environment: + BRANCH_NAME: ${DRONE_BRANCH} + REPO_NAME: openware/rango + GITHUB_API_KEY: + from_secret: kite_bot_key + commands: + - BUNDLE_GEMFILE=/sdk/Gemfile bundle exec rake --rakefile=/sdk/Rakefile ci:prebuild + when: + branch: + - master + - "*-stable" + + - name: Build container + image: plugins/docker + settings: + username: + from_secret: quay_username + password: + from_secret: quay_password + repo: quay.io/openware/rango + registry: quay.io + when: + branch: + - master + - "*-stable" + + - name: "Update global OpenDAX version" + image: quay.io/openware/goci:0.0.11 + environment: + GIT_TOKEN: + from_secret: gh_api_token + GIT_USER: + from_secret: gh_api_user + commands: + - export GIT_URL=https://${GIT_USER}:${GIT_TOKEN}@github.com/openware/versions-aurora.git + - /app/goci versions + when: + branch: + - "*-stable" + - master + + - name: "Notify" + image: quay.io/openware/sdk-citools:2.6.7 + environment: + BOT_USERNAME: kite-bot + BOT_NAME: Kite Bot + BOT_EMAIL: kite-bot@heliostech.fr + BRANCH_NAME: ${DRONE_BRANCH} + REPO_NAME: ${DRONE_REPO} + SDK_BRANCH: ${DRONE_BRANCH} + SLACK_CHANNEL: peatio + SLACK_TOKEN: + from_secret: slack_token + GITHUB_API_KEY: + from_secret: kite_bot_key + TELEGRAM_BOT_TOKEN: + from_secret: telegram_bot_token + TELEGRAM_CHAT_ID: + from_secret: telegram_chat_id + commands: + - BUNDLE_GEMFILE=/sdk/Gemfile bundle exec rake --rakefile=/sdk/Rakefile ci:postbuild[/drone/src] + when: + branch: + - master + - "*-stable" + +trigger: + event: + - push + +image_pull_secrets: + - dockerconfigjson diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..889bc3d --- /dev/null +++ b/.gitignore @@ -0,0 +1 @@ +config/rsa* diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..1e3718d --- /dev/null +++ b/Dockerfile @@ -0,0 +1,30 @@ +FROM golang:1.20-alpine AS builder + +RUN apk update && apk add curl gcc g++ libc-dev + +ARG KAIGARA_VERSION=v1.0.34 +# Install Kaigara +RUN curl -Lso /usr/bin/kaigara https://github.com/openware/kaigara/releases/download/${KAIGARA_VERSION}/kaigara \ + && chmod +x /usr/bin/kaigara + +WORKDIR /build +ENV CGO_ENABLED=1 \ + GOOS=linux \ + GOARCH=amd64 + +COPY go.mod go.sum ./ +RUN go mod download + +COPY . . +RUN go build ./cmd/rango + + +FROM alpine:3.18 + +RUN apk add ca-certificates +WORKDIR app +COPY --from=builder /build/rango ./ +COPY --from=builder /usr/bin/kaigara /usr/bin/kaigara +RUN mkdir -p /app/config + +CMD ["./rango"] diff --git a/LICENSE.md b/LICENSE.md new file mode 100644 index 0000000..e72929e --- /dev/null +++ b/LICENSE.md @@ -0,0 +1,202 @@ + Apache License + Version 2.0, January 2004 + http://www.apache.org/licenses/ + + TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION + + 1. Definitions. + + "License" shall mean the terms and conditions for use, reproduction, + and distribution as defined by Sections 1 through 9 of this document. + + "Licensor" shall mean the copyright owner or entity authorized by + the copyright owner that is granting the License. + + "Legal Entity" shall mean the union of the acting entity and all + other entities that control, are controlled by, or are under common + control with that entity. For the purposes of this definition, + "control" means (i) the power, direct or indirect, to cause the + direction or management of such entity, whether by contract or + otherwise, or (ii) ownership of fifty percent (50%) or more of the + outstanding shares, or (iii) beneficial ownership of such entity. + + "You" (or "Your") shall mean an individual or Legal Entity + exercising permissions granted by this License. + + "Source" form shall mean the preferred form for making modifications, + including but not limited to software source code, documentation + source, and configuration files. + + "Object" form shall mean any form resulting from mechanical + transformation or translation of a Source form, including but + not limited to compiled object code, generated documentation, + and conversions to other media types. + + "Work" shall mean the work of authorship, whether in Source or + Object form, made available under the License, as indicated by a + copyright notice that is included in or attached to the work + (an example is provided in the Appendix below). + + "Derivative Works" shall mean any work, whether in Source or Object + form, that is based on (or derived from) the Work and for which the + editorial revisions, annotations, elaborations, or other modifications + represent, as a whole, an original work of authorship. For the purposes + of this License, Derivative Works shall not include works that remain + separable from, or merely link (or bind by name) to the interfaces of, + the Work and Derivative Works thereof. + + "Contribution" shall mean any work of authorship, including + the original version of the Work and any modifications or additions + to that Work or Derivative Works thereof, that is intentionally + submitted to Licensor for inclusion in the Work by the copyright owner + or by an individual or Legal Entity authorized to submit on behalf of + the copyright owner. For the purposes of this definition, "submitted" + means any form of electronic, verbal, or written communication sent + to the Licensor or its representatives, including but not limited to + communication on electronic mailing lists, source code control systems, + and issue tracking systems that are managed by, or on behalf of, the + Licensor for the purpose of discussing and improving the Work, but + excluding communication that is conspicuously marked or otherwise + designated in writing by the copyright owner as "Not a Contribution." + + "Contributor" shall mean Licensor and any individual or Legal Entity + on behalf of whom a Contribution has been received by Licensor and + subsequently incorporated within the Work. + + 2. Grant of Copyright License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + copyright license to reproduce, prepare Derivative Works of, + publicly display, publicly perform, sublicense, and distribute the + Work and such Derivative Works in Source or Object form. + + 3. Grant of Patent License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + (except as stated in this section) patent license to make, have made, + use, offer to sell, sell, import, and otherwise transfer the Work, + where such license applies only to those patent claims licensable + by such Contributor that are necessarily infringed by their + Contribution(s) alone or by combination of their Contribution(s) + with the Work to which such Contribution(s) was submitted. If You + institute patent litigation against any entity (including a + cross-claim or counterclaim in a lawsuit) alleging that the Work + or a Contribution incorporated within the Work constitutes direct + or contributory patent infringement, then any patent licenses + granted to You under this License for that Work shall terminate + as of the date such litigation is filed. + + 4. Redistribution. You may reproduce and distribute copies of the + Work or Derivative Works thereof in any medium, with or without + modifications, and in Source or Object form, provided that You + meet the following conditions: + + (a) You must give any other recipients of the Work or + Derivative Works a copy of this License; and + + (b) You must cause any modified files to carry prominent notices + stating that You changed the files; and + + (c) You must retain, in the Source form of any Derivative Works + that You distribute, all copyright, patent, trademark, and + attribution notices from the Source form of the Work, + excluding those notices that do not pertain to any part of + the Derivative Works; and + + (d) If the Work includes a "NOTICE" text file as part of its + distribution, then any Derivative Works that You distribute must + include a readable copy of the attribution notices contained + within such NOTICE file, excluding those notices that do not + pertain to any part of the Derivative Works, in at least one + of the following places: within a NOTICE text file distributed + as part of the Derivative Works; within the Source form or + documentation, if provided along with the Derivative Works; or, + within a display generated by the Derivative Works, if and + wherever such third-party notices normally appear. The contents + of the NOTICE file are for informational purposes only and + do not modify the License. You may add Your own attribution + notices within Derivative Works that You distribute, alongside + or as an addendum to the NOTICE text from the Work, provided + that such additional attribution notices cannot be construed + as modifying the License. + + You may add Your own copyright statement to Your modifications and + may provide additional or different license terms and conditions + for use, reproduction, or distribution of Your modifications, or + for any such Derivative Works as a whole, provided Your use, + reproduction, and distribution of the Work otherwise complies with + the conditions stated in this License. + + 5. Submission of Contributions. Unless You explicitly state otherwise, + any Contribution intentionally submitted for inclusion in the Work + by You to the Licensor shall be under the terms and conditions of + this License, without any additional terms or conditions. + Notwithstanding the above, nothing herein shall supersede or modify + the terms of any separate license agreement you may have executed + with Licensor regarding such Contributions. + + 6. Trademarks. This License does not grant permission to use the trade + names, trademarks, service marks, or product names of the Licensor, + except as required for reasonable and customary use in describing the + origin of the Work and reproducing the content of the NOTICE file. + + 7. Disclaimer of Warranty. Unless required by applicable law or + agreed to in writing, Licensor provides the Work (and each + Contributor provides its Contributions) on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + implied, including, without limitation, any warranties or conditions + of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A + PARTICULAR PURPOSE. You are solely responsible for determining the + appropriateness of using or redistributing the Work and assume any + risks associated with Your exercise of permissions under this License. + + 8. Limitation of Liability. In no event and under no legal theory, + whether in tort (including negligence), contract, or otherwise, + unless required by applicable law (such as deliberate and grossly + negligent acts) or agreed to in writing, shall any Contributor be + liable to You for damages, including any direct, indirect, special, + incidental, or consequential damages of any character arising as a + result of this License or out of the use or inability to use the + Work (including but not limited to damages for loss of goodwill, + work stoppage, computer failure or malfunction, or any and all + other commercial damages or losses), even if such Contributor + has been advised of the possibility of such damages. + + 9. Accepting Warranty or Additional Liability. While redistributing + the Work or Derivative Works thereof, You may choose to offer, + and charge a fee for, acceptance of support, warranty, indemnity, + or other liability obligations and/or rights consistent with this + License. However, in accepting such obligations, You may act only + on Your own behalf and on Your sole responsibility, not on behalf + of any other Contributor, and only if You agree to indemnify, + defend, and hold each Contributor harmless for any liability + incurred by, or claims asserted against, such Contributor by reason + of your accepting any such warranty or additional liability. + + END OF TERMS AND CONDITIONS + + APPENDIX: How to apply the Apache License to your work. + + To apply the Apache License to your work, attach the following + boilerplate notice, with the fields enclosed by brackets "[]" + replaced with your own identifying information. (Don't include + the brackets!) The text should be enclosed in the appropriate + comment syntax for the file format. We also recommend that a + file or class name and description of purpose be included on the + same "printed page" as the copyright notice for easier + identification within third-party archives. + + Copyright [yyyy] [name of copyright owner] + + Licensed 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. + \ No newline at end of file diff --git a/README.md b/README.md new file mode 100644 index 0000000..aea20c2 --- /dev/null +++ b/README.md @@ -0,0 +1,72 @@ +# Fibitex Rango + +Fork of [OpenWare Rango](https://github.com/openware/rango) based on upstream release **3.1.3**, patched for the Fibitex / OpenDAX 2.6 stack (Barong RS256 JWT + incremental order book on market switch). + +## Fibitex documentation + +| Doc | Topic | +|-----|--------| +| [`Docs/Fibitex-staging-deploy.md`](../Docs/Fibitex-staging-deploy.md) | Staging deploy | +| [`Alvand-P/README.md`](../Alvand-P/README.md) | Compose & image `custom/rango:1` | +| [`Docs/README.md`](../Docs/README.md) | Central docs | +| [`Docs/06-setup-local.md`](../Docs/06-setup-local.md) | Local setup | + +Upstream repository: https://github.com/openware/rango +Base version: **3.1.3** (Fibitex build: see `VERSION`) + +## Fibitex patches + +1. **RS256 JWT (Barong 2.6)** — upstream 3.1.2+ accepts EdDSA only; this fork validates Barong `JWT_PUBLIC_KEY` (RSA) via `pkg/auth/validator.go`. +2. **Order book on market switch** — replay `ob-snap` and buffered `ob-inc` before adding the client to the topic, so live increments cannot arrive before the snapshot (`pkg/routing/hub.go`). + +## Docker image + +Alvand-P uses: `custom/rango:1` + +```powershell +cd Rango +docker build -t custom/rango:1 . +``` + +## Alvand-P configuration + +In `Alvand-P/config/app.local.yml`: + +```yaml +images: + rango: custom/rango:1 +``` + +Restart: + +```powershell +cd Alvand-P +docker compose up -d rango +``` + +## Environment + +Same as upstream Rango — see upstream README sections for RabbitMQ, scopes, and WebSocket paths (`/public`, `/private`). + +| Variable | Fibitex note | +| -------- | ------------ | +| `JWT_PUBLIC_KEY` | Base64 PEM of Barong RSA public key (from `rango.env`) | +| `LOG_LEVEL` | Use `info` temporarily to see `JWT validator mode: rsa` | + +## Verify + +After login on `www.app.local`: + +- Order cancel / create updates in UI (private `order` stream) +- Switching markets loads order book without refresh +- No repeated `OrderBook increment received before snapshot` in browser console + +--- + +## Upstream Rango (reference) + +![Cryptocurrency Exchange Platform - OpenDAX](https://github.com/openware/meta/raw/main/images/github_opendax.png) + +Rango is a general purpose websocket server which dispatches public and private messages from RabbitMQ. It is designed as a drop-in replacement of Ranger built in Ruby. + +For full upstream documentation (metrics, scopes, RBAC, message format), see the [openware/rango README](https://github.com/openware/rango/blob/master/README.md). diff --git a/VERSION b/VERSION new file mode 100644 index 0000000..62bd6f6 --- /dev/null +++ b/VERSION @@ -0,0 +1 @@ +3.1.3-fibitex.2 diff --git a/cmd/rango/rango.go b/cmd/rango/rango.go new file mode 100644 index 0000000..ff857e2 --- /dev/null +++ b/cmd/rango/rango.go @@ -0,0 +1,208 @@ +package main + +import ( + "flag" + "fmt" + "net/http" + "os" + "strings" + "time" + + "math/rand" + + "github.com/prometheus/client_golang/prometheus/promhttp" + "github.com/rs/zerolog" + "github.com/rs/zerolog/log" + + "github.com/openware/rango/pkg/amqp" + "github.com/openware/rango/pkg/auth" + "github.com/openware/rango/pkg/metrics" + "github.com/openware/rango/pkg/routing" +) + +var ( + wsAddr = flag.String("ws-addr", "", "http service address") + amqpAddr = flag.String("amqp-addr", "", "AMQP server address") + pubKey = flag.String("pubKey", "config/ed25519-key.pub", "Path to public key") + exName = flag.String("exchange", "peatio.events.ranger", "Exchange name of upstream messages") +) + +const prefix = "Bearer " + +type httpHanlder func(w http.ResponseWriter, r *http.Request) + +func token(r *http.Request) string { + authHeader := r.Header.Get("Authorization") + if !strings.HasPrefix(string(authHeader), prefix) { + return "" + } + + return authHeader[len(prefix):] +} + +func authHandler(h httpHanlder, validator auth.Validator, mustAuth bool) httpHanlder { + return func(w http.ResponseWriter, r *http.Request) { + claims, err := validator.ParseAndValidate(token(r)) + + if err != nil && mustAuth { + w.WriteHeader(http.StatusUnauthorized) + return + } + + if err == nil { + r.Header.Set("JwtUID", claims.UID) + r.Header.Set("JwtRole", claims.Role) + } else { + r.Header.Del("JwtUID") + r.Header.Del("JwtRole") + } + h(w, r) + } +} + +func setupLogger() { + logLevel, ok := os.LookupEnv("LOG_LEVEL") + if ok { + level, err := zerolog.ParseLevel(strings.ToLower(logLevel)) + if err != nil { + panic(err) + } + + zerolog.SetGlobalLevel(level) + return + } + + zerolog.SetGlobalLevel(zerolog.DebugLevel) +} + +func getEnv(name, value string) string { + v := os.Getenv(name) + if v == "" { + return value + } + return v +} + +func getAMQPConnectionURL() string { + if *amqpAddr != "" { + return *amqpAddr + } + + user := getEnv("RABBITMQ_USER", "guest") + pass := getEnv("RABBITMQ_PASSWORD", "guest") + host := getEnv("RABBITMQ_HOST", "localhost") + port := getEnv("RABBITMQ_PORT", "5672") + + return fmt.Sprintf("amqp://%s:%s@%s:%s", user, pass, host, port) +} + +func getServerAddress() string { + if *wsAddr != "" { + return *wsAddr + } + host := getEnv("RANGER_HOST", "0.0.0.0") + port := getEnv("RANGER_PORT", "8080") + return fmt.Sprintf("%s:%s", host, port) +} + +func getRBACConfig() map[string][]string { + envs := os.Environ() + + rbacEnv := filterPrefixed("RANGO_RBAC_", envs) + + return envToMatrix(rbacEnv, "RANGO_RBAC_") +} + +func envToMatrix(env []string, trimPrefix string) map[string][]string { + matr := make(map[string][]string) + + for _, rec := range env { + kv := strings.Split(rec, "=") + key := strings.ToLower(strings.TrimPrefix(kv[0], trimPrefix)) + value := strings.Split(kv[1], ",") + + matr[key] = value + } + + return matr +} + +func filterPrefixed(prefix string, arr []string) []string { + var res []string + + for _, rec := range arr { + if strings.HasPrefix(rec, prefix) { + res = append(res, rec) + } + } + + return res +} + +func main() { + flag.Parse() + + setupLogger() + + metrics.Enable() + + rbac := getRBACConfig() + hub := routing.NewHub(rbac) + validator, err := auth.LoadValidator(*pubKey) + if err != nil { + log.Error().Msgf("Loading public key failed: %s", err.Error()) + time.Sleep(2 * time.Second) + return + } + log.Info().Msgf("JWT validator mode: %s", validator.Mode) + + rand.Seed(time.Now().UnixNano()) + globalQName := fmt.Sprintf("rango.instance.%d", rand.Int()) + privateQName := fmt.Sprintf("rango.instance.private-%d", rand.Int()) + + // Establish AMQP session for all non private events + globalMq, err := amqp.NewAMQPSession(getAMQPConnectionURL()) + if err != nil { + log.Fatal().Msgf("creating new AMQP session failed: %s", err.Error()) + return + } + err = globalMq.Stream(*exName, globalQName, "#", hub.SkipPrivateMsg) + defer globalMq.Close(globalQName) + + if err != nil { + log.Fatal().Msgf("AMQP init failed: %s", err.Error()) + return + } + + // Establish AMQP session for private events + privateMq, err := amqp.NewAMQPSession(getAMQPConnectionURL()) + if err != nil { + log.Fatal().Msgf("creating new AMQP session failed: %s", err.Error()) + return + } + err = privateMq.Stream(*exName, privateQName, "private.#", hub.ReceiveMsg) + defer privateMq.Close(privateQName) + + if err != nil { + log.Fatal().Msgf("AMQP init failed: %s", err.Error()) + return + } + + go hub.ListenWebsocketEvents() + + wsHandler := func(w http.ResponseWriter, r *http.Request) { + routing.NewClient(hub, w, r) + } + + http.HandleFunc("/private", authHandler(wsHandler, validator, true)) + http.HandleFunc("/public", authHandler(wsHandler, validator, false)) + http.HandleFunc("/", authHandler(wsHandler, validator, false)) + + go http.ListenAndServe(":4242", promhttp.Handler()) + + log.Printf("Listenning on %s", getServerAddress()) + err = http.ListenAndServe(getServerAddress(), nil) + if err != nil { + log.Fatal().Msg("ListenAndServe failed: " + err.Error()) + } +} diff --git a/cmd/rango/rango_test.go b/cmd/rango/rango_test.go new file mode 100644 index 0000000..b692bc2 --- /dev/null +++ b/cmd/rango/rango_test.go @@ -0,0 +1,22 @@ +package main + +import ( + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestRango_envToMatrix(t *testing.T) { + env := []string{ + "PREFIXED_HELLO=world", + "PREFIXED_ONE=TWO,three,four", + "PREFIXED_FOO=bar", + } + + matrix := envToMatrix(env, "PREFIXED_") + + assert.Len(t, matrix, 3) + assert.Equal(t, "world", matrix["hello"][0]) + assert.Equal(t, []string{"TWO", "three", "four"}, matrix["one"]) + assert.Equal(t, "bar", matrix["foo"][0]) +} diff --git a/config/.keep b/config/.keep new file mode 100644 index 0000000..e69de29 diff --git a/docs/troubleshoot.md b/docs/troubleshoot.md new file mode 100644 index 0000000..c5cbca5 --- /dev/null +++ b/docs/troubleshoot.md @@ -0,0 +1,17 @@ +# Troubleshooting docs + +## "message":"Websocket upgrade failed: websocket: request origin not allowed by Upgrader.CheckOrigin" + +If you deploy your Rango somewhere, lets say to _www.rango.host_, and try to connect with some local +frontend app you develop, you'll probably face this problem. + +It is caused by [CORS](https://developer.mozilla.org/en-US/docs/Web/HTTP/CORS), to resolve this issue +specify a list of domains, allowed to connect to server in Rango `API_CORS_ORIGINS` env var. + +By default, if this env is not set, only requests from _www.rango.host_ are enabled. + +So, if I have some react-app running on _localhost:3000_: + +``` +API_CORS_ORIGINS=www.rango.host,localhost:3000 +``` diff --git a/go.mod b/go.mod new file mode 100644 index 0000000..d9f6006 --- /dev/null +++ b/go.mod @@ -0,0 +1,29 @@ +module github.com/openware/rango + +go 1.18 + +require ( + github.com/gorilla/websocket v1.4.2 + github.com/openware/pkg v0.1.6 + github.com/prometheus/client_golang v1.11.1 + github.com/rs/zerolog v1.18.0 + github.com/streadway/amqp v0.0.0-20200108173154-1c71cc93ed71 + github.com/stretchr/testify v1.8.1 +) + +require ( + github.com/beorn7/perks v1.0.1 // indirect + github.com/cespare/xxhash/v2 v2.1.1 // indirect + github.com/davecgh/go-spew v1.1.1 // indirect + github.com/golang-jwt/jwt v3.2.2+incompatible // indirect + github.com/golang/protobuf v1.5.2 // indirect + github.com/matttproud/golang_protobuf_extensions v1.0.1 // indirect + github.com/pmezard/go-difflib v1.0.0 // indirect + github.com/prometheus/client_model v0.2.0 // indirect + github.com/prometheus/common v0.26.0 // indirect + github.com/prometheus/procfs v0.6.0 // indirect + github.com/stretchr/objx v0.5.0 // indirect + golang.org/x/sys v0.6.0 // indirect + google.golang.org/protobuf v1.28.0 // indirect + gopkg.in/yaml.v3 v3.0.1 // indirect +) diff --git a/go.sum b/go.sum new file mode 100644 index 0000000..8d53f62 --- /dev/null +++ b/go.sum @@ -0,0 +1,175 @@ +cloud.google.com/go v0.34.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw= +github.com/alecthomas/template v0.0.0-20160405071501-a0175ee3bccc/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc= +github.com/alecthomas/template v0.0.0-20190718012654-fb15b899a751/go.mod h1:LOuyumcjzFXgccqObfd/Ljyb9UuFJ6TxHnclSeseNhc= +github.com/alecthomas/units v0.0.0-20151022065526-2efee857e7cf/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0= +github.com/alecthomas/units v0.0.0-20190717042225-c3de453c63f4/go.mod h1:ybxpYRFXyAe+OPACYpWeL0wqObRcbAqCMya13uyzqw0= +github.com/alecthomas/units v0.0.0-20190924025748-f65c72e2690d/go.mod h1:rBZYJk541a8SKzHPHnH3zbiI+7dagKZ0cgpgrD7Fyho= +github.com/beorn7/perks v0.0.0-20180321164747-3a771d992973/go.mod h1:Dwedo/Wpr24TaqPxmxbtue+5NUziq4I4S80YR8gNf3Q= +github.com/beorn7/perks v1.0.0/go.mod h1:KWe93zE9D1o94FZ5RNwFwVgaQK1VOXiVxmqh+CedLV8= +github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM= +github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw= +github.com/cespare/xxhash/v2 v2.1.1 h1:6MnRN8NT7+YBpUIWxHtefFZOKTAPgGjpQSxqLNn0+qY= +github.com/cespare/xxhash/v2 v2.1.1/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= +github.com/coreos/go-systemd v0.0.0-20190321100706-95778dfbb74e/go.mod h1:F5haX7vjVVG0kc13fIWeqUViNPyEJxv/OmvnBo0Yme4= +github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/go-kit/kit v0.8.0/go.mod h1:xBxKIO96dXMWWy0MnWVtmwkA9/13aqxPnvrjFYMA2as= +github.com/go-kit/kit v0.9.0/go.mod h1:xBxKIO96dXMWWy0MnWVtmwkA9/13aqxPnvrjFYMA2as= +github.com/go-kit/log v0.1.0/go.mod h1:zbhenjAZHb184qTLMA9ZjW7ThYL0H2mk7Q6pNt4vbaY= +github.com/go-logfmt/logfmt v0.3.0/go.mod h1:Qt1PoO58o5twSAckw1HlFXLmHsOX5/0LbT9GBnD5lWE= +github.com/go-logfmt/logfmt v0.4.0/go.mod h1:3RMwSq7FuexP4Kalkev3ejPJsZTpXXBr9+V4qmtdjCk= +github.com/go-logfmt/logfmt v0.5.0/go.mod h1:wCYkCAKZfumFQihp8CzCvQ3paCTfi41vtzG1KdI/P7A= +github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY= +github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ= +github.com/golang-jwt/jwt v3.2.2+incompatible h1:IfV12K8xAKAnZqdXVzCZ+TOjboZ2keLg81eXfW3O+oY= +github.com/golang-jwt/jwt v3.2.2+incompatible/go.mod h1:8pz2t5EyA70fFQQSrl6XZXzqecmYZeUEB8OUGHkxJ+I= +github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U= +github.com/golang/protobuf v1.3.1/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U= +github.com/golang/protobuf v1.3.2/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U= +github.com/golang/protobuf v1.4.0-rc.1/go.mod h1:ceaxUfeHdC40wWswd/P6IGgMaK3YpKi5j83Wpe3EHw8= +github.com/golang/protobuf v1.4.0-rc.1.0.20200221234624-67d41d38c208/go.mod h1:xKAWHe0F5eneWXFV3EuXVDTCmh+JuBKY0li0aMyXATA= +github.com/golang/protobuf v1.4.0-rc.2/go.mod h1:LlEzMj4AhA7rCAGe4KMBDvJI+AwstrUpVNzEA03Pprs= +github.com/golang/protobuf v1.4.0-rc.4.0.20200313231945-b860323f09d0/go.mod h1:WU3c8KckQ9AFe+yFwt9sWVRKCVIyN9cPHBJSNnbL67w= +github.com/golang/protobuf v1.4.0/go.mod h1:jodUvKwWbYaEsadDk5Fwe5c77LiNKVO9IDvqG2KuDX0= +github.com/golang/protobuf v1.4.2/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw735rRwI= +github.com/golang/protobuf v1.4.3/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw735rRwI= +github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk= +github.com/golang/protobuf v1.5.2 h1:ROPKBNFfQgOUMifHyP+KYbvpjbdoFNs+aK7DXlji0Tw= +github.com/golang/protobuf v1.5.2/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY= +github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU= +github.com/google/go-cmp v0.3.1/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU= +github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= +github.com/google/go-cmp v0.5.4/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= +github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= +github.com/google/go-cmp v0.5.6 h1:BKbKCqvP6I+rmFHt06ZmyQtvB8xAkWdhFyr0ZUNZcxQ= +github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg= +github.com/gorilla/websocket v1.4.2 h1:+/TMaTYc4QFitKJxsQ7Yye35DkWvkdLcvGKqM+x0Ufc= +github.com/gorilla/websocket v1.4.2/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE= +github.com/jpillora/backoff v1.0.0/go.mod h1:J/6gKK9jxlEcS3zixgDgUAsiuZ7yrSoa/FX5e0EB2j4= +github.com/json-iterator/go v1.1.6/go.mod h1:+SdeFBvtyEkXs7REEP0seUULqWtbJapLOCVDaaPEHmU= +github.com/json-iterator/go v1.1.10/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4= +github.com/json-iterator/go v1.1.11/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4= +github.com/julienschmidt/httprouter v1.2.0/go.mod h1:SYymIcj16QtmaHHD7aYtjjsJG7VTCxuUUipMqKk8s4w= +github.com/julienschmidt/httprouter v1.3.0/go.mod h1:JR6WtHb+2LUe8TCKY3cZOxFyyO8IZAc4RVcycCCAKdM= +github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ= +github.com/konsorten/go-windows-terminal-sequences v1.0.3/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ= +github.com/kr/logfmt v0.0.0-20140226030751-b84e30acd515/go.mod h1:+0opPa2QZZtGFBFZlji/RkVcI2GknAs/DXo4wKdlNEc= +github.com/kr/pretty v0.1.0 h1:L/CwN0zerZDmRFUapSPitk6f+Q3+0za1rQkzVuMiMFI= +github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo= +github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ= +github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE= +github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= +github.com/matttproud/golang_protobuf_extensions v1.0.1 h1:4hp9jkHxhMHkqkrB3Ix0jegS5sx/RkqARlsWZ6pIwiU= +github.com/matttproud/golang_protobuf_extensions v1.0.1/go.mod h1:D8He9yQNgCq6Z5Ld7szi9bcBfOoFv/3dc6xSMkL2PC0= +github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= +github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= +github.com/modern-go/reflect2 v0.0.0-20180701023420-4b7aa43c6742/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0= +github.com/modern-go/reflect2 v1.0.1/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0= +github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U= +github.com/mwitkow/go-conntrack v0.0.0-20190716064945-2f068394615f/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U= +github.com/openware/pkg v0.1.6 h1:gBwG7ugO6f09W8MQn67cZBF6NuKBf540KuCQuVJrKj4= +github.com/openware/pkg v0.1.6/go.mod h1:MfXQPnKaCG0JxOST4b4tHUOtff/pEai9dMd7teUqs9g= +github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= +github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= +github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/prometheus/client_golang v0.9.1/go.mod h1:7SWBe2y4D6OKWSNQJUaRYU/AaXPKyh/dDVn+NZz0KFw= +github.com/prometheus/client_golang v1.0.0/go.mod h1:db9x61etRT2tGnBNRi70OPL5FsnadC4Ky3P0J6CfImo= +github.com/prometheus/client_golang v1.7.1/go.mod h1:PY5Wy2awLA44sXw4AOSfFBetzPP4j5+D6mVACh+pe2M= +github.com/prometheus/client_golang v1.11.1 h1:+4eQaD7vAZ6DsfsxB15hbE0odUjGI5ARs9yskGu1v4s= +github.com/prometheus/client_golang v1.11.1/go.mod h1:Z6t4BnS23TR94PD6BsDNk8yVqroYurpAkEiz0P2BEV0= +github.com/prometheus/client_model v0.0.0-20180712105110-5c3871d89910/go.mod h1:MbSGuTsp3dbXC40dX6PRTWyKYBIrTGTE9sqQNg2J8bo= +github.com/prometheus/client_model v0.0.0-20190129233127-fd36f4220a90/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA= +github.com/prometheus/client_model v0.2.0 h1:uq5h0d+GuxiXLJLNABMgp2qUWDPiLvgCzz2dUR+/W/M= +github.com/prometheus/client_model v0.2.0/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA= +github.com/prometheus/common v0.4.1/go.mod h1:TNfzLD0ON7rHzMJeJkieUDPYmFC7Snx/y86RQel1bk4= +github.com/prometheus/common v0.10.0/go.mod h1:Tlit/dnDKsSWFlCLTWaA1cyBgKHSMdTB80sz/V91rCo= +github.com/prometheus/common v0.26.0 h1:iMAkS2TDoNWnKM+Kopnx/8tnEStIfpYA0ur0xQzzhMQ= +github.com/prometheus/common v0.26.0/go.mod h1:M7rCNAaPfAosfx8veZJCuw84e35h3Cfd9VFqTh1DIvc= +github.com/prometheus/procfs v0.0.0-20181005140218-185b4288413d/go.mod h1:c3At6R/oaqEKCNdg8wHV1ftS6bRYblBhIjjI8uT2IGk= +github.com/prometheus/procfs v0.0.2/go.mod h1:TjEm7ze935MbeOT/UhFTIMYKhuLP4wbCsTZCD3I8kEA= +github.com/prometheus/procfs v0.1.3/go.mod h1:lV6e/gmhEcM9IjHGsFOCxxuZ+z1YqCvr4OA4YeYWdaU= +github.com/prometheus/procfs v0.6.0 h1:mxy4L2jP6qMonqmq+aTtOx1ifVWUgG/TAmntgbh3xv4= +github.com/prometheus/procfs v0.6.0/go.mod h1:cz+aTbrPOrUb4q7XlbU9ygM+/jj0fzG6c1xBZuNvfVA= +github.com/rs/xid v1.2.1/go.mod h1:+uKXf+4Djp6Md1KODXJxgGQPKngRmWyn10oCKFzNHOQ= +github.com/rs/zerolog v1.18.0 h1:CbAm3kP2Tptby1i9sYy2MGRg0uxIN9cyDb59Ys7W8z8= +github.com/rs/zerolog v1.18.0/go.mod h1:9nvC1axdVrAHcu/s9taAVfBuIdTZLVQmKQyvrUjF5+I= +github.com/sirupsen/logrus v1.2.0/go.mod h1:LxeOpSwHxABJmUn/MG1IvRgCAasNZTLOkJPxbbu5VWo= +github.com/sirupsen/logrus v1.4.2/go.mod h1:tLMulIdttU9McNUspp0xgXVQah82FyeX6MwdIuYE2rE= +github.com/sirupsen/logrus v1.6.0/go.mod h1:7uNnSEd1DgxDLC74fIahvMZmmYsHGZGEOFrfsX/uA88= +github.com/streadway/amqp v0.0.0-20200108173154-1c71cc93ed71 h1:2MR0pKUzlP3SGgj5NYJe/zRYDwOu9ku6YHy+Iw7l5DM= +github.com/streadway/amqp v0.0.0-20200108173154-1c71cc93ed71/go.mod h1:AZpEONHx3DKn8O/DFsRAY58/XVQiIPMTMB1SddzLXVw= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/objx v0.1.1/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw= +github.com/stretchr/objx v0.5.0 h1:1zr/of2m5FGMsad5YfcqgdqdWrIhu+EBEJRhR1U7z/c= +github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo= +github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs= +github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= +github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4= +github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= +github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU= +github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKsk= +github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4= +github.com/zenazn/goji v0.9.0/go.mod h1:7S9M489iMyHBNxwZnk9/EHS098H4/F6TATF2mIxtB1Q= +golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4= +golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= +golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= +golang.org/x/net v0.0.0-20180724234803-3673e40ba225/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= +golang.org/x/net v0.0.0-20181114220301-adae6a3d119a/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= +golang.org/x/net v0.0.0-20190108225652-1e06a53dbb7e/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= +golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= +golang.org/x/net v0.0.0-20190613194153-d28f0bde5980/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= +golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= +golang.org/x/net v0.0.0-20200625001655-4c5254603344/go.mod h1:/O7V0waA8r7cgGh81Ro3o1hOxt32SMVPicZroKQ2sZA= +golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw= +golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20201207232520-09787c993a3a/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= +golang.org/x/sys v0.0.0-20181116152217-5ac8a444bdc5/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= +golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= +golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20190422165155-953cdadca894/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20200106162015-b016eb3dc98e/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20200323222414-85ca7c5b95cd/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20200615200032-f1bc736245b1/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20200625212154-ddb9806d33ae/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20210124154548-22da62e12c0c/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20210603081109-ebe580a85c40/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.6.0 h1:MVltZSvRTcU2ljQOhs94SXPftV6DCNnZViHeQps87pQ= +golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= +golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk= +golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= +golang.org/x/tools v0.0.0-20190828213141-aed303cbaa74/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= +golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543 h1:E7g+9GITq07hpfrRu66IVDexMakfv52eLZ2CXBWiKr4= +golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +google.golang.org/appengine v1.4.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7/EB5XEv4= +google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8= +google.golang.org/protobuf v0.0.0-20200221191635-4d8936d0db64/go.mod h1:kwYJMbMJ01Woi6D6+Kah6886xMZcty6N08ah7+eCXa0= +google.golang.org/protobuf v0.0.0-20200228230310-ab0ca4ff8a60/go.mod h1:cfTl7dwQJ+fmap5saPgwCLgHXTUD7jkjRqWcaiX5VyM= +google.golang.org/protobuf v1.20.1-0.20200309200217-e05f789c0967/go.mod h1:A+miEFZTKqfCUM6K7xSMQL9OKL/b6hQv+e19PK+JZNE= +google.golang.org/protobuf v1.21.0/go.mod h1:47Nbq4nVaFHyn7ilMalzfO3qCViNmqZ2kzikPIcrTAo= +google.golang.org/protobuf v1.23.0/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU= +google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw= +google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc= +google.golang.org/protobuf v1.28.0 h1:w43yiav+6bVFTBQFZX0r7ipe9JQ1QsbMgHwbBziscLw= +google.golang.org/protobuf v1.28.0/go.mod h1:HV8QOd/L58Z+nl8r43ehVNZIU/HEI6OcFqwMG9pJV4I= +gopkg.in/alecthomas/kingpin.v2 v2.2.6/go.mod h1:FMv+mEhP44yOT+4EoQTLFTRgOQ1FBLkstjWtayDeSgw= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 h1:YR8cESwS4TdDjEe65xsg0ogRM/Nc3DYOhEAlW+xobZo= +gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/yaml.v2 v2.2.1/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= +gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= +gopkg.in/yaml.v2 v2.2.4/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= +gopkg.in/yaml.v2 v2.2.5/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= +gopkg.in/yaml.v2 v2.3.0/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= +gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/msg.txt b/msg.txt new file mode 100644 index 0000000..a9122f3 --- /dev/null +++ b/msg.txt @@ -0,0 +1,11 @@ +public.global.tickers {"eurusd":{"name":"EUR/USD","base_unit":"eur","quote_unit":"usd","low":"1000.0","high":"10000.0","last":"1000.0","open":1000.0,"volume":"0.0","sell":"1020.0","buy":"1000.0","at":1588001398}} +public.eurusd.update {"asks":[["1020.0","0.005"],["1026.0","0.03"]],"bids":[["1000.0","0.25"],["999.0","0.005"],["994.0","0.005"],["1.0","11.0"]]} +private.IDABC0000001.order {"id":22,"at":1588000798,"market":"eurusd","kind":"bid","price":"1026.0","state":"wait","volume":"0.001","origin_volume":"0.001"} +private.IDABC0000001.trade {"id":7,"kind":"ask","at":1588000798,"price":"1020.0","volume":"0.001","ask_id":15,"bid_id":22,"market":"eurusd"} +private.IDABC0000002.trade {"id":7,"kind":"bid","at":1588000798,"price":"1020.0","volume":"0.001","ask_id":15,"bid_id":22,"market":"eurusd"} +public.eurusd.trades {"trades":[{"tid":7,"taker_type":"buy","date":1588000798,"price":"1020.0","amount":"0.001"}]} +public.eurusd.ob-inc {"asks":[["1020.0","0.015"]],"sequence":497773} +public.eurusd.ob-snap {"bids":[["205.61","3.65889708"],["205.6","0.04510722"],["205.59","0.069"],["205.56","2.069"],["205.53","0.0467993"],["205.48","0.069"],["205.47","0.11749138"],["205.46","16.43759934"],["205.45","13"],["205.44","12.5"],["205.43","16.069"],["205.42","0.207"],["205.41","5.207"],["205.4","0.05018346"],["205.39","7.4525"],["205.38","9.73757"],["205.37","68.58682725"],["205.35","8.854"],["205.34","22.4"],["205.33","166.05087553"],["205.32","9.73899"],["205.31","39.38"],["205.3","88.88"],["205.29","1.02"],["205.28","3.40537786"]],"asks":[["205.62","161.8301188"],["205.63","0.04652703"],["205.69","66.10131708"],["205.71","1"],["205.72","8.32"],["205.74","9.72147"],["205.75","54.73161449"],["205.77","23.66494524"],["205.8","9.71864"],["205.82","12.5"],["205.83","18.991"],["205.84","0.87023637"],["205.85","4.009"],["205.86","38.0919318"],["205.87","3.02781953"],["205.88","12.5"],["205.89","133.2103"],["205.9","47.419"],["205.91","0.2775"],["205.92","123.54929228"],["205.93","0.279"],["205.94","170.6108218"],["205.95","16"],["205.97","21.58822214"],["205.98","159.28014829"]],"sequence":497774} +admin.eurusd.ob-inc {"bids":[["1000.0","0"]],"sequence":497775} +public.eurusd.ob-inc {"bids":[["999.0","0.001"]],"sequence":497776} +public.eurusd.kline-12h [1588204800,215.3,226.82,201.91566271,212.43,168224.36548725003] diff --git a/pkg/amqp/amqp.go b/pkg/amqp/amqp.go new file mode 100644 index 0000000..52520bc --- /dev/null +++ b/pkg/amqp/amqp.go @@ -0,0 +1,323 @@ +package amqp + +import ( + "errors" + "sync" + "time" + + "github.com/rs/zerolog/log" + "github.com/streadway/amqp" +) + +type AMQPSession struct { + connection *amqp.Connection + channel *amqp.Channel + done chan bool + streamsReInit []chan bool + ready chan bool + notifyConnClose chan *amqp.Error + notifyChanClose chan *amqp.Error + notifyConfirm chan amqp.Confirmation + isready bool + mutex sync.Mutex + mutexCh sync.Mutex +} + +const ( + // When reconnecting to the server after connection failure + reconnectDelay = 5 * time.Second + + // When setting up the channel after a channel exception + reInitDelay = 2 * time.Second + + // When resending messages the server didn't confirm + resendDelay = 5 * time.Second +) + +var ( + errNotConnected = errors.New("not connected to a server") + errAlreadyClosed = errors.New("already closed: not connected to the server") + errShutdown = errors.New("session is shutting down") +) + +// NewAMQPSession creates a new consumer state instance, and automatically +// attempts to connect to the server. +func NewAMQPSession(addr string) (*AMQPSession, error) { + session := AMQPSession{ + ready: make(chan bool, 1), + streamsReInit: make([]chan bool, 0), + } + go session.handleReconnect(addr) + return &session, nil +} + +func (session *AMQPSession) waitChannelReady() { + session.mutex.Lock() + + if !session.isready { + session.mutex.Unlock() + select { + case <-session.ready: + return + } + } else { + session.mutex.Unlock() + } +} + +func (session *AMQPSession) setReady(ready bool) { + session.mutex.Lock() + session.isready = ready + session.mutex.Unlock() + if ready { + session.ready <- true + } +} + +// connect will create a new AMQP connection +func (session *AMQPSession) connect(addr string) (*amqp.Connection, error) { + conn, err := amqp.Dial(addr) + + if err != nil { + return nil, err + } + + session.changeConnection(conn) + log.Info().Msg("AMQP Connected to AMQP!") + + return conn, nil +} + +// handleReconnect will wait for a connection error on +// notifyConnClose, and then continuously attempt to reconnect. +func (session *AMQPSession) handleReconnect(addr string) { + for { + session.setReady(false) + log.Info().Msg("AMQP Attempting to connect") + + conn, err := session.connect(addr) + + if err != nil { + log.Info().Msg("AMQP Failed to connect. Retrying...") + + select { + case <-session.done: + return + case <-time.After(reconnectDelay): + } + continue + } + + if done := session.handleReInit(conn); done { + break + } + } + log.Fatal().Msg("AMQP stopped") +} + +// handleReconnect will wait for a channel error +// and then continuously attempt to re-initialize both channels +func (session *AMQPSession) handleReInit(conn *amqp.Connection) bool { + for { + session.setReady(false) + + err := session.init(conn) + + if err != nil { + log.Info().Msg("AMQP Failed to initialize channel. Retrying...") + + select { + case <-session.done: + log.Info().Msg("AMQP Connection closed. Done") + return true + case <-time.After(reInitDelay): + } + continue + } + + select { + case <-session.done: + log.Info().Msg("AMQP Connection closed. Done") + return true + case <-session.notifyConnClose: + log.Info().Msg("AMQP Connection closed. Reconnecting...") + for _, ch := range session.streamsReInit { + ch <- true + } + return false + case <-session.notifyChanClose: + log.Info().Msg("AMQP Channel closed. Re-running init...") + for _, ch := range session.streamsReInit { + ch <- true + } + } + } +} + +// init will initialize channel +func (session *AMQPSession) init(conn *amqp.Connection) error { + ch, err := conn.Channel() + + if err != nil { + return err + } + + err = ch.Confirm(false) + + if err != nil { + return err + } + + session.changeChannel(ch) + session.setReady(true) + log.Info().Msg("AMQP Setup complete!") + + return nil +} + +// changeConnection takes a new connection to the queue, +// and updates the close listener to reflect this. +func (session *AMQPSession) changeConnection(connection *amqp.Connection) { + session.connection = connection + session.notifyConnClose = make(chan *amqp.Error) + session.connection.NotifyClose(session.notifyConnClose) +} + +// changeChannel takes a new channel to the queue, +// and updates the channel listeners to reflect this. +func (session *AMQPSession) changeChannel(channel *amqp.Channel) { + session.mutexCh.Lock() + session.channel = channel + session.notifyChanClose = make(chan *amqp.Error) + session.notifyConfirm = make(chan amqp.Confirmation, 1) + session.channel.NotifyClose(session.notifyChanClose) + session.channel.NotifyPublish(session.notifyConfirm) + session.mutexCh.Unlock() +} + +// Push will push data onto the queue, and wait for a confirm. +// If no confirms are received until within the resendTimeout, +// it continuously re-sends messages until a confirm is received. +// This will block until the server sends a confirm. Errors are +// only returned if the push action itself fails, see UnsafePush. +func (session *AMQPSession) Push(ex, rt string, data []byte) error { + for { + err := session.UnsafePush(ex, rt, data) + if err != nil { + log.Info().Msg("AMQP Push failed. Retrying...") + select { + case <-session.done: + return errShutdown + case <-time.After(resendDelay): + } + continue + } + select { + case confirm := <-session.notifyConfirm: + if confirm.Ack { + log.Info().Msg("AMQP Push confirmed!") + return nil + } + case <-time.After(resendDelay): + } + log.Info().Msg("AMQP Push didn't confirm. Retrying...") + } +} + +// Push will push to the queue without checking for +// confirmation. It returns an error if it fails to connect. +// No guarantees are provided for whether the server will +// recieve the message. +func (session *AMQPSession) UnsafePush(ex, rt string, data []byte) error { + session.waitChannelReady() + session.mutexCh.Lock() + defer session.mutexCh.Unlock() + return session.channel.Publish( + ex, // Exchange + rt, // Routing key + false, // Mandatory + false, // Immediate + amqp.Publishing{ + ContentType: "text/plain", + Body: data, + }, + ) +} + +// Stream will continuously put queue items on the channel. +// It is required to call delivery.Ack when it has been +// successfully processed, or delivery.Nack when it fails. +// Ignoring this will cause data to build up on the server. +func (session *AMQPSession) Stream(exName, qName, rKey string, consumer func(amqp.Delivery)) error { + session.mutex.Lock() + reinit := make(chan bool, 1) + session.streamsReInit = append(session.streamsReInit, reinit) + session.mutex.Unlock() + go func() { + for { + session.waitChannelReady() + + _, err := session.channel.QueueDeclare( + qName, + false, // Durable + true, // Delete when unused + true, // Exclusive + false, // No-wait + nil, // Arguments + ) + + if err != nil { + panic("AMQP failed to declare queue " + qName + ": " + err.Error()) + } + + session.channel.ExchangeDeclare(exName, "topic", false, false, false, false, nil) + session.channel.QueueBind(qName, rKey, exName, false, nil) + + ch, err := session.channel.Consume( + qName, + "", // Consumer + true, // Auto-Ack + true, // Exclusive + false, // No-local + false, // No-Wait + nil, // Args + ) + + if err != nil { + panic("AMQP failed to consume queue " + err.Error()) + } + + func() { + for { + select { + case delivery := <-ch: + consumer(delivery) + case <-reinit: + return + } + } + }() + + log.Warn().Msg("AMQP Stream loop interrupted") + } + }() + return nil +} + +// Close will delete the queue, close the channel and the connection. +func (session *AMQPSession) Close(qName string) error { + log.Error().Msg("Closing connection to RabbitMQ") + _, err := session.channel.QueueDelete(qName, false, false, false) + if err != nil { + return err + } + err = session.channel.Close() + if err != nil { + return err + } + err = session.connection.Close() + if err != nil { + return err + } + return nil +} diff --git a/pkg/auth/hmac.go b/pkg/auth/hmac.go new file mode 100644 index 0000000..ad8895e --- /dev/null +++ b/pkg/auth/hmac.go @@ -0,0 +1,47 @@ +package auth + +import ( + "crypto/hmac" + "crypto/sha256" + "encoding/hex" + "fmt" + "net/http" + "time" +) + +// APIKeyHMAC contains API keys credential to authenticate to HTTP API using HMAC +type APIKeyHMAC struct { + AccessKey string + SecretKey string +} + +// NewAPIKeyHMAC creates an instance of APIKeyHMAC +func NewAPIKeyHMAC(accessKey, secretKey string) *APIKeyHMAC { + return &APIKeyHMAC{ + AccessKey: accessKey, + SecretKey: secretKey, + } +} + +// GetSignature return a signature for the given nonce, if nonce is zero it use the current time in millisecond +func (key *APIKeyHMAC) GetSignature(nonce int64) string { + if nonce == 0 { + nonce = int64(time.Now().UnixNano() * 1000000) + } + mac := hmac.New(sha256.New, []byte(key.SecretKey)) + mac.Write([]byte(fmt.Sprintf("%d%s", nonce, key.AccessKey))) + return hex.EncodeToString(mac.Sum(nil)) +} + +// GetSignedHeader returns a header with valid HMAC authorization fields +func (key *APIKeyHMAC) GetSignedHeader(nonce int64) http.Header { + if nonce == 0 { + nonce = int64(time.Now().UnixNano() * 1000000) + } + + return http.Header{ + "X-Auth-Apikey": {key.AccessKey}, + "X-Auth-Nonce": {fmt.Sprintf("%d", nonce)}, + "X-Auth-Signature": {key.GetSignature(nonce)}, + } +} diff --git a/pkg/auth/hmac_test.go b/pkg/auth/hmac_test.go new file mode 100644 index 0000000..546d741 --- /dev/null +++ b/pkg/auth/hmac_test.go @@ -0,0 +1,34 @@ +package auth + +import ( + "net/http" + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestAPIKeyHMACGetSignature(t *testing.T) { + accessKey := "61d025b8573501c2" + secretKey := "2d0b4979c7fe6986daa8e21d1dc0644f" + + k := NewAPIKeyHMAC(accessKey, secretKey) + nonce := int64(1584524005143) + signature := k.GetSignature(nonce) + assert.Equal(t, "bd42b945e095880e28d046846dbecf655fdf09d95a396a24fe6fe1df42f15d13", signature) +} + +func TestAPIKeyHMACGetSignedHeader(t *testing.T) { + accessKey := "61d025b8573501c2" + secretKey := "2d0b4979c7fe6986daa8e21d1dc0644f" + + k := NewAPIKeyHMAC(accessKey, secretKey) + nonce := int64(1584524005143) + headers := k.GetSignedHeader(nonce) + + assert.Equal(t, + http.Header{ + "X-Auth-Apikey": {accessKey}, + "X-Auth-Nonce": {"1584524005143"}, + "X-Auth-Signature": {"bd42b945e095880e28d046846dbecf655fdf09d95a396a24fe6fe1df42f15d13"}, + }, headers) +} diff --git a/pkg/auth/jwt.go b/pkg/auth/jwt.go new file mode 100644 index 0000000..dab755a --- /dev/null +++ b/pkg/auth/jwt.go @@ -0,0 +1,31 @@ +package auth + +import ( + "crypto/rsa" + "encoding/json" + + "github.com/golang-jwt/jwt" +) + +// Auth struct represents parsed jwt information. +type Auth struct { + UID string `json:"uid"` + State string `json:"state"` + Email string `json:"email"` + Role string `json:"role"` + ReferralID json.Number `json:"referral_id"` + Level json.Number `json:"level"` + Audience []string `json:"aud,omitempty"` + jwt.StandardClaims +} + +// ParseAndValidate parses token and validates RS256 signature with Barong RSA public key. +func ParseAndValidate(token string, key *rsa.PublicKey) (Auth, error) { + auth := Auth{} + + _, err := jwt.ParseWithClaims(token, &auth, func(t *jwt.Token) (interface{}, error) { + return key, nil + }) + + return auth, err +} diff --git a/pkg/auth/key_store.go b/pkg/auth/key_store.go new file mode 100644 index 0000000..50a60f5 --- /dev/null +++ b/pkg/auth/key_store.go @@ -0,0 +1,150 @@ +package auth + +import ( + "crypto/rand" + "crypto/rsa" + "crypto/x509" + "encoding/base64" + "encoding/pem" + "io/ioutil" + "os" + + "github.com/golang-jwt/jwt" +) + +type KeyStore struct { + PublicKey *rsa.PublicKey + PrivateKey *rsa.PrivateKey +} + +func fileExist(path string) bool { + _, err := os.Stat(path) + return err == nil +} + +func (ks *KeyStore) LoadPublicKeyFromFile(path string) error { + pemBytes, err := ioutil.ReadFile(path) + if err != nil { + return err + } + + key, err := jwt.ParseRSAPublicKeyFromPEM(pemBytes) + if err != nil { + return err + } + + ks.PublicKey = key + + return nil +} + +func (ks *KeyStore) LoadPublicKeyFromString(str string) error { + pemBytes, err := base64.StdEncoding.DecodeString(str) + if err != nil { + return err + } + + key, err := jwt.ParseRSAPublicKeyFromPEM(pemBytes) + if err != nil { + return err + } + + ks.PublicKey = key + + return nil +} + +func (ks *KeyStore) LoadPrivateKey(path string) error { + pemBytes, err := ioutil.ReadFile(path) + if err != nil { + return err + } + + key, err := jwt.ParseRSAPrivateKeyFromPEM(pemBytes) + if err != nil { + return err + } + + ks.PrivateKey = key + + return nil +} + +func (ks *KeyStore) GenerateKeys() error { + key, err := rsa.GenerateKey(rand.Reader, 2048) + if err != nil { + return err + } + + ks.PrivateKey = key + ks.PublicKey = &key.PublicKey + + return nil +} + +func (ks *KeyStore) SavePrivateKey(path string) error { + file, err := os.Create(path) + if err != nil { + return err + } + defer file.Close() + + block := &pem.Block{ + Type: "RSA PRIVATE KEY", + Bytes: x509.MarshalPKCS1PrivateKey(ks.PrivateKey), + } + + return pem.Encode(file, block) +} + +func (ks *KeyStore) SavePublicKey(path string) error { + bytes, err := x509.MarshalPKIXPublicKey(ks.PublicKey) + if err != nil { + return err + } + + file, err := os.Create(path) + if err != nil { + return err + } + defer file.Close() + + block := &pem.Block{ + Type: "PUBLIC KEY", + Bytes: bytes, + } + + return pem.Encode(file, block) +} + +func LoadOrGenerateKeys(privPath, pubPath string) (*KeyStore, error) { + ks := &KeyStore{} + + if fileExist(privPath) { + if err := ks.LoadPrivateKey(privPath); err != nil { + return ks, err + } + } else { + if err := ks.GenerateKeys(); err != nil { + return ks, err + } + if err := ks.SavePrivateKey(privPath); err != nil { + return ks, err + } + } + + if fileExist(pubPath) { + if err := ks.LoadPublicKeyFromFile(pubPath); err != nil { + return ks, err + } + } else { + if ks.PublicKey == nil { + ks.PublicKey = &ks.PrivateKey.PublicKey + } + if err := ks.SavePublicKey(pubPath); err != nil { + return ks, err + } + } + + return ks, nil +} diff --git a/pkg/auth/validator.go b/pkg/auth/validator.go new file mode 100644 index 0000000..394adcf --- /dev/null +++ b/pkg/auth/validator.go @@ -0,0 +1,83 @@ +package auth + +import ( + "crypto/ed25519" + "crypto/rsa" + "fmt" + "os" + + pkgjwt "github.com/openware/pkg/jwt" +) + +const ( + ValidatorRSA = "rsa" + ValidatorEdDSA = "eddsa" +) + +// Validator verifies Barong JWT tokens using RS256 or EdDSA depending on configured key material. +type Validator struct { + Mode string + RSA *rsa.PublicKey + EdDSA ed25519.PublicKey +} + +type Claims struct { + UID string + Role string +} + +// LoadValidator reads JWT_PUBLIC_KEY (base64 PEM) or -pubKey file path. +// OpenDAX / Barong 2.6 uses RS256; upstream Rango 3.1.2+ defaults to EdDSA. +func LoadValidator(pubKeyPath string) (Validator, error) { + if encPem := os.Getenv("JWT_PUBLIC_KEY"); encPem != "" { + rsaStore := KeyStore{} + if err := rsaStore.LoadPublicKeyFromString(encPem); err == nil && rsaStore.PublicKey != nil { + return Validator{Mode: ValidatorRSA, RSA: rsaStore.PublicKey}, nil + } + + edStore := pkgjwt.KeyStoreEdDSA{} + if err := edStore.LoadPublicKeyFromString(encPem); err == nil && edStore.PublicKey != nil { + return Validator{Mode: ValidatorEdDSA, EdDSA: edStore.PublicKey}, nil + } + + return Validator{}, fmt.Errorf("JWT_PUBLIC_KEY is neither a valid RSA nor EdDSA public key") + } + + rsaStore := KeyStore{} + if err := rsaStore.LoadPublicKeyFromFile("config/rsa-key.pub"); err == nil && rsaStore.PublicKey != nil { + return Validator{Mode: ValidatorRSA, RSA: rsaStore.PublicKey}, nil + } + + edStore := pkgjwt.KeyStoreEdDSA{} + if err := edStore.LoadPublicKeyFromFile(pubKeyPath); err != nil { + return Validator{}, fmt.Errorf("load public key from %q: %w", pubKeyPath, err) + } + if edStore.PublicKey == nil { + return Validator{}, fmt.Errorf("public key not loaded from %q", pubKeyPath) + } + + return Validator{Mode: ValidatorEdDSA, EdDSA: edStore.PublicKey}, nil +} + +func (v Validator) ParseAndValidate(token string) (Claims, error) { + if token == "" { + return Claims{}, fmt.Errorf("missing token") + } + + switch v.Mode { + case ValidatorRSA: + auth, err := ParseAndValidate(token, v.RSA) + if err != nil { + return Claims{}, err + } + return Claims{UID: auth.UID, Role: auth.Role}, nil + case ValidatorEdDSA: + auth, err := pkgjwt.ParseAndValidateEdDSA(token, v.EdDSA) + if err != nil { + return Claims{}, err + } + return Claims{UID: auth.UID, Role: auth.Role}, nil + default: + return Claims{}, fmt.Errorf("unknown validator mode %q", v.Mode) + } +} diff --git a/pkg/auth/validator_test.go b/pkg/auth/validator_test.go new file mode 100644 index 0000000..d8556d3 --- /dev/null +++ b/pkg/auth/validator_test.go @@ -0,0 +1,31 @@ +package auth + +import ( + "crypto/rand" + "crypto/rsa" + "encoding/base64" + "os" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestLoadValidator_RSAFromEnv(t *testing.T) { + key, err := rsa.GenerateKey(rand.Reader, 2048) + require.NoError(t, err) + + store := KeyStore{PublicKey: &key.PublicKey} + path := t.TempDir() + "/pub.pem" + require.NoError(t, store.SavePublicKey(path)) + + pemBytes, err := os.ReadFile(path) + require.NoError(t, err) + + t.Setenv("JWT_PUBLIC_KEY", base64.StdEncoding.EncodeToString(pemBytes)) + + validator, err := LoadValidator("config/ed25519-key.pub") + require.NoError(t, err) + assert.Equal(t, ValidatorRSA, validator.Mode) + assert.NotNil(t, validator.RSA) +} diff --git a/pkg/message/msg.go b/pkg/message/msg.go new file mode 100644 index 0000000..dd4dfae --- /dev/null +++ b/pkg/message/msg.go @@ -0,0 +1,26 @@ +package message + +import ( + "encoding/json" +) + +type Request struct { + Method string + Streams []string +} + +func PackOutgoingResponse(err error, message interface{}) ([]byte, error) { + res := make(map[string]interface{}, 1) + if err != nil { + res["error"] = err.Error() + } else { + res["success"] = message + } + return json.Marshal(res) +} + +func PackOutgoingEvent(channel string, data interface{}) ([]byte, error) { + resp := make(map[string]interface{}, 1) + resp[channel] = data + return json.Marshal(resp) +} diff --git a/pkg/message/msg_test.go b/pkg/message/msg_test.go new file mode 100644 index 0000000..5249dd1 --- /dev/null +++ b/pkg/message/msg_test.go @@ -0,0 +1,48 @@ +package message + +import ( + "errors" + "fmt" + "testing" +) + +func TestMsg_Response(t *testing.T) { + t.Run("no error", func(t *testing.T) { + res, err := PackOutgoingResponse(nil, "ok") + fmt.Println(string(res)) + + if err != nil { + t.Fatal("Should not return error") + } + + if string(res) != `{"success":"ok"}` { + t.Fatal("Response invalid") + } + }) + + t.Run("Some error", func(t *testing.T) { + res, err := PackOutgoingResponse(errors.New("Some Error"), "ok") + fmt.Println(string(res)) + + if err != nil { + t.Fatal("Should not return error") + } + + if string(res) != `{"error":"Some Error"}` { + t.Fatal("Response invalid") + } + }) +} + +func TestMsg_Event(t *testing.T) { + res, err := PackOutgoingEvent("someMethod", "Hello") + fmt.Println(string(res)) + + if err != nil { + t.Fatal("Should not return error") + } + + if string(res) != `{"someMethod":"Hello"}` { + t.Fatal("Event invalid") + } +} diff --git a/pkg/message/parser.go b/pkg/message/parser.go new file mode 100644 index 0000000..7ae95d5 --- /dev/null +++ b/pkg/message/parser.go @@ -0,0 +1,59 @@ +package message + +import ( + "encoding/json" + "errors" + "fmt" + "reflect" +) + +func ParseRequest(msg []byte) (Request, error) { + request, err := Parse(msg) + if err != nil { + return request, err + } + + return request, nil +} + +func Parse(msg []byte) (Request, error) { + var v map[string]interface{} + var parsed Request + + if err := json.Unmarshal(msg, &v); err != nil { + return parsed, fmt.Errorf("Could not parse message: %w", err) + } + + switch v["event"] { + case "subscribe": + parsed.Method = "subscribe" + streams, ok := v["streams"] + if !ok { + return parsed, fmt.Errorf("No streams provided") + } + switch reflect.TypeOf(streams).Kind() { + case reflect.Slice: + streams := reflect.ValueOf(v["streams"]) + for i := 0; i < streams.Len(); i++ { + parsed.Streams = append(parsed.Streams, streams.Index(i).Interface().(string)) + } + } + case "unsubscribe": + parsed.Method = "unsubscribe" + streams, ok := v["streams"] + if !ok { + return parsed, fmt.Errorf("No streams provided") + } + switch reflect.TypeOf(streams).Kind() { + case reflect.Slice: + streams := reflect.ValueOf(v["streams"]) + for i := 0; i < streams.Len(); i++ { + parsed.Streams = append(parsed.Streams, streams.Index(i).Interface().(string)) + } + } + default: + return parsed, errors.New("Could not parse Type: Invalid event") + } + + return parsed, nil +} diff --git a/pkg/metrics/metrics.go b/pkg/metrics/metrics.go new file mode 100644 index 0000000..4e6d11b --- /dev/null +++ b/pkg/metrics/metrics.go @@ -0,0 +1,63 @@ +package metrics + +import ( + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/client_golang/prometheus/promauto" +) + +var defaultMetrics *Metrics + +type Metrics struct { + clients prometheus.Gauge + subs *prometheus.GaugeVec +} + +func Enable() { + defaultMetrics = &Metrics{} + registerMetrics() +} + +func registerMetrics() { + defaultMetrics.clients = promauto.NewGauge( + prometheus.GaugeOpts{ + Name: "rango_hub_clients_count", + Help: "Number of clients currently connected", + }, + ) + + defaultMetrics.subs = promauto.NewGaugeVec( + prometheus.GaugeOpts{ + Name: "rango_hub_subscriptions_count", + Help: "Number of user subscribed to a topic", + }, + []string{"type", "topic"}, + ) +} + +func RecordHubClientNew() { + if defaultMetrics == nil { + return + } + defaultMetrics.clients.Inc() +} + +func RecordHubClientClose() { + if defaultMetrics == nil { + return + } + defaultMetrics.clients.Dec() +} + +func RecordHubSubscription(typ, topic string) { + if defaultMetrics == nil { + return + } + defaultMetrics.subs.WithLabelValues(typ, topic).Inc() +} + +func RecordHubUnsubscription(typ, topic string) { + if defaultMetrics == nil { + return + } + defaultMetrics.subs.WithLabelValues(typ, topic).Dec() +} diff --git a/pkg/routing/client.go b/pkg/routing/client.go new file mode 100644 index 0000000..7bdde55 --- /dev/null +++ b/pkg/routing/client.go @@ -0,0 +1,331 @@ +package routing + +import ( + "bytes" + "net/http" + "net/url" + "os" + "strings" + "time" + + "github.com/gorilla/websocket" + msg "github.com/openware/rango/pkg/message" + "github.com/openware/rango/pkg/metrics" + "github.com/rs/zerolog/log" +) + +const ( + // Time allowed to write a message to the peer. + writeWait = 10 * time.Second + + // Time allowed to read the next pong message from the peer. + pongWait = 60 * time.Second + + // Send pings to peer with this period. Must be less than pongWait. + pingPeriod = (pongWait * 9) / 10 + + // Maximum message size allowed from peer. + maxMessageSize = 512 +) + +var ( + newline = []byte{'\n'} + space = []byte{' '} +) + +var upgrader = websocket.Upgrader{ + ReadBufferSize: 1024, + WriteBufferSize: 1024, + CheckOrigin: checkSameOrigin(os.Getenv("API_CORS_ORIGINS")), +} + +var maxBufferedMessages = 256 + +type Auth struct { + UID string + Role string +} + +// FIXME: IClient looks very wrong. +type IClient interface { + Send(string) + Close() + GetAuth() Auth + GetSubscriptions() []string + SubscribePublic(string) + SubscribePrivate(string) + UnsubscribePublic(string) + UnsubscribePrivate(string) +} + +// Client is a middleman between the websocket connection and the hub. +type Client struct { + hub *Hub + + // User ID if authorized + Auth Auth + + pubSub []string + privSub []string + + // The websocket connection. + conn *websocket.Conn + + // Buffered channel of outbound messages. + send chan []byte +} + +func checkSameOrigin(origins string) func(r *http.Request) bool { + if origins == "" { + return func(r *http.Request) bool { + origin := r.Header["Origin"] + if len(origin) == 0 { + return true + } + u, err := url.Parse(origin[0]) + if err != nil { + return false + } + return strings.EqualFold(u.Host, r.Host) + } + } + + hosts := []string{} + + for _, o := range strings.Split(origins, ",") { + o = strings.TrimSpace(o) + if strings.HasPrefix(o, "http://") || strings.HasPrefix(o, "https://") { + u, err := url.Parse(o) + if err != nil || u.Host == "" { + panic("Failed to parse url in API_CORS_ORIGINS: " + o) + } + hosts = append(hosts, u.Host) + } else { + hosts = append(hosts, o) + } + } + + return func(r *http.Request) bool { + origin := r.Header["Origin"] + if len(origin) == 0 { + return true + } + u, err := url.Parse(origin[0]) + if err != nil { + return false + } + + for _, host := range hosts { + if strings.EqualFold(u.Host, host) { + return true + } + } + return false + } +} + +// NewClient handles websocket requests from the peer. +func NewClient(hub *Hub, w http.ResponseWriter, r *http.Request) { + conn, err := upgrader.Upgrade(w, r, nil) + if err != nil { + log.Error().Msg("Websocket upgrade failed: " + err.Error()) + return + } + client := &Client{ + hub: hub, + conn: conn, + send: make(chan []byte, maxBufferedMessages), + Auth: Auth{ + UID: r.Header.Get("JwtUID"), + Role: r.Header.Get("JwtRole"), + }, + pubSub: []string{}, + privSub: []string{}, + } + + if client.Auth.UID == "" { + log.Info().Msgf("New anonymous connection") + } else { + log.Info().Msgf("New authenticated connection: %s", client.Auth.UID) + } + + hub.handleSubscribe(&Request{ + client: client, + Request: msg.Request{ + Streams: parseStreamsFromURI(r.RequestURI), + }, + }) + + metrics.RecordHubClientNew() + + // Allow collection of memory referenced by the caller by doing all work in + // new goroutines. + go client.write() + go client.read() +} + +func (c *Client) Send(s string) { + if len(c.send) == maxBufferedMessages { + log.Warn().Msg("Closing slow websocket connection") + c.conn.Close() + } else { + c.send <- []byte(s) + } +} + +func (c *Client) Close() { + close(c.send) +} + +func (c *Client) GetAuth() Auth { + return c.Auth +} + +func (c *Client) GetSubscriptions() []string { + return append(c.pubSub, c.privSub...) +} + +func (c *Client) SubscribePublic(s string) { + if !contains(c.pubSub, s) { + c.pubSub = append(c.pubSub, s) + } +} + +func (c *Client) SubscribePrivate(s string) { + if !contains(c.privSub, s) { + c.privSub = append(c.privSub, s) + } +} + +func (c *Client) UnsubscribePublic(s string) { + l := make([]string, len(c.pubSub)-1) + i := 0 + for _, el := range c.pubSub { + if s != el { + l[i] = el + i++ + } + } + c.pubSub = l +} + +func (c *Client) UnsubscribePrivate(s string) { + l := make([]string, len(c.privSub)-1) + i := 0 + for _, el := range c.privSub { + if s != el { + l[i] = el + i++ + } + } + c.privSub = l +} + +func parseStreamsFromURI(uri string) []string { + streams := make([]string, 0) + path := strings.Split(uri, "?") + if len(path) != 2 { + return streams + } + for _, up := range strings.Split(path[1], "&") { + p := strings.Split(up, "=") + if len(p) != 2 || p[0] != "stream" { + continue + } + streams = append(streams, strings.Split(p[1], ",")...) + + } + return streams +} + +// read pumps messages from the websocket connection to the hub. +// +// The application runs read in a per-connection goroutine. The application +// ensures that there is at most one reader on a connection by executing all +// reads from this goroutine. +func (c *Client) read() { + defer func() { + log.Debug().Msgf("Closing client read (%s)", c.GetAuth().UID) + c.hub.Unregister <- c + metrics.RecordHubClientClose() + c.conn.Close() + }() + + c.conn.SetReadLimit(maxMessageSize) + c.conn.SetReadDeadline(time.Now().Add(pongWait)) + c.conn.SetPongHandler(func(string) error { + c.conn.SetReadDeadline(time.Now().Add(pongWait)) + return nil + }) + + for { + _, message, err := c.conn.ReadMessage() + if err != nil { + if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) { + log.Info().Msgf("error: %v", err) + } + break + } + message = bytes.TrimSpace(bytes.Replace(message, newline, space, -1)) + if len(message) == 0 { + continue + } + if isDebug() { + log.Debug().Msgf("Received message %s", message) + } + + // handle ping + if string(message) == "ping" { + c.send <- []byte("pong") + continue + } + + req, err := msg.ParseRequest(message) + if err != nil { + c.send <- []byte(responseMust(err, nil)) + continue + } + + c.hub.Requests <- Request{c, req} + } +} + +// write pumps messages from the hub to the websocket connection. +// +// A goroutine running write is started for each connection. The +// application ensures that there is at most one writer to a connection by +// executing all writes from this goroutine. +func (c *Client) write() { + ticker := time.NewTicker(pingPeriod) + defer func() { + log.Debug().Msgf("Closing client write (%s)", c.GetAuth().UID) + ticker.Stop() + c.conn.Close() + }() + + for { + select { + case message, ok := <-c.send: + c.conn.SetWriteDeadline(time.Now().Add(writeWait)) + if !ok { + // The hub closed the channel. + c.conn.WriteMessage(websocket.CloseMessage, []byte{}) + return + } + + w, err := c.conn.NextWriter(websocket.TextMessage) + if err != nil { + return + } + w.Write(message) + if err := w.Close(); err != nil { + return + } + case <-ticker.C: + c.conn.SetWriteDeadline(time.Now().Add(writeWait)) + if err := c.conn.WriteMessage(websocket.PingMessage, nil); err != nil { + return + } + } + } +} diff --git a/pkg/routing/client_test.go b/pkg/routing/client_test.go new file mode 100644 index 0000000..c8c24d0 --- /dev/null +++ b/pkg/routing/client_test.go @@ -0,0 +1,139 @@ +package routing + +import ( + "net/http" + "testing" + + "github.com/stretchr/testify/assert" +) + +func TestClient(t *testing.T) { + hub := NewHub(nil) + client := &Client{ + hub: hub, + send: make(chan []byte, 256), + Auth: Auth{UID: "UIDABC001", Role: "admin"}, + pubSub: []string{}, + privSub: []string{}, + } + + assert.Equal(t, "UIDABC001", client.GetAuth().UID) + assert.Equal(t, []string{}, client.GetSubscriptions()) + + client.SubscribePublic("a.x") + assert.Equal(t, []string{"a.x"}, client.GetSubscriptions()) + assert.Equal(t, []string{"a.x"}, client.pubSub) + assert.Equal(t, []string{}, client.privSub) + + client.SubscribePublic("a.y") + assert.Equal(t, []string{"a.x", "a.y"}, client.GetSubscriptions()) + assert.Equal(t, []string{"a.x", "a.y"}, client.pubSub) + assert.Equal(t, []string{}, client.privSub) + + client.UnsubscribePublic("a.y") + assert.Equal(t, []string{"a.x"}, client.GetSubscriptions()) + assert.Equal(t, []string{"a.x"}, client.pubSub) + assert.Equal(t, []string{}, client.privSub) + + client.SubscribePrivate("b") + assert.Equal(t, []string{"a.x", "b"}, client.GetSubscriptions()) + assert.Equal(t, []string{"a.x"}, client.pubSub) + assert.Equal(t, []string{"b"}, client.privSub) + + client.SubscribePrivate("c") + assert.Equal(t, []string{"a.x", "b", "c"}, client.GetSubscriptions()) + assert.Equal(t, []string{"a.x"}, client.pubSub) + assert.Equal(t, []string{"b", "c"}, client.privSub) + + client.UnsubscribePrivate("b") + assert.Equal(t, []string{"a.x", "c"}, client.GetSubscriptions()) + assert.Equal(t, []string{"a.x"}, client.pubSub) + assert.Equal(t, []string{"c"}, client.privSub) + + client.UnsubscribePrivate("c") + assert.Equal(t, []string{"a.x"}, client.GetSubscriptions()) + assert.Equal(t, []string{"a.x"}, client.pubSub) + assert.Equal(t, []string{}, client.privSub) + + client.UnsubscribePublic("a.x") + assert.Equal(t, []string{}, client.GetSubscriptions()) + assert.Equal(t, []string{}, client.pubSub) + assert.Equal(t, []string{}, client.privSub) +} + +func TestParseStreamsFromURI(t *testing.T) { + assert.Equal(t, []string{}, parseStreamsFromURI("/?")) + assert.Equal(t, []string{}, parseStreamsFromURI("")) + assert.Equal(t, []string{"aaa", "bbb"}, parseStreamsFromURI("/?stream=aaa&stream=bbb")) + assert.Equal(t, []string{"aaa", "bbb"}, parseStreamsFromURI("/?stream=aaa,bbb")) + assert.Equal(t, []string{"aaa", "bbb"}, parseStreamsFromURI("/public/?stream=aaa,bbb")) +} + +func TestCheckSameOriginEmpty(t *testing.T) { + var checkSameOriginTests = []struct { + ok bool + r *http.Request + }{ + {false, &http.Request{Host: "example.org", Header: map[string][]string{"Origin": {"https://other.org"}}}}, + {true, &http.Request{Host: "example.org", Header: map[string][]string{"Origin": {"https://example.org"}}}}, + {true, &http.Request{Host: "Example.org", Header: map[string][]string{"Origin": {"https://example.org"}}}}, + } + + for _, tt := range checkSameOriginTests { + ok := checkSameOrigin("")(tt.r) + if tt.ok != ok { + t.Errorf("checkSameOrigin(%+v) returned %v, want %v", tt.r, ok, tt.ok) + } + } +} + +func TestCheckSameOriginDomainsSetup(t *testing.T) { + var checkSameOriginTests = []struct { + ok bool + r *http.Request + }{ + {false, &http.Request{Host: "example.org", Header: map[string][]string{"Origin": {"https://other.org"}}}}, + {true, &http.Request{Host: "whatever.org", Header: map[string][]string{"Origin": {"https://example.org"}}}}, + {true, &http.Request{Host: "whatever.org", Header: map[string][]string{"Origin": {"https://Example.org"}}}}, + {true, &http.Request{Host: "whatever.org", Header: map[string][]string{"Origin": {"https://example.com"}}}}, + {true, &http.Request{Host: "whatever.org", Header: map[string][]string{"Origin": {"https://Example.com"}}}}, + {true, &http.Request{Host: "whatever.org", Header: map[string][]string{"Origin": {}}}}, + } + + checker := checkSameOrigin("example.org,example.com") + for _, tt := range checkSameOriginTests { + ok := checker(tt.r) + if tt.ok != ok { + t.Errorf("checkSameOrigin(%+v) returned %v, want %v", tt.r, ok, tt.ok) + } + } + + checker = checkSameOrigin("example.org, example.com") + for _, tt := range checkSameOriginTests { + ok := checker(tt.r) + if tt.ok != ok { + t.Errorf("checkSameOrigin(%+v) returned %v, want %v", tt.r, ok, tt.ok) + } + } + + checker = checkSameOrigin("https://example.org,https://example.com") + for _, tt := range checkSameOriginTests { + ok := checker(tt.r) + if tt.ok != ok { + t.Errorf("checkSameOrigin(%+v) returned %v, want %v", tt.r, ok, tt.ok) + } + } + + checker = checkSameOrigin("https://example.org, https://example.com") + for _, tt := range checkSameOriginTests { + ok := checker(tt.r) + if tt.ok != ok { + t.Errorf("checkSameOrigin(%+v) returned %v, want %v", tt.r, ok, tt.ok) + } + } +} + +func TestCheckSameOriginBadConfiguration(t *testing.T) { + assert.Panics(t, func() { checkSameOrigin("https://ex ample.org") }) + assert.Panics(t, func() { checkSameOrigin("https://ex:ample.org") }) +} diff --git a/pkg/routing/hub.go b/pkg/routing/hub.go new file mode 100644 index 0000000..41cb496 --- /dev/null +++ b/pkg/routing/hub.go @@ -0,0 +1,572 @@ +package routing + +import ( + "encoding/json" + "errors" + "fmt" + "strings" + "sync" + + msg "github.com/openware/rango/pkg/message" + "github.com/openware/rango/pkg/metrics" + "github.com/rs/zerolog" + "github.com/rs/zerolog/log" + "github.com/streadway/amqp" +) + +type Request struct { + client IClient + msg.Request +} + +// Hub maintains the set of active clients and broadcasts messages to the +// clients. +type Hub struct { + // Register Requests from the clients. + Requests chan Request + + // Unregister requests from clients. + Unregister chan IClient + + // List of clients registered to public topics + PublicTopics map[string]*Topic + + // List of clients registered to private topics + PrivateTopics map[string]map[string]*Topic + + // map[prefix -> map[topic -> *Topic]] + PrefixedTopics map[string]map[string]*Topic + + // Storage for incremental objects + IncrementalObjects map[string]*IncrementalObject + + // map[prefix -> allowed roles] + RBAC map[string][]string + + mutex sync.Mutex +} + +type Event struct { + Scope string // global, public, private + Stream string // channel routing key + Type string // event type + Topic string // topic routing key (stream.type) + Body interface{} // event json body +} + +type IncrementalObject struct { + Snapshot string + Increments []string +} + +func NewHub(rbac map[string][]string) *Hub { + return &Hub{ + Requests: make(chan Request), + Unregister: make(chan IClient), + PublicTopics: make(map[string]*Topic, 100), + PrivateTopics: make(map[string]map[string]*Topic, 1000), + PrefixedTopics: make(map[string]map[string]*Topic, 100), + IncrementalObjects: make(map[string]*IncrementalObject, 5), + RBAC: rbac, + } +} + +func isIncrementObject(s string) bool { + return strings.HasSuffix(s, "-inc") +} + +func isSnapshotObject(s string) bool { + return strings.HasSuffix(s, "-snap") +} + +func isDebug() bool { + return log.Logger.GetLevel() <= zerolog.DebugLevel +} + +func isTrace() bool { + return log.Logger.GetLevel() <= zerolog.TraceLevel +} + +func getTopic(scope, stream, typ string) string { + if isSnapshotObject(typ) { + typ = strings.Replace(typ, "-snap", "-inc", 1) + } + if scope == "private" { + return typ + } + return stream + "." + typ +} + +func (h *Hub) ListenWebsocketEvents() { + for { + select { + case req := <-h.Requests: + h.handleRequest(&req) + + case client := <-h.Unregister: + log.Info().Msgf("Unregistering client (%s)", client.GetAuth().UID) + h.unsubscribeAll(client) + client.Close() + } + } +} + +// ReceiveMsg handles AMQP messages +func (h *Hub) ReceiveMsg(delivery amqp.Delivery) { + if isTrace() { + log.Trace().Msgf("AMQP msg received: %s -> %s", delivery.RoutingKey, delivery.Body) + } + s := strings.Split(delivery.RoutingKey, ".") + + var o interface{} + err := json.Unmarshal(delivery.Body, &o) + + if err != nil { + log.Error().Msgf("JSON parse error: %s, msg: %s", err.Error(), delivery.Body) + return + } + + switch len(s) { + case 2: + msg := Event{ + Scope: s[0], + Stream: "", + Type: s[1], + Topic: getTopic(s[0], s[0], s[1]), + Body: o, + } + + h.routeMessage(&msg) + + case 3: + msg := Event{ + Scope: s[0], + Stream: s[1], + Type: s[2], + Topic: getTopic(s[0], s[1], s[2]), + Body: o, + } + + h.routeMessage(&msg) + + default: + log.Error().Msgf("Bad routing key: %s", delivery.RoutingKey) + } +} + +func (h *Hub) SkipPrivateMsg(delivery amqp.Delivery) { + if strings.HasPrefix(delivery.RoutingKey, "private") { + return + } + + h.ReceiveMsg(delivery) +} + +func (h *Hub) handleSnapshot(msg *Event) (string, error) { + topic := msg.Stream + "." + msg.Type + body, err := json.Marshal(map[string]interface{}{ + topic: msg.Body, + }) + + if err != nil { + return "", err + } + + o, ok := h.IncrementalObjects[msg.Topic] + if !ok { + o = &IncrementalObject{} + h.IncrementalObjects[msg.Topic] = o + } + o.Snapshot = string(body) + o.Increments = []string{} + + return string(body), nil +} + +func (h *Hub) handleIncrement(msg *Event) (string, error) { + body, err := json.Marshal(map[string]interface{}{ + msg.Topic: msg.Body, + }) + + if err != nil { + return "", err + } + + o, ok := h.IncrementalObjects[msg.Topic] + if !ok { + return "", fmt.Errorf("No snapshot received before the increment for topic %s, ignoring", msg.Topic) + } + o.Increments = append(o.Increments, string(body)) + return string(body), nil + +} + +func (h *Hub) handleMessage(topic *Topic, ok bool, msg *Event) { + switch { + case isIncrementObject(msg.Type): + rm, err := h.handleIncrement(msg) + if err != nil { + log.Error().Msgf("handleIncrement failed: %s", err.Error()) + return + } + if ok { + topic.broadcastRaw(rm) + } + + case isSnapshotObject(msg.Type): + _, err := h.handleSnapshot(msg) + if err != nil { + log.Error().Msgf("handleSnapshot failed: %s", err.Error()) + return + } + + default: + if ok { + topic.broadcast(msg) + } + } +} + +func (h *Hub) routeMessage(msg *Event) { + if isTrace() { + log.Trace().Msgf("Routing message %v", msg) + } + h.mutex.Lock() + defer h.mutex.Unlock() + + switch msg.Scope { + case "public", "global": + topic, ok := h.PublicTopics[msg.Topic] + h.handleMessage(topic, ok, msg) + + if !ok { + if isTrace() { + log.Trace().Msgf("No public registration to %s", msg.Topic) + log.Trace().Msgf("Public topics: %v", h.PublicTopics) + } + } + + case "private": + uid := msg.Stream + uTopic, ok := h.PrivateTopics[uid] + if ok { + topic, ok := uTopic[msg.Topic] + if ok { + topic.broadcast(msg) + break + } + } + if isTrace() { + log.Trace().Msgf("No private registration to %s", msg.Topic) + log.Trace().Msgf("Private topics: %v", h.PrivateTopics) + } + + default: + scope, ok := h.PrefixedTopics[msg.Scope] + if !ok { + return + } + + topic, ok := scope[msg.Topic] + if !ok { + return + } + + topic.broadcast(msg) + + log.Trace().Msgf("Broadcasted message scope %s", msg.Scope) + } + +} + +func (h *Hub) unsubscribeAll(client IClient) { + h.mutex.Lock() + defer h.mutex.Unlock() + + for t, topic := range h.PublicTopics { + if topic.unsubscribe(client) { + metrics.RecordHubUnsubscription("public", t) + } + if topic.len() == 0 { + delete(h.PublicTopics, t) + } + } + + for k, scope := range h.PrefixedTopics { + for t, topic := range scope { + if topic.unsubscribe(client) { + metrics.RecordHubUnsubscription("prefixed", t) + } + + if topic.len() == 0 { + delete(scope, t) + } + } + + if len(scope) == 0 { + delete(h.PrefixedTopics, k) + } + } + + uid := client.GetAuth().UID + topics, ok := h.PrivateTopics[uid] + if !ok { + return + } + + for t, topic := range topics { + if topic.unsubscribe(client) { + metrics.RecordHubUnsubscription("private", t) + } + if topic.len() == 0 { + delete(topics, t) + } + } + + if len(topics) == 0 { + delete(h.PrivateTopics, uid) + } + +} + +func responseMust(e error, r interface{}) string { + res, err := msg.PackOutgoingResponse(e, r) + if err != nil { + log.Panic().Msg("responseMust failed:" + err.Error()) + panic(err.Error()) + } + + return string(res) +} + +func isPrivateStream(s string) bool { + return strings.Count(s, ".") == 0 +} +func isPrefixedStream(s string) bool { + return strings.Count(s, ".") == 2 +} + +func (h *Hub) handleRequest(req *Request) { + switch req.Method { + case "subscribe": + h.handleSubscribe(req) + case "unsubscribe": + h.handleUnsubscribe(req) + default: + req.client.Send(responseMust(errors.New("unsupported method"), nil)) + } +} + +func (h *Hub) subscribePrivate(t string, req *Request) { + uid := req.client.GetAuth().UID + if uid == "" { + log.Error().Msgf("Anonymous user tried to subscribe to private stream %s", t) + return + } + + uTopics, ok := h.PrivateTopics[uid] + if !ok { + uTopics = make(map[string]*Topic, 3) + h.PrivateTopics[uid] = uTopics + } + + topic, ok := uTopics[t] + if !ok { + topic = NewTopic(h) + uTopics[t] = topic + } + + if topic.subscribe(req.client) { + metrics.RecordHubSubscription("private", t) + req.client.SubscribePrivate(t) + } +} + +func (h *Hub) subscribePublic(t string, req *Request) { + topic, ok := h.PublicTopics[t] + if !ok { + topic = NewTopic(h) + h.PublicTopics[t] = topic + } + + // Replay snapshot and buffered increments before joining the topic, so live + // broadcasts cannot interleave ahead of the initial ob-snap on market switch. + if isIncrementObject(t) { + o, ok := h.IncrementalObjects[t] + if ok && o.Snapshot != "" { + req.client.Send(o.Snapshot) + for _, inc := range o.Increments { + req.client.Send(inc) + } + } + } + + if topic.subscribe(req.client) { + metrics.RecordHubSubscription("public", t) + req.client.SubscribePublic(t) + } +} + +func (h *Hub) premittedRBAC(prefix string, auth Auth) bool { + rbac := h.RBAC[prefix] + + for _, role := range rbac { + if role == auth.Role { + return true + } + } + + return false +} + +func splitPrefixedTopic(prefixed string) (string, string) { + spl := strings.Split(prefixed, ".") + prefix := spl[0] + t := strings.TrimPrefix(prefixed, prefix+".") + + return prefix, t +} + +func (h *Hub) subscribePrefixed(prefixed string, req *Request) { + prefix, t := splitPrefixedTopic(prefixed) + + if !h.premittedRBAC(prefix, req.client.GetAuth()) { + req.client.Send(responseMust(nil, map[string]interface{}{ + "message": "cannot subscribe to " + prefixed, + })) + + return + } + + topics, ok := h.PrefixedTopics[prefix] + if !ok { + topics := make(map[string]*Topic, 0) + h.PrefixedTopics[prefix] = topics + } + + topic, ok := topics[t] + if !ok { + topic = NewTopic(h) + h.PrefixedTopics[prefix][t] = topic + } + + if isIncrementObject(t) { + o, ok := h.IncrementalObjects[t] + if ok && o.Snapshot != "" { + req.client.Send(o.Snapshot) + for _, inc := range o.Increments { + req.client.Send(inc) + } + } + } + + if topic.subscribe(req.client) { + metrics.RecordHubSubscription("prefixed", prefixed) + req.client.SubscribePublic(prefixed) + } +} + +func (h *Hub) handleSubscribe(req *Request) { + h.mutex.Lock() + defer h.mutex.Unlock() + + for _, t := range req.Streams { + switch { + case isPrivateStream(t): + h.subscribePrivate(t, req) + case isPrefixedStream(t): + h.subscribePrefixed(t, req) + default: + h.subscribePublic(t, req) + } + } + + req.client.Send(responseMust(nil, map[string]interface{}{ + "message": "subscribed", + "streams": req.client.GetSubscriptions(), + })) +} + +func (h *Hub) unsubscribePrivate(t string, req *Request) { + uid := req.client.GetAuth().UID + if uid == "" { + return + } + uTopics, ok := h.PrivateTopics[uid] + if !ok { + return + } + + topic, ok := uTopics[t] + if ok { + if topic.unsubscribe(req.client) { + metrics.RecordHubUnsubscription("private", t) + req.client.UnsubscribePrivate(t) + } + + if topic.len() == 0 { + delete(uTopics, t) + } + } + + uTopics, ok = h.PrivateTopics[uid] + if ok && len(uTopics) == 0 { + delete(h.PrivateTopics, uid) + } +} + +func (h *Hub) unsubscribePrefixed(prefixed string, req *Request) { + scope, t := splitPrefixedTopic(prefixed) + topics, ok := h.PrefixedTopics[scope] + if !ok { + return + } + + topic, ok := topics[t] + if ok { + if topic.unsubscribe(req.client) { + metrics.RecordHubUnsubscription("prefixed", t) + req.client.UnsubscribePublic(t) + } + + if topic.len() == 0 { + delete(topics, t) + h.PrefixedTopics[scope] = topics + } + } +} + +func (h *Hub) unsubscribePublic(t string, req *Request) { + topic, ok := h.PublicTopics[t] + if ok { + if topic.unsubscribe(req.client) { + metrics.RecordHubUnsubscription("public", t) + req.client.UnsubscribePublic(t) + } + + if topic.len() == 0 { + delete(h.PublicTopics, t) + } + } +} + +func (h *Hub) handleUnsubscribe(req *Request) { + h.mutex.Lock() + defer h.mutex.Unlock() + + for _, t := range req.Streams { + switch { + case isPrivateStream(t): + h.unsubscribePrivate(t, req) + case isPrefixedStream(t): + h.unsubscribePrefixed(t, req) + default: + h.unsubscribePublic(t, req) + } + } + + req.client.Send(responseMust(nil, map[string]interface{}{ + "message": "unsubscribed", + "streams": req.client.GetSubscriptions(), + })) +} diff --git a/pkg/routing/hub_test.go b/pkg/routing/hub_test.go new file mode 100644 index 0000000..5c0264a --- /dev/null +++ b/pkg/routing/hub_test.go @@ -0,0 +1,355 @@ +package routing + +import ( + "testing" + + "github.com/openware/rango/pkg/message" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" +) + +type MockedClient struct { + mock.Mock +} + +func (c *MockedClient) Send(m string) { + c.Called(m) +} + +func (c *MockedClient) Close() { +} + +func (c *MockedClient) GetAuth() Auth { + args := c.Called() + return args.Get(0).(Auth) +} + +func (c *MockedClient) GetSubscriptions() []string { + args := c.Called() + return args.Get(0).([]string) +} + +func (c *MockedClient) SubscribePublic(s string) { + c.Called(s) +} + +func (c *MockedClient) SubscribePrivate(s string) { + c.Called(s) +} + +func (c *MockedClient) UnsubscribePublic(s string) { + c.Called(s) +} + +func (c *MockedClient) UnsubscribePrivate(s string) { + c.Called(s) +} + +func setup(c *MockedClient, streams []string) *Hub { + h := NewHub(nil) + h.handleSubscribe(&Request{ + client: c, + Request: message.Request{ + Streams: streams, + }, + }) + return h +} + +func teardown(h *Hub, c *MockedClient, streams []string) { + h.handleUnsubscribe(&Request{ + client: c, + Request: message.Request{ + Streams: streams, + }, + }) +} + +func TestAnonymous(t *testing.T) { + t.Run("subscribe to a public single stream", func(t *testing.T) { + c := &MockedClient{} + + streams := []string{ + "eurusd.trades", + } + + c.On("GetAuth").Return(Auth{}) + c.On("GetSubscriptions").Return(streams).Once() + c.On("SubscribePublic", streams[0]).Return().Once() + c.On("Send", `{"success":{"message":"subscribed","streams":["`+streams[0]+`"]}}`).Return() + + h := setup(c, streams) + assert.Equal(t, 1, len(h.PublicTopics)) + assert.Equal(t, 0, len(h.PrivateTopics)) + + c.On("UnsubscribePublic", streams[0]).Return() + c.On("GetSubscriptions").Return([]string{}).Once() + c.On("Send", `{"success":{"message":"unsubscribed","streams":[]}}`).Return() + + teardown(h, c, streams) + assert.Equal(t, 0, len(h.PublicTopics)) + assert.Equal(t, 0, len(h.PrivateTopics)) + }) + + t.Run("subscribe to multiple public streams", func(t *testing.T) { + c := &MockedClient{} + streams := []string{ + "eurusd.trades", + "eurusd.updates", + } + + c.On("GetAuth").Return(Auth{}) + c.On("GetSubscriptions").Return(streams).Once() + c.On("SubscribePublic", "eurusd.trades").Return() + c.On("SubscribePublic", "eurusd.updates").Return() + c.On("Send", `{"success":{"message":"subscribed","streams":["eurusd.trades","eurusd.updates"]}}`).Return() + + h := setup(c, []string{ + "eurusd.trades", + "eurusd.updates", + }) + + assert.Equal(t, 2, len(h.PublicTopics)) + assert.Equal(t, 0, len(h.PrivateTopics)) + + c.On("UnsubscribePublic", streams[0]).Return().Once() + c.On("UnsubscribePublic", streams[1]).Return().Once() + c.On("GetSubscriptions").Return([]string{}).Once() + c.On("Send", `{"success":{"message":"unsubscribed","streams":[]}}`).Return() + + teardown(h, c, streams) + assert.Equal(t, 0, len(h.PublicTopics)) + assert.Equal(t, 0, len(h.PrivateTopics)) + + }) + + t.Run("subscribe to a private single stream", func(t *testing.T) { + c := MockedClient{} + + c.On("GetAuth").Return(Auth{}) + c.On("GetSubscriptions").Return([]string{}) + c.On("SubscribePrivate", "trades").Return() + c.On("Send", `{"success":{"message":"subscribed","streams":[]}}`).Return() + + h := setup(&c, []string{ + "trades", + }) + + assert.Equal(t, 0, len(h.PublicTopics)) + assert.Equal(t, 0, len(h.PrivateTopics)) + }) +} +func TestAuthenticated(t *testing.T) { + t.Run("subscribe to a private single stream", func(t *testing.T) { + c := &MockedClient{} + + c.On("GetAuth").Return(Auth{UID: "UIDABC00001"}) + c.On("GetSubscriptions").Return([]string{"trades"}).Once() + c.On("SubscribePrivate", "trades").Return() + c.On("Send", `{"success":{"message":"subscribed","streams":["trades"]}}`).Return() + + h := setup(c, []string{ + "trades", + }) + assert.Equal(t, 0, len(h.PublicTopics)) + assert.Equal(t, 1, len(h.PrivateTopics)) + + c.On("UnsubscribePrivate", "trades").Return().Once() + c.On("GetSubscriptions").Return([]string{}).Once() + c.On("Send", `{"success":{"message":"unsubscribed","streams":[]}}`).Return() + + teardown(h, c, []string{"trades"}) + assert.Equal(t, 0, len(h.PublicTopics)) + assert.Equal(t, 0, len(h.PrivateTopics)) + }) + + t.Run("subscribe to multiple private streams", func(t *testing.T) { + c := &MockedClient{} + + c.On("GetSubscriptions").Return([]string{"trades", "orders"}).Once() + c.On("GetAuth").Return(Auth{UID: "UIDABC00001"}) + c.On("SubscribePrivate", "trades").Return() + c.On("SubscribePrivate", "orders").Return() + c.On("Send", `{"success":{"message":"subscribed","streams":["trades","orders"]}}`).Return() + + h := setup(c, []string{"trades", "orders"}) + assert.Equal(t, 0, len(h.PublicTopics)) + assert.Equal(t, 1, len(h.PrivateTopics)) + + uTopics, ok := h.PrivateTopics["UIDABC00001"] + require.True(t, ok) + assert.Equal(t, 2, len(uTopics)) + + c.On("UnsubscribePrivate", "trades").Return().Once() + c.On("UnsubscribePrivate", "orders").Return().Once() + c.On("GetSubscriptions").Return([]string{}).Once() + c.On("Send", `{"success":{"message":"unsubscribed","streams":[]}}`).Return() + + teardown(h, c, []string{"trades", "orders"}) + assert.Equal(t, 0, len(h.PublicTopics)) + assert.Equal(t, 0, len(h.PrivateTopics)) + + }) + + t.Run("subscribe to multiple private and public streams", func(t *testing.T) { + c := &MockedClient{} + + c.On("GetSubscriptions").Return([]string{"trades", "orders", "eurusd.updates"}).Once() + c.On("GetAuth").Return(Auth{UID: "UIDABC00001"}) + c.On("SubscribePrivate", "trades").Return() + c.On("SubscribePrivate", "orders").Return() + c.On("SubscribePublic", "eurusd.updates").Return() + c.On("Send", `{"success":{"message":"subscribed","streams":["trades","orders","eurusd.updates"]}}`).Return() + + h := setup(c, []string{"trades", "orders", "eurusd.updates"}) + assert.Equal(t, 1, len(h.PublicTopics)) + assert.Equal(t, 1, len(h.PrivateTopics)) + + uTopics, ok := h.PrivateTopics["UIDABC00001"] + require.True(t, ok) + assert.Equal(t, 2, len(uTopics)) + + c.On("UnsubscribePrivate", "trades").Return().Once() + c.On("UnsubscribePrivate", "orders").Return().Once() + c.On("UnsubscribePublic", "eurusd.updates").Return().Once() + c.On("GetSubscriptions").Return([]string{}).Once() + c.On("Send", `{"success":{"message":"unsubscribed","streams":[]}}`).Return() + + teardown(h, c, []string{"trades", "orders", "eurusd.updates"}) + assert.Equal(t, 0, len(h.PublicTopics)) + assert.Equal(t, 0, len(h.PrivateTopics)) + }) +} + +func TestIsIncremental(t *testing.T) { + assert.True(t, isIncrementObject("public.eurusd.ob-inc")) + assert.False(t, isIncrementObject("public.eurusd.ob-snap")) + assert.False(t, isIncrementObject("public.eurusd.ob")) + + assert.True(t, isSnapshotObject("public.eurusd.ob-snap")) + assert.False(t, isSnapshotObject("public.eurusd.ob-inc")) + assert.False(t, isSnapshotObject("public.eurusd.ob")) +} + +func TestGetTopic(t *testing.T) { + assert.Equal(t, "abc.count", getTopic("public", "abc", "count")) + assert.Equal(t, "count", getTopic("private", "abc", "count")) + assert.Equal(t, "abc.count-inc", getTopic("public", "abc", "count-inc")) + assert.Equal(t, "abc.count-inc", getTopic("public", "abc", "count-snap")) +} + +func TestHandleMessage(t *testing.T) { + h := NewHub(nil) + c := &MockedClient{} + c.On("SubscribePublic", "abc.ticker").Return() + c.On("Send", "{\"abc.ticker\":{\"some\":\"data\"}}").Return() + + h.subscribePublic("abc.ticker", &Request{ + client: c, + }) + + h.routeMessage(&Event{ + Scope: "public", + Stream: "abc", + Type: "ticker", + Topic: "abc.ticker", + Body: map[string]interface{}{ + "some": "data", + }, + }) + + c.AssertExpectations(t) +} + +func TestIncrementalObjectStorage(t *testing.T) { + h := NewHub(nil) + + // Increments before the first snapshot must be ignored + h.routeMessage(&Event{ + Scope: "public", + Stream: "abc", + Type: "count-inc", + Topic: "abc.count-inc", + Body: map[string]interface{}{ + "data": 1, + "sequence": 11, + }, + }) + + require.Equal(t, 0, len(h.IncrementalObjects)) + + // Initial snapshot + h.routeMessage(&Event{ + Scope: "public", + Stream: "abc", + Type: "count-snap", + Topic: "abc.count-inc", + Body: map[string]interface{}{ + "data": []int{2, 3, 4}, + "sequence": 12, + }, + }) + + require.Equal(t, 1, len(h.IncrementalObjects)) + + o, ok := h.IncrementalObjects["abc.count-inc"] + require.True(t, ok) + require.Equal(t, 0, len(o.Increments)) + require.Equal(t, `{"abc.count-snap":{"data":[2,3,4],"sequence":12}}`, o.Snapshot) + + // First Increment + h.routeMessage(&Event{ + Scope: "public", + Stream: "abc", + Type: "count-inc", + Topic: "abc.count-inc", + Body: map[string]interface{}{ + "data": 5, + "sequence": 13, + }, + }) + require.Equal(t, 1, len(h.IncrementalObjects)) + o, ok = h.IncrementalObjects["abc.count-inc"] + require.True(t, ok) + require.Equal(t, 1, len(o.Increments)) + require.Equal(t, `{"abc.count-snap":{"data":[2,3,4],"sequence":12}}`, o.Snapshot) + require.Equal(t, `{"abc.count-inc":{"data":5,"sequence":13}}`, o.Increments[0]) + + // Second Increment + h.routeMessage(&Event{ + Scope: "public", + Stream: "abc", + Type: "count-inc", + Topic: "abc.count-inc", + Body: map[string]interface{}{ + "data": 6, + "sequence": 14, + }, + }) + require.Equal(t, 1, len(h.IncrementalObjects)) + o, ok = h.IncrementalObjects["abc.count-inc"] + require.True(t, ok) + require.Equal(t, 2, len(o.Increments)) + require.Equal(t, `{"abc.count-snap":{"data":[2,3,4],"sequence":12}}`, o.Snapshot) + require.Equal(t, `{"abc.count-inc":{"data":5,"sequence":13}}`, o.Increments[0]) + require.Equal(t, `{"abc.count-inc":{"data":6,"sequence":14}}`, o.Increments[1]) + + // Second snapshot + h.routeMessage(&Event{ + Scope: "public", + Stream: "abc", + Type: "count-snap", + Topic: "abc.count-inc", + Body: map[string]interface{}{ + "data": []int{2, 3, 4, 5, 6}, + "sequence": 14, + }, + }) + + require.Equal(t, 1, len(h.IncrementalObjects)) + o, ok = h.IncrementalObjects["abc.count-inc"] + require.True(t, ok) + require.Equal(t, 0, len(o.Increments)) + require.Equal(t, `{"abc.count-snap":{"data":[2,3,4,5,6],"sequence":14}}`, o.Snapshot) +} diff --git a/pkg/routing/topic.go b/pkg/routing/topic.go new file mode 100644 index 0000000..fee1014 --- /dev/null +++ b/pkg/routing/topic.go @@ -0,0 +1,79 @@ +package routing + +import ( + "encoding/json" + + msg "github.com/openware/rango/pkg/message" + "github.com/rs/zerolog/log" +) + +type Topic struct { + hub *Hub + clients map[IClient]struct{} +} + +func NewTopic(h *Hub) *Topic { + return &Topic{ + clients: make(map[IClient]struct{}), + hub: h, + } +} + +func eventMust(method string, data interface{}) []byte { + ev, err := msg.PackOutgoingEvent(method, data) + if err != nil { + log.Panic().Msg(err.Error()) + } + + return ev +} + +func contains(list []string, el string) bool { + for _, l := range list { + if l == el { + return true + } + } + return false +} + +func (t *Topic) len() int { + return len(t.clients) +} + +func (t *Topic) broadcast(message *Event) { + body, err := json.Marshal(map[string]interface{}{ + message.Topic: message.Body, + }) + + if err != nil { + log.Error().Msgf("Fail to JSON marshal: %s", err.Error()) + return + } + + for client := range t.clients { + client.Send(string(body)) + } +} + +func (t *Topic) broadcastRaw(msgBody string) { + for client := range t.clients { + client.Send(msgBody) + } +} + +func (t *Topic) subscribe(c IClient) bool { + if _, ok := t.clients[c]; ok { + return false + } + t.clients[c] = struct{}{} + + return true +} + +func (t *Topic) unsubscribe(c IClient) bool { + _, ok := t.clients[c] + delete(t.clients, c) + + return ok +} diff --git a/tools/README.md b/tools/README.md new file mode 100644 index 0000000..caa3624 --- /dev/null +++ b/tools/README.md @@ -0,0 +1,5 @@ +## Connect to private channel + +```bash +wscat --connect localhost:8080/private --header "Authorization: Bearer $(go run ./tools/jwt)" +``` diff --git a/tools/inject-msg/inject-msg.go b/tools/inject-msg/inject-msg.go new file mode 100644 index 0000000..09d1c02 --- /dev/null +++ b/tools/inject-msg/inject-msg.go @@ -0,0 +1,56 @@ +package main + +import ( + "bufio" + "flag" + "fmt" + "os" + "strings" + "time" + + "github.com/openware/rango/pkg/amqp" + "github.com/rs/zerolog/log" +) + +var ( + ex = flag.String("exchange", "peatio.events.ranger", "Exchange name of upstream messages") + amqpAddr = flag.String("amqp-addr", "amqp://localhost:5672", "AMQP server address") + wait = flag.Float64("wait", 2, "Time to wait between submit batch of messages") +) + +func main() { + flag.Parse() + + mq, err := amqp.NewAMQPSession(*amqpAddr) + + if err != nil { + log.Error().Msg(err.Error()) + return + } + + for { + file, err := os.Open("msg.txt") + if err != nil { + panic(err.Error()) + } + scanner := bufio.NewScanner(file) + for scanner.Scan() { + if err := scanner.Err(); err != nil { + panic(err.Error()) + } + + msg := strings.Split(scanner.Text(), " ") + + log.Info().Msgf("Pushing to %s", msg[0]) + + if err := mq.Push(*ex, msg[0], []byte(msg[1])); err != nil { + fmt.Printf("Push failed: %s\n", err) + } else { + log.Info().Msgf("Pushed to %s", msg[0]) + } + } + file.Close() + log.Info().Msgf("Waiting %f seconds", *wait) + time.Sleep(time.Duration(float64(time.Second) * *wait)) + } +} diff --git a/tools/jwt/jwt.go b/tools/jwt/jwt.go new file mode 100644 index 0000000..ef0aa24 --- /dev/null +++ b/tools/jwt/jwt.go @@ -0,0 +1,25 @@ +package main + +import ( + "flag" + "fmt" + + "github.com/openware/pkg/jwt" +) + +var ( + port = flag.String("port", "7070", "Port to bind") + uid = flag.String("uid", "IDABC0000001", "UID") + email = flag.String("email", "admin@barong.io", "Email") + role = flag.String("role", "admin", "Role") + level = flag.Int("level", 3, "Level") + ref = flag.Int("ref", 1, "Referral id") +) + +func main() { + flag.Parse() + + ks, _ := jwt.LoadOrGenerateKeysEdDSA("config/ed25519-key", "config/ed25519-key.pub") + t, _ := jwt.ForgeTokenEdDSA(*uid, *email, *role, *level, *ref, ks.PrivateKey, nil) + fmt.Print(t) +} diff --git a/tools/ws-client/ws-client.go b/tools/ws-client/ws-client.go new file mode 100644 index 0000000..3f77a7a --- /dev/null +++ b/tools/ws-client/ws-client.go @@ -0,0 +1,87 @@ +package main + +import ( + "flag" + "log" + "net/url" + "os" + "os/signal" + "time" + + "github.com/gorilla/websocket" +) + +var ( + addr = flag.String("addr", "localhost:8080", "http service address") + streams = flag.String("streams", "eurusd.ob-inc,eurusd.kline-12h", "streams to connect to") + wait = flag.Float64("wait", 2, "Time to wait between submit batch of messages") +) + +func main() { + flag.Parse() + interrupt := make(chan os.Signal, 1) + signal.Notify(interrupt, os.Interrupt) + + u := url.URL{ + Scheme: "ws", + Host: *addr, + Path: "", + RawQuery: "stream=" + *streams, + } + log.Printf("connecting to %s", u.String()) + + dialer := websocket.DefaultDialer + dialer.ReadBufferSize = 1 + dialer.WriteBufferSize = 1 + + c, _, err := dialer.Dial(u.String(), nil) + if err != nil { + log.Fatal("dial:", err) + } + defer c.Close() + + done := make(chan bool) + + go func() { + defer close(done) + for { + _, message, err := c.ReadMessage() + if err != nil { + log.Println("read:", err) + return + } + log.Printf("recv: %s", message) + } + }() + + ticker := time.NewTicker(time.Second) + defer ticker.Stop() + + for { + select { + case <-done: + return + // case t := <-ticker.C: + // err := c.WriteMessage(websocket.TextMessage, []byte(t.String())) + // if err != nil { + // log.Println("write:", err) + // return + // } + case <-interrupt: + log.Println("interrupt") + + // Cleanly close the connection by sending a close message and then + // waiting (with timeout) for the server to close the connection. + err := c.WriteMessage(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseNormalClosure, "")) + if err != nil { + log.Println("write close:", err) + return + } + select { + case <-done: + case <-time.After(time.Second): + } + return + } + } +}