ingest_operations.py

import connexion
from flask import Flask
import os
import re
import traceback
import urllib.parse

import auth
import katsu_ingest
import htsget_ingest
from opa_ingest import remove_user_from_dataset, add_user_to_dataset
import config
import tempfile
import uuid
import json


app = Flask(__name__)

ERROR_CODES = {
    "SUCCESS": 0,
    "UNAUTHORIZED": 1,
    "VALIDATION": 2,
    "PROGRAMEXISTS": 3,
    "INTERNAL": 4,
    "AUTHORIZATIONERR": 5
}

def generateResponse(result, response_code):
    response_mapping = {
        0: ("Success", 200),
        1: ("Unauthorized", 403),
        2: ("Validation error", 422),
        3: ("Program exists", 422),
        4: ("Internal CanDIG error", 500),
        5: ("Authorization error", 401)
    }
    return {"result": result, "response_code": response_code,
            "response_message": response_mapping[response_code][0]}, response_mapping[response_code][1]

def get_headers():
    headers = {}
    if "Authorization" not in connexion.request.headers:
        return generateResponse("Bearer token required", ERROR_CODES["UNAUTHORIZED"])
    try:
        # New auth model
        # refresh_token = connexion.request.headers["Authorization"].split("Bearer ")[1]
        # token = auth.get_bearer_from_refresh(refresh_token)
        if not connexion.request.headers["Authorization"].startswith("Bearer "):
            return generateResponse("Invalid bearer token", ERROR_CODES["UNAUTHORIZED"])
        token = connexion.request.headers["Authorization"].split("Bearer ")[1]
        headers["Authorization"] = "Bearer %s" % token
    except Exception as e:
        if "Invalid bearer token" in str(e):
            return generateResponse("Bearer token invalid or unauthorized", ERROR_CODES["UNAUTHORIZED"])
        return generateResponse("Unknown error during authorization", ERROR_CODES["AUTHORIZATIONERR"])
    headers["Content-Type"] = "application/json"
    return headers


def check_default_site_admin(response):
    if auth.is_default_site_admin_set():
        if "warnings" not in response:
            response["warnings"] = []
        response["warnings"].append(f"Default site administrator {os.getenv('DEFAULT_SITE_ADMIN_USER')} is still configured. Use the /ingest/site-role/site_admin endpoint to set a different site admin.")


# API endpoints
def get_service_info():
    return {
        "id": "org.candig.ingest",
        "name": "CanDIG Ingest Passthrough Service",
        "description": "A microservice used as a processing intermediary for ingesting data into Katsu and htsget",
        "organization": {
            "name": "CanDIG",
            "url": "https://www.distributedgenomics.ca"
        },
        "version": config.VERSION
    }


####
# S3 credentials
####

async def add_s3_credential():
    data = await connexion.request.json()
    token = connexion.request.headers['Authorization'].split("Bearer ")[1]
    return auth.store_s3_credential(data["endpoint"], data["bucket"], data["access_key"], data["secret_key"], token)


@app.route('/s3-credential/endpoint/<path:endpoint_id>/bucket/<path:bucket_id>')
def get_s3_credential(endpoint_id, bucket_id):
    token = connexion.request.headers['Authorization'].split("Bearer ")[1]
    endpoint_cleaned = re.sub(r"\W", "_", endpoint_id)
    return auth.get_s3_credential(endpoint_cleaned, bucket_id, token)


@app.route('/s3-credential/endpoint/<path:endpoint_id>/bucket/<path:bucket_id>')
def delete_s3_credential(endpoint_id, bucket_id):
    token = connexion.request.headers['Authorization'].split("Bearer ")[1]
    endpoint_cleaned = re.sub(r"\W", "_", endpoint_id)
    return auth.remove_s3_credential(endpoint_cleaned, bucket_id, token)


####
# Site roles
####

@app.route('/site-role/<path:role_type>')
def list_role(role_type):
    try:
        token = connexion.request.headers['Authorization'].split("Bearer ")[1]
        result, status_code = auth.get_role_type_in_opa(role_type, token)
        return result, status_code
    except Exception as e:
        return {"error": str(e)}, 500


@app.route('/site-role/<path:role_type>')
async def update_role(role_type):
    role_members = await connexion.request.json()
    try:
        token = connexion.request.headers['Authorization'].split("Bearer ")[1]
        result, status_code = auth.set_role_type_in_opa(role_type, role_members, token)
        return result, status_code
    except Exception as e:
        return {"error": str(e)}, 500


@app.route('/site-role/<path:role_type>/email/<path:email>')
def is_user_in_role(role_type, email):
    try:
        token = connexion.request.headers['Authorization'].split("Bearer ")[1]
        result, status_code = auth.get_role_type_in_opa(role_type, token)
        if status_code == 200:
            return (email in result[role_type]), 200
        return result, status_code
    except Exception as e:
        return {"error": str(e)}, 500


@app.route('/site-role/<path:role_type>/email/<path:email>')
def add_user_to_role(role_type, email):
    try:
        token = connexion.request.headers['Authorization'].split("Bearer ")[1]
        result, status_code = auth.get_role_type_in_opa(role_type, token)
        if status_code == 200:
            if email not in result[role_type]:
                result[role_type].append(email)
                result, status_code = auth.set_role_type_in_opa(role_type, result[role_type], token)
        return result, status_code
    except Exception as e:
        return {"error": str(e)}, 500


@app.route('/site-role/<path:role_type>/email/<path:email>')
def remove_user_from_role(role_type, email):
    try:
        token = connexion.request.headers['Authorization'].split("Bearer ")[1]
        result, status_code = auth.get_role_type_in_opa(role_type, token)
        if status_code == 200:
            if email in result[role_type]:
                result[role_type].remove(email)
                result, status_code = auth.set_role_type_in_opa(role_type, result[role_type], token)
            else:
                return {"error": f"User {email} not found in role {role_type}"}, 404
        return result, status_code
    except Exception as e:
        return {"error": str(e)}, 500

####
# Data ingest
####

async def add_genomic_linkages():
    dataset = await connexion.request.json()
    do_not_index = bool(connexion.request.query_params.get("do_not_index", False))
    headers = get_headers()
    token = connexion.request.headers['Authorization'].split("Bearer ")[1]
    response, status_code = htsget_ingest.check_genomic_data(dataset, token)
    if status_code == 200:
        ingest_uuid = add_to_queue({"htsget": response, "do_not_index": do_not_index})
        response = {"queue_id": ingest_uuid}
    check_default_site_admin(response)
    return response, status_code


async def add_clinical_donors():
    dataset = await connexion.request.json()
    batch_size = int(connexion.request.query_params.get("batch_size", 1000))
    headers = get_headers()
    token = connexion.request.headers['Authorization'].split("Bearer ")[1]
    response, status_code = katsu_ingest.prep_check_clinical_data(dataset, token, batch_size)
    if status_code == 200:
        ingest_uuid = add_to_queue({"katsu": response})
        response = {"queue_id": ingest_uuid}
    check_default_site_admin(response)
    return response, status_code


def add_to_queue(ingest_json):
    queue_id = str(uuid.uuid1())
    with tempfile.NamedTemporaryFile(delete_on_close=False, mode="w") as f:
        json.dump(ingest_json, f, indent=4)
        os.rename(f.name, os.path.join(config.DAEMON_PATH, "to_ingest", queue_id))
    results_path = os.path.join(config.DAEMON_PATH, "results", queue_id)
    with open(results_path, "w") as f:
        json.dump({"status": "still in queue"}, f)
    return queue_id


@app.route('/status/<path:queue_id>')
def get_ingest_status(queue_id):
    try:
        results_path = os.path.join(config.DAEMON_PATH, "results", queue_id)
        with open(results_path) as f:
            json_data = json.load(f)
            # os.remove(results_path)
            return json_data, 200
    except:
        return {"error": f"no such queue_id {queue_id}"}, 404


####
# Program authorizations
####

def list_program_authorizations():
    token = connexion.request.headers['Authorization'].split("Bearer ")[1]

    response, status_code = auth.list_programs_in_opa(token)
    return response, status_code


async def add_program_authorization():
    program = await connexion.request.json()
    token = connexion.request.headers['Authorization'].split("Bearer ")[1]

    response, status_code = auth.add_program_to_opa(program, token)
    check_default_site_admin(response)
    return response, status_code


@app.route('/program/<path:program_id>')
def get_program_authorization(program_id):
    token = connexion.request.headers['Authorization'].split("Bearer ")[1]

    response, status_code = auth.get_program_in_opa(program_id, token)
    return response, status_code


@app.route('/program/<path:program_id>')
def remove_program(program_id):
    token = connexion.request.headers['Authorization'].split("Bearer ")[1]
    response = {"errors": {}}
    check_default_site_admin(response)

    opa_response, opa_status = auth.remove_program_from_opa(program_id, token)
    katsu_response = katsu_ingest.delete_program(program_id, token)
    htsget_response = htsget_ingest.delete_program(program_id, token)

    if opa_status == 404:
        # htsget status is not included here because it doesn't have a 404 response
        return {"message": f"Program {program_id} not found"}, 404

    if opa_status != 200:
        response["errors"]["opa"] = {"message": opa_response, "status_code": opa_status}

    if katsu_response.status_code != 204 and katsu_response.status_code != 404:
        response["errors"]["katsu"] = {"message": katsu_response.text, "status_code": katsu_response.status_code}

    if htsget_response.status_code != 200:
        response["errors"]["htsget"] = {"message": htsget_response.text, "status_code": htsget_response.status_code}

    if len(response["errors"]) == 0:
        response.pop("errors")
        response["message"] = f"Program {program_id} successfully deleted"
        return response, 200

    return response, 500


####
# Pending users: approving a pending user creates a CanDIG-authorized user
####

def add_pending_user():
    token = connexion.request.headers['Authorization'].split("Bearer ")[1]

    response, status_code = auth.add_pending_user_to_opa(token)
    return response, status_code


def list_pending_users():
    token = connexion.request.headers['Authorization'].split("Bearer ")[1]

    response, status_code = auth.list_pending_users_in_opa(token)
    return {"results": response}, status_code


@app.route('/user/pending/<path:user_id>')
def approve_pending_user(user_id):
    token = connexion.request.headers['Authorization'].split("Bearer ")[1]
    user_name = urllib.parse.unquote_plus(user_id)

    response, status_code = auth.approve_pending_user_in_opa(user_name, token)
    return response, status_code


@app.route('/user/pending/<path:user_id>')
def reject_pending_user(user_id):
    token = connexion.request.headers['Authorization'].split("Bearer ")[1]
    user_name = urllib.parse.unquote_plus(user_id)

    response, status_code = auth.reject_pending_user_in_opa(user_name, token)
    return response, status_code


async def approve_pending_users():
    users = await connexion.request.json()
    token = connexion.request.headers['Authorization'].split("Bearer ")[1]

    rejected = []
    for user_id in users:
        response, status_code = auth.approve_pending_user_in_opa(user_id, token)
        if status_code != 200:
            rejected.append(user_id)
    if len(rejected) > 0:
        status_code = 401
        response = {"message": f"The following requested user IDs could not be approved: {rejected}"}

    return response, status_code


def clear_pending_users():
    token = connexion.request.headers['Authorization'].split("Bearer ")[1]

    response, status_code = auth.clear_pending_users_in_opa(token)
    return response, status_code


####
# DAC authorization for users
####

def list_programs_for_self(token):
    response, status_code = auth.get_self_in_opa(token)
    if status_code == 404:
        # We next check if the user is pending
        response, status_code = auth.is_self_pending(token)
        # NB: The results is a string if unauthorized or pending, and a list otherwise
        return "Pending" if response else "Unauthorized", status_code
    print(response)
    # NB: The results is a list if authorized, and a string otherwise
    return list(response["programs"].values()), status_code


@app.route('/user/<path:user_id>/authorize')
def list_programs_for_user(user_id):
    token = connexion.request.headers['Authorization'].split("Bearer ")[1]
    response = ""
    status_code = 0
    if user_id == "me":
        # Grab the user's own authorization
        response, status_code = list_programs_for_self(token)
    else:
        user_name = urllib.parse.unquote_plus(user_id)
        response, status_code = auth.get_user_in_opa(user_name, token)
        if status_code != 200:
            return response, status_code
        response = list(response["programs"].values())
    print(response)
    return {"results": response}, status_code


@app.route('/user/<path:user_id>/authorize')
async def authorize_program_for_user(user_id):
    program_dict = await connexion.request.json()
    token = connexion.request.headers['Authorization'].split("Bearer ")[1]
    user_name = urllib.parse.unquote_plus(user_id)
    response, status_code = auth.get_user_in_opa(user_name, token)
    if status_code != 200:
        return response, status_code

    # we need to check to see if the program even exists in the system
    all_programs, status_code = auth.list_programs_in_opa(token)
    if status_code != 200:
        return all_programs, status_code
    if program_dict["program_id"] not in all_programs:
        return {"error": f"Program {program_dict['program_id']} does not exist in {all_programs}"}
    response["programs"][program_dict["program_id"]] = program_dict
    response, status_code = auth.write_user_in_opa(response, token)
    return response, status_code


@app.route('/user/<path:user_id>/authorize/<path:program_id>')
def get_program_for_user(user_id, program_id):
    token = connexion.request.headers['Authorization'].split("Bearer ")[1]
    user_name = urllib.parse.unquote_plus(user_id)

    response, status_code = auth.get_user_in_opa(user_name, token)
    if status_code != 200:
        return response, status_code
    for p in response["programs"]:
        if p == program_id:
            return p, 200
    return {"error": f"No program {program_id} found for user"}, status_code


@app.route('/user/<path:user_id>/authorize/<path:program_id>')
def remove_program_for_user(user_id, program_id):
    token = connexion.request.headers['Authorization'].split("Bearer ")[1]
    user_name = urllib.parse.unquote_plus(user_id)

    response, status_code = auth.get_user_in_opa(user_name, token)
    if status_code != 200:
        return response, status_code
    for p in response["programs"]:
        if p == program_id:
            response["programs"].pop(program_id)
            response, status_code = auth.write_user_in_opa(response, token)
            return response, status_code
    return {"error": f"No program {program_id} found for user"}, status_code


@app.route('/get-token')
def get_token():
    # Attempt to grab the token via session_id
    if not hasattr(connexion.request, 'cookies'):
        return {'error': 'Unable to use the get-token endpoint without cookies'}, 200
    token = connexion.request.cookies['session_id']

    return {"token": token}, 200

    # Uncomment the below to exchange for a new token and return
    # that, instead
    # try:
    #    response = auth.get_refresh_token(token)
    #    if "error" in response:
    #        return {"error": response["error"]}, 500
    #    return {"token": response["refresh_token"]}, 200
    #except Exception as e:
    #    return {"error": str(e)}, 500