Skip to content

Commit 06c3fc4

Browse files
idanasulin2706idanasulin2706
committed
bugfixes
1 parent fb1a5f2 commit 06c3fc4

File tree

10 files changed

+190
-55
lines changed

10 files changed

+190
-55
lines changed

‎conf/config.go‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@ type Configuration struct {
2222
ROOT_PASSWORD string
2323
DEBUG bool
2424
CLOUD_ENV bool
25+
REST_GW_UPDATES_SUBJ string
2526
}
2627

2728
func GetConfig() Configuration {

‎conf/config.json‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
{
2-
"VERSION": "1.2.0",
2+
"VERSION": "1.2.1",
33
"JWT_EXPIRES_IN_MINUTES": 15,
4-
"REFRESH_JWT_EXPIRES_IN_MINUTES": 300
4+
"REFRESH_JWT_EXPIRES_IN_MINUTES": 300,
5+
"REST_GW_UPDATES_SUBJ": "$memphis_restgw_updates"
56
}

‎go.mod‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,7 @@ require (
66
github.com/go-playground/validator/v10 v10.11.1
77
github.com/gofiber/fiber/v2 v2.48.0
88
github.com/golang-jwt/jwt/v4 v4.5.0
9-
github.com/memphisdev/memphis.go v1.1.1
9+
github.com/memphisdev/memphis.go v1.1.2
1010
github.com/nats-io/nats.go v1.25.0
1111
github.com/tkanos/gonfig v0.0.0-20210106201359-53e13348de2f
1212
)

‎go.sum‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -57,8 +57,8 @@ github.com/mattn/go-isatty v0.0.19 h1:JITubQf0MOLdlGRuRq+jtsDlekdYPia9ZFsB8h/APP
5757
github.com/mattn/go-isatty v0.0.19/go.mod h1:W+V8PltTTMOvKvAeJH7IuucS94S2C6jfK/D7dTCTo3Y=
5858
github.com/mattn/go-runewidth v0.0.15 h1:UNAjwbU9l54TA3KzvqLGxwWjHmMgBUVhBiTjelZgg3U=
5959
github.com/mattn/go-runewidth v0.0.15/go.mod h1:Jdepj2loyihRzMpdS35Xk/zdY8IAYHsh153qUoGf23w=
60-
github.com/memphisdev/memphis.go v1.1.1 h1:FsSU3+YaqsM1ni6A5am5jyb0CoKtSBhnDxgrGJ+4GbU=
61-
github.com/memphisdev/memphis.go v1.1.1/go.mod h1:M/VqaTMdzYS4UJgo0h7JtLQmQrdXz3lvAO/+LhhaGbY=
60+
github.com/memphisdev/memphis.go v1.1.2 h1:dWhPQSJ+NmnP3eSLPHOR+TZnI0AXbX1Mb/5kIpac/NM=
61+
github.com/memphisdev/memphis.go v1.1.2/go.mod h1:M/VqaTMdzYS4UJgo0h7JtLQmQrdXz3lvAO/+LhhaGbY=
6262
github.com/minio/highwayhash v1.0.2 h1:Aak5U0nElisjDCfPSG79Tgzkn2gl66NxOMspRrKnA/g=
6363
github.com/minio/highwayhash v1.0.2/go.mod h1:BQskDq+xkJ12lmlUUi7U0M5Swg3EWR+dLTk+kldvVxY=
6464
github.com/mitchellh/mapstructure v1.5.0 h1:jeMsZIYE/09sWLaz43PL7Gy6RuMjD2eJVyuac5Z2hdY=

‎handlers/auth.go‎

Lines changed: 102 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,10 @@
11
package handlers
22

33
import (
4+
"encoding/json"
45
"rest-gateway/conf"
56
"rest-gateway/logger"
7+
"rest-gateway/memphisSingleton"
68
"rest-gateway/models"
79
"rest-gateway/utils"
810
"strconv"
@@ -13,6 +15,7 @@ import (
1315
"github.com/gofiber/fiber/v2"
1416
"github.com/golang-jwt/jwt/v4"
1517
"github.com/memphisdev/memphis.go"
18+
"github.com/nats-io/nats.go"
1619
)
1720

1821
var configuration = conf.GetConfig()
@@ -82,6 +85,7 @@ func (ah AuthHandler) Authenticate(c *fiber.Ctx) error {
8285
accountId = accId
8386
}
8487
}
88+
8589
conn, err := Connect(body.Password, body.Username, body.ConnectionToken, accountId)
8690
if err != nil {
8791
errMsg := strings.ToLower(err.Error())
@@ -119,6 +123,43 @@ func (ah AuthHandler) Authenticate(c *fiber.Ctx) error {
119123
ConnectionsCacheLock.Lock()
120124
ConnectionsCache[accountIdStr][username] = Connection{Connection: conn, ExpirationTime: tokenExpiry}
121125
ConnectionsCacheLock.Unlock()
126+
127+
mc, err := memphisSingleton.GetMemphisConnection("", "", "") // already initialized on logger creation
128+
if err != nil {
129+
log.Errorf("Authenticate: %s", err.Error())
130+
return c.Status(fiber.StatusInternalServerError).JSON(fiber.Map{
131+
"message": "Server error",
132+
})
133+
}
134+
135+
update := models.RestGwUpdate{
136+
Type: "update_connection",
137+
Update: map[string]interface{}{
138+
"password": body.Password,
139+
"username": body.Username,
140+
"connection_token": body.ConnectionToken,
141+
"account_id": accountId,
142+
"token_expiry": tokenExpiry,
143+
},
144+
}
145+
146+
msg, err := json.Marshal(update)
147+
if err != nil {
148+
log.Errorf("Authenticate: %s", err.Error())
149+
return c.Status(fiber.StatusInternalServerError).JSON(fiber.Map{
150+
"message": "Server error",
151+
})
152+
}
153+
154+
// send to other rest GWs to update their cache
155+
err = mc.Publish(configuration.REST_GW_UPDATES_SUBJ, msg)
156+
if err != nil {
157+
log.Errorf("Authenticate: %s", err.Error())
158+
return c.Status(fiber.StatusInternalServerError).JSON(fiber.Map{
159+
"message": "Server error",
160+
})
161+
}
162+
122163
return c.Status(fiber.StatusOK).JSON(fiber.Map{
123164
"jwt": token,
124165
"expires_in": tokenExpiry * 60 * 1000,
@@ -242,7 +283,11 @@ func CleanConnectionsCache() {
242283
currentTime := time.Now()
243284
unixTimeNow := currentTime.Unix()
244285
conn := ConnectionsCache[t][u].Connection
245-
if unixTimeNow > int64(user.ExpirationTime) {
286+
if !conn.IsConnected() {
287+
ConnectionsCacheLock.Lock()
288+
delete(ConnectionsCache[t], u)
289+
ConnectionsCacheLock.Unlock()
290+
} else if unixTimeNow > int64(user.ExpirationTime) {
246291
conn.Close()
247292
ConnectionsCacheLock.Lock()
248293
delete(ConnectionsCache[t], u)
@@ -257,3 +302,59 @@ func CleanConnectionsCache() {
257302
}
258303
}
259304
}
305+
306+
func ListenForUpdates(log *logger.Logger) error {
307+
mc, err := memphisSingleton.GetMemphisConnection("", "", "") // already initialized on logger creation
308+
if err != nil {
309+
return err
310+
}
311+
312+
_, err = mc.Subscribe(configuration.REST_GW_UPDATES_SUBJ, func(msg *nats.Msg) {
313+
var update models.RestGwUpdate
314+
err := json.Unmarshal(msg.Data, &update)
315+
if err != nil {
316+
log.Errorf("update unmarshal error: %v\n", err.Error())
317+
return
318+
}
319+
320+
switch update.Type {
321+
case "update_connection":
322+
username := update.Update["username"].(string)
323+
accountId := int(update.Update["account_id"].(float64))
324+
username = strings.ToLower(username)
325+
accountIdStr := strconv.Itoa(accountId)
326+
327+
if ConnectionsCache[accountIdStr] != nil {
328+
_, exists := ConnectionsCache[accountIdStr][username]
329+
if exists {
330+
return // connection already exists, nothing to update
331+
}
332+
}
333+
334+
conn, err := Connect(update.Update["password"].(string), username, update.Update["connection_token"].(string), accountId)
335+
if err != nil {
336+
errMsg := strings.ToLower(err.Error())
337+
if strings.Contains(errMsg, ErrorMsgAuthorizationViolation) || strings.Contains(errMsg, "token") || strings.Contains(errMsg, ErrorMsgMissionAccountId) {
338+
return
339+
}
340+
341+
log.Errorf("ListenForUpdates: %s", err.Error())
342+
return
343+
}
344+
345+
if ConnectionsCache[accountIdStr] == nil {
346+
ConnectionsCacheLock.Lock()
347+
ConnectionsCache[accountIdStr] = make(map[string]Connection)
348+
ConnectionsCacheLock.Unlock()
349+
}
350+
351+
ConnectionsCacheLock.Lock()
352+
ConnectionsCache[accountIdStr][username] = Connection{Connection: conn, ExpirationTime: int64(update.Update["token_expiry"].(float64))}
353+
ConnectionsCacheLock.Unlock()
354+
}
355+
})
356+
if err != nil {
357+
return err
358+
}
359+
return nil
360+
}

‎logger/log.go‎

Lines changed: 3 additions & 47 deletions
Original file line numberDiff line numberDiff line change
@@ -1,14 +1,11 @@
11
package logger
22

33
import (
4-
"crypto/tls"
5-
"crypto/x509"
64
"fmt"
7-
"io/ioutil"
85
"log"
96
"os"
107
"rest-gateway/conf"
11-
"time"
8+
"rest-gateway/memphisSingleton"
129

1310
"github.com/gofiber/fiber/v2"
1411
"github.com/nats-io/nats.go"
@@ -62,48 +59,7 @@ func (sw streamWriter) Write(p []byte) (int, error) {
6259
}
6360

6461
func CreateLogger(hostname string, username string, creds string) (*Logger, error) {
65-
configuration := conf.GetConfig()
66-
var nc *nats.Conn
67-
var err error
68-
69-
natsOpts := nats.Options{
70-
Url: hostname + ":6666",
71-
AllowReconnect: true,
72-
MaxReconnect: 10,
73-
ReconnectWait: 3 * time.Second,
74-
Name: "MEMPHIS HTTP LOGGER",
75-
}
76-
77-
if configuration.USER_PASS_BASED_AUTH {
78-
natsOpts.Password = creds
79-
natsOpts.User = username
80-
} else {
81-
natsOpts.Token = username + "::" + creds
82-
}
83-
84-
if configuration.CLIENT_CERT_PATH != "" && configuration.CLIENT_KEY_PATH != "" && configuration.ROOT_CA_PATH != "" {
85-
cert, err := tls.LoadX509KeyPair(configuration.CLIENT_CERT_PATH, configuration.CLIENT_KEY_PATH)
86-
if err != nil {
87-
return nil, err
88-
}
89-
cert.Leaf, err = x509.ParseCertificate(cert.Certificate[0])
90-
if err != nil {
91-
return nil, err
92-
}
93-
TLSConfig := &tls.Config{MinVersion: tls.VersionTLS12}
94-
TLSConfig.Certificates = []tls.Certificate{cert}
95-
certs := x509.NewCertPool()
96-
97-
pemData, err := ioutil.ReadFile(configuration.ROOT_CA_PATH)
98-
if err != nil {
99-
return nil, err
100-
}
101-
certs.AppendCertsFromPEM(pemData)
102-
TLSConfig.RootCAs = certs
103-
natsOpts.TLSConfig = TLSConfig
104-
}
105-
106-
nc, err = natsOpts.Connect()
62+
mc, err := memphisSingleton.GetMemphisConnection(hostname, creds, username)
10763
if err != nil {
10864
return nil, err
10965
}
@@ -113,7 +69,7 @@ func CreateLogger(hostname string, username string, creds string) (*Logger, erro
11369
labelStart := len(pidPrefix) + 28
11470

11571
sw := streamWriter{
116-
nc: nc,
72+
nc: mc,
11773
labelStart: labelStart,
11874
}
11975

‎main.go‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,10 @@ func initializeLogger() *logger.Logger {
3535
func main() {
3636
configuration := conf.GetConfig()
3737
l := initializeLogger()
38+
err := handlers.ListenForUpdates(l)
39+
if err != nil {
40+
panic("Error while listening for updates - " + err.Error())
41+
}
3842
go handlers.CleanConnectionsCache()
3943
app := router.SetupRoutes(l)
4044
l.Noticef("Memphis REST gateway is up and running")

‎memphisSingleton/memphisSingleton.go‎

Lines changed: 67 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,67 @@
1+
package memphisSingleton
2+
3+
import (
4+
"crypto/tls"
5+
"crypto/x509"
6+
"os"
7+
"rest-gateway/conf"
8+
"time"
9+
10+
"github.com/nats-io/nats.go"
11+
)
12+
13+
var mc *nats.Conn
14+
15+
func GetMemphisConnection(hostname, creds, username string) (*nats.Conn, error) {
16+
if mc == nil {
17+
configuration := conf.GetConfig()
18+
var nc *nats.Conn
19+
var err error
20+
21+
natsOpts := nats.Options{
22+
Url: hostname + ":6666",
23+
AllowReconnect: true,
24+
MaxReconnect: 10,
25+
ReconnectWait: 3 * time.Second,
26+
Name: "MEMPHIS HTTP LOGGER",
27+
}
28+
29+
if configuration.USER_PASS_BASED_AUTH {
30+
natsOpts.Password = creds
31+
natsOpts.User = username
32+
} else {
33+
natsOpts.Token = username + "::" + creds
34+
}
35+
36+
if configuration.CLIENT_CERT_PATH != "" && configuration.CLIENT_KEY_PATH != "" && configuration.ROOT_CA_PATH != "" {
37+
cert, err := tls.LoadX509KeyPair(configuration.CLIENT_CERT_PATH, configuration.CLIENT_KEY_PATH)
38+
if err != nil {
39+
return nil, err
40+
}
41+
cert.Leaf, err = x509.ParseCertificate(cert.Certificate[0])
42+
if err != nil {
43+
return nil, err
44+
}
45+
TLSConfig := &tls.Config{MinVersion: tls.VersionTLS12}
46+
TLSConfig.Certificates = []tls.Certificate{cert}
47+
certs := x509.NewCertPool()
48+
49+
pemData, err := os.ReadFile(configuration.ROOT_CA_PATH)
50+
if err != nil {
51+
return nil, err
52+
}
53+
certs.AppendCertsFromPEM(pemData)
54+
TLSConfig.RootCAs = certs
55+
natsOpts.TLSConfig = TLSConfig
56+
}
57+
58+
nc, err = natsOpts.Connect()
59+
if err != nil {
60+
return nil, err
61+
}
62+
63+
mc = nc
64+
}
65+
66+
return mc, nil
67+
}

‎models/models.go‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1 +1,6 @@
1-
package models
1+
package models
2+
3+
type RestGwUpdate struct {
4+
Type string `json:"type"`
5+
Update map[string]interface{} `json:"update"`
6+
}

‎version.conf‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1 +1 @@
1-
1.2.0
1+
1.2.1

0 commit comments

Comments
 (0)