From 02ffbed7bee6e9809453e72f75b0f7f5c53862e7 Mon Sep 17 00:00:00 2001 From: BRAINIALL Team Date: Tue, 28 Jul 2026 09:20:26 -0300 Subject: [PATCH] Reject credential-bearing redirects --- CHANGES.md | 6 ++ actions/lib/brainiall_action.py | 16 +++++- pack.yaml | 2 +- scripts/validate_pack.py | 1 + tests/test_pdf_to_markdown.py | 16 +++--- tests/test_redirect_policy.py | 97 +++++++++++++++++++++++++++++++++ tests/test_text_to_speech.py | 18 +++--- 7 files changed, 135 insertions(+), 21 deletions(-) create mode 100644 tests/test_redirect_policy.py diff --git a/CHANGES.md b/CHANGES.md index 860ce74..68cc3d4 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -1,5 +1,11 @@ # Changelog +## 0.1.1 + +- Reject HTTP redirects so API credentials cannot be forwarded to another origin. +- Add an integration test proving a redirect target receives no request. +- Attribute GitHub installs accurately instead of implying Exchange publication. + ## 0.1.0 - Add an explicit text-to-speech action that writes private WAV output. diff --git a/actions/lib/brainiall_action.py b/actions/lib/brainiall_action.py index d830589..f7553a2 100644 --- a/actions/lib/brainiall_action.py +++ b/actions/lib/brainiall_action.py @@ -16,13 +16,23 @@ import socket import tempfile from urllib.error import HTTPError, URLError -from urllib.request import Request, urlopen +from urllib.request import HTTPRedirectHandler, Request, build_opener from st2common.runners.base_action import Action API_BASE_URL = "https://api.brainiall.com" -SOURCE = "stackstorm-exchange-c9" +SOURCE = "stackstorm-github-c9" + + +class _NoRedirectHandler(HTTPRedirectHandler): + """Fail closed instead of forwarding credentials to a redirect target.""" + + def redirect_request(self, req, fp, code, msg, headers, newurl): + return None + + +NO_REDIRECT_OPENER = build_opener(_NoRedirectHandler()) class BrainiallRequestError(RuntimeError): @@ -57,7 +67,7 @@ def _post(self, path, payload, content_type, timeout, max_response_bytes): ) try: - response = urlopen(request, timeout=timeout) + response = NO_REDIRECT_OPENER.open(request, timeout=timeout) except HTTPError as exc: exc.close() raise BrainiallRequestError( diff --git a/pack.yaml b/pack.yaml index 33d7c9f..323c8aa 100644 --- a/pack.yaml +++ b/pack.yaml @@ -8,7 +8,7 @@ keywords: - tts - pdf - markdown -version: 0.1.0 +version: 0.1.1 python_versions: - "3" author: BRAINIALL Team diff --git a/scripts/validate_pack.py b/scripts/validate_pack.py index a428eed..42159bb 100644 --- a/scripts/validate_pack.py +++ b/scripts/validate_pack.py @@ -41,6 +41,7 @@ def validate_files(): "actions/pdf_to_markdown.yaml", "tests/test_text_to_speech.py", "tests/test_pdf_to_markdown.py", + "tests/test_redirect_policy.py", ] missing = [path for path in required if not (ROOT / path).is_file()] if missing: diff --git a/tests/test_pdf_to_markdown.py b/tests/test_pdf_to_markdown.py index a7125ab..27a4ad9 100644 --- a/tests/test_pdf_to_markdown.py +++ b/tests/test_pdf_to_markdown.py @@ -43,12 +43,12 @@ def test_sends_bounded_multipart_and_writes_private_markdown(self): action = PdfToMarkdownAction({"api_key": TEST_KEY}) with mock.patch( - "lib.brainiall_action.urlopen", + "lib.brainiall_action.NO_REDIRECT_OPENER.open", return_value=FakeResponse(MARKDOWN_BYTES), - ) as urlopen_mock: + ) as open_mock: result = action.run(pdf_path, output_path, "1-2", "markdown") - request = urlopen_mock.call_args[0][0] + request = open_mock.call_args[0][0] headers = lower_headers(request) self.assertEqual( request.full_url, @@ -77,12 +77,12 @@ def test_rejects_bad_page_range_and_symlink_before_network(self): os.symlink(pdf_path, symlink_path) action = PdfToMarkdownAction({"api_key": TEST_KEY}) - with mock.patch("lib.brainiall_action.urlopen") as urlopen_mock: + with mock.patch("lib.brainiall_action.NO_REDIRECT_OPENER.open") as open_mock: with self.assertRaises(ValueError): action.run(pdf_path, output_path, "3-1") with self.assertRaises(ValueError): action.run(symlink_path, output_path) - urlopen_mock.assert_not_called() + open_mock.assert_not_called() def test_rejects_output_extension_before_metered_request(self): with tempfile.TemporaryDirectory() as directory: @@ -90,10 +90,10 @@ def test_rejects_output_extension_before_metered_request(self): output_path = os.path.join(directory, "output.txt") self._write(pdf_path, PDF_BYTES) action = PdfToMarkdownAction({"api_key": TEST_KEY}) - with mock.patch("lib.brainiall_action.urlopen") as urlopen_mock: + with mock.patch("lib.brainiall_action.NO_REDIRECT_OPENER.open") as open_mock: with self.assertRaises(ValueError): action.run(pdf_path, output_path) - urlopen_mock.assert_not_called() + open_mock.assert_not_called() def test_http_error_does_not_echo_key_or_replace_output(self): with tempfile.TemporaryDirectory() as directory: @@ -110,7 +110,7 @@ def test_http_error_does_not_echo_key_or_replace_output(self): ) action = PdfToMarkdownAction({"api_key": TEST_KEY}) - with mock.patch("lib.brainiall_action.urlopen", side_effect=error): + with mock.patch("lib.brainiall_action.NO_REDIRECT_OPENER.open", side_effect=error): with self.assertRaises(BrainiallRequestError) as raised: action.run(pdf_path, output_path) diff --git a/tests/test_redirect_policy.py b/tests/test_redirect_policy.py new file mode 100644 index 0000000..9585e2c --- /dev/null +++ b/tests/test_redirect_policy.py @@ -0,0 +1,97 @@ +# Copyright 2026 BRAINIALL +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import contextlib +import http.server +import socketserver +import threading +import unittest + +from common import install_st2_stub + + +install_st2_stub() + +from lib.brainiall_action import BrainiallAction, BrainiallRequestError # noqa: E402 + + +TEST_KEY = "redirect-policy-test-key" + + +class _RedirectSource(http.server.BaseHTTPRequestHandler): + target_url = None + + def do_POST(self): + self.send_response(302) + self.send_header("Location", self.target_url) + self.end_headers() + + def log_message(self, format_string, *args): + pass + + +class _RedirectTarget(http.server.BaseHTTPRequestHandler): + requests = [] + + def do_GET(self): + self.requests.append(dict(self.headers.items())) + self.send_response(200) + self.end_headers() + self.wfile.write(b"unexpected") + + def do_POST(self): + self.requests.append(dict(self.headers.items())) + self.send_response(200) + self.end_headers() + self.wfile.write(b"unexpected") + + def log_message(self, format_string, *args): + pass + + +@contextlib.contextmanager +def _server(handler): + instance = socketserver.TCPServer(("127.0.0.1", 0), handler) + thread = threading.Thread(target=instance.serve_forever) + thread.daemon = True + thread.start() + try: + yield instance + finally: + instance.shutdown() + instance.server_close() + thread.join() + + +class RedirectPolicyTest(unittest.TestCase): + def test_redirect_is_rejected_without_contacting_target(self): + _RedirectTarget.requests = [] + with _server(_RedirectTarget) as target: + _RedirectSource.target_url = "http://127.0.0.1:{0}/capture".format( + target.server_address[1] + ) + with _server(_RedirectSource) as source: + action = BrainiallAction({"api_key": TEST_KEY}) + source_url = "http://127.0.0.1:{0}".format(source.server_address[1]) + + with unittest.mock.patch("lib.brainiall_action.API_BASE_URL", source_url): + with self.assertRaises(BrainiallRequestError) as raised: + action._post("/redirect", b"{}", "application/json", 3, 100) + + self.assertEqual(str(raised.exception), "Brainiall API request failed with HTTP 302.") + self.assertEqual(_RedirectTarget.requests, []) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_text_to_speech.py b/tests/test_text_to_speech.py index 71d5c36..a52929a 100644 --- a/tests/test_text_to_speech.py +++ b/tests/test_text_to_speech.py @@ -42,19 +42,19 @@ def test_writes_private_wav_and_returns_only_metadata(self): action = TextToSpeechAction({"api_key": TEST_KEY}) with mock.patch( - "lib.brainiall_action.urlopen", + "lib.brainiall_action.NO_REDIRECT_OPENER.open", return_value=FakeResponse(WAV_BYTES), - ) as urlopen_mock: + ) as open_mock: result = action.run(TEST_TEXT, output_path, "af_heart", 1.25) - request = urlopen_mock.call_args[0][0] + request = open_mock.call_args[0][0] headers = lower_headers(request) request_json = json.loads(request.data.decode("utf-8")) self.assertEqual(request.full_url, "https://api.brainiall.com/v1/tts/synthesize") self.assertEqual(headers["authorization"], "Bearer " + TEST_KEY) self.assertEqual(headers["ocp-apim-subscription-key"], TEST_KEY) - self.assertEqual(headers["x-brainiall-source"], "stackstorm-exchange-c9") + self.assertEqual(headers["x-brainiall-source"], "stackstorm-github-c9") self.assertEqual(request_json["text"], TEST_TEXT) self.assertNotIn(TEST_KEY.encode("utf-8"), request.data) self.assertNotIn(TEST_KEY, request.full_url) @@ -78,7 +78,7 @@ def test_http_error_does_not_echo_key_content_or_replace_output(self): ) action = TextToSpeechAction({"api_key": TEST_KEY}) - with mock.patch("lib.brainiall_action.urlopen", side_effect=error): + with mock.patch("lib.brainiall_action.NO_REDIRECT_OPENER.open", side_effect=error): with self.assertRaises(BrainiallRequestError) as raised: action.run(TEST_TEXT, output_path) @@ -91,19 +91,19 @@ def test_rejects_invalid_voice_before_network(self): with tempfile.TemporaryDirectory() as directory: output_path = os.path.join(directory, "speech.wav") action = TextToSpeechAction({"api_key": TEST_KEY}) - with mock.patch("lib.brainiall_action.urlopen") as urlopen_mock: + with mock.patch("lib.brainiall_action.NO_REDIRECT_OPENER.open") as open_mock: with self.assertRaises(ValueError): action.run(TEST_TEXT, output_path, "voice\nheader") - urlopen_mock.assert_not_called() + open_mock.assert_not_called() def test_rejects_invalid_output_path_before_metered_request(self): with tempfile.TemporaryDirectory() as directory: output_path = os.path.join(directory, "speech.txt") action = TextToSpeechAction({"api_key": TEST_KEY}) - with mock.patch("lib.brainiall_action.urlopen") as urlopen_mock: + with mock.patch("lib.brainiall_action.NO_REDIRECT_OPENER.open") as open_mock: with self.assertRaises(ValueError): action.run(TEST_TEXT, output_path) - urlopen_mock.assert_not_called() + open_mock.assert_not_called() class PathReader(object):