Skip to content

Commit ff98e50

Browse files
arkady-emelyanovarkady-emelyanov
committed
KIP-554: Add Broker-side SCRAM Config API
https://cwiki.apache.org/confluence/display/KAFKA/KIP-554%3A+Add+Broker-side+SCRAM+Config+API
1 parent 9d205e2 commit ff98e50

19 files changed

+936
-0
lines changed

‎admin.go

+67
Original file line numberDiff line numberDiff line change
@@ -104,6 +104,15 @@ type ClusterAdmin interface {
104104
// Get information about all log directories on the given set of brokers
105105
DescribeLogDirs(brokers []int32) (map[int32][]DescribeLogDirsResponseDirMetadata, error)
106106

107+
// Get information about SCRAM users
108+
DescribeUserScramCredentials(users []string) ([]*DescribeUserScramCredentialsResult, error)
109+
110+
// Delete SCRAM users
111+
DeleteUserScramCredentials(delete []AlterUserScramCredentialsDelete) ([]*AlterUserScramCredentialsResult, error)
112+
113+
// Upsert SCRAM users
114+
UpsertUserScramCredentials(upsert []AlterUserScramCredentialsUpsert) ([]*AlterUserScramCredentialsResult, error)
115+
107116
// Close shuts down the admin and closes underlying client.
108117
Close() error
109118
}
@@ -936,3 +945,61 @@ func (ca *clusterAdmin) DescribeLogDirs(brokerIds []int32) (allLogDirs map[int32
936945
err = <-errChan
937946
return
938947
}
948+
949+
func (ca *clusterAdmin) DescribeUserScramCredentials(users []string) ([]*DescribeUserScramCredentialsResult, error) {
950+
req := &DescribeUserScramCredentialsRequest{}
951+
for _, u := range users {
952+
req.DescribeUsers = append(req.DescribeUsers, DescribeUserScramCredentialsRequestUser{
953+
Name: u,
954+
})
955+
}
956+
957+
b, err := ca.Controller()
958+
if err != nil {
959+
return nil, err
960+
}
961+
962+
rsp, err := b.DescribeUserScramCredentials(req)
963+
if err != nil {
964+
return nil, err
965+
}
966+
967+
return rsp.Results, nil
968+
}
969+
970+
func (ca *clusterAdmin) UpsertUserScramCredentials(upsert []AlterUserScramCredentialsUpsert) ([]*AlterUserScramCredentialsResult, error) {
971+
res, err := ca.AlterUserScramCredentials(upsert, nil)
972+
if err != nil {
973+
return nil, err
974+
}
975+
976+
return res, nil
977+
}
978+
979+
func (ca *clusterAdmin) DeleteUserScramCredentials(delete []AlterUserScramCredentialsDelete) ([]*AlterUserScramCredentialsResult, error) {
980+
res, err := ca.AlterUserScramCredentials(nil, delete)
981+
if err != nil {
982+
return nil, err
983+
}
984+
985+
return res, nil
986+
}
987+
988+
func (ca *clusterAdmin) AlterUserScramCredentials(u []AlterUserScramCredentialsUpsert, d []AlterUserScramCredentialsDelete) ([]*AlterUserScramCredentialsResult, error) {
989+
req := &AlterUserScramCredentialsRequest{
990+
Deletions: d,
991+
Upsertions: u,
992+
}
993+
994+
b, err := ca.Controller()
995+
if err != nil {
996+
return nil, err
997+
}
998+
999+
rsp, err := b.AlterUserScramCredentials(req)
1000+
if err != nil {
1001+
return nil, err
1002+
}
1003+
1004+
return rsp.Results, nil
1005+
}

‎alter_user_scram_credentials_request.go

+142
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,142 @@
1+
package sarama
2+
3+
type AlterUserScramCredentialsRequest struct {
4+
Version int16
5+
6+
// Deletions represent list of SCRAM credentials to remove
7+
Deletions []AlterUserScramCredentialsDelete
8+
9+
// Upsertions represent list of SCRAM credentials to update/insert
10+
Upsertions []AlterUserScramCredentialsUpsert
11+
}
12+
13+
type AlterUserScramCredentialsDelete struct {
14+
Name string
15+
Mechanism ScramMechanismType
16+
}
17+
18+
type AlterUserScramCredentialsUpsert struct {
19+
Name string
20+
Mechanism ScramMechanismType
21+
Iterations int32
22+
Salt []byte
23+
saltedPassword []byte
24+
25+
// This field is never transmitted over the wire
26+
// @see: https://tools.ietf.org/html/rfc5802
27+
Password []byte
28+
}
29+
30+
func (r *AlterUserScramCredentialsRequest) encode(pe packetEncoder) error {
31+
pe.putCompactArrayLength(len(r.Deletions))
32+
for _, d := range r.Deletions {
33+
if err := pe.putCompactString(d.Name); err != nil {
34+
return err
35+
}
36+
pe.putInt8(int8(d.Mechanism))
37+
pe.putEmptyTaggedFieldArray()
38+
}
39+
40+
pe.putCompactArrayLength(len(r.Upsertions))
41+
for _, u := range r.Upsertions {
42+
if err := pe.putCompactString(u.Name); err != nil {
43+
return err
44+
}
45+
pe.putInt8(int8(u.Mechanism))
46+
pe.putInt32(u.Iterations)
47+
48+
if err := pe.putCompactBytes(u.Salt); err != nil {
49+
return err
50+
}
51+
52+
// do not transmit the password over the wire
53+
formatter := scramFormatter{mechanism: u.Mechanism}
54+
salted, err := formatter.saltedPassword(u.Password, u.Salt, int(u.Iterations))
55+
if err != nil {
56+
return err
57+
}
58+
59+
if err := pe.putCompactBytes(salted); err != nil {
60+
return err
61+
}
62+
pe.putEmptyTaggedFieldArray()
63+
}
64+
65+
pe.putEmptyTaggedFieldArray()
66+
return nil
67+
}
68+
69+
func (r *AlterUserScramCredentialsRequest) decode(pd packetDecoder, version int16) error {
70+
numDeletions, err := pd.getCompactArrayLength()
71+
if err != nil {
72+
return err
73+
}
74+
75+
r.Deletions = make([]AlterUserScramCredentialsDelete, numDeletions)
76+
for i := 0; i < numDeletions; i++ {
77+
r.Deletions[i] = AlterUserScramCredentialsDelete{}
78+
if r.Deletions[i].Name, err = pd.getCompactString(); err != nil {
79+
return err
80+
}
81+
mechanism, err := pd.getInt8()
82+
if err != nil {
83+
return err
84+
}
85+
r.Deletions[i].Mechanism = ScramMechanismType(mechanism)
86+
if _, err = pd.getEmptyTaggedFieldArray(); err != nil {
87+
return err
88+
}
89+
}
90+
91+
numUpsertions, err := pd.getCompactArrayLength()
92+
if err != nil {
93+
return err
94+
}
95+
96+
r.Upsertions = make([]AlterUserScramCredentialsUpsert, numUpsertions)
97+
for i := 0; i < numUpsertions; i++ {
98+
r.Upsertions[i] = AlterUserScramCredentialsUpsert{}
99+
if r.Upsertions[i].Name, err = pd.getCompactString(); err != nil {
100+
return err
101+
}
102+
mechanism, err := pd.getInt8()
103+
if err != nil {
104+
return err
105+
}
106+
107+
r.Upsertions[i].Mechanism = ScramMechanismType(mechanism)
108+
if r.Upsertions[i].Iterations, err = pd.getInt32(); err != nil {
109+
return err
110+
}
111+
if r.Upsertions[i].Salt, err = pd.getCompactBytes(); err != nil {
112+
return err
113+
}
114+
if r.Upsertions[i].saltedPassword, err = pd.getCompactBytes(); err != nil {
115+
return err
116+
}
117+
if _, err = pd.getEmptyTaggedFieldArray(); err != nil {
118+
return err
119+
}
120+
}
121+
122+
if _, err = pd.getEmptyTaggedFieldArray(); err != nil {
123+
return err
124+
}
125+
return nil
126+
}
127+
128+
func (r *AlterUserScramCredentialsRequest) key() int16 {
129+
return 51
130+
}
131+
132+
func (r *AlterUserScramCredentialsRequest) version() int16 {
133+
return r.Version
134+
}
135+
136+
func (r *AlterUserScramCredentialsRequest) headerVersion() int16 {
137+
return 2
138+
}
139+
140+
func (r *AlterUserScramCredentialsRequest) requiredVersion() KafkaVersion {
141+
return V2_7_0_0
142+
}

‎alter_user_scram_credentials_request_test.go

+60
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,60 @@
1+
package sarama
2+
3+
import "testing"
4+
5+
var (
6+
emptyAlterUserScramCredentialsRequest = []byte{
7+
1, // Deletions
8+
1, // Upsertions
9+
0, // empty tagged fields
10+
}
11+
userAlterUserScramCredentialsRequest = []byte{
12+
2, // Deletions array, length 1
13+
7, // User name length 6
14+
'd', 'e', 'l', 'e', 't', 'e', // User name
15+
2, // SCRAM_SHA_512
16+
0, // empty tagged fields
17+
2, // Upsertions array, length 1
18+
7, // User name length 6
19+
'u', 'p', 's', 'e', 'r', 't',
20+
1, // SCRAM_SHA_256
21+
0, 0, 16, 0, // iterations: 4096
22+
// salt bytes:
23+
6, 119, 111, 114, 108, 100,
24+
// saltedPassword:
25+
33, 193, 85, 83, 3, 218, 48, 159, 107, 125, 30, 143,
26+
228, 86, 54, 191, 221, 220, 75, 245, 100, 5, 231,
27+
233, 78, 157, 21, 240, 231, 185, 203, 211, 128,
28+
0, // empty tagged fields
29+
0, // empty tagged fields
30+
}
31+
)
32+
33+
func TestAlterUserScramCredentialsRequest(t *testing.T) {
34+
request := &AlterUserScramCredentialsRequest{
35+
Version: 0,
36+
Deletions: []AlterUserScramCredentialsDelete{},
37+
Upsertions: []AlterUserScramCredentialsUpsert{},
38+
}
39+
40+
// Password is not transmitted, will fail with `testRequest` and `DeepEqual` check
41+
testRequestEncode(t, "no upsertions/deletions", request, emptyAlterUserScramCredentialsRequest)
42+
43+
request.Deletions = []AlterUserScramCredentialsDelete{
44+
{
45+
Name: "delete",
46+
Mechanism: SCRAM_MECHANISM_SHA_512,
47+
},
48+
}
49+
request.Upsertions = []AlterUserScramCredentialsUpsert{
50+
{
51+
Name: "upsert",
52+
Mechanism: SCRAM_MECHANISM_SHA_256,
53+
Iterations: 4096,
54+
Salt: []byte("world"),
55+
Password: []byte("hello"),
56+
},
57+
}
58+
// Password is not transmitted, will fail with `testRequest` and `DeepEqual` check
59+
testRequestEncode(t, "single deletion and upsertion", request, userAlterUserScramCredentialsRequest)
60+
}

‎alter_user_scram_credentials_response.go

+94
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,94 @@
1+
package sarama
2+
3+
import "time"
4+
5+
type AlterUserScramCredentialsResponse struct {
6+
Version int16
7+
8+
ThrottleTime time.Duration
9+
10+
Results []*AlterUserScramCredentialsResult
11+
}
12+
13+
type AlterUserScramCredentialsResult struct {
14+
User string
15+
16+
ErrorCode KError
17+
ErrorMessage *string
18+
}
19+
20+
func (r *AlterUserScramCredentialsResponse) encode(pe packetEncoder) error {
21+
pe.putInt32(int32(r.ThrottleTime / time.Millisecond))
22+
pe.putCompactArrayLength(len(r.Results))
23+
24+
for _, u := range r.Results {
25+
if err := pe.putCompactString(u.User); err != nil {
26+
return err
27+
}
28+
pe.putInt16(int16(u.ErrorCode))
29+
if err := pe.putNullableCompactString(u.ErrorMessage); err != nil {
30+
return err
31+
}
32+
pe.putEmptyTaggedFieldArray()
33+
}
34+
35+
pe.putEmptyTaggedFieldArray()
36+
return nil
37+
}
38+
39+
func (r *AlterUserScramCredentialsResponse) decode(pd packetDecoder, version int16) error {
40+
throttleTime, err := pd.getInt32()
41+
if err != nil {
42+
return err
43+
}
44+
r.ThrottleTime = time.Duration(throttleTime) * time.Millisecond
45+
46+
numResults, err := pd.getCompactArrayLength()
47+
if err != nil {
48+
return err
49+
}
50+
51+
if numResults > 0 {
52+
r.Results = make([]*AlterUserScramCredentialsResult, numResults)
53+
for i := 0; i < numResults; i++ {
54+
r.Results[i] = &AlterUserScramCredentialsResult{}
55+
if r.Results[i].User, err = pd.getCompactString(); err != nil {
56+
return err
57+
}
58+
59+
kerr, err := pd.getInt16()
60+
if err != nil {
61+
return err
62+
}
63+
64+
r.Results[i].ErrorCode = KError(kerr)
65+
if r.Results[i].ErrorMessage, err = pd.getCompactNullableString(); err != nil {
66+
return err
67+
}
68+
if _, err := pd.getEmptyTaggedFieldArray(); err != nil {
69+
return err
70+
}
71+
}
72+
}
73+
74+
if _, err := pd.getEmptyTaggedFieldArray(); err != nil {
75+
return err
76+
}
77+
return nil
78+
}
79+
80+
func (r *AlterUserScramCredentialsResponse) key() int16 {
81+
return 51
82+
}
83+
84+
func (r *AlterUserScramCredentialsResponse) version() int16 {
85+
return r.Version
86+
}
87+
88+
func (r *AlterUserScramCredentialsResponse) headerVersion() int16 {
89+
return 2
90+
}
91+
92+
func (r *AlterUserScramCredentialsResponse) requiredVersion() KafkaVersion {
93+
return V2_7_0_0
94+
}

0 commit comments

Comments
 (0)