diff --git a/CHANGELOG.md b/CHANGELOG.md index 6376587..5ea2e93 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## Unreleased +### Added + +- NGS360: a launcher running as a WES run (e.g. an AWS Batch job) finds its own run + through `WES_RUN_ID`: `get_current_task()` returns it, and `submit_task()` tags each + run it submits with `ParentRunId` (PR #132) +- NGS360: when `NGS360_AUTH_KEY_ID` names the API key in `NGS360_AUTH_TOKEN`, the key is + retired when the process exits, so a key minted for one launch does not outlive it + (PR #132) + +### Fixed + +- `PlatformFactory.detect_platform()` no longer logs credential values in the + environment dump it writes when no platform is detected (PR #132) + ## [v0.6.1](https://github.com/NGS360/PAML/releases/tag/v0.6.1) - 2026-09-21 ### Changed diff --git a/src/cwl_platform/__init__.py b/src/cwl_platform/__init__.py index ea6fdba..18e4f44 100644 --- a/src/cwl_platform/__init__.py +++ b/src/cwl_platform/__init__.py @@ -21,6 +21,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 ''' @@ -42,7 +62,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 ffe52e9..ede179c 100644 --- a/src/cwl_platform/ngs360_platform.py +++ b/src/cwl_platform/ngs360_platform.py @@ -3,8 +3,24 @@ 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 (or ngs360_ API key) of the user the + launcher runs as; sent to both WES and the NGS360 API + NGS360_AUTH_KEY_ID Id of that API key, when it was minted for this one + launch. The key is retired when the launcher exits, so + it does not outlive the run; its expiry covers a + launcher that is killed before it can retire it. + 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 atexit import os import json import logging @@ -86,6 +102,7 @@ def __init__(self, name): self.ngs360_endpoint = None self._ngs360_auth_config = {} + self._auth_key_id = None self.projects = {} # Map project names to project objects self.workflows = {} # Map workflow names to workflow objects @@ -98,6 +115,8 @@ def connect(self, **kwargs): :param kwargs: Connection parameters - api_endpoint: WES API endpoint URL - ngs360_auth_token: Authentication token for the WES API + - ngs360_auth_key_id: Id of the API key in ngs360_auth_token, to be + retired when this process exits """ self.connected = False @@ -137,6 +156,16 @@ def connect(self, **kwargs): if not self.ngs360_endpoint: raise ValueError("NGS360 API endpoint URL is required") + # Registered before the /auth/me probe so that a launcher which fails to + # connect still retires its key on the way out. + auth_key_id = kwargs.get( + "ngs360_auth_key_id", + os.environ.get("NGS360_AUTH_KEY_ID") + ) + if auth_key_id and not self._auth_key_id: + self._auth_key_id = auth_key_id + atexit.register(self.retire_auth_key) + # Test connection by getting current user info try: response = requests.get( @@ -163,6 +192,32 @@ def _auth_headers(self): headers["Authorization"] = f"Bearer {self._ngs360_auth_config['token']}" return headers + def retire_auth_key(self): + """ + Retire the API key this launcher was given for its run + + Runs at process exit when NGS360_AUTH_KEY_ID is set. The key authenticates + its own retirement. Failure is logged, not raised: the key's expiry still + bounds it, and an exit handler has no caller to report to. + + :return: True if the key was retired, False otherwise + """ + if not self._auth_key_id: + return False + try: + response = requests.delete( + f"{self.ngs360_endpoint}/api/v1/auth/api-keys/{self._auth_key_id}", + headers=self._auth_headers(), + timeout=30 + ) + response.raise_for_status() + except requests.RequestException as e: + self.logger.warning("Failed to retire API key %s: %s", self._auth_key_id, e) + return False + self.logger.info("Retired API key %s", self._auth_key_id) + self._auth_key_id = None + return True + def _make_request(self, method, path, **kwargs): """ Make a request to the WES API @@ -407,9 +462,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): """ @@ -654,16 +731,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 fb700db..a014ac9 100644 --- a/tests/test_ngs360_ga4gh_platform.py +++ b/tests/test_ngs360_ga4gh_platform.py @@ -5,11 +5,34 @@ import os import tempfile 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', + 'NGS360_AUTH_KEY_ID', + '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): ''' @@ -134,6 +157,74 @@ def test_connect_ngs360_failure(self, mock_request, mock_get): self.assertFalse(result) + @patch('atexit.register') + @patch('requests.get') + @patch('requests.request') + def test_connect_registers_auth_key_retirement(self, mock_request, mock_get, mock_register): + ''' + Test a per-launch API key is set to be retired at exit, once + ''' + mock_request.return_value.json.return_value = {} + mock_get.return_value.status_code = 200 + + platform = NGS360Platform('WES') + with hidden_launcher_env(): + for _ in range(2): + platform.connect( + api_endpoint='https://wes.example.com/ga4gh/wes/v1', + ngs360_auth_token='ngs360_key', + ngs360_endpoint='https://ngs360.example.com', + ngs360_auth_key_id='key-123' + ) + + mock_register.assert_called_once_with(platform.retire_auth_key) + + @patch('atexit.register') + @patch('requests.get') + @patch('requests.request') + def test_connect_without_auth_key_id_registers_nothing(self, mock_request, mock_get, mock_register): + ''' + Test a user's own token is never retired at exit + ''' + mock_request.return_value.json.return_value = {} + mock_get.return_value.status_code = 200 + + with hidden_launcher_env(): + NGS360Platform('WES').connect( + api_endpoint='https://wes.example.com/ga4gh/wes/v1', + ngs360_auth_token='test_token', + ngs360_endpoint='https://ngs360.example.com' + ) + + mock_register.assert_not_called() + + @patch('requests.delete') + def test_retire_auth_key(self, mock_delete): + ''' + Test the key retires itself through the API key endpoint + ''' + self.platform._auth_key_id = 'key-123' # pylint: disable=protected-access + + self.assertTrue(self.platform.retire_auth_key()) + + args, kwargs = mock_delete.call_args + self.assertEqual(args[0], 'https://ngs360.example.com/api/v1/auth/api-keys/key-123') + self.assertEqual(kwargs['headers']['Authorization'], 'Bearer test_token') + # Retiring twice is a no-op rather than a second call + self.assertFalse(self.platform.retire_auth_key()) + mock_delete.assert_called_once() + + @patch('requests.delete') + def test_retire_auth_key_failure_is_not_raised(self, mock_delete): + ''' + Test a failed retirement is logged, not raised, from the exit handler + ''' + self.platform._auth_key_id = 'key-123' # pylint: disable=protected-access + mock_delete.side_effect = requests.RequestException("unreachable") + + with self.assertLogs(level='WARNING'): + self.assertFalse(self.platform.retire_auth_key()) + @patch('requests.request') def test_submit_task(self, mock_request): ''' @@ -149,13 +240,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( @@ -182,6 +274,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_submit_task_writes_no_files(self, mock_request): ''' @@ -1031,19 +1152,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 0000000..54de5c8 --- /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', + 'ARVADOS_API_TOKEN', + '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', + 'NGS360_AUTH_TOKEN': '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('NGS360_AUTH_TOKEN: ', 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()