[apple/security.git] / OSX / sec / SOSCircle / SecureObjectSync / SOSTransportMessageKVS.c

#include <Security/SecureObjectSync/SOSTransport.h>
#include <Security/SecureObjectSync/SOSTransportMessage.h>
#include <Security/SecureObjectSync/SOSTransportMessageKVS.h>
#include <Security/SecureObjectSync/SOSKVSKeys.h>
#include <Security/SecureObjectSync/SOSAccountPriv.h>
#include <utilities/SecCFWrappers.h>
#include <SOSInternal.h>
#include <AssertMacros.h>
#include <SOSCloudKeychainClient.h>

struct __OpaqueSOSTransportMessageKVS {
    struct __OpaqueSOSTransportMessage          m;
    CFMutableDictionaryRef  pending_changes;
    
};

//
// V-table implementation forward declarations
//
static bool sendToPeer(SOSTransportMessageRef transport, CFStringRef circleName, CFStringRef peerID, CFDataRef message, CFErrorRef *error);
static bool syncWithPeers(SOSTransportMessageRef transport, CFSetRef peers, CFErrorRef *error);
static bool sendMessages(SOSTransportMessageRef transport, CFDictionaryRef peersToMessage, CFErrorRef *error);
static bool cleanupAfterPeer(SOSTransportMessageRef transport, CFDictionaryRef circle_to_peer_ids, CFErrorRef *error);
static void destroy(SOSTransportMessageRef transport);
static inline CFIndex getTransportType(SOSTransportMessageRef transport, CFErrorRef *error);

static CF_RETURNS_RETAINED
CFDictionaryRef handleMessages(SOSTransportMessageRef transport, CFMutableDictionaryRef circle_peer_messages_table, CFErrorRef *error);
    
static bool flushChanges(SOSTransportMessageRef transport, CFErrorRef *error);

SOSTransportMessageKVSRef SOSTransportMessageKVSCreate(SOSAccountRef account, CFStringRef circleName, CFErrorRef *error){
    SOSTransportMessageKVSRef tkvs = (SOSTransportMessageKVSRef) SOSTransportMessageCreateForSubclass(sizeof(struct __OpaqueSOSTransportMessageKVS) - sizeof(CFRuntimeBase), account, circleName, error);
    
    if (tkvs) {
        // Fill in vtable:
        tkvs->m.sendMessages = sendMessages;
        tkvs->m.syncWithPeers = syncWithPeers;
        tkvs->m.flushChanges = flushChanges;
        tkvs->m.cleanupAfterPeerMessages = cleanupAfterPeer;
        tkvs->m.destroy = destroy;
        tkvs->m.handleMessages = handleMessages;
        tkvs->m.getTransportType = getTransportType;
        // Initialize ourselves
        tkvs->pending_changes = CFDictionaryCreateMutableForCFTypes(kCFAllocatorDefault);
        SOSRegisterTransportMessage((SOSTransportMessageRef)tkvs);
    }
    
    return tkvs;
}

bool SOSTransportMessageKVSAppendKeyInterest(SOSTransportMessageKVSRef transport, CFMutableArrayRef alwaysKeys, CFMutableArrayRef afterFirstUnlockKeys, CFMutableArrayRef unlockedKeys, CFErrorRef *localError){
    SOSEngineRef engine = SOSTransportMessageGetEngine((SOSTransportMessageRef)transport);
    require_quiet(engine, fail);
    
    CFArrayRef peerInfos = SOSAccountCopyPeersToListenTo(SOSTransportMessageGetAccount((SOSTransportMessageRef) transport), localError);
    
    if(peerInfos){
        CFArrayForEach(peerInfos, ^(const void *value) {
            SOSPeerInfoRef peer = (SOSPeerInfoRef)value;
            CFStringRef peerID = SOSPeerInfoGetPeerID(peer);
            CFStringRef peerMessage = SOSMessageKeyCreateFromPeerToTransport((SOSTransportMessageRef)transport, peerID);
            if(peerMessage != NULL)
                CFArrayAppendValue(unlockedKeys, peerMessage);
            CFReleaseNull(peerMessage);
        });
        CFReleaseNull(peerInfos);
    }
    return true;
fail:
    return false;
}
static void destroy(SOSTransportMessageRef transport){
    SOSTransportMessageKVSRef tkvs = (SOSTransportMessageKVSRef)transport;
    CFReleaseNull(tkvs->pending_changes);
    SOSUnregisterTransportMessage((SOSTransportMessageRef)tkvs);
    
}

static inline CFIndex getTransportType(SOSTransportMessageRef transport, CFErrorRef *error){
    return kKVS;
}

static bool SOSTransportMessageKVSUpdateKVS(SOSTransportMessageKVSRef transport, CFDictionaryRef changes, CFErrorRef *error){

    CloudKeychainReplyBlock log_error = ^(CFDictionaryRef returnedValues __unused, CFErrorRef block_error) {
        if (block_error) {
            secerror("Error putting: %@", block_error);
        }
    };
    
    SOSCloudKeychainPutObjectsInCloud(changes, dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0), log_error);
    return true;
}

static bool SOSTransportMessageKVSSendPendingChanges(SOSTransportMessageKVSRef transport, CFErrorRef *error) {
    CFErrorRef changeError = NULL;
    
    if (transport->pending_changes == NULL || CFDictionaryGetCount(transport->pending_changes) == 0) {
        CFReleaseNull(transport->pending_changes);
        return true;
    }
    SOSAccountRef account = SOSTransportMessageGetAccount((SOSTransportMessageRef)transport);
    CFTypeRef dsid = SOSAccountGetValue(account, kSOSDSIDKey, error);
    
    if(dsid == NULL)
        dsid = kCFNull;
    
    CFDictionaryAddValue(transport->pending_changes, kSOSKVSRequiredKey, dsid);
    
    bool success = SOSTransportMessageKVSUpdateKVS(transport, transport->pending_changes, &changeError);
    if (success) {
        CFDictionaryRemoveAllValues(transport->pending_changes);
    } else {
        SOSCreateErrorWithFormat(kSOSErrorSendFailure, changeError, error, NULL,
                                 CFSTR("Send changes block failed [%@]"), transport->pending_changes);
    }
    
    return success;
}

static void SOSTransportMessageKVSAddToPendingChanges(SOSTransportMessageKVSRef transport, CFStringRef message_key, CFDataRef message_data){
    if (transport->pending_changes == NULL) {
        transport->pending_changes = CFDictionaryCreateMutableForCFTypes(kCFAllocatorDefault);
    }
    if (message_data == NULL) {
        CFDictionarySetValue(transport->pending_changes, message_key, kCFNull);
    } else {
        CFDictionarySetValue(transport->pending_changes, message_key, message_data);
    }
}

static bool SOSTransportMessageKVSCleanupAfterPeerMessages(SOSTransportMessageKVSRef transport, CFDictionaryRef circle_to_peer_ids, CFErrorRef *error)
{
    CFArrayRef enginePeers = SOSEngineGetPeerIDs(SOSTransportMessageGetEngine((SOSTransportMessageRef)transport));
    __block SOSAccountRef account = SOSTransportMessageGetAccount((SOSTransportMessageRef)transport);
 
    CFDictionaryForEach(circle_to_peer_ids, ^(const void *key, const void *value) {
        if (isString(key) && isArray(value)) {
            CFStringRef circle_name = (CFStringRef) key;
            CFArrayRef peers_to_cleanup_after = (CFArrayRef) value;
            
            CFArrayForEach(peers_to_cleanup_after, ^(const void *value) {
                if (isString(value)) {
                    CFStringRef cleanup_id = (CFStringRef) value;
                    // TODO: Since the enginePeers list is not authorative (the Account is) this could inadvertently clean up active peers or leave behind stale peers
                    if (enginePeers) CFArrayForEach(enginePeers, ^(const void *value) {
                        if (isString(value)) {
                            CFStringRef in_circle_id = (CFStringRef) value;
                            
                            CFStringRef kvsKey = SOSMessageKeyCreateWithCircleNameAndPeerNames(circle_name, cleanup_id, in_circle_id);
                            SOSTransportMessageKVSAddToPendingChanges(transport, kvsKey, NULL);
                            CFReleaseSafe(kvsKey);
                            
                            CFStringRef lastCirclePushedKey = SOSLastCirclePushedKeyCreateWithCircleNameAndPeerID(circle_name, cleanup_id);
                            SOSTransportMessageKVSAddToPendingChanges(transport, lastCirclePushedKey, NULL);
                            CFReleaseSafe(lastCirclePushedKey);
                            
                            CFStringRef lastKeyParameterPushedKey = SOSLastKeyParametersPushedKeyCreateWithPeerID(cleanup_id);
                            SOSTransportMessageKVSAddToPendingChanges(transport, lastKeyParameterPushedKey, NULL);
                            CFReleaseSafe(lastKeyParameterPushedKey);
                    
			    CFStringRef lastCirclePushedWithAccountGestaltKey = SOSLastCirclePushedKeyCreateWithAccountGestalt(account);
	                    SOSTransportMessageKVSAddToPendingChanges(transport, lastCirclePushedKey, NULL);
        		    CFReleaseSafe(lastCirclePushedWithAccountGestaltKey);

		            CFStringRef lastKeyParameterWithAccountGestaltKey = SOSLastKeyParametersPushedKeyCreateWithAccountGestalt(account);
			    SOSTransportMessageKVSAddToPendingChanges(transport, lastCirclePushedKey, NULL);
                            CFReleaseSafe(lastKeyParameterWithAccountGestaltKey);

                            kvsKey = SOSMessageKeyCreateWithCircleNameAndPeerNames(circle_name, in_circle_id, cleanup_id);
                            SOSTransportMessageKVSAddToPendingChanges(transport, kvsKey, NULL);
                            CFReleaseSafe(kvsKey);
                        }
                    });
                    
                }
            });
        }
    });
    
    return SOSTransportMessageFlushChanges((SOSTransportMessageRef)transport, error);
}

static CF_RETURNS_RETAINED
CFDictionaryRef handleMessages(SOSTransportMessageRef transport, CFMutableDictionaryRef circle_peer_messages_table, CFErrorRef *error) {

    CFMutableDictionaryRef handled = CFDictionaryCreateMutableForCFTypes(kCFAllocatorDefault);
    CFDictionaryRef peerToMessage = CFDictionaryGetValue(circle_peer_messages_table, transport->circleName);
    CFMutableArrayRef handled_peers = CFArrayCreateMutableForCFTypes(kCFAllocatorDefault);
    
    if(peerToMessage){
        CFDictionaryForEach(peerToMessage, ^(const void *key, const void *value) {
            CFStringRef peer_id = asString(key, NULL);
            CFDataRef peer_message = asData(value, NULL);
            CFErrorRef localError = NULL;
            
            if (peer_id && peer_message && SOSTransportMessageHandlePeerMessage(transport, peer_id, peer_message, &localError)) {
                CFArrayAppendValue(handled_peers, key);
            } else {
                secnotice("transport", "%@ KVSTransport handle message failed: %@", peer_id, localError);
            }
            CFReleaseNull(localError);
        });
    }
    CFDictionaryAddValue(handled, transport->circleName, handled_peers);
    CFReleaseNull(handled_peers);
    
    return handled;
}


static bool sendToPeer(SOSTransportMessageRef transport, CFStringRef circleName, CFStringRef peerID, CFDataRef message, CFErrorRef *error) {
    SOSTransportMessageKVSRef kvsTransport = (SOSTransportMessageKVSRef) transport;
    bool result = true;
    SOSAccountRef account = SOSTransportMessageGetAccount((SOSTransportMessageRef)transport);
    CFTypeRef dsid = SOSAccountGetValue(account, kSOSDSIDKey, error);
    
    if(dsid == NULL)
        dsid = kCFNull;

    CFStringRef message_to_peer_key = SOSMessageKeyCreateFromTransportToPeer((SOSTransportMessageRef)kvsTransport, peerID);

    CFTypeRef messageToSend = message != NULL ? (CFTypeRef) message : (CFTypeRef) kCFNull;
    CFDictionaryRef a_message_to_a_peer = CFDictionaryCreateForCFTypes(NULL,
                                                                       message_to_peer_key, messageToSend,
                                                                       kSOSKVSRequiredKey, dsid,
                                                                       NULL);
    
    if (!SOSTransportMessageKVSUpdateKVS(kvsTransport, a_message_to_a_peer, error)) {
        secerror("Sync with peers failed to send to %@ [%@], %@", peerID, a_message_to_a_peer, *error);
        result = false;
    }
    CFReleaseNull(a_message_to_a_peer);
    CFReleaseNull(message_to_peer_key);
    
    return result;
}

static bool syncWithPeers(SOSTransportMessageRef transport, CFSetRef peers, CFErrorRef *error) {
    // Each entry is keyed by circle name and contains a list of peerIDs
    __block bool result = true;

    CFSetForEach(peers, ^(const void *value) {
        CFStringRef peerID = asString(value, NULL);
        result &= SOSTransportMessageSendMessageIfNeeded(transport, transport->circleName, peerID, error);
    });

    return result;
}

static bool sendMessages(SOSTransportMessageRef transport, CFDictionaryRef peersToMessage, CFErrorRef *error) {
    __block bool result = true;
    
    CFStringRef circleName = transport->circleName;
    CFDictionaryForEach(peersToMessage, ^(const void *key, const void *value) {
        CFStringRef peerID = asString(key, NULL);
        CFDataRef message = asData(value,NULL);
        if (peerID && message) {
            bool rx = sendToPeer(transport, circleName, peerID, message, error);
            result &= rx;
        }
    });

    return true;
}

static bool flushChanges(SOSTransportMessageRef transport, CFErrorRef *error)
{
    return SOSTransportMessageKVSSendPendingChanges((SOSTransportMessageKVSRef) transport, error);
}

static bool cleanupAfterPeer(SOSTransportMessageRef transport, CFDictionaryRef circle_to_peer_ids, CFErrorRef *error)
{
    return SOSTransportMessageKVSCleanupAfterPeerMessages((SOSTransportMessageKVSRef) transport, circle_to_peer_ids, error);
}
Commit	Line	Data
5c19dc3a A	1	#include <Security/SecureObjectSync/SOSTransport.h>
	2	#include <Security/SecureObjectSync/SOSTransportMessage.h>
	3	#include <Security/SecureObjectSync/SOSTransportMessageKVS.h>
	4	#include <Security/SecureObjectSync/SOSKVSKeys.h>
	5	#include <Security/SecureObjectSync/SOSAccountPriv.h>
d8f41ccd A	6	#include <utilities/SecCFWrappers.h>
	7	#include <SOSInternal.h>
	8	#include <AssertMacros.h>
	9	#include <SOSCloudKeychainClient.h>
	10
	11	struct __OpaqueSOSTransportMessageKVS {
	12	struct __OpaqueSOSTransportMessage m;
d8f41ccd A	13	CFMutableDictionaryRef pending_changes;
	14
	15	};
	16
	17	//
	18	// V-table implementation forward declarations
	19	//
	20	static bool sendToPeer(SOSTransportMessageRef transport, CFStringRef circleName, CFStringRef peerID, CFDataRef message, CFErrorRef *error);
6b200bc3 A	21	static bool syncWithPeers(SOSTransportMessageRef transport, CFSetRef peers, CFErrorRef *error);
6b200bc3 A	22	static bool sendMessages(SOSTransportMessageRef transport, CFDictionaryRef peersToMessage, CFErrorRef *error);
d8f41ccd A	23	static bool cleanupAfterPeer(SOSTransportMessageRef transport, CFDictionaryRef circle_to_peer_ids, CFErrorRef *error);
d8f41ccd A	24	static void destroy(SOSTransportMessageRef transport);
5c19dc3a A	25	static inline CFIndex getTransportType(SOSTransportMessageRef transport, CFErrorRef *error);
5c19dc3a A	26
d8f41ccd A	27	static CF_RETURNS_RETAINED
	28	CFDictionaryRef handleMessages(SOSTransportMessageRef transport, CFMutableDictionaryRef circle_peer_messages_table, CFErrorRef *error);
	29
	30	static bool flushChanges(SOSTransportMessageRef transport, CFErrorRef *error);
	31
d8f41ccd A	32	SOSTransportMessageKVSRef SOSTransportMessageKVSCreate(SOSAccountRef account, CFStringRef circleName, CFErrorRef *error){
	33	SOSTransportMessageKVSRef tkvs = (SOSTransportMessageKVSRef) SOSTransportMessageCreateForSubclass(sizeof(struct __OpaqueSOSTransportMessageKVS) - sizeof(CFRuntimeBase), account, circleName, error);
	34
	35	if (tkvs) {
	36	// Fill in vtable:
	37	tkvs->m.sendMessages = sendMessages;
	38	tkvs->m.syncWithPeers = syncWithPeers;
	39	tkvs->m.flushChanges = flushChanges;
	40	tkvs->m.cleanupAfterPeerMessages = cleanupAfterPeer;
	41	tkvs->m.destroy = destroy;
	42	tkvs->m.handleMessages = handleMessages;
5c19dc3a	43	tkvs->m.getTransportType = getTransportType;
d8f41ccd A	44	// Initialize ourselves
d8f41ccd A	45	tkvs->pending_changes = CFDictionaryCreateMutableForCFTypes(kCFAllocatorDefault);
d8f41ccd A	46	SOSRegisterTransportMessage((SOSTransportMessageRef)tkvs);
	47	}
	48
	49	return tkvs;
	50	}
5c19dc3a	51
d8f41ccd A	52	bool SOSTransportMessageKVSAppendKeyInterest(SOSTransportMessageKVSRef transport, CFMutableArrayRef alwaysKeys, CFMutableArrayRef afterFirstUnlockKeys, CFMutableArrayRef unlockedKeys, CFErrorRef *localError){
	53	SOSEngineRef engine = SOSTransportMessageGetEngine((SOSTransportMessageRef)transport);
	54	require_quiet(engine, fail);
fa7225c8	55
5c19dc3a A	56	CFArrayRef peerInfos = SOSAccountCopyPeersToListenTo(SOSTransportMessageGetAccount((SOSTransportMessageRef) transport), localError);
	57
	58	if(peerInfos){
	59	CFArrayForEach(peerInfos, ^(const void *value) {
	60	SOSPeerInfoRef peer = (SOSPeerInfoRef)value;
	61	CFStringRef peerID = SOSPeerInfoGetPeerID(peer);
fa7225c8 A	62	CFStringRef peerMessage = SOSMessageKeyCreateFromPeerToTransport((SOSTransportMessageRef)transport, peerID);
	63	if(peerMessage != NULL)
	64	CFArrayAppendValue(unlockedKeys, peerMessage);
d8f41ccd A	65	CFReleaseNull(peerMessage);
d8f41ccd A	66	});
5c19dc3a	67	CFReleaseNull(peerInfos);
d8f41ccd A	68	}
	69	return true;
	70	fail:
	71	return false;
	72	}
	73	static void destroy(SOSTransportMessageRef transport){
	74	SOSTransportMessageKVSRef tkvs = (SOSTransportMessageKVSRef)transport;
d8f41ccd A	75	CFReleaseNull(tkvs->pending_changes);
	76	SOSUnregisterTransportMessage((SOSTransportMessageRef)tkvs);
	77
	78	}
	79
5c19dc3a A	80	static inline CFIndex getTransportType(SOSTransportMessageRef transport, CFErrorRef *error){
	81	return kKVS;
	82	}
	83
d8f41ccd	84	static bool SOSTransportMessageKVSUpdateKVS(SOSTransportMessageKVSRef transport, CFDictionaryRef changes, CFErrorRef *error){
5c19dc3a	85
6b200bc3 A	86	CloudKeychainReplyBlock log_error = ^(CFDictionaryRef returnedValues __unused, CFErrorRef block_error) {
	87	if (block_error) {
	88	secerror("Error putting: %@", block_error);
d8f41ccd A	89	}
	90	};
	91
	92	SOSCloudKeychainPutObjectsInCloud(changes, dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0), log_error);
	93	return true;
	94	}
	95
	96	static bool SOSTransportMessageKVSSendPendingChanges(SOSTransportMessageKVSRef transport, CFErrorRef *error) {
	97	CFErrorRef changeError = NULL;
	98
	99	if (transport->pending_changes == NULL \|\| CFDictionaryGetCount(transport->pending_changes) == 0) {
	100	CFReleaseNull(transport->pending_changes);
	101	return true;
	102	}
5c19dc3a A	103	SOSAccountRef account = SOSTransportMessageGetAccount((SOSTransportMessageRef)transport);
	104	CFTypeRef dsid = SOSAccountGetValue(account, kSOSDSIDKey, error);
	105
	106	if(dsid == NULL)
	107	dsid = kCFNull;
	108
	109	CFDictionaryAddValue(transport->pending_changes, kSOSKVSRequiredKey, dsid);
d8f41ccd A	110
	111	bool success = SOSTransportMessageKVSUpdateKVS(transport, transport->pending_changes, &changeError);
	112	if (success) {
	113	CFDictionaryRemoveAllValues(transport->pending_changes);
	114	} else {
	115	SOSCreateErrorWithFormat(kSOSErrorSendFailure, changeError, error, NULL,
	116	CFSTR("Send changes block failed [%@]"), transport->pending_changes);
	117	}
	118
	119	return success;
	120	}
	121
	122	static void SOSTransportMessageKVSAddToPendingChanges(SOSTransportMessageKVSRef transport, CFStringRef message_key, CFDataRef message_data){
	123	if (transport->pending_changes == NULL) {
	124	transport->pending_changes = CFDictionaryCreateMutableForCFTypes(kCFAllocatorDefault);
	125	}
	126	if (message_data == NULL) {
	127	CFDictionarySetValue(transport->pending_changes, message_key, kCFNull);
	128	} else {
	129	CFDictionarySetValue(transport->pending_changes, message_key, message_data);
	130	}
	131	}
	132
	133	static bool SOSTransportMessageKVSCleanupAfterPeerMessages(SOSTransportMessageKVSRef transport, CFDictionaryRef circle_to_peer_ids, CFErrorRef *error)
	134	{
	135	CFArrayRef enginePeers = SOSEngineGetPeerIDs(SOSTransportMessageGetEngine((SOSTransportMessageRef)transport));
5c19dc3a A	136	__block SOSAccountRef account = SOSTransportMessageGetAccount((SOSTransportMessageRef)transport);
5c19dc3a A	137
d8f41ccd A	138	CFDictionaryForEach(circle_to_peer_ids, ^(const void key, const void value) {
	139	if (isString(key) && isArray(value)) {
	140	CFStringRef circle_name = (CFStringRef) key;
	141	CFArrayRef peers_to_cleanup_after = (CFArrayRef) value;
	142
	143	CFArrayForEach(peers_to_cleanup_after, ^(const void *value) {
	144	if (isString(value)) {
	145	CFStringRef cleanup_id = (CFStringRef) value;
	146	// TODO: Since the enginePeers list is not authorative (the Account is) this could inadvertently clean up active peers or leave behind stale peers
	147	if (enginePeers) CFArrayForEach(enginePeers, ^(const void *value) {
	148	if (isString(value)) {
	149	CFStringRef in_circle_id = (CFStringRef) value;
	150
	151	CFStringRef kvsKey = SOSMessageKeyCreateWithCircleNameAndPeerNames(circle_name, cleanup_id, in_circle_id);
	152	SOSTransportMessageKVSAddToPendingChanges(transport, kvsKey, NULL);
	153	CFReleaseSafe(kvsKey);
	154
5c19dc3a A	155	CFStringRef lastCirclePushedKey = SOSLastCirclePushedKeyCreateWithCircleNameAndPeerID(circle_name, cleanup_id);
	156	SOSTransportMessageKVSAddToPendingChanges(transport, lastCirclePushedKey, NULL);
	157	CFReleaseSafe(lastCirclePushedKey);
	158
	159	CFStringRef lastKeyParameterPushedKey = SOSLastKeyParametersPushedKeyCreateWithPeerID(cleanup_id);
	160	SOSTransportMessageKVSAddToPendingChanges(transport, lastKeyParameterPushedKey, NULL);
	161	CFReleaseSafe(lastKeyParameterPushedKey);
	162
	163	CFStringRef lastCirclePushedWithAccountGestaltKey = SOSLastCirclePushedKeyCreateWithAccountGestalt(account);
	164	SOSTransportMessageKVSAddToPendingChanges(transport, lastCirclePushedKey, NULL);
	165	CFReleaseSafe(lastCirclePushedWithAccountGestaltKey);
	166
	167	CFStringRef lastKeyParameterWithAccountGestaltKey = SOSLastKeyParametersPushedKeyCreateWithAccountGestalt(account);
	168	SOSTransportMessageKVSAddToPendingChanges(transport, lastCirclePushedKey, NULL);
	169	CFReleaseSafe(lastKeyParameterWithAccountGestaltKey);
	170
d8f41ccd A	171	kvsKey = SOSMessageKeyCreateWithCircleNameAndPeerNames(circle_name, in_circle_id, cleanup_id);
	172	SOSTransportMessageKVSAddToPendingChanges(transport, kvsKey, NULL);
	173	CFReleaseSafe(kvsKey);
	174	}
	175	});
	176
	177	}
	178	});
	179	}
	180	});
	181
	182	return SOSTransportMessageFlushChanges((SOSTransportMessageRef)transport, error);
	183	}
	184
	185	static CF_RETURNS_RETAINED
	186	CFDictionaryRef handleMessages(SOSTransportMessageRef transport, CFMutableDictionaryRef circle_peer_messages_table, CFErrorRef *error) {
5c19dc3a	187
d8f41ccd	188	CFMutableDictionaryRef handled = CFDictionaryCreateMutableForCFTypes(kCFAllocatorDefault);
5c19dc3a	189	CFDictionaryRef peerToMessage = CFDictionaryGetValue(circle_peer_messages_table, transport->circleName);
d8f41ccd	190	CFMutableArrayRef handled_peers = CFArrayCreateMutableForCFTypes(kCFAllocatorDefault);
d8f41ccd A	191
	192	if(peerToMessage){
	193	CFDictionaryForEach(peerToMessage, ^(const void key, const void value) {
fa7225c8 A	194	CFStringRef peer_id = asString(key, NULL);
fa7225c8 A	195	CFDataRef peer_message = asData(value, NULL);
d8f41ccd A	196	CFErrorRef localError = NULL;
d8f41ccd A	197
fa7225c8	198	if (peer_id && peer_message && SOSTransportMessageHandlePeerMessage(transport, peer_id, peer_message, &localError)) {
d8f41ccd A	199	CFArrayAppendValue(handled_peers, key);
d8f41ccd A	200	} else {
5c19dc3a	201	secnotice("transport", "%@ KVSTransport handle message failed: %@", peer_id, localError);
d8f41ccd A	202	}
	203	CFReleaseNull(localError);
	204	});
	205	}
5c19dc3a	206	CFDictionaryAddValue(handled, transport->circleName, handled_peers);
d8f41ccd A	207	CFReleaseNull(handled_peers);
	208
	209	return handled;
	210	}
	211
	212
	213	static bool sendToPeer(SOSTransportMessageRef transport, CFStringRef circleName, CFStringRef peerID, CFDataRef message, CFErrorRef *error) {
	214	SOSTransportMessageKVSRef kvsTransport = (SOSTransportMessageKVSRef) transport;
	215	bool result = true;
5c19dc3a A	216	SOSAccountRef account = SOSTransportMessageGetAccount((SOSTransportMessageRef)transport);
	217	CFTypeRef dsid = SOSAccountGetValue(account, kSOSDSIDKey, error);
	218
	219	if(dsid == NULL)
	220	dsid = kCFNull;
6b200bc3	221
fa7225c8	222	CFStringRef message_to_peer_key = SOSMessageKeyCreateFromTransportToPeer((SOSTransportMessageRef)kvsTransport, peerID);
6b200bc3 A	223
	224	CFTypeRef messageToSend = message != NULL ? (CFTypeRef) message : (CFTypeRef) kCFNull;
	225	CFDictionaryRef a_message_to_a_peer = CFDictionaryCreateForCFTypes(NULL,
	226	message_to_peer_key, messageToSend,
	227	kSOSKVSRequiredKey, dsid,
	228	NULL);
d8f41ccd A	229
	230	if (!SOSTransportMessageKVSUpdateKVS(kvsTransport, a_message_to_a_peer, error)) {
	231	secerror("Sync with peers failed to send to %@ [%@], %@", peerID, a_message_to_a_peer, *error);
	232	result = false;
	233	}
	234	CFReleaseNull(a_message_to_a_peer);
	235	CFReleaseNull(message_to_peer_key);
	236
	237	return result;
	238	}
	239
6b200bc3	240	static bool syncWithPeers(SOSTransportMessageRef transport, CFSetRef peers, CFErrorRef *error) {
d8f41ccd	241	// Each entry is keyed by circle name and contains a list of peerIDs
d8f41ccd	242	__block bool result = true;
6b200bc3 A	243
	244	CFSetForEach(peers, ^(const void *value) {
	245	CFStringRef peerID = asString(value, NULL);
	246	result &= SOSTransportMessageSendMessageIfNeeded(transport, transport->circleName, peerID, error);
d8f41ccd	247	});
6b200bc3	248
d8f41ccd A	249	return result;
	250	}
	251
6b200bc3	252	static bool sendMessages(SOSTransportMessageRef transport, CFDictionaryRef peersToMessage, CFErrorRef *error) {
d8f41ccd A	253	__block bool result = true;
d8f41ccd A	254
6b200bc3 A	255	CFStringRef circleName = transport->circleName;
	256	CFDictionaryForEach(peersToMessage, ^(const void key, const void value) {
	257	CFStringRef peerID = asString(key, NULL);
	258	CFDataRef message = asData(value,NULL);
	259	if (peerID && message) {
	260	bool rx = sendToPeer(transport, circleName, peerID, message, error);
	261	result &= rx;
d8f41ccd A	262	}
d8f41ccd A	263	});
6b200bc3	264
d8f41ccd A	265	return true;
	266	}
	267
	268	static bool flushChanges(SOSTransportMessageRef transport, CFErrorRef *error)
	269	{
	270	return SOSTransportMessageKVSSendPendingChanges((SOSTransportMessageKVSRef) transport, error);
	271	}
	272
	273	static bool cleanupAfterPeer(SOSTransportMessageRef transport, CFDictionaryRef circle_to_peer_ids, CFErrorRef *error)
	274	{
	275	return SOSTransportMessageKVSCleanupAfterPeerMessages((SOSTransportMessageKVSRef) transport, circle_to_peer_ids, error);
	276	}
	277