diff --git a/go.mod b/go.mod index ece2c22f..6c7cab94 100644 --- a/go.mod +++ b/go.mod @@ -23,13 +23,13 @@ require ( github.com/ncruces/go-sqlite3 v0.29.0 github.com/ncruces/zenity v0.10.14 github.com/prometheus/client_golang v1.22.0 - github.com/stretchr/testify v1.10.0 github.com/vearutop/statigz v1.5.0 go.uber.org/multierr v1.11.0 go.uber.org/zap v1.27.0 golang.org/x/crypto v0.41.0 golang.org/x/net v0.43.0 golang.org/x/sync v0.17.0 + golang.org/x/sys v0.36.0 google.golang.org/genproto/googleapis/api v0.0.0-20250505200425-f936aa4a68b2 google.golang.org/grpc v1.72.0 google.golang.org/protobuf v1.36.6 @@ -40,7 +40,6 @@ require ( github.com/akavel/rsrc v0.10.2 // indirect github.com/beorn7/perks v1.0.1 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect - github.com/davecgh/go-spew v1.1.1 // indirect github.com/dchest/jsmin v1.0.0 // indirect github.com/fatih/color v1.18.0 // indirect github.com/getlantern/context v0.0.0-20220418194847-3d5e7a086201 // indirect @@ -59,7 +58,6 @@ require ( github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect github.com/ncruces/julianday v1.0.0 // indirect github.com/oxtoacart/bpool v0.0.0-20190530202638-03653db5a59c // indirect - github.com/pmezard/go-difflib v1.0.0 // indirect github.com/prometheus/client_model v0.6.2 // indirect github.com/prometheus/common v0.63.0 // indirect github.com/prometheus/procfs v0.16.1 // indirect @@ -70,15 +68,6 @@ require ( go.opentelemetry.io/otel/metric v1.35.0 // indirect go.opentelemetry.io/otel/trace v1.35.0 // indirect golang.org/x/image v0.27.0 // indirect - golang.org/x/sys v0.36.0 // indirect golang.org/x/text v0.29.0 // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20250505200425-f936aa4a68b2 // indirect -<<<<<<< HEAD -======= - gopkg.in/yaml.v3 v3.0.1 // indirect - modernc.org/libc v1.65.0 // indirect - modernc.org/mathutil v1.7.1 // indirect - modernc.org/memory v1.10.0 // indirect - modernc.org/sqlite v1.37.0 // indirect ->>>>>>> 9041d3c (improve sync api security by using 'Authorization' headers for initial key exchange) ) diff --git a/go.sum b/go.sum index a73fdcab..78f2904b 100644 --- a/go.sum +++ b/go.sum @@ -93,12 +93,8 @@ github.com/josephspurrier/goversioninfo v1.5.0/go.mod h1:6MoTvFZ6GKJkzcdLnU5T/RG github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo= github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ= github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo= -github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= -github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ= github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= -github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= -github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc= github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw= github.com/lxn/walk v0.0.0-20210112085537-c389da54e794/go.mod h1:E23UucZGqpuUANJooIbHWCufXvOcT6E7Stq81gU+CSQ= @@ -136,13 +132,6 @@ github.com/prometheus/procfs v0.16.1 h1:hZ15bTNuirocR6u0JZ6BAHHmwS1p8B4P6MRqxtzM github.com/prometheus/procfs v0.16.1/go.mod h1:teAbpZRB1iIAJYREa1LsoWUXykVXA1KlTmWl8x/U+Is= github.com/randall77/makefat v0.0.0-20210315173500-7ddd0e42c844 h1:GranzK4hv1/pqTIhMTXt2X8MmMOuH3hMeUR0o9SP5yc= github.com/randall77/makefat v0.0.0-20210315173500-7ddd0e42c844/go.mod h1:T1TLSfyWVBRXVGzWd0o9BI4kfoO9InEgfQe4NV3mLz8= -<<<<<<< HEAD -======= -github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec h1:W09IVJc94icq4NjY3clb7Lk8O1qJ8BdBEF8z0ibU0rE= -github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec/go.mod h1:qqbHyh8v60DhA7CoWK5oRCqLrMHRGoxYCSS9EjAz6Eo= -github.com/rogpeppe/go-internal v1.13.1 h1:KvO1DLK/DRN07sQ1LQKScxyZJuNnedQ5/wKSR38lUII= -github.com/rogpeppe/go-internal v1.13.1/go.mod h1:uMEvuHeurkdAXX61udpOXGD/AzZDWNMNyH2VO9fmH0o= ->>>>>>> 9041d3c (improve sync api security by using 'Authorization' headers for initial key exchange) github.com/skratchdot/open-golang v0.0.0-20200116055534-eef842397966/go.mod h1:sUM3LWHvSMaG192sy56D9F7CNvL7jUJVXoqM1QKLnog= github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw= @@ -235,8 +224,6 @@ google.golang.org/protobuf v1.36.6/go.mod h1:jduwjTPXsFjZGTmRluh+L6NjiWu7pchiJ2/ gopkg.in/Knetic/govaluate.v3 v3.0.0/go.mod h1:csKLBORsPbafmSCGTEh3U7Ozmsuq8ZSIlKk1bcqph0E= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= -gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= -gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= gopkg.in/natefinch/lumberjack.v2 v2.2.1 h1:bBRl1b0OH9s/DuPhuXpNl+VtCaJXFZ5/uEFST95x9zc= gopkg.in/natefinch/lumberjack.v2 v2.2.1/go.mod h1:YD8tP3GAjkrDg1eZH7EGmyESg/lsYskCTPBJVb9jqSc= gopkg.in/yaml.v2 v2.2.8/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= diff --git a/internal/api/syncapi/authentication.go b/internal/api/syncapi/authentication.go deleted file mode 100644 index 4109f354..00000000 --- a/internal/api/syncapi/authentication.go +++ /dev/null @@ -1,192 +0,0 @@ -package syncapi - -import ( - "context" - "encoding/base64" - "errors" - "fmt" - "net/http" - "slices" - "strings" - - v1 "github.com/garethgeorge/backrest/gen/go/v1" - "github.com/garethgeorge/backrest/internal/config" - "github.com/garethgeorge/backrest/internal/cryptoutil" - "go.uber.org/zap" - "google.golang.org/protobuf/proto" -) - -type peerContextKey struct{} -type peerPublicKeyContextKey struct{} - -func PeerFromContext(ctx context.Context) *v1.Multihost_Peer { - return ctx.Value(peerContextKey{}).(*v1.Multihost_Peer) -} - -func PeerPublicKeyFromContext(ctx context.Context) *cryptoutil.PublicKey { - return ctx.Value(peerPublicKeyContextKey{}).(*cryptoutil.PublicKey) -} - -func ContextWithPeer(ctx context.Context, peer *v1.Multihost_Peer, publicKey *cryptoutil.PublicKey) context.Context { - if peer == nil { - return ctx - } - ctx = context.WithValue(ctx, peerContextKey{}, peer) - ctx = context.WithValue(ctx, peerPublicKeyContextKey{}, publicKey) - return ctx -} - -// HTTP decorator for authentication middleware. -func AuthenticationMiddleware(configManager *config.ConfigManager, handler http.Handler) http.Handler { - return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { - authHeader := r.Header.Get("Authorization") - if authHeader == "" { - http.Error(w, "Unauthorized: missing authentication header", http.StatusUnauthorized) - return - } - - config, err := configManager.Get() - if err != nil { - http.Error(w, "Internal error", http.StatusInternalServerError) - return - } - authorizedClientPeers := config.GetMultihost().GetAuthorizedClients() - - peerKey, instanceID, err := verifyAuthenticationHeader(authHeader) - if err != nil { - http.Error(w, fmt.Sprintf("Unauthorized: %v", err), http.StatusUnauthorized) - return - } - - authorizedPeerIdx := slices.IndexFunc(authorizedClientPeers, func(peer *v1.Multihost_Peer) bool { - return peer.Keyid == peerKey.KeyID() - }) - if authorizedPeerIdx == -1 { - zap.S().Errorf("peer key %q is not listed in authorized clients", peerKey.KeyID()) - http.Error(w, fmt.Sprintf("Unauthorized: peer key %q is not listed in authorized clients", peerKey.KeyID()), http.StatusUnauthorized) - return - } - authorizedPeer := authorizedClientPeers[authorizedPeerIdx] - if authorizedPeer.InstanceId != instanceID { - zap.S().Errorf("instance ID mismatch for peer key %q, expected %q, got %q", peerKey.KeyID(), authorizedPeer.InstanceId, instanceID) - http.Error(w, fmt.Sprintf("Unauthorized: instance ID mismatch for peer key %q, expected %q, got %q", peerKey.KeyID(), authorizedPeer.InstanceId, instanceID), http.StatusUnauthorized) - return - } - ctx := ContextWithPeer(r.Context(), authorizedPeer, peerKey) - handler.ServeHTTP(w, r.WithContext(ctx)) - }) -} - -func createHandshakePacket(instanceID string, identity *cryptoutil.PrivateKey) (*v1.SyncStreamItem, error) { - signedMessage, err := createSignedMessage([]byte(instanceID), identity) - if err != nil { - return nil, fmt.Errorf("signing instance ID: %w", err) - } - - return &v1.SyncStreamItem{ - Action: &v1.SyncStreamItem_Handshake{ - Handshake: &v1.SyncStreamItem_SyncActionHandshake{ - ProtocolVersion: SyncProtocolVersion, - InstanceId: signedMessage, - PublicKey: identity.PublicKeyProto(), - }, - }, - }, nil -} - -// send authentication in a header rather than in the stream. -func createAuthenticationHeader(instanceID string, identity *cryptoutil.PrivateKey) (string, error) { - signedMessage, err := createSignedMessage([]byte(instanceID), identity) - if err != nil { - return "", fmt.Errorf("signing instance ID for authentication header: %w", err) - } - - handshakePacket := &v1.SyncStreamItem_SyncActionHandshake{ - ProtocolVersion: SyncProtocolVersion, - InstanceId: signedMessage, - PublicKey: identity.PublicKeyProto(), - } - - encodedHandshake, err := proto.Marshal(handshakePacket) - if err != nil { - return "", fmt.Errorf("marshalling handshake packet: %w", err) - } - - if len(encodedHandshake) > 512 { - return "", fmt.Errorf("authorization header is too large, max size is 512 bytes, got %d bytes", len(encodedHandshake)) - } - - base64Handshake := base64.StdEncoding.EncodeToString(encodedHandshake) - return fmt.Sprintf("Backrest-Sync-Auth %s", base64Handshake), nil -} - -// verifyHandshakePacket verifies that -// - the signature on the instance ID is valid against the public key provided in the handshake -// - that the public key's ID is as attested in the handshake packet e.g. matches handshake.PublicKey.Keyid -// -// To authenticate, the caller must then check that the public key is trusted by checking the key ID against a local list. -func verifyHandshakePacket(item *v1.SyncStreamItem) (*cryptoutil.PublicKey, error) { - handshake := item.GetHandshake() - if handshake == nil { - return nil, fmt.Errorf("empty or nil handshake, handshake packet must be sent first") - } - - if handshake.ProtocolVersion != SyncProtocolVersion { - return nil, fmt.Errorf("protocol version mismatch: expected %d, got %d", SyncProtocolVersion, handshake.ProtocolVersion) - } - - if len(handshake.InstanceId.GetPayload()) == 0 || len(handshake.InstanceId.GetSignature()) == 0 { - return nil, errors.New("instance ID payload and signature must not be empty") - } - - if len(handshake.PublicKey.Keyid) == 0 { - return nil, errors.New("public key ID must not be empty") - } - - peerKey, err := cryptoutil.NewPublicKey(handshake.PublicKey) - if err != nil { - return nil, fmt.Errorf("loading peer public key: %w", err) - } - - if err := verifySignedMessage(handshake.InstanceId, peerKey); err != nil { - return nil, fmt.Errorf("verifying instance ID signature: %w", err) - } - - return peerKey, nil -} - -func verifyAuthenticationHeader(header string) (*cryptoutil.PublicKey, string, error) { - if len(header) == 0 { - return nil, "", errors.New("authentication header must not be empty") - } - - // The header is expected to be in the format "Backrest-Sync-Auth " - if !strings.HasPrefix(header, "Backrest-Sync-Auth ") { - return nil, "", fmt.Errorf("invalid authentication header format, expected 'Backrest-Sync-Auth ', got %s", header) - } - - // Extract the base64-encoded handshake packet - header = header[len("Backrest-Sync-Auth "):] - decoded, err := base64.StdEncoding.DecodeString(header) - if err != nil { - return nil, "", fmt.Errorf("decoding authentication header: %w", err) - } - - // Unmarshal the decoded header into a handshake packet - var handshakePacket v1.SyncStreamItem_SyncActionHandshake - if err := proto.Unmarshal(decoded, &handshakePacket); err != nil { - return nil, "", fmt.Errorf("unmarshalling handshake packet: %w", err) - } - - // Verify the handshake packet - peerKey, err := verifyHandshakePacket(&v1.SyncStreamItem{ - Action: &v1.SyncStreamItem_Handshake{ - Handshake: &handshakePacket, - }, - }) - if err != nil { - return nil, "", fmt.Errorf("verifying handshake packet: %w", err) - } - - return peerKey, string(handshakePacket.GetInstanceId().GetPayload()), nil -} diff --git a/internal/api/syncapi/cryptoutil.go b/internal/api/syncapi/cryptoutil.go deleted file mode 100644 index 04ba28a0..00000000 --- a/internal/api/syncapi/cryptoutil.go +++ /dev/null @@ -1,68 +0,0 @@ -package syncapi - -import ( - "encoding/binary" - "errors" - "fmt" - "time" - - v1 "github.com/garethgeorge/backrest/gen/go/v1" - "github.com/garethgeorge/backrest/internal/cryptoutil" -) - -func createSignedMessage(payload []byte, identity *cryptoutil.PrivateKey) (*v1.SignedMessage, error) { - if len(payload) == 0 { - return nil, errors.New("payload must not be empty") - } - - timestampMillis := time.Now().UnixMilli() - - payloadWithTimestamp := make([]byte, 0, len(payload)+8) - binary.BigEndian.AppendUint64(payloadWithTimestamp, uint64(timestampMillis)) - payloadWithTimestamp = append(payloadWithTimestamp, payload...) - - signature, err := identity.Sign(payloadWithTimestamp) - if err != nil { - return nil, fmt.Errorf("signing payload: %w", err) - } - - return &v1.SignedMessage{ - Payload: payload, - Signature: signature, - Keyid: identity.KeyID(), - TimestampMillis: timestampMillis, - }, nil -} - -func verifySignedMessage(msg *v1.SignedMessage, publicKey *cryptoutil.PublicKey) error { - if msg == nil { - return errors.New("signed message must not be nil") - } - if len(msg.GetPayload()) == 0 { - return errors.New("signed message payload must not be empty") - } - if len(msg.GetSignature()) == 0 { - return errors.New("signed message signature must not be empty") - } - if len(msg.GetKeyid()) == 0 { - return errors.New("signed message key ID must not be empty") - } - - if publicKey.KeyID() != msg.GetKeyid() { - return fmt.Errorf("public key ID mismatch: expected %s, got %s", publicKey.KeyID(), msg.GetKeyid()) - } - - payloadWithTimestamp := make([]byte, 0, len(msg.GetPayload())+8) - binary.BigEndian.AppendUint64(payloadWithTimestamp, uint64(msg.GetTimestampMillis())) - payloadWithTimestamp = append(payloadWithTimestamp, msg.GetPayload()...) - - if err := publicKey.Verify(payloadWithTimestamp, msg.GetSignature()); err != nil { - return fmt.Errorf("verifying signed message: %w", err) - } - - if time.Since(time.UnixMilli(msg.GetTimestampMillis())) > maxSignatureAge { - return fmt.Errorf("signature is too old, max age is %s. Is the clock out of sync?", maxSignatureAge) - } - - return nil -} diff --git a/internal/api/syncapi/errors.go b/internal/api/syncapi/errors.go index cd2872c0..d847a7de 100644 --- a/internal/api/syncapi/errors.go +++ b/internal/api/syncapi/errors.go @@ -42,6 +42,20 @@ func NewSyncErrorPending(message error) *SyncError { } } +func NewSyncErrorConnected(message error) *SyncError { + return &SyncError{ + State: v1.SyncConnectionState_CONNECTION_STATE_CONNECTED, + Message: message, + } +} + +func NewSyncErrorRetryWait(message error) *SyncError { + return &SyncError{ + State: v1.SyncConnectionState_CONNECTION_STATE_RETRY_WAIT, + Message: message, + } +} + func NewSyncErrorAuth(message error) *SyncError { return &SyncError{ State: v1.SyncConnectionState_CONNECTION_STATE_ERROR_AUTH, diff --git a/internal/api/syncapi/peerstate.go b/internal/api/syncapi/peerstate.go index 8260afa3..a00fd8fe 100644 --- a/internal/api/syncapi/peerstate.go +++ b/internal/api/syncapi/peerstate.go @@ -25,8 +25,8 @@ type PeerState struct { ConnectionStateMessage string // Plans and repos available on this peer - KnownRepos map[string]*v1.SyncRepoMetadata - KnownPlans map[string]*v1.SyncPlanMetadata + KnownRepos map[string]struct{} + KnownPlans map[string]struct{} // Partial configuration available for this peer Config *v1.RemoteConfig @@ -39,8 +39,8 @@ func newPeerState(instanceID, keyID string) *PeerState { LastHeartbeat: time.Now(), ConnectionState: v1.SyncConnectionState_CONNECTION_STATE_DISCONNECTED, ConnectionStateMessage: "disconnected", - KnownRepos: make(map[string]*v1.SyncRepoMetadata), - KnownPlans: make(map[string]*v1.SyncPlanMetadata), + KnownRepos: make(map[string]struct{}), + KnownPlans: make(map[string]struct{}), Config: nil, // Will be set when the config is received } } @@ -68,8 +68,8 @@ func peerStateToProto(state *PeerState) *v1.PeerState { LastHeartbeatMillis: state.LastHeartbeat.UnixMilli(), State: state.ConnectionState, StatusMessage: state.ConnectionStateMessage, - KnownRepos: slices.Collect(maps.Values(state.KnownRepos)), - KnownPlans: slices.Collect(maps.Values(state.KnownPlans)), + KnownRepos: slices.Collect(maps.Keys(state.KnownRepos)), + KnownPlans: slices.Collect(maps.Keys(state.KnownPlans)), RemoteConfig: state.Config, } } @@ -78,13 +78,13 @@ func peerStateFromProto(state *v1.PeerState) *PeerState { if state.PeerInstanceId == "" || state.PeerKeyid == "" { return nil } - knownRepos := make(map[string]*v1.SyncRepoMetadata, len(state.KnownRepos)) + knownRepos := make(map[string]struct{}, len(state.KnownRepos)) for _, repo := range state.KnownRepos { - knownRepos[repo.Id] = repo + knownRepos[repo] = struct{}{} } - knownPlans := make(map[string]*v1.SyncPlanMetadata, len(state.KnownPlans)) + knownPlans := make(map[string]struct{}, len(state.KnownPlans)) for _, plan := range state.KnownPlans { - knownPlans[plan.Id] = plan + knownPlans[plan] = struct{}{} } return &PeerState{ @@ -103,7 +103,6 @@ type PeerStateManager interface { GetPeerState(keyID string) *PeerState GetAll() []*PeerState SetPeerState(keyID string, state *PeerState) - UpdatePeerState(keyID string, instanceID string, updateFn func(state *PeerState)) OnStateChanged() eventemitter.Receiver[*PeerState] Close() error } @@ -159,23 +158,6 @@ func (m *InMemoryPeerStateManager) SetPeerState(keyID string, state *PeerState) m.onStateChanged.Emit(copy) } -func (m *InMemoryPeerStateManager) UpdatePeerState(keyID string, instanceID string, updateFn func(state *PeerState)) { - m.mu.Lock() - defer m.mu.Unlock() - - var state *PeerState - if existingState, exists := m.peerStates[keyID]; exists { - state = existingState.Clone() - } else { - state = newPeerState(instanceID, keyID) - } - - updateFn(state) - - m.peerStates[keyID] = state - m.onStateChanged.Emit(state.Clone()) -} - func (m *InMemoryPeerStateManager) Close() error { return nil } @@ -264,41 +246,6 @@ func (m *SqlitePeerStateManager) SetPeerState(keyID string, state *PeerState) { m.onStateChanged.Emit(state.Clone()) } -func (m *SqlitePeerStateManager) UpdatePeerState(keyID string, instanceID string, updateFn func(state *PeerState)) { - m.mu.Lock() - defer m.mu.Unlock() - - var state *PeerState - if stateBytes, err := m.kvstore.Get(keyID); err == nil { - var stateProto v1.PeerState - if err := proto.Unmarshal(stateBytes, &stateProto); err != nil { - zap.S().Warnf("error unmarshalling peer state for key %s: %v", keyID, err) - } else { - state = peerStateFromProto(&stateProto) - } - } else { - zap.S().Warnf("error getting peer state for key %s: %v", keyID, err) - } - if state == nil { - state = newPeerState(instanceID, keyID) - } - - updateFn(state) - - stateProto := peerStateToProto(state) - newStateBytes, err := proto.Marshal(stateProto) - if err != nil { - zap.S().Warnf("error marshalling peer state for key %s: %v", keyID, err) - return - } - - if err := m.kvstore.Set(keyID, newStateBytes); err != nil { - zap.S().Warnf("error setting peer state for key %s: %v", keyID, err) - return - } - m.onStateChanged.Emit(state.Clone()) -} - func (m *SqlitePeerStateManager) Close() error { return nil } diff --git a/internal/api/syncapi/peerstate_test.go b/internal/api/syncapi/peerstate_test.go index 697f8c86..a83cf8fa 100644 --- a/internal/api/syncapi/peerstate_test.go +++ b/internal/api/syncapi/peerstate_test.go @@ -1,26 +1,15 @@ package syncapi import ( - "database/sql" "testing" "time" -<<<<<<< HEAD + "github.com/garethgeorge/backrest/internal/kvstore" "github.com/google/go-cmp/cmp" - _ "github.com/ncruces/go-sqlite3/driver" - "github.com/ncruces/go-sqlite3/vfs/memdb" -======= - v1 "github.com/garethgeorge/backrest/gen/go/v1" - "github.com/garethgeorge/backrest/internal/cryptoutil" - "github.com/google/go-cmp/cmp" - "google.golang.org/protobuf/testing/protocmp" - "zombiezen.com/go/sqlite" - "zombiezen.com/go/sqlite/sqlitex" ->>>>>>> 9041d3c (improve sync api security by using 'Authorization' headers for initial key exchange) ) func PeerStateManagersForTest(t testing.TB) map[string]PeerStateManager { - dbpool := newDbForTest(t) + dbpool := kvstore.NewInMemorySqliteDbForKvStore(t) t.Cleanup(func() { dbpool.Close() }) @@ -44,18 +33,12 @@ func TestPeerStateManager_GetSet(t *testing.T) { InstanceID: "testInstance", KeyID: keyID, LastHeartbeat: time.Now().Round(time.Millisecond), - KnownRepos: map[string]*v1.SyncRepoMetadata{ - "repo1": {Id: "repo1"}, - "repo2": {Id: "repo2"}, - }, - KnownPlans: map[string]*v1.SyncPlanMetadata{ - "plan1": {Id: "plan1"}, - "plan2": {Id: "plan2"}, - }, + KnownRepos: map[string]struct{}{"repo1": {}, "repo2": {}}, + KnownPlans: map[string]struct{}{"plan1": {}, "plan2": {}}, } psm.SetPeerState(keyID, state) gotState := psm.GetPeerState(keyID) - if diff := cmp.Diff(state, gotState, protocmp.Transform()); diff != "" { + if diff := cmp.Diff(state, gotState, cmp.AllowUnexported(PeerState{})); diff != "" { t.Errorf("unexpected diff: %v", diff) } }) @@ -95,7 +78,7 @@ func TestPeerStateManager_OnStateChanged(t *testing.T) { select { case changedState := <-ch: - if diff := cmp.Diff(state, changedState, protocmp.Transform()); diff != "" { + if diff := cmp.Diff(state, changedState, cmp.AllowUnexported(PeerState{})); diff != "" { t.Errorf("unexpected diff: %v", diff) } case <-time.After(1 * time.Second): @@ -104,12 +87,3 @@ func TestPeerStateManager_OnStateChanged(t *testing.T) { }) } } - -func newDbForTest(t testing.TB) *sql.DB { - t.Helper() - dbpool, err := sql.Open("sqlite3", memdb.TestDB(t)) - if err != nil { - t.Fatalf("error creating sqlite pool: %s", err) - } - return dbpool -} diff --git a/internal/api/syncapi/syncapi_test.go b/internal/api/syncapi/syncapi_test.go index fb0ce1eb..121a9d75 100644 --- a/internal/api/syncapi/syncapi_test.go +++ b/internal/api/syncapi/syncapi_test.go @@ -539,9 +539,6 @@ func tryExpectExactOperations(t *testing.T, ctx context.Context, peer *peerUnder for _, op := range ops { op.Modno = 0 } - for _, op := range wantOps { - op.Modno = 0 - } if diff := cmp.Diff(ops, wantOps, protocmp.Transform()); diff != "" { return fmt.Errorf("unexpected diff: %v", diff) } @@ -626,13 +623,13 @@ func waitForConnectionState(t *testing.T, ctx context.Context, peer *peerUnderTe onStateChanged := peer.manager.peerStateManager.OnStateChanged().Subscribe() defer peer.manager.peerStateManager.OnStateChanged().Unsubscribe(onStateChanged) - // First check if the peer is already in the desired state. + // First check if the peer is already connected. state := peer.manager.peerStateManager.GetPeerState(hostPeer.Keyid) - if state != nil && state.ConnectionState == wantState { - return // Already in desired state, nothing to do + if state != nil && state.ConnectionState == v1.SyncConnectionState_CONNECTION_STATE_CONNECTED { + return // Already connected, nothing to do } - // If not in desired state, wait for a connection event + // If not connected, wait for a connection event var lastState *PeerState stop := false for !stop { @@ -644,7 +641,7 @@ func waitForConnectionState(t *testing.T, ctx context.Context, peer *peerUnderTe } if state.KeyID == hostPeer.Keyid && state.InstanceID == hostPeer.InstanceId { lastState = state - if state.ConnectionState == wantState { + if state.ConnectionState == v1.SyncConnectionState_CONNECTION_STATE_CONNECTED { stop = true continue } @@ -656,8 +653,8 @@ func waitForConnectionState(t *testing.T, ctx context.Context, peer *peerUnderTe } if lastState == nil { t.Fatalf("timeout waiting for connection to host peer %s", hostPeer.InstanceId) - } else if lastState.ConnectionState != wantState { - t.Fatalf("expected connection state to be %v, got %v (reason: %q)", wantState, lastState.ConnectionState, lastState.ConnectionStateMessage) + } else if lastState.ConnectionState != v1.SyncConnectionState_CONNECTION_STATE_CONNECTED { + t.Fatalf("expected connection state to be CONNECTED, got %v (reason: %q)", lastState.ConnectionState, lastState.ConnectionStateMessage) } } @@ -668,8 +665,8 @@ func tryConnect(t *testing.T, ctx context.Context, peer *peerUnderTest, hostPeer func runSyncAPIWithCtx(ctx context.Context, peer *peerUnderTest, bindAddr string) { mux := http.NewServeMux() syncHandler := NewBackrestSyncHandler(peer.manager) - endPoint, syncServiceHandler := v1connect.NewBackrestSyncServiceHandler(syncHandler) - mux.Handle(endPoint, AuthenticationMiddleware(peer.configMgr, syncServiceHandler)) + mux.Handle(v1connect.NewBackrestSyncServiceHandler(syncHandler)) + server := &http.Server{ Addr: bindAddr, Handler: h2c.NewHandler(mux, &http2.Server{}), // h2c is HTTP/2 without TLS for grpc-connect support. @@ -768,7 +765,7 @@ func newPeerUnderTest(t *testing.T, initialConfig *v1.Config) *peerUnderTest { t.Fatalf("failed to create peer state manager: %v", err) } - manager := NewSyncManager(configMgr, oplog, logStore, orchestrator, peerStateManager) + manager := NewSyncManager(configMgr, oplog, orchestrator, peerStateManager) manager.syncClientRetryDelay = 250 * time.Millisecond return &peerUnderTest{ diff --git a/internal/api/syncapi/syncclient.go b/internal/api/syncapi/syncclient.go index 1cdb0efc..6083d80f 100644 --- a/internal/api/syncapi/syncclient.go +++ b/internal/api/syncapi/syncclient.go @@ -74,9 +74,7 @@ func NewSyncClient( oplog: oplog, l: zap.L().Named(fmt.Sprintf("syncclient for %q", peer.GetInstanceId())), } - c.mgr.peerStateManager.UpdatePeerState(peer.Keyid, peer.InstanceId, func(peerState *PeerState) { - // this will create a new peer state if one doesn't exist - }) + c.mgr.peerStateManager.SetPeerState(peer.Keyid, newPeerState(peer.InstanceId, peer.Keyid)) return c, nil } @@ -101,92 +99,47 @@ func (c *SyncClient) RunSync(ctx context.Context) { c.l.Sugar().Infof("connecting to peer %q (%s) at %s", c.peer.InstanceId, c.peer.Keyid, c.peer.GetInstanceUrl()) var wg sync.WaitGroup - - // Attach the 'Authorization' header to the sync stream as we create it. - syncStream := c.client.Sync(ctx) - if authHeader, err := createAuthenticationHeader(c.syncConfigSnapshot.config.Instance, c.syncConfigSnapshot.identityKey); err != nil { - c.l.Sugar().Errorf("failed to create authentication header: %v", err) - } else { - syncStream.RequestHeader().Set("Authorization", authHeader) - } - - // Start a thread to handle the stream IO e.g. send and receive messages and pass them into the cmdStream stream abstraction. wg.Add(1) go func() { defer wg.Done() - if err := cmdStream.ConnectStream(ctx, syncStream); err != nil { - c.l.Sugar().Infof("lost stream connection to peer %q (%s): %v", c.peer.InstanceId, c.peer.Keyid, err) - var syncErr *SyncError - c.mgr.peerStateManager.UpdatePeerState(c.peer.Keyid, c.peer.InstanceId, func(state *PeerState) { - state.LastHeartbeat = time.Now() - if errors.As(err, &syncErr) { - state.ConnectionState = syncErr.State - state.ConnectionStateMessage = syncErr.Message.Error() - } else { - state.ConnectionState = v1.SyncConnectionState_CONNECTION_STATE_ERROR_INTERNAL - state.ConnectionStateMessage = err.Error() - } - }) - } else { - c.l.Sugar().Infof("stream connection to peer %q (%s) closed", c.peer.InstanceId, c.peer.Keyid) - c.reconnectAttempts = 0 - c.mgr.peerStateManager.UpdatePeerState(c.peer.Keyid, c.peer.InstanceId, func(state *PeerState) { - state.ConnectionState = v1.SyncConnectionState_CONNECTION_STATE_DISCONNECTED - state.ConnectionStateMessage = "disconnected" - state.LastHeartbeat = time.Now() - }) - } - }() - - // Start a thread to handle the API loop, this is fundamentally a switch statement operating on incomming commands. - wg.Add(1) - go func() { - defer wg.Done() - - // Get the headers from the stream - headers := syncStream.ResponseHeader() - authHeader := headers.Get("Authorization") - - peerKey, instanceID, err := verifyAuthenticationHeader(authHeader) - if err != nil { - cmdStream.SendErrorAndTerminate(NewSyncErrorAuth(fmt.Errorf("verifying authentication header: %w", err))) - return - } - if peerKey.KeyID() != c.peer.Keyid { - cmdStream.SendErrorAndTerminate(NewSyncErrorAuth(fmt.Errorf("remote peer key ID mismatch: expected %q, got %q", c.peer.Keyid, peerKey.KeyID()))) - return - } - if instanceID != c.peer.InstanceId { - cmdStream.SendErrorAndTerminate(NewSyncErrorAuth(fmt.Errorf("remote instance ID mismatch: expected %q, got %q", c.peer.InstanceId, instanceID))) - return - } - ctx := ContextWithPeer(context.Background(), c.peer, peerKey) - - // Run sync once the first packet has been received allowing us to read the request headers. - // This means it is essential that the server always sends some initial packet, today this is just a heartbeat to start off the connection. - err = runSync( + err := runSync( ctx, c.localInstanceID, c.syncConfigSnapshot.identityKey, cmdStream, syncSessionHandler, + c.syncConfigSnapshot.config.GetMultihost().GetKnownHosts(), ) cmdStream.SendErrorAndTerminate(err) }() - // Send a heartbeat packet to trigger establishing the connection. - cmdStream.Send(&v1.SyncStreamItem{ - Action: &v1.SyncStreamItem_Heartbeat{}, - }) + if err := cmdStream.ConnectStream(ctx, c.client.Sync(ctx)); err != nil { + c.l.Sugar().Infof("lost stream connection to peer %q (%s): %v", c.peer.InstanceId, c.peer.Keyid, err) + var syncErr *SyncError + state := c.mgr.peerStateManager.GetPeerState(c.peer.Keyid).Clone() + if state == nil { + state = newPeerState(c.peer.InstanceId, c.peer.Keyid) + } + state.LastHeartbeat = time.Now() + if errors.As(err, &syncErr) { + state.ConnectionState = syncErr.State + state.ConnectionStateMessage = syncErr.Message.Error() + } else { + state.ConnectionState = v1.SyncConnectionState_CONNECTION_STATE_ERROR_INTERNAL + state.ConnectionStateMessage = err.Error() + } + c.mgr.peerStateManager.SetPeerState(c.peer.Keyid, state) + } else { + c.reconnectAttempts = 0 + } - // Wait for the thread running the API loop and the thread running the stream connection to finish. wg.Wait() - reconnectDelayWithBackoff := c.reconnectDelay + delay := c.reconnectDelay - time.Since(lastConnect) if c.reconnectAttempts > 0 { - reconnectDelayWithBackoff *= time.Duration(1 << min(c.reconnectAttempts, 5)) // 2^reconnectAttempts, max 32 + backoff := time.Duration(1< 0 { - stream.Send(&v1.SyncStreamItem{ - Action: &v1.SyncStreamItem_SendLogData{ - SendLogData: &v1.SyncStreamItem_SyncActionSendLogData{ - LogId: item.GetLogId(), - Chunk: buf[:n], - }, - }, - }) - } - if err != nil { - if errors.Is(err, net.ErrClosed) || errors.Is(err, context.Canceled) { - c.l.Sugar().Debugf("finished reading log %q, context cancelled or stream closed", item.GetLogId()) - return nil - } - c.l.Sugar().Warnf("error reading log %q: %v", item.GetLogId(), err) - return fmt.Errorf("reading log %q: %w", item.GetLogId(), err) - } - if n == 0 { - c.l.Sugar().Debugf("finished reading log %q, no more data", item.GetLogId()) - // No more data to read, we can close the stream. - stream.Send(&v1.SyncStreamItem{ - Action: &v1.SyncStreamItem_SendLogData{ - SendLogData: &v1.SyncStreamItem_SyncActionSendLogData{ - LogId: item.GetLogId(), - Chunk: nil, // Send an empty chunk to signal the end of the log. - }, - }, - }) - } - } - } - - return nil -} diff --git a/internal/api/syncapi/synccommon.go b/internal/api/syncapi/synccommon.go index 1195270e..b01e32ae 100644 --- a/internal/api/syncapi/synccommon.go +++ b/internal/api/syncapi/synccommon.go @@ -2,12 +2,14 @@ package syncapi import ( "context" + "encoding/binary" + "errors" "fmt" + "slices" "time" v1 "github.com/garethgeorge/backrest/gen/go/v1" "github.com/garethgeorge/backrest/internal/cryptoutil" - "go.uber.org/zap" ) var maxSignatureAge = 5 * time.Minute @@ -18,18 +20,41 @@ func runSync( localKey *cryptoutil.PrivateKey, commandStream *bidiSyncCommandStream, handler syncSessionHandler, + knownPeers []*v1.Multihost_Peer, // could be known hosts or authorized clients, doesn't matter. This is used to verify the handshake packet, authorization comes later. ) error { - peer := PeerFromContext(ctx) - peerPublicKey := PeerPublicKeyFromContext(ctx) - if peer == nil || peerPublicKey == nil { - return NewSyncErrorAuth(fmt.Errorf("peer not found in context, ensure authentication middleware is applied before sync handlers")) + // send the initial handshake packet to the peer to establish the connection. + go func() { + handshakePacket, err := createHandshakePacket(localInstanceID, localKey) + if err != nil { + commandStream.SendErrorAndTerminate(fmt.Errorf("creating handshake packet: %w", err)) + return + } + commandStream.Send(handshakePacket) + }() + + // Wait for the handshake packet to be acknowledged by the peer. + handshake := commandStream.ReceiveWithinDuration(15 * time.Second) + if handshake == nil { + return NewSyncErrorAuth(fmt.Errorf("no handshake packet received from peer within timeout")) + } + if _, err := verifyHandshakePacket(handshake); err != nil { + return NewSyncErrorAuth(fmt.Errorf("verifying handshake packet: %w", err)) } - defer func() { - if err := handler.OnConnectionClosed(ctx, commandStream); err != nil { - zap.L().Error("error handling connection closed", zap.Error(err)) - } - }() + // Find the peer definition in the known peers list. + var peer *v1.Multihost_Peer + peerIdx := slices.IndexFunc(knownPeers, func(p *v1.Multihost_Peer) bool { + return p.Keyid == handshake.GetHandshake().GetPublicKey().GetKeyid() + }) + if peerIdx >= 0 { + peer = knownPeers[peerIdx] + } else { + return NewSyncErrorAuth(fmt.Errorf("peer public key ID %s (instance ID %s) not found in known peers", handshake.GetHandshake().GetPublicKey().GetKeyid(), string(handshake.GetHandshake().GetInstanceId().GetPayload()))) + } + + if err := authorizeHandshakeAsPeer(handshake, peer); err != nil { + return NewSyncErrorAuth(fmt.Errorf("authorizing handshake as peer: %w", err)) + } if err := handler.OnConnectionEstablished(ctx, commandStream, peer); err != nil { return err @@ -65,14 +90,6 @@ func runSync( if err := handler.HandleThrottle(ctx, commandStream, item.GetThrottle()); err != nil { return fmt.Errorf("handling throttle: %w", err) } - case *v1.SyncStreamItem_GetLog: - if err := handler.HandleGetLog(ctx, commandStream, item.GetGetLog()); err != nil { - return fmt.Errorf("handling get log: %w", err) - } - case *v1.SyncStreamItem_SendLogData: - if err := handler.HandleSendLogData(ctx, commandStream, item.GetSendLogData()); err != nil { - return fmt.Errorf("handling send log data: %w", err) - } default: return NewSyncErrorProtocol(fmt.Errorf("unknown action type %T in sync stream item", item.GetAction())) } @@ -97,6 +114,131 @@ func tryReceiveWithinDuration(ctx context.Context, receiveChan chan *v1.SyncStre } } +func createSignedMessage(payload []byte, identity *cryptoutil.PrivateKey) (*v1.SignedMessage, error) { + if len(payload) == 0 { + return nil, errors.New("payload must not be empty") + } + + timestampMillis := time.Now().UnixMilli() + + payloadWithTimestamp := make([]byte, 0, len(payload)+8) + binary.BigEndian.AppendUint64(payloadWithTimestamp, uint64(timestampMillis)) + payloadWithTimestamp = append(payloadWithTimestamp, payload...) + + signature, err := identity.Sign(payloadWithTimestamp) + if err != nil { + return nil, fmt.Errorf("signing payload: %w", err) + } + + return &v1.SignedMessage{ + Payload: payload, + Signature: signature, + Keyid: identity.KeyID(), + TimestampMillis: timestampMillis, + }, nil +} + +func verifySignedMessage(msg *v1.SignedMessage, publicKey *cryptoutil.PublicKey) error { + if msg == nil { + return errors.New("signed message must not be nil") + } + if len(msg.GetPayload()) == 0 { + return errors.New("signed message payload must not be empty") + } + if len(msg.GetSignature()) == 0 { + return errors.New("signed message signature must not be empty") + } + if len(msg.GetKeyid()) == 0 { + return errors.New("signed message key ID must not be empty") + } + + if publicKey.KeyID() != msg.GetKeyid() { + return fmt.Errorf("public key ID mismatch: expected %s, got %s", publicKey.KeyID(), msg.GetKeyid()) + } + + payloadWithTimestamp := make([]byte, 0, len(msg.GetPayload())+8) + binary.BigEndian.AppendUint64(payloadWithTimestamp, uint64(msg.GetTimestampMillis())) + payloadWithTimestamp = append(payloadWithTimestamp, msg.GetPayload()...) + + if err := publicKey.Verify(payloadWithTimestamp, msg.GetSignature()); err != nil { + return fmt.Errorf("verifying signed message: %w", err) + } + + if time.Since(time.UnixMilli(msg.GetTimestampMillis())) > maxSignatureAge { + return fmt.Errorf("signature is too old, max age is %s. Is the clock out of sync?", maxSignatureAge) + } + + return nil +} + +func createHandshakePacket(instanceID string, identity *cryptoutil.PrivateKey) (*v1.SyncStreamItem, error) { + signedMessage, err := createSignedMessage([]byte(instanceID), identity) + if err != nil { + return nil, fmt.Errorf("signing instance ID: %w", err) + } + + return &v1.SyncStreamItem{ + Action: &v1.SyncStreamItem_Handshake{ + Handshake: &v1.SyncStreamItem_SyncActionHandshake{ + ProtocolVersion: SyncProtocolVersion, + InstanceId: signedMessage, + PublicKey: identity.PublicKeyProto(), + }, + }, + }, nil +} + +// verifyHandshakePacket verifies that +// - the signature on the instance ID is valid against the public key provided in the handshake +// - that the public key's ID is as attested in the handshake packet e.g. matches handshake.PublicKey.Keyid +// +// To authenticate, the caller must then check that the public key is trusted by checking the key ID against a local list. +func verifyHandshakePacket(item *v1.SyncStreamItem) (*cryptoutil.PublicKey, error) { + handshake := item.GetHandshake() + if handshake == nil { + return nil, fmt.Errorf("empty or nil handshake, handshake packet must be sent first") + } + + if handshake.ProtocolVersion != SyncProtocolVersion { + return nil, fmt.Errorf("protocol version mismatch: expected %d, got %d", SyncProtocolVersion, handshake.ProtocolVersion) + } + + if len(handshake.InstanceId.GetPayload()) == 0 || len(handshake.InstanceId.GetSignature()) == 0 { + return nil, errors.New("instance ID payload and signature must not be empty") + } + + if len(handshake.PublicKey.Keyid) == 0 { + return nil, errors.New("public key ID must not be empty") + } + + peerKey, err := cryptoutil.NewPublicKey(handshake.PublicKey) + if err != nil { + return nil, fmt.Errorf("loading peer public key: %w", err) + } + + if err := verifySignedMessage(handshake.InstanceId, peerKey); err != nil { + return nil, fmt.Errorf("verifying instance ID signature: %w", err) + } + + return peerKey, nil +} + +// authorizeHandshakeAsPeer checks that the handshake packet has the expected key ID and instance ID. +// If this succeeds and the handshake is verified, then it is safe to assume the identity we are talking to. +func authorizeHandshakeAsPeer(item *v1.SyncStreamItem, peer *v1.Multihost_Peer) error { + handshake := item.GetHandshake() + if handshake == nil { + return fmt.Errorf("empty or nil handshake, handshake packet must be sent first") + } + if string(handshake.GetInstanceId().GetPayload()) != peer.InstanceId { + return fmt.Errorf("instance ID mismatch: expected %s, got %s", peer.InstanceId, string(handshake.InstanceId.GetPayload())) + } + if handshake.GetPublicKey().GetKeyid() != peer.Keyid { + return fmt.Errorf("public key ID mismatch: expected %s, got %s", peer.Keyid, handshake.PublicKey.Keyid) + } + return nil +} + // sendHeartbeats sends a heartbeat message to the stream at regular intervals. // This is useful for keeping the connection alive and ensuring that the peer is still responsive. func sendHeartbeats(ctx context.Context, stream *bidiSyncCommandStream, interval time.Duration) { @@ -120,11 +262,7 @@ func sendHeartbeats(ctx context.Context, stream *bidiSyncCommandStream, interval // syncSessionHandler is a stateful handler for the messages within the context of a sync stream session. // the handler does not need to be thread safe as it is guaranteed to be called from a single thread. type syncSessionHandler interface { - // OnConnectionEstablished is called when a new sync connection is established, provides the peer information once identified. OnConnectionEstablished(ctx context.Context, stream *bidiSyncCommandStream, peer *v1.Multihost_Peer) error - // OnConnectionClosed is called when the sync connection is closed, e.g. when the peer disconnects. Guaranteed to be called after OnConnectionEstablished. - OnConnectionClosed(ctx context.Context, stream *bidiSyncCommandStream) error - // Handle* methods are called for each action type in the sync stream. HandleHeartbeat(ctx context.Context, stream *bidiSyncCommandStream, item *v1.SyncStreamItem_SyncActionHeartbeat) error HandleDiffOperations(ctx context.Context, stream *bidiSyncCommandStream, item *v1.SyncStreamItem_SyncActionDiffOperations) error HandleSendOperations(ctx context.Context, stream *bidiSyncCommandStream, item *v1.SyncStreamItem_SyncActionSendOperations) error @@ -132,18 +270,12 @@ type syncSessionHandler interface { HandleSetConfig(ctx context.Context, stream *bidiSyncCommandStream, item *v1.SyncStreamItem_SyncActionSetConfig) error HandleListResources(ctx context.Context, stream *bidiSyncCommandStream, item *v1.SyncStreamItem_SyncActionListResources) error HandleThrottle(ctx context.Context, stream *bidiSyncCommandStream, item *v1.SyncStreamItem_SyncActionThrottle) error - HandleGetLog(ctx context.Context, stream *bidiSyncCommandStream, item *v1.SyncStreamItem_SyncActionGetLog) error - HandleSendLogData(ctx context.Context, stream *bidiSyncCommandStream, item *v1.SyncStreamItem_SyncActionSendLogData) error } type unimplementedSyncSessionHandler struct{} func (h *unimplementedSyncSessionHandler) OnConnectionEstablished(ctx context.Context, stream *bidiSyncCommandStream, peer *v1.Multihost_Peer) error { - return nil // no-op by default. -} - -func (h *unimplementedSyncSessionHandler) OnConnectionClosed(ctx context.Context, stream *bidiSyncCommandStream) error { - panic("must not be unimplemented") + return NewSyncErrorProtocol(fmt.Errorf("OnConnectionEstablished not implemented")) } func (h *unimplementedSyncSessionHandler) HandleHeartbeat(ctx context.Context, stream *bidiSyncCommandStream, item *v1.SyncStreamItem_SyncActionHeartbeat) error { @@ -173,11 +305,3 @@ func (h *unimplementedSyncSessionHandler) HandleListResources(ctx context.Contex func (h *unimplementedSyncSessionHandler) HandleThrottle(ctx context.Context, stream *bidiSyncCommandStream, item *v1.SyncStreamItem_SyncActionThrottle) error { return NewSyncErrorProtocol(fmt.Errorf("HandleThrottle not implemented")) } - -func (h *unimplementedSyncSessionHandler) HandleGetLog(ctx context.Context, stream *bidiSyncCommandStream, item *v1.SyncStreamItem_SyncActionGetLog) error { - return NewSyncErrorProtocol(fmt.Errorf("HandleGetLog not implemented")) -} - -func (h *unimplementedSyncSessionHandler) HandleSendLogData(ctx context.Context, stream *bidiSyncCommandStream, item *v1.SyncStreamItem_SyncActionSendLogData) error { - return NewSyncErrorProtocol(fmt.Errorf("HandleSendLogData not implemented")) -} diff --git a/internal/api/syncapi/synchandler.go b/internal/api/syncapi/synchandler.go index 40eea085..279a1f09 100644 --- a/internal/api/syncapi/synchandler.go +++ b/internal/api/syncapi/synchandler.go @@ -4,10 +4,8 @@ import ( "context" "errors" "fmt" - "io" "slices" "sort" - "sync" "time" "connectrpc.com/connect" @@ -26,8 +24,6 @@ const SyncProtocolVersion = 1 type BackrestSyncHandler struct { v1connect.UnimplementedBackrestSyncServiceHandler mgr *SyncManager - - mu sync.Mutex } var _ v1connect.BackrestSyncServiceHandler = &BackrestSyncHandler{} @@ -46,25 +42,9 @@ func (h *BackrestSyncHandler) Sync(ctx context.Context, stream *connect.BidiStre return connect.NewError(connect.CodePermissionDenied, errors.New("sync server is not configured")) } - // Attach the 'Authorization' header to the sync stream as we create it. - if authHeader, err := createAuthenticationHeader(snapshot.config.Instance, snapshot.identityKey); err != nil { - zap.S().Errorf("failed to create authentication header: %v", err) - return connect.NewError(connect.CodeInternal, fmt.Errorf("creating authentication header: %w", err)) - } else { - stream.ResponseHeader().Set("Authorization", authHeader) - } - - // Setup to read from the stream and handle commands - // Note that runSync will perform the authentication check, which in this case is asserting that a peer is present. - // The peer must be provided in the context by the authentication middleware. sessionHandler := newSyncHandlerServer(h.mgr, snapshot) cmdStream := newBidiSyncCommandStream() - // Send a heartbeat packet to send the initial headers to the client and establish the connection. - cmdStream.Send(&v1.SyncStreamItem{ - Action: &v1.SyncStreamItem_Heartbeat{}, - }) - go func() { err := runSync( ctx, @@ -72,29 +52,25 @@ func (h *BackrestSyncHandler) Sync(ctx context.Context, stream *connect.BidiStre snapshot.identityKey, cmdStream, sessionHandler, + snapshot.config.GetMultihost().GetAuthorizedClients(), ) cmdStream.SendErrorAndTerminate(err) }() if err := cmdStream.ConnectStream(ctx, stream); err != nil { - if sessionHandler.peer != nil { - zap.S().Errorf("sync handler stream error for client %q: %v", sessionHandler.peer.InstanceId, err) - h.mgr.peerStateManager.UpdatePeerState(sessionHandler.peer.Keyid, sessionHandler.peer.InstanceId, func(peerState *PeerState) { - peerState.LastHeartbeat = time.Now() - peerState.ConnectionState = v1.SyncConnectionState_CONNECTION_STATE_DISCONNECTED - peerState.ConnectionStateMessage = err.Error() - var syncErr *SyncError - if errors.As(err, &syncErr) { - peerState.ConnectionState = syncErr.State - peerState.ConnectionStateMessage = syncErr.Message.Error() - } - }) - } else { - zap.S().Errorf("sync handler stream error for unestablished session: %v", err) - } - + zap.S().Errorf("sync handler stream error: %v", err) var syncErr *SyncError if errors.As(err, &syncErr) { + if sessionHandler.peer != nil { + peerState := h.mgr.peerStateManager.GetPeerState(sessionHandler.peer.Keyid).Clone() + if peerState == nil { + peerState = newPeerState(sessionHandler.peer.InstanceId, sessionHandler.peer.Keyid) + } + peerState.ConnectionState = syncErr.State + peerState.ConnectionStateMessage = syncErr.Message.Error() + peerState.LastHeartbeat = time.Now() + h.mgr.peerStateManager.SetPeerState(sessionHandler.peer.Keyid, peerState) + } switch syncErr.State { case v1.SyncConnectionState_CONNECTION_STATE_ERROR_AUTH: return connect.NewError(connect.CodePermissionDenied, syncErr.Message) @@ -104,19 +80,12 @@ func (h *BackrestSyncHandler) Sync(ctx context.Context, stream *connect.BidiStre return connect.NewError(connect.CodeInternal, syncErr.Message) } } - return connect.NewError(connect.CodeInternal, err) - } else { - h.mgr.peerStateManager.UpdatePeerState(sessionHandler.peer.Keyid, sessionHandler.peer.InstanceId, func(peerState *PeerState) { - peerState.LastHeartbeat = time.Now() - peerState.ConnectionState = v1.SyncConnectionState_CONNECTION_STATE_DISCONNECTED - peerState.ConnectionStateMessage = "disconnected" - }) - zap.S().Infof("sync handler stream closed for client %q", sessionHandler.peer.InstanceId) } return nil } +// syncSessionHandlerServer is a syncSessionHandler implementation for servers. type syncSessionHandlerServer struct { unimplementedSyncSessionHandler @@ -129,9 +98,6 @@ type syncSessionHandlerServer struct { opIDLru *lru.Cache[int64, int64] // original ID -> local ID flowIDLru *lru.Cache[int64, int64] // original flow ID -> local flow ID - requestedLogStreams map[string]struct{} - activeLogStreams map[string]io.WriteCloser - configWatchCh chan struct{} // Channel for configuration updates } @@ -144,24 +110,12 @@ func newSyncHandlerServer(mgr *SyncManager, snapshot *syncConfigSnapshot) *syncS snapshot: *snapshot, opIDLru: opIDLru, flowIDLru: flowIDLru, - - requestedLogStreams: make(map[string]struct{}), - activeLogStreams: make(map[string]io.WriteCloser), } } var _ syncSessionHandler = (*syncSessionHandlerServer)(nil) func (h *syncSessionHandlerServer) OnConnectionEstablished(ctx context.Context, stream *bidiSyncCommandStream, peer *v1.Multihost_Peer) error { - // Check if the peer is already connected, and then store the connection. - h.mgr.mu.Lock() - if _, exists := h.mgr.sessionHandlerMap[peer.Keyid]; exists { - h.mgr.mu.Unlock() - return NewSyncErrorAuth(fmt.Errorf("client %q is already connected", peer.InstanceId)) - } - h.mgr.sessionHandlerMap[peer.Keyid] = h - h.mgr.mu.Unlock() - // Verify that the peer is in our authorized clients list authorizedClientPeerIdx := slices.IndexFunc(h.snapshot.config.Multihost.GetAuthorizedClients(), func(p *v1.Multihost_Peer) bool { return p.InstanceId == peer.InstanceId && p.Keyid == peer.Keyid @@ -185,11 +139,11 @@ func (h *syncSessionHandlerServer) OnConnectionEstablished(ctx context.Context, } // Configure the state for the connected peer. - h.mgr.peerStateManager.UpdatePeerState(h.peer.Keyid, peer.InstanceId, func(peerState *PeerState) { - peerState.ConnectionStateMessage = "connected" - peerState.ConnectionState = v1.SyncConnectionState_CONNECTION_STATE_CONNECTED - peerState.LastHeartbeat = time.Now() - }) + peerState := newPeerState(peer.InstanceId, h.peer.Keyid) + peerState.ConnectionStateMessage = "connected" + peerState.ConnectionState = v1.SyncConnectionState_CONNECTION_STATE_CONNECTED + peerState.LastHeartbeat = time.Now() + h.mgr.peerStateManager.SetPeerState(h.peer.Keyid, peerState) zap.S().Infof("syncserver accepted a connection from client instance ID %q", h.peer.InstanceId) @@ -220,38 +174,16 @@ func (h *syncSessionHandlerServer) OnConnectionEstablished(ctx context.Context, }() // Send initial configuration to client - if err := h.sendConfigToClient(stream, h.snapshot.config); err != nil { - return err - } - - return nil -} - -func (h *syncSessionHandlerServer) OnConnectionClosed(ctx context.Context, stream *bidiSyncCommandStream) error { - if h.peer != nil { - zap.S().Infof("syncserver connection closed for client %q", h.peer.InstanceId) - h.mgr.mu.Lock() - delete(h.mgr.sessionHandlerMap, h.peer.Keyid) - h.mgr.mu.Unlock() - } - - // Close any active resources e.g. sinks for active log streams. - for logID, logSink := range h.activeLogStreams { - if err := logSink.Close(); err != nil { - return fmt.Errorf("action SendLogData: closing log stream %q: %w", logID, err) - } - } - - return nil + return h.sendConfigToClient(stream, h.snapshot.config) } func (h *syncSessionHandlerServer) HandleHeartbeat(ctx context.Context, stream *bidiSyncCommandStream, item *v1.SyncStreamItem_SyncActionHeartbeat) error { - h.mgr.peerStateManager.UpdatePeerState(h.peer.Keyid, h.peer.InstanceId, func(peerState *PeerState) { - if peerState == nil { - return // this should not happen - } - peerState.LastHeartbeat = time.Now() - }) + peerState := h.mgr.peerStateManager.GetPeerState(h.peer.Keyid).Clone() + if peerState == nil { + return NewSyncErrorInternal(fmt.Errorf("peer state for %q not found", h.peer.Keyid)) + } + peerState.LastHeartbeat = time.Now() + h.mgr.peerStateManager.SetPeerState(h.peer.Keyid, peerState) return nil } @@ -369,7 +301,7 @@ func (h *syncSessionHandlerServer) HandleDiffOperations(ctx context.Context, str func (h *syncSessionHandlerServer) HandleSendOperations(ctx context.Context, stream *bidiSyncCommandStream, item *v1.SyncStreamItem_SyncActionSendOperations) error { switch event := item.GetEvent().Event.(type) { case *v1.OperationEvent_CreatedOperations: - zap.L().Debug("syncserver received create operations", zap.Any("operations", event.CreatedOperations.GetOperations())) + zap.L().Debug("syncserver received created operations", zap.Any("operations", event.CreatedOperations.GetOperations())) for _, op := range event.CreatedOperations.GetOperations() { if err := h.insertOrUpdate(op); err != nil { return fmt.Errorf("action SendOperations: operation event create %+v: %w", op, err) @@ -397,79 +329,32 @@ func (h *syncSessionHandlerServer) HandleSendOperations(ctx context.Context, str } func (h *syncSessionHandlerServer) HandleSendConfig(ctx context.Context, stream *bidiSyncCommandStream, item *v1.SyncStreamItem_SyncActionSendConfig) error { - h.mgr.peerStateManager.UpdatePeerState(h.peer.Keyid, h.peer.InstanceId, func(peerState *PeerState) { - if peerState == nil { - return // this should not happen - } - peerState.Config = item.GetConfig() - }) + peerState := h.mgr.peerStateManager.GetPeerState(h.peer.Keyid).Clone() + if peerState == nil { + return NewSyncErrorInternal(fmt.Errorf("peer state for %q not found", h.peer.Keyid)) + } + peerState.Config = item.GetConfig() + h.mgr.peerStateManager.SetPeerState(h.peer.Keyid, peerState) return nil } func (h *syncSessionHandlerServer) HandleListResources(ctx context.Context, stream *bidiSyncCommandStream, item *v1.SyncStreamItem_SyncActionListResources) error { zap.L().Debug("syncserver received resource list from client", zap.String("client_instance_id", h.peer.InstanceId), - zap.Any("repos", item.GetRepos()), - zap.Any("plans", item.GetPlans())) - h.mgr.peerStateManager.UpdatePeerState(h.peer.Keyid, h.peer.InstanceId, func(peerState *PeerState) { - if peerState == nil { - return // this should not happen - } - - repos := item.GetRepos() - plans := item.GetPlans() - for _, repo := range repos { - peerState.KnownRepos[repo.Id] = repo - } - for _, plan := range plans { - peerState.KnownPlans[plan.Id] = plan - } - }) - return nil -} - -func (h *syncSessionHandlerServer) HandleSendLogData(ctx context.Context, stream *bidiSyncCommandStream, item *v1.SyncStreamItem_SyncActionSendLogData) error { - h.mgr.mu.Lock() - defer h.mgr.mu.Unlock() - - logID := item.GetLogId() - if logID == "" { - return NewSyncErrorProtocol(errors.New("action SendLogData: log ID is required")) + zap.Any("repos", item.GetRepoIds()), + zap.Any("plans", item.GetPlanIds())) + peerState := h.mgr.peerStateManager.GetPeerState(h.peer.Keyid).Clone() + if peerState == nil { + return NewSyncErrorInternal(fmt.Errorf("peer state for %q not found", h.peer.Keyid)) } - if _, exists := h.requestedLogStreams[logID]; !exists { - return NewSyncErrorProtocol(fmt.Errorf("action SendLogData: log ID %q was not requested", logID)) + repos := item.GetRepoIds() + plans := item.GetPlanIds() + for _, repoID := range repos { + peerState.KnownRepos[repoID] = struct{}{} } - - var logSink io.WriteCloser - if s, exists := h.activeLogStreams[logID]; !exists { - // Check if there are too many active log streams - if len(h.activeLogStreams) >= 16 { - return NewSyncErrorProtocol(fmt.Errorf("action SendLogData: too many active log streams, limit is 16")) - } - - // If the log stream is not active, we need to create a new one. - f, err := h.mgr.logStore.Create(logID, 0, 24*3600) // 24 hour retention, will just be re-requested next time it's wanted. - if err != nil { - return fmt.Errorf("action SendLogData: creating log stream %q: %w", logID, err) - } - logSink = f - h.activeLogStreams[logID] = logSink - } else { - logSink = s + for _, planID := range plans { + peerState.KnownPlans[planID] = struct{}{} } - - if len(item.GetChunk()) == 0 { - delete(h.activeLogStreams, logID) - delete(h.requestedLogStreams, logID) - if err := logSink.Close(); err != nil { - return fmt.Errorf("action SendLogData: closing log stream %q: %w", logID, err) - } - return nil - } - - if _, err := logSink.Write(item.GetChunk()); err != nil { - return fmt.Errorf("action SendLogData: writing to log stream %q: %w", logID, err) - } - + h.mgr.peerStateManager.SetPeerState(h.peer.Keyid, peerState) return nil } @@ -545,14 +430,14 @@ func (h *syncSessionHandlerServer) sendConfigToClient(stream *bidiSyncCommandStr for _, repo := range config.Repos { if h.permissions.CheckPermissionForRepo(repo.Id, v1.Multihost_Permission_PERMISSION_READ_CONFIG) { remoteConfig.Repos = append(remoteConfig.Repos, repo) - resourceListMsg.Repos = append(resourceListMsg.Repos, &v1.SyncRepoMetadata{Id: repo.Id, Guid: repo.Guid}) + resourceListMsg.RepoIds = append(resourceListMsg.RepoIds, repo.Id) allowedRepoIDs = append(allowedRepoIDs, repo.Id) } } for _, plan := range config.Plans { if h.permissions.CheckPermissionForPlan(plan.Id, v1.Multihost_Permission_PERMISSION_READ_CONFIG) { remoteConfig.Plans = append(remoteConfig.Plans, plan) - resourceListMsg.Plans = append(resourceListMsg.Plans, &v1.SyncPlanMetadata{Id: plan.Id}) + resourceListMsg.PlanIds = append(resourceListMsg.PlanIds, plan.Id) allowedPlanIDs = append(allowedPlanIDs, plan.Id) } } diff --git a/internal/api/syncapi/syncmanager.go b/internal/api/syncapi/syncmanager.go index a96923e4..550156c0 100644 --- a/internal/api/syncapi/syncmanager.go +++ b/internal/api/syncapi/syncmanager.go @@ -11,7 +11,6 @@ import ( v1 "github.com/garethgeorge/backrest/gen/go/v1" "github.com/garethgeorge/backrest/internal/config" "github.com/garethgeorge/backrest/internal/cryptoutil" - "github.com/garethgeorge/backrest/internal/logstore" "github.com/garethgeorge/backrest/internal/oplog" "github.com/garethgeorge/backrest/internal/orchestrator" "go.uber.org/zap" @@ -21,7 +20,6 @@ type SyncManager struct { configMgr *config.ConfigManager orchestrator *orchestrator.Orchestrator oplog *oplog.OpLog - logStore *logstore.LogStore // mutable properties mu sync.Mutex @@ -30,27 +28,32 @@ type SyncManager struct { syncClientRetryDelay time.Duration // the default retry delay for sync clients, protected by mu - syncClients map[string]*SyncClient // current sync clients, protected by mu - sessionHandlerMap map[string]*syncSessionHandlerServer // handlers for sync sessions keyed by peer KeyID, protected by mu + syncClients map[string]*SyncClient // current sync clients, protected by mu peerStateManager PeerStateManager } -func NewSyncManager(configMgr *config.ConfigManager, oplog *oplog.OpLog, logStore *logstore.LogStore, orchestrator *orchestrator.Orchestrator, peerStateManager PeerStateManager) *SyncManager { +func NewSyncManager(configMgr *config.ConfigManager, oplog *oplog.OpLog, orchestrator *orchestrator.Orchestrator, peerStateManager PeerStateManager) *SyncManager { // Fetch the config, and mark all sync clients and known hosts as disconnected (but preserve other fields). config, err := configMgr.Get() if err == nil { for _, knownHostPeer := range config.GetMultihost().GetKnownHosts() { - peerStateManager.UpdatePeerState(knownHostPeer.Keyid, knownHostPeer.InstanceId, func(state *PeerState) { - state.ConnectionState = v1.SyncConnectionState_CONNECTION_STATE_DISCONNECTED - state.ConnectionStateMessage = "disconnected" - }) + state := peerStateManager.GetPeerState(knownHostPeer.Keyid).Clone() + if state == nil { + state = newPeerState(knownHostPeer.InstanceId, knownHostPeer.Keyid) + } + state.ConnectionState = v1.SyncConnectionState_CONNECTION_STATE_DISCONNECTED + state.ConnectionStateMessage = "disconnected" + peerStateManager.SetPeerState(knownHostPeer.Keyid, state) } for _, authorizedClient := range config.GetMultihost().GetAuthorizedClients() { - peerStateManager.UpdatePeerState(authorizedClient.Keyid, authorizedClient.InstanceId, func(state *PeerState) { - state.ConnectionState = v1.SyncConnectionState_CONNECTION_STATE_DISCONNECTED - state.ConnectionStateMessage = "disconnected" - }) + state := peerStateManager.GetPeerState(authorizedClient.Keyid).Clone() + if state == nil { + state = newPeerState(authorizedClient.InstanceId, authorizedClient.Keyid) + } + state.ConnectionState = v1.SyncConnectionState_CONNECTION_STATE_DISCONNECTED + state.ConnectionStateMessage = "disconnected" + peerStateManager.SetPeerState(authorizedClient.Keyid, state) } } else { zap.S().Errorf("syncmanager failed to get initial config: %v", err) @@ -59,11 +62,9 @@ func NewSyncManager(configMgr *config.ConfigManager, oplog *oplog.OpLog, logStor configMgr: configMgr, orchestrator: orchestrator, oplog: oplog, - logStore: logStore, syncClientRetryDelay: 60 * time.Second, syncClients: make(map[string]*SyncClient), - sessionHandlerMap: make(map[string]*syncSessionHandlerServer), peerStateManager: peerStateManager, } @@ -151,7 +152,6 @@ func (m *SyncManager) RunSync(ctx context.Context) { for { select { case <-ctx.Done(): - zap.S().Debugf("syncmanager context canceled for instance %q, stopping sync", m.snapshot.config.GetInstance()) return case <-configWatchCh: runSyncWithNewConfig() diff --git a/internal/kvstore/sqlitedb.go b/internal/kvstore/sqlitedb.go index a940fc9f..d01f4221 100644 --- a/internal/kvstore/sqlitedb.go +++ b/internal/kvstore/sqlitedb.go @@ -42,10 +42,10 @@ func NewSqliteDbForKvStore(db string) (*sql.DB, error) { return dbpool, nil } -func NewInMemorySqliteDbForKvStore(t testing.TB) (*sql.DB, error) { +func NewInMemorySqliteDbForKvStore(t testing.TB) *sql.DB { dbpool, err := sql.Open("sqlite3", memdb.TestDB(t)) if err != nil { - return nil, fmt.Errorf("open sqlite pool: %v", err) + t.Fatalf("failed to open db: %v", err) } - return dbpool, nil + return dbpool } diff --git a/internal/logstore/logstore.go b/internal/logstore/logstore.go index fe30f7fc..35d695a0 100644 --- a/internal/logstore/logstore.go +++ b/internal/logstore/logstore.go @@ -270,23 +270,23 @@ func (ls *LogStore) SelectAll(f func(id string, parentID int64)) error { // Find logs owned by a specific operation ID. func (ls *LogStore) FindLogsWithParent(parentOpID int64) ([]string, error) { - conn, err := ls.dbpool.Take(context.Background()) + rows, err := ls.dbpool.QueryContext(context.Background(), "SELECT id FROM logs WHERE owner_opid = ?", parentOpID) if err != nil { - return nil, fmt.Errorf("take connection: %v", err) - } - defer ls.dbpool.Put(conn) - - var logs []string - if err := sqlitex.Execute(conn, "SELECT id FROM logs WHERE owner_opid = ?", &sqlitex.ExecOptions{ - Args: []any{parentOpID}, - ResultFunc: func(stmt *sqlite.Stmt) error { - logs = append(logs, stmt.ColumnText(0)) - return nil - }, - }); err != nil { return nil, fmt.Errorf("select logs: %v", err) } + defer rows.Close() + var logs []string + for rows.Next() { + var id string + if err := rows.Scan(&id); err != nil { + return nil, fmt.Errorf("scan log id: %v", err) + } + logs = append(logs, id) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("iterate logs: %v", err) + } return logs, nil }