diff --git a/Makefile b/Makefile index 964498f47..6ccb8ebac 100644 --- a/Makefile +++ b/Makefile @@ -16,6 +16,7 @@ export PYTHONPATH = ${API}:${CHECK_PROCESSING_TIME}:${GET_FILES}:${HANDLE_BATCH_ build: render + cp -r job_spec apps/api/src/hyp3_api/job_spec python -m pip install --upgrade -r requirements-apps-api.txt -t ${API}; \ python -m pip install --upgrade -r requirements-apps-api-binary.txt --platform manylinux2014_x86_64 --only-binary=:all: -t ${API}; \ python -m pip install --upgrade -r requirements-apps-handle-batch-event.txt -t ${HANDLE_BATCH_EVENT}; \ diff --git a/apps/api/api-cf.yml.j2 b/apps/api/api-cf.yml.j2 index 1fbe7dd71..f8de94fde 100644 --- a/apps/api/api-cf.yml.j2 +++ b/apps/api/api-cf.yml.j2 @@ -182,6 +182,11 @@ Resources: Action: - dynamodb:GetItem Resource: !Sub "arn:aws:dynamodb:${AWS::Region}:${AWS::AccountId}:table/${AccessCodesTable}*" + - Effect: Allow + Action: + - s3:PutObject + - s3:PutObjectTagging + Resource: !Sub "arn:aws:s3:::${ContentBucket}/*" Lambda: Type: AWS::Lambda::Function diff --git a/apps/api/src/hyp3_api/api-spec/job_parameters.yml.j2 b/apps/api/src/hyp3_api/api-spec/job_parameters.yml.j2 index d99538a24..8e7ef4480 100644 --- a/apps/api/src/hyp3_api/api-spec/job_parameters.yml.j2 +++ b/apps/api/src/hyp3_api/api-spec/job_parameters.yml.j2 @@ -15,7 +15,19 @@ components: - {{ job_type }} {% endfor %} + upload_job_type: + description: Type of process to run this job. + type: string + example: {{ job_types.__iter__().__next__() }} + enum: + {% for job_type, job_spec in job_types.items() %} + {% if 'files' in job_spec.keys() %} + - {{ job_type }} + {% endif %} + {% endfor %} + job_parameters: + type: object anyOf: {% for job_type in job_types %} - $ref: "#/components/schemas/{{ job_type }}Parameters" @@ -23,6 +35,7 @@ components: {% for job_type, job_spec in job_types.items() %} {{ job_type }}Parameters: + title: {{ job_type }}Parameters description: Parameters for running {{ job_type }} jobs type: object additionalProperties: false @@ -41,12 +54,13 @@ components: {% endfor %} {{ job_type }}Job: + title: {{ job_type }}Job description: Contains user provided information on running a new {{ job_type }} job. type: object - additionalProperties: false required: - job_type - job_parameters + additionalProperties: false properties: job_type: type: string @@ -60,5 +74,4 @@ components: $ref: "./openapi-spec.yml#/components/schemas/bucket_prefix" job_parameters: $ref: "#/components/schemas/{{ job_type }}Parameters" - {% endfor %} diff --git a/apps/api/src/hyp3_api/api-spec/openapi-spec.yml.j2 b/apps/api/src/hyp3_api/api-spec/openapi-spec.yml.j2 index b90ec04e1..b2ea433db 100644 --- a/apps/api/src/hyp3_api/api-spec/openapi-spec.yml.j2 +++ b/apps/api/src/hyp3_api/api-spec/openapi-spec.yml.j2 @@ -16,7 +16,6 @@ servers: {% endif %} paths: - /costs: get: description: Get table of job costs. @@ -28,8 +27,54 @@ paths: schema: $ref: "#/components/schemas/costs_response" - /jobs: + /upload-job: + post: + description: Submits a list of jobs for processing. + requestBody: + content: + multipart/form-data: + schema: + type: object + required: + - job_type + properties: + job_type: + $ref: "./job_parameters.yml#/components/schemas/upload_job_type" + name: + $ref: "#/components/schemas/name" + bucket: + $ref: "#/components/schemas/bucket" + bucket_prefix: + $ref: "#/components/schemas/bucket_prefix" + job_parameters: + $ref: "./job_parameters.yml#/components/schemas/job_parameters" + {% for job_type, job_spec in job_types.items() %} + {% if 'files' in job_spec.keys() %} + {% for parameter, parameter_spec in job_spec['files'].items() %} + {{ parameter }}: + {{ json.dumps(parameter_spec['api_schema']) }} + {% endfor %} + {% endif %} + {% endfor %} + encoding: + {% for job_type, job_spec in job_types.items() %} + {% if 'files' in job_spec.keys() %} + {% for parameter, parameter_spec in job_spec['files'].items() %} + {{ parameter }}: + contentType: {{ parameter_spec['allowed_types'] | join(', ') }} + {% endfor %} + {% endif %} + {% endfor %} + required: true + responses: + "200": + description: 200 response + content: + application/json: + schema: + $ref: "#/components/schemas/jobs_response" + /jobs: post: description: Submits a list of jobs for processing. requestBody: diff --git a/apps/api/src/hyp3_api/handlers.py b/apps/api/src/hyp3_api/handlers.py index 66d52fe57..68e172023 100644 --- a/apps/api/src/hyp3_api/handlers.py +++ b/apps/api/src/hyp3_api/handlers.py @@ -1,6 +1,8 @@ +import json from http.client import responses +from uuid import uuid4 -from flask import Response, abort, jsonify, request +from flask import Request, Response, abort, jsonify, request import dynamo from dynamo.exceptions import ( @@ -13,7 +15,7 @@ ) from hyp3_api import util from hyp3_api.multi_burst_validation import MultiBurstValidationError -from hyp3_api.validation import CmrError, ValidationError, validate_jobs +from hyp3_api.validation import CmrError, ValidationError, validate_files, validate_job_parameters, validate_jobs def problem_format(status: int, message: str) -> Response: @@ -23,6 +25,50 @@ def problem_format(status: int, message: str) -> Response: return response +def _get_request_dict(request: Request) -> dict: + """Retrieve and sanitize request dictionary from `/upload-job` form.""" + request_form = dict(request.form) + allowed_params = ['job_type', 'name', 'bucket', 'bucket_prefix', 'job_parameters'] + + # File params from other job types with files will be included in the request and need to be removed + params = list(request_form.keys()) + for param in params: + if param not in allowed_params: + request_form.pop(param) + + request_form['job_parameters'] = json.loads(request_form['job_parameters']) + + return request_form + + +def post_upload_job(request: Request, user: str) -> dict: + request_dict = _get_request_dict(request) + request_dict['job_id'] = str(uuid4()) + try: + validate_files(request) + validate_job_parameters(request_dict) + validate_jobs([request_dict]) + except CmrError as e: + abort(problem_format(503, str(e))) + except (ValidationError, MultiBurstValidationError) as e: + abort(problem_format(400, str(e))) + try: + request_dict = dynamo.jobs.handle_content_bucket(request_dict) + for file_param, file_obj in request.files.items(): + s3_uri = util.save_and_upload_to_s3( + file_obj=file_obj, bucket=request_dict['bucket'], bucket_prefix=request_dict['bucket_prefix'] + ) + request_dict['job_parameters'][file_param] = s3_uri + request_dict = dynamo.jobs.put_jobs(user, [request_dict])[0] + except UnexpectedApplicationStatusError as e: + abort(problem_format(403, str(e))) + except InsufficientCreditsError as e: + abort(problem_format(400, str(e))) + except CustomPrefixForDefaultBucketError as e: + abort(problem_format(400, str(e))) + return request_dict + + def post_jobs(body: dict, user: str) -> dict: print(body) try: diff --git a/apps/api/src/hyp3_api/routes.py b/apps/api/src/hyp3_api/routes.py index f4cd9c228..7b338ce80 100644 --- a/apps/api/src/hyp3_api/routes.py +++ b/apps/api/src/hyp3_api/routes.py @@ -26,7 +26,7 @@ JWKS_CLIENT = auth.get_jwks_client() -AUTHENTICATED_ROUTES = ['/jobs', '/user'] +AUTHENTICATED_ROUTES = ['/jobs', '/user', '/upload-job'] @app.before_request @@ -139,6 +139,12 @@ def costs_get() -> Response: return jsonify(dynamo.jobs.COSTS) +@app.route('/upload-job', methods=['POST']) +@openapi +def upload_job_post() -> Response: + return jsonify(handlers.post_upload_job(request, g.user)) + + @app.route('/jobs', methods=['POST']) @openapi def jobs_post() -> Response: diff --git a/apps/api/src/hyp3_api/util.py b/apps/api/src/hyp3_api/util.py index eeefe0b57..0d2dcbf23 100644 --- a/apps/api/src/hyp3_api/util.py +++ b/apps/api/src/hyp3_api/util.py @@ -1,10 +1,17 @@ import binascii import json from base64 import b64decode, b64encode +from pathlib import Path +from tempfile import TemporaryDirectory from typing import Any from urllib.parse import parse_qsl, urlencode, urlparse, urlunparse import boto3 +from boto3.s3.transfer import TransferConfig +from werkzeug.datastructures import FileStorage + + +S3_CLIENT = boto3.client('s3') class TokenDeserializeError(Exception): @@ -53,3 +60,22 @@ def build_next_url(url: str, start_token: str, x_forwarded_host: str | None = No def get_current_account_arn() -> str: sts = boto3.client('sts') return sts.get_caller_identity()['Account'] + + +def _upload_file_to_s3( + path_to_file: Path, content_type: str, bucket: str, prefix: str = '', chunk_size: int = 8_388_608 +) -> None: + key = str(Path(prefix) / path_to_file.name) + extra_args = {'ContentType': content_type} + config = TransferConfig(multipart_threshold=chunk_size, multipart_chunksize=chunk_size) + S3_CLIENT.upload_file(str(path_to_file), bucket, key, extra_args, Config=config) + + +def save_and_upload_to_s3(file_obj: FileStorage, bucket: str, bucket_prefix: str) -> str: + filename = file_obj.filename + assert filename + with TemporaryDirectory() as temp_dir: + filepath = Path(temp_dir) / filename + file_obj.save(filepath) + _upload_file_to_s3(filepath, file_obj.mimetype, bucket, bucket_prefix) + return f's3://{bucket}/{bucket_prefix}/{filename}' diff --git a/apps/api/src/hyp3_api/validation.py b/apps/api/src/hyp3_api/validation.py index 6218cd10c..8161040dd 100644 --- a/apps/api/src/hyp3_api/validation.py +++ b/apps/api/src/hyp3_api/validation.py @@ -6,9 +6,12 @@ import requests import yaml +from flask import Request +from jsonschema import Draft7Validator from shapely.geometry import MultiPolygon, Polygon, shape from hyp3_api import CMR_URL, multi_burst_validation +from hyp3_api.openapi import get_spec_yaml from hyp3_api.util import get_granules @@ -94,6 +97,70 @@ def _make_sure_granules_exist(granules: Iterable[str], granule_metadata: list[di raise ValidationError(f'Some requested scenes could not be found: {", ".join(not_found_granules)}') +def validate_job_parameters(request_dict: dict) -> None: + """Manual validation of job parameters for `/upload-job`. + + Args: + request_dict: Sanitized request dictionary to be validated + """ + api_spec_dict = get_spec_yaml(Path(__file__).parent / 'api-spec' / 'openapi-spec.yml') + job_parameters_by_title = { + x['title']: x for x in api_spec_dict['components']['schemas']['job']['properties']['job_parameters']['anyOf'] + } + + job_type = request_dict['job_type'] + job_parameter_schema = job_parameters_by_title.get(f'{job_type}Parameters') + + validator = Draft7Validator(job_parameter_schema) # type: ignore + errors = sorted(validator.iter_errors(request_dict['job_parameters']), key=lambda e: e.path) + + if errors: + raise ValidationError(errors[-1]) + + +def _check_for_missing_files(request_files: dict, file_spec: dict) -> None: + missing_files = [] + for key in file_spec.keys(): + if 'required' in file_spec[key].keys() and file_spec[key]['required']: + if key not in request_files.keys(): + missing_files.append(key) + if len(missing_files) > 0: + msg = f'Missing required file(s): {", ".join(missing_files)}' + raise ValidationError(msg) + + +def _check_for_irrelevant_files(request_files: dict, file_spec: dict, job_type: str) -> None: + for file_param in request_files.keys(): + if file_param not in file_spec.keys(): + msg = f'Invalid file included for {job_type} job type: ({file_param}) {request_files[file_param].filename}' + raise ValidationError(msg) + + +def _check_correct_file_type(request_files: dict, file_spec: dict) -> None: + for param, file_obj in request_files.items(): + filetype = file_obj.mimetype + allowed_types = file_spec[param]['allowed_types'] + if filetype not in allowed_types: + msg = f"Invalid file type for {param}: '{filetype}' is not one of {allowed_types}." + raise ValidationError(msg) + + +def validate_files(request: Request) -> None: + """Check that files have been included correctly in an `/upload-job` request.""" + job_type = request.form['job_type'] + job_spec_path = Path(__file__).parent / f'job_spec/{job_type}.yml' + + with Path.open(job_spec_path) as file: + job_spec = yaml.safe_load(file) + + file_spec = job_spec[job_type]['files'] + request_files = dict(request.files) + + _check_for_missing_files(request_files, file_spec) + _check_for_irrelevant_files(request_files, file_spec, job_type) + _check_correct_file_type(request_files, file_spec) + + def check_cmr_query_succeeded(job: dict, granule_metadata: list[dict]) -> None: if not granule_metadata: raise CmrError( diff --git a/apps/render_cf.py b/apps/render_cf.py index 2327c2d92..f92033f94 100644 --- a/apps/render_cf.py +++ b/apps/render_cf.py @@ -236,9 +236,11 @@ def render_costs(job_types: dict, cost_profile: str) -> None: def validate_job_spec(job_type: str, job_spec: dict) -> None: - expected_fields = sorted(['required_parameters', 'parameters', 'cost_profiles', 'validators', 'steps']) - actual_fields = sorted(job_spec.keys()) - if actual_fields != expected_fields: + expected_fields = {'required_parameters', 'parameters', 'cost_profiles', 'validators', 'steps'} + optional_fields = {'files'} + actual_fields = set(job_spec.keys()) + + if actual_fields - optional_fields != expected_fields: raise ValueError(f'{job_type} has fields {actual_fields} but should have {expected_fields}') if 'job_id' in job_spec['parameters']: diff --git a/job_spec/AK_FIRE_SAFE.yml b/job_spec/AK_FIRE_SAFE.yml index 33a56b2bd..fc9ee73f5 100644 --- a/job_spec/AK_FIRE_SAFE.yml +++ b/job_spec/AK_FIRE_SAFE.yml @@ -54,6 +54,22 @@ AK_FIRE_SAFE: description: Prefix to pull the fire detection products. type: string example: "txt" + ak_fire_safe_file: + api_schema: + description: S3 uri for the file + type: string + nullable: true + default: null + example: "" + files: + ak_fire_safe_file: + allowed_types: ["image/jpeg", "image/png"] + required: true + api_schema: + description: Test file to upload + type: string + format: binary + cost_profiles: DEFAULT: cost: 1.0 diff --git a/lib/dynamo/dynamo/jobs.py b/lib/dynamo/dynamo/jobs.py index 5a44b0040..118242c26 100644 --- a/lib/dynamo/dynamo/jobs.py +++ b/lib/dynamo/dynamo/jobs.py @@ -85,7 +85,7 @@ def _raise_for_application_status(application_status: str, user_id: str) -> None raise InvalidApplicationStatusError(user_id, application_status) -def _handle_content_bucket(job: dict) -> dict: +def handle_content_bucket(job: dict) -> dict: content_bucket = environ['CONTENT_BUCKET'] example_bucket = 'my-example-bucket' @@ -121,7 +121,6 @@ def _prepare_job_for_database( else: priority = min(round(remaining_credits - running_cost), 9999) prepared_job = { - 'job_id': str(uuid4()), 'user_id': user_id, 'status_code': 'PENDING', 'execution_started': False, @@ -129,7 +128,10 @@ def _prepare_job_for_database( 'priority': priority, **job, } - prepared_job = _handle_content_bucket(prepared_job) + # `/upload-job` creates a job_id and handles the bucket in the API handler + if 'job_id' not in prepared_job.keys(): + prepared_job['job_id'] = str(uuid4()) + prepared_job = handle_content_bucket(prepared_job) if 'job_type' in prepared_job: prepared_job['job_parameters'] = { **DEFAULT_PARAMS_BY_JOB_TYPE[prepared_job['job_type']],