From 2384ce0e67a63ff8c21eb1a5b92e02a48b706244 Mon Sep 17 00:00:00 2001 From: vasques1 Date: Mon, 17 Aug 2026 22:26:09 -0400 Subject: [PATCH 1/5] feat(ngs360): accept a WES service key as a fallback credential connect() required NGS360_AUTH_TOKEN. That is fine for a launcher started from a user's session, but a launcher running as an AWS Batch job has no session to source a token from -- 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 (X-Internal-Service-Key plus X-On-Behalf-Of for the audit trail). The bearer token still 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. When authenticating with the service key, skip the /api/v1/auth/me probe: that endpoint validates a *user's* bearer token, so there is nothing for it to check. The service-info call already proved we can reach and authenticate against WES. Tests hide only the four launcher environment variables (via hidden_launcher_env) rather than clearing os.environ, so a variable left set in the developer's shell cannot change what the tests exercise while everything else in the environment is left alone. Co-Authored-By: Claude Opus 5 --- src/cwl_platform/ngs360_platform.py | 40 +++++++++- tests/test_ngs360_ga4gh_platform.py | 114 ++++++++++++++++++++++++++++ 2 files changed, 153 insertions(+), 1 deletion(-) diff --git a/src/cwl_platform/ngs360_platform.py b/src/cwl_platform/ngs360_platform.py index 10e44974..9e3e77ce 100644 --- a/src/cwl_platform/ngs360_platform.py +++ b/src/cwl_platform/ngs360_platform.py @@ -86,6 +86,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 +100,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 +117,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 +160,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 +211,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, diff --git a/tests/test_ngs360_ga4gh_platform.py b/tests/test_ngs360_ga4gh_platform.py index a8f7be4f..800b9b67 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): ''' From 8c7d4f5d65709d65b2ac7f47d685b1e4fee1d086 Mon Sep 17 00:00:00 2001 From: vasques1 Date: Mon, 17 Aug 2026 22:36:58 -0400 Subject: [PATCH 2/5] feat(ngs360): implement get_current_task from WES_RUN_ID get_current_task() returned None with the note that "WES API doesn't have a concept of current task". That was accurate when it was written: a run could not be identified from inside its own container. The corresponding WES service change makes a launcher execution a first-class WES run, so the launcher can be told its own run id -- WES_RUN_ID, the same shape as SevenBridges' TASK_ID -- and read its own run record back. This is what the RNA-Seq launcher needs to reach its own inputs (platform.get_task_input(current_task, "factors_file")). The inputs come from request.workflow_params, which is what the run was submitted with. Raises ValueError when WES_RUN_ID is unset, matching what the SevenBridges and Arvados platforms do for their equivalent variables -- a launcher that cannot find itself is misconfigured, not idle. test_current_task_and_users loses its assertIsNone half and becomes test_project_users; the current-task behaviour now has its own two tests. Co-Authored-By: Claude Opus 5 --- src/cwl_platform/ngs360_platform.py | 28 +++++++++++++-- tests/test_ngs360_ga4gh_platform.py | 53 +++++++++++++++++++++++++---- 2 files changed, 72 insertions(+), 9 deletions(-) diff --git a/src/cwl_platform/ngs360_platform.py b/src/cwl_platform/ngs360_platform.py index 9e3e77ce..5a5f0cc0 100644 --- a/src/cwl_platform/ngs360_platform.py +++ b/src/cwl_platform/ngs360_platform.py @@ -437,9 +437,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): """ diff --git a/tests/test_ngs360_ga4gh_platform.py b/tests/test_ngs360_ga4gh_platform.py index 800b9b67..1d1d9b7d 100644 --- a/tests/test_ngs360_ga4gh_platform.py +++ b/tests/test_ngs360_ga4gh_platform.py @@ -1039,19 +1039,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 From 10f8809cead91dc04a9fa5ade4131d98d1ad7032 Mon Sep 17 00:00:00 2001 From: vasques1 Date: Mon, 17 Aug 2026 22:39:08 -0400 Subject: [PATCH 3/5] feat(ngs360): tag submitted tasks with the launcher's ParentRunId Nothing recorded which launcher execution submitted a given run, so a launcher's progress could only be answered by asking the launcher. Adding ParentRunId to the tags submit_task already sends -- ProjectId and TaskName -- links each child back to the launcher run that created it. WES promotes ParentRunId into an indexed column, so the service can roll up launcher progress from persisted rows instead of the launcher exposing a status endpoint, and a restarted launcher can find the work it already submitted. Only set when WES_RUN_ID is present, so a launcher run outside WES (a developer's shell, an external platform) submits exactly the tags it does today. test_submit_task hides the launcher variables so its exact-tags assertion does not depend on the developer's shell. Co-Authored-By: Claude Opus 5 --- src/cwl_platform/ngs360_platform.py | 17 ++++++++--- tests/test_ngs360_ga4gh_platform.py | 44 ++++++++++++++++++++++++----- 2 files changed, 50 insertions(+), 11 deletions(-) diff --git a/src/cwl_platform/ngs360_platform.py b/src/cwl_platform/ngs360_platform.py index 5a5f0cc0..c226efbe 100644 --- a/src/cwl_platform/ngs360_platform.py +++ b/src/cwl_platform/ngs360_platform.py @@ -699,16 +699,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 1d1d9b7d..186b95e1 100644 --- a/tests/test_ngs360_ga4gh_platform.py +++ b/tests/test_ngs360_ga4gh_platform.py @@ -261,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( @@ -294,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): ''' From 7229a62130f198b005cf9c776d747856ac4f5f99 Mon Sep 17 00:00:00 2001 From: vasques1 Date: Mon, 17 Aug 2026 22:42:42 -0400 Subject: [PATCH 4/5] fix(factory): redact credentials in detect_platform's environment dump When platform detection fails, PlatformFactory logs every environment variable so whoever is debugging can see what was set. That dump goes wherever the container's logs go -- CloudWatch for a launcher running as an AWS Batch job -- which is a wider audience than whoever provisioned the credential, and the launcher's environment now carries WES_SERVICE_KEY alongside the pre-existing NGS360_AUTH_TOKEN and SB_AUTH_TOKEN. Mask values whose variable name contains TOKEN, KEY, SECRET, or PASSWORD, keeping the character count: "set but empty" and "set to a truncated value" are both real misconfigurations, and the dump exists to tell them apart. Everything else is logged unchanged. Adds tests/test_platform_factory.py -- the factory had no test of its own. Co-Authored-By: Claude Opus 5 --- src/cwl_platform/__init__.py | 22 ++++++++++- tests/test_platform_factory.py | 71 ++++++++++++++++++++++++++++++++++ 2 files changed, 92 insertions(+), 1 deletion(-) create mode 100644 tests/test_platform_factory.py 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/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() From 9ba67608046baaeab6c82f4349526c66b1130773 Mon Sep 17 00:00:00 2001 From: vasques1 Date: Mon, 17 Aug 2026 22:45:52 -0400 Subject: [PATCH 5/5] docs(ngs360): document the launcher environment contract The module reads six environment variables across connect(), get_current_task() and submit_task(), so the full contract was only discoverable by reading the whole file. Whoever writes the AWS Batch job definition -- or debugs a launcher that starts and cannot find itself -- needs it in one place. Co-Authored-By: Claude Opus 5 --- src/cwl_platform/ngs360_platform.py | 14 ++++++++++++++ 1 file changed, 14 insertions(+) diff --git a/src/cwl_platform/ngs360_platform.py b/src/cwl_platform/ngs360_platform.py index c226efbe..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