FilesExpand file tree

master

/

directory.go

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

134 lines (122 loc) · 4.14 KB

master

/

directory.go

Copy path

Top

File metadata and controls

• Code
• Blame

134 lines (122 loc) · 4.14 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

50

51

52

53

54

55

56

57

58

59

60

61

62

63

64

65

66

67

68

69

70

71

72

73

74

75

76

77

78

79

80

81

82

83

84

85

86

87

88

89

90

91

92

93

94

95

96

97

98

99

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

package fleet

import (

"bytes"

"context"

"crypto/sha256"

"crypto/x509"

"encoding/base64"

"encoding/hex"

"fmt"

"log/slog"

"runtime/debug"

"time"

"github.com/BottleFmt/gobottle"

"github.com/KarpelesLab/jwt"

)

func (a *Agent) directoryThread() {

a.spot.WaitOnline(context.Background())

if a.directoryThreadStart() {

return

}

go func() {

for {

// wait

time.Sleep(15 * time.Minute)

// and retry

if a.directoryThreadStart() {

return

}

}

}()

}

func (a *Agent) directoryThreadStart() bool {

// this is run in its own gorouting after db is setup

defer func() {

// ensure this thread crashing doesn't take the whole process

if e := recover(); e != nil {

slog.Error(fmt.Sprintf("[fleet] directory thread panic'd, will retry later. Error: %s\n%s", e, debug.Stack()), "event", "fleet:directory:panic", "category", "go.panic")

}

}()

// attempt to load jwt

jwtData, err := a.dbFleetGet("internal_key:jwt")

if err != nil {

slog.Info(fmt.Sprintf("[fleet] failed to load jwt: %s (will retry soon)", err), "event", "fleet:directory:no_jwt")

// attempt to get issuer to give us a key

err = a.performSelfIdentificationAttempt()

if err == nil {

// re-attempt to get jwt

jwtData, err = a.dbFleetGet("internal_key:jwt")

}

if err != nil {

slog.Info(fmt.Sprintf("[fleet] failed to id: %s (will retry later)", err), "event", "fleet:directory:no_selfid")

return false

}

}

// decode jwt

jwtInfo, err := jwt.ParseString(string(jwtData))

if err != nil {

slog.Error(fmt.Sprintf("[fleet] failed to decode jwt: %s", err), "event", "fleet:directory:jwt_invalid")

return false

}

// our tokens have the actual key stored in kid

key, err := base64.RawURLEncoding.DecodeString(jwtInfo.GetKeyId())

if err != nil {

slog.Error(fmt.Sprintf("[fleet] failed to decode kid: %s", err), "event", "fleet:directory:jwt_kid_invalid")

return false

}

keyObj, err := x509.ParsePKIXPublicKey(key)

if err != nil {

slog.Error(fmt.Sprintf("[fleet] failed to parse jwt key: %s", err), "event", "fleet:directory:jwt_pkix_invalid")

slog.Info("[fleet] removing invalid jwt from database", "event", "fleet:directory:jwt_expunge")

a.dbFleetDel("internal_key:jwt")

return false

}

// keyObj is a *rsa.PublicKey, *dsa.PublicKey, *ecdsa.PublicKey, or ed25519.PublicKey

err = jwtInfo.Verify(jwt.VerifySignature(keyObj), jwt.VerifyTime(time.Now(), false))

if err != nil {

slog.Error(fmt.Sprintf("[fleet] failed to verify jwt: %s", err), "event", "fleet:jwt:verify_fail")

slog.Info("[fleet] removing invalid jwt from database", "event", "fleet:directory:jwt_expunge")

a.dbFleetDel("internal_key:jwt")

return false

}

err = a.doInit(jwtInfo)

if err != nil {

slog.Warn(fmt.Sprintf("[agent] failed to init agent: %s", err), "event", "fleet:directory:agent_init_fail")

}

sgr := jwtInfo.Payload().GetString("sgr") // Spot Group (sha256 hash as hex)

if sgr == "" {

slog.Error("[fleet] JWT missing SpotGroup", "event", "fleet:directory:jwt_sgr_missing")

slog.Info("[fleet] removing invalid jwt from database", "event", "fleet:directory:jwt_expunge")

a.dbFleetDel("internal_key:jwt")

return false

}

// new process, use spot instead of directory to find & talk with other peers

groupHash, err := hex.DecodeString(sgr)

if err != nil || len(groupHash) != 32 {

slog.Error(fmt.Sprintf("[fleet] bad groupHash value: %s", err), "event", "fleet:directory:jwt_sgr_decode_error")

slog.Info("[fleet] removing invalid jwt from database", "event", "fleet:directory:jwt_expunge")

a.dbFleetDel("internal_key:jwt")

return false

}

// let's make sure we're in the group

id := a.spot.IDCard()

var group []byte

found := false

for _, m := range id.Groups {

mh := sha256.Sum256(m.Key)

if bytes.Equal(mh[:], groupHash) {

group = m.Key

found = true

}

}

if !found {

selfId := base64.RawURLEncoding.EncodeToString(gobottle.Hash(id.Self, sha256.New))

groupId := base64.RawURLEncoding.EncodeToString(groupHash)

slog.Error(fmt.Sprintf("[fleet] unable to join group as not member of the group myself; self=%s group=%s", selfId, groupId), "event", "fleet:directory:group_join_fail")

return false

}

// new process

a.setGroup(group)

a.setStatus(1)

return true

}