# -*- coding: utf-8 -*-
# __init__.py
# Copyright (C) 2017 LEAP
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see .
"""
Clientside BlobBackend Storage.
"""
from urlparse import urljoin
import os
import json
import base64
import struct
from collections import defaultdict
from io import BytesIO
from twisted.logger import Logger
from twisted.internet import defer
import treq
from leap.soledad.common.errors import SoledadError
from leap.common.files import mkdir_p
from leap.soledad.client._crypto import DocInfo
from leap.soledad.client._crypto import InvalidBlob
from leap.soledad.client._crypto import BlobEncryptor
from leap.soledad.client._crypto import BlobDecryptor
from leap.soledad.client._crypto import EncryptionSchemeNotImplementedException
from leap.soledad.client._crypto import get_unarmored_ciphertext_size
from leap.soledad.client._http import HTTPClient
from leap.soledad.client._pipes import TruncatedTailPipe
from leap.soledad.client._pipes import PreamblePipe
from .sql import SyncStatus
from .sql import Priority
from .sql import SQLiteBlobBackend
from .sync import BlobsSynchronizer
from .upstream_producer import BlobsUpstreamProducer
from .errors import (
BlobAlreadyExistsError, MaximumRetriesError,
RetriableTransferError, BlobNotFoundError, InvalidFlagsError)
logger = Logger()
FIXED_REV = 'ImmutableRevision' # Blob content is immutable
SEPARATOR = ' '
def check_http_status(code, blob_id=None, flags=None):
if code == 404:
raise BlobNotFoundError(blob_id)
if code == 409:
raise BlobAlreadyExistsError(blob_id)
elif code == 406:
raise InvalidFlagsError((blob_id, flags))
elif code != 200:
raise SoledadError("Server Error: %s" % code)
class DecrypterBuffer(object):
def __init__(self, blob_id, secret, tag):
self.doc_info = DocInfo(blob_id, FIXED_REV)
self.secret = secret
self.tag = tag
self.preamble_pipe = PreamblePipe(self._make_decryptor)
self.decrypter = None
def _make_decryptor(self, preamble):
try:
self.decrypter = BlobDecryptor(
self.doc_info, preamble,
secret=self.secret,
armor=False,
start_stream=False,
tag=self.tag)
return TruncatedTailPipe(self.decrypter, tail_size=len(self.tag))
except EncryptionSchemeNotImplementedException:
# If we do not support the provided encryption scheme, then that's
# something for the application using soledad to handle. This is
# the case on asymmetrically encrypted documents on IncomingBox.
self.raw_data = BytesIO()
return self.raw_data
def write(self, data):
self.preamble_pipe.write(data)
def close(self):
if self.decrypter:
real_size = self.decrypter.decrypted_content_size
return self.decrypter.endStream(), real_size
elif hasattr(self, 'raw_data'):
# blob was not symmetrically encrypted, see _make_decryptor() above
return self.raw_data, self.raw_data.tell()
else:
# the only case that can lead here is if the preamble was not
# completelly consumed by the pipe.
msg = "Incomplete Blob: %s" % self.doc_info.doc_id
raise RetriableTransferError(msg)
class StreamDecrypterBuffer(object):
size_pack = struct.Struct('= max_retries:
failed_download = SyncStatus.FAILED_DOWNLOAD
yield self.local.update_sync_status(
blob_id, failed_download, namespace=namespace)
raise MaximumRetriesError(e)
raise RetriableTransferError(e)
if not result:
defer.returnValue(None)
blob, size = result
if blob:
logger.info("Got decrypted blob of type: %s" % type(blob))
blob.seek(0)
yield self.local.put(blob_id, blob, size=size, namespace=namespace)
yield self.local.update_sync_status(blob_id, SyncStatus.SYNCED,
namespace=namespace)
local_blob = yield self.local.get(blob_id, namespace=namespace)
defer.returnValue(local_blob)
else:
# XXX we shouldn't get here, but we will...
# lots of ugly error handling possible:
# 1. retry, might be network error
# 2. try later, maybe didn't finished streaming
# 3.. resignation, might be error while verifying
logger.error('sorry, dunno what happened')
@defer.inlineCallbacks
def _encrypt_and_upload(self, blob_id, fd, namespace=''):
# TODO ------------------------------------------
# this is wrong, is doing 2 stages.
# the crypto producer can be passed to
# the uploader and react as data is written.
# try to rewrite as a tube: pass the fd to aes and let aes writer
# produce data to the treq request fd.
# ------------------------------------------------
logger.info("Staring upload of blob: %s" % blob_id)
doc_info = DocInfo(blob_id, FIXED_REV)
uri = urljoin(self.remote, self.user + "/" + blob_id)
crypter = BlobEncryptor(doc_info, fd, secret=self.secret,
armor=False)
fd = yield crypter.encrypt()
params = {'namespace': namespace} if namespace else None
response = yield self._client.put(uri, data=fd, params=params)
check_http_status(response.code, blob_id)
logger.info("Finished upload: %s" % (blob_id,))
@defer.inlineCallbacks
def _downstream(self, blobs_id_list, namespace=''):
uri = urljoin(self.remote_stream, self.user)
params = {'namespace': namespace} if namespace else {}
params['direction'] = 'download'
data = BytesIO(json.dumps(blobs_id_list))
response = yield self._client.post(uri, params=params, data=data)
deferreds = []
def done_cb(blob_id, blobfd, size):
d = self.local.put(blob_id, blobfd, size=size, namespace=namespace)
deferreds.append(d)
buf = StreamDecrypterBuffer(self.secret, blobs_id_list, done_cb)
yield treq.collect(response, buf.write)
yield defer.gatherResults(deferreds, consumeErrors=True)
buf.close()
@defer.inlineCallbacks
def _upstream(self, blobs_id_list, namespace=''):
local, secret = self.local, self.secret
uri = urljoin(self.remote_stream, self.user)
sizes = yield self.local.get_size_list(blobs_id_list, namespace)
convert = get_unarmored_ciphertext_size
sizes = map(lambda (blob_id, size): (blob_id, convert(size)), sizes)
producer = BlobsUpstreamProducer(local, sizes, namespace, secret)
params = {'namespace': namespace} if namespace else {}
params['direction'] = 'upload'
response = yield self._client.post(uri, data=producer, params=params)
check_http_status(response.code, 'stream')
logger.info("Finished stream up: %s" % (blobs_id_list,))
@defer.inlineCallbacks
def _download_and_decrypt(self, blob_id, namespace=''):
logger.info("Staring download of blob: %s" % blob_id)
# TODO this needs to be connected in a tube
uri = urljoin(self.remote, self.user + '/' + blob_id)
params = {'namespace': namespace} if namespace else None
response = yield self._client.get(uri, params=params)
check_http_status(response.code, blob_id=blob_id)
if not response.headers.hasHeader('Tag'):
msg = "Server didn't send a tag header for: %s" % blob_id
logger.error(msg)
raise SoledadError(msg)
tag = response.headers.getRawHeaders('Tag')[0]
tag = base64.urlsafe_b64decode(tag)
buf = DecrypterBuffer(blob_id, self.secret, tag)
# incrementally collect the body of the response
yield treq.collect(response, buf.write)
fd, size = buf.close()
logger.info("Finished download: (%s, %d)" % (blob_id, size))
defer.returnValue((fd, size))
def delete(self, blob_id, namespace=''):
"""
Delete a blob from local and remote storages.
:param blob_id:
Unique identifier of a blob.
:type blob_id: str
:param namespace:
Optional parameter to restrict operation to a given namespace.
:type namespace: str
:return: A deferred that fires when the operation finishes.
:rtype: twisted.internet.defer.Deferred
"""
return self.semaphore.run(self._delete, blob_id, namespace)
@defer.inlineCallbacks
def _delete(self, blob_id, namespace):
logger.info("Marking blobs as PENDING_DELETE: %s" % blob_id)
yield self.local.update_sync_status(
blob_id, SyncStatus.PENDING_DELETE, namespace=namespace)
logger.info("Staring deletion of blob: %s" % blob_id)
yield self._delete_from_remote(blob_id, namespace=namespace)
if (yield self.local.exists(blob_id, namespace=namespace)):
yield self.local.delete(blob_id, namespace=namespace)
yield self.local.update_sync_status(
blob_id, SyncStatus.SYNCED, namespace=namespace)
@defer.inlineCallbacks
def _delete_from_remote(self, blob_id, namespace=''):
# TODO this needs to be connected in a tube
uri = urljoin(self.remote, self.user + '/' + blob_id)
params = {'namespace': namespace} if namespace else None
response = yield self._client.delete(uri, params=params)
check_http_status(response.code, blob_id=blob_id)
defer.returnValue(response)
# TODO: evaluate if the following get/set priority methods are needed in
# the public interface of then blob manager, and remove if not.
def _set_priority(self, blob_id, priority, namespace=''):
"""
Set the transfer priority for a certain blob.
:param blob_id: Unique identifier of a blob.
:type blob_id: str
:param priority: The numerical priority to be set.
:type priority: int
:param namespace: Optional parameter to restrict operation to a given
namespace.
:type namespace: str
:return: A deferred that fires after the priority has been set.
:rtype: twisted.internet.defer.Deferred
"""
prio = _parse_priority(priority)
d = self.local.update_priority(blob_id, prio, namespace=namespace)
return d
def _get_priority(self, blob_id, namespace=''):
"""
Get the transfer priority for a certain blob.
:param blob_id: Unique identifier of a blob.
:type blob_id: str
:param namespace: Optional parameter to restrict operation to a given
namespace.
:type namespace: str
:return: A deferred that fires with the current transfer priority of
the blob.
:rtype: twisted.internet.defer.Deferred
"""
d = self.local.get_priority(blob_id, namespace=namespace)
return d
def _parse_priority(prio):
if isinstance(prio, int):
if Priority.LOW <= prio <= Priority.URGENT:
return prio
else:
raise ValueError()
elif isinstance(prio, str):
if prio == 'low':
return Priority.LOW
elif prio == 'medium':
return Priority.MEDIUM
elif prio == 'high':
return Priority.HIGH
elif prio == 'urgent':
return Priority.URGENT
raise ValueError()
raise ValueError()