1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
|
try:
import simplejson as json
except ImportError:
import json # noqa
from u1db import Document
from u1db.remote.http_target import HTTPSyncTarget
import base64
class LeapDocument(Document):
"""
LEAP Documents are standard u1db documents with cabability of returning an
encrypted version of the document json string as well as setting document
content based on an encrypted version of json string.
"""
def __init__(self, doc_id=None, rev=None, json='{}', has_conflicts=False,
encrypted_json=None):
super(Document, self).__init__(doc_id, rev, json, has_conflicts)
if encrypted_json:
self.set_encrypted_json(encrypted_json)
def get_encrypted_json(self):
"""
Returns document's json serialization encrypted with user's public key.
"""
# TODO: replace for openpgp encryption with users's pub key.
return base64.b64encode(self.get_json())
def set_encrypted_json(self):
"""
Set document's content based on encrypted version of json string.
"""
# TODO:
# - replace for openpgp decryption using user's priv key.
# - raise error if unsuccessful.
return self.set_json(base64.b64decode(self.get_json()))
class LeapSyncTarget(HTTPSyncTarget):
def _parse_sync_stream(self, data, return_doc_cb, ensure_callback=None):
parts = data.splitlines() # one at a time
if not parts or parts[0] != '[':
raise BrokenSyncStream
data = parts[1:-1]
comma = False
if data:
line, comma = utils.check_and_strip_comma(data[0])
res = json.loads(line)
if ensure_callback and 'replica_uid' in res:
ensure_callback(res['replica_uid'])
for entry in data[1:]:
if not comma: # missing in between comma
raise BrokenSyncStream
line, comma = utils.check_and_strip_comma(entry)
entry = json.loads(line)
doc = LeapDocument(entry['id'], entry['rev'],
encrypted_json=entry['content'])
return_doc_cb(doc, entry['gen'], entry['trans_id'])
if parts[-1] != ']':
try:
partdic = json.loads(parts[-1])
except ValueError:
pass
else:
if isinstance(partdic, dict):
self._error(partdic)
raise BrokenSyncStream
if not data or comma: # no entries or bad extra comma
raise BrokenSyncStream
return res
def sync_exchange(self, docs_by_generations, source_replica_uid,
last_known_generation, last_known_trans_id,
return_doc_cb, ensure_callback=None):
self._ensure_connection()
if self._trace_hook: # for tests
self._trace_hook('sync_exchange')
url = '%s/sync-from/%s' % (self._url.path, source_replica_uid)
self._conn.putrequest('POST', url)
self._conn.putheader('content-type', 'application/x-u1db-sync-stream')
for header_name, header_value in self._sign_request('POST', url, {}):
self._conn.putheader(header_name, header_value)
entries = ['[']
size = 1
def prepare(**dic):
entry = comma + '\r\n' + json.dumps(dic)
entries.append(entry)
return len(entry)
comma = ''
size += prepare(
last_known_generation=last_known_generation,
last_known_trans_id=last_known_trans_id,
ensure=ensure_callback is not None)
comma = ','
for doc, gen, trans_id in docs_by_generations:
size += prepare(id=doc.doc_id, rev=doc.rev,
content=doc.get_encrypted_json(),
gen=gen, trans_id=trans_id)
entries.append('\r\n]')
size += len(entries[-1])
self._conn.putheader('content-length', str(size))
self._conn.endheaders()
for entry in entries:
self._conn.send(entry)
entries = None
data, _ = self._response()
res = self._parse_sync_stream(data, return_doc_cb, ensure_callback)
data = None
return res['new_generation'], res['new_transaction_id']
|
