diff options
Diffstat (limited to 'common/src/leap/soledad/common/l2db/remote/http_app.py')
| -rw-r--r-- | common/src/leap/soledad/common/l2db/remote/http_app.py | 660 |
1 files changed, 0 insertions, 660 deletions
diff --git a/common/src/leap/soledad/common/l2db/remote/http_app.py b/common/src/leap/soledad/common/l2db/remote/http_app.py deleted file mode 100644 index a4eddb36..00000000 --- a/common/src/leap/soledad/common/l2db/remote/http_app.py +++ /dev/null @@ -1,660 +0,0 @@ -# Copyright 2011-2012 Canonical Ltd. -# Copyright 2016 LEAP Encryption Access Project -# -# This file is part of u1db. -# -# u1db is free software: you can redistribute it and/or modify -# it under the terms of the GNU Lesser General Public License version 3 -# as published by the Free Software Foundation. -# -# u1db 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 Lesser General Public License for more details. -# -# You should have received a copy of the GNU Lesser General Public License -# along with u1db. If not, see <http://www.gnu.org/licenses/>. - -""" -HTTP Application exposing U1DB. -""" -# TODO -- deprecate, use twisted/txaio. - -import functools -import six.moves.http_client as httplib -import inspect -import json -import sys -import six.moves.urllib.parse as urlparse - -import routes.mapper - -from leap.soledad.common.l2db import ( - __version__ as _u1db_version, - DBNAME_CONSTRAINTS, Document, - errors, sync) -from leap.soledad.common.l2db.remote import http_errors, utils - - -def parse_bool(expression): - """Parse boolean querystring parameter.""" - if expression == 'true': - return True - return False - - -def parse_list(expression): - if not expression: - return [] - return [t.strip() for t in expression.split(',')] - - -def none_or_str(expression): - if expression is None: - return None - return str(expression) - - -class BadRequest(Exception): - """Bad request.""" - - -class _FencedReader(object): - """Read and get lines from a file but not past a given length.""" - - MAXCHUNK = 8192 - - def __init__(self, rfile, total, max_entry_size): - self.rfile = rfile - self.remaining = total - self.max_entry_size = max_entry_size - self._kept = None - - def read_chunk(self, atmost): - if self._kept is not None: - # ignore atmost, kept data should be a subchunk anyway - kept, self._kept = self._kept, None - return kept - if self.remaining == 0: - return '' - data = self.rfile.read(min(self.remaining, atmost)) - self.remaining -= len(data) - return data - - def getline(self): - line_parts = [] - size = 0 - while True: - chunk = self.read_chunk(self.MAXCHUNK) - if chunk == '': - break - nl = chunk.find("\n") - if nl != -1: - size += nl + 1 - if size > self.max_entry_size: - raise BadRequest - line_parts.append(chunk[:nl + 1]) - rest = chunk[nl + 1:] - self._kept = rest or None - break - else: - size += len(chunk) - if size > self.max_entry_size: - raise BadRequest - line_parts.append(chunk) - return ''.join(line_parts) - - -def http_method(**control): - """Decoration for handling of query arguments and content for a HTTP - method. - - args and content here are the query arguments and body of the incoming - HTTP requests. - - Match query arguments to python method arguments: - w = http_method()(f) - w(self, args, content) => args["content"]=content; - f(self, **args) - - JSON deserialize content to arguments: - w = http_method(content_as_args=True,...)(f) - w(self, args, content) => args.update(json.loads(content)); - f(self, **args) - - Support conversions (e.g int): - w = http_method(Arg=Conv,...)(f) - w(self, args, content) => args["Arg"]=Conv(args["Arg"]); - f(self, **args) - - Enforce no use of query arguments: - w = http_method(no_query=True,...)(f) - w(self, args, content) raises BadRequest if args is not empty - - Argument mismatches, deserialisation failures produce BadRequest. - """ - content_as_args = control.pop('content_as_args', False) - no_query = control.pop('no_query', False) - conversions = control.items() - - def wrap(f): - argspec = inspect.getargspec(f) - assert argspec.args[0] == "self" - nargs = len(argspec.args) - ndefaults = len(argspec.defaults or ()) - required_args = set(argspec.args[1:nargs - ndefaults]) - all_args = set(argspec.args) - - @functools.wraps(f) - def wrapper(self, args, content): - if no_query and args: - raise BadRequest() - if content is not None: - if content_as_args: - try: - args.update(json.loads(content)) - except ValueError: - raise BadRequest() - else: - args["content"] = content - if not (required_args <= set(args) <= all_args): - raise BadRequest("Missing required arguments.") - for name, conv in conversions: - if name not in args: - continue - try: - args[name] = conv(args[name]) - except ValueError: - raise BadRequest() - return f(self, **args) - - return wrapper - - return wrap - - -class URLToResource(object): - """Mappings from URLs to resources.""" - - def __init__(self): - self._map = routes.mapper.Mapper(controller_scan=None) - - def register(self, resource_cls): - # register - self._map.connect(None, resource_cls.url_pattern, - resource_cls=resource_cls, - requirements={"dbname": DBNAME_CONSTRAINTS}) - self._map.create_regs() - return resource_cls - - def match(self, path): - params = self._map.match(path) - if params is None: - return None, None - resource_cls = params.pop('resource_cls') - return resource_cls, params - - -url_to_resource = URLToResource() - - -@url_to_resource.register -class GlobalResource(object): - """Global (root) resource.""" - - url_pattern = "/" - - def __init__(self, state, responder): - self.state = state - self.responder = responder - - @http_method() - def get(self): - info = self.state.global_info() - info['version'] = _u1db_version - self.responder.send_response_json(**info) - - -@url_to_resource.register -class DatabaseResource(object): - """Database resource.""" - - url_pattern = "/{dbname}" - - def __init__(self, dbname, state, responder): - self.dbname = dbname - self.state = state - self.responder = responder - - @http_method() - def get(self): - self.state.check_database(self.dbname) - self.responder.send_response_json(200) - - @http_method(content_as_args=True) - def put(self): - self.state.ensure_database(self.dbname) - self.responder.send_response_json(200, ok=True) - - @http_method() - def delete(self): - self.state.delete_database(self.dbname) - self.responder.send_response_json(200, ok=True) - - -@url_to_resource.register -class DocsResource(object): - """Documents resource.""" - - url_pattern = "/{dbname}/docs" - - def __init__(self, dbname, state, responder): - self.responder = responder - self.db = state.open_database(dbname) - - @http_method(doc_ids=parse_list, check_for_conflicts=parse_bool, - include_deleted=parse_bool) - def get(self, doc_ids=None, check_for_conflicts=True, - include_deleted=False): - if doc_ids is None: - raise errors.MissingDocIds - docs = self.db.get_docs(doc_ids, include_deleted=include_deleted) - self.responder.content_type = 'application/json' - self.responder.start_response(200) - self.responder.start_stream(), - for doc in docs: - entry = dict( - doc_id=doc.doc_id, doc_rev=doc.rev, content=doc.get_json(), - has_conflicts=doc.has_conflicts) - self.responder.stream_entry(entry) - self.responder.end_stream() - self.responder.finish_response() - - -@url_to_resource.register -class AllDocsResource(object): - """All Documents resource.""" - - url_pattern = "/{dbname}/all-docs" - - def __init__(self, dbname, state, responder): - self.responder = responder - self.db = state.open_database(dbname) - - @http_method(include_deleted=parse_bool) - def get(self, include_deleted=False): - gen, docs = self.db.get_all_docs(include_deleted=include_deleted) - self.responder.content_type = 'application/json' - # returning a x-u1db-generation header is optional - # HTTPDatabase will fallback to return -1 if it's missing - self.responder.start_response(200, - headers={'x-u1db-generation': str(gen)}) - self.responder.start_stream(), - for doc in docs: - entry = dict( - doc_id=doc.doc_id, doc_rev=doc.rev, content=doc.get_json(), - has_conflicts=doc.has_conflicts) - self.responder.stream_entry(entry) - self.responder.end_stream() - self.responder.finish_response() - - -@url_to_resource.register -class DocResource(object): - """Document resource.""" - - url_pattern = "/{dbname}/doc/{id:.*}" - - def __init__(self, dbname, id, state, responder): - self.id = id - self.responder = responder - self.db = state.open_database(dbname) - - @http_method(old_rev=str) - def put(self, content, old_rev=None): - doc = Document(self.id, old_rev, content) - doc_rev = self.db.put_doc(doc) - if old_rev is None: - status = 201 # created - else: - status = 200 - self.responder.send_response_json(status, rev=doc_rev) - - @http_method(old_rev=str) - def delete(self, old_rev=None): - doc = Document(self.id, old_rev, None) - self.db.delete_doc(doc) - self.responder.send_response_json(200, rev=doc.rev) - - @http_method(include_deleted=parse_bool) - def get(self, include_deleted=False): - doc = self.db.get_doc(self.id, include_deleted=include_deleted) - if doc is None: - wire_descr = errors.DocumentDoesNotExist.wire_description - self.responder.send_response_json( - http_errors.wire_description_to_status[wire_descr], - error=wire_descr, - headers={ - 'x-u1db-rev': '', - 'x-u1db-has-conflicts': 'false' - }) - return - headers = { - 'x-u1db-rev': doc.rev, - 'x-u1db-has-conflicts': json.dumps(doc.has_conflicts) - } - if doc.is_tombstone(): - self.responder.send_response_json( - http_errors.wire_description_to_status[ - errors.DOCUMENT_DELETED], - error=errors.DOCUMENT_DELETED, - headers=headers) - else: - self.responder.send_response_content( - doc.get_json(), headers=headers) - - -@url_to_resource.register -class SyncResource(object): - """Sync endpoint resource.""" - - # maximum allowed request body size - max_request_size = 15 * 1024 * 1024 # 15Mb - # maximum allowed entry/line size in request body - max_entry_size = 10 * 1024 * 1024 # 10Mb - - url_pattern = "/{dbname}/sync-from/{source_replica_uid}" - - # pluggable - sync_exchange_class = sync.SyncExchange - - def __init__(self, dbname, source_replica_uid, state, responder): - self.source_replica_uid = source_replica_uid - self.responder = responder - self.state = state - self.dbname = dbname - self.replica_uid = None - - def get_target(self): - return self.state.open_database(self.dbname).get_sync_target() - - @http_method() - def get(self): - result = self.get_target().get_sync_info(self.source_replica_uid) - self.responder.send_response_json( - target_replica_uid=result[0], target_replica_generation=result[1], - target_replica_transaction_id=result[2], - source_replica_uid=self.source_replica_uid, - source_replica_generation=result[3], - source_transaction_id=result[4]) - - @http_method(generation=int, - content_as_args=True, no_query=True) - def put(self, generation, transaction_id): - self.get_target().record_sync_info(self.source_replica_uid, - generation, - transaction_id) - self.responder.send_response_json(ok=True) - - # Implements the same logic as LocalSyncTarget.sync_exchange - - @http_method(last_known_generation=int, last_known_trans_id=none_or_str, - content_as_args=True) - def post_args(self, last_known_generation, last_known_trans_id=None, - ensure=False): - if ensure: - db, self.replica_uid = self.state.ensure_database(self.dbname) - else: - db = self.state.open_database(self.dbname) - db.validate_gen_and_trans_id( - last_known_generation, last_known_trans_id) - self.sync_exch = self.sync_exchange_class( - db, self.source_replica_uid, last_known_generation) - - @http_method(content_as_args=True) - def post_stream_entry(self, id, rev, content, gen, trans_id): - doc = Document(id, rev, content) - self.sync_exch.insert_doc_from_source(doc, gen, trans_id) - - def post_end(self): - - def send_doc(doc, gen, trans_id): - entry = dict(id=doc.doc_id, rev=doc.rev, content=doc.get_json(), - gen=gen, trans_id=trans_id) - self.responder.stream_entry(entry) - - new_gen = self.sync_exch.find_changes_to_return() - self.responder.content_type = 'application/x-u1db-sync-stream' - self.responder.start_response(200) - self.responder.start_stream(), - header = {"new_generation": new_gen, - "new_transaction_id": self.sync_exch.new_trans_id} - if self.replica_uid is not None: - header['replica_uid'] = self.replica_uid - self.responder.stream_entry(header) - self.sync_exch.return_docs(send_doc) - self.responder.end_stream() - self.responder.finish_response() - - -class HTTPResponder(object): - """Encode responses from the server back to the client.""" - - # a multi document response will put args and documents - # each on one line of the response body - - def __init__(self, start_response): - self._started = False - self._stream_state = -1 - self._no_initial_obj = True - self.sent_response = False - self._start_response = start_response - self._write = None - self.content_type = 'application/json' - self.content = [] - - def start_response(self, status, obj_dic=None, headers={}): - """start sending response with optional first json object.""" - if self._started: - return - self._started = True - status_text = httplib.responses[status] - self._write = self._start_response( - '%d %s' % (status, status_text), - [('content-type', self.content_type), - ('cache-control', 'no-cache')] + - headers.items()) - # xxx version in headers - if obj_dic is not None: - self._no_initial_obj = False - self._write(json.dumps(obj_dic) + "\r\n") - - def finish_response(self): - """finish sending response.""" - self.sent_response = True - - def send_response_json(self, status=200, headers={}, **kwargs): - """send and finish response with json object body from keyword args.""" - content = json.dumps(kwargs) + "\r\n" - self.send_response_content(content, headers=headers, status=status) - - def send_response_content(self, content, status=200, headers={}): - """send and finish response with content""" - headers['content-length'] = str(len(content)) - self.start_response(status, headers=headers) - if self._stream_state == 1: - self.content = [',\r\n', content] - else: - self.content = [content] - self.finish_response() - - def start_stream(self): - "start stream (array) as part of the response." - assert self._started and self._no_initial_obj - self._stream_state = 0 - self._write("[") - - def stream_entry(self, entry): - "send stream entry as part of the response." - assert self._stream_state != -1 - if self._stream_state == 0: - self._stream_state = 1 - self._write('\r\n') - else: - self._write(',\r\n') - if type(entry) == dict: - entry = json.dumps(entry) - self._write(entry) - - def end_stream(self): - "end stream (array)." - assert self._stream_state != -1 - self._write("\r\n]\r\n") - - -class HTTPInvocationByMethodWithBody(object): - """Invoke methods on a resource.""" - - def __init__(self, resource, environ, parameters): - self.resource = resource - self.environ = environ - self.max_request_size = getattr( - resource, 'max_request_size', parameters.max_request_size) - self.max_entry_size = getattr( - resource, 'max_entry_size', parameters.max_entry_size) - - def _lookup(self, method): - try: - return getattr(self.resource, method) - except AttributeError: - raise BadRequest() - - def __call__(self): - args = urlparse.parse_qsl(self.environ['QUERY_STRING'], - strict_parsing=False) - try: - args = dict( - (k.decode('utf-8'), v.decode('utf-8')) for k, v in args) - except ValueError: - raise BadRequest() - method = self.environ['REQUEST_METHOD'].lower() - if method in ('get', 'delete'): - meth = self._lookup(method) - return meth(args, None) - else: - # we expect content-length > 0, reconsider if we move - # to support chunked enconding - try: - content_length = int(self.environ['CONTENT_LENGTH']) - except (ValueError, KeyError): - raise BadRequest - if content_length <= 0: - raise BadRequest - if content_length > self.max_request_size: - raise BadRequest - reader = _FencedReader(self.environ['wsgi.input'], content_length, - self.max_entry_size) - content_type = self.environ.get('CONTENT_TYPE', '') - content_type = content_type.split(';', 1)[0].strip() - if content_type == 'application/json': - meth = self._lookup(method) - body = reader.read_chunk(sys.maxint) - return meth(args, body) - elif content_type == 'application/x-u1db-sync-stream': - meth_args = self._lookup('%s_args' % method) - meth_entry = self._lookup('%s_stream_entry' % method) - meth_end = self._lookup('%s_end' % method) - body_getline = reader.getline - if body_getline().strip() != '[': - raise BadRequest() - line = body_getline() - line, comma = utils.check_and_strip_comma(line.strip()) - meth_args(args, line) - while True: - line = body_getline() - entry = line.strip() - if entry == ']': - break - if not entry or not comma: # empty or no prec comma - raise BadRequest - entry, comma = utils.check_and_strip_comma(entry) - meth_entry({}, entry) - if comma or body_getline(): # extra comma or data - raise BadRequest - return meth_end() - else: - raise BadRequest() - - -class HTTPApp(object): - - # maximum allowed request body size - max_request_size = 15 * 1024 * 1024 # 15Mb - # maximum allowed entry/line size in request body - max_entry_size = 10 * 1024 * 1024 # 10Mb - - def __init__(self, state): - self.state = state - - def _lookup_resource(self, environ, responder): - resource_cls, params = url_to_resource.match(environ['PATH_INFO']) - if resource_cls is None: - raise BadRequest # 404 instead? - resource = resource_cls( - state=self.state, responder=responder, **params) - return resource - - def __call__(self, environ, start_response): - responder = HTTPResponder(start_response) - self.request_begin(environ) - try: - resource = self._lookup_resource(environ, responder) - HTTPInvocationByMethodWithBody(resource, environ, self)() - except errors.U1DBError as e: - self.request_u1db_error(environ, e) - status = http_errors.wire_description_to_status.get( - e.wire_description, 500) - responder.send_response_json(status, error=e.wire_description) - except BadRequest: - self.request_bad_request(environ) - responder.send_response_json(400, error="bad request") - except KeyboardInterrupt: - raise - except: - self.request_failed(environ) - raise - else: - self.request_done(environ) - return responder.content - - # hooks for tracing requests - - def request_begin(self, environ): - """Hook called at the beginning of processing a request.""" - pass - - def request_done(self, environ): - """Hook called when done processing a request.""" - pass - - def request_u1db_error(self, environ, exc): - """Hook called when processing a request resulted in a U1DBError. - - U1DBError passed as exc. - """ - pass - - def request_bad_request(self, environ): - """Hook called when processing a bad request. - - No actual processing was done. - """ - pass - - def request_failed(self, environ): - """Hook called when processing a request failed unexpectedly. - - Invoked from an except block, so there's interpreter exception - information available. - """ - pass |