Skip to content

Commit e91862d

Browse files
abelanger5abelanger5
authored
refactor: make rbac package generic and reusable (#3581)
* refactor: make rbac package generic and reusable * remove sqlcv1 dep * chore: generate * revert to github actions * try to remove rampup test * revert GHA changes * fix: tests * add 5 second context to security check * disable security check for testing harness * fix: race conditions in msgqueue
1 parent f5a9dbd commit e91862d

10 files changed

Lines changed: 68 additions & 42 deletions

File tree

‎api/v1/server/authz/middleware.go‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,7 @@ import (
88
"github.com/rs/zerolog"
99

1010
"github.com/hatchet-dev/hatchet/api/v1/server/middleware"
11-
"github.com/hatchet-dev/hatchet/api/v1/server/rbac"
11+
"github.com/hatchet-dev/hatchet/pkg/auth/rbac"
1212
"github.com/hatchet-dev/hatchet/pkg/config/server"
1313
"github.com/hatchet-dev/hatchet/pkg/repository/sqlcv1"
1414
)
@@ -20,7 +20,7 @@ type AuthZ struct {
2020
}
2121

2222
func NewAuthZ(config *server.ServerConfig) (*AuthZ, error) {
23-
rbacAuthorizer, err := rbac.NewAuthorizer()
23+
rbacAuthorizer, err := newHatchetAuthorizer()
2424
if err != nil {
2525
return nil, err
2626
}
@@ -197,7 +197,7 @@ func (a *AuthZ) authorizeTenantOperations(tenantMemberRole sqlcv1.TenantMemberRo
197197
}
198198

199199
// at the moment, tenant members are only restricted from creating other tenant users.
200-
if !a.rbac.IsAuthorized(tenantMemberRole, r.OperationID) {
200+
if !a.rbac.IsAuthorized(string(tenantMemberRole), r.OperationID) {
201201
return echo.NewHTTPError(http.StatusUnauthorized, "Not authorized to perform this operation")
202202
}
203203

‎api/v1/server/authz/rbac.go‎

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,25 @@
1+
package authz
2+
3+
import (
4+
_ "embed"
5+
6+
"github.com/hatchet-dev/hatchet/api/v1/server/oas/gen"
7+
8+
"github.com/hatchet-dev/hatchet/pkg/auth/rbac"
9+
)
10+
11+
//go:embed rbac.yaml
12+
var yamlFile []byte
13+
14+
func newHatchetAuthorizer() (*rbac.Authorizer, error) {
15+
permMap, err := rbac.LoadPermissionMap(yamlFile)
16+
if err != nil {
17+
return nil, err
18+
}
19+
spec, err := gen.GetSwagger()
20+
if err != nil {
21+
return nil, err
22+
}
23+
24+
return rbac.NewAuthorizer(permMap, spec)
25+
}

api/v1/server/rbac/rbac.yaml renamed to api/v1/server/authz/rbac.yaml

File renamed without changes.

api/v1/server/rbac/rbac_test.go renamed to api/v1/server/authz/rbac_test.go

Lines changed: 9 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,10 @@
1-
package rbac
1+
package authz
22

33
import (
44
"testing"
55

66
"github.com/hatchet-dev/hatchet/api/v1/server/oas/gen"
7+
"github.com/hatchet-dev/hatchet/pkg/auth/rbac"
78
"github.com/hatchet-dev/hatchet/pkg/repository/sqlcv1"
89

910
"github.com/stretchr/testify/assert"
@@ -34,21 +35,21 @@ func operationIdsFromSpec() []string {
3435
}
3536

3637
func TestAuthorizeTenantOperations(t *testing.T) {
37-
r, err := NewAuthorizer()
38+
r, err := newHatchetAuthorizer()
3839
assert.Nil(t, err)
3940
allOperations := operationIdsFromSpec()
4041
for _, operationId := range allOperations {
41-
assert.Equal(t, r.IsAuthorized(sqlcv1.TenantMemberRoleADMIN, operationId), true)
42-
assert.Equal(t, r.IsAuthorized(sqlcv1.TenantMemberRoleOWNER, operationId), true)
43-
if OperationIn(operationId, adminAndOwnerOnly) {
44-
assert.Equal(t, r.IsAuthorized(sqlcv1.TenantMemberRoleMEMBER, operationId), false)
42+
assert.Equal(t, r.IsAuthorized(string(sqlcv1.TenantMemberRoleADMIN), operationId), true)
43+
assert.Equal(t, r.IsAuthorized(string(sqlcv1.TenantMemberRoleOWNER), operationId), true)
44+
if rbac.OperationIn(operationId, adminAndOwnerOnly) {
45+
assert.Equal(t, r.IsAuthorized(string(sqlcv1.TenantMemberRoleMEMBER), operationId), false)
4546
} else {
46-
assert.Equal(t, r.IsAuthorized(sqlcv1.TenantMemberRoleMEMBER, operationId), true)
47+
assert.Equal(t, r.IsAuthorized(string(sqlcv1.TenantMemberRoleMEMBER), operationId), true)
4748
}
4849
}
4950
}
5051

5152
func TestValidateSpec(t *testing.T) {
52-
_, err := NewAuthorizer()
53+
_, err := newHatchetAuthorizer()
5354
assert.Nil(t, err)
5455
}

‎internal/msgqueue/rabbitmq/rabbitmq.go‎

Lines changed: 5 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -754,13 +754,10 @@ func (t *MessageQueueImpl) subscribe(
754754
return err
755755
}
756756

757-
wg.Add(1) // we add an extra delta for the deliveries channel to be closed
758-
defer wg.Done()
759-
760757
for rabbitMsg := range deliveries {
761758
wg.Add(1)
762759

763-
go func(rabbitMsg amqp.Delivery) {
760+
go func(rabbitMsg amqp.Delivery, session int) {
764761
defer wg.Done()
765762

766763
msg := &msgqueue.Message{}
@@ -831,7 +828,7 @@ func (t *MessageQueueImpl) subscribe(
831828
}
832829
}
833830

834-
t.l.Debug().Msgf("(session: %d) got msg", sessionCount)
831+
t.l.Debug().Msgf("(session: %d) got msg", session)
835832

836833
if err := preAck(msg); err != nil {
837834
if isPermanentPreAckError(err) {
@@ -868,15 +865,17 @@ func (t *MessageQueueImpl) subscribe(
868865
t.l.Error().Msgf("error in post-ack: %v", err)
869866
return
870867
}
871-
}(rabbitMsg)
868+
}(rabbitMsg, sessionCount)
872869
}
873870

874871
t.l.Info().Msg("deliveries channel closed")
875872

876873
return nil
877874
}
878875

876+
wg.Add(1)
879877
go func() {
878+
defer wg.Done()
880879
retryCount := 0
881880
lastRetry := time.Now()
882881

‎internal/services/partition/partition.go‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -284,6 +284,8 @@ func (p *Partition) StartSchedulerPartition(ctx context.Context) (func() error,
284284

285285
p.schedulerCron.Start()
286286

287+
rebalanceInactiveSchedulerPartitions(ctx, p.l, p.repo) // nolint: errcheck
288+
287289
return cleanup, nil
288290
}
289291

api/v1/server/rbac/rbac.go renamed to pkg/auth/rbac/rbac.go

Lines changed: 9 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -1,17 +1,12 @@
11
package rbac
22

33
import (
4-
_ "embed"
54
"fmt"
65
"maps"
76
"strings"
87

98
"github.com/getkin/kin-openapi/openapi3"
109
"go.yaml.in/yaml/v3"
11-
12-
"github.com/hatchet-dev/hatchet/api/v1/server/oas/gen"
13-
14-
"github.com/hatchet-dev/hatchet/pkg/repository/sqlcv1"
1510
)
1611

1712
func OperationIn(operationId string, operationIds []string) bool {
@@ -28,16 +23,11 @@ type Authorizer struct {
2823
permissionMap PermissionMap
2924
}
3025

31-
func NewAuthorizer() (*Authorizer, error) {
32-
permMap, err := LoadYaml()
33-
if err != nil {
34-
return nil, err
35-
}
36-
spec, err := gen.GetSwagger()
37-
if err != nil {
38-
return nil, err
39-
}
40-
err = permMap.ValidateSpec(*spec)
26+
func NewAuthorizer(
27+
permMap *PermissionMap,
28+
spec *openapi3.T,
29+
) (*Authorizer, error) {
30+
err := permMap.ValidateSpec(*spec)
4131
if err != nil {
4232
return nil, err
4333
}
@@ -46,8 +36,8 @@ func NewAuthorizer() (*Authorizer, error) {
4636
}, nil
4737
}
4838

49-
func (a *Authorizer) IsAuthorized(role sqlcv1.TenantMemberRole, operation string) bool {
50-
return a.permissionMap.HasPermission(string(role), operation)
39+
func (a *Authorizer) IsAuthorized(role string, operation string) bool {
40+
return a.permissionMap.HasPermission(role, operation)
5141
}
5242

5343
type Role struct {
@@ -146,12 +136,9 @@ func (p *PermissionMap) ValidateSpec(spec openapi3.T) error {
146136
return nil
147137
}
148138

149-
//go:embed rbac.yaml
150-
var yamlFile []byte
151-
152-
func LoadYaml() (*PermissionMap, error) {
139+
func LoadPermissionMap(yamlBytes []byte) (*PermissionMap, error) {
153140
var yamlContents PermissionMap
154-
err := yaml.Unmarshal(yamlFile, &yamlContents)
141+
err := yaml.Unmarshal(yamlBytes, &yamlContents)
155142
if err != nil {
156143
return nil, err
157144
}

‎pkg/config/loader/loader.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -512,7 +512,7 @@ func createControllerLayer(dc *database.Layer, cf *server.ServerConfigFile, vers
512512
Version: version,
513513
}, dc.V1.SecurityCheck())
514514

515-
defer securityCheck.Check()
515+
go securityCheck.Check()
516516
}
517517

518518
var analyticsEmitter analytics.Analytics

‎pkg/security/security.go‎

Lines changed: 13 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,11 @@
11
package security
22

33
import (
4+
"context"
45
"fmt"
56
"io"
67
"net/http"
8+
"time"
79

810
v1 "github.com/hatchet-dev/hatchet/pkg/repository"
911

@@ -51,8 +53,17 @@ func (a DefaultSecurityCheck) Check() {
5153
return
5254
}
5355

54-
req := fmt.Sprintf("%s/check?version=%s&tag=%s", a.Endpoint, a.Version, ident)
55-
resp, err := http.Get(req) // #nosec
56+
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
57+
defer cancel()
58+
59+
reqURL := fmt.Sprintf("%s/check?version=%s&tag=%s", a.Endpoint, a.Version, ident)
60+
req, err := http.NewRequestWithContext(ctx, http.MethodGet, reqURL, nil)
61+
if err != nil {
62+
a.Logger.Debug().Msgf("Error creating security check request: %s", err)
63+
return
64+
}
65+
66+
resp, err := http.DefaultClient.Do(req) // #nosec
5667
if err != nil {
5768
a.Logger.Debug().Msgf("Error making request to security endpoint: %s", err)
5869
return

‎pkg/testing/harness/engine.go‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -147,6 +147,7 @@ func startEngine() func() {
147147
os.Setenv("SERVER_LOGGER_FORMAT", "console")
148148
os.Setenv("DATABASE_LOGGER_LEVEL", "error")
149149
os.Setenv("DATABASE_LOGGER_FORMAT", "console")
150+
os.Setenv("SERVER_SECURITY_CHECK_ENABLED", "false")
150151
os.Setenv("SERVER_ADDITIONAL_LOGGERS_QUEUE_LEVEL", "error")
151152
os.Setenv("SERVER_ADDITIONAL_LOGGERS_QUEUE_FORMAT", "console")
152153
os.Setenv("SERVER_ADDITIONAL_LOGGERS_PGXSTATS_LEVEL", "error")

0 commit comments

Comments
 (0)