diff --git a/src/cwl_platform/__init__.py b/src/cwl_platform/__init__.py index 21be0147..a4d7f7f4 100644 --- a/src/cwl_platform/__init__.py +++ b/src/cwl_platform/__init__.py @@ -15,6 +15,26 @@ 'NGS360': NGS360Platform } +# Substrings that mark an environment variable as carrying a credential. The +# environment dump below goes to whatever collects the container's logs, which +# is a wider audience than whoever provisioned the credential. +_SECRET_NAME_MARKERS = ('TOKEN', 'KEY', 'SECRET', 'PASSWORD') + + +def _redact(name, value): + ''' + Return an environment variable's value, masked if the name marks it secret + + :param name: Environment variable name + :param value: Environment variable value + :return: The value, or a mask of it + ''' + if any(marker in name.upper() for marker in _SECRET_NAME_MARKERS): + # Length is kept because "set but empty" and "set to a truncated value" + # are both real misconfigurations worth telling apart in a log. + return f"" + return value + class PlatformFactory(): ''' PlatformFactory ''' @@ -36,7 +56,7 @@ def detect_platform(self): # print out environment variables and raise an error logging.info("Environment Variables:") for name, value in os.environ.items(): - logging.info("%s: %s", name, value) + logging.info("%s: %s", name, _redact(name, value)) raise ValueError("Unable to detect platform") diff --git a/src/cwl_platform/ngs360_platform.py b/src/cwl_platform/ngs360_platform.py index 10e44974..db4a146b 100644 --- a/src/cwl_platform/ngs360_platform.py +++ b/src/cwl_platform/ngs360_platform.py @@ -3,6 +3,20 @@ This module implements the Platform abstract base class using the GA4GH Workflow Execution Service (WES) API. + +Environment contract, set by whoever starts the launcher container: + + WES_API_ENDPOINT WES API base URL, including the /ga4gh/wes/v1 prefix + NGS360_API_ENDPOINT NGS360 API base URL (projects, file upload) + NGS360_AUTH_TOKEN Bearer token for a user; preferred when available + WES_SERVICE_KEY Shared service key, used only when no bearer token is + set -- for headless runs with no user session to source + a token from. See connect() for the trust implications. + WES_ON_BEHALF_OF User the service key is acting for; audit trail only + WES_RUN_ID This launcher's own WES run id, when the launcher is + itself running as a WES run (e.g. as an AWS Batch job). + Makes get_current_task() work and links every task this + launcher submits back to it. """ import os @@ -86,6 +100,8 @@ def __init__(self, name): self.ngs360_endpoint = None self._ngs360_auth_config = {} + self._wes_service_key = None + self._wes_on_behalf_of = None self.projects = {} # Map project names to project objects self.workflows = {} # Map workflow names to workflow objects @@ -98,6 +114,9 @@ def connect(self, **kwargs): :param kwargs: Connection parameters - api_endpoint: WES API endpoint URL - ngs360_auth_token: Authentication token for the WES API + - wes_service_key: Shared service key, used only when no + ngs360_auth_token is available + - wes_on_behalf_of: User the service key is acting for """ self.connected = False @@ -112,10 +131,28 @@ def connect(self, **kwargs): "ngs360_auth_token", os.environ.get("NGS360_AUTH_TOKEN") ) + # The service key is a fallback for headless runs: a launcher running as an + # AWS Batch job has no user session to source a token from, so whoever + # submits the job has to inject a credential, and the service key is the + # one WES already provisions for callers acting without a user session. + # A bearer token wins when both are set: it names one real user and is + # scoped to them, whereas a holder of the service key is trusted to + # assert whatever identity it likes, so injecting the key into a + # container widens WES's trust boundary to that container. + service_key = kwargs.get( + "wes_service_key", + os.environ.get("WES_SERVICE_KEY") + ) if auth_token: self._ngs360_auth_config['token'] = auth_token + elif service_key: + self._wes_service_key = service_key + self._wes_on_behalf_of = kwargs.get( + "wes_on_behalf_of", + os.environ.get("WES_ON_BEHALF_OF") + ) else: - raise ValueError("NGS360 AUTH TOKEN is required") + raise ValueError("NGS360 AUTH TOKEN or WES SERVICE KEY is required") # Test connection by getting service info try: @@ -137,6 +174,17 @@ def connect(self, **kwargs): if not self.ngs360_endpoint: raise ValueError("NGS360 API endpoint URL is required") + # /api/v1/auth/me validates a user's bearer token, so there is nothing to + # probe when authenticating with the service key. The service-info call + # above already proved we can reach and authenticate against WES. + if self._wes_service_key: + self.logger.info( + "Authenticated to WES with a service key on behalf of '%s'", + self._wes_on_behalf_of or "(unspecified)" + ) + self.connected = True + return self.connected + # Test connection by getting current user info try: response = requests.get( @@ -177,6 +225,10 @@ def _make_request(self, method, path, **kwargs): if 'token' in self._ngs360_auth_config: headers["Authorization"] = f"Bearer {self._ngs360_auth_config['token']}" + elif self._wes_service_key: + headers["X-Internal-Service-Key"] = self._wes_service_key + if self._wes_on_behalf_of: + headers["X-On-Behalf-Of"] = self._wes_on_behalf_of response = requests.request( method=method, @@ -399,9 +451,31 @@ def get_current_task(self): """ Get the current task - Note: WES API doesn't have a concept of current task, so this returns None - """ - return None + A launcher that is itself a WES run -- e.g. one running as an AWS Batch + job -- is told its own run id through WES_RUN_ID, the same way the + SevenBridges platform uses TASK_ID. Its own run record is where the + launcher reads back the inputs it was started with. + + :return: WESTask object for this launcher's own run + :raises ValueError: If WES_RUN_ID is not set + """ + run_id = os.environ.get("WES_RUN_ID") + if not run_id: + raise ValueError("ERROR: Environment variable WES_RUN_ID not set.") + self.logger.info("WES_RUN_ID: %s", run_id) + + response = self._make_request("GET", f"runs/{run_id}") + request = response.get("request") or {} + + return WESTask( + run_id=run_id, + name=response.get("name") or "", + state=self.STATE_MAP.get(response.get("state", "UNKNOWN"), "Unknown"), + # workflow_params are what the launcher was submitted with, which is + # what get_task_input reads back. + inputs=request.get("workflow_params") or {}, + outputs=response.get("outputs") or {}, + ) def get_task_cost(self, task): """ @@ -639,16 +713,25 @@ def submit_task(self, name, project, workflow, parameters, execution_settings=No else: files = None + tags = { + "ProjectId": project["project_id"], + "TaskName": name, + } + # When this launcher is itself a WES run, link what it submits back to it. + # WES promotes ParentRunId into an indexed column, which is what lets it + # roll up launcher progress without the launcher reporting its own status, + # and what lets a restarted launcher find work it already submitted. + launcher_run_id = os.environ.get("WES_RUN_ID") + if launcher_run_id: + tags["ParentRunId"] = launcher_run_id + # Prepare the request data data = { "workflow_params": json.dumps(parameters), "workflow_type": workflow_type, "workflow_type_version": workflow_type_version, "workflow_url": workflow_url, - "tags": json.dumps({ - "ProjectId": project["project_id"], - "TaskName": name, - }), + "tags": json.dumps(tags), "workflow_engine_parameters": json.dumps( workflow_engine_parameters ), diff --git a/tests/test_ngs360_ga4gh_platform.py b/tests/test_ngs360_ga4gh_platform.py index a8f7be4f..186b95e1 100644 --- a/tests/test_ngs360_ga4gh_platform.py +++ b/tests/test_ngs360_ga4gh_platform.py @@ -2,12 +2,37 @@ Test WES Platform implementation ''' import json +import os import unittest +from contextlib import contextmanager from unittest.mock import patch, MagicMock import requests from cwl_platform.ngs360_platform import NGS360Platform, WESTask +# connect() and submit_task() read these as fallbacks, so one left set in the +# shell running the tests would change what those tests exercise. +LAUNCHER_ENV_VARS = ( + 'NGS360_AUTH_TOKEN', + 'WES_SERVICE_KEY', + 'WES_ON_BEHALF_OF', + 'WES_RUN_ID', +) + + +@contextmanager +def hidden_launcher_env(): + ''' + Hide the launcher's environment variables for the duration of a test + + Removes only the variables in LAUNCHER_ENV_VARS and restores the + environment on exit; everything else is left alone. + ''' + with patch.dict('os.environ'): + for var in LAUNCHER_ENV_VARS: + os.environ.pop(var, None) + yield + class TestNGS360Platform(unittest.TestCase): ''' @@ -90,6 +115,95 @@ def test_connect(self, mock_request, mock_get): _, kwargs = mock_request.call_args self.assertEqual(kwargs['headers']['Authorization'], 'Bearer test_token') + @patch('requests.get') + @patch('requests.request') + def test_connect_with_service_key(self, mock_request, mock_get): + ''' + Test connect method with a service key instead of a bearer token + ''' + # Mock WES API response + mock_response = MagicMock() + mock_response.json.return_value = { + 'workflow_type_versions': { + 'CWL': {'workflow_type_version': ['v1.0']} + } + } + mock_request.return_value = mock_response + + platform = NGS360Platform('WES') + with hidden_launcher_env(): + result = platform.connect( + api_endpoint='https://wes.example.com/ga4gh/wes/v1', + ngs360_endpoint='https://ngs360.example.com', + wes_service_key='test_service_key', + wes_on_behalf_of='someuser' + ) + self.assertTrue(result) + + # The service key is not a user token, so /auth/me is not probed + mock_get.assert_not_called() + + # Verify the service key headers are sent instead of a bearer token + _, kwargs = mock_request.call_args + self.assertEqual(kwargs['headers']['X-Internal-Service-Key'], 'test_service_key') + self.assertEqual(kwargs['headers']['X-On-Behalf-Of'], 'someuser') + self.assertNotIn('Authorization', kwargs['headers']) + + @patch('requests.get') + @patch('requests.request') + def test_connect_prefers_token_over_service_key(self, mock_request, mock_get): + ''' + Test connect method uses the bearer token when both credentials are given + ''' + # Mock WES API response + mock_response = MagicMock() + mock_response.json.return_value = { + 'workflow_type_versions': { + 'CWL': {'workflow_type_version': ['v1.0']} + } + } + mock_request.return_value = mock_response + + # Mock NGS360 API response + mock_ngs360_response = MagicMock() + mock_ngs360_response.status_code = 200 + mock_get.return_value = mock_ngs360_response + + platform = NGS360Platform('WES') + result = platform.connect( + api_endpoint='https://wes.example.com/ga4gh/wes/v1', + ngs360_endpoint='https://ngs360.example.com', + ngs360_auth_token='test_token', + wes_service_key='test_service_key' + ) + self.assertTrue(result) + + _, kwargs = mock_request.call_args + self.assertEqual(kwargs['headers']['Authorization'], 'Bearer test_token') + self.assertNotIn('X-Internal-Service-Key', kwargs['headers']) + + @patch('requests.request') + def test_connect_requires_a_credential(self, mock_request): + ''' + Test connect method raises when neither credential is available + ''' + # Mock WES API response + mock_response = MagicMock() + mock_response.json.return_value = { + 'workflow_type_versions': { + 'CWL': {'workflow_type_version': ['v1.0']} + } + } + mock_request.return_value = mock_response + + platform = NGS360Platform('WES') + with hidden_launcher_env(): + with self.assertRaises(ValueError): + platform.connect( + api_endpoint='https://wes.example.com/ga4gh/wes/v1', + ngs360_endpoint='https://ngs360.example.com' + ) + @patch('requests.request') def test_connect_wes_failure(self, mock_request): ''' @@ -147,13 +261,14 @@ def test_submit_task(self, mock_request): mock_request.return_value = mock_response # Test - task = self.platform.submit_task( - name='Test Task', - project={'project_id': "P-1234567", 'name': 'Test Project'}, - workflow=workflow_url, - parameters=workflow_parameters, - execution_settings={"use_spot_instance": False} - ) + with hidden_launcher_env(): + task = self.platform.submit_task( + name='Test Task', + project={'project_id': "P-1234567", 'name': 'Test Project'}, + workflow=workflow_url, + parameters=workflow_parameters, + execution_settings={"use_spot_instance": False} + ) # Verify GA4GH API request withing submit_task was made correctly mock_request.assert_called_with( @@ -180,6 +295,35 @@ def test_submit_task(self, mock_request): self.assertEqual(task.state, 'Queued') self.assertEqual(task.inputs, workflow_parameters) + @patch.dict('os.environ', {'WES_RUN_ID': 'launcher_run_id'}) + @patch('requests.request') + def test_submit_task_links_to_the_launcher_run(self, mock_request): + ''' + Test submit_task method tags submitted tasks with the launcher's run id + ''' + # Mock the GA4GH response for submit_task + mock_response = MagicMock() + mock_response.json.return_value = {'run_id': 'test_run_id'} + mock_request.return_value = mock_response + + self.platform.submit_task( + name='Test Task', + project={'project_id': "P-1234567", 'name': 'Test Project'}, + workflow='workflow_id', + parameters={'input': 'value'}, + execution_settings={"use_spot_instance": False} + ) + + _, kwargs = mock_request.call_args + self.assertEqual( + json.loads(kwargs['data']['tags']), + { + 'ProjectId': 'P-1234567', + 'TaskName': 'Test Task', + 'ParentRunId': 'launcher_run_id', + } + ) + @patch('requests.request') def test_get_task_state(self, mock_request): ''' @@ -925,19 +1069,60 @@ def test_cost_methods(self): # Test get_task_cost self.assertIsNone(self.platform.get_task_cost(task)) - def test_current_task_and_users(self): + def test_project_users(self): ''' - Test get_current_task and get_project_users methods + Test get_project_users method ''' project = {'project_id': 'test_project'} - # Test get_current_task - self.assertIsNone(self.platform.get_current_task()) - - # Test get_project_users result = self.platform.get_project_users(project) self.assertEqual(result, []) + @patch.dict('os.environ', {'WES_RUN_ID': 'launcher_run_id'}) + @patch('requests.request') + def test_get_current_task(self, mock_request): + ''' + Test get_current_task method returns the launcher's own run + ''' + # Mock the WES response for the launcher's own run + mock_response = MagicMock() + mock_response.json.return_value = { + 'run_id': 'launcher_run_id', + 'name': 'rnaseq-launcher', + 'state': 'RUNNING', + 'request': { + 'workflow_params': {'factors_file': 'ngs360://file-1'} + } + } + mock_request.return_value = mock_response + + task = self.platform.get_current_task() + + # Verify the launcher's own run was fetched + _, kwargs = mock_request.call_args + self.assertEqual( + kwargs['url'], + 'https://wes.example.com/ga4gh/wes/v1/runs/launcher_run_id' + ) + + self.assertEqual(task.run_id, 'launcher_run_id') + self.assertEqual(task.name, 'rnaseq-launcher') + self.assertEqual(task.state, 'Running') + + # The launcher reads its own inputs back through get_task_input + self.assertEqual( + self.platform.get_task_input(task, 'factors_file'), + 'ngs360://file-1' + ) + + def test_get_current_task_requires_run_id(self): + ''' + Test get_current_task method raises when WES_RUN_ID is not set + ''' + with hidden_launcher_env(): + with self.assertRaises(ValueError): + self.platform.get_current_task() + def test_detect_method(self): ''' Test detect class method diff --git a/tests/test_platform_factory.py b/tests/test_platform_factory.py new file mode 100644 index 00000000..994c24d6 --- /dev/null +++ b/tests/test_platform_factory.py @@ -0,0 +1,71 @@ +''' +Test PlatformFactory +''' +import unittest +from unittest.mock import patch + +from cwl_platform import ( + ArvadosPlatform, + NGS360Platform, + PlatformFactory, + SevenBridgesPlatform, + _redact, +) + + +class TestRedact(unittest.TestCase): + ''' + Test the environment variable redaction helper + ''' + + def test_secret_names_are_masked(self): + ''' Test values of credential-looking variables are masked ''' + for name in ( + 'NGS360_AUTH_TOKEN', + 'WES_SERVICE_KEY', + 'SB_AUTH_TOKEN', + 'AWS_SECRET_ACCESS_KEY', + 'DB_PASSWORD', + 'api_key' + ): + with self.subTest(name=name): + self.assertEqual(_redact(name, 'supersecret'), '') + + def test_other_names_are_left_alone(self): + ''' Test values that are not credentials are reported as-is ''' + self.assertEqual( + _redact('WES_API_ENDPOINT', 'https://wes.example.com'), + 'https://wes.example.com' + ) + + +class TestDetectPlatform(unittest.TestCase): + ''' + Test PlatformFactory.detect_platform + ''' + + def test_undetected_platform_dump_is_redacted(self): + ''' + Test the environment dump on failure does not log credentials + ''' + env = { + 'WES_API_ENDPOINT': 'https://wes.example.com', + 'WES_SERVICE_KEY': 'supersecret', + } + with patch.dict('os.environ', env), \ + patch.object(ArvadosPlatform, 'detect', return_value=False), \ + patch.object(SevenBridgesPlatform, 'detect', return_value=False), \ + patch.object(NGS360Platform, 'detect', return_value=False): + with self.assertLogs(level='INFO') as logs: + with self.assertRaises(ValueError): + PlatformFactory().detect_platform() + + output = '\n'.join(logs.output) + self.assertNotIn('supersecret', output) + self.assertIn('WES_SERVICE_KEY: ', output) + # Non-credential variables are still there to diagnose the failure with + self.assertIn('WES_API_ENDPOINT: https://wes.example.com', output) + + +if __name__ == '__main__': + unittest.main()