diff --git a/fence/blueprints/data/blueprint.py b/fence/blueprints/data/blueprint.py index d8041b949..ad292ddf0 100755 --- a/fence/blueprints/data/blueprint.py +++ b/fence/blueprints/data/blueprint.py @@ -1,4 +1,6 @@ import flask +import json +import hashlib from cdislogging import get_logger from cdispyutils.config import get_value @@ -7,8 +9,10 @@ from fence.authz.auth import check_arborist_auth from fence.blueprints.data.indexd import ( BlankIndex, + EmbeddingIndex, IndexedFile, get_signed_url_for_file, + bulk_get_signed_url_for_file, verify_data_upload_bucket_configuration, ) from fence.config import config @@ -197,6 +201,101 @@ def upload_data_file(): return flask.jsonify(response), 201 +@blueprint.route("/upload/vector", methods=["POST"]) +@require_auth_header(scope={"data"}) +@login_required({"data"}) +def upload_vector(): + """ + Return a presigned URL for use with uploading a data file. + + See the documentation on the entire flow here for more info: + + https://github.com/uc-cdis/cdis-wiki/tree/master/dev/gen3/data_upload + + """ + # make new record in indexd, with just the `uploader` field (and a GUID) + + params = flask.request.get_json() + if not params: + raise UserError("wrong Content-Type; expected application/json") + + if "model" not in params: + raise UserError("missing required argument `model`") + + if "embedding" not in params: + raise UserError("missing required argument `embedding`") + + if "file_id" not in params: + raise UserError("missing required argument `file_id`") + + if "authz" not in params: + raise UserError("missing required argument `authz`") + + authorized = False + authz_err_msg = "Auth error when attempting to get a presigned URL for upload. User must have '{}' access on '{}'." + + authz = params.get("authz") + uploader = None + + guid = params.get("guid") + + if authz: + # if requesting an authz field, using new authorization method which doesn't + # rely on uploader field, so clear it out + uploader = "" + authorized = flask.current_app.arborist.auth_request( + jwt=get_jwt(), + service="fence", + methods=["create", "write-storage"], + resources=authz, + ) + if not authorized: + logger.error(authz_err_msg.format("create' and 'write-storage", authz)) + else: + # no 'authz' was provided, so fall back on 'file_upload' logic + authorized = flask.current_app.arborist.auth_request( + jwt=get_jwt(), + service="fence", + methods=["file_upload"], + resources=["/data_file"], + ) + if not authorized: + logger.error(authz_err_msg.format("file_upload", "/data_file")) + + if not authorized: + raise Forbidden( + "You do not have access to upload data. You either need " + "general file uploader permissions or create and write-storage permissions " + "on the authz resources you specified (if you specified any)." + ) + + # token = get_jwt() + + model = params.get("model") + embedding = params.get("embedding") + file_id = params.get("file_id") + + EMS = EmbeddingIndex( + authz=authz, + model=model, + embedding=embedding, + file_id=file_id, + uploader=uploader, + ) + + # create embedding record and get id from service + embedding_id, md5_hash = EMS.create_embedding_record() + + guid = EMS.create_indexd_record(embedding_id, md5_hash) + + response = { + "message": "An embedding was successfully added to the embedding management service and inserted into indexd", + "guid": guid, + } + + return flask.jsonify(response), 201 + + @blueprint.route("/multipart/init", methods=["POST"]) @require_auth_header(scope={"data"}) @login_required({"data"}) @@ -348,6 +447,35 @@ def download_file(file_id): return flask.redirect(result["url"]) +@blueprint.route("/download/bulk", methods=["POST"]) +def download_bulk_files(): + """ + Get a presigned url to download a file given by file_id. + """ + # {"guids": ["1234", "4567"]} + params = flask.request.get_json() + if not params: + raise UserError("wrong Content-Type; expected application/json") + + if "guids" not in params: + raise UserError("missing required argument `guids`") + + guids = params["guids"] + results = bulk_get_signed_url_for_file(guids) + + return flask.jsonify(results) + # results = {} + # results["urls"] = [] + # for g in guids: + # result = get_signed_url_for_file("download", g) + # if not "redirect" in flask.request.args or not "url" in result: + # # return flask.jsonify(result) + # results["urls"].append(result) + # else: + # results["urls"].append(result["url"]) + # return flask.jsonify(results) + + @blueprint.route( "/buckets", methods=["GET"], diff --git a/fence/blueprints/data/indexd.py b/fence/blueprints/data/indexd.py index c8293044c..92840621d 100755 --- a/fence/blueprints/data/indexd.py +++ b/fence/blueprints/data/indexd.py @@ -1,6 +1,7 @@ import re import time import json +import hashlib import boto3 from botocore.client import Config from urllib.parse import urlparse, ParseResult, urlunparse @@ -65,7 +66,7 @@ "az": {"upload": "PUT", "download": "GET"}, } -SUPPORTED_PROTOCOLS = ["s3", "http", "ftp", "https", "gs", "az"] +SUPPORTED_PROTOCOLS = ["s3", "http", "ftp", "https", "gs", "az", "vec"] SUPPORTED_ACTIONS = ["upload", "download"] ANONYMOUS_USER_ID = "-1" ANONYMOUS_USERNAME = "anonymous" @@ -182,6 +183,57 @@ def get_signed_url_for_file( return {"url": signed_url} +@enable_audit_logging +def bulk_get_signed_url_for_file( + file_ids, + requested_protocol=None, + db_session=None, + bucket=None, + drs="False", +): + requested_protocol = requested_protocol or flask.request.args.get("protocol", None) + r_pays_project = flask.request.args.get("userProject", None) + db_session = db_session or current_app.scoped_session() + + # default to signing the URL + force_signed_url = True + no_force_sign_param = flask.request.args.get("no_force_sign") + if no_force_sign_param and no_force_sign_param.lower() == "true": + force_signed_url = False + + # Collect X-Forwarded headers + x_forwarded_headers = [ + f"{header}:{value}" for header, value in flask.request.headers if "X-" in header + ] + + auth_info = _get_auth_info_for_id_or_from_request( + sub_type=int, db_session=db_session + ) + flask.g.audit_data = { + "username": auth_info["username"], + "sub": auth_info["user_id"], + "additional_data": x_forwarded_headers, + } + + indexed_files = BulkIndexedFiles(file_ids) + + default_expires_in = config.get("MAX_PRESIGNED_URL_TTL", 3600) + expires_in = get_valid_expiration_from_request( + max_limit=default_expires_in, + default=default_expires_in, + ) + + # prepare_presigned_url_audit_log(requested_protocol, indexed_file) + signed_urls, users = indexed_files.get_signed_urls( + requested_protocol, + expires_in, + force_signed_url=force_signed_url, + r_pays_project=r_pays_project, + ) + + return {"urls": signed_urls} + + def get_bucket_from_urls(urls, protocol): """ Return the bucket name from the first of the provided URLs that starts with the given protocol (usually `gs`, `s3`, `az`...) @@ -261,6 +313,122 @@ def prepare_presigned_url_audit_log(protocol, indexed_file): flask.g.audit_data["protocol"] = protocol +class EmbeddingIndex(object): + def __init__( + self, + authz=None, + model=None, + embedding=None, + file_id=None, + uploader=None, + logger_=None, + ): + self.logger = logger_ or logger + self.indexd = ( + flask.current_app.config.get("INDEXD") + or flask.current_app.config["BASE_URL"] + "/index" + ) + + # allow passing "" empty string to signify you do NOT want + # uploader to be populated. If nothing is provided, default + # to parsing from token + if uploader == "": + self.uploader = None + elif uploader: + self.uploader = uploader + else: + self.uploader = current_token["context"]["user"]["name"] + + self.authz = authz + self.model = model + self.embedding = embedding + self.file_id = file_id + + def create_embedding_record(self): + """ + create a record in the embedding management service + """ + + embedding_management_service_url = f"http://embedding-management-service/vector/indexes/{self.model}/embeddings" + params = { + "file_id": self.file_id, + "vector": self.embedding, + } + + ems_response = requests.post(embedding_management_service_url, json=params) + if ems_response.status_code not in [200, 201]: + try: + data = ems_response.json() + except ValueError: + data = ems_response.text + self.logger.error( + "could not create new record in embedding management service; got response: {}".format( + data + ) + ) + raise InternalError("received error from embedding management service") + else: + dict_str = json.dumps(params, sort_keys=True) + dict_bytes = dict_str.encode("utf-8") + md5_hash = hashlib.md5(dict_bytes, usedforsecurity=False).hexdigest() + + emsID = ems_response.text + logger.info( + f"successfully added vector to embedding service with id {emsID}" + ) + + return emsID, md5_hash + + def create_indexd_record(self, embedding_id, md5_hash): + """ + create an indexd record for a vector corresponding to the embedding_id + """ + index_url = self.indexd.rstrip("/") + "/index/" + escaped_id = embedding_id.strip('"') + vec_url = f"vec://{self.model}/{escaped_id}" + + params = { + "authz": self.authz, + "form": "object", + "hashes": {"md5": md5_hash}, + "size": 0, + "urls": [vec_url], + } + + if self.authz: + params["authz"] = self.authz + token = get_jwt() + + auth = None + headers = {"Authorization": f"bearer {token}"} + logger.info("passing users authorization header to create blank record") + else: + logger.info("using indexd basic auth to create blank record") + auth = (config["INDEXD_USERNAME"], config["INDEXD_PASSWORD"]) + headers = {} + + indexd_response = requests.post( + index_url, json=params, headers=headers, auth=auth + ) + if indexd_response.status_code not in [200, 201]: + try: + data = indexd_response.json() + except ValueError: + data = indexd_response.text + self.logger.error( + "could not create new record in indexd; got response: {}".format(data) + ) + raise InternalError( + "received error from indexd trying to create blank record" + ) + + document = indexd_response.json() + guid = document["did"] + self.logger.info("created an index record with GUID {} for upload".format(guid)) + + return guid + + class BlankIndex(object): """ A blank record in indexd, to use for the data upload flow. @@ -470,6 +638,239 @@ def generate_aws_presigned_url_for_part( ) +class BulkIndexedFiles(object): + """ + For bulk handeling of files + + Args: + file_ids (list(str)): A list of GUIDs coresponding to files + """ + + def __init__(self, file_ids): + self.file_ids = file_ids + self.auth_roles = [] + + @cached_property + def indexd_server(self): + indexd_server = ( + flask.current_app.config.get("INDEXD") + or flask.current_app.config["BASE_URL"] + "/index" + ) + return indexd_server.rstrip("/") + + @cached_property + def index_document(self): + index_dict = {} + for file_id in self.file_ids: + indexd_server = config.get("INDEXD") or config["BASE_URL"] + "/index" + url = indexd_server + "/index/" + try: + res = requests.get(url + file_id) + except Exception as e: + logger.error( + "failed to reach indexd at {0}: {1}".format(url + file_id, e) + ) + raise UnavailableError("Fail to reach id service to find data location") + if res.status_code == 200: + try: + json_response = res.json() + if "urls" not in json_response: + logger.error( + "URLs are not included in response from " + "indexd: {}".format(url + file_id) + ) + raise InternalError("URLs and metadata not found") + + # indexd can resolve GUIDs without prefix, but cannot perform other operations + # (such as delete) without the prefix, so make sure `file_id` is the whole GUID + real_guid = json_response.get("did") + if real_guid and real_guid != file_id: + file_id = real_guid + + index_dict[file_id] = json_response + except Exception as e: + logger.error( + "indexd response missing JSON field {}".format(url + file_id) + ) + raise InternalError("internal error from indexd: {}".format(e)) + elif res.status_code == 404: + logger.error( + "Not Found. indexd could not find {}: {}".format( + url + file_id, res.text + ) + ) + raise NotFound("No indexed document found with id {}".format(file_id)) + else: + raise UnavailableError(res.text) + + return index_dict + + def get_signed_urls( + self, + protocol, + expires_in, + force_signed_url=True, + r_pays_project=None, + users_from_passports=None, + ): + users_from_passports = users_from_passports or {} + signed_urls = [] + users = [] + for file_id in self.file_ids: + authorized_user = None + file_authz = self.index_document.get(file_id).get("authz") + + if file_authz: + # check if user has already been authorized in this bulk operation + if self.index_document.get(file_id).get("authz") not in self.auth_roles: + action_to_permission = { + "download": "read-storage", + } + is_authorized, authorized_username = ( + self.get_authorized_with_username( + action_to_permission["download"], + file_id, + # keys are usernames + usernames_from_passports=list(users_from_passports.keys()), + ) + ) + if not is_authorized: + msg = ( + f"Either you weren't authenticated successfully or you don't have " + f"read-storage permission " + f"on authorization resource: {self.index_document['authz']}." + ) + logger.debug( + f"denied. authorized_username: {authorized_username}\nmsg:\n{msg}" + ) + raise Unauthorized(msg) + self.auth_roles.append( + self.index_document.get(file_id).get("authz") + ) + authorized_user = users_from_passports.get(authorized_username) + + signed_url, user = ( + self._get_signed_urls( + protocol, + file_id, + expires_in, + force_signed_url, + r_pays_project, + authorized_user, + ), + authorized_user, + ) + signed_urls.append(signed_url) + users.append(user) + + return signed_urls, users + + def _get_signed_urls( + self, + protocol, + file_id, + expires_in, + force_signed_url, + r_pays_project, + authorized_user=None, + ): + # we only allow for download with bulk operations + file_urls = self.index_document.get(file_id).get("urls", []) + file_locations = list(map(IndexedFileLocation.from_url, file_urls)) + if not protocol: + # no protocol specified, return first location as signed url + try: + return file_locations[0].get_signed_url( + "download", + expires_in, + force_signed_url=force_signed_url, + r_pays_project=r_pays_project, + authorized_user=authorized_user, + ) + except IndexError: + raise NotFound("Can't find any file locations.") + + for file_location in file_locations: + # allow file location to be https, even if they specific http + if (file_location.protocol == protocol) or ( + protocol == "http" and file_location.protocol == "https" + ): + return file_location.get_signed_url( + "download", + expires_in, + force_signed_url=force_signed_url, + r_pays_project=r_pays_project, + authorized_user=authorized_user, + ) + + raise NotFound( + "File {} does not have a location with specified " + "protocol {}.".format(file_id, protocol) + ) + + def get_authorized_with_username( + self, action, file_id, usernames_from_passports=None + ): + """ + Return a tuple of (boolean, str) which represents whether they're authorized + and their username. username is only returned if `usernames_from_passports` + is provided and one of the usernames from the passports is authorized. + + Args: + action (str): Authorization action being performed + usernames_from_passports (list[str], optional): List of user usernames parsed + from validated passports + + Returns: + tuple of (boolean, str): which represents whether they're authorized + and their username. username is only returned if `usernames_from_passports` + is provided and one of the usernames from the passports is authorized. + """ + if not self.index_document.get(file_id).get("authz"): + raise ValueError("index record missing `authz`") + + logger.debug( + f"authz check can user {action} on {self.index_document[file_id]['authz']} for fence? " + f"if passport provided, IDs parsed: {usernames_from_passports}" + ) + + # handle multiple GA4GH passports as a means of authn/z + if usernames_from_passports: + authorized = False + for username in usernames_from_passports: + authorized = flask.current_app.arborist.auth_request( + jwt=None, + user_id=username, + service="fence", + methods=action, + resources=self.index_document[file_id]["authz"], + ) + # if any passport provides access, user is authorized + if authorized: + # for google proxy groups and future use: we need to know which + # user_id actually gave access + return authorized, username + return authorized, None + else: + try: + token = get_jwt() + except Unauthorized: + # get_jwt raises an Unauthorized error when user is anonymous (no + # available token), so to allow anonymous users possible access to + # public data, we still make the request to Arborist + token = None + + return ( + flask.current_app.arborist.auth_request( + jwt=token, + service="fence", + methods=action, + resources=self.index_document[file_id]["authz"], + ), + None, + ) + + class IndexedFile(object): """ A file from the index service that will contain information about access and where @@ -853,6 +1254,8 @@ def from_url(url): return GoogleStorageIndexedFileLocation(url) elif protocol == "az": return AzureBlobStorageIndexedFileLocation(url) + elif protocol == "vec": + return VecIndexdFileLocation(url) return IndexedFileLocation(url) def get_signed_url( @@ -866,6 +1269,26 @@ def get_signed_url( return self.url +class VecIndexdFileLocation(IndexedFileLocation): + def get_signed_url( + self, + action, + expires_in, + force_signed_url=True, + r_pays_project=None, + authorized_user=None, + ): + # need to return the vector from the embedding management service + model, emsID = str(self.url).replace("vec://", "").split("/") + emsUrl = f"http://embedding-management-service/vector/indexes/{model}/embeddings/{emsID}" + try: + req = requests.get(emsUrl) + return req.json() + except Exception as e: + print(f"error {e}") + raise NotFound(f"No embedding found with id {emsID}") + + class S3IndexedFileLocation(IndexedFileLocation): """ An indexed file that lives in an AWS S3 bucket.