benjamyn/any-sync
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #40 from anyproto/acl-change

Browse Source
This commit is contained in:
Mikhail Rakhmanov 2023-07-02 15:54:07 +02:00 committed by GitHub
commit b10d72a092
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
39 changed files with 7590 additions and 3794 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
Makefile
View File

@ -20,6 +20,7 @@ proto:
protoc --gogofaster_out=$(PKGMAP):. --go-drpc_out=protolib=github.com/gogo/protobuf:. net/streampool/testservice/protos/*.proto
protoc --gogofaster_out=:. net/secureservice/handshake/handshakeproto/protos/*.proto
protoc --gogofaster_out=$(PKGMAP):. --go-drpc_out=protolib=github.com/gogo/protobuf:. coordinator/coordinatorproto/protos/*.proto
protoc --gogofaster_out=:. --go-drpc_out=protolib=github.com/gogo/protobuf:. consensus/consensusproto/protos/*.proto
deps:
go mod download

9
commonspace/headsync/diffsyncer_test.go
View File

@ -4,18 +4,19 @@ import (
"bytes"
"context"
"fmt"
"testing"
"time"
"github.com/anyproto/any-sync/app/ldiff"
"github.com/anyproto/any-sync/commonspace/object/acl/aclrecordproto"
"github.com/anyproto/any-sync/commonspace/object/acl/liststorage/mock_liststorage"
"github.com/anyproto/any-sync/commonspace/object/tree/treechangeproto"
"github.com/anyproto/any-sync/commonspace/object/tree/treestorage/mock_treestorage"
"github.com/anyproto/any-sync/commonspace/spacesyncproto"
"github.com/anyproto/any-sync/consensus/consensusproto"
"github.com/anyproto/any-sync/net/peer"
"github.com/golang/mock/gomock"
"github.com/stretchr/testify/require"
"storj.io/drpc"
"testing"
"time"
)
type pushSpaceRequestMatcher struct {
@ -169,7 +170,7 @@ func TestDiffSyncer(t *testing.T) {
settingsStorage := mock_treestorage.NewMockTreeStorage(fx.ctrl)
settingsId := "settingsId"
aclRootId := "aclRootId"
aclRoot := &aclrecordproto.RawAclRecordWithId{
aclRoot := &consensusproto.RawRecordWithId{
Id: aclRootId,
}
settingsRoot := &treechangeproto.RawTreeChangeWithId{

5904
commonspace/object/acl/aclrecordproto/aclrecord.pb.go
View File

File diff suppressed because it is too large Load Diff

132
commonspace/object/acl/aclrecordproto/protos/aclrecord.proto
View File

@ -2,26 +2,7 @@ syntax = "proto3";
package aclrecord;
option go_package = "commonspace/object/acl/aclrecordproto";
message RawAclRecord {
bytes payload = 1;
bytes signature = 2;
bytes acceptorIdentity = 3;
bytes acceptorSignature = 4;
}
message RawAclRecordWithId {
bytes payload = 1;
string id = 2;
}
message AclRecord {
string prevId = 1;
bytes identity = 2;
bytes data = 3;
string readKeyId = 4;
int64 timestamp = 5;
}
// AclRoot is a root of access control list
message AclRoot {
bytes identity = 1;
bytes masterKey = 2;
@ -31,82 +12,95 @@ message AclRoot {
bytes identitySignature = 6;
}
message AclContentValue {
oneof value {
AclUserAdd userAdd = 1;
AclUserRemove userRemove = 2;
AclUserPermissionChange userPermissionChange = 3;
AclUserInvite userInvite = 4;
AclUserJoin userJoin = 5;
}
// AclAccountInvite contains the public invite key, the private part of which is sent to the user directly
message AclAccountInvite {
bytes inviteKey = 1;
}
message AclData {
repeated AclContentValue aclContent = 1;
// AclAccountRequestJoin contains the reference to the invite record and the data of the person who wants to join, confirmed by the private invite key
message AclAccountRequestJoin {
bytes inviteIdentity = 1;
string inviteRecordId = 2;
bytes inviteIdentitySignature = 3;
bytes metadata = 4;
}
message AclState {
repeated string readKeyIds = 1;
repeated AclUserState userStates = 2;
map<string, AclUserInvite> invites = 3;
}
message AclUserState {
// AclAccountRequestAccept contains the reference to join record and all read keys, encrypted with the identity of the requestor
message AclAccountRequestAccept {
bytes identity = 1;
AclUserPermissions permissions = 2;
string requestRecordId = 2;
repeated AclReadKeyWithRecord encryptedReadKeys = 3;
AclUserPermissions permissions = 4;
}
message AclUserAdd {
bytes identity = 1;
repeated bytes encryptedReadKeys = 2;
AclUserPermissions permissions = 3;
// AclAccountRequestDecline contains the reference to join record
message AclAccountRequestDecline {
string requestRecordId = 1;
}
message AclUserInvite {
bytes acceptPublicKey = 1;
repeated bytes encryptedReadKeys = 2;
AclUserPermissions permissions = 3;
// AclAccountInviteRevoke revokes the invite record
message AclAccountInviteRevoke {
string inviteRecordId = 1;
}
message AclUserJoin {
bytes identity = 1;
bytes acceptSignature = 2;
bytes acceptPubKey = 3;
repeated bytes encryptedReadKeys = 4;
// AclReadKeys are a read key with record id
message AclReadKeyWithRecord {
string recordId = 1;
bytes encryptedReadKey = 2;
}
message AclUserRemove {
bytes identity = 1;
repeated AclReadKeyReplace readKeyReplaces = 2;
}
message AclReadKeyReplace {
// AclEncryptedReadKeys are new key for specific identity
message AclEncryptedReadKey {
bytes identity = 1;
bytes encryptedReadKey = 2;
}
message AclUserPermissionChange {
// AclAccountPermissionChange changes permissions of specific account
message AclAccountPermissionChange {
bytes identity = 1;
AclUserPermissions permissions = 2;
}
enum AclUserPermissions {
Admin = 0;
Writer = 1;
Reader = 2;
// AclReadKeyChange changes the key for a space
message AclReadKeyChange {
repeated AclEncryptedReadKey accountKeys = 1;
}
message AclSyncMessage {
AclSyncContentValue content = 1;
// AclAccountRemove removes an account and changes read key for space
message AclAccountRemove {
repeated bytes identities = 1;
repeated AclEncryptedReadKey accountKeys = 2;
}
// AclSyncContentValue provides different types for acl sync
message AclSyncContentValue {
// AclAccountRequestRemove adds a request to remove an account
message AclAccountRequestRemove {
}
// AclContentValue contains possible values for Acl
message AclContentValue {
oneof value {
AclAddRecords addRecords = 1;
AclAccountInvite invite = 1;
AclAccountInviteRevoke inviteRevoke = 2;
AclAccountRequestJoin requestJoin = 3;
AclAccountRequestAccept requestAccept = 4;
AclAccountPermissionChange permissionChange = 5;
AclAccountRemove accountRemove = 6;
AclReadKeyChange readKeyChange = 7;
AclAccountRequestDecline requestDecline = 8;
AclAccountRequestRemove accountRequestRemove = 9;
}
}
message AclAddRecords {
repeated RawAclRecordWithId records = 1;
// AclData contains different acl content
message AclData {
repeated AclContentValue aclContent = 1;
}
// AclUserPermissions contains different possible user roles
enum AclUserPermissions {
None = 0;
Owner = 1;
Admin = 2;
Writer = 3;
Reader = 4;
}

402
commonspace/object/acl/list/aclrecordbuilder.go
View File

@ -1,11 +1,14 @@
package list
import (
"time"
"github.com/anyproto/any-sync/commonspace/object/accountdata"
"github.com/anyproto/any-sync/commonspace/object/acl/aclrecordproto"
"github.com/anyproto/any-sync/consensus/consensusproto"
"github.com/anyproto/any-sync/util/cidutil"
"github.com/anyproto/any-sync/util/crypto"
"github.com/gogo/protobuf/proto"
"time"
)
type RootContent struct {
@ -15,26 +18,387 @@ type RootContent struct {
EncryptedReadKey []byte
}
type RequestJoinPayload struct {
InviteRecordId string
InviteKey crypto.PrivKey
Metadata []byte
}
type RequestAcceptPayload struct {
RequestRecordId string
Permissions AclPermissions
}
type PermissionChangePayload struct {
Identity crypto.PubKey
Permissions AclPermissions
}
type AccountRemovePayload struct {
Identities []crypto.PubKey
ReadKey crypto.SymKey
}
type InviteResult struct {
InviteRec *consensusproto.RawRecord
InviteKey crypto.PrivKey
}
type AclRecordBuilder interface {
Unmarshall(rawIdRecord *aclrecordproto.RawAclRecordWithId) (rec *AclRecord, err error)
BuildRoot(content RootContent) (rec *aclrecordproto.RawAclRecordWithId, err error)
UnmarshallWithId(rawIdRecord *consensusproto.RawRecordWithId) (rec *AclRecord, err error)
Unmarshall(rawRecord *consensusproto.RawRecord) (rec *AclRecord, err error)
BuildRoot(content RootContent) (rec *consensusproto.RawRecordWithId, err error)
BuildInvite() (res InviteResult, err error)
BuildInviteRevoke(inviteRecordId string) (rawRecord *consensusproto.RawRecord, err error)
BuildRequestJoin(payload RequestJoinPayload) (rawRecord *consensusproto.RawRecord, err error)
BuildRequestAccept(payload RequestAcceptPayload) (rawRecord *consensusproto.RawRecord, err error)
BuildRequestDecline(requestRecordId string) (rawRecord *consensusproto.RawRecord, err error)
BuildRequestRemove() (rawRecord *consensusproto.RawRecord, err error)
BuildPermissionChange(payload PermissionChangePayload) (rawRecord *consensusproto.RawRecord, err error)
BuildReadKeyChange(newKey crypto.SymKey) (rawRecord *consensusproto.RawRecord, err error)
BuildAccountRemove(payload AccountRemovePayload) (rawRecord *consensusproto.RawRecord, err error)
}
type aclRecordBuilder struct {
id string
keyStorage crypto.KeyStorage
accountKeys *accountdata.AccountKeys
verifier AcceptorVerifier
state *AclState
}
func NewAclRecordBuilder(id string, keyStorage crypto.KeyStorage) AclRecordBuilder {
func NewAclRecordBuilder(id string, keyStorage crypto.KeyStorage, keys *accountdata.AccountKeys, verifier AcceptorVerifier) AclRecordBuilder {
return &aclRecordBuilder{
id: id,
keyStorage: keyStorage,
accountKeys: keys,
verifier: verifier,
}
}
func (a *aclRecordBuilder) Unmarshall(rawIdRecord *aclrecordproto.RawAclRecordWithId) (rec *AclRecord, err error) {
func (a *aclRecordBuilder) buildRecord(aclContent *aclrecordproto.AclContentValue) (rawRec *consensusproto.RawRecord, err error) {
aclData := &aclrecordproto.AclData{AclContent: []*aclrecordproto.AclContentValue{
aclContent,
}}
marshalledData, err := aclData.Marshal()
if err != nil {
return
}
protoKey, err := a.accountKeys.SignKey.GetPublic().Marshall()
if err != nil {
return
}
rec := &consensusproto.Record{
PrevId: a.state.lastRecordId,
Identity: protoKey,
Data: marshalledData,
Timestamp: time.Now().Unix(),
}
marshalledRec, err := rec.Marshal()
if err != nil {
return
}
signature, err := a.accountKeys.SignKey.Sign(marshalledRec)
if err != nil {
return
}
rawRec = &consensusproto.RawRecord{
Payload: marshalledRec,
Signature: signature,
}
return
}
func (a *aclRecordBuilder) BuildInvite() (res InviteResult, err error) {
if !a.state.Permissions(a.state.pubKey).CanManageAccounts() {
err = ErrInsufficientPermissions
return
}
privKey, pubKey, err := crypto.GenerateRandomEd25519KeyPair()
if err != nil {
return
}
invitePubKey, err := pubKey.Marshall()
if err != nil {
return
}
inviteRec := &aclrecordproto.AclAccountInvite{InviteKey: invitePubKey}
content := &aclrecordproto.AclContentValue{Value: &aclrecordproto.AclContentValue_Invite{Invite: inviteRec}}
rawRec, err := a.buildRecord(content)
if err != nil {
return
}
res.InviteKey = privKey
res.InviteRec = rawRec
return
}
func (a *aclRecordBuilder) BuildInviteRevoke(inviteRecordId string) (rawRecord *consensusproto.RawRecord, err error) {
if !a.state.Permissions(a.state.pubKey).CanManageAccounts() {
err = ErrInsufficientPermissions
return
}
_, exists := a.state.inviteKeys[inviteRecordId]
if !exists {
err = ErrNoSuchInvite
return
}
revokeRec := &aclrecordproto.AclAccountInviteRevoke{InviteRecordId: inviteRecordId}
content := &aclrecordproto.AclContentValue{Value: &aclrecordproto.AclContentValue_InviteRevoke{InviteRevoke: revokeRec}}
return a.buildRecord(content)
}
func (a *aclRecordBuilder) BuildRequestJoin(payload RequestJoinPayload) (rawRecord *consensusproto.RawRecord, err error) {
key, exists := a.state.inviteKeys[payload.InviteRecordId]
if !exists {
err = ErrNoSuchInvite
return
}
if !payload.InviteKey.GetPublic().Equals(key) {
err = ErrIncorrectInviteKey
}
rawIdentity, err := a.accountKeys.SignKey.GetPublic().Raw()
if err != nil {
return
}
signature, err := payload.InviteKey.Sign(rawIdentity)
if err != nil {
return
}
protoIdentity, err := a.accountKeys.SignKey.GetPublic().Marshall()
if err != nil {
return
}
joinRec := &aclrecordproto.AclAccountRequestJoin{
InviteIdentity: protoIdentity,
InviteRecordId: payload.InviteRecordId,
InviteIdentitySignature: signature,
Metadata: payload.Metadata,
}
content := &aclrecordproto.AclContentValue{Value: &aclrecordproto.AclContentValue_RequestJoin{RequestJoin: joinRec}}
return a.buildRecord(content)
}
func (a *aclRecordBuilder) BuildRequestAccept(payload RequestAcceptPayload) (rawRecord *consensusproto.RawRecord, err error) {
if !a.state.Permissions(a.state.pubKey).CanManageAccounts() {
err = ErrInsufficientPermissions
return
}
request, exists := a.state.requestRecords[payload.RequestRecordId]
if !exists {
err = ErrNoSuchRequest
return
}
var encryptedReadKeys []*aclrecordproto.AclReadKeyWithRecord
for keyId, key := range a.state.userReadKeys {
rawKey, err := key.Raw()
if err != nil {
return nil, err
}
enc, err := request.RequestIdentity.Encrypt(rawKey)
if err != nil {
return nil, err
}
encryptedReadKeys = append(encryptedReadKeys, &aclrecordproto.AclReadKeyWithRecord{
RecordId: keyId,
EncryptedReadKey: enc,
})
}
if err != nil {
return
}
requestIdentityProto, err := request.RequestIdentity.Marshall()
if err != nil {
return
}
acceptRec := &aclrecordproto.AclAccountRequestAccept{
Identity: requestIdentityProto,
RequestRecordId: payload.RequestRecordId,
EncryptedReadKeys: encryptedReadKeys,
Permissions: aclrecordproto.AclUserPermissions(payload.Permissions),
}
content := &aclrecordproto.AclContentValue{Value: &aclrecordproto.AclContentValue_RequestAccept{RequestAccept: acceptRec}}
return a.buildRecord(content)
}
func (a *aclRecordBuilder) BuildRequestDecline(requestRecordId string) (rawRecord *consensusproto.RawRecord, err error) {
if !a.state.Permissions(a.state.pubKey).CanManageAccounts() {
err = ErrInsufficientPermissions
return
}
_, exists := a.state.requestRecords[requestRecordId]
if !exists {
err = ErrNoSuchRequest
return
}
declineRec := &aclrecordproto.AclAccountRequestDecline{RequestRecordId: requestRecordId}
content := &aclrecordproto.AclContentValue{Value: &aclrecordproto.AclContentValue_RequestDecline{RequestDecline: declineRec}}
return a.buildRecord(content)
}
func (a *aclRecordBuilder) BuildPermissionChange(payload PermissionChangePayload) (rawRecord *consensusproto.RawRecord, err error) {
permissions := a.state.Permissions(a.state.pubKey)
if !permissions.CanManageAccounts() || payload.Identity.Equals(a.state.pubKey) {
err = ErrInsufficientPermissions
return
}
if payload.Permissions.IsOwner() {
err = ErrIsOwner
return
}
protoIdentity, err := payload.Identity.Marshall()
if err != nil {
return
}
permissionRec := &aclrecordproto.AclAccountPermissionChange{
Identity: protoIdentity,
Permissions: aclrecordproto.AclUserPermissions(payload.Permissions),
}
content := &aclrecordproto.AclContentValue{Value: &aclrecordproto.AclContentValue_PermissionChange{PermissionChange: permissionRec}}
return a.buildRecord(content)
}
func (a *aclRecordBuilder) BuildReadKeyChange(newKey crypto.SymKey) (rawRecord *consensusproto.RawRecord, err error) {
if !a.state.Permissions(a.state.pubKey).CanManageAccounts() {
err = ErrInsufficientPermissions
return
}
rawKey, err := newKey.Raw()
if err != nil {
return
}
if len(rawKey) != crypto.KeyBytes {
err = ErrIncorrectReadKey
return
}
var aclReadKeys []*aclrecordproto.AclEncryptedReadKey
for _, st := range a.state.userStates {
protoIdentity, err := st.PubKey.Marshall()
if err != nil {
return nil, err
}
enc, err := st.PubKey.Encrypt(rawKey)
if err != nil {
return nil, err
}
aclReadKeys = append(aclReadKeys, &aclrecordproto.AclEncryptedReadKey{
Identity: protoIdentity,
EncryptedReadKey: enc,
})
}
readRec := &aclrecordproto.AclReadKeyChange{AccountKeys: aclReadKeys}
content := &aclrecordproto.AclContentValue{Value: &aclrecordproto.AclContentValue_ReadKeyChange{ReadKeyChange: readRec}}
return a.buildRecord(content)
}
func (a *aclRecordBuilder) BuildAccountRemove(payload AccountRemovePayload) (rawRecord *consensusproto.RawRecord, err error) {
deletedMap := map[string]struct{}{}
for _, key := range payload.Identities {
permissions := a.state.Permissions(key)
if permissions.IsOwner() {
return nil, ErrInsufficientPermissions
}
if permissions.NoPermissions() {
return nil, ErrNoSuchAccount
}
deletedMap[mapKeyFromPubKey(key)] = struct{}{}
}
if !a.state.Permissions(a.state.pubKey).CanManageAccounts() {
err = ErrInsufficientPermissions
return
}
rawKey, err := payload.ReadKey.Raw()
if err != nil {
return
}
if len(rawKey) != crypto.KeyBytes {
err = ErrIncorrectReadKey
return
}
var aclReadKeys []*aclrecordproto.AclEncryptedReadKey
for _, st := range a.state.userStates {
if _, exists := deletedMap[mapKeyFromPubKey(st.PubKey)]; exists {
continue
}
protoIdentity, err := st.PubKey.Marshall()
if err != nil {
return nil, err
}
enc, err := st.PubKey.Encrypt(rawKey)
if err != nil {
return nil, err
}
aclReadKeys = append(aclReadKeys, &aclrecordproto.AclEncryptedReadKey{
Identity: protoIdentity,
EncryptedReadKey: enc,
})
}
var marshalledIdentities [][]byte
for _, key := range payload.Identities {
protoIdentity, err := key.Marshall()
if err != nil {
return nil, err
}
marshalledIdentities = append(marshalledIdentities, protoIdentity)
}
removeRec := &aclrecordproto.AclAccountRemove{AccountKeys: aclReadKeys, Identities: marshalledIdentities}
content := &aclrecordproto.AclContentValue{Value: &aclrecordproto.AclContentValue_AccountRemove{AccountRemove: removeRec}}
return a.buildRecord(content)
}
func (a *aclRecordBuilder) BuildRequestRemove() (rawRecord *consensusproto.RawRecord, err error) {
permissions := a.state.Permissions(a.state.pubKey)
if permissions.NoPermissions() {
err = ErrNoSuchAccount
return
}
if permissions.IsOwner() {
err = ErrIsOwner
return
}
removeRec := &aclrecordproto.AclAccountRequestRemove{}
content := &aclrecordproto.AclContentValue{Value: &aclrecordproto.AclContentValue_AccountRequestRemove{AccountRequestRemove: removeRec}}
return a.buildRecord(content)
}
func (a *aclRecordBuilder) Unmarshall(rawRecord *consensusproto.RawRecord) (rec *AclRecord, err error) {
aclRecord := &consensusproto.Record{}
err = proto.Unmarshal(rawRecord.Payload, aclRecord)
if err != nil {
return
}
pubKey, err := a.keyStorage.PubKeyFromProto(aclRecord.Identity)
if err != nil {
return
}
aclData := &aclrecordproto.AclData{}
err = proto.Unmarshal(aclRecord.Data, aclData)
if err != nil {
return
}
rec = &AclRecord{
PrevId: aclRecord.PrevId,
Timestamp: aclRecord.Timestamp,
Data: aclRecord.Data,
Signature: rawRecord.Signature,
Identity: pubKey,
Model: aclData,
}
res, err := pubKey.Verify(rawRecord.Payload, rawRecord.Signature)
if err != nil {
return
}
if !res {
err = ErrInvalidSignature
return
}
return
}
func (a *aclRecordBuilder) UnmarshallWithId(rawIdRecord *consensusproto.RawRecordWithId) (rec *AclRecord, err error) {
var (
rawRec = &aclrecordproto.RawAclRecord{}
rawRec = &consensusproto.RawRecord{}
pubKey crypto.PubKey
)
err = proto.Unmarshal(rawIdRecord.Payload, rawRec)
@ -53,14 +417,17 @@ func (a *aclRecordBuilder) Unmarshall(rawIdRecord *aclrecordproto.RawAclRecordWi
}
rec = &AclRecord{
Id: rawIdRecord.Id,
ReadKeyId: rawIdRecord.Id,
Timestamp: aclRoot.Timestamp,
Signature: rawRec.Signature,
Identity: pubKey,
Model: aclRoot,
}
} else {
aclRecord := &aclrecordproto.AclRecord{}
err = a.verifier.VerifyAcceptor(rawRec)
if err != nil {
return
}
aclRecord := &consensusproto.Record{}
err = proto.Unmarshal(rawRec.Payload, aclRecord)
if err != nil {
return
@ -69,14 +436,19 @@ func (a *aclRecordBuilder) Unmarshall(rawIdRecord *aclrecordproto.RawAclRecordWi
if err != nil {
return
}
aclData := &aclrecordproto.AclData{}
err = proto.Unmarshal(aclRecord.Data, aclData)
if err != nil {
return
}
rec = &AclRecord{
Id: rawIdRecord.Id,
PrevId: aclRecord.PrevId,
ReadKeyId: aclRecord.ReadKeyId,
Timestamp: aclRecord.Timestamp,
Data: aclRecord.Data,
Signature: rawRec.Signature,
Identity: pubKey,
Model: aclData,
}
}
@ -84,7 +456,7 @@ func (a *aclRecordBuilder) Unmarshall(rawIdRecord *aclrecordproto.RawAclRecordWi
return
}
func (a *aclRecordBuilder) BuildRoot(content RootContent) (rec *aclrecordproto.RawAclRecordWithId, err error) {
func (a *aclRecordBuilder) BuildRoot(content RootContent) (rec *consensusproto.RawRecordWithId, err error) {
rawIdentity, err := content.PrivKey.GetPublic().Raw()
if err != nil {
return
@ -118,8 +490,8 @@ func (a *aclRecordBuilder) BuildRoot(content RootContent) (rec *aclrecordproto.R
func verifyRaw(
pubKey crypto.PubKey,
rawRec *aclrecordproto.RawAclRecord,
recWithId *aclrecordproto.RawAclRecordWithId) (err error) {
rawRec *consensusproto.RawRecord,
recWithId *consensusproto.RawRecordWithId) (err error) {
// verifying signature
res, err := pubKey.Verify(rawRec.Payload, rawRec.Signature)
if err != nil {
@ -137,7 +509,7 @@ func verifyRaw(
return
}
func marshalAclRoot(aclRoot *aclrecordproto.AclRoot, key crypto.PrivKey) (rawWithId *aclrecordproto.RawAclRecordWithId, err error) {
func marshalAclRoot(aclRoot *aclrecordproto.AclRoot, key crypto.PrivKey) (rawWithId *consensusproto.RawRecordWithId, err error) {
marshalledRoot, err := aclRoot.Marshal()
if err != nil {
return
@ -146,7 +518,7 @@ func marshalAclRoot(aclRoot *aclrecordproto.AclRoot, key crypto.PrivKey) (rawWit
if err != nil {
return
}
raw := &aclrecordproto.RawAclRecord{
raw := &consensusproto.RawRecord{
Payload: marshalledRoot,
Signature: signature,
}
@ -158,7 +530,7 @@ func marshalAclRoot(aclRoot *aclrecordproto.AclRoot, key crypto.PrivKey) (rawWit
if err != nil {
return
}
rawWithId = &aclrecordproto.RawAclRecordWithId{
rawWithId = &consensusproto.RawRecordWithId{
Payload: marshalledRaw,
Id: aclHeadId,
}

9
commonspace/object/acl/list/aclrecordbuilder_test.go
View File

@ -1,9 +0,0 @@
package list
import (
"testing"
)
func TestAclRecordBuilder_BuildUserJoin(t *testing.T) {
return
}

308
commonspace/object/acl/list/aclstate.go
View File

@ -2,7 +2,7 @@ package list
import (
"errors"
"fmt"
"github.com/anyproto/any-sync/app/logger"
"github.com/anyproto/any-sync/commonspace/object/acl/aclrecordproto"
"github.com/anyproto/any-sync/util/crypto"
@ -13,16 +13,21 @@ import (
var log = logger.NewNamedSugared("common.commonspace.acllist")
var (
ErrNoSuchUser = errors.New("no such user")
ErrNoSuchAccount = errors.New("no such account")
ErrPendingRequest = errors.New("already exists pending request")
ErrUnexpectedContentType = errors.New("unexpected content type")
ErrIncorrectIdentity = errors.New("incorrect identity")
ErrIncorrectInviteKey = errors.New("incorrect invite key")
ErrFailedToDecrypt = errors.New("failed to decrypt key")
ErrUserRemoved = errors.New("user was removed from the document")
ErrDocumentForbidden = errors.New("your user was forbidden access to the document")
ErrUserAlreadyExists = errors.New("user already exists")
ErrNoSuchRecord = errors.New("no such record")
ErrNoSuchRequest = errors.New("no such request")
ErrNoSuchInvite = errors.New("no such invite")
ErrOldInvite = errors.New("invite is too old")
ErrInsufficientPermissions = errors.New("insufficient permissions")
ErrIsOwner = errors.New("can't be made by owner")
ErrIncorrectNumberOfAccounts = errors.New("incorrect number of accounts")
ErrDuplicateAccounts = errors.New("duplicate accounts")
ErrNoReadKey = errors.New("acl state doesn't have a read key")
ErrIncorrectReadKey = errors.New("incorrect read key")
ErrInvalidSignature = errors.New("signature is invalid")
ErrIncorrectRoot = errors.New("incorrect root")
ErrIncorrectRecordSequence = errors.New("incorrect prev id of a record")
@ -36,37 +41,71 @@ type UserPermissionPair struct {
type AclState struct {
id string
currentReadKeyId string
// userReadKeys is a map recordId -> read key which tells us about every read key
userReadKeys map[string]crypto.SymKey
// userStates is a map pubKey -> state which defines current user state
userStates map[string]AclUserState
// statesAtRecord is a map recordId -> state which define user state at particular record
// probably this can grow rather large at some point, so we can maybe optimise later to have:
// - map pubKey -> []recordIds (where recordIds is an array where such identity permissions were changed)
statesAtRecord map[string][]AclUserState
// inviteKeys is a map recordId -> invite
inviteKeys map[string]crypto.PubKey
// requestRecords is a map recordId -> RequestRecord
requestRecords map[string]RequestRecord
// pendingRequests is a map pubKey -> recordId
pendingRequests map[string]string
key crypto.PrivKey
pubKey crypto.PubKey
keyStore crypto.KeyStorage
totalReadKeys int
lastRecordId string
contentValidator ContentValidator
}
func newAclStateWithKeys(
id string,
key crypto.PrivKey) (*AclState, error) {
return &AclState{
st := &AclState{
id: id,
key: key,
pubKey: key.GetPublic(),
userReadKeys: make(map[string]crypto.SymKey),
userStates: make(map[string]AclUserState),
statesAtRecord: make(map[string][]AclUserState),
}, nil
inviteKeys: make(map[string]crypto.PubKey),
requestRecords: make(map[string]RequestRecord),
pendingRequests: make(map[string]string),
keyStore: crypto.NewKeyStorage(),
}
st.contentValidator = &contentValidator{
keyStore: st.keyStore,
aclState: st,
}
return st, nil
}
func newAclState(id string) *AclState {
return &AclState{
st := &AclState{
id: id,
userReadKeys: make(map[string]crypto.SymKey),
userStates: make(map[string]AclUserState),
statesAtRecord: make(map[string][]AclUserState),
inviteKeys: make(map[string]crypto.PubKey),
requestRecords: make(map[string]RequestRecord),
pendingRequests: make(map[string]string),
keyStore: crypto.NewKeyStorage(),
}
st.contentValidator = &contentValidator{
keyStore: st.keyStore,
aclState: st,
}
return st
}
func (st *AclState) Validator() ContentValidator {
return st.contentValidator
}
func (st *AclState) CurrentReadKeyId() string {
@ -74,7 +113,7 @@ func (st *AclState) CurrentReadKeyId() string {
}
func (st *AclState) CurrentReadKey() (crypto.SymKey, error) {
key, exists := st.userReadKeys[st.currentReadKeyId]
key, exists := st.userReadKeys[st.CurrentReadKeyId()]
if !exists {
return nil, ErrNoReadKey
}
@ -97,7 +136,7 @@ func (st *AclState) StateAtRecord(id string, pubKey crypto.PubKey) (AclUserState
return perm, nil
}
}
return AclUserState{}, ErrNoSuchUser
return AclUserState{}, ErrNoSuchAccount
}
func (st *AclState) applyRecord(record *AclRecord) (err error) {
@ -110,17 +149,18 @@ func (st *AclState) applyRecord(record *AclRecord) (err error) {
err = ErrIncorrectRecordSequence
return
}
// if the record is root record
if record.Id == st.id {
err = st.applyRoot(record)
if err != nil {
return
}
st.statesAtRecord[record.Id] = []AclUserState{
{PubKey: record.Identity, Permissions: aclrecordproto.AclUserPermissions_Admin},
st.userStates[mapKeyFromPubKey(record.Identity)],
}
return
}
// if the model is not cached
if record.Model == nil {
aclData := &aclrecordproto.AclData{}
err = proto.Unmarshal(record.Data, aclData)
@ -129,18 +169,16 @@ func (st *AclState) applyRecord(record *AclRecord) (err error) {
}
record.Model = aclData
}
// applying records contents
err = st.applyChangeData(record)
if err != nil {
return
}
// getting all states for users at record
// getting all states for users at record and saving them
var states []AclUserState
for _, state := range st.userStates {
states = append(states, state)
}
st.statesAtRecord[record.Id] = states
return
}
@ -156,9 +194,9 @@ func (st *AclState) applyRoot(record *AclRecord) (err error) {
// adding user to the list
userState := AclUserState{
PubKey: record.Identity,
Permissions: aclrecordproto.AclUserPermissions_Admin,
Permissions: AclPermissions(aclrecordproto.AclUserPermissions_Owner),
}
st.currentReadKeyId = record.ReadKeyId
st.currentReadKeyId = record.Id
st.userStates[mapKeyFromPubKey(record.Identity)] = userState
st.totalReadKeys++
return
@ -181,92 +219,191 @@ func (st *AclState) saveReadKeyFromRoot(record *AclRecord) (err error) {
return
}
}
st.userReadKeys[record.Id] = readKey
return
}
func (st *AclState) applyChangeData(record *AclRecord) (err error) {
defer func() {
if err != nil {
return
}
if record.ReadKeyId != st.currentReadKeyId {
st.totalReadKeys++
st.currentReadKeyId = record.ReadKeyId
}
}()
model := record.Model.(*aclrecordproto.AclData)
if !st.isUserJoin(model) {
// we check signature when we add this to the List, so no need to do it here
if _, exists := st.userStates[mapKeyFromPubKey(record.Identity)]; !exists {
err = ErrNoSuchUser
return
}
// only Admins can do non-user join changes
if !st.HasPermission(record.Identity, aclrecordproto.AclUserPermissions_Admin) {
// TODO: add string encoding
err = fmt.Errorf("user %s must have admin permissions", record.Identity.Account())
return
}
}
for _, ch := range model.GetAclContent() {
if err = st.applyChangeContent(ch, record.Id); err != nil {
if err = st.applyChangeContent(ch, record.Id, record.Identity); err != nil {
log.Info("error while applying changes: %v; ignore", zap.Error(err))
return err
}
}
return nil
}
func (st *AclState) applyChangeContent(ch *aclrecordproto.AclContentValue, recordId string) error {
func (st *AclState) applyChangeContent(ch *aclrecordproto.AclContentValue, recordId string, authorIdentity crypto.PubKey) error {
switch {
case ch.GetUserPermissionChange() != nil:
return st.applyUserPermissionChange(ch.GetUserPermissionChange(), recordId)
case ch.GetUserAdd() != nil:
return st.applyUserAdd(ch.GetUserAdd(), recordId)
case ch.GetUserRemove() != nil:
return st.applyUserRemove(ch.GetUserRemove(), recordId)
case ch.GetUserInvite() != nil:
return st.applyUserInvite(ch.GetUserInvite(), recordId)
case ch.GetUserJoin() != nil:
return st.applyUserJoin(ch.GetUserJoin(), recordId)
case ch.GetPermissionChange() != nil:
return st.applyPermissionChange(ch.GetPermissionChange(), recordId, authorIdentity)
case ch.GetInvite() != nil:
return st.applyInvite(ch.GetInvite(), recordId, authorIdentity)
case ch.GetInviteRevoke() != nil:
return st.applyInviteRevoke(ch.GetInviteRevoke(), recordId, authorIdentity)
case ch.GetRequestJoin() != nil:
return st.applyRequestJoin(ch.GetRequestJoin(), recordId, authorIdentity)
case ch.GetRequestAccept() != nil:
return st.applyRequestAccept(ch.GetRequestAccept(), recordId, authorIdentity)
case ch.GetRequestDecline() != nil:
return st.applyRequestDecline(ch.GetRequestDecline(), recordId, authorIdentity)
case ch.GetAccountRemove() != nil:
return st.applyAccountRemove(ch.GetAccountRemove(), recordId, authorIdentity)
case ch.GetReadKeyChange() != nil:
return st.applyReadKeyChange(ch.GetReadKeyChange(), recordId, authorIdentity)
case ch.GetAccountRequestRemove() != nil:
return st.applyRequestRemove(ch.GetAccountRequestRemove(), recordId, authorIdentity)
default:
return fmt.Errorf("unexpected change type: %v", ch)
return ErrUnexpectedContentType
}
}
func (st *AclState) applyUserPermissionChange(ch *aclrecordproto.AclUserPermissionChange, recordId string) error {
func (st *AclState) applyPermissionChange(ch *aclrecordproto.AclAccountPermissionChange, recordId string, authorIdentity crypto.PubKey) error {
chIdentity, err := st.keyStore.PubKeyFromProto(ch.Identity)
if err != nil {
return err
}
state, exists := st.userStates[mapKeyFromPubKey(chIdentity)]
if !exists {
return ErrNoSuchUser
err = st.contentValidator.ValidatePermissionChange(ch, authorIdentity)
if err != nil {
return err
}
state.Permissions = ch.Permissions
stringKey := mapKeyFromPubKey(chIdentity)
state, _ := st.userStates[stringKey]
state.Permissions = AclPermissions(ch.Permissions)
st.userStates[stringKey] = state
return nil
}
func (st *AclState) applyUserInvite(ch *aclrecordproto.AclUserInvite, recordId string) error {
// TODO: check old code and bring it back :-)
func (st *AclState) applyInvite(ch *aclrecordproto.AclAccountInvite, recordId string, authorIdentity crypto.PubKey) error {
inviteKey, err := st.keyStore.PubKeyFromProto(ch.InviteKey)
if err != nil {
return err
}
err = st.contentValidator.ValidateInvite(ch, authorIdentity)
if err != nil {
return err
}
st.inviteKeys[recordId] = inviteKey
return nil
}
func (st *AclState) applyUserJoin(ch *aclrecordproto.AclUserJoin, recordId string) error {
func (st *AclState) applyInviteRevoke(ch *aclrecordproto.AclAccountInviteRevoke, recordId string, authorIdentity crypto.PubKey) error {
err := st.contentValidator.ValidateInviteRevoke(ch, authorIdentity)
if err != nil {
return err
}
delete(st.inviteKeys, ch.InviteRecordId)
return nil
}
func (st *AclState) applyUserAdd(ch *aclrecordproto.AclUserAdd, recordId string) error {
func (st *AclState) applyRequestJoin(ch *aclrecordproto.AclAccountRequestJoin, recordId string, authorIdentity crypto.PubKey) error {
err := st.contentValidator.ValidateRequestJoin(ch, authorIdentity)
if err != nil {
return err
}
st.pendingRequests[mapKeyFromPubKey(authorIdentity)] = recordId
st.requestRecords[recordId] = RequestRecord{
RequestIdentity: authorIdentity,
RequestMetadata: ch.Metadata,
Type: RequestTypeJoin,
}
return nil
}
func (st *AclState) applyUserRemove(ch *aclrecordproto.AclUserRemove, recordId string) error {
func (st *AclState) applyRequestAccept(ch *aclrecordproto.AclAccountRequestAccept, recordId string, authorIdentity crypto.PubKey) error {
err := st.contentValidator.ValidateRequestAccept(ch, authorIdentity)
if err != nil {
return err
}
acceptIdentity, err := st.keyStore.PubKeyFromProto(ch.Identity)
if err != nil {
return err
}
record, _ := st.requestRecords[ch.RequestRecordId]
st.userStates[mapKeyFromPubKey(acceptIdentity)] = AclUserState{
PubKey: acceptIdentity,
Permissions: AclPermissions(ch.Permissions),
RequestMetadata: record.RequestMetadata,
}
delete(st.pendingRequests, mapKeyFromPubKey(st.requestRecords[ch.RequestRecordId].RequestIdentity))
if !st.pubKey.Equals(acceptIdentity) {
return nil
}
for _, key := range ch.EncryptedReadKeys {
decrypted, err := st.key.Decrypt(key.EncryptedReadKey)
if err != nil {
return err
}
sym, err := crypto.UnmarshallAESKey(decrypted)
if err != nil {
return err
}
st.userReadKeys[key.RecordId] = sym
}
return nil
}
func (st *AclState) applyRequestDecline(ch *aclrecordproto.AclAccountRequestDecline, recordId string, authorIdentity crypto.PubKey) error {
err := st.contentValidator.ValidateRequestDecline(ch, authorIdentity)
if err != nil {
return err
}
delete(st.pendingRequests, mapKeyFromPubKey(st.requestRecords[ch.RequestRecordId].RequestIdentity))
delete(st.requestRecords, ch.RequestRecordId)
return nil
}
func (st *AclState) applyRequestRemove(ch *aclrecordproto.AclAccountRequestRemove, recordId string, authorIdentity crypto.PubKey) error {
err := st.contentValidator.ValidateRequestRemove(ch, authorIdentity)
if err != nil {
return err
}
st.requestRecords[recordId] = RequestRecord{
RequestIdentity: authorIdentity,
Type: RequestTypeRemove,
}
st.pendingRequests[mapKeyFromPubKey(authorIdentity)] = recordId
return nil
}
func (st *AclState) applyAccountRemove(ch *aclrecordproto.AclAccountRemove, recordId string, authorIdentity crypto.PubKey) error {
err := st.contentValidator.ValidateAccountRemove(ch, authorIdentity)
if err != nil {
return err
}
for _, rawIdentity := range ch.Identities {
identity, err := st.keyStore.PubKeyFromProto(rawIdentity)
if err != nil {
return err
}
idKey := mapKeyFromPubKey(identity)
delete(st.userStates, idKey)
delete(st.pendingRequests, idKey)
}
return st.updateReadKey(ch.AccountKeys, recordId)
}
func (st *AclState) applyReadKeyChange(ch *aclrecordproto.AclReadKeyChange, recordId string, authorIdentity crypto.PubKey) error {
err := st.contentValidator.ValidateReadKeyChange(ch, authorIdentity)
if err != nil {
return err
}
return st.updateReadKey(ch.AccountKeys, recordId)
}
func (st *AclState) updateReadKey(keys []*aclrecordproto.AclEncryptedReadKey, recordId string) error {
for _, accKey := range keys {
identity, _ := st.keyStore.PubKeyFromProto(accKey.Identity)
if st.pubKey.Equals(identity) {
res, err := st.decryptReadKey(accKey.EncryptedReadKey)
if err != nil {
return err
}
st.userReadKeys[recordId] = res
}
}
st.currentReadKeyId = recordId
return nil
}
@ -275,7 +412,6 @@ func (st *AclState) decryptReadKey(msg []byte) (crypto.SymKey, error) {
if err != nil {
return nil, ErrFailedToDecrypt
}
key, err := crypto.UnmarshallAESKey(decrypted)
if err != nil {
return nil, ErrFailedToDecrypt
@ -283,29 +419,31 @@ func (st *AclState) decryptReadKey(msg []byte) (crypto.SymKey, error) {
return key, nil
}
func (st *AclState) HasPermission(identity crypto.PubKey, permission aclrecordproto.AclUserPermissions) bool {
func (st *AclState) Permissions(identity crypto.PubKey) AclPermissions {
state, exists := st.userStates[mapKeyFromPubKey(identity)]
if !exists {
return false
return AclPermissions(aclrecordproto.AclUserPermissions_None)
}
return state.Permissions
}
return state.Permissions == permission
func (st *AclState) JoinRecords() (records []RequestRecord) {
for _, recId := range st.pendingRequests {
rec := st.requestRecords[recId]
if rec.Type == RequestTypeJoin {
records = append(records, rec)
}
}
return
}
func (st *AclState) isUserJoin(data *aclrecordproto.AclData) bool {
// if we have a UserJoin, then it should always be the first one applied
return data.GetAclContent() != nil && data.GetAclContent()[0].GetUserJoin() != nil
func (st *AclState) RemoveRecords() (records []RequestRecord) {
for _, recId := range st.pendingRequests {
rec := st.requestRecords[recId]
if rec.Type == RequestTypeRemove {
records = append(records, rec)
}
func (st *AclState) isUserAdd(data *aclrecordproto.AclData, identity []byte) bool {
return false
}
func (st *AclState) UserStates() map[string]AclUserState {
return st.userStates
}
func (st *AclState) Invite(acceptPubKey []byte) (invite *aclrecordproto.AclUserInvite, err error) {
return
}

122
commonspace/object/acl/list/list.go
View File

@ -5,16 +5,20 @@ import (
"context"
"errors"
"fmt"
"github.com/anyproto/any-sync/commonspace/object/accountdata"
"github.com/anyproto/any-sync/commonspace/object/acl/aclrecordproto"
"github.com/anyproto/any-sync/commonspace/object/acl/liststorage"
"github.com/anyproto/any-sync/util/crypto"
"sync"
"github.com/anyproto/any-sync/commonspace/object/accountdata"
"github.com/anyproto/any-sync/commonspace/object/acl/liststorage"
"github.com/anyproto/any-sync/consensus/consensusproto"
"github.com/anyproto/any-sync/util/crypto"
)
type IterFunc = func(record *AclRecord) (IsContinue bool)
var ErrIncorrectCID = errors.New("incorrect CID")
var (
ErrIncorrectCID = errors.New("incorrect CID")
ErrRecordAlreadyExists = errors.New("record already exists")
)
type RWLocker interface {
sync.Locker
@ -22,26 +26,41 @@ type RWLocker interface {
RUnlock()
}
type AcceptorVerifier interface {
VerifyAcceptor(rec *consensusproto.RawRecord) (err error)
}
type NoOpAcceptorVerifier struct {
}
func (n NoOpAcceptorVerifier) VerifyAcceptor(rec *consensusproto.RawRecord) (err error) {
return nil
}
type AclList interface {
RWLocker
Id() string
Root() *aclrecordproto.RawAclRecordWithId
Root() *consensusproto.RawRecordWithId
Records() []*AclRecord
AclState() *AclState
IsAfter(first string, second string) (bool, error)
Head() *AclRecord
Get(id string) (*AclRecord, error)
GetIndex(idx int) (*AclRecord, error)
Iterate(iterFunc IterFunc)
IterateFrom(startId string, iterFunc IterFunc)
KeyStorage() crypto.KeyStorage
AddRawRecord(rawRec *aclrecordproto.RawAclRecordWithId) (added bool, err error)
KeyStorage() crypto.KeyStorage
RecordBuilder() AclRecordBuilder
ValidateRawRecord(record *consensusproto.RawRecord) (err error)
AddRawRecord(rawRec *consensusproto.RawRecordWithId) (err error)
Close() (err error)
}
type aclList struct {
root *aclrecordproto.RawAclRecordWithId
root *consensusproto.RawRecordWithId
records []*AclRecord
indexes map[string]int
id string
@ -55,18 +74,45 @@ type aclList struct {
sync.RWMutex
}
func BuildAclListWithIdentity(acc *accountdata.AccountKeys, storage liststorage.ListStorage) (AclList, error) {
builder := newAclStateBuilderWithIdentity(acc)
keyStorage := crypto.NewKeyStorage()
return build(storage.Id(), keyStorage, builder, NewAclRecordBuilder(storage.Id(), keyStorage), storage)
type internalDeps struct {
storage liststorage.ListStorage
keyStorage crypto.KeyStorage
stateBuilder *aclStateBuilder
recordBuilder AclRecordBuilder
acceptorVerifier AcceptorVerifier
}
func BuildAclList(storage liststorage.ListStorage) (AclList, error) {
func BuildAclListWithIdentity(acc *accountdata.AccountKeys, storage liststorage.ListStorage, verifier AcceptorVerifier) (AclList, error) {
keyStorage := crypto.NewKeyStorage()
return build(storage.Id(), keyStorage, newAclStateBuilder(), NewAclRecordBuilder(storage.Id(), crypto.NewKeyStorage()), storage)
deps := internalDeps{
storage: storage,
keyStorage: keyStorage,
stateBuilder: newAclStateBuilderWithIdentity(acc),
recordBuilder: NewAclRecordBuilder(storage.Id(), keyStorage, acc, verifier),
acceptorVerifier: verifier,
}
return build(deps)
}
func build(id string, keyStorage crypto.KeyStorage, stateBuilder *aclStateBuilder, recBuilder AclRecordBuilder, storage liststorage.ListStorage) (list AclList, err error) {
func BuildAclList(storage liststorage.ListStorage, verifier AcceptorVerifier) (AclList, error) {
keyStorage := crypto.NewKeyStorage()
deps := internalDeps{
storage: storage,
keyStorage: keyStorage,
stateBuilder: newAclStateBuilder(),
recordBuilder: NewAclRecordBuilder(storage.Id(), keyStorage, nil, verifier),
acceptorVerifier: verifier,
}
return build(deps)
}
func build(deps internalDeps) (list AclList, err error) {
var (
storage = deps.storage
id = deps.storage.Id()
recBuilder = deps.recordBuilder
stateBuilder = deps.stateBuilder
)
head, err := storage.Head()
if err != nil {
return
@ -77,7 +123,7 @@ func build(id string, keyStorage crypto.KeyStorage, stateBuilder *aclStateBuilde
return
}
record, err := recBuilder.Unmarshall(rawRecordWithId)
record, err := recBuilder.UnmarshallWithId(rawRecordWithId)
if err != nil {
return
}
@ -89,7 +135,7 @@ func build(id string, keyStorage crypto.KeyStorage, stateBuilder *aclStateBuilde
return
}
record, err = recBuilder.Unmarshall(rawRecordWithId)
record, err = recBuilder.UnmarshallWithId(rawRecordWithId)
if err != nil {
return
}
@ -119,6 +165,7 @@ func build(id string, keyStorage crypto.KeyStorage, stateBuilder *aclStateBuilde
return
}
recBuilder.(*aclRecordBuilder).state = state
list = &aclList{
root: rootWithId,
records: records,
@ -132,15 +179,27 @@ func build(id string, keyStorage crypto.KeyStorage, stateBuilder *aclStateBuilde
return
}
func (a *aclList) RecordBuilder() AclRecordBuilder {
return a.recordBuilder
}
func (a *aclList) Records() []*AclRecord {
return a.records
}
func (a *aclList) AddRawRecord(rawRec *aclrecordproto.RawAclRecordWithId) (added bool, err error) {
if _, ok := a.indexes[rawRec.Id]; ok {
func (a *aclList) ValidateRawRecord(rawRec *consensusproto.RawRecord) (err error) {
record, err := a.recordBuilder.Unmarshall(rawRec)
if err != nil {
return
}
record, err := a.recordBuilder.Unmarshall(rawRec)
return a.aclState.Validator().ValidateAclRecordContents(record)
}
func (a *aclList) AddRawRecord(rawRec *consensusproto.RawRecordWithId) (err error) {
if _, ok := a.indexes[rawRec.Id]; ok {
return ErrRecordAlreadyExists
}
record, err := a.recordBuilder.UnmarshallWithId(rawRec)
if err != nil {
return
}
@ -155,15 +214,6 @@ func (a *aclList) AddRawRecord(rawRec *aclrecordproto.RawAclRecordWithId) (added
if err = a.storage.SetHead(rawRec.Id); err != nil {
return
}
return true, nil
}
func (a *aclList) IsValidNext(rawRec *aclrecordproto.RawAclRecordWithId) (err error) {
_, err = a.recordBuilder.Unmarshall(rawRec)
if err != nil {
return
}
// TODO: change state and add "check" method for records
return
}
@ -171,7 +221,7 @@ func (a *aclList) Id() string {
return a.id
}
func (a *aclList) Root() *aclrecordproto.RawAclRecordWithId {
func (a *aclList) Root() *consensusproto.RawRecordWithId {
return a.root
}
@ -199,11 +249,19 @@ func (a *aclList) Head() *AclRecord {
func (a *aclList) Get(id string) (*AclRecord, error) {
recIdx, ok := a.indexes[id]
if !ok {
return nil, fmt.Errorf("no such record")
return nil, ErrNoSuchRecord
}
return a.records[recIdx], nil
}
func (a *aclList) GetIndex(idx int) (*AclRecord, error) {
// TODO: when we add snapshots we will have to monitor record num in snapshots
if idx < 0 || idx >= len(a.records) {
return nil, ErrNoSuchRecord
}
return a.records[idx], nil
}
func (a *aclList) Iterate(iterFunc IterFunc) {
for _, rec := range a.records {
if !iterFunc(rec) {

297
commonspace/object/acl/list/list_test.go
View File

@ -2,11 +2,114 @@ package list
import (
"fmt"
"github.com/anyproto/any-sync/commonspace/object/accountdata"
"github.com/stretchr/testify/require"
"testing"
"github.com/anyproto/any-sync/commonspace/object/accountdata"
"github.com/anyproto/any-sync/commonspace/object/acl/aclrecordproto"
"github.com/anyproto/any-sync/consensus/consensusproto"
"github.com/anyproto/any-sync/util/cidutil"
"github.com/anyproto/any-sync/util/crypto"
"github.com/stretchr/testify/require"
)
func wrapRecord(rawRec *consensusproto.RawRecord) *consensusproto.RawRecordWithId {
payload, err := rawRec.Marshal()
if err != nil {
panic(err)
}
id, err := cidutil.NewCidFromBytes(payload)
if err != nil {
panic(err)
}
return &consensusproto.RawRecordWithId{
Payload: payload,
Id: id,
}
}
type aclFixture struct {
ownerKeys *accountdata.AccountKeys
accountKeys *accountdata.AccountKeys
ownerAcl *aclList
accountAcl *aclList
spaceId string
}
func newFixture(t *testing.T) *aclFixture {
ownerKeys, err := accountdata.NewRandom()
require.NoError(t, err)
accountKeys, err := accountdata.NewRandom()
require.NoError(t, err)
spaceId := "spaceId"
ownerAcl, err := NewTestDerivedAcl(spaceId, ownerKeys)
require.NoError(t, err)
accountAcl, err := NewTestAclWithRoot(accountKeys, ownerAcl.Root())
require.NoError(t, err)
return &aclFixture{
ownerKeys: ownerKeys,
accountKeys: accountKeys,
ownerAcl: ownerAcl.(*aclList),
accountAcl: accountAcl.(*aclList),
spaceId: spaceId,
}
}
func (fx *aclFixture) addRec(t *testing.T, rec *consensusproto.RawRecordWithId) {
err := fx.ownerAcl.AddRawRecord(rec)
require.NoError(t, err)
err = fx.accountAcl.AddRawRecord(rec)
require.NoError(t, err)
}
func (fx *aclFixture) inviteAccount(t *testing.T, perms AclPermissions) {
var (
ownerAcl = fx.ownerAcl
ownerState = fx.ownerAcl.aclState
accountAcl = fx.accountAcl
accountState = fx.accountAcl.aclState
)
// building invite
inv, err := ownerAcl.RecordBuilder().BuildInvite()
require.NoError(t, err)
inviteRec := wrapRecord(inv.InviteRec)
fx.addRec(t, inviteRec)
// building request join
requestJoin, err := accountAcl.RecordBuilder().BuildRequestJoin(RequestJoinPayload{
InviteRecordId: inviteRec.Id,
InviteKey: inv.InviteKey,
})
require.NoError(t, err)
requestJoinRec := wrapRecord(requestJoin)
fx.addRec(t, requestJoinRec)
// building request accept
requestAccept, err := ownerAcl.RecordBuilder().BuildRequestAccept(RequestAcceptPayload{
RequestRecordId: requestJoinRec.Id,
Permissions: perms,
})
require.NoError(t, err)
// validate
err = ownerAcl.ValidateRawRecord(requestAccept)
require.NoError(t, err)
requestAcceptRec := wrapRecord(requestAccept)
fx.addRec(t, requestAcceptRec)
// checking acl state
require.True(t, ownerState.Permissions(ownerState.pubKey).IsOwner())
require.True(t, ownerState.Permissions(accountState.pubKey).CanWrite())
require.Equal(t, 0, len(ownerState.pendingRequests))
require.Equal(t, 0, len(accountState.pendingRequests))
require.True(t, accountState.Permissions(ownerState.pubKey).IsOwner())
require.True(t, accountState.Permissions(accountState.pubKey).CanWrite())
_, err = ownerState.StateAtRecord(requestJoinRec.Id, accountState.pubKey)
require.Equal(t, ErrNoSuchAccount, err)
stateAtRec, err := ownerState.StateAtRecord(requestAcceptRec.Id, accountState.pubKey)
require.NoError(t, err)
require.True(t, stateAtRec.Permissions == perms)
}
func TestAclList_BuildRoot(t *testing.T) {
randomKeys, err := accountdata.NewRandom()
require.NoError(t, err)
@ -14,3 +117,193 @@ func TestAclList_BuildRoot(t *testing.T) {
require.NoError(t, err)
fmt.Println(randomAcl.Id())
}
func TestAclList_InvitePipeline(t *testing.T) {
fx := newFixture(t)
fx.inviteAccount(t, AclPermissions(aclrecordproto.AclUserPermissions_Writer))
}
func TestAclList_InviteRevoke(t *testing.T) {
fx := newFixture(t)
var (
ownerState = fx.ownerAcl.aclState
accountState = fx.accountAcl.aclState
)
// building invite
inv, err := fx.ownerAcl.RecordBuilder().BuildInvite()
require.NoError(t, err)
inviteRec := wrapRecord(inv.InviteRec)
fx.addRec(t, inviteRec)
// building invite revoke
inviteRevoke, err := fx.ownerAcl.RecordBuilder().BuildInviteRevoke(ownerState.lastRecordId)
require.NoError(t, err)
inviteRevokeRec := wrapRecord(inviteRevoke)
fx.addRec(t, inviteRevokeRec)
// checking acl state
require.True(t, ownerState.Permissions(ownerState.pubKey).IsOwner())
require.True(t, ownerState.Permissions(accountState.pubKey).NoPermissions())
require.Empty(t, ownerState.inviteKeys)
require.Empty(t, accountState.inviteKeys)
}
func TestAclList_RequestDecline(t *testing.T) {
fx := newFixture(t)
var (
ownerAcl = fx.ownerAcl
ownerState = fx.ownerAcl.aclState
accountAcl = fx.accountAcl
accountState = fx.accountAcl.aclState
)
// building invite
inv, err := ownerAcl.RecordBuilder().BuildInvite()
require.NoError(t, err)
inviteRec := wrapRecord(inv.InviteRec)
fx.addRec(t, inviteRec)
// building request join
requestJoin, err := accountAcl.RecordBuilder().BuildRequestJoin(RequestJoinPayload{
InviteRecordId: inviteRec.Id,
InviteKey: inv.InviteKey,
})
require.NoError(t, err)
requestJoinRec := wrapRecord(requestJoin)
fx.addRec(t, requestJoinRec)
// building request decline
requestDecline, err := ownerAcl.RecordBuilder().BuildRequestDecline(ownerState.lastRecordId)
require.NoError(t, err)
requestDeclineRec := wrapRecord(requestDecline)
fx.addRec(t, requestDeclineRec)
// checking acl state
require.True(t, ownerState.Permissions(ownerState.pubKey).IsOwner())
require.True(t, ownerState.Permissions(accountState.pubKey).NoPermissions())
require.Empty(t, ownerState.pendingRequests)
require.Empty(t, accountState.pendingRequests)
}
func TestAclList_Remove(t *testing.T) {
fx := newFixture(t)
var (
ownerState = fx.ownerAcl.aclState
accountState = fx.accountAcl.aclState
)
fx.inviteAccount(t, AclPermissions(aclrecordproto.AclUserPermissions_Writer))
newReadKey := crypto.NewAES()
remove, err := fx.ownerAcl.RecordBuilder().BuildAccountRemove(AccountRemovePayload{
Identities: []crypto.PubKey{fx.accountKeys.SignKey.GetPublic()},
ReadKey: newReadKey,
})
require.NoError(t, err)
removeRec := wrapRecord(remove)
fx.addRec(t, removeRec)
// checking acl state
require.True(t, ownerState.Permissions(ownerState.pubKey).IsOwner())
require.True(t, ownerState.Permissions(accountState.pubKey).NoPermissions())
require.True(t, ownerState.userReadKeys[removeRec.Id].Equals(newReadKey))
require.NotNil(t, ownerState.userReadKeys[fx.ownerAcl.Id()])
require.Equal(t, 0, len(ownerState.pendingRequests))
require.Equal(t, 0, len(accountState.pendingRequests))
require.True(t, accountState.Permissions(ownerState.pubKey).IsOwner())
require.True(t, accountState.Permissions(accountState.pubKey).NoPermissions())
require.Nil(t, accountState.userReadKeys[removeRec.Id])
require.NotNil(t, accountState.userReadKeys[fx.ownerAcl.Id()])
}
func TestAclList_ReadKeyChange(t *testing.T) {
fx := newFixture(t)
var (
ownerState = fx.ownerAcl.aclState
accountState = fx.accountAcl.aclState
)
fx.inviteAccount(t, AclPermissions(aclrecordproto.AclUserPermissions_Admin))
newReadKey := crypto.NewAES()
readKeyChange, err := fx.ownerAcl.RecordBuilder().BuildReadKeyChange(newReadKey)
require.NoError(t, err)
readKeyRec := wrapRecord(readKeyChange)
fx.addRec(t, readKeyRec)
// checking acl state
require.True(t, ownerState.Permissions(ownerState.pubKey).IsOwner())
require.True(t, ownerState.Permissions(accountState.pubKey).CanManageAccounts())
require.True(t, ownerState.userReadKeys[readKeyRec.Id].Equals(newReadKey))
require.True(t, accountState.userReadKeys[readKeyRec.Id].Equals(newReadKey))
require.NotNil(t, ownerState.userReadKeys[fx.ownerAcl.Id()])
require.NotNil(t, accountState.userReadKeys[fx.ownerAcl.Id()])
readKey, err := ownerState.CurrentReadKey()
require.NoError(t, err)
require.True(t, newReadKey.Equals(readKey))
require.Equal(t, 0, len(ownerState.pendingRequests))
require.Equal(t, 0, len(accountState.pendingRequests))
}
func TestAclList_PermissionChange(t *testing.T) {
fx := newFixture(t)
var (
ownerState = fx.ownerAcl.aclState
accountState = fx.accountAcl.aclState
)
fx.inviteAccount(t, AclPermissions(aclrecordproto.AclUserPermissions_Admin))
permissionChange, err := fx.ownerAcl.RecordBuilder().BuildPermissionChange(PermissionChangePayload{
Identity: fx.accountKeys.SignKey.GetPublic(),
Permissions: AclPermissions(aclrecordproto.AclUserPermissions_Writer),
})
require.NoError(t, err)
permissionChangeRec := wrapRecord(permissionChange)
fx.addRec(t, permissionChangeRec)
// checking acl state
require.True(t, ownerState.Permissions(ownerState.pubKey).IsOwner())
require.True(t, ownerState.Permissions(accountState.pubKey) == AclPermissions(aclrecordproto.AclUserPermissions_Writer))
require.True(t, accountState.Permissions(ownerState.pubKey).IsOwner())
require.True(t, accountState.Permissions(accountState.pubKey) == AclPermissions(aclrecordproto.AclUserPermissions_Writer))
require.NotNil(t, ownerState.userReadKeys[fx.ownerAcl.Id()])
require.NotNil(t, accountState.userReadKeys[fx.ownerAcl.Id()])
require.Equal(t, 0, len(ownerState.pendingRequests))
require.Equal(t, 0, len(accountState.pendingRequests))
}
func TestAclList_RequestRemove(t *testing.T) {
fx := newFixture(t)
var (
ownerState = fx.ownerAcl.aclState
accountState = fx.accountAcl.aclState
)
fx.inviteAccount(t, AclPermissions(aclrecordproto.AclUserPermissions_Writer))
removeRequest, err := fx.accountAcl.RecordBuilder().BuildRequestRemove()
require.NoError(t, err)
removeRequestRec := wrapRecord(removeRequest)
fx.addRec(t, removeRequestRec)
recs := fx.accountAcl.AclState().RemoveRecords()
require.Len(t, recs, 1)
require.True(t, accountState.pubKey.Equals(recs[0].RequestIdentity))
newReadKey := crypto.NewAES()
remove, err := fx.ownerAcl.RecordBuilder().BuildAccountRemove(AccountRemovePayload{
Identities: []crypto.PubKey{recs[0].RequestIdentity},
ReadKey: newReadKey,
})
require.NoError(t, err)
removeRec := wrapRecord(remove)
fx.addRec(t, removeRec)
// checking acl state
require.True(t, ownerState.Permissions(ownerState.pubKey).IsOwner())
require.True(t, ownerState.Permissions(accountState.pubKey).NoPermissions())
require.True(t, ownerState.userReadKeys[removeRec.Id].Equals(newReadKey))
require.NotNil(t, ownerState.userReadKeys[fx.ownerAcl.Id()])
require.Equal(t, 0, len(ownerState.pendingRequests))
require.Equal(t, 0, len(accountState.pendingRequests))
require.True(t, accountState.Permissions(ownerState.pubKey).IsOwner())
require.True(t, accountState.Permissions(accountState.pubKey).NoPermissions())
require.Nil(t, accountState.userReadKeys[removeRec.Id])
require.NotNil(t, accountState.userReadKeys[fx.ownerAcl.Id()])
}

18
commonspace/object/acl/list/listutils.go
View File

@ -2,13 +2,13 @@ package list
import (
"github.com/anyproto/any-sync/commonspace/object/accountdata"
"github.com/anyproto/any-sync/commonspace/object/acl/aclrecordproto"
"github.com/anyproto/any-sync/commonspace/object/acl/liststorage"
"github.com/anyproto/any-sync/consensus/consensusproto"
"github.com/anyproto/any-sync/util/crypto"
)
func NewTestDerivedAcl(spaceId string, keys *accountdata.AccountKeys) (AclList, error) {
builder := NewAclRecordBuilder("", crypto.NewKeyStorage())
builder := NewAclRecordBuilder("", crypto.NewKeyStorage(), keys, NoOpAcceptorVerifier{})
masterKey, _, err := crypto.GenerateRandomEd25519KeyPair()
if err != nil {
return nil, err
@ -21,11 +21,21 @@ func NewTestDerivedAcl(spaceId string, keys *accountdata.AccountKeys) (AclList,
if err != nil {
return nil, err
}
st, err := liststorage.NewInMemoryAclListStorage(root.Id, []*aclrecordproto.RawAclRecordWithId{
st, err := liststorage.NewInMemoryAclListStorage(root.Id, []*consensusproto.RawRecordWithId{
root,
})
if err != nil {
return nil, err
}
return BuildAclListWithIdentity(keys, st)
return BuildAclListWithIdentity(keys, st, NoOpAcceptorVerifier{})
}
func NewTestAclWithRoot(keys *accountdata.AccountKeys, root *consensusproto.RawRecordWithId) (AclList, error) {
st, err := liststorage.NewInMemoryAclListStorage(root.Id, []*consensusproto.RawRecordWithId{
root,
})
if err != nil {
return nil, err
}
return BuildAclListWithIdentity(keys, st, NoOpAcceptorVerifier{})
}

56
commonspace/object/acl/list/mock_list/mock_list.go
View File

@ -7,8 +7,8 @@ package mock_list
import (
reflect "reflect"
aclrecordproto "github.com/anyproto/any-sync/commonspace/object/acl/aclrecordproto"
list "github.com/anyproto/any-sync/commonspace/object/acl/list"
consensusproto "github.com/anyproto/any-sync/consensus/consensusproto"
crypto "github.com/anyproto/any-sync/util/crypto"
gomock "github.com/golang/mock/gomock"
)
@ -51,12 +51,11 @@ func (mr *MockAclListMockRecorder) AclState() *gomock.Call {
}
// AddRawRecord mocks base method.
func (m *MockAclList) AddRawRecord(arg0 *aclrecordproto.RawAclRecordWithId) (bool, error) {
func (m *MockAclList) AddRawRecord(arg0 *consensusproto.RawRecordWithId) error {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "AddRawRecord", arg0)
ret0, _ := ret[0].(bool)
ret1, _ := ret[1].(error)
return ret0, ret1
ret0, _ := ret[0].(error)
return ret0
}
// AddRawRecord indicates an expected call of AddRawRecord.
@ -94,6 +93,21 @@ func (mr *MockAclListMockRecorder) Get(arg0 interface{}) *gomock.Call {
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Get", reflect.TypeOf((*MockAclList)(nil).Get), arg0)
}
// GetIndex mocks base method.
func (m *MockAclList) GetIndex(arg0 int) (*list.AclRecord, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "GetIndex", arg0)
ret0, _ := ret[0].(*list.AclRecord)
ret1, _ := ret[1].(error)
return ret0, ret1
}
// GetIndex indicates an expected call of GetIndex.
func (mr *MockAclListMockRecorder) GetIndex(arg0 interface{}) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetIndex", reflect.TypeOf((*MockAclList)(nil).GetIndex), arg0)
}
// Head mocks base method.
func (m *MockAclList) Head() *list.AclRecord {
m.ctrl.T.Helper()
@ -211,6 +225,20 @@ func (mr *MockAclListMockRecorder) RUnlock() *gomock.Call {
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RUnlock", reflect.TypeOf((*MockAclList)(nil).RUnlock))
}
// RecordBuilder mocks base method.
func (m *MockAclList) RecordBuilder() list.AclRecordBuilder {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "RecordBuilder")
ret0, _ := ret[0].(list.AclRecordBuilder)
return ret0
}
// RecordBuilder indicates an expected call of RecordBuilder.
func (mr *MockAclListMockRecorder) RecordBuilder() *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "RecordBuilder", reflect.TypeOf((*MockAclList)(nil).RecordBuilder))
}
// Records mocks base method.
func (m *MockAclList) Records() []*list.AclRecord {
m.ctrl.T.Helper()
@ -226,10 +254,10 @@ func (mr *MockAclListMockRecorder) Records() *gomock.Call {
}
// Root mocks base method.
func (m *MockAclList) Root() *aclrecordproto.RawAclRecordWithId {
func (m *MockAclList) Root() *consensusproto.RawRecordWithId {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "Root")
ret0, _ := ret[0].(*aclrecordproto.RawAclRecordWithId)
ret0, _ := ret[0].(*consensusproto.RawRecordWithId)
return ret0
}
@ -250,3 +278,17 @@ func (mr *MockAclListMockRecorder) Unlock() *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Unlock", reflect.TypeOf((*MockAclList)(nil).Unlock))
}
// ValidateRawRecord mocks base method.
func (m *MockAclList) ValidateRawRecord(arg0 *consensusproto.RawRecord) error {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "ValidateRawRecord", arg0)
ret0, _ := ret[0].(error)
return ret0
}
// ValidateRawRecord indicates an expected call of ValidateRawRecord.
func (mr *MockAclListMockRecorder) ValidateRawRecord(arg0 interface{}) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "ValidateRawRecord", reflect.TypeOf((*MockAclList)(nil).ValidateRawRecord), arg0)
}

51
commonspace/object/acl/list/models.go
View File

@ -8,7 +8,6 @@ import (
type AclRecord struct {
Id string
PrevId string
ReadKeyId string
Timestamp int64
Data []byte
Identity crypto.PubKey
@ -16,7 +15,55 @@ type AclRecord struct {
Signature []byte
}
type RequestRecord struct {
RequestIdentity crypto.PubKey
RequestMetadata []byte
Type RequestType
}
type AclUserState struct {
PubKey crypto.PubKey
Permissions aclrecordproto.AclUserPermissions
Permissions AclPermissions
RequestMetadata []byte
}
type RequestType int
const (
RequestTypeRemove RequestType = iota
RequestTypeJoin
)
type AclPermissions aclrecordproto.AclUserPermissions
func (p AclPermissions) NoPermissions() bool {
return aclrecordproto.AclUserPermissions(p) == aclrecordproto.AclUserPermissions_None
}
func (p AclPermissions) IsOwner() bool {
return aclrecordproto.AclUserPermissions(p) == aclrecordproto.AclUserPermissions_Owner
}
func (p AclPermissions) CanWrite() bool {
switch aclrecordproto.AclUserPermissions(p) {
case aclrecordproto.AclUserPermissions_Admin:
return true
case aclrecordproto.AclUserPermissions_Writer:
return true
case aclrecordproto.AclUserPermissions_Owner:
return true
default:
return false
}
}
func (p AclPermissions) CanManageAccounts() bool {
switch aclrecordproto.AclUserPermissions(p) {
case aclrecordproto.AclUserPermissions_Admin:
return true
case aclrecordproto.AclUserPermissions_Owner:
return true
default:
return false
}
}

218
commonspace/object/acl/list/validator.go Normal file
View File

@ -0,0 +1,218 @@
package list
import (
"github.com/anyproto/any-sync/commonspace/object/acl/aclrecordproto"
"github.com/anyproto/any-sync/util/crypto"
)
type ContentValidator interface {
ValidateAclRecordContents(ch *AclRecord) (err error)
ValidatePermissionChange(ch *aclrecordproto.AclAccountPermissionChange, authorIdentity crypto.PubKey) (err error)
ValidateInvite(ch *aclrecordproto.AclAccountInvite, authorIdentity crypto.PubKey) (err error)
ValidateInviteRevoke(ch *aclrecordproto.AclAccountInviteRevoke, authorIdentity crypto.PubKey) (err error)
ValidateRequestJoin(ch *aclrecordproto.AclAccountRequestJoin, authorIdentity crypto.PubKey) (err error)
ValidateRequestAccept(ch *aclrecordproto.AclAccountRequestAccept, authorIdentity crypto.PubKey) (err error)
ValidateRequestDecline(ch *aclrecordproto.AclAccountRequestDecline, authorIdentity crypto.PubKey) (err error)
ValidateAccountRemove(ch *aclrecordproto.AclAccountRemove, authorIdentity crypto.PubKey) (err error)
ValidateRequestRemove(ch *aclrecordproto.AclAccountRequestRemove, authorIdentity crypto.PubKey) (err error)
ValidateReadKeyChange(ch *aclrecordproto.AclReadKeyChange, authorIdentity crypto.PubKey) (err error)
}
type contentValidator struct {
keyStore crypto.KeyStorage
aclState *AclState
}
func (c *contentValidator) ValidateAclRecordContents(ch *AclRecord) (err error) {
if ch.PrevId != c.aclState.lastRecordId {
return ErrIncorrectRecordSequence
}
aclData := ch.Model.(*aclrecordproto.AclData)
for _, content := range aclData.AclContent {
err = c.validateAclRecordContent(content, ch.Identity)
if err != nil {
return
}
}
return
}
func (c *contentValidator) validateAclRecordContent(ch *aclrecordproto.AclContentValue, authorIdentity crypto.PubKey) (err error) {
switch {
case ch.GetPermissionChange() != nil:
return c.ValidatePermissionChange(ch.GetPermissionChange(), authorIdentity)
case ch.GetInvite() != nil:
return c.ValidateInvite(ch.GetInvite(), authorIdentity)
case ch.GetInviteRevoke() != nil:
return c.ValidateInviteRevoke(ch.GetInviteRevoke(), authorIdentity)
case ch.GetRequestJoin() != nil:
return c.ValidateRequestJoin(ch.GetRequestJoin(), authorIdentity)
case ch.GetRequestAccept() != nil:
return c.ValidateRequestAccept(ch.GetRequestAccept(), authorIdentity)
case ch.GetRequestDecline() != nil:
return c.ValidateRequestDecline(ch.GetRequestDecline(), authorIdentity)
case ch.GetAccountRemove() != nil:
return c.ValidateAccountRemove(ch.GetAccountRemove(), authorIdentity)
case ch.GetAccountRequestRemove() != nil:
return c.ValidateRequestRemove(ch.GetAccountRequestRemove(), authorIdentity)
case ch.GetReadKeyChange() != nil:
return c.ValidateReadKeyChange(ch.GetReadKeyChange(), authorIdentity)
default:
return ErrUnexpectedContentType
}
}
func (c *contentValidator) ValidatePermissionChange(ch *aclrecordproto.AclAccountPermissionChange, authorIdentity crypto.PubKey) (err error) {
if !c.aclState.Permissions(authorIdentity).CanManageAccounts() {
return ErrInsufficientPermissions
}
chIdentity, err := c.keyStore.PubKeyFromProto(ch.Identity)
if err != nil {
return err
}
_, exists := c.aclState.userStates[mapKeyFromPubKey(chIdentity)]
if !exists {
return ErrNoSuchAccount
}
return
}
func (c *contentValidator) ValidateInvite(ch *aclrecordproto.AclAccountInvite, authorIdentity crypto.PubKey) (err error) {
if !c.aclState.Permissions(authorIdentity).CanManageAccounts() {
return ErrInsufficientPermissions
}
_, err = c.keyStore.PubKeyFromProto(ch.InviteKey)
return
}
func (c *contentValidator) ValidateInviteRevoke(ch *aclrecordproto.AclAccountInviteRevoke, authorIdentity crypto.PubKey) (err error) {
if !c.aclState.Permissions(authorIdentity).CanManageAccounts() {
return ErrInsufficientPermissions
}
_, exists := c.aclState.inviteKeys[ch.InviteRecordId]
if !exists {
return ErrNoSuchInvite
}
return
}
func (c *contentValidator) ValidateRequestJoin(ch *aclrecordproto.AclAccountRequestJoin, authorIdentity crypto.PubKey) (err error) {
inviteKey, exists := c.aclState.inviteKeys[ch.InviteRecordId]
if !exists {
return ErrNoSuchInvite
}
inviteIdentity, err := c.keyStore.PubKeyFromProto(ch.InviteIdentity)
if err != nil {
return
}
if _, exists := c.aclState.pendingRequests[mapKeyFromPubKey(inviteIdentity)]; exists {
return ErrPendingRequest
}
if !authorIdentity.Equals(inviteIdentity) {
return ErrIncorrectIdentity
}
rawInviteIdentity, err := inviteIdentity.Raw()
if err != nil {
return err
}
ok, err := inviteKey.Verify(rawInviteIdentity, ch.InviteIdentitySignature)
if err != nil {
return ErrInvalidSignature
}
if !ok {
return ErrInvalidSignature
}
return
}
func (c *contentValidator) ValidateRequestAccept(ch *aclrecordproto.AclAccountRequestAccept, authorIdentity crypto.PubKey) (err error) {
if !c.aclState.Permissions(authorIdentity).CanManageAccounts() {
return ErrInsufficientPermissions
}
record, exists := c.aclState.requestRecords[ch.RequestRecordId]
if !exists {
return ErrNoSuchRequest
}
acceptIdentity, err := c.keyStore.PubKeyFromProto(ch.Identity)
if err != nil {
return
}
if !acceptIdentity.Equals(record.RequestIdentity) {
return ErrIncorrectIdentity
}
if ch.Permissions == aclrecordproto.AclUserPermissions_Owner {
return ErrInsufficientPermissions
}
return
}
func (c *contentValidator) ValidateRequestDecline(ch *aclrecordproto.AclAccountRequestDecline, authorIdentity crypto.PubKey) (err error) {
if !c.aclState.Permissions(authorIdentity).CanManageAccounts() {
return ErrInsufficientPermissions
}
_, exists := c.aclState.requestRecords[ch.RequestRecordId]
if !exists {
return ErrNoSuchRequest
}
return
}
func (c *contentValidator) ValidateAccountRemove(ch *aclrecordproto.AclAccountRemove, authorIdentity crypto.PubKey) (err error) {
if !c.aclState.Permissions(authorIdentity).CanManageAccounts() {
return ErrInsufficientPermissions
}
seenIdentities := map[string]struct{}{}
for _, rawIdentity := range ch.Identities {
identity, err := c.keyStore.PubKeyFromProto(rawIdentity)
if err != nil {
return err
}
if identity.Equals(authorIdentity) {
return ErrInsufficientPermissions
}
permissions := c.aclState.Permissions(identity)
if permissions.NoPermissions() {
return ErrNoSuchAccount
}
if permissions.IsOwner() {
return ErrInsufficientPermissions
}
idKey := mapKeyFromPubKey(identity)
if _, exists := seenIdentities[idKey]; exists {
return ErrDuplicateAccounts
}
seenIdentities[mapKeyFromPubKey(identity)] = struct{}{}
}
return c.validateAccountReadKeys(ch.AccountKeys, len(c.aclState.userStates)-len(ch.Identities))
}
func (c *contentValidator) ValidateRequestRemove(ch *aclrecordproto.AclAccountRequestRemove, authorIdentity crypto.PubKey) (err error) {
if c.aclState.Permissions(authorIdentity).NoPermissions() {
return ErrInsufficientPermissions
}
if _, exists := c.aclState.pendingRequests[mapKeyFromPubKey(authorIdentity)]; exists {
return ErrPendingRequest
}
return
}
func (c *contentValidator) ValidateReadKeyChange(ch *aclrecordproto.AclReadKeyChange, authorIdentity crypto.PubKey) (err error) {
return c.validateAccountReadKeys(ch.AccountKeys, len(c.aclState.userStates))
}
func (c *contentValidator) validateAccountReadKeys(accountKeys []*aclrecordproto.AclEncryptedReadKey, usersNum int) (err error) {
if len(accountKeys) != usersNum {
return ErrIncorrectNumberOfAccounts
}
for _, encKeys := range accountKeys {
identity, err := c.keyStore.PubKeyFromProto(encKeys.Identity)
if err != nil {
return err
}
_, exists := c.aclState.userStates[mapKeyFromPubKey(identity)]
if !exists {
return ErrNoSuchAccount
}
}
return
}

18
commonspace/object/acl/liststorage/inmemory.go
View File

@ -3,24 +3,26 @@ package liststorage
import (
"context"
"fmt"
"github.com/anyproto/any-sync/commonspace/object/acl/aclrecordproto"
"github.com/anyproto/any-sync/consensus/consensusproto"
"sync"
)
type inMemoryAclListStorage struct {
id string
root *aclrecordproto.RawAclRecordWithId
root *consensusproto.RawRecordWithId
head string
records map[string]*aclrecordproto.RawAclRecordWithId
records map[string]*consensusproto.RawRecordWithId
sync.RWMutex
}
func NewInMemoryAclListStorage(
id string,
records []*aclrecordproto.RawAclRecordWithId) (ListStorage, error) {
records []*consensusproto.RawRecordWithId) (ListStorage, error) {
allRecords := make(map[string]*aclrecordproto.RawAclRecordWithId)
allRecords := make(map[string]*consensusproto.RawRecordWithId)
for _, ch := range records {
allRecords[ch.Id] = ch
}
@ -41,7 +43,7 @@ func (t *inMemoryAclListStorage) Id() string {
return t.id
}
func (t *inMemoryAclListStorage) Root() (*aclrecordproto.RawAclRecordWithId, error) {
func (t *inMemoryAclListStorage) Root() (*consensusproto.RawRecordWithId, error) {
t.RLock()
defer t.RUnlock()
return t.root, nil
@ -60,7 +62,7 @@ func (t *inMemoryAclListStorage) SetHead(head string) error {
return nil
}
func (t *inMemoryAclListStorage) AddRawRecord(ctx context.Context, record *aclrecordproto.RawAclRecordWithId) error {
func (t *inMemoryAclListStorage) AddRawRecord(ctx context.Context, record *consensusproto.RawRecordWithId) error {
t.Lock()
defer t.Unlock()
// TODO: better to do deep copy
@ -68,7 +70,7 @@ func (t *inMemoryAclListStorage) AddRawRecord(ctx context.Context, record *aclre
return nil
}
func (t *inMemoryAclListStorage) GetRawRecord(ctx context.Context, recordId string) (*aclrecordproto.RawAclRecordWithId, error) {
func (t *inMemoryAclListStorage) GetRawRecord(ctx context.Context, recordId string) (*consensusproto.RawRecordWithId, error) {
t.RLock()
defer t.RUnlock()
if res, exists := t.records[recordId]; exists {

11
commonspace/object/acl/liststorage/liststorage.go
View File

@ -4,7 +4,8 @@ package liststorage
import (
"context"
"errors"
"github.com/anyproto/any-sync/commonspace/object/acl/aclrecordproto"
"github.com/anyproto/any-sync/consensus/consensusproto"
)
var (
@ -14,15 +15,15 @@ var (
)
type Exporter interface {
ListStorage(root *aclrecordproto.RawAclRecordWithId) (ListStorage, error)
ListStorage(root *consensusproto.RawRecordWithId) (ListStorage, error)
}
type ListStorage interface {
Id() string
Root() (*aclrecordproto.RawAclRecordWithId, error)
Root() (*consensusproto.RawRecordWithId, error)
Head() (string, error)
SetHead(headId string) error
GetRawRecord(ctx context.Context, id string) (*aclrecordproto.RawAclRecordWithId, error)
AddRawRecord(ctx context.Context, rec *aclrecordproto.RawAclRecordWithId) error
GetRawRecord(ctx context.Context, id string) (*consensusproto.RawRecordWithId, error)
AddRawRecord(ctx context.Context, rec *consensusproto.RawRecordWithId) error
}

12
commonspace/object/acl/liststorage/mock_liststorage/mock_liststorage.go
View File

@ -8,7 +8,7 @@ import (
context "context"
reflect "reflect"
aclrecordproto "github.com/anyproto/any-sync/commonspace/object/acl/aclrecordproto"
consensusproto "github.com/anyproto/any-sync/consensus/consensusproto"
gomock "github.com/golang/mock/gomock"
)
@ -36,7 +36,7 @@ func (m *MockListStorage) EXPECT() *MockListStorageMockRecorder {
}
// AddRawRecord mocks base method.
func (m *MockListStorage) AddRawRecord(arg0 context.Context, arg1 *aclrecordproto.RawAclRecordWithId) error {
func (m *MockListStorage) AddRawRecord(arg0 context.Context, arg1 *consensusproto.RawRecordWithId) error {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "AddRawRecord", arg0, arg1)
ret0, _ := ret[0].(error)
@ -50,10 +50,10 @@ func (mr *MockListStorageMockRecorder) AddRawRecord(arg0, arg1 interface{}) *gom
}
// GetRawRecord mocks base method.
func (m *MockListStorage) GetRawRecord(arg0 context.Context, arg1 string) (*aclrecordproto.RawAclRecordWithId, error) {
func (m *MockListStorage) GetRawRecord(arg0 context.Context, arg1 string) (*consensusproto.RawRecordWithId, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "GetRawRecord", arg0, arg1)
ret0, _ := ret[0].(*aclrecordproto.RawAclRecordWithId)
ret0, _ := ret[0].(*consensusproto.RawRecordWithId)
ret1, _ := ret[1].(error)
return ret0, ret1
}
@ -94,10 +94,10 @@ func (mr *MockListStorageMockRecorder) Id() *gomock.Call {
}
// Root mocks base method.
func (m *MockListStorage) Root() (*aclrecordproto.RawAclRecordWithId, error) {
func (m *MockListStorage) Root() (*consensusproto.RawRecordWithId, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "Root")
ret0, _ := ret[0].(*aclrecordproto.RawAclRecordWithId)
ret0, _ := ret[0].(*consensusproto.RawRecordWithId)
ret1, _ := ret[1].(error)
return ret0, ret1
}

3
commonspace/object/acl/syncacl/syncacl.go
View File

@ -2,6 +2,7 @@ package syncacl
import (
"context"
"github.com/anyproto/any-sync/accountservice"
"github.com/anyproto/any-sync/app"
"github.com/anyproto/any-sync/commonspace/object/acl/list"
@ -34,7 +35,7 @@ func (s *SyncAcl) Init(a *app.App) (err error) {
return err
}
acc := a.MustComponent(accountservice.CName).(accountservice.Service)
s.AclList, err = list.BuildAclListWithIdentity(acc.Account(), aclStorage)
s.AclList, err = list.BuildAclListWithIdentity(acc.Account(), aclStorage, list.NoOpAcceptorVerifier{})
return err
}

19
commonspace/object/acl/syncacl/syncaclhandler.go
View File

@ -2,8 +2,7 @@ package syncacl
import (
"context"
"fmt"
"github.com/anyproto/any-sync/commonspace/object/acl/aclrecordproto"
"github.com/anyproto/any-sync/commonspace/object/acl/list"
"github.com/anyproto/any-sync/commonspace/spacesyncproto"
)
@ -13,19 +12,5 @@ type syncAclHandler struct {
}
func (s *syncAclHandler) HandleMessage(ctx context.Context, senderId string, req *spacesyncproto.ObjectSyncMessage) (err error) {
aclMsg := &aclrecordproto.AclSyncMessage{}
if err = aclMsg.Unmarshal(req.Payload); err != nil {
return
}
content := aclMsg.GetContent()
switch {
case content.GetAddRecords() != nil:
return s.handleAddRecords(ctx, senderId, content.GetAddRecords())
default:
return fmt.Errorf("unexpected aclSync message: %T", content.Value)
}
}
func (s *syncAclHandler) handleAddRecords(ctx context.Context, senderId string, addRecord *aclrecordproto.AclAddRecords) (err error) {
return
return nil
}

2
commonspace/object/tree/exporter/treeimport.go
View File

@ -15,7 +15,7 @@ type TreeImportParams struct {
}
func ImportHistoryTree(params TreeImportParams) (tree objecttree.ReadableObjectTree, err error) {
aclList, err := list.BuildAclList(params.ListStorage)
aclList, err := list.BuildAclList(params.ListStorage, list.NoOpAcceptorVerifier{})
if err != nil {
return
}

8
commonspace/object/tree/objecttree/objecttree.go
View File

@ -4,11 +4,11 @@ package objecttree
import (
"context"
"errors"
"github.com/anyproto/any-sync/util/crypto"
"sync"
"time"
"github.com/anyproto/any-sync/commonspace/object/acl/aclrecordproto"
"github.com/anyproto/any-sync/util/crypto"
"github.com/anyproto/any-sync/commonspace/object/acl/list"
"github.com/anyproto/any-sync/commonspace/object/tree/treechangeproto"
"github.com/anyproto/any-sync/commonspace/object/tree/treestorage"
@ -248,9 +248,7 @@ func (ot *objectTree) prepareBuilderContent(content SignableChangeContent) (cnt
pubKey = content.Key.GetPublic()
readKeyId string
)
canWrite := state.HasPermission(pubKey, aclrecordproto.AclUserPermissions_Writer) ||
state.HasPermission(pubKey, aclrecordproto.AclUserPermissions_Admin)
if !canWrite {
if !state.Permissions(pubKey).CanWrite() {
err = list.ErrInsufficientPermissions
return
}

10
commonspace/object/tree/objecttree/objecttreevalidator.go
View File

@ -3,7 +3,7 @@ package objecttree
import (
"context"
"fmt"
"github.com/anyproto/any-sync/commonspace/object/acl/aclrecordproto"
"github.com/anyproto/any-sync/commonspace/object/acl/list"
"github.com/anyproto/any-sync/commonspace/object/tree/treestorage"
"github.com/anyproto/any-sync/util/slice"
@ -52,20 +52,18 @@ func (v *objectTreeValidator) ValidateNewChanges(tree *Tree, aclList list.AclLis
func (v *objectTreeValidator) validateChange(tree *Tree, aclList list.AclList, c *Change) (err error) {
var (
perm list.AclUserState
userState list.AclUserState
state = aclList.AclState()
)
// checking if the user could write
perm, err = state.StateAtRecord(c.AclHeadId, c.Identity)
userState, err = state.StateAtRecord(c.AclHeadId, c.Identity)
if err != nil {
return
}
if perm.Permissions != aclrecordproto.AclUserPermissions_Writer && perm.Permissions != aclrecordproto.AclUserPermissions_Admin {
if !userState.Permissions.CanWrite() {
err = list.ErrInsufficientPermissions
return
}
if c.Id == tree.RootId() {
return
}

20
commonspace/payloads.go
View File

@ -2,20 +2,22 @@ package commonspace
import (
"errors"
"hash/fnv"
"math/rand"
"strconv"
"strings"
"time"
"github.com/anyproto/any-sync/commonspace/object/acl/aclrecordproto"
"github.com/anyproto/any-sync/commonspace/object/acl/list"
"github.com/anyproto/any-sync/commonspace/object/tree/objecttree"
"github.com/anyproto/any-sync/commonspace/object/tree/treechangeproto"
"github.com/anyproto/any-sync/commonspace/spacestorage"
"github.com/anyproto/any-sync/commonspace/spacesyncproto"
"github.com/anyproto/any-sync/consensus/consensusproto"
"github.com/anyproto/any-sync/util/cidutil"
"github.com/anyproto/any-sync/util/crypto"
"github.com/gogo/protobuf/proto"
"hash/fnv"
"math/rand"
"strconv"
"strings"
"time"
)
const (
@ -71,7 +73,7 @@ func storagePayloadForSpaceCreate(payload SpaceCreatePayload) (storagePayload sp
// building acl root
keyStorage := crypto.NewKeyStorage()
aclBuilder := list.NewAclRecordBuilder("", keyStorage)
aclBuilder := list.NewAclRecordBuilder("", keyStorage, nil, list.NoOpAcceptorVerifier{})
aclRoot, err := aclBuilder.BuildRoot(list.RootContent{
PrivKey: payload.SigningKey,
MasterKey: payload.MasterKey,
@ -158,7 +160,7 @@ func storagePayloadForSpaceDerive(payload SpaceDerivePayload) (storagePayload sp
// building acl root
keyStorage := crypto.NewKeyStorage()
aclBuilder := list.NewAclRecordBuilder("", keyStorage)
aclBuilder := list.NewAclRecordBuilder("", keyStorage, nil, list.NoOpAcceptorVerifier{})
aclRoot, err := aclBuilder.BuildRoot(list.RootContent{
PrivKey: payload.SigningKey,
MasterKey: payload.MasterKey,
@ -254,12 +256,12 @@ func ValidateSpaceHeader(rawHeaderWithId *spacesyncproto.RawSpaceHeaderWithId, i
return
}
func validateCreateSpaceAclPayload(rawWithId *aclrecordproto.RawAclRecordWithId) (spaceId string, err error) {
func validateCreateSpaceAclPayload(rawWithId *consensusproto.RawRecordWithId) (spaceId string, err error) {
if !cidutil.VerifyCid(rawWithId.Payload, rawWithId.Id) {
err = objecttree.ErrIncorrectCid
return
}
var rawAcl aclrecordproto.RawAclRecord
var rawAcl consensusproto.RawRecord
err = proto.Unmarshal(rawWithId.Payload, &rawAcl)
if err != nil {
return

28
commonspace/payloads_test.go
View File

@ -2,20 +2,22 @@ package commonspace
import (
"fmt"
"math/rand"
"strconv"
"testing"
"time"
"github.com/anyproto/any-sync/commonspace/object/accountdata"
"github.com/anyproto/any-sync/commonspace/object/acl/aclrecordproto"
"github.com/anyproto/any-sync/commonspace/object/tree/objecttree"
"github.com/anyproto/any-sync/commonspace/object/tree/treechangeproto"
"github.com/anyproto/any-sync/commonspace/spacestorage"
"github.com/anyproto/any-sync/commonspace/spacesyncproto"
"github.com/anyproto/any-sync/consensus/consensusproto"
"github.com/anyproto/any-sync/util/cidutil"
"github.com/anyproto/any-sync/util/crypto"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"math/rand"
"strconv"
"testing"
"time"
)
func TestSuccessHeaderPayloadForSpaceCreate(t *testing.T) {
@ -188,14 +190,14 @@ func TestFailAclPayloadSpace_IncorrectCid(t *testing.T) {
marshalled, err := aclRoot.Marshal()
require.NoError(t, err)
signature, err := accountKeys.SignKey.Sign(marshalled)
rawAclRecord := &aclrecordproto.RawAclRecord{
rawAclRecord := &consensusproto.RawRecord{
Payload: marshalled,
Signature: signature,
}
marshalledRaw, err := rawAclRecord.Marshal()
require.NoError(t, err)
aclHeadId := "rand"
rawWithId := &aclrecordproto.RawAclRecordWithId{
rawWithId := &consensusproto.RawRecordWithId{
Payload: marshalledRaw,
Id: aclHeadId,
}
@ -230,7 +232,7 @@ func TestFailedAclPayloadSpace_IncorrectSignature(t *testing.T) {
}
marshalled, err := aclRoot.Marshal()
require.NoError(t, err)
rawAclRecord := &aclrecordproto.RawAclRecord{
rawAclRecord := &consensusproto.RawRecord{
Payload: marshalled,
Signature: marshalled,
}
@ -238,7 +240,7 @@ func TestFailedAclPayloadSpace_IncorrectSignature(t *testing.T) {
require.NoError(t, err)
aclHeadId, err := cidutil.NewCidFromBytes(marshalledRaw)
require.NoError(t, err)
rawWithId := &aclrecordproto.RawAclRecordWithId{
rawWithId := &consensusproto.RawRecordWithId{
Payload: marshalledRaw,
Id: aclHeadId,
}
@ -286,7 +288,7 @@ func TestFailedAclPayloadSpace_IncorrectIdentitySignature(t *testing.T) {
return
}
signature, err := accountKeys.SignKey.Sign(marshalled)
rawAclRecord := &aclrecordproto.RawAclRecord{
rawAclRecord := &consensusproto.RawRecord{
Payload: marshalled,
Signature: signature,
}
@ -298,7 +300,7 @@ func TestFailedAclPayloadSpace_IncorrectIdentitySignature(t *testing.T) {
if err != nil {
return
}
rawWithId := &aclrecordproto.RawAclRecordWithId{
rawWithId := &consensusproto.RawRecordWithId{
Payload: marshalledRaw,
Id: aclHeadId,
}
@ -540,7 +542,7 @@ func rawSettingsPayload(accountKeys *accountdata.AccountKeys, spaceId, aclHeadId
return
}
func rawAclWithId(accountKeys *accountdata.AccountKeys, spaceId string) (aclHeadId string, rawWithId *aclrecordproto.RawAclRecordWithId, err error) {
func rawAclWithId(accountKeys *accountdata.AccountKeys, spaceId string) (aclHeadId string, rawWithId *consensusproto.RawRecordWithId, err error) {
// TODO: use same storage creation methods as we use in spaces
readKeyBytes := make([]byte, 32)
_, err = rand.Read(readKeyBytes)
@ -582,7 +584,7 @@ func rawAclWithId(accountKeys *accountdata.AccountKeys, spaceId string) (aclHead
return
}
signature, err := accountKeys.SignKey.Sign(marshalled)
rawAclRecord := &aclrecordproto.RawAclRecord{
rawAclRecord := &consensusproto.RawRecord{
Payload: marshalled,
Signature: signature,
}
@ -594,7 +596,7 @@ func rawAclWithId(accountKeys *accountdata.AccountKeys, spaceId string) (aclHead
if err != nil {
return
}
rawWithId = &aclrecordproto.RawAclRecordWithId{
rawWithId = &consensusproto.RawRecordWithId{
Payload: marshalledRaw,
Id: aclHeadId,
}

11
commonspace/requestmanager/requestmanager_test.go
View File

@ -2,6 +2,9 @@ package requestmanager
import (
"context"
"sync"
"testing"
"github.com/anyproto/any-sync/commonspace/objectsync"
"github.com/anyproto/any-sync/commonspace/objectsync/mock_objectsync"
"github.com/anyproto/any-sync/commonspace/spacesyncproto"
@ -13,9 +16,6 @@ import (
"github.com/stretchr/testify/require"
"storj.io/drpc"
"storj.io/drpc/drpcconn"
"sync"
"testing"
"time"
)
type fixture struct {
@ -146,6 +146,7 @@ func TestRequestManager_QueueRequest(t *testing.T) {
_, ok = msgs.Load("otherId1")
require.True(t, ok)
close(msgRelease)
fx.requestManager.Close(context.Background())
})
t.Run("no requests after close", func(t *testing.T) {
@ -179,11 +180,7 @@ func TestRequestManager_QueueRequest(t *testing.T) {
fx.requestManager.Close(context.Background())
close(msgRelease)
// waiting to know if the second one is not taken
// because the manager is now closed
time.Sleep(100 * time.Millisecond)
_, ok = msgs.Load("id2")
require.False(t, ok)
})
}

9
commonspace/spaceservice.go
View File

@ -2,6 +2,8 @@ package commonspace
import (
"context"
"sync/atomic"
"github.com/anyproto/any-sync/accountservice"
"github.com/anyproto/any-sync/app"
"github.com/anyproto/any-sync/app/logger"
@ -9,7 +11,6 @@ import (
"github.com/anyproto/any-sync/commonspace/credentialprovider"
"github.com/anyproto/any-sync/commonspace/deletionstate"
"github.com/anyproto/any-sync/commonspace/headsync"
"github.com/anyproto/any-sync/commonspace/object/acl/aclrecordproto"
"github.com/anyproto/any-sync/commonspace/object/acl/syncacl"
"github.com/anyproto/any-sync/commonspace/object/tree/objecttree"
"github.com/anyproto/any-sync/commonspace/object/tree/treechangeproto"
@ -24,13 +25,13 @@ import (
"github.com/anyproto/any-sync/commonspace/spacestorage"
"github.com/anyproto/any-sync/commonspace/spacesyncproto"
"github.com/anyproto/any-sync/commonspace/syncstatus"
"github.com/anyproto/any-sync/consensus/consensusproto"
"github.com/anyproto/any-sync/metric"
"github.com/anyproto/any-sync/net/peer"
"github.com/anyproto/any-sync/net/pool"
"github.com/anyproto/any-sync/net/rpc/rpcerr"
"github.com/anyproto/any-sync/nodeconf"
"storj.io/drpc"
"sync/atomic"
)
const CName = "common.commonspace"
@ -193,7 +194,7 @@ func (s *spaceService) NewSpace(ctx context.Context, id string) (Space, error) {
func (s *spaceService) addSpaceStorage(ctx context.Context, spaceDescription SpaceDescription) (st spacestorage.SpaceStorage, err error) {
payload := spacestorage.SpaceStorageCreatePayload{
AclWithId: &aclrecordproto.RawAclRecordWithId{
AclWithId: &consensusproto.RawRecordWithId{
Payload: spaceDescription.AclPayload,
Id: spaceDescription.AclId,
},
@ -240,7 +241,7 @@ func (s *spaceService) getSpaceStorageFromRemote(ctx context.Context, id string)
}
st, err = s.createSpaceStorage(spacestorage.SpaceStorageCreatePayload{
AclWithId: &aclrecordproto.RawAclRecordWithId{
AclWithId: &consensusproto.RawRecordWithId{
Payload: res.Payload.AclPayload,
Id: res.Payload.AclPayloadId,
},

6
commonspace/spacestorage/inmemorystorage.go
View File

@ -2,12 +2,14 @@ package spacestorage
import (
"context"
"github.com/anyproto/any-sync/app"
"github.com/anyproto/any-sync/commonspace/object/acl/aclrecordproto"
"github.com/anyproto/any-sync/commonspace/object/acl/liststorage"
"github.com/anyproto/any-sync/commonspace/object/tree/treechangeproto"
"github.com/anyproto/any-sync/commonspace/object/tree/treestorage"
"github.com/anyproto/any-sync/commonspace/spacesyncproto"
"github.com/anyproto/any-sync/consensus/consensusproto"
"sync"
)
@ -40,7 +42,7 @@ func (i *InMemorySpaceStorage) Name() (name string) {
}
func NewInMemorySpaceStorage(payload SpaceStorageCreatePayload) (SpaceStorage, error) {
aclStorage, err := liststorage.NewInMemoryAclListStorage(payload.AclWithId.Id, []*aclrecordproto.RawAclRecordWithId{payload.AclWithId})
aclStorage, err := liststorage.NewInMemoryAclListStorage(payload.AclWithId.Id, []*consensusproto.RawRecordWithId{payload.AclWithId})
if err != nil {
return nil, err
}

5
commonspace/spacestorage/spacestorage.go
View File

@ -4,12 +4,13 @@ package spacestorage
import (
"context"
"errors"
"github.com/anyproto/any-sync/app"
"github.com/anyproto/any-sync/commonspace/object/acl/aclrecordproto"
"github.com/anyproto/any-sync/commonspace/object/acl/liststorage"
"github.com/anyproto/any-sync/commonspace/object/tree/treechangeproto"
"github.com/anyproto/any-sync/commonspace/object/tree/treestorage"
"github.com/anyproto/any-sync/commonspace/spacesyncproto"
"github.com/anyproto/any-sync/consensus/consensusproto"
)
const CName = "common.commonspace.spacestorage"
@ -47,7 +48,7 @@ type SpaceStorage interface {
}
type SpaceStorageCreatePayload struct {
AclWithId *aclrecordproto.RawAclRecordWithId
AclWithId *consensusproto.RawRecordWithId
SpaceHeaderWithId *spacesyncproto.RawSpaceHeaderWithId
SpaceSettingsWithId *treechangeproto.RawTreeChangeWithId
}

8
commonspace/syncstatus/syncstatus.go
View File

@ -3,11 +3,12 @@ package syncstatus
import (
"context"
"fmt"
"github.com/anyproto/any-sync/app"
"github.com/anyproto/any-sync/commonspace/spacestate"
"sync"
"time"
"github.com/anyproto/any-sync/app"
"github.com/anyproto/any-sync/commonspace/spacestate"
"github.com/anyproto/any-sync/app/logger"
"github.com/anyproto/any-sync/commonspace/object/tree/treestorage"
"github.com/anyproto/any-sync/commonspace/spacestorage"
@ -178,8 +179,9 @@ func (s *syncStatusService) update(ctx context.Context) (err error) {
}
s.treeStatusBuf = append(s.treeStatusBuf, treeStatus{treeId, treeHeads.syncStatus, treeHeads.heads})
}
nodesOnline := s.nodesOnline
s.Unlock()
s.updateReceiver.UpdateNodeConnection(s.nodesOnline)
s.updateReceiver.UpdateNodeConnection(nodesOnline)
for _, entry := range s.treeStatusBuf {
err = s.updateReceiver.UpdateTree(ctx, entry.treeId, entry.status)
if err != nil {

241
consensus/consensusclient/client.go Normal file
View File

@ -0,0 +1,241 @@
//go:generate mockgen -destination mock_consensusclient/mock_consensusclient.go github.com/anyproto/any-sync/consensus/consensusclient Service
package consensusclient
import (
"context"
"errors"
"github.com/anyproto/any-sync/app"
"github.com/anyproto/any-sync/app/logger"
"github.com/anyproto/any-sync/consensus/consensusproto"
"github.com/anyproto/any-sync/net/pool"
"github.com/anyproto/any-sync/net/rpc/rpcerr"
"github.com/anyproto/any-sync/nodeconf"
"go.uber.org/zap"
"sync"
"time"
)
const CName = "consensus.consensusclient"
var log = logger.NewNamed(CName)
var (
ErrWatcherExists = errors.New("watcher exists")
ErrWatcherNotExists = errors.New("watcher not exists")
)
func New() Service {
return new(service)
}
// Watcher watches new events by specified logId
type Watcher interface {
AddConsensusRecords(recs []*consensusproto.RawRecordWithId)
AddConsensusError(err error)
}
type Service interface {
// AddLog adds new log to consensus servers
AddLog(ctx context.Context, clog *consensusproto.Log) (err error)
// AddRecord adds new record to consensus servers
AddRecord(ctx context.Context, logId []byte, clog *consensusproto.RawRecord) (record *consensusproto.RawRecordWithId, err error)
// Watch starts watching to given logId and calls watcher when any relative event received
Watch(logId []byte, w Watcher) (err error)
// UnWatch stops watching given logId and removes watcher
UnWatch(logId []byte) (err error)
app.ComponentRunnable
}
type service struct {
pool pool.Pool
nodeconf nodeconf.Service
watchers map[string]Watcher
stream *stream
close chan struct{}
mu sync.Mutex
}
func (s *service) Init(a *app.App) (err error) {
s.pool = a.MustComponent(pool.CName).(pool.Pool)
s.nodeconf = a.MustComponent(nodeconf.CName).(nodeconf.Service)
s.watchers = make(map[string]Watcher)
s.close = make(chan struct{})
return nil
}
func (s *service) Name() (name string) {
return CName
}
func (s *service) Run(_ context.Context) error {
go s.streamWatcher()
return nil
}
func (s *service) doClient(ctx context.Context, fn func(cl consensusproto.DRPCConsensusClient) error) error {
peer, err := s.pool.GetOneOf(ctx, s.nodeconf.ConsensusPeers())
if err != nil {
return err
}
dc, err := peer.AcquireDrpcConn(ctx)
if err != nil {
return err
}
defer peer.ReleaseDrpcConn(dc)
return fn(consensusproto.NewDRPCConsensusClient(dc))
}
func (s *service) AddLog(ctx context.Context, clog *consensusproto.Log) (err error) {
return s.doClient(ctx, func(cl consensusproto.DRPCConsensusClient) error {
if _, err = cl.LogAdd(ctx, &consensusproto.LogAddRequest{
Log: clog,
}); err != nil {
return rpcerr.Unwrap(err)
}
return nil
})
}
func (s *service) AddRecord(ctx context.Context, logId []byte, clog *consensusproto.RawRecord) (record *consensusproto.RawRecordWithId, err error) {
err = s.doClient(ctx, func(cl consensusproto.DRPCConsensusClient) error {
if record, err = cl.RecordAdd(ctx, &consensusproto.RecordAddRequest{
LogId: logId,
Record: clog,
}); err != nil {
return rpcerr.Unwrap(err)
}
return nil
})
return
}
func (s *service) Watch(logId []byte, w Watcher) (err error) {
s.mu.Lock()
defer s.mu.Unlock()
if _, ok := s.watchers[string(logId)]; ok {
return ErrWatcherExists
}
s.watchers[string(logId)] = w
if s.stream != nil {
if wErr := s.stream.WatchIds([][]byte{logId}); wErr != nil {
log.Warn("WatchIds error", zap.Error(wErr))
}
}
return
}
func (s *service) UnWatch(logId []byte) (err error) {
s.mu.Lock()
defer s.mu.Unlock()
if _, ok := s.watchers[string(logId)]; !ok {
return ErrWatcherNotExists
}
delete(s.watchers, string(logId))
if s.stream != nil {
if wErr := s.stream.UnwatchIds([][]byte{logId}); wErr != nil {
log.Warn("UnWatchIds error", zap.Error(wErr))
}
}
return
}
func (s *service) openStream(ctx context.Context) (st *stream, err error) {
pr, err := s.pool.GetOneOf(ctx, s.nodeconf.ConsensusPeers())
if err != nil {
return nil, err
}
dc, err := pr.AcquireDrpcConn(ctx)
if err != nil {
return nil, err
}
rpcStream, err := consensusproto.NewDRPCConsensusClient(dc).LogWatch(ctx)
if err != nil {
return nil, rpcerr.Unwrap(err)
}
return runStream(rpcStream), nil
}
func (s *service) streamWatcher() {
var (
err error
st *stream
i int
)
for {
// open stream
if st, err = s.openStream(context.Background()); err != nil {
// can't open stream, we will retry until success connection or close
if i < 60 {
i++
}
sleepTime := time.Second * time.Duration(i)
log.Error("watch log error", zap.Error(err), zap.Duration("waitTime", sleepTime))
select {
case <-time.After(sleepTime):
continue
case <-s.close:
return
}
}
i = 0
// collect ids and setup stream
s.mu.Lock()
var logIds = make([][]byte, 0, len(s.watchers))
for id := range s.watchers {
logIds = append(logIds, []byte(id))
}
s.stream = st
s.mu.Unlock()
// restore subscriptions
if len(logIds) > 0 {
if err = s.stream.WatchIds(logIds); err != nil {
log.Error("watch ids error", zap.Error(err))
continue
}
}
// read stream
if err = s.streamReader(); err != nil {
log.Error("stream read error", zap.Error(err))
continue
}
return
}
}
func (s *service) streamReader() error {
for {
events := s.stream.WaitLogs()
if len(events) == 0 {
return s.stream.Err()
}
for _, e := range events {
if w, ok := s.watchers[string(e.LogId)]; ok {
if e.Error == nil {
w.AddConsensusRecords(e.Records)
} else {
w.AddConsensusError(rpcerr.Err(uint64(e.Error.Error)))
}
} else {
log.Warn("received unexpected log id", zap.Binary("logId", e.LogId))
}
}
}
}
func (s *service) Close(_ context.Context) error {
s.mu.Lock()
if s.stream != nil {
_ = s.stream.Close()
}
s.mu.Unlock()
select {
case <-s.close:
default:
close(s.close)
}
return nil
}

241
consensus/consensusclient/client_test.go Normal file
View File

@ -0,0 +1,241 @@
package consensusclient
import (
"context"
"fmt"
"github.com/anyproto/any-sync/app"
"github.com/anyproto/any-sync/consensus/consensusproto"
"github.com/anyproto/any-sync/consensus/consensusproto/consensuserr"
"github.com/anyproto/any-sync/net/pool"
"github.com/anyproto/any-sync/net/rpc/rpctest"
"github.com/anyproto/any-sync/nodeconf"
"github.com/anyproto/any-sync/nodeconf/mock_nodeconf"
"github.com/anyproto/any-sync/testutil/accounttest"
"github.com/anyproto/any-sync/util/cidutil"
"github.com/golang/mock/gomock"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"sync"
"testing"
"time"
)
func TestService_Watch(t *testing.T) {
t.Run("not found error", func(t *testing.T) {
fx := newFixture(t).run(t)
defer fx.Finish()
var logId = []byte{'1'}
w := &testWatcher{ready: make(chan struct{})}
require.NoError(t, fx.Watch(logId, w))
st := fx.testServer.waitStream(t)
req, err := st.Recv()
require.NoError(t, err)
assert.Equal(t, [][]byte{logId}, req.WatchIds)
require.NoError(t, st.Send(&consensusproto.LogWatchEvent{
LogId: logId,
Error: &consensusproto.Err{
Error: consensusproto.ErrCodes_ErrorOffset + consensusproto.ErrCodes_LogNotFound,
},
}))
<-w.ready
assert.Equal(t, consensuserr.ErrLogNotFound, w.err)
fx.testServer.releaseStream <- nil
})
t.Run("watcherExists error", func(t *testing.T) {
fx := newFixture(t).run(t)
defer fx.Finish()
var logId = []byte{'1'}
w := &testWatcher{}
require.NoError(t, fx.Watch(logId, w))
require.Error(t, fx.Watch(logId, w))
st := fx.testServer.waitStream(t)
st.Recv()
fx.testServer.releaseStream <- nil
})
t.Run("watch", func(t *testing.T) {
fx := newFixture(t).run(t)
defer fx.Finish()
var logId1 = []byte{'1'}
w := &testWatcher{}
require.NoError(t, fx.Watch(logId1, w))
st := fx.testServer.waitStream(t)
req, err := st.Recv()
require.NoError(t, err)
assert.Equal(t, [][]byte{logId1}, req.WatchIds)
var logId2 = []byte{'2'}
w = &testWatcher{}
require.NoError(t, fx.Watch(logId2, w))
req, err = st.Recv()
require.NoError(t, err)
assert.Equal(t, [][]byte{logId2}, req.WatchIds)
fx.testServer.releaseStream <- nil
})
}
func TestService_UnWatch(t *testing.T) {
t.Run("no watcher", func(t *testing.T) {
fx := newFixture(t).run(t)
defer fx.Finish()
require.Error(t, fx.UnWatch([]byte{'1'}))
})
t.Run("success", func(t *testing.T) {
fx := newFixture(t).run(t)
defer fx.Finish()
w := &testWatcher{}
require.NoError(t, fx.Watch([]byte{'1'}, w))
assert.NoError(t, fx.UnWatch([]byte{'1'}))
})
}
func TestService_Init(t *testing.T) {
t.Run("reconnect on watch err", func(t *testing.T) {
fx := newFixture(t)
fx.testServer.watchErrOnce = true
fx.run(t)
defer fx.Finish()
fx.testServer.waitStream(t)
fx.testServer.releaseStream <- nil
})
t.Run("reconnect on start", func(t *testing.T) {
fx := newFixture(t)
fx.a.MustComponent(pool.CName).(*rpctest.TestPool).WithServer(nil)
fx.run(t)
defer fx.Finish()
time.Sleep(time.Millisecond * 50)
fx.a.MustComponent(pool.CName).(*rpctest.TestPool).WithServer(fx.drpcTS)
fx.testServer.waitStream(t)
fx.testServer.releaseStream <- nil
})
}
func TestService_AddLog(t *testing.T) {
fx := newFixture(t).run(t)
defer fx.Finish()
assert.NoError(t, fx.AddLog(ctx, &consensusproto.Log{}))
}
func TestService_AddRecord(t *testing.T) {
fx := newFixture(t).run(t)
defer fx.Finish()
rec, err := fx.AddRecord(ctx, []byte{'1'}, &consensusproto.RawRecord{})
require.NoError(t, err)
assert.NotEmpty(t, rec)
}
var ctx = context.Background()
func newFixture(t *testing.T) *fixture {
fx := &fixture{
Service: New(),
a: &app.App{},
ctrl: gomock.NewController(t),
testServer: &testServer{
stream: make(chan consensusproto.DRPCConsensus_LogWatchStream),
releaseStream: make(chan error),
},
}
fx.nodeconf = mock_nodeconf.NewMockService(fx.ctrl)
fx.nodeconf.EXPECT().Name().Return(nodeconf.CName).AnyTimes()
fx.nodeconf.EXPECT().Init(gomock.Any()).AnyTimes()
fx.nodeconf.EXPECT().Run(gomock.Any()).AnyTimes()
fx.nodeconf.EXPECT().Close(gomock.Any()).AnyTimes()
fx.nodeconf.EXPECT().ConsensusPeers().Return([]string{"c1", "c2", "c3"}).AnyTimes()
fx.drpcTS = rpctest.NewTestServer()
require.NoError(t, consensusproto.DRPCRegisterConsensus(fx.drpcTS.Mux, fx.testServer))
fx.a.Register(fx.Service).
Register(&accounttest.AccountTestService{}).
Register(fx.nodeconf).
Register(rpctest.NewTestPool().WithServer(fx.drpcTS))
return fx
}
type fixture struct {
Service
a *app.App
ctrl *gomock.Controller
testServer *testServer
drpcTS *rpctest.TestServer
nodeconf *mock_nodeconf.MockService
}
func (fx *fixture) run(t *testing.T) *fixture {
require.NoError(t, fx.a.Start(ctx))
return fx
}
func (fx *fixture) Finish() {
assert.NoError(fx.ctrl.T, fx.a.Close(ctx))
fx.ctrl.Finish()
}
type testServer struct {
stream chan consensusproto.DRPCConsensus_LogWatchStream
addLog func(ctx context.Context, req *consensusproto.LogAddRequest) error
addRecord func(ctx context.Context, req *consensusproto.RecordAddRequest) error
releaseStream chan error
watchErrOnce bool
}
func (t *testServer) LogAdd(ctx context.Context, req *consensusproto.LogAddRequest) (*consensusproto.Ok, error) {
if t.addLog != nil {
if err := t.addLog(ctx, req); err != nil {
return nil, err
}
}
return &consensusproto.Ok{}, nil
}
func (t *testServer) RecordAdd(ctx context.Context, req *consensusproto.RecordAddRequest) (*consensusproto.RawRecordWithId, error) {
if t.addRecord != nil {
if err := t.addRecord(ctx, req); err != nil {
return nil, err
}
}
data, _ := req.Record.Marshal()
id, _ := cidutil.NewCidFromBytes(data)
return &consensusproto.RawRecordWithId{Id: id, Payload: data}, nil
}
func (t *testServer) LogWatch(stream consensusproto.DRPCConsensus_LogWatchStream) error {
if t.watchErrOnce {
t.watchErrOnce = false
return fmt.Errorf("error")
}
t.stream <- stream
return <-t.releaseStream
}
func (t *testServer) waitStream(test *testing.T) consensusproto.DRPCConsensus_LogWatchStream {
select {
case <-time.After(time.Second * 5):
test.Fatalf("waiteStream timeout")
case st := <-t.stream:
return st
}
return nil
}
type testWatcher struct {
recs [][]*consensusproto.RawRecordWithId
err error
ready chan struct{}
once sync.Once
}
func (t *testWatcher) AddConsensusRecords(recs []*consensusproto.RawRecordWithId) {
t.recs = append(t.recs, recs)
t.once.Do(func() {
close(t.ready)
})
}
func (t *testWatcher) AddConsensusError(err error) {
t.err = err
t.once.Do(func() {
close(t.ready)
})
}

151
consensus/consensusclient/mock_consensusclient/mock_consensusclient.go Normal file
View File

@ -0,0 +1,151 @@
// Code generated by MockGen. DO NOT EDIT.
// Source: github.com/anyproto/any-sync/consensus/consensusclient (interfaces: Service)
// Package mock_consensusclient is a generated GoMock package.
package mock_consensusclient
import (
context "context"
reflect "reflect"
app "github.com/anyproto/any-sync/app"
consensusclient "github.com/anyproto/any-sync/consensus/consensusclient"
consensusproto "github.com/anyproto/any-sync/consensus/consensusproto"
gomock "github.com/golang/mock/gomock"
)
// MockService is a mock of Service interface.
type MockService struct {
ctrl *gomock.Controller
recorder *MockServiceMockRecorder
}
// MockServiceMockRecorder is the mock recorder for MockService.
type MockServiceMockRecorder struct {
mock *MockService
}
// NewMockService creates a new mock instance.
func NewMockService(ctrl *gomock.Controller) *MockService {
mock := &MockService{ctrl: ctrl}
mock.recorder = &MockServiceMockRecorder{mock}
return mock
}
// EXPECT returns an object that allows the caller to indicate expected use.
func (m *MockService) EXPECT() *MockServiceMockRecorder {
return m.recorder
}
// AddLog mocks base method.
func (m *MockService) AddLog(arg0 context.Context, arg1 *consensusproto.Log) error {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "AddLog", arg0, arg1)
ret0, _ := ret[0].(error)
return ret0
}
// AddLog indicates an expected call of AddLog.
func (mr *MockServiceMockRecorder) AddLog(arg0, arg1 interface{}) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AddLog", reflect.TypeOf((*MockService)(nil).AddLog), arg0, arg1)
}
// AddRecord mocks base method.
func (m *MockService) AddRecord(arg0 context.Context, arg1 []byte, arg2 *consensusproto.RawRecord) (*consensusproto.RawRecordWithId, error) {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "AddRecord", arg0, arg1, arg2)
ret0, _ := ret[0].(*consensusproto.RawRecordWithId)
ret1, _ := ret[1].(error)
return ret0, ret1
}
// AddRecord indicates an expected call of AddRecord.
func (mr *MockServiceMockRecorder) AddRecord(arg0, arg1, arg2 interface{}) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "AddRecord", reflect.TypeOf((*MockService)(nil).AddRecord), arg0, arg1, arg2)
}
// Close mocks base method.
func (m *MockService) Close(arg0 context.Context) error {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "Close", arg0)
ret0, _ := ret[0].(error)
return ret0
}
// Close indicates an expected call of Close.
func (mr *MockServiceMockRecorder) Close(arg0 interface{}) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Close", reflect.TypeOf((*MockService)(nil).Close), arg0)
}
// Init mocks base method.
func (m *MockService) Init(arg0 *app.App) error {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "Init", arg0)
ret0, _ := ret[0].(error)
return ret0
}
// Init indicates an expected call of Init.
func (mr *MockServiceMockRecorder) Init(arg0 interface{}) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Init", reflect.TypeOf((*MockService)(nil).Init), arg0)
}
// Name mocks base method.
func (m *MockService) Name() string {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "Name")
ret0, _ := ret[0].(string)
return ret0
}
// Name indicates an expected call of Name.
func (mr *MockServiceMockRecorder) Name() *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Name", reflect.TypeOf((*MockService)(nil).Name))
}
// Run mocks base method.
func (m *MockService) Run(arg0 context.Context) error {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "Run", arg0)
ret0, _ := ret[0].(error)
return ret0
}
// Run indicates an expected call of Run.
func (mr *MockServiceMockRecorder) Run(arg0 interface{}) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Run", reflect.TypeOf((*MockService)(nil).Run), arg0)
}
// UnWatch mocks base method.
func (m *MockService) UnWatch(arg0 []byte) error {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "UnWatch", arg0)
ret0, _ := ret[0].(error)
return ret0
}
// UnWatch indicates an expected call of UnWatch.
func (mr *MockServiceMockRecorder) UnWatch(arg0 interface{}) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UnWatch", reflect.TypeOf((*MockService)(nil).UnWatch), arg0)
}
// Watch mocks base method.
func (m *MockService) Watch(arg0 []byte, arg1 consensusclient.Watcher) error {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "Watch", arg0, arg1)
ret0, _ := ret[0].(error)
return ret0
}
// Watch indicates an expected call of Watch.
func (mr *MockServiceMockRecorder) Watch(arg0, arg1 interface{}) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Watch", reflect.TypeOf((*MockService)(nil).Watch), arg0, arg1)
}

70
consensus/consensusclient/stream.go Normal file
View File

@ -0,0 +1,70 @@
package consensusclient
import (
"context"
"github.com/anyproto/any-sync/consensus/consensusproto"
"github.com/cheggaaa/mb/v3"
"sync"
)
func runStream(rpcStream consensusproto.DRPCConsensus_LogWatchClient) *stream {
st := &stream{
rpcStream: rpcStream,
mb: mb.New[*consensusproto.LogWatchEvent](100),
}
go st.readStream()
return st
}
type stream struct {
rpcStream consensusproto.DRPCConsensus_LogWatchClient
mb *mb.MB[*consensusproto.LogWatchEvent]
mu sync.Mutex
err error
}
func (s *stream) WatchIds(logIds [][]byte) (err error) {
return s.rpcStream.Send(&consensusproto.LogWatchRequest{
WatchIds: logIds,
})
}
func (s *stream) UnwatchIds(logIds [][]byte) (err error) {
return s.rpcStream.Send(&consensusproto.LogWatchRequest{
UnwatchIds: logIds,
})
}
func (s *stream) WaitLogs() []*consensusproto.LogWatchEvent {
events, _ := s.mb.Wait(context.TODO())
return events
}
func (s *stream) Err() error {
s.mu.Lock()
defer s.mu.Unlock()
return s.err
}
func (s *stream) readStream() {
defer s.Close()
for {
event, err := s.rpcStream.Recv()
if err != nil {
s.mu.Lock()
s.err = err
s.mu.Unlock()
return
}
if err = s.mb.Add(s.rpcStream.Context(), event); err != nil {
return
}
}
}
func (s *stream) Close() error {
if err := s.mb.Close(); err == nil {
return s.rpcStream.Close()
}
return nil
}

2572
consensus/consensusproto/consensus.pb.go Normal file
View File

File diff suppressed because it is too large Load Diff

232
consensus/consensusproto/consensus_drpc.pb.go Normal file
View File

@ -0,0 +1,232 @@
// Code generated by protoc-gen-go-drpc. DO NOT EDIT.
// protoc-gen-go-drpc version: v0.0.33
// source: consensus/consensusproto/protos/consensus.proto
package consensusproto
import (
bytes "bytes"
context "context"
errors "errors"
jsonpb "github.com/gogo/protobuf/jsonpb"
proto "github.com/gogo/protobuf/proto"
drpc "storj.io/drpc"
drpcerr "storj.io/drpc/drpcerr"
)
type drpcEncoding_File_consensus_consensusproto_protos_consensus_proto struct{}
func (drpcEncoding_File_consensus_consensusproto_protos_consensus_proto) Marshal(msg drpc.Message) ([]byte, error) {
return proto.Marshal(msg.(proto.Message))
}
func (drpcEncoding_File_consensus_consensusproto_protos_consensus_proto) Unmarshal(buf []byte, msg drpc.Message) error {
return proto.Unmarshal(buf, msg.(proto.Message))
}
func (drpcEncoding_File_consensus_consensusproto_protos_consensus_proto) JSONMarshal(msg drpc.Message) ([]byte, error) {
var buf bytes.Buffer
err := new(jsonpb.Marshaler).Marshal(&buf, msg.(proto.Message))
if err != nil {
return nil, err
}
return buf.Bytes(), nil
}
func (drpcEncoding_File_consensus_consensusproto_protos_consensus_proto) JSONUnmarshal(buf []byte, msg drpc.Message) error {
return jsonpb.Unmarshal(bytes.NewReader(buf), msg.(proto.Message))
}
type DRPCConsensusClient interface {
DRPCConn() drpc.Conn
LogAdd(ctx context.Context, in *LogAddRequest) (*Ok, error)
RecordAdd(ctx context.Context, in *RecordAddRequest) (*RawRecordWithId, error)
LogWatch(ctx context.Context) (DRPCConsensus_LogWatchClient, error)
}
type drpcConsensusClient struct {
cc drpc.Conn
}
func NewDRPCConsensusClient(cc drpc.Conn) DRPCConsensusClient {
return &drpcConsensusClient{cc}
}
func (c *drpcConsensusClient) DRPCConn() drpc.Conn { return c.cc }
func (c *drpcConsensusClient) LogAdd(ctx context.Context, in *LogAddRequest) (*Ok, error) {
out := new(Ok)
err := c.cc.Invoke(ctx, "/consensusProto.Consensus/LogAdd", drpcEncoding_File_consensus_consensusproto_protos_consensus_proto{}, in, out)
if err != nil {
return nil, err
}
return out, nil
}
func (c *drpcConsensusClient) RecordAdd(ctx context.Context, in *RecordAddRequest) (*RawRecordWithId, error) {
out := new(RawRecordWithId)
err := c.cc.Invoke(ctx, "/consensusProto.Consensus/RecordAdd", drpcEncoding_File_consensus_consensusproto_protos_consensus_proto{}, in, out)
if err != nil {
return nil, err
}
return out, nil
}
func (c *drpcConsensusClient) LogWatch(ctx context.Context) (DRPCConsensus_LogWatchClient, error) {
stream, err := c.cc.NewStream(ctx, "/consensusProto.Consensus/LogWatch", drpcEncoding_File_consensus_consensusproto_protos_consensus_proto{})
if err != nil {
return nil, err
}
x := &drpcConsensus_LogWatchClient{stream}
return x, nil
}
type DRPCConsensus_LogWatchClient interface {
drpc.Stream
Send(*LogWatchRequest) error
Recv() (*LogWatchEvent, error)
}
type drpcConsensus_LogWatchClient struct {
drpc.Stream
}
func (x *drpcConsensus_LogWatchClient) GetStream() drpc.Stream {
return x.Stream
}
func (x *drpcConsensus_LogWatchClient) Send(m *LogWatchRequest) error {
return x.MsgSend(m, drpcEncoding_File_consensus_consensusproto_protos_consensus_proto{})
}
func (x *drpcConsensus_LogWatchClient) Recv() (*LogWatchEvent, error) {
m := new(LogWatchEvent)
if err := x.MsgRecv(m, drpcEncoding_File_consensus_consensusproto_protos_consensus_proto{}); err != nil {
return nil, err
}
return m, nil
}
func (x *drpcConsensus_LogWatchClient) RecvMsg(m *LogWatchEvent) error {
return x.MsgRecv(m, drpcEncoding_File_consensus_consensusproto_protos_consensus_proto{})
}
type DRPCConsensusServer interface {
LogAdd(context.Context, *LogAddRequest) (*Ok, error)
RecordAdd(context.Context, *RecordAddRequest) (*RawRecordWithId, error)
LogWatch(DRPCConsensus_LogWatchStream) error
}
type DRPCConsensusUnimplementedServer struct{}
func (s *DRPCConsensusUnimplementedServer) LogAdd(context.Context, *LogAddRequest) (*Ok, error) {
return nil, drpcerr.WithCode(errors.New("Unimplemented"), drpcerr.Unimplemented)
}
func (s *DRPCConsensusUnimplementedServer) RecordAdd(context.Context, *RecordAddRequest) (*RawRecordWithId, error) {
return nil, drpcerr.WithCode(errors.New("Unimplemented"), drpcerr.Unimplemented)
}
func (s *DRPCConsensusUnimplementedServer) LogWatch(DRPCConsensus_LogWatchStream) error {
return drpcerr.WithCode(errors.New("Unimplemented"), drpcerr.Unimplemented)
}
type DRPCConsensusDescription struct{}
func (DRPCConsensusDescription) NumMethods() int { return 3 }
func (DRPCConsensusDescription) Method(n int) (string, drpc.Encoding, drpc.Receiver, interface{}, bool) {
switch n {
case 0:
return "/consensusProto.Consensus/LogAdd", drpcEncoding_File_consensus_consensusproto_protos_consensus_proto{},
func(srv interface{}, ctx context.Context, in1, in2 interface{}) (drpc.Message, error) {
return srv.(DRPCConsensusServer).
LogAdd(
ctx,
in1.(*LogAddRequest),
)
}, DRPCConsensusServer.LogAdd, true
case 1:
return "/consensusProto.Consensus/RecordAdd", drpcEncoding_File_consensus_consensusproto_protos_consensus_proto{},
func(srv interface{}, ctx context.Context, in1, in2 interface{}) (drpc.Message, error) {
return srv.(DRPCConsensusServer).
RecordAdd(
ctx,
in1.(*RecordAddRequest),
)
}, DRPCConsensusServer.RecordAdd, true
case 2:
return "/consensusProto.Consensus/LogWatch", drpcEncoding_File_consensus_consensusproto_protos_consensus_proto{},
func(srv interface{}, ctx context.Context, in1, in2 interface{}) (drpc.Message, error) {
return nil, srv.(DRPCConsensusServer).
LogWatch(
&drpcConsensus_LogWatchStream{in1.(drpc.Stream)},
)
}, DRPCConsensusServer.LogWatch, true
default:
return "", nil, nil, nil, false
}
}
func DRPCRegisterConsensus(mux drpc.Mux, impl DRPCConsensusServer) error {
return mux.Register(impl, DRPCConsensusDescription{})
}
type DRPCConsensus_LogAddStream interface {
drpc.Stream
SendAndClose(*Ok) error
}
type drpcConsensus_LogAddStream struct {
drpc.Stream
}
func (x *drpcConsensus_LogAddStream) SendAndClose(m *Ok) error {
if err := x.MsgSend(m, drpcEncoding_File_consensus_consensusproto_protos_consensus_proto{}); err != nil {
return err
}
return x.CloseSend()
}
type DRPCConsensus_RecordAddStream interface {
drpc.Stream
SendAndClose(*RawRecordWithId) error
}
type drpcConsensus_RecordAddStream struct {
drpc.Stream
}
func (x *drpcConsensus_RecordAddStream) SendAndClose(m *RawRecordWithId) error {
if err := x.MsgSend(m, drpcEncoding_File_consensus_consensusproto_protos_consensus_proto{}); err != nil {
return err
}
return x.CloseSend()
}
type DRPCConsensus_LogWatchStream interface {
drpc.Stream
Send(*LogWatchEvent) error
Recv() (*LogWatchRequest, error)
}
type drpcConsensus_LogWatchStream struct {
drpc.Stream
}
func (x *drpcConsensus_LogWatchStream) Send(m *LogWatchEvent) error {
return x.MsgSend(m, drpcEncoding_File_consensus_consensusproto_protos_consensus_proto{})
}
func (x *drpcConsensus_LogWatchStream) Recv() (*LogWatchRequest, error) {
m := new(LogWatchRequest)
if err := x.MsgRecv(m, drpcEncoding_File_consensus_consensusproto_protos_consensus_proto{}); err != nil {
return nil, err
}
return m, nil
}
func (x *drpcConsensus_LogWatchStream) RecvMsg(m *LogWatchRequest) error {
return x.MsgRecv(m, drpcEncoding_File_consensus_consensusproto_protos_consensus_proto{})
}

16
consensus/consensusproto/consensuserr/errors.go Normal file
View File

@ -0,0 +1,16 @@
package consensuserr
import (
"fmt"
"github.com/anyproto/any-sync/consensus/consensusproto"
"github.com/anyproto/any-sync/net/rpc/rpcerr"
)
var (
errGroup = rpcerr.ErrGroup(consensusproto.ErrCodes_ErrorOffset)
ErrUnexpected = errGroup.Register(fmt.Errorf("unexpected consensus error"), uint64(consensusproto.ErrCodes_Unexpected))
ErrConflict = errGroup.Register(fmt.Errorf("records conflict"), uint64(consensusproto.ErrCodes_RecordConflict))
ErrLogExists = errGroup.Register(fmt.Errorf("log exists"), uint64(consensusproto.ErrCodes_LogExists))
ErrLogNotFound = errGroup.Register(fmt.Errorf("log not found"), uint64(consensusproto.ErrCodes_LogNotFound))
)

77
consensus/consensusproto/protos/consensus.proto Normal file
View File

@ -0,0 +1,77 @@
syntax = "proto3";
package consensusProto;
option go_package = "consensus/consensusproto";
enum ErrCodes {
Unexpected = 0;
LogExists = 1;
LogNotFound = 2;
RecordConflict = 3;
ErrorOffset = 500;
}
message Log {
bytes id = 1;
bytes payload = 2;
repeated RawRecordWithId records = 3;
}
// RawRecord is a proto message containing the payload in bytes, signature of the account who added it and signature of the acceptor
message RawRecord {
bytes payload = 1;
bytes signature = 2;
bytes acceptorIdentity = 3;
bytes acceptorSignature = 4;
}
// RawRecordWithId is a raw record and the id for convenience
message RawRecordWithId {
bytes payload = 1;
string id = 2;
}
// Record is a record containing a data
message Record {
string prevId = 1;
bytes identity = 2;
bytes data = 3;
int64 timestamp = 4;
}
service Consensus {
// AddLog adds new log to consensus
rpc LogAdd(LogAddRequest) returns (Ok);
// AddRecord adds new record to log
rpc RecordAdd(RecordAddRequest) returns (RawRecordWithId);
// WatchLog fetches log and subscribes for a changes
rpc LogWatch(stream LogWatchRequest) returns (stream LogWatchEvent);
}
message Ok {}
message LogAddRequest {
Log log = 1;
}
message RecordAddRequest {
bytes logId = 1;
RawRecord record = 2;
}
message LogWatchRequest {
repeated bytes watchIds = 1;
repeated bytes unwatchIds = 2;
}
message LogWatchEvent {
bytes logId = 1;
repeated RawRecordWithId records = 2;
Err error = 3;
}
message Err {
ErrCodes error = 1;
}

11
net/peer/peer.go
View File

@ -3,6 +3,12 @@ package peer
import (
"context"
"io"
"net"
"sync"
"sync/atomic"
"time"
"github.com/anyproto/any-sync/app/logger"
"github.com/anyproto/any-sync/app/ocache"
"github.com/anyproto/any-sync/net/connutil"
@ -11,16 +17,11 @@ import (
"github.com/anyproto/any-sync/net/secureservice/handshake/handshakeproto"
"github.com/anyproto/any-sync/net/transport"
"go.uber.org/zap"
"io"
"net"
"storj.io/drpc"
"storj.io/drpc/drpcconn"
"storj.io/drpc/drpcmanager"
"storj.io/drpc/drpcstream"
"storj.io/drpc/drpcwire"
"sync"
"sync/atomic"
"time"
)
var log = logger.NewNamed("common.net.peer")

3
util/crypto/ed25519_test.go
View File

@ -2,8 +2,9 @@ package crypto
import (
"crypto/rand"
"github.com/stretchr/testify/require"
"testing"
"github.com/stretchr/testify/require"
)
func Test_EncryptDecrypt(t *testing.T) {

3
util/strkey/strkey_test.go
View File

@ -1,9 +1,10 @@
package strkey
import (
"testing"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"testing"
)
func TestDecode(t *testing.T) {